@Override
    public void findDeletedRecords(ColumnVector[] cols, int size, BitSet selectedBitSet)
        throws IOException {
      if (!isDeleteRecordAvailable) {
        return;
      }

      long[] originalWriteId =
          cols[OrcRecordUpdater.ORIGINAL_WRITEID].isRepeating ? null
              : ((LongColumnVector) cols[OrcRecordUpdater.ORIGINAL_WRITEID]).vector;
      long[] bucket =
          cols[OrcRecordUpdater.BUCKET].isRepeating ? null
              : ((LongColumnVector) cols[OrcRecordUpdater.BUCKET]).vector;
      long[] rowId =
          cols[OrcRecordUpdater.ROW_ID].isRepeating ? null
              : ((LongColumnVector) cols[OrcRecordUpdater.ROW_ID]).vector;

      // The following repeatedX values will be set, if any of the columns are repeating.
      long repeatedOriginalWriteId = (originalWriteId != null) ? -1
          : ((LongColumnVector) cols[OrcRecordUpdater.ORIGINAL_WRITEID]).vector[0];
      long repeatedBucket = (bucket != null) ? -1
          : ((LongColumnVector) cols[OrcRecordUpdater.BUCKET]).vector[0];
      long repeatedRowId = (rowId != null) ? -1
          : ((LongColumnVector) cols[OrcRecordUpdater.ROW_ID]).vector[0];
      LongColumnVector currentWriteIdVector = (LongColumnVector) cols[OrcRecordUpdater.CURRENT_WRITEID];

      // Get the first valid row in the batch still available.
      int firstValidIndex = selectedBitSet.nextSetBit(0);
      if (firstValidIndex == -1) {
        return; // Everything in the batch has already been filtered out.
      }
      RecordIdentifier firstRecordIdInBatch =
          new RecordIdentifier(
              originalWriteId != null ? originalWriteId[firstValidIndex] : repeatedOriginalWriteId,
              bucket != null ? (int) bucket[firstValidIndex] : (int) repeatedBucket,
              rowId != null ? (int)  rowId[firstValidIndex] : repeatedRowId);

      // Get the last valid row in the batch still available.
      int lastValidIndex = selectedBitSet.previousSetBit(size - 1);
      RecordIdentifier lastRecordIdInBatch =
          new RecordIdentifier(
              originalWriteId != null ? originalWriteId[lastValidIndex] : repeatedOriginalWriteId,
              bucket != null ? (int) bucket[lastValidIndex] : (int) repeatedBucket,
              rowId != null ? (int)  rowId[lastValidIndex] : repeatedRowId);

      // We must iterate over all the delete records, until we find one record with
      // deleteRecord >= firstRecordInBatch or until we exhaust all the delete records.
      while (deleteRecordKey.compareRow(firstRecordIdInBatch) == -1) {
        isDeleteRecordAvailable = deleteRecords.next(deleteRecordKey, deleteRecordValue);
        if (!isDeleteRecordAvailable) return; // exhausted all delete records, return.
      }

      // If we are here, then we have established that firstRecordInBatch <= deleteRecord.
      // Now continue marking records which have been deleted until we reach the end of the batch
      // or we exhaust all the delete records.

      int currIndex = firstValidIndex;
      RecordIdentifier currRecordIdInBatch = new RecordIdentifier();
      while (isDeleteRecordAvailable && currIndex != -1 && currIndex <= lastValidIndex) {
        currRecordIdInBatch.setValues(
            (originalWriteId != null) ? originalWriteId[currIndex] : repeatedOriginalWriteId,
            (bucket != null) ? (int) bucket[currIndex] : (int) repeatedBucket,
            (rowId != null) ? rowId[currIndex] : repeatedRowId);

        if (deleteRecordKey.compareRow(currRecordIdInBatch) == 0) {
          // When deleteRecordId == currRecordIdInBatch, this record in the batch has been deleted.
          selectedBitSet.clear(currIndex);
          if (fetchDeletedRows) {
            currentWriteIdVector.vector[currIndex] = deleteRecordKey.getCurrentWriteId();
            cols[OrcRecordUpdater.CURRENT_WRITEID].isRepeating = false;
          }
          currIndex = selectedBitSet.nextSetBit(currIndex + 1); // Move to next valid index.
        } else if (deleteRecordKey.compareRow(currRecordIdInBatch) == 1) {
          // When deleteRecordId > currRecordIdInBatch, we have to move on to look at the
          // next record in the batch.
          // But before that, can we short-circuit and skip the entire batch itself
          // by checking if the deleteRecordId > lastRecordInBatch?
          if (deleteRecordKey.compareRow(lastRecordIdInBatch) == 1) {
            return; // Yay! We short-circuited, skip everything remaining in the batch and return.
          }
          currIndex = selectedBitSet.nextSetBit(currIndex + 1); // Move to next valid index.
        } else {
          // We have deleteRecordId < currRecordIdInBatch, we must now move on to find
          // next the larger deleteRecordId that can possibly match anything in the batch.
          isDeleteRecordAvailable = deleteRecords.next(deleteRecordKey, deleteRecordValue);
        }
      }
    }
An implementation for DeleteEventRegistry that opens the delete delta files all
at once, and then uses the sort-merge algorithm to maintain a sorted list of
delete events. This internally uses the OrcRawRecordMerger and maintains a constant
amount of memory usage, given the number of delete delta files. Therefore, this
implementation will be picked up when the memory pressure is high.

Don't bother to use KeyInterval from split here because since this doesn't
buffer delete events in memory.
ql/src/java/org/apache/hadoop/hive/ql/io/orc/VectorizedOrcAcidRowBatchReader.java