You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "sunchao (via GitHub)" <gi...@apache.org> on 2026/09/29 22:28:27 UTC
[PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
sunchao opened a new pull request, #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431
## Which issue does this PR close?
No linked issue.
## Rationale for this change
A join can often rule out most probe-side rows once its build side is complete. Comet already uses that information to prune eligible Parquet readers, but an intervening predicate such as `quantity > 0` stops propagation. The reader may therefore scan many rows that cannot join, even though the predicate is deterministic.
Crossing that predicate is an execution-policy choice. Consider a residual `10 / key > 0`: a row with `key = 0` can raise a division-by-zero error even when the build contains only key 3. If reader pruning removes that row first, its expression is never evaluated. Determinism alone does not make that change invisible, so broader propagation should require an explicit opt-in.
## What changes are included in this PR?
This PR adds `spark.comet.exec.join.dynamicFilter.allowDeterministicFilterPushdown`, defaulting to **false**. With both this setting and `spark.comet.exec.join.dynamicFilter.enabled` enabled, a join runtime filter may cross a deterministic probe-side residual and reach the existing Parquet reader path. For example, a build containing item IDs 17 and 42 can let the reader skip a row group containing only item 99 before evaluating its quantity predicate.
The residual stays in the plan and still checks every retained row. A row that passes the runtime filter can still fail the residual, and errors on retained rows still propagate. The new setting explicitly permits skipping expression errors on rows eliminated earlier. With the setting disabled, propagation remains limited to direct-column `IS NOT NULL` checks combined with `AND`.
Spark certifies that the complete residual is deterministic and sends that permission with the filter. Native planning uses the metadata from the actual probe side, including build-right joins, and preserves it when rebuilding the filter for an execution. Nondeterministic predicates, limits, computed projections, and execution boundaries remain barriers. Existing per-file Parquet schema-conversion safeguards are preserved, including when the new setting is enabled.
## How are these changes tested?
- All 15 `CometJoinSuite` tests selected by `join dynamic filter` pass on Spark 4.1.3. The new regression covers both build sides, compares the option off/on, checks residual rejection of a key present in the build, and verifies reader attachment, row-group pruning, scan bytes, and metric ownership. The suite also checks seeded-random evaluation and Parquet schema errors with the option enabled.
- All 40 native tests selected by `filter` pass. A real Parquet regression verifies that opt-in pruning skips division by zero only for an eliminated row; the default path and a retained zero-key row still fail. Additional tests cover metadata selection and preservation during plan rebuilding.
- `cargo clippy --locked --offline --all-targets --workspace -- -D warnings`, the native build, Maven Spotless, Markdown formatting, and `git diff --check` pass. The Spark run used the newly built JNI library.
The scan-byte assertion is regression coverage for the fixture, not an end-to-end performance claim. Broader Spark SQL coverage is requested through the `run-spark-4.1-tests` CI label.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao merged PR #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431#issuecomment-6108429257
cc @andygrove @viirya @comphead
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431#discussion_r4243806645
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -544,14 +613,18 @@ class CometJoinSuite extends CometTestBase {
.parquet(probePath.getCanonicalPath)
.createOrReplaceTempView("dynamic_schema_probe")
withParquetTable(Seq(Tuple1(0)), "dynamic_schema_build") {
- for ((comet, dynamicFilter) <- Seq((false, false), (true, false), (true, true))) {
+ for {
+ (comet, dynamicFilter) <- Seq((false, false), (true, false), (true, true))
+ allowed <- Seq(false, true)
+ } {
withSQLConf(
CometConf.COMET_ENABLED.key -> comet.toString,
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN.key -> allowed.toString,
CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.key -> dynamicFilter.toString) {
val df = sql(
"SELECT /*+ BROADCAST(b) */ p.probe_key, p.payload " +
"FROM dynamic_schema_probe p JOIN dynamic_schema_build b " +
- "ON p.probe_key = b._1")
+ "ON p.probe_key = b._1 WHERE pmod(p.payload, 2) >= 0")
Review Comment:
With the `pmod` residual added, the `dynamicFilter=true, allowed=false` case no longer reaches the attached-reader path. It now exercises the same path as `dynamicFilter=false`. The attached path is only covered by `allowed=true`, and nothing confirms that the reader filter actually attaches there. Metrics from a failed task aren't merged, so that's hard to assert after the fact. Could we keep the original residual-free query in the loop as well? That keeps the schema-conversion guard covered on the attached path regardless of the new flag.
##########
native/core/src/execution/planner.rs:
##########
@@ -2541,12 +2542,21 @@ impl PhysicalPlanner {
}
/// Keep the Spark filter's metric identity when its reader is replaced for an execution.
- fn prepare_probe_filter_for_runtime_reader(plan: Arc<SparkPlan>) -> Arc<SparkPlan> {
+ fn prepare_probe_filter_for_runtime_reader(
Review Comment:
This conflicts with main after #6433, which turned this into the recursive `prepare_probe_for_runtime_filters`. That version wraps nested `FilterExec` and `ProjectionExec` nodes too, not just the probe root. Right now the permission comes from a single `source: &Operator` matched against the top node. When rebasing, could the proto `Operator` children be walked alongside the `SparkPlan` children, so each wrapped filter reads its own `Filter` metadata? A mismatch falls back to `false` today, which is safe, but a misaligned walk could hand one filter's permission to another.
It would also help to say whether `place` in `early.rs` should consult `allows_runtime_filter_pushdown`. I'd lean towards no, since early consumers don't prune readers and the config is described in terms of reaching Parquet readers. A short comment there would keep the two paths from drifting. After the rebase, could we add native tests for a `Projection -> Filter -> Scan` probe and a `Filter -> Filter -> Scan` probe so those shapes are pinned down?
##########
docs/source/user-guide/latest/tuning/operators.md:
##########
@@ -59,6 +59,29 @@ boundaries. A shuffled hash join can still filter probe batches after shuffle, b
its filter back to an earlier scan stage. Compare the [runtime-filter and scan metrics](../metrics.md#hash-joins)
with the setting disabled to distinguish reduced hash-probe work from reader I/O savings.
+### Join runtime filters through residual predicates
+
+Native join runtime filtering is enabled with
+`spark.comet.exec.join.dynamicFilter.enabled`. It can reject probe rows whose keys
+are absent from the completed build side and use eligible Parquet readers to prune
+row groups. By default, an intervening residual filter permits reader propagation
+only for direct-column `IS NOT NULL` checks combined with `AND`.
+
+Set `spark.comet.exec.join.dynamicFilter.allowDeterministicFilterPushdown=true` to
+let the join filter pass through other deterministic residual predicates, such as
+`quantity > 0` or a Spark-generated Bloom check. The original residual and the
+exact join remain in place. For example, when the build contains only item IDs
+17 and 42, a row for item 99 can be rejected before evaluating its quantity.
+Nondeterministic predicates, limits, computed projections, and execution boundaries
Review Comment:
`try_attach_parquet_reader_filter` stops at any `ProjectionExec`, not only computed ones. Spark's `FileSourceStrategy` adds a `ProjectExec` above the filter whenever a residual column isn't part of the output. So a query like `SELECT p.item_id ... WHERE p.quantity > 0` won't prune, and that is the example this section uses. The existing test at `CometJoinSuite.scala:489` keeps `payload` in the output to avoid that projection, and the new test does the same.
Would you consider letting reader attachment cross column-only `CometProjectionExec` nodes? `early.rs` on main already remaps the key through column-only projections, so the logic could probably be shared. If that's better done separately, could the docs say that any projection between the residual and the join stops propagation? Could we also open a tracking issue for crossing column-only projections and link it here?
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -489,6 +489,20 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
+ val COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN: ConfigEntry[Boolean] =
Review Comment:
The same "computed projections" wording appears in this description a few lines down, so it would need the same fix as the docs.
Separately, I'd like to understand why this needs its own key. Spark already skips residual conjuncts on rows eliminated by DPP, or by Parquet row-group pruning on other pushed predicates. A `10 / b > 0` conjunct is never evaluated for rows in a group pruned by `a > 5`. So skipping errors on rows that can't join seems to fit Spark's existing semantics. `dynamicFilter.enabled` is already opt-in, and this key would be covered by the versioning policy. Is the second key worth the long-term commitment, or could this behavior come with `dynamicFilter.enabled`? I'm open to either. I'd just like the reasoning on record.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on code in PR #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431#discussion_r4315280302
##########
native/core/src/execution/planner.rs:
##########
@@ -2541,12 +2542,21 @@ impl PhysicalPlanner {
}
/// Keep the Spark filter's metric identity when its reader is replaced for an execution.
- fn prepare_probe_filter_for_runtime_reader(plan: Arc<SparkPlan>) -> Arc<SparkPlan> {
+ fn prepare_probe_filter_for_runtime_reader(
Review Comment:
Rebased onto current main in b9230be2f. `prepare_probe_for_runtime_filters` now walks the serialized Operator and SparkPlan children together, checking plan ID, operator type, child counts, and native/Spark child identity before reading each filter's permission. A mismatch defaults permissions to false for that subtree while retaining the existing metric wrappers.
Added native tests for `Projection -> Filter -> Scan`, `Filter -> Filter -> Scan` with all combinations of permissions, and mismatched metadata/tree boundaries. Added the comment in `early.rs`: this permission controls reader pruning and does not relax early-consumer expression barriers.
##########
docs/source/user-guide/latest/tuning/operators.md:
##########
@@ -59,6 +59,29 @@ boundaries. A shuffled hash join can still filter probe batches after shuffle, b
its filter back to an earlier scan stage. Compare the [runtime-filter and scan metrics](../metrics.md#hash-joins)
with the setting disabled to distinguish reduced hash-probe work from reader I/O savings.
+### Join runtime filters through residual predicates
+
+Native join runtime filtering is enabled with
+`spark.comet.exec.join.dynamicFilter.enabled`. It can reject probe rows whose keys
+are absent from the completed build side and use eligible Parquet readers to prune
+row groups. By default, an intervening residual filter permits reader propagation
+only for direct-column `IS NOT NULL` checks combined with `AND`.
+
+Set `spark.comet.exec.join.dynamicFilter.allowDeterministicFilterPushdown=true` to
+let the join filter pass through other deterministic residual predicates, such as
+`quantity > 0` or a Spark-generated Bloom check. The original residual and the
+exact join remain in place. For example, when the build contains only item IDs
+17 and 42, a row for item 99 can be rejected before evaluating its quantity.
+Nondeterministic predicates, limits, computed projections, and execution boundaries
Review Comment:
Kept column-only reader traversal as a separate follow-up and opened #7113. The docs now state that every standalone projection blocks reader attachment, including column-only projections Spark inserts to drop residual-only columns. The example explicitly retains `quantity` alongside `item_id`; projection inside the scan remains supported. The follow-up is linked from the tuning guide.
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -489,6 +489,20 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(false)
+ val COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN: ConfigEntry[Boolean] =
Review Comment:
Updated the configuration description to say all standalone projections, including column-only projections.
Agreed that Spark can already avoid these errors through DPP or pushed predicates. The additional key is a compatibility and tuning choice, rather than a claim that Spark must evaluate every residual. Existing dynamic filtering preserves residual evaluation except direct null checks; this traversal changes observable errors for plans without those other pruning opportunities. I kept the separate supported knob so users can retain that behavior while enabling dynamic filtering, accepting its versioning commitment. The tuning guide now records that rationale, and the native division-by-zero regression pins the behavior with the flag off/on and with an error on a retained row.
##########
spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala:
##########
@@ -544,14 +613,18 @@ class CometJoinSuite extends CometTestBase {
.parquet(probePath.getCanonicalPath)
.createOrReplaceTempView("dynamic_schema_probe")
withParquetTable(Seq(Tuple1(0)), "dynamic_schema_build") {
- for ((comet, dynamicFilter) <- Seq((false, false), (true, false), (true, true))) {
+ for {
+ (comet, dynamicFilter) <- Seq((false, false), (true, false), (true, true))
+ allowed <- Seq(false, true)
+ } {
withSQLConf(
CometConf.COMET_ENABLED.key -> comet.toString,
+ CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ALLOW_DETERMINISTIC_FILTER_PUSHDOWN.key -> allowed.toString,
CometConf.COMET_EXEC_JOIN_DYNAMIC_FILTER_ENABLED.key -> dynamicFilter.toString) {
val df = sql(
"SELECT /*+ BROADCAST(b) */ p.probe_key, p.payload " +
"FROM dynamic_schema_probe p JOIN dynamic_schema_build b " +
- "ON p.probe_key = b._1")
+ "ON p.probe_key = b._1 WHERE pmod(p.payload, 2) >= 0")
Review Comment:
Restored the residual-free query alongside the `pmod` residual in the schema-conversion test. Both now run for every existing Comet/dynamic-filter/permission combination, so the original attached-reader case remains covered when the new flag is false as well as true. Failure clues include the permission and residual. Included in b9230be2f.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431#issuecomment-5982634329
Rebased onto current main in `0843ef7d74bb4b7174d04b8d5c0e37c5ef0057b0`. Resolved the extracted `CometFilterExec` API and test imports, kept the current public operators crate boundary, and moved the residual-filter documentation into the new operator tuning page.
Native core and test compilation (`cargo check --locked --offline -p datafusion-comet --tests`), the operators crate check, Rust formatting, and whitespace checks pass. Spark 4.1 SQL CI remains enabled. Runtime CI at the new head is pending; no runtime test or performance result from the older head is being claimed for this rebase.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431#issuecomment-6270307672
Updated in b9230be2f: rebased onto current main, aligned recursive per-filter permissions with the serialized plan, added nested-plan/mismatch regressions, restored residual-free schema-error coverage, and corrected the projection limits and configuration rationale. Replied to all four review threads; column-only reader traversal is tracked in #7113.
Local validation passed: 64 focused native tests, native test compilation, Clippy for core/operators with warnings denied, Rust formatting, full Markdown formatting, Spotless, and whitespace checks. Spark validation is blocked locally by DNS failure at the configured Maven mirror while resolving `jackson-bom:2.21.2`.
The branch is conflict-free. Current-head CI is running with Spark 4.1 SQL coverage enabled; Preflight and CodeQL passed, and full runtime CI is still pending.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat: opt in to join filter pushdown through deterministic residuals [datafusion-comet]
Posted by "sunchao (via GitHub)" <gi...@apache.org>.
sunchao commented on PR #6431:
URL: https://github.com/apache/datafusion-comet/pull/6431#issuecomment-6273940250
Merging, thanks!
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org