feat(datafusion): runtime filter / dynamic predicate pushdown into IcebergTableScan

[jverhoeks]

Is your feature request related to a problem or challenge?

IcebergTableScan (the DataFusion ExecutionPlan in crates/integrations/datafusion/src/physical_plan/scan.rs) does not participate in DataFusion's runtime / dynamic filter pushdown. Filters generated by HashJoinExec build sides at execution time never reach the Iceberg scan, so lineitem-style probe-side scans cannot prune row groups using build-side join key sets. This leaves a real performance gap on TPC-H joins.

Concrete observation

In TPC-H q14 (lineitem JOIN part ON l_partkey = p_partkey) at SF10, with everything DataFusion offers turned on (enable_dynamic_filter_pushdown = true, parquet.bloom_filter_on_read = true, parquet.pushdown_filters = true, parquet.reorder_filters = true), IcebergTableScan still leaves the full lineitem scan unpruned. The post-scan FilterExec evaluates the dynamic filter row by row, which doesn't skip Parquet row groups.

For comparison, the same query on Trino's Iceberg connector finishes ~5x faster on the same data because Trino's connector receives a DynamicFilter, samples it once at split open, and feeds the result into Parquet row-group statistics + bloom filters before any data is read.

Why the existing API can't carry this

IcebergTableScan::new takes filters: &[Expr] once and converts them via convert_filters_to_predicate. TableScanBuilder::with_filter accepts a single Predicate. Both fire at scan construction time. There is no surface for a predicate that only becomes useful mid-execution (when the join build side completes).

Internally the reader is already capable of applying the predicate at all three pruning levels (see crates/iceberg/src/arrow/reader.rs):

// reader.rs:398-444 — when final_predicate is Some:
//   1. with_row_filter(row_filter)               // post-decode RowFilter
//   2. get_selected_row_group_indices(...)       // row-group min/max skip
//   3. get_row_selection_for_filter_predicate    // page-index row selection

So the missing piece is purely the entry point — there is no way to feed the latest dynamic predicate to process_file_scan_task once the scan is in flight.

Describe the solution you'd like

A Trino-style "sample once per file scan task" hook. Sketch:

/// User-supplied factory consulted right before each FileScanTask
/// opens its Parquet file. Returns None when the dynamic source is
/// not yet useful (e.g. join build side has not completed).
pub trait DynamicPredicate: Send + Sync {
    fn current(&self) -> Option<Predicate>;
}

impl TableScanBuilder {
    pub fn with_dynamic_predicate(self, dp: Arc<dyn DynamicPredicate>) -> Self;
}

In ArrowReader::process_file_scan_task, before the existing final_predicate block, intersect the static predicate with dynamic_predicate.current(). The combined predicate flows into the three pruning paths the reader already implements. No reader-side changes beyond the intersection.

IcebergTableScan (or any DataFusion ExecutionPlan that owns DF runtime filters) wraps Vec<Arc<dyn PhysicalExpr>> as a DynamicPredicate impl that translates each filter to Predicate via the existing expr_to_predicate paths and returns None when no filter is yet selective.

Reference implementations elsewhere

	Sample point	What it prunes
