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