[SPARK-57379][SQL] Block temporary variables in CHECK constraint expressions

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala

@@ -947,6 +947,20 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
               messageParameters = Map("checkCondition" -> a.checkConstraint.condition)
             )
 
+          // Spark currently only supports session/temporary SQL variables
+          // (TempVariableManager); every VariableReference is therefore session-scoped and
+          // invalid in a persisted CHECK constraint. If persistent SQL variables are added
+          // later, tighten this to fire only on session-scoped variables.
+          case a: AddCheckConstraint =>
+            a.checkConstraint.child.foreach {
+              case v: VariableReference =>
+                throw QueryCompilationErrors
+                  .notAllowedToCreateCheckConstraintReferencingTempVarError(
+                    Option(a.checkConstraint.userProvidedName).getOrElse(""),
+                    v.originalNameParts)
+              case _ =>
+            }
+
           case alter: AlterTableCommand =>
             checkAlterTableCommand(alter)
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableSpec.scala

@@ -95,6 +95,18 @@ object ResolveTableSpec extends Rule[LogicalPlan] {
               messageParameters = Map("checkCondition" -> check.condition)
             )
           }
+          // Spark currently only supports session/temporary SQL variables
+          // (TempVariableManager); every VariableReference is therefore session-scoped and
+          // invalid in a persisted CHECK constraint. If persistent SQL variables are added
+          // later, tighten this to fire only on session-scoped variables.
+          check.child.foreach {
+            case v: VariableReference =>
+              throw QueryCompilationErrors
+                .notAllowedToCreateCheckConstraintReferencingTempVarError(
+                  Option(check.userProvidedName).getOrElse(""),
+                  v.originalNameParts)
+            case _ =>
+          }
         case _ =>
       }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala

@@ -3578,6 +3578,18 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
         "tempObjName" -> toSQLId(varName)))
   }
 
+  def notAllowedToCreateCheckConstraintReferencingTempVarError(
+      constraintName: String,
+      varName: Seq[String]): Throwable = {
+    new AnalysisException(
+      errorClass = "INVALID_TEMP_OBJ_REFERENCE",
+      messageParameters = Map(
+        "obj" -> "CHECK CONSTRAINT",
+        "objName" -> toSQLId(constraintName),
+        "tempObj" -> "VARIABLE",
+        "tempObjName" -> toSQLId(varName)))
+  }
+
   def queryFromRawFilesIncludeCorruptRecordColumnError(): Throwable = {
     new AnalysisException(
       errorClass = "UNSUPPORTED_FEATURE.QUERY_ONLY_CORRUPT_RECORD_COLUMN",

sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala

@@ -78,6 +78,93 @@ class CheckConstraintSuite extends QueryTest with CommandSuiteBase with DDLComma
     }
   }
 
+  test("SPARK-57379: Temporary variable in check constraint -- alter table") {
+    withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+      sql(s"CREATE TABLE $t (i INT) $defaultUsing")
+      withSessionVariable("my_var") {
+        sql("DECLARE OR REPLACE VARIABLE my_var INT DEFAULT 1")
+        Seq(
+          s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (i > my_var)",
+          s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (i > CAST(my_var AS BIGINT))"
+        ).foreach { query =>
+          val error = intercept[AnalysisException] {
+            sql(query)
+          }
+          checkError(
+            exception = error,
+            condition = "INVALID_TEMP_OBJ_REFERENCE",
+            sqlState = "42K0F",
+            parameters = Map(
+              "obj" -> "CHECK CONSTRAINT",
+              "objName" -> "`c1`",
+              "tempObj" -> "VARIABLE",
+              "tempObjName" -> "`my_var`")
+          )
+        }
+      }
+    }
+  }
+
+  test("SPARK-57379: Temporary variable in check constraint -- create table") {
+    withSessionVariable("my_var") {
+      sql("DECLARE OR REPLACE VARIABLE my_var INT DEFAULT 1")
+      Seq(
+        ("CREATE TABLE t(i INT CHECK (i > my_var))", "``"),
+        ("CREATE TABLE t(i INT, CONSTRAINT c1 CHECK (i > my_var))", "`c1`"),
+        ("REPLACE TABLE t(i INT CHECK (i > my_var))", "``"),
+        ("REPLACE TABLE t(i INT, CONSTRAINT c1 CHECK (i > my_var))", "`c1`"),
+        ("CREATE TABLE t(i INT CHECK (i > CAST(my_var AS BIGINT)))", "``")
+      ).foreach { case (query, expectedObjName) =>
+        withTable("t") {
+          val error = intercept[AnalysisException] {
+            sql(query)
+          }
+          checkError(
+            exception = error,
+            condition = "INVALID_TEMP_OBJ_REFERENCE",
+            sqlState = "42K0F",
+            parameters = Map(
+              "obj" -> "CHECK CONSTRAINT",
+              "objName" -> expectedObjName,
+              "tempObj" -> "VARIABLE",
+              "tempObjName" -> "`my_var`")
+          )
+        }
+      }
+    }
+  }
+
+  test("SPARK-57379: Temporary variable in check constraint -- qualified variable name") {
+    withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+      sql(s"CREATE TABLE $t (i INT) $defaultUsing")
+      withSessionVariable("my_var") {
+        sql("DECLARE OR REPLACE VARIABLE my_var INT DEFAULT 1")
+        val error = intercept[AnalysisException] {
+          sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (i > session.my_var)")
+        }
+        checkError(
+          exception = error,
+          condition = "INVALID_TEMP_OBJ_REFERENCE",
+          sqlState = "42K0F",
+          parameters = Map(
+            "obj" -> "CHECK CONSTRAINT",
+            "objName" -> "`c1`",
+            "tempObj" -> "VARIABLE",
+            "tempObjName" -> "`session`.`my_var`")
+        )
+      }
+    }
+  }
+
+  test("SPARK-57379: Temporary variable in check constraint -- happy path without variable") {
+    withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+      sql(s"CREATE TABLE $t (i INT, CONSTRAINT c1 CHECK (i > 0)) $defaultUsing")
+      sql(s"ALTER TABLE $t ADD CONSTRAINT c2 CHECK (i > -1)")
+    }
+  }
+
+  // CTAS/RTAS cannot carry CHECK constraints -- the parser rejects them -- so no coverage needed.
+
   test("Expression referring a column of another table -- alter table") {
     withNamespaceAndTable("ns", "tbl_1", nonPartitionCatalog) { t1 =>
       withNamespaceAndTable("ns", "tbl_2", nonPartitionCatalog) { t2 =>