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>
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. The gate is a per-stream state machine (Evaluate / Paused with exponential backoff) that restarts evaluation when the filter generation changes. Gates of one plan site share lock-free pooled statistics, which seed the state of new gates. Add the `datafusion.execution.optional_filter_mode` (always | adaptive | pruning_only, default always) and `datafusion.execution.optional_filter_max_pass_ratio` (default 0.8) options. No operator uses the gate yet, so behavior does not 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 #25674 +/- ##
========================================
Coverage 82.48% 82.49%
========================================
Files 1140 1142 +2
Lines 437614 438513 +899
Branches 437614 438513 +899
========================================
+ Hits 360986 361766 +780
- Misses 54834 54925 +91
- Partials 21794 21822 +28 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
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. Review only the top commit.Rationale for this change
An optional filter (#25673) can be skipped without changing the result. Consumers (the Parquet row filter,
FilterExec) need one shared way to decide when to skip it: stop evaluating a filter that removes too few rows, and start again when it may have become useful (for example, when a TopK threshold tightens).Earlier attempts (#20160, #22236) either decided only when a file opened, or kept state in a shared expression so that one partition's measurements paused the filter in another partition. This PR keeps the decision per stream and has no wall clock, so the tests are deterministic.
What changes are included in this PR?
datafusion_physical_expr::optional_filter_gate:OptionalFilterGate(one per stream, not shared). A state machine with two states:sample_batchesbatches, if the pass ratio is abovemax_pass_ratio, pause.max_pause_batches). N resets when a probe finds the filter selective.snapshot_generation, for example a dynamic filter update), the gate goes back to Evaluate and resets N.begin_batch(num_rows) -> GateDecisionplusrecord(rows_in, rows_out)for consumers that evaluate the filter themselves (for example anArrowPredicate), andevaluate(batch) -> Option<BooleanArray>for the simple case.OptionalFilterSiteStats(one per operator and filter, shared by its streams). A new stream starts from the pooled verdict for the current generation. So a new file does not pay for a new sample. A running stream is never changed by another stream.Config (no behavior change yet, no consumer uses the gate in this PR):
datafusion.execution.optional_filter_mode:always(default, today's behavior) |adaptive|pruning_only.datafusion.execution.optional_filter_max_pass_ratio: default0.8.What is the testing strategy for this PR?
14 deterministic unit tests: pause on a non-selective filter, stay on for a selective one, backoff doubling and reset, generation reset (with a real
DynamicFilterPhysicalExpr), a TopK-like tightening pattern, skewed input, pooled seeding, and the "no backoff growth while paused" regression. Config docs andinformation_schema.sltare updated.Are there any user-facing changes?
Two new config options, with defaults that keep today's behavior. New public API in
datafusion-physical-expr.🤖 Generated with Claude Code