adriangb opened a new pull request, #25674: URL: https://github.com/apache/datafusion/pull/25674
## Which issue does this PR close? - Part of #22883 (experimental "optional filters" stack, design notes: https://claude.ai/artifact/SSz7t6hPyhFWp1MDPecVqt). - **Depends on #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: - **Evaluate**: evaluate the filter and count rows in and out. After `sample_batches` batches, if the pass ratio is above `max_pass_ratio`, pause. - **Paused**: skip the filter for N batches, then probe again. N doubles after each pause that the probe confirms (up to `max_pause_batches`). N resets when a probe finds the filter selective. - When the filter's generation changes (`snapshot_generation`, for example a dynamic filter update), the gate goes back to Evaluate and resets N. - API: `begin_batch(num_rows) -> GateDecision` plus `record(rows_in, rows_out)` for consumers that evaluate the filter themselves (for example an `ArrowPredicate`), and `evaluate(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. - While paused, the gate does not change its counters or backoff (a regression test covers this). 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`: default `0.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 and `information_schema.slt` are 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](https://claude.com/claude-code) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
