Skip to content

Commit 76dca41

Browse files
committed
[SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown
1 parent af55029 commit 76dca41

5 files changed

Lines changed: 555 additions & 0 deletions

File tree

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -141,6 +141,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
141141
BooleanSimplification,
142142
SimplifyConditionals,
143143
PushFoldableIntoBranches,
144+
DecomposeStructComparison,
144145
SimplifyBinaryComparison,
145146
ReplaceNullWithFalseInPredicate,
146147
PruneFilters,

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

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -564,6 +564,47 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
564564
}
565565

566566

567+
/**
568+
* Decomposes struct-level equality comparisons into conjunctions of field-level equalities.
569+
* This enables filter pushdown for individual struct fields.
570+
* For example, `struct_col = struct(1, 'a')` becomes
571+
* `struct_col.field1 = 1 AND struct_col.field2 = 'a'`.
572+
*/
573+
object DecomposeStructComparison extends Rule[LogicalPlan] {
574+
def apply(plan: LogicalPlan): LogicalPlan = plan.transformWithPruning(
575+
_.containsPattern(FILTER), ruleId) {
576+
case f @ Filter(condition, _) =>
577+
f.copy(condition = decomposeCondition(condition))
578+
}
579+
580+
private def decomposeCondition(expr: Expression): Expression = expr.transformWithPruning(
581+
_.containsPattern(BINARY_COMPARISON)) {
582+
case EqualTo(left, right) if canDecompose(left, right) =>
583+
decompose(left, right, EqualNullSafe)
584+
case EqualNullSafe(left, right) if canDecompose(left, right) =>
585+
decompose(left, right, EqualNullSafe)
586+
}
587+
588+
private def canDecompose(left: Expression, right: Expression): Boolean = {
589+
(left.dataType, right.dataType) match {
590+
case (l: StructType, r: StructType) =>
591+
l.length > 0 && l.length <= 100 && l.length == r.length &&
592+
left.deterministic && right.deterministic
593+
case _ => false
594+
}
595+
}
596+
597+
private def decompose(
598+
left: Expression,
599+
right: Expression,
600+
cmp: (Expression, Expression) => Expression): Expression = {
601+
val fields = left.dataType.asInstanceOf[StructType].fields
602+
fields.indices.map { i =>
603+
cmp(GetStructField(left, i), GetStructField(right, i))
604+
}.reduceLeft(And)
605+
}
606+
}
607+
567608
/**
568609
* Simplifies binary comparisons with semantically-equal expressions:
569610
* 1) Replace '<=>' with 'true' literal.

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -182,6 +182,7 @@ object RuleIdCollection {
182182
"org.apache.spark.sql.catalyst.optimizer.RewriteExceptAll" ::
183183
"org.apache.spark.sql.catalyst.optimizer.RewriteIntersectAll" ::
184184
"org.apache.spark.sql.catalyst.optimizer.SimplifyBinaryComparison" ::
185+
"org.apache.spark.sql.catalyst.optimizer.DecomposeStructComparison" ::
185186
"org.apache.spark.sql.catalyst.optimizer.SimplifyCaseConversionExpressions" ::
186187
"org.apache.spark.sql.catalyst.optimizer.SimplifyCasts" ::
187188
"org.apache.spark.sql.catalyst.optimizer.SimplifyConditionals" ::

0 commit comments

Comments
 (0)