[SPARK-56660][SQL][FOLLOWUP] Add EqualNullSafe eval test and clarifying comments for the null-guard path

yadavay-amzn · yadavay-amzn · commit d1c02ec9bb60 · 2026-06-30T17:22:22.000Z
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala
@@ -606,18 +606,15 @@ object DecomposeStructComparison extends Rule[LogicalPlan] with PredicateHelper
         // there NULL vs FALSE is observable -- so those go through the null-preserving
         // `decomposeCondition` path unchanged.
         val rewritten = splitConjunctivePredicates(condition).map {
-          case EqualTo(l, r) if canDecompose(l, r) && filterEquivalentToConjunction(l, r) =>
+          case Equality(l, r)
+              if canDecompose(l, r) && filterEquivalentToConjunction(l, r) =>
             // Guard: AND IsNotNull(<nullable operand>) to preserve whole-null vs
             // all-null-fields semantics. Without it, a whole-null struct row would
             // pass the fieldConjunction when the literal has all-null fields.
-            val nullGuards = (if (l.nullable) Seq(IsNotNull(l)) else Nil) ++
-              (if (r.nullable) Seq(IsNotNull(r)) else Nil)
-            val conjunction = fieldConjunction(l, r)
-            if (nullGuards.isEmpty) conjunction
-            else (nullGuards :+ conjunction).reduce(And)
-          case EqualNullSafe(l, r) if canDecompose(l, r) =>
-            // <=> on the pushable path (one foldable, one cheap). Guard nullable operands
-            // with IsNotNull so that a whole-null struct is not conflated with struct(nulls).
+            // Note: canDecompose requires one foldable operand (always non-nullable),
+            // so both-nullable is unreachable and filterEquivalentToConjunction is
+            // always true here -- the guard is a safety net that also makes the <=>
+            // path fall through to decomposeEqualNullSafe when needed.
             val nullGuards = (if (l.nullable) Seq(IsNotNull(l)) else Nil) ++
               (if (r.nullable) Seq(IsNotNull(r)) else Nil)
             val conjunction = fieldConjunction(l, r)
@@ -640,6 +637,8 @@ object DecomposeStructComparison extends Rule[LogicalPlan] with PredicateHelper
    * shape has a non-nullable literal side, so it takes the pushable path.
    */
   private def filterEquivalentToConjunction(left: Expression, right: Expression): Boolean =
+    // Safety net: foldable structs (CreateNamedStruct) are non-nullable, so both-nullable is
+    // currently unreachable given the foldable+cheap scope-narrowing invariant in canDecompose.
     !(left.nullable && right.nullable)
 
   private def decomposeCondition(expr: Expression): Expression = expr.transformWithPruning(
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/StructPredicateDecomposeSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/StructPredicateDecomposeSuite.scala
@@ -432,6 +432,39 @@ class StructPredicateDecomposeSuite extends PlanTest {
         s"  The IsNotNull guard should prevent the row from being kept.")
   }
 
+  test("SPARK-56660 FIXED: pushable EqualNullSafe with all-null-fields literal correctly " +
+      "excludes whole-null struct row (IsNotNull guard)") {
+    // Analogous to the EqualTo eval test above, but for the <=> path.
+    // s <=> named_struct('a', CAST(NULL AS INT)) on a whole-null struct row:
+    // original <=> yields FALSE (null is not null-struct), rewritten must agree.
+    val structType = StructType(Seq(StructField("a", IntegerType, nullable = true)))
+    val nullableRelation = LocalRelation(AttributeReference("s", structType, nullable = true)())
+    val sAttr = nullableRelation.output.head
+    val literal = CreateNamedStruct(Seq(Literal("a"), Literal(null, IntegerType)))
+    val original = EqualNullSafe(sAttr, literal)
+
+    val plan = nullableRelation.where(original).analyze
+    val rewritten = Optimize.execute(plan).collect { case f: Filter => f.condition }.head
+
+    // Evaluate on a whole-null-struct row
+    val row = InternalRow(null)
+    val boundOriginal = BindReferences.bindReference(original, Seq(sAttr))
+    val boundRewritten = BindReferences.bindReference(rewritten, Seq(sAttr))
+
+    val origResult = boundOriginal.eval(row)
+    val rewrittenResult = boundRewritten.eval(row)
+
+    // Original <=> returns FALSE (null != struct(null)). Rewritten must also yield FALSE.
+    def keptByFilter(v: Any): Boolean = v == true
+    assert(!keptByFilter(origResult),
+      s"Original <=> should not keep whole-null row, got: $origResult")
+    assert(keptByFilter(origResult) === keptByFilter(rewrittenResult),
+      s"Filter outcome mismatch on whole-null row with all-null-fields literal (<=> path).\n" +
+        s"  original: $original -> $origResult\n" +
+        s"  rewritten: $rewritten -> $rewrittenResult\n" +
+        s"  The IsNotNull guard should prevent the row from being kept.")
+  }
+
   test("SPARK-56660 FIXED: pushable EqualNullSafe with two nullable struct columns no longer " +
       "decomposes (scope narrowing prevents the bug)") {
     // After scope narrowing, col <=> col doesn't pass canDecompose (neither is foldable),
@@ -455,6 +488,230 @@ class StructPredicateDecomposeSuite extends PlanTest {
       s"col <=> col should NOT be decomposed (no foldable operand). Plan condition: $rewritten")
   }
 
+  // -----------------------------------------------------------------------------------
+  // Null-preserving (non-pushable) path tests: these exercise `decomposeCondition`,
+  // which fires for comparisons NOT in top-level filter position (e.g. under Not/Or).
+  // The null-preserving form wraps the conjunction in If(...) to preserve NULL vs FALSE
+  // distinctions that are observable under negation.
+  // -----------------------------------------------------------------------------------
+
+  /**
+   * Helper: wraps `original` in a Filter, runs the optimizer, and returns the rewritten
+   * condition. Uses a relation with the given attributes so references resolve.
+   */
+  private def optimizeWithRelation(
+      original: Expression,
+      relation: LocalRelation): Expression = {
+    val plan = relation.where(original).analyze
+    Optimize.execute(plan).collect { case f: Filter => f.condition }.head
+  }
+
+  test("SPARK-56660 oracle: Not(EqualTo(nullableAttr, literal)) reaches decomposeEqualTo " +
+      "nullable If branch") {
+    // This exercises decomposeEqualTo's `If(anyNull, NULL, conjunction)` branch
+    // (expressions.scala ~line 726) via the decomposeCondition path.
+    val structType = StructType(Seq(
+      StructField("a", IntegerType, nullable = false),
+      StructField("b", IntegerType, nullable = false)))
+    val nullableRelation = LocalRelation(
+      AttributeReference("s", structType, nullable = true)())
+    val sAttr = nullableRelation.output.head
+    val literal = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+    val original = Not(EqualTo(sAttr, literal))
+
+    // Verify the rule fires and produces the null-preserving If form
+    val rewritten = optimizeWithRelation(original, nullableRelation)
+
+    // The rewritten form must contain an If (the null-preserving wrapper)
+    val hasIfNull = rewritten.collect {
+      case If(IsNull(_), Literal(null, _), _) => true
+    }
+    assert(hasIfNull.nonEmpty,
+      s"decomposeEqualTo nullable branch should produce If(IsNull, NULL, conjunction). " +
+        s"Got: $rewritten")
+
+    // Semantic check: evaluate on three representative rows
+    val nonNullRow = InternalRow(InternalRow(1, 2))  // s = struct(1, 2): = is TRUE, Not = FALSE
+    val nullRow = InternalRow(null)                    // s = NULL: = is NULL, Not(NULL) = NULL
+    val mismatchRow = InternalRow(InternalRow(1, 99)) // s = struct(1,99): = is FALSE, Not = TRUE
+
+    val boundOriginal = BindReferences.bindReference(original, Seq(sAttr))
+    val boundRewritten = BindReferences.bindReference(rewritten, Seq(sAttr))
+
+    // Non-null matching row: Not(TRUE) = FALSE
+    assert(boundOriginal.eval(nonNullRow) === boundRewritten.eval(nonNullRow),
+      s"Mismatch on non-null matching row")
+    // Whole-null row: Not(NULL) = NULL (not FALSE!)
+    assert(boundOriginal.eval(nullRow) === boundRewritten.eval(nullRow),
+      s"Mismatch on whole-null row: original=${boundOriginal.eval(nullRow)}, " +
+        s"rewritten=${boundRewritten.eval(nullRow)}")
+    assert(boundOriginal.eval(nullRow) == null,
+      "Not(NULL struct = literal) should be NULL, not FALSE")
+    // Mismatching row: Not(FALSE) = TRUE
+    assert(boundOriginal.eval(mismatchRow) === boundRewritten.eval(mismatchRow),
+      s"Mismatch on non-matching row")
+  }
+
+  test("SPARK-56660 oracle: Not(EqualNullSafe(nullableAttr, literal)) reaches " +
+      "decomposeEqualNullSafe (true, false) branch") {
+    // This exercises decomposeEqualNullSafe's case (true, false) branch
+    // (expressions.scala ~line 756): If(IsNull(left), FalseLiteral, conjunction)
+    val structType = StructType(Seq(
+      StructField("a", IntegerType, nullable = false),
+      StructField("b", IntegerType, nullable = false)))
+    val nullableRelation = LocalRelation(
+      AttributeReference("s", structType, nullable = true)())
+    val sAttr = nullableRelation.output.head
+    val literal = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+    val original = Not(EqualNullSafe(sAttr, literal))
+
+    val rewritten = optimizeWithRelation(original, nullableRelation)
+
+    // The rewritten form must contain If(IsNull(...), FALSE, conjunction)
+    val hasIfFalse = rewritten.collect {
+      case If(IsNull(_), Literal(false, _), _) => true
+    }
+    assert(hasIfFalse.nonEmpty,
+      s"decomposeEqualNullSafe (true, false) branch should produce " +
+        s"If(IsNull, FALSE, conjunction). Got: $rewritten")
+
+    // Semantic check on three row kinds
+    val nonNullRow = InternalRow(InternalRow(1, 2))   // s <=> struct(1,2) = TRUE, Not = FALSE
+    val nullRow = InternalRow(null)                     // s <=> struct(1,2) = FALSE, Not = TRUE
+    val mismatchRow = InternalRow(InternalRow(1, 99))  // s <=> struct(1,99) = FALSE, Not = TRUE
+
+    val boundOriginal = BindReferences.bindReference(original, Seq(sAttr))
+    val boundRewritten = BindReferences.bindReference(rewritten, Seq(sAttr))
+
+    assert(boundOriginal.eval(nonNullRow) === boundRewritten.eval(nonNullRow))
+    assert(boundOriginal.eval(nullRow) === boundRewritten.eval(nullRow))
+    assert(boundOriginal.eval(nullRow) == true,
+      "Not(NULL <=> literal) should be TRUE (since NULL <=> non-null = FALSE)")
+    assert(boundOriginal.eval(mismatchRow) === boundRewritten.eval(mismatchRow))
+  }
+
+  test("SPARK-56660 oracle: Not(EqualNullSafe(literal, nullableAttr)) reaches " +
+      "decomposeEqualNullSafe (false, true) branch") {
+    // This exercises decomposeEqualNullSafe's case (false, true) branch
+    // (expressions.scala ~line 759): If(IsNull(right), FalseLiteral, conjunction)
+    val structType = StructType(Seq(
+      StructField("a", IntegerType, nullable = false),
+      StructField("b", IntegerType, nullable = false)))
+    val nullableRelation = LocalRelation(
+      AttributeReference("s", structType, nullable = true)())
+    val sAttr = nullableRelation.output.head
+    val literal = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+    // Note: literal on LEFT, attribute on RIGHT
+    val original = Not(EqualNullSafe(literal, sAttr))
+
+    val rewritten = optimizeWithRelation(original, nullableRelation)
+
+    // The rewritten form must contain If(IsNull(...), FALSE, conjunction)
+    val hasIfFalse = rewritten.collect {
+      case If(IsNull(_), Literal(false, _), _) => true
+    }
+    assert(hasIfFalse.nonEmpty,
+      s"decomposeEqualNullSafe (false, true) branch should produce " +
+        s"If(IsNull, FALSE, conjunction). Got: $rewritten")
+
+    val nonNullRow = InternalRow(InternalRow(1, 2))
+    val nullRow = InternalRow(null)
+    val mismatchRow = InternalRow(InternalRow(1, 99))
+
+    val boundOriginal = BindReferences.bindReference(original, Seq(sAttr))
+    val boundRewritten = BindReferences.bindReference(rewritten, Seq(sAttr))
+
+    assert(boundOriginal.eval(nonNullRow) === boundRewritten.eval(nonNullRow))
+    assert(boundOriginal.eval(nullRow) === boundRewritten.eval(nullRow))
+    assert(boundOriginal.eval(mismatchRow) === boundRewritten.eval(mismatchRow))
+  }
+
+  test("SPARK-56660 oracle: Or(EqualTo(nullableAttr, lit), other) reaches null-preserving " +
+      "decomposeEqualTo path") {
+    // Under Or, NULL vs FALSE is observable: Or(NULL, FALSE) = NULL (row dropped),
+    // Or(FALSE, FALSE) = FALSE (row dropped too, but conceptually different).
+    // The null-preserving If wrapper in decomposeEqualTo ensures correctness here.
+    val structType = StructType(Seq(StructField("a", IntegerType, nullable = false)))
+    val nullableRelation = LocalRelation(
+      AttributeReference("s", structType, nullable = true)(),
+      AttributeReference("flag", IntegerType, nullable = false)())
+    val sAttr = nullableRelation.output(0)
+    val flagAttr = nullableRelation.output(1)
+    val literal = litStruct(("a", 42, IntegerType))
+    val original = Or(EqualTo(sAttr, literal), EqualTo(flagAttr, Literal(1)))
+
+    val rewritten = optimizeWithRelation(original, nullableRelation)
+
+    // When s is NULL and flag != 1: Or(NULL, FALSE) = NULL (row dropped by filter)
+    // The rewrite must NOT turn this into Or(FALSE, FALSE) = FALSE unless it correctly
+    // preserves the If wrapper.
+    val nullRow = InternalRow(null, 0)
+    val boundOriginal = BindReferences.bindReference(original, nullableRelation.output)
+    val boundRewritten = BindReferences.bindReference(rewritten, nullableRelation.output)
+
+    // For this specific case, NULL and FALSE have the same filter outcome (both drop),
+    // but the value must match for correctness under further composition.
+    assert(boundOriginal.eval(nullRow) === boundRewritten.eval(nullRow),
+      s"Or with null struct: original=${boundOriginal.eval(nullRow)}, " +
+        s"rewritten=${boundRewritten.eval(nullRow)}")
+
+    // When s is NULL and flag == 1: Or(NULL, TRUE) = TRUE (row kept)
+    val nullKeepRow = InternalRow(null, 1)
+    assert(boundOriginal.eval(nullKeepRow) === boundRewritten.eval(nullKeepRow))
+    assert(boundOriginal.eval(nullKeepRow) == true)
+  }
+
+  test("SPARK-56660 oracle: decomposeEqualNullSafe (false, false) branch with non-nullable " +
+      "struct attribute") {
+    // When both operands are non-nullable, decomposeEqualNullSafe returns bare conjunction.
+    val structType = StructType(Seq(
+      StructField("a", IntegerType, nullable = false),
+      StructField("b", IntegerType, nullable = false)))
+    val nonNullableRelation = LocalRelation(
+      AttributeReference("s", structType, nullable = false)())
+    val sAttr = nonNullableRelation.output.head
+    val literal = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+    val original = Not(EqualNullSafe(sAttr, literal))
+
+    val rewritten = optimizeWithRelation(original, nonNullableRelation)
+
+    // No If wrapper needed when both are non-nullable
+    val hasIf = rewritten.collect { case _: If => true }
+    assert(hasIf.isEmpty,
+      s"Non-nullable <=> should decompose to bare conjunction (no If). Got: $rewritten")
+
+    val matchRow = InternalRow(InternalRow(1, 2))
+    val noMatchRow = InternalRow(InternalRow(3, 4))
+    val boundOriginal = BindReferences.bindReference(original, Seq(sAttr))
+    val boundRewritten = BindReferences.bindReference(rewritten, Seq(sAttr))
+
+    assert(boundOriginal.eval(matchRow) === boundRewritten.eval(matchRow))
+    assert(boundOriginal.eval(noMatchRow) === boundRewritten.eval(noMatchRow))
+  }
+
+  test("SPARK-56660: decomposeEqualNullSafe (true, true) is unreachable with current " +
+      "canDecompose invariant") {
+    // Document that (true, true) in decomposeEqualNullSafe is unreachable because
+    // canDecompose requires one foldable operand, and foldable CreateNamedStruct is
+    // always non-nullable. Two nullable attributes fail canDecompose (neither is foldable).
+    val structType = StructType(Seq(StructField("a", IntegerType, nullable = true)))
+    val rel = LocalRelation(
+      AttributeReference("s1", structType, nullable = true)(),
+      AttributeReference("s2", structType, nullable = true)())
+    val s1 = rel.output(0)
+    val s2 = rel.output(1)
+    val original = Not(EqualNullSafe(s1, s2))
+
+    val rewritten = optimizeWithRelation(original, rel)
+
+    // The rule should NOT fire (neither is foldable) -- condition stays as Not(s1 <=> s2)
+    val hasStructEns = rewritten.collect {
+      case EqualNullSafe(l, _) if l.dataType.isInstanceOf[StructType] => true
+    }
+    assert(hasStructEns.nonEmpty,
+      s"Both-nullable col <=> col should NOT be decomposed. Got: $rewritten")
+  }
+
   // Note on schema mismatch: the analyzer rejects struct comparisons with different
   // field types at type-check time (see DATATYPE_MISMATCH.BINARY_OP_DIFF_TYPES) before
   // the optimizer ever runs. The `sameType` check inside `canDecompose` is therefore a
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/StructPredicateDecomposeE2ESuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/StructPredicateDecomposeE2ESuite.scala
