[SPARK-XXXXX][SQL] Guard against unresolved attributes in constraint inference

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

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)
+  }
 }

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)
+  }
 }