dongjoon-hyun commented on code in PR #59545:
URL: https://github.com/apache/spark/pull/59545#discussion_r4349708579
##########
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(
+ readsCurrentContext = !hasParameterMarker &&
CheckConstraint.readsCurrentContext(child))
+
+ // ALTER TABLE builds the constraint from the optimized plan, where the
current time, date,
+ // catalog, user and the like are replaced by their values at the time of
the ALTER TABLE. A
+ // condition that `readsCurrentContext` finds reading one is stored without
a predicate, so that
+ // writes check `predicateSql`.
+ // The predicate is not built from the analyzed condition, as the default
value of a column is,
+ // because `V2ExpressionBuilder` folds `CAST('today' AS DATE)` there as an
ordinary cast. CREATE
+ // and REPLACE TABLE convert their constraints before analysis finishes,
while
+ // `readsCurrentContext` is still false.
def toV2Constraint: Constraint = {
- val predicate = new V2ExpressionBuilder(child,
true).buildPredicate().orNull
+ val predicate = if (readsCurrentContext) {
Review Comment:
**With the predicate dropped, a CHAR(n) column compared with a shorter
literal loses its padding, so valid rows are rejected, or the check is silently
skipped.**
At ALTER time, `ApplyCharTypePadding` pads a literal that is compared with a
CHAR column, because the read-side projection keeps the CHAR metadata
([ApplyCharTypePadding.scala
L63-L78](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/ApplyCharTypePadding.scala#L63-L78),
[ApplyCharTypePaddingHelper.scala
L57](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyCharTypePaddingHelper.scala#L57),
[L164-L190](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ApplyCharTypePaddingHelper.scala#L164-L190)).
So before this PR the stored predicate was `status = 'OPEN ' OR closed <=
DATE '<ALTER day>'`, and only the date was wrong.
A write now re-parses `predicateSql` against the write query, where the
padding is lost:
- The output of that query has its CHAR metadata cleaned
([TableOutputResolver.scala
L289-L293](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala#L289-L293),
[L331](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala#L331)).
- The written value, however, is right-padded ([CharVarcharCodegenUtils.java
L37-L42](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/java/org/apache/spark/sql/catalyst/util/CharVarcharCodegenUtils.java#L37-L42)).
- `ResolveTableConstraints` also runs in the Resolution batch, before the
post-hoc `ApplyCharTypePadding` ([Analyzer.scala
L708-L709](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L708-L709),
[L747-L749](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L747-L749)).
From reading the code:
```sql
CREATE TABLE t (id INT, status CHAR(10), closed DATE);
ALTER TABLE t ADD CONSTRAINT c CHECK (status = 'OPEN' OR closed <=
current_date());
INSERT INTO t VALUES (1, 'OPEN', DATE'2999-01-01');
-- 'OPEN ' = 'OPEN' is false: CHECK_CONSTRAINT_VIOLATION, while the row
passed before this PR
```
With `status <> 'CLOSED' OR closed <= current_date()`, the first term is
always true, so the constraint is never enforced. `IN` and `NOT IN` behave the
same way.
CREATE TABLE constraints seem to have this problem already, since they
resolve against columns without the CHAR metadata ([ResolveCatalogs.scala
L158-L165](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveCatalogs.scala#L158-L165)),
so the general fix could be a separate JIRA. For this PR, could the
description list the CHAR case among the user-facing changes, or could a
condition that compares a CHAR column keep its predicate?
##########
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(
+ readsCurrentContext = !hasParameterMarker &&
CheckConstraint.readsCurrentContext(child))
Review Comment:
**A session or script variable in such a condition is no longer frozen, so
writes from other sessions fail, and a writer's own variable silently changes
the check.**
`markAsAnalyzed` keeps the frozen predicate only for legacy parameter
markers, and L139-L141 give the reason: a marker "stays in `condition`, which a
write cannot bind". The same holds for a session variable, a variable local to
a SQL script, and `IDENTIFIER(var)`, but nothing guards them.
- ALTER TABLE resolves a bare name to a variable inside the CHECK
([Analyzer.scala
L2060-L2062](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala#L2060-L2062),
[ColumnResolutionHelper.scala
L508-L510](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala#L508-L510)).
- The only check on the condition is determinism ([CheckAnalysis.scala
L1030-L1034](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala#L1030-L1034)).
- Before this PR, the optimizer folded the variable into the stored
predicate, since it is foldable ([VariableReference.scala
L42-L47](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/VariableReference.scala#L42-L47)).
Now the predicate is null, and every write re-parses the name
([ResolveTableConstraints.scala
L76-L80](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala#L76-L80)).
From reading the code:
```sql
DECLARE VARIABLE days INT DEFAULT 30;
ALTER TABLE t ADD CONSTRAINT recent CHECK (ts >= current_timestamp() -
make_interval(0, 0, 0, days));
-- predicate() is null, and predicateSql still says `days`
```
- Every INSERT, UPDATE or MERGE fails with UNRESOLVED_COLUMN when it runs in
another session, after `DROP TEMPORARY VARIABLE days`, or after the `BEGIN ...
END` script that declared the variable. Nobody can write to `t` until the
constraint is dropped.
- A writer that has its own `days` silently checks against its own value,
and a NULL value turns the check off.
- A column named `days` wins over the variable, whether it is added to `t`
later or comes from the source of `INSERT INTO t SELECT ... FROM src`
([ColumnResolutionHelper.scala
L50-L99](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala#L50-L99)).
- `CHECK (d <= CAST(v AS DATE))`, with a STRING variable `v` holding
'today', takes the same path without any `current_*` call, because L213-L215
evaluate the variable.
A temporary Hive UDF called with constant arguments behaves the same way, as
does an unqualified persistent or V2 function. Such a call was folded too
([hiveUDFs.scala
L63](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/hive/src/main/scala/org/apache/spark/sql/hive/hiveUDFs.scala#L63)),
and is now resolved again in the writer's session and current namespace.
Generated columns reject variables for this reason
([GeneratedColumnExpression.scala
L81-L86](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/GeneratedColumnExpression.scala#L81-L86)),
and so do DEFAULT values ([ResolveDefaultColumnsUtil.scala
L566-L573](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/ResolveDefaultColumnsUtil.scala#L566-L573)).
Could `markAsAnalyzed` also keep the predicate when the analyzed child
contains VARIABLE_REFERENCE? That would need:
- a parse-time check for UNRESOLVED_IDENTIFIER, since `IDENTIFIER(var)` is
gone by the end of analysis ([ResolveIdentifierClause.scala
L143-L156](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveIdentifierClause.scala#L143-L156));
- a test that inserts from `spark.newSession()`.
Rejecting variables in CHECK altogether would also cover CREATE TABLE, which
already re-parses such conditions. That could be a separate JIRA.
##########
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()
Review Comment:
Minor: this evaluates every constant string under any cast, before it looks
at the target type and without catching errors, so a condition that the
optimizer accepts can now fail the ALTER TABLE.
`ConstantFolding` does not fail on an error inside a branch that may not
run. It tags the expression and keeps it ([expressions.scala
L63-L80](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/expressions.scala#L63-L80)),
and `V2ExpressionBuilder` leaves it alone ([V2ExpressionBuilder.scala
L108-L119](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala#L108-L119)).
L213-L215 run `s.eval()` from `HandleSpecialCommand` without such a guard.
From reading the code, with ANSI on:
```sql
ALTER TABLE t ADD CONSTRAINT c CHECK (coalesce(id, CAST(CAST(1/0 AS STRING)
AS INT)) > 0);
ALTER TABLE t ADD CONSTRAINT c CHECK (qty < IF(false, CAST(CAST(1/0 AS
STRING) AS INT), 100));
-- Before this PR both succeeded; the first one needed no row with a NULL id.
-- Now both fail with DIVIDE_BY_ZERO during analysis.
```
The error is also thrown before `CheckAnalysis` runs ([HybridAnalyzer.scala
L280-L285](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/HybridAnalyzer.scala#L280-L285)).
So `CHECK (rand() > 0 OR CAST(CAST(1/0 AS STRING) AS INT) > 0)` reports
DIVIDE_BY_ZERO instead of NON_DETERMINISTIC_CHECK_CONSTRAINT.
Casts to DATE, TIMESTAMP and TIMESTAMP_NTZ already failed, because
`SpecialDatetimeValues` evaluates the string without a guard too
([finishAnalysis.scala
L316-L323](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala#L316-L323)).
Could the evaluation be wrapped per cast in `try ... catch { case NonFatal(_)
=> false }`, and maybe skipped when `to` is not a datetime type?
##########
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(
+ readsCurrentContext = !hasParameterMarker &&
CheckConstraint.readsCurrentContext(child))
+
+ // ALTER TABLE builds the constraint from the optimized plan, where the
current time, date,
+ // catalog, user and the like are replaced by their values at the time of
the ALTER TABLE. A
+ // condition that `readsCurrentContext` finds reading one is stored without
a predicate, so that
+ // writes check `predicateSql`.
+ // The predicate is not built from the analyzed condition, as the default
value of a column is,
+ // because `V2ExpressionBuilder` folds `CAST('today' AS DATE)` there as an
ordinary cast. CREATE
+ // and REPLACE TABLE convert their constraints before analysis finishes,
while
+ // `readsCurrentContext` is still false.
def toV2Constraint: Constraint = {
- val predicate = new V2ExpressionBuilder(child,
true).buildPredicate().orNull
+ val predicate = if (readsCurrentContext) {
+ null
Review Comment:
Minor: when a re-parsed condition uses `current_timestamp`, `current_date`
or `current_user` without parentheses, the violation message now lists them as
columns.
- By default these keywords parse as an `UnresolvedAttribute`, and resolve
to the function only when no column matches ([AstBuilder.scala
L4082-L4090](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala#L4082-L4090),
[ColumnResolutionHelper.scala
L164](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ColumnResolutionHelper.scala#L164)).
- `buildColumnExtractors` treats every `UnresolvedAttribute` as a column
([ResolveTableConstraints.scala
L137-L138](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala#L137-L138)).
- `checkViolation` prints each of them ([QueryExecutionErrors.scala
L3469-L3472](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala#L3469-L3472)).
From reading the code:
```sql
ALTER TABLE t ADD CONSTRAINT c CHECK (created <= current_timestamp);
INSERT INTO t VALUES (1, current_timestamp() + INTERVAL 1 DAY);
-- values: " - created : <micros>" and " - current_timestamp : <micros>",
-- and the second value changes on every run
```
Before this PR, the stored predicate referenced only real columns
([V2ExpressionUtils.scala
L267](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/V2ExpressionUtils.scala#L267)),
so only `created` was listed. CREATE TABLE constraints already behave this
way, and variable names show up in the same way.
Could `buildColumnExtractors` keep only the extractors that resolve to
columns or fields, perhaps under a separate JIRA? Could a test also cover the
paren-less form?
##########
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:
Nit: this field holds "store no predicate" rather than "reads the current
context", and it shares its name with the method at L208.
`markAsAnalyzed` stores `!hasParameterMarker &&
CheckConstraint.readsCurrentContext(child)`, and L154 is its only reader. So
under the legacy parameter substitution, `ts >= current_timestamp() -
make_interval(0, 0, 0, :d)` reads the current time while the field is false, as
the new parameter-marker test asserts. In the comment at L145-L152, the first
`readsCurrentContext` means the method and the second one means the field.
Could the field keep the plain result, with `toV2Constraint` checking
`readsCurrentContext && !hasParameterMarker` next to the marker comment? Or
could it be renamed to something like `omitPredicate`?
Also, unlike `DefaultValueExpression` and `GeneratedColumnExpression`
([ColumnDefinition.scala
L275](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/ColumnDefinition.scala#L275),
[GeneratedColumnExpression.scala
L44](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/GeneratedColumnExpression.scala#L44)),
there is no `stringArgs` override. So the analyzed plan of the ALTER TABLE,
for example in EXPLAIN EXTENDED, now prints both flags, ending with `, false,
true)`.
##########
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:
Nit: the ANSI loop doubles this test's runtime without reaching another
branch, and L206 repeats L195.
Nothing in the new detection reads the eval mode:
- L213 matches any `Cast`.
- `to_timestamp('now')` becomes a `Cast` in both modes
([datetimeExpressions.scala
L3261-L3264](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/datetimeExpressions.scala#L3261-L3264)).
- Every row compares values of the same type.
The ALTER TABLE and the INSERTs also run under the same `withSQLConf`, so
the loop does not cover a writer whose settings differ (SPARK-60453). L206
checks the same `predicateSql` as L195, since the message's `expression` comes
from it ([ResolveTableConstraints.scala
L52](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala#L52)).
Could the loop and L206 be dropped? Or could the loop stay, with a comment
on what it guards?
##########
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] =>
Review Comment:
Minor: the detection covers only a cast from a string or TIME whose
top-level target is a timestamp type. Casts that read the current date per
element or field still store the ALTER-time value.
- A cast into an ARRAY, MAP or STRUCT of timestamps casts each element with
the scalar cast ([Cast.scala
L1491-L1505](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L1491-L1505),
[L1517-L1520](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L1517-L1520)).
- That scalar cast completes a time-only string with `LocalDate.now`
([SparkDateTimeUtils.scala
L906-L911](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L906-L911)),
and a TIME value with the current date ([Cast.scala
L923-L930](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L923-L930)).
- L213 requires a string source. `ComputeCurrentTime` rewrites only
top-level TIME casts, since `CAST_TO_TIMESTAMP` is keyed on the top-level
target ([Cast.scala
L701-L714](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L701-L714)).
- A VARIANT source goes through the same string cast ([Cast.scala
L1545-L1548](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L1545-L1548),
[variantExpressions.scala
L743-L746](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/variant/variantExpressions.scala#L743-L746)).
From reading the code, each of these is folded at ALTER time into a literal
that the predicate keeps:
```sql
CHECK (ts <= CAST(array('23:59:59') AS ARRAY<TIMESTAMP>)[0])
CHECK (ts >= CAST(named_struct('t', '08:00:00') AS STRUCT<t: TIMESTAMP>).t)
CHECK (ts <= CAST(parse_json('"23:59:59"') AS TIMESTAMP))
CHECK (ts <= CAST(coalesce('23:59:59', s) AS TIMESTAMP)) -- foldable only
after NullPropagation
```
These were frozen before this PR too, so this is not a regression, and the
inputs are unusual. Could the comment at L220-L224 or the description say that
only top-level casts are detected? Or could the check also look at nested types?
##########
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:
Minor: the description says typed literals and conditions folded under the
legacy `earlyEvalCurrentTime` are "still stored with their values at the time
of the ALTER TABLE". That holds only when nothing else in the condition reads
the current context.
`markAsAnalyzed` decides for the whole condition, and a write re-parses the
whole `predicateSql`. At that point the parser folds typed literals again
([AstBuilder.scala
L4775-L4778](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala#L4775-L4778),
[L4815-L4818](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala#L4815-L4818)),
and the legacy branch of `ResolveTableSpec` applies only `ComputeCurrentTime`
([ResolveTableSpec.scala
L39-L45](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableSpec.scala#L39-L45)).
From reading the code:
- `CHECK (d >= DATE 'today' - 7)` stays at the ALTER date. But `CHECK (d >=
DATE 'today' - 7 AND ts <= current_timestamp())` gets no predicate, so each
write folds `DATE 'today'` to its own date.
- With `spark.sql.legacy.earlyEvalCurrentTime=true`, `CHECK (ts <=
current_timestamp())` stays frozen. But `CHECK (ts <= current_timestamp() AND
owner = current_user())` gets no predicate because of `current_user()`, so
`current_timestamp()` is evaluated at each write too.
This extends the earlier thread on typed literals, and the resulting
behavior is what users want. Could the description add "unless another part of
the condition reads the current context"? Otherwise the tests for SPARK-60452
and SPARK-60456 may expect the frozen value.
##########
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:
Nit: `values` is deterministic here (` - db : ns`). So `checkError` with
`sqlState` and `parameters` would pin the whole message, as the existing
current_database test at L709-L718 does.
As written, a regression that reports the wrong column or value on the
re-parse path would still pass. The time-dependent rows of the first test can
keep the current style; L577-L579 does the same.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala:
##########
@@ -852,6 +852,32 @@ object DateTimeUtils extends SparkDateTimeUtils {
}
}
+ /**
+ * Whether `input` is one of the special values that read the current date
or time: 'now',
+ * 'today', 'tomorrow' and 'yesterday'.
+ *
+ * @param input A string to check. It can contain trailing or leading
whitespaces.
+ */
+ private[sql] def isCurrentSpecialValue(input: String): Boolean = {
+ extractSpecialValue(input.trim).exists {
+ case "now" | "today" | "tomorrow" | "yesterday" => true
+ case _ => false
+ }
+ }
+
+ /**
+ * Whether `input` is a timestamp string with only a time part, which a cast
to a timestamp with
+ * local time zone completes with the current date.
+ */
+ private[sql] def isTimeOnlyTimestamp(input: UTF8String): Boolean = {
+ try {
+ val (segments, _, justTime) = parseTimestampString(input)
+ segments.nonEmpty && justTime
Review Comment:
Nit: this also returns true for out-of-range times such as '24:00:00',
'23:59:60' or '12:60'. The cast rejects those without reading the date.
`parseTimestampString` only counts digits ([SparkDateTimeUtils.scala
L718](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L718),
[L765-L773](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L765-L773)).
The range check is `LocalTime.of` in the cast path, which runs before
`LocalDate.now`
([L906-L911](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L906-L911)),
and a failure there gives None
([L922-L930](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/api/src/main/scala/org/apache/spark/sql/catalyst/util/SparkDateTimeUtils.scala#L922-L930)).
So the Scaladoc promises more than the code checks.
The effect is small but not zero:
- With ANSI off, `CHECK (ts IS NULL OR ts <= CAST('25:00:00' AS TIMESTAMP))`
loses its predicate (`isnull(ts) OR null`).
- A writer with ANSI on then fails every INSERT with CAST_INVALID_INPUT,
even for a NULL `ts`.
Could this also require `stringToTimestamp(input, ZoneOffset.UTC).isDefined`
inside the `try`? And could '24:00:00' and '12:60' be added to the negative
cases of the new test?
##########
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:
Nit: this copies the guard and the target types of `SpecialDatetimeValues`.
If that rule changes, the detection drifts silently, and on drift it falls back
to freezing.
L213 repeats the guard at [finishAnalysis.scala
L316-L318](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala#L316-L318),
and L226-L229 restate its `conv` map
([L298-L301](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/finishAnalysis.scala#L298-L301)).
Nothing there points back here.
Both lists do change:
- SPARK-59112 changed this guard from `== StringType` to
`isInstanceOf[StringType]`.
- `ComputeCurrentTime` already handles the nanosecond timestamp types.
If `conv` gains `TimestampLTZNanosType`, `CAST('now' AS TIMESTAMP_LTZ(9))`
is folded at ALTER time. L229 checks only time-only strings, so the predicate
is frozen again, and no test fails. `Cast` warns about the same hazard for
`CAST_TO_TIMESTAMP` ([Cast.scala
L701-L704](https://github.com/apache/spark/blob/605deea50b574d87d32903936e9fbf5d80c36820/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala#L701-L704)).
Could the special-value part live next to `SpecialDatetimeValues`? For
example, a `private[sql]` helper could say whether the rule rewrites a cast to
a value other than 'epoch'. Or could a test check `castReadsCurrentDate`
against `SpecialDatetimeValues` for every datetime type?
--
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]
