[SPARK-56660][SQL][FOLLOWUP] Preserve NULL semantics; gate behind SQLConf; rework tests

Addresses review feedback on PR #56244:

1. Correctness fix for NULL handling. The original decomposition rewrote
   EqualTo(struct, struct) into a plain conjunction of per-field EqualTo
   comparisons, which silently changed semantics for non-null structs that
   contained NULL fields:
     - Before this PR: struct(1, null) = struct(1, null) returned TRUE
       (Spark's whole-struct EqualTo evaluates ordering.equiv on the row,
       which treats per-field NULL == NULL as equal).
     - With original PR #56244: returned NULL.

   The fix wraps the conjunction with a null-check that mirrors the
   original outer null behavior:
     - EqualTo(L, R) over nullable structs: IF (L IS NULL OR R IS NULL)
       THEN NULL ELSE And(EqualNullSafe(L.fi, R.fi)).
     - EqualNullSafe(L, R): IF (L IS NULL AND R IS NULL) THEN TRUE
       ELSE IF (L IS NULL OR R IS NULL) THEN FALSE
       ELSE And(EqualNullSafe(L.fi, R.fi)).
   The wrappers fold out cleanly when either operand is non-nullable,
   leaving the simple conjunction in the common
   `CreateNamedStruct = column` pushdown case.

2. SQLConf gate. Add `spark.sql.optimizer.decomposeStructComparison.enabled`
   (default false) so users opt in once the behavior has soaked. Add
   `spark.sql.optimizer.decomposeStructComparison.maxFields` (default 1000)
   that bounds total decomposed predicates including recursively nested
   struct fields, replacing the unprincipled per-level field cap of 100.

3. Scaladoc explaining Filter scope. Document why join conditions and
   aggregate grouping keys are deliberately not rewritten.

4. Tests reworked as oracle tests. The original suite asserted post-rewrite
   NULL behavior directly, which codified the regression as expected. The
   rewritten suite uses two patterns:
     - Catalyst-level: build expressions and assert eval result of original
       expression equals eval result of rewritten expression on representative
       inputs (struct(1, null), whole-struct null, Not wrapper, etc.).
     - End-to-end: run each query with the rule enabled and with the conf
       disabled; assert row sets are identical.
   Added tests for: Not(struct = struct) with NULL fields, whole-struct null
   on one side, conf gating. Removed: 3 wrong-oracle NULL tests, structural-
   only "nullable fields decomposes" test, duplicate LessThan, duplicate
   3-level nested, single-field, duplicate join test in catalyst suite.

Author: yadavay-amzn

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

