sunchao commented on code in PR #4587:
URL: https://github.com/apache/datafusion-comet/pull/4587#discussion_r4042586357
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2428,6 +2428,8 @@ trait CometHashJoin {
case FullOuter => JoinType.FullOuter
case LeftSemi => JoinType.LeftSemi
case LeftAnti => JoinType.LeftAnti
+ case ExistenceJoin(_) if
CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.get(join.conf) =>
Review Comment:
[P2] Preserve first-match evaluation for residual predicates
Could we keep hash existence joins with residual predicates in Spark until
this evaluation difference is handled? Spark stops after the first qualifying
candidate, but DataFusion 55.1 evaluates the hash-join filter on the candidate
batch before reducing it to a marker. With equal right-hand keys, a successful
candidate can therefore be followed by a candidate that throws, although Spark
never evaluates it.
For example, I ran Spark 4.1.3 with a single-file left input `(id=1,
vals=[1.0], threshold=-0.0)` and right rows `(id=1, idx=0), (id=1, idx=1)`. An
`EXISTS` correlation on `r.id=l.id AND element_at(l.vals,r.idx)>l.threshold`
returns `true`: Spark visits index 1 first and stops. The native
candidate-batch evaluation reaches index 0, which Comet's `ListExtract` rejects
even with ANSI disabled.
There is also a wrong-marker case for the residual `r.value > l.value` with
equal integer keys, `r.value=+0.0`, and `l.value=-0.0`. Spark returns false,
while the pinned Arrow 59.3 comparison returns true. `normalizePlan` only
rewrites project/filter expressions, so it does not normalize this join
condition.
These checks used fresh Spark/Arrow probes and the pinned DataFusion source,
not an end-to-end run of this Comet head.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2943,6 +2955,8 @@ object CometSortMergeJoinExec extends
CometOperatorSerde[SortMergeJoinExec] {
case FullOuter => JoinType.FullOuter
case LeftSemi => JoinType.LeftSemi
case LeftAnti => JoinType.LeftAnti
+ case ExistenceJoin(_) if
CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.get(join.conf) =>
Review Comment:
[P2] Keep existence SMJ behind fallback until output retention is bounded
Could we leave existence sort-merge joins in Spark until the pinned
DataFusion implementation yields completed output inside its spanning-key and
exhausted-inner loops? In DataFusion 55.1,
`BitwiseSortMergeJoinStream::process_unfiltered_match_loop` and `drain_outer`
repeatedly push outer batches into `BatchCoalescer`, but
`emit_completed_batches` only runs after those loops return. The reservation
tracks the inner key buffer, not that queued output.
Consequently, a large equal-key group or an empty right input can retain an
entire left partition outside the configured memory budget before returning its
first output batch. That can cause executor OOM even though the join reports
little or no reserved memory.
I checked this control flow in the exact 55.1 source matching `Cargo.lock`.
A separate cached 54.1 probe corroborated the pattern: all 102,400 outer rows
were consumed before the first 1,024-row result with a one-byte pool and zero
reported reservation. That cached execution is not a run of the current
dependency. A regression should verify incremental output and bounded retained
memory for both empty-right and skewed-key inputs.
##########
spark/src/main/scala/org/apache/spark/sql/comet/operators.scala:
##########
@@ -2428,6 +2428,8 @@ trait CometHashJoin {
case FullOuter => JoinType.FullOuter
case LeftSemi => JoinType.LeftSemi
case LeftAnti => JoinType.LeftAnti
+ case ExistenceJoin(_) if
CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.get(join.conf) =>
+ JoinType.Existence
Review Comment:
[P2] Check computed-key evaluation safety on both inputs
Could we add a recursive safety check for existence join keys, beyond
checking whether each expression can be serialized? Spark packs pairs of
integer keys and can skip the second expression when the first is NULL, whereas
the native join evaluates the key expressions independently over the batch.
With ANSI enabled, left rows `(id=1, prefix=NULL, txt='bad'), (id=2,
prefix=1, txt='1')`, and right row `(a=1,b=1)`, Spark 4.1.3 returns
`(1,false),(2,true)` for `EXISTS (... WHERE r.a=l.prefix AND r.b=CAST(l.txt AS
INT))`. Native key evaluation reaches the invalid cast on the first row.
Similarly, `LIMIT 1` can stop Spark after a valid probe row while a native key
batch still evaluates a later invalid row.
A computed integer key can also conceal a floating comparison: `CASE WHEN x
< 0.0 THEN 1 ELSE 2 END` yields 2 in Spark for `x=-0.0`, but the pinned Arrow
comparison selects 1. Normalizing the final floating join key does not help
because this key is an integer. The guard needs to inspect descendants on both
inputs and retain native execution for expressions safe under batch evaluation.
The Spark baseline and Arrow 59.3 comparison were executed locally. The
current native key path was source-checked; full Comet integration was not run.
##########
spark/src/test/scala/org/apache/spark/sql/benchmark/CometExistenceJoinBenchmark.scala:
##########
@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.spark.sql.benchmark
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.internal.SQLConf
+
+import org.apache.comet.{CometConf, CometSparkSessionExtensions}
+
+/**
+ * Benchmark to measure performance of Comet's ExistenceJoin support across
the three join
+ * physical operators (BHJ, SHJ, SMJ). To run this benchmark:
+ * {{{
+ * SPARK_GENERATE_BENCHMARK_FILES=1 make
benchmark-org.apache.spark.sql.benchmark.CometExistenceJoinBenchmark
+ * }}}
+ * Results will be written to
"spark/benchmarks/CometExistenceJoinBenchmark-**results.txt".
+ */
+object CometExistenceJoinBenchmark extends CometBenchmarkBase {
+
+ override def getSparkSession: SparkSession = {
+ val conf = new SparkConf()
+ .setAppName("CometExistenceJoinBenchmark")
+ .set("spark.master", "local[5]")
+ .setIfMissing("spark.driver.memory", "3g")
+ .setIfMissing("spark.executor.memory", "3g")
+ .set(
+ "spark.shuffle.manager",
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
+
+ val sparkSession = SparkSession.builder
+ .config(conf)
+ .withExtensions(new CometSparkSessionExtensions)
+ .getOrCreate()
+
+ sparkSession.conf.set(CometConf.COMET_ENABLED.key, "false")
+ sparkSession.conf.set(CometConf.COMET_EXEC_ENABLED.key, "false")
+ sparkSession.conf.set(SQLConf.ANSI_ENABLED.key, "false")
+ sparkSession.conf.set("spark.sql.shuffle.partitions", "2")
+
+ sparkSession
+ }
+
+ override def runCometBenchmark(mainArgs: Array[String]): Unit = {
+ val probeRows = 1024 * 1024
+ val buildRows = 10000
+
+ withTempPath { dir =>
+ withTempTable("probe", "build") {
+ spark
+ .range(probeRows)
+ .selectExpr("id AS k", "CASE WHEN id % 3 = 0 THEN 'US' ELSE 'EU' END
AS region")
+ .write
+ .parquet(s"${dir.getAbsolutePath}/probe")
+ spark
+ .range(buildRows)
+ .selectExpr("id * 7 AS k")
+ .write
+ .parquet(s"${dir.getAbsolutePath}/build")
+
+
spark.read.parquet(s"${dir.getAbsolutePath}/probe").createOrReplaceTempView("probe")
+
spark.read.parquet(s"${dir.getAbsolutePath}/build").createOrReplaceTempView("build")
+
+ val query =
+ "SELECT count(*) FROM probe p " +
+ "WHERE p.region = 'US' OR EXISTS (SELECT 1 FROM build b WHERE b.k
= p.k)"
+
+ runBenchmark("ExistenceJoin - BroadcastHashJoin") {
+ runExpressionBenchmark(
+ "exists OR predicate (BHJ)",
+ probeRows,
+ query,
+ Map(
+ CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.key -> "true",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+ SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB"))
+ }
+
+ runBenchmark("ExistenceJoin - ShuffledHashJoin") {
+ runExpressionBenchmark(
+ "exists OR predicate (SHJ)",
+ probeRows,
+ query,
+ Map(
+ CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.key -> "true",
+ SQLConf.PREFER_SORTMERGEJOIN.key -> "false",
Review Comment:
[P2] Apply benchmark strategy settings to the Spark baseline too
This is the same benchmark mismatch noted in the [existing
discussion](https://github.com/apache/datafusion-comet/pull/4587#issuecomment-5723010195),
anchored at the config map that causes it. `runExpressionBenchmark` treats
this map as `extraCometConfigs`, so the SHJ/SMJ planning settings apply only to
Comet. The Spark arm retains normal broadcast planning for the small build
table, and the reported SHJ/SMJ ratios compare different strategies between
engines.
Could we wrap each case in shared `withSQLConf` strategy settings, leaving
the Comet feature flag in `extraCometConfigs`, and assert the actual join class
for both arms? Subquery-local hints are also viable on Spark 4.1.3, as verified
in the SQL-fixture comment. This would make the benchmark labels and
performance comparisons reflect the strategy being measured.
##########
spark/src/test/resources/sql-tests/join/existence_join.sql:
##########
@@ -0,0 +1,167 @@
+-- Licensed to the Apache Software Foundation (ASF) under one
+-- or more contributor license agreements. See the NOTICE file
+-- distributed with this work for additional information
+-- regarding copyright ownership. The ASF licenses this file
+-- to you under the Apache License, Version 2.0 (the
+-- "License"); you may not use this file except in compliance
+-- with the License. You may obtain a copy of the License at
+--
+-- http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing,
+-- software distributed under the License is distributed on an
+-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+-- KIND, either express or implied. See the License for the
+-- specific language governing permissions and limitations
+-- under the License.
+
+-- Tests for ExistenceJoin: produced when EXISTS / NOT EXISTS is combined
+-- with another predicate via OR, preventing rewrite to LeftSemi / LeftAnti.
+-- Each query runs against the three physical join strategies (BHJ, SHJ,
+-- SMJ) via hints, so we exercise CometBroadcastHashJoinExec,
+-- CometHashJoinExec, and CometSortMergeJoinExec all carrying joinType =
+-- ExistenceJoin.
+
+-- Native ExistenceJoin support is experimental and disabled by default.
+-- Config: spark.comet.exec.existenceJoin.enabled=true
+
+-- ============================================================
+-- Setup: covers NULLs, duplicates, empty build side
+-- ============================================================
+
+statement
+CREATE TABLE ex_left(id int, k int, region string) USING parquet
+
+statement
+INSERT INTO ex_left VALUES
+ (1, 1, 'US'),
+ (2, 2, 'EU'),
+ (3, NULL, 'US'),
+ (4, 4, 'EU'),
+ (5, 5, 'EU')
+
+statement
+CREATE TABLE ex_right(id int, k int) USING parquet
+
+statement
+INSERT INTO ex_right VALUES (10, 1), (11, 2), (12, 2), (13, NULL)
+
+statement
+CREATE TABLE ex_right_no_nulls(id int, k int) USING parquet
+
+statement
+INSERT INTO ex_right_no_nulls VALUES (10, 1), (11, 5)
+
+statement
+CREATE TABLE ex_right_empty(id int, k int) USING parquet
+
+statement
+CREATE TABLE ex_right_dups(id int, k int) USING parquet
+
+statement
+INSERT INTO ex_right_dups VALUES (10, 1), (11, 1), (12, 1), (13, 2)
+
+-- ============================================================
+-- EXISTS with OR: BHJ build-right
+-- ============================================================
+
+query
+SELECT /*+ BROADCAST(ex_right) */ * FROM ex_left l
+WHERE l.region = 'US' OR EXISTS (SELECT 1 FROM ex_right r WHERE r.k = l.k)
+ORDER BY l.id
+
+-- ============================================================
+-- EXISTS with OR: SHJ build-right
+-- ============================================================
+
+query
+SELECT /*+ SHUFFLE_HASH(ex_right) */ * FROM ex_left l
Review Comment:
[P2] Place strategy hints in the subquery that owns the relation
Following up on the [fixture concern already
raised](https://github.com/apache/datafusion-comet/pull/4587#issuecomment-5723010195):
I confirmed on Spark 4.1.3 with AQE disabled that the outer
`SHUFFLE_HASH(ex_right)` and `MERGE(ex_right)` hints cannot resolve `ex_right`,
emit warnings, and both produce a broadcast hash join.
One correction to that discussion: putting the hint inside `EXISTS` does
work in this version. `EXISTS (SELECT /*+ SHUFFLE_HASH(r) */ 1 FROM ex_right r
WHERE r.k=l.k)` selects `ShuffledHashJoin`, and the corresponding `MERGE(r)`
selects `SortMergeJoin` in the fresh plan probes.
Could we move the hints into their subqueries and assert the selected
physical strategy, or use a configuration matrix that enforces the same
coverage? The current fixture does not establish null/duplicate/empty-input
coverage across the named strategies. The separate Scala tests do cover all
three operators for their simple unique, non-null integer case.
##########
native/core/src/execution/planner.rs:
##########
@@ -2699,6 +2699,7 @@ impl PhysicalPlanner {
Ok(JoinType::FullOuter) => DFJoinType::Full,
Ok(JoinType::LeftSemi) => DFJoinType::LeftSemi,
Ok(JoinType::LeftAnti) => DFJoinType::LeftAnti,
+ Ok(JoinType::Existence) => DFJoinType::LeftMark,
Review Comment:
[P2] Avoid enumerating every duplicate build match
For Spark's `BuildRight`, the later input swap turns this `LeftMark` into
`RightMark`. The pinned DataFusion 55.1 hash lookup enumerates every equal-key
candidate pair before the RightMark path reduces duplicate probe indices to
Boolean markers. With N identical build keys and M matching probe rows, that
means N*M candidate matches for M output rows, rather than Spark's existence
lookup stopping after a match.
Could we deduplicate the evaluated build-key tuples for residual-free
existence joins, or use a mark-join implementation that stops after the first
match? For example, 10,000 identical build keys and 1,000 matching probes imply
10 million candidates even though only 1,000 markers are needed. The new
benchmark's distinct build keys do not exercise this case.
I verified that Spark 4.1.3 preserves duplicate build rows in this plan
without inserting a distinct aggregate, and checked the duplicate-chain
traversal in the exact DataFusion 55.1 source. Any deduplication helper should
account for its memory and support bounded spilling. It must not discard
distinct right payloads while residual predicates can still use them. No
current-version performance timing is claimed here.
--
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]
