perf(physical-expr): cache remapped expression in DynamicFilterPhysicalExpr::current()#23532
Merged
adriangb merged 6 commits intoJul 14, 2026
Merged
Conversation
…alExpr::current() `DynamicFilterPhysicalExpr::current()` is called per batch on the RowFilter path (via `PhysicalExpr::evaluate`). It internally calls `remap_children`, which does a `transform_up` tree walk that, for filters carrying a large `InListExpr` (HashJoin's `col IN build_keys`, ~150+ values), is dominated by `InListExpr::with_new_children` cloning the whole list on every call. The inner generation only changes when the filter is `update`d — once per HashJoin build or once per TopK threshold refresh — while `current()` fires per batch. Cache the remapped expression per generation so per-batch calls short-circuit the walk. Reset semantics: * `update()` bumps `inner.generation`, invalidating the cache lookup on the next `current()`. * `with_new_children` clones the outer struct with its own remapped children; each derived filter gets its own fresh cache slot. * `from_parts` (proto deserialization) starts with an empty cache. TPCH SF1 (10 iterations): * Q17: 128ms -> 53ms (2.4x) * Total (22 queries): 743ms -> 644ms (-13%) TPCH row counts identical across all 22 queries pre/post fix. Adds three regression tests: * cache hits within a generation (same Arc returned) * update() invalidates (new Arc returned) * derived filters via with_new_children have independent caches
Contributor
There was a problem hiding this comment.
Pull request overview
This PR optimizes DynamicFilterPhysicalExpr::current() on the hot RowFilter per-batch evaluation path by caching the remapped expression per inner-generation, avoiding repeated transform_up remap walks for unchanged dynamic filters.
Changes:
- Add a per-generation
(generation, Arc<dyn PhysicalExpr>)cache forDynamicFilterPhysicalExpr::current(). - Update
current()to return the cached remapped expression when the generation matches, otherwise recompute and store. - Add regression tests covering cache hits within a generation, invalidation on
update(), and per-derived-filter cache isolation.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
…obber A slow `current()` call that observed an older `inner` generation could finish after a concurrent caller has published a newer entry, and its unconditional write would replace that newer entry with a stale one. Semantics were still correct (later readers see the mismatch and recompute), but subsequent generation-matched calls would miss the cache and redo the remap unnecessarily. Guard the write: only publish if the observed generation is at least as fresh as whatever is currently cached.
…s stable
`datafusion-proto` roundtrip tests compare `format!("{:?}", expr)` on
freshly-deserialized vs post-`current()`-called filters. Deriving
`Debug` for `DynamicFilterPhysicalExpr` prints the cache slot, which
differs across the assertion because one side has populated the cache
and the other hasn't. The cache is a pure optimization artifact —
exclude it from Debug output so equality by Debug format is preserved.
Fixes `test_dynamic_filter_plan_roundtrip_dedupe` and
`test_custom_node_with_dynamic_filter_dedup_roundtrip` regressions
observed on the "cargo test hash collisions (amd64)" CI job.
Contributor
|
run benchmark tcph10 env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/cache-dynamic-filter-current (ac24b01) to 65f9726 (merge-base) diff using: tcph10 File an issue against this benchmark runner |
Contributor
|
Note that this seems to be attacking the same thing as #20353, I don't remember why I gave up on that but I probably just missed something. |
Contributor
Author
|
Benchmark for this request failed. Last 20 lines of output: Click to expandFile an issue against this benchmark runner |
Contributor
Author
|
run benchmark tpch10 env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/cache-dynamic-filter-current (12ee030) to 21986cf (merge-base) diff using: tpch10 File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch10 — base (merge-base)
tpch10 — branch
File an issue against this benchmark runner |
Contributor
Author
|
run benchmark tpch env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/cache-dynamic-filter-current (12ee030) to 21986cf (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
Contributor
Author
|
The benchmark result is good: │ QQuery 17 │ 148.46 / 153.38 ±2.83 / 156.51 ms │ 71.07 / 72.02 ±0.57 / 72.84 ms │ +2.13x faster │
│ QQuery 18 │ 79.77 / 81.44 ±1.45 / 83.84 ms │ 74.40 / 74.66 ±0.20 / 74.89 ms │ +1.09x faster │ |
adriangb
approved these changes
Jul 13, 2026
adriangb
left a comment
adriangb
left a comment
Contributor
There was a problem hiding this comment.
Generally looks good to me! Leaving some minor nits.
| { | ||
| let mut cache = self.current_cache.write(); | ||
| let should_write = match cache.as_ref() { | ||
| Some((cached_gen, _)) => generation >= *cached_gen, |
Contributor
There was a problem hiding this comment.
Is >= the right choice here or would > result in less updates and still be correct?
Contributor
Author
There was a problem hiding this comment.
Thanks @adriangb addressed it now.
= → >: You're right, same-generation writes are redundant (identical remap output). Tightened the guard so it only skips + saves the write-lock take when there's already an entry at the same generation.
Comment on lines
+275
to
+287
| // Fast path: cache hit for the current generation. | ||
| let (expr, generation) = { | ||
| let inner = self.inner.read(); | ||
| (Arc::clone(inner.expr()), inner.generation) | ||
| }; | ||
| if let Some((cached_gen, cached_expr)) = self.current_cache.read().as_ref() | ||
| && *cached_gen == generation | ||
| { | ||
| return Ok(Arc::clone(cached_expr)); | ||
| } | ||
| // Slow path: (re)compute the remap and store it under a write lock. | ||
| let remapped = | ||
| Self::remap_children(&self.children, self.remapped_children.as_ref(), expr)?; |
Contributor
There was a problem hiding this comment.
Do we have tests that hit this under concurrency? I'm not sure how to inject behavior to make sure we hit all concurrency scenarios. We could also make a test that launches many threads and hot loop races them and asserts that some property is upheld. I just feel we should derisk this looking okay under single threaded test scenarios but then breaking under production when someone runs with 128 partitions or whatever.
Contributor
Author
There was a problem hiding this comment.
Good suggestion, addressed in latest PR, thanks!
Concurrency test: Added test_current_cache_concurrent_readers_and_writer
… test Two review nits from apache#23532: 1. Switch cache-write guard from `generation >= cached_gen` to `generation > cached_gen`. Same-generation writes are redundant — the cached and about-to-write remaps are semantically identical (same input expr, same remapped_children) — so skipping saves one write-lock take when multiple readers race a miss on the same generation. 2. Add `test_current_cache_concurrent_readers_and_writer`: spawns 8 reader threads hot-looping `current()` while a writer fires 200 `update()`s. Asserts: (a) `current()` never fails, (b) the cached generation observed by each reader is monotonically non-decreasing, (c) after the writer finishes and a final `current()` flushes, the cache generation equals `inner.generation`. Derisks the "single-thread test OK, 128-partition prod hangs / corrupts" class of bugs. Test helper `clone_with_remapped_children_for_test` (test-cfg only) lets the stress test obtain multiple `Arc<Self>` references without going through `with_new_children` (which would clear `remapped_children`).
Contributor
Author
Contributor
|
thanks, lgtm! |
Contributor
Author
|
run benchmarks env:
DATAFUSION_EXECUTION_PARQUET_PUSHDOWN_FILTERS: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/cache-dynamic-filter-current (41a43d3) to d0304b3 (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/cache-dynamic-filter-current (41a43d3) to d0304b3 (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing perf/cache-dynamic-filter-current (41a43d3) to d0304b3 (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpch — base (merge-base)
tpch — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usageclickbench_partitioned — base (merge-base)
clickbench_partitioned — branch
File an issue against this benchmark runner |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Rationale for this change
DynamicFilterPhysicalExpr::current()is invoked per batch on the RowFilter path (viaPhysicalExpr::evaluate). It callsremap_children, which does atransform_uptree walk. For dynamic filters that carry a largeInListExpr— e.g.HashJoinExecpushing down acol IN build_keyslist of ~150+ values — that walk is dominated byInListExpr::with_new_childrencloning the whole list on every call.The
inner.generationonly changes when the filter isupdated — once per HashJoin build, or once per TopK threshold refresh. Between updates everycurrent()call recomputes an identical remapped tree. Caching per generation lets per-batch calls short-circuit the walk.Measured impact (TPCH SF1, 10 iterations)
pushdown=truepushdown=truepushdown=falseQ17 (Nation-in-orders subquery) matches the
pushdown=falsebaseline after the fix — the DynamicFilter tax is fully eliminated on that shape. Row counts identical across all 22 queries pre/post fix.Profile before the fix:
InListExpr::with_new_childrenand itsdrop_gluesat at the top of Q17's samply flamegraph.Part of #20324 (parquet filter-pushdown regression EPIC).
What changes are included in this PR?
current_cache: Arc<RwLock<Option<(u64, Arc<dyn PhysicalExpr>)>>>onDynamicFilterPhysicalExpr(aliased asCurrentExprCacheto keep the type readable).current(): read the current(expr, generation)frominner, return the cachedArcif the cached generation matches; otherwise recompute viaremap_childrenand store the new pair under a write lock.update()bumpsinner.generation, so the nextcurrent()misses the cache and recomputes.with_new_childrenclones the outer struct with newremapped_children; the derived filter gets its own fresh cache slot.from_parts(proto deserialization) starts with an empty cache.Are these changes tested?
Yes — three new regression tests in
expressions::dynamic_filters::test:test_current_cache_hits_within_generation— three consecutivecurrent()calls return the sameArc(pointer-equal) at the same generation, proving the remap walk did not re-run.test_current_cache_invalidates_on_update— afterupdate()bumps the generation,current()returns a freshArc(not pointer-equal, and stringly distinct).test_current_cache_is_per_derived_filter— two filters derived viawith_new_childrenwith differentremapped_childrenproduce distinct cachedArcs and each hits its own cache on subsequent calls.All 21 tests in
expressions::dynamic_filters(18 existing + 3 new) pass.Are there any user-facing changes?
No. Internal caching only. No public API changes; behavior is identical modulo the generation-tied fast path.