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.