You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "jverhoeks (via GitHub)" <gi...@apache.org> on 2026/04/27 18:51:22 UTC
[I] feat(datafusion): runtime filter / dynamic predicate pushdown into IcebergTableScan [iceberg-rust]
jverhoeks opened a new issue, #2376:
URL: https://github.com/apache/iceberg-rust/issues/2376
### 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`):
```rust
// 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:
```rust
/// 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] feat(datafusion): runtime filter / dynamic predicate pushdown into IcebergTableScan [iceberg-rust]
Posted by "alexanderbianchi (via GitHub)" <gi...@apache.org>.
alexanderbianchi commented on issue #2376:
URL: https://github.com/apache/iceberg-rust/issues/2376#issuecomment-6113668862
@jverhoeks I think we should move this into https://github.com/apache/datafusion-iceberg, would you like to open an issue there for dynamic filtering support?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] feat(datafusion): runtime filter / dynamic predicate pushdown into IcebergTableScan [iceberg-rust]
Posted by "jverhoeks (via GitHub)" <gi...@apache.org>.
jverhoeks commented on issue #2376:
URL: https://github.com/apache/iceberg-rust/issues/2376#issuecomment-6120476214
Thanks @alexanderbianchi, good idea. I've opened https://github.com/apache/datafusion-iceberg/issues/68 to continue there. The iceberg-rust part (a `DynamicPredicate` hook on `TableScanBuilder`) is described in that issue too, and I can split it back out here when it's time for a PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] feat(datafusion): runtime filter / dynamic predicate pushdown into IcebergTableScan [iceberg-rust]
Posted by "jverhoeks (via GitHub)" <gi...@apache.org>.
jverhoeks commented on issue #2376:
URL: https://github.com/apache/iceberg-rust/issues/2376#issuecomment-4330331459
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] feat(datafusion): runtime filter / dynamic predicate pushdown into IcebergTableScan [iceberg-rust]
Posted by "jverhoeks (via GitHub)" <gi...@apache.org>.
jverhoeks commented on issue #2376:
URL: https://github.com/apache/iceberg-rust/issues/2376#issuecomment-4330042368
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.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] feat(datafusion): runtime filter / dynamic predicate pushdown into IcebergTableScan [iceberg-rust]
Posted by "jverhoeks (via GitHub)" <gi...@apache.org>.
jverhoeks commented on issue #2376:
URL: https://github.com/apache/iceberg-rust/issues/2376#issuecomment-4337066868
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org