dongjoon-hyun commented on code in PR #59965:
URL: https://github.com/apache/spark/pull/59965#discussion_r4436655467
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -123,23 +131,41 @@ 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),
- columnExtractors,
- "Generated Column",
- s"${col.name} <=> $genExprSql")
+ analyzeGeneratedColumnCheck(
+ CheckInvariant(
+ EqualNullSafe(colRef, catalystExpr),
+ columnExtractors,
+ "Generated Column",
+ s"${col.name} <=> $genExprSql"),
+ v2Write.query)
}
}.toSeq
}
+ /**
+ * 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. What
+ * those settings cannot resolve is left to the analysis of the write, as
for an auto-filled
+ * value.
+ */
+ private def analyzeGeneratedColumnCheck(check: Expression, query:
LogicalPlan): Expression = {
+ SQLConf.withExistingConf(StoredExpressionText.conf(conf)) {
+ analyze(Filter(check, LocalRelation(query.output)))
+ } match {
+ case Filter(analyzed, _: LocalRelation) => analyzed
+ // A catalog can store a text that analyzes to another plan. The write
analyzes it as is.
+ case _ => check
+ }
+ }
+
/**
* Convert a V2 GenerationExpression to a Catalyst expression.
* Try the V2 expression first, fall back to parsing the SQL string.
*/
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 on `testcat`, and fixed it in 958490e358a as
you suggested. The check now reads the text first and uses the V2 form only
when there is no text, so it agrees with the auto-fill. Your second insert now
passes. A provided `007` also passes, because the check compares it with the
`BIGINT` result of the expression.
I updated the `GenerationExpression` javadoc and the comment above this
method. The test `SPARK-60757: the validation of a provided value reads the
text of a table with a V2 form` covers it. I did not run the `DELETE` either,
for the reason you mention.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -4122,11 +4122,37 @@ 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 {
+ // A text these settings cannot resolve stays unresolved here, and
the analysis of the
Review Comment:
I reproduced both, and fixed them in 958490e358a with your first option, per
column. A column that the fixed settings do not resolve is now read entirely
under the session, for both the auto-fill and the check. This also replaces
`case _ => p`.
Both of your cases now match the base. `t1` stores `NULL` under an ANSI-off
writer and fails under an ANSI-on writer, and `t2` stores `-2147483648`. I
chose this over rejecting at CREATE, because existing tables keep working and
it adds no new rejection. Such a text keeps following the session, as before
this PR, and the migration guide and the PR description now say that.
The test `SPARK-60757: a text the fixed settings cannot resolve is read
under the session` covers it, with a second generated column in the same write
that stays with the fixed settings.
##########
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")
+ c.setConfString(SQLConf.SESSION_LOCAL_TIMEZONE.key,
session.sessionLocalTimeZone)
Review Comment:
I reproduced this, and fixed it in 958490e358a with your first option.
`spark.sql.timestampType` now stays with the session, for the same reason as
the time zone. Your run stores `02:30` again, and the check rejects `03:30`.
As you note, `CAST(a AS TIMESTAMP)` follows the writer again, as on the
base. So I removed that case from the rejection test, and added this setting to
the migration guide and the PR description. The test `SPARK-60757: a generation
expression reads the timestamp type of the session` covers 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")
+ c.setConfString(SQLConf.SESSION_LOCAL_TIMEZONE.key,
session.sessionLocalTimeZone)
+ c.setConfString(SQLConf.CASE_SENSITIVE.key,
session.caseSensitiveAnalysis.toString)
+ c
+ }
+
+ /** Parses `text` under [[conf]]. */
+ def parse(parser: ParserInterface, text: String): Expression = {
+ SQLConf.withExistingConf(conf(SQLConf.get)) {
+ parser.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(parser: ParserInterface, text: String): Boolean = {
+ val session = SQLConf.get
+ val fixed = conf(session)
+ fixed.setConfString(SQLConf.ANSI_ENABLED.key, session.ansiEnabled.toString)
Review Comment:
I reproduced the rejection, but did not apply the suggestion as is, because
it lets a decimal value change silently. With the setting off and `b
DECIMAL(38, 20) GENERATED ALWAYS AS (a * a)` on `a DECIMAL(20, 10)`, the
suggestion stores `1.26215515697488188000` for `a = 1.1234567891`, where the
base stores `1.26215515697488187881`.
In 958490e358a, `StoredExpressionText.conf` keeps this setting from the
session instead, so both the CREATE and the value match the base. A write with
another value of this setting than the DDL still reads the text differently, as
on the base, and the migration guide and the PR description list that. The test
`SPARK-60757: a generation expression reads the decimal precision loss of the
session` covers both.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala:
##########
@@ -123,23 +131,41 @@ 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),
- columnExtractors,
- "Generated Column",
- s"${col.name} <=> $genExprSql")
+ analyzeGeneratedColumnCheck(
+ CheckInvariant(
+ EqualNullSafe(colRef, catalystExpr),
+ columnExtractors,
+ "Generated Column",
+ s"${col.name} <=> $genExprSql"),
+ v2Write.query)
}
}.toSeq
}
+ /**
+ * 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. What
+ * those settings cannot resolve is left to the analysis of the write, as
for an auto-filled
+ * value.
+ */
+ private def analyzeGeneratedColumnCheck(check: Expression, query:
LogicalPlan): Expression = {
+ SQLConf.withExistingConf(StoredExpressionText.conf(conf)) {
+ analyze(Filter(check, LocalRelation(query.output)))
+ } match {
+ case Filter(analyzed, _: LocalRelation) => analyzed
+ // A catalog can store a text that analyzes to another plan. The write
analyzes it as is.
+ case _ => check
+ }
+ }
+
/**
* Convert a V2 GenerationExpression to a Catalyst expression.
* Try the V2 expression first, fall back to parsing the SQL string.
*/
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:
Fixed in 958490e358a with `CatalystSqlParser`, together with finding 5.
Every reader uses it now, so I also removed the parser parameter from
`StoredExpressionText`. The test `SPARK-60757: the validation of a provided
value does not substitute a variable` covers it.
##########
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 session time zone, case sensitivity, and the configs that a
view also takes from the session that reads it follow that session. A text that
does not resolve with ANSI mode on, such as `flag = 1` on a `BOOLEAN` column,
also still follows 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 created 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
row that Spark 4.3 stored that way can fail with such an error or with `CHECK_C
ONSTRAINT_VIOLATION` when a later write, such as a `DELETE` that rewrites
rows, carries it.
Review Comment:
Updated in 958490e358a. The entry now lists what still follows the session,
including `SET PATH`, the settings read during evaluation,
`spark.sql.timestampType` and `spark.sql.decimalOperations.allowPrecisionLoss`.
It also says that a table created under a non-default parser setting gets the
default reading without an error, and how to fix such a table.
I did not add a legacy config, because the tables from finding 6 can be
written again.
--
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]
