dongjoon-hyun commented on code in PR #59965: URL: https://github.com/apache/spark/pull/59965#discussion_r4451017138
########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/StoredExpressionText.scala: ########## @@ -0,0 +1,117 @@ +/* + * 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 = { Review Comment: Thank you. I reproduced it: with `negativeIndexInArrayInsert=true`, the base stores `[a, x, b]` and this PR stores `[a, b, x]`, and CREATE accepts it. I agree that more exemptions don't scale. So, as you suggested, the scaladoc and the migration entry in 0a3dd82c527 state the general rule: any other config that the parsing or the analysis reads takes its default value, also one that a function reads while it is built. The new `spark.sql.legacy.useCurrentConfigsForGeneratedColumn` restores the session reading for a deployment that sets such a config. I replied about capturing the DDL's configs on the review. ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala: ########## @@ -4122,11 +4124,45 @@ class Analyzer( cleanedRelation case other => other } - v2Write.withNewQuery(projection).withNewTable(cleanedTable) + v2Write + .withNewQuery(analyzeAutoFilledGeneratedColumns( + projection, + autoFilledGenCols, + resolveOutput(StoredExpressionText.parseWithSessionAnsi)._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 its text is read + * with the ANSI value of the session, 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( Review Comment: Agreed. In 0a3dd82c527, both go through `StoredExpressionText.analyze`. It takes the expression read under the fixed settings and the one read under the session, analyzes the first in a `Project` under the fixed settings, and returns the second when that fails, does not resolve, or analyzes to another plan. `analyzeGeneratedColumnCheck` is gone. ########## 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: Both applied in 0a3dd82c527. The comment now says that the rule also covers a table that computes the values itself, so that the text means the same to any reader. The reason now reads "a write reads it with the default values of the parser configs, such as spark.sql.parser.escapedStringLiterals and spark.sql.ansi.enforceReservedKeywords, and this session reads it differently". ########## sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveTableConstraints.scala: ########## @@ -119,27 +135,59 @@ 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(r, 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), + // Compare with the value an auto-fill stores, which store assignment casts to the column + // type. Otherwise, the comparison can cast the provided value to the expression type. + val colType = CharVarcharUtils.replaceCharVarcharWithString(col.dataType()) + def check(expr: Expression): Expression = CheckInvariant( + EqualNullSafe( + colRef, Cast(expr, colType, Option(conf.sessionLocalTimeZone), ansiEnabled = true)), Review Comment: I reproduced it with a catalog that stores a narrower column type: a provided value failed with `CAST_OVERFLOW`, and an auto-filled one with `CAST_OVERFLOW_IN_TABLE_INSERT`. In 0a3dd82c527 the check uses `TableOutputResolver.cast`, so both fail with `CAST_OVERFLOW_IN_TABLE_INSERT`, and a test covers it. The check casts the expression before it is resolved, so `canCauseCastOverflow` now treats an unresolved child as one that can overflow. The existing callers always pass a resolved one. ########## sql/core/src/test/scala/org/apache/spark/sql/connector/GeneratedColumnWriteSuite.scala: ########## @@ -1290,4 +1302,545 @@ 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 " + Review Comment: Removed in 0a3dd82c527. -- 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]