@@ -565,43 +565,149 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
 
 
 /**
- * Decomposes struct-level equality comparisons into conjunctions of field-level equalities.
- * This enables filter pushdown for individual struct fields.
- * For example, `struct_col = struct(1, 'a')` becomes
- * `struct_col.field1 = 1 AND struct_col.field2 = 'a'`.
+ * Decomposes struct-level equality comparisons appearing in Filter conditions into a
+ * conjunction of field-level equalities. This enables per-field filter pushdown to data
+ * sources (Parquet row group skipping, Iceberg/Delta column statistics, partition pruning).
+ *
+ * For a non-nullable struct comparison `struct_col = struct(1, 'a')`, the rewrite produces
+ * `struct_col.field1 = 1 AND struct_col.field2 = 'a'`. When either operand is nullable,
+ * the conjunction is wrapped in a null-check expression that mirrors the original
+ * struct-level comparison's NULL semantics (see comments on `decomposeEqualTo` and
+ * `decomposeEqualNullSafe`).
+ *
+ * Scope: the rule only rewrites comparisons appearing inside `Filter` conditions. It
+ * deliberately does NOT rewrite:
+ *   - Join conditions: an equi-join key on a struct is matched by the planner as a single
+ *     equality and routes to BroadcastHashJoin / SortMergeJoin. Decomposing it would
+ *     break key matching for those join strategies.
+ *   - Aggregate grouping expressions: structurally cannot be reached because the
+ *     transformation is scoped to `case Filter`.
+ *   - Project expressions: outside the pushdown path; rewriting them does not enable
+ *     pushdown and would expand the projection unnecessarily.
+ *
+ * The rewrite is gated on `spark.sql.optimizer.decomposeStructComparison.enabled`
+ * (default off) so users can opt in once the behavior has soaked in their workloads.
+ * The total number of decomposed predicates per top-level comparison is bounded by
+ * `spark.sql.optimizer.decomposeStructComparison.maxFields` to prevent runaway expansion
+ * on deeply nested or wide structs.
  */
 object DecomposeStructComparison extends Rule[LogicalPlan] {
-  def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
-    _.containsPattern(FILTER), ruleId) {
-    case f @ Filter(condition, _) =>
-      f.copy(condition = decomposeCondition(condition))
+  def apply(plan: LogicalPlan): LogicalPlan = {
+    if (!conf.decomposeStructComparisonEnabled) {
+      return plan
+    }
+    plan.transformWithPruning(_.containsPattern(FILTER), ruleId) {
+      case f @ Filter(condition, _) =>
+        f.copy(condition = decomposeCondition(condition))
+    }
   }
 
   private def decomposeCondition(expr: Expression): Expression = expr.transformWithPruning(
     _.containsPattern(BINARY_COMPARISON)) {
     case EqualTo(left, right) if canDecompose(left, right) =>
-      decompose(left, right, EqualTo)
+      decomposeEqualTo(left, right)
     case EqualNullSafe(left, right) if canDecompose(left, right) =>
-      decompose(left, right, EqualNullSafe)
+      decomposeEqualNullSafe(left, right)
   }
 
   private def canDecompose(left: Expression, right: Expression): Boolean = {
     (left.dataType, right.dataType) match {
       case (l: StructType, r: StructType) =>
-        l.length > 0 && l.length <= 100 && l.length == r.length &&
+        // 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.
+        l.length > 0 && l.length == r.length &&
+          totalLeafFields(l) <= conf.decomposeStructComparisonMaxFields &&
           left.deterministic && right.deterministic
       case _ => false
     }
   }
 
-  private def decompose(
-      left: Expression,
-      right: Expression,
-      cmp: (Expression, Expression) => Expression): Expression = {
+  /**
+   * Counts the total number of leaf (non-struct) fields in a struct type, recursing into
+   * nested structs. Used to bound rewrite expansion on deeply nested structs: a 3-level
+   * nested struct of 100 fields per level expands to 1M field-level predicates, which
+   * `maxFields` is intended to prevent.
+   */
+  private def totalLeafFields(t: StructType): Int = {
+    t.fields.iterator.map { f =>
+      f.dataType match {
+        case s: StructType => totalLeafFields(s)
+        case _ => 1
+      }
+    }.sum
+  }
+
+  /**
+   * Builds the field-by-field conjunction `(L.f0 <=> R.f0) AND ... AND (L.fn <=> R.fn)`.
+   * Per-field comparisons use `EqualNullSafe` so that NULL fields at the same position
+   * compare as equal -- this matches Spark's existing struct equality semantics, where
+   * `InterpretedOrdering.compare` treats two NULLs at the same field position as equal
+   * and continues comparing remaining fields.
+   */
+  private def fieldConjunction(left: Expression, right: Expression): Expression = {
     val fields = left.dataType.asInstanceOf[StructType].fields
     fields.indices.map { i =>
-      cmp(GetStructField(left, i), GetStructField(right, i))
-    }.reduceLeft(And)
+      EqualNullSafe(GetStructField(left, i), GetStructField(right, i)).asInstanceOf[Expression]
+    }.reduce(And)
+  }
+
+  /**
+   * Decomposes `EqualTo(L, R)` over struct types preserving original semantics:
+   *   - If either L or R is NULL (the whole struct), the result is NULL.
+   *   - Otherwise, two struct values are equal iff each pair of fields at the same
+   *     position is equal, with NULL at the same field position counting as equal
+   *     (per `InterpretedOrdering`).
+   *
+   * Lowering: `IF (L IS NULL OR R IS NULL) THEN NULL ELSE And(EqualNullSafe(L.fi, R.fi))`.
+   * The outer null-check is omitted when both operands are non-nullable.
+   */
+  private def decomposeEqualTo(left: Expression, right: Expression): Expression = {
+    val conjunction = fieldConjunction(left, right)
+    val nullableSides = (if (left.nullable) Seq(IsNull(left)) else Nil) ++
+      (if (right.nullable) Seq(IsNull(right)) else Nil)
+    if (nullableSides.isEmpty) {
+      conjunction
+    } else {
+      val anyNull = nullableSides.reduce(Or)
+      If(anyNull, Literal(null, BooleanType), conjunction)
+    }
+  }
+
+  /**
+   * Decomposes `EqualNullSafe(L, R)` (`<=>`) over struct types preserving original semantics:
+   *   - `null <=> null` -> TRUE
+   *   - exactly one side null -> FALSE
+   *   - both sides non-null -> equal iff each field pair is equal under `<=>`
+   *
+   * Lowering:
+   *   IF (L IS NULL AND R IS NULL) THEN TRUE
+   *   ELSE IF (L IS NULL OR R IS NULL) THEN FALSE
+   *   ELSE And(EqualNullSafe(L.fi, R.fi))
+   *
+   * Branches whose preconditions are statically false (because the corresponding operand is
+   * non-nullable) are omitted.
+   */
+  private def decomposeEqualNullSafe(left: Expression, right: Expression): Expression = {
+    val conjunction = fieldConjunction(left, right)
+    (left.nullable, right.nullable) match {
+      case (false, false) =>
+        conjunction
+      case (true, false) =>
+        // Right cannot be null. If left is null, result is FALSE; else conjunction.
+        If(IsNull(left), Literal.FalseLiteral, conjunction)
+      case (false, true) =>
+        // Left cannot be null. If right is null, result is FALSE; else conjunction.
+        If(IsNull(right), Literal.FalseLiteral, conjunction)
+      case (true, true) =>
+        // Both nullable. Need to distinguish both-null (TRUE) from exactly-one-null (FALSE).
+        If(And(IsNull(left), IsNull(right)),
+          Literal.TrueLiteral,
+          If(Or(IsNull(left), IsNull(right)),
+            Literal.FalseLiteral,
+            conjunction))
+    }
   }
 }
 

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

