[SPARK-51905][SQL] Disallow NOT ENFORCED CHECK constraint

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

@@ -150,8 +150,20 @@ case class CheckConstraint(
 
   override def withTableName(tableName: String): TableConstraint = copy(tableName = tableName)
 
-  override def withUserProvidedCharacteristic(c: ConstraintCharacteristic): TableConstraint =
+  override def withUserProvidedCharacteristic(c: ConstraintCharacteristic): TableConstraint = {
+    if (c.enforced.contains(false)) {
+      val origin = CurrentOrigin.get
+      throw new ParseException(
+        command = origin.sqlText,
+        start = origin,
+        errorClass = "UNSUPPORTED_CONSTRAINT_CHARACTERISTIC",
+        messageParameters = Map(
+          "characteristic" -> "NOT ENFORCED",
+          "constraintType" -> "CHECK")
+      )
+    }
     copy(userProvidedCharacteristic = c)
+  }
 }
 
 // scalastyle:off line.size.limit

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

@@ -316,4 +316,87 @@ class CheckConstraintParseSuite extends ConstraintParseSuiteBase {
     }
   }
 
+  test("NOT ENFORCED is not supported for CHECK -- table level") {
+    notEnforcedConstraintCharacteristics.foreach { case (c1, c2, _) =>
+      val characteristic = if (c2.isEmpty) {
+        c1
+      } else {
+        s"$c1 $c2"
+      }
+      val sql =
+        s"""
+           |CREATE TABLE a.b.t (a INT, b STRING, CONSTRAINT C1 CHECK (a > 0) $characteristic)
+           |""".stripMargin
+
+      val expectedContext = ExpectedContext(
+        fragment = s"CONSTRAINT C1 CHECK (a > 0) $characteristic"
+      )
+
+      checkError(
+        exception = intercept[ParseException] {
+          parsePlan(sql)
+        },
+        condition = "UNSUPPORTED_CONSTRAINT_CHARACTERISTIC",
+        parameters = Map(
+          "characteristic" -> "NOT ENFORCED",
+          "constraintType" -> "CHECK"),
+        queryContext = Array(expectedContext))
+    }
+  }
+
+  test("NOT ENFORCED is not supported for CHECK -- column level") {
+    notEnforcedConstraintCharacteristics.foreach { case (c1, c2, _) =>
+      val characteristic = if (c2.isEmpty) {
+        c1
+      } else {
+        s"$c1 $c2"
+      }
+      val sql =
+        s"""
+           |CREATE TABLE a.b.t (a INT CHECK (a > 0) $characteristic, b STRING)
+           |""".stripMargin
+
+      val expectedContext = ExpectedContext(
+        fragment = s"CHECK (a > 0) $characteristic"
+      )
+
+      checkError(
+        exception = intercept[ParseException] {
+          parsePlan(sql)
+        },
+        condition = "UNSUPPORTED_CONSTRAINT_CHARACTERISTIC",
+        parameters = Map(
+          "characteristic" -> "NOT ENFORCED",
+          "constraintType" -> "CHECK"),
+        queryContext = Array(expectedContext))
+    }
+  }
+
+  test("NOT ENFORCED is not supported for CHECK -- ALTER TABLE") {
+    notEnforcedConstraintCharacteristics.foreach { case (c1, c2, _) =>
+      val characteristic = if (c2.isEmpty) {
+        c1
+      } else {
+        s"$c1 $c2"
+      }
+      val sql =
+        s"""
+           |ALTER TABLE a.b.t ADD CONSTRAINT C1 CHECK (a > 0) $characteristic
+           |""".stripMargin
+
+      val expectedContext = ExpectedContext(
+        fragment = s"CONSTRAINT C1 CHECK (a > 0) $characteristic"
+      )
+
+      checkError(
+        exception = intercept[ParseException] {
+          parsePlan(sql)
+        },
+        condition = "UNSUPPORTED_CONSTRAINT_CHARACTERISTIC",
+        parameters = Map(
+          "characteristic" -> "NOT ENFORCED",
+          "constraintType" -> "CHECK"),
+        queryContext = Array(expectedContext))
+    }
+  }
 }

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

