dongjoon-hyun commented on code in PR #59965:
URL: https://github.com/apache/spark/pull/59965#discussion_r4432488136
##########
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:
Thank you. I reproduced this, including the re-insert, and fixed it in
045f67e3d98. It also happens for a table created with ANSI on when the
expression has no V2 form, such as `upper(a) + 1` under an ANSI-off writer.
`ResolveTableConstraints` now analyzes the whole check of a provided value
under the same settings as the auto-fill, through a new `analyze` parameter.
The test `SPARK-60757: a later write carries a row that a non-ANSI writer
auto-filled` covers it.
The cost is that a row stored earlier under ANSI off can now fail when a
later write carries it, whichever ANSI setting that write uses. Before this PR,
that happens only when the later write runs with ANSI on. I described this in
the migration guide and the PR description.
##########
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:
Added two entries under "Upgrading from Spark SQL 4.3 to 4.4" in
045f67e3d98, one for the writes and one for the DDL rejection. The DDL entry
also covers a table without `GENERATE_COLUMN_VALUES_ON_WRITE`, which I
confirmed by running it.
##########
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:
Agreed, so I kept the behavior. In 045f67e3d98, I added the comment on this
case instead of next to `case _ => p`, because an unresolved result goes
through this case. The PR description now lists this under "The following still
depend on the session that writes", and it no longer says "the same way
`ResolveSQLFunctions`".
##########
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:
Applied in 045f67e3d98. With the legacy config, I confirmed that the base
returns `NULL` with ANSI on and `-1` with ANSI off, and this PR returns `NULL`
for both.
--
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]
