private void testNewBaseAndDelta(boolean use130Format) throws Exception {
    final int BUCKET = 10;
    String[] values = new String[]{"first", "second", "third", "fourth",
                                   "fifth", "sixth", "seventh", "eighth",
                                   "ninth", "tenth"};
    Configuration conf = new Configuration();
    OrcOutputFormat of = new OrcOutputFormat();
    FileSystem fs = FileSystem.getLocal(conf);
    Path root = new Path(tmpDir, "testNewBaseAndDelta").makeQualified(fs);
    fs.delete(root, true);
    ObjectInspector inspector;
    synchronized (TestOrcFile.class) {
      inspector = ObjectInspectorFactory.getReflectionObjectInspector
          (MyRow.class, ObjectInspectorFactory.ObjectInspectorOptions.JAVA);
    }

    // write the base
    AcidOutputFormat.Options options = new AcidOutputFormat.Options(conf)
        .inspector(inspector).bucket(BUCKET).finalDestination(root);
    final int BUCKET_PROPERTY = BucketCodec.V1.encode(options);
    if(!use130Format) {
      options.statementId(-1);
    }
    RecordUpdater ru = of.getRecordUpdater(root,
        options.writingBase(true).maximumWriteId(100));
    for(String v: values) {
      ru.insert(0, new MyRow(v));
    }
    ru.close(false);

    // write a delta
    ru = of.getRecordUpdater(root, options.writingBase(false)
        .minimumWriteId(200).maximumWriteId(200).recordIdColumn(1));
    ru.update(200, new MyRow("update 1", 0, 0, BUCKET_PROPERTY));
    ru.update(200, new MyRow("update 2", 2, 0, BUCKET_PROPERTY));
    ru.update(200, new MyRow("update 3", 3, 0, BUCKET_PROPERTY));
    ru.delete(200, new MyRow("", 7, 0, BUCKET_PROPERTY));
    ru.delete(200, new MyRow("", 8, 0, BUCKET_PROPERTY));
    ru.close(false);

    conf.set(ValidTxnList.VALID_TXNS_KEY,
        new ValidReadTxnList(new long[0], new BitSet(), 1000, Long.MAX_VALUE).writeToString());
    ValidWriteIdList writeIdList = ValidReaderWriteIdList.fromValue("testNewBaseAndDelta:200:" + Long.MAX_VALUE);
    AcidDirectory directory = AcidUtils.getAcidState(fs, root, conf, writeIdList, null, use130Format);

    assertEquals(new Path(root, "base_0000100"), directory.getBaseDirectory());
    assertEquals(new Path(root, use130Format ?
        AcidUtils.deleteDeltaSubdir(200,200,0) : AcidUtils.deleteDeltaSubdir(200,200)),
        directory.getCurrentDirectories().get(0).getPath());
    assertEquals(new Path(root, use130Format ?
        AcidUtils.deltaSubdir(200,200,0) : AcidUtils.deltaSubdir(200,200)),
      directory.getCurrentDirectories().get(1).getPath());

    Path basePath = AcidUtils.createBucketFile(directory.getBaseDirectory(),
        BUCKET);
    Path deltaPath = AcidUtils.createBucketFile(directory.getCurrentDirectories().get(1).getPath(),
      BUCKET);
    Path deleteDeltaDir = directory.getCurrentDirectories().get(0).getPath();

    conf.set(IOConstants.SCHEMA_EVOLUTION_COLUMNS, MyRow.getColumnNamesProperty());
    conf.set(IOConstants.SCHEMA_EVOLUTION_COLUMNS_TYPES, MyRow.getColumnTypesProperty());
    AcidUtils.setAcidOperationalProperties(conf, true, null);
    conf.setBoolean(hive_metastoreConstants.TABLE_IS_TRANSACTIONAL, true);

    //the first "split" is for base/
    Reader baseReader = OrcFile.createReader(basePath,
        OrcFile.readerOptions(conf));
    OrcRawRecordMerger merger =
        new OrcRawRecordMerger(conf, true, baseReader, false, BUCKET,
            createMaximalTxnList(), new Reader.Options(),
            new Path[] {deleteDeltaDir}, new OrcRawRecordMerger.Options().isCompacting(false));
    assertNull(merger.getMinKey());
    assertNull(merger.getMaxKey());
    RecordIdentifier id = merger.createKey();
    OrcStruct event = merger.createValue();

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 0, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "second",
        new ReaderKey(0, BUCKET_PROPERTY, 1, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 2, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 3, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "fifth",
        new ReaderKey(0, BUCKET_PROPERTY, 4, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "sixth",
        new ReaderKey(0, BUCKET_PROPERTY, 5, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "seventh",
        new ReaderKey(0, BUCKET_PROPERTY, 6, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 7, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 8, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "tenth",
        new ReaderKey(0, BUCKET_PROPERTY, 9, 0), id, event);

    assertFalse(merger.next(id, event));
    merger.close();

    //second "split" is delta_200_200
    baseReader = OrcFile.createReader(deltaPath,
      OrcFile.readerOptions(conf));
    merger =
      new OrcRawRecordMerger(conf, true, baseReader, false, BUCKET,
        createMaximalTxnList(), new Reader.Options(),
        new Path[] {deleteDeltaDir}, new OrcRawRecordMerger.Options().isCompacting(false));
    assertNull(merger.getMinKey());
    assertNull(merger.getMaxKey());

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 0, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 2, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 3, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 7, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 8, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 1",
        new ReaderKey(200, BUCKET_PROPERTY, 0, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 2",
        new ReaderKey(200, BUCKET_PROPERTY, 1, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 3",
        new ReaderKey(200, BUCKET_PROPERTY, 2, 200), id, event);

    assertFalse(merger.next(id, event));
    merger.close();

    //now run as if it's a minor Compaction so we don't collapse events
    //here there is only 1 "split" since we only have data for 1 bucket
    merger =
      new OrcRawRecordMerger(conf, false, null, false, BUCKET,
        createMaximalTxnList(), new Reader.Options(),
        AcidUtils.getPaths(directory.getCurrentDirectories()), new OrcRawRecordMerger.Options().isCompacting(true));
    assertNull(merger.getMinKey());
    assertNull(merger.getMaxKey());

    assertTrue(merger.next(id, event));
    //minor comp, so we ignore 'base_0000100' files so all Deletes end up first since
    // they all modify primordial rows
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 0, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 2, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 3, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 7, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 8, 200), id, event);

    //data from delta_200_200
    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 1",
        new ReaderKey(200, BUCKET_PROPERTY, 0, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 2",
        new ReaderKey(200, BUCKET_PROPERTY, 1, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 3",
        new ReaderKey(200, BUCKET_PROPERTY, 2, 200), id, event);

    assertFalse(merger.next(id, event));
    merger.close();

    //now run as if it's a major Compaction so we collapse events
    //here there is only 1 "split" since we only have data for 1 bucket
    baseReader = OrcFile.createReader(basePath,
      OrcFile.readerOptions(conf));
    merger =
      new OrcRawRecordMerger(conf, true, null, false, BUCKET,
        createMaximalTxnList(), new Reader.Options(),
        AcidUtils.getPaths(directory.getCurrentDirectories()), new OrcRawRecordMerger.Options()
        .isCompacting(true).isMajorCompaction(true).baseDir(new Path(root, "base_0000100")));
    assertNull(merger.getMinKey());
    assertNull(merger.getMaxKey());

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 0, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "second",
        new ReaderKey(0, BUCKET_PROPERTY, 1, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 2, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 3, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "fifth",
        new ReaderKey(0, BUCKET_PROPERTY, 4, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "sixth",
        new ReaderKey(0, BUCKET_PROPERTY, 5, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "seventh",
        new ReaderKey(0, BUCKET_PROPERTY, 6, 0), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 7, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.DELETE_OPERATION, null,
        new ReaderKey(0, BUCKET_PROPERTY, 8, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "tenth",
        new ReaderKey(0, BUCKET_PROPERTY, 9, 0), id, event);

    //data from delta_200_200
    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 1",
        new ReaderKey(200, BUCKET_PROPERTY, 0, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 2",
        new ReaderKey(200, BUCKET_PROPERTY, 1, 200), id, event);

    assertTrue(merger.next(id, event));
    checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, "update 3",
        new ReaderKey(200, BUCKET_PROPERTY, 2, 200), id, event);

    assertFalse(merger.next(id, event));
    merger.close();

    // try ignoring the 200 transaction and make sure it works still
    ValidWriteIdList writeIds = ValidReaderWriteIdList.fromValue("testNewBaseAndDelta:2000:200:200");
    //again 1st split is for base/
    baseReader = OrcFile.createReader(basePath,
      OrcFile.readerOptions(conf));
    merger =
      new OrcRawRecordMerger(conf, false, baseReader, false, BUCKET,
              writeIds, new Reader.Options(),
        new Path[] {deleteDeltaDir}, new OrcRawRecordMerger.Options().isCompacting(false));

    assertNull(merger.getMinKey());
    assertNull(merger.getMaxKey());

    for(int i=0; i < values.length; ++i) {
      assertTrue(merger.next(id, event));
      LOG.info("id = " + id + "event = " + event);
      checkMergerRecord(OrcRecordUpdater.INSERT_OPERATION, values[i],
          new ReaderKey(0, BUCKET_PROPERTY, i, 0), id, event);
    }
    assertFalse(merger.next(id, event));
    merger.close();

    // 2nd split is for delta_200_200 which is filtered out entirely by "txns"
    baseReader = OrcFile.createReader(deltaPath,
      OrcFile.readerOptions(conf));
    merger =
      new OrcRawRecordMerger(conf, false, baseReader, false, BUCKET,
              writeIds, new Reader.Options(),
        new Path[] {deleteDeltaDir}, new OrcRawRecordMerger.Options().isCompacting(false));

    assertNull(merger.getMinKey());
    assertNull(merger.getMaxKey());
    assertFalse(merger.next(id, event));
    merger.close();
  }
Test the OrcRecordUpdater with the OrcRawRecordMerger when there is
a base and a delta.
@throws Exception
ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestOrcRawRecordMerger.java