Adaptive (runtime, stats-based) conjunct reordering for FilterExec#22698
Open
adriangb wants to merge 9 commits into
Open
Adaptive (runtime, stats-based) conjunct reordering for FilterExec#22698adriangb wants to merge 9 commits into
adriangb wants to merge 9 commits into
Conversation
github-actions
Bot
added
documentation
Contributor
Author
|
run benchmarks |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing lift-selectivity-stats (5e71ea4) to 85bc5ef (merge-base) diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing lift-selectivity-stats (5e71ea4) to 85bc5ef (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing lift-selectivity-stats (5e71ea4) to 85bc5ef (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
Thank you for opening this pull request! Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch). Details |
|
🤖 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 |
Contributor
Author
|
run benchmarks env:
DATAFUSION_EXECUTION_ADAPTIVE_FILTER_REORDERING: true |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing lift-selectivity-stats (5e71ea4) to 85bc5ef (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing lift-selectivity-stats (5e71ea4) to 85bc5ef (merge-base) diff using: tpch File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing lift-selectivity-stats (5e71ea4) to 85bc5ef (merge-base) diff using: clickbench_partitioned 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 |
Contributor
Author
|
run benchmarks env:
DATAFUSION_EXECUTION_ADAPTIVE_FILTER_REORDERING: true |
|
Benchmark for this request failed. Last 20 lines of output: Click to expandFile an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing lift-selectivity-stats (5e71ea4) to 85bc5ef (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
Benchmark for this request failed. Last 20 lines of output: Click to expandFile 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 |
adriangb
force-pushed
the
lift-selectivity-stats
branch
3 times, most recently
from
a24471d to
4d7b733
Compare
|
🤖 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 |
|
🤖 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 |
Contributor
Author
Contributor
Author
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing a456356 (a456356) to a456356 diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing a456356 (a456356) to a456356 diff using: tpch10 File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing a456356 (a456356) to a456356 diff using: clickbench_partitioned File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing a456356 (a456356) to a456356 diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing a456356 (a456356) to a456356 diff using: tpch10 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 Usagetpch10 — base (merge-base)
tpch10 — branch
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 |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing a456356 (a456356) to a456356 diff using: tpcds 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 |
|
🤖 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 |
|
🤖 Benchmark completed (GKE) | trigger Instance: CPU Details (lscpu)Details
Resource Usagetpcds — base (merge-base)
tpcds — branch
File an issue against this benchmark runner |
Contributor
Author
Benchmark summary (adaptive filter reordering)Ran two matched experiments on the PR head (
I ran it twice; the control is what makes the numbers trustworthy, because the Wins (reproduced across both runs)
Both are multi-conjunct No regressions
Net: two solid wins on the target case, no confirmed regressions, no |
…act-once core) Add runtime, statistics-based conjunct reordering for `FilterExec`, off by default behind `datafusion.execution.adaptive_filter_reordering`. A conjunctive predicate is evaluated through a compact-once loop: conjunct masks are AND-combined and the working batch is physically compacted to the surviving rows once the accumulated mask is selective enough, so a selective conjunct shrinks the batch the conjuncts after it must decode. This compaction — not reordering a fused `BinaryExpr` AND, which does not compact between conjuncts — is the source of the win, and reordering compounds it. Each conjunct is timed and counted on the rows that reach it during a short warm-up; the conjuncts are then ranked by rows discarded per nanosecond (`(1 - pass_rate) / cost_per_row`) and, if the ranked order is materially cheaper than the written one, it is adopted and frozen. Results, plan, and EXPLAIN are unchanged; volatile predicates are never reordered. This is the minimal core. Benchmarks (predicate_eval) confirm it captures the "buried selective conjunct" wins (costsel_q01 ~-14%, width ~-12%) but also that compact-once regresses cheap-predicate conjunctions (cardinality k8 ~+37%) where the compaction overhead is not repaid — a guard that keeps the plain fused evaluation for those is added in the next commit. Cross-stream sharing, drift re-measurement, and confidence-interval statistics are later layers. Tested by unit tests (compact-once result-equivalence in any order, ranking, expected-cost weighting, adopt/keep decisions) and an end-to-end `adaptive_filter.slt` asserting identical results and EXPLAIN with the flag on and off. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Lh7i9DyeFWuTFWjogrVNkb
A `FilterExec` is split across many partition streams, each seeing a slice of the data. With per-stream warm-up, every stream pays its own measurement cost, and when each stream is only a handful of batches long that warm-up is most of its work — so the reordering win never materialises (and the warm-up overhead shows up as a regression). Benchmarked: at 12 partitions the costsel_q01 win collapsed from -67% (single stream) to -14%. Share the measurements. `AdaptiveFilterShared` holds a per-conjunct stats pool plus a settled-order epoch, common to every stream of one `FilterExec`. Each stream measures a batch into a local accumulator and folds it into the pool; the first stream to reach `WARMUP_BATCHES` pooled batches decides the order and publishes it by bumping the epoch. Other streams poll the epoch with one relaxed atomic load per batch and adopt the published order without paying warm-up. The warm-up is thus paid ~once per query, not once per stream. Restores the recovered win at default partitioning: width -56..-66%, costsel_q01 -58%, cardinality k16 -26% (was -12%, -14%, -6% without sharing), matching or beating the full design. Steady-state regressions on conjunctions where compaction does not pay (neutral_q61 ~+11%, cardinality k4 ~+5%, costsel_q02/q03 ~+3-5%) remain — a Fused-vs-CompactOnce guard addresses those in the next commit. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Lh7i9DyeFWuTFWjogrVNkb
The compact-once loop wins by gating expensive conjuncts behind a selective one, but its per-conjunct bookkeeping (mask AND, true_count, the compaction copy) is pure overhead when there is nothing to gate. On a conjunction of interchangeable predicates — several equally expensive, equally unselective regexps, say — the warm-up settles on the written order (nothing to reorder) yet still paid compact-once on every batch, regressing ~11% vs the plain predicate (predicate_eval neutral_q61). Guard it: compact-once is adopted only when the warm-up actually reorders the conjuncts. When the settled order equals the written order, evaluate the predicate as-is — byte-for-byte the flag-off path, zero overhead. Since every real win reorders (a selective conjunct moves toward the front), this keeps the full win while removing the no-reorder regression. predicate_eval (vs flag off): neutral_q61 +11% -> ~0; wins preserved (costsel_q01 -60%, width -58..-66%, cardinality k16 -31%). A small residual remains on low-cardinality cheap conjunctions that do reorder (k4/k8 ~+3-4%), where compaction's cost is not repaid by gating so few/cheap predicates; the full champion/challenger arbiter regresses these more (~+10%), so a heavier guard is not worth it here. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Lh7i9DyeFWuTFWjogrVNkb
Two points raised by @xudong963 that carried over into the compact-once rewrite of the adaptive `FilterExec` conjunct evaluator: - Replace `.expect("u32 live")` on the live-row index downcast with a let-else returning `internal_err!`, so a broken invariant surfaces as a clean error rather than a panic. - Add a `debug_assert!` documenting that live-row indices are tracked in arrow's `u32` `filter`/`take` index space, making the `num_rows as u32` cast's precondition explicit. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01R1mvYrjFyTy2kbBoGrWzT6
adriangb
force-pushed
the
lift-selectivity-stats
branch
from
a456356 to
511ed51
Compare
@xudong963 noted the pooled adaptive-conjunct measurements live on the `FilterExec` plan node and are reused by every `execute()` call, leaking the learned conjunct order across independent executions. Implement `ExecutionPlan::reset_state` for `FilterExec` — the sanctioned mechanism for exactly this (its trait docs cite `DynamicFilterPhysicalExpr`; `CrossJoinExec`, `HashJoinExec`, and `SortExec` use it for their build-side / dynamic-filter state). It returns a fresh node with a new `AdaptiveFilterShared` (and fresh metrics), so a re-executed plan re-learns from scratch, while preserving the still-valid predicate, input, and cached plan properties. Reordering only ever affects performance, never results, so this closes a perf-staleness gap, not a correctness bug. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01R1mvYrjFyTy2kbBoGrWzT6
- An empty batch no longer consumes the warm-up: a run of empty batches would settle the written order on no evidence, permanently disabling adaptation for the stream. - A conjunct evaluated faster than the timer's resolution now clamps its cost to 1ns instead of dropping out of the ranking as unmeasured (which sorted the cheapest conjunct last — backwards). - The u32::MAX row-count guard is now a real internal error instead of a debug_assert; in release the indices would have silently wrapped. Also documents the known limitations of the one-shot, conditional-stats settle (correlated conjuncts, no drift re-measurement) in the module doc. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DgXQmKrKgre4epbcSxUHNo
The previous test stored the table as a single batch, so with WARMUP_BATCHES = 8 the flag-on queries only ever exercised the measuring path. Store 4000 rows as 64-row batches so the warm-up completes and the settled (possibly reordered) path runs end-to-end, and add a query whose conjunct produces NULLs to exercise the null-mask path through real SQL. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DgXQmKrKgre4epbcSxUHNo
The config docs claimed reordering was the only observable difference; in fact side effects of fallible predicates can change even when no reorder is adopted, because while measuring (and after a reorder) conjuncts are evaluated only on rows that survived the conjuncts before them. Say so explicitly, with an example, and note in FilterExec that Clone sharing the pooled measurements is deliberate. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DgXQmKrKgre4epbcSxUHNo
adriangb
force-pushed
the
lift-selectivity-stats
branch
from
511ed51 to
b42f823
Compare
…atch strategy Two review responses: - AdaptiveConjunction::try_new no longer takes an `enabled` bool that short-circuits to None; whether the feature is on is FilterExec's policy, so the flag check moves to FilterExec::execute and try_new answers only the structural question (reorderable, non-volatile conjunction). - The evaluator's per-batch behaviour is now observable: evaluate is a thin wrapper over evaluate_traced, which also reports the BatchStrategy used (Measure / Fused / Reordered). Two scenario tests exercise the input/output contract end to end — batches in, masks + strategy trace out — with per-conjunct costs injected by seeding the shared pool with synthetic measurements (the stand-in for a mocked clock), so which strategy gets adopted is deterministic: warm-up settles on a reorder for cheap-unselective + expensive-selective conjuncts, and on the written fused predicate for interchangeable ones. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DgXQmKrKgre4epbcSxUHNo
adriangb
force-pushed
the
lift-selectivity-stats
branch
from
49dd3d8 to
a991d38
Compare
Contributor
Author
|
@alamb @xudong963 I think this is ready for review. The meat of the complexity is in |
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
4 participants
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?
(static cheap/expensive heuristic reordering). No single issue; happy to file
one if useful.
Rationale for this change
Predicate evaluation order matters: running a selective predicate first lets it
gate the work of the predicates after it. The static cheap/expensive heuristic
(#22343) sorts conjuncts into two cost classes and stable-sorts within each, so
it does nothing to order multiple similarly-expensive predicates; and
BinaryExpr'sANDshort-circuit only gates on a leftmost selective conjunct.So a conjunction of several expensive predicates whose selective member is not
written first is evaluated with every predicate scanning ~every row — and
neither mechanism fixes it.
This PR adds runtime, statistics-based conjunct reordering for
FilterExec:it measures each conjunct's selectivity and cost on the rows that actually reach
it and runs the ones that discard the most rows per unit of CPU time first.
Maximising discards-per-second is exactly minimising
cost_per_row / (1 - pass_rate),the classic optimal ordering key for independent conjuncts.
It is off by default (
datafusion.execution.adaptive_filter_reordering).What changes are included in this PR?
Everything lives in a new private module,
datafusion/physical-plan/src/adaptive_filter.rs;FilterExecitself gains onefield, an
Optionon the stream, and a two-arm match in the poll loop. No plan,EXPLAIN, statistics, or public-API changes beyond the config flag.masks combined with cheap bitwise
ANDs, and the working batch is physicallycompacted to the survivors once the accumulated mask keeps ≤20% of rows. This
compaction is what makes ordering pay off — a fused
BinaryExprANDchainevaluates ~every conjunct on ~every row regardless of order. (An earlier
revision of this PR validated adoptions with A/B trials and a
champion/challenger arbiter; benchmarking showed the compact-once loop is
where the win actually comes from, so this PR ships the simple one-shot
cost-model design and leaves A/B validation as possible future work.)
are measured for 8 batches; conjuncts are then ranked by discards-per-
nanosecond, and the ranked order is adopted only if its expected cost is
materially (≥5%) cheaper than the written order. If not, the settled path
evaluates the original fused predicate — identical to the feature being off.
Compact-once is used only in service of a reorder.
AdaptiveFilterShared): all partition streams ofone
FilterExecpool their measurements; the first to finish the warm-uppublishes the settled order via an epoch atomic (one relaxed load per batch
for the others). The warm-up is paid roughly once per query, not once per
stream — which is what makes the win materialise when each stream only sees a
handful of batches.
reset_stategives re-executions (e.g. recursive queries) fresh measurements; predicate
rewrites reset the pooled stats; empty batches don't consume the warm-up;
results are provably order-independent (a conjunction's value doesn't depend
on evaluation order).
execution.adaptive_filter_reordering(experimental,default false), plus regenerated
configs.md/information_schema.Known limitations (documented in the module): the statistics are
conditional — each conjunct is measured on the rows that survived the ones
before it — so correlated conjuncts can be misjudged, and the settle is
one-shot with no drift re-measurement. The material-win guard makes adoption
conservative. Drift re-thaw, confidence intervals, and A/B-validated adoption
are explicitly future work layered on this core.
Reuse note (#22237 / #22144): the adaptive parquet filter-placement work
needs the same ideas (per-predicate selectivity/cost accumulators, a
measure → settle → publish pattern) but a different arrangement type and
lifetime. Everything here is deliberately
pub(crate): rather than pre-build ashared abstraction in
physical-expr-commonfrom a single consumer, the planis to lift the small shared pieces once the second consumer's shape is
concrete. Nothing in this PR commits any public API beyond the config flag.
Are these changes tested?
Yes:
(mask always equals the plain predicate's, before and after settling),
no-reorder-runs-plain-predicate, cross-stream pooling/adoption, empty-batch
and timer-resolution edge cases, and
reset_state.adaptive_filter.slt: the table is stored as 64-row batches so the flag-onqueries cross the warm-up and exercise the settled (reordered) path
end-to-end, including a NULL-producing conjunct; results and
EXPLAINareidentical with the flag on and off.
Are there any user-facing changes?
One new config option,
datafusion.execution.adaptive_filter_reordering(experimental, default false). When enabled, query results never change, but
observable side effects of fallible predicates can — even when no reorder is
adopted — because while measuring (and after a reorder) conjuncts are evaluated
only on rows that survived the conjuncts before them, so an error the fused
predicate would have raised on an already-filtered row (e.g.
b <> 0 AND 1/b > 2evaluating1/bon every row) may not occur. Predicatescontaining volatile expressions are never reordered.