refactor check constraint

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

@@ -25,17 +25,20 @@ public interface Constraint {
     boolean rely(); // indicates whether the constraint is believed to be true
     boolean enforced(); // indicates whether the constraint must be enforced
 
-    static Constraint check(String name, Predicate predicate) {
-      return new Check(name, predicate);
+    static Constraint check(String name, String sql, Predicate predicate) {
+      return new Check(name, sql, predicate);
     }
 
     final class Check implements Constraint {
         private final String name;
+        private final String sql;
         private final Predicate predicate;
-        private Check(String name, Predicate predicate) {
+        private Check(String name, String sql, Predicate predicate) {
           this.name = name;
+          this.sql = sql;
           this.predicate = predicate;
         }
+
         @Override public String name() {
             return name;
         }
@@ -46,7 +49,7 @@ private Check(String name, Predicate predicate) {
 
         @Override
         public String toDDL() {
-            return "CHECK (" + predicate.toString() + ")";
+            return "CHECK (" + sql + ")";
         }
 
         @Override public boolean rely() {
@@ -56,5 +59,9 @@ public String toDDL() {
         @Override public boolean enforced() {
             return true;
         }
+
+        public String sql() { return sql; }
+
+        public Predicate predicate() { return predicate; }
     }
 }

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

@@ -1191,7 +1191,7 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
           case _ =>
         }
 
-      case addConstraint @ AddCheckConstraint(table: ResolvedTable, _, constraintExpr) =>
+      case addConstraint @ AddCheckConstraint(table: ResolvedTable, _, _, constraintExpr) =>
         if (!constraintExpr.resolved) {
           constraintExpr.failAnalysis(
             errorClass = "INVALID_CHECK_CONSTRAINT.UNRESOLVED",
@@ -1205,13 +1205,6 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
             messageParameters = Map.empty
           )
         }
-
-        if (addConstraint.predicate.isEmpty) {
-          constraintExpr.failAnalysis(
-            errorClass = "INVALID_CHECK_CONSTRAINT.INVALID_V2_PREDICATE",
-            messageParameters = Map.empty
-          )
-        }
       case _ =>
     }
   }

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

@@ -4642,10 +4642,10 @@ class AstBuilder extends DataTypeAstBuilder
       }
     }
 
-    val constraints = ctx.constraintSpec().asScala.map(visitConstraintSpec).toSeq
+    // val constraints = ctx.constraintSpec().asScala.map(visitConstraintSpec).toSeq
 
     (partTransforms, partCols, bucketSpec, cleanedProperties, cleanedOptions, newLocation, comment,
-      collation, serdeInfo, clusterBySpec, constraints)
+      collation, serdeInfo, clusterBySpec, Seq.empty)
   }
 
   protected def getSerdeInfo(
@@ -5241,14 +5241,6 @@ class AstBuilder extends DataTypeAstBuilder
       AlterTableCollation(table, visitCollationSpec(ctx.collationSpec()))
     }
 
