This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new d3f4a73abebf [SPARK-59342][SQL] Allow configuring schema alignment
behavior for DSv2 writes
d3f4a73abebf is described below
commit d3f4a73abebf70c8dd2d61f94bbcc3972bbf76f6
Author: Johan Lasperas <[email protected]>
AuthorDate: Wed Oct 7 15:57:36 2026 -0700
[SPARK-59342][SQL] Allow configuring schema alignment behavior for DSv2
writes
### What changes were proposed in this pull request?
This change extends the DSv2 connector interface to allow connectors to
configure schema alignment behavior. For DSv2 batch / row-level writes, under
`spark.sql.storeAssignmentPolicy=ANSI`, implicit cast are validated at analysis
time, rejecting casts that are valid but may fail, e.g. `STRING`->`INT`.
This change allows connectors to opt into runtime checks instead of
analysis check: allow more casts during analysis (`STRING`-> any, complex type
-> STRING, ..) and fail at runtime on malformed / overflowing values.
New internal DSV2 interface:
```
public interface ConfigurableSchemaAlignment extends Table {
SchemaAlignmentConfig schemaAlignmentConfig();
}
public interface SchemaAlignmentConfig {
default AnsiStoreAssignmentCastCheck ansiStoreAssignmentCastCheck() {
// AT_ANALYSIS or AT_RUNTIME.
return AnsiStoreAssignmentCastCheck.AT_ANALYSIS;
}
}
```
### Why are the changes needed?
This allows connector to configure and opt out of strict schema alignement
behavior when needed.
### How was this patch tested?
Add Spark suite SchemaAlignmentConfigSuite covering the configuration
mechanism
Closes #58632 from johanl-db/schema-alignment-config.
Authored-by: Johan Lasperas <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
---
.../internal/connector/SchemaAlignmentConfig.java | 57 +++
.../SupportsConfigurableSchemaAlignment.java | 35 ++
.../spark/sql/catalyst/analysis/Analyzer.scala | 11 +-
.../sql/catalyst/analysis/AssignmentUtils.scala | 44 ++-
.../ResolveRowLevelCommandAssignments.scala | 32 +-
.../catalyst/analysis/TableOutputResolver.scala | 120 ++++---
.../spark/sql/catalyst/types/DataTypeUtils.scala | 45 ++-
.../connector/write/RowLevelOperationTable.scala | 9 +-
.../sql/connector/catalog/InMemoryBaseTable.scala | 7 +-
.../catalog/InMemoryRowLevelOperationTable.scala | 22 +-
.../InMemoryRowLevelOperationTableCatalog.scala | 6 +-
.../sql/connector/catalog/InMemoryTable.scala | 9 +-
.../connector/catalog/InMemoryTableCatalog.scala | 7 +-
...InMemoryTableCatalystRuntimeFilterCatalog.scala | 5 +-
.../catalog/InMemoryTableWithV2FilterCatalog.scala | 5 +-
.../apache/spark/sql/connector/catalog/txns.scala | 3 +-
.../types/DataTypeWriteCompatibilitySuite.scala | 148 ++++++++
.../connector/KeyGroupedPartitioningSuite.scala | 6 +-
.../catalog/SchemaAlignmentConfigSuite.scala | 389 +++++++++++++++++++++
19 files changed, 859 insertions(+), 101 deletions(-)
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/SchemaAlignmentConfig.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/SchemaAlignmentConfig.java
new file mode 100644
index 000000000000..16ef14670cb0
--- /dev/null
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/SchemaAlignmentConfig.java
@@ -0,0 +1,57 @@
+/*
+ * 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.internal.connector;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.catalog.Table;
+
+/**
+ * Schema alignment configuration for DSv2 batch/row-level writes to a {@link
Table}, exposed via
+ * {@link SupportsConfigurableSchemaAlignment}.
+ *
+ * @since 4.4.0
+ */
+@Evolving
+public class SchemaAlignmentConfig {
+
+ /** The default data source v2 configuration. */
+ public static final SchemaAlignmentConfig DEFAULT = new
SchemaAlignmentConfig();
+
+ /**
+ * When the {@code ANSI} store-assignment cast check runs for writes to a
table, under
+ * {@code spark.sql.storeAssignmentPolicy=ANSI}. Has no effect under the
{@code STRICT} policy.
+ */
+ public enum AnsiStoreAssignmentCastCheck {
+ /** (default) Reject incompatible casts at analysis. See {@code
Cast.canANSIStoreAssign}. */
+ ANALYSIS,
+
+ /**
+ * Insert an ANSI cast check, so malformed values or overflows fail at
execution time instead
+ * of being rejected during analysis.
+ */
+ RUNTIME;
+
+ /** The default ANSI store-assignment cast check. */
+ public static final AnsiStoreAssignmentCastCheck DEFAULT = ANALYSIS;
+ }
+
+ /** When the {@code ANSI} store-assignment cast check runs. */
+ public AnsiStoreAssignmentCastCheck ansiStoreAssignmentCastCheck() {
+ return AnsiStoreAssignmentCastCheck.DEFAULT;
+ }
+}
diff --git
a/sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/SupportsConfigurableSchemaAlignment.java
b/sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/SupportsConfigurableSchemaAlignment.java
new file mode 100644
index 000000000000..a92a431113c9
--- /dev/null
+++
b/sql/catalyst/src/main/java/org/apache/spark/sql/internal/connector/SupportsConfigurableSchemaAlignment.java
@@ -0,0 +1,35 @@
+/*
+ * 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.internal.connector;
+
+import org.apache.spark.annotation.Evolving;
+import org.apache.spark.sql.connector.catalog.Table;
+
+/**
+ * Allows connectors to configure schema alignment and casting behavior for
DSv2 batch/row-level
+ * writes to a {@link Table}. It is not consulted for streaming writes, where
alignment is
+ * delegated to the connector.
+ *
+ * @since 4.4.0
+ */
+@Evolving
+public interface SupportsConfigurableSchemaAlignment extends Table {
+
+ /** The schema alignment configuration for writes to this table. */
+ SchemaAlignmentConfig schemaAlignmentConfig();
+}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index 0efeda570b55..73f25cd9c2e4 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -59,10 +59,10 @@ import
org.apache.spark.sql.connector.catalog.functions.UnboundFunction
import org.apache.spark.sql.connector.catalog.procedures.{BoundProcedure,
ProcedureParameter, UnboundProcedure}
import org.apache.spark.sql.connector.expressions.{FieldReference,
IdentityTransform}
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation}
+import org.apache.spark.sql.execution.datasources.v2.{ChangelogTable,
DataSourceV2Relation, ExtractV2Table}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.{PartitionOverwriteMode,
StoreAssignmentPolicy}
-import org.apache.spark.sql.internal.connector.V1Function
+import org.apache.spark.sql.internal.connector.{SchemaAlignmentConfig,
SupportsConfigurableSchemaAlignment, V1Function}
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.util.ArrayImplicits._
@@ -4053,10 +4053,15 @@ class Analyzer(
case r: DataSourceV2Relation =>
GeneratedColumn.attachGenerationExpressions(r)
case _ => v2Write.table.output
}
+ val schemaAlignmentConfig = v2Write.table match {
+ case ExtractV2Table(table: SupportsConfigurableSchemaAlignment) =>
+ table.schemaAlignmentConfig()
+ case _ => SchemaAlignmentConfig.DEFAULT
+ }
val (projection, autoFilledGenCols) =
TableOutputResolver.resolveOutputColumnsWithGeneratedInfo(
v2Write.table.name, expected, v2Write.query, v2Write.isByName,
conf,
- defaultValueFillMode)
+ defaultValueFillMode,
schemaAlignmentConfig.ansiStoreAssignmentCastCheck())
if (projection != v2Write.query) {
val cleanedTable = v2Write.table match {
case r: DataSourceV2Relation =>
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala
index df4b0646ed42..b35c824e6e91 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/AssignmentUtils.scala
@@ -30,6 +30,7 @@ import
org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getDefaultValueE
import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
+import
org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
import org.apache.spark.sql.types.{DataType, StructType}
import org.apache.spark.util.ArrayImplicits._
@@ -57,13 +58,17 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
* (preserving existing fields).
* @param coerceNestedTypes whether to coerce nested types to match the
target type
* for complex types
+ * @param ansiStoreAssignmentCastCheck when the ANSI store-assignment cast
check runs:
+ * ANALYSIS (default) rejects an unsafe
cast at analysis
+ * time; RUNTIME rejects invalid values
at execution time.
* @return aligned update assignments that match table attributes
*/
def alignUpdateAssignments(
attrs: Seq[Attribute],
assignments: Seq[Assignment],
fromStar: Boolean,
- coerceNestedTypes: Boolean): Seq[Assignment] = {
+ coerceNestedTypes: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck):
Seq[Assignment] = {
val errors = new mutable.ArrayBuffer[String]()
@@ -75,7 +80,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport
{
addError = err => errors += err,
colPath = Seq(attr.name),
coerceNestedTypes,
- fromStar)
+ fromStar,
+ ansiStoreAssignmentCastCheck)
}
if (errors.nonEmpty) {
@@ -98,12 +104,16 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
* @param assignments insert assignments to align
* @param coerceNestedTypes whether to coerce nested types to match the
target type
* for complex types
+ * @param ansiStoreAssignmentCastCheck when the ANSI store-assignment cast
check runs:
+ * ANALYSIS (default) rejects an unsafe
cast at analysis
+ * time; RUNTIME rejects invalid values
at execution time.
* @return aligned insert assignments that match table attributes
*/
def alignInsertAssignments(
attrs: Seq[Attribute],
assignments: Seq[Assignment],
- coerceNestedTypes: Boolean = false): Seq[Assignment] = {
+ coerceNestedTypes: Boolean = false,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck):
Seq[Assignment] = {
val errors = new mutable.ArrayBuffer[String]()
@@ -137,7 +147,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,
+ ansiStoreAssignmentCastCheck)
}
Assignment(attr, resolvedValue)
}
@@ -160,7 +171,8 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
addError: String => Unit,
colPath: Seq[String],
coerceNestedTypes: Boolean = false,
- updateStar: Boolean = false): Expression = {
+ updateStar: Boolean = false,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression
= {
val (exactAssignments, otherAssignments) = assignments.partition {
assignment =>
assignment.key.semanticEquals(colExpr)
@@ -188,21 +200,22 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
case _: StructType =>
// Expand assignments to leaf fields (fixNullExpansion is applied
inside)
applyNestedFieldAssignments(col, colExpr, value, addError, colPath,
- coerceNestedTypes)
+ coerceNestedTypes, ansiStoreAssignmentCastCheck)
case _ =>
// For non-struct types, resolve directly
val coerceMode = if (coerceNestedTypes) RECURSE else NONE
TableOutputResolver.resolveUpdate("", value, col, conf, addError,
colPath,
- coerceMode)
+ coerceMode, ansiStoreAssignmentCastCheck)
}
} else {
val value = exactAssignments.head.value
val coerceMode = if (coerceNestedTypes) RECURSE else NONE
TableOutputResolver.resolveUpdate("", value, col, conf, addError,
- colPath, coerceMode)
+ colPath, coerceMode, ansiStoreAssignmentCastCheck)
}
} else {
- applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath,
coerceNestedTypes)
+ applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath,
coerceNestedTypes,
+ ansiStoreAssignmentCastCheck)
}
}
@@ -212,7 +225,8 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
assignments: Seq[Assignment],
addError: String => Unit,
colPath: Seq[String],
- coerceNestedTypes: Boolean): Expression = {
+ coerceNestedTypes: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression
= {
col.dataType match {
case structType: StructType =>
@@ -222,7 +236,7 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
}
val updatedFieldExprs = fieldAttrs.zip(fieldExprs).map { case
(fieldAttr, fieldExpr) =>
applyAssignments(fieldAttr, fieldExpr, assignments, addError,
colPath :+ fieldAttr.name,
- coerceNestedTypes)
+ coerceNestedTypes, ansiStoreAssignmentCastCheck =
ansiStoreAssignmentCastCheck)
}
toNamedStruct(structType, updatedFieldExprs)
@@ -240,7 +254,8 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
value: Expression,
addError: String => Unit,
colPath: Seq[String],
- coerceNestedTypes: Boolean): Expression = {
+ coerceNestedTypes: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression
= {
col.dataType match {
case structType: StructType =>
@@ -273,12 +288,13 @@ object AssignmentUtils extends SQLConfHelper with
CastSupport {
case _: StructType =>
// Field is a struct, recurse
applyNestedFieldAssignments(fieldAttr, targetFieldExpr,
- sourceFieldValue, addError, fieldPath, coerceNestedTypes)
+ sourceFieldValue, addError, fieldPath, coerceNestedTypes,
+ ansiStoreAssignmentCastCheck)
case _ =>
// Field is not a struct, resolve with TableOutputResolver
val coerceMode = if (coerceNestedTypes) RECURSE else NONE
TableOutputResolver.resolveUpdate("", sourceFieldValue,
fieldAttr, conf, addError,
- fieldPath, coerceMode)
+ fieldPath, coerceMode, ansiStoreAssignmentCastCheck)
}
}
val namedStruct = toNamedStruct(structType, updatedFieldExprs)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
index 76035ea819ff..a31ac092e279 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveRowLevelCommandAssignments.scala
@@ -25,8 +25,11 @@ import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.COMMAND
import org.apache.spark.sql.catalyst.util.CharVarcharUtils
import org.apache.spark.sql.errors.QueryCompilationErrors
-import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.execution.datasources.v2.{DataSourceV2Relation,
ExtractV2Table}
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig
+import
org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
+import
org.apache.spark.sql.internal.connector.SupportsConfigurableSchemaAlignment
/**
* A rule that resolves assignments in row-level commands.
@@ -43,7 +46,8 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
validateStoreAssignmentPolicy()
val newTable = cleanAttrMetadata(u.table)
val newAssignments =
AssignmentUtils.alignUpdateAssignments(u.table.output, u.assignments,
- fromStar = false, coerceNestedTypes = false)
+ fromStar = false, coerceNestedTypes = false,
+ ansiStoreAssignmentCastCheck = ansiStoreAssignmentCastCheck(u.table))
u.copy(table = newTable, assignments = newAssignments)
case u: UpdateTable if !u.skipSchemaResolution && u.resolved && !u.aligned
=>
@@ -52,20 +56,24 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
case m: MergeIntoTable if m.rewritable && shouldAlignAssignments(m) &&
containsFinalSchema(m) =>
validateStoreAssignmentPolicy()
val coerceNestedTypes = conf.coerceMergeNestedTypes &&
m.withSchemaEvolution
+ val castCheck = ansiStoreAssignmentCastCheck(m.targetTable)
m.copy(
targetTable = cleanAttrMetadata(m.targetTable),
matchedActions = alignActions(
m.targetTable.output,
m.matchedActions,
- coerceNestedTypes),
+ coerceNestedTypes,
+ castCheck),
notMatchedActions = alignActions(
m.targetTable.output,
m.notMatchedActions,
- coerceNestedTypes),
+ coerceNestedTypes,
+ castCheck),
notMatchedBySourceActions = alignActions(
m.targetTable.output,
m.notMatchedBySourceActions,
- coerceNestedTypes))
+ coerceNestedTypes,
+ castCheck))
case m: MergeIntoTable if shouldAlignAssignments(m) &&
containsFinalSchema(m) =>
resolveAssignments(m)
@@ -79,6 +87,13 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
!m.schemaEvolutionEnabled || (m.schemaEvolutionReady &&
m.pendingSchemaChanges.isEmpty)
}
+ private def ansiStoreAssignmentCastCheck(target: LogicalPlan):
AnsiStoreAssignmentCastCheck = {
+ target.collectFirst {
+ case ExtractV2Table(table: SupportsConfigurableSchemaAlignment) =>
+ table.schemaAlignmentConfig().ansiStoreAssignmentCastCheck()
+ }.getOrElse(SchemaAlignmentConfig.DEFAULT.ansiStoreAssignmentCastCheck())
+ }
+
private def validateStoreAssignmentPolicy(): Unit = {
// SPARK-28730: LEGACY store assignment policy is disallowed in data
source v2
if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY) {
@@ -127,16 +142,17 @@ object ResolveRowLevelCommandAssignments extends
Rule[LogicalPlan] {
private def alignActions(
attrs: Seq[Attribute],
actions: Seq[MergeAction],
- coerceNestedTypes: Boolean): Seq[MergeAction] = {
+ coerceNestedTypes: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck):
Seq[MergeAction] = {
actions.map {
case u @ UpdateAction(_, assignments, fromStar) =>
u.copy(assignments = AssignmentUtils.alignUpdateAssignments(attrs,
assignments,
- fromStar, coerceNestedTypes))
+ fromStar, coerceNestedTypes, ansiStoreAssignmentCastCheck))
case d: DeleteAction =>
d
case i @ InsertAction(_, assignments) =>
i.copy(assignments = AssignmentUtils.alignInsertAssignments(attrs,
assignments,
- coerceNestedTypes))
+ coerceNestedTypes, ansiStoreAssignmentCastCheck))
case other =>
throw new AnalysisException(
errorClass = "_LEGACY_ERROR_TEMP_3052",
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
index 933513654dc1..719e32b02c34 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/TableOutputResolver.scala
@@ -36,6 +36,7 @@ import
org.apache.spark.sql.connector.catalog.CatalogV2Implicits._
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import
org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
import org.apache.spark.sql.types.{ArrayType, DataType, DecimalType,
IntegralType, MapType, StructType, UserDefinedType}
object TableOutputResolver extends SQLConfHelper with Logging {
@@ -94,9 +95,12 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
query: LogicalPlan,
byName: Boolean,
conf: SQLConf,
- defaultValueFillMode: DefaultValueFillMode.Value = NONE): LogicalPlan = {
+ defaultValueFillMode: DefaultValueFillMode.Value = NONE,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck =
+ AnsiStoreAssignmentCastCheck.DEFAULT): LogicalPlan = {
resolveOutputColumnsInternal(
- tableName, expected, query, byName, conf, defaultValueFillMode)._1
+ tableName, expected, query, byName, conf, defaultValueFillMode,
+ ansiStoreAssignmentCastCheck)._1
}
/**
@@ -111,10 +115,12 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
query: LogicalPlan,
byName: Boolean,
conf: SQLConf,
- defaultValueFillMode: DefaultValueFillMode.Value = NONE
+ defaultValueFillMode: DefaultValueFillMode.Value = NONE,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck
): (LogicalPlan, Set[String]) = {
resolveOutputColumnsInternal(
- tableName, expected, query, byName, conf, defaultValueFillMode)
+ tableName, expected, query, byName, conf, defaultValueFillMode,
+ ansiStoreAssignmentCastCheck)
}
private def resolveOutputColumnsInternal(
@@ -123,7 +129,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
query: LogicalPlan,
byName: Boolean,
conf: SQLConf,
- defaultValueFillMode: DefaultValueFillMode.Value
+ defaultValueFillMode: DefaultValueFillMode.Value,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck
): (LogicalPlan, Set[String]) = {
if (expected.size < query.output.size) {
@@ -149,7 +156,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
errors += _,
Nil,
defaultValueFillMode,
- enforceFullOutput = true)
+ enforceFullOutput = true,
+ ansiStoreAssignmentCastCheck)
} else {
if (expected.size > query.output.size && !fillDefaultValue) {
throw QueryCompilationErrors.cannotWriteNotEnoughColumnsToTableError(
@@ -157,7 +165,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
}
resolveColumnsByPosition(
tableName, query.output, expected, conf, errors += _,
- fillDefaultValue = fillDefaultValue)
+ fillDefaultValue = fillDefaultValue,
+ ansiStoreAssignmentCastCheck = ansiStoreAssignmentCastCheck)
}
if (errors.nonEmpty) {
@@ -180,14 +189,16 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String],
- defaultValueFillMode: DefaultValueFillMode.Value): Expression = {
+ defaultValueFillMode: DefaultValueFillMode.Value,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression
= {
val fillChildDefaultValue = defaultValueFillMode == RECURSE
(value.dataType, col.dataType) match {
// no need to reorder inner fields or cast if types are already
compatible
case (valueType, colType) if
DataType.equalsIgnoreCompatibleNullability(valueType, colType) =>
val canWriteExpr = canWrite(
- tableName, valueType, colType, byName = true, conf, addError,
colPath)
+ tableName, valueType, colType, byName = true, conf, addError,
colPath,
+ ansiStoreAssignmentCastCheck)
if (canWriteExpr) {
val nullsHandled = checkNullability(value, col, conf, colPath)
applyColumnMetadata(nullsHandled, col)
@@ -196,21 +207,22 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
}
case (valueType: StructType, colType: StructType) =>
val resolvedValue = resolveStructType(
- tableName, value, valueType, col, colType,
- byName = true, conf, addError, colPath, fillChildDefaultValue,
enforceFullOutput = false)
+ tableName, value, valueType, col, colType, byName = true, conf,
addError, colPath,
+ fillChildDefaultValue, enforceFullOutput = false,
ansiStoreAssignmentCastCheck)
resolvedValue.getOrElse(value)
case (valueType: ArrayType, colType: ArrayType) =>
val resolvedValue = resolveArrayType(
- tableName, value, valueType, col, colType,
- byName = true, conf, addError, colPath, fillChildDefaultValue,
enforceFullOutput = false)
+ tableName, value, valueType, col, colType, byName = true, conf,
addError, colPath,
+ fillChildDefaultValue, enforceFullOutput = false,
ansiStoreAssignmentCastCheck)
resolvedValue.getOrElse(value)
case (valueType: MapType, colType: MapType) =>
val resolvedValue = resolveMapType(
- tableName, value, valueType, col, colType,
- byName = true, conf, addError, colPath, fillChildDefaultValue,
enforceFullOutput = false)
+ tableName, value, valueType, col, colType, byName = true, conf,
addError, colPath,
+ fillChildDefaultValue, enforceFullOutput = false,
ansiStoreAssignmentCastCheck)
resolvedValue.getOrElse(value)
case _ =>
- checkUpdate(tableName, value, col, conf, addError, colPath)
+ checkUpdate(
+ tableName, value, col, conf, addError, colPath,
ansiStoreAssignmentCastCheck)
}
}
@@ -220,7 +232,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
attr: Attribute,
conf: SQLConf,
addError: String => Unit,
- colPath: Seq[String]): Expression = {
+ colPath: Seq[String],
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Expression
= {
val attrTypeHasCharVarchar = CharVarcharUtils.hasCharVarchar(attr.dataType)
val attrTypeWithoutCharVarchar = if (attrTypeHasCharVarchar) {
@@ -231,7 +244,7 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
val canWriteValue = canWrite(
tableName, value.dataType, attrTypeWithoutCharVarchar,
- byName = true, conf, addError, colPath)
+ byName = true, conf, addError, colPath, ansiStoreAssignmentCastCheck)
if (canWriteValue) {
val nullCheckedValue = checkNullability(value, attr, conf, colPath)
@@ -354,12 +367,13 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
byName: Boolean,
conf: SQLConf,
addError: String => Unit,
- colPath: Seq[String]): Boolean = {
+ colPath: Seq[String],
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck): Boolean = {
conf.storeAssignmentPolicy match {
case StoreAssignmentPolicy.STRICT | StoreAssignmentPolicy.ANSI =>
DataTypeUtils.canWrite(
tableName, valueType, expectedType, byName, conf.resolver,
colPath.quoted,
- conf.storeAssignmentPolicy, addError)
+ conf.storeAssignmentPolicy, addError, ansiStoreAssignmentCastCheck)
case _ =>
true
}
@@ -373,7 +387,9 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
addError: String => Unit,
colPath: Seq[String] = Nil,
defaultValueFillMode: DefaultValueFillMode.Value,
- enforceFullOutput: Boolean = false): (Seq[NamedExpression], Set[String])
= {
+ enforceFullOutput: Boolean = false,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck
+ ): (Seq[NamedExpression], Set[String]) = {
// Names of generated columns that were auto-filled (not provided by the
user). Only populated
// for top-level columns, since generated columns cannot be nested.
val autoFilledGenCols = mutable.Set.empty[String]
@@ -423,18 +439,22 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
case (matchedType: StructType, expectedType: StructType) =>
resolveStructType(
tableName, matchedCol, matchedType, actualExpectedCol,
expectedType,
- byName = true, conf, addError, newColPath,
childFillDefaultValue, enforceFullOutput)
+ byName = true, conf, addError, newColPath,
childFillDefaultValue, enforceFullOutput,
+ ansiStoreAssignmentCastCheck)
case (matchedType: ArrayType, expectedType: ArrayType) =>
resolveArrayType(
tableName, matchedCol, matchedType, actualExpectedCol,
expectedType,
- byName = true, conf, addError, newColPath,
childFillDefaultValue, enforceFullOutput)
+ byName = true, conf, addError, newColPath,
childFillDefaultValue, enforceFullOutput,
+ ansiStoreAssignmentCastCheck)
case (matchedType: MapType, expectedType: MapType) =>
resolveMapType(
tableName, matchedCol, matchedType, actualExpectedCol,
expectedType,
- byName = true, conf, addError, newColPath,
childFillDefaultValue, enforceFullOutput)
+ byName = true, conf, addError, newColPath,
childFillDefaultValue, enforceFullOutput,
+ ansiStoreAssignmentCastCheck)
case _ =>
checkField(
- tableName, actualExpectedCol, matchedCol, byName = true, conf,
addError, newColPath)
+ tableName, actualExpectedCol, matchedCol, byName = true, conf,
addError, newColPath,
+ ansiStoreAssignmentCastCheck)
}
}
}
@@ -488,7 +508,9 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
conf: SQLConf,
addError: String => Unit,
colPath: Seq[String] = Nil,
- fillDefaultValue: Boolean = false): (Seq[NamedExpression], Set[String])
= {
+ fillDefaultValue: Boolean = false,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck
+ ): (Seq[NamedExpression], Set[String]) = {
// Names of generated columns that were auto-filled (not provided by the
user). Only populated
// for top-level columns, since generated columns cannot be nested.
val autoFilledGenCols = mutable.Set.empty[String]
@@ -528,17 +550,21 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
case (inputType: StructType, expectedType: StructType) =>
resolveStructType(
tableName, inputCol, inputType, expectedCol, expectedType,
- byName = false, conf, addError, newColPath, fillDefaultValue,
enforceFullOutput = true)
+ byName = false, conf, addError, newColPath, fillDefaultValue,
enforceFullOutput = true,
+ ansiStoreAssignmentCastCheck)
case (inputType: ArrayType, expectedType: ArrayType) =>
resolveArrayType(
tableName, inputCol, inputType, expectedCol, expectedType,
- byName = false, conf, addError, newColPath, fillDefaultValue,
enforceFullOutput = true)
+ byName = false, conf, addError, newColPath, fillDefaultValue,
enforceFullOutput = true,
+ ansiStoreAssignmentCastCheck)
case (inputType: MapType, expectedType: MapType) =>
resolveMapType(
tableName, inputCol, inputType, expectedCol, expectedType,
- byName = false, conf, addError, newColPath, fillDefaultValue,
enforceFullOutput = true)
+ byName = false, conf, addError, newColPath, fillDefaultValue,
enforceFullOutput = true,
+ ansiStoreAssignmentCastCheck)
case _ =>
- checkField(tableName, expectedCol, inputCol, byName = false, conf,
addError, newColPath)
+ checkField(tableName, expectedCol, inputCol, byName = false, conf,
addError, newColPath,
+ ansiStoreAssignmentCastCheck)
}
}
@@ -622,7 +648,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
addError: String => Unit,
colPath: Seq[String],
fillDefaultValue: Boolean,
- enforceFullOutput: Boolean): Option[NamedExpression] = {
+ enforceFullOutput: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck):
Option[NamedExpression] = {
val nullCheckedInput = checkNullability(input, expected, conf, colPath)
val fields = inputType.zipWithIndex.map { case (f, i) =>
Alias(GetStructField(nullCheckedInput, i, Some(f.name)), f.name)()
@@ -631,10 +658,11 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
// Generated columns cannot be nested, so the auto-filled set is always
empty here.
val (resolved, _) = if (byName) {
reorderColumnsByName(tableName, fields, toAttributes(expectedType),
conf, addError, colPath,
- defaultValueMode, enforceFullOutput)
+ defaultValueMode, enforceFullOutput, ansiStoreAssignmentCastCheck)
} else {
resolveColumnsByPosition(
- tableName, fields, toAttributes(expectedType), conf, addError,
colPath, fillDefaultValue)
+ tableName, fields, toAttributes(expectedType), conf, addError,
colPath, fillDefaultValue,
+ ansiStoreAssignmentCastCheck)
}
if (resolved.length == expectedType.length) {
val struct = CreateStruct(resolved)
@@ -665,7 +693,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
addError: String => Unit,
colPath: Seq[String],
fillDefaultValue: Boolean,
- enforceFullOutput: Boolean): Option[NamedExpression] = {
+ enforceFullOutput: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck):
Option[NamedExpression] = {
val nullCheckedInput = checkNullability(input, expected, conf, colPath)
val param = NamedLambdaVariable("element", inputType.elementType,
inputType.containsNull)
val fakeAttr =
@@ -674,10 +703,11 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
val (res, _) = if (byName) {
val defaultValueMode = if (fillDefaultValue) RECURSE else NONE
reorderColumnsByName(tableName, Seq(param), Seq(fakeAttr), conf,
addError, colPath,
- defaultValueMode, enforceFullOutput)
+ defaultValueMode, enforceFullOutput, ansiStoreAssignmentCastCheck)
} else {
resolveColumnsByPosition(
- tableName, Seq(param), Seq(fakeAttr), conf, addError, colPath,
fillDefaultValue)
+ tableName, Seq(param), Seq(fakeAttr), conf, addError, colPath,
fillDefaultValue,
+ ansiStoreAssignmentCastCheck)
}
if (res.length == 1) {
val castedArray =
@@ -708,7 +738,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
addError: String => Unit,
colPath: Seq[String],
fillDefaultValue: Boolean,
- enforceFullOutput: Boolean): Option[NamedExpression] = {
+ enforceFullOutput: Boolean,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck):
Option[NamedExpression] = {
val nullCheckedInput = checkNullability(input, expected, conf, colPath)
val keyParam = NamedLambdaVariable("key", inputType.keyType, nullable =
false)
@@ -717,10 +748,11 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
// Generated columns cannot be nested, so the auto-filled set is always
empty here.
val (resKey, _) = if (byName) {
reorderColumnsByName(tableName, Seq(keyParam), Seq(fakeKeyAttr), conf,
addError, colPath,
- defaultValueFillMode, enforceFullOutput)
+ defaultValueFillMode, enforceFullOutput, ansiStoreAssignmentCastCheck)
} else {
resolveColumnsByPosition(
- tableName, Seq(keyParam), Seq(fakeKeyAttr), conf, addError, colPath,
fillDefaultValue)
+ tableName, Seq(keyParam), Seq(fakeKeyAttr), conf, addError, colPath,
fillDefaultValue,
+ ansiStoreAssignmentCastCheck)
}
val valueParam =
@@ -730,10 +762,11 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
// Generated columns cannot be nested, so the auto-filled set is always
empty here.
val (resValue, _) = if (byName) {
reorderColumnsByName(tableName, Seq(valueParam), Seq(fakeValueAttr),
conf, addError, colPath,
- defaultValueFillMode, enforceFullOutput)
+ defaultValueFillMode, enforceFullOutput, ansiStoreAssignmentCastCheck)
} else {
resolveColumnsByPosition(
- tableName, Seq(valueParam), Seq(fakeValueAttr), conf, addError,
colPath, fillDefaultValue)
+ tableName, Seq(valueParam), Seq(fakeValueAttr), conf, addError,
colPath, fillDefaultValue,
+ ansiStoreAssignmentCastCheck)
}
if (resKey.length == 1 && resValue.length == 1) {
@@ -833,7 +866,8 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
byName: Boolean,
conf: SQLConf,
addError: String => Unit,
- colPath: Seq[String]): Option[NamedExpression] = {
+ colPath: Seq[String],
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck):
Option[NamedExpression] = {
val attrTypeHasCharVarchar =
CharVarcharUtils.hasCharVarchar(tableAttr.dataType)
val attrTypeWithoutCharVarchar = if (attrTypeHasCharVarchar) {
@@ -844,7 +878,7 @@ object TableOutputResolver extends SQLConfHelper with
Logging {
val canWriteExpr = canWrite(
tableName, queryExpr.dataType, attrTypeWithoutCharVarchar,
- byName, conf, addError, colPath)
+ byName, conf, addError, colPath, ansiStoreAssignmentCastCheck)
if (canWriteExpr) {
val prepared =
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/DataTypeUtils.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/DataTypeUtils.scala
index f7ab8b06baf5..625a6720c03b 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/DataTypeUtils.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/types/DataTypeUtils.scala
@@ -25,6 +25,8 @@ import org.apache.spark.sql.connector.catalog.Column
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy.{ANSI,
STRICT}
+import
org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
+import
org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck.{ANALYSIS,
RUNTIME}
import org.apache.spark.sql.types._
import org.apache.spark.sql.types.DecimalType.{forType, fromDecimal}
@@ -111,13 +113,15 @@ object DataTypeUtils {
resolver: Resolver,
context: String,
storeAssignmentPolicy: StoreAssignmentPolicy.Value,
- addError: String => Unit): Boolean = {
+ addError: String => Unit,
+ ansiStoreAssignmentCastCheck: AnsiStoreAssignmentCastCheck =
+ AnsiStoreAssignmentCastCheck.DEFAULT): Boolean = {
(write, read) match {
case (wArr: ArrayType, rArr: ArrayType) =>
// run compatibility check first to produce all error messages
val typesCompatible = canWrite(
tableName, wArr.elementType, rArr.elementType, byName, resolver,
context + ".element",
- storeAssignmentPolicy, addError)
+ storeAssignmentPolicy, addError, ansiStoreAssignmentCastCheck)
if (wArr.containsNull && !rArr.containsNull) {
throw
QueryCompilationErrors.incompatibleDataToTableNullableArrayElementsError(
@@ -134,10 +138,10 @@ object DataTypeUtils {
// run compatibility check first to produce all error messages
val keyCompatible = canWrite(
tableName, wMap.keyType, rMap.keyType, byName, resolver, context +
".key",
- storeAssignmentPolicy, addError)
+ storeAssignmentPolicy, addError, ansiStoreAssignmentCastCheck)
val valueCompatible = canWrite(
tableName, wMap.valueType, rMap.valueType, byName, resolver, context
+ ".value",
- storeAssignmentPolicy, addError)
+ storeAssignmentPolicy, addError, ansiStoreAssignmentCastCheck)
if (wMap.valueContainsNull && !rMap.valueContainsNull) {
throw
QueryCompilationErrors.incompatibleDataToTableNullableMapValuesError(
@@ -155,7 +159,7 @@ object DataTypeUtils {
val fieldContext = s"$context.${rField.name}"
val typesCompatible = canWrite(
tableName, wField.dataType, rField.dataType, byName, resolver,
fieldContext,
- storeAssignmentPolicy, addError)
+ storeAssignmentPolicy, addError, ansiStoreAssignmentCastCheck)
if (byName && !nameMatch) {
throw
QueryCompilationErrors.incompatibleDataToTableUnexpectedColumnNameError(
@@ -198,7 +202,8 @@ object DataTypeUtils {
case (_: NullType, _) if storeAssignmentPolicy == ANSI => true
- case (w: AtomicType, r: AtomicType) if storeAssignmentPolicy == ANSI =>
+ case (w: AtomicType, r: AtomicType)
+ if storeAssignmentPolicy == ANSI && ansiStoreAssignmentCastCheck ==
ANALYSIS =>
if (!Cast.canANSIStoreAssign(w, r)) {
throw
QueryCompilationErrors.incompatibleDataToTableCannotSafelyCastError(
tableName, context, w.catalogString, r.catalogString
@@ -213,13 +218,34 @@ object DataTypeUtils {
// If write-side data type is a user-defined type, check with its
underlying data type.
case (w, r) if w.isInstanceOf[UserDefinedType[_]] &&
!r.isInstanceOf[UserDefinedType[_]] =>
canWrite(tableName, w.asInstanceOf[UserDefinedType[_]].sqlType, r,
byName, resolver,
- context, storeAssignmentPolicy, addError)
+ context, storeAssignmentPolicy, addError,
ansiStoreAssignmentCastCheck)
// If read-side data type is a user-defined type, check with its
underlying data type.
case (w, r) if r.isInstanceOf[UserDefinedType[_]] &&
!w.isInstanceOf[UserDefinedType[_]] =>
canWrite(tableName, w, r.asInstanceOf[UserDefinedType[_]].sqlType,
byName, resolver,
- context, storeAssignmentPolicy, addError)
-
+ context, storeAssignmentPolicy, addError,
ansiStoreAssignmentCastCheck)
+
+ // RUNTIME defers to the inserted ANSI cast, which fails on malformed or
overflowing
+ // values at runtime. Clearly invalid conversion are still rejected
either in an earlier
+ // branch of this match clause, or in checkAnalysis.
+ case (w, r) if storeAssignmentPolicy == ANSI &&
ansiStoreAssignmentCastCheck == RUNTIME =>
+ (w, r) match {
+ // Some casts that pass checkAnalysis don't behave correctly at
runtime. Keep rejecting
+ // them for now:
+ // - Long/decimal -> timestamp can silently overflow.
+ // - Variant -> complex types doesn't enforce nested field
non-nullability.
+ case (LongType | _: DecimalType, TimestampType) |
+ (VariantType, _: StructType | _: ArrayType | _: MapType) =>
+ throw
QueryCompilationErrors.incompatibleDataToTableCannotSafelyCastError(
+ tableName, context, w.catalogString, r.catalogString)
+ // - Building a variant from TIME or nanosecond timestamps isn't
supported, whether the
+ // value is top-level or nested in an array.
+ case (_, VariantType) if w.existsRecursively(
+ t => t.isInstanceOf[TimeType] ||
t.isInstanceOf[AnyTimestampNanoType]) =>
+ throw
QueryCompilationErrors.incompatibleDataToTableCannotSafelyCastError(
+ tableName, context, w.catalogString, r.catalogString)
+ case _ => true
+ }
case (w, r) =>
throw
QueryCompilationErrors.incompatibleDataToTableCannotSafelyCastError(
tableName, context, w.catalogString, r.catalogString
@@ -331,4 +357,3 @@ object DataTypeUtils {
}
}
}
-
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala
index 0ed78b667db3..f8a903476819 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/connector/write/RowLevelOperationTable.scala
@@ -23,6 +23,7 @@ import org.apache.spark.sql.connector.catalog.{Column,
SupportsRead, SupportsRow
import org.apache.spark.sql.connector.catalog.constraints.Constraint
import org.apache.spark.sql.connector.expressions.Transform
import org.apache.spark.sql.connector.read.ScanBuilder
+import org.apache.spark.sql.internal.connector.{SchemaAlignmentConfig,
SupportsConfigurableSchemaAlignment}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
/**
@@ -35,7 +36,8 @@ import org.apache.spark.sql.util.CaseInsensitiveStringMap
*/
private[sql] case class RowLevelOperationTable(
table: Table with SupportsRowLevelOperations,
- operation: RowLevelOperation) extends Table with SupportsRead with
SupportsWrite {
+ operation: RowLevelOperation)
+ extends Table with SupportsRead with SupportsWrite with
SupportsConfigurableSchemaAlignment {
override def name: String = table.name
override def columns: Array[Column] = table.columns()
@@ -44,6 +46,11 @@ private[sql] case class RowLevelOperationTable(
override def partitioning(): Array[Transform] = table.partitioning()
override def toString: String = table.toString
+ override def schemaAlignmentConfig(): SchemaAlignmentConfig = table match {
+ case t: SupportsConfigurableSchemaAlignment => t.schemaAlignmentConfig()
+ case _ => SchemaAlignmentConfig.DEFAULT
+ }
+
override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder
= {
operation.newScanBuilder(options)
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
index 7f2f57e3438e..091a1a441f3b 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryBaseTable.scala
@@ -44,7 +44,7 @@ import
org.apache.spark.sql.connector.read.streaming.{MicroBatchStream, Offset}
import org.apache.spark.sql.connector.write._
import
org.apache.spark.sql.connector.write.streaming.{StreamingDataWriterFactory,
StreamingWrite}
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.internal.connector.{ColumnImpl,
SupportsRuntimeCatalystFiltering, SupportsStreamingUpdateAsAppend}
+import org.apache.spark.sql.internal.connector.{ColumnImpl,
SchemaAlignmentConfig, SupportsConfigurableSchemaAlignment,
SupportsRuntimeCatalystFiltering, SupportsStreamingUpdateAsAppend}
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -65,9 +65,10 @@ abstract class InMemoryBaseTable(
val numPartitions: Option[Int] = None,
val advisoryPartitionSize: Option[Long] = None,
val isDistributionStrictlyRequired: Boolean = true,
- val numRowsPerSplit: Int = Int.MaxValue)
+ val numRowsPerSplit: Int = Int.MaxValue,
+ override val schemaAlignmentConfig: SchemaAlignmentConfig =
SchemaAlignmentConfig.DEFAULT)
extends Table with SupportsRead with SupportsWrite with
SupportsMetadataColumns
- with SupportsSchemaEvolution {
+ with SupportsSchemaEvolution with SupportsConfigurableSchemaAlignment {
// Tracks the current version number of the table.
protected var tableVersion: Int = 0
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
index c08dfef04a5b..3e04ab719b7b 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTable.scala
@@ -28,6 +28,7 @@ import
org.apache.spark.sql.connector.expressions.filter.Predicate
import org.apache.spark.sql.connector.read.{InputPartition, Scan, ScanBuilder}
import org.apache.spark.sql.connector.write.{BatchWrite, DeltaBatchWrite,
DeltaWrite, DeltaWriteBuilder, DeltaWriter, DeltaWriterFactory,
LogicalWriteInfo, PhysicalWriteInfo, RequiresDistributionAndOrdering,
RowLevelOperation, RowLevelOperationBuilder, RowLevelOperationInfo,
SupportsDelta, Write, WriteBuilder, WriterCommitMessage}
import org.apache.spark.sql.connector.write.RowLevelOperation.Command
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.unsafe.types.UTF8String
@@ -47,14 +48,16 @@ class InMemoryRowLevelOperationTable private (
partitioning: Array[Transform],
properties: util.Map[String, String],
constraints: Array[Constraint],
- tableId: String)
+ tableId: String,
+ schemaAlignmentConfig: SchemaAlignmentConfig)
extends InMemoryTable(
name,
columns,
partitioning,
properties,
constraints,
- id = tableId)
+ id = tableId,
+ schemaAlignmentConfig = schemaAlignmentConfig)
with SupportsRowLevelOperations {
def this(
@@ -63,14 +66,16 @@ class InMemoryRowLevelOperationTable private (
partitioning: Array[Transform],
properties: util.Map[String, String],
constraints: Array[Constraint] = Array.empty,
- tableId: String = java.util.UUID.randomUUID().toString) = {
+ tableId: String = java.util.UUID.randomUUID().toString,
+ schemaAlignmentConfig: SchemaAlignmentConfig =
SchemaAlignmentConfig.DEFAULT) = {
this(
name = name,
columns = CatalogV2Util.structTypeToV2Columns(schema),
partitioning = partitioning,
properties = properties,
constraints = constraints,
- tableId = tableId)
+ tableId = tableId,
+ schemaAlignmentConfig = schemaAlignmentConfig)
}
private final val PARTITION_COLUMN_REF =
FieldReference(PartitionKeyColumn.name)
@@ -106,7 +111,8 @@ class InMemoryRowLevelOperationTable private (
partitioning = partitioning,
properties = properties,
constraints = constraints,
- tableId = id)
+ tableId = id,
+ schemaAlignmentConfig = schemaAlignmentConfig)
dataMap.synchronized {
dataMap.foreach { case (key, splits) =>
val copiedSplits = splits.map { bufferedRows =>
@@ -373,9 +379,11 @@ object InMemoryRowLevelOperationTable {
partitioning: Array[Transform],
properties: util.Map[String, String],
constraints: Array[Constraint] = Array.empty,
- tableId: String = java.util.UUID.randomUUID().toString):
InMemoryRowLevelOperationTable = {
+ tableId: String = java.util.UUID.randomUUID().toString,
+ schemaAlignmentConfig: SchemaAlignmentConfig =
SchemaAlignmentConfig.DEFAULT)
+ : InMemoryRowLevelOperationTable = {
new InMemoryRowLevelOperationTable(
- name, columns, partitioning, properties, constraints, tableId)
+ name, columns, partitioning, properties, constraints, tableId,
schemaAlignmentConfig)
}
}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTableCatalog.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTableCatalog.scala
index 95e6659956ca..5c9060fcca6c 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTableCatalog.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryRowLevelOperationTableCatalog.scala
@@ -97,7 +97,8 @@ class InMemoryRowLevelOperationTableCatalog
partitioning = partitioning,
properties = properties,
constraints = constraints,
- tableId = table.id)
+ tableId = table.id,
+ schemaAlignmentConfig = table.schemaAlignmentConfig)
newTable.alterTableWithData(table.data, schema)
newTable.setVersionAndValidatedVersionFrom(table)
@@ -139,7 +140,8 @@ class PartialSchemaEvolutionCatalog extends
InMemoryRowLevelOperationTableCatalo
schema = schema,
partitioning = table.partitioning,
properties = properties,
- constraints = table.constraints)
+ constraints = table.constraints,
+ schemaAlignmentConfig = table.schemaAlignmentConfig)
newTable.alterTableWithData(table.data, table.schema)
newTable.setVersionAndValidatedVersionFrom(table)
tables.put(ident, newTable)
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala
index c783bfbece14..20ce2af03ac6 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala
@@ -26,6 +26,7 @@ import
org.apache.spark.sql.connector.distributions.{Distribution, Distributions
import org.apache.spark.sql.connector.expressions.{SortOrder, Transform}
import org.apache.spark.sql.connector.read._
import org.apache.spark.sql.connector.write.{LogicalWriteInfo,
SupportsOverwrite, WriteBuilder, WriterCommitMessage}
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig
import org.apache.spark.sql.sources._
import org.apache.spark.sql.types.{LongType, StructField, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -47,10 +48,11 @@ class InMemoryTable(
advisoryPartitionSize: Option[Long] = None,
isDistributionStrictlyRequired: Boolean = true,
override val numRowsPerSplit: Int = Int.MaxValue,
- override val id: String = UUID.randomUUID().toString)
+ override val id: String = UUID.randomUUID().toString,
+ override val schemaAlignmentConfig: SchemaAlignmentConfig =
SchemaAlignmentConfig.DEFAULT)
extends InMemoryBaseTable(name, columns, partitioning, properties,
constraints, distribution,
ordering, numPartitions, advisoryPartitionSize,
isDistributionStrictlyRequired,
- numRowsPerSplit) with SupportsDelete {
+ numRowsPerSplit, schemaAlignmentConfig) with SupportsDelete {
def this(
name: String,
@@ -143,7 +145,8 @@ class InMemoryTable(
advisoryPartitionSize,
isDistributionStrictlyRequired,
numRowsPerSplit,
- id)
+ id,
+ schemaAlignmentConfig = schemaAlignmentConfig)
dataMap.synchronized {
dataMap.foreach { case (key, splits) =>
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
index cc5c75dd779b..dc9253e42540 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalog.scala
@@ -32,6 +32,7 @@ import
org.apache.spark.sql.connector.catalog.procedures.{BoundProcedure, Proced
import org.apache.spark.sql.connector.distributions.{Distribution,
Distributions}
import org.apache.spark.sql.connector.expressions.{SortOrder, Transform}
import org.apache.spark.sql.connector.read.{LocalScan, Scan}
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig
import org.apache.spark.sql.types.{DataTypes, StructType}
import org.apache.spark.sql.util.CaseInsensitiveStringMap
@@ -219,11 +220,13 @@ class BasicInMemoryTableCatalog extends TableCatalog {
advisoryPartitionSize: Option[Long],
distributionStrictlyRequired: Boolean,
numRowsPerSplit: Int,
- id: String): InMemoryBaseTable = {
+ id: String,
+ schemaAlignmentConfig: SchemaAlignmentConfig =
SchemaAlignmentConfig.DEFAULT
+ ): InMemoryBaseTable = {
// scalastyle:on argcount
new InMemoryTable(name, columns, partitioning, properties, constraints,
distribution,
ordering, requiredNumPartitions, advisoryPartitionSize,
distributionStrictlyRequired,
- numRowsPerSplit, id)
+ numRowsPerSplit, id, schemaAlignmentConfig)
}
override def alterTable(ident: Identifier, changes: TableChange*): Table = {
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalystRuntimeFilterCatalog.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalystRuntimeFilterCatalog.scala
index e699f439223f..63389e3f941d 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalystRuntimeFilterCatalog.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableCatalystRuntimeFilterCatalog.scala
@@ -22,6 +22,7 @@ import java.util
import org.apache.spark.sql.connector.catalog.constraints.Constraint
import org.apache.spark.sql.connector.distributions.Distribution
import org.apache.spark.sql.connector.expressions.{SortOrder, Transform}
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig
/**
* Mix-in that constructs [[InMemoryCatalystRuntimeFilterTable]] from the
shared in-memory
@@ -41,7 +42,9 @@ trait InMemoryCatalystRuntimeFilterTableFactory { self:
BasicInMemoryTableCatalo
advisoryPartitionSize: Option[Long],
distributionStrictlyRequired: Boolean,
numRowsPerSplit: Int,
- id: String): InMemoryBaseTable = {
+ id: String,
+ // InMemoryCatalystRuntimeFilterTable carries no schema-alignment
config, so this is ignored.
+ schemaAlignmentConfig: SchemaAlignmentConfig): InMemoryBaseTable = {
// scalastyle:on argcount
new InMemoryCatalystRuntimeFilterTable(
name, columns, partitioning, properties, constraints, distribution,
ordering,
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2FilterCatalog.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2FilterCatalog.scala
index be5a907f5212..0c0a79975ac1 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2FilterCatalog.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTableWithV2FilterCatalog.scala
@@ -22,6 +22,7 @@ import java.util
import org.apache.spark.sql.connector.catalog.constraints.Constraint
import org.apache.spark.sql.connector.distributions.Distribution
import org.apache.spark.sql.connector.expressions.{SortOrder, Transform}
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig
class InMemoryTableWithV2FilterCatalog extends InMemoryTableCatalog {
// scalastyle:off argcount
@@ -37,7 +38,9 @@ class InMemoryTableWithV2FilterCatalog extends
InMemoryTableCatalog {
advisoryPartitionSize: Option[Long],
distributionStrictlyRequired: Boolean,
numRowsPerSplit: Int,
- id: String): InMemoryBaseTable = {
+ id: String,
+ // InMemoryTableWithV2Filter carries no schema-alignment config, so this
is ignored.
+ schemaAlignmentConfig: SchemaAlignmentConfig): InMemoryBaseTable = {
// scalastyle:on argcount
new InMemoryTableWithV2Filter(
name, columns, partitioning, properties, constraints, distribution,
ordering,
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala
index f48f99994a8c..0a18f32e7029 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/txns.scala
@@ -110,7 +110,8 @@ class TxnTable(
schema,
delegate.partitioning,
delegate.properties,
- delegate.constraints) {
+ delegate.constraints,
+ schemaAlignmentConfig = delegate.schemaAlignmentConfig) {
// Expose the same id as the delegate so that identity checks during
transaction re-resolution
// don't false-positive on the TxnTable wrapper having a different UUID.
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeWriteCompatibilitySuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeWriteCompatibilitySuite.scala
index 131eab34762f..b30e0d8fd7c1 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeWriteCompatibilitySuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/types/DataTypeWriteCompatibilitySuite.scala
@@ -27,6 +27,7 @@ import org.apache.spark.sql.catalyst.types.DataTypeUtils
import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLType
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy
+import
org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
class StrictDataTypeWriteCompatibilitySuite extends
DataTypeWriteCompatibilityBaseSuite {
override def storeAssignmentPolicy: SQLConf.StoreAssignmentPolicy.Value =
@@ -152,6 +153,153 @@ class ANSIDataTypeWriteCompatibilitySuite extends
DataTypeWriteCompatibilityBase
)
}
+ test("RUNTIME does not relax struct field ordering") {
+ val writeStruct = StructType(Seq(
+ StructField("first", FloatType, nullable = false),
+ StructField("second", StringType, nullable = false)))
+ val readStruct = StructType(Seq(
+ StructField("first", FloatType, nullable = false),
+ StructField("renamed", IntegerType, nullable = false)))
+
+ val errs = new mutable.ArrayBuffer[String]()
+ checkError(
+ exception = intercept[AnalysisException](
+ DataTypeUtils.canWrite("", writeStruct, readStruct, byName = true,
+ analysis.caseSensitiveResolution, "t", storeAssignmentPolicy,
+ errMsg => errs += errMsg,
+ ansiStoreAssignmentCastCheck = AnsiStoreAssignmentCastCheck.RUNTIME)
+ ),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.UNEXPECTED_COLUMN_NAME",
+ parameters = Map(
+ "tableName" -> "``",
+ "colName" -> "`t`",
+ "order" -> "1",
+ "expected" -> "`renamed`",
+ "found" -> "`second`")
+ )
+ }
+
+ test("RUNTIME does not relax array element nullability") {
+ val errs = new mutable.ArrayBuffer[String]()
+ checkError(
+ exception = intercept[AnalysisException](
+ DataTypeUtils.canWrite("", ArrayType(LongType, containsNull = true),
+ ArrayType(LongType, containsNull = false), byName = true,
+ analysis.caseSensitiveResolution, "arr", storeAssignmentPolicy,
+ errMsg => errs += errMsg,
+ ansiStoreAssignmentCastCheck =
AnsiStoreAssignmentCastCheck.RUNTIME)),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.NULLABLE_ARRAY_ELEMENTS",
+ parameters = Map("tableName" -> "``", "colName" -> "`arr`"))
+ }
+
+ test("RUNTIME does not relax map value nullability") {
+ val errs = new mutable.ArrayBuffer[String]()
+ checkError(
+ exception = intercept[AnalysisException](
+ DataTypeUtils.canWrite("", MapType(StringType, LongType,
valueContainsNull = true),
+ MapType(StringType, LongType, valueContainsNull = false), byName =
true,
+ analysis.caseSensitiveResolution, "m", storeAssignmentPolicy,
+ errMsg => errs += errMsg,
+ ansiStoreAssignmentCastCheck =
AnsiStoreAssignmentCastCheck.RUNTIME)),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.NULLABLE_MAP_VALUES",
+ parameters = Map("tableName" -> "``", "colName" -> "`m`"))
+ }
+
+ test("RUNTIME does not relax struct field nullability") {
+ val writeType = StructType(Seq(StructField("x", LongType, nullable =
true)))
+ val readType = StructType(Seq(StructField("x", LongType, nullable =
false)))
+ val errs = new mutable.ArrayBuffer[String]()
+ checkError(
+ exception = intercept[AnalysisException](
+ DataTypeUtils.canWrite("", writeType, readType, byName = true,
+ analysis.caseSensitiveResolution, "t", storeAssignmentPolicy,
+ errMsg => errs += errMsg,
+ ansiStoreAssignmentCastCheck =
AnsiStoreAssignmentCastCheck.RUNTIME)),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.NULLABLE_COLUMN",
+ parameters = Map("tableName" -> "``", "colName" -> "`t`.`x`"))
+ }
+
+ test("RUNTIME allows complex-to-string casts") {
+ Seq(
+ ArrayType(IntegerType),
+ MapType(StringType, IntegerType),
+ StructType(Seq(StructField("a", IntegerType)))).foreach { writeType =>
+ assert(
+ DataTypeUtils.canWrite("", writeType, StringType, byName = true,
+ analysis.caseSensitiveResolution, "c", storeAssignmentPolicy,
+ _ => (),
+ ansiStoreAssignmentCastCheck = AnsiStoreAssignmentCastCheck.RUNTIME),
+ s"$writeType -> string should be allowed under RUNTIME")
+ }
+ }
+
+ test("RUNTIME rejects unsupported datetime-to-variant casts") {
+ val unsupportedTypes = Seq(
+ TimeType(TimeType.DEFAULT_PRECISION),
+ TimeType(TimeType.MAX_PRECISION),
+ TimestampLTZNanosType(TimestampLTZNanosType.MAX_PRECISION),
+ TimestampNTZNanosType(TimestampNTZNanosType.MAX_PRECISION))
+ val arrays = unsupportedTypes.map(ArrayType(_))
+ val nestedArrays = unsupportedTypes.map(t => ArrayType(ArrayType(t)))
+ (unsupportedTypes ++ arrays ++ nestedArrays).foreach { writeType =>
+ val errs = new mutable.ArrayBuffer[String]()
+ checkError(
+ exception = intercept[AnalysisException](
+ DataTypeUtils.canWrite("", writeType, VariantType, byName = true,
+ analysis.caseSensitiveResolution, "v", storeAssignmentPolicy,
+ errMsg => errs += errMsg,
+ ansiStoreAssignmentCastCheck =
AnsiStoreAssignmentCastCheck.RUNTIME)),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST",
+ parameters = Map(
+ "tableName" -> "``",
+ "colName" -> "`v`",
+ "srcType" -> toSQLType(writeType),
+ "targetType" -> toSQLType(VariantType)))
+ }
+ }
+
+ test("RUNTIME allows supported array-to-variant casts") {
+ Seq(ArrayType(DateType), ArrayType(ArrayType(TimestampType))).foreach {
writeType =>
+ assert(
+ DataTypeUtils.canWrite("", writeType, VariantType, byName = true,
+ analysis.caseSensitiveResolution, "v", storeAssignmentPolicy,
+ _ => (),
+ ansiStoreAssignmentCastCheck = AnsiStoreAssignmentCastCheck.RUNTIME),
+ s"$writeType -> variant should be allowed under RUNTIME")
+ }
+ }
+
+ test("RUNTIME does not relax UDT field-name validation") {
+ // A UDT backed by STRUCT<b, a> written by name into STRUCT<a, b> must be
rejected on the
+ // field-name mismatch; deferring the ANSI cast check must not unwrap the
UDT and silently swap
+ // fields positionally.
+ val udtType = new UserDefinedType[Any] {
+ override def sqlType: DataType = StructType(Seq(
+ StructField("b", IntegerType, nullable = false),
+ StructField("a", IntegerType, nullable = false)))
+ override def userClass: java.lang.Class[Any] = null
+ override def serialize(obj: Any): Any = null
+ override def deserialize(datum: Any): Any = null
+ }
+ val readType = StructType(Seq(
+ StructField("a", IntegerType, nullable = false),
+ StructField("b", IntegerType, nullable = false)))
+ val errs = new mutable.ArrayBuffer[String]()
+ checkError(
+ exception = intercept[AnalysisException](
+ DataTypeUtils.canWrite("", udtType, readType, byName = true,
+ analysis.caseSensitiveResolution, "t", storeAssignmentPolicy,
+ errMsg => errs += errMsg,
+ ansiStoreAssignmentCastCheck =
AnsiStoreAssignmentCastCheck.RUNTIME)),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.UNEXPECTED_COLUMN_NAME",
+ parameters = Map(
+ "tableName" -> "``",
+ "colName" -> "`t`",
+ "order" -> "0",
+ "expected" -> "`a`",
+ "found" -> "`b`"))
+ }
+
private val stringPoint2 = StructType(Seq(
StructField("x", StringType, nullable = false),
StructField("y", StringType, nullable = false)))
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
index 0b037ed6829b..fb92a17161e0 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/KeyGroupedPartitioningSuite.scala
@@ -60,6 +60,7 @@ import org.apache.spark.sql.execution.window.{Final, Partial,
WindowGroupLimitEx
import org.apache.spark.sql.functions.{col, max}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.internal.SQLConf._
+import org.apache.spark.sql.internal.connector.SchemaAlignmentConfig
import org.apache.spark.sql.types._
import org.apache.spark.sql.util.CaseInsensitiveStringMap
import org.apache.spark.tags.ExtendedSQLTest
@@ -9944,11 +9945,12 @@ class CustomReportingCatalog extends
InMemoryTableCatalog {
advisoryPartitionSize: Option[Long],
distributionStrictlyRequired: Boolean,
numRowsPerSplit: Int,
- id: String): InMemoryBaseTable = {
+ id: String,
+ schemaAlignmentConfig: SchemaAlignmentConfig): InMemoryBaseTable = {
// scalastyle:on argcount
new InMemoryTable(name, columns, partitioning, properties, constraints,
distribution,
ordering, requiredNumPartitions, advisoryPartitionSize,
distributionStrictlyRequired,
- numRowsPerSplit, id) {
+ numRowsPerSplit, id, schemaAlignmentConfig) {
override def newScanBuilder(options: CaseInsensitiveStringMap):
ScanBuilder =
new InMemoryScanBuilder(schema(), options) {
override def build(): Scan =
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala
new file mode 100644
index 000000000000..536d10e93fd0
--- /dev/null
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala
@@ -0,0 +1,389 @@
+/*
+ * 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, Try}
+
+import org.apache.spark.SparkConf
+import org.apache.spark.SparkThrowable
+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.internal.connector.SchemaAlignmentConfig
+import
org.apache.spark.sql.internal.connector.SchemaAlignmentConfig.AnsiStoreAssignmentCastCheck
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{IntegerType, StructType}
+
+/**
+ * A catalog that creates [[InMemoryRowLevelOperationTable]]s carrying the
fixed
+ * [[SchemaAlignmentConfig]] supplied by the concrete subclass.
+ */
+abstract class SchemaAlignmentTestCatalog extends
InMemoryRowLevelOperationTableCatalog {
+
+ protected def tableConfig: SchemaAlignmentConfig
+
+ 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 defer the ANSI store-assignment cast check to
runtime. */
+class RelaxedSchemaAlignmentCatalog extends SchemaAlignmentTestCatalog {
+ override protected def tableConfig: SchemaAlignmentConfig = new
SchemaAlignmentConfig {
+ override def ansiStoreAssignmentCastCheck(): AnsiStoreAssignmentCastCheck =
+ AnsiStoreAssignmentCastCheck.RUNTIME
+ }
+}
+
+/** 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 withAnsiPolicy(f: => Unit): Unit =
+ withSQLConf(
+ SQLConf.STORE_ASSIGNMENT_POLICY.key ->
StoreAssignmentPolicy.ANSI.toString)(f)
+
+ test("RUNTIME: 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("RUNTIME: 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("RUNTIME: 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("RUNTIME: 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)")
+ }
+ }
+ }
+ }
+
+ test("RUNTIME: complex-to-string cast is deferred") {
+ withTable(s"$relaxed.t", s"$strict.t") {
+ sql(s"CREATE TABLE $relaxed.t (c STRING) USING foo")
+ sql(s"CREATE TABLE $strict.t (c STRING) USING foo")
+ withAnsiPolicy {
+ sql(s"INSERT INTO $relaxed.t VALUES (array(1, 2))")
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row("[1, 2]"))
+ val strictInsert = s"INSERT INTO $strict.t VALUES (array(1, 2))"
+ checkError(
+ exception = intercept[AnalysisException](sql(strictInsert)),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST",
+ parameters = Map(
+ "tableName" -> s"`$strict`.`t`",
+ "colName" -> "`c`",
+ "srcType" -> "\"ARRAY<INT>\"",
+ "targetType" -> "\"STRING\""))
+ }
+ }
+ }
+
+ test("RUNTIME: long and decimal to timestamp casts are rejected") {
+ withTable(s"$relaxed.t") {
+ sql(s"CREATE TABLE $relaxed.t (c TIMESTAMP) USING foo")
+ withAnsiPolicy {
+ Seq(
+ "9223372036854775807L",
+ "cast(99999999999999999999999999999999999999 as decimal(38, 0))"
+ ).foreach { value =>
+ val error = intercept[AnalysisException] {
+ sql(s"INSERT INTO $relaxed.t VALUES ($value)")
+ }
+ assert(error.getCondition ==
"INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST")
+ }
+ }
+ }
+ }
+
+ test("RUNTIME: variant-to-complex casts are rejected") {
+ Seq(
+ "STRUCT<a: INT>" -> "parse_json('{}')",
+ "ARRAY<INT>" -> "parse_json('[]')",
+ "MAP<STRING, INT>" -> "parse_json('{}')"
+ ).foreach { case (targetType, value) =>
+ withTable(s"$relaxed.t") {
+ sql(s"CREATE TABLE $relaxed.t (c $targetType) USING foo")
+ withAnsiPolicy {
+ val error = intercept[AnalysisException] {
+ sql(s"INSERT INTO $relaxed.t VALUES ($value)")
+ }
+ assert(error.getCondition ==
"INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST")
+ }
+ }
+ }
+ }
+
+ test("schema alignment config survives ALTER TABLE ADD COLUMNS") {
+ 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")
+ sql(s"ALTER TABLE $relaxed.t ADD COLUMNS (data INT)")
+ sql(s"ALTER TABLE $strict.t ADD COLUMNS (data INT)")
+ withAnsiPolicy {
+ sql(s"INSERT INTO $relaxed.t VALUES (1, '5')")
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1, 5))
+ checkError(
+ exception = intercept[AnalysisException](sql(s"INSERT INTO $strict.t
VALUES (1, '5')")),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST",
+ parameters = Map(
+ "tableName" -> s"`$strict`.`t`",
+ "colName" -> "`data`",
+ "srcType" -> "\"STRING\"",
+ "targetType" -> "\"INT\""))
+ }
+ }
+ }
+
+ test("schema alignment config survives schema evolution") {
+ 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")
+ val evolving = spark.createDataFrame(
+ java.util.Arrays.asList(Row(1, 5)),
+ new StructType().add("id", IntegerType).add("data", IntegerType))
+
evolving.write.mode("append").withSchemaEvolution().insertInto(s"$relaxed.t")
+
evolving.write.mode("append").withSchemaEvolution().insertInto(s"$strict.t")
+ withAnsiPolicy {
+ sql(s"INSERT INTO $relaxed.t VALUES (2, '5')")
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Seq(Row(1, 5), Row(2,
5)))
+ checkError(
+ exception = intercept[AnalysisException](sql(s"INSERT INTO $strict.t
VALUES (2, '5')")),
+ condition = "INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_SAFELY_CAST",
+ parameters = Map(
+ "tableName" -> s"`$strict`.`t`",
+ "colName" -> "`data`",
+ "srcType" -> "\"STRING\"",
+ "targetType" -> "\"INT\""))
+ }
+ }
+ }
+
+ 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 assertBothReject(
+ targetSchema: StructType, source: DataFrame, errorClass: String): Unit =
+ withAnsiPolicy {
+ Seq(relaxed, strict).foreach { catalog =>
+ appendByName(catalog, targetSchema, source) match {
+ case Failure(error: SparkThrowable) =>
+ assert(error.getCondition == errorClass,
+ s"$catalog: expected $errorClass, got ${error.getCondition}")
+ case other =>
+ fail(s"$catalog: expected rejection with $errorClass, got $other")
+ }
+ }
+ }
+
+ test("RUNTIME: renamed nested struct field is still rejected") {
+ val target = new StructType()
+ .add("s", new StructType().add("a", IntegerType).add("b", IntegerType))
+ val source = spark.createDataFrame(
+ java.util.Arrays.asList(Row(Row(1, 2))),
+ new StructType().add("s", new StructType().add("a",
IntegerType).add("c", IntegerType)))
+ assertBothReject(target, source,
"INCOMPATIBLE_DATA_FOR_TABLE.CANNOT_FIND_DATA")
+ }
+
+ // Invalid source values and the error the cast or null check inserted under
RUNTIME raises. The
+ // values are read from a table, so the casts aren't constant-folded during
analysis.
+ private val invalidValues = Seq(
+ // Malformed string.
+ "'abc'" -> "CAST_INVALID_INPUT",
+ // Out-of-range numeric string.
+ "'9999999999'" -> "CAST_INVALID_INPUT",
+ // NULL written to a NOT NULL column.
+ "CAST(NULL AS STRING)" -> "NOT_NULL_ASSERT_VIOLATION")
+
+ Seq(true, false).foreach { ansiEnabled =>
+ test(s"RUNTIME: INSERT of invalid values fails at execution,
ansi.enabled=$ansiEnabled") {
+ invalidValues.foreach { case (value, expectedCondition) =>
+ withTable(s"$relaxed.src", s"$relaxed.t") {
+ sql(s"CREATE TABLE $relaxed.src (v STRING) USING foo")
+ sql(s"INSERT INTO $relaxed.src VALUES ($value)")
+ sql(s"CREATE TABLE $relaxed.t (c INT NOT NULL) USING foo")
+ withSQLConf(
+ SQLConf.STORE_ASSIGNMENT_POLICY.key ->
StoreAssignmentPolicy.ANSI.toString,
+ SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ val error = intercept[SparkThrowable] {
+ sql(s"INSERT INTO $relaxed.t SELECT v FROM $relaxed.src")
+ }
+ assert(error.getCondition == expectedCondition, s"value: $value")
+ }
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Seq.empty)
+ }
+ }
+ }
+
+ test(s"RUNTIME: UPDATE with invalid values fails at execution,
ansi.enabled=$ansiEnabled") {
+ invalidValues.foreach { case (value, expectedCondition) =>
+ withTable(s"$relaxed.t") {
+ sql(s"CREATE TABLE $relaxed.t (id INT, data INT NOT NULL, s STRING)
USING foo")
+ sql(s"INSERT INTO $relaxed.t VALUES (1, 0, $value)")
+ withSQLConf(
+ SQLConf.STORE_ASSIGNMENT_POLICY.key ->
StoreAssignmentPolicy.ANSI.toString,
+ SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ val error = intercept[SparkThrowable] {
+ sql(s"UPDATE $relaxed.t SET data = s WHERE id = 1")
+ }
+ assert(error.getCondition == expectedCondition, s"value: $value")
+ }
+ checkAnswer(sql(s"SELECT id, data FROM $relaxed.t"), Row(1, 0))
+ }
+ }
+ }
+
+ test(s"RUNTIME: MERGE with invalid values fails at execution,
ansi.enabled=$ansiEnabled") {
+ invalidValues.foreach { case (value, expectedCondition) =>
+ Seq(
+ "WHEN MATCHED THEN UPDATE SET t.data = s.v" -> 1,
+ "WHEN NOT MATCHED THEN INSERT (id, data) VALUES (s.id, s.v)" -> 2
+ ).foreach { case (clause, sourceId) =>
+ withTable(s"$relaxed.src", s"$relaxed.t") {
+ sql(s"CREATE TABLE $relaxed.src (id INT, v STRING) USING foo")
+ sql(s"INSERT INTO $relaxed.src VALUES ($sourceId, $value)")
+ sql(s"CREATE TABLE $relaxed.t (id INT, data INT NOT NULL) USING
foo")
+ sql(s"INSERT INTO $relaxed.t VALUES (1, 0)")
+ withSQLConf(
+ SQLConf.STORE_ASSIGNMENT_POLICY.key ->
StoreAssignmentPolicy.ANSI.toString,
+ SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
+ val error = intercept[SparkThrowable] {
+ sql(
+ s"""MERGE INTO $relaxed.t t
+ |USING $relaxed.src s
+ |ON t.id = s.id
+ |$clause""".stripMargin)
+ }
+ assert(error.getCondition == expectedCondition, s"value: $value,
clause: $clause")
+ }
+ checkAnswer(sql(s"SELECT * FROM $relaxed.t"), Row(1, 0))
+ }
+ }
+ }
+ }
+ }
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