@@ -522,6 +522,30 @@ object SQLConf {
     .booleanConf
     .createWithDefault(true)
 
+  val DECOMPOSE_STRUCT_COMPARISON_ENABLED =
+    buildConf("spark.sql.optimizer.decomposeStructComparison.enabled")
+      .doc("When true, the optimizer rewrites struct equality (= and <=>) appearing in Filter " +
+        "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")
+      .withBindingPolicy(ConfigBindingPolicy.SESSION)
+      .booleanConf
+      .createWithDefault(false)
+
+  val DECOMPOSE_STRUCT_COMPARISON_MAX_FIELDS =
+    buildConf("spark.sql.optimizer.decomposeStructComparison.maxFields")
+      .internal()
+      .doc("The maximum total number of field-level predicates a single struct comparison may " +
+        "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")
+      .withBindingPolicy(ConfigBindingPolicy.SESSION)
+      .intConf
+      .checkValue(_ > 0, "The threshold must be positive.")
+      .createWithDefault(1000)
+
   val OPTIMIZER_EXCLUDED_RULES = buildConf("spark.sql.optimizer.excludedRules")
     .doc("Configures a list of rules to be disabled in the optimizer, in which the rules are " +
       "specified by their rule names and separated by comma. It is not guaranteed that all the " +
@@ -8581,6 +8605,10 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def avoidDoubleFilterEval: Boolean = getConf(AVOID_DOUBLE_FILTER_EVAL)
 
+  def decomposeStructComparisonEnabled: Boolean = getConf(DECOMPOSE_STRUCT_COMPARISON_ENABLED)
+
+  def decomposeStructComparisonMaxFields: Int = getConf(DECOMPOSE_STRUCT_COMPARISON_MAX_FIELDS)
+
   def readSideCharPadding: Boolean = getConf(SQLConf.READ_SIDE_CHAR_PADDING)
 
   def cliPrintHeader: Boolean = getConf(SQLConf.CLI_PRINT_HEADER)