LuciferYang commented on code in PR #60035:
URL: https://github.com/apache/spark/pull/60035#discussion_r4451126170
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala:
##########
@@ -337,6 +337,23 @@ class GeneratedColumnWriteSuite extends QueryTest with
DatasourceV2SQLBase {
}
}
+ test("SPARK-60817: INSERT by name computes generated column from the stored
(post-cast) " +
+ "value in a lambda") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a ARRAY<INT>,
+ | i INT,
+ | g1 ARRAY<INT> GENERATED ALWAYS AS (transform(a, x -> x + i *
10)),
+ | g2 ARRAY<ARRAY<INT>> GENERATED ALWAYS AS (transform(a, i ->
transform(a, x -> i)))
+ |) USING foo""".stripMargin)
+ // 5.7 is cast to INT (= 5) when stored in `i`. `g1` must read the
stored value (1 + 5 * 10),
+ // not the pre-cast value (1 + 5.7 * 10). In `g2`, `i` is the lambda
variable, not the column.
+ sql(s"INSERT INTO testcat.$tblName(a, i) SELECT array(1), 5.7D")
Review Comment:
The new test has no DEFAULT column, so the default-value case this PR fixes
is not covered. The description also understates when this failed before the
fix: when the DEFAULT column is declared after the generated column, the write
failed with `UNRESOLVED_COLUMN` under the default configuration too. For
example, `(a ARRAY<INT>, g ARRAY<INT> GENERATED ALWAYS AS (transform(a, x -> x
+ d)), d INT DEFAULT 5)` with `INSERT INTO t(a) SELECT array(1)` now succeeds
and stores `[6]`.
Could you add a test where `d` is declared after `g`, and update that
sentence in the description?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala:
##########
@@ -831,15 +831,26 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
genExpr: Expression,
matched: Seq[NamedExpression],
conf: SQLConf): Expression = {
- genExpr.transform {
- case u @ UnresolvedAttribute(name +: nestedFields) =>
- matched.collectFirst {
- case alias: Alias if conf.resolver(alias.name, name) =>
- nestedFields.foldLeft(alias.child) { (child, field) =>
- UnresolvedExtractValue(child, Literal(field))
- }
- }.getOrElse(u)
+ def storedValue(nameParts: Seq[String]): Option[Expression] =
matched.collectFirst {
+ case alias: Alias if conf.resolver(alias.name, nameParts.head) =>
+ nameParts.tail.foldLeft(alias.child) { (child, field) =>
+ UnresolvedExtractValue(child, Literal(field))
+ }
}
+
+ // The parser turns every reference in a lambda function into a lambda
variable, so a lambda
+ // variable that no enclosing lambda function declares refers to a column.
+ def resolve(e: Expression, lambdaVars: Seq[String]): Expression = e match {
+ case u: UnresolvedAttribute => storedValue(u.nameParts).getOrElse(u)
+ case v: UnresolvedNamedLambdaVariable
+ if !lambdaVars.exists(conf.resolver(_, v.nameParts.head)) =>
Review Comment:
This uses `conf.resolver` to decide whether a name is a lambda variable,
while `ResolveLambdaVariables` uses `conf.canonicalize`, and the two disagree
on characters such as `ı` and `İ`. I tried two cases locally. With a lambda
parameter named `ı`, the rewrite skips the column `i` in the body, so it stores
`[58]` instead of `[51]`. With a column named `i̇` and a parameter named `İ`, a
name that the analyzer binds to the lambda variable is replaced with the column
value, so it now stores `[70]` instead of `[10]`.
Matching the analyzer fixes both: keep `conf.canonicalize(a.name)` in
`lambdaVars` and check
`!lambdaVars.contains(conf.canonicalize(v.nameParts.head))`. If the expression
is resolved against the stored values as suggested in the comment on line 847,
this check goes away.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala:
##########
@@ -831,15 +831,26 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
genExpr: Expression,
matched: Seq[NamedExpression],
conf: SQLConf): Expression = {
- genExpr.transform {
- case u @ UnresolvedAttribute(name +: nestedFields) =>
- matched.collectFirst {
- case alias: Alias if conf.resolver(alias.name, name) =>
- nestedFields.foldLeft(alias.child) { (child, field) =>
- UnresolvedExtractValue(child, Literal(field))
- }
- }.getOrElse(u)
+ def storedValue(nameParts: Seq[String]): Option[Expression] =
matched.collectFirst {
+ case alias: Alias if conf.resolver(alias.name, nameParts.head) =>
+ nameParts.tail.foldLeft(alias.child) { (child, field) =>
+ UnresolvedExtractValue(child, Literal(field))
+ }
}
+
+ // The parser turns every reference in a lambda function into a lambda
variable, so a lambda
+ // variable that no enclosing lambda function declares refers to a column.
+ def resolve(e: Expression, lambdaVars: Seq[String]): Expression = e match {
+ case u: UnresolvedAttribute => storedValue(u.nameParts).getOrElse(u)
+ case v: UnresolvedNamedLambdaVariable
+ if !lambdaVars.exists(conf.resolver(_, v.nameParts.head)) =>
+ storedValue(v.nameParts).getOrElse(v)
Review Comment:
The stored-value expression of a referenced collection column is now inlined
into the lambda body, so it is recomputed for every element of the outer array.
For example, when an `ARRAY<STRING>` is written into `b ARRAY<VARCHAR(10)>`,
`transform(a, x -> x + size(b))` goes from O(|a|+|b|) to O(|a|·|b|) per row.
Locally, 100 rows with 2000 elements in each array took 3781 ms to write,
versus 92 ms before this PR.
Could the generation expression be resolved against the stored values
instead of substituting `alias.child`? For by-name writes, wrap the query in
`Project(storedCols, query)` and let the outer Project reference its output, so
the analyzer binds the names itself. I tried a draft of this locally: this PR's
test and the cases in the description give the right values, the write above
drops to 47 ms, and a column reference that the rewrite misses today, such as
`IDENTIFIER('i')`, also reads the stored value.
--
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]
