[SPARK-51771][SQL] Add DSv2 APIs for ALTER TABLE ADD/DROP CONSTRAINT

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

This PR adds the following DSv2 TableChange as per SPIP [doc](https://docs.google.com/document/d/1EHjB4W1LjiXxsK_G7067j9pPX0y15LUF1Z5DlUPoPIo/):
* AddConstraint
* DropConstraint

### Why are the changes needed?

For constraints support in Spark

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

No

### How was this patch tested?

New unit tests

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

No

Closes #50561 from gengliangwang/alterConstraint.

Authored-by: Gengliang Wang <gengliang@apache.org>
Signed-off-by: Gengliang Wang <gengliang@apache.org>

Author: gengliangwang

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

@@ -811,6 +811,20 @@
     },
     "sqlState" : "XX000"
   },
+  "CONSTRAINT_ALREADY_EXISTS" : {
+    "message" : [
+      "Constraint '<constraintName>' already exists. Please delete the existing constraint first.",
+      "Existing constraint:",
+      "<oldConstraint>"
+    ],
+    "sqlState" : "42710"
+  },
+  "CONSTRAINT_DOES_NOT_EXIST" : {
+    "message" : [
+      "Cannot drop nonexistent constraint <constraintName> from table <tableName>."
+    ],
+    "sqlState" : "42704"
+  },
   "CONVERSION_INVALID_INPUT" : {
     "message" : [
       "The value <str> (<fmt>) cannot be converted to <targetType> because it is malformed. Correct the value as per the syntax, or change its format. Use <suggestion> to tolerate malformed input and return NULL instead."

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java

@@ -89,4 +89,10 @@ default Map<String, String> properties() {
    * Returns the constraints for this table.
    */
   default Constraint[] constraints() { return new Constraint[0]; }
+
+  /**
+   * Returns the current table version if implementation supports versioning.
+   * If the table is not versioned, null can be returned here.
+   */
+  default String currentVersion() { return null; }
 }

sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/TableChange.java

@@ -22,6 +22,7 @@
 import javax.annotation.Nullable;
 
 import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.catalog.constraints.Constraint;
 import org.apache.spark.sql.connector.expressions.NamedReference;
 import org.apache.spark.sql.types.DataType;
 
@@ -297,6 +298,21 @@ public int hashCode() {
     }
   }
 
+  /**
+   * Create a TableChange for adding a new table constraint
+   */
+  static TableChange addConstraint(Constraint constraint, String validatedTableVersion) {
+    return new AddConstraint(constraint, validatedTableVersion);
+  }
+
+  /**
+   * Create a TableChange for dropping a table constraint
+   */
+  static TableChange dropConstraint(String name, boolean ifExists, boolean cascade) {
+    DropConstraint.Mode mode = cascade ? DropConstraint.Mode.CASCADE : DropConstraint.Mode.RESTRICT;
+    return new DropConstraint(name, ifExists, mode);
+  }
+
   /**
    * A TableChange to remove a table property.
    * <p>
@@ -787,4 +803,82 @@ public int hashCode() {
       return Arrays.hashCode(clusteringColumns);
     }
   }
+
+  /** A TableChange to alter table and add a constraint. */
+  final class AddConstraint implements TableChange {
+    private final Constraint constraint;
+    private final String validatedTableVersion;
+
+    private AddConstraint(Constraint constraint, String validatedTableVersion) {
+      this.constraint = constraint;
+      this.validatedTableVersion = validatedTableVersion;
+    }
+
+    public Constraint constraint() {
+      return constraint;
+    }
+
+    public String validatedTableVersion() {
+      return validatedTableVersion;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      AddConstraint that = (AddConstraint) o;
+      return constraint.equals(that.constraint) &&
+              Objects.equals(validatedTableVersion, that.validatedTableVersion);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(constraint, validatedTableVersion);
+    }
+  }
+
+  /** A TableChange to alter table and drop a constraint. */
+  final class DropConstraint implements TableChange {
+    private final String name;
+    private final boolean ifExists;
+    private final Mode mode;
+
+    /**
+     * Defines modes for dropping a constraint.
+     * <p>
+     * RESTRICT - Prevents dropping a constraint if it is referenced by other objects.
+     * CASCADE - Automatically drops objects that depend on the constraint.
+     */
+    public enum Mode { RESTRICT, CASCADE }
+
+    private DropConstraint(String name, boolean ifExists, Mode mode) {
+      this.name = name;
+      this.ifExists = ifExists;
+      this.mode = mode;
+    }
+
+    public String name() {
+      return name;
+    }
+
+    public boolean ifExists() {
+      return ifExists;
+    }
+
+    public Mode mode() {
+      return mode;
+    }
+
+    public boolean equals(Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      DropConstraint that = (DropConstraint) o;
+      return that.name.equals(name) && that.ifExists == ifExists && mode == that.mode;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(name, ifExists, mode);
+    }
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveTableConstraints.scala

@@ -16,12 +16,13 @@
  */
 package org.apache.spark.sql.catalyst.util
 
-import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog, TableCatalogCapability}
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog, TableCatalogCapability, TableChange}
 import org.apache.spark.sql.connector.catalog.constraints.Constraint
 import org.apache.spark.sql.errors.QueryCompilationErrors
 
 object ResolveTableConstraints {
-  // Fails if the given catalog does not support table constraint.
+  // Validates that the catalog supports create/replace table with constraints.
+  // Throws an exception if unsupported
   def validateCatalogForTableConstraint(
       constraints: Seq[Constraint],
       catalog: TableCatalog,
@@ -32,4 +33,23 @@ object ResolveTableConstraints {
         catalog, ident, "table constraint")
     }
   }
