Flink: Support aggregate push down for Iceberg table source in batch

[Guosmilesmile]

Summary

Adds aggregate push down to the Flink Iceberg source (IcebergTableSource). For batch queries that aggregate the whole table without GROUP BY or LIMIT, COUNT(*), COUNT(col), MAX(col) and MIN(col) can now be answered from file-level metrics alone, without reading any data files.

The feature is opt-in and disabled by default via a new Flink configuration:

SET table.exec.iceberg.aggregate-push-down-enabled = true;

Supported scenarios

• Batch (bounded) reads only; streaming reads never take this path.
• Query-wide aggregates without GROUP BY or LIMIT.
• COUNT(*), COUNT(col), MAX(col), MIN(col).

When push down is skipped (falls back to a regular scan)

• GROUP BY or LIMIT present.
• Streaming/unbounded reads.
• Any filter that is not guaranteed to select whole partitions (a file may contain rows the filter would remove).
• Row-level deletes present on planned files, or a file still needs row-level filtering (non-TRUE residual).
• Metrics configuration cannot produce the requested aggregate for a column (e.g. count-only mode for MIN/MAX, or truncated bounds on string/binary columns).
• Metadata tables; aggregates over types that cannot be derived from file metrics.

Alignment with Spark

Spark already supports aggregate push down via spark.sql.iceberg.aggregate-push-down.enabled . This change mirrors that design:

• The metrics-mode feasibility check is extracted into a shared, engine-agnostic helper in core (AggregatePushDownUtil.metricsModeSupportsAggregatePushDown), now used by both Spark 4.1 and Flink to avoid duplicated logic.
• The per-file validation (row-level deletes, residual, AggregateEvaluator.update) follows the same pattern as Spark's pushAggregation.

[wombatu-kun]

applyFilters now drops partition-aligned filters from the remaining set for every query, and RowDataFileScanTaskReader grew a wider read schema to compensate - both are always on regardless of the new flag, and neither is in the summary. Split the filter push-down change into its own PR so it can be reviewed and released independently of the opt-in aggregate feature.

[Guosmilesmile]

I’m planning to keep this as is for now and see what everyone thinks about the overall direction. If the PR is considered large, we can split it up later.

[wombatu-kun]

In the table source path the projection can only be missing a filter column when applyFilters already proved that filter selects whole partitions, and those files all have a TRUE residual, so the row filter can never drop a row. Consider keeping such filters out of the row filter so the extra columns are not read at all; follow-up, not a blocker.

[Guosmilesmile]

I'd suggest keeping the current logic as-is for now and handling it in a follow-up.

[wombatu-kun]

This is the only read option here declared without a connector.iceberg. ConfigOption, so the Flink config spelling that works for every neighbouring read option does nothing for this one. Was the table.exec.iceberg namespace a deliberate choice here?

[Guosmilesmile]

Yes, because this is more of an optimization at the execution layer.