feat(physical-expr): DynamicFilterTracker for cheap dynamic-filter change detection by adriangb · Pull Request #22460 · apache/datafusion

adriangb · 2026-05-22T14:57:33Z

Extracted from a design discussion around the duplicated "does this filter have a dynamic portion that might change?" / "has the filter changed?" patterns (e.g. #22450, FilePruner).

Rationale for this change

DynamicFilterPhysicalExpr has a rich producer API (update(), mark_complete(), wait_update(), wait_complete()), but consumers that hold a predicate which contains dynamic filters only had a bare, recursive snapshot_generation() -> u64. Call sites hand-rolled the same boilerplate around it: store a last_generation, recompute snapshot_generation(&predicate) (a full tree walk) on every check, diff it, and rebuild an expensive PruningPredicate on change. FilePruner did exactly this, and none of these consumers exploited mark_complete().

This adds a small consumer-side counterpart so the pattern lives in one place, driven by the existing watch channel rather than by re-walking the tree.

This immediately eliminates some tree traversals (we were constantly traversing the expression tree to check if any filters updated). Long term I hope this makes changes like #22450 easier.

What changes are included in this PR?

New public API (datafusion_physical_expr):

DynamicFilterTracking (classify → Static / AllComplete / Watching, plus contains_dynamic_filter / watcher) and DynamicFilterTracker (changed). A tracker walks a (possibly composite) predicate once, subscribes to every still-incomplete dynamic filter, and answers changed() by polling only that shrinking set — steady-state is one atomic load per filter, no tree walk, no lock until something actually moves.
Lives in the expressions::dynamic_filters module (dynamic_filters.rs → dynamic_filters/mod.rs, tracker in dynamic_filters/tracker.rs). The subscription plumbing (subscribe, DynamicFilterSubscription, DynamicFilterChange, observe, is_complete) is pub(crate); test-only constructors are #[cfg(test)].

Consumers:

FilePruner is driven by DynamicFilterTracking instead of snapshot_generation polling, and now decides its own existence in try_new (a static predicate with no usable stats builds no pruner).
The Parquet opener skips wrapping the scan in EarlyStoppingStream when nothing can change, and no longer needs an "is it dynamic?" gate.

Are these changes tested?

Yes — unit tests for the tracker (classification, detect-update-once, mark_complete is not a change, coalesced update+complete, multi-filter), plus the existing datafusion-pruning / datafusion-datasource-parquet suites (incl. the static/dynamic/partition opener pruning test) pass unchanged.

Are there any user-facing changes?

New public API as above (additive). One deprecation and one behavior change, both documented in the DataFusion 55.0.0 upgrade guide:

datafusion_physical_expr_common::physical_expr::is_dynamic_physical_expr is deprecated (since 55.0.0) — downcast to DynamicFilterPhysicalExpr or use DynamicFilterTracking. (snapshot_generation itself is unchanged — still backing the FFI vtable and proto roundtrip.)
FilePruner::try_new now returns None for a purely static predicate over a file with no usable column statistics (previously Some whenever a statistics struct was present).

Followups

I noticed a possible follow-up gate refinement, tracked in #22495.

This also opens up the possibility to deprecate / remove the snapshot / generation machinery from the public physical expr APIs. These new APIs (the watchers, tracker) subsumes much of the functionality, and I don't think we want to add PhysicalExpr::watch. And after several releases the only thing using it right now is dynamic filters, i.e. no other legitimate use case has materialized.

🤖 Generated with Claude Code

adriangb · 2026-05-24T14:07:47Z

Follow-up on the FilePruner creation gate → #22495

This PR keeps the gate contains_dynamic_filter(p) || has_statistics() (swapping the old is_dynamic_physical_expr for the equivalent downcast so the opener no longer depends on snapshot_generation).

(Correction to an earlier version of this comment: I had suggested replacing the dynamic-filter check with a structural "references only partition columns" analysis. That's unsound — a dynamic filter's expression changes at runtime and is usually lit(true) at file-open, e.g. the TopK filter on part_col in SELECT * FROM t ORDER BY part_col LIMIT 10. "Has a (non-terminated) dynamic filter" is a correct, first-class gate condition, not a proxy. #22495 has been rewritten with the corrected analysis.)

zhuqi-lucas

LGTM, minor question left.

adriangb · 2026-05-25T04:30:41Z

Thanks for the review @zhuqi-lucas !

Do you think this will help with #22450? My thought was that it would make it very cheap to check e.g. on every batch, row group boundary, etc. if we need to re-prune using stats. Does that pan out you think?

…r change detection Introduce a consumer-side counterpart to the producer API on `DynamicFilterPhysicalExpr` (`update`/`mark_complete`/`wait_*`), living in a new `expressions::dynamic_filters` module (`dynamic_filters.rs` becomes `dynamic_filters/mod.rs`, with the tracker in `dynamic_filters/tracker.rs`). `DynamicFilterPhysicalExpr::subscribe()` returns a `DynamicFilterSubscription` that observes one filter through its existing `watch` channel: steady-state polling is a single atomic load, the lock is taken only when the filter actually moved, and a bare `mark_complete()` (which re-broadcasts the current generation) is distinguished from a real expression change. This subscription plumbing is `pub(crate)` — the public surface is the tracker. `DynamicFilterTracker` walks a (possibly composite) predicate once, subscribing to every still-incomplete dynamic filter, then answers `changed()` by polling only that shrinking set. `DynamicFilterTracking::classify` distinguishes Static / AllComplete / Watching in a single traversal. Test-only constructors are gated behind `#[cfg(test)]`. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Replace `FilePruner`'s hand-rolled `snapshot_generation()` polling (store last `u64`, recompute + diff on every `should_prune`) with a `DynamicFilterTracking` classification computed once at construction. The pruner rebuilds the pruning predicate on the first check and thereafter only when a watched dynamic filter has actually moved. This also lets the Parquet opener skip wrapping the scan in `EarlyStoppingStream` when the predicate is static or its dynamic filters are already complete: the up-front `prune_file` check already captured everything such a predicate can prune, so per-batch re-checking was pure overhead. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

…s_dynamic_physical_expr The Parquet opener gated `FilePruner` creation on `is_dynamic_physical_expr(p) || has_statistics()`, where `is_dynamic_physical_expr` answered "is it dynamic?" via `snapshot_generation(p) != 0`. `FilePruner` already classifies its predicate once (`DynamicFilterTracking`) to drive the change tracker, so the same classification can answer the gate. Move the decision into `FilePruner::try_new`: it returns `None` when the file has no statistics struct, or when the predicate is purely static and the file has no usable column statistics (planning already did everything such a pruner could). A dynamic predicate is still accepted (it may prune via partition-value folding even without column statistics). The opener now just calls `try_new`. With its last internal caller gone, `is_dynamic_physical_expr` is deprecated (since 55.0.0) rather than removed; downstream users should downcast to `DynamicFilterPhysicalExpr` or use `DynamicFilterTracking`. See the 55.0.0 upgrade guide. `snapshot_generation` itself is unchanged (FFI vtable + proto). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

zhuqi-lucas · 2026-05-25T05:30:40Z

Thanks for the review @zhuqi-lucas !
Do you think this will help with #22450? My thought was that it would make it very cheap to check e.g. on every batch, row group boundary, etc. if we need to re-prune using stats. Does that pan out you think?

Thank you @adriangb, yes , this should fit #22450 cleanly. The RG-boundary "did the filter change?" check is exactly the pattern this tracker is built for — steady-state O(1) atomic loads per still-incomplete filter, no tree walk. The complete flag is a bonus on top: once all dynamic filters upstream have called mark_complete(), the RG pruner can stop polling and stay on the fast path.

I'll wire this into #22450 once this PR merged.

…ange detection apache#22460 added a consumer-side DynamicFilterTracker that subscribes once to every still-incomplete DynamicFilterPhysicalExpr inside a predicate and answers "has it changed?" with a single atomic load per filter — no tree walk on every check. Three call sites on this branch were polling the predicate's snapshot_generation on every batch / row-group boundary; switch them over: - RowGroupPruner (push_decoder.rs): owns a DynamicFilterTracking and rebuilds its PruningPredicate only when tracker.changed() reports an update. Falls back to a single up-front build for Static / AllComplete predicates. - Parquet opener force_per_rg_runs gate: "dynamic AND not yet complete" is exactly the Watching variant of DynamicFilterTracking — match on that instead of is_dynamic_physical_expr + is_filter_complete. - ParquetSource fmt_extra marker (dynamic_rg_pruning=eligible): use DynamicFilterTracking::classify(...).contains_dynamic_filter() so the deprecated is_dynamic_physical_expr can go. is_filter_complete and its unit tests in opener/mod.rs are removed — the semantics they asserted are now covered by DynamicFilterTracking tests in datafusion_physical_expr::expressions::dynamic_filters. SLT expectations for sort_pushdown.slt are also brought in line with upstream main (the SortExec-elimination changes from apache#22493 landed between this branch's last push and now).

…tes only Address adriangb's review on apache#22450: - Build `RowGroupPruner` only when `DynamicFilterTracking::classify` reports `Watching`. Static and already-complete predicates were fully consumed by `prune_by_statistics` at file open, so re-evaluating them per RG boundary was wasted work. - Refresh stale comments left over from the multi-decoder design that the `into_builder` refactor superseded (push_decoder.rs module doc + struct docs, source.rs `fmt_extra` marker comment, decoder_projection.rs doc). - Drop the `snapshot_generation` reference from opener/mod.rs — the pruner uses the `DynamicFilterTracker` watch channel from apache#22460.

Closesapache#22407. DataFusion already prunes parquet at three granularities — file (`EarlyStoppingStream` + `FilePruner`), row group at scan-startup (`PruningPredicate` → `RowGroupAccessPlanFilter`), and row inside an open RG (`RowFilter`). There's a gap in the middle: once row-group pruning runs at file open, that decision is frozen because any dynamic filter is still `lit(true)` at that point. As `TopK` tightens its threshold at runtime, subsequent RGs in the already-opened file keep getting decoded even when their stats already prove they cannot beat the threshold. This is the dominant cost for `ORDER BY ... LIMIT` queries on multi-RG files where file-level pruning can't help. This PR closes the gap with a single decoder paused at row-group boundaries, a pruner consulted between row groups, and the decoder rebuilt via `into_builder()` to skip the row groups the pruner just rejected. Three coordinated pieces: 1. `RowGroupPruner` (`push_decoder.rs`) mirrors `FilePruner` at row-group granularity. Uses `DynamicFilterTracker` (apache#22460) to subscribe once to every not-yet-complete dynamic filter; `tracker.changed()` is a single atomic load — no tree traversal per check. The cached `PruningPredicate` is rebuilt only when a watched filter has actually moved. Predicate construction errors and predicate evaluation errors are counted into two separate metrics. 2. Single-decoder iteration model (`PushDecoderStreamState::transition`). The opener builds one `ParquetPushDecoder` from the prepared access plan, and the stream uses arrow-rs 59's `ParquetRecordBatchReader` iterator to pause at row-group boundaries. At each boundary the pruner is consulted against the head of `rg_plan`; pruned indices are dropped and the decoder is rebuilt via `decoder.into_builder().with_row_groups(remaining).build()` so the skipped RGs are bypassed entirely. Already-fetched buffered bytes for downstream RGs carry across the rebuild. 3. Gate: build the pruner only when `DynamicFilterTracking::classify(&predicate)` reports `Watching` AND more than one row group remains. Static or already-complete predicates were fully consumed by `prune_by_statistics` at file open, so re-evaluating them per RG boundary would be wasted work. Observability: - New `Count` metric `row_groups_pruned_dynamic_filter` on `ParquetFileMetrics`. - New `dynamic_rg_pruning=eligible` marker on `ParquetSource::fmt_extra` (Default + Verbose), emitted when the predicate has a still-watching dynamic portion. Benchmarks (`benchmarks/sort_pushdown_inexact`, 5 iterations): | Query | main | this PR | Δ | |---|---|---|---| | Q1 `ORDER BY l_orderkey DESC LIMIT 100` | 6.99 ms | 3.80 ms | -46% | | Q2 `ORDER BY l_orderkey DESC LIMIT 1000` | 3.29 ms | 1.33 ms | -60% | | Q3 `SELECT * ... DESC LIMIT 100` | 11.17 ms | 9.91 ms | -11% | | Q4 `SELECT * ... DESC LIMIT 1000` | 9.28 ms | 7.95 ms | -14% | Tests: - 6 unit tests (3 in push_decoder.rs::tests for RowGroupPruner; 3 in source.rs::tests for the EXPLAIN marker). - 3 integration tests in `datafusion/core/tests/parquet/dynamic_row_group_pruning.rs`. - New SLT `dynamic_row_group_pruning.slt` covering both EXPLAIN surfaces. - `cargo clippy --all-targets --all-features -- -D warnings` clean.

…ters (apache#22450) ## Which issue does this PR close? Closesapache#22407. ## Rationale for this change DataFusion already prunes parquet at three granularities — **file** (`EarlyStoppingStream` + `FilePruner`), **row group at scan-startup** (`PruningPredicate` → `RowGroupAccessPlanFilter`), and **row inside an open RG** (`RowFilter`). There's a gap in the middle: once row-group pruning runs at file open, that decision is **frozen** because any dynamic filter is still `lit(true)` at that point. As `TopK` tightens its threshold at runtime, subsequent RGs in the already-opened file keep getting decoded even when their stats already prove they cannot beat the threshold. This is the dominant cost for `ORDER BY ... LIMIT` queries on multi-RG files where file-level pruning can't help (single large file, or scrambled-RG multi-file). See the issue for a full architectural diagram and a concrete trace showing where the wasted I/O / decompression / decode lives. ## What changes are included in this PR? A single decoder paused at row-group boundaries, with a pruner consulted between row groups and the decoder rebuilt via `into_builder()` to skip the row groups the pruner just rejected. Three coordinated pieces: 1. **`RowGroupPruner`** (`datafusion/datasource-parquet/src/push_decoder.rs`) mirrors `FilePruner` at row-group granularity. It uses the `DynamicFilterTracker` API from apache#22460 to subscribe once to every not-yet-complete dynamic filter in the predicate; `tracker.changed()` is a single atomic load — no tree traversal per check. The cached `PruningPredicate` is rebuilt only when a watched filter has actually moved, then evaluated against the next pending row group's statistics via the existing `RowGroupPruningStatistics` adapter. Predicate construction errors and predicate evaluation errors are counted into two separate metrics so a flaky predicate path can never silently drop data. 2. **Single-decoder iteration model** (`PushDecoderStreamState::transition`). The opener builds **one** `ParquetPushDecoder` from the prepared access plan, and the stream uses arrow-rs 59's `ParquetRecordBatchReader` iterator to pause at row-group boundaries. At each boundary the pruner is consulted against the head of `rg_plan` (the remaining row-group indices). If the pruner proves the head RG unwinnable, that index is dropped from the plan and the decoder is **rebuilt via** `decoder.into_builder().with_row_groups(remaining).build()` so the skipped RGs are bypassed entirely — no decode, no row-filter eval. Already-fetched buffered bytes for downstream RGs carry across the rebuild. 3. **Gate: build the pruner only when the predicate actually moves.** The opener creates a `RowGroupPruner` only when `DynamicFilterTracking::classify(&predicate)` reports `Watching` (at least one not-yet-complete dynamic filter) **and** more than one row group remains in the access plan. Static or already-complete predicates were fully consumed by `prune_by_statistics` at file open, so re-evaluating them per RG boundary would be wasted work. The earlier multi-decoder design (`PendingDecoderRun`, `ParquetAccessPlan::split_runs`, `force_per_row_group`) is removed — arrow-rs 59's `into_builder` + `with_row_groups` makes a single decoder strictly more capable. ### Observability - New `Count` metric `row_groups_pruned_dynamic_filter` on `ParquetFileMetrics` surfaces the runtime saving. - New `dynamic_rg_pruning=eligible` marker on `ParquetSource`'s `EXPLAIN` (`fmt_extra` Default + Verbose) signals plan-time eligibility, emitted whenever the predicate has a still-watching dynamic portion. **Eligible** rather than **true** because the static plan can't predict the runtime outcome. ### Benchmarks (`benchmarks/sort_pushdown_inexact`, 5 iterations) | Query | main | this PR | Δ | |---|---|---|---| | Q1 `ORDER BY l_orderkey DESC LIMIT 100` | 6.99 ms | 3.80 ms | **−46%** | | Q2 `ORDER BY l_orderkey DESC LIMIT 1000` | 3.29 ms | 1.33 ms | **−60%** | | Q3 `SELECT * ... DESC LIMIT 100` | 11.17 ms | 9.91 ms | −11% | | Q4 `SELECT * ... DESC LIMIT 1000` | 9.28 ms | 7.95 ms | −14% | Narrow-projection queries gain the most — their per-RG cost is dominated by metadata + sort-column read, which this PR eliminates for unwinnable RGs. Wide-projection queries gain less because the *kept* RG's all-column decode dominates total time, but still see meaningful savings. ## Are these changes tested? Three layers: - **6 unit tests**: - 3 in `push_decoder.rs::tests`: `RowGroupPruner` basic pruning, tracker-driven dynamic-filter updates, fallback when the predicate has no analyzable bounds. - 3 in `source.rs::tests`: `dynamic_rg_pruning=eligible` marker present on dynamic predicate, absent on static predicate, absent when there is no predicate at all. - **3 integration tests** in `datafusion/core/tests/parquet/dynamic_row_group_pruning.rs`: asserts `row_groups_pruned_dynamic_filter >= 1` end-to-end on a 5-RG `ORDER BY DESC LIMIT 5` scan; a regression test for the `prepare_access_plan` reorder bug that uses `ORDER BY ASC` against a file written in descending value order so the sort-pushdown reorder is exercised; and a quiet-without-TopK test that asserts the metric stays at 0 (no spurious firing). - **New SLT** `datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt`: asserts both `EXPLAIN` surfaces — plain `EXPLAIN` shows `dynamic_rg_pruning=eligible`, and `EXPLAIN ANALYZE` pins `row_groups_pruned_dynamic_filter=4` (five RGs, four pruned at runtime). `cargo clippy --all-targets --all-features -- -D warnings` clean. ## Are there any user-facing changes? Two visible additions, both opt-in via existing dynamic-filter infrastructure: - New `row_groups_pruned_dynamic_filter` counter visible in `EXPLAIN ANALYZE` for queries whose plan carries a `DynamicFilterPhysicalExpr` (today: only TopK with `enable_topk_dynamic_filter_pushdown=true`, which is the default). - New `dynamic_rg_pruning=eligible` marker visible in `EXPLAIN` output for the same queries. No config changes, no API breakage, no behavior change for queries without a dynamic predicate.

github-actionsBot added physical-expr Changes to the physical-expr crates datasource Changes to the datasource crate labels May 22, 2026

adriangb mentioned this pull request May 22, 2026
feat(parquet): intra-file early stopping via statistics + dynamic filters #22450
Merged

adriangb force-pushed the dynamic-filter-tracker branch from 189218c to 6ef286fCompare May 24, 2026 03:51

adriangb mentioned this pull request May 24, 2026
FilePruner creation gate: the dynamic-filter check is correct (only a narrow no-stats optimization may remain) #22495
Closed

adriangb force-pushed the dynamic-filter-tracker branch 2 times, most recently from df15a2e to 783f761Compare May 24, 2026 14:56

github-actionsBot added the documentation Improvements or additions to documentation label May 24, 2026

adriangb added the api change Changes the API exposed to users of the crate label May 24, 2026

adriangb force-pushed the dynamic-filter-tracker branch 2 times, most recently from ef5217a to 5a3ae2cCompare May 24, 2026 15:44

adriangb requested review from xudong963 and zhuqi-lucas May 24, 2026 15:45

adriangb marked this pull request as ready for review May 24, 2026 15:45

zhuqi-lucas approved these changes May 25, 2026
View reviewed changes

Comment threaddatafusion/physical-expr/src/expressions/dynamic_filters/mod.rs
Comment threaddatafusion/physical-expr/src/expressions/dynamic_filters/mod.rs Outdated

adriangband others added 3 commits May 24, 2026 23:35

adriangb force-pushed the dynamic-filter-tracker branch from 5a3ae2c to 489b6a6Compare May 25, 2026 04:35

adriangb added this pull request to the merge queue May 25, 2026

Merged via the queue into apache:main with commit 4c0f944May 25, 2026
40 checks passed

adriangb deleted the dynamic-filter-tracker branch May 25, 2026 12:41

adriangb mentioned this pull request May 26, 2026
docs: drop TableSchema upgrade-guide note pydantic/datafusion#62
Merged

Navigation Menu

Uh oh!

feat(physical-expr): DynamicFilterTracker for cheap dynamic-filter change detection - #22460

feat(physical-expr): DynamicFilterTracker for cheap dynamic-filter change detection#22460
adriangb merged 3 commits into
apache:mainfrom
adriangb:dynamic-filter-tracker

adriangb commented May 22, 2026•
edited
Loading

Uh oh!

adriangb commented May 24, 2026•
edited
Loading

Uh oh!

zhuqi-lucas left a comment

Uh oh!

Uh oh!

Uh oh!

adriangb commented May 25, 2026

Uh oh!

zhuqi-lucas commented May 25, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

adriangb commented May 22, 2026• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Rationale for this change

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

Followups

Uh oh!

adriangb commented May 24, 2026• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

zhuqi-lucas left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

adriangb commented May 25, 2026

Uh oh!

zhuqi-lucas commented May 25, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

adriangb commented May 22, 2026•
edited
Loading

adriangb commented May 24, 2026•
edited
Loading