ulysses-you commented on code in PR #57181:
URL: https://github.com/apache/spark/pull/57181#discussion_r3568251644
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -1337,6 +1337,30 @@ object SQLConf {
.bytesConf(ByteUnit.BYTE)
.createWithDefault(0L)
+ val ADAPTIVE_CONVERT_SORT_MERGE_JOIN_TO_SHUFFLED_HASH_JOIN_ENABLED =
+
buildConf("spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled")
+ .doc("When true, Spark converts a sort merge join to a shuffled hash
join during adaptive " +
+ "execution when the build side's materialized per-partition sizes are
all within " +
+ s"${ADAPTIVE_MAX_SHUFFLE_HASH_JOIN_LOCAL_MAP_THRESHOLD.key}, even when
non-shuffle " +
+ "operators sit between the join and its input shuffle.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.SESSION)
+ .booleanConf
+ .createWithDefault(false)
+
+ val ADAPTIVE_COST_EVALUATOR_COUNT_LOCAL_SORT_ENABLED =
Review Comment:
Made `spark.sql.adaptive.costEvaluator.countLocalSort.enabled` a
`fallbackConf` to `...convertSortMergeJoinToShuffledHashJoin.enabled`, so by
default it is enabled exactly together with the conversion. I kept it as a
separate overridable flag rather than defaulting it to `true` globally, because
flipping it on unconditionally is not a no-op - it broke the existing "Change
broadcast join to merge join" test, where a tied BHJ-vs-SMJ re-optimization now
prefers the sort-free BHJ. The fallback gives the "enabled together" behavior
you suggested without that regression.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/ReplaceSortMergeJoinToShuffledHashJoin.scala:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.spark.sql.execution.adaptive
+
+import scala.annotation.tailrec
+
+import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight,
BuildSide, JoinSelectionHelper}
+import org.apache.spark.sql.catalyst.plans.LeftExistence
+import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.execution.{FilterExec, ProjectExec, SortExec,
SparkPlan}
+import org.apache.spark.sql.execution.aggregate.BaseAggregateExec
+import org.apache.spark.sql.execution.exchange.{ENSURE_REQUIREMENTS,
EnsureRequirements}
+import org.apache.spark.sql.execution.joins.{BaseJoinExec,
ShuffledHashJoinExec, SortMergeJoinExec}
+import org.apache.spark.sql.execution.window.{WindowExecBase,
WindowGroupLimitExec}
+
+/**
+ * Converts a [[SortMergeJoinExec]] into a [[ShuffledHashJoinExec]] during
adaptive execution when
+ * a build side's materialized shuffle statistics show it is small enough for
a local hash map.
+ * Unlike [[DynamicJoinSelection]], this runs on the physical plan, so it can
reach the input
+ * shuffle through operators (aggregate, project, filter, window, etc...)
sitting above it.
+ *
+ * The swap is shuffle-free since both joins are `ShuffledJoin`s with the same
distribution and
+ * partitioning; only the child sorts become unnecessary. As a shuffled hash
join loses the sort
+ * merge join's output ordering, [[EnsureRequirements]] is re-run to restore
any ordering an
+ * ancestor still needs, and AQE's [[CostEvaluator]] decides whether to adopt
the converted plan.
+ */
+case class ReplaceSortMergeJoinToShuffledHashJoin(ensureRequirements:
EnsureRequirements)
Review Comment:
Renamed to `ConvertSortMergeJoinToShuffledHashJoin` to match the config's
`convert...To...` naming.
##########
docs/sql-performance-tuning.md:
##########
@@ -375,6 +375,22 @@ AQE converts sort-merge join to shuffled hash join when
all post shuffle partiti
</td>
<td>3.2.0</td>
</tr>
+ <tr>
+
<td><code>spark.sql.adaptive.convertSortMergeJoinToShuffledHashJoin.enabled</code></td>
Review Comment:
Updated - the entry now notes that `preferShuffledHashJoin` also requires
`spark.sql.adaptive.advisoryPartitionSizeInBytes` to not be larger than
`maxShuffledHashJoinLocalMapThreshold`, mirroring the 3.2.0 entry. Also added
the two new configs to the table.
--
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]
