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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
