@Override
  public InputSplit[] getSplits(JobConf job, int numSplits) throws IOException {
    Expression filter = icebergDataFilterFromHiveConf(job);
    if (filter != null) {
      job.set(InputFormatConfig.FILTER_EXPRESSION, SerializationUtil.serializeToBase64(filter));
    }

    job.set(InputFormatConfig.SELECTED_COLUMNS, job.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR, ""));
    job.set(InputFormatConfig.GROUPING_PARTITION_COLUMNS, job.get(TableScanDesc.GROUPING_PARTITION_COLUMNS, ""));
    job.setBoolean(InputFormatConfig.FETCH_VIRTUAL_COLUMNS,
            job.getBoolean(ColumnProjectionUtils.FETCH_VIRTUAL_COLUMNS_CONF_STR, false));
    job.set(InputFormatConfig.AS_OF_TIMESTAMP, job.get(TableScanDesc.AS_OF_TIMESTAMP, "-1"));
    job.set(InputFormatConfig.SNAPSHOT_ID, job.get(TableScanDesc.AS_OF_VERSION, "-1"));
    job.set(InputFormatConfig.SNAPSHOT_ID_INTERVAL_FROM, job.get(TableScanDesc.FROM_VERSION, "-1"));
    job.set(InputFormatConfig.OUTPUT_TABLE_SNAPSHOT_REF, job.get(TableScanDesc.SNAPSHOT_REF, ""));

    String location = job.get(InputFormatConfig.TABLE_LOCATION);
    int numBuckets = job.getInt(TableScanDesc.GROUPING_NUM_BUCKETS, -1);
    return Arrays.stream(super.getSplits(job, numSplits))
                 .map(split -> new HiveIcebergSplit((IcebergSplit) split, location, numBuckets))
                 .toArray(InputSplit[]::new);
  }
Converts Hive filter found in the passed job conf to an Iceberg filter expression. Then evaluates this
against the task's partition value producing a residual filter expression.
@param task - file scan task to evaluate the expression against
@param conf - job conf
@return - Iceberg residual filter expression
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/HiveIcebergInputFormat.java