[VL] support build left when Leftsemi appears

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxBackend.scala

@@ -34,7 +34,7 @@ import org.apache.gluten.utils._
 import org.apache.spark.sql.catalyst.catalog.BucketSpec
 import org.apache.spark.sql.catalyst.expressions.{Alias, CumeDist, DenseRank, Descending, Expression, Lag, Lead, NamedExpression, NthValue, NTile, PercentRank, RangeFrame, Rank, RowNumber, SortOrder, SpecialFrameBoundary, SpecifiedWindowFrame}
 import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, ApproximatePercentile, Count, HyperLogLogPlusPlus, Percentile}
-import org.apache.spark.sql.catalyst.plans.{JoinType, LeftOuter, RightOuter}
+import org.apache.spark.sql.catalyst.plans.{JoinType, LeftOuter, LeftSemi, RightOuter}
 import org.apache.spark.sql.catalyst.util.{CaseInsensitiveMap, CharVarcharUtils}
 import org.apache.spark.sql.connector.read.Scan
 import org.apache.spark.sql.execution.{ColumnarCachedBatchSerializer, SparkPlan}
@@ -506,15 +506,17 @@ object VeloxBackendSettings extends BackendSettingsApi {
         true
       } else {
         t match {
-          // OPPRO-266: For Velox backend, build right and left are both supported for
-          // LeftOuter.
-          // TODO: Support LeftSemi after resolve issue
-          // https://github.com/facebookincubator/velox/issues/9980
           case LeftOuter => true
+          case LeftSemi
+              if GlutenConfig.get.shjLeftSemiBuildLeftEnabled &&
+                SQLConf.get.adaptiveExecutionEnabled => true
           case _ => false
         }
       }
   }
+
+  override def semiAntiBuildLeftOutputsBuildOnly: Boolean = true
+
   override def supportHashBuildJoinTypeOnRight: JoinType => Boolean = {
     t =>
       if (super.supportHashBuildJoinTypeOnRight(t)) {

backends-velox/src/main/scala/org/apache/gluten/backendsapi/velox/VeloxRuleApi.scala

@@ -52,6 +52,9 @@ object VeloxRuleApi {
    * injected through [[injectLegacy]].
    */
   private def injectSpark(injector: SparkInjector): Unit = {
+    // QueryStagePrepRule: guards LeftSemi BuildLeft against unsafe conditions.
+    injector.injectQueryStagePrepRule(LeftSemiBuildLeftGuardRule.apply)
+
     // Inject the regular Spark rules directly.
     injector.injectOptimizerRule(CollectRewriteRule.apply)
     injector.injectOptimizerRule(HLLRewriteRule.apply)

backends-velox/src/main/scala/org/apache/gluten/extension/LeftSemiBuildLeftGuardRule.scala

@@ -0,0 +1,111 @@
+/*
+ * 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.gluten.extension
+
+import org.apache.gluten.config.GlutenConfig
+import org.apache.gluten.extension.columnar.rewrite.RewriteJoin
+import org.apache.gluten.extension.columnar.util.ShuffleSkewDetector
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.LeftSemi
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.joins.{ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.internal.SQLConf
+
+case class LeftSemiBuildLeftGuardRule(session: SparkSession)
+  extends Rule[SparkPlan]
+  with Logging {
+
+  override def apply(plan: SparkPlan): SparkPlan = {
+    val glutenConf = GlutenConfig.get
+    if (!glutenConf.shjLeftSemiBuildLeftEnabled) {
+      return plan
+    }
+    val skewJudgement = buildSkewJudgement()
+    val minRatio = glutenConf.shjLeftSemiBuildLeftMinRightToLeftRatio
+    val minRightBytes = glutenConf.shjLeftSemiBuildLeftMinRightBytes
+    plan.foreachUp {
+      case smj: SortMergeJoinExec if smj.joinType == LeftSemi =>
+        maybeTag(smj, smj.left, smj.right, "SMJ", skewJudgement, minRatio, minRightBytes)
+      case shj: ShuffledHashJoinExec if shj.joinType == LeftSemi =>
+        maybeTag(shj, shj.left, shj.right, "SHJ", skewJudgement, minRatio, minRightBytes)
+      case _ =>
+    }
+    plan
+  }
+
+  private def maybeTag(
+      join: SparkPlan,
+      leftSide: SparkPlan,
+      rightSide: SparkPlan,
+      kind: String,
+      skewJudgement: ShuffleSkewDetector.SkewJudgement,
+      minRatio: Double,
+      minRightBytes: Long): Unit = {
+    if (join.getTagValue(RewriteJoin.ForceShjBuildLeftTag).getOrElse(false)) {
+      return
+    }
+
+    val rightTotalBytes = ShuffleSkewDetector.totalBytes(rightSide) match {
+      case Some(bytes) => bytes
+      case None => return
+    }
+    if (rightTotalBytes < minRightBytes) {
+      logDebug(
+        s"LeftSemiBuildLeftGuardRule: right-side too small on LeftSemi $kind " +
+          s"(rightBytes=$rightTotalBytes < minRightBytes=$minRightBytes); skipping BuildLeft.")
+      return
+    }
+
+    val leftTotalBytes = ShuffleSkewDetector.totalBytes(leftSide) match {
+      case Some(bytes) => bytes
+      case None => return
+    }
+    if (leftTotalBytes > 0) {
+      val ratio = rightTotalBytes.toDouble / leftTotalBytes.toDouble
+      if (ratio < minRatio) {
+        logDebug(
+          f"LeftSemiBuildLeftGuardRule: insufficient right/left ratio on LeftSemi $kind " +
+            f"(rightBytes=$rightTotalBytes / leftBytes=$leftTotalBytes " +
+            f"= $ratio%.1f < minRatio=$minRatio%.1f); skipping BuildLeft.")
+        return
+      }
+    }
+
+    val rightStats = ShuffleSkewDetector.analyze(rightSide, skewJudgement)
+    if (rightStats.isSkewed) {
+      logDebug(
+        s"LeftSemiBuildLeftGuardRule: right-side partition skew on LeftSemi $kind " +
+          s"(totalBytes=${rightStats.totalBytes}, max=${rightStats.maxBytes}, " +
+          s"median=${rightStats.medianBytes}); skipping BuildLeft.")
+      return
+    }
+
+    join.setTagValue(RewriteJoin.ForceShjBuildLeftTag, true)
+  }
+
+  private def buildSkewJudgement(): ShuffleSkewDetector.SkewJudgement = {
+    val sqlConf = SQLConf.get
+    ShuffleSkewDetector.SkewJudgement(
+      factor = sqlConf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_FACTOR),
+      partitionThresholdBytes = sqlConf.getConf(SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD),
+      minTotalBytes = 0L
+    )
+  }
+}

backends-velox/src/test/scala/org/apache/gluten/execution/VeloxHashJoinSuite.scala

@@ -22,6 +22,7 @@ import org.apache.gluten.sql.shims.SparkShimLoader
 import org.apache.spark.SparkConf
 import org.apache.spark.sql.Row
 import org.apache.spark.sql.catalyst.expressions.AttributeReference
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight}
 import org.apache.spark.sql.execution.{ColumnarBroadcastExchangeExec, ColumnarSubqueryBroadcastExec, InputIteratorTransformer, SerializedHashTableBroadcastRelation}
 import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
 import org.apache.spark.sql.execution.joins.BuildSideRelation
@@ -655,4 +656,104 @@ class VeloxHashJoinSuite extends VeloxWholeStageTransformerSuite {
         })
   }
 
+  test("LeftSemi BuildLeft output correctness") {
+    withSQLConf(
+      ("spark.sql.autoBroadcastJoinThreshold", "-1"),
+      ("spark.sql.adaptive.enabled", "true"),
+      (GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true"),
+      (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key, "true"),
+      (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key, "0"),
+      (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key, "1.0")
+    ) {
+      withTable("semi_left", "semi_right") {
+        spark.sql("""
+          CREATE TABLE semi_left USING PARQUET
+          AS SELECT id as key, id * 10 as value FROM range(100)
+        """)
+        spark.sql("""
+          CREATE TABLE semi_right USING PARQUET
+          AS SELECT id * 2 as key FROM range(1000)
+        """)
+
+        val query = "SELECT key, value FROM semi_left WHERE key IN (SELECT key FROM semi_right)"
+        runQueryAndCompare(query) {
+          df =>
+            val plan = df.queryExecution.executedPlan
+            val shjJoins = plan.collect {
+              case shj: ShuffledHashJoinExecTransformer => shj
+            }
+            assert(shjJoins.nonEmpty, "Should use ShuffledHashJoin")
+            assert(
+              shjJoins.exists(_.joinBuildSide == BuildLeft),
+              "Should have at least one LeftSemi BuildLeft join")
+        }
+      }
+    }
+  }
+
+  test("LeftSemi BuildRight output correctness (regression)") {
+    withSQLConf(
+      ("spark.sql.autoBroadcastJoinThreshold", "-1"),
+      ("spark.sql.adaptive.enabled", "true"),
+      (GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true"),
+      (GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key, "false")
+    ) {
+      withTable("semi_left_br", "semi_right_br") {
+        spark.sql("""
+          CREATE TABLE semi_left_br USING PARQUET
+          AS SELECT id as key, id * 10 as value FROM range(100)
+        """)
+        spark.sql("""
+          CREATE TABLE semi_right_br USING PARQUET
+          AS SELECT id * 2 as key FROM range(1000)
+        """)
+
+        val query =
+          "SELECT key, value FROM semi_left_br WHERE key IN (SELECT key FROM semi_right_br)"
+        runQueryAndCompare(query) {
+          df =>
+            val plan = df.queryExecution.executedPlan
+            val shjJoins = plan.collect {
+              case shj: ShuffledHashJoinExecTransformer => shj
+            }
+            assert(shjJoins.nonEmpty, "Should use ShuffledHashJoin")
+            assert(
+              shjJoins.forall(_.joinBuildSide == BuildRight),
+              "All LeftSemi joins should use BuildRight when feature is disabled")
+        }
+      }
+    }
+  }
+
+  test("LeftAnti with ShuffledHashJoin output correctness") {
+    withSQLConf(
+      ("spark.sql.autoBroadcastJoinThreshold", "-1"),
+      ("spark.sql.adaptive.enabled", "true"),
+      (GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key, "true")
+    ) {
+      withTable("anti_left", "anti_right") {
+        spark.sql("""
+          CREATE TABLE anti_left USING PARQUET
+          AS SELECT id as key, id * 10 as value FROM range(100)
+        """)
+        spark.sql("""
+          CREATE TABLE anti_right USING PARQUET
+          AS SELECT id * 2 as key FROM range(1000)
+        """)
+
+        val query =
+          """SELECT key, value FROM anti_left a
+            |WHERE NOT EXISTS (SELECT 1 FROM anti_right b WHERE a.key = b.key)""".stripMargin
+        runQueryAndCompare(query) {
+          df =>
+            val plan = df.queryExecution.executedPlan
+            val shjJoins = plan.collect {
+              case shj: ShuffledHashJoinExecTransformer => shj
+            }
+            assert(shjJoins.nonEmpty, "Should use ShuffledHashJoin")
+        }
+      }
+    }
+  }
+
 }