This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 5f1f8da99f50 [SPARK-57442][SQL] Add opt-in nullability check to
logical plan integrity validation
5f1f8da99f50 is described below
commit 5f1f8da99f5039460828632ef37b4ffa5a999aea
Author: Wenchen Fan <[email protected]>
AuthorDate: Tue Jun 16 14:09:12 2026 -0700
[SPARK-57442][SQL] Add opt-in nullability check to logical plan integrity
validation
### What changes were proposed in this pull request?
This PR adds an opt-in nullability check to logical plan integrity
validation.
- A new internal config `spark.sql.planChangeValidation.nullability`
(default `false`) gates the check. It is a sub-flag of the existing
`spark.sql.planChangeValidation`.
- A new `LogicalPlanIntegrity.validateNullability` walks the plan and flags
any operator that references one of its child's output attributes with a
**narrower** nullability than the child produces — i.e. declaring the attribute
`nullable = false` while the child produces it as `nullable = true`. It is
wired into `validateOptimizedPlan`'s existing check chain.
- References declared **wider** (nullable over a non-nullable child) are
safe and not flagged; references whose `ExprId` is not produced by any child
(e.g. outer references) are ignored.
### Why are the changes needed?
When a rule synthesizes operators, every reference to a child output must
carry the nullability the child actually produces. A classic miss is an
outer-join rewrite that references the null-widened side with its original
non-nullable attributes — for example SPARK-56395 in `RewriteNearestByJoin`,
where the synthesized `Aggregate` referenced the right side (widened to
nullable by the synthetic `LEFT OUTER` join) with its original non-nullable
metadata.
Such a mismatch is a latent plan-integrity defect: the rows are unchanged,
so it is invisible to result-equivalence reasoning, to `validateSchemaOutput`
(which compares types `asNullable`), and to green CI. Yet a later optimization
that trusts the non-nullable declaration can drop a null check and produce
wrong results. There is currently no integrity check that catches this class of
defect; this adds one.
The check is disabled by default — even though
`spark.sql.planChangeValidation` defaults to on in test runs — because not all
existing rules maintain precise nullability today, and enabling it globally
would surface pre-existing benign mismatches. Suites that have been audited (or
rules that have been fixed) can opt in to guard against regressions.
### Does this PR introduce _any_ user-facing change?
No. The config is internal and disabled by default; the check only runs
under plan-change validation when explicitly enabled.
### How was this patch tested?
New unit test in `LogicalPlanIntegritySuite` that, with the config enabled,
asserts:
- a reference declared non-nullable over a nullable child is flagged (and
the message names the offending attribute);
- a faithful reference (same nullability as the child) passes;
- a wider reference (nullable over a non-nullable child) passes;
and that the check is a no-op when the config is disabled.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Opus 4.8)
This pull request and its description were written by Isaac.
Closes #56502 from cloud-fan/SPARK-oss-nullability-plan-validation.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/catalyst/plans/logical/LogicalPlan.scala | 61 +++++++++++
.../org/apache/spark/sql/internal/SQLConf.scala | 16 +++
.../plans/logical/LogicalPlanIntegritySuite.scala | 111 ++++++++++++++++++++-
3 files changed, 187 insertions(+), 1 deletion(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
index f0020688eceb..d573f4886254 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlan.scala
@@ -31,6 +31,7 @@ import
org.apache.spark.sql.catalyst.trees.TreePattern.{LOGICAL_QUERY_STAGE, Tre
import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.MetadataColumnHelper
import org.apache.spark.sql.errors.{QueryCompilationErrors,
QueryExecutionErrors}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.{DataType, StructType}
@@ -436,6 +437,59 @@ object LogicalPlanIntegrity {
}
}
+ /**
+ * This method validates that no operator references one of its child's
output attributes with a
+ * narrower nullability than the child produces -- declaring the attribute
non-nullable while the
+ * child produces it as nullable. Such a mismatch (for example, an
outer-join-widened column that
+ * a synthesized operator references with its original non-nullable
metadata) is a latent
+ * plan-integrity defect: the rows are unchanged, so result-equivalence
reasoning and
+ * `validateSchemaOutput` (which compares types `asNullable`) both miss it,
yet a later
+ * optimization that trusts the non-nullable declaration may drop a null
check and produce wrong
+ * results. Declaring a reference *wider* (nullable over a non-nullable
child) is safe and is not
+ * flagged. References whose `ExprId` is not produced by any child (e.g.
outer references) are
+ * ignored.
+ *
+ * Scope limitations:
+ * - Only top-level `AttributeReference.nullable` is compared. Nested
nullability (struct
+ * fields, array elements, map values) is not validated, so a child
producing
+ * `struct<a: long /* nullable */>` referenced as `struct<a: long /*
non-null */>` keeps the
+ * same top-level `nullable` and passes. This mirrors
`validateSchemaOutput`, which compares
+ * types `asNullable`.
+ * - The check is strictly local: it compares a parent operator against its
immediate children
+ * over `p.expressions`. A narrowing introduced at a pure pass-through
boundary (an operator
+ * that narrows an attribute it only forwards via `output`, without
re-referencing it in
+ * `expressions`) is not flagged at that node, and the parent then treats
the already-narrowed
+ * value as the child's truth. The targeted SPARK-56395 shape is caught
because the synthesized
+ * references appear in `expressions`.
+ * - Subquery plans are not reached by `plan.collect` (they live inside
expressions, not as
+ * children), so narrowing inside a subquery is not validated.
+ *
+ * The check is disabled unless `spark.sql.planChangeValidation.nullability`
is set, because not
+ * all rules maintain precise nullability today; it is intended to be
enabled by targeted suites.
+ * Returns the error message if the check does not pass, or None otherwise.
+ */
+ def validateNullability(plan: LogicalPlan): Option[String] = {
+ if (!SQLConf.get.getConf(SQLConf.PLAN_CHANGE_VALIDATION_NULLABILITY)) {
+ return None
+ }
+ plan.collect {
+ case p if p.childrenResolved =>
+ val childNullable = p.children.flatMap(_.output).groupBy(_.exprId).map
{
+ // If multiple child attributes share an ExprId, treat the column as
nullable if any is.
+ case (exprId, attrs) => exprId -> attrs.exists(_.nullable)
+ }
+ p.expressions.flatMap(_.collect {
+ case ref: AttributeReference
+ if !ref.nullable && childNullable.getOrElse(ref.exprId, false) =>
+ s"${p.nodeName} references ${ref.name}#${ref.exprId.id} as
non-nullable but its " +
+ "child produces it as nullable"
+ })
+ }.flatten.headOption.map { mismatch =>
+ "The plan declares an attribute non-nullable while its child produces it
as nullable, " +
+ s"which is a plan-integrity defect: $mismatch. The
plan:\n${plan.treeString}"
+ }
+ }
+
/**
* Validate the structural integrity of an optimized plan.
* For example, we can check after the execution of each rule that each plan:
@@ -444,7 +498,13 @@ object LogicalPlanIntegrity {
* - has globally-unique attribute IDs
* - has the same result schema as the previous plan
* - has no dangling attribute references
+ * - has valid aggregate expressions
+ * - references each child output attribute with consistent (or wider)
nullability
+ * (only when `spark.sql.planChangeValidation.nullability` is enabled)
* If `lightweight` is true, we only run the first check above.
+ *
+ * Note that these checks run only during plan-change validation around the
optimizer, so defects
+ * introduced during analysis (rather than optimization) are not caught here.
*/
def validateOptimizedPlan(
previousPlan: LogicalPlan,
@@ -474,6 +534,7 @@ object LogicalPlanIntegrity {
.orElse(LogicalPlanIntegrity.validateSchemaOutput(previousPlan,
currentPlan))
.orElse(LogicalPlanIntegrity.validateNoDanglingReferences(currentPlan))
.orElse(LogicalPlanIntegrity.validateAggregateExpressions(currentPlan))
+ .orElse(LogicalPlanIntegrity.validateNullability(currentPlan))
.map(err => s"${err}\nPrevious schema:${previousPlan.output.mkString(",
")}" +
s"\nPrevious plan: ${previousPlan.treeString}")
validation
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index 4f76dec323ef..e017a9076b1d 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -619,6 +619,22 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PLAN_CHANGE_VALIDATION_NULLABILITY =
+ buildConf("spark.sql.planChangeValidation.nullability")
+ .internal()
+ .doc(s"When true, and ${PLAN_CHANGE_VALIDATION.key} is enabled, plan
change validation " +
+ "additionally checks that no operator references a child output
attribute with a " +
+ "narrower nullability than the child produces (declaring it
non-nullable while the child " +
+ "produces it as nullable). Such a mismatch is a latent plan-integrity
defect that can " +
+ "enable unsafe optimizations. The check runs only during plan change
validation around " +
+ "the optimizer, so nullability defects introduced during analysis are
not caught. " +
+ "Disabled by default because not all rules maintain precise
nullability today; intended " +
+ "to be enabled by targeted test suites.")
+ .version("4.3.0")
+ .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)
+ .booleanConf
+ .createWithDefault(false)
+
val ALLOW_NAMED_FUNCTION_ARGUMENTS =
buildConf("spark.sql.allowNamedFunctionArguments")
.doc("If true, Spark will turn on support for named parameters for all
functions that has" +
" it implemented.")
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanIntegritySuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanIntegritySuite.scala
index 4c4152a1ecbd..71650f37249f 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanIntegritySuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/logical/LogicalPlanIntegritySuite.scala
@@ -20,7 +20,8 @@ package org.apache.spark.sql.catalyst.plans.logical
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.dsl.plans._
import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute,
AttributeReference}
-import org.apache.spark.sql.catalyst.plans.PlanTest
+import org.apache.spark.sql.catalyst.plans.{LeftOuter, PlanTest}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.types.LongType
class LogicalPlanIntegritySuite extends PlanTest {
@@ -32,6 +33,18 @@ class LogicalPlanIntegritySuite extends PlanTest {
copy(child = newChild)
}
+ // A binary node that references an attribute (via the `ref` field, picked
up by `expressions`)
+ // and exposes both children's outputs. Used to exercise nullability
validation when the same
+ // `ExprId` is produced by more than one child.
+ case class RefTestPlan(left: LogicalPlan, right: LogicalPlan, ref:
AttributeReference)
+ extends BinaryNode {
+ override val analyzed = true
+ override def output: Seq[Attribute] = left.output ++ right.output
+ override protected def withNewChildrenInternal(
+ newLeft: LogicalPlan, newRight: LogicalPlan): LogicalPlan =
+ copy(left = newLeft, right = newRight)
+ }
+
test("Checks if the same `ExprId` refers to a semantically-equal attribute
in a plan output") {
val t = LocalRelation($"a".int, $"b".int)
assert(hasUniqueExprIdsForOutput(OutputTestPlan(t, t.output)).isEmpty)
@@ -50,4 +63,100 @@ class LogicalPlanIntegritySuite extends PlanTest {
assert(checkIfSameExprIdNotReused(t.select(Alias(a + b, "ab")(exprId =
a.exprId))).isDefined)
assert(checkIfSameExprIdNotReused(t.select(Alias(a + b, "ab")(exprId =
b.exprId))).isDefined)
}
+
+ test("Checks if an attribute is referenced as non-nullable over a nullable
child") {
+ val nullableChild = LocalRelation(AttributeReference("a", LongType,
nullable = true)())
+ val childAttr = nullableChild.output.head
+ // A Project that references the child's column but declares it
non-nullable (same ExprId) --
+ // the SPARK-56395 plan-integrity defect, where a synthesized operator
references an
+ // outer-join-widened (nullable) column with its original non-nullable
metadata.
+ val narrowedRef = AttributeReference("a", LongType, nullable =
false)(childAttr.exprId)
+ val badPlan = Project(Seq(narrowedRef), nullableChild)
+
+ // Disabled by default: even the inconsistent plan passes.
+ assert(validateNullability(badPlan).isEmpty)
+
+ withSQLConf(SQLConf.PLAN_CHANGE_VALIDATION_NULLABILITY.key -> "true") {
+ // A faithful reference (same nullability as the child) passes.
+ assert(validateNullability(Project(Seq(childAttr),
nullableChild)).isEmpty)
+ // The narrowed reference is flagged, naming the offending attribute.
+
assert(validateNullability(badPlan).exists(_.contains(s"a#${childAttr.exprId.id}")))
+ // Declaring a reference *wider* (nullable over a non-nullable child) is
safe.
+ val nonNullChild = LocalRelation(AttributeReference("b", LongType,
nullable = false)())
+ val widerRef =
+ AttributeReference("b", LongType, nullable =
true)(nonNullChild.output.head.exprId)
+ assert(validateNullability(Project(Seq(widerRef), nonNullChild)).isEmpty)
+ }
+ }
+
+ test("Nullability check flags the motivating outer-join shape over a
multi-child plan") {
+ // SPARK-56395: an operator over a LEFT OUTER join references the
null-widened (right) side
+ // with its original non-nullable metadata. This is the realistic
motivating shape -- a
+ // synthesized Aggregate over a join whose output widened the right side
to nullable.
+ val left = LocalRelation(AttributeReference("a", LongType, nullable =
false)())
+ val right = LocalRelation(AttributeReference("b", LongType, nullable =
false)())
+ val a = left.output.head
+ val b = right.output.head
+ // The join widens `b` to nullable in its output, but `b` here still
carries the original
+ // non-nullable metadata -- the synthesized, stale reference.
+ val join = left.join(right, LeftOuter)
+ val agg = Aggregate(Seq(a), Seq(a, b), join)
+
+ withSQLConf(SQLConf.PLAN_CHANGE_VALIDATION_NULLABILITY.key -> "true") {
+ assert(validateNullability(agg).exists(_.contains(s"b#${b.exprId.id}")))
+ }
+ }
+
+ test("Nullability check finds a narrowed reference nested in a compound
expression") {
+ val nullableChild = LocalRelation(AttributeReference("a", LongType,
nullable = true)())
+ val childAttr = nullableChild.output.head
+ val narrowedRef = AttributeReference("a", LongType, nullable =
false)(childAttr.exprId)
+ // The offending reference is buried inside `narrowedRef + 1L`, wrapped in
an Alias, to
+ // confirm the inner `_.collect` descends into compound expressions.
+ val plan = Project(Seq(Alias(narrowedRef + 1L, "x")()), nullableChild)
+
+ withSQLConf(SQLConf.PLAN_CHANGE_VALIDATION_NULLABILITY.key -> "true") {
+
assert(validateNullability(plan).exists(_.contains(s"a#${childAttr.exprId.id}")))
+ }
+ }
+
+ test("Nullability check traverses below the root to find the offending
node") {
+ val nullableChild = LocalRelation(AttributeReference("a", LongType,
nullable = true)())
+ val childAttr = nullableChild.output.head
+ val narrowedRef = AttributeReference("a", LongType, nullable =
false)(childAttr.exprId)
+ val badPlan = Project(Seq(narrowedRef), nullableChild)
+ // The root Project is a faithful pass-through (references the
already-narrowed attribute as
+ // non-nullable, matching its child's output); only the inner Project is
inconsistent.
+ val plan = badPlan.select(narrowedRef)
+
+ withSQLConf(SQLConf.PLAN_CHANGE_VALIDATION_NULLABILITY.key -> "true") {
+
assert(validateNullability(plan).exists(_.contains(s"a#${childAttr.exprId.id}")))
+ }
+ }
+
+ test("Nullability check ignores references not produced by any child") {
+ val nullableChild = LocalRelation(AttributeReference("a", LongType,
nullable = true)())
+ // `o` is declared non-nullable but its ExprId is not produced by the
child (e.g. an outer
+ // reference), so it must be ignored rather than flagged.
+ val outerRef = AttributeReference("o", LongType, nullable = false)()
+ val plan = Project(Seq(outerRef), nullableChild)
+
+ withSQLConf(SQLConf.PLAN_CHANGE_VALIDATION_NULLABILITY.key -> "true") {
+ assert(validateNullability(plan).isEmpty)
+ }
+ }
+
+ test("Nullability check treats a duplicated ExprId across children as
nullable if any is") {
+ // Two children produce the same ExprId with differing nullability; the
inline "any nullable
+ // wins" rule must treat the column as nullable, so a non-nullable
reference is flagged.
+ val sharedId = AttributeReference("k", LongType, nullable = false)().exprId
+ val nonNullChild = LocalRelation(AttributeReference("k", LongType,
nullable = false)(sharedId))
+ val nullableChild = LocalRelation(AttributeReference("k", LongType,
nullable = true)(sharedId))
+ val ref = AttributeReference("k", LongType, nullable = false)(sharedId)
+ val plan = RefTestPlan(nonNullChild, nullableChild, ref)
+
+ withSQLConf(SQLConf.PLAN_CHANGE_VALIDATION_NULLABILITY.key -> "true") {
+ assert(validateNullability(plan).exists(_.contains(s"k#${sharedId.id}")))
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
