adriangb opened a new pull request, #25682: URL: https://github.com/apache/datafusion/pull/25682
## Which issue does this PR close? - Part of #22883 (experimental "optional filters" stack, design notes: https://claude.ai/artifact/SSz7t6hPyhFWp1MDPecVqt). Related to #20324, #3463, #19858. - **Depends on #25673 and #25674.** Review only the top commit. Works without the producer PR, but real queries only produce optional filters once that PR lands. ## Rationale for this change With `datafusion.execution.parquet.pushdown_filters = true`, the Parquet scan evaluates pushed-down dynamic filters row by row. When a dynamic filter removes few rows (for example a hash join where most probe keys match), we evaluate it and then the join checks the same rows again. This is a main cause of the regressions that keep `pushdown_filters` off by default (#20324, #19858). ## What changes are included in this PR? The Parquet scan uses `datafusion.execution.optional_filter_mode` (from #25674) for optional conjuncts (found with `split_optional`, root AND chain only). Required conjuncts do not change. | mode | optional conjuncts | |---|---| | `always` (default) | exactly today's behavior | | `pruning_only` | not added to the `RowFilter`; still used for file, row group, page and bloom filter pruning | | `adaptive` | each optional conjunct gets its own `RowFilter` predicate behind an `OptionalFilterGate`. A paused gate returns an all-true mask without evaluating. | Details: - Optional predicates always run after the required ones, so the gate measures only what the optional filter adds. - Pooled gate statistics live on the `ParquetSource` and are shared by all partitions and files. A dynamic filter is keyed by `expression_id()`, other optional filters by position. A new file starts from the pooled verdict. - A gate lives for one file and survives per-row-group row filter rebuilds. The dynamic filter inside `Optional` stays live, so the gate resets when the filter changes. - An optional conjunct that the row filter rejects for a file (for example after schema evolution) is dropped for that file. This is safe because it is optional (compare #25268). - New metrics: `optional_filter_rows_skipped`, `optional_filter_pauses`. - When `pushdown_filters = false` nothing changes. ## What is the testing strategy for this PR? - Unit tests for candidate building per mode, ordering, dropped rejected conjuncts, pausing with exact row counts, pooled seeding, site keys, and config plumbing. - Integration tests (`core/tests/parquet/optional_filters.rs`, 10 row groups × 2000 rows): identical results in all modes, metrics per mode, a selective filter never pauses, and a dynamic filter updated while its gate is paused takes effect at the next row group. - `optional_filters.slt`: join and TopK queries give the same results in all three modes. This file will show the new metrics once the producer PR lands. Benchmarks (TPC-H, TPC-DS, ClickBench with `pushdown_filters = true` and each mode) will follow once the producer PR is stacked in. ## Are there any user-facing changes? New behavior only when `optional_filter_mode` is set to `adaptive` or `pruning_only`. The default keeps today's behavior. Two new scan metrics. Known limitation: the mode is not serialized to protobuf (a decoded `ParquetSource` uses `always`). 🤖 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]