+
+  // Validates that the catalog supports ALTER TABLE ADD/DROP CONSTRAINT operations.
+  // Throws an exception if unsupported.
+  def validateCatalogForTableChange(
+      tableChanges: Seq[TableChange],
+      catalog: TableCatalog,
+      ident: Identifier): Unit = {
+    // Check if the table changes contain table constraints.
+    val hasTableConstraint = tableChanges.exists {
+      case _: TableChange.AddConstraint => true
+      case _: TableChange.DropConstraint => true
+      case _ => false
+    }
+    if (hasTableConstraint &&
+      !catalog.capabilities().contains(TableCatalogCapability.SUPPORT_TABLE_CONSTRAINT)) {
+      throw QueryCompilationErrors.unsupportedTableOperationError(
+        catalog, ident, "table constraint")
+    }
+  }
 }

sql/catalyst/src/main/scala/org/apache/spark/sql/connector/catalog/CatalogV2Util.scala

@@ -18,7 +18,7 @@
 package org.apache.spark.sql.connector.catalog
 
 import java.util
-import java.util.Collections
+import java.util.{Collections, Locale}
 
 import scala.jdk.CollectionConverters._
 
@@ -32,6 +32,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{SerdeInfo, TableSpec}
 import org.apache.spark.sql.catalyst.util.{GeneratedColumn, IdentityColumn}
 import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns._
 import org.apache.spark.sql.connector.catalog.TableChange._
+import org.apache.spark.sql.connector.catalog.constraints.Constraint
 import org.apache.spark.sql.connector.catalog.functions.UnboundFunction
 import org.apache.spark.sql.connector.expressions.{ClusterByTransform, LiteralValue, Transform}
 import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
@@ -296,6 +297,49 @@ private[sql] object CatalogV2Util {
     }
   }
 
