Change default behavior of DROP CONSTRAINT to differ from IF EXISTS

Changes the behavior of DROP CONSTRAINT to throw an error by default if the constraint does not exist. The error will not be thrown if the user provides the argument IF EXISTS.

Unit test

GitOrigin-RevId: 969e7d41899ed911f596ee2c9221e6c73e2d444d

Author: karenfeng

core/src/main/resources/error/delta-error-classes.json

@@ -27,6 +27,10 @@
     "message" : [ "There is a conflict from these SET columns: %s." ],
     "sqlState" : "42000"
   },
+  "CONSTRAINT_DOES_NOT_EXIST" : {
+    "message" : [ "Cannot drop nonexistent constraint %s from table %s. To avoid throwing an error, provide the parameter IF EXISTS or set the SQL session configuration %s to %s." ],
+    "sqlState" : "42000"
+  },
   "DELTA_CANNOT_CHANGE_DATA_TYPE" : {
     "message" : [ "Cannot change data type: %s" ],
     "sqlState" : "22000"

core/src/main/scala/io/delta/sql/parser/DeltaSqlParser.scala

@@ -297,7 +297,8 @@ class DeltaSqlAstBuilder extends DeltaSqlBaseBaseVisitor[AnyRef] {
     AlterTableDropConstraint(
       createUnresolvedTable(ctx.table.identifier.asScala.map(_.getText).toSeq,
         "ALTER TABLE ... DROP CONSTRAINT"),
-      ctx.name.getText)
+      ctx.name.getText,
+      ifExists = ctx.EXISTS != null)
   }
 
   protected def typedVisit[T](ctx: ParseTree): T = {

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

@@ -34,8 +34,8 @@ case class AlterTableAddConstraint(
  * The logical plan of the ALTER TABLE ... DROP CONSTRAINT command.
  */
 case class AlterTableDropConstraint(
-    table: LogicalPlan, constraintName: String) extends AlterTableCommand {
-  override def changes: Seq[TableChange] = Seq(DropConstraint(constraintName))
+    table: LogicalPlan, constraintName: String, ifExists: Boolean) extends AlterTableCommand {
+  override def changes: Seq[TableChange] = Seq(DropConstraint(constraintName, ifExists))
 
   protected def withNewChildInternal(newChild: LogicalPlan): LogicalPlan = copy(table = newChild)
 }

core/src/main/scala/org/apache/spark/sql/delta/DeltaErrors.scala

@@ -210,6 +210,16 @@ object DeltaErrors
     new AnalysisException(s"Cannot use '$name' as the name of a CHECK constraint.")
   }
 
+  def nonexistentConstraint(constraintName: String, tableName: String): AnalysisException = {
+    new DeltaAnalysisException(
+      errorClass = "CONSTRAINT_DOES_NOT_EXIST",
+      messageParameters = Array(
+        constraintName,
+        tableName,
+        DeltaSQLConf.DELTA_ASSUMES_DROP_CONSTRAINT_IF_EXISTS.key,
+        "true"))
+  }
+
   def checkConstraintNotBoolean(name: String, expr: String): AnalysisException = {
     new AnalysisException(s"CHECK constraint '$name' ($expr) should be a boolean expression.'")
   }

core/src/main/scala/org/apache/spark/sql/delta/catalog/DeltaCatalog.scala

@@ -544,7 +544,7 @@ class DeltaCatalog extends DelegatingCatalogExtension
       case (t, constraints) if t == classOf[DropConstraint] =>
         constraints.foreach { constraint =>
           val c = constraint.asInstanceOf[DropConstraint]
-          AlterTableDropConstraintDeltaCommand(table, c.constraintName).run(spark)
+          AlterTableDropConstraintDeltaCommand(table, c.constraintName, c.ifExists).run(spark)
         }
     }
 

core/src/main/scala/org/apache/spark/sql/delta/commands/alterDeltaTableCommands.scala

@@ -629,7 +629,8 @@ case class AlterTableAddConstraintDeltaCommand(
  */
 case class AlterTableDropConstraintDeltaCommand(
     table: DeltaTableV2,
-    name: String)
+    name: String,
+    ifExists: Boolean)
   extends LeafRunnableCommand with AlterDeltaTableCommand with IgnoreCachedData {
 
   override def run(sparkSession: SparkSession): Seq[Row] = {
@@ -638,6 +639,14 @@ case class AlterTableDropConstraintDeltaCommand(
       val txn = startTransaction()
 
       val oldExprText = Constraints.getExprTextByName(name, txn.metadata, sparkSession)
+      if (oldExprText.isEmpty && !ifExists && !sparkSession.sessionState.conf.getConf(
+        DeltaSQLConf.DELTA_ASSUMES_DROP_CONSTRAINT_IF_EXISTS)) {
+        val quotedTableName = table.getTableIdentifierIfExists.map(_.quotedString)
+          .orElse(table.catalogTable.map(_.identifier.quotedString))
+          .getOrElse(table.name())
+        throw DeltaErrors.nonexistentConstraint(name, quotedTableName)
+      }
+
       val newMetadata = txn.metadata.copy(
         configuration = txn.metadata.configuration - Constraints.checkConstraintPropertyName(name))
 

core/src/main/scala/org/apache/spark/sql/delta/constraints/tableChanges.scala

@@ -32,5 +32,6 @@ case class AddConstraint(constraintName: String, expr: String) extends TableChan
  * will be thrown if the constraint doesn't exist.
  *
  * @param constraintName the name of the constraint to drop - case insensitive
+ * @param ifExists if false, throws an error if the constraint to be dropped does not exist
  */
-case class DropConstraint(constraintName: String) extends TableChange {}
+case class DropConstraint(constraintName: String, ifExists: Boolean) extends TableChange {}

core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala

@@ -284,6 +284,14 @@ trait DeltaSQLConfBase {
       .booleanConf
       .createWithDefault(false)
 
+  val DELTA_ASSUMES_DROP_CONSTRAINT_IF_EXISTS =
+    buildConf("constraints.assumesDropIfExists.enabled")
+      .doc("""If true, DROP CONSTRAINT quietly drops nonexistent constraints even without
+             |IF EXISTS.
+           """)
+      .booleanConf
+      .createWithDefault(false)
+
   val DELTA_STATE_CORRUPTION_IS_FATAL =
     buildConf("state.corruptionIsFatal")
       .internal()

core/src/test/scala/org/apache/spark/sql/delta/schema/CheckConstraintsSuite.scala

@@ -136,18 +136,37 @@ class CheckConstraintsSuite extends QueryTest
     }
   }
 
-  test("can drop constraint that doesn't exist") {
+  test("drop constraint that doesn't exist throws an exception") {
     withTestTable { table =>
-      sql(s"ALTER TABLE $table DROP CONSTRAINT IF EXISTS myConstraint")
+      intercept[AnalysisException] {
+        sql(s"ALTER TABLE $table DROP CONSTRAINT myConstraint")
+      }
+    }
+
+    withSQLConf((DeltaSQLConf.DELTA_ASSUMES_DROP_CONSTRAINT_IF_EXISTS.key, "false")) {
+      withTestTable { table =>
+        val e = intercept[AnalysisException] {
+          sql(s"ALTER TABLE $table DROP CONSTRAINT myConstraint")
+        }
+        assert(e.getErrorClass == "CONSTRAINT_DOES_NOT_EXIST")
+        errorContains(e.getMessage,
+          "nonexistent constraint myconstraint from table `default`.`checkconstraintstest`")
+        errorContains(e.getMessage,
+          "databricks.spark.delta.constraints.assumesDropIfExists.enabled to true")
+      }
     }
   }
 
-  // IF EXISTS is provided only for parallelism with existing DataSourceV2 commands that support it
-  // as a stub. It doesn't change any behavior.
   test("can drop constraint that doesn't exist with IF EXISTS") {
     withTestTable { table =>
       sql(s"ALTER TABLE $table DROP CONSTRAINT IF EXISTS myConstraint")
     }
+
+    withSQLConf((DeltaSQLConf.DELTA_ASSUMES_DROP_CONSTRAINT_IF_EXISTS.key, "true")) {
+      withTestTable { table =>
+        sql(s"ALTER TABLE $table DROP CONSTRAINT myConstraint")
+      }
+    }
   }