From d8c218ab56cb979fc4b069516cef5bddfcd8116c Mon Sep 17 00:00:00 2001 From: Johan Lasperas Date: Tue, 8 Sep 2026 16:00:36 +0000 Subject: [PATCH] Allow configuring schema alignment behavior for DSv2 writes --- .../catalog/SchemaAlignmentConfig.java | 52 ++++ .../spark/sql/connector/catalog/Table.java | 10 + .../sql/catalyst/analysis/Analyzer.scala | 15 +- .../catalyst/analysis/AssignmentUtils.scala | 44 ++- .../ResolveRowLevelCommandAssignments.scala | 38 ++- .../analysis/TableOutputResolver.scala | 120 ++++--- .../write/RowLevelOperationTable.scala | 3 +- .../connector/catalog/InMemoryBaseTable.scala | 3 +- .../InMemoryRowLevelOperationTable.scala | 21 +- .../sql/connector/catalog/InMemoryTable.scala | 5 +- .../spark/sql/connector/catalog/txns.scala | 3 +- .../InsertSchemaEvolutionSuite.scala | 192 ++++++++++++ .../catalog/SchemaAlignmentConfigSuite.scala | 294 ++++++++++++++++++ .../command/AlignAssignmentsSuiteBase.scala | 10 +- .../command/PlanResolutionSuite.scala | 9 +- 15 files changed, 734 insertions(+), 85 deletions(-) create mode 100644 sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/connector/InsertSchemaEvolutionSuite.scala create mode 100644 sql/core/src/test/scala/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfigSuite.scala diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/SchemaAlignmentConfig.java new file mode 100644 index 0000000000000..2351f9a6229c7 --- /dev/null +++ b/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 + */ +@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 deferCastValidationToRuntime() { + return false; + } +} diff --git a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java index ec27bcf6c82e2..1c740afbf80a7 100644 --- a/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java +++ b/sql/catalyst/src/main/java/org/apache/spark/sql/connector/catalog/Table.java @@ -112,4 +112,14 @@ default Map properties() { * the version that corresponds to the current state of this table instance. */ default String version() { return null; } + + /** + * Returns the schema-alignment configuration for writes to this table. This allows connectors to + * configure casting behavior and handling of schema mismatches during writes. + * It is recommended to use the DEFAULT configuration to provide a unified behavior across data + * sources, but some connectors may require deviating from the default behavior. + */ + default SchemaAlignmentConfig schemaAlignmentConfig() { + return SchemaAlignmentConfig.DEFAULT; + } } 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 35b9052686dcf..6322fccd74948 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 @@ -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) TableOutputResolver.suitableForByNameCheck(v2Write.isByName, expected = v2Write.table.output, queryOutput = v2Write.query.output) // With schema evolution + coercion flag, missing top-level columns AND missing nested @@ -3965,7 +3968,8 @@ class Analyzer( val (projection, autoFilledGenCols) = TableOutputResolver.resolveOutputColumnsWithGeneratedInfo( v2Write.table.name, expected, v2Write.query, v2Write.isByName, conf, - defaultValueFillMode) + defaultValueFillMode, + deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime()) if (projection != v2Write.query) { val cleanedTable = v2Write.table match { case r: DataSourceV2Relation => @@ -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()) { throw QueryCompilationErrors.legacyStoreAssignmentPolicyError() } } 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 df4b0646ed42f..42065caebb683 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 @@ -28,6 +28,7 @@ import org.apache.spark.sql.catalyst.types.DataTypeUtils import org.apache.spark.sql.catalyst.util.CharVarcharUtils import org.apache.spark.sql.catalyst.util.ResolveDefaultColumns.getDefaultValueExprOrNullLit import org.apache.spark.sql.connector.catalog.CatalogV2Implicits._ +import org.apache.spark.sql.connector.catalog.SchemaAlignmentConfig import org.apache.spark.sql.errors.QueryCompilationErrors import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{DataType, StructType} @@ -63,7 +64,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport { attrs: Seq[Attribute], assignments: Seq[Assignment], fromStar: Boolean, - coerceNestedTypes: Boolean): Seq[Assignment] = { + coerceNestedTypes: Boolean, + schemaAlignment: SchemaAlignmentConfig = SchemaAlignmentConfig.DEFAULT): Seq[Assignment] = { val errors = new mutable.ArrayBuffer[String]() @@ -75,7 +77,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport { addError = err => errors += err, colPath = Seq(attr.name), coerceNestedTypes, - fromStar) + fromStar, + schemaAlignment = schemaAlignment) } if (errors.nonEmpty) { @@ -103,7 +106,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport { def alignInsertAssignments( attrs: Seq[Attribute], assignments: Seq[Assignment], - coerceNestedTypes: Boolean = false): Seq[Assignment] = { + coerceNestedTypes: Boolean = false, + schemaAlignment: SchemaAlignmentConfig = SchemaAlignmentConfig.DEFAULT): Seq[Assignment] = { val errors = new mutable.ArrayBuffer[String]() @@ -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, + deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime()) } Assignment(attr, resolvedValue) } @@ -160,7 +165,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport { addError: String => Unit, colPath: Seq[String], coerceNestedTypes: Boolean = false, - updateStar: Boolean = false): Expression = { + updateStar: Boolean = false, + schemaAlignment: SchemaAlignmentConfig): Expression = { val (exactAssignments, otherAssignments) = assignments.partition { assignment => assignment.key.semanticEquals(colExpr) @@ -188,21 +194,24 @@ 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, schemaAlignment = schemaAlignment) case _ => // For non-struct types, resolve directly val coerceMode = if (coerceNestedTypes) RECURSE else NONE TableOutputResolver.resolveUpdate("", value, col, conf, addError, colPath, - coerceMode) + coerceMode, + deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime()) } } else { val value = exactAssignments.head.value val coerceMode = if (coerceNestedTypes) RECURSE else NONE TableOutputResolver.resolveUpdate("", value, col, conf, addError, - colPath, coerceMode) + colPath, coerceMode, + deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime()) } } else { - applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes) + applyFieldAssignments(col, colExpr, fieldAssignments, addError, colPath, coerceNestedTypes, + schemaAlignment = schemaAlignment) } } @@ -212,7 +221,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport { assignments: Seq[Assignment], addError: String => Unit, colPath: Seq[String], - coerceNestedTypes: Boolean): Expression = { + coerceNestedTypes: Boolean, + schemaAlignment: SchemaAlignmentConfig): Expression = { col.dataType match { case structType: StructType => @@ -222,7 +232,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, schemaAlignment = schemaAlignment) } toNamedStruct(structType, updatedFieldExprs) @@ -240,7 +250,8 @@ object AssignmentUtils extends SQLConfHelper with CastSupport { value: Expression, addError: String => Unit, colPath: Seq[String], - coerceNestedTypes: Boolean): Expression = { + coerceNestedTypes: Boolean, + schemaAlignment: SchemaAlignmentConfig): Expression = { col.dataType match { case structType: StructType => @@ -273,12 +284,15 @@ object AssignmentUtils extends SQLConfHelper with CastSupport { case _: StructType => // Field is a struct, recurse applyNestedFieldAssignments(fieldAttr, targetFieldExpr, - sourceFieldValue, addError, fieldPath, coerceNestedTypes) + sourceFieldValue, addError, fieldPath, coerceNestedTypes, + schemaAlignment = schemaAlignment) case _ => // Field is not a struct, resolve with TableOutputResolver val coerceMode = if (coerceNestedTypes) RECURSE else NONE - TableOutputResolver.resolveUpdate("", sourceFieldValue, fieldAttr, conf, addError, - fieldPath, coerceMode) + TableOutputResolver.resolveUpdate( + "", sourceFieldValue, fieldAttr, conf, addError, + fieldPath, coerceMode, + deferCastValidationToRuntime = schemaAlignment.deferCastValidationToRuntime()) } } 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 76035ea819ff5..ca04b577c5822 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 @@ -24,6 +24,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{Assignment, DeleteAction, In 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.connector.catalog.SchemaAlignmentConfig import org.apache.spark.sql.errors.QueryCompilationErrors import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation import org.apache.spark.sql.internal.SQLConf.StoreAssignmentPolicy @@ -40,32 +41,37 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] { override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsWithPruning( _.containsPattern(COMMAND), ruleId) { case u: UpdateTable if !u.skipSchemaResolution && u.resolved && u.rewritable && !u.aligned => - validateStoreAssignmentPolicy() + val schemaAlignment = schemaAlignmentConfig(u.table) + validateStoreAssignmentPolicy(schemaAlignment) val newTable = cleanAttrMetadata(u.table) val newAssignments = AssignmentUtils.alignUpdateAssignments(u.table.output, u.assignments, - fromStar = false, coerceNestedTypes = false) + fromStar = false, coerceNestedTypes = false, schemaAlignment = schemaAlignment) u.copy(table = newTable, assignments = newAssignments) case u: UpdateTable if !u.skipSchemaResolution && u.resolved && !u.aligned => resolveAssignments(u) case m: MergeIntoTable if m.rewritable && shouldAlignAssignments(m) && containsFinalSchema(m) => - validateStoreAssignmentPolicy() + val schemaAlignment = schemaAlignmentConfig(m.targetTable) + validateStoreAssignmentPolicy(schemaAlignment) val coerceNestedTypes = conf.coerceMergeNestedTypes && m.withSchemaEvolution m.copy( targetTable = cleanAttrMetadata(m.targetTable), matchedActions = alignActions( m.targetTable.output, m.matchedActions, - coerceNestedTypes), + coerceNestedTypes, + schemaAlignment), notMatchedActions = alignActions( m.targetTable.output, m.notMatchedActions, - coerceNestedTypes), + coerceNestedTypes, + schemaAlignment), notMatchedBySourceActions = alignActions( m.targetTable.output, m.notMatchedBySourceActions, - coerceNestedTypes)) + coerceNestedTypes, + schemaAlignment)) case m: MergeIntoTable if shouldAlignAssignments(m) && containsFinalSchema(m) => resolveAssignments(m) @@ -79,9 +85,16 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] { !m.schemaEvolutionEnabled || (m.schemaEvolutionReady && m.pendingSchemaChanges.isEmpty) } - private def validateStoreAssignmentPolicy(): Unit = { - // SPARK-28730: LEGACY store assignment policy is disallowed in data source v2 - if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY) { + private def schemaAlignmentConfig(target: LogicalPlan): SchemaAlignmentConfig = + target.collectFirst { + case relation: DataSourceV2Relation => relation.table.schemaAlignmentConfig() + }.getOrElse(SchemaAlignmentConfig.DEFAULT) + + private def validateStoreAssignmentPolicy(schemaAlignment: SchemaAlignmentConfig): Unit = { + // SPARK-28730: LEGACY store assignment policy is disallowed in data source v2, unless the + // target table opts into it via its SchemaAlignmentConfig. + if (conf.storeAssignmentPolicy == StoreAssignmentPolicy.LEGACY && + !schemaAlignment.allowLegacyStoreAssignmentPolicy()) { throw QueryCompilationErrors.legacyStoreAssignmentPolicyError() } } @@ -127,16 +140,17 @@ object ResolveRowLevelCommandAssignments extends Rule[LogicalPlan] { private def alignActions( attrs: Seq[Attribute], actions: Seq[MergeAction], - coerceNestedTypes: Boolean): Seq[MergeAction] = { + coerceNestedTypes: Boolean, + schemaAlignment: SchemaAlignmentConfig): Seq[MergeAction] = { actions.map { case u @ UpdateAction(_, assignments, fromStar) => u.copy(assignments = AssignmentUtils.alignUpdateAssignments(attrs, assignments, - fromStar, coerceNestedTypes)) + fromStar, coerceNestedTypes, schemaAlignment = schemaAlignment)) case d: DeleteAction => d case i @ InsertAction(_, assignments) => i.copy(assignments = AssignmentUtils.alignInsertAssignments(attrs, assignments, - coerceNestedTypes)) + coerceNestedTypes, schemaAlignment = schemaAlignment)) 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 933513654dc15..5c1958b1b221a 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 @@ -94,9 +94,11 @@ object TableOutputResolver extends SQLConfHelper with Logging { query: LogicalPlan, byName: Boolean, conf: SQLConf, - defaultValueFillMode: DefaultValueFillMode.Value = NONE): LogicalPlan = { + defaultValueFillMode: DefaultValueFillMode.Value = NONE, + deferCastValidationToRuntime: Boolean = false): LogicalPlan = { resolveOutputColumnsInternal( - tableName, expected, query, byName, conf, defaultValueFillMode)._1 + tableName, expected, query, byName, conf, defaultValueFillMode, + deferCastValidationToRuntime)._1 } /** @@ -111,10 +113,11 @@ object TableOutputResolver extends SQLConfHelper with Logging { query: LogicalPlan, byName: Boolean, conf: SQLConf, - defaultValueFillMode: DefaultValueFillMode.Value = NONE + defaultValueFillMode: DefaultValueFillMode.Value = NONE, + deferCastValidationToRuntime: Boolean = false ): (LogicalPlan, Set[String]) = { resolveOutputColumnsInternal( - tableName, expected, query, byName, conf, defaultValueFillMode) + tableName, expected, query, byName, conf, defaultValueFillMode, deferCastValidationToRuntime) } private def resolveOutputColumnsInternal( @@ -123,7 +126,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { query: LogicalPlan, byName: Boolean, conf: SQLConf, - defaultValueFillMode: DefaultValueFillMode.Value + defaultValueFillMode: DefaultValueFillMode.Value, + deferCastValidationToRuntime: Boolean ): (LogicalPlan, Set[String]) = { if (expected.size < query.output.size) { @@ -149,7 +153,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { errors += _, Nil, defaultValueFillMode, - enforceFullOutput = true) + enforceFullOutput = true, + deferCastValidationToRuntime = deferCastValidationToRuntime) } else { if (expected.size > query.output.size && !fillDefaultValue) { throw QueryCompilationErrors.cannotWriteNotEnoughColumnsToTableError( @@ -157,7 +162,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { } resolveColumnsByPosition( tableName, query.output, expected, conf, errors += _, - fillDefaultValue = fillDefaultValue) + fillDefaultValue = fillDefaultValue, + deferCastValidationToRuntime = deferCastValidationToRuntime) } if (errors.nonEmpty) { @@ -180,14 +186,16 @@ object TableOutputResolver extends SQLConfHelper with Logging { conf: SQLConf, addError: String => Unit, colPath: Seq[String], - defaultValueFillMode: DefaultValueFillMode.Value): Expression = { + defaultValueFillMode: DefaultValueFillMode.Value, + deferCastValidationToRuntime: Boolean = false): 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) if (canWriteExpr) { val nullsHandled = checkNullability(value, col, conf, colPath) applyColumnMetadata(nullsHandled, col) @@ -197,20 +205,23 @@ 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) + byName = true, conf, addError, colPath, fillChildDefaultValue, enforceFullOutput = false, + deferCastValidationToRuntime = deferCastValidationToRuntime) resolvedValue.getOrElse(value) case (valueType: ArrayType, colType: ArrayType) => val resolvedValue = resolveArrayType( tableName, value, valueType, col, colType, - byName = true, conf, addError, colPath, fillChildDefaultValue, enforceFullOutput = false) + byName = true, conf, addError, colPath, fillChildDefaultValue, enforceFullOutput = false, + deferCastValidationToRuntime = deferCastValidationToRuntime) resolvedValue.getOrElse(value) case (valueType: MapType, colType: MapType) => val resolvedValue = resolveMapType( tableName, value, valueType, col, colType, - byName = true, conf, addError, colPath, fillChildDefaultValue, enforceFullOutput = false) + byName = true, conf, addError, colPath, fillChildDefaultValue, enforceFullOutput = false, + deferCastValidationToRuntime = deferCastValidationToRuntime) resolvedValue.getOrElse(value) case _ => - checkUpdate(tableName, value, col, conf, addError, colPath) + checkUpdate(tableName, value, col, conf, addError, colPath, deferCastValidationToRuntime) } } @@ -220,7 +231,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { attr: Attribute, conf: SQLConf, addError: String => Unit, - colPath: Seq[String]): Expression = { + colPath: Seq[String], + deferCastValidationToRuntime: Boolean): Expression = { val attrTypeHasCharVarchar = CharVarcharUtils.hasCharVarchar(attr.dataType) val attrTypeWithoutCharVarchar = if (attrTypeHasCharVarchar) { @@ -231,7 +243,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { val canWriteValue = canWrite( tableName, value.dataType, attrTypeWithoutCharVarchar, - byName = true, conf, addError, colPath) + byName = true, conf, addError, colPath, + deferCastValidationToRuntime = deferCastValidationToRuntime) if (canWriteValue) { val nullCheckedValue = checkNullability(value, attr, conf, colPath) @@ -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 => + // The target validates casts at runtime, so skip the ANSI store-assignment analysis check + // and let the inserted cast surface overflows / malformed values at execution time. Casts + // between structurally incompatible types are still rejected when the cast is resolved. + true case StoreAssignmentPolicy.STRICT | StoreAssignmentPolicy.ANSI => DataTypeUtils.canWrite( tableName, valueType, expectedType, byName, conf.resolver, colPath.quoted, @@ -373,7 +392,8 @@ 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, + deferCastValidationToRuntime: Boolean): (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 +443,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, + deferCastValidationToRuntime = deferCastValidationToRuntime) 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) case _ => checkField( - tableName, actualExpectedCol, matchedCol, byName = true, conf, addError, newColPath) + tableName, actualExpectedCol, matchedCol, byName = true, conf, addError, newColPath, + deferCastValidationToRuntime = deferCastValidationToRuntime) } } } @@ -488,7 +512,8 @@ 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, + deferCastValidationToRuntime: Boolean): (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 +553,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, + deferCastValidationToRuntime = deferCastValidationToRuntime) 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) case _ => - checkField(tableName, expectedCol, inputCol, byName = false, conf, addError, newColPath) + checkField(tableName, expectedCol, inputCol, byName = false, conf, addError, newColPath, + deferCastValidationToRuntime = deferCastValidationToRuntime) } } @@ -622,7 +651,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { addError: String => Unit, colPath: Seq[String], fillDefaultValue: Boolean, - enforceFullOutput: Boolean): Option[NamedExpression] = { + enforceFullOutput: Boolean, + deferCastValidationToRuntime: Boolean): 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 +661,12 @@ 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) } else { resolveColumnsByPosition( - tableName, fields, toAttributes(expectedType), conf, addError, colPath, fillDefaultValue) + tableName, fields, toAttributes(expectedType), conf, addError, colPath, fillDefaultValue, + deferCastValidationToRuntime = deferCastValidationToRuntime) } if (resolved.length == expectedType.length) { val struct = CreateStruct(resolved) @@ -665,7 +697,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { addError: String => Unit, colPath: Seq[String], fillDefaultValue: Boolean, - enforceFullOutput: Boolean): Option[NamedExpression] = { + enforceFullOutput: Boolean, + deferCastValidationToRuntime: Boolean): Option[NamedExpression] = { val nullCheckedInput = checkNullability(input, expected, conf, colPath) val param = NamedLambdaVariable("element", inputType.elementType, inputType.containsNull) val fakeAttr = @@ -674,10 +707,12 @@ 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) } else { resolveColumnsByPosition( - tableName, Seq(param), Seq(fakeAttr), conf, addError, colPath, fillDefaultValue) + tableName, Seq(param), Seq(fakeAttr), conf, addError, colPath, fillDefaultValue, + deferCastValidationToRuntime = deferCastValidationToRuntime) } if (res.length == 1) { val castedArray = @@ -708,7 +743,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { addError: String => Unit, colPath: Seq[String], fillDefaultValue: Boolean, - enforceFullOutput: Boolean): Option[NamedExpression] = { + enforceFullOutput: Boolean, + deferCastValidationToRuntime: Boolean): Option[NamedExpression] = { val nullCheckedInput = checkNullability(input, expected, conf, colPath) val keyParam = NamedLambdaVariable("key", inputType.keyType, nullable = false) @@ -717,10 +753,12 @@ 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) } else { resolveColumnsByPosition( - tableName, Seq(keyParam), Seq(fakeKeyAttr), conf, addError, colPath, fillDefaultValue) + tableName, Seq(keyParam), Seq(fakeKeyAttr), conf, addError, colPath, fillDefaultValue, + deferCastValidationToRuntime = deferCastValidationToRuntime) } val valueParam = @@ -730,10 +768,12 @@ 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, + deferCastValidationToRuntime = deferCastValidationToRuntime) } else { resolveColumnsByPosition( - tableName, Seq(valueParam), Seq(fakeValueAttr), conf, addError, colPath, fillDefaultValue) + tableName, Seq(valueParam), Seq(fakeValueAttr), conf, addError, colPath, fillDefaultValue, + deferCastValidationToRuntime = deferCastValidationToRuntime) } if (resKey.length == 1 && resValue.length == 1) { @@ -833,7 +873,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { byName: Boolean, conf: SQLConf, addError: String => Unit, - colPath: Seq[String]): Option[NamedExpression] = { + colPath: Seq[String], + deferCastValidationToRuntime: Boolean): Option[NamedExpression] = { val attrTypeHasCharVarchar = CharVarcharUtils.hasCharVarchar(tableAttr.dataType) val attrTypeWithoutCharVarchar = if (attrTypeHasCharVarchar) { @@ -844,7 +885,8 @@ object TableOutputResolver extends SQLConfHelper with Logging { val canWriteExpr = canWrite( tableName, queryExpr.dataType, attrTypeWithoutCharVarchar, - byName, conf, addError, colPath) + byName, conf, addError, colPath, + deferCastValidationToRuntime = deferCastValidationToRuntime) if (canWriteExpr) { val prepared = 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 50179824e255f..04169c0e927d5 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 @@ -19,7 +19,7 @@ package org.apache.spark.sql.connector.write import java.util -import org.apache.spark.sql.connector.catalog.{Column, SupportsRead, SupportsRowLevelOperations, SupportsWrite, Table, TableCapability} +import org.apache.spark.sql.connector.catalog.{Column, SchemaAlignmentConfig, SupportsRead, SupportsRowLevelOperations, SupportsWrite, Table, TableCapability} import org.apache.spark.sql.connector.catalog.constraints.Constraint import org.apache.spark.sql.connector.read.ScanBuilder import org.apache.spark.sql.util.CaseInsensitiveStringMap @@ -40,6 +40,7 @@ private[sql] case class RowLevelOperationTable( override def columns: Array[Column] = table.columns() override def capabilities: util.Set[TableCapability] = table.capabilities override def constraints(): Array[Constraint] = table.constraints() + override def schemaAlignmentConfig(): SchemaAlignmentConfig = table.schemaAlignmentConfig() override def toString: String = table.toString override def newScanBuilder(options: CaseInsensitiveStringMap): ScanBuilder = { 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 39634eb4c8343..056cc35e835ff 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 @@ -64,7 +64,8 @@ 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 { 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 7556bc96912f6..b79bd5fbb3d48 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 @@ -47,14 +47,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 +65,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 +110,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 +378,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/InMemoryTable.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/connector/catalog/InMemoryTable.scala index c783bfbece149..f4a55cd402fbe 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 @@ -47,10 +47,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, 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 f48f99994a8cc..0a18f32e70295 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/core/src/test/scala/org/apache/spark/sql/connector/InsertSchemaEvolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/connector/InsertSchemaEvolutionSuite.scala new file mode 100644 index 0000000000000..f10cf3e317346 --- /dev/null +++ b/sql/core/src/test/scala/org/apache/spark/sql/connector/InsertSchemaEvolutionSuite.scala @@ -0,0 +1,192 @@ +/* + * 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 + +import org.scalatest.BeforeAndAfter + +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.connector.catalog.InMemoryCatalog +import org.apache.spark.sql.internal.SQLConf.{PARTITION_OVERWRITE_MODE, PartitionOverwriteMode} +import org.apache.spark.sql.test.SharedSparkSession +import org.apache.spark.sql.types._ + +class InsertSchemaEvolutionSuite + extends QueryTest with SharedSparkSession with BeforeAndAfter { + + private val catalogName = "testcat" + private val namespace = "ns" + private val tableIdent = s"$catalogName.$namespace.test_table" + + before { + spark.conf.set(s"spark.sql.catalog.$catalogName", classOf[InMemoryCatalog].getName) + } + + after { + spark.sessionState.catalogManager.reset() + spark.sessionState.conf.unsetConf(s"spark.sql.catalog.$catalogName") + } + + test("INSERT BY NAME with extra source column adds column to table") { + withTable(tableIdent) { + sql(s"CREATE TABLE $tableIdent (id INT, data STRING)") + sql( + s"""INSERT WITH SCHEMA EVOLUTION INTO $tableIdent BY NAME + |SELECT * FROM VALUES (1, 'a', CAST(10.0 AS DOUBLE)), + | (2, 'b', CAST(20.0 AS DOUBLE)) AS t(id, data, amount) + |""".stripMargin) + + val result = spark.table(tableIdent) + checkAnswer(result, Seq(Row(1, "a", 10.0d), Row(2, "b", 20.0d))) + assert(result.schema == StructType(Seq( + StructField("id", IntegerType), + StructField("data", StringType), + StructField("amount", DoubleType)))) + } + } + + test("INSERT BY NAME with type widening updates column type") { + withTable(tableIdent) { + sql(s"CREATE TABLE $tableIdent (id INT, value INT)") + sql( + s"""INSERT WITH SCHEMA EVOLUTION INTO $tableIdent BY NAME + |SELECT * FROM VALUES (1, CAST(100 AS LONG)), + | (2, CAST(200 AS LONG)) AS t(id, value) + |""".stripMargin) + + val result = spark.table(tableIdent) + checkAnswer(result, Seq(Row(1, 100L), Row(2, 200L))) + assert(result.schema == StructType(Seq( + StructField("id", IntegerType), + StructField("value", LongType)))) + } + } + + test("INSERT BY NAME with nested struct evolution") { + withTable(tableIdent) { + sql(s"CREATE TABLE $tableIdent (id INT, info STRUCT)") + sql( + s"""INSERT WITH SCHEMA EVOLUTION INTO $tableIdent BY NAME + |SELECT id, named_struct('name', name, 'age', age) AS info + |FROM VALUES (1, 'Alice', 30), (2, 'Bob', 25) AS t(id, name, age) + |""".stripMargin) + + val result = spark.table(tableIdent) + checkAnswer(result, Seq(Row(1, Row("Alice", 30)), Row(2, Row("Bob", 25)))) + val expectedInfoType = StructType(Seq( + StructField("name", StringType), + StructField("age", IntegerType))) + assert(result.schema == StructType(Seq( + StructField("id", IntegerType), + StructField("info", expectedInfoType)))) + } + } + + test("INSERT BY NAME with matching schema - no evolution needed") { + withTable(tableIdent) { + sql(s"CREATE TABLE $tableIdent (id INT, data STRING)") + sql( + s"""INSERT WITH SCHEMA EVOLUTION INTO $tableIdent BY NAME + |SELECT * FROM VALUES (1, 'a'), (2, 'b') AS t(id, data) + |""".stripMargin) + + val result = spark.table(tableIdent) + checkAnswer(result, Seq(Row(1, "a"), Row(2, "b"))) + assert(result.schema == StructType(Seq( + StructField("id", IntegerType), + StructField("data", StringType)))) + } + } + + test("INSERT BY POSITION with schema evolution adds extra columns") { + withTable(tableIdent) { + sql(s"CREATE TABLE $tableIdent (id INT, data STRING)") + sql( + s"""INSERT WITH SCHEMA EVOLUTION INTO $tableIdent + |SELECT * FROM VALUES (1, 'a', CAST(10.0 AS DOUBLE)), + | (2, 'b', CAST(20.0 AS DOUBLE)) AS t(id, data, amount) + |""".stripMargin) + + val result = spark.table(tableIdent) + checkAnswer(result, Seq(Row(1, "a", 10.0d), Row(2, "b", 20.0d))) + assert(result.schema == StructType(Seq( + StructField("id", IntegerType), + StructField("data", StringType), + StructField("amount", DoubleType)))) + } + } + + test("table without AUTOMATIC_SCHEMA_EVOLUTION - no evolution") { + withTable(tableIdent) { + sql( + s"""CREATE TABLE $tableIdent (id INT, data STRING) + |TBLPROPERTIES ('auto-schema-evolution' = 'false')""".stripMargin) + + intercept[Exception] { + sql( + s"""INSERT WITH SCHEMA EVOLUTION INTO $tableIdent BY NAME + |SELECT * FROM VALUES (1, 'a', CAST(10.0 AS DOUBLE)), + | (2, 'b', CAST(20.0 AS DOUBLE)) AS t(id, data, amount) + |""".stripMargin) + } + } + } + + test("OVERWRITE BY EXPRESSION with schema evolution") { + withTable(tableIdent) { + sql(s"CREATE TABLE $tableIdent (id INT, data STRING)") + sql(s"INSERT INTO $tableIdent VALUES (1, 'a'), (2, 'b')") + + withSQLConf(PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.STATIC.toString) { + sql( + s"""INSERT WITH SCHEMA EVOLUTION OVERWRITE $tableIdent BY NAME + |SELECT * FROM VALUES (3, 'c', CAST(30.0 AS DOUBLE)), + | (4, 'd', CAST(40.0 AS DOUBLE)) AS t(id, data, amount) + |""".stripMargin) + } + + val result = spark.table(tableIdent) + checkAnswer(result, Seq(Row(3, "c", 30.0d), Row(4, "d", 40.0d))) + assert(result.schema == StructType(Seq( + StructField("id", IntegerType), + StructField("data", StringType), + StructField("amount", DoubleType)))) + } + } + + test("OVERWRITE PARTITIONS DYNAMIC with schema evolution") { + withTable(tableIdent) { + sql(s"CREATE TABLE $tableIdent (id INT, data STRING) PARTITIONED BY (id)") + sql(s"INSERT INTO $tableIdent VALUES (1, 'a'), (2, 'b')") + + withSQLConf(PARTITION_OVERWRITE_MODE.key -> PartitionOverwriteMode.DYNAMIC.toString) { + sql( + s"""INSERT WITH SCHEMA EVOLUTION OVERWRITE $tableIdent BY NAME + |SELECT * FROM VALUES (1, 'c', CAST(30.0 AS DOUBLE)) AS t(id, data, amount) + |""".stripMargin) + } + + val result = spark.table(tableIdent) + checkAnswer(result.orderBy("id"), + Seq(Row(1, "c", 30.0d), Row(2, "b", null))) + assert(result.schema == StructType(Seq( + StructField("id", IntegerType), + StructField("data", StringType), + StructField("amount", DoubleType)))) + } + } +} 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 0000000000000..f1827d3ca4cf7 --- /dev/null +++ b/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(_)) => + case _ => + fail(s"relaxed and strict diverged: relaxed=$fromRelaxed strict=$fromStrict") + } + } + + test("deferCastValidationToRuntime: 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))) + assertRelaxedMatchesStrict(target, source) + } + + test("deferCastValidationToRuntime: nullable array element into non-null element type") { + val target = new StructType().add("a", ArrayType(IntegerType, containsNull = false)) + val source = spark.createDataFrame( + java.util.Arrays.asList(Row(Seq(1, 2))), + new StructType().add("a", ArrayType(IntegerType, containsNull = true))) + assertRelaxedMatchesStrict(target, source) + } + + test("deferCastValidationToRuntime: nullable map value into non-null value type") { + val target = new StructType() + .add("m", MapType(StringType, IntegerType, valueContainsNull = false)) + val source = spark.createDataFrame( + java.util.Arrays.asList(Row(Map("k" -> 1))), + new StructType().add("m", MapType(StringType, IntegerType, valueContainsNull = true))) + assertRelaxedMatchesStrict(target, source) + } + + test("deferCastValidationToRuntime: nullable child into non-null struct field") { + val target = new StructType() + .add("s", new StructType().add("a", IntegerType, nullable = false)) + val source = spark.createDataFrame( + java.util.Arrays.asList(Row(Row(1))), + new StructType().add("s", new StructType().add("a", IntegerType, nullable = true))) + assertRelaxedMatchesStrict(target, source) + } +} diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala index bcc4895616bdc..a8c7e3b559aaf 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/AlignAssignmentsSuiteBase.scala @@ -30,7 +30,7 @@ import org.apache.spark.sql.catalyst.expressions.objects.AssertNotNull import org.apache.spark.sql.catalyst.parser.CatalystSqlParser import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan import org.apache.spark.sql.catalyst.rules.Rule -import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SupportsRowLevelOperations, TableCapability, TableCatalog, TableContext, TableWritePrivilege} +import org.apache.spark.sql.connector.catalog.{CatalogManager, CatalogV2Util, Column, ColumnDefaultValue, Identifier, SchemaAlignmentConfig, SupportsRowLevelOperations, TableCapability, TableCatalog, TableContext, TableWritePrivilege} import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform} import org.apache.spark.sql.errors.QueryExecutionErrors import org.apache.spark.sql.execution.datasources.v2.V2SessionCatalog @@ -48,6 +48,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { .add("txt", "STRING") when(t.columns()).thenReturn(CatalogV2Util.structTypeToV2Columns(schema)) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -59,6 +60,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { .add("i", "INT") when(t.columns()).thenReturn(CatalogV2Util.structTypeToV2Columns(schema)) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -73,6 +75,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { .add("txt", "STRING") when(t.columns()).thenReturn(CatalogV2Util.structTypeToV2Columns(schema)) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -84,6 +87,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { .add("txt", "STRING") when(t.columns()).thenReturn(CatalogV2Util.structTypeToV2Columns(schema)) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -96,6 +100,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { .add("txt", "STRING") when(t.columns()).thenReturn(CatalogV2Util.structTypeToV2Columns(schema)) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -121,6 +126,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { nullable = false) when(t.columns()).thenReturn(CatalogV2Util.structTypeToV2Columns(schema)) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -132,6 +138,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { .add("txt", "STRING") when(t.columns()).thenReturn(CatalogV2Util.structTypeToV2Columns(schema)) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) when(t.capabilities()).thenReturn(Collections.singleton(TableCapability.ACCEPT_ANY_SCHEMA)) t } @@ -143,6 +150,7 @@ abstract class AlignAssignmentsSuiteBase extends AnalysisTest { Column.create("b", BooleanType, true, null, null), Column.create("i", IntegerType, true, null, iDefault, null))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala index f960507b300e5..09f2abb00cf84 100644 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala +++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/command/PlanResolutionSuite.scala @@ -36,7 +36,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{AlterColumns, AlterColumnSpe import org.apache.spark.sql.catalyst.rules.Rule import org.apache.spark.sql.catalyst.util.TypeUtils.toSQLId import org.apache.spark.sql.connector.FakeV2Provider -import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SupportsDelete, Table, TableCapability, TableCatalog, TableChange, TableContext, TableWritePrivilege, V1Table} +import org.apache.spark.sql.connector.catalog.{CatalogManager, Column, ColumnDefaultValue, Identifier, SchemaAlignmentConfig, SupportsDelete, Table, TableCapability, TableCatalog, TableChange, TableContext, TableWritePrivilege, V1Table} import org.apache.spark.sql.connector.catalog.CatalogManager.SESSION_CATALOG_NAME import org.apache.spark.sql.connector.expressions.{LiteralValue, Transform} import org.apache.spark.sql.errors.QueryExecutionErrors @@ -62,6 +62,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { when(t.columns()).thenReturn( Array(Column.create("i", IntegerType), Column.create("s", StringType))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -70,6 +71,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { when(t.columns()).thenReturn( Array(Column.create("s", StringType), Column.create("i", IntegerType))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -78,6 +80,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { when(t.columns()).thenReturn( Array(Column.create("i", IntegerType), Column.create("x", StringType, false))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) when(t.name()).thenReturn("tab2") t } @@ -95,6 +98,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { when(t.columns()).thenReturn( Array(Column.create("c1", CharType(5)), Column.create("c2", VarcharType(5)))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -106,6 +110,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { Column.create("i", BooleanType, true, null, default1, null), Column.create("s", IntegerType, true, null, default2, null))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -117,6 +122,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { Column.create("i", StringType), Column.create("e", StringType, true, null, default, null))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t } @@ -126,6 +132,7 @@ class PlanResolutionSuite extends SharedSparkSession with AnalysisTest { Column.create("s", StringType), Column.create("default", StringType))) when(t.partitioning()).thenReturn(Array.empty[Transform]) + when(t.schemaAlignmentConfig()).thenReturn(SchemaAlignmentConfig.DEFAULT) t }