dongjoon-hyun commented on code in PR #59965:
URL: https://github.com/apache/spark/pull/59965#discussion_r4447196215
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -123,23 +132,47 @@ class ResolveTableConstraints(val catalogManager:
CatalogManager) extends Rule[L
val colRef = UnresolvedAttribute.quoted(col.name)
val columnExtractors = Seq(col.name -> colRef)
val genExprSql = Option(genExpr.getSql()).getOrElse(catalystExpr.sql)
- CheckInvariant(
- EqualNullSafe(colRef, catalystExpr),
+ def check(expr: Expression): Expression = CheckInvariant(
+ EqualNullSafe(colRef, expr),
Review Comment:
You're right that this PR already changed the ANSI-off mode, so keeping the
default mode did not avoid a behavior change. I applied your fix in
e4311984c08. The check now casts the expression to the column type, as store
assignment does for an auto-filled value, so only `7` passes in both modes.
The migration guide and the PR description describe it. The test
`SPARK-60757: the validation of a provided value compares it with the stored
value` covers both modes.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala:
##########
@@ -1290,4 +1302,497 @@ 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: a later write carries a row that a non-ANSI writer
auto-filled") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b DOUBLE GENERATED ALWAYS AS (a + 1)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES ('9007199254740993')")
+ sql(s"INSERT INTO testcat.$tblName SELECT * FROM testcat.$tblName")
+ }
+ val row = Row("9007199254740993", 9007199254740994D)
+ checkAnswer(spark.table(s"testcat.$tblName"), Seq(row, row))
+ }
+ }
+
+ test("SPARK-60757: a DELETE does not check the rows that it carries") {
+ withRowLevelCatalog {
+ val tblName = "my_tab"
+ withTable(s"$rowLevelCat.$tblName") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ // Without the capability, the rows go in as a non-ANSI writer
stored them before.
+ sql(s"""CREATE TABLE $rowLevelCat.$tblName(
+ | id INT,
+ | a STRING,
+ | b DOUBLE GENERATED ALWAYS AS (a + 1)
+ |) USING foo TBLPROPERTIES ('$noWriteCapProp' =
'false')""".stripMargin)
+ sql(s"INSERT INTO $rowLevelCat.$tblName VALUES " +
+ "(1, 'abc', NULL), (2, '2', 3D), (3, '3', 4D)")
+ sql(s"ALTER TABLE $rowLevelCat.$tblName SET TBLPROPERTIES
('$noWriteCapProp' = 'true')")
+ sql(s"DELETE FROM $rowLevelCat.$tblName WHERE id >= 3")
+ }
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ sql(s"DELETE FROM $rowLevelCat.$tblName WHERE id >= 2")
+ }
+ checkAnswer(spark.table(s"$rowLevelCat.$tblName"), Row(1, "abc", null))
+ assertWriteFails("CAST_INVALID_INPUT") {
+ sql(s"INSERT INTO $rowLevelCat.$tblName SELECT * FROM
$rowLevelCat.$tblName")
+ }
+ }
+ }
+ }
+
+ 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.ANSI_ENABLED.key -> "false",
+ SQLConf.LEGACY_SIZE_OF_NULL.key -> "true") {
+ sql(s"INSERT INTO testcat.$tblName(a, arr) VALUES (1, NULL)")
+ }
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(1, null, 1, null,
null))
+ }
+ }
+
+ test("SPARK-60757: a temporary function does not replace a function of an
auto-filled column") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b STRING GENERATED ALWAYS AS (upper(a))
+ |) USING foo""".stripMargin)
+ spark.udf.register("upper", (s: String) => s + "-temp")
+ try {
+ withSQLConf(SQLConf.SESSION_FUNCTION_RESOLUTION_ORDER.key -> "first") {
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES ('x')")
+ }
+ } finally {
+ sql("DROP TEMPORARY FUNCTION IF EXISTS upper")
+ }
+ checkAnswer(spark.table(s"testcat.$tblName"), Row("x", "X"))
+ }
+ }
+
+ test("SPARK-60757: an IDENTIFIER clause is not a different reading of a
generation expression") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a INT,
+ | b INT GENERATED ALWAYS AS (IDENTIFIER('a') + 1)
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES (1)")
+ checkAnswer(spark.table(s"testcat.$tblName"), Row(1, 2))
+ }
+ }
+
+ test("SPARK-60757: CREATE and REPLACE TABLE reject a generation expression
the session reads " +
+ "differently") {
+ val tblName = "my_tab"
+ for {
+ (confs, genExpr) <- Seq(
+ (Seq(SQLConf.ESCAPED_STRING_LITERALS.key -> "true"), "concat(a,
'x\\\\y')"),
+ (Seq(SQLConf.ANSI_ENABLED.key -> "true",
SQLConf.DOUBLE_QUOTED_IDENTIFIERS.key -> "true"),
+ "concat(\"a\", 'x')"))
+ create <- Seq("CREATE TABLE", "CREATE OR REPLACE TABLE")
+ } withSQLConf(confs: _*) {
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"""$create testcat.$tblName(
+ | a STRING,
+ | b STRING GENERATED ALWAYS AS ($genExpr)
+ |) USING foo""".stripMargin)
+ },
+ condition = "UNSUPPORTED_EXPRESSION_GENERATED_COLUMN",
+ parameters = Map(
+ "fieldName" -> "b",
+ "expressionStr" -> genExpr,
+ "reason" -> readsDifferentlyReason))
+ }
+ }
+
+ test("SPARK-60757: createTable with a schema rejects a generation expression
the session " +
+ "reads differently") {
+ val metadata = new MetadataBuilder()
+ .putString(GeneratedColumn.GENERATION_EXPRESSION_METADATA_KEY,
"concat(a, 'x\\\\y')")
+ .build()
+ val schema = new StructType()
+ .add("a", StringType)
+ .add("b", StringType, nullable = true, metadata)
+ val tblName = "my_tab"
+ withSQLConf(SQLConf.ESCAPED_STRING_LITERALS.key -> "true") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.catalog.createTable(s"testcat.$tblName", "foo", schema,
Map.empty[String, String])
+ },
+ condition = "UNSUPPORTED_EXPRESSION_GENERATED_COLUMN",
+ parameters = Map(
+ "fieldName" -> "b",
+ "expressionStr" -> "concat(a, 'x\\\\y')",
+ "reason" -> readsDifferentlyReason))
+ }
+ }
+
+ test("SPARK-60757: a stored text that is not a scalar expression fails the
analysis of the " +
+ "write") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ // A catalog can store a text that CREATE TABLE rejects.
+ catalog("testcat").asInstanceOf[InMemoryCatalog].createTable(
+ Identifier.of(Array(), tblName),
+ Array(
+ Column.create("i", IntegerType),
+ Column.create("b", IntegerType, true, null, "sum(i)", null)),
+ Array.empty[Transform],
+ Collections.emptyMap[String, String])
+ assertWriteFails("MISSING_GROUP_BY") {
+ sql(s"INSERT INTO testcat.$tblName(i) VALUES (1)")
+ }
+ }
+ }
+
+ test("SPARK-60757: a parser setting of the session does not reject an
expression that reads " +
+ "the same") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ withSQLConf(SQLConf.ESCAPED_STRING_LITERALS.key -> "true") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b STRING GENERATED ALWAYS AS (concat(a, 'xy'))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName(a) VALUES ('v')")
+ }
+ checkAnswer(spark.table(s"testcat.$tblName"), Row("v", "vxy"))
+ }
+ }
+
+ test("SPARK-60757: the validation of a provided value reads the text of a
table with a V2 form") {
+ val tblName = "my_tab"
+ withTable(s"testcat.$tblName") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ // The V2 form is COALESCE(s, '1'), with the implicit cast of a
non-ANSI session.
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | s STRING,
+ | b STRING GENERATED ALWAYS AS (coalesce(s, 1))
+ |) USING foo""".stripMargin)
+ val table =
catalog("testcat").asTableCatalog.loadTable(Identifier.of(Array(), tblName))
+ val genExpr = table.columns().find(_.name ==
"b").get.columnGenerationExpression()
+ assert(genExpr.getExpression != null)
+ sql(s"INSERT INTO testcat.$tblName VALUES ('007', '7')")
+ sql(s"INSERT INTO testcat.$tblName(s) VALUES ('008')")
+ checkAnswer(spark.table(s"testcat.$tblName"), Seq(Row("007", "7"),
Row("008", "8")))
+ }
+ }
+ }
+
+ test("SPARK-60757: the validation of a provided value does not substitute a
variable") {
+ val tblName = "my_tab"
+ val variable = s"$${system:java.version}"
+ withTable(s"testcat.$tblName") {
+ withSQLConf(SQLConf.VARIABLE_SUBSTITUTE_ENABLED.key -> "false") {
+ sql(s"""CREATE TABLE testcat.$tblName(
+ | a STRING,
+ | b STRING GENERATED ALWAYS AS (concat(a, '$variable'))
+ |) USING foo""".stripMargin)
+ sql(s"INSERT INTO testcat.$tblName VALUES ('w', 'w$variable')")
Review Comment:
Thank you. I confirmed that the base fails your DataFrame write with
`CHECK_CONSTRAINT_VIOLATION` and this PR passes it. I replaced the test with
your version in e4311984c08, and it now fails on the base.
##########
docs/sql-migration-guide.md:
##########
@@ -70,6 +70,8 @@ license: |
- Since Spark 4.4, `CREATE TABLE`, `REPLACE TABLE`, and `ALTER TABLE ... ADD
CONSTRAINT` on a Data Source V2 table fail with
`USER_DEFINED_FUNCTION_IN_CHECK_CONSTRAINT` when the condition of a CHECK
constraint calls a user-defined function, including a Data Source V2 function.
`ALTER TABLE ... ADD CONSTRAINT` keeps failing with `UNSUPPORTED_SQL_UDF_USAGE`
on a SQL function and with `INVALID_WHERE_CONDITION` on an aggregate function.
Previously, a write could resolve the function name in its own session, so that
it enforced another rule or failed (SPARK-60777). To run these statements,
change the condition to call only built-in functions. Existing constraints that
call a user-defined function are not rejected.
- Since Spark 4.4, a write to a Data Source V2 table that parses the SQL text
of a CHECK constraint binds an unqualified built-in function name to
`system.builtin`, regardless of the SQL path and
`spark.sql.functionResolution.sessionOrder` (SPARK-60777). Previously, an
existing constraint could call another function with the same name.
- Since Spark 4.4, `CREATE TABLE`, `REPLACE TABLE`, and `ALTER TABLE ... ADD
CONSTRAINT` on a Data Source V2 table fail with `SUBQUERY_IN_CHECK_CONSTRAINT`
when the condition of a CHECK constraint contains a subquery. `ALTER TABLE ...
ADD CONSTRAINT` keeps failing with
`UNSUPPORTED_SUBQUERY_EXPRESSION_CATEGORY.HIGHER_ORDER_FUNCTION` on a subquery
in a lambda function. Previously, most of these statements failed with
`INTERNAL_ERROR`, but a subquery that needed no resolution, such as `id >
(SELECT 0 AS a)`, was accepted and every write evaluated it again
(SPARK-60781). Existing constraints that contain a subquery are not rejected.
+- Since Spark 4.4, a write to a Data Source V2 table parses and analyzes the
stored text of a generation expression with ANSI mode on and the default values
of the SQL configs that change how a text reads, such as
`spark.sql.parser.escapedStringLiterals`, regardless of the session of the
write. Only the following still follow that session: the session time zone,
case sensitivity, `spark.sql.timestampType`,
`spark.sql.decimalOperations.allowPrecisionLoss`, the configs that a view also
takes from the session that reads it, the functions found through `SET PATH`,
and the configs that an expression reads during evaluation, such as
`spark.sql.legacy.timeParserPolicy`. A text that does not parse or resolve
under these settings, such as `flag = 1` on a `BOOLEAN` column, is read under
that session. Previously, the session of the write decided how the text read,
so a write could silently store a wrong value (SPARK-60757). As a result, with
`spark.sql.ansi.enabled=false`, also for a table cre
ated with that setting, a write can fail with an error such as
`CAST_INVALID_INPUT` where Spark 4.3 stored `NULL`, and an implicit cast can
change the stored value. A table created with a non-default value of a config
such as `spark.sql.parser.escapedStringLiterals` also gets the default reading
of its text on every write, without an error. A row that Spark 4.3 stored
differently can fail with such an error or with `CHECK_CONSTRAINT_VIOLATION`
when a later write provides it again, such as an `INSERT ... SELECT` from the
same table. A `DELETE` does not check the rows that it carries. To fix such a
table, create a new table with a text that reads as intended under these
settings, and copy the other columns into it with `INSERT ... SELECT`.
Review Comment:
I reproduced this on the base and this PR, and added the sentence to the
migration entry in e4311984c08. The PR description lists it too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StoredExpressionText.scala:
##########
@@ -0,0 +1,105 @@
+/*
+ * 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 = {
+ val c = new SQLConf
+ Analyzer.retainResolutionConfigsForAnalysis(newConf = c, existingConf =
session)
+ c.setConfString(SQLConf.ANSI_ENABLED.key, "true")
Review Comment:
Confirmed on both the base and this PR: an ANSI-off writer stores `d = 1`,
and an ANSI writer fails with `CAST_INVALID_INPUT`. This predates this PR and
is not specific to generated columns, so I kept it out of this one. I am
working on it separately, and will file a JIRA ticket for it. The PR
description now lists this shape as one that still depends on the writer.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -4122,11 +4125,43 @@ class Analyzer(
cleanedRelation
case other => other
}
- v2Write.withNewQuery(projection).withNewTable(cleanedTable)
+ v2Write
+ .withNewQuery(analyzeAutoFilledGeneratedColumns(
+ projection, autoFilledGenCols,
resolveOutput(CatalystSqlParser.parseExpression)._1))
Review Comment:
I reproduced this, and applied your suggestion in e4311984c08 as
`StoredExpressionText.parseWithSessionAnsi`, at both fallbacks. Both of your
writes now store `true`, and a provided `true` under the escaped-literal writer
passes too. The test `SPARK-60757: a text the fixed settings cannot resolve
keeps their parser settings` covers the auto-fill and the check.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -119,27 +135,56 @@ 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(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),
+ def check(expr: Expression): Expression = CheckInvariant(
+ EqualNullSafe(colRef, expr),
columnExtractors,
"Generated Column",
s"${col.name} <=> $genExprSql")
+ analyzeGeneratedColumnCheck(check(catalystExpr),
v2Write.query).getOrElse {
+ // The analysis of the write resolves it under the session that
writes.
+ check(Option(genExpr.getSql).map(CatalystSqlParser.parseExpression)
+ .getOrElse(catalystExpr))
+ }
}
}.toSeq
}
/**
- * Convert a V2 GenerationExpression to a Catalyst expression.
- * Try the V2 expression first, fall back to parsing the SQL string.
+ * Analyzes the check of a provided value under the settings an auto-filled
value is analyzed
+ * under, so that a row Spark auto-filled passes the check when a later
write carries it.
+ *
+ * @return the analyzed check, or None if these settings cannot resolve it.
A catalog can also
+ * store a text that analyzes to another plan.
*/
- private def buildGenerationCatalystExpression(genExpr:
GenerationExpression): Expression = {
- Option(genExpr.getExpression)
- .flatMap(V2ExpressionUtils.toCatalyst)
-
.getOrElse(catalogManager.v1SessionCatalog.parser.parseExpression(genExpr.getSql))
+ private def analyzeGeneratedColumnCheck(
+ check: Expression,
+ query: LogicalPlan): Option[Expression] = {
+ StoredExpressionText.analyze(Filter(check, LocalRelation(query.output)),
analyze) match {
+ case Some(Filter(analyzed, _: LocalRelation)) if analyzed.resolved =>
Some(analyzed)
+ case _ => None
+ }
+ }
+
+ /**
+ * Convert a V2 GenerationExpression to a Catalyst expression. Read the SQL
string first, as an
+ * auto-filled value does, and fall back to the V2 expression.
+ */
+ private def buildGenerationCatalystExpression(
+ colName: String,
+ genExpr: GenerationExpression): Expression = {
+ Option(genExpr.getSql)
+ .map(StoredExpressionText.parse)
+
.orElse(Option(genExpr.getExpression).flatMap(V2ExpressionUtils.toCatalyst))
+ .getOrElse(throw new AnalysisException(
+ errorClass = "UNSUPPORTED_EXPRESSION_GENERATED_COLUMN",
Review Comment:
Changed in e4311984c08 to `UNSUPPORTED_FEATURE.TABLE_OPERATION`, which the
other limits of generated columns on a write, such as MERGE and UPDATE, also
use. The message now starts with "Table `testcat`.`my_tab` does not support a
value for the generated column `b`, because Spark cannot convert its expression
NOT_A_FUNCTION()", and the test checks it.
--
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]
