johanl-db commented on code in PR #58632:
URL: https://github.com/apache/spark/pull/58632#discussion_r4060128746
##########
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:
Good catch, I'm now threading the decision to `DataTypeUtils.canWrite` to
only allow specific cases.
E.p., struct field ordering and UDT validation are still enforced
##########
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 and other comments made me realize `LEGACY` isn't properly supported in
DSv2, so it's not just a matter of allowing it.
I completely removed allowing `LEGACY` from this PR, it now solely focuses
on configuring implicit cast validation. Leaving `LEGACY` for a follow up - if
it can be reasonably supported
##########
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:
Calling out explicitly that this applies to v2 batch write / row-level
writes and not streaming - schema alignment is the responsibility of the
connector in streaming today.
I made the interface internal in later changes so not documenting it in the
guide.
##########
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:
Removed, the API is now internal
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -3981,9 +3985,10 @@ class Analyzer(
}
}
- private def validateStoreAssignmentPolicy(): Unit = {
- // SPARK-28730: LEGACY store assignment policy is disallowed in data
source v2.
- if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY) {
+ private def validateStoreAssignmentPolicy(schemaAlignment:
SchemaAlignmentConfig): Unit = {
+ // SPARK-28730: LEGACY store assignment policy is disallowed in data
source v2 by default.
+ if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY &&
+ !schemaAlignment.allowLegacyStoreAssignmentPolicy()) {
Review Comment:
Not relevant anymore: I removed allowing `LEGACY` from this PR.
##########
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:
Removed override and now threading the config as needed
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala:
##########
@@ -137,7 +141,8 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
val value = matchingAssignments.head.value
val coerceMode = if (coerceNestedTypes) RECURSE else NONE
TableOutputResolver.resolveUpdate(
- "", value, actualAttr, conf, err => errors += err, colPath,
coerceMode)
+ "", value, actualAttr, conf, err => errors += err, colPath,
coerceMode,
+ deferAnsiCastValidationToRuntime =
schemaAlignment.deferAnsiCastValidationToRuntime())
Review Comment:
`chemaAlignment.deferAnsiCastValidationToRuntime()` is now called once and
passed down
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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 batch/row-level writes to a {@link
Table}. This allows
+ * connectors to configure casting behavior and handling of schema mismatches
during DSv2 writes.
+ * It is not consulted for streaming writes, which do not go through this
alignment path.
+ *
+ * @since 4.4.0
+ */
+@Evolving
+public interface SchemaAlignmentConfig {
Review Comment:
This is now an interface extending `Table` instead of directly adding a new
method to `Table` itself.
Also made it internal in the process.
##########
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:
Updated to make check stricter
##########
docs/sql-ref-ansi-compliance.md:
##########
@@ -226,6 +226,9 @@ INSERT INTO test VALUES (2147483648L);
org.apache.spark.SparkArithmeticException: [CAST_OVERFLOW_IN_TABLE_INSERT]
Fail to insert a value of "BIGINT" type into the "INT" type column `i` due to
an overflow. Use `try_cast` on the input value to tolerate overflow and return
NULL instead.
```
+By default, invalid source/target combinations are rejected during analysis.
+Data sources may instead defer this validation to execution time, so an
insertion is rejected only when a value is actually malformed or overflows, not
during analysis.
Review Comment:
I removed the doc since I made the interface internal in later changes.
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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 batch/row-level writes to a {@link
Table}. This allows
+ * connectors to configure casting behavior and handling of schema mismatches
during DSv2 writes.
+ * It is not consulted for streaming writes, which do not go through this
alignment path.
+ *
+ * @since 4.4.0
+ */
+@Evolving
+public interface SchemaAlignmentConfig {
+
+ /** The strict data source v2 configuration, returned by {@link Table} by
default. */
+ SchemaAlignmentConfig DEFAULT = new SchemaAlignmentConfig() {};
+
+ /**
+ * Whether {@code spark.sql.storeAssignmentPolicy=LEGACY} is allowed for
writes and row-level
+ * operations targeting this table. Data source v2 rejects LEGACY by
default; a table can decide
+ * to opt-out from this restriction.
+ */
+ default boolean allowLegacyStoreAssignmentPolicy() {
+ return false;
+ }
+
+ /**
+ * Whether the {@code ANSI} store-assignment cast check is deferred from
analysis to runtime under
+ * {@code spark.sql.storeAssignmentPolicy=ANSI}. When {@code true}, the
analyzer skips the
+ * store-assignment compatibility check and inserts an ANSI cast, so
malformed values or
+ * overflows surface at execution time.
+ */
+ default boolean deferAnsiCastValidationToRuntime() {
Review Comment:
I reworked the structure of the interface / config
##########
sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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 batch/row-level writes to a {@link
Table}. This allows
+ * connectors to configure casting behavior and handling of schema mismatches
during DSv2 writes.
+ * It is not consulted for streaming writes, which do not go through this
alignment path.
Review Comment:
Not that I know of.
The interplay between schema alignment and schema evolution is particularly
interesting, I suspect if we ever want to support schema evolution in DSv2
streaming, we may have to implement schema alignment also.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala:
##########
@@ -354,12 +369,17 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
byName: Boolean,
conf: SQLConf,
addError: String => Unit,
- colPath: Seq[String]): Boolean = {
+ colPath: Seq[String],
+ deferAnsiCastValidationToRuntime: Boolean): Boolean = {
conf.storeAssignmentPolicy match {
case StoreAssignmentPolicy.STRICT | StoreAssignmentPolicy.ANSI =>
+ // Always delegate to DataTypeUtils.canWrite so structural checks
(struct field names and
+ // nullability, array/map element nullability, field counts) still
run. Only the atomic
+ // ANSI store-assignment cast check is relaxed, via
deferAnsiCastValidationToRuntime.
DataTypeUtils.canWrite(
tableName, valueType, expectedType, byName, conf.resolver,
colPath.quoted,
- conf.storeAssignmentPolicy, addError)
+ conf.storeAssignmentPolicy, addError,
+ deferAnsiCastValidationToRuntime = deferAnsiCastValidationToRuntime)
case _ =>
true
Review Comment:
I dropped allowing LEGACY from this PR, it would require more work to be
properly supported.
--
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]
