[GitHub] [spark] wForget commented on a diff in pull request #41609: [SPARK-44065][SQL] Optimize BroadcastHashJoin skew in OptimizeSkewedJoin

["wForget (via GitHub)" <gi...@apache.org>]

wForget commented on code in PR #41609:
URL: https://github.com/apache/spark/pull/41609#discussion_r1271629202


##########
sql/core/src/test/scala/org/apache/spark/sql/execution/adaptive/AdaptiveQueryExecSuite.scala:
##########
@@ -2851,6 +2851,68 @@ class AdaptiveQueryExecSuite
     val unionDF = aggDf1.union(aggDf2)
     checkAnswer(unionDF.select("id").distinct, Seq(Row(null)))
   }
+
+  test("SPARK-44065: Optimize BroadcastHashJoin skew when localShuffleReader is disabled") {
+    withSQLConf(
+      SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "true",
+      SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+      SQLConf.ADAPTIVE_AUTO_BROADCASTJOIN_THRESHOLD.key -> "1000",
+      SQLConf.LOCAL_SHUFFLE_READER_ENABLED.key -> "false",
+      SQLConf.SHUFFLE_PARTITIONS.key -> "10",
+      SQLConf.SKEW_JOIN_SKEWED_PARTITION_THRESHOLD.key -> "600",
+      SQLConf.ADVISORY_PARTITION_SIZE_IN_BYTES.key -> "600") {
+      withTempView("skewData", "smallData") {
+        spark
+          .range(0, 1000, 1, 10)
+          .selectExpr("if(id >= 5, 5, id) as key1", "id as value1")
+          .createOrReplaceTempView("skewData1")
+        spark
+          .range(0, 5, 1, 10)
+          .selectExpr("id key2", "id as value2")
+          .createOrReplaceTempView("smallData")
+
+        Seq(true, false).foreach { localShuffleReader =>
+          withSQLConf(SQLConf.LOCAL_SHUFFLE_READER_ENABLED.key -> localShuffleReader.toString) {
+            val sqlText =
+              s"""
+                 |select * from skewData1 a join smallData b on a.key1 = b.key2
+                 |""".stripMargin
+            val (_, plan) = runAdaptiveAndVerifyResult(sqlText)
+            val bhjs = findTopLevelBroadcastHashJoin(plan)
+            assert(bhjs.nonEmpty)
+
+            if (localShuffleReader) {
+              val localShuffleReaders = collect(plan) {
+                case c: AQEShuffleReadExec if c.isLocalRead => c

Review Comment:
   > it seems not addressed, can you take a look ? @wForget
   
   Sorry, I'm still confused about this so belatedly did it. Do we still need to follow advisoryParallelism if we consider block size?



-- 
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: reviews-unsubscribe@spark.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org