[Perseus]

Builds on #5; review that first.

What and why

Iceberg scans already keep the filters pushed into them, including dynamic filters such as a hash join's build-side bounds, and those filters reach worker scans with their live updates. But the reader never used them: every row group of every probe-side file was still read and then thrown away by the FilterExec. This change lets the Iceberg Arrow reader prune with the dynamic filters' current values.

How it works

• Provider (iceberg/src/common/runtime_predicate.rs): DynamicFilterPredicateProvider implements iceberg-rust's RuntimePredicateProvider (from the fork pinned in build: point iceberg-rust at a fork carrying the execution-time predicate API #5). It holds the DynamicFilterPhysicalExprs that are kept filters or top-level AND conjuncts of one. A dynamic filter under OR, NOT or any other expression doesn't restrict the rows the filter matches, so it is ignored (fail open). It also holds the scan's Arrow output schema, and the Iceberg schema derived from it (the output schema carries Iceberg field ids).
  • generation(): the sum of the filters' DataFusion generations. Each starts at 1 and grows on every update, so the sum grows on any change and is never reused. It is cheap (a read lock per filter).
  • snapshot(): reads the generation first, then ANDs the filters and converts them with physical_expr_to_predicate (which reads each filter's current() value). So a snapshot may carry a value newer than its generation, but never an older one. If nothing converts, it publishes no predicate.
• Where it's installed: ScanConfig::open, shared by the lazy and eager sources, which now pass their kept filters in. The provider goes to ArrowReaderBuilder::with_runtime_predicate_provider. The reader refreshes the predicate when each data-file task starts and only when the generation changes. It then uses it for row-group pruning, and also for page pruning when iceberg.row_selection_enabled is set (still off by default; unchanged).
• Flag: new iceberg.dynamic_filter_pruning_enabled, default false. When it is off, or the scan keeps no dynamic filter, no provider is installed and the reader works exactly as before.
• No waiting: each file uses whatever value the filters hold when its read starts. Like DataFusion's Parquet scan, it never waits on wait_update/wait_complete.
• Dynamic filters only: static filters already reach the reader as the task predicate at planning time, so they aren't added again.
• Fail open: filters that can't be converted publish no predicate. The reader treats provider, binding and page-pruning errors as "no runtime pruning". If the output schema can't be mapped to Iceberg fields, no provider is installed.
• The converter's dead_code allowance is removed now that it's used.

Correctness

• The converted predicate matches every row the filter could keep (it may match more), so pruning only skips data that no passing row could be in.
• The provider contract requires published predicates to only tighten. DataFusion's dynamic filters (hash join build bounds/values, TopK, aggregate bounds) only tighten, which is the same assumption DataFusion's own Parquet scan makes. This is documented on the provider.
• Filters are still reported as PushedDown::No, so the FilterExec stays and evaluates every row returned. Results are the same with the flag on and off.

Tests

New iceberg/tests/dynamic_filter_pruning.rs, on the taxi fixture (7 files, one per day) with a semi-join whose build side is one day:

• lazy_scan_prunes_with_dynamic_filter / eager_scan_prunes_with_dynamic_filter: single node, so the build side finishes before the probe scan reads and the result is deterministic. Each test runs the same query with the flag off and on and snapshots every scan's output rows:
  • flag off: probe scan output_rows=175000 (all 7 files)
  • flag on: probe scan output_rows=25000 (only the joined day's row groups)
  • the results are equal (trips = 25000).
• Unit tests next to the provider: after a dynamic filter is updated, generation() grows and snapshot() returns the new value (TRUE, then id <= 10). A dynamic filter under OR or NOT installs no provider, while one ANDed with a static filter does.
• distributed_lazy_… / distributed_eager_…_returns_same_rows: 2 workers with the flag on. These check only the result, because pruning there depends on timing.

Checks run: cargo fmt --all, cargo clippy -p datafusion-distributed-iceberg --all-targets (no new warnings), cargo test -p datafusion-distributed-iceberg, cargo test --features integration --test dynamic_filtering. All pass.