dongjoon-hyun commented on code in PR #59545:
URL: https://github.com/apache/spark/pull/59545#discussion_r4464512056
##########
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:
Major: on the latest master, `HandleSpecialCommand` no longer calls this
method for `ALTER TABLE ... ADD CONSTRAINT ... CHECK`, so every such constraint
keeps the predicate again.
- SPARK-60824
([b7557f1](https://github.com/apache/spark/commit/b7557f1b5df91a6942edacc2e1dfd8f965e2333b),
#60042) made `AddCheckConstraint` a `KeepAnalyzedQuery`
([v2AlterTableCommands.scala
L383-L390](https://github.com/apache/spark/blob/11e5a465c516f67cb6357e56a19b482d2ccfb138/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/v2AlterTableCommands.scala#L383-L390)).
It landed after the CI of this head ran, so the CI merged an earlier master.
- `HandleSpecialCommand` matches `KeepAnalyzedQuery` before the arm that
calls `markAsAnalyzed` ([Analyzer.scala
L4487-L4495](https://github.com/apache/spark/blob/11e5a465c516f67cb6357e56a19b482d2ccfb138/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L4487-L4495)).
For `AddCheckConstraint`, it only runs `storeAnalyzedQuery()` and then
descends into the child `Filter`, which does not hold the `CheckConstraint`.
So `omitPredicate` stays false, and `toV2Constraint` builds the predicate
from the optimized condition, as on master. There is no textual conflict, so
GitHub still shows the PR as mergeable. From reading the merged code, the new
assertions on `predicate() == null`, such as [CheckConstraintSuite.scala
L271](https://github.com/apache/spark/blob/c6c8935a247d8fbb94401b7846e82ecdcd59ca09/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala#L271)
and
[L379-L380](https://github.com/apache/spark/blob/c6c8935a247d8fbb94401b7846e82ecdcd59ca09/sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala#L379-L380),
would fail after merging the latest master.
One way is to mark the constraint where the analyzed query is stored:
```scala
override def storeAnalyzedQuery(): Command =
copy(checkConstraint = checkConstraint.markAsAnalyzed(), analyzedQuery =
Some(child))
```
Alternatively, `HandleSpecialCommand` could also mark the analysis-aware
expressions of a `KeepAnalyzedQuery`, which would cover other commands of that
kind too. Could you merge the latest master and re-run the CI?
##########
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:
Minor: besides variables and CHAR columns, the settings of the writer can
change how the rest of a condition is parsed and resolved. A condition that
ALTER TABLE stores without a predicate is then checked differently, while the
predicate stored on master was not affected.
From reading the code:
```sql
CREATE TABLE t (id INT, created TIMESTAMP);
-- In a session with the default spark.sql.caseSensitive=false:
ALTER TABLE t ADD CONSTRAINT c CHECK (Created <= current_timestamp());
-- In a session with spark.sql.caseSensitive=true:
INSERT INTO t VALUES (1, current_timestamp()); -- UNRESOLVED_COLUMN for
`Created`
```
- On master, the predicate holds `FieldReference(created)`, the name of the
resolved column, so a case-sensitive session resolves it too.
- With this PR, each write parses `Created` again and resolves it with its
own resolver. A case-sensitive writer finds no column, so every write fails.
- Under `spark.sql.ansi.doubleQuotedIdentifiers`, `"CLOSED"` in `s <>
"CLOSED" OR d <= current_date()` becomes a column name, with the same result.
- Under `spark.sql.parser.escapedStringLiterals`, `'A\\B'` in `s <> 'A\\B'
OR d <= current_date()` is read as a different string, so the check changes
silently.
The description already says that such a constraint is re-parsed with each
writer's session settings, so the parser settings may be fine as a documented
limitation. The case mismatch is the same kind of difference that the PR
already handles for `` `USER` ``, and it can be detected here: the parser could
record the single-part names it sees, and `markAsAnalyzed` could keep the
predicate when one of them differs, case-sensitively, from the name of the
column it resolves to.
Could you either add that check, or mention the case-sensitivity and
parser-setting effects in the description?
##########
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:
Nit: no test fails if this `CURRENT_PATH` exclusion is removed.
Without it, `CHECK (p = current_path)` would keep the predicate, which
stores the SQL path of the ALTER TABLE as a literal. The new tests cover the
paren-less `user` and `current_date`, but not `current_path`. An assertion that
such a constraint has `predicate() == null`, for example in the test reading
the current database, would cover this branch.
##########
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:
Nit: the testing section says that under ANSI mode, ALTER TABLE accepts
conditions with a division by zero inside a cast in a branch that may not run.
As this comment says, that holds only for a cast to `TIMESTAMP_LTZ(9)`. With a
cast to DATE, TIMESTAMP or TIMESTAMP_NTZ, `SpecialDatetimeValues` evaluates the
string, and the ALTER TABLE fails with DIVIDE_BY_ZERO. That failure is on
master too, so I am not asking to change it here.
After the `coalesce` case was removed, this test covers only the
`TIMESTAMP_LTZ(9)` cast. Could the testing section say so, for example "accepts
such a cast to `TIMESTAMP_LTZ(9)` in a branch that may not run"?
##########
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:
Nit: this INSERT is expected to fail because `'nobody'` differs from
`current_user`. However, `nobody` is a standard Unix account, and some
sandboxed or container builds run as that user. Then `current_user` returns
`nobody` (from `SPARK_USER` or the UGI short name), the row passes, and
`intercept` fails.
Could the row use a value that cannot be a user name, such as an empty
string?
--
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]
