You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "yadavay-amzn (via GitHub)" <gi...@apache.org> on 2026/06/01 06:44:34 UTC
[PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
yadavay-amzn opened a new pull request, #56244:
URL: https://github.com/apache/spark/pull/56244
### What changes were proposed in this pull request?
Add optimizer rule `DecomposeStructComparison` that rewrites struct-level equality (`=` and `<=>`) into a conjunction 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'`.
### Why are the changes needed?
Struct literal comparisons and tuple comparisons are treated as opaque predicates by the optimizer. Data source filter pushdown only understands scalar predicates, so struct equality cannot be pushed down for file pruning (Parquet row group skipping, partition pruning, etc.), even though the equivalent scalar predicates would be pushed.
### Does this PR introduce _any_ user-facing change?
Yes — queries filtering on struct equality will now benefit from file pruning and filter pushdown, improving performance on large tables.
### How was this patch tested?
Added `StructPredicateDecomposeSuite` with tests covering EqualTo, EqualNullSafe, nested structs, single-field structs, empty structs, tuple comparisons, non-deterministic guard, and GreaterThan exclusion.
### Was this patch authored or co-authored using generative AI tooling?
Yes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3503668827
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -69,7 +69,12 @@ object PushDownUtils extends Logging {
// Catalyst filter expression that can't be translated to data source filters.
val untranslatableExprs = mutable.ArrayBuffer.empty[Expression]
- for (filterExpr <- filters) {
+ // Expand struct equality predicates into field-level predicates for pushdown.
+ // Original struct predicates remain in the list and end up as post-scan filters.
+ val expandedFilters = DataSourceStrategy.expandStructPredicatesForPushdown(
Review Comment:
**The V2 wiring is on the wrong `pushFilters` branch, so it does not reach any V2 file source.**
`expandedFilters` is only consumed by the `SupportsPushDownFilters` arm (the loop just below). But all V2 file sources — Parquet, ORC, CSV, JSON, Avro — extend `FileScanBuilder`, which implements `SupportsPushDownCatalystFilters`, handled by the *separate* branch at lines 141-143:
```scala
case r: SupportsPushDownCatalystFilters =>
val postScanFilters = r.pushFilters(filters) // <- un-expanded `filters`, not `expandedFilters`
```
`SupportsPushDownFilters` is implemented only by `PythonScanBuilder` (and the V1Scan fallback in `V2ScanRelationPushDown`), neither of which does Parquet/ORC row-group pruning. So for the sources that would actually benefit from field-level pruning, the decomposition never runs — the V2 half of the feature the PR describes is inert.
The natural home for V2 file sources is `FileScanBuilder.pushFilters` (expand `dataFilters` before the translate loop there), or the `SupportsPushDownCatalystFilters` branch here. Worth confirming which interface the intended V2 sources actually use before landing.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/StructPredicatePushdownSuite.scala:
##########
@@ -0,0 +1,354 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution.datasources
+
+import org.apache.spark.sql.{DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.expressions.{EqualTo, Expression, GetStructField, Literal}
+import org.apache.spark.sql.execution.{FileSourceScanExec, FilterExec}
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types._
+
+/**
+ * Tests for struct equality predicate decomposition at the pushdown layer.
+ *
+ * Validates that:
+ * - Decomposed field-level predicates reach the scan (PushedFilters/DataFilters)
+ * - The original struct predicate is retained as a post-scan filter (correctness)
+ * - Null-valued literal fields are NOT pushed as `= null` (soundness)
+ * - maxFields bound is respected
+ * - Conf on/off behavior works correctly
+ * - Results are identical with rule on vs off (parity)
+ */
+class StructPredicatePushdownSuite extends QueryTest with SharedSparkSession {
Review Comment:
**No test covers the V2 datasource pushdown path.** Every test here uses `spark.read.parquet(...)` under the default `spark.sql.sources.useV1SourceList` (which includes `parquet`), so all of them exercise the **V1** `FileSourceStrategy` path. None routes through the DSv2 reader.
This matters because it's exactly the gap that hides the design issue on the V2 wiring (see the `PushDownUtils.scala` comment): a test on the V2 path would assert field predicates reach the scan and find that none do. Consider adding a case that forces the V2 path (e.g. `withSQLConf(SQLConf.USE_V1_SOURCE_LIST.key -> "")` so parquet resolves to DSv2) and checks the pushed filters — once the wiring reaches V2 file sources.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4851281487
Thanks @cloud-fan, both addressed:
- **V2 wiring:** moved the expansion into `FileScanBuilder.pushFilters` (`SupportsPushDownCatalystFilters`, which all V2 file sources use). Only the translate-and-push loop sees the expanded filters; the returned post-scan filters stay the original struct predicates, so the pushed field predicates remain a sound over-approximation like the V1 path. Removed the now-redundant expansion from `PushDownUtils.pushFilters`.
- **V2 test gap:** added a DSv2 case (`useV1SourceList=''`) asserting the decomposed field predicates reach `ParquetScan`'s pushed filters, plus a DSv2 on/off parity test. Confirmed the new test fails without the `FileScanBuilder` change and passes with it.
Also refreshed the PR description to reflect the V2 path via `FileScanBuilder`. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4862574769
**Merge summary** (posted by `merge_spark_pr.py`):
- merged into master https://github.com/apache/spark/commit/c3f6abd0
- merged into branch-4.x https://github.com/apache/spark/commit/6907555c
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3502688751
##########
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,
Review Comment:
**Design question: should this decomposition live at the pushdown/translation layer instead of as a blanket logical rule?**
`DecomposeStructComparison` matches `Filter` and rewrites its condition (`expressions.scala:600`) without inspecting the child relation, and runs unconditionally in the optimizer batch. So by the time `PushDownUtils.pushFilters` consults the concrete source (`PushDownUtils.scala:56`), the whole-struct `EqualTo` no longer exists -- the source only ever sees the field predicates. Two consequences:
1. **Silent pushdown regression for struct-capable sources.** A DSv2 connector that natively supports a whole-struct equality but not nested-leaf predicates (common -- many connectors only push top-level columns) now pushes *nothing* for `s = struct(...)`, where it previously pushed the whole struct. Since the rule is relation-blind, such a source can't opt out. This isn't a wrong-results bug (the rewrite is filter-equivalent) but a capability regression, and no test covers it.
2. **The exact-NULL-equivalence complexity is a consequence of this layer choice.** Because the rule mutates the real logical predicate, it must be exactly filter-equivalent -- which is the only reason the `IsNotNull` guard / `filterEquivalentToConjunction` / both-nullable handling are needed. Field decomposition already exists at the translation layer (`PushableColumn` extracts `GetStructField` chains, `DataSourceStrategy.scala` ~L915), and pushed filters are inexact over-approximations with the original `Filter` retained post-scan (`FileSourceStrategy.scala` ~L212) -- so a decomposition done there only needs to be a sound superset, not NULL-exact.
Would it make sense to perform the decomposition at the filter-translation / pushdown layer (`PushDownUtils`), scoping it to sources that lack whole-struct support? That preserves whole-struct pushdown for capable sources and drops the NULL-equivalence surface. (Gating the *logical* rule on child-relation type would instead be a layering smell -- the point is to move the work down, not teach the optimizer about physical sources.) There are tradeoffs worth your call: a logical rewrite also exposes field predicates to other optimizer rules, and any DSv2 source reporting exact pushdown would need the decomposed form kept inexact.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -564,6 +564,196 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
}
+/**
+ * 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] with PredicateHelper {
+ def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!conf.decomposeStructComparisonEnabled) {
+ return plan
+ }
+ plan.transformWithPruning(_.containsPattern(FILTER), ruleId) {
+ case f @ Filter(condition, _) =>
+ // A Filter drops rows whose condition is not TRUE, so at a top-level conjunct
+ // NULL and FALSE are indistinguishable. That lets us emit the bare, *pushable*
+ // field conjunction for a struct comparison sitting directly in the WHERE,
+ // instead of the null-preserving `If(...)` form (which is opaque to filter
+ // pushdown). Comparisons nested under Not/Or/etc. are NOT in filter position --
+ // there NULL vs FALSE is observable -- so those go through the null-preserving
+ // `decomposeCondition` path unchanged.
+ val rewritten = splitConjunctivePredicates(condition).map {
+ 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.
+ // 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)
+ if (nullGuards.isEmpty) conjunction
Review Comment:
Nit: verbless sentence fragment -- "`<=>` on the pushable path (one foldable, one cheap)." "Guard nullable operands" already flows into line 622, so just give the fragment a verb. (If @peter-toth's suggestion on line 618 to merge this arm with the `EqualTo` arm lands, this comment is removed entirely, so feel free to skip.)
```suggestion
// This is the <=> pushable path (one operand foldable, the other cheap). Guard nullable operands
```
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/StructPredicateDecomposeSuite.scala:
##########
@@ -0,0 +1,719 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.dsl.expressions._
+import org.apache.spark.sql.catalyst.dsl.plans._
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest}
+import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, Filter, Join, LocalRelation, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.RuleExecutor
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{IntegerType, StructField, StructType}
+
+/**
+ * Tests for [[DecomposeStructComparison]]. Most semantic-correctness tests are oracle
+ * tests that compare results of evaluating the original expression against the rewritten
+ * expression: the rule must never change the value of an existing predicate, only its
+ * shape. End-to-end Filter behavior is exercised in `StructPredicateDecomposeE2ESuite`.
+ */
+class StructPredicateDecomposeSuite extends PlanTest {
+
+ // The rule is gated by `decomposeStructComparison.enabled`, default off. Tests in
+ // this suite enable it for the rule to fire.
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ SQLConf.get.setConf(SQLConf.DECOMPOSE_STRUCT_COMPARISON_ENABLED, true)
+ }
+
+ override def afterAll(): Unit = {
+ SQLConf.get.unsetConf(SQLConf.DECOMPOSE_STRUCT_COMPARISON_ENABLED)
+ super.afterAll()
+ }
+
+ object Optimize extends RuleExecutor[LogicalPlan] {
+ val batches =
+ Batch("Optimizer", FixedPoint(10),
+ DecomposeStructComparison,
+ ConstantFolding,
+ BooleanSimplification,
+ SimplifyBinaryComparison) :: Nil
+ }
+
+ private val structRelation = LocalRelation(
+ $"nested".struct($"col1".string, $"col2".string),
+ $"id".int
+ )
+
+ // -----------------------------------------------------------------------------------
+ // Structural tests: verify the rewrite happens (or does not happen) for the right
+ // input shapes. Semantic preservation is covered by the oracle tests further below.
+ // -----------------------------------------------------------------------------------
+
+ test("SPARK-56660: struct equality is decomposed into field-level predicates") {
+ val structLiteral = CreateNamedStruct(Seq(
+ Literal("col1"), Literal("foo"),
+ Literal("col2"), Literal("bar")
+ ))
+ val plan = structRelation
+ .where(EqualTo($"nested", structLiteral))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+ val filter = optimized.collect { case f: Filter => f }.head
+
+ // After decomposition + ConstantFolding + BooleanSimplification, no struct-level
+ // EqualTo should remain in the filter condition.
+ val structComparisons = filter.condition.collect {
+ case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(structComparisons.isEmpty,
+ s"Should not have struct-level comparison. Plan:\n${optimized.treeString}")
+ }
+
+ test("SPARK-56660: struct EqualNullSafe is decomposed into field-level predicates") {
+ val structLiteral = CreateNamedStruct(Seq(
+ Literal("col1"), Literal("foo"),
+ Literal("col2"), Literal("bar")
+ ))
+ val plan = structRelation
+ .where(EqualNullSafe($"nested", structLiteral))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+ val filter = optimized.collect { case f: Filter => f }.head
+
+ val structComparisons = filter.condition.collect {
+ case EqualNullSafe(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(structComparisons.isEmpty,
+ s"Should not have struct-level EqualNullSafe. Plan:\n${optimized.treeString}")
+ }
+
+ test("SPARK-56660: nested struct equality is recursively decomposed") {
+ val nestedStructRelation = LocalRelation(
+ $"outer".struct($"inner".struct($"a".string, $"b".string), $"c".int),
+ $"id".int
+ )
+ val structLiteral = CreateNamedStruct(Seq(
+ Literal("inner"), CreateNamedStruct(Seq(
+ Literal("a"), Literal("x"),
+ Literal("b"), Literal("y")
+ )),
+ Literal("c"), Literal(42)
+ ))
+ val plan = nestedStructRelation
+ .where(EqualTo($"outer", structLiteral))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+ val filter = optimized.collect { case f: Filter => f }.head
+
+ val hasStructComparison = filter.condition.collect {
+ case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
+ case EqualNullSafe(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(hasStructComparison.isEmpty,
+ s"Nested struct should be fully decomposed. Plan:\n${optimized.treeString}")
+ }
+
+ // 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")))
+ val plan = relation
+ .where(EqualTo(leftStruct, rightStruct))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+ val filter = optimized.collect { case f: Filter => f }.head
+
+ val structComparisons = filter.condition.collect {
+ case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(structComparisons.nonEmpty,
+ s"Tuple comparison should NOT be decomposed (left is not cheap). " +
+ s"Plan:\n${optimized.treeString}")
+ }
+
+ // 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)
+ )
+ val plan = relation
+ .where(EqualTo($"s1", $"s2"))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+ val filter = optimized.collect { case f: Filter => f }.head
+
+ val structComparisons = filter.condition.collect {
+ case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(structComparisons.nonEmpty,
+ s"Struct-to-struct column comparison should NOT be decomposed (no foldable operand). " +
+ s"Plan:\n${optimized.treeString}")
+ }
+
+ // -----------------------------------------------------------------------------------
+ // Negative tests: verify the rule does NOT fire for shapes outside its scope.
+ // -----------------------------------------------------------------------------------
+
+ // GreaterThan/LessThan on structs use lexicographic ordering across fields. The rule
+ // intentionally only matches EqualTo / EqualNullSafe; one negative test for the
+ // comparator family is sufficient since all non-equality cases share a code path.
+ test("SPARK-56660: non-equality struct comparisons are NOT decomposed") {
+ val plan = structRelation
+ .where(GreaterThan($"nested",
+ CreateNamedStruct(Seq(Literal("col1"), Literal("a"), Literal("col2"), Literal("b")))))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+ val filter = optimized.collect { case f: Filter => f }.head
+
+ val hasStructGreaterThan = filter.condition.collect {
+ case GreaterThan(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(hasStructGreaterThan.nonEmpty,
+ "GreaterThan on structs should NOT be decomposed")
+ }
+
+ // Non-deterministic expressions cannot be decomposed: each field-level comparison
+ // would re-evaluate the expression independently with a different value.
+ test("SPARK-56660: non-deterministic expressions are not decomposed") {
+ val doubleStructRelation = LocalRelation(
+ $"s".struct($"a".double, $"b".double),
+ $"id".int
+ )
+ val nonDetStruct = CreateNamedStruct(Seq(
+ Literal("a"), Rand(0),
+ Literal("b"), Rand(1)
+ ))
+ val plan = doubleStructRelation
+ .where(EqualTo($"s", nonDetStruct))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+ val filter = optimized.collect { case f: Filter => f }.head
+
+ val hasStructComparison = filter.condition.collect {
+ case EqualTo(_, _: CreateNamedStruct) => true
+ case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(hasStructComparison.nonEmpty,
+ "Non-deterministic struct comparison should NOT be decomposed")
+ }
+
+ // Critical: join conditions must NOT be decomposed. The rule is Filter-scoped.
+ // Decomposing join keys would break equi-join detection for SMJ/BHJ planning.
+ test("SPARK-56660: join condition struct equality is NOT decomposed") {
+ val left = LocalRelation(
+ $"s1".struct($"a".string, $"b".int),
+ $"id".int
+ )
+ val right = LocalRelation(
+ $"s2".struct($"a".string, $"b".int),
+ $"id".int
+ )
+ val plan = left.join(right, Inner,
+ Some(EqualTo($"s1", $"s2"))).analyze
+
+ val optimized = Optimize.execute(plan)
+
+ val joins = optimized.collect { case j: Join => j }
+ assert(joins.nonEmpty, "Plan should contain a Join")
+ val joinCondition = joins.head.condition.get
+ val structComparisons = joinCondition.collect {
+ case EqualTo(l, _) if l.dataType.isInstanceOf[StructType] => true
+ }
+ assert(structComparisons.nonEmpty,
+ s"Join condition should NOT be decomposed. Plan:\n${optimized.treeString}")
+ }
+
+ // The rule is Filter-scoped via the outer `case Filter` pattern, so an Aggregate
+ // grouping key on a struct is structurally unreachable. This test guards against
+ // accidentally broadening the rule's scope in a future change.
+ test("SPARK-56660: aggregate grouping struct key is NOT decomposed (scope guard)") {
+ val relation = LocalRelation(
+ $"s".struct($"a".string, $"b".int),
+ $"value".int
+ )
+ val plan = relation
+ .groupBy($"s")(count($"value").as("cnt"))
+ .analyze
+
+ val optimized = Optimize.execute(plan)
+
+ val aggregates = optimized.collect { case a: Aggregate => a }
+ assert(aggregates.nonEmpty, "Plan should contain an Aggregate")
+ val groupingExprs = aggregates.head.groupingExpressions
+ val hasStructGroupKey = groupingExprs.exists(_.dataType.isInstanceOf[StructType])
+ assert(hasStructGroupKey,
+ s"Aggregate grouping key should NOT be decomposed. Plan:\n${optimized.treeString}")
+ }
+
+ // -----------------------------------------------------------------------------------
+ // Oracle tests: build an `EqualTo` / `EqualNullSafe` directly, fold the rewrite
+ // through Optimize, and assert the rewritten expression evaluates to the same value
+ // as the original on representative inputs. These are the tests that catch
+ // correctness regressions in NULL handling.
+ // -----------------------------------------------------------------------------------
+
+ /**
+ * Returns the optimized condition produced by the rule for `original`, by wrapping it
+ * in a `Filter` over a single-column relation. The relation type does not matter; the
+ * rule rewrites the condition expression in place.
+ */
+ private def optimizeAsFilter(original: Expression): Expression = {
+ val dummy = LocalRelation($"id".int)
+ val plan = dummy.where(original).analyze
+ Optimize.execute(plan).collect { case f: Filter => f.condition }.head
+ }
+
+ private def assertSameSemantics(original: Expression): Unit = {
+ val rewritten = optimizeAsFilter(original)
+ val expected = original.eval(InternalRow.empty)
+ val actual = rewritten.eval(InternalRow.empty)
+ assert(expected === actual,
+ s"Rewritten expression changed semantics.\n" +
+ s" original: $original -> $expected\n" +
+ s" rewritten: $rewritten -> $actual")
+ }
+
+ private def litStruct(values: (String, Any, org.apache.spark.sql.types.DataType)*)
+ : CreateNamedStruct = {
+ CreateNamedStruct(values.flatMap {
+ case (n, v, t) => Seq(Literal(n), Literal.create(v, t))
+ })
+ }
+
+ test("SPARK-56660 oracle: EqualTo(struct(1, null), struct(1, null)) preserves TRUE") {
+ val left = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ val right = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ assertSameSemantics(EqualTo(left, right))
+ }
+
+ test("SPARK-56660 oracle: EqualTo(struct(1, null), struct(1, 2)) preserves FALSE") {
+ val left = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ val right = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+ assertSameSemantics(EqualTo(left, right))
+ }
+
+ test("SPARK-56660 oracle: EqualTo(struct(1, 2), struct(1, 2)) preserves TRUE") {
+ val left = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+ val right = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+ assertSameSemantics(EqualTo(left, right))
+ }
+
+ test("SPARK-56660 oracle: EqualTo with nullable left struct (null = struct(...)) " +
+ "is filter-equivalent") {
+ // For a struct EqualTo sitting directly in a Filter, the rule emits the bare, pushable
+ // field conjunction. On a whole-struct-null row the original yields NULL while the
+ // rewrite yields FALSE -- these are NOT value-equal, but they ARE filter-equivalent
+ // (a Filter drops the row in both cases, since neither is TRUE). Assert that filter
+ // outcome, which is the contract the rule actually guarantees in filter position.
+ val structType = StructType(Seq(StructField("a", IntegerType, nullable = false)))
+ val nullableRelation = LocalRelation(AttributeReference("s", structType, nullable = true)())
+ val sAttr = nullableRelation.output.head
+ val literal = litStruct(("a", 1, IntegerType))
+ val original = EqualTo(sAttr, literal)
+
+ val rewritten = Optimize.execute(nullableRelation.where(original).analyze).collect {
+ case f: Filter => f.condition
+ }.head
+ val row = InternalRow(null)
+ val boundOriginal = BindReferences.bindReference(original, Seq(sAttr))
+ val boundRewritten = BindReferences.bindReference(rewritten, Seq(sAttr))
+
+ // A Filter keeps a row iff the predicate evaluates to TRUE; both NULL and FALSE drop it.
+ def keptByFilter(v: Any): Boolean = v == true
+ assert(keptByFilter(boundOriginal.eval(row)) === keptByFilter(boundRewritten.eval(row)),
+ s"Filter outcome mismatch on whole-struct-null row.\n" +
+ s" original: $original -> ${boundOriginal.eval(row)}\n" +
+ s" rewritten: $rewritten -> ${boundRewritten.eval(row)}")
+ // Document the value-level divergence that makes the rewrite pushable.
+ assert(boundOriginal.eval(row) == null && boundRewritten.eval(row) == false,
+ "Expected original NULL and rewritten FALSE on the whole-struct-null row")
+ }
+
+ test("SPARK-56660 oracle: EqualNullSafe(struct(1, null), struct(1, null)) preserves TRUE") {
+ val left = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ val right = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ assertSameSemantics(EqualNullSafe(left, right))
+ }
+
+ test("SPARK-56660 oracle: EqualNullSafe(struct(1, null), struct(1, 2)) preserves FALSE") {
+ val left = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ val right = litStruct(("a", 1, IntegerType), ("b", 2, IntegerType))
+ assertSameSemantics(EqualNullSafe(left, right))
+ }
+
+ test("SPARK-56660 oracle: EqualNullSafe with one nullable struct null preserves FALSE") {
+ val structType = StructType(Seq(StructField("a", IntegerType, nullable = false)))
+ val nullableRelation = LocalRelation(AttributeReference("s", structType, nullable = true)())
+ val sAttr = nullableRelation.output.head
+ val literal = litStruct(("a", 1, IntegerType))
+ val original = EqualNullSafe(sAttr, literal)
+
+ val plan = nullableRelation.where(original).analyze
+ val rewritten = Optimize.execute(plan).collect {
+ case f: Filter => f.condition
+ }.head
+ val row = InternalRow(null)
+ val boundOriginal = BindReferences.bindReference(original, Seq(sAttr))
+ val boundRewritten = BindReferences.bindReference(rewritten, Seq(sAttr))
+ assert(boundOriginal.eval(row) === boundRewritten.eval(row),
+ s"EqualNullSafe one-side-null mismatch.\n original: $original\n rewritten: $rewritten")
+ }
+
+ test("SPARK-56660 oracle: Not(EqualTo) preserves semantics with null fields") {
+ // Negation is the canonical place where field-level decomposition can quietly diverge
+ // from struct-level semantics if NULL handling is wrong: Not(NULL) is NULL, not TRUE.
+ val left = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ val right = litStruct(("a", 1, IntegerType), ("b", null, IntegerType))
+ 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 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),
+ // 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")
+ }
+
+ // -----------------------------------------------------------------------------------
+ // 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
Review Comment:
`canDecompose` has no `sameType` check (the only `sameType` token in this file is here in the comment) -- type compatibility is enforced upstream by the analyzer (`DATATYPE_MISMATCH.BINARY_OP_DIFF_TYPES`, as the first sentence notes) plus the `StructType` match and `l.length == r.length` arity check inside `canDecompose`. Naming a check that doesn't exist will mislead a future reader.
```suggestion
// the optimizer ever runs. The `StructType` match and `l.length == r.length` arity check
// inside `canDecompose` are therefore defense-in-depth guards, not directly testable at
// the rule level.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4855856666
let's make the CI happy.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4627464289
@yyanyy Great catch on the NULL semantics — you're right. Spark's struct equality uses `InterpretedOrdering` which treats null=null within fields as equal (returns TRUE), while `EqualTo(null, null)` returns NULL.
Fixed: the decomposition now uses `EqualNullSafe` (`<=>`) for per-field comparisons, which matches the struct equality semantics exactly:
- `null <=> null` → true (matches struct behavior)
- `null <=> 2` → false (matches struct behavior)
The only remaining discrepancy is when the entire struct itself is null (original returns NULL, decomposed returns FALSE) — but since our rule only fires in Filter context, this is harmless (both NULL and FALSE exclude the row from WHERE).
Also added a width guard (max 100 fields) to prevent stack overflow on very wide/deeply nested structs, per your second concern.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "yyanyy (via GitHub)" <gi...@apache.org>.
yyanyy commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3359726511
##########
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(
Review Comment:
I'm a little worried that combining this 1/ recursive transform of nested struct and 2/ unconditionally AND'ing all field together within struct regardless of how many fields the struct hold, together could cause stack overflow error for deeply nested and/or huge struct, and worst case pose DOS threat to the hosting system. Do we think if we should guard towards these by setting an upper limit, and/or in general have this feature behind a config flag to guard against this and the below issue?
##########
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))
Review Comment:
Apparently there's a nuance case in spark - while `1 = NULL` returns null, `struct(1, null) = struct(1, null)` returns true; with this code change, this behavior is lost. I think we will need to be careful in null handling in general
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3452216031
##########
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")
Review Comment:
`4.2.0` is already released; this and the `maxFields` config (line 543) should be `4.3.0` (the next release).
```suggestion
.version("4.3.0")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4812396915
Thanks @yadavay-amzn for the update, let me take a look at the PR next week.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4846468053
Thanks @peter-toth - addressed all three:
- Folded the EqualTo/EqualNullSafe arms into one `case Equality(l, r)` guarded by `canDecompose && filterEquivalentToConjunction`. Adding the guard to the `<=>` path is safe: `canDecompose` requires a foldable (hence non-nullable) operand, so both-nullable is unreachable and falls through to `decomposeEqualNullSafe`; the two-nullable-columns test confirms it.
- Untested paths: you were right the Not/NULL e2e tests were vacuous - `withColumn("s", struct(...))` inlines to a non-cheap `CreateNamedStruct`, so `canDecompose` was false and the rule never fired. Replaced them with Parquet-backed tables where `s` is a nullable struct attribute, and added catalyst tests under `Not`/`Or` that actually reach `decomposeEqualTo`'s nullable `If(anyNull, NULL, conjunction)` branch and `decomposeEqualNullSafe`'s branches, asserting NULL-vs-FALSE parity across whole-null / all-null-fields / non-null rows (verified by breaking each branch).
- Refreshed the PR description: scope narrowing (foldable + isCheap), the tuple-not-decomposed note, the IsNotNull-guarded pushable form, and the How-tested list.
PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4795027584
Thanks for the careful review, @peter-toth - both points were spot on.
**Scope:** narrowed `canDecompose` to fire only when one operand is foldable and the other is `CollapseProject.isCheap`. So `struct_col1 = struct_col2` (no pushdown benefit) and `udf() = literal` (non-cheap, would duplicate the UDF) no longer decompose; it now only produces the pushable, non-duplicating `col <=> literal` shape it was meant for.
**NULL correctness:** you're right the pushable shortcut dropped the whole-null vs all-null-fields distinction. I now AND `IsNotNull(<nullable operand>)` into the conjunction on both the EqualTo and EqualNullSafe pushable paths (it pushes down too), so an `s IS NULL` row is correctly excluded for `s = named_struct('a', null)`. Verified with eval-level tests and end-to-end `assertSameUnderRule` over Parquet data containing genuine null structs (these fail without the guard). The non-pushable paths were already correct and are unchanged.
Also bumped both confs to 4.3.0. PTAL.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3503431104
##########
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,
Review Comment:
@cloud-fan thanks, agreed the placement was the core issue. Reworked to do the decomposition at the pushdown/translation layer instead of as a logical rule:
- Removed the DecomposeStructComparison logical rule entirely, along with its exact-NULL-equivalence machinery (IsNotNull guard / filterEquivalentToConjunction / decomposeEqualTo / decomposeEqualNullSafe).
- Added DataSourceStrategy.expandStructPredicatesForPushdown, wired into FileSourceStrategy (V1) and PushDownUtils.pushFilters (V2). It adds field-level equality predicates as pushdown candidates; the original struct predicate is left untouched and retained post-scan, so the pushed field predicates only need to be a sound over-approximation, not NULL-exact. This also fixes the capability regression you raised: the whole-struct predicate is still present for a source that can push it natively.
- Soundness: for a struct literal with a NULL field we do not push s.fi = null (that would drop valid rows); only non-null fields are pushed, still a sound superset. Both = and <=> are handled.
- Renamed the confs to spark.sql.sources.structPredicateDecompose.{enabled (default true), maxFields (default 100)} to reflect the sources/pushdown placement.
- New StructPredicatePushdownSuite asserts the field predicates reach PushedFilters, the original filter is retained post-scan, null-valued literal fields are not pushed, maxFields is bounded, conf on/off, and rule-on == rule-off parity across whole-null / all-null-fields / non-null rows.
Does this placement look right to you? Happy to adjust (e.g. default OFF, or scoping by source capability) if you would prefer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #56244: [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer
URL: https://github.com/apache/spark/pull/56244
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3499453511
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala:
##########
@@ -564,6 +564,201 @@ object BooleanSimplification extends Rule[LogicalPlan] with PredicateHelper {
}
+/**
+ * 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] with PredicateHelper {
+ def apply(plan: LogicalPlan): LogicalPlan = {
+ if (!conf.decomposeStructComparisonEnabled) {
+ return plan
+ }
+ plan.transformWithPruning(_.containsPattern(FILTER), ruleId) {
+ case f @ Filter(condition, _) =>
+ // A Filter drops rows whose condition is not TRUE, so at a top-level conjunct
+ // NULL and FALSE are indistinguishable. That lets us emit the bare, *pushable*
+ // field conjunction for a struct comparison sitting directly in the WHERE,
+ // instead of the null-preserving `If(...)` form (which is opaque to filter
+ // pushdown). Comparisons nested under Not/Or/etc. are NOT in filter position --
+ // 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) =>
+ // 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) =>
Review Comment:
Nit: These two arms (the `EqualTo` case above and this `EqualNullSafe` case) now have byte-identical bodies -- in Filter position `s = lit` and `s <=> lit` decompose to the same `IsNotNull-guards AND fieldConjunction`. They could fold into one:
```scala
case EqualTo(l, r) | EqualNullSafe(l, r)
if canDecompose(l, r) && filterEquivalentToConjunction(l, r) =>
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)
```
Beyond de-duplication, this closes a small asymmetry: the `EqualTo` arm is structurally guarded by `filterEquivalentToConjunction` (a both-nullable pair falls through to the null-preserving `decomposeEqualTo`), but this `<=>` arm has no such guard -- its correctness for a both-nullable pair relies on `NullPropagation` (ordered before this rule) having already rewritten `<=> null` to `IsNull`. I verified that holds today (`s <=> CAST(NULL AS struct<...>)` optimizes to `isnull(s)`, and both-nullable columns are rejected by `canDecompose`), so this is hardening, not a bugfix -- but routing both arms through the same guard makes the `<=>` path self-evidently correct without depending on batch ordering. (A both-nullable `<=>` correctly falls through to `decomposeEqualNullSafe`, which already handles it.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4759550793
@yaooqinn @peter-toth Could you please review when you get a chance. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on code in PR #56244:
URL: https://github.com/apache/spark/pull/56244#discussion_r3471299366
##########
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")
Review Comment:
Done - both confs now use 4.3.0.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "yadavay-amzn (via GitHub)" <gi...@apache.org>.
yadavay-amzn commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4858610056
Thanks for reviewing @cloud-fan @peter-toth and @yyanyy !
Fixed the CI issue, should be green now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field predicates at the data source pushdown layer [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4862562101
thanks, merging to master/4.x
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-56660][SQL] Decompose struct equality into field-level predicates for filter pushdown [spark]
Posted by "peter-toth (via GitHub)" <gi...@apache.org>.
peter-toth commented on PR #56244:
URL: https://github.com/apache/spark/pull/56244#issuecomment-4844497927
cc @cloud-fan
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org