LuciferYang commented on code in PR #59545:
URL: https://github.com/apache/spark/pull/59545#discussion_r4374701627
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -120,16 +123,39 @@ case class CheckConstraint(
condition: String,
override val userProvidedName: String = null,
override val tableName: String = null,
- override val userProvidedCharacteristic: ConstraintCharacteristic =
ConstraintCharacteristic.empty)
+ override val userProvidedCharacteristic: ConstraintCharacteristic =
ConstraintCharacteristic.empty,
+ hasParameterMarker: Boolean = false,
+ readsCurrentContext: Boolean = false)
extends UnaryExpression
with TableConstraint
- with ImplicitCastInputTypes {
+ with ImplicitCastInputTypes
+ with AnalysisAwareExpression[CheckConstraint] {
// scalastyle:on line.size.limit
+ final override val nodePatterns: Seq[TreePattern] =
Seq(ANALYSIS_AWARE_EXPRESSION)
+
override def inputTypes: Seq[AbstractDataType] = Seq(BooleanType)
+ // Under the legacy parameter substitution, a parameter marker stays in
`condition`, which a write
+ // cannot bind, so a condition with one keeps the predicate built from the
optimized plan, as
+ // before.
+ override def markAsAnalyzed(): CheckConstraint = copy(
Review Comment:
Updated the description: those conditions keep their values from the time of
the ALTER TABLE, unless another part of the condition gets the constraint
stored without a predicate, in which case each write evaluates them too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -171,6 +197,40 @@ 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 Cast(s, to, _, _) if s.foldable &&
s.dataType.isInstanceOf[StringType] =>
+ val v = s.eval()
+ v != null && castReadsCurrentDate(v.asInstanceOf[UTF8String], to)
+ case _ => false
+ }
+ }
+
+ /**
+ * Whether casting `s` to `to` reads the current date: `s` is 'now',
'today', 'tomorrow' or
+ * 'yesterday', which `SpecialDatetimeValues` replaces ('epoch' is a fixed
value), or `s` has only
+ * a time part, which a cast to a timestamp with local time zone completes
with the current date.
+ */
+ private def castReadsCurrentDate(s: UTF8String, to: DataType): Boolean = to
match {
Review Comment:
Done in 57f662dda0b: the special-value part is now
`SpecialDatetimeValues.readsCurrentDate`, which shares the rule's guard and the
types in `conv` through one extractor, so a type added to `conv` is detected as
well.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -120,16 +123,39 @@ case class CheckConstraint(
condition: String,
override val userProvidedName: String = null,
override val tableName: String = null,
- override val userProvidedCharacteristic: ConstraintCharacteristic =
ConstraintCharacteristic.empty)
+ override val userProvidedCharacteristic: ConstraintCharacteristic =
ConstraintCharacteristic.empty,
+ hasParameterMarker: Boolean = false,
+ readsCurrentContext: Boolean = false)
Review Comment:
Renamed the field to `omitPredicate` in 57f662dda0b, since it now also
depends on the checks from your other comments, and the parser flag to
`keepPredicate`, since it now also covers `IDENTIFIER(...)` and names with more
than one part. A `stringArgs` override leaves both out of plan strings.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -170,6 +170,146 @@ 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()`.
+ 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 1 DAY"),
+ ("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()", "1")
+ ).foreach { case (dataType, now, current, step) =>
+ Seq("true", "false").foreach { ansi =>
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> ansi) {
+ 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")
+ assert(error.getMessageParameters.get("expression") == s"created
<= $now")
+ }
+ }
+ }
+ }
+ }
+
+ 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)))
+ }
+ }
+
+ 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')")
+ val error = intercept[SparkRuntimeException] {
+ sql(s"INSERT INTO $t VALUES (2, 'ns')")
+ }
+ assert(error.getCondition == "CHECK_CONSTRAINT_VIOLATION")
Review Comment:
Done in 57f662dda0b.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -170,6 +170,146 @@ 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()`.
+ 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 1 DAY"),
+ ("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()", "1")
+ ).foreach { case (dataType, now, current, step) =>
+ Seq("true", "false").foreach { ansi =>
Review Comment:
Dropped both in 57f662dda0b.
--
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]
