gengliangwang commented on code in PR #58632:
URL: https://github.com/apache/spark/pull/58632#discussion_r4008547414
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala:
##########
@@ -354,8 +367,14 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
byName: Boolean,
conf: SQLConf,
addError: String => Unit,
- colPath: Seq[String]): Boolean = {
+ colPath: Seq[String],
+ deferCastValidationToRuntime: Boolean): Boolean = {
conf.storeAssignmentPolicy match {
+ case StoreAssignmentPolicy.ANSI if deferCastValidationToRuntime =>
Review Comment:
Returning `true` here bypasses more than the ANSI leaf-cast check.
`DataTypeUtils.canWrite` also validates struct field names after unwrapping a
UDT. With a source UDT backed by `STRUCT<b: INT, a: INT>` and a by-name target
`STRUCT<a: INT, b: INT>`, `checkField` unwraps the UDT and `Cast.castStruct`
zips fields positionally, so this mode silently writes `b` into `a` and `a`
into `b`. Please preserve recursive/name validation and relax only the atomic
store-assignment compatibility check; add this as a regression test.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3945,7 +3945,10 @@ class Analyzer(
case v2Write: V2WriteCommand
if v2Write.table.resolved && v2Write.query.resolved &&
!v2Write.outputResolved &&
v2Write.pendingSchemaChanges.isEmpty =>
- validateStoreAssignmentPolicy()
+ val schemaAlignment = v2Write.table.collectFirst {
+ case r: DataSourceV2Relation => r.table.schemaAlignmentConfig()
+ }.getOrElse(SchemaAlignmentConfig.DEFAULT)
+ validateStoreAssignmentPolicy(schemaAlignment)
Review Comment:
This validation is only reached under `!v2Write.outputResolved`. An
exact-name/type `DataFrameWriterV2.append` is already output-resolved, so a
table returning `DEFAULT` accepts LEGACY without this method ever being
consulted. The row-level rule has the same issue because validation is inside
`!aligned`. That contradicts `allowLegacyStoreAssignmentPolicy`'s contract.
Please run policy validation independently of whether alignment is needed and
cover exact-schema append plus fully aligned UPDATE/MERGE.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala:
##########
@@ -0,0 +1,294 @@
+/*
+ * 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.connector.catalog
+
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType,
StringType, StructType}
+
+/**
+ * A catalog that creates [[InMemoryRowLevelOperationTable]]s carrying a fixed
+ * [[SchemaAlignmentConfig]] supplied by the concrete subclass. It returns the
live table instance
+ * on load (rather than a copy) so the config is preserved for the analyzer.
+ */
+abstract class SchemaAlignmentTestCatalog extends
InMemoryRowLevelOperationTableCatalog {
+
+ protected def tableConfig: SchemaAlignmentConfig
+
+ override def loadTable(ident: Identifier): Table = liveTable(ident)
Review Comment:
This override papers over config loss by avoiding the catalog's normal copy
path. `InMemoryTable.copy()` omits the new config, and
`InMemoryRowLevelOperationTableCatalog.alterTable` reconstructs via
`withColumns` without passing it, resetting relaxed tables to `DEFAULT` after
ALTER/schema evolution. Please propagate the config through all
copy/reconstruction paths, remove this workaround, and cover relaxed writes
after schema evolution.
##########
sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala:
##########
@@ -0,0 +1,294 @@
+/*
+ * 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.connector.catalog
+
+import scala.util.{Failure, Success, Try}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.{AnalysisException, DataFrame, QueryTest, Row}
+import org.apache.spark.sql.catalyst.analysis.TableAlreadyExistsException
+import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{ArrayType, IntegerType, MapType,
StringType, StructType}
+
+/**
+ * A catalog that creates [[InMemoryRowLevelOperationTable]]s carrying a fixed
+ * [[SchemaAlignmentConfig]] supplied by the concrete subclass. It returns the
live table instance
+ * on load (rather than a copy) so the config is preserved for the analyzer.
+ */
+abstract class SchemaAlignmentTestCatalog extends
InMemoryRowLevelOperationTableCatalog {
+
+ protected def tableConfig: SchemaAlignmentConfig
+
+ override def loadTable(ident: Identifier): Table = liveTable(ident)
+
+ override def createTable(ident: Identifier, tableInfo: TableInfo): Table = {
+ if (tables.containsKey(ident)) {
+ throw new TableAlreadyExistsException(ident.asMultipartIdentifier)
+ }
+ val name = s"${this.name}.${ident.quoted}"
+ val schema = CatalogV2Util.v2ColumnsToStructType(tableInfo.columns)
+ val table = new InMemoryRowLevelOperationTable(
+ name, schema, tableInfo.partitions, tableInfo.properties,
tableInfo.constraints(),
+ schemaAlignmentConfig = tableConfig)
+ tables.put(ident, table)
+ namespaces.putIfAbsent(ident.namespace.toList, Map())
+ table
+ }
+}
+
+/** A catalog whose tables opt into every [[SchemaAlignmentConfig]]
relaxation. */
+class RelaxedSchemaAlignmentCatalog extends SchemaAlignmentTestCatalog {
+ override protected def tableConfig: SchemaAlignmentConfig = new
SchemaAlignmentConfig {
+ override def allowLegacyStoreAssignmentPolicy(): Boolean = true
+ override def deferCastValidationToRuntime(): Boolean = true
+ }
+}
+
+/** A catalog whose tables keep the strict data source v2 defaults. */
+class StrictSchemaAlignmentCatalog extends SchemaAlignmentTestCatalog {
+ override protected def tableConfig: SchemaAlignmentConfig =
SchemaAlignmentConfig.DEFAULT
+}
+
+/**
+ * End-to-end coverage for [[SchemaAlignmentConfig]]: a table that opts into a
relaxation gets the
+ * more permissive analyzer behavior, while an otherwise identical table using
the default (strict)
+ * config keeps the data source v2 behavior. Exercised on both the INSERT path
+ * ([[org.apache.spark.sql.catalyst.analysis.Analyzer.ResolveOutputRelation]])
and the row-level
+ * path
([[org.apache.spark.sql.catalyst.analysis.ResolveRowLevelCommandAssignments]]).
+ */
+class SchemaAlignmentConfigSuite extends QueryTest with SharedSparkSession {
+
+ private val relaxed = "relaxed"
+ private val strict = "strict"
+
+ override def sparkConf: SparkConf =
+ super.sparkConf
+ .set(s"spark.sql.catalog.$relaxed",
classOf[RelaxedSchemaAlignmentCatalog].getName)
+ .set(s"spark.sql.catalog.$strict",
classOf[StrictSchemaAlignmentCatalog].getName)
+
+ private def withLegacyPolicy(f: => Unit): Unit =
+ withSQLConf(
+ SQLConf.STORE_ASSIGNMENT_POLICY.key ->
StoreAssignmentPolicy.LEGACY.toString)(f)
+
+ private def withAnsiPolicy(f: => Unit): Unit =
+ withSQLConf(
+ SQLConf.STORE_ASSIGNMENT_POLICY.key ->
StoreAssignmentPolicy.ANSI.toString)(f)
+
+ private def legacyRejected(f: => Unit): Unit =
+ checkError(
+ exception = intercept[AnalysisException](f),
+ condition = "_LEGACY_ERROR_TEMP_1000",
+ parameters = Map("configKey" -> SQLConf.STORE_ASSIGNMENT_POLICY.key))
+
+ test("allowLegacyStoreAssignmentPolicy: INSERT under LEGACY policy") {
+ withTable(s"$relaxed.t", s"$strict.t") {
+ sql(s"CREATE TABLE $relaxed.t (id INT) USING foo")
+ sql(s"CREATE TABLE $strict.t (id INT) USING foo")
+ withLegacyPolicy {
+ sql(s"INSERT INTO $relaxed.t VALUES (1)")
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1))
+ legacyRejected(sql(s"INSERT INTO $strict.t VALUES (1)"))
+ }
+ }
+ }
+
+ test("allowLegacyStoreAssignmentPolicy: UPDATE under LEGACY policy") {
+ withTable(s"$relaxed.t", s"$strict.t") {
+ sql(s"CREATE TABLE $relaxed.t (id INT, data STRING) USING foo")
+ sql(s"CREATE TABLE $strict.t (id INT, data STRING) USING foo")
+ sql(s"INSERT INTO $relaxed.t VALUES (1, 'a')")
+ sql(s"INSERT INTO $strict.t VALUES (1, 'a')")
+ withLegacyPolicy {
+ sql(s"UPDATE $relaxed.t SET data = 'b' WHERE id = 1")
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1, "b"))
+ legacyRejected(sql(s"UPDATE $strict.t SET data = 'b' WHERE id = 1"))
+ }
+ }
+ }
+
+ test("deferCastValidationToRuntime: INSERT of an ANSI-incompatible cast") {
+ withTable(s"$relaxed.t", s"$strict.t") {
+ sql(s"CREATE TABLE $relaxed.t (id INT) USING foo")
+ sql(s"CREATE TABLE $strict.t (id INT) USING foo")
+ withAnsiPolicy {
+ sql(s"INSERT INTO $relaxed.t VALUES ('1')")
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1))
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"INSERT INTO $strict.t VALUES ('1')")
+ },
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST",
+ parameters = Map(
+ "tableName" -> s"`$strict`.`t`",
+ "colName" -> "`id`",
+ "srcType" -> "\"STRING\"",
+ "targetType" -> "\"INT\""))
+ }
+ }
+ }
+
+ test("deferCastValidationToRuntime: UPDATE with an ANSI-incompatible cast") {
+ withTable(s"$relaxed.t", s"$strict.t") {
+ sql(s"CREATE TABLE $relaxed.t (id INT, data INT) USING foo")
+ sql(s"CREATE TABLE $strict.t (id INT, data INT) USING foo")
+ sql(s"INSERT INTO $relaxed.t VALUES (1, 0)")
+ sql(s"INSERT INTO $strict.t VALUES (1, 0)")
+ withAnsiPolicy {
+ sql(s"UPDATE $relaxed.t SET data = '5' WHERE id = 1")
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1, 5))
+ checkError(
+ exception = intercept[AnalysisException] {
+ sql(s"UPDATE $strict.t SET data = '5' WHERE id = 1")
+ },
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST",
+ parameters = Map(
+ "tableName" -> "``",
+ "colName" -> "`data`",
+ "srcType" -> "\"STRING\"",
+ "targetType" -> "\"INT\""))
+ }
+ }
+ }
+
+ test("allowLegacyStoreAssignmentPolicy: MERGE under LEGACY policy") {
+ withTable(s"$relaxed.t", s"$strict.t") {
+ sql(s"CREATE TABLE $relaxed.t (id INT, data STRING) USING foo")
+ sql(s"CREATE TABLE $strict.t (id INT, data STRING) USING foo")
+ sql(s"INSERT INTO $relaxed.t VALUES (1, 'a')")
+ sql(s"INSERT INTO $strict.t VALUES (1, 'a')")
+ def merge(target: String): String =
+ s"""MERGE INTO $target t
+ |USING (SELECT 1 AS id, 'b' AS data) s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.data = s.data""".stripMargin
+ withLegacyPolicy {
+ sql(merge(s"$relaxed.t"))
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1, "b"))
+ legacyRejected(sql(merge(s"$strict.t")))
+ }
+ }
+ }
+
+ test("deferCastValidationToRuntime: MERGE with an ANSI-incompatible cast") {
+ withTable(s"$relaxed.t", s"$strict.t") {
+ sql(s"CREATE TABLE $relaxed.t (id INT, data INT) USING foo")
+ sql(s"CREATE TABLE $strict.t (id INT, data INT) USING foo")
+ sql(s"INSERT INTO $relaxed.t VALUES (1, 0)")
+ sql(s"INSERT INTO $strict.t VALUES (1, 0)")
+ def merge(target: String): String =
+ s"""MERGE INTO $target t
+ |USING (SELECT 1 AS id, '5' AS data) s
+ |ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.data = s.data""".stripMargin
+ withAnsiPolicy {
+ sql(merge(s"$relaxed.t"))
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1, 5))
+ checkError(
+ exception = intercept[AnalysisException](sql(merge(s"$strict.t"))),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST",
+ parameters = Map(
+ "tableName" -> "``",
+ "colName" -> "`data`",
+ "srcType" -> "\"STRING\"",
+ "targetType" -> "\"INT\""))
+ }
+ }
+ }
+
+ test("deferCastValidationToRuntime: structurally impossible casts are still
rejected") {
+ withTable(s"$relaxed.t") {
+ sql(s"CREATE TABLE $relaxed.t (d DATE) USING foo")
+ withAnsiPolicy {
+ // BOOLEAN cannot be cast to DATE at all, so the write is rejected
even though the table
+ // defers store-assignment cast validation to runtime.
+ intercept[AnalysisException] {
+ sql(s"INSERT INTO $relaxed.t VALUES (true)")
+ }
+ }
+ }
+ }
+
+ private def appendByName(
+ catalog: String, targetSchema: StructType, source: DataFrame):
Try[Seq[Row]] = {
+ var result: Try[Seq[Row]] = Try(Seq.empty[Row])
+ withTable(s"$catalog.t") {
+ spark.createDataFrame(new java.util.ArrayList[Row](), targetSchema)
+ .writeTo(s"$catalog.t").create()
+ result = Try {
+ source.writeTo(s"$catalog.t").append()
+ spark.table(s"$catalog.t").collect().toSeq
+ }
+ }
+ result
+ }
+
+ private def assertRelaxedMatchesStrict(targetSchema: StructType, source:
DataFrame): Unit =
+ withAnsiPolicy {
+ val fromRelaxed = appendByName(relaxed, targetSchema, source)
+ val fromStrict = appendByName(strict, targetSchema, source)
+ (fromRelaxed, fromStrict) match {
+ case (Success(relaxedRows), Success(strictRows)) =>
+ assert(relaxedRows.map(_.toString).sorted ==
strictRows.map(_.toString).sorted,
+ s"relaxed=$relaxedRows strict=$strictRows")
+ case (Failure(relaxedError: AnalysisException), Failure(strictError:
AnalysisException)) =>
+ assert(relaxedError.getCondition == strictError.getCondition,
+ s"relaxed=${relaxedError.getCondition}
strict=${strictError.getCondition}")
+ case (Failure(_), Failure(_)) =>
Review Comment:
This accepts any two failures, including an NPE/runtime failure on the
relaxed side and the expected analysis failure on the strict side. The
`Success/Success` branch also means the test named `renamed nested struct field
is still rejected` does not actually require rejection. Please assert exact
outcomes per test. Also add non-foldable malformed/overflow values and
null-bearing inputs so the promised runtime failures and null checks are
exercised.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.connector.catalog;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Schema-alignment configuration for writes to a {@link Table}. This allows
connectors to
Review Comment:
This says the config applies to writes to a Table, but only `V2WriteCommand`
and row-level paths consult it. Micro-batch table sinks use the separate
`V2StreamingWriteCommand`, so `STREAMING_WRITE` never sees this config. Either
wire streaming schema alignment into the hook or explicitly scope the public
contract to batch/row-level writes. Please also document the changed ANSI
behavior in the DSv2 and ANSI guides.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java:
##########
@@ -0,0 +1,52 @@
+/*
+ * 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.connector.catalog;
+
+import org.apache.spark.annotation.Evolving;
+
+/**
+ * Schema-alignment configuration for writes to a {@link Table}. This allows
connectors to
+ * configure casting behavior and handling of schema mismatches during writes.
+ *
+ * @since 4.3.0
Review Comment:
The version is stale. `branch-4.x` had already moved to `4.4.0-SNAPSHOT`
before this PR opened, so this normal backportable API should say `@since
4.4.0` (or `5.0.0` only if this is explicitly master-only).
--
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]
