Flink: Enable Cost Based Optimizaton in Flink's planner for batch jobs

[talatuyarer]

Implements SupportsStatisticReport in IcebergTableSource so that in batch execution mode the Flink planner receives table statistics computed from Iceberg metadata: the row count, and per-column null counts, min/max values, and NDV. These feed Flink's cost-based optimizations to filter selectivity estimation, join reordering, and broadcast-join selection. previously code ran on defaults because the Iceberg source reported no statistics at all.

All statistics come from metadata already maintained by Iceberg, no data files are read. For streaming reads, time-travel/incremental options, missing record counts, row-count overflow, and any exception I logged as WARN log and all report TableStats.UNKNOWN rather than failing or misleading the planner.

Because column statistics reporting is enabled by default, I measured the planning-time overhead with a local JMH benchmark to validate that default, I did not include for this PR. But If you want I can share.

Setup: metadata-only data files appended in ~1000-entry manifests, so manifest I/O is measured without writing actual data files; single-shot mode, since planning is a once-per-query cost:

files	row count only	full column stats
100	~10 µs	4 ms
1,000	~10 µs	8 ms
10,000	~10 µs	21 ms
100,000	~10 µs	106 ms
1,000,000	~10 µs	996 ms

The row-count-only path is constant time regardless of table size. Full column stats cost ~1 µs per data file single-digit milliseconds at typical table sizes, ~100 ms at 100K files, ~1 s at 1M files (warm-cache local FS, so this is manifest decode cost, a lower bound vs. object storage).

I consider this acceptable for a default configuration: tables where the cost is noticeable (100K+ files) are exactly the tables that benefit most from statistics-driven plans, and latency-sensitive users can opt out with table.exec.iceberg.report-column-statistics=false. I am also happy your feedbacks. I dont have any strong preferences.

[talatuyarer]

To assist with the review process, I have created a demo using Google’s public Iceberg datasets. By using Flink’s EXPLAIN command, I have demonstrated how this feature significantly improves query plans. For example, the optimizer was able to reduce the estimated scan count from 323,000,000 to 12,919 rows and successfully chose a broadcast join over a costly shuffle for large datasets. You are welcome to explore these results and test the implementation yourself. Table Stats Demo