+  /**
+   * Extracts and validates table constraints from a sequence of table changes.
+   */
+  def collectConstraintChanges(
+      table: Table,
+      changes: Seq[TableChange]): Array[Constraint] = {
+    val constraints = table.constraints()
+
+    def findExistingConstraint(name: String): Option[Constraint] = {
+      constraints.find(_.name.toLowerCase(Locale.ROOT) == name.toLowerCase(Locale.ROOT))
+    }
+
+    changes.foldLeft(constraints) { (constraints, change) =>
+      change match {
+        case add: AddConstraint =>
+          val newConstraint = add.constraint
+          val existingConstraint = findExistingConstraint(newConstraint.name)
+          if (existingConstraint.isDefined) {
+            throw new AnalysisException(
+              errorClass = "CONSTRAINT_ALREADY_EXISTS",
+              messageParameters =
+                Map("constraintName" -> existingConstraint.get.name,
+                  "oldConstraint" -> existingConstraint.get.toDDL))
+          }
+          constraints :+ newConstraint
+
+        case drop: DropConstraint =>
+          val existingConstraint = findExistingConstraint(drop.name)
+          if (existingConstraint.isEmpty && !drop.ifExists) {
+            throw new AnalysisException(
+              errorClass = "CONSTRAINT_DOES_NOT_EXIST",
+              messageParameters =
+                Map("constraintName" -> drop.name, "tableName" -> table.name()))
+          }
+          constraints.filterNot(_.name == drop.name)
+
+        case _ =>
+          // ignore non-constraint changes
+          constraints
+      }
+    }.toArray
+  }
+
   private def addField(
       schema: StructType,
       field: StructField,

sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/CatalogSuite.scala

@@ -23,6 +23,7 @@ import java.util.Collections
 import scala.jdk.CollectionConverters._
 
 import org.apache.spark.{SparkFunSuite, SparkIllegalArgumentException, SparkUnsupportedOperationException}
+import org.apache.spark.sql.AnalysisException
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.catalyst.analysis.{NamespaceAlreadyExistsException, NoSuchFunctionException, NoSuchNamespaceException, NoSuchTableException, TableAlreadyExistsException}
 import org.apache.spark.sql.catalyst.parser.CatalystSqlParser
@@ -58,6 +59,14 @@ class CatalogSuite extends SparkFunSuite {
   private val testIdentNewQuoted = testIdentNew.asMultipartIdentifier
     .map(part => quoteIdentifier(part)).mkString(".")
 
+  private val constraints: Array[Constraint] = Array(
+    Constraint.primaryKey("pk", Array(FieldReference.column("id"))).build(),
+    Constraint.check("chk").predicateSql("id > 0").build(),
+    Constraint.unique("uk", Array(FieldReference.column("data"))).build(),
+    Constraint.foreignKey("fk", Array(FieldReference.column("data")), testIdentNew,
+      Array(FieldReference.column("id"))).build()
+  )
+
   test("Catalogs can load the catalog") {
     val catalog = newCatalog()
 
@@ -174,13 +183,6 @@ class CatalogSuite extends SparkFunSuite {
     val columns = Array(
       Column.create("id", IntegerType, false),
       Column.create("data", StringType))
-    val constraints: Array[Constraint] = Array(
-      Constraint.primaryKey("pk", Array(FieldReference.column("id"))).build(),
-      Constraint.check("chk").predicateSql("id > 0").build(),
-      Constraint.unique("uk", Array(FieldReference.column("data"))).build(),
-      Constraint.foreignKey("fk", Array(FieldReference.column("data")), testIdentNew,
-        Array(FieldReference.column("id"))).build()
-    )
     val tableInfo = new TableInfo.Builder()
       .withColumns(columns)
       .withPartitions(emptyTrans)
@@ -845,6 +847,109 @@ class CatalogSuite extends SparkFunSuite {
     checkErrorTableNotFound(exc, testIdentQuoted)
   }
 
+  test("alterTable: add constraint") {
+    val catalog = newCatalog()
+
+    val tableColumns = Array(
+      Column.create("id", IntegerType, false),
+      Column.create("data", StringType))
+    val tableInfo = new TableInfo.Builder()
+      .withColumns(tableColumns)
+      .build()
+    val table = catalog.createTable(testIdent, tableInfo)
+
+    assert(table.constraints.isEmpty)
+
+    for ((constraint, index) <- constraints.zipWithIndex) {
+      val updated = catalog.alterTable(testIdent, TableChange.addConstraint(constraint, null))
+      assert(updated.constraints.length === index + 1)
+      assert(updated.constraints.apply(index) === constraint)
+    }
+  }
+
+  test("alterTable: add existing constraint should fail") {
+    val catalog = newCatalog()
+
+    val tableColumns = Array(
+      Column.create("id", IntegerType, false),
+      Column.create("data", StringType))
+    val tableInfo = new TableInfo.Builder()
+      .withColumns(tableColumns)
+      .withConstraints(constraints)
+      .build()
+    val table = catalog.createTable(testIdent, tableInfo)
+
+    assert(table.constraints.length === constraints.length)
+
+    for (constraint <- constraints) {
+      checkError(
+        exception = intercept[AnalysisException] {
+          catalog.alterTable(testIdent, TableChange.addConstraint(constraint, null))
+        },
+        condition = "CONSTRAINT_ALREADY_EXISTS",
+        parameters = Map("constraintName" -> constraint.name, "oldConstraint" -> constraint.toDDL))
+    }
+  }
+
+  test("alterTable: drop constraint") {
+    val catalog = newCatalog()
+
+    val tableColumns = Array(
+      Column.create("id", IntegerType, false),
+      Column.create("data", StringType))
+    val tableInfo = new TableInfo.Builder()
+      .withColumns(tableColumns)
+      .withConstraints(constraints)
+      .build()
+    val table = catalog.createTable(testIdent, tableInfo)
+
+    assert(table.constraints.length === constraints.length)
+
+    for ((constraint, index) <- constraints.zipWithIndex) {
+      val updated =
+        catalog.alterTable(testIdent, TableChange.dropConstraint(constraint.name(), false, false))
+      assert(updated.constraints.length === constraints.length - index -1)
+    }
+  }
+
+  test("alterTable: drop non-existing constraint") {
+    val catalog = newCatalog()
+
+    val tableColumns = Array(
+      Column.create("id", IntegerType, false),
+      Column.create("data", StringType))
+    val tableInfo = new TableInfo.Builder()
+      .withColumns(tableColumns)
+      .withConstraints(constraints)
+      .build()
+    val table = catalog.createTable(testIdent, tableInfo)
+
+    checkError(
+      exception = intercept[AnalysisException] {
+        catalog.alterTable(testIdent,
+          TableChange.dropConstraint("missing_constraint", false, false))
+      },
+      condition = "CONSTRAINT_DOES_NOT_EXIST",
+      parameters = Map("constraintName" -> "missing_constraint",
+        "tableName" -> table.name()))
+  }
+
+  test("alterTable: drop non-existing constraint if exists") {
+    val catalog = newCatalog()
+
+    val tableColumns = Array(
+      Column.create("id", IntegerType, false),
+      Column.create("data", StringType))
+    val tableInfo = new TableInfo.Builder()
+      .withColumns(tableColumns)
+      .withConstraints(constraints)
+      .build()
+    catalog.createTable(testIdent, tableInfo)
+    val updated = catalog.alterTable(testIdent,
+      TableChange.dropConstraint("missing_constraint", true, false))
+    assert(updated.constraints.length === constraints.length)
+  }
+
   test("dropTable") {
     val catalog = newCatalog()