-  override def visitConstraintSpec(ctx: ConstraintSpecContext): Expression = {
-    ctx.constraintExpression() match {
-      case c: CheckConstraintContext => expression(c.booleanExpression())
-      case other =>
-        throw QueryParsingErrors.constraintNotSupportedError(ctx, other.getText)
-    }
-  }
-
   /**
    * Parse a [[AddCheckConstraint]] command.
    *
@@ -5261,8 +5253,16 @@ class AstBuilder extends DataTypeAstBuilder
     withOrigin(ctx) {
       val table = createUnresolvedTable(
         ctx.identifierReference, "ALTER TABLE ... ADD CONSTRAINT")
-      val constraintExpression = visitConstraintSpec(ctx.constraintSpec())
-      AddCheckConstraint(table, ctx.constraintSpec().constraintName.getText, constraintExpression)
+      ctx.constraintSpec.constraintExpression() match {
+        case c: CheckConstraintContext =>
+          AddCheckConstraint(
+            table,
+            ctx.constraintSpec().constraintName.getText,
+            c.booleanExpression().getText,
+            expression(c.booleanExpression()))
+        case other =>
+          throw QueryParsingErrors.constraintNotSupportedError(ctx, other.getText)
+      }
     }
 
 

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

@@ -295,12 +295,13 @@ case class AlterTableCollation(
 case class AddCheckConstraint(
     table: LogicalPlan,
     name: String,
+    constraintText: String,
     constraintExpr: Expression) extends AlterTableCommand {
 
   lazy val predicate = new V2ExpressionBuilder(constraintExpr, true).buildPredicate()
 
   override def changes: Seq[TableChange] = {
-    val constraint = Constraint.check(name, predicate.get)
+    val constraint = Constraint.check(name, constraintText, predicate.orNull)
     Seq(TableChange.addCheckConstraint(constraint, constraint.enforced()))
   }
 

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

@@ -1521,7 +1521,8 @@ case class UnresolvedTableSpec(
     collation: Option[String],
     serde: Option[SerdeInfo],
     external: Boolean,
-    constraints: Seq[Expression]) extends UnaryExpression with Unevaluable with TableSpecBase {
+    constraints: Seq[Expression] = Seq.empty)
+  extends UnaryExpression with Unevaluable with TableSpecBase {
 
   override def dataType: DataType =
     throw new SparkUnsupportedOperationException("_LEGACY_ERROR_TEMP_3113")
@@ -1568,7 +1569,7 @@ case class TableSpec(
     collation: Option[String],
     serde: Option[SerdeInfo],
     external: Boolean,
-    constraints: Seq[Constraint]) extends TableSpecBase {
+    constraints: Seq[Constraint] = Seq.empty) extends TableSpecBase {
   def withNewLocation(newLocation: Option[String]): TableSpec = {
     TableSpec(properties, provider, options, newLocation,
       comment, collation, serde, external, constraints)

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

@@ -36,6 +36,7 @@ class AlterTableAddConstraintParseSuite extends AnalysisTest with SharedSparkSes
         Seq("a", "b", "c"),
         "ALTER TABLE ... ADD CONSTRAINT"),
       "c1",
+      "d > 0",
       GreaterThan(UnresolvedAttribute("d"), Literal(0)))
     comparePlans(parsed, expected)
   }

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

@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.command.v2
 
 import org.apache.spark.sql.{AnalysisException, QueryTest}
 import org.apache.spark.sql.connector.catalog.Constraint.Check
+import org.apache.spark.sql.connector.catalog.Table
 import org.apache.spark.sql.execution.command.DDLCommandTestUtils
 
 class CheckConstraintSuite extends QueryTest with CommandSuiteBase with DDLCommandTestUtils {
@@ -73,27 +74,26 @@ class CheckConstraintSuite extends QueryTest with CommandSuiteBase with DDLComma
     }
   }
 
-  test("Can't convert expression to V2 predicate") {
-    withTable("t") {
-      sql("create table t(i string) using parquet")
-      val query =
-        """
-          |ALTER TABLE t ADD CONSTRAINT c1 CHECK (from_json(i, 'a INT').a > 1)
-          |""".stripMargin
-      val error = intercept[AnalysisException] {
-        sql(query)
-      }
-      checkError(
-        exception = error,
-        condition = "INVALID_CHECK_CONSTRAINT.INVALID_V2_PREDICATE",
-        sqlState = "42621",
-        parameters = Map.empty,
-        context = ExpectedContext(
-          fragment = "from_json(i, 'a INT').a > 1",
-          start = 40,
-          stop = 66
-        )
-      )
+  private def getCheckConstraint(table: Table): Check = {
+    assert(table.constraints.length == 1)
+    assert(table.constraints.head.isInstanceOf[Check])
+    table.constraints.head.asInstanceOf[Check]
+    val constraint = table.constraints.head.asInstanceOf[Check]
+    assert(constraint.rely())
+    assert(constraint.enforced())
+    constraint
+  }
+
+  test("Predicate should be null if it can't be converted to V2 predicate") {
+    withNamespaceAndTable("ns", "tbl", catalog) { t =>
+      sql(s"CREATE TABLE $t (id bigint, j string) $defaultUsing")
+      sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (from_json(j, 'a INT').a > 1)")
+      val table = loadTable(catalog, "ns", "tbl")
+      val constraint = getCheckConstraint(table)
+      assert(constraint.name() == "c1")
+      assert(constraint.toDDL == "CHECK (from_json(j,'a INT').a>1)")
+      assert(constraint.sql() == "from_json(j,'a INT').a>1")
+      assert(constraint.predicate() == null)
     }
   }
 
@@ -104,13 +104,9 @@ class CheckConstraintSuite extends QueryTest with CommandSuiteBase with DDLComma
 
       sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (id > 0)")
       val table = loadTable(catalog, "ns", "tbl")
-      assert(table.constraints.length == 1)
-      assert(table.constraints.head.isInstanceOf[Check])
-      val constraint = table.constraints.head.asInstanceOf[Check]
+      val constraint = getCheckConstraint(table)
       assert(constraint.name() == "c1")
-      assert(constraint.rely())
-      assert(constraint.enforced())
-      assert(constraint.toDDL == "CHECK (id > 0)")
+      assert(constraint.toDDL == "CHECK (id>0)")
     }
   }
 
@@ -123,13 +119,13 @@ class CheckConstraintSuite extends QueryTest with CommandSuiteBase with DDLComma
       // Constraint names are case-insensitive
       Seq("abc", "ABC").foreach { name =>
         val error = intercept[AnalysisException] {
-          sql(s"ALTER TABLE $t ADD CONSTRAINT $name CHECK (id > 0)")
+          sql(s"ALTER TABLE $t ADD CONSTRAINT $name CHECK (id>0)")
         }
         checkError(
           exception = error,
           condition = "CONSTRAINT_ALREADY_EXISTS",
           sqlState = "42710",
-          parameters = Map("constraintName" -> "abc", "oldConstraint" -> "CHECK (id > 0)")
+          parameters = Map("constraintName" -> "abc", "oldConstraint" -> "CHECK (id>0)")
         )
       }
     }