public static CloseableIterable<HiveBatchContext> reader(Table table, Path path, FileScanTask task,
      Map<Integer, ?> idToConstant, TaskAttemptContext context, Expression residual, Schema readSchema) {

    HiveDeleteFilter deleteFilter = null;
    Schema requiredSchema = readSchema;

    if (!task.deletes().isEmpty()) {
      deleteFilter = new HiveDeleteFilter(table.io(), task, table.schema(), prepareSchemaForDeleteFilter(readSchema),
          context.getConfiguration());
      requiredSchema = deleteFilter.requiredSchema();
      // TODO: take requiredSchema and adjust readColumnIds below accordingly for equality delete cases
      // and remove below limitation
      if (task.deletes().stream().anyMatch(d -> d.content() == FileContent.EQUALITY_DELETES)) {
        throw new UnsupportedOperationException("Vectorized reading with equality deletes is not supported yet.");
      }
    }


    // Tweaks on jobConf here are relevant for this task only, so we need to copy it first as context's conf is reused..
    JobConf job = new JobConf(context.getConfiguration());
    FileFormat format = task.file().format();
    Reporter reporter = ((MapredIcebergInputFormat.CompatibilityTaskAttemptContextImpl) context).getLegacyReporter();

    // Hive by default requires partition columns to be read too. This is not required for identity partition
    // columns, as we will add this as constants later.

    int[] partitionColIndices = null;
    Object[] partitionValues = null;
    PartitionSpec partitionSpec = task.spec();
    List<Integer> readColumnIds = ColumnProjectionUtils.getReadColumnIDs(job);

    if (!partitionSpec.isUnpartitioned()) {

      List<PartitionField> fields = partitionSpec.fields();
      List<Integer> partitionColIndicesList = Lists.newLinkedList();
      List<Object> partitionValuesList = Lists.newLinkedList();

      for (PartitionField partitionField : fields) {
        if (partitionField.transform().isIdentity()) {

          // Get columns in read schema order (which matches those of readColumnIds) to find partition column indices
          List<Types.NestedField> columns = task.spec().schema().columns();
          for (int colIdx = 0; colIdx < columns.size(); ++colIdx) {
            if (columns.get(colIdx).fieldId() == partitionField.sourceId()) {
              // Skip reading identity partition columns from source file...
              readColumnIds.remove((Integer) colIdx);

              // ...and use the corresponding constant value instead
              partitionColIndicesList.add(colIdx);
              partitionValuesList.add(idToConstant.get(partitionField.sourceId()));
              break;
            }
          }
        }
      }

      partitionColIndices = ArrayUtils.toPrimitive(partitionColIndicesList.toArray(new Integer[0]));
      partitionValues = partitionValuesList.toArray(new Object[0]);

      ColumnProjectionUtils.setReadColumns(job, readColumnIds);
    }

    try {

      long start = task.start();
      long length = task.length();

      // TODO: Iceberg currently does not track the last modification time of a file. Until that's added,
      // we need to set Long.MIN_VALUE as last modification time in the fileId triplet.
      SyntheticFileId fileId = new SyntheticFileId(path, task.file().fileSizeInBytes(), Long.MIN_VALUE);
      fileId.toJobConf(job);
      RecordReader<NullWritable, VectorizedRowBatch> recordReader = null;

      switch (format) {
        case ORC:
          recordReader = orcRecordReader(job, reporter, task, path, start, length, readColumnIds,
              fileId, residual, table.name());
          break;

        case PARQUET:
          recordReader = parquetRecordReader(job, reporter, task, path, start, length, fileId);
          break;
        default:
          throw new UnsupportedOperationException("Vectorized Hive reading unimplemented for format: " + format);
      }

      CloseableIterable<HiveBatchContext> vrbIterable =
          createVectorizedRowBatchIterable(recordReader, job, partitionColIndices, partitionValues, idToConstant);

      return deleteFilter != null ? deleteFilter.filterBatch(vrbIterable) : vrbIterable;

    } catch (IOException ioe) {
      throw new RuntimeException("Error creating vectorized record reader for " + path, ioe);
    }
  }
Utility class to create vectorized readers for Hive.
As per the file format of the task, it will create a matching vectorized record reader that is already implemented
in Hive. It will also do some tweaks on the produced vectors for Iceberg's use e.g. partition column handling.
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java