Copilot commented on code in PR #12513: URL: https://github.com/apache/gluten/pull/12513#discussion_r3584740482
########## 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 + ) + } Review Comment: `LeftSemiBuildLeftGuardRule` takes a `SparkSession` but never uses it (the rule reads SQLConf via `SQLConf.get`). This leaves an unused param and can also pick up the wrong conf if multiple sessions are active. Prefer reading conf from the provided session to both use the parameter and bind to the correct session. ########## gluten-ut/spark35/src/test/scala/org/apache/spark/sql/execution/adaptive/velox/VeloxAdaptiveQueryExecSuite.scala: ########## @@ -1502,6 +1502,167 @@ class VeloxAdaptiveQueryExecSuite extends AdaptiveQueryExecSuite with GlutenSQLT } } + testGluten("LeftSemi BuildLeft guard: enabled with large ratio chooses BuildLeft") { + withTempView("big", "small") { + // big: 1000 rows across 10 partitions, small: 10 rows across 5 partitions. + // This creates a large right/left ratio so BuildLeft should activate. + spark.sparkContext + .parallelize((1 to 1000).map(i => TestData(i % 50, i.toString)), 10) + .toDF("c1", "c2") + .createOrReplaceTempView("big") + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "true", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key -> "1", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key -> "2.0", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true" + ) { + val (_, adaptive) = runAdaptiveAndVerifyResult( + "SELECT small.c1 FROM small LEFT SEMI JOIN big ON small.c1 = big.c1") + val shj = findTopLevelShuffledHashJoinTransform(adaptive) + assert(shj.size === 1) + assert(shj.head.joinBuildSide == BuildLeft) + } + } + } + + testGluten("LeftSemi BuildLeft guard: right-side too small forces BuildRight") { + withTempView("big", "small") { + spark.sparkContext + .parallelize((1 to 1000).map(i => TestData(i % 50, i.toString)), 10) + .toDF("c1", "c2") + .createOrReplaceTempView("big") + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "true", + // Set minRightBytes very high so guard forces BuildRight + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key -> "100GB", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key -> "1.0", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true" + ) { + val (_, adaptive) = runAdaptiveAndVerifyResult( + "SELECT small.c1 FROM small LEFT SEMI JOIN big ON small.c1 = big.c1") + val shj = findTopLevelShuffledHashJoinTransform(adaptive) + assert(shj.size === 1) + assert(shj.head.joinBuildSide == BuildRight) + } + } + } + + testGluten("LeftSemi BuildLeft guard: insufficient ratio forces BuildRight") { + withTempView("small", "medium") { + // small: 10 rows, medium: 50 rows. Right is larger so unguarded getOptimalBuildSide + // would pick BuildLeft (left is smaller). But ratio = 50/10 = 5x < minRatio(10.0), + // so the ratio guard forces BuildRight. + spark.sparkContext + .parallelize((1 to 10).map(i => TestData(i, i.toString)), 5) + .toDF("c1", "c2") + .createOrReplaceTempView("small") + spark.sparkContext + .parallelize((1 to 50).map(i => TestData(i % 10, i.toString)), 10) + .toDF("c1", "c2") + .createOrReplaceTempView("medium") + + withSQLConf( + SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true", + SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1", + SQLConf.SHUFFLE_PARTITIONS.key -> "5", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_ENABLED.key -> "true", + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_BYTES.key -> "1", + // Ratio ~5x won't meet this threshold + GlutenConfig.COLUMNAR_SHJ_LEFTSEMI_BUILDLEFT_MIN_RIGHT_TO_LEFT_RATIO.key -> "10.0", + GlutenConfig.COLUMNAR_FORCE_SHUFFLED_HASH_JOIN_ENABLED.key -> "true" Review Comment: This test expects the ratio guard to trigger based on row-count intuition ("~5x"), but the guard logic uses *shuffle bytes* (`ShuffleSkewDetector.totalBytes`) rather than row counts. Byte totals can vary with serialization overhead and partition emptiness, which can make a 10.0 threshold flaky. Consider using an intentionally huge `minRightToLeftRatio` so the ratio guard deterministically forces BuildRight regardless of the exact byte ratio. -- 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]
