peter-toth commented on code in PR #59965:
URL: https://github.com/apache/spark/pull/59965#discussion_r4440779653
##########
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:
**Finding 24.** This write runs with `spark.sql.variable.substitute=false`,
so the test passes on base too. Base only substitutes when the setting is on,
which is the default. I ran a write under the default setting on base and head,
with the DDL as here:
```scala
Seq(("v", s"v$variable")).toDF("a",
"b").writeTo(s"testcat.$tblName").append()
// base: CHECK_CONSTRAINT_VIOLATION
// head: succeeds
```
A re-insert of an auto-filled row fails the same way on base. So the fix is
real, but no test pins it. A DataFrame write outside the `withSQLConf` block
would. A SQL `INSERT` would get its own `${...}` substituted. I tried this
version, and it fails on base and passes on head:
```scala
test("SPARK-60757: the validation of a provided value does not substitute
a variable") {
import testImplicits._
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')")
}
// Substitution is on by default. A DataFrame write keeps the variable
in its values.
Seq(("v", s"v$variable")).toDF("a",
"b").writeTo(s"testcat.$tblName").append()
checkAnswer(
spark.table(s"testcat.$tblName"),
Seq(Row("w", s"w$variable"), Row("v", s"v$variable")))
}
}
```
##########
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:
**Finding 25.** A table CHECK constraint is still read under the writer's
session, which [SPARK-60453](https://issues.apache.org/jira/browse/SPARK-60453)
tracks. The generated value no longer is. So a write that passes on base now
fails when a constraint repeats the generation text. On `testcat`, on base and
head:
```sql
CREATE TABLE t (a STRING, b STRING GENERATED ALWAYS AS (concat(a, 'x\\y')));
ALTER TABLE t ADD CONSTRAINT c CHECK (b = concat(a, 'x\\y'));
SET spark.sql.parser.escapedStringLiterals=true;
INSERT INTO t (a) VALUES ('q');
-- base: stores b = qx\\y and passes
-- head: computes b = qx\y, and c fails it with CHECK_CONSTRAINT_VIOLATION
```
Base's value is the wrong one, so this is SPARK-60453 showing up rather than
a new defect. But a writer with a non-default parser setting now gets a failure
where 4.3 succeeded. Until that ticket lands, one sentence in this entry would
tell them why.
##########
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:
**Finding 26.** This pins ANSI for the analysis, but the optimizer of the
write still runs under the writer's session. `ConstantFolding.hasNoSideEffect`
(`optimizer/expressions.scala:58`) treats a `Cast` as free of side effects
whenever the session has ANSI off, whatever the cast's own eval mode. So under
such a session, `ConstantFolding` folds `size(array(CAST(a AS INT)))` to `1`
and the ANSI cast is gone. On `testcat`, on base and head:
```sql
CREATE TABLE t (a STRING, d INT GENERATED ALWAYS AS (size(array(CAST(a AS
INT)))));
INSERT INTO t (a) VALUES ('abc');
-- ANSI-off writer: d = 1
-- ANSI writer: CAST_INVALID_INPUT
```
So this shape still depends on the writer. It predates the PR, since base
gives the same, and it is not specific to generated columns. I found no JIRA
ticket for it. [SPARK-60749](https://issues.apache.org/jira/browse/SPARK-60749)
is the same kind of problem in another optimizer rule. Could you file one?
Checking the cast's own eval mode there instead of `conf.ansiEnabled` looks
like the fix. I haven't tried it.
##########
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:
**Finding 27.** Finding 6 suggested reading a column that the fixed settings
don't resolve entirely under the session. This does that, but it also gives
such a column the writer's parser settings, not only its ANSI mode. With ANSI
off for the DDL and the writes, on `testcat`, on base and head:
```sql
CREATE TABLE t (d DATE, s STRING,
b BOOLEAN GENERATED ALWAYS AS (CAST(d AS BOOLEAN) OR s = 'x\\y'));
INSERT INTO t (d, s) SELECT DATE'2024-01-01', concat('x', chr(92), 'y'); --
b = true
SET spark.sql.parser.escapedStringLiterals=true;
INSERT INTO t (d, s) SELECT DATE'2024-01-02', concat('x', chr(92), 'y'); --
b = NULL
```
Only the ANSI part needs the session here. So the fallback can parse under
the fixed settings with the session's ANSI value, as the conf in
`readsAsInSession` does. It then needs the session parse only for a text those
settings can't parse:
```scala
def parseWithSessionAnsi(text: String): Expression = {
val session = SQLConf.get
val c = conf(session)
c.setConfString(SQLConf.ANSI_ENABLED.key, session.ansiEnabled.toString)
try {
SQLConf.withExistingConf(c) {
CatalystSqlParser.parseExpression(text)
}
} catch {
case _: ParseException => CatalystSqlParser.parseExpression(text)
}
}
```
I tried it here and at the fallback at `ResolveTableConstraints.scala:149`.
Both writes above then store `true`, and `GeneratedColumnWriteSuite` and
`DataSourceV2DataFrameSuite` still pass.
##########
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:
**Finding 28.** `UNSUPPORTED_EXPRESSION_GENERATED_COLUMN` reads "Cannot
create generated column <fieldName> with generation expression <expressionStr>
because <reason>". So this `INSERT` now says that it cannot create column `b`,
and the new test pins that message. A condition worded for the write, or a
sub-condition of an existing write error, would fit better than the DDL one.
##########
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:
**Finding 29.** "A `DELETE` does not check the rows that it carries" holds
for the generated columns only. A copy-on-write `DELETE` still checks the
table's CHECK constraints on the rows it carries
(`ResolveTableConstraints.scala:54`). Only a `WriteDelta` `DELETE` skips both
(`:47`). "does not check the generated columns of the rows that it carries"
would be exact.
##########
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:
**Finding 15 (round 4).** This PR already changes the ANSI-off mode. Base
rejects `'07'` and `' 7'` there, and head accepts them. So keeping the default
mode as it is doesn't avoid a behavior change. It picks the side where the
column can hold a value that its own expression doesn't produce. Base's
acceptance with ANSI on is the same defect. The fix moves both modes to the
value the auto-fill stores.
If it stays out of this PR anyway, two things would cover it:
- The migration entry needs the sentence that the description already has
("With `spark.sql.ansi.enabled=false`, the validation of a provided value now
uses the ANSI implicit casts ..."). An ANSI-off writer now gets provided values
accepted that used to fail, and the entry doesn't say so.
- A JIRA ticket for the check comparing at the expression's type, so the
default-mode fix has a home.
--
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]
