[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}
@@ -508,13 +508,22 @@ object VeloxBackendSettings extends BackendSettingsApi {
         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
+          // Velox maps LeftSemi BuildLeft to kRightSemiFilter (via Substrait
+          // JOIN_TYPE_RIGHT_SEMI) and fully supports it in both HashBuild and HashProbe.
+          // Requires both opt-in config AND AQE enabled: the runtime guard rule
+          // (LeftSemiBuildLeftGuardRule) only fires under AQE, so without AQE there is
+          // no safety net against skew/size regressions.
+          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

@@ -51,6 +51,12 @@ object VeloxRuleApi {
    * injected through [[injectLegacy]].
    */
   private def injectSpark(injector: SparkInjector): Unit = {
+    // QueryStagePrepRule: fires after each AQE shuffle stage materializes, so mapStats are
+    // available. Guards LeftSemi BuildLeft against unsafe conditions (right-side too small,
+    // insufficient right/left ratio, or probe-side partition skew) by tagging the join to
+    // force BuildRight.
+    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,147 @@
+/*
+ * 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
+
+/**
+ * QueryStagePrepRule that fires after each AQE shuffle stage completes, i.e. once `mapStats` are
+ * available.
+ *
+ * For a LeftSemi join (either [[SortMergeJoinExec]] or already-rewritten [[ShuffledHashJoinExec]])
+ * it decides whether BuildLeft is unsafe and, if so, sets [[RewriteJoin.ForceShjBuildRightTag]] on
+ * the join node. The tag is later consumed by:
+ *   - [[RewriteJoin.getSmjBuildSide]] (SMJ path).
+ *   - [[org.apache.gluten.extension.columnar.offload.OffloadJoin.getShjBuildSide]] (SHJ path).
+ *
+ * Three reasons to reject BuildLeft (OR-ed together):
+ *   - right-side shuffle too small (`< minRightBytes`): BuildLeft is only profitable at scale; on
+ *     small shuffles the plan is fragile and BuildLeft can regress badly.
+ *   - right-side partition skew: AQE `OptimizeSkewedJoin` cannot split the probe side of a
+ *     ShuffledHashJoin, so a skewed probe materializes as straggler tasks.
+ *   - insufficient size ratio: BuildLeft carries a fixed streamed-probe overhead on the right side.
+ *     When right/left is small (the two sides are close in size), that overhead outweighs the
+ *     hash-build win. Only when right >> left is BuildLeft profitable.
+ *
+ * Skew thresholds reuse Spark's `spark.sql.adaptive.skewJoin.*` so we don't fight AQE's own
+ * definition of skew. The size ratio and right-side floor are Gluten-owned knobs
+ * (`minRightToLeftRatio` and `minRightBytes`).
+ */
+case class LeftSemiBuildLeftGuardRule(session: SparkSession)
+  extends Rule[SparkPlan]
+  with Logging {
+
+  override def apply(plan: SparkPlan): SparkPlan = {
+    // Fast path: if the feature is disabled, avoid the transformUp entirely.
+    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.ForceShjBuildRightTag).getOrElse(false)) {
+      // Already tagged (e.g. by RewriteJoin propagating from a tagged SMJ). Nothing to do.
+      return
+    }
+
+    // Step 1 (cheap, O(N) fold): right-side totalBytes. Force BuildRight when stats are
+    // unavailable -- without concrete byte counts we cannot verify safety.
+    val rightTotalBytes = ShuffleSkewDetector.totalBytes(rightSide) match {
+      case Some(bytes) => bytes
+      case None =>
+        join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true)
+        return
+    }
+    if (rightTotalBytes < minRightBytes) {
+      logDebug(
+        s"LeftSemiBuildLeftGuardRule: right-side too small on LeftSemi $kind " +
+          s"(rightBytes=$rightTotalBytes < minRightBytes=$minRightBytes); " +
+          "forcing BuildRight.")
+      join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true)
+      return
+    }
+
+    // Step 2 (cheap, O(N) fold): ratio guard. Force BuildRight when left stats unavailable.
+    val leftTotalBytes = ShuffleSkewDetector.totalBytes(leftSide) match {
+      case Some(bytes) => bytes
+      case None =>
+        join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true)
+        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); forcing BuildRight.")
+        join.setTagValue(RewriteJoin.ForceShjBuildRightTag, true)
+        return
+      }
+    }
+
+    // Step 3 (expensive, O(N log N) sort for median): skew analysis. Only reached when
+    // both size and ratio passed, so on regressive shapes (which we saw dominate q14a-like
+    // plans) the sort is skipped.
+    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}); forcing BuildRight.")
+      join.setTagValue(RewriteJoin.ForceShjBuildRightTag, 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
+    )
+  }
+}