public void testDeleteEventOriginalFiltering() throws Exception {
    boolean filterOn =
        HiveConf.getBoolVar(conf, HiveConf.ConfVars.FILTER_DELETE_EVENTS);

    conf.setBoolean(hive_metastoreConstants.TABLE_IS_TRANSACTIONAL, false);

    // Create 3 original files with 3 rows each
    Properties properties = new Properties();
    properties.setProperty("columns", DummyOriginalRow.getColumnNamesProperty());
    properties.setProperty("columns.types", DummyOriginalRow.getColumnTypesProperty());

    OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(properties, conf);
    writerOptions.inspector(originalInspector);

    Path testFilePath = new Path(root, "000000_0");
    Writer writer = OrcFile.createWriter(testFilePath, writerOptions);

    writer.addRow(new DummyOriginalRow(0));
    writer.addRow(new DummyOriginalRow(0));
    writer.addRow(new DummyOriginalRow(0));
    writer.close();

    testFilePath = new Path(root, "000000_0_copy_1");

    writer = OrcFile.createWriter(testFilePath, writerOptions);
    writer.addRow(new DummyOriginalRow(0));
    writer.addRow(new DummyOriginalRow(0));
    writer.addRow(new DummyOriginalRow(0));
    writer.close();

    testFilePath = new Path(root, "000000_0_copy_2");

    writer = OrcFile.createWriter(testFilePath, writerOptions);
    writer.addRow(new DummyOriginalRow(0));
    writer.addRow(new DummyOriginalRow(0));
    writer.addRow(new DummyOriginalRow(0));
    writer.close();

    conf.setBoolean(hive_metastoreConstants.TABLE_IS_TRANSACTIONAL, true);
    conf.set(ValidTxnList.VALID_TXNS_KEY,
        new ValidReadTxnList(new long[0], new BitSet(), 1000, Long.MAX_VALUE).writeToString());

    int bucket = 0;

    AcidOutputFormat.Options options = new AcidOutputFormat.Options(conf)
        .filesystem(fs)
        .bucket(bucket)
        .writingBase(false)
        .minimumWriteId(1)
        .maximumWriteId(1)
        .inspector(inspector)
        .reporter(Reporter.NULL)
        .recordIdColumn(1)
        .finalDestination(root);

    int bucketProperty = BucketCodec.V1.encode(options);

    RecordUpdater updater = new OrcRecordUpdater(root, options);

    //delete 1 row from each of the original files
    // Delete the last record in this split to test boundary conditions. It should not be present in the delete event
    // registry for the next split
    updater.delete(options.getMinimumWriteId(), new DummyRow(-1, 2, 0, bucket));
    // Delete the first record in this split to test boundary conditions. It should not be present in the delete event
    // registry for the previous split
    updater.delete(options.getMinimumWriteId(), new DummyRow(-1, 3, 0, bucket));
    updater.delete(options.getMinimumWriteId(), new DummyRow(-1, 7, 0, bucket));
    updater.close(false);

    //HWM is not important - just make sure deltas created above are read as if committed
    conf.set(ValidWriteIdList.VALID_WRITEIDS_KEY, "tbl:2:" + Long.MAX_VALUE + "::");

    // Set vector mode to true int the map work so that we recognize this as a vector mode execution during the split
    // generation. Without this we will not compute the offset for the synthetic row ids.
    MapWork mapWork = new MapWork();
    mapWork.setVectorMode(true);
    VectorizedRowBatchCtx vrbContext = new VectorizedRowBatchCtx();
    mapWork.setVectorizedRowBatchCtx(vrbContext);
    HiveConf.setVar(conf, HiveConf.ConfVars.PLAN, "//tmp");
    Utilities.setMapWork(conf, mapWork);

    // now we have 3 delete events total, but for each split we should only
    // load 1 into DeleteRegistry (if filtering is on)
    List<OrcInputFormat.SplitStrategy<?>> splitStrategies = getSplitStrategies();
    assertEquals(1, splitStrategies.size());
    List<OrcSplit> splits = ((OrcInputFormat.ACIDSplitStrategy)splitStrategies.get(0)).getSplits();

    assertEquals(3, splits.size());
    assertEquals(root.toUri().toString() + File.separator + "000000_0",
        splits.get(0).getPath().toUri().toString());
    assertTrue(splits.get(0).isOriginal());

    assertEquals(root.toUri().toString() + File.separator + "000000_0_copy_1",
        splits.get(1).getPath().toUri().toString());
    assertTrue(splits.get(1).isOriginal());

    assertEquals(root.toUri().toString() + File.separator + "000000_0_copy_2",
        splits.get(2).getPath().toUri().toString());
    assertTrue(splits.get(2).isOriginal());

    VectorizedOrcAcidRowBatchReader vectorizedReader =
        new VectorizedOrcAcidRowBatchReader(splits.get(0), conf, Reporter.NULL, vrbContext);
    ColumnizedDeleteEventRegistry deleteEventRegistry =
        (ColumnizedDeleteEventRegistry) vectorizedReader.getDeleteEventRegistry();
    assertEquals("number of delete events for stripe 1", filterOn ? 1 : 3, deleteEventRegistry.size());
    OrcRawRecordMerger.KeyInterval keyInterval = vectorizedReader.getKeyInterval();
    if(filterOn) {
      assertEquals(new OrcRawRecordMerger.KeyInterval(
              new RecordIdentifier(0, bucketProperty, 0),
              new RecordIdentifier(0, bucketProperty, 2)),
          keyInterval);
    } else {
      assertEquals(new OrcRawRecordMerger.KeyInterval(null, null), keyInterval);
    }

    vectorizedReader = new VectorizedOrcAcidRowBatchReader(splits.get(1), conf, Reporter.NULL, vrbContext);
    deleteEventRegistry = (ColumnizedDeleteEventRegistry) vectorizedReader.getDeleteEventRegistry();
    assertEquals("number of delete events for stripe 2", filterOn ? 1 : 3, deleteEventRegistry.size());
    keyInterval = vectorizedReader.getKeyInterval();
    if(filterOn) {
      assertEquals(new OrcRawRecordMerger.KeyInterval(
              new RecordIdentifier(0, bucketProperty, 3),
              new RecordIdentifier(0, bucketProperty, 5)),
          keyInterval);
    } else {
      assertEquals(new OrcRawRecordMerger.KeyInterval(null, null), keyInterval);
    }

    vectorizedReader = new VectorizedOrcAcidRowBatchReader(splits.get(2), conf, Reporter.NULL, vrbContext);
    deleteEventRegistry = (ColumnizedDeleteEventRegistry) vectorizedReader.getDeleteEventRegistry();
    assertEquals("number of delete events for stripe 3", filterOn ? 1 : 3, deleteEventRegistry.size());
    keyInterval = vectorizedReader.getKeyInterval();
    if(filterOn) {
      assertEquals(new OrcRawRecordMerger.KeyInterval(
          new RecordIdentifier(0, bucketProperty, 6),
          new RecordIdentifier(0, bucketProperty, 8)), keyInterval);
    } else {
      assertEquals(new OrcRawRecordMerger.KeyInterval(null, null), keyInterval);
    }
  }
Tests that we can figure out min/max ROW__ID for each split and then use
that to only load delete events between min/max.
This test doesn't actually check what is read - that is done more E2E
unit tests.
@throws Exception
ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestVectorizedOrcAcidRowBatchReader.java