[Perseus]

Builds on #8; review that first.

What and why

When DataFusion plans a partitioned hash join, which is the usual plan for large and distributed joins, the dynamic filter it pushes into the probe-side scan is a routing CASE expression. Until now, Iceberg scans got nothing out of it:

• the conversion to an Iceberg predicate (used to prune row groups and rows while reading a file) dropped CASE;
• partition-based file skipping (from feat(iceberg): skip data files whose partition values can't match dynamic filters #8) uses DataFusion's pruning predicate, which can't evaluate a CASE against statistics either.

So the probe side of a partitioned join read every row group of every file. With this change, both paths prune with these filters. Query results are unchanged.

The CASE shape

A partitioned hash join builds one filter per build partition and routes each probe row to its partition's filter:

CASE hash_repartition(keys) % N
  WHEN 0 THEN F0
  WHEN 1 THEN F1
  ...
  ELSE <true if a build partition was canceled, else false>
END

Each Fi holds that partition's bounds (key >= min AND key <= max) and its membership test (an IN list, or a hash lookup for large builds). With range partitioning the shape is CASE range_partition(keys) WHEN i THEN Fi ... ELSE Fn END. Joins that need probe-side NULLs wrap the CASE in ... OR key IS NULL. When only one build partition has rows, DataFusion skips the CASE and pushes that partition's filter directly.

In a distributed query, the coordinator merges the partitioned-join filters reported by every producer task with OR, so workers receive an OR of per-task CASE expressions.

Why the OR of the branches is safe

A CASE evaluates, for each row, exactly one of its THEN branches or its ELSE. If the CASE is true for a row, the branch that row took is true for it, so the OR of all THEN branches and the ELSE is true too. That holds whatever the base and WHEN expressions are, even when their values can't be listed or evaluated against statistics.

The OR can match rows the CASE doesn't, because a row is checked against every partition's filter instead of only its own. It never misses a row the CASE keeps. A missing ELSE yields NULL, which never passes a filter, so it is left out.

Conversion rules

Every converted subtree is tracked as exact or approximate. A predicate is approximate when it may match more rows than the expression; it never matches fewer.

• CASE converts to the OR of its branches and is always approximate. If any branch can't be converted, the whole CASE can't be, because a branch can't be dropped from an OR. An ELSE true gives always-true, so there is nothing to prune; false branches drop out of the OR.
• OR now converts when every child converts, exact or approximate, and is approximate if any child is. Before, it required every child to be exact. An OR of over-approximations still over-approximates.
• NOT and negated IN still require an exact child. Negating an over-approximation would under-approximate and could skip data a row passing the filter is in. NOT (... OR <approximate>) therefore still doesn't convert.

How both pruning paths use it

iceberg/src/common/case_as_or.rs holds the shared rewrite:

• case_branches_disjunction builds the OR of a CASE's branches.
• rewrite_case_as_or applies it to every CASE reachable from the root through AND and OR, including CASEs nested in the branches. AND and OR only ever widen when an operand widens, so the result still matches every row the original keeps. A CASE anywhere else, for example under NOT, is left alone.

Row groups and rows. physical_expr_to_predicate treats a CASE as case_branches_disjunction(case) marked approximate. The runtime predicate provider from #6 publishes that predicate to the Iceberg reader as before.

Whole files. PartitionFilePruner used to hand the dynamic filter to DataFusion's FilePruner, which takes its own snapshot of the filter and can't accept a rewritten expression. Now it:

1. reads the filters' generation;
2. takes a snapshot of their current value;
3. rewrites it with rewrite_case_as_or;
4. builds DataFusion's PruningPredicate from the result;
5. evaluates it against each file's partition-value statistics.

The built predicate is cached and rebuilt only when the filters' generation changes, which reproduces the change tracking FilePruner did internally. A comment in the code explains this.

Both paths keep the existing behavior otherwise: same single flag, iceberg.dynamic_filter_pruning_enabled; nothing waits for a filter; only dynamic filters are used; any error keeps the data; and the FilterExec above the scan still applies every filter.

Reused from DataFusion

DataFusion 55 has no simplification or pruning support for CASE, so the CASE-to-OR rewrite is new. Everything else is DataFusion's:

• PruningPredicateBuilder and PruningPredicate for file-level pruning;
• PrunableStatistics to present a file's statistics;
• snapshot_physical_expr and snapshot_generation to read the dynamic filters and detect changes;
• CaseExpr's accessors to read the branches.

Tests

Converter unit tests, each showing the predicate produced:

• a hash-routed CASE with ELSE false;
• a range-routed CASE;
• a CASE with ELSE true, which gives always-true;
• a CASE with no ELSE;
• a CASE with an unconvertible branch, which doesn't convert;
• the ... OR key IS NULL wrapping;
• the distributed merged shape, an OR of two per-task CASEs;
• an OR with approximate children, which now converts;
• NOT of an approximate OR or of a CASE, which doesn't convert.

A unit test for the file pruner uses one pruner throughout. It checks a file, updates the dynamic filter to a CASE, and shows the decision change. It then shows that an unchanged generation reuses the cached predicate, and that a later update changes the decision again.

Integration tests on the 7-day taxi table (one file per day), with target_partitions = 4 and both hash_join_single_partition_threshold settings at 0 so the join is partitioned. The plan is asserted to use a partitioned join. The build side selects two days that hash to different build partitions, so the filter is a routing CASE. Each test runs with pruning off and on and checks that the results are identical (50000 trips).

Scenario	Probe-scan rows, off → on	Data files read, off → on
Partitioned table, lazy file planning	175000 → 50000	9 → 4
Partitioned table, eager file planning	175000 → 50000	9 → 4
Same data in an unpartitioned table (row groups only)	175000 → 50000	9 → 9

The data-file counts include the build side's 2 files. The unpartitioned case isolates the row-group path: no file can be skipped by partition value, so the drop in rows comes only from row-group pruning.

A 2-worker distributed run of the partitioned join with pruning on returns the correct result.

Each new integration test was checked against a broken implementation:

• disabling the rewrite on the file path makes the partitioned-table test fail;
• disabling CASE conversion makes the unpartitioned-table test fail;
• making the pruner's cache always hit makes the cache unit test fail.

Checks run:

• cargo fmt --all --check
• cargo clippy -p datafusion-distributed-iceberg -p datafusion-distributed-iceberg-test-tools --all-targets: no warnings
• cargo test -p datafusion-distributed-iceberg: all pass
• cargo test --features integration --test dynamic_filtering: 18/18 with --test-threads=1. Run in parallel, the pre-existing flaky aggregates::tests::remote_dynamic_filters fails intermittently.