Adaptive (runtime, stats-based) conjunct reordering for FilterExec — minimal core [benchmarking] - #14
Draft
adriangb wants to merge 3 commits into
Draft
Adaptive (runtime, stats-based) conjunct reordering for FilterExec — minimal core [benchmarking]#14adriangb wants to merge 3 commits into
adriangb wants to merge 3 commits into
Conversation
…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
github-actions
Bot
added
documentation
Owner
Author
|
run benchmark tpch10 clickbench_partitioned tpcds baseline:
ref: adaptive-filter-simple
changed:
ref: adaptive-filter-simple
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 |
2 similar comments
|
Benchmark for this request failed. Last 20 lines of output: Click to expandFile 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 |
Owner
Author
|
run benchmark tpch10 clickbench_partitioned tpcds changed:
env:
DATAFUSION_EXECUTION_ADAPTIVE_FILTER_REORDERING: "true" |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing adaptive-filter-simple (e3182a0) to ea2ffdb (merge-base) diff using: tpcds File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing adaptive-filter-simple (e3182a0) to ea2ffdb (merge-base) diff using: tpch10 File an issue against this benchmark runner |
|
🤖 Benchmark running (GKE) | trigger CPU Details (lscpu)Comparing adaptive-filter-simple (e3182a0) to ea2ffdb (merge-base) diff using: clickbench_partitioned 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 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
2 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.
Draft for benchmarking only. A dwindled-down version of
apache#22698, split into a 3-commit stack.
What changes are included in this PR?
Runtime, statistics-based conjunct reordering for
FilterExec, off by defaultbehind
datafusion.execution.adaptive_filter_reordering, in three reviewablecommits:
warm-up, rank by rows-discarded-per-ns, evaluate via a compact-once loop in
the better order.
first to settle publishes the order, others adopt it for one atomic load per
batch.
evaluate the plain predicate (byte-for-byte the flag-off path).
~906 lines vs the full PR's 3024. Drops the A/B arbiter, concurrent registry,
Welford/CI stats, and drift re-thaw, which benchmarking showed are not needed
for the core win.
Are these changes tested?
Yes — unit tests (compact-once result-equivalence, ranking, cross-stream
sharing, the no-reorder guard) and an end-to-end
adaptive_filter.sltassertingidentical results and EXPLAIN with the flag on and off.
Are there any user-facing changes?
One new experimental config option (default false). When enabled, the evaluation
order of a conjunction may change at runtime; results are unchanged.
🤖 Generated with Claude Code
https://claude.ai/code/session_01Lh7i9DyeFWuTFWjogrVNkb