[SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown

Author: yadavay-amzn

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

@@ -141,6 +141,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
         BooleanSimplification,
         SimplifyConditionals,
         PushFoldableIntoBranches,
+        DecomposeStructComparison,
         SimplifyBinaryComparison,
         ReplaceNullWithFalseInPredicate,
         PruneFilters,

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

@@ -564,6 +564,46 @@ 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'`.
+ */
+object DecomposeStructComparison extends Rule[LogicalPlan] {
+  def apply(plan: LogicalPlan): LogicalPlan = 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)
+    case EqualNullSafe(left, right) if canDecompose(left, right) =>
+      decompose(left, right, EqualNullSafe)
+  }
+
+  private def canDecompose(left: Expression, right: Expression): Boolean = {
+    (left.dataType, right.dataType) match {
+      case (l: StructType, r: StructType) =>
+        l.length > 0 && l.length == r.length && left.deterministic && right.deterministic
+      case _ => false
+    }
+  }
+
+  private def decompose(
+      left: Expression,
+      right: Expression,
+      cmp: (Expression, Expression) => Expression): Expression = {
+    val fields = left.dataType.asInstanceOf[StructType].fields
+    fields.indices.map { i =>
+      cmp(GetStructField(left, i), GetStructField(right, i))
+    }.reduceLeft(And)
+  }
+}
+
 /**
  * Simplifies binary comparisons with semantically-equal expressions:
  * 1) Replace '<=>' with 'true' literal.

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala

@@ -182,6 +182,7 @@ object RuleIdCollection {
       "org.apache.spark.sql.catalyst.optimizer.RewriteExceptAll" ::
       "org.apache.spark.sql.catalyst.optimizer.RewriteIntersectAll" ::
       "org.apache.spark.sql.catalyst.optimizer.SimplifyBinaryComparison" ::
+      "org.apache.spark.sql.catalyst.optimizer.DecomposeStructComparison" ::
       "org.apache.spark.sql.catalyst.optimizer.SimplifyCaseConversionExpressions" ::
       "org.apache.spark.sql.catalyst.optimizer.SimplifyCasts" ::
       "org.apache.spark.sql.catalyst.optimizer.SimplifyConditionals" ::