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: [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]
