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 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/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/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 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: