LuciferYang commented on code in PR #59545:
URL: https://github.com/apache/spark/pull/59545#discussion_r4454091859
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -7189,9 +7189,16 @@ 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) ||
+ child.exists {
+ case u: UnresolvedAttribute =>
+ u.nameParts.length > 1 ||
LiteralFunctionResolution.resolve(u.nameParts).isDefined
Review Comment:
I tried keeping the matched names and checking `child.references`, but a
write resolves the name with its own settings. With
`spark.sql.caseSensitive=true`, a `` `USER` `` that ALTER TABLE resolved to the
column `user` becomes `current_user()`, and with
`spark.sql.ansi.enforceReservedKeywords=true`, so does an unquoted `user`.
Either way the row with a NULL `user` then passes. So ff1f973919d keeps the
predicate for a column with such a name too, as on master, and the description
says so. The parser now also recognizes the keyword itself, so an ALTER TABLE
run with `enforceReservedKeywords`, where `user` parses as the function, keeps
the predicate as well.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -205,6 +239,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`.
Review Comment:
Done in ff1f973919d: `markAsAnalyzed` skips the check when the condition
calls a user-defined function, with the same match as
`failIfCallsUserDefinedFunction`. A new test checks that the function is not
called before the error.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -144,16 +144,22 @@ class ResolveTableConstraints(val catalogManager:
CatalogManager) extends Rule[L
private def buildColumnExtractors(
expr: Expression,
+ query: LogicalPlan,
columnExtractors: mutable.Map[String, Expression]): Unit = {
expr match {
case u: UnresolvedExtractValue =>
// When extracting a value from a Map or Array type, we display only
the specific extracted
// value rather than the entire Map or Array structure for clarity and
readability.
columnExtractors(u.sql) = u
+ // A name that is not a column of the write, such as `current_timestamp`
without parentheses
+ // or a variable, is not a value of the row.
case u: UnresolvedAttribute =>
- columnExtractors(u.name) = u
+ val resolver = catalogManager.v1SessionCatalog.conf.resolver
+ if (withPosition(u)(query.resolve(u.nameParts, resolver)).isDefined) {
Review Comment:
Done in ff1f973919d: `buildColumnExtractors` is back to what master does,
and once the analysis resolves the `CheckInvariant`, `ResolveTableConstraints`
drops the extractors that reference no attribute of the row, such as a
paren-less `current_user` or a variable. This also drops the `withPosition`
from the last round. The new test pins the message of a CREATE TABLE constraint
with both.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -231,6 +231,289 @@ 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") {
+ Seq(
+ "coalesce(id, CAST(CAST(1/0 AS STRING) AS INT)) > 0",
+ "id < IF(false, hour(CAST(CAST(1/0 AS STRING) AS TIMESTAMP_LTZ(9))),
100)"
+ ).foreach { cond =>
+ 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)")
+ sql(s"INSERT INTO $t VALUES (1, current_timestamp())")
+ checkAnswer(spark.table(t).select("id"), Seq(Row(1)))
+ }
+ }
+ Seq("INT", "DATE").foreach { to =>
Review Comment:
Done in ff1f973919d: the `coalesce` case is gone, and the nondeterministic
case uses TIMESTAMP and DATE, which both reach the catch.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala:
##########
@@ -7189,9 +7189,16 @@ 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) ||
+ child.exists {
+ case u: UnresolvedAttribute =>
+ u.nameParts.length > 1 ||
LiteralFunctionResolution.resolve(u.nameParts).isDefined
Review Comment:
Done in ff1f973919d.
--
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]
