[SPARK-51903][SQL] Validate data on adding a CHECK constraint

### What changes were proposed in this pull request?

This PR adds support for enforcing CHECK constraints when executing the `ALTER TABLE ... ADD CONSTRAINT` statement on V2 tables. Specifically, before adding a new CHECK constraint, Spark will validate the constraint against existing table data to ensure data integrity and compliance with the constraint definition.

### Why are the changes needed?

CHECK constraint enforcement in the `ALTER TABLE ... ADD CONSTRAINT` statement is part of the ANSI SQL standard. Supporting this feature ensures Spark's compliance with ANSI SQL and enhances data integrity by preventing invalid constraints from being added to existing tables.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

New UT
### Was this patch authored or co-authored using generative AI tooling?

No

Closes #50839 from gengliangwang/alterAddNewCheck.

Authored-by: Gengliang Wang <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>

Author: gengliangwang

common/utils/src/main/resources/error/error-conditions.json

@@ -4137,6 +4137,12 @@
     ],
     "sqlState" : "07501"
   },
+  "NEW_CHECK_CONSTRAINT_VIOLATION" : {
+    "message" : [
+      "The new check constraint (<expression>) cannot be added because it would be violated by existing data in table <tableName>. Please ensure all existing rows satisfy the constraint before adding it."
+    ],
+    "sqlState" : "23512"
+  },
   "NONEXISTENT_FIELD_NAME_IN_LIST" : {
     "message" : [
       "Field(s) <nonExistFields> do(es) not exist. Available fields: <fieldNames>"

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

@@ -736,6 +736,12 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
           case write: V2WriteCommand if write.resolved =>
             write.query.schema.foreach(f => TypeUtils.failWithIntervalType(f.dataType))
 
+          case a: AddCheckConstraint if !a.checkConstraint.deterministic =>
+            a.checkConstraint.child.failAnalysis(
+              errorClass = "NON_DETERMINISTIC_CHECK_CONSTRAINT",
+              messageParameters = Map("checkCondition" -> a.checkConstraint.condition)
+            )
+
           case alter: AlterTableCommand =>
             checkAlterTableCommand(alter)
 
@@ -1037,12 +1043,6 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
       case RenameColumn(table: ResolvedTable, col: ResolvedFieldName, newName) =>
         checkColumnNotExists("rename", col.path :+ newName, table.schema)
 
-      case AddConstraint(_: ResolvedTable, check: CheckConstraint) if !check.deterministic =>
-        check.child.failAnalysis(
-          errorClass = "NON_DETERMINISTIC_CHECK_CONSTRAINT",
-          messageParameters = Map("checkCondition" -> check.condition)
-        )
-
       case AlterColumns(table: ResolvedTable, specs) =>
         val groupedColumns = specs.groupBy(_.column.name)
         groupedColumns.collect {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala

@@ -129,15 +129,13 @@ case class CheckConstraint(
     val predicate = new V2ExpressionBuilder(child, true).buildPredicate().orNull
     val enforced = userProvidedCharacteristic.enforced.getOrElse(true)
     val rely = userProvidedCharacteristic.rely.getOrElse(false)
-    // TODO(SPARK-51903): Change the status to VALIDATED when we support validation on ALTER TABLE
-    val validateStatus = Constraint.ValidationStatus.UNVALIDATED
     Constraint
       .check(name)
       .predicateSql(condition)
       .predicate(predicate)
       .rely(rely)
       .enforced(enforced)
-      .validationStatus(validateStatus)
+      .validationStatus(Constraint.ValidationStatus.VALID)
       .build()
   }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -5490,9 +5490,16 @@ class AstBuilder extends DataTypeAstBuilder
     withOrigin(ctx) {
       val tableConstraint = visitTableConstraintDefinition(ctx.tableConstraintDefinition())
       withIdentClause(ctx.identifierReference, identifiers => {
-        val table = UnresolvedTable(identifiers, "ALTER TABLE ... ADD CONSTRAINT")
         val namedConstraint = tableConstraint.withTableName(identifiers.last)
-        AddConstraint(table, namedConstraint)
+        namedConstraint match {
+          case c: CheckConstraint =>
+            val relation = createUnresolvedRelation(ctx.identifierReference)
+            val child = Filter(Not(c.child), relation)
+            AddCheckConstraint(child, c)
+          case _ =>
+            val child = UnresolvedTable(identifiers, "ALTER TABLE ... ADD CONSTRAINT")
+            AddConstraint(child, namedConstraint)
+        }
       })
     }
 

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2AlterTableCommands.scala

@@ -17,10 +17,10 @@
 
 package org.apache.spark.sql.catalyst.plans.logical
 
-import org.apache.spark.sql.catalyst.analysis.{FieldName, FieldPosition, ResolvedFieldName, ResolvedTable, UnresolvedException}
+import org.apache.spark.sql.catalyst.analysis.{FieldName, FieldPosition, ResolvedFieldName, UnresolvedException}
 import org.apache.spark.sql.catalyst.catalog.CatalogTypes.TablePartitionSpec
 import org.apache.spark.sql.catalyst.catalog.ClusterBySpec
-import org.apache.spark.sql.catalyst.expressions.{Expression, TableConstraint, Unevaluable}
+import org.apache.spark.sql.catalyst.expressions.{CheckConstraint, Expression, TableConstraint, Unevaluable}
 import org.apache.spark.sql.catalyst.util.{ResolveDefaultColumns, TypeUtils}
 import org.apache.spark.sql.connector.catalog.{TableCatalog, TableChange}
 import org.apache.spark.sql.errors.QueryCompilationErrors
@@ -290,23 +290,34 @@ case class AlterTableCollation(
 }
 
 /**
- * The logical plan of the ALTER TABLE ... ADD CONSTRAINT command.
+ * The logical plan of the ALTER TABLE ... ADD CONSTRAINT command for Primary Key, Foreign Key,
+ * and Unique constraints.
  */
 case class AddConstraint(
     table: LogicalPlan,
     tableConstraint: TableConstraint) extends AlterTableCommand {
+
   override def changes: Seq[TableChange] = {
     val constraint = tableConstraint.toV2Constraint
-    val validatedTableVersion = table match {
-      case t: ResolvedTable if constraint.enforced() =>
-        t.table.currentVersion()
-      case _ =>
-        null
-    }
-    Seq(TableChange.addConstraint(constraint, validatedTableVersion))
+    // The table version is null because the constraint is not enforced.
+    Seq(TableChange.addConstraint(constraint, null))
   }
 
-  protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = copy(table = newChild)
+  override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan =
+    copy(table = newChild)
+}
+
+/**
+ * The logical plan of the ALTER TABLE ... ADD CONSTRAINT command for Check constraints.
+ * It doesn't extend [[AlterTableCommand]] because its child is a filtered table scan rather than
+ * a table reference.
+ */
+case class AddCheckConstraint(
+    child: LogicalPlan,
+    checkConstraint: CheckConstraint) extends UnaryCommand {
+
+  override protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan =
+    copy(child = newChild)
 }
 
 /**

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

@@ -3026,4 +3026,14 @@ private[sql] object QueryExecutionErrors extends QueryErrorsBase with ExecutionE
       values: java.util.List[Any]): SparkRuntimeException = {
     checkViolation(constraintName, sqlStr, columns.asScala.zip(values.asScala).toMap)
   }
+
+  def newCheckViolation(sqlStr: String, tableName: String): SparkRuntimeException = {
+    new SparkRuntimeException(
+      errorClass = "NEW_CHECK_CONSTRAINT_VIOLATION",
+      messageParameters = Map(
+        "expression" -> sqlStr,
+        "tableName" -> tableName
+      )
+    )
+  }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/AlterTableExec.scala

@@ -21,7 +21,9 @@ import org.apache.spark.SparkThrowable
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.expressions.Attribute
 import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog, TableChange}
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits.{IdentifierHelper, MultipartIdentifierHelper}
 import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.{SparkPlan, UnaryExecNode}
 
 /**
  * Physical plan node for altering a table.
@@ -44,3 +46,35 @@ case class AlterTableExec(
     Seq.empty
   }
 }
+
+/**
+ * Physical plan node for adding a check constraint with validation.
+ */
+case class AddCheckConstraintExec(
+    catalog: TableCatalog,
+    ident: Identifier,
+    change: TableChange,
+    condition: String,
+    child: SparkPlan) extends V2CommandExec with UnaryExecNode {
+
+  override def output: Seq[Attribute] = Seq.empty
+
+  override protected def run(): Seq[InternalRow] = {
+    if (child.executeTake(1).nonEmpty) {
+      throw QueryExecutionErrors.newCheckViolation(
+        condition, ident.toQualifiedNameParts(catalog).quoted)
+    }
+    try {
+     catalog.alterTable(ident, change)
+    } catch {
+      case e: IllegalArgumentException if !e.isInstanceOf[SparkThrowable] =>
+        throw QueryExecutionErrors.unsupportedTableChangeError(e)
+    }
+
+    Seq.empty
+  }
+
+  override protected def withNewChildInternal(newChild: SparkPlan): SparkPlan = {
+    copy(child = newChild)
+  }
+}

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2Strategy.scala

@@ -35,6 +35,7 @@ import org.apache.spark.sql.catalyst.plans.logical._
 import org.apache.spark.sql.catalyst.util.{toPrettySQL, GeneratedColumn, IdentityColumn, ResolveDefaultColumns, ResolveTableConstraints, V2ExpressionBuilder}
 import org.apache.spark.sql.classic.SparkSession
 import org.apache.spark.sql.connector.catalog.{Identifier, StagingTableCatalog, SupportsDeleteV2, SupportsNamespaces, SupportsPartitionManagement, SupportsWrite, Table, TableCapability, TableCatalog, TruncatableTable}
+import org.apache.spark.sql.connector.catalog.TableChange
 import org.apache.spark.sql.connector.catalog.index.SupportsIndex
 import org.apache.spark.sql.connector.expressions.{FieldReference, LiteralValue}
 import org.apache.spark.sql.connector.expressions.filter.{And => V2And, Not => V2Not, Or => V2Or, Predicate}
@@ -534,6 +535,18 @@ class DataSourceV2Strategy(session: SparkSession) extends Strategy with Predicat
       }
       UncacheTableExec(r.table, cascade = !isTempView(r.table)) :: Nil
 
+    case a @ AddCheckConstraint(PhysicalOperation(_, _, d: DataSourceV2ScanRelation), check) =>
+      assert(d.relation.catalog.isDefined, "Catalog should be defined after analysis")
+      assert(d.relation.identifier.isDefined, "Identifier should be defined after analysis")
+      val catalog = d.relation.catalog.get.asTableCatalog
+      val ident = d.relation.identifier.get
+      val condition = a.checkConstraint.condition
+      val change = TableChange.addConstraint(
+        check.toV2Constraint,
+        d.relation.table.currentVersion)
+      ResolveTableConstraints.validateCatalogForTableChange(Seq(change), catalog, ident)
+      AddCheckConstraintExec(catalog, ident, change, condition, planLater(a.child)) :: Nil
+
     case a: AlterTableCommand if a.table.resolved =>
       val table = a.table.asInstanceOf[ResolvedTable]
       ResolveTableConstraints.validateCatalogForTableChange(

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

@@ -17,11 +17,11 @@
 
 package org.apache.spark.sql.execution.command
 
-import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedTable}
+import org.apache.spark.sql.catalyst.analysis.{UnresolvedAttribute, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.expressions._
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser.parsePlan
 import org.apache.spark.sql.catalyst.parser.ParseException
-import org.apache.spark.sql.catalyst.plans.logical.{AddConstraint, ColumnDefinition, CreateTable, ReplaceTable, UnresolvedTableSpec}
+import org.apache.spark.sql.catalyst.plans.logical.{AddCheckConstraint, ColumnDefinition, CreateTable, Filter, ReplaceTable, UnresolvedTableSpec}
 import org.apache.spark.sql.types.StringType
 
 class CheckConstraintParseSuite extends ConstraintParseSuiteBase {
@@ -187,11 +187,13 @@ class CheckConstraintParseSuite extends ConstraintParseSuiteBase {
         |ALTER TABLE a.b.t ADD CONSTRAINT c1 CHECK (a > 0)
         |""".stripMargin
     val parsed = parsePlan(sql)
-    val expected = AddConstraint(
-      UnresolvedTable(
-        Seq("a", "b", "t"),
-        "ALTER TABLE ... ADD CONSTRAINT"),
-      constraint1)
+    val expected = AddCheckConstraint(
+      Filter(
+        Not(GreaterThan(UnresolvedAttribute("a"), Literal(0))),
+        UnresolvedRelation(Seq("a", "b", "t"))
+      ),
+      constraint1
+    )
     comparePlans(parsed, expected)
   }
 
@@ -224,17 +226,20 @@ class CheckConstraintParseSuite extends ConstraintParseSuiteBase {
            |ALTER TABLE a.b.t ADD CONSTRAINT c1 CHECK (d > 0) $enforcedStr $relyStr
            |""".stripMargin
       val parsed = parsePlan(sql)
-      val expected = AddConstraint(
-        UnresolvedTable(
-          Seq("a", "b", "t"),
-          "ALTER TABLE ... ADD CONSTRAINT"),
-        CheckConstraint(
-          child = GreaterThan(UnresolvedAttribute("d"), Literal(0)),
-          condition = "d > 0",
-          userProvidedName = "c1",
-          tableName = "t",
-          userProvidedCharacteristic = characteristic
-        ))
+      val expectedConstraint = CheckConstraint(
+        child = GreaterThan(UnresolvedAttribute("d"), Literal(0)),
+        condition = "d > 0",
+        userProvidedName = "c1",
+        tableName = "t",
+        userProvidedCharacteristic = characteristic
+      )
+      val expected = AddCheckConstraint(
+        Filter(
+          Not(GreaterThan(UnresolvedAttribute("d"), Literal(0))),
+          UnresolvedRelation(Seq("a", "b", "t"))
+        ),
+        expectedConstraint
+      )
       comparePlans(parsed, expected)
     }
   }
@@ -305,11 +310,12 @@ class CheckConstraintParseSuite extends ConstraintParseSuiteBase {
         |""".stripMargin
     val plan = parsePlan(sql)
     plan match {
-      case a: AddConstraint =>
-        val table = a.table.asInstanceOf[UnresolvedTable]
-        assert(table.multipartIdentifier == Seq("a", "b", "t"))
-        assert(a.tableConstraint == unnamedConstraint)
-        assert(a.tableConstraint.name.matches("t_chk_[0-9a-f]{7}"))
+      case a: AddCheckConstraint =>
+        comparePlans(a.child, Filter(
+          Not(GreaterThan(UnresolvedAttribute("a"), Literal(0))),
+          UnresolvedRelation(Seq("a", "b", "t"))))
+        assert(a.checkConstraint == unnamedConstraint)
+        assert(a.checkConstraint.name.matches("t_chk_[0-9a-f]{7}"))
 
       case other =>
         fail(s"Expected AddConstraint, but got: $other")