[Perseus]

Builds on #9; review that first.

What and why

With iceberg.dynamic_filter_pruning_enabled on, the scan already skips whole data files before opening them, but only by their identity partition values. A file without useful partition values, for example any file of an unpartitioned table, was never skipped, even when its column values clearly can't match the dynamic filter. On the unpartitioned taxi table every data file was read (9 → 9).

This change also uses each file's per-column lower/upper bounds from its manifest entry, so such files are skipped too.

How statistics are requested, and their cost

• Only when the flag is on, file listing calls TableScanBuilder::with_column_stats(names), in both places files are listed: the lazy work-unit feed (lazy/work_unit_feed.rs, at execution time) and eager planning (table_provider/scan.rs).
• The names are the scan's projected columns (all columns with no projection), restricted to primitive fields of the scanned snapshot's schema (column_stats_columns). These are the only columns a filter on the scan output can reference; metadata and nested columns have no statistics and would make the scan fail to build.
• Flag off: nothing is requested, so tasks carry no statistics and behaviour and task size are exactly as before.
• Cost with the flag on: each file task carries, per projected column, its lower/upper bound and value/null/NaN/size counts. That is a few tens of bytes per column per file, more for long string bounds (Iceberg truncates them, by default to 16 characters). It adds to coordinator memory for eagerly planned files and to each task sent to a worker.

How they reach workers

Tasks travel as rmp_serde named-struct encodings of iceberg-rust's FileScanTask, in both the lazy feed's SerializedFileScanTask and the eager codec's assigned files. The FileScanTask serde in the iceberg-rust revision we depend on already includes column_stats (with its bounds), so no change to it was needed. A new unit test (column_stats_survive_serialization) round-trips a task through the work-unit encoding and checks the statistics arrive unchanged.

Conversion rules (partition_pruning.rs)

file_statistics now fills each output column (matched by Iceberg field id) from the task's column statistics, then lets identity partition values replace them where present:

• Lower bound → min, upper bound → max, both inexact. Iceberg may truncate string and binary bounds; a truncated lower bound still bounds values from below and a truncated upper bound is rounded up, and DataFusion compares strings byte-wise as Iceberg does.
• A bound is used only when its type exactly matches the output Arrow type (iceberg-rust already promotes bounds to the scan schema's type); otherwise the column stays unknown. Binary bounds were added to the shared converter.
• Missing bounds (for example metrics mode none or counts) are unknown, never empty. A column that is all nulls may have no bounds; its exact null count alone describes it.
• Null count comes from the manifest; row count stays the task's record_count.
• Floats and doubles: Iceberg bounds exclude NaN, and DataFusion orders NaN above every value (and a negative NaN below every value), so float bounds are used only when the NaN count is known to be 0. If it is non-zero or unknown, both min and max stay unknown (this is a little stricter than leaving just the max unknown, to cover negative NaN). Zero bounds are widened: a lower bound of ±0.0 becomes -0.0 and an upper bound of ±0.0 becomes +0.0.
• Identity partition values are exact and take precedence over bounds for their column; other columns keep their bounds. Neither source narrows the other.
• Delete files only remove rows, so manifest statistics cover more rows than the scan returns, which can only keep files.

DataFusion's PrunableStatistics only reads exact min/max, so a small FileStatistics adapter implements PruningStatistics for one file: it returns min/max whether exact or inexact (pruning only needs bounds) and delegates null and row counts to PrunableStatistics, which still requires them to be exact.

Pruner path

Unchanged from #9: PartitionFilePruner snapshots the dynamic filters, rewrites a partitioned join's routing CASE as an OR, and builds a PruningPredicate, cached by the filters' generation. Only the statistics it evaluates against changed.

Correctness

Same single flag, off by default. Files are checked when the reader takes each task, with the filter's current value; nothing waits for it. Any conversion gap or pruning error keeps the file. The FilterExec above the scan still applies the filter, so results are identical with the flag on and off; the tests assert this.

Tests

• iceberg/tests/dynamic_filter_pruning.rs, single node, unpartitioned taxi table (one day per file), counting data-file reads with CountingStorageFactory, flag off vs on, identical results:
  • CollectLeft join, lazy and eager: 8 → 2 files read (1 build + 1 probe file instead of 1 + 7).
  • Partitioned join (routing CASE), lazy and eager: 9 → 4 files read (2 build + 2 probe instead of 2 + 7). The existing lazy test previously showed 9 → 9.
  • 2-worker distributed join with the flag on: result only.
• Unit tests for the conversion: int/date/string bounds, truncated string bounds (filters between and beyond them), type mismatch, missing bounds and all-null columns, floats with zero/non-zero/unknown NaN counts and ±0 widening, merging with partition values, and the requested column list.
• Round-trip test for the statistics on serialized file tasks.

Checks: cargo fmt --all, cargo clippy -p datafusion-distributed-iceberg -p datafusion-distributed-iceberg-test-tools --all-targets (no new warnings), cargo test -p datafusion-distributed-iceberg, cargo test --features integration --test dynamic_filtering (one known flaky test failed in a parallel run, aggregates::tests::remote_dynamic_filters; all 18 pass with --test-threads=1).