This is an automated email from the ASF dual-hosted git repository.
szehon-ho pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 0bfdb323c3 Spark 4.2: Implement SupportsSchemaEvolution for automatic
schema evolution (#17957)
0bfdb323c3 is described below
commit 0bfdb323c3db7828be93a33bad52c42db4d011ce
Author: Rahul Shivu Mahadev <[email protected]>
AuthorDate: Wed Oct 14 15:31:22 2026 -0700
Spark 4.2: Implement SupportsSchemaEvolution for automatic schema evolution
(#17957)
---
.../spark/extensions/TestMergeSchemaEvolution.java | 30 +++
.../iceberg/spark/source/SparkSchemaEvolution.java | 170 ++++++++++++++
.../apache/iceberg/spark/source/SparkTable.java | 15 +-
.../iceberg/spark/source/TestSparkTable.java | 243 +++++++++++++++++++++
.../spark/sql/TestInsertSchemaEvolution.java | 178 +++++++++++++++
5 files changed, 635 insertions(+), 1 deletion(-)
diff --git
a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java
b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java
index fa8314ecae..ad438aa78b 100644
---
a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java
+++
b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestMergeSchemaEvolution.java
@@ -27,6 +27,7 @@ import org.apache.iceberg.ParameterizedTestExtension;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.spark.SparkTableProperties;
+import org.apache.spark.sql.types.DataTypes;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.TestTemplate;
@@ -262,6 +263,35 @@ public class TestMergeSchemaEvolution extends
SparkRowLevelOperationsTestBase {
sql("SELECT id, value FROM %s ORDER BY id", selectTarget()));
}
+ @TestTemplate
+ void mergeWithSchemaEvolutionCastsSmallIntToInt() {
+ assumeThat(branch).as("Schema evolution does not work for branches
currently").isNull();
+
+ createAndInitTable(
+ "id INT, value INT", "{ \"id\": 1, \"value\": 100 }\n" + "{ \"id\": 2,
\"value\": 200 }");
+
+ createOrReplaceView(
+ "source",
+ "id INT, value SMALLINT",
+ "{ \"id\": 1, \"value\": 10 }\n" + "{ \"id\": 3, \"value\": 30 }");
+
+ sql(
+ "MERGE WITH SCHEMA EVOLUTION INTO %s AS t USING source AS s "
+ + "ON t.id == s.id "
+ + "WHEN MATCHED THEN "
+ + " UPDATE SET * "
+ + "WHEN NOT MATCHED THEN "
+ + " INSERT *",
+ commitTarget());
+
+ assertThat(spark.table(tableName).schema().apply("value").dataType())
+ .isEqualTo(DataTypes.IntegerType);
+ assertEquals(
+ "Should cast SMALLINT source values to INT",
+ ImmutableList.of(row(1, 10), row(2, 200), row(3, 30)),
+ sql("SELECT id, value FROM %s ORDER BY id", selectTarget()));
+ }
+
@TestTemplate
public void testMergeWithSchemaEvolutionDisabledByTableProperty() {
assumeThat(branch).as("Schema evolution does not work for branches
currently").isNull();
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkSchemaEvolution.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkSchemaEvolution.java
new file mode 100644
index 0000000000..4cad991668
--- /dev/null
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkSchemaEvolution.java
@@ -0,0 +1,170 @@
+/*
+ * 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.iceberg.spark.source;
+
+import java.util.Arrays;
+import java.util.Set;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.spark.SparkSchemaUtil;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.spark.sql.connector.catalog.TableChange;
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * Decides which column changes Spark may apply to a table through automatic
schema evolution.
+ *
+ * <p>The changes are checked against the schema pinned when the table was
loaded.
+ */
+class SparkSchemaEvolution {
+ private final Table table;
+ private final Schema schema;
+ private volatile Set<Integer> mapKeyFieldIds;
+
+ SparkSchemaEvolution(Table table, Schema schema) {
+ this.table = table;
+ this.schema = schema;
+ }
+
+ boolean supportsColumnChange(TableChange.ColumnChange change) {
+ if (isMapKeyChange(change)) {
+ return false;
+ }
+
+ if (change instanceof TableChange.AddColumn) {
+ TableChange.AddColumn add = (TableChange.AddColumn) change;
+ if (!add.isNullable() || add.defaultValue() != null) {
+ return false;
+ }
+
+ Type type = tryConvert(add.dataType());
+ return type != null && isSupportedAtFormatVersion(type);
+ } else if (change instanceof TableChange.UpdateColumnType) {
+ return supportsTypeUpdate((TableChange.UpdateColumnType) change);
+ } else if (change instanceof TableChange.UpdateColumnNullability) {
+ return supportsNullabilityUpdate((TableChange.UpdateColumnNullability)
change);
+ } else if (change instanceof TableChange.DeleteColumn) {
+ return supportsDeleteColumn((TableChange.DeleteColumn) change);
+ } else {
+ return change instanceof TableChange.RenameColumn
+ || change instanceof TableChange.UpdateColumnComment
+ || change instanceof TableChange.UpdateColumnPosition;
+ }
+ }
+
+ private Set<Integer> mapKeyFieldIds() {
+ if (mapKeyFieldIds == null) {
+ this.mapKeyFieldIds = mapKeyFieldIds(schema);
+ }
+
+ return mapKeyFieldIds;
+ }
+
+ private static Set<Integer> mapKeyFieldIds(Schema schema) {
+ Set<Integer> keyFieldIds = Sets.newHashSet();
+ for (Types.NestedField field :
TypeUtil.indexById(schema.asStruct()).values()) {
+ if (field.type().isMapType()) {
+ Types.MapType map = field.type().asMapType();
+
keyFieldIds.addAll(TypeUtil.indexById(Types.StructType.of(map.fields().get(0))).keySet());
+ }
+ }
+
+ return keyFieldIds;
+ }
+
+ private boolean isMapKeyChange(TableChange.ColumnChange change) {
+ String[] fieldNames = change.fieldNames();
+ int pathLength =
+ change instanceof TableChange.AddColumn ? fieldNames.length - 1 :
fieldNames.length;
+ if (pathLength == 0) {
+ return false;
+ }
+
+ Types.NestedField field =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames,
pathLength)));
+ return field != null && mapKeyFieldIds().contains(field.fieldId());
+ }
+
+ private boolean supportsTypeUpdate(TableChange.UpdateColumnType update) {
+ Types.NestedField field = schema.findField(String.join(".",
update.fieldNames()));
+ if (field == null) {
+ return false;
+ }
+
+ Type newType = tryConvert(update.newDataType());
+ return newType != null
+ && newType.isPrimitiveType()
+ && TypeUtil.isPromotionAllowed(field.type(), newType.asPrimitiveType())
+ && SparkSchemaUtil.convert(newType).equals(update.newDataType());
+ }
+
+ private boolean
supportsNullabilityUpdate(TableChange.UpdateColumnNullability update) {
+ Types.NestedField field = schema.findField(String.join(".",
update.fieldNames()));
+ return update.nullable() && field != null &&
!containsIdentifierField(field);
+ }
+
+ private boolean supportsDeleteColumn(TableChange.DeleteColumn delete) {
+ Types.NestedField field = schema.findField(String.join(".",
delete.fieldNames()));
+ if (field == null || isCollectionElement(delete.fieldNames())) {
+ return false;
+ }
+
+ // Iceberg rejects dropping an identifier field or a field whose subtree
contains one
+ return !containsIdentifierField(field);
+ }
+
+ // a list element or a map key or value can only be dropped together with
its collection
+ private boolean isCollectionElement(String[] fieldNames) {
+ if (fieldNames.length < 2) {
+ return false;
+ }
+
+ Types.NestedField parent =
+ schema.findField(String.join(".", Arrays.copyOf(fieldNames,
fieldNames.length - 1)));
+ return parent != null && (parent.type().isListType() ||
parent.type().isMapType());
+ }
+
+ private boolean containsIdentifierField(Types.NestedField field) {
+ Set<Integer> identifierFieldIds = schema.identifierFieldIds();
+ return TypeUtil.indexById(Types.StructType.of(field)).keySet().stream()
+ .anyMatch(identifierFieldIds::contains);
+ }
+
+ private boolean isSupportedAtFormatVersion(Type type) {
+ try {
+ Schema typeSchema = new Schema(Types.NestedField.optional(-1, "value",
type));
+ Schema.checkCompatibility(typeSchema, TableUtil.formatVersion(table));
+ return true;
+ } catch (IllegalArgumentException | IllegalStateException e) {
+ return false;
+ }
+ }
+
+ private static Type tryConvert(DataType sparkType) {
+ try {
+ return SparkSchemaUtil.convert(sparkType);
+ } catch (IllegalArgumentException | UnsupportedOperationException e) {
+ return null;
+ }
+ }
+}
diff --git
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
index 1e767f2e8c..2162d825e0 100644
---
a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
+++
b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/source/SparkTable.java
@@ -60,8 +60,10 @@ import org.apache.iceberg.util.SnapshotUtil;
import org.apache.spark.sql.connector.catalog.SupportsDeleteV2;
import org.apache.spark.sql.connector.catalog.SupportsRead;
import org.apache.spark.sql.connector.catalog.SupportsRowLevelOperations;
+import org.apache.spark.sql.connector.catalog.SupportsSchemaEvolution;
import org.apache.spark.sql.connector.catalog.SupportsWrite;
import org.apache.spark.sql.connector.catalog.TableCapability;
+import org.apache.spark.sql.connector.catalog.TableChange;
import org.apache.spark.sql.connector.catalog.constraints.Constraint;
import
org.apache.spark.sql.connector.catalog.constraints.Constraint.ValidationStatus;
import org.apache.spark.sql.connector.expressions.filter.Predicate;
@@ -80,7 +82,11 @@ import org.slf4j.LoggerFactory;
* <p>Note the table state (e.g. schema, snapshot) is pinned upon loading and
must not change.
*/
public class SparkTable extends BaseSparkTable
- implements SupportsRead, SupportsWrite, SupportsDeleteV2,
SupportsRowLevelOperations {
+ implements SupportsRead,
+ SupportsWrite,
+ SupportsDeleteV2,
+ SupportsRowLevelOperations,
+ SupportsSchemaEvolution {
private static final Logger LOG = LoggerFactory.getLogger(SparkTable.class);
@@ -94,6 +100,7 @@ public class SparkTable extends BaseSparkTable
TableCapability.OVERWRITE_DYNAMIC);
private final Schema schema; // effective schema (not necessarily current
table schema)
+ private final SparkSchemaEvolution schemaEvolution;
private final Snapshot snapshot; // always set unless table is empty
private final String branch; // set if table is loaded for specific branch
private final TimeTravel timeTravel; // set if table is loaded for time
travel
@@ -125,6 +132,7 @@ public class SparkTable extends BaseSparkTable
Table table, Schema schema, Snapshot snapshot, String branch, TimeTravel
timeTravel) {
super(table, schema);
this.schema = schema;
+ this.schemaEvolution = new SparkSchemaEvolution(table, schema);
this.snapshot = snapshot;
this.branch = branch;
this.timeTravel = timeTravel;
@@ -163,6 +171,11 @@ public class SparkTable extends BaseSparkTable
return capabilities;
}
+ @Override
+ public boolean supportsColumnChange(TableChange.ColumnChange change) {
+ return schemaEvolution.supportsColumnChange(change);
+ }
+
@Override
public Constraint[] constraints() {
List<Constraint> constraints = Lists.newArrayList();
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkTable.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkTable.java
index 106d908b49..4b1d2e5f65 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkTable.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkTable.java
@@ -29,13 +29,21 @@ import org.apache.iceberg.TableProperties;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.spark.CatalogTestBase;
import org.apache.iceberg.spark.SparkSQLProperties;
+import org.apache.iceberg.spark.SparkTableProperties;
+import org.apache.iceberg.types.Types;
import org.apache.spark.sql.catalyst.analysis.NoSuchTableException;
import org.apache.spark.sql.connector.catalog.CatalogManager;
import org.apache.spark.sql.connector.catalog.Identifier;
+import org.apache.spark.sql.connector.catalog.TableCapability;
import org.apache.spark.sql.connector.catalog.TableCatalog;
+import org.apache.spark.sql.connector.catalog.TableChange;
import org.apache.spark.sql.connector.catalog.constraints.Constraint;
import org.apache.spark.sql.connector.catalog.constraints.PrimaryKey;
import org.apache.spark.sql.connector.expressions.NamedReference;
+import org.apache.spark.sql.types.DataType;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.types.VariantType$;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.TestTemplate;
@@ -56,6 +64,241 @@ public class TestSparkTable extends CatalogTestBase {
sql("DROP TABLE IF EXISTS %s", tableName);
}
+ @TestTemplate
+ public void testSupportedSchemaEvolutionChanges() {
+ SparkTable table = loadSparkTable();
+
+
assertThat(table.capabilities()).contains(TableCapability.AUTOMATIC_SCHEMA_EVOLUTION);
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.addColumn(new String[] {"new_col"},
DataTypes.IntegerType, true)))
+ .isTrue();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.addColumn(new String[] {"new_col"},
DataTypes.IntegerType, false)))
+ .isFalse();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(new String[] {"id"},
DataTypes.LongType)))
+ .isTrue();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(new String[] {"id"},
DataTypes.IntegerType)))
+ .isFalse();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnNullability(new String[] {"data"},
true)))
+ .isTrue();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnNullability(new String[] {"data"},
false)))
+ .isFalse();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnDefaultValue(new String[]
{"data"}, "'default'")))
+ .isFalse();
+ }
+
+ @TestTemplate
+ void addColumnsRespectFormatVersion() {
+ List<DataType> types =
+ Arrays.asList(
+ VariantType$.MODULE$,
+ DataTypes.NullType,
+ new StructType().add("v", VariantType$.MODULE$),
+ DataTypes.createMapType(DataTypes.StringType,
VariantType$.MODULE$),
+ DataTypes.createArrayType(VariantType$.MODULE$));
+
+ for (int formatVersion : new int[] {2, 3}) {
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = '%s')",
+ tableName, TableProperties.FORMAT_VERSION, formatVersion);
+ SparkTable table = loadSparkTable();
+
+ for (DataType type : types) {
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.addColumn(new String[] {"new_col"}, type,
true)))
+ .as("Adding %s to a v%s table", type, formatVersion)
+ .isEqualTo(formatVersion >= 3);
+ }
+ }
+ }
+
+ @TestTemplate
+ void variantTypeUpdatesAreNotSupported() {
+ for (int formatVersion : new int[] {2, 3}) {
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = '%s')",
+ tableName, TableProperties.FORMAT_VERSION, formatVersion);
+ SparkTable table = loadSparkTable();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(new String[] {"data"},
VariantType$.MODULE$)))
+ .as("Updating a column to variant in a v%s table", formatVersion)
+ .isFalse();
+ }
+ }
+
+ @TestTemplate
+ void identifierColumnCannotBecomeNullable() {
+ SparkTable table = loadSparkTable();
+
table.table().updateSchema().allowIncompatibleChanges().setIdentifierFields("id").commit();
+ table = loadSparkTable();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnNullability(new String[] {"id"},
true)))
+ .isFalse();
+ }
+
+ @TestTemplate
+ void identifierStructCannotBecomeNullable() {
+ SparkTable table = loadSparkTable();
+ table
+ .table()
+ .updateSchema()
+ .allowIncompatibleChanges()
+ .addRequiredColumn(
+ "parent",
+ Types.StructType.of(Types.NestedField.required(1, "id",
Types.LongType.get())))
+ .setIdentifierFields("parent.id")
+ .commit();
+ table = loadSparkTable();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnNullability(new String[]
{"parent"}, true)))
+ .isFalse();
+ }
+
+ @TestTemplate
+ void missingColumnCannotBecomeNullable() {
+ SparkTable table = loadSparkTable();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnNullability(new String[]
{"missing"}, true)))
+ .isFalse();
+ }
+
+ @TestTemplate
+ void collectionElementsCannotBeDeleted() {
+ sql("ALTER TABLE %s ADD COLUMN items array<struct<x: int, y: int>>",
tableName);
+ sql("ALTER TABLE %s ADD COLUMN m map<int, string>", tableName);
+ SparkTable table = loadSparkTable();
+
+ for (String[] path : new String[][] {{"items", "element"}, {"m", "key"},
{"m", "value"}}) {
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange) TableChange.deleteColumn(path,
false)))
+ .as("Deleting %s", String.join(".", path))
+ .isFalse();
+ }
+
+ // a field of a list element struct can still be deleted
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.deleteColumn(new String[] {"items", "element",
"x"}, false)))
+ .isTrue();
+ }
+
+ @TestTemplate
+ public void testMapKeySchemaEvolutionChanges() {
+ sql("ALTER TABLE %s ADD COLUMN m map<int, string>", tableName);
+ sql("ALTER TABLE %s ADD COLUMN value_map map<int, int>", tableName);
+ sql("ALTER TABLE %s ADD COLUMN struct_key_map map<struct<key_field: int>,
string>", tableName);
+ sql("ALTER TABLE %s ADD COLUMN struct_value_map map<int,
struct<value_field: int>>", tableName);
+ SparkTable table = loadSparkTable();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(new String[] {"m", "key"},
DataTypes.LongType)))
+ .isFalse();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(
+ new String[] {"struct_key_map", "key", "key_field"},
DataTypes.LongType)))
+ .isFalse();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.addColumn(
+ new String[] {"struct_key_map", "key", "new_field"},
+ DataTypes.IntegerType,
+ true)))
+ .isFalse();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(
+ new String[] {"value_map", "value"},
DataTypes.LongType)))
+ .isTrue();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.addColumn(
+ new String[] {"struct_value_map", "value",
"new_field"},
+ DataTypes.IntegerType,
+ true)))
+ .isTrue();
+ }
+
+ @TestTemplate
+ void typeUpdatesMustRoundTripToSparkType() {
+ sql("ALTER TABLE %s ADD COLUMN int_col int", tableName);
+ SparkTable table = loadSparkTable();
+
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(new String[] {"int_col"},
DataTypes.ShortType)))
+ .isFalse();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(new String[] {"int_col"},
DataTypes.ByteType)))
+ .isFalse();
+ assertThat(
+ table.supportsColumnChange(
+ (TableChange.ColumnChange)
+ TableChange.updateColumnType(new String[] {"int_col"},
DataTypes.LongType)))
+ .isTrue();
+ }
+
+ @TestTemplate
+ void schemaEvolutionWithAcceptAnySchema() {
+ sql(
+ "ALTER TABLE %s SET TBLPROPERTIES ('%s' = 'true')",
+ tableName, SparkTableProperties.WRITE_ACCEPT_ANY_SCHEMA);
+ SparkTable table = loadSparkTable();
+ TableChange.ColumnChange change =
+ (TableChange.ColumnChange)
+ TableChange.addColumn(new String[] {"new_col"},
DataTypes.IntegerType, true);
+
+ assertThat(table.supportsColumnChange(change)).isTrue();
+ assertThat(table.supportsColumnChange(change)).isTrue();
+ }
+
@TestTemplate
public void testTableEquality() {
SparkTable table1 = loadSparkTable();
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestInsertSchemaEvolution.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestInsertSchemaEvolution.java
new file mode 100644
index 0000000000..c190f4e087
--- /dev/null
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestInsertSchemaEvolution.java
@@ -0,0 +1,178 @@
+/*
+ * 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.iceberg.spark.sql;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.Parameter;
+import org.apache.iceberg.ParameterizedTestExtension;
+import org.apache.iceberg.Parameters;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.CatalogTestBase;
+import org.apache.iceberg.spark.SparkSQLProperties;
+import org.apache.iceberg.spark.SparkTableProperties;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.TestTemplate;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+@ExtendWith(ParameterizedTestExtension.class)
+class TestInsertSchemaEvolution extends CatalogTestBase {
+
+ @Parameter(index = 3)
+ private boolean byName;
+
+ @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2},
byName = {3}")
+ protected static Object[][] parameters() {
+ List<Object[]> parameters = Lists.newArrayList();
+ for (Object[] catalogParams : CatalogTestBase.parameters()) {
+ for (boolean byName : new boolean[] {false, true}) {
+ parameters.add(new Object[] {catalogParams[0], catalogParams[1],
catalogParams[2], byName});
+ }
+ }
+
+ return parameters.toArray(new Object[0][]);
+ }
+
+ @AfterEach
+ void removeTable() {
+ sql("DROP TABLE IF EXISTS %s", tableName);
+ }
+
+ @TestTemplate
+ void addColumnWhileCastingSmallintToInt() {
+ sql(
+ "CREATE TABLE %s (id INT, data STRING) USING iceberg TBLPROPERTIES
('%s' = 'false')",
+ tableName, SparkTableProperties.WRITE_ACCEPT_ANY_SCHEMA);
+ sql("INSERT INTO %s VALUES (1, 'old')", tableName);
+
+ sql(
+ "INSERT WITH SCHEMA EVOLUTION INTO %s %s SELECT %s "
+ + "FROM VALUES (CAST(2 AS SMALLINT), 'new', 'added') AS source(id,
data, extra)",
+ tableName, byName ? "BY NAME" : "", byName ? "extra, data, id" : "id,
data, extra");
+
+ Schema expectedSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+ assertThat(validationCatalog.loadTable(tableIdent).schema().asStruct())
+ .isEqualTo(expectedSchema.asStruct());
+ assertEquals(
+ "Should retain INT values and add a nullable column",
+ Lists.newArrayList(row(1, "old", null), row(2, "new", "added")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ }
+
+ @TestTemplate
+ void widenIntToBigint() {
+ sql(
+ "CREATE TABLE %s (id INT, data STRING) USING iceberg TBLPROPERTIES
('%s' = 'false')",
+ tableName, SparkTableProperties.WRITE_ACCEPT_ANY_SCHEMA);
+ sql("INSERT INTO %s VALUES (1, 'old')", tableName);
+
+ long newId = (long) Integer.MAX_VALUE + 1;
+ sql(
+ "INSERT WITH SCHEMA EVOLUTION INTO %s %s SELECT %s "
+ + "FROM VALUES (CAST(%d AS BIGINT), 'new') AS source(id, data)",
+ tableName, byName ? "BY NAME" : "", byName ? "data, id" : "id, data",
newId);
+
+ Schema expectedSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()));
+ assertThat(validationCatalog.loadTable(tableIdent).schema().asStruct())
+ .isEqualTo(expectedSchema.asStruct());
+ assertEquals(
+ "Should preserve old and new values after widening to BIGINT",
+ Lists.newArrayList(row(1L, "old"), row(newId, "new")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ }
+
+ @TestTemplate
+ void castBigintMapKeysToInt() {
+ sql(
+ "CREATE TABLE %s (id INT, data MAP<INT, STRING>) USING iceberg "
+ + "TBLPROPERTIES ('%s' = 'false')",
+ tableName, SparkTableProperties.WRITE_ACCEPT_ANY_SCHEMA);
+ sql("INSERT INTO %s SELECT 1, map(1, 'old')", tableName);
+
+ sql(
+ "INSERT WITH SCHEMA EVOLUTION INTO %s %s SELECT %s "
+ + "FROM VALUES (2, map(CAST(2 AS BIGINT), 'new', CAST(3 AS
BIGINT), 'other')) "
+ + "AS source(id, data)",
+ tableName, byName ? "BY NAME" : "", byName ? "data, id" : "id, data");
+
+ Schema expectedSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(
+ 2,
+ "data",
+ Types.MapType.ofOptional(3, 4, Types.IntegerType.get(),
Types.StringType.get())));
+ assertThat(validationCatalog.loadTable(tableIdent).schema().asStruct())
+ .isEqualTo(expectedSchema.asStruct());
+ Map<Integer, String> oldMap = Maps.newHashMap();
+ oldMap.put(1, "old");
+ Map<Integer, String> newMap = Maps.newHashMap();
+ newMap.put(2, "new");
+ newMap.put(3, "other");
+ assertEquals(
+ "Should preserve map values and retain INT keys",
+ Lists.newArrayList(row(1, oldMap), row(2, newMap)),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ }
+
+ @TestTemplate
+ void evolveSchemaWithAcceptAnySchemaAndMergeSchema() {
+ sql(
+ "CREATE TABLE %s (id INT, data STRING) USING iceberg TBLPROPERTIES
('%s' = 'true')",
+ tableName, SparkTableProperties.WRITE_ACCEPT_ANY_SCHEMA);
+ sql("INSERT INTO %s SELECT 1 AS id, 'old' AS data", tableName);
+
+ long newId = (long) Integer.MAX_VALUE + 1;
+ Map<String, String> conf = Maps.newHashMap();
+ conf.put(SparkSQLProperties.MERGE_SCHEMA, "true");
+ // accept-any-schema skips Spark's by-name column alignment and Iceberg
checks the column order
+ // by default, so the source columns stay in table order for both insert
modes
+ withSQLConf(
+ conf,
+ () ->
+ sql(
+ "INSERT WITH SCHEMA EVOLUTION INTO %s %s SELECT id, data,
extra "
+ + "FROM VALUES (CAST(%d AS BIGINT), 'new', 'added') AS
source(id, data, extra)",
+ tableName, byName ? "BY NAME" : "", newId));
+
+ Schema expectedSchema =
+ new Schema(
+ Types.NestedField.optional(1, "id", Types.LongType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get()),
+ Types.NestedField.optional(3, "extra", Types.StringType.get()));
+ assertThat(validationCatalog.loadTable(tableIdent).schema().asStruct())
+ .isEqualTo(expectedSchema.asStruct());
+ assertEquals(
+ "Should widen and add columns with native and legacy evolution
enabled",
+ Lists.newArrayList(row(1L, "old", null), row(newId, "new", "added")),
+ sql("SELECT * FROM %s ORDER BY id", tableName));
+ }
+}
