dongjoon-hyun commented on code in PR #59965:
URL: https://github.com/apache/spark/pull/59965#discussion_r4437393966
##########
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),
columnExtractors,
"Generated Column",
s"${col.name} <=> $genExprSql")
+ analyzeGeneratedColumnCheck(check(catalystExpr),
v2Write.query).getOrElse {
Review Comment:
I reproduced this with rows written the way a 4.3 writer stores them, and
fixed it in 0fd998404af as you suggested. The check of generated columns is now
skipped for a row-level `DELETE`, which writes no new generated value. The
`DELETE` passes with ANSI off, and also with ANSI on, where the base fails.
An `INSERT ... SELECT` of such a row still goes through the check and fails,
and the migration guide now says that. `SPARK-60757: a later write carries a
row that a non-ANSI writer auto-filled` now uses that re-insert, and
`SPARK-60757: a DELETE does not check the rows that it carries` covers the
`DELETE`.
##########
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:
I reproduced this, and your fix works in my run too: only `7` passes for
every DDL and writer setting. I did not apply it in this PR, because it also
changes the default mode. With ANSI on, the base accepts `07`, ` 7` and `007`,
and the fix rejects them. So I kept it out of this PR, and the PR description
now lists the behavior under an ANSI-off writer.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -4122,11 +4125,44 @@ class Analyzer(
cleanedRelation
case other => other
}
- v2Write.withNewQuery(projection).withNewTable(cleanedTable)
+ v2Write
+ .withNewQuery(analyzeAutoFilledGeneratedColumns(
+ projection, autoFilledGenCols,
resolveOutput(CatalystSqlParser.parseExpression)._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 the session reads
+ * its text, 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(
+ projection: LogicalPlan,
+ autoFilled: Set[String],
+ readInSession: => LogicalPlan): LogicalPlan = projection match {
+ case p @ Project(projectList, child) if autoFilled.nonEmpty =>
+ lazy val sessionCols = readInSession.expressions.collect {
+ case c: NamedExpression => c.name -> c
+ }.toMap
+ p.copy(projectList = projectList.map {
+ case c if autoFilled.contains(c.name) =>
+ SQLConf.withExistingConf(StoredExpressionText.conf(conf)) {
+ execute(Project(Seq(c), LocalRelation(child.output)))
Review Comment:
I confirmed this at the analysis step, and fixed it in 0fd998404af as you
suggested. `StoredExpressionText.analyze` catches an `AnalysisException` from
the analysis under the fixed settings, and both places then take the session
reading. The test `SPARK-60757: a text the fixed settings cannot analyze is
read under the session` covers an auto-filled and a provided value. My run and
the test stop at the analysis, so I did not run these writes.
##########
docs/sql-migration-guide.md:
##########
@@ -69,6 +69,8 @@ license: |
- Since Spark 4.4, `UPDATE` and `MERGE` on a Data Source V2 table fail with
`DATATYPE_MISMATCH.INVALID_ROW_LEVEL_OPERATION_ASSIGNMENTS` when they assign to
a metadata column of the table. This includes `UPDATE SET *` and `INSERT *` in
`MERGE WITH SCHEMA EVOLUTION` when the source has a column with the name of a
metadata column of the target table. Previously, these assignments passed
analysis and were usually ignored. To restore the previous behavior, set
`spark.sql.legacy.ignoreMetadataColumnAssignments` to `true`.
- 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, 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, such as a `DELETE` that rewrites rows, carries it. To fix
such a table, recreate the column with a text that reads as intended under
these settings.
Review Comment:
Applied your wording in 0fd998404af. I also confirmed that both `ALTER
TABLE` forms fail with `PARSE_SYNTAX_ERROR`.
##########
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:
Confirmed, the V2 form stays `COALESCE(s, '1')` while the write stores `7`.
I did not change this in this PR. It only matters for a table created with ANSI
off, on a connector that evaluates the V2 form, and Spark no longer reads that
form when there is a text. The PR description now lists it. As you note,
finding 12 would fix it by construction.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StoredExpressionText.scala:
##########
@@ -0,0 +1,89 @@
+/*
+ * 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.{CatalystSqlParser, ParseException}
+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")
+ c.setConfString(SQLConf.SESSION_LOCAL_TIMEZONE.key,
session.sessionLocalTimeZone)
+ c.setConfString(SQLConf.CASE_SENSITIVE.key,
session.caseSensitiveAnalysis.toString)
+ c.setConf(SQLConf.TIMESTAMP_TYPE, session.getConf(SQLConf.TIMESTAMP_TYPE))
+ c.setConf(
+ SQLConf.DECIMAL_OPERATIONS_ALLOW_PREC_LOSS,
session.decimalOperationsAllowPrecisionLoss)
+ c
+ }
+
+ /**
+ * Parses `text` under [[conf]]. A text these settings cannot parse is
parsed under the active
+ * session, as a text they cannot resolve is analyzed under it.
+ */
+ def parse(text: String): Expression = {
+ try {
+ SQLConf.withExistingConf(conf(SQLConf.get)) {
+ CatalystSqlParser.parseExpression(text)
+ }
+ } catch {
+ case _: ParseException => CatalystSqlParser.parseExpression(text)
+ }
+ }
+
+ /**
+ * Whether `text` reads the same under [[conf]] as it does in the active
session. The comparison
+ * uses the ANSI value of the session, so that it is about how the text
reads.
+ */
+ def readsAsInSession(text: String): Boolean = {
+ val session = SQLConf.get
+ val fixed = conf(session)
+ fixed.setConfString(SQLConf.ANSI_ENABLED.key, session.ansiEnabled.toString)
+ val parsedInSession = CatalystSqlParser.parseExpression(text)
+ try {
+ val parsed = SQLConf.withExistingConf(fixed) {
+ CatalystSqlParser.parseExpression(text)
+ }
+ normalize(parsed) == normalize(parsedInSession)
Review Comment:
Confirmed, and `curdate()` and `now()` are accepted under that setting. I
kept this as is. The rejection happens at CREATE and only under that setting,
no wrong value is stored, and the PR description already lists the alternative.
Accepting it would need a special case for these keywords in the comparison.
--
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]
