private void testRecordReaderIncompleteDelta(boolean use130Format) throws Exception {
    final int BUCKET = 1;
    Configuration conf = new Configuration();
    OrcOutputFormat of = new OrcOutputFormat();
    FileSystem fs = FileSystem.getLocal(conf).getRaw();
    Path root = new Path(tmpDir, "testRecordReaderIncompleteDelta").makeQualified(fs);
    fs.delete(root, true);
    ObjectInspector inspector;
    synchronized (TestOrcFile.class) {
      inspector = ObjectInspectorFactory.getReflectionObjectInspector
          (MyRow.class, ObjectInspectorFactory.ObjectInspectorOptions.JAVA);
    }

    // write a base
    AcidOutputFormat.Options options =
        new AcidOutputFormat.Options(conf)
            .writingBase(true).minimumWriteId(0).maximumWriteId(0)
            .bucket(BUCKET).inspector(inspector).filesystem(fs).finalDestination(root);
    if(!use130Format) {
      options.statementId(-1);
    }
    RecordUpdater ru = of.getRecordUpdater(root, options);
    String[] values= new String[]{"1", "2", "3", "4", "5"};
    for (String s : values) {
      ru.insert(0, new MyRow(s));
    }
    ru.close(false);

    // write a delta
    options.writingBase(false).minimumWriteId(10)
        .maximumWriteId(19);
    ru = of.getRecordUpdater(root, options);
    values = new String[]{"6", "7", "8"};
    for (String s : values) {
      ru.insert(1, new MyRow(s));
    }
    InputFormat inf = new OrcInputFormat();
    JobConf job = new JobConf();
    job.set(ValidTxnList.VALID_TXNS_KEY,
        new ValidReadTxnList(new long[0], new BitSet(), 1000, Long.MAX_VALUE).writeToString());
    job.set("mapred.input.dir", root.toString());
    job.set("bucket_count", "2");
    job.set(IOConstants.SCHEMA_EVOLUTION_COLUMNS, MyRow.getColumnNamesProperty());
    job.set(IOConstants.SCHEMA_EVOLUTION_COLUMNS_TYPES, MyRow.getColumnTypesProperty());
    AcidUtils.setAcidOperationalProperties(job, true, null);

    job.setBoolean(hive_metastoreConstants.TABLE_IS_TRANSACTIONAL, true);

    // read the keys before the delta is flushed
    InputSplit[] splits = inf.getSplits(job, 1);
    //1 split since we only have 1 bucket file in base/.  delta is not flushed (committed) yet, i.e. empty
    assertEquals(1, splits.length);
    org.apache.hadoop.mapred.RecordReader<NullWritable, OrcStruct> rr =
        inf.getRecordReader(splits[0], job, Reporter.NULL);
    NullWritable key = rr.createKey();
    OrcStruct value = rr.createValue();
    System.out.println("Looking at split " + splits[0]);
    for(int i=1; i < 6; ++i) {
      System.out.println("Checking row " + i);
      assertTrue(rr.next(key, value));
      assertEquals(Integer.toString(i), value.getFieldValue(0).toString());
    }
    assertFalse(rr.next(key, value));

    ru.flush();
    ru.flush();
    values = new String[]{"9", "10"};
    for (String s : values) {
      ru.insert(3, new MyRow(s));
    }
    ru.flush();

    splits = inf.getSplits(job, 1);
    assertEquals(2, splits.length);
    Path sideFile = new Path(root + "/" + (use130Format ? AcidUtils.deltaSubdir(10,19,0) :
      AcidUtils.deltaSubdir(10,19)) + "/bucket_00001_flush_length");
    assertTrue(fs.exists(sideFile));
    assertEquals(32, fs.getFileStatus(sideFile).getLen());

    rr = inf.getRecordReader(splits[0], job, Reporter.NULL);
    for(int i=1; i <= 5; ++i) {
      assertTrue(rr.next(key, value));
      assertEquals(Integer.toString(i), value.getFieldValue(0).toString());
    }
    assertFalse(rr.next(key, value));

    rr = inf.getRecordReader(splits[1], job, Reporter.NULL);
    for(int i=6; i < 11; ++i) {
      assertTrue("i=" + i, rr.next(key, value));
      assertEquals(Integer.toString(i), value.getFieldValue(0).toString());
    }
    assertFalse(rr.next(key, value));
  }
@param use130Format true means use delta_0001_0001_0000 format, else delta_0001_00001
ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestOrcRawRecordMerger.java