From a3f50eab4cfd5ad890251b00632fbc2b320741ae Mon Sep 17 00:00:00 2001
From: Mihailo Timotic <mihailo.timotic@databricks.com>
Date: Fri, 20 Mar 2026 13:22:28 +0000
Subject: [PATCH] fix

---
 .../plans/logical/QueryPlanConstraints.scala  |  2 +-
 .../InferFiltersFromConstraintsSuite.scala    | 27 +++++++++++++++++++
 .../plans/ConstraintPropagationSuite.scala    | 11 ++++++++
 3 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
index 5769f006ccbc3..252a1e5496038 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/QueryPlanConstraints.scala
@@ -100,7 +100,7 @@ trait ConstraintHelper {
 
     // Second, we infer additional constraints from non-nullable attributes that are part of the
     // operator's output
-    val nonNullableAttributes = output.filterNot(_.nullable)
+    val nonNullableAttributes = output.filter(a => a.resolved && !a.nullable)
     isNotNullConstraints ++= nonNullableAttributes.map(IsNotNull)
 
     isNotNullConstraints -- constraints
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala
index d8d8a2b333bcd..0690110d5bf7c 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/InferFiltersFromConstraintsSuite.scala
@@ -17,6 +17,7 @@
 
 package org.apache.spark.sql.catalyst.optimizer
 
+import org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
 import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.dsl.plans._
 import org.apache.spark.sql.catalyst.expressions._
@@ -399,4 +400,30 @@ class InferFiltersFromConstraintsSuite extends PlanTest {
     comparePlans(optimizedQuery, correctAnswer)
     comparePlans(InferFiltersFromConstraints(optimizedQuery), correctAnswer)
   }
+
+  test("SPARK-XXXXX: should not crash when plan output contains unresolved attributes") {
+    // Reproduces an issue where a REPLACE TABLE AS SELECT query submitted via Spark Connect
+    // hit UnresolvedException during optimizer constraint inference. The stack trace:
+    //   UnresolvedAttribute.nullable -> constructIsNotNullConstraints -> constraints ->
+    //   InferFiltersFromConstraints.inferFilters
+    //
+    // Root cause: a plan node's output contained UnresolvedAttribute (produced by
+    // Alias.toAttribute when the Alias is unresolved), and constructIsNotNullConstraints
+    // called output.filterNot(_.nullable) which threw on the UnresolvedAttribute.
+    val resolvedAttr = testRelation.output.head
+    val unresolvedAlias = Alias(UnresolvedAttribute("unknown_col"), "x")()
+    val projectWithUnresolved = Project(
+      Seq(resolvedAttr, unresolvedAlias),
+      testRelation
+    )
+    val filterPlan = Filter(
+      GreaterThan(resolvedAttr, Literal(5)),
+      projectWithUnresolved
+    )
+
+    // Without the fix, this throws:
+    //   org.apache.spark.sql.catalyst.analysis.UnresolvedException: nullable
+    val result = InferFiltersFromConstraints(filterPlan)
+    assert(result != null)
+  }
 }
diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
index fb5ab31350b33..3f6f39f857089 100644
--- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
+++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/plans/ConstraintPropagationSuite.scala
@@ -428,4 +428,15 @@ class ConstraintPropagationSuite extends SparkFunSuite with PlanTest {
       assert(aliasedRelation.analyze.constraints.isEmpty)
     }
   }
+
+  test("constructIsNotNullConstraints should not crash on unresolved attributes") {
+    val helper = new ConstraintHelper {}
+    val resolved = AttributeReference("a", IntegerType, nullable = false)()
+    val unresolved = UnresolvedAttribute("b")
+    val output = Seq(resolved, unresolved)
+
+    val result = helper.constructIsNotNullConstraints(ExpressionSet(), output)
+    assert(result.contains(IsNotNull(resolved)))
+    assert(result.size === 1)
+  }
 }