peter-toth commented on code in PR #59965:
URL: https://github.com/apache/spark/pull/59965#discussion_r4432308025
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -123,7 +123,7 @@ class ResolveTableConstraints(val catalogManager:
CatalogManager) extends Rule[L
private def buildGenerationCatalystExpression(genExpr:
GenerationExpression): Expression = {
Option(genExpr.getExpression)
.flatMap(V2ExpressionUtils.toCatalyst)
-
.getOrElse(catalogManager.v1SessionCatalog.parser.parseExpression(genExpr.getSql))
+
.getOrElse(StoredExpressionText.parse(catalogManager.v1SessionCatalog.parser,
genExpr.getSql))
Review Comment:
**Finding 1.** For a table created with ANSI off, the auto-fill and the
validation of a provided value now read the same text differently:
- The auto-fill analyzes it under `StoredExpressionText.conf`, so with ANSI
on (`Analyzer.scala:4143`).
- The validation parses it here, and the outer analysis then resolves it
under the session. The V2 form doesn't take over. `V2ExpressionBuilder` skips a
non-ANSI `Add` or `Cast`, so the DDL below stores none.
Base reads both under the session, so they agree. On head a row that Spark
auto-filled can fail the check when a later write carries it. I ran this with
ANSI off for the DDL and for every write, on
`InMemoryRowLevelOperationTableCatalog`:
```sql
CREATE TABLE t (id INT, a STRING, b DOUBLE GENERATED ALWAYS AS (a + 1));
INSERT INTO t (id, a) VALUES (1, '9007199254740993'), (2, '2');
-- b of row 1: 9.007199254740994E15 on head (STRING to BIGINT),
9.007199254740992E15 on base (STRING to DOUBLE)
DELETE FROM t WHERE id >= 2;
-- head: CHECK_CONSTRAINT_VIOLATION on the carried row 1, base: succeeds
```
Re-inserting the table's own rows with `INSERT ... SELECT` fails the same
way on head and passes on base. The DELETE passes on head when it runs with
ANSI on. So analyzing this expression under the same settings as the auto-fill
should fix it. `ResolveTableConstraints` is created inside the `Analyzer`
(`Analyzer.scala:717`), so it can be given the nested analysis that
`analyzeAutoFilledGeneratedColumns` does. A test with the DELETE above would
pin it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StoredExpressionText.scala:
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.catalyst.analysis.{Analyzer,
ExpressionWithUnresolvedIdentifier}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Expression, ExprId}
+import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
+import org.apache.spark.sql.internal.SQLConf
+
+/**
+ * Reads the SQL text of an expression stored with a table, such as a CHECK
condition or a
+ * generation expression, independently of the settings of the session that
reads it.
+ */
+object StoredExpressionText {
+
+ /**
+ * The settings a stored text is read under. The session time zone and case
sensitivity 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 non-UTC or a case-sensitive
deployment.
+ */
+ def conf(session: SQLConf): SQLConf = {
+ val c = new SQLConf
+ Analyzer.retainResolutionConfigsForAnalysis(newConf = c, existingConf =
session)
+ c.setConfString(SQLConf.ANSI_ENABLED.key, "true")
Review Comment:
**Finding 2.** This changes the behaviour of the released 4.3.0, and
`docs/sql-migration-guide.md` has no entry for it. These two changes need one
under "Upgrading from Spark SQL 4.3 to 4.4":
- CREATE and REPLACE TABLE reject a generation expression that the session
parses differently, e.g. with `spark.sql.parser.escapedStringLiterals=true` or
`spark.sql.ansi.enforceReservedKeywords=true`. The check runs for every catalog
that supports generated columns, also for a table that computes the values
itself (no `GENERATE_COLUMN_VALUES_ON_WRITE`).
- With `spark.sql.ansi.enabled=false`, writes compute generated values in
ANSI mode, also for tables created with that setting. They can fail where 4.3.0
stored `NULL`, and the implicit casts can change the stored value.
Your "Does this PR introduce _any_ user-facing change?" section already has
the wording for both.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -4122,11 +4122,35 @@ class Analyzer(
cleanedRelation
case other => other
}
- v2Write.withNewQuery(projection).withNewTable(cleanedTable)
+ v2Write
+ .withNewQuery(analyzeAutoFilledGeneratedColumns(projection,
autoFilledGenCols))
+ .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.
+ */
+ private def analyzeAutoFilledGeneratedColumns(
+ projection: LogicalPlan,
+ autoFilled: Set[String]): LogicalPlan = projection match {
+ case p @ Project(projectList, child) if autoFilled.nonEmpty =>
+ val filled = projectList.filter(c => autoFilled.contains(c.name))
+ SQLConf.withExistingConf(StoredExpressionText.conf(conf)) {
+ execute(Project(filled, LocalRelation(child.output)))
+ } match {
+ case Project(analyzedCols, _: LocalRelation) =>
Review Comment:
**Finding 3.** This case also matches when the nested analysis leaves the
columns unresolved. The outer analysis then resolves them under the writer's
session, so the fixed settings don't apply. With ANSI off for the DDL:
```sql
CREATE TABLE t (flag BOOLEAN, b BOOLEAN GENERATED ALWAYS AS (flag = 1));
INSERT INTO t (flag) VALUES (true); -- ANSI-off writer: b = true
INSERT INTO t (flag) VALUES (false); -- ANSI-on writer:
DATATYPE_MISMATCH.BINARY_OP_DIFF_TYPES
```
Base does the same, so this is not a regression. But two sentences of the
description don't hold for it:
- "The settings of the writer no longer change an auto-filled value through
... the ANSI mode".
- "the same way `ResolveSQLFunctions` analyzes a SQL function body".
`ResolveSQLFunctions.resolve` ends with `SimpleAnalyzer.checkAnalysis`.
Failing eagerly here would break the ANSI-off writer, which works on base.
So I'd add this case to "The following still depend on the session that
writes", and mention it in the comment next to `case _ => p`.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala:
##########
@@ -1267,4 +1279,284 @@ 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 " +
+ "writer") {
+ withRowLevelCatalog {
+ val tblName = "my_tab"
+ withTable(s"$rowLevelCat.$tblName") {
+ sql(s"""CREATE TABLE $rowLevelCat.$tblName(
+ | id INT,
+ | a STRING,
+ | b STRING GENERATED ALWAYS AS (concat(a, 'x\\\\y'))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO $rowLevelCat.$tblName(id, a) VALUES (1, 'p'), (2,
'q')")
+ withSQLConf(SQLConf.ESCAPED_STRING_LITERALS.key -> "true") {
+ sql(s"DELETE FROM $rowLevelCat.$tblName WHERE id <= 1")
+ }
+ checkAnswer(spark.table(s"$rowLevelCat.$tblName"), Row(2, "q",
"qx\\y"))
+ }
+ }
+ }
+
+ test("SPARK-60757: an explicit cast and arithmetic keep failing under a
non-ANSI writer") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b INT GENERATED ALWAYS AS (CAST(a AS INT)),
+ | c INT,
+ | d INT GENERATED ALWAYS AS (c + 1)
+ |) USING foo""".stripMargin)
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ assertWriteFails("CAST_INVALID_INPUT") {
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES ('abc', 1)")
+ }
+ assertWriteFails("ARITHMETIC_OVERFLOW") {
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES ('1',
${Int.MaxValue})")
+ }
+ sql(s"INSERT INTO testcat.$tblName(a, c) VALUES ('1', 1)")
+ }
+ checkAnswer(spark.table(s"testcat.$tblName"), Row("1", 1, 1, 2))
+ }
+ }
+
+ test("SPARK-60757: the validation of a provided value keeps failing under a
non-ANSI writer") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ // UPPER has no way back from its V2 form, so the validation reads the
SQL text.
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b INT GENERATED ALWAYS AS (CAST(upper(a) AS INT))
+ |) USING foo""".stripMargin)
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ assertWriteFails("CAST_INVALID_INPUT") {
+ sql(s"INSERT INTO testcat.$tblName(a, b) VALUES ('abc', NULL)")
+ }
+ }
+ checkAnswer(spark.table(s"testcat.$tblName"), Nil)
+ }
+ }
+
+ test("SPARK-60757: a reserved keyword of an auto-filled column does not
follow the settings " +
+ "of the writer") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | `current_date` DATE,
+ | b DATE GENERATED ALWAYS AS (current_date)
+ |) USING foo""".stripMargin)
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.ENFORCE_RESERVED_KEYWORDS.key -> "true") {
+ sql(s"INSERT INTO testcat.$tblName(`current_date`) VALUES
(DATE'2024-06-15')")
+ }
+ val date = sqlDate("2024-06-15")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(date, date))
+ }
+ }
+
+ test("SPARK-60757: an implicit cast of an auto-filled column does not follow
a non-ANSI writer") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b BIGINT GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ }
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ assertWriteFails("CAST_INVALID_INPUT") {
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES ('abc')")
+ }
+ // A non-ANSI cast to DOUBLE would lose the last digit.
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES ('9007199254740993')")
+ }
+ checkAnswer(
+ spark.table(s"testcat.$tblName"),
+ Row("9007199254740993", 9007199254740994L))
+ }
+ }
+
+ test("SPARK-60757: a function of an auto-filled column does not follow the
settings of the " +
+ "writer") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | arr ARRAY<INT>,
+ | b INT GENERATED ALWAYS AS (abs(a)),
+ | c INT GENERATED ALWAYS AS (element_at(arr, 2)),
+ | d INT GENERATED ALWAYS AS (size(arr))
+ |) USING foo""".stripMargin)
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ assertWriteFails("ARITHMETIC_OVERFLOW") {
+ sql(s"INSERT INTO testcat.$tblName(a, arr) VALUES (${Int.MinValue},
array(1, 2))")
+ }
+ assertWriteFails("INVALID_ARRAY_INDEX_IN_ELEMENT_AT") {
+ sql(s"INSERT INTO testcat.$tblName(a, arr) VALUES (1, array(1))")
+ }
+ }
+ withSQLConf(SQLConf.LEGACY_SIZE_OF_NULL.key -> "true") {
Review Comment:
**Finding 4.** `legacySizeOfNull` is `LEGACY_SIZE_OF_NULL && !ANSI_ENABLED`.
So in the default ANSI job base returns `NULL` here too, and this step only
tests something in the `SPARK_ANSI_SQL_MODE=false` job. With ANSI off in this
block, base returns `-1` in both jobs, and the test still passes on head.
```suggestion
withSQLConf(
SQLConf.ANSI_ENABLED.key -> "false",
SQLConf.LEGACY_SIZE_OF_NULL.key -> "true") {
```
--
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]
