[SPARK-52143][TESTS] Add test for table constraints with current_* functions

gengliangwang · gengliangwang · commit 80da53ed1dfd · 2025-05-23T11:58:37.000-07:00
### What changes were proposed in this pull request? Add test for table constraints with `current_*` functions ### Why are the changes needed? Improve test coverage ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? UT ### Was this patch authored or co-authored using generative AI tooling? No Closes #50993 from gengliangwang/testCurrent. Authored-by: Gengliang Wang <gengliang@apache.org> Signed-off-by: Gengliang Wang <gengliang@apache.org>
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala
@@ -371,6 +371,248 @@ class CheckConstraintSuite extends QueryTest with CommandSuiteBase with DDLComma
     }
   }
 
+  test("Check constraint with current_timestamp function") {
+    withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+      // Create table with a constraint using current_timestamp
+      sql(s"CREATE TABLE $t (id INT, creation_time TIMESTAMP, " +
+        s"CONSTRAINT valid_time CHECK (creation_time <= current_timestamp())) $defaultUsing")
+
+      // Insert valid data (current or past timestamp)
+      sql(s"INSERT INTO $t VALUES (1, current_timestamp()), (2, TIMESTAMP '2020-01-01 00:00:00')")
+      checkAnswer(spark.table(t).select("id"), Seq(Row(1), Row(2)))
+
+      // Insert valid data with null timestamp
+      sql(s"INSERT INTO $t VALUES (3, null)")
+      checkAnswer(spark.table(t).select("id"), Seq(Row(1), Row(2), Row(3)))
+
+      // Future timestamp should fail validation
+      val tomorrow = "current_timestamp() + INTERVAL 1 DAY"
+      val error = intercept[SparkRuntimeException] {
+        sql(s"INSERT INTO $t VALUES (4, $tomorrow)")
+      }
+      assert(error.getMessage.contains("CHECK_CONSTRAINT_VIOLATION"))
+      assert(error.getMessageParameters.get("constraintName") == "valid_time")
+      assert(error.getMessageParameters.get("expression") == "creation_time <= current_timestamp()")
+    }
+  }
+
+  test("Check constraint with current_timestamp function - update operation") {
+    withNamespaceAndTable("ns", "tbl", rowLevelOPCatalog) { t =>
+      // Create table with a constraint using current_timestamp
+      sql(s"CREATE TABLE $t (id INT, creation_time TIMESTAMP, " +
+        s"CONSTRAINT valid_time CHECK (creation_time <= current_timestamp())) $defaultUsing")
+
+      // Insert initial data
+      sql(s"INSERT INTO $t VALUES (1, current_timestamp()), (2, TIMESTAMP '2020-01-01 00:00:00')")
+
+      // Valid update with current or past timestamp
+      sql(s"UPDATE $t SET creation_time = TIMESTAMP '2021-01-01 00:00:00' WHERE id = 1")
+      checkAnswer(spark.table(t).select("id")
+        .where("creation_time = TIMESTAMP '2021-01-01 00:00:00'"),
+        Seq(Row(1)))
+
+      // Valid update with null timestamp
+      sql(s"UPDATE $t SET creation_time = null WHERE id = 2")
+      checkAnswer(sql(s"SELECT * FROM $t WHERE id = 2"), Seq(Row(2, null)))
+
+      // Future timestamp should fail validation
+      val tomorrow = "current_timestamp() + INTERVAL 1 DAY"
+      val error = intercept[SparkRuntimeException] {
+        sql(s"UPDATE $t SET creation_time = $tomorrow WHERE id = 1")
+      }
+      assert(error.getMessage.contains("CHECK_CONSTRAINT_VIOLATION"))
+      assert(error.getMessageParameters.get("constraintName") == "valid_time")
+      assert(error.getMessageParameters.get("expression") == "creation_time <= current_timestamp()")
+    }
+  }
+
+  test("Check constraint with current_timestamp function - merge operation") {
+    withNamespaceAndTable("ns", "tbl", rowLevelOPCatalog) { target =>
+      withNamespaceAndTable("ns", "source", rowLevelOPCatalog) { source =>
+        // Create target table with constraint using current_timestamp
+        sql(s"CREATE TABLE $target (id INT, creation_time TIMESTAMP, " +
+          s"CONSTRAINT valid_time CHECK (creation_time <= current_timestamp())) $defaultUsing")
+
+        // Create source table without constraints
+        sql(s"CREATE TABLE $source (id INT, creation_time TIMESTAMP) $defaultUsing")
+
+        // Insert initial data
+        sql(s"INSERT INTO $target VALUES (1, TIMESTAMP '2020-01-01 00:00:00')")
+        sql(s"INSERT INTO $source VALUES " +
+          s"(2, TIMESTAMP '2021-01-01 00:00:00'), " +
+          s"(3, current_timestamp()), " +
+          s"(4, null)")
+
+        // Valid merge with past timestamps or null
+        sql(
+          s"""
+             |MERGE INTO $target t
+             |USING (SELECT * FROM $source WHERE id IN (2, 4)) s
+             |ON t.id = s.id
+             |WHEN MATCHED THEN UPDATE SET creation_time = s.creation_time
+             |WHEN NOT MATCHED THEN INSERT (id, creation_time) VALUES (s.id, s.creation_time)
+             |""".stripMargin)
+
+        checkAnswer(spark.table(target).orderBy("id"),
+          Seq(Row(1, java.sql.Timestamp.valueOf("2020-01-01 00:00:00")),
+              Row(2, java.sql.Timestamp.valueOf("2021-01-01 00:00:00")),
+              Row(4, null)))
+
+        // Future timestamp should fail validation
+        val tomorrow = "current_timestamp() + INTERVAL 1 DAY"
+        sql(s"INSERT INTO $source VALUES (5, $tomorrow)")
+
+        val error = intercept[SparkRuntimeException] {
+          sql(
+            s"""
+               |MERGE INTO $target t
+               |USING (SELECT * FROM $source WHERE id = 5) s
+               |ON t.id = s.id
+               |WHEN NOT MATCHED THEN INSERT (id, creation_time) VALUES (s.id, s.creation_time)
+               |""".stripMargin)
+        }
+
+        assert(error.getMessage.contains("CHECK_CONSTRAINT_VIOLATION"))
+        assert(error.getMessageParameters.get("constraintName") == "valid_time")
+        assert(error.getMessageParameters.get("expression") ==
+          "creation_time <= current_timestamp()")
+      }
+    }
+  }
+
+  test("Check constraint with current_date function") {
+    // Create another table with other current_* functions
+    withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+      sql(s"CREATE TABLE $t (id INT, creation_date DATE, " +
+        s"CONSTRAINT valid_date CHECK (creation_date <= current_date())) $defaultUsing")
+
+      // Insert valid data (current or past timestamp)
+      sql(s"INSERT INTO $t VALUES (1, current_date()), (2, DATE'2020-01-01')")
+      checkAnswer(spark.table(t).select("id"), Seq(Row(1), Row(2)))
+
+      // Future date should fail validation
+      val tomorrow = "DATE'9999-12-31'"
+      val error = intercept[SparkRuntimeException] {
+        sql(s"INSERT INTO $t VALUES (3, $tomorrow)")
+      }
+      checkError(
+        exception = error,
+        condition = "CHECK_CONSTRAINT_VIOLATION",
+        sqlState = "23001",
+        parameters = Map(
+          "constraintName" -> "valid_date",
+          "expression" -> "creation_date <= current_date()",
+          "values" -> " - creation_date : 2932896"
+        )
+      )
+    }
+  }
+
+  test("Check constraint with current_database function") {
+    withNamespaceAndTable("test_db", "tbl", nonPartitionCatalog) { t =>
+      sql(s"USE $nonPartitionCatalog.test_db")
+      sql(s"CREATE TABLE $t (id INT, db STRING, " +
+        s"CONSTRAINT valid_db CHECK (db = current_database())) $defaultUsing")
+
+      // Insert valid data (current database)
+      sql(s"INSERT INTO $t VALUES (1, current_database()), (2, 'test_db')")
+      checkAnswer(spark.table(t).select("id"), Seq(Row(1), Row(2)))
+
+      // Invalid database should fail validation
+      val error = intercept[SparkRuntimeException] {
+        sql(s"INSERT INTO $t VALUES (3, 'invalid_db')")
+      }
+      checkError(
+        exception = error,
+        condition = "CHECK_CONSTRAINT_VIOLATION",
+        sqlState = "23001",
+        parameters = Map(
+          "constraintName" -> "valid_db",
+          "expression" -> "db = current_database()",
+          "values" -> " - db : invalid_db"
+        )
+      )
+    }
+  }
+
+  test("Check constraint with current_database function - update operation") {
+    withNamespaceAndTable("test_db", "tbl", rowLevelOPCatalog) { t =>
+      sql(s"USE $rowLevelOPCatalog.test_db")
+      sql(s"CREATE TABLE $t (id INT, db STRING, " +
+        s"CONSTRAINT valid_db CHECK (db = current_database())) $defaultUsing")
+
+      // Insert initial valid data
+      sql(s"INSERT INTO $t VALUES (1, current_database()), (2, 'test_db')")
+      checkAnswer(spark.table(t).select("id"), Seq(Row(1), Row(2)))
+
+      // Valid update with current database value
+      sql(s"UPDATE $t SET db = 'test_db' WHERE id = 1")
+      checkAnswer(spark.table(t).where("id = 1"), Seq(Row(1, "test_db")))
+
+      // Invalid database should fail validation
+      val error = intercept[SparkRuntimeException] {
+        sql(s"UPDATE $t SET db = 'invalid_db' WHERE id = 2")
+      }
+      checkError(
+        exception = error,
+        condition = "CHECK_CONSTRAINT_VIOLATION",
+        sqlState = "23001",
+        parameters = Map(
+          "constraintName" -> "valid_db",
+          "expression" -> "db = current_database()",
+          "values" -> " - db : invalid_db"
+        )
+      )
+    }
+  }
+
+  test("Check constraint with current_database function - merge operation") {
+    withNamespaceAndTable("test_db", "target", rowLevelOPCatalog) { target =>
+      withNamespaceAndTable("test_db", "source", rowLevelOPCatalog) { source =>
+        sql(s"USE $rowLevelOPCatalog.test_db")
+        sql(s"CREATE TABLE $target (id INT, db STRING, " +
+          s"CONSTRAINT valid_db CHECK (db = current_database())) $defaultUsing")
+        sql(s"CREATE TABLE $source (id INT, db STRING) $defaultUsing")
+
+        // Insert initial valid data
+        sql(s"INSERT INTO $target VALUES (1, current_database()), (2, 'test_db')")
+        sql(s"INSERT INTO $source VALUES (3, 'test_db'), (4, 'invalid_db')")
+
+        // Valid merge with current database value
+        sql(
+          s"""
+             |MERGE INTO $target t
+             |USING (SELECT * FROM $source WHERE id = 3) s
+             |ON t.id = s.id
+             |WHEN NOT MATCHED THEN INSERT (id, db) VALUES (s.id, s.db)
+             |""".stripMargin)
+        checkAnswer(spark.table(target).orderBy("id"),
+          Seq(Row(1, "test_db"), Row(2, "test_db"), Row(3, "test_db")))
+
+        // Invalid database should fail validation
+        val error = intercept[SparkRuntimeException] {
+          sql(
+            s"""
+               |MERGE INTO $target t
+               |USING (SELECT * FROM $source WHERE id = 4) s
+               |ON t.id = s.id
+               |WHEN NOT MATCHED THEN INSERT (id, db) VALUES (s.id, s.db)
+               |""".stripMargin)
+        }
+        checkError(
+          exception = error,
+          condition = "CHECK_CONSTRAINT_VIOLATION",
+          sqlState = "23001",
+          parameters = Map(
+            "constraintName" -> "valid_db",
+            "expression" -> "db = current_database()",
+            "values" -> " - db : invalid_db"
+          )
+        )
+      }
+    }
+  }
+
   test("Check constraint violation on table insert - nested column") {
     withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
       sql(s"CREATE TABLE $t (id INT," +
