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 6cd6183336 Spark 4.2: Set Iceberg field IDs as Spark column IDs for
DSv2 validation (#17958)
6cd6183336 is described below
commit 6cd6183336c86999648164fd2128c4985c17fb30
Author: Rahul Shivu Mahadev <[email protected]>
AuthorDate: Mon Oct 19 15:57:08 2026 -0700
Spark 4.2: Set Iceberg field IDs as Spark column IDs for DSv2 validation
(#17958)
---
.../TestIncrementallyConstructedQueries.java | 11 ++++----
.../org/apache/iceberg/spark/TypeToSparkType.java | 4 ++-
.../iceberg/spark/TestSparkCatalogOperations.java | 4 ++-
.../apache/iceberg/spark/TestSparkSchemaUtil.java | 30 ++++++++++++++++++++++
.../TestMetadataTablesWithPartitionEvolution.java | 10 ++++++--
5 files changed, 49 insertions(+), 10 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 a5b63c3580..6c4c1985a0 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 class TestIncrementallyConstructedQueries extends
ExtensionsTestBase {
// 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
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 dc07793757..7290099d5b 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 @@ class TypeToSparkType extends
TypeUtil.SchemaVisitor<DataType> {
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 27d99127b7..b1f2137073 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 class TestSparkCatalogOperations extends
CatalogTestBase {
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 ba0dfd8411..d1c01d4e4d 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.AttributeReference;
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 class TestSparkSchemaUtil {
}
}
+ @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 a417454b45..c55b569107 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.Row;
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 @@ public class TestMetadataTablesWithPartitionEvolution
extends CatalogTestBase {
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:
