LuciferYang commented on code in PR #59545:
URL: https://github.com/apache/spark/pull/59545#discussion_r4467344158
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -160,8 +169,38 @@ case class CheckConstraint(
}
}
+ override def stringArgs: Iterator[Any] =
+ Iterator(child, condition, userProvidedName, tableName,
userProvidedCharacteristic)
+
+ // A write parses `condition` again, with its own settings. It cannot bind a
parameter marker
+ // that the legacy parameter substitution leaves there, resolves an
IDENTIFIER clause in its own
+ // session, and does not resolve a column qualified by the table name the
way ALTER TABLE does.
+ // It also resolves a name such as `current_date` or `user` to the function
or to a column,
+ // depending on its case sensitivity, its reserved keywords and the columns
of the table at that
+ // time. None of them can be told apart after analysis, so the parser sets
`keepPredicate` for
+ // them, and for any name with more than one part, since it cannot tell a
table qualifier from a
+ // struct column. Such a condition, or one that
`resolvesDifferentlyOnWrite`, keeps the predicate
+ // built from the optimized plan, as before. So does one that calls a
user-defined function,
+ // which the analysis rejects next and which is not evaluated here.
+ override def markAsAnalyzed(): CheckConstraint = copy(omitPredicate =
Review Comment:
Thanks for catching this. Merged master in db3b39f31c6 and fixed in
1c0a8266e50: `AddCheckConstraint.storeAnalyzedQuery` now marks the constraint
as you suggested. `CheckConstraint` is no longer an `AnalysisAwareExpression`,
since nothing else calls `markAsAnalyzed` now. On the merged master without the
fix, the three tests that assert `predicate() == null` failed, and they pass
with it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -203,6 +242,58 @@ case class CheckConstraint(
}
}
+object CheckConstraint {
+ /**
+ * Whether `e`, with its `RuntimeReplaceable`s replaced, reads the current
time, date, catalog,
+ * user and the like, which the optimizer replaces with their values. The
CURRENT_LIKE pattern
+ * stands in for `ReplaceCurrentLike`, which needs a `CatalogManager`, and
for the functions
+ * `ComputeCurrentTime` replaces. `ComputeCurrentTime` is still applied for
the casts from TIME
+ * to a timestamp it rewrites, which carry no CURRENT_LIKE bit.
+ */
+ private def readsCurrentContext(e: Expression): Boolean = {
+ val replaced = ReplaceExpressions.replace(e)
+ replaced.containsPattern(CURRENT_LIKE) ||
+ (ComputeCurrentTime.applyForExpression(replaced) ne replaced) ||
+ replaced.exists {
+ case c: Cast => castReadsCurrentDate(c)
+ case _ => false
+ }
+ }
+
+ /**
+ * Whether `cast` reads the current date: `SpecialDatetimeValues` replaces
it with the value of
+ * 'now', 'today', 'tomorrow' or 'yesterday', or it casts a string with only
a time part to a
+ * timestamp with local time zone, which completes it with the current date.
Only a cast of a
+ * constant string at the top level counts, not one into an array, map or
struct of timestamps,
+ * from a VARIANT, or of a string that only the optimizer makes constant.
+ */
+ private def castReadsCurrentDate(cast: Cast): Boolean = {
+ val completesTime = cast.dataType match {
+ case TimestampType | _: TimestampLTZNanosType => true
+ case _ => false
+ }
+ (completesTime || SpecialDatetimeValues.rewrites(cast.dataType)) &&
+ SpecialDatetimeValues.constantString(cast).exists { s =>
+ SpecialDatetimeValues.readsCurrentDate(s, cast.dataType) ||
+ (completesTime && DateTimeUtils.isTimeOnlyTimestamp(s))
+ }
+ }
+
+ /**
+ * Whether a write that parses the condition resolves `e` differently: `e`
reads a variable, which
+ * the write looks up in its own session, or a column whose type is or
contains CHAR, which the
+ * write reads without the CHAR type that pads the literals compared with
it. A user-defined
+ * function is rejected by `failIfCallsUserDefinedFunction`.
+ */
+ private def resolvesDifferentlyOnWrite(e: Expression): Boolean = {
Review Comment:
I tried the case check, but master behaves the same here: the stored
predicate refers to the column by the name written in the condition, not by its
name in the table. With `CHECK (Created <= current_timestamp())` and the
predicate kept, as on master, a case-sensitive INSERT fails with
UNRESOLVED_COLUMN for `Created` in the same way, since `V2ExpressionBuilder`
builds the `FieldReference` from the attribute, which keeps the written name.
So keeping the predicate would not help that writer, and I left the check out.
In 1c0a8266e50, the description names `spark.sql.caseSensitive`,
`spark.sql.ansi.doubleQuotedIdentifiers` and
`spark.sql.parser.escapedStringLiterals` among the writer's settings that the
re-parse uses (SPARK-60453).
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -232,6 +244,319 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-60032: Check constraint added by ALTER TABLE reading the current
time") {
+ // The check compares with the time of the write, not the time of the
ALTER TABLE. 'now' is
+ // read after the snapshot `current_timestamp()` takes, so only `<=` holds
for a row written
+ // with `current_timestamp()`. A string with only a time part and 'today'
read the date after
+ // that snapshot too, so their rows step two days ahead in case the date
changes in between.
+ Seq(
+ ("TIMESTAMP", "current_timestamp()", "current_timestamp()", "INTERVAL 1
DAY"),
+ ("TIMESTAMP", "CAST('now' AS TIMESTAMP)", "current_timestamp()",
"INTERVAL 1 DAY"),
+ ("TIMESTAMP", "to_timestamp('now')", "current_timestamp()", "INTERVAL 1
DAY"),
+ ("TIMESTAMP", "CAST('23:59:59.999999' AS TIMESTAMP)",
"current_timestamp()",
+ "INTERVAL 2 DAYS"),
+ ("TIMESTAMP_LTZ(9)", "CAST('23:59:59.999999' AS TIMESTAMP_LTZ(9))",
"current_timestamp()",
+ "INTERVAL 2 DAYS"),
+ ("TIMESTAMP", "CAST(TIME '23:59:59.999999' AS TIMESTAMP)",
"current_timestamp()",
+ "INTERVAL 1 DAY"),
+ ("TIMESTAMP_NTZ", "CAST('now' AS TIMESTAMP_NTZ)", "localtimestamp()",
"INTERVAL 1 DAY"),
+ ("DATE", "current_date()", "current_date()", "1"),
+ ("DATE", "CAST('today' AS DATE)", "current_date()", "2")
+ ).foreach { case (dataType, now, current, step) =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, created $dataType) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT valid_time CHECK (created <=
$now)")
+ val check = getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl"))
+ assert(check.predicateSql() == s"created <= $now")
+ assert(check.predicate() == null, now)
+
+ sql(s"INSERT INTO $t VALUES (1, $current)")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, $current + $step)")
+ }
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ assert(error.getMessageParameters.get("constraintName") ==
"valid_time")
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE on a column
qualified by the table") {
+ // ALTER TABLE resolves `tbl.id` as the column `id` of the table `tbl`,
not as the field `id` of
+ // its struct column `tbl`. A write that parses `predicateSql` reads
`tbl.id` as the field, so
+ // the test fails if this constraint is stored without a predicate.
`nullif` has no V2 form
+ // before the optimizer replaces it, so the test also fails if the
predicate is built from the
+ // analyzed condition.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id BIGINT, tbl STRUCT<id: BIGINT>) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (nullif(tbl.id, 3) > 0)")
+ sql(s"INSERT INTO $t VALUES (3, named_struct('id', -1L))")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (-1, named_struct('id', 3L))")
+ }
+ checkError(
+ exception = error,
+ condition = "CHECK_CONSTRAINT_VIOLATION",
+ sqlState = "23001",
+ parameters = Map(
+ "constraintName" -> "c1",
+ "expression" -> "nullif(tbl.id, 3) > 0",
+ "values" -> " - id : -1"))
+ checkAnswer(spark.table(t).select("id"), Seq(Row(3)))
+ }
+ // The same holds for a condition that also reads the current time: it
keeps the predicate.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id BIGINT, ts TIMESTAMP, tbl STRUCT<id: BIGINT>)
$defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK " +
+ "(tbl.id > 0 AND ts <= current_timestamp() + INTERVAL 1 DAY)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null)
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp(), named_struct('id',
-1L))")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE on a fixed
datetime") {
+ // 'epoch' and a full datetime string do not read the current date, so the
predicate is kept.
+ Seq(
+ "created >= CAST('epoch' AS TIMESTAMP)",
+ "created >= 'epoch'",
+ "created >= CAST('2020-01-01 00:00:00' AS TIMESTAMP)").foreach { cond =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, created TIMESTAMP) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($cond)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ cond)
+ }
+ }
+ // A write that parses `predicateSql` fails to resolve `tbl.created`, so
this fails if the
+ // constraint is stored without a predicate.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, created TIMESTAMP) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (tbl.created >= 'epoch')")
+ sql(s"INSERT INTO $t VALUES (1, TIMESTAMP '2020-01-01 00:00:00')")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, TIMESTAMP '1960-01-01 00:00:00')")
+ }
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE with a parameter
marker") {
+ // Under the legacy parameter substitution, the marker stays in
`predicateSql`, which a write
+ // cannot bind, so the constraint keeps the predicate built from the
optimized plan. Otherwise
+ // the marker is substituted before parsing and the constraint is stored
without a predicate.
+ Seq(true, false).foreach { legacy =>
+ withSQLConf(SQLConf.LEGACY_PARAMETER_SUBSTITUTION_CONSTANTS_ONLY.key ->
legacy.toString) {
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, ts TIMESTAMP) $defaultUsing")
+ spark.sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK " +
+ "(ts >= current_timestamp() - make_interval(0, 0, 0, :d))",
Map("d" -> 1))
+ val check = getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl"))
+ assert(check.predicateSql().contains(":d") == legacy,
check.predicateSql())
+ assert((check.predicate() != null) == legacy, check.predicateSql())
+
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp())")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, current_timestamp() - INTERVAL 2
DAYS)")
+ }
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE reading the current
database") {
+ withNamespace(s"$nonPartitionCatalog.ns2") {
+ sql(s"CREATE NAMESPACE $nonPartitionCatalog.ns2")
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ withCurrentCatalogAndNamespace {
+ sql(s"CREATE TABLE $t (id INT, db STRING) $defaultUsing")
+ sql(s"USE $nonPartitionCatalog.ns")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT valid_db CHECK (db =
current_database())")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() ==
+ null)
+
+ sql(s"USE $nonPartitionCatalog.ns2")
+ sql(s"INSERT INTO $t VALUES (1, 'ns2')")
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, 'ns')")
+ },
+ condition = "CHECK_CONSTRAINT_VIOLATION",
+ sqlState = "23001",
+ parameters = Map(
+ "constraintName" -> "valid_db",
+ "expression" -> "db = current_database()",
+ "values" -> " - db : ns"))
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE resolving names in
the session") {
+ // A write would resolve a variable or an IDENTIFIER clause in its own
session, so such a
+ // condition keeps the predicate built from the optimized plan. The name
is gone by the time of
+ // the write, so the write fails if the constraint is stored without a
predicate.
+ def checkKeepsPredicate(cond: String)(dropName: => Unit): Unit = {
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, ts TIMESTAMP) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($cond)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ cond)
+ dropName
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp())")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+ withSessionVariable("days", "c") {
+ sql("DECLARE VARIABLE days INT DEFAULT 30")
+ checkKeepsPredicate("ts >= current_timestamp() - make_interval(0, 0, 0,
days)") {
+ sql("DROP TEMPORARY VARIABLE days")
+ }
+ sql("DECLARE VARIABLE c STRING DEFAULT 'ts'")
+ checkKeepsPredicate("IDENTIFIER(c) <= current_timestamp() + INTERVAL 1
DAY") {
+ sql("DROP TEMPORARY VARIABLE c")
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE comparing a CHAR
column") {
+ // ALTER TABLE pads 'OPEN' to the length of `status`, but a write that
parses `predicateSql`
+ // does not, so the constraint keeps the predicate built from the
optimized plan.
+ Seq(
+ ("status = 'OPEN' OR closed <= current_date()", "OPEN", true),
+ ("status <> 'CLOSED' OR closed <= current_date()", "CLOSED", false)
+ ).foreach { case (cond, status, accepted) =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, status CHAR(10), closed DATE)
$defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($cond)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ cond)
+ val insert = s"INSERT INTO $t VALUES (1, '$status', DATE'2999-01-01')"
+ if (accepted) {
+ sql(insert)
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ } else {
+ val error = intercept[SparkRuntimeException](sql(insert))
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION", cond)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE with an error in a
branch") {
+ // The optimizer keeps an expression that fails in a branch that may not
run, and so does the
+ // check for the current date.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ // A cast to DATE, TIMESTAMP or TIMESTAMP_NTZ fails in
`SpecialDatetimeValues`, so this one
Review Comment:
Updated the testing section as you suggested.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -232,6 +244,319 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-60032: Check constraint added by ALTER TABLE reading the current
time") {
+ // The check compares with the time of the write, not the time of the
ALTER TABLE. 'now' is
+ // read after the snapshot `current_timestamp()` takes, so only `<=` holds
for a row written
+ // with `current_timestamp()`. A string with only a time part and 'today'
read the date after
+ // that snapshot too, so their rows step two days ahead in case the date
changes in between.
+ Seq(
+ ("TIMESTAMP", "current_timestamp()", "current_timestamp()", "INTERVAL 1
DAY"),
+ ("TIMESTAMP", "CAST('now' AS TIMESTAMP)", "current_timestamp()",
"INTERVAL 1 DAY"),
+ ("TIMESTAMP", "to_timestamp('now')", "current_timestamp()", "INTERVAL 1
DAY"),
+ ("TIMESTAMP", "CAST('23:59:59.999999' AS TIMESTAMP)",
"current_timestamp()",
+ "INTERVAL 2 DAYS"),
+ ("TIMESTAMP_LTZ(9)", "CAST('23:59:59.999999' AS TIMESTAMP_LTZ(9))",
"current_timestamp()",
+ "INTERVAL 2 DAYS"),
+ ("TIMESTAMP", "CAST(TIME '23:59:59.999999' AS TIMESTAMP)",
"current_timestamp()",
+ "INTERVAL 1 DAY"),
+ ("TIMESTAMP_NTZ", "CAST('now' AS TIMESTAMP_NTZ)", "localtimestamp()",
"INTERVAL 1 DAY"),
+ ("DATE", "current_date()", "current_date()", "1"),
+ ("DATE", "CAST('today' AS DATE)", "current_date()", "2")
+ ).foreach { case (dataType, now, current, step) =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, created $dataType) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT valid_time CHECK (created <=
$now)")
+ val check = getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl"))
+ assert(check.predicateSql() == s"created <= $now")
+ assert(check.predicate() == null, now)
+
+ sql(s"INSERT INTO $t VALUES (1, $current)")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, $current + $step)")
+ }
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ assert(error.getMessageParameters.get("constraintName") ==
"valid_time")
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE on a column
qualified by the table") {
+ // ALTER TABLE resolves `tbl.id` as the column `id` of the table `tbl`,
not as the field `id` of
+ // its struct column `tbl`. A write that parses `predicateSql` reads
`tbl.id` as the field, so
+ // the test fails if this constraint is stored without a predicate.
`nullif` has no V2 form
+ // before the optimizer replaces it, so the test also fails if the
predicate is built from the
+ // analyzed condition.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id BIGINT, tbl STRUCT<id: BIGINT>) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (nullif(tbl.id, 3) > 0)")
+ sql(s"INSERT INTO $t VALUES (3, named_struct('id', -1L))")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (-1, named_struct('id', 3L))")
+ }
+ checkError(
+ exception = error,
+ condition = "CHECK_CONSTRAINT_VIOLATION",
+ sqlState = "23001",
+ parameters = Map(
+ "constraintName" -> "c1",
+ "expression" -> "nullif(tbl.id, 3) > 0",
+ "values" -> " - id : -1"))
+ checkAnswer(spark.table(t).select("id"), Seq(Row(3)))
+ }
+ // The same holds for a condition that also reads the current time: it
keeps the predicate.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id BIGINT, ts TIMESTAMP, tbl STRUCT<id: BIGINT>)
$defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK " +
+ "(tbl.id > 0 AND ts <= current_timestamp() + INTERVAL 1 DAY)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null)
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp(), named_struct('id',
-1L))")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE on a fixed
datetime") {
+ // 'epoch' and a full datetime string do not read the current date, so the
predicate is kept.
+ Seq(
+ "created >= CAST('epoch' AS TIMESTAMP)",
+ "created >= 'epoch'",
+ "created >= CAST('2020-01-01 00:00:00' AS TIMESTAMP)").foreach { cond =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, created TIMESTAMP) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($cond)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ cond)
+ }
+ }
+ // A write that parses `predicateSql` fails to resolve `tbl.created`, so
this fails if the
+ // constraint is stored without a predicate.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, created TIMESTAMP) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (tbl.created >= 'epoch')")
+ sql(s"INSERT INTO $t VALUES (1, TIMESTAMP '2020-01-01 00:00:00')")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, TIMESTAMP '1960-01-01 00:00:00')")
+ }
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE with a parameter
marker") {
+ // Under the legacy parameter substitution, the marker stays in
`predicateSql`, which a write
+ // cannot bind, so the constraint keeps the predicate built from the
optimized plan. Otherwise
+ // the marker is substituted before parsing and the constraint is stored
without a predicate.
+ Seq(true, false).foreach { legacy =>
+ withSQLConf(SQLConf.LEGACY_PARAMETER_SUBSTITUTION_CONSTANTS_ONLY.key ->
legacy.toString) {
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, ts TIMESTAMP) $defaultUsing")
+ spark.sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK " +
+ "(ts >= current_timestamp() - make_interval(0, 0, 0, :d))",
Map("d" -> 1))
+ val check = getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl"))
+ assert(check.predicateSql().contains(":d") == legacy,
check.predicateSql())
+ assert((check.predicate() != null) == legacy, check.predicateSql())
+
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp())")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, current_timestamp() - INTERVAL 2
DAYS)")
+ }
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE reading the current
database") {
+ withNamespace(s"$nonPartitionCatalog.ns2") {
+ sql(s"CREATE NAMESPACE $nonPartitionCatalog.ns2")
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ withCurrentCatalogAndNamespace {
+ sql(s"CREATE TABLE $t (id INT, db STRING) $defaultUsing")
+ sql(s"USE $nonPartitionCatalog.ns")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT valid_db CHECK (db =
current_database())")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() ==
+ null)
+
+ sql(s"USE $nonPartitionCatalog.ns2")
+ sql(s"INSERT INTO $t VALUES (1, 'ns2')")
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, 'ns')")
+ },
+ condition = "CHECK_CONSTRAINT_VIOLATION",
+ sqlState = "23001",
+ parameters = Map(
+ "constraintName" -> "valid_db",
+ "expression" -> "db = current_database()",
+ "values" -> " - db : ns"))
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE resolving names in
the session") {
+ // A write would resolve a variable or an IDENTIFIER clause in its own
session, so such a
+ // condition keeps the predicate built from the optimized plan. The name
is gone by the time of
+ // the write, so the write fails if the constraint is stored without a
predicate.
+ def checkKeepsPredicate(cond: String)(dropName: => Unit): Unit = {
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, ts TIMESTAMP) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($cond)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ cond)
+ dropName
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp())")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+ withSessionVariable("days", "c") {
+ sql("DECLARE VARIABLE days INT DEFAULT 30")
+ checkKeepsPredicate("ts >= current_timestamp() - make_interval(0, 0, 0,
days)") {
+ sql("DROP TEMPORARY VARIABLE days")
+ }
+ sql("DECLARE VARIABLE c STRING DEFAULT 'ts'")
+ checkKeepsPredicate("IDENTIFIER(c) <= current_timestamp() + INTERVAL 1
DAY") {
+ sql("DROP TEMPORARY VARIABLE c")
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE comparing a CHAR
column") {
+ // ALTER TABLE pads 'OPEN' to the length of `status`, but a write that
parses `predicateSql`
+ // does not, so the constraint keeps the predicate built from the
optimized plan.
+ Seq(
+ ("status = 'OPEN' OR closed <= current_date()", "OPEN", true),
+ ("status <> 'CLOSED' OR closed <= current_date()", "CLOSED", false)
+ ).foreach { case (cond, status, accepted) =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, status CHAR(10), closed DATE)
$defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($cond)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ cond)
+ val insert = s"INSERT INTO $t VALUES (1, '$status', DATE'2999-01-01')"
+ if (accepted) {
+ sql(insert)
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ } else {
+ val error = intercept[SparkRuntimeException](sql(insert))
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION", cond)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE with an error in a
branch") {
+ // The optimizer keeps an expression that fails in a branch that may not
run, and so does the
+ // check for the current date.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ // A cast to DATE, TIMESTAMP or TIMESTAMP_NTZ fails in
`SpecialDatetimeValues`, so this one
+ // casts to TIMESTAMP_LTZ(9).
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, ts TIMESTAMP) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK " +
+ "(id < IF(false, hour(CAST(CAST(1/0 AS STRING) AS
TIMESTAMP_LTZ(9))), 100))")
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp())")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ Seq("TIMESTAMP", "DATE").foreach { to =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT) $defaultUsing")
+ val error = intercept[AnalysisException] {
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK " +
+ s"(rand() > 0 OR CAST(CAST(1/0 AS STRING) AS $to) IS NOT NULL)")
+ }
+ assert(error.getCondition == "NON_DETERMINISTIC_CHECK_CONSTRAINT",
to)
+ }
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE with a function
without parentheses") {
+ // A write that parses `predicateSql` resolves `user` or `current_date` to
a column of that name
+ // added later, so such a condition keeps the predicate built from the
optimized plan.
+ Seq(
+ ("owner STRING", "owner = user", "user STRING", "'', ''"),
+ ("d DATE", "d <= current_date", "current_date DATE", "DATE'2999-01-01',
DATE'2999-12-31'"),
+ ("d DATE", "d <= `current_date`", "current_date DATE",
"DATE'2999-01-01', DATE'2999-12-31'")
+ ).foreach { case (column, cond, added, values) =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, $column) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($cond)")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ cond)
+ sql(s"ALTER TABLE $t ADD COLUMN $added")
+ val error = intercept[SparkRuntimeException](sql(s"INSERT INTO $t
VALUES (1, $values)"))
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION", cond)
+ }
+ }
+ // With `spark.sql.ansi.enforceReservedKeywords`, the keyword parses as
the function, which a
+ // write with the default settings resolves to such a column.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, owner STRING) $defaultUsing")
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.ENFORCE_RESERVED_KEYWORDS.key -> "true") {
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (owner = user)")
+ }
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null)
+ sql(s"ALTER TABLE $t ADD COLUMN user STRING")
+ val error = intercept[SparkRuntimeException](sql(s"INSERT INTO $t VALUES
(1, '', '')"))
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
+ }
+ // A column with such a name keeps the predicate too. A write with
+ // `spark.sql.ansi.enforceReservedKeywords` parses `user` as the function,
and a case-sensitive
+ // one resolves `USER` to the function, so both would accept a NULL row
without it.
+ Seq("user", "`USER`").foreach { name =>
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, user STRING, ts TIMESTAMP)
$defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK ($name IS NOT NULL AND ts
<= now())")
+ assert(getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl")).predicate() != null,
+ name)
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint added by ALTER TABLE does not call a
user-defined function") {
+ // `HandleSpecialCommand` runs before `CheckAnalysis` rejects such a
condition, so it must not
+ // evaluate the function.
+ withV2Function("strlen", StrLen(CountingString)) {
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, ts TIMESTAMP) $defaultUsing")
+ CountingString.calls = 0
+ val error = intercept[AnalysisException] {
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK " +
+ s"(ts <= CAST(fun_$catalog.ns.strlen('12:00:00') AS TIMESTAMP))")
+ }
+ assert(error.getCondition ==
"USER_DEFINED_FUNCTION_IN_CHECK_CONSTRAINT")
+ assert(CountingString.calls == 0)
+ }
+ }
+ }
+
+ test("SPARK-60032: Check constraint violation lists only the values of the
row") {
+ // A write that parses `predicateSql` resolves `current_user` without
parentheses and a
+ // variable to values that are not columns of the row.
+ withSessionVariable("v") {
+ sql("DECLARE VARIABLE v STRING DEFAULT 'x'")
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id INT, owner STRING, " +
+ s"CONSTRAINT c1 CHECK (owner = current_user OR owner = v))
$defaultUsing")
+ checkError(
+ exception = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (1, 'nobody')")
Review Comment:
Done in 1c0a8266e50: the row uses an empty string.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -7189,11 +7189,27 @@ class AstBuilder extends DataTypeAstBuilder
override def visitCheckConstraint(ctx: CheckConstraintContext):
CheckConstraint =
withOrigin(ctx) {
val condition = getOriginalText(ctx.expr)
+ val child = expression(ctx.booleanExpression())
CheckConstraint(
- child = expression(ctx.booleanExpression()),
- condition = condition)
+ child = child,
+ condition = condition,
+ keepPredicate = child.containsAnyPattern(PARAMETER,
UNRESOLVED_IDENTIFIER) ||
+ hasCurrentLikeKeyword(ctx.booleanExpression()) ||
+ child.exists {
+ case u: UnresolvedAttribute =>
+ u.nameParts.length > 1 ||
LiteralFunctionResolution.isLiteralFunction(u.nameParts)
+ case _ => false
+ })
}
+ /**
+ * Whether `tree` has a keyword such as `current_date` or `user` without
parentheses, which parses
+ * as a column name or as the function depending on
`spark.sql.ansi.enforceReservedKeywords`.
+ */
+ private def hasCurrentLikeKeyword(tree: ParseTree): Boolean = tree match {
+ case c: CurrentLikeContext => c.name.getType != SqlBaseParser.CURRENT_PATH
Review Comment:
Done in 1c0a8266e50: the current database test also checks that `p =
current_path` is stored without a predicate. Removing the exclusion fails it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
