(spark) branch branch-4.3 updated: [SPARK-58384][SQL] Preserve null semantics in OptimizeJoinCondition

[we...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.3 by this push:
     new f67d154fb179 [SPARK-58384][SQL] Preserve null semantics in OptimizeJoinCondition
f67d154fb179 is described below

commit f67d154fb1792996cf88cdf4b85b4c78542ca17c
Author: xuanyulu <35...@qq.com>
AuthorDate: Fri Aug 7 22:21:11 2026 +0800

    [SPARK-58384][SQL] Preserve null semantics in OptimizeJoinCondition
    
    ### What changes were proposed in this pull request?
    
    Restrict `OptimizeJoinCondition` to rewriting null-safe equality patterns only at the root of a join condition or beneath `AND`/`OR`.
    
    The rule no longer performs this rewrite beneath `NOT`, where the difference between `NULL` and `FALSE` is observable.
    
    ### Why are the changes needed?
    
    When exactly one operand is `NULL`, the original pattern returns `NULL`, while `<=>` returns `FALSE`.
    
    This difference does not matter at the root of a join condition, but under `NOT` it can cause Spark to incorrectly keep extra rows.
    
    ### Does this PR introduce *any* user-facing change?
    
    Yes. Join conditions containing this pattern beneath `NOT` now return the correct rows.
    
    For the SPARK-58384 reproduction, the result changes from four rows to only:
    
    ```text
    [0,10,1,22]
    ```
    
    ### How was this patch tested?
    
    Added tests covering:
    
    - No rewrite beneath `NOT`.
    - Continued rewriting beneath `AND` and `OR`.
    - The end-to-end join result with null values.
    
    Ran:
    
    ```text
    build/sbt 'catalyst/testOnly org.apache.spark.sql.catalyst.optimizer.OptimizeJoinConditionSuite'
    build/sbt 'sql/testOnly org.apache.spark.sql.DataFrameJoinSuite -- -z "SPARK-58384"'
    build/sbt 'catalyst/scalastyle' 'catalyst/Test/scalastyle' 'sql/Test/scalastyle'
    ```
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    co-authored by Codex (GPT-5)
    
    Closes #57791 from MrHappyEnding/SPARK-58384.
    
    Authored-by: xuanyulu <35...@qq.com>
    Signed-off-by: Wenchen Fan <we...@databricks.com>
    (cherry picked from commit cea566b22b3de6879f3b38e6def2cef0d417a731)
    Signed-off-by: Wenchen Fan <we...@databricks.com>
---
 .../catalyst/optimizer/OptimizeJoinCondition.scala | 28 +++++++++++++---
 .../optimizer/OptimizeJoinConditionSuite.scala     | 37 ++++++++++++++++++++++
 .../org/apache/spark/sql/DataFrameJoinSuite.scala  |  9 ++++++
 3 files changed, 70 insertions(+), 4 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinCondition.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinCondition.scala
index 7c41ebea050b..86248cf30828 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinCondition.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinCondition.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.optimizer
 
-import org.apache.spark.sql.catalyst.expressions.{And, EqualNullSafe, EqualTo, IsNull, Or, PredicateHelper}
+import org.apache.spark.sql.catalyst.expressions.{And, EqualNullSafe, EqualTo, Expression, IsNull, Or, PredicateHelper}
 import org.apache.spark.sql.catalyst.plans.logical.{Join, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreePattern.{JOIN, OR}
@@ -30,7 +30,16 @@ object OptimizeJoinCondition extends Rule[LogicalPlan] with PredicateHelper {
   override def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
     _.containsPattern(JOIN), ruleId) {
     case j @ Join(_, _, _, condition, _) if condition.nonEmpty =>
-      val newCondition = condition.map(_.transformWithPruning(_.containsPattern(OR), ruleId) {
+      val newCondition = condition.map(optimizeCondition)
+      j.copy(condition = newCondition)
+  }
+
+  // Rewriting the pattern to EqualNullSafe maps NULL to FALSE, so only recurse through And/Or.
+  private def optimizeCondition(condition: Expression): Expression = {
+    if (!condition.containsPattern(OR)) {
+      condition
+    } else {
+      condition match {
         case Or(EqualTo(l, r), And(IsNull(c1), IsNull(c2)))
           if (l.semanticEquals(c1) && r.semanticEquals(c2))
             || (l.semanticEquals(c2) && r.semanticEquals(c1)) =>
@@ -39,7 +48,18 @@ object OptimizeJoinCondition extends Rule[LogicalPlan] with PredicateHelper {
           if (l.semanticEquals(c1) && r.semanticEquals(c2))
             || (l.semanticEquals(c2) && r.semanticEquals(c1)) =>
           EqualNullSafe(l, r)
-      })
-      j.copy(condition = newCondition)
+        case and @ And(left, right) =>
+          val newLeft = optimizeCondition(left)
+          val newRight = optimizeCondition(right)
+          if (newLeft.fastEquals(left) && newRight.fastEquals(right)) and
+          else And(newLeft, newRight)
+        case or @ Or(left, right) =>
+          val newLeft = optimizeCondition(left)
+          val newRight = optimizeCondition(right)
+          if (newLeft.fastEquals(left) && newRight.fastEquals(right)) or
+          else Or(newLeft, newRight)
+        case other => other
+      }
+    }
   }
 }
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinConditionSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinConditionSuite.scala
index e7f090ec4d0d..c686eb6e680d 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinConditionSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/OptimizeJoinConditionSuite.scala
@@ -46,4 +46,41 @@ class OptimizeJoinConditionSuite extends PlanTest {
       comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
     })
   }
+
+  test("SPARK-58384: do not replace null-safe equality pattern under NOT") {
+    val x = testRelation.subquery("x")
+    val y = testRelation1.subquery("y")
+    val originalQuery =
+      x.join(y, Inner, Option(!($"a" === $"c" || ($"a".isNull && $"c".isNull))))
+
+    comparePlans(Optimize.execute(originalQuery.analyze), originalQuery.analyze)
+  }
+
+  test("SPARK-58384: replace null-safe equality pattern under AND and OR") {
+    val x = testRelation.subquery("x")
+    val y = testRelation1.subquery("y")
+    val pattern = $"a" === $"c" || ($"a".isNull && $"c".isNull)
+    val optimizedPattern = $"a" <=> $"c"
+    val otherCondition = $"b" === $"d"
+    val conditions = Seq(
+      (pattern && otherCondition) -> (optimizedPattern && otherCondition),
+      (pattern || otherCondition) -> (optimizedPattern || otherCondition))
+
+    conditions.foreach { case (condition, optimizedCondition) =>
+      val originalQuery = x.join(y, Inner, Option(condition))
+      val correctAnswer = x.join(y, Inner, Option(optimizedCondition))
+      comparePlans(Optimize.execute(originalQuery.analyze), correctAnswer.analyze)
+    }
+  }
+
+  test("SPARK-58384: preserve unrelated AND and OR nodes") {
+    val x = testRelation.subquery("x")
+    val y = testRelation1.subquery("y")
+    val condition = ($"a" === $"c") && (($"b" === $"d") || ($"a" === 1))
+    val originalQuery = x.join(y, Inner, Option(condition)).analyze.asInstanceOf[Join]
+
+    val optimized = OptimizeJoinCondition(originalQuery).asInstanceOf[Join]
+
+    assert(optimized.condition.get eq originalQuery.condition.get)
+  }
 }
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
index 129d6bb68676..1d93006342ab 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameJoinSuite.scala
@@ -630,4 +630,13 @@ class DataFrameJoinSuite extends SharedSparkSession
       checkAnswer(df1, df2)
     })
   }
+
+  test("SPARK-58384: preserve null semantics under NOT in join condition") {
+    val left = Seq((Some(0), 10), (None, 11)).toDF("a", "b").as("left")
+    val right = Seq((None, 20), (Some(0), 21), (Some(1), 22)).toDF("x", "y").as("right")
+    val condition = !(($"left.a" === $"right.x") ||
+      ($"left.a".isNull && $"right.x".isNull))
+
+    checkAnswer(left.join(right, condition), Row(0, 10, 1, 22))
+  }
 }


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