Trino (IcebergPageSourceProvider#createPageSource)	once per split via DynamicFilter.getCurrentPredicate()	file-level + row-group + Parquet bloom
Spark (SupportsRuntimeV2Filtering#filter(Predicate[]))	once per partition task, can re-plan	partition-task pruning
iceberg-rust today	n/a	static only

The Trino model fits iceberg-rust's reader cleanly because the reader already has the static-predicate plumbing.

Willingness to contribute

Yes — I have a working prototype on a downstream fork that overrides gather_filters_for_pushdown / handle_child_pushdown_result on IcebergTableScan and applies runtime filters per-batch (post-decode). It produces a real win at SF1 (-21% on TPC-H) and SF10 (-9.4%), but it doesn't reach scan-time pruning since iceberg-rust's TableScan API is single-shot. The follow-up is the DynamicPredicate API above; happy to draft a PR if there's interest.

Related issue: #2363 (static Inexact -> Exact pushdown) — addresses a different bug but cleans up the same pushdown surface.

[jverhoeks]

Update — tried two follow-up "fixes" on top of the per-task DynamicPredicate impl, both backfired and got reverted. Sharing the lessons so anyone working on the actual API in iceberg-rust doesn't have to relearn:

Attempt 1 — cap on Predicate::Set size in the converter
A HashJoinExec build side can produce a runtime InListExpr with hundreds of thousands of values (e.g. TPC-H q04 at SF10: ~580K orderkeys probed back into lineitem). Setting a cap (4096) so oversized lists fall through to a per-batch evaluator looked safe but was net-negative: queries with multiple runtime filters keep paying construction cost on the unaffected filters while losing the row-group pruning that the big filter was actually delivering. Worst-of-both-worlds.

Attempt 2 — Arc::ptr_eq cache keyed on DynamicFilterPhysicalExpr::current()
The intention was to skip retranslation when the filter hadn't changed between tasks. In DataFusion 53 current() calls remap_children, which in the join-pushdown path returns a freshly-built Arc<dyn PhysicalExpr> each call (because column indices are remapped to the probe-side schema). So Arc::ptr_eq essentially never matched, the cache never hit, and I just added Mutex contention across the scan's concurrent FileScanTask processors. Net SF10 wallclock: ~+5% vs no cache.

Practical takeaway for the API design

Per-task current() calls hit a cost floor of the underlying Predicate construction + bind() + row-group eval. For multi-join queries with many filters above one big scan, that floor is non-trivial.

Two ways to lower it without breaking caching:

1. Expose DynamicFilterPhysicalExpr generation: u64 (or any cheap monotonic) so consumers can skip retranslation on the no-change path.
2. Or have the DynamicPredicate trait return an opaque "version" alongside the predicate, letting the reader make its own cache key.

Either is a small extension to what's proposed in this issue, and both are forward-compatible. Happy to fold this into the PR draft when we get to it.

(Bench evidence for the original Path B-2 working as designed and these failed follow-ups is in our downstream branch; can share if useful.)

[jverhoeks]

Update 2: tried a third follow-up that also got reverted. Cataloguing here so the next person doesn't burn a bench cycle on it.

Attempt 3: OnceLock<Predicate> first-success cache

To avoid the Mutex contention from attempt 2 and the conversion cost on every per-task call, I tried a std::sync::OnceLock<Predicate> that fills on the first call returning a non-None predicate and is read lock-free thereafter. The trade-off was acknowledged: we lock in the first snapshot we observe.

Why it failed in a new way: multi-filter scans (TPC-H q04 / q06 / q08 / q09) have several DynamicFilterPhysicalExpr runtime filters that seal at different times as the upstream hash joins finish. The first task that observes ANY filter sealed converts to a Predicate that AND-combines whatever sealed so far. The OnceLock then locks that partial predicate in for the rest of the scan, so later tasks miss the additional pruning that would have come from filters sealing later. SF10: q04 +34%, q06 +74%, q16 +42% vs Path B-2 alone. Net total +5% vs Path B-2.

The composite lesson across all three attempts: any cache-based optimisation here depends on knowing whether all dynamic filters in the scan's bundle are sealed before populating the cache. A single OnceLock keyed on "first non-None" is unsafe for multi-filter scans. A correct cache needs either:

• per-filter OnceLock slots keyed on stable filter identity (which we don't have, because current() returns a fresh Arc from remap_children), or
• a "this filter is sealed" predicate exposed by DynamicFilterPhysicalExpr so the cache can wait until all bundled filters have sealed before populating.

The latter is the API ask: an is_sealed() (or equivalent monotonic state-completion signal) on DynamicFilterPhysicalExpr would let downstream consumers cache safely without observing the inner expression. Combined with the generation/version ask from the previous comment, that gives consumers everything they need: "has this changed since I last cached?" plus "is this final yet?".

Branch is back at the clean Path B-2 implementation. Three failed attempts logged in our downstream feature doc; happy to share the doc as a gist if useful for the upstream discussion.

[jverhoeks]

Closing the loop on this issue with hard data and a sharper API ask.

What we shipped (downstream fork): a DynamicPredicate trait in iceberg::expr plus TableScanBuilder::with_dynamic_predicate(...) that wires runtime filters into the reader's existing row-group / page-index / row-filter pruning paths. Sample once per FileScanTask, intersect with the static predicate, no other reader changes. Net -12.4% on TPC-H SF10 (22/22 match, 20.3 s saved) in the iceberg-datafusion integration.

What we couldn't fix without upstream support: caching the bound predicate across tasks. Five different cache designs (Mutex+ptr_eq, OnceLock, IN-list cap=4096, IN-list cap=200, per-filter Mutex<CachedBound>) all regressed because of one unsolvable problem: in multi-join scans, DynamicFilterPhysicalExpr instances seal at staggered times, and there's no signal we can cheaply observe to know when the inner expression is finalized. Every cache either (a) populates with a partial predicate, (b) never hits because current()'s remap_children rebuilds the inner Arc, or (c) pays Mutex contention on every per-task call.

Microbench data (committed in our fork, runnable upstream): Predicate::bind(schema) is O(N) at 61 ms for N=580K on a hash-join probe-set runtime filter, dominating the 19 ms conversion cost. With ~88 lineitem files at SF10 and parallelism 12, that's ~560 ms wall-clock per scan that we cannot amortize without an upstream signal.

Concrete API ask: an is_sealed() (or monotonic generation() counter) on DynamicFilterPhysicalExpr so consumers can populate a cache exactly once after the build side completes. With that, the bound-predicate cache becomes lock-free read-only after the first sealed observation. We're happy to draft a PR if there's interest.