diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala index aae092bcb2632..3ca99ecfb702b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelation.scala @@ -33,6 +33,8 @@ import org.apache.spark.sql.catalyst.trees.TreePattern.{LOCAL_RELATION, REPARTIT * - Union with all empty children. * 2. Binary-node Logical Plans * - Join with one or two empty children (including Intersect/Except). + * - Full outer join with a false condition + * Rewrite to a UNION ALL of both sides padded with nulls. * - Left semi Join * Right side is non-empty and condition is empty. Eliminate join to its left side. * - Left anti join @@ -123,6 +125,13 @@ abstract class PropagateEmptyRelationBase extends Rule[LogicalPlan] with CastSup Project(p.left.output ++ nullValueProjectList(p.right), p.left) case RightOuter if isFalseCondition && canExecuteWithoutJoin(p.right) => Project(nullValueProjectList(p.left) ++ p.right.output, p.right) + // No row of either side can find a match, so the join degenerates into both sides + // padded with nulls and concatenated. + case FullOuter if isFalseCondition && canExecuteWithoutJoin(p.left) && + canExecuteWithoutJoin(p.right) => + Union( + Project(p.left.output ++ nullValueProjectList(p.right), p.left), + Project(nullValueProjectList(p.left) ++ p.right.output, p.right)) case _ => p } } else if (joinType == LeftSemi && conditionOpt.isEmpty && diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelationSuite.scala index 723d0db4f0838..b9359ede7dc32 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelationSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/PropagateEmptyRelationSuite.scala @@ -24,7 +24,7 @@ import org.apache.spark.sql.catalyst.dsl.plans._ import org.apache.spark.sql.catalyst.expressions.{EqualTo, Literal, UnspecifiedFrame} import org.apache.spark.sql.catalyst.expressions.Literal.FalseLiteral import org.apache.spark.sql.catalyst.plans._ -import org.apache.spark.sql.catalyst.plans.logical.{Expand, Filter, LocalRelation, LogicalPlan, Project} +import org.apache.spark.sql.catalyst.plans.logical.{Expand, Filter, LocalRelation, LogicalPlan, Project, Union} import org.apache.spark.sql.catalyst.rules.RuleExecutor import org.apache.spark.sql.catalyst.types.DataTypeUtils import org.apache.spark.sql.internal.SQLConf @@ -172,7 +172,11 @@ class PropagateEmptyRelationSuite extends PlanTest { (RightOuter, Some(Project(Seq(Literal(null).cast(IntegerType).as("a"), $"b"), testRelation2) .analyze)), - (FullOuter, None), + (FullOuter, + Some(Union( + Project(Seq($"a", Literal(null).cast(IntegerType).as("b")), testRelation1), + Project(Seq(Literal(null).cast(IntegerType).as("a"), $"b"), testRelation2)) + .analyze)), (LeftAnti, Some(testRelation1)), (LeftSemi, Some(LocalRelation($"a".int))) ) diff --git a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala index 3c0b00793ca5e..25f01cc48eb95 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala @@ -28,8 +28,8 @@ import org.apache.spark.sql.catalyst.TableIdentifier import org.apache.spark.sql.catalyst.analysis.UnresolvedRelation import org.apache.spark.sql.catalyst.expressions.{Ascending, GenericRow, SortOrder} import org.apache.spark.sql.catalyst.optimizer.{BuildLeft, BuildRight, JoinSelectionHelper} -import org.apache.spark.sql.catalyst.plans.logical.{Filter, HintInfo, Join, JoinHint, NO_BROADCAST_AND_REPLICATION} -import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec, ProjectExec, SortExec, SparkPlan, WholeStageCodegenExec} +import org.apache.spark.sql.catalyst.plans.logical.{Filter, HintInfo, Join, JoinHint, NO_BROADCAST_AND_REPLICATION, Union} +import org.apache.spark.sql.execution.{BinaryExecNode, FilterExec, ProjectExec, SortExec, SparkPlan, UnionExec, WholeStageCodegenExec} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.execution.exchange.{ShuffleExchangeExec, ShuffleExchangeLike} import org.apache.spark.sql.execution.joins._ @@ -1838,6 +1838,29 @@ class JoinSuite extends SharedSparkSession with AdaptiveSparkPlanHelper cached.unpersist() } } + + test("SPARK-53618: full outer join with a false condition is rewritten to a union") { + val df = sql( + """ + |SELECT t1.id AS a, t2.id AS b + |FROM range(0, 2) t1 FULL OUTER JOIN range(10, 12) t2 ON 1 = 0 + |""".stripMargin) + + val optimized = df.queryExecution.optimizedPlan + assert(!optimized.exists(_.isInstanceOf[Join])) + assert(optimized.exists(_.isInstanceOf[Union])) + + // `range` produces a non-nullable column, but a full outer join makes both sides + // nullable and the union has to keep it that way + assert(df.schema.fields.forall(_.nullable)) + + checkAnswer(df, Row(0, null) :: Row(1, null) :: Row(null, 10) :: Row(null, 11) :: Nil) + + // the point of the rewrite: the nested loop join and its broadcast are gone + val executed = df.queryExecution.executedPlan + assert(find(executed)(_.isInstanceOf[UnionExec]).isDefined) + assert(find(executed)(_.isInstanceOf[BroadcastNestedLoopJoinExec]).isEmpty) + } } class ThreadLeakInSortMergeJoinSuite diff --git a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala index ea15f13c225cf..09da2d0f36041 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/MergeIntoTableSuiteBase.scala @@ -27,7 +27,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{ReplaceData, WriteDelta} import org.apache.spark.sql.connector.catalog.{Aborted, Column, ColumnDefaultValue, Committed, InMemoryBaseTable, InMemoryTable, TableInfo} import org.apache.spark.sql.connector.expressions.{GeneralScalarExpression, LiteralValue} import org.apache.spark.sql.connector.write.MergeSummary -import org.apache.spark.sql.execution.SparkPlan +import org.apache.spark.sql.execution.{SparkPlan, UnionExec} import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation, DataSourceV2ScanRelation, InsertOnlyMergeExec, MergeRowsExec} import org.apache.spark.sql.execution.joins.{BroadcastHashJoinExec, BroadcastNestedLoopJoinExec, CartesianProductExec} @@ -3041,6 +3041,41 @@ abstract class MergeIntoTableSuiteBase extends RowLevelOperationSuiteBase } } + test("SPARK-53618: merge with a never matching condition avoids a nested loop join") { + withTempView("source") { + createAndInitTable("pk INT NOT NULL, salary INT, dep STRING", + """{ "pk": 1, "salary": 100, "dep": "hr" } + |{ "pk": 2, "salary": 200, "dep": "software" } + |{ "pk": 3, "salary": 300, "dep": "hr" } + |""".stripMargin) + + Seq((4, 400, "hr"), (5, 500, "software")).toDF("pk", "salary", "dep") + .createOrReplaceTempView("source") + + // a replaceWhere-style merge: the ON condition never matches, so the merge is + // rewritten to a full outer join with a false condition, which in turn becomes a + // union of both sides padded with nulls + val executedPlan = executeAndKeepPlan { + sql( + s"""MERGE INTO $tableNameAsString t + |USING source s + |ON 1 = 0 + |WHEN NOT MATCHED THEN + | INSERT * + |WHEN NOT MATCHED BY SOURCE AND t.dep = 'hr' THEN + | DELETE + |""".stripMargin) + } + + assert(collect(executedPlan) { case j: BroadcastNestedLoopJoinExec => j }.isEmpty) + assert(collect(executedPlan) { case u: UnionExec => u }.size == 1) + + checkAnswer( + sql(s"SELECT * FROM $tableNameAsString"), + Row(2, 200, "software") :: Row(4, 400, "hr") :: Row(5, 500, "software") :: Nil) + } + } + private def assertMetric( mergeExec: MergeRowsExec, metricName: String,