Conversation
…orrectness Add `OptionalFilterPhysicalExpr`, a transparent wrapper that marks a filter as optional: a consumer can skip it without changing the query result. A consumer can skip it only when the wrapper is a direct conjunct of the root AND chain of its predicate. In all other positions the wrapper is transparent, because `evaluate()` always evaluates the inner expression. `snapshot()` returns the inner expression, so pruning sees through it. Also add: - `split_optional`, `is_optional_filter`, `as_dynamic_filter` and `debug_assert_optional_on_root_chain` helpers in `physical_expr::utils` - `PhysicalOptionalFilterNode` proto message (field 29 in `PhysicalExprNode`) with self-encoding `try_to_proto`/`try_from_proto` No producer uses the wrapper yet, so there is no behavior change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
github-actions
Bot
added
documentation
|
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 |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #25683 +/- ##
==========================================
+ Coverage 82.48% 82.54% +0.05%
==========================================
Files 1140 1144 +4
Lines 437614 440482 +2868
Branches 437614 440482 +2868
==========================================
+ Hits 360986 363591 +2605
- Misses 54834 54970 +136
- Partials 21794 21921 +127 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
adriangb
force-pushed
the
filter-exec-adaptive
branch
2 times, most recently
from
September 24, 2026 06:11
98e8970 to
50eff49
Compare
Add a runtime gate that pauses optional filters (filters that are not needed for correctness, such as hash join and TopK dynamic filters) when they remove too few rows or when they cost more than they save. The gate is a per-stream state machine (Evaluate / Paused with exponential backoff) that restarts evaluation when the filter changes. The gate finds the dynamic filters one time with `DynamicFilterTracking::classify` and then polls their subscriptions, so a check does not walk the filter tree. Gates of one plan site share lock-free pooled statistics (rows and evaluation time), which seed the state of new gates. At the end of each window of evaluated batches the gate pauses the filter if its pass ratio is larger than `max_pass_ratio`, or if its evaluation time is larger than the work that the removed rows save: `(rows_in - rows_out) * saving_ns_per_row`. The saving for each row is the configured minimum plus an optional value that the consumer measures and updates (`MeasuredRowSaving`). The cost check has a margin (pause above 1.1x the saving, resume below 0.9x) so that a filter does not switch on and off when cost and saving are almost equal. Add the `filter_stats` module with the shared measurement primitives: an injectable monotonic `Clock` (`SystemClock`, and `ManualClock` for deterministic tests) and `FilterCost` (rows in, rows out, time, and the derived pass ratio, cost per row and rows removed per nanosecond). Add a crate-private generation tag to `DynamicFilterTracker` (the sum of the latest observed generations of the dynamic filters, including complete ones), so the pooled statistics can tag verdicts by filter version without calling `snapshot_generation`. Add the `datafusion.execution.optional_filter_mode` (always | adaptive | pruning_only, default always), `datafusion.execution.optional_filter_max_pass_ratio` (default 0.8) and `datafusion.execution.optional_filter_min_saving_ns_per_row` (default 20) options. No operator uses the gate yet, so behavior does not change. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
`FilterExec` now handles optional conjuncts (see `OptionalFilterPhysicalExpr`) according to `datafusion.execution.optional_filter_mode`: - `always` (default): the predicate is not split; today's behavior. - `pruning_only`: optional conjuncts are not evaluated. - `adaptive`: required conjuncts run as one ordinary `BinaryExpr` AND chain, and each optional conjunct runs behind its own `OptionalFilterGate`, so optional conjuncts that remove too few rows, or that cost more than they save, are paused. The gates measure the evaluation time; the saving for each removed row is `datafusion.execution.optional_filter_min_saving_ns_per_row`. Pooled gate statistics are tied to the current `TaskContext`, so a new execution of the same plan starts fresh, and `reset_state` clears them. New metrics: `optional_filter_rows_skipped`, `optional_filter_pauses`, `optional_filter_eval_time`. Also add adaptive conjunct reordering behind `datafusion.execution.adaptive_filter_reordering` (default `false`): each stream measures its conjuncts over a warm-up, ranks them by rows dropped per nanosecond, and adopts a new order (built as a plain `BinaryExpr` AND chain) only if the estimated cost, using `BinaryExpr`'s pre-selection rule, is at least 5% lower. The measurements use the shared `Clock` and `FilterCost` of `datafusion_physical_expr::filter_stats`, so tests use a `ManualClock`. New metric: `adaptive_reorders`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
adriangb
force-pushed
the
filter-exec-adaptive
branch
from
September 24, 2026 15:24
50eff49 to
8730312
Compare
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This was referenced Sep 24, 2026
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.
Which issue does this PR close?
OptionalFilterPhysicalExprto mark filters not needed for correctness #25673 and feat: addOptionalFilterGateandoptional_filter_modeconfig #25674. Review only the top two commits (the feature and a license header fix).Rationale for this change
FilterExecis the second consumer of optional filters. It is also where the order of conjuncts matters most. This PR starts from the review of #22698:BinaryExprAND chainBinaryExprpre-selectionPRE_SELECTION_THRESHOLD(nowpub,#[doc(hidden)])TaskContext;reset_stateclears themClock(ManualClockin tests)unwrap, noUInt32ArraydowncastWhat changes are included in this PR?
Part 1: optional conjuncts (
datafusion.execution.optional_filter_mode, from #25674)FilterExecalways(default)pruning_onlyFilterExeccannot prune)adaptiveBinaryExprAND chain; each optional conjunct runs behind anOptionalFilterGate; the saving per removed row isoptional_filter_min_saving_ns_per_rowOptional(a = b)is ignored by predicate analysis).optional_filter_rows_skipped,optional_filter_pauses,optional_filter_eval_time(registered only when a stream gates optional filters).Part 2: adaptive conjunct reordering (
datafusion.execution.adaptive_filter_reordering, defaultfalse)graph LR W["8 warm-up batches:<br/>measure each conjunct"] --> R["rank by rows dropped per ns"] R --> D{"estimated cost ≥ 5% lower?"} D -- yes --> N["new BinaryExpr AND chain"] D -- no --> O["keep the original predicate"]filter_stats::{FilterCost, Clock}(feat: addOptionalFilterGateandoptional_filter_modeconfig #25674). State is per stream.adaptive_reorders.What is the testing strategy for this PR?
NOT(Optional(x))is never skipped; wrapped equalities add no equivalences; statistics per context, reset byreset_stateadaptive_filter_reordering.sltcargo test -p datafusion-physical-plan filter, the physical-plan and physical-expr lib tests, proto tests and the full sqllogictest suite pass.Known gaps: in
adaptivemode, optional conjuncts run after the required chain and are not ranked with it. Reordering decides once per stream, but a gate can change its decision, so ranking them together needs a gate-aware node in the AND chain.Are there any user-facing changes?
One new config option (
adaptive_filter_reordering, default off) and new metrics. The default behavior does not change.🤖 Generated with Claude Code