@@ -26,13 +26,8 @@ import org.apache.spark.sql.types.{IntegerType, StringType}
 abstract class ConstraintParseSuiteBase extends AnalysisTest with SharedSparkSession {
   protected def validConstraintCharacteristics = Seq(
     ("", "", ConstraintCharacteristic(enforced = None, rely = None)),
-    ("NOT ENFORCED", "", ConstraintCharacteristic(enforced = Some(false), rely = None)),
     ("", "RELY", ConstraintCharacteristic(enforced = None, rely = Some(true))),
-    ("", "NORELY", ConstraintCharacteristic(enforced = None, rely = Some(false))),
-    ("NOT ENFORCED", "RELY",
-      ConstraintCharacteristic(enforced = Some(false), rely = Some(true))),
-    ("NOT ENFORCED", "NORELY",
-      ConstraintCharacteristic(enforced = Some(false), rely = Some(false)))
+    ("", "NORELY", ConstraintCharacteristic(enforced = None, rely = Some(false)))
   )
 
   protected def enforcedConstraintCharacteristics = Seq(
@@ -43,6 +38,19 @@ abstract class ConstraintParseSuiteBase extends AnalysisTest with SharedSparkSes
     ("NORELY", "ENFORCED", ConstraintCharacteristic(enforced = Some(true), rely = Some(false)))
   )
 
+  protected def notEnforcedConstraintCharacteristics = Seq(
+    ("NOT ENFORCED", "RELY",
+      ConstraintCharacteristic(enforced = Some(false), rely = Some(true))),
+    ("NOT ENFORCED", "NORELY",
+      ConstraintCharacteristic(enforced = Some(false), rely = Some(false))),
+    ("RELY", "NOT ENFORCED",
+      ConstraintCharacteristic(enforced = Some(false), rely = Some(true))),
+    ("NORELY", "NOT ENFORCED",
+      ConstraintCharacteristic(enforced = Some(false), rely = Some(false))),
+    ("NOT ENFORCED", "",
+      ConstraintCharacteristic(enforced = Some(false), rely = None))
+  )
+
   protected val invalidConstraintCharacteristics = Seq(
     ("ENFORCED", "ENFORCED"),
     ("ENFORCED", "NOT ENFORCED"),

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

@@ -23,6 +23,10 @@ import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.logical.AddConstraint
 
 class ForeignKeyConstraintParseSuite extends ConstraintParseSuiteBase {
+
+  override val validConstraintCharacteristics =
+    super.validConstraintCharacteristics ++ notEnforcedConstraintCharacteristics
+
   test("Create table with foreign key - table level") {
     val sql = "CREATE TABLE t (a INT, b STRING," +
       " FOREIGN KEY (a) REFERENCES parent(id)) USING parquet"

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

@@ -24,6 +24,8 @@ import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.logical.AddConstraint
 
 class PrimaryKeyConstraintParseSuite extends ConstraintParseSuiteBase {
+  override val validConstraintCharacteristics =
+    super.validConstraintCharacteristics ++ notEnforcedConstraintCharacteristics
 
   test("Create table with primary key - table level") {
     val sql = "CREATE TABLE t (a INT, b STRING, PRIMARY KEY (a)) USING parquet"

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

@@ -23,6 +23,9 @@ import org.apache.spark.sql.catalyst.parser.ParseException
 import org.apache.spark.sql.catalyst.plans.logical.AddConstraint
 
 class UniqueConstraintParseSuite extends ConstraintParseSuiteBase {
+  override val validConstraintCharacteristics =
+    super.validConstraintCharacteristics ++ notEnforcedConstraintCharacteristics
+
   test("Create table with unnamed unique constraint") {
     Seq(
       "CREATE TABLE t (a INT, b STRING, UNIQUE (a)) USING parquet",

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

@@ -143,15 +143,11 @@ class CheckConstraintSuite extends QueryTest with CommandSuiteBase with DDLComma
     val validStatus = "UNVALIDATED"
     Seq(
       ("", s"ENFORCED $validStatus NORELY"),
-      ("NOT ENFORCED", s"NOT ENFORCED $validStatus NORELY"),
-      ("NOT ENFORCED NORELY", s"NOT ENFORCED $validStatus NORELY"),
-      ("NORELY NOT ENFORCED", s"NOT ENFORCED $validStatus NORELY"),
       ("NORELY", s"ENFORCED $validStatus NORELY"),
-      ("NOT ENFORCED RELY", s"NOT ENFORCED $validStatus RELY"),
-      ("RELY NOT ENFORCED", s"NOT ENFORCED $validStatus RELY"),
-      ("NOT ENFORCED RELY", s"NOT ENFORCED $validStatus RELY"),
-      ("RELY NOT ENFORCED", s"NOT ENFORCED $validStatus RELY"),
-      ("RELY", s"ENFORCED $validStatus RELY")
+      ("RELY", s"ENFORCED $validStatus RELY"),
+      ("ENFORCED", s"ENFORCED $validStatus NORELY"),
+      ("ENFORCED NORELY", s"ENFORCED $validStatus NORELY"),
+      ("ENFORCED RELY", s"ENFORCED $validStatus RELY")
     )
   }