[Perseus]

Builds on #10; review that first.

What and why

With iceberg.dynamic_filter_pruning_enabled on, several TPC-H SF10 Iceberg queries got slower, not faster. Before this change (median of 3 alternating rounds, 10 iterations each):

• q17 (single node): 600 ms with the setting off, 5,907 ms with it on.
• q20 (8 workers): 360 ms off, 1,148 ms on.
• q8 and q9 were 1.4–2.3× slower with it on.

The cause is the shape of the predicate the Iceberg reader was handed. Its row filter costs O(rows × predicate size): every IN literal becomes one equality and one OR, and AND/OR don't short-circuit. The partitioned-join rewrite from #9 made that predicate an OR of every build partition's filter:

• For q17 that was 16 branches, 15 IN lists and 1,892 literals.
• One q17 branch, a hash-table lookup that can't be converted, became a bare wide range, so the whole OR was close to TRUE (59.99M → 59.54M rows).
• q20 got a 128-branch OR with 4,054 literals.

Design

Before either consumer uses a dynamic filter, its current value is normalized into a domain for each probe column:

• the non-null values that may pass: a range [min, max], or an exact value set, which then implies its own range;
• whether NULL may pass.

The domain is then rebuilt as a small expression. For each constrained column:

col >= min AND col <= max [AND col IN (set)] [OR col IS NULL]

A single value is written col = v, and the per-column constraints are AND-ed together.

Rules:

• AND intersects domains: ranges intersect, sets intersect, and a set is filtered by a range. When a value set exists, the range is derived from the set's min and max, so the output is as tight as the domain allows. An empty intersection becomes literal false.

• Unconvertible conjuncts constrain nothing and are dropped. This covers hash lookups, casts, NOT, columns compared with columns, and literals of another type.

• OR and CASE branches are unioned:

  • The range spans every branch.
  • NULL passes if it passes in any branch.
  • A value set survives only if every branch has one and the union stays within the cap.
  • A column that is unconstrained in any branch is unconstrained in the result.
  • A false (or NULL) branch adds nothing. A true branch, e.g. the ELSE of a canceled build partition, removes the constraint.
  • For CASE, the THEN and ELSE branches are unioned and the WHEN and base expressions are ignored. A missing ELSE yields NULL, so it adds nothing.

  These rules also cover the merged distributed shape, an OR of per-task CASEs.

• Multi-column keys become one constraint per column. That is wider than the tuples but still holds all of them.

• The cap reuses datafusion.optimizer.hash_join_inlist_pushdown_max_distinct_values (default 150), so no new configuration is added. It applies to unions and to IN lists.

• Literals follow the same rules as the existing predicate converter:

  • A literal is used only if its type is exactly the column's type.
  • Float NaN and ±0.0 literals are not used.
  • An IN list containing such an item is not used at all, because dropping the item would narrow the set.

Why it's conservative

Every rule only keeps or widens the set of rows that pass, so the domain contains every row the filter keeps. Widening is only safe where the filter appears in a monotonic position, so normalization is applied:

• only to dynamic filters that are top-level AND conjuncts of the scan's kept filters (the existing selection rule), never to one under NOT or OR;
• inside a filter, only through AND, OR and CASE branches. Anything else constrains nothing.

The FilterExec above the scan still evaluates the original filters, so this only changes what is read. Results are identical with the setting on and off.

Where it's used, and caching

CompactedDynamicFilters (the new dynamic_domain.rs) is built once per scan stream and shared by both consumers:

• the reader's runtime predicate, which converts it with the existing predicate converter;
• the file pruner, which builds its PruningPredicate from it.

It is recomputed only when the filters' generation changes, so per-file work is a cache hit. It replaces the OR-of-branches rewrite: case_as_or.rs and the predicate converter's CASE handling are removed, because they are now dead code.

Design note: comparison with runtime-filter domains

This is the same idea as compacting runtime filters into per-column domains (as Trino does):

• Same: collected filters are summarized per column as a range plus an optional bounded value set, and a value set that grows past a threshold collapses to its min/max.
• Different: the inputs here are DataFusion's already-built physical filter expressions, including the hash-routed CASE of partitioned joins, rather than raw build-side values. So the domain is derived structurally (intersect under AND, union under OR/CASE), and anything it can't read is treated as unconstrained rather than failing.
• Different: the output is again an expression, so the existing Iceberg predicate converter and DataFusion's PruningPredicate consume it unchanged.

Tests

• Unit tests (dynamic_domain.rs) snapshot the compacted expression for:
  • a hash-routed CASE with ELSE false (union range plus union IN);
  • a union over the cap (range only);
  • a branch without an IN list, i.e. a dropped hash lookup (range only);
  • ELSE true (no constraint);
  • a range-routed CASE;
  • the merged OR-of-CASEs shape;
  • multi-column keys;
  • NULL-admitting wrapping;
  • AND intersection, including an empty intersection that becomes false;
  • unconvertible conjuncts being dropped;
  • unusable literals (another type, NaN, zero);
  • selection, i.e. filters under NOT or OR are not used;
  • caching once per generation, including a caller delayed on the cache while the filter is updated, which must reuse the newer expression rather than overwrite it.
• New reader-input snapshot (runtime_predicate.rs). It runs a forced partitioned join over the taxi fixture through the normal planning path. The join's filter is a 2-branch CASE carrying 50,000 IN-list items; the reader receives pickup_date >= 2024-01-08 AND pickup_date <= 2024-01-14 AND pickup_date IN (2024-01-14, 2024-01-08).
• Existing dynamic_filter_pruning.rs integration tests pass with unchanged snapshots, so rows and files read are the same as before this change. That includes the partitioned-join and file-skipping cases, and the 2-worker distributed correctness run.
• Checks:
  • cargo fmt --all;
  • cargo clippy -p datafusion-distributed-iceberg -p datafusion-distributed-iceberg-test-tools --all-targets, with no warnings;
  • cargo test -p datafusion-distributed-iceberg;
  • cargo test --features integration --test dynamic_filtering, with 18 passed.

Benchmarks

Setup:

• TPC-H SF10 on Iceberg, local, run from a temporary local patch that exposes the setting through an environment variable; the patch is not part of this PR.
• base is this PR's base and head is this PR.
• Each cell is the median over 3 rounds of the per-round median of 10 iterations, with the run order reversed every other round.
• The machine was under heavy background load: 1-minute load average 33–51 at the start of runs.

query	workers	base off	base on	head off	head on
q17	0	600	5,907	462	500
q20	8	360	1,148	357	317
q8	0	328	741	392	454
q9	0	510	699	589	702

Deterministic metrics from one --debug run each:

query, scan	base off	base on	head off	head on
q17 lineitem rows out of the scan	59.99M	59.54M	59.99M	59.91M
q17 lineitem fetch_time (2 scans)	2.62 s + 0.53 s	28.94 s + 43.95 s	1.98 s + 0.54 s	0.85 s + 0.85 s
q8 lineitem fetch_time	1.27 s	3.43 s	1.56 s	2.25 s
q9 lineitem fetch_time	1.47 s	5.48 s	1.72 s	1.77 s
q20 partsupp rows out of the scan	8.0M	324K	8.0M	8.0M
q20 supplier rows out of the scan	100K	4.05K	100K	4.05K

How to read the rows rows:

• With the setting on at base, fewer rows leave the scan because the reader's row filter evaluates the large OR, and that is the expensive part.
• At head, row-level filtering of keys whose value sets are over the cap is left to the FilterExec. Range-based row-group, page and file pruning still applies.
• q20's partsupp scan shows the trade-off: its union value set is over the cap, so the scan returns all rows again, and the query is still faster than with the setting off.

One alternating full-suite pass at head (single node, 2 rounds × 3 iterations, order reversed in round 2). The total of per-query medians was:

• setting off: 14.26 s and 9.26 s;
• setting on: 10.10 s and 8.59 s.

q8 and q9 were slower with the setting on in both rounds (q8 576 → 750 ms, q9 896 → 981 ms), which matches the focused rounds above: their remaining cost with the setting on is the reader evaluating the compacted ranges. No other query was slower in both rounds. The per-query noise under this load is large (±30–40%).

With the setting off, base and head run the same code path (the shared filter state is never created, so no provider or pruner is installed). A dedicated re-run of q8 and q9 (single node, 4 alternating rounds × 10 iterations, 1-minute load 28–37) confirms that the setting-off difference above is noise:

query	base off	head off
q8	409 ms [339, 403, 453, 415]	443 ms [513, 386, 411, 474]
q9	696 ms [631, 703, 688, 802]	723 ms [822, 588, 750, 695]