[SPARK-53618][SQL] Turn a FULL JOIN with a false condition into a UNION ALL

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 &&

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)))
     )

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

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,