From f9d1338eae1c3b644b4c313017e6da80883a3995 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Fri, 4 Sep 2026 20:59:17 +0000 Subject: [PATCH 1/3] Spark 4.2: Set Iceberg field IDs as Spark column IDs for DSv2 validation Set StructField.withId with each Iceberg field ID during Schema to StructType conversion, so SparkTable.columns() exposes stable Column.id() for top-level and nested struct fields and Spark 4.2's nested-field column-ID validation (SPARK-57544) works with Iceberg tables. Adds a TestSparkSchemaUtil test. --- .../apache/iceberg/spark/TypeToSparkType.java | 4 ++- .../iceberg/spark/TestSparkSchemaUtil.java | 30 +++++++++++++++++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java index dc077937577c..7290099d5b75 100644 --- a/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java +++ b/spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java @@ -74,7 +74,9 @@ public DataType struct(Types.StructType struct, List fieldResults) { Types.NestedField field = fields.get(i); DataType type = fieldResults.get(i); Metadata metadata = fieldMetadata(field.fieldId()); - StructField sparkField = StructField.apply(field.name(), type, field.isOptional(), metadata); + StructField sparkField = + StructField.apply(field.name(), type, field.isOptional(), metadata) + .withId(Integer.toString(field.fieldId())); if (field.doc() != null) { sparkField = sparkField.withComment(field.doc()); } diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index ba0dfd84110b..d1c01d4e4d10 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -37,6 +37,8 @@ import org.apache.spark.sql.catalyst.expressions.MetadataAttribute; import org.apache.spark.sql.catalyst.types.DataTypeUtils; import org.apache.spark.sql.catalyst.util.ResolveDefaultColumnsUtils$; +import org.apache.spark.sql.connector.catalog.CatalogV2Util; +import org.apache.spark.sql.connector.catalog.Column; import org.apache.spark.sql.types.DataType; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.GeographyType; @@ -101,6 +103,34 @@ public void testSchemaConversionWithMetaDataColumnSchema() { } } + @Test + public void testFieldIDsInSparkColumns() { + Schema schema = + new Schema( + optional( + 10, + "location", + Types.StructType.of( + optional(11, "latitude", Types.DoubleType.get()), + optional(12, "longitude", Types.DoubleType.get())))); + + StructType sparkSchema = SparkSchemaUtil.convert(schema); + StructField location = sparkSchema.apply("location"); + StructType locationType = (StructType) location.dataType(); + + assertThat(location.id().get()).isEqualTo("10"); + assertThat(locationType.apply("latitude").id().get()).isEqualTo("11"); + assertThat(locationType.apply("longitude").id().get()).isEqualTo("12"); + + Column[] columns = CatalogV2Util.structTypeToV2Columns(sparkSchema, true); + assertThat(columns).hasSize(1); + assertThat(columns[0].id()).isEqualTo("10"); + + StructType columnType = (StructType) columns[0].dataType(); + assertThat(columnType.apply("latitude").id().get()).isEqualTo("11"); + assertThat(columnType.apply("longitude").id().get()).isEqualTo("12"); + } + @Test public void testGeospatialTypeConversion() { // a default-CRS geometry round-trips through the null <-> OGC:CRS84 normalization From 1ea141bbaa1e46eaee4bac8ab9692ae67a8632a5 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Thu, 8 Oct 2026 22:59:35 +0000 Subject: [PATCH 2/3] Spark 4.2: Expect column ID mismatch after external drop and re-add Dropping and re-adding salary gives it a new field ID, which Spark 4.2 now detects through the column IDs, so the stale DataFrame fails with INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH. --- .../TestIncrementallyConstructedQueries.java | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) diff --git a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestIncrementallyConstructedQueries.java b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestIncrementallyConstructedQueries.java index a5b63c358058..6c4c1985a0f0 100644 --- a/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestIncrementallyConstructedQueries.java +++ b/spark/v4.2/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestIncrementallyConstructedQueries.java @@ -239,12 +239,11 @@ public void testIncrementalQueryWithExternalDropAndAddColumnSameNameSameType() // join two DataFrames (QueryExecution #3) Dataset joined = df1.join(df2, df1.col("id").equalTo(df2.col("id"))); - // query should resolve columns by name and show null for old data - List result = rowsToJava(joined.collectAsList()); - assertThat(result) - .as("Join should resolve columns by name") - .hasSize(2) - .containsExactlyInAnyOrder(row(1, null, 1, null), row(2, 200, 2, 200)); + // the re-added column has a new field ID, so Spark detects that df1 is out of date + assertThatThrownBy(joined::collect) + .isInstanceOf(AnalysisException.class) + .hasMessageContaining("INCOMPATIBLE_TABLE_CHANGE_AFTER_ANALYSIS.COLUMNS_MISMATCH") + .hasMessageContaining("`salary` field ID has changed from 2 to 3"); } @TestTemplate From a8853cfe78867b5f4fab512f54411b33f250a2e0 Mon Sep 17 00:00:00 2001 From: rahulsmahadev Date: Sat, 10 Oct 2026 00:32:37 +0000 Subject: [PATCH 3/3] Spark 4.2: Account for Iceberg field IDs in column and partition type tests Spark columns and struct fields now carry Iceberg field IDs, so the column added by ALTER TABLE is expected with its ID, and metadata table partition types are compared without IDs because the expected types are parsed from DDL. --- .../iceberg/spark/TestSparkCatalogOperations.java | 4 +++- .../TestMetadataTablesWithPartitionEvolution.java | 10 ++++++++-- 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java index 27d99127b764..b1f21370730d 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/TestSparkCatalogOperations.java @@ -134,7 +134,9 @@ public void testAlterTable() throws NoSuchTableException { assertThat(table).as("Should return updated table").isNotNull(); - Column expectedField = Column.create(fieldName, DataTypes.StringType, true); + // the new column carries the field ID that Iceberg assigned to it + Column expectedField = + Column.builderFor(fieldName, DataTypes.StringType).nullable(true).id("3").build(); assertThat(table.columns()) .as("Adding a column to a table should return the updated table with the new column") .contains(expectedField, atIndex(2)); diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java index a417454b45dc..c55b569107da 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestMetadataTablesWithPartitionEvolution.java @@ -58,6 +58,7 @@ import org.apache.spark.sql.catalyst.parser.ParseException; import org.apache.spark.sql.types.DataType; import org.apache.spark.sql.types.StructType; +import org.apache.spark.sql.util.SchemaUtils; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; import org.junit.jupiter.api.extension.ExtendWith; @@ -670,20 +671,25 @@ private void assertPartitions( df = df.filter(filter); } + // the expected type is parsed from DDL and has no field IDs, so compare without them DataType expectedType = spark.sessionState().sqlParser().parseDataType(expectedTypeAsString); switch (tableType) { case PARTITIONS: case FILES: case ALL_DATA_FILES: DataType actualFilesType = df.schema().apply("partition").dataType(); - assertThat(actualFilesType).as("Partition type must match").isEqualTo(expectedType); + assertThat(SchemaUtils.clearFieldIds(actualFilesType)) + .as("Partition type must match") + .isEqualTo(expectedType); break; case ENTRIES: case ALL_ENTRIES: StructType dataFileType = (StructType) df.schema().apply("data_file").dataType(); DataType actualEntriesType = dataFileType.apply("partition").dataType(); - assertThat(actualEntriesType).as("Partition type must match").isEqualTo(expectedType); + assertThat(SchemaUtils.clearFieldIds(actualEntriesType)) + .as("Partition type must match") + .isEqualTo(expectedType); break; default: