dongjoon-hyun commented on code in PR #59217:
URL: https://github.com/apache/spark/pull/59217#discussion_r4199939279
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -147,6 +147,21 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-59962: Predicate of a check reading a computed value twice") {
+ // NULLIF reads `id + 1` twice. The predicate stored in the catalog is
built from the check as
+ // the optimizer leaves it, so it must not refer to a column pre-evaluated
under the command.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (nullif(id + 1, 3) > 0)")
+ val constraint = getCheckConstraint(loadTable(nonPartitionCatalog, "ns",
"tbl"))
+ assert(constraint.predicateSql() == "nullif(id + 1, 3) > 0")
+ assert(constraint.predicate() != null)
Review Comment:
**This assertion fails in the scheduled Non-ANSI build: `id + 1` translates
into a V2 expression only in ANSI mode.**
`build_non_ansi.yml` (L42) runs the sql tests with
`SPARK_ANSI_SQL_MODE=false`, which turns `spark.sql.ansi.enabled` off by
default (`SQLConf.scala` L6014). `id + 1` is then an `Add` in
`EvalMode.LEGACY`, and `V2ExpressionBuilder` translates `Add` only in ANSI mode
(L80), so with the definition inlined `toV2Constraint` stores a null predicate.
Before this PR the predicate read `_common_expr_N`, which translates in both
modes, so the failure is new. Neither this suite nor `CommandSuiteBase` pins
ANSI, unlike `DataSourceV2SQLSuite` (L62-63), where the new DELETE test is
fine. The PR's CI runs with ANSI on, so this shows up only after the merge;
ec120fddcf6 had to add the same guard to earlier tests of V2 translation.
Could the body, including the ALTER, run under
`withSQLConf(SQLConf.ANSI_ENABLED.key -> "true")`, or assert the enforcement
with INSERTs instead (my comment on L150), which doesn't depend on the mode?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. A
+ // definition hoisted into a child would leave them reading a column
only that `Project`
+ // produces, or leave the command over a child its planner does not
expect, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
Review Comment:
**This also inlines `groupFilterCondition`, which nothing translates: only
the runtime group filter reads it, and it evaluates it.**
This is about item 2 of my previous review and the reply that keeping a
`With` in that filter "would need the condition in two forms, since the same
expression is also translated for pushdown". `ReplaceData` and `WriteDelta`
already hold two forms. `condition` is what
`GroupBasedRowLevelOperationScanPlanning` pushes (L48, L56) and
`OptimizeMetadataOnlyDeleteFromTable` translates (L116). `groupFilterCondition`
(`v2Commands.scala` L390, L480), set from the same `cond` by
`RewriteDeleteFromTable` (L92) and `RewriteUpdateTable` (L82, L117), is read
only by `RowLevelOperationRuntimeGroupFiltering` (L56), which wraps it in
`Filter(cond, relation)` (L105) and runs `optimizeSubqueries` on it (L95), and
so this rule again. A `With` left there would be pre-evaluated under that
filter, and `spark.sql.optimizer.avoidDoubleFilterEval`, on by default, keeps
an expensive alias such as a Scala UDF memoized.
Inlining it also regresses a statement that works on master. On a
group-based table, `DELETE FROM t WHERE CASE WHEN dep = 'hr' THEN my_udf(id)
BETWEEN 2 AND 4 ELSE false END` keeps a memoizing `With` in both slots on
master (the `ConditionalExpression` path, L391-402), so the runtime group
filter calls the UDF once per row. Here it is called up to twice per row that
reaches the branch, and subexpression elimination doesn't merge two uses in one
branch (`EquivalentExpressions.scala` L171, L181).
Could this case inline only `condition` of `ReplaceData` and `WriteDelta`,
and leave `groupFilterCondition` to the rewrite of the injected filter? That
would also save the runtime filter's extra calls in the top-level BETWEEN you
measured.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -147,6 +147,21 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-59962: Predicate of a check reading a computed value twice") {
Review Comment:
Minor: this test checks the shape of the stored predicate, but not what
users hit on master. From reading the code, after this ALTER every later write
to the table fails there: `ResolveTableConstraints` builds the write-time check
from `c.predicate()` before `predicateSql` (L77), and
`V2ExpressionUtils.toCatalyst` turns the stored `_common_expr_1` into an
`UnresolvedAttribute` (`V2ExpressionUtils.scala` L260), so INSERT, UPDATE,
MERGE and a group-based DELETE fail analysis with UNRESOLVED_COLUMN. The
description mentions only the stored predicate.
Could the test also insert a row the check accepts and one it rejects, e.g.
`(2, 'a')`, where `nullif(3, 3)` is NULL and a CHECK accepts it, and `(-5,
'b')`, which fails with CHECK_CONSTRAINT_VIOLATION? That fails on master and
passes here in both ANSI modes.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala:
##########
@@ -456,6 +456,23 @@ class GeneratedColumnWriteSuite extends QueryTest with
DatasourceV2SQLBase {
}
}
+ test("SPARK-59962: CREATE TABLE with a generation expression reading a value
twice") {
Review Comment:
Minor: REPLACE TABLE fails the same way on master and is fixed by this PR as
well, but isn't tested. `ResolveCatalogs` gives `ReplaceTable` the same
column-output `ResolvedIdentifier` as `CreateTable` (L123-125), `ReplaceTable`
is a `UnaryCommand`, and `DataSourceV2Strategy` plans it only over a
`ResolvedIdentifier` (L312-313), so on master `CREATE OR REPLACE TABLE
testcat.my_tab (a INT, b INT, c INT GENERATED ALWAYS AS (nullif(a + b, 3)))
USING foo` should hit "No plan for ReplaceTable". If SPARK-59994 turns the
`CreateVariable` check into a per-command property and misses `ReplaceTable`,
every test would still pass. Could this test run for both `CREATE TABLE` and
`CREATE OR REPLACE TABLE`?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. A
+ // definition hoisted into a child would leave them reading a column
only that `Project`
+ // produces, or leave the command over a child its planner does not
expect, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
Review Comment:
**A CHECK predicate stored by ALTER TABLE is evaluated after all, too: every
later write reads it back and checks it per row.**
`ResolveTableConstraints` adds a `CheckInvariant` filter to each
`V2WriteCommand` (L39-61), built from `c.predicate()` before `predicateSql`
(L77), and `V2ExpressionUtils.toCatalyst` round-trips this PR's CASE WHEN, `+`,
`=` and `>`. So the inlined form is what every later INSERT, UPDATE and MERGE
evaluates, at each reference, and the catalog keeps it. For one top-level
NULLIF, which fails on master, that costs little. But it regresses a definition
in a branch: for `ALTER TABLE t ADD CONSTRAINT c CHECK (CASE WHEN id > 0 THEN
nullif(id + 1, 3) > 0 ELSE true END)`, master keeps the `With` in the branch,
`V2ExpressionBuilder` returns None for it, and every write re-parses
`predicateSql`, where NULLIF memoizes; here the inlined predicate is stored and
`id + 1` is computed twice per written row. The same CHECK in CREATE TABLE
still takes the `predicateSql` path (`ResolveTableSpec.scala` L110).
Could the comment name this path too, or could the predicate be built from
the analyzed expression (my comment on `CheckConstraintSuite` L152)?
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -147,6 +147,21 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-59962: Predicate of a check reading a computed value twice") {
+ // NULLIF reads `id + 1` twice. The predicate stored in the catalog is
built from the check as
+ // the optimizer leaves it, so it must not refer to a column pre-evaluated
under the command.
+ withNamespaceAndTable("ns", "tbl", nonPartitionCatalog) { t =>
+ sql(s"CREATE TABLE $t (id bigint, data string) $defaultUsing")
+ sql(s"ALTER TABLE $t ADD CONSTRAINT c1 CHECK (nullif(id + 1, 3) > 0)")
Review Comment:
Minor: a table that got such a constraint on 4.1 keeps failing every write
after upgrading, since this PR changes only what a new ALTER stores. v4.1.0
already hoists under `AddCheckConstraint`, builds the predicate from the
optimized plan, and prefers it on writes (`ResolveTableConstraints.scala`
L70-72 there). A catalog that persists the V2 predicate keeps returning `CASE
WHEN _common_expr_1 = 3 ...`, `toCatalyst` turns `_common_expr_1` into an
`UnresolvedAttribute` instead of failing, the `predicateSql` fallback is never
reached, and INSERT, UPDATE and MERGE keep failing with UNRESOLVED_COLUMN.
In-tree only `InMemoryTableCatalog` keeps the predicate object, and dropping
and re-adding the constraint works around it, so a note may be enough. Would
falling back to `predicateSql` when the converted predicate names a column the
table doesn't have be worth a separate JIRA?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. A
+ // definition hoisted into a child would leave them reading a column
only that `Project`
+ // produces, or leave the command over a child its planner does not
expect, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
+ case w: With =>
+ val multiplyReferenced = multiplyReferencedIds(w.child, w.defs)
+ inlineDefs(w)(d => d.child.deterministic ||
!multiplyReferenced.contains(d.id))
Review Comment:
**Nested NULLIF/BETWEEN grow as 2^k here, since every level is inlined in
one bottom-up pass whatever it costs.**
Both builders read their definition twice (`nullExpressions.scala` L203-204,
`Between.scala` L49-50), and the inner `With` is inlined first, so each
enclosing definition copies the already expanded tree into both of its
references. `nullif(nullif(nullif(trim(s), ''), 'NA'), 'N/A') IS NULL` puts 8
copies of `trim(s)` into `condition` and `groupFilterCondition`, into the V2
predicates handed to `canDeleteWhere` and the scan, and into a CHECK predicate
stored by ALTER TABLE; 10 levels give 1024. V2 translation builds every copy,
and subexpression elimination can't merge NULLIF's second read, which sits in a
branch. The generic path stays linear by handling one level per iteration
(L320).
Most of these statements fail on master, so there this is the cost of the
fix, the same tree `spark.sql.alwaysInlineCommonExpr=true` builds. Two that
work on master grow as well, at planning time: a delta-based DELETE, and a
group-based DELETE without runtime group filtering. Could the comment above say
that the copies compound with nesting, and would it be worth keeping the `With`
past a small size?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. A
+ // definition hoisted into a child would leave them reading a column
only that `Project`
+ // produces, or leave the command over a child its planner does not
expect, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
Review Comment:
Minor: keying this case on `Command` leaves `Call`, which isn't a `Command`,
with the same hoisting. This relates to item 5 of my previous review and
SPARK-59994. `Call` is a `UnaryNode` (`v2Commands.scala` L2340),
`InvokeProcedures` runs the optimizer on it during analysis and requires
`Call(ResolvedProcedure(...), ...)` back (`InvokeProcedures.scala` L38-42), and
`ResolvedProcedure` has no output, so the generic path hoists a definition
without references into a `Project` over it (L339-341). With `ProcedureSuite`'s
`cat.ns.sum` and a Scala UDF `my_udf`, `CALL cat.ns.sum(if(nullif(my_udf(1), 2)
IS NULL OR true, 1, 0), 1)` fails with INTERNAL_ERROR "Unexpected plan for
optimized CALL statement": the arguments fold to literals, but the `Project`
stays. It should work with `spark.sql.alwaysInlineCommonExpr=true`. The trigger
is contrived, since an argument that doesn't fold fails "args must be foldable"
anyway (L81 there), so a note in SPARK-59994, or extending this case to `Call`,
would be enough.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. A
+ // definition hoisted into a child would leave them reading a column
only that `Project`
+ // produces, or leave the command over a child its planner does not
expect, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
Review Comment:
Minor: this points to `canSubstitute` for a narrower gap than this case has,
and the selector at L87 is `canSubstitute` without its `isCheap` term, written
out again. This is about the comment added for item 1 of my previous review.
`canSubstitute`'s note (L239-243) covers an impure *foldable* definition,
because `isCheap` admits foldables. This case also duplicates impure
definitions that read columns, which `canSubstitute` memoizes as not cheap:
`aes_encrypt(data, k)`, the example in that item, is a deterministic
`StaticInvoke` that isn't foldable over a column. And since L87 doesn't go
through `canSubstitute`, fixing purity "in one purity predicate shared with
`isSafeToDuplicate`", as that note suggests, would leave this case behind, with
nothing there pointing here. Could the selector go through `canSubstitute` with
a flag that skips the cost check, or could this comment say that it includes
definitions reading columns, and `canSubstitute` point back here?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. A
+ // definition hoisted into a child would leave them reading a column
only that `Project`
+ // produces, or leave the command over a child its planner does not
expect, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
Review Comment:
Nit: `c.mapExpressions(_.transformUpWithPruning(...))` is what
`QueryPlan.transformExpressionsUpWithPruning` does (`QueryPlan.scala`
L250-254), so
`c.transformExpressionsUpWithPruning(_.containsPattern(WITH_EXPRESSION)) { case
w: With => ... }` would read like the other rules.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpression.scala:
##########
@@ -67,6 +69,23 @@ object RewriteWithExpression extends Rule[LogicalPlan] {
// changed anything: `mapExpressions` and `withNewChildren` preserve
reference equality when
// they rewrite nothing, which is what makes this detectable.
if ((rewrittenAgg eq agg) && (rewrittenProj eq proj)) p else
rewrittenProj
+ // A command's own expressions are not evaluated over its children's
rows: most are stored as
+ // metadata or turned into source predicates, neither of which can hold
a memoized value. A
+ // definition hoisted into a child would leave them reading a column
only that `Project`
+ // produces, or leave the command over a child its planner does not
expect, so a
+ // deterministic one is substituted whatever it costs. Where the
expression is evaluated
+ // after all, as by the runtime group filter of a row-level command,
that evaluates it at
+ // every reference, and it duplicates an impure one that declares itself
deterministic (see
+ // `canSubstitute`). Analysis lets no command but `CreateVariable` hold
a nondeterministic
+ // expression; one read more than once keeps its `With`.
`CreateVariable` is left out: its
+ // default is evaluated rather than stored, so a definition there should
be memoized.
+ case c: Command if !c.isInstanceOf[CreateVariable] &&
+ c.expressions.exists(_.containsPattern(WITH_EXPRESSION)) =>
+
c.mapExpressions(_.transformUpWithPruning(_.containsPattern(WITH_EXPRESSION)) {
+ case w: With =>
+ val multiplyReferenced = multiplyReferencedIds(w.child, w.defs)
Review Comment:
Nit: this comes from the change for item 6 of my previous review.
`multiplyReferencedIds` walks `w.child` and every definition (L258-273) for
each `With`, but the selector reads the result only for a nondeterministic
definition, which analysis keeps out of these commands, so it is computed for
nothing in practice, and with nested NULLIF/BETWEEN it walks the expanded tree
once more. A `lazy val` would skip it without changing the result.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/DataSourceV2SQLSuite.scala:
##########
@@ -2683,6 +2683,23 @@ class DataSourceV2SQLSuiteV1Filter
}
}
+ test("SPARK-59962: DeleteFrom: a condition reading a value twice reaches the
source inlined") {
+ // BETWEEN reads `id + p` twice. Pre-evaluating it would put a `Project`
where the planner
+ // expects the table, so the condition has to reach the source inlined.
+ val t = "testcat.ns1.ns2.tbl"
+ withTable(t) {
+ sql(s"CREATE TABLE $t (id bigint, data string, p int) USING foo
PARTITIONED BY (id, p)")
+ sql(s"INSERT INTO $t VALUES (2L, 'a', 2), (2L, 'b', 3), (3L, 'c', 3)")
+ checkError(
+ exception = analysisException(s"DELETE FROM $t WHERE (id + p) BETWEEN
0 AND 4"),
+ condition = "_LEGACY_ERROR_TEMP_1110",
+ parameters = Map(
+ "table" -> "testcat.ns1.ns2.tbl",
+ "filters" -> "[(id + CAST(p AS long)) >= 0, (id + CAST(p AS long))
<= 4]"))
+ assert(spark.table(t).count() === 3)
Review Comment:
Nit: this test repeats "DeleteFrom: - delete with invalid predicate" above
it except for the condition and `filters`, and this count can't fail once
`checkError` has passed: `_LEGACY_ERROR_TEMP_1110` is thrown at planning, from
`canDeleteWhere`, which looks only at the filters (`DataSourceV2Strategy.scala`
L570). Could the two share one test over `(condition, filters)` pairs, or could
this one drop the INSERT and the count?
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/RewriteWithExpressionSuite.scala:
##########
@@ -486,6 +486,53 @@ class RewriteWithExpressionSuite extends PlanTest {
)
}
+ test("SPARK-59962: a With in a command's condition is inlined instead of
hoisted") {
+ val a = testRelation.output.head
+ // The condition is translated into source predicates, not evaluated over
the child's rows, so a
+ // `Project` under the command would sit where the planner expects the
table.
+ val condition = With(a + a) { case Seq(ref) => ref < 10 && ref > 0 }
+ comparePlans(
+ Optimizer.execute(DeleteFromTable(testRelation, condition)),
+ DeleteFromTable(testRelation, (a + a) < 10 && (a + a) > 0))
+
+ // Nor is one in a conditional branch kept: a `With` translates to no
source predicate.
+ val inBranch = If(a > 0, With(a + a) { case Seq(ref) => ref < 10 && ref >
0 }, Literal(false))
Review Comment:
Nit: this rebuilds `condition` from L493, L502 repeats the expected
expression of L496, and `Rand(Literal(0L)) + a` is built three times at
L509-517. `If(a > 0, condition, Literal(false))` and a shared `val` for the
expected expression would show directly that the second case is the first one
moved into a branch.
##########
sql/core/src/test/scala/org/apache/spark/sql/execution/command/v2/CheckConstraintSuite.scala:
##########
@@ -147,6 +147,21 @@ class CheckConstraintSuite extends QueryTest with
CommandSuiteBase with DDLComma
}
}
+ test("SPARK-59962: Predicate of a check reading a computed value twice") {
+ // NULLIF reads `id + 1` twice. The predicate stored in the catalog is
built from the check as
+ // the optimizer leaves it, so it must not refer to a column pre-evaluated
under the command.
Review Comment:
**The CHECK part of this fix sits one level too low: `toV2Constraint` builds
the stored predicate from the optimized expression, which is what let
`_common_expr_N` in, and this test pins that design.**
`CheckConstraint.toV2Constraint` translates its current `child`
(`constraints.scala` L131-132), and ALTER TABLE calls it on the optimized plan
(`DataSourceV2Strategy.scala` L810-817). `DefaultValueExpression` and
`GeneratedColumnExpression` avoid this as `AnalysisAwareExpression`s whose
`toV2` uses `analyzedChild` (`ColumnDefinition.scala` L284,
`GeneratedColumnExpression.scala` L135), for the reason the trait's doc gives:
the optimizer folds context-dependent functions into definition-time literals
(`AnalysisAwareExpression.scala` L21-25).
That reason already bites here, before this PR. From reading the code,
`ALTER TABLE t ADD CONSTRAINT valid_time CHECK (creation_time <=
current_timestamp())` stores the ALTER time as a literal
(`ComputeCurrentTime`), and since writes prefer `predicate()`
(`ResolveTableConstraints.scala` L77), a later `INSERT INTO t VALUES (1,
current_timestamp())` fails with CHECK_CONSTRAINT_VIOLATION, while the message
shows `current_timestamp()`. `current_user()` is frozen the same way. The
`current_*` tests at L410-645 use CREATE TABLE only, whose predicate
`ResolveTableSpec` builds during analysis (L110), as peter-toth noted.
Building from the analyzed expression would fix both, but it makes this
constraint's predicate null, as it already is for CREATE TABLE, since
`V2ExpressionBuilder` doesn't translate `NullIf`. Could this test assert the
enforcement instead (my comment on L150), rather than `predicate() != null` and
"as the optimizer leaves it", and the analyzed-expression change go to a
separate JIRA?
--
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]
