private List<InputSplit> planInputSplits(Table table, Configuration conf, ExecutorService workerPool) {
    List<InputSplit> splits = Lists.newArrayList();
    boolean applyResidual = !conf.getBoolean(InputFormatConfig.SKIP_RESIDUAL_FILTERING, false);
    InputFormatConfig.InMemoryDataModel model = conf.getEnum(InputFormatConfig.IN_MEMORY_DATA_MODEL,
        InputFormatConfig.InMemoryDataModel.GENERIC);

    long fromVersion = conf.getLong(InputFormatConfig.SNAPSHOT_ID_INTERVAL_FROM, -1);
    Scan<? extends Scan, FileScanTask, CombinedScanTask> scan;
    if (fromVersion != -1) {
      scan = applyConfig(conf, createIncrementalAppendScan(table, conf));
    } else {
      scan = applyConfig(conf, createTableScan(table, conf));
    }
    scan = scan.planWith(workerPool);

    boolean allowDataFilesWithinTableLocationOnly =
        conf.getBoolean(HiveConf.ConfVars.HIVE_ICEBERG_ALLOW_DATAFILES_IN_TABLE_LOCATION_ONLY.varname,
            HiveConf.ConfVars.HIVE_ICEBERG_ALLOW_DATAFILES_IN_TABLE_LOCATION_ONLY.defaultBoolVal);
    Path tableLocation = new Path(conf.get(InputFormatConfig.TABLE_LOCATION));

    String[] groupingPartitionColumns = conf.getStrings(InputFormatConfig.GROUPING_PARTITION_COLUMNS);
    generateInputSplits(scan, table, groupingPartitionColumns, taskGroup -> {
      if (applyResidual && (model == InputFormatConfig.InMemoryDataModel.HIVE ||
          model == InputFormatConfig.InMemoryDataModel.PIG)) {
        // TODO: We do not support residual evaluation for HIVE and PIG in memory data model yet
        checkResiduals(taskGroup);
      }
      if (allowDataFilesWithinTableLocationOnly) {
        validateFileLocations(taskGroup, tableLocation);
      }
      splits.add(new IcebergSplit(conf, taskGroup));
    });

    // If enabled, do not serialize FileIO hadoop config to decrease split size
    // However, do not skip serialization for metatable queries, because some metadata tasks cache the IO object and we
    // wouldn't be able to inject the config into these tasks on the deserializer-side, unlike for standard queries
    if (scan instanceof DataTableScan) {
      HiveIcebergStorageHandler.checkAndSkipIoConfigSerialization(conf, table);
    }

    return splits;
  }
Configures the {@code Job} to use the {@code IcebergInputFormat} and
returns a helper to add further configuration.

@param job the {@code Job} to configure
iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergInputFormat.java