[SPARK-56660][SQL][FOLLOWUP] Narrow struct decomposition to foldable+cheap operands and guard NULL semantics on the pushable path

Author: yadavay-amzn

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala

@@ -607,12 +607,22 @@ object DecomposeStructComparison extends Rule[LogicalPlan] with PredicateHelper
         // `decomposeCondition` path unchanged.
         val rewritten = splitConjunctivePredicates(condition).map {
           case EqualTo(l, r) if canDecompose(l, r) && filterEquivalentToConjunction(l, r) =>
-            fieldConjunction(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) =>
-            // <=> is filter-equivalent to the bare conjunction for every nullability
-            // combination (both-null -> TRUE under <=>, and the conjunction's
-            // `null <=> null` per field also yields TRUE).
-            fieldConjunction(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).
+            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 other =>
             decomposeCondition(other)
         }.reduce(And)
@@ -643,11 +653,15 @@ object DecomposeStructComparison extends Rule[LogicalPlan] with PredicateHelper
   private def canDecompose(left: Expression, right: Expression): Boolean = {
     (left.dataType, right.dataType) match {
       case (l: StructType, r: StructType) =>
-        // The analyzer has already validated the operand types match structurally
-        // (BinaryComparison uses equalsStructurally, which checks types positionally
-        // and ignores field names). We bound expansion via maxFields and require
-        // determinism so the rewrite does not duplicate side effects.
+        // Gate: one operand must be foldable (literal/constant) and the other must be cheap
+        // (attribute/extractvalue -- see CollapseProject.isCheap). This avoids:
+        //  - col=col shapes that yield no pushdown benefit (field conjunction is just expansion)
+        //  - non-cheap operands (e.g. UDFs) that would be duplicated N times
+        val hasFoldableAndCheap =
+          (left.foldable && CollapseProject.isCheap(right)) ||
+          (right.foldable && CollapseProject.isCheap(left))
         l.length > 0 && l.length == r.length &&
+          hasFoldableAndCheap &&
           totalLeafFields(l) <= conf.decomposeStructComparisonMaxFields &&
           left.deterministic && right.deterministic
       case _ => false

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -528,7 +528,7 @@ object SQLConf {
         "conditions into a conjunction of field-level equalities, enabling per-field filter " +
         "pushdown to data sources. The rewrite preserves NULL semantics by wrapping the " +
         "conjunction in a null-check that mirrors the original struct-level comparison.")
-      .version("4.2.0")
+      .version("4.3.0")
       .withBindingPolicy(ConfigBindingPolicy.SESSION)
       .booleanConf
       .createWithDefault(false)
@@ -540,7 +540,7 @@ object SQLConf {
         "be decomposed into, including fields from recursively nested structs. Comparisons " +
         "exceeding this limit are left as struct-level predicates. Used to bound rewrite cost " +
         "and memory for deeply nested or wide structs.")
-      .version("4.2.0")
+      .version("4.3.0")
       .withBindingPolicy(ConfigBindingPolicy.SESSION)
       .intConf
       .checkValue(_ > 0, "The threshold must be positive.")

sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/StructPredicateDecomposeSuite.scala

@@ -133,7 +133,10 @@ class StructPredicateDecomposeSuite extends PlanTest {
       s"Nested struct should be fully decomposed. Plan:\n${optimized.treeString}")
   }
 
-  test("SPARK-56660: tuple comparison (CreateStruct = CreateStruct) is decomposed") {
+  // After scope narrowing, CreateStruct over attributes is NOT cheap (it's not an Attribute
+  // or ExtractValue), so this no longer decomposes even though the right side is foldable.
+  test("SPARK-56660: tuple comparison (CreateStruct = CreateStruct) is NOT decomposed " +
+      "(scope narrowing)") {
     val relation = LocalRelation($"col1".string, $"col2".string)
     val leftStruct = CreateStruct(Seq($"col1", $"col2"))
     val rightStruct = CreateStruct(Seq(Literal("a"), Literal("b")))
@@ -147,11 +150,15 @@ class StructPredicateDecomposeSuite extends PlanTest {
     val structComparisons = filter.condition.collect {
       case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
     }
-    assert(structComparisons.isEmpty,
-      s"Tuple comparison should be decomposed. Plan:\n${optimized.treeString}")
+    assert(structComparisons.nonEmpty,
+      s"Tuple comparison should NOT be decomposed (left is not cheap). " +
+        s"Plan:\n${optimized.treeString}")
   }
 
-  test("SPARK-56660: struct compared to another struct column is decomposed") {
+  // After scope narrowing (SPARK-56660 FOLLOWUP), col=col no longer decomposes because
+  // neither operand is foldable. This is intentional: col=col yields no pushdown benefit
+  // and just expands the plan.
+  test("SPARK-56660: struct col=col is NOT decomposed (scope narrowing)") {
     val relation = LocalRelation(
       $"s1".struct($"a".string, $"b".int),
       $"s2".struct($"a".string, $"b".int)
@@ -166,8 +173,9 @@ class StructPredicateDecomposeSuite extends PlanTest {
     val structComparisons = filter.condition.collect {
       case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
     }
-    assert(structComparisons.isEmpty,
-      s"Struct-to-struct column comparison should be decomposed. Plan:\n${optimized.treeString}")
+    assert(structComparisons.nonEmpty,
+      s"Struct-to-struct column comparison should NOT be decomposed (no foldable operand). " +
+        s"Plan:\n${optimized.treeString}")
   }
 
   // -----------------------------------------------------------------------------------
@@ -389,6 +397,64 @@ class StructPredicateDecomposeSuite extends PlanTest {
     assertSameSemantics(Not(EqualTo(left, right)))
   }
 
+  test("SPARK-56660 FIXED: pushable EqualTo with all-null-fields literal correctly excludes " +
+      "whole-null struct row (IsNotNull guard)") {
+    // Previously the pushable path emitted `s.a <=> null` without a null guard.
+    // On a whole-null-struct row, GetStructField(null, 0) returns null, so
+    // `null <=> null` -> TRUE (row wrongly kept). The IsNotNull(s) guard now
+    // ensures the row is correctly excluded.
+    val structType = StructType(Seq(StructField("a", IntegerType, nullable = true)))
+    val nullableRelation = LocalRelation(AttributeReference("s", structType, nullable = true)())
+    val sAttr = nullableRelation.output.head
+    // named_struct('a', CAST(NULL AS INT)) -- non-nullable wrapper, all-null fields
+    val literal = CreateNamedStruct(Seq(Literal("a"), Literal(null, IntegerType)))
+    val original = EqualTo(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)
+
+    // A Filter keeps a row iff predicate is TRUE. Original yields NULL (drop).
+    // Rewritten must also NOT keep the row.
+    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.\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),
+    // so the buggy fieldConjunction is never emitted. The plan is left intact.
+    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 = EqualNullSafe(s1, s2)
+
+    val plan = rel.where(original).analyze
+    val rewritten = Optimize.execute(plan).collect { case f: Filter => f.condition }.head
+
+    // The rule should NOT have fired; condition should still contain struct-level <=>
+    val hasStructEns = rewritten.collect {
+      case EqualNullSafe(l, _) if l.dataType.isInstanceOf[StructType] => true
+    }
+    assert(hasStructEns.nonEmpty,
+      s"col <=> col should NOT be decomposed (no foldable operand). Plan condition: $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

sql/core/src/test/scala/org/apache/spark/sql/StructPredicateDecomposeE2ESuite.scala

@@ -66,7 +66,11 @@ class StructPredicateDecomposeE2ESuite extends QueryTest with SharedSparkSession
 
     assertSameUnderRule(df.filter($"s" === struct(lit("a"), lit("x"))))
 
-    // Plan inspection: with the rule on, no struct-level EqualTo should remain.
+    // After scope narrowing, the rule requires one foldable + one cheap operand.
+    // When CollapseProject inlines the struct projection, the filter sees
+    // CreateNamedStruct(f1, f2) = CreateNamedStruct(lit_a, lit_x) -- the left side
+    // is not cheap (not an attribute). So decomposition does NOT fire here.
+    // The rule fires for base attributes from scans (e.g. Parquet), tested separately.
     withSQLConf(SQLConf.DECOMPOSE_STRUCT_COMPARISON_ENABLED.key -> "true") {
       val optimizedPlan = df.filter($"s" === struct(lit("a"), lit("x")))
         .queryExecution.optimizedPlan
@@ -76,8 +80,8 @@ class StructPredicateDecomposeE2ESuite extends QueryTest with SharedSparkSession
         case e: org.apache.spark.sql.catalyst.expressions.EqualTo
           if e.left.dataType.isInstanceOf[org.apache.spark.sql.types.StructType] => true
       }.nonEmpty)
-      assert(!hasStructComparison,
-        "Struct equality should be decomposed (no struct-level EqualTo in optimized plan)")
+      assert(hasStructComparison,
+        "Inline struct projection should NOT be decomposed (left operand is not cheap)")
     }
   }
 
@@ -217,6 +221,55 @@ class StructPredicateDecomposeE2ESuite extends QueryTest with SharedSparkSession
     }
   }
 
+  test("SPARK-56660: all-null-fields literal vs whole-null struct (pushable EqualTo path)") {
+    // Bug: when the literal has ALL fields null, the pushable fieldConjunction emits
+    // `s.a <=> null` which evaluates to TRUE on a whole-null-struct row (GetStructField
+    // on null struct returns null). The original `s = named_struct('a', null)` returns
+    // NULL on a whole-null row (dropped by WHERE). The rewrite incorrectly keeps it.
+    withTempPath { path =>
+      val dir = path.getCanonicalPath
+      // Create data with a genuine whole-null struct row using when/otherwise
+      Seq((1, 1), (2, -1), (3, 2))
+        .toDF("id", "a")
+        .withColumn("s", when($"a" =!= -1, struct($"a")))  // id=2 gets NULL struct
+        .select("id", "s")
+        .write.parquet(dir)
+
+      val df = spark.read.parquet(dir)
+      assert(df.schema("s").nullable, "precondition: struct column should be nullable")
+
+      // The correct result: `s = named_struct('a', CAST(NULL AS INT))` on a whole-null
+      // struct row returns NULL (row dropped). No row has s=struct(null) with s non-null,
+      // so the correct result is EMPTY.
+      assertSameUnderRule(
+        df.filter($"s" === struct(lit(null).cast("int"))))
+    }
+  }
+
+  test("SPARK-56660: all-null-fields EqualNullSafe with two nullable struct columns") {
+    // Bug: the <=> branch in the pushable path (expressions.scala:611) is unguarded.
+    // Two nullable struct columns both null: original `null <=> null` -> TRUE (keep),
+    // but if one is null and the other is struct(null), original is FALSE (drop).
+    // The fieldConjunction can't distinguish these cases.
+    withTempPath { path =>
+      val dir = path.getCanonicalPath
+      // Row 1: s1=struct(null), s2=struct(null) -> s1 <=> s2 is TRUE (keep)
+      // Row 2: s1=NULL (whole), s2=struct(null) -> s1 <=> s2 is FALSE (drop)
+      // Row 3: s1=struct(1), s2=struct(1) -> s1 <=> s2 is TRUE (keep)
+      Seq((1, 1), (2, -1), (3, 2))
+        .toDF("id", "v")
+        .withColumn("s1", when($"v" =!= -1, struct(lit(null).cast("int").as("a")))
+                          .otherwise(lit(null).cast("struct<a:int>")))
+        .withColumn("s2", struct(lit(null).cast("int").as("a")))  // always struct(null)
+        .select("id", "s1", "s2")
+        .write.parquet(dir)
+
+      val df = spark.read.parquet(dir)
+      // Correct: rows 1 and 3 match. Row 2 must NOT match.
+      assertSameUnderRule(df.filter($"s1" <=> $"s2"))
+    }
+  }
+
   test("SPARK-56660: rule is gated by spark.sql.optimizer.decomposeStructComparison.enabled") {
     val df = Seq(
       (1, "a", "x"),