Spark 4.2: Set Iceberg field IDs as Spark column IDs for DSv2 validation

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<Row> joined = df1.join(df2, df1.col("id").equalTo(df2.col("id")));
 
-    // query should resolve columns by name and show null for old data
-    List<Object[]> 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

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java

@@ -74,7 +74,9 @@ public DataType struct(Types.StructType struct, List<DataType> 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());
       }

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));

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

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: