yadavay-amzn commented on code in PR #56719:
URL: https://github.com/apache/spark/pull/56719#discussion_r3583737944
##########
sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala:
##########
@@ -58,6 +59,117 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
assert(planned.size === 1)
}
+ test("SPARK-57648: spread rows rejected by a safe left outer join residual")
{
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
Review Comment:
Non-blocking: the tests all run with AQE off, but it's on by default, and
AQE is where re-planning and coalescing (`CoalescedNullAwareHashPartitioning`)
happen - and where the expression tag is most likely to get dropped. Might be
worth one case with AQE on that checks the answer and the spreading.
##########
sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala:
##########
@@ -58,6 +59,117 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
assert(planned.size === 1)
}
+ test("SPARK-57648: spread rows rejected by a safe left outer join residual")
{
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "4",
+ SQLConf.SHUFFLE_SPREAD_NULL_JOIN_KEYS_ENABLED.key -> "true") {
+ val left = Seq(
+ (1, java.lang.Boolean.TRUE, "match"),
+ (1, java.lang.Boolean.FALSE, "false"),
+ (2, null.asInstanceOf[java.lang.Boolean], "null")).toDF("k",
"eligible", "lv").as("l")
+ val right = Seq((1, "right-1"), (2, "right-2")).toDF("k", "rv").as("r")
+ val joined = left.join(
+ right, $"l.k" === $"r.k" && $"l.eligible", "left_outer")
+
+ val sortMergeJoin = joined.queryExecution.sparkPlan.collectFirst {
+ case join: SortMergeJoinExec => join
+ }.getOrElse(fail("Expected a sort-merge join"))
+ assert(sortMergeJoin.leftKeys.size == 2)
+ assert(sortMergeJoin.rightKeys.size == 2)
+ assert(sortMergeJoin.leftKeys.last.isInstanceOf[If])
+ assert(sortMergeJoin.rightKeys.last == Literal.TrueLiteral)
+ assert(sortMergeJoin.condition.nonEmpty)
+ assert(sortMergeJoin.requiredChildDistribution.forall {
+ case ClusteredDistribution(_, true, _, _) => true
+ case _ => false
+ })
+
+ val compoundPredicate = !($"l.lv".isin("false")) ||
+ (functions.coalesce($"l.eligible", functions.lit(false)) &&
$"l.lv".isNotNull)
+ val compoundJoin = left.join(
+ right, $"l.k" === $"r.k" && compoundPredicate, "left_outer")
+ val compoundSortMergeJoin =
compoundJoin.queryExecution.sparkPlan.collectFirst {
+ case join: SortMergeJoinExec => join
+ }.getOrElse(fail("Expected a sort-merge join"))
+ assert(compoundSortMergeJoin.leftKeys.size == 2)
+ assert(compoundSortMergeJoin.leftKeys.last.isInstanceOf[If])
+
+ checkAnswer(joined, Seq(
+ Row(1, true, "match", 1, "right-1"),
+ Row(1, false, "false", null, null),
+ Row(2, null, "null", null, null)))
+
+ val shufflePartitionings = joined.queryExecution.executedPlan.collect {
+ case exchange: ShuffleExchangeExec => exchange.outputPartitioning
+ }
Review Comment:
Non-blocking: this checks the plan uses a 2-expression
`NullAwareHashPartitioning`, but not that the rows actually spread - which is
the point of the change. A version that added the guard but didn't spread would
still pass. Maybe throw a bunch of residual-false rows on one hot key and check
they land on more than one partition.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala:
##########
@@ -286,7 +324,12 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
def createSortMergeJoin() = {
if (canMerge(joinType) && RowOrdering.isOrderable(leftKeys)) {
Some(Seq(joins.SortMergeJoinExec(
- leftKeys, rightKeys, joinType, nonEquiCond, planLater(left),
planLater(right))))
+ shuffledLeftKeys,
Review Comment:
Nit: this checks `RowOrdering.isOrderable(leftKeys)` (the original keys) but
builds the join with `shuffledLeftKeys`. Doesn't matter today since the guard
is Boolean and always orderable, but the two would be out of sync if the guard
type ever changed - might be cleaner to check `shuffledLeftKeys`.
##########
sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala:
##########
@@ -58,6 +59,117 @@ class JoinSuite extends SharedSparkSession with
AdaptiveSparkPlanHelper
assert(planned.size === 1)
}
+ test("SPARK-57648: spread rows rejected by a safe left outer join residual")
{
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "4",
+ SQLConf.SHUFFLE_SPREAD_NULL_JOIN_KEYS_ENABLED.key -> "true") {
+ val left = Seq(
+ (1, java.lang.Boolean.TRUE, "match"),
+ (1, java.lang.Boolean.FALSE, "false"),
+ (2, null.asInstanceOf[java.lang.Boolean], "null")).toDF("k",
"eligible", "lv").as("l")
+ val right = Seq((1, "right-1"), (2, "right-2")).toDF("k", "rv").as("r")
+ val joined = left.join(
+ right, $"l.k" === $"r.k" && $"l.eligible", "left_outer")
+
+ val sortMergeJoin = joined.queryExecution.sparkPlan.collectFirst {
+ case join: SortMergeJoinExec => join
+ }.getOrElse(fail("Expected a sort-merge join"))
+ assert(sortMergeJoin.leftKeys.size == 2)
+ assert(sortMergeJoin.rightKeys.size == 2)
+ assert(sortMergeJoin.leftKeys.last.isInstanceOf[If])
+ assert(sortMergeJoin.rightKeys.last == Literal.TrueLiteral)
+ assert(sortMergeJoin.condition.nonEmpty)
+ assert(sortMergeJoin.requiredChildDistribution.forall {
+ case ClusteredDistribution(_, true, _, _) => true
+ case _ => false
+ })
+
+ val compoundPredicate = !($"l.lv".isin("false")) ||
+ (functions.coalesce($"l.eligible", functions.lit(false)) &&
$"l.lv".isNotNull)
+ val compoundJoin = left.join(
+ right, $"l.k" === $"r.k" && compoundPredicate, "left_outer")
+ val compoundSortMergeJoin =
compoundJoin.queryExecution.sparkPlan.collectFirst {
+ case join: SortMergeJoinExec => join
+ }.getOrElse(fail("Expected a sort-merge join"))
+ assert(compoundSortMergeJoin.leftKeys.size == 2)
+ assert(compoundSortMergeJoin.leftKeys.last.isInstanceOf[If])
+
+ checkAnswer(joined, Seq(
+ Row(1, true, "match", 1, "right-1"),
+ Row(1, false, "false", null, null),
+ Row(2, null, "null", null, null)))
+
+ val shufflePartitionings = joined.queryExecution.executedPlan.collect {
+ case exchange: ShuffleExchangeExec => exchange.outputPartitioning
+ }
+ assert(shufflePartitionings.size == 2)
+ assert(shufflePartitionings.forall {
+ case NullAwareHashPartitioning(expressions, _) => expressions.size == 2
+ case _ => false
+ })
+ }
+ }
+
+ test("SPARK-57648: spread unmatchable rows for shuffled hash join") {
+ withSQLConf(
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> "false",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1",
+ SQLConf.SHUFFLE_PARTITIONS.key -> "4",
+ SQLConf.SHUFFLE_SPREAD_NULL_JOIN_KEYS_ENABLED.key -> "true") {
+ val left = Seq((1, true), (1, false)).toDF("k", "eligible").as("l")
+ val right = Seq((1, "right-1")).toDF("k",
"rv").as("r").hint("SHUFFLE_HASH")
+ val joined = left.join(
+ right, $"l.k" === $"r.k" && $"l.eligible", "left_outer")
+
Review Comment:
Non-blocking: every test has one right row per key, so we don't cover a
match producing multiple rows. Something like left `(1,true)` against right
`(1,"a"),(1,"b")` expecting two rows would confirm the guard doesn't break the
matched fan-out.
--
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]
