feat: implement native native existence join support

native/core/src/execution/planner.rs

@@ -2199,6 +2199,7 @@ impl PhysicalPlanner {
             Ok(JoinType::FullOuter) => DFJoinType::Full,
             Ok(JoinType::LeftSemi) => DFJoinType::LeftSemi,
             Ok(JoinType::LeftAnti) => DFJoinType::LeftAnti,
+            Ok(JoinType::Existence) => DFJoinType::LeftMark,
             Err(_) => {
                 return Err(GeneralError(format!(
                     "Unsupported join type: {join_type:?}"

native/proto/src/proto/operator.proto

@@ -414,6 +414,7 @@ enum JoinType {
   FullOuter = 3;
   LeftSemi = 4;
   LeftAnti = 5;
+  Existence = 6;
 }
 
 enum BuildSide {

spark/src/main/scala/org/apache/comet/CometConf.scala

@@ -261,6 +261,13 @@ object CometConf extends ShimCometConf {
     createExecEnabledConfig("broadcastNestedLoopJoin", defaultValue = true)
   val COMET_EXEC_SORT_MERGE_JOIN_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("sortMergeJoin", defaultValue = true)
+  val COMET_EXEC_EXISTENCE_JOIN_ENABLED: ConfigEntry[Boolean] =
+    createExecEnabledConfig(
+      "existenceJoin",
+      defaultValue = false,
+      notes = Some(
+        "This enables native ExistenceJoin support (EXISTS/NOT EXISTS combined with OR). " +
+          "This is highly experimental and disabled by default"))
   val COMET_EXEC_AGGREGATE_ENABLED: ConfigEntry[Boolean] =
     createExecEnabledConfig("aggregate", defaultValue = true)
   val COMET_EXEC_COLLECT_LIMIT_ENABLED: ConfigEntry[Boolean] =

spark/src/main/scala/org/apache/spark/sql/comet/operators.scala

@@ -2056,6 +2056,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
         case _ =>
           // Spark doesn't support other join types
           withFallbackReason(join, s"Unsupported join type ${join.joinType}")
@@ -2327,6 +2329,11 @@ case class CometHashJoinExec(
   override def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =
     this.copy(left = newLeft, right = newRight)
 
+  override def producedAttributes: AttributeSet = joinType match {
+    case ExistenceJoin(exists) => AttributeSet(exists)
+    case _ => AttributeSet.empty
+  }
+
   override def stringArgs: Iterator[Any] =
     Iterator(leftKeys, rightKeys, joinType, buildSide, condition, left, right)
 
@@ -2468,6 +2475,11 @@ case class CometBroadcastHashJoinExec(
   override def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =
     this.copy(left = newLeft, right = newRight)
 
+  override def producedAttributes: AttributeSet = joinType match {
+    case ExistenceJoin(exists) => AttributeSet(exists)
+    case _ => AttributeSet.empty
+  }
+
   override def stringArgs: Iterator[Any] =
     Iterator(leftKeys, rightKeys, joinType, condition, buildSide, left, right)
 
@@ -2549,6 +2561,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) =>
+          JoinType.Existence
         case _ =>
           // Spark doesn't support other join types
           withFallbackReason(join, s"Unsupported join type ${join.joinType}")
@@ -2658,6 +2672,11 @@ case class CometSortMergeJoinExec(
   override def withNewChildrenInternal(newLeft: SparkPlan, newRight: SparkPlan): SparkPlan =
     this.copy(left = newLeft, right = newRight)
 
+  override def producedAttributes: AttributeSet = joinType match {
+    case ExistenceJoin(exists) => AttributeSet(exists)
+    case _ => AttributeSet.empty
+  }
+
   override def stringArgs: Iterator[Any] =
     Iterator(leftKeys, rightKeys, joinType, condition, left, right)
 

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
+WHERE l.region = 'US' OR EXISTS (SELECT 1 FROM ex_right r WHERE r.k = l.k)
+ORDER BY l.id
+
+-- ============================================================
+-- EXISTS with OR: SMJ
+-- ============================================================
+
+query
+SELECT /*+ MERGE(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
+
+-- ============================================================
+-- Empty build: every left row is unmatched, only OR-arm rows survive
+-- ============================================================
+
+query
+SELECT /*+ BROADCAST(ex_right_empty) */ * FROM ex_left l
+WHERE l.region = 'US' OR EXISTS (SELECT 1 FROM ex_right_empty r WHERE r.k = l.k)
+ORDER BY l.id
+
+query
+SELECT /*+ MERGE(ex_right_empty) */ * FROM ex_left l
+WHERE l.region = 'US' OR EXISTS (SELECT 1 FROM ex_right_empty r WHERE r.k = l.k)
+ORDER BY l.id
+
+-- ============================================================
+-- Right side has no NULL: NULL-keyed left row reaches the marker
+-- evaluation but cannot match (NULL = anything is NULL → false),
+-- so its exists tag is false.
+-- ============================================================
+
+query
+SELECT /*+ BROADCAST(ex_right_no_nulls) */ * FROM ex_left l
+WHERE l.region = 'US' OR EXISTS (SELECT 1 FROM ex_right_no_nulls r WHERE r.k = l.k)
+ORDER BY l.id
+
+-- ============================================================
+-- NOT EXISTS combined with OR: also lowers to ExistenceJoin
+-- (the optimizer flips the marker via NOT in the filter).
+-- ============================================================
+
+query
+SELECT /*+ BROADCAST(ex_right) */ * FROM ex_left l
+WHERE l.region = 'US' OR NOT EXISTS (SELECT 1 FROM ex_right r WHERE r.k = l.k)
+ORDER BY l.id
+
+-- ============================================================
+-- Build with duplicate keys: marker is "at least one match", so duplicates
+-- on the right must not multiply the output.
+-- ============================================================
+
+query
+SELECT /*+ BROADCAST(ex_right_dups) */ * FROM ex_left l
+WHERE l.region = 'US' OR EXISTS (SELECT 1 FROM ex_right_dups r WHERE r.k = l.k)
+ORDER BY l.id
+
+-- ============================================================
+-- Marker used inside a more complex predicate (NOT exists OR ...).
+-- ============================================================
+
+query
+SELECT /*+ BROADCAST(ex_right) */ id, k, region FROM ex_left l
+WHERE l.id > 1
+  AND (l.region = 'US' OR NOT EXISTS (SELECT 1 FROM ex_right r WHERE r.k = l.k))
+ORDER BY l.id
+
+-- ============================================================
+-- Multi-column correlation
+-- ============================================================
+
+statement
+CREATE TABLE ex_left_multi(id int, k1 int, k2 int) USING parquet
+
+statement
+INSERT INTO ex_left_multi VALUES (1, 1, 100), (2, 2, 200), (3, 1, 300)
+
+statement
+CREATE TABLE ex_right_multi(k1 int, k2 int) USING parquet
+
+statement
+INSERT INTO ex_right_multi VALUES (1, 100), (2, 999)
+
+query
+SELECT /*+ BROADCAST(ex_right_multi) */ * FROM ex_left_multi l
+WHERE l.id > 0
+  AND (l.k1 = 1
+       OR EXISTS (SELECT 1 FROM ex_right_multi r WHERE r.k1 = l.k1 AND r.k2 = l.k2))
+ORDER BY l.id

spark/src/test/scala/org/apache/comet/exec/CometJoinSuite.scala

@@ -25,7 +25,7 @@ import org.scalatest.Tag
 import org.apache.spark.sql.CometTestBase
 import org.apache.spark.sql.catalyst.TableIdentifier
 import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation
-import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometSortMergeJoinExec}
+import org.apache.spark.sql.comet.{CometBroadcastExchangeExec, CometBroadcastHashJoinExec, CometBroadcastNestedLoopJoinExec, CometHashJoinExec, CometSortMergeJoinExec}
 import org.apache.spark.sql.execution.adaptive.AQEShuffleReadExec
 import org.apache.spark.sql.internal.SQLConf
 
@@ -945,4 +945,56 @@ class CometJoinSuite extends CometTestBase {
       }
     }
   }
+
+  test("ExistenceJoin via BroadcastHashJoin (EXISTS combined with OR)") {
+    withSQLConf(
+      CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB",
+      SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "10MB") {
+      withParquetTable((0 until 100).map(i => (i, if (i % 3 == 0) "US" else "EU")), "tbl_a") {
+        withParquetTable((0 until 30).map(i => (i, i + 1)), "tbl_b") {
+          val df = sql("SELECT * FROM tbl_a a " +
+            "WHERE a._2 = 'US' OR EXISTS (SELECT /*+ BROADCAST(b) */ 1 FROM tbl_b b WHERE b._1 = a._1)")
+          checkSparkAnswerAndOperator(
+            df,
+            Seq(classOf[CometBroadcastExchangeExec], classOf[CometBroadcastHashJoinExec]))
+        }
+      }
+    }
+  }
+
+  test("ExistenceJoin via ShuffledHashJoin (EXISTS combined with OR)") {
+    withSQLConf(
+      CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.key -> "true",
+      SQLConf.PREFER_SORTMERGEJOIN.key -> "false",
+      "spark.sql.join.forceApplyShuffledHashJoin" -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+      withParquetTable((0 until 100).map(i => (i, if (i % 3 == 0) "US" else "EU")), "tbl_a") {
+        withParquetTable((0 until 30).map(i => (i, i + 1)), "tbl_b") {
+          val df = sql(
+            "SELECT * FROM tbl_a a " +
+              "WHERE a._2 = 'US' OR EXISTS (SELECT 1 FROM tbl_b b WHERE b._1 = a._1)")
+          checkSparkAnswerAndOperator(df, Seq(classOf[CometHashJoinExec]))
+        }
+      }
+    }
+  }
+
+  test("ExistenceJoin via SortMergeJoin (EXISTS combined with OR)") {
+    withSQLConf(
+      CometConf.COMET_EXEC_EXISTENCE_JOIN_ENABLED.key -> "true",
+      SQLConf.PREFER_SORTMERGEJOIN.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
+      withParquetTable((0 until 100).map(i => (i, if (i % 3 == 0) "US" else "EU")), "tbl_a") {
+        withParquetTable((0 until 30).map(i => (i, i + 1)), "tbl_b") {
+          val df = sql(
+            "SELECT * FROM tbl_a a " +
+              "WHERE a._2 = 'US' OR EXISTS (SELECT 1 FROM tbl_b b WHERE b._1 = a._1)")
+          checkSparkAnswerAndOperator(df, Seq(classOf[CometSortMergeJoinExec]))
+        }
+      }
+    }
+  }
 }