(spark) branch branch-4.x updated: [SPARK-58365][SQL] Build the NOT-IN-in-disjunction join condition from the deduplicated subquery output

[ya...@apache.org]

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

LuciferYang pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new 9255e0e2d992 [SPARK-58365][SQL] Build the NOT-IN-in-disjunction join condition from the deduplicated subquery output
9255e0e2d992 is described below

commit 9255e0e2d99250b711aed1bbbd3ad9c80bea04c1
Author: YangJie <ya...@baidu.com>
AuthorDate: Thu Jul 30 14:31:05 2026 +0800

    [SPARK-58365][SQL] Build the NOT-IN-in-disjunction join condition from the deduplicated subquery output
    
    ### What changes were proposed in this pull request?
    
    In `RewritePredicateSubquery.rewriteExistentialExprWithAttrs`, the `Not(InSubquery(...))` branch (the one handling a NOT IN nested inside a disjunction, e.g. `v > 0 OR x NOT IN (...)`) calls `dedupSubqueryOnSelfJoin` to alias the subquery's attributes when they conflict with the outer plan, but then builds the IN equality conditions from the pre-dedup `sub.output` instead of the deduplicated `newSub.output`. The join's right child uses `newSub`, so the condition can reference attribut [...]
    
    ### Why are the changes needed?
    
    When `dedupSubqueryOnSelfJoin` fires, it rebinds the conflicting subquery attributes to fresh exprIds. Building the condition from `sub.output` then uses the stale ids, which only exist on the outer side, so the null-aware anti-join condition collapses to trivially-true self-equalities like `id#2 = id#2` and no longer references the join's right child. That is exactly the SPARK-26078 defect the `dedupSubqueryOnSelfJoin` call is there to prevent, so today that call is dead weight on th [...]
    
    ### Does this PR introduce _any_ user-facing change?
    
    No. When dedup does not fire, `newSub` is the same object as `sub`, so the change is a no-op on every plan reachable from user SQL today.
    
    ### How was this patch tested?
    
    Added a `RewriteSubquerySuite` case that builds the colliding-attribute plan directly (bypassing the analyzer's `DeduplicateRelations`, which would otherwise renew the ids) and asserts the rewritten join condition references the deduplicated right-side output. It fails on the unfixed tree (the condition is `(a#0 = a#0) OR isnull((a#0 = a#0))`, referencing nothing on the right) and passes with the fix.
    
    ### Was this patch authored or co-authored using generative AI tooling?
    
    Generated-by: Claude Code (Opus 4.8)
    
    Closes #57558 from LuciferYang/SPARK-58365-notin-dedup.
    
    Authored-by: YangJie <ya...@baidu.com>
    Signed-off-by: yangjie01 <ya...@baidu.com>
    (cherry picked from commit 6e8829dbccd7d8afa61a504389eb48084f0d999b)
    Signed-off-by: yangjie01 <ya...@baidu.com>
---
 .../spark/sql/catalyst/optimizer/subquery.scala    |  2 +-
 .../catalyst/optimizer/RewriteSubquerySuite.scala  | 31 ++++++++++++++++++++--
 2 files changed, 30 insertions(+), 3 deletions(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala
index bf3e63571dbb..32554e575a3a 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/subquery.scala
@@ -418,7 +418,7 @@ object RewritePredicateSubquery extends Rule[LogicalPlan] with PredicateHelper {
           val exists = AttributeReference("exists", BooleanType, nullable = false)()
           // Deduplicate conflicting attributes if any.
           val newSub = dedupSubqueryOnSelfJoin(newPlan, sub, Some(values))
-          val inConditions = values.zip(sub.output).map(EqualTo.tupled)
+          val inConditions = values.zip(newSub.output).map(EqualTo.tupled)
           // To handle a null-aware predicate not-in-subquery in nested conditions
           // (e.g., `v > 0 OR t1.id NOT IN (SELECT id FROM t2)`), we transform
           // `inCondition` (t1.id=t2.id) into `(inCondition) OR ISNULL(inCondition)`.
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteSubquerySuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteSubquerySuite.scala
index dca1d503e3fd..d98c04396082 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteSubquerySuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteSubquerySuite.scala
@@ -20,9 +20,9 @@ package org.apache.spark.sql.catalyst.optimizer
 import org.apache.spark.sql.catalyst.QueryPlanningTracker
 import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
-import org.apache.spark.sql.catalyst.expressions.{Cast, Exists, IsNull, ListQuery, Literal, Not}
+import org.apache.spark.sql.catalyst.expressions.{Cast, EqualTo, Exists, InSubquery, IsNull, ListQuery, Literal, Not, Or}
 import org.apache.spark.sql.catalyst.plans.{ExistenceJoin, LeftSemi, PlanTest}
-import org.apache.spark.sql.catalyst.plans.logical.{LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.plans.logical.{Filter, Join, LocalRelation, LogicalPlan}
 import org.apache.spark.sql.catalyst.rules.RuleExecutor
 import org.apache.spark.sql.types.LongType
 
@@ -72,6 +72,33 @@ class RewriteSubquerySuite extends PlanTest {
     comparePlans(optimized, correctAnswer)
   }
 
+  test("SPARK-58365: NOT-IN nested in OR deduplicates conflicting attrs on the join right side") {
+    // Nesting the NOT-IN under an OR routes it through rewriteExistentialExprWithAttrs,
+    // whose join condition must be built from the deduplicated subquery output. Build the
+    // colliding-attribute plan directly: the analyzer's DeduplicateRelations would renew
+    // the shared exprId before the optimizer runs, hiding the rule's behavior.
+    val a = $"a".int
+    val b = $"b".int
+    val relation = LocalRelation(a, b)
+    // The subquery reuses `a`, so its output conflicts with the outer plan by exprId.
+    val subquery = relation.select(a)
+    val query = Filter(
+      Or(EqualTo(b, Literal(1)), Not(InSubquery(Seq(a), ListQuery(subquery, numCols = 1)))),
+      relation)
+
+    val optimized = Optimize.execute(query)
+
+    val join = optimized.collectFirst { case j: Join => j }.get
+    // dedupSubqueryOnSelfJoin aliases the conflicting attr to a fresh exprId, so the join
+    // is duplicate-resolved regardless. The real check: the condition must reference that
+    // deduplicated right-side attr. The pre-fix code zips against the stale pre-dedup
+    // output, so the condition only touches the outer side and the right side dangles.
+    assert(join.duplicateResolved)
+    assert(join.condition.get.references.intersect(join.right.outputSet).nonEmpty,
+      s"join condition ${join.condition.get} must reference the right child output " +
+        s"${join.right.outputSet}")
+  }
+
   test("SPARK-34598: Filters without subquery must not be modified by RewritePredicateSubquery") {
     val relation = LocalRelation($"a".int, $"b".int, $"c".int, $"d".int)
     val query = relation.where(($"a" === 1 || $"b" === 2)


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