@Test
  public void testAutoRollTransactionBatch() throws Exception {
    queryTable(driver, "drop table if exists default.streamingnobuckets");
    queryTable(driver, "create table default.streamingnobuckets (a string, b string) stored as orc " +
      "TBLPROPERTIES('transactional'='true')");
    queryTable(driver, "insert into default.streamingnobuckets values('foo','bar')");
    List<String> rs = queryTable(driver, "select * from default.streamingnobuckets");
    Assert.assertEquals(1, rs.size());
    Assert.assertEquals("foo\tbar", rs.get(0));
    StrictDelimitedInputWriter wr = StrictDelimitedInputWriter.newBuilder()
      .withFieldDelimiter(',')
      .build();
    HiveStreamingConnection connection = HiveStreamingConnection.newBuilder()
      .withDatabase("default")
      .withTable("streamingnobuckets")
      .withAgentInfo("UT_" + Thread.currentThread().getName())
      .withRecordWriter(wr)
      .withHiveConf(conf)
      .withTransactionBatchSize(2)
      .connect();

    connection.beginTransaction();
    connection.write("a1,b2".getBytes());
    connection.write("a3,b4".getBytes());
    connection.commitTransaction();
    connection.beginTransaction();
    connection.write("a5,b6".getBytes());
    connection.write("a7,b8".getBytes());
    connection.commitTransaction();
    // should have rolled over to next transaction batch
    connection.beginTransaction();
    connection.write("a9,b10".getBytes());
    connection.write("a11,b12".getBytes());
    connection.commitTransaction();
    connection.beginTransaction();
    connection.write("a13,b14".getBytes());
    connection.write("a15,b16".getBytes());
    connection.commitTransaction();
    connection.close();

    Assert.assertEquals("", 0, BucketCodec.determineVersion(536870912).decodeWriterId(536870912));
    rs = queryTable(driver, "select ROW__ID, a, b, INPUT__FILE__NAME from default.streamingnobuckets order by ROW__ID");

    Assert.assertTrue(rs.get(0), rs.get(0).startsWith("{\"writeid\":1,\"bucketid\":536870912,\"rowid\":0}\tfoo\tbar"));
    Assert.assertTrue(rs.get(0), rs.get(0).endsWith("streamingnobuckets/delta_0000001_0000001_0000/bucket_00000_0"));
    Assert.assertTrue(rs.get(1), rs.get(1).startsWith("{\"writeid\":2,\"bucketid\":536870912,\"rowid\":0}\ta1\tb2"));
    Assert.assertTrue(rs.get(1), rs.get(1).endsWith("streamingnobuckets/delta_0000002_0000003/bucket_00000"));
    Assert.assertTrue(rs.get(2), rs.get(2).startsWith("{\"writeid\":2,\"bucketid\":536870912,\"rowid\":1}\ta3\tb4"));
    Assert.assertTrue(rs.get(2), rs.get(2).endsWith("streamingnobuckets/delta_0000002_0000003/bucket_00000"));
    Assert.assertTrue(rs.get(3), rs.get(3).startsWith("{\"writeid\":3,\"bucketid\":536870912,\"rowid\":0}\ta5\tb6"));
    Assert.assertTrue(rs.get(3), rs.get(3).endsWith("streamingnobuckets/delta_0000002_0000003/bucket_00000"));
    Assert.assertTrue(rs.get(4), rs.get(4).startsWith("{\"writeid\":3,\"bucketid\":536870912,\"rowid\":1}\ta7\tb8"));
    Assert.assertTrue(rs.get(4), rs.get(4).endsWith("streamingnobuckets/delta_0000002_0000003/bucket_00000"));

    Assert.assertTrue(rs.get(5), rs.get(5).startsWith("{\"writeid\":4,\"bucketid\":536870912,\"rowid\":0}\ta9\tb10"));
    Assert.assertTrue(rs.get(5), rs.get(5).endsWith("streamingnobuckets/delta_0000004_0000005/bucket_00000"));
    Assert.assertTrue(rs.get(6), rs.get(6).startsWith("{\"writeid\":4,\"bucketid\":536870912,\"rowid\":1}\ta11\tb12"));
    Assert.assertTrue(rs.get(6), rs.get(6).endsWith("streamingnobuckets/delta_0000004_0000005/bucket_00000"));
    Assert.assertTrue(rs.get(7), rs.get(7).startsWith("{\"writeid\":5,\"bucketid\":536870912,\"rowid\":0}\ta13\tb14"));
    Assert.assertTrue(rs.get(7), rs.get(7).endsWith("streamingnobuckets/delta_0000004_0000005/bucket_00000"));
    Assert.assertTrue(rs.get(8), rs.get(8).startsWith("{\"writeid\":5,\"bucketid\":536870912,\"rowid\":1}\ta15\tb16"));
    Assert.assertTrue(rs.get(8), rs.get(8).endsWith("streamingnobuckets/delta_0000004_0000005/bucket_00000"));

    queryTable(driver, "update default.streamingnobuckets set a=0, b=0 where a='a7'");
    queryTable(driver, "delete from default.streamingnobuckets where a='a1'");
    queryTable(driver, "update default.streamingnobuckets set a=0, b=0 where a='a15'");
    queryTable(driver, "delete from default.streamingnobuckets where a='a9'");
    rs = queryTable(driver, "select a, b from default.streamingnobuckets order by a, b");
    int row = 0;
    Assert.assertEquals("at row=" + row, "0\t0", rs.get(row++));
    Assert.assertEquals("at row=" + row, "0\t0", rs.get(row++));
    Assert.assertEquals("at row=" + row, "a11\tb12", rs.get(row++));
    Assert.assertEquals("at row=" + row, "a13\tb14", rs.get(row++));
    Assert.assertEquals("at row=" + row, "a3\tb4", rs.get(row++));
    Assert.assertEquals("at row=" + row, "a5\tb6", rs.get(row++));
    Assert.assertEquals("at row=" + row, "foo\tbar", rs.get(row++));

    queryTable(driver, "alter table default.streamingnobuckets compact 'major'");
    runWorker(conf);
    rs = queryTable(driver, "select ROW__ID, a, b, INPUT__FILE__NAME from default.streamingnobuckets order by ROW__ID");

    Assert.assertTrue(rs.get(0), rs.get(0).startsWith("{\"writeid\":1,\"bucketid\":536870912,\"rowid\":0}\tfoo\tbar"));
    Assert.assertTrue(rs.get(0), rs.get(0).endsWith("streamingnobuckets/base_0000009_v0000028/bucket_00000"));
    Assert.assertTrue(rs.get(1), rs.get(1).startsWith("{\"writeid\":2,\"bucketid\":536870912,\"rowid\":1}\ta3\tb4"));
    Assert.assertTrue(rs.get(1), rs.get(1).endsWith("streamingnobuckets/base_0000009_v0000028/bucket_00000"));
    Assert.assertTrue(rs.get(2), rs.get(2).startsWith("{\"writeid\":3,\"bucketid\":536870912,\"rowid\":0}\ta5\tb6"));
    Assert.assertTrue(rs.get(2), rs.get(2).endsWith("streamingnobuckets/base_0000009_v0000028/bucket_00000"));
    Assert.assertTrue(rs.get(3), rs.get(3).startsWith("{\"writeid\":4,\"bucketid\":536870912,\"rowid\":1}\ta11\tb12"));
    Assert.assertTrue(rs.get(3), rs.get(3).endsWith("streamingnobuckets/base_0000009_v0000028/bucket_00000"));
    Assert.assertTrue(rs.get(4), rs.get(4).startsWith("{\"writeid\":5,\"bucketid\":536870912,\"rowid\":0}\ta13\tb14"));
    Assert.assertTrue(rs.get(4), rs.get(4).endsWith("streamingnobuckets/base_0000009_v0000028/bucket_00000"));
    Assert.assertTrue(rs.get(5), rs.get(5).startsWith("{\"writeid\":6,\"bucketid\":536936449,\"rowid\":0}\t0\t0"));
    Assert.assertTrue(rs.get(5), rs.get(5).endsWith("streamingnobuckets/base_0000009_v0000028/bucket_00001"));
  }
Test that streaming can write to unbucketed table.
streaming/src/test/org/apache/hive/streaming/TestStreaming.java