viirya commented on code in PR #59965:
URL: https://github.com/apache/spark/pull/59965#discussion_r4450626924
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -4122,11 +4124,45 @@ class Analyzer(
cleanedRelation
case other => other
}
- v2Write.withNewQuery(projection).withNewTable(cleanedTable)
+ v2Write
+ .withNewQuery(analyzeAutoFilledGeneratedColumns(
+ projection,
+ autoFilledGenCols,
+ resolveOutput(StoredExpressionText.parseWithSessionAnsi)._1))
+ .withNewTable(cleanedTable)
} else {
v2Write
}
}
+
+ /**
+ * Analyzes the auto-filled generated columns of a write under the
settings their stored text
+ * is read under, so that the implicit casts and the functions of a
generation expression do
+ * not follow the session that writes.
+ *
+ * A column these settings cannot resolve comes from `readInSession`,
where its text is read
+ * with the ANSI value of the session, and the analysis of the write
resolves it under the
+ * session. So does a text a catalog stores that analyzes to another plan.
+ */
+ private def analyzeAutoFilledGeneratedColumns(
Review Comment:
This fallback ladder is also implemented in `analyzeGeneratedColumnCheck`
and its caller, once wrapped in a `Project` and once in a `Filter`. Findings 1,
5, 9 and 16 all came from the two copies drifting apart. Could both go through
one helper in `StoredExpressionText` that takes the text and returns the
analyzed expression or the session reading?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StoredExpressionText.scala:
##########
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.util
+
+import org.apache.spark.sql.AnalysisException
+import org.apache.spark.sql.catalyst.analysis.{Analyzer,
ExpressionWithUnresolvedIdentifier}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, ExprId}
+import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
+import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Reads the SQL text of a generation expression stored with a table,
independently of the
+ * settings of the session that reads it.
+ */
+object StoredExpressionText {
+
+ /**
+ * The settings a stored text is read under. The session time zone, case
sensitivity, timestamp
+ * type and decimal precision loss stay with the given session, because the
settings of the DDL
+ * are not known here and pinning them would silently change existing tables
of a deployment that
+ * sets them.
+ */
+ def conf(session: SQLConf): SQLConf = {
Review Comment:
The reason given for keeping `timestampType` and `allowPrecisionLoss` with
the session also applies to every config that an expression reads in its
constructor and that has no `SESSION` binding. Those are reset to their
defaults here. Examples are `legacy.negativeIndexInArrayInsert`
(`ArrayInsert`), `legacy.followThreeValuedLogicInArrayExists` (`ArrayExists`)
and `legacy.allowNullComparisonResultInArraySort` (`ArraySort`). At
e4311984c08, on `testcat`, with `negativeIndexInArrayInsert=true` for the DDL
and the writes:
```sql
CREATE TABLE t (a ARRAY<STRING>, b ARRAY<STRING> GENERATED ALWAYS AS
(array_insert(a, -1, 'x')));
INSERT INTO t (a) VALUES (array('a', 'b')); -- stores [a,
b, x]
SELECT array_insert(array('a', 'b'), -1, 'x'); -- [a, x, b]
INSERT INTO t VALUES (array('a', 'b'), array('a', 'x', 'b')); --
CHECK_CONSTRAINT_VIOLATION
```
`readsAsInSession` compares parse trees only, so CREATE accepts this. Adding
more exemptions doesn't scale, which is why I'd prefer capturing the DDL's
configs. If that stays out of scope, this scaladoc and the migration entry
should state the general rule: any config read during analysis takes its
default, except the ones listed.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/GeneratedColumnExpression.scala:
##########
@@ -124,6 +125,14 @@ case class GeneratedColumnExpression(
throw unsupportedExpressionError(
"generation expression cannot contain non utf8 binary collated string
type")
}
+
+ // A write reads the stored text under fixed settings, so that a value
does not depend on the
+ // session that writes it. The text has to read the same there.
+ if (!StoredExpressionText.readsAsInSession(originalSQL)) {
Review Comment:
- This also rejects tables without `GENERATE_COLUMN_VALUES_ON_WRITE`, where
Spark never reads the text on a write. If the rule is that the text has to mean
one thing for any reader, a comment saying so would help.
- The reason doesn't tell the user what to change. Naming the parser configs
and `enforceReservedKeywords` would make it actionable.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala:
##########
@@ -1290,4 +1302,545 @@ class GeneratedColumnWriteSuite extends QueryTest with
DatasourceV2SQLBase {
"No CheckInvariant should be added when the table lacks the write
capability")
}
}
+
+ test("SPARK-60757: an auto-filled value does not depend on the parser
settings of the writer") {
+ import testImplicits._
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b STRING GENERATED ALWAYS AS (concat(a, 'x\\\\y'))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES ('default')")
+ withSQLConf(SQLConf.ESCAPED_STRING_LITERALS.key -> "true") {
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES ('insert')")
+ Seq("append").toDF("a").writeTo(s"testcat.$tblName").append()
+ }
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Seq("default", "insert", "append").map(a => Row(a, a + "x\\y")))
+
+ withSQLConf(SQLConf.ESCAPED_STRING_LITERALS.key -> "true") {
+ sql(s"INSERT OVERWRITE testcat.$tblName(a) VALUES ('overwrite')")
+ }
+ checkAnswer(spark.table(s"testcat.$tblName"), Row("overwrite",
"overwritex\\y"))
+ }
+ }
+
+ test("SPARK-60757: the validation of a provided value does not depend on the
parser settings " +
+ "of the writer") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b STRING GENERATED ALWAYS AS (concat(a, 'x\\\\y'))
+ |) USING foo""".stripMargin)
+ withSQLConf(SQLConf.ESCAPED_STRING_LITERALS.key -> "true") {
+ sql(s"INSERT INTO testcat.$tblName(a, b) SELECT 'p', concat('p', 'x',
chr(92), 'y')")
+ assertWriteFails("CHECK_CONSTRAINT_VIOLATION") {
+ sql(s"INSERT INTO testcat.$tblName(a, b) " +
+ "SELECT 'q', concat('q', 'x', chr(92), chr(92), 'y')")
+ }
+ }
+ checkAnswer(spark.table(s"testcat.$tblName"), Row("p", "px\\y"))
+ }
+ }
+
+ test("SPARK-60757: a DELETE that rewrites rows does not depend on the parser
settings of the " +
Review Comment:
Since the DELETE skip, this goes through the same early return as "a DELETE
does not check the rows that it carries", so it can be dropped.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -119,27 +135,59 @@ class ResolveTableConstraints(val catalogManager:
CatalogManager) extends Rule[L
Option(col.columnGenerationExpression())
.filter(_ => !autoFilledGenCols.exists(n => resolver(n, col.name)))
.map { genExpr =>
- val catalystExpr = buildGenerationCatalystExpression(genExpr)
+ val catalystExpr = buildGenerationCatalystExpression(r, col.name,
genExpr)
val colRef = UnresolvedAttribute.quoted(col.name)
val columnExtractors = Seq(col.name -> colRef)
val genExprSql = Option(genExpr.getSql()).getOrElse(catalystExpr.sql)
- CheckInvariant(
- EqualNullSafe(colRef, catalystExpr),
+ // Compare with the value an auto-fill stores, which store
assignment casts to the column
+ // type. Otherwise, the comparison can cast the provided value to
the expression type.
+ val colType =
CharVarcharUtils.replaceCharVarcharWithString(col.dataType())
+ def check(expr: Expression): Expression = CheckInvariant(
+ EqualNullSafe(
+ colRef, Cast(expr, colType, Option(conf.sessionLocalTimeZone),
ansiEnabled = true)),
Review Comment:
The auto-fill gets this cast from `TableOutputResolver.cast`, which also
adds `BY_TABLE_INSERTION` and `CheckOverflowInTableInsert`. When the expression
overflows the column type, a provided value now fails with `CAST_OVERFLOW`,
while the auto-fill fails with `CAST_OVERFLOW_IN_TABLE_INSERT`. Sharing that
helper would keep the two in sync.
--
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]
