@Override
  public RowReader<OrcStruct> getReader(InputSplit inputSplit,
                                        Options options)
                                            throws IOException {

    final OrcSplit split = (OrcSplit) inputSplit;

    // Retrieve the acidOperationalProperties for the table, initialized in HiveInputFormat.
    AcidUtils.AcidOperationalProperties acidOperationalProperties
            = AcidUtils.getAcidOperationalProperties(options.getConfiguration());
    if(!acidOperationalProperties.isSplitUpdate()) {
      throw new IllegalStateException("Expected SpliUpdate table: " + split.getPath());
    }

    Map<String, AcidInputFormat.DeltaMetaData> pathToDeltaMetaData = new HashMap<>();
    final Path[] deltas = VectorizedOrcAcidRowBatchReader.getDeleteDeltaDirsFromSplit(split, pathToDeltaMetaData);
    final Configuration conf = options.getConfiguration();

    final Reader reader = OrcInputFormat.createOrcReaderForSplit(conf, split);
    OrcRawRecordMerger.Options mergerOptions = new OrcRawRecordMerger.Options().isCompacting(false);
    mergerOptions.rootPath(split.getRootDir());
    mergerOptions.bucketPath(split.getPath());
    final int bucket;
    if (split.hasBase()) {
      AcidOutputFormat.Options acidIOOptions =
        AcidUtils.parseBaseOrDeltaBucketFilename(split.getPath(), conf);
      if(acidIOOptions.getBucketId() < 0) {
        LOG.warn("Can't determine bucket ID for " + split.getPath() + "; ignoring");
      }
      bucket = acidIOOptions.getBucketId();
    } else {
      bucket = (int) split.getStart();
      assert false : "We should never have a split w/o base in acid 2.0 for full acid: " + split.getPath();
    }
    //todo: createOptionsForReader() assumes it's !isOriginal.... why?
    final Reader.Options readOptions = OrcInputFormat.createOptionsForReader(conf);
    readOptions.range(split.getStart(), split.getLength());

    String txnString = conf.get(ValidWriteIdList.VALID_WRITEIDS_KEY);
    ValidWriteIdList validWriteIdList
            = (txnString == null) ? new ValidReaderWriteIdList() : ValidReaderWriteIdList.fromValue(txnString);
    if (LOG.isDebugEnabled()) {
      LOG.debug("getReader:: Read ValidWriteIdList: " + validWriteIdList.toString()
            + " isTransactionalTable: " + HiveConf.getBoolVar(conf, ConfVars.HIVE_TRANSACTIONAL_TABLE_SCAN));
      LOG.debug("Creating merger for {} and {}", split.getPath(), Arrays.toString(deltas));
    }
    boolean fetchDeletedRows = acidOperationalProperties.isFetchDeletedRows();

    Map<String, Integer> deltaToAttemptId = AcidUtils.getDeltaToAttemptIdMap(pathToDeltaMetaData, deltas, bucket);
    final OrcRawRecordMerger records;
    if (!fetchDeletedRows) {
      records = new OrcRawRecordMerger(conf, true, reader, split.isOriginal(), bucket,
              validWriteIdList, readOptions, deltas, mergerOptions, deltaToAttemptId);
    } else {
      records = new OrcRawRecordMerger(conf, true, reader, split.isOriginal(), bucket,
              validWriteIdList, readOptions, deltas, mergerOptions, deltaToAttemptId) {
        @Override
        protected boolean collapse(RecordIdentifier recordIdentifier) {
          ((ReaderKey) recordIdentifier).setValues(
                  prevKey.getCurrentWriteId(),
                  prevKey.getBucketProperty(),
                  prevKey.getRowId(),
                  prevKey.getCurrentWriteId(),
                  true);
          return false;
        }
      };
    }
    return new OrcRowReader(records, readOptions);
  }
Return a RecordReader that is compatible with the Hive 0.12 reader
with NullWritable for the key instead of RecordIdentifier.
ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java