dongjoon-hyun commented on code in PR #59545:
URL: https://github.com/apache/spark/pull/59545#discussion_r4425480938
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -170,6 +174,275 @@ 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()",
Review Comment:
Nit: this row and the `CAST('today' AS DATE)` row at L191 fail if the clock
passes midnight during the rejected INSERT.
The row value comes from the `ComputeCurrentTime` snapshot, and the
condition reads the date later: `SpecialDatetimeValues` runs after that rule
([Optimizer.scala
L371-L373](https://github.com/apache/spark/blob/57f662dda0b4c42e5e6d8adc59d4617dabd8dd2d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala#L371-L373)),
and `ConstantFolding` evaluates the time-only cast after that. If the date
changes in between, the row written a day ahead is no longer past the
condition, so the INSERT succeeds and `intercept` fails.
The window is a few milliseconds a day. The `'now'` rows and the TIME cast
row do not have it. Could these two rows use a step of two days?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -120,16 +127,47 @@ 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,
+ keepPredicate: Boolean = false,
+ omitPredicate: 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)
+ override def stringArgs: Iterator[Any] =
+ Iterator(child, condition, userProvidedName, tableName,
userProvidedCharacteristic)
+
+ // A write parses `condition` again. 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.
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.
+ override def markAsAnalyzed(): CheckConstraint = copy(omitPredicate =
Review Comment:
Minor: a paren-less `current_date`, `current_timestamp`, `current_user` or
`user` in such a condition can now bind to a column added later, and the check
then changes silently.
- By default these keywords parse as an `UnresolvedAttribute`
([AstBuilder.scala
L4082-L4090](https://github.com/apache/spark/blob/57f662dda0b4c42e5e6d8adc59d4617dabd8dd2d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala#L4082-L4090)).
- A write resolves that name to a column first, and to the function only
when no column matches ([ColumnResolutionHelper.scala
L158-L165](https://github.com/apache/spark/blob/57f662dda0b4c42e5e6d8adc59d4617dabd8dd2d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala#L158-L165)).
- `markAsAnalyzed` stores such a condition without a predicate, and ADD
COLUMN does not look at the constraints.
Before this PR, the stored predicate held a literal, so the new column
changed nothing. From reading the code:
```sql
CREATE TABLE t (id INT, owner STRING);
ALTER TABLE t ADD CONSTRAINT c CHECK (owner = user);
ALTER TABLE t ADD COLUMN user STRING;
INSERT INTO t VALUES (1, 'bob', NULL); -- owner = NULL is NULL, so the row
passes
INSERT INTO t VALUES (2, 'bob', 'bob'); -- passes: the check now compares
two columns of the row
```
`d <= current_date` behaves the same way after `ADD COLUMN current_date
DATE`. This is the effect I noted for variables in [the earlier
thread](https://github.com/apache/spark/pull/59545#discussion_r4349708565),
which keeping the predicate now avoids for them. CREATE TABLE constraints
without a predicate already take this path.
I don't see a small fix here, because the name exists only in the condition
text. Could the user-facing section mention this next to the note about
re-parsing, with SPARK-60453 or a separate JIRA tracking it?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -171,6 +209,65 @@ 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 = cast match {
Review Comment:
Nit: the constant-string logic now exists twice.
L239-L240 repeat the foldable and `StringType` guard of
`FoldableStringCast`, and L241-L246 repeat the evaluation and the catch of
`readsCurrentDate` ([finishAnalysis.scala
L321-L329](https://github.com/apache/spark/blob/57f662dda0b4c42e5e6d8adc59d4617dabd8dd2d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala#L321-L329)).
So `CAST('12:00:00' AS TIMESTAMP)` evaluates its string twice.
Could one `private[sql]` helper next to `FoldableStringCast` return the
constant string of a cast, with the guard and the catch, so that this method
only decides on the value?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/constraints.scala:
##########
@@ -171,6 +209,65 @@ 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 = cast match {
+ case c if SpecialDatetimeValues.readsCurrentDate(c) => true
+ case Cast(s, TimestampType | _: TimestampLTZNanosType, _, _)
Review Comment:
Nit: no test fails if `| _: TimestampLTZNanosType` is removed here.
The only row with this type is the `IF(false, ...)` one in "with an error in
a branch", which passes either way. The rows of the current-time test use only
TIMESTAMP, TIMESTAMP_NTZ and DATE. Without this arm, `CAST('23:59:59' AS
TIMESTAMP_LTZ(9))` is frozen at the ALTER TABLE date again, since that cast
also completes the string with the current date ([SparkDateTimeUtils.scala
L906-L911](https://github.com/apache/spark/blob/57f662dda0b4c42e5e6d8adc59d4617dabd8dd2d/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L906-L911)).
Could the current-time test get a row with this cast?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -128,16 +128,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 (query.resolve(u.nameParts, resolver).isDefined) {
Review Comment:
Nit: `query.resolve` throws when a name goes through a field that does not
exist, and here that error loses its position in the condition.
`ColumnResolutionHelper` runs the same lookup inside `withPosition`
([L152](https://github.com/apache/spark/blob/57f662dda0b4c42e5e6d8adc59d4617dabd8dd2d/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala#L152)),
so before this change the error pointed into `predicateSql`. For example, take
`from_json(s.j, 'a INT').a > 1`, which is stored without a predicate. After
`ALTER TABLE t DROP COLUMN s.j`, a write now fails on this line with the same
error but without the query context.
```suggestion
if (withPosition(u)(query.resolve(u.nameParts, resolver)).isDefined)
{
```
--
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]
