From b9e112bce689985464b30417b7020832f9752be1 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Wed, 17 Jun 2026 09:12:45 -0700 Subject: [PATCH 1/9] Spark 4.1: Map geo Spark types Add schema conversion support between Iceberg geometry/geography types and Spark 4.1 geospatial DataTypes, and allow pruning projections over those types. --- .../spark/PruneColumnsWithoutReordering.java | 4 ++ .../apache/iceberg/spark/SparkTypeToType.java | 6 +++ .../apache/iceberg/spark/TypeToSparkType.java | 8 ++++ .../iceberg/spark/TestSparkSchemaUtil.java | 39 +++++++++++++++++++ 4 files changed, 57 insertions(+) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java index f4323f1c0350..003ba218cc85 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java @@ -38,6 +38,8 @@ import org.apache.spark.sql.types.DecimalType; import org.apache.spark.sql.types.DoubleType$; import org.apache.spark.sql.types.FloatType$; +import org.apache.spark.sql.types.GeographyType; +import org.apache.spark.sql.types.GeometryType; import org.apache.spark.sql.types.IntegerType$; import org.apache.spark.sql.types.LongType$; import org.apache.spark.sql.types.MapType; @@ -244,6 +246,8 @@ public Type primitive(Type.PrimitiveType primitive) { .put(TypeID.STRING, ImmutableSet.of(StringType$.class)) .put(TypeID.FIXED, ImmutableSet.of(BinaryType$.class)) .put(TypeID.BINARY, ImmutableSet.of(BinaryType$.class)) + .put(TypeID.GEOMETRY, ImmutableSet.of(GeometryType.class)) + .put(TypeID.GEOGRAPHY, ImmutableSet.of(GeographyType.class)) .put(TypeID.UNKNOWN, ImmutableSet.of(NullType$.class)) .buildOrThrow(); } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java index 54ad899ade77..0060b6ee6e91 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java @@ -32,6 +32,8 @@ import org.apache.spark.sql.types.DecimalType; import org.apache.spark.sql.types.DoubleType; import org.apache.spark.sql.types.FloatType; +import org.apache.spark.sql.types.GeographyType; +import org.apache.spark.sql.types.GeometryType; import org.apache.spark.sql.types.IntegerType; import org.apache.spark.sql.types.LongType; import org.apache.spark.sql.types.MapType; @@ -162,6 +164,10 @@ public Type atomic(DataType atomic) { ((DecimalType) atomic).precision(), ((DecimalType) atomic).scale()); } else if (atomic instanceof BinaryType) { return Types.BinaryType.get(); + } else if (atomic instanceof GeometryType) { + return Types.GeometryType.of(((GeometryType) atomic).crs()); + } else if (atomic instanceof GeographyType) { + return Types.GeographyType.of(((GeographyType) atomic).crs()); } else if (atomic instanceof NullType) { return Types.UnknownType.get(); } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java index 09c89bbba813..0b0f1ec083fb 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java @@ -34,6 +34,8 @@ import org.apache.spark.sql.types.DecimalType$; import org.apache.spark.sql.types.DoubleType$; import org.apache.spark.sql.types.FloatType$; +import org.apache.spark.sql.types.GeographyType$; +import org.apache.spark.sql.types.GeometryType$; import org.apache.spark.sql.types.IntegerType$; import org.apache.spark.sql.types.LongType$; import org.apache.spark.sql.types.MapType$; @@ -145,6 +147,12 @@ public DataType primitive(Type.PrimitiveType primitive) { return BinaryType$.MODULE$; case BINARY: return BinaryType$.MODULE$; + case GEOMETRY: + Types.GeometryType geometry = (Types.GeometryType) primitive; + return GeometryType$.MODULE$.apply(geometry.crs()); + case GEOGRAPHY: + Types.GeographyType geography = (Types.GeographyType) primitive; + return GeographyType$.MODULE$.apply(geography.crs()); case DECIMAL: Types.DecimalType decimal = (Types.DecimalType) primitive; return DecimalType$.MODULE$.apply(decimal.precision(), decimal.scale()); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index d5f407a715ef..fd6a98cb25d4 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -35,6 +35,11 @@ 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.types.DataType; +import org.apache.spark.sql.types.GeographyType; +import org.apache.spark.sql.types.GeographyType$; +import org.apache.spark.sql.types.GeometryType; +import org.apache.spark.sql.types.GeometryType$; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; import org.apache.spark.sql.types.StructType; @@ -93,6 +98,40 @@ public void testSchemaConversionWithMetaDataColumnSchema() { } } + @Test + public void testGeospatialTypeConversion() { + Types.GeometryType geometry = Types.GeometryType.of("EPSG:3857"); + DataType sparkGeometry = SparkSchemaUtil.convert(geometry); + assertThat(sparkGeometry).isInstanceOf(GeometryType.class); + assertThat(((GeometryType) sparkGeometry).crs()).isEqualTo("EPSG:3857"); + assertThat(SparkSchemaUtil.convert(sparkGeometry)).isEqualTo(geometry); + + Types.GeographyType geography = Types.GeographyType.of("EPSG:4326"); + DataType sparkGeography = SparkSchemaUtil.convert(geography); + assertThat(sparkGeography).isInstanceOf(GeographyType.class); + assertThat(((GeographyType) sparkGeography).crs()).isEqualTo("EPSG:4326"); + assertThat(SparkSchemaUtil.convert(sparkGeography)).isEqualTo(geography); + + assertThat(SparkSchemaUtil.convert(GeometryType$.MODULE$.apply("EPSG:3857"))) + .isEqualTo(geometry); + assertThat(SparkSchemaUtil.convert(GeographyType$.MODULE$.apply("EPSG:4326"))) + .isEqualTo(geography); + } + + @Test + public void testPruneGeospatialTypes() { + Schema schema = + new Schema( + optional(1, "geom", Types.GeometryType.of("EPSG:3857")), + optional(2, "geog", Types.GeographyType.of("EPSG:4326")), + optional(3, "id", Types.LongType.get())); + + StructType requestedType = SparkSchemaUtil.convert(schema); + Schema pruned = SparkSchemaUtil.prune(schema, requestedType); + + assertThat(pruned.asStruct()).isEqualTo(schema.asStruct()); + } + @Test public void testSchemaConversionWithOnlyWriteDefault() { Schema schema = From 129153df6887c20613aa9408763bfea8a03d3900 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Wed, 17 Jun 2026 22:16:13 -0700 Subject: [PATCH 2/9] Spark 4.1: Reject non-spherical geography mapping Avoid silently coercing Iceberg geography types with non-spherical edge algorithms when converting to Spark's geography type. --- .../java/org/apache/iceberg/spark/TypeToSparkType.java | 5 +++++ .../org/apache/iceberg/spark/TestSparkSchemaUtil.java | 8 ++++++++ 2 files changed, 13 insertions(+) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java index 0b0f1ec083fb..eb307d4cb4d7 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java @@ -22,6 +22,7 @@ import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.Schema; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.EdgeAlgorithm; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; @@ -152,6 +153,10 @@ public DataType primitive(Type.PrimitiveType primitive) { return GeometryType$.MODULE$.apply(geometry.crs()); case GEOGRAPHY: Types.GeographyType geography = (Types.GeographyType) primitive; + if (geography.algorithm() != EdgeAlgorithm.SPHERICAL) { + throw new UnsupportedOperationException( + "Spark does not support geography edge algorithm: " + geography.algorithm()); + } return GeographyType$.MODULE$.apply(geography.crs()); case DECIMAL: Types.DecimalType decimal = (Types.DecimalType) primitive; diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index fd6a98cb25d4..d4f734e913d8 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -20,6 +20,7 @@ import static org.apache.iceberg.types.Types.NestedField.optional; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.math.BigDecimal; import java.nio.ByteBuffer; @@ -29,6 +30,7 @@ import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.Schema; import org.apache.iceberg.expressions.Literal; +import org.apache.iceberg.types.EdgeAlgorithm; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.spark.sql.catalyst.expressions.AttributeReference; @@ -116,6 +118,12 @@ public void testGeospatialTypeConversion() { .isEqualTo(geometry); assertThat(SparkSchemaUtil.convert(GeographyType$.MODULE$.apply("EPSG:4326"))) .isEqualTo(geography); + + Types.GeographyType vincentyGeography = + Types.GeographyType.of("EPSG:4326", EdgeAlgorithm.VINCENTY); + assertThatThrownBy(() -> SparkSchemaUtil.convert(vincentyGeography)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Spark does not support geography edge algorithm: vincenty"); } @Test From 582c75961f0a598420294b9d7db11b708a36db1c Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Tue, 23 Jun 2026 20:08:45 -0700 Subject: [PATCH 3/9] Spark 4.1: Use OGC:CRS84 for geography type conversion tests Spark 4.1 only accepts OGC:CRS84 as the geography CRS, so the schema conversion tests now use it instead of EPSG:4326. Also document why the Spark edge algorithm is not propagated back to Iceberg. --- .../java/org/apache/iceberg/spark/SparkTypeToType.java | 3 +++ .../org/apache/iceberg/spark/TestSparkSchemaUtil.java | 10 +++++----- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java index 0060b6ee6e91..4df125deb89b 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java @@ -167,6 +167,9 @@ public Type atomic(DataType atomic) { } else if (atomic instanceof GeometryType) { return Types.GeometryType.of(((GeometryType) atomic).crs()); } else if (atomic instanceof GeographyType) { + // Spark only supports the spherical edge-interpolation algorithm, which matches the Iceberg + // default, so the Spark algorithm is intentionally not propagated here. Revisit if Spark + // starts supporting additional algorithms. return Types.GeographyType.of(((GeographyType) atomic).crs()); } else if (atomic instanceof NullType) { return Types.UnknownType.get(); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index d4f734e913d8..0ca42a5c6d5a 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -108,19 +108,19 @@ public void testGeospatialTypeConversion() { assertThat(((GeometryType) sparkGeometry).crs()).isEqualTo("EPSG:3857"); assertThat(SparkSchemaUtil.convert(sparkGeometry)).isEqualTo(geometry); - Types.GeographyType geography = Types.GeographyType.of("EPSG:4326"); + Types.GeographyType geography = Types.GeographyType.of("OGC:CRS84"); DataType sparkGeography = SparkSchemaUtil.convert(geography); assertThat(sparkGeography).isInstanceOf(GeographyType.class); - assertThat(((GeographyType) sparkGeography).crs()).isEqualTo("EPSG:4326"); + assertThat(((GeographyType) sparkGeography).crs()).isEqualTo("OGC:CRS84"); assertThat(SparkSchemaUtil.convert(sparkGeography)).isEqualTo(geography); assertThat(SparkSchemaUtil.convert(GeometryType$.MODULE$.apply("EPSG:3857"))) .isEqualTo(geometry); - assertThat(SparkSchemaUtil.convert(GeographyType$.MODULE$.apply("EPSG:4326"))) + assertThat(SparkSchemaUtil.convert(GeographyType$.MODULE$.apply("OGC:CRS84"))) .isEqualTo(geography); Types.GeographyType vincentyGeography = - Types.GeographyType.of("EPSG:4326", EdgeAlgorithm.VINCENTY); + Types.GeographyType.of("OGC:CRS84", EdgeAlgorithm.VINCENTY); assertThatThrownBy(() -> SparkSchemaUtil.convert(vincentyGeography)) .isInstanceOf(UnsupportedOperationException.class) .hasMessage("Spark does not support geography edge algorithm: vincenty"); @@ -131,7 +131,7 @@ public void testPruneGeospatialTypes() { Schema schema = new Schema( optional(1, "geom", Types.GeometryType.of("EPSG:3857")), - optional(2, "geog", Types.GeographyType.of("EPSG:4326")), + optional(2, "geog", Types.GeographyType.of("OGC:CRS84")), optional(3, "id", Types.LongType.get())); StructType requestedType = SparkSchemaUtil.convert(schema); From 55af6ce4a3656b788707c9d2fc7fa8118d276211 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Wed, 24 Jun 2026 21:56:43 -0700 Subject: [PATCH 4/9] Spark 4.1: Cover default-CRS geo round-trip and incompatible prune projection --- .../iceberg/spark/TestSparkSchemaUtil.java | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index 0ca42a5c6d5a..b4b6944fdacd 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -38,6 +38,7 @@ import org.apache.spark.sql.catalyst.types.DataTypeUtils; import org.apache.spark.sql.catalyst.util.ResolveDefaultColumnsUtils$; import org.apache.spark.sql.types.DataType; +import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.GeographyType; import org.apache.spark.sql.types.GeographyType$; import org.apache.spark.sql.types.GeometryType; @@ -102,6 +103,14 @@ public void testSchemaConversionWithMetaDataColumnSchema() { @Test public void testGeospatialTypeConversion() { + // a default-CRS geometry round-trips through the null <-> OGC:CRS84 normalization + Types.GeometryType defaultGeometry = Types.GeometryType.crs84(); + DataType sparkDefaultGeometry = SparkSchemaUtil.convert(defaultGeometry); + assertThat(sparkDefaultGeometry).isInstanceOf(GeometryType.class); + assertThat(((GeometryType) sparkDefaultGeometry).crs()) + .isEqualTo(Types.GeometryType.DEFAULT_CRS); + assertThat(SparkSchemaUtil.convert(sparkDefaultGeometry)).isEqualTo(defaultGeometry); + Types.GeometryType geometry = Types.GeometryType.of("EPSG:3857"); DataType sparkGeometry = SparkSchemaUtil.convert(geometry); assertThat(sparkGeometry).isInstanceOf(GeometryType.class); @@ -140,6 +149,19 @@ public void testPruneGeospatialTypes() { assertThat(pruned.asStruct()).isEqualTo(schema.asStruct()); } + @Test + public void testPruneGeospatialTypeWithIncompatibleRequestedType() { + Schema schema = new Schema(optional(1, "geom", Types.GeometryType.of("EPSG:3857"))); + + // requesting a non-geo Spark type for a geometry column must be rejected + StructType incompatibleType = new StructType().add("geom", DataTypes.BinaryType, true); + + assertThatThrownBy(() -> SparkSchemaUtil.prune(schema, incompatibleType)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Cannot project") + .hasMessageContaining("incompatible type"); + } + @Test public void testSchemaConversionWithOnlyWriteDefault() { Schema schema = From 5b01562900605dbd9595bdd81a5e66a66c1992d8 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Mon, 29 Jun 2026 13:19:53 -0700 Subject: [PATCH 5/9] Spark 4.1: Reject unsupported CRS and mixed SRID at the Spark boundary Iceberg permits any non-empty CRS, but Spark 4.1 recognizes only a fixed set (geometry: SRID:0 / EPSG:3857 / OGC:CRS84; geography: OGC:CRS84) and throws an opaque ST_INVALID_CRS_VALUE for the rest. Translate that into a clear UnsupportedOperationException when converting an Iceberg geo type to Spark, mirroring the existing edge-algorithm rejection. In the reverse direction a Spark mixed-SRID geometry/geography carries the sentinel CRS "SRID:ANY"; reject it instead of persisting a spec-invalid Iceberg CRS, since an Iceberg geo CRS is column-level. --- .../apache/iceberg/spark/SparkTypeToType.java | 14 +++++++-- .../apache/iceberg/spark/TypeToSparkType.java | 23 ++++++++++++-- .../iceberg/spark/TestSparkSchemaUtil.java | 30 +++++++++++++++++++ 3 files changed, 63 insertions(+), 4 deletions(-) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java index 4df125deb89b..833ebcda4a23 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java @@ -165,12 +165,22 @@ public Type atomic(DataType atomic) { } else if (atomic instanceof BinaryType) { return Types.BinaryType.get(); } else if (atomic instanceof GeometryType) { - return Types.GeometryType.of(((GeometryType) atomic).crs()); + GeometryType geometry = (GeometryType) atomic; + if (geometry.isMixedSrid()) { + throw new UnsupportedOperationException( + "Cannot convert Spark geometry with mixed SRID to Iceberg"); + } + return Types.GeometryType.of(geometry.crs()); } else if (atomic instanceof GeographyType) { + GeographyType geography = (GeographyType) atomic; + if (geography.isMixedSrid()) { + throw new UnsupportedOperationException( + "Cannot convert Spark geography with mixed SRID to Iceberg"); + } // Spark only supports the spherical edge-interpolation algorithm, which matches the Iceberg // default, so the Spark algorithm is intentionally not propagated here. Revisit if Spark // starts supporting additional algorithms. - return Types.GeographyType.of(((GeographyType) atomic).crs()); + return Types.GeographyType.of(geography.crs()); } else if (atomic instanceof NullType) { return Types.UnknownType.get(); } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java index eb307d4cb4d7..b6fe1a9e2201 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java @@ -26,6 +26,7 @@ import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; +import org.apache.spark.SparkIllegalArgumentException; import org.apache.spark.sql.catalyst.expressions.Literal$; import org.apache.spark.sql.types.ArrayType$; import org.apache.spark.sql.types.BinaryType$; @@ -150,14 +151,14 @@ public DataType primitive(Type.PrimitiveType primitive) { return BinaryType$.MODULE$; case GEOMETRY: Types.GeometryType geometry = (Types.GeometryType) primitive; - return GeometryType$.MODULE$.apply(geometry.crs()); + return geometryType(geometry.crs()); case GEOGRAPHY: Types.GeographyType geography = (Types.GeographyType) primitive; if (geography.algorithm() != EdgeAlgorithm.SPHERICAL) { throw new UnsupportedOperationException( "Spark does not support geography edge algorithm: " + geography.algorithm()); } - return GeographyType$.MODULE$.apply(geography.crs()); + return geographyType(geography.crs()); case DECIMAL: Types.DecimalType decimal = (Types.DecimalType) primitive; return DecimalType$.MODULE$.apply(decimal.precision(), decimal.scale()); @@ -169,6 +170,24 @@ public DataType primitive(Type.PrimitiveType primitive) { } } + private DataType geometryType(String crs) { + // Iceberg allows any non-empty CRS, but Spark only recognizes a fixed set and throws an opaque + // ST_INVALID_CRS_VALUE for the rest; surface it as a clear unsupported-CRS error instead. + try { + return GeometryType$.MODULE$.apply(crs); + } catch (SparkIllegalArgumentException e) { + throw new UnsupportedOperationException("Spark does not support geometry CRS: " + crs, e); + } + } + + private DataType geographyType(String crs) { + try { + return GeographyType$.MODULE$.apply(crs); + } catch (SparkIllegalArgumentException e) { + throw new UnsupportedOperationException("Spark does not support geography CRS: " + crs, e); + } + } + private Metadata fieldMetadata(int fieldId) { if (MetadataColumns.metadataFieldIds().contains(fieldId)) { return new MetadataBuilder().putBoolean(METADATA_COL_ATTR_KEY, true).build(); diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index b4b6944fdacd..097942f5f8b9 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -135,6 +135,36 @@ public void testGeospatialTypeConversion() { .hasMessage("Spark does not support geography edge algorithm: vincenty"); } + @Test + public void testGeospatialCrsUnsupportedBySparkIsRejected() { + // Iceberg permits any non-empty CRS, but Spark only recognizes a fixed set; a CRS Spark cannot + // resolve must fail with a clear Iceberg error rather than an opaque Spark exception. + Types.GeometryType geometry = Types.GeometryType.of("EPSG:4269"); + assertThatThrownBy(() -> SparkSchemaUtil.convert(geometry)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Spark does not support geometry CRS: EPSG:4269"); + + Types.GeographyType geography = Types.GeographyType.of("EPSG:4269"); + assertThatThrownBy(() -> SparkSchemaUtil.convert(geography)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Spark does not support geography CRS: EPSG:4269"); + } + + @Test + public void testGeospatialMixedSridIsRejected() { + // Spark models a mixed-SRID column with a sentinel CRS ("SRID:ANY"); Iceberg requires a + // concrete column-level CRS, so a mixed-SRID Spark type must be rejected, not persisted. + DataType mixedGeometry = GeometryType$.MODULE$.apply("ANY"); + assertThatThrownBy(() -> SparkSchemaUtil.convert(mixedGeometry)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot convert Spark geometry with mixed SRID to Iceberg"); + + DataType mixedGeography = GeographyType$.MODULE$.apply("ANY"); + assertThatThrownBy(() -> SparkSchemaUtil.convert(mixedGeography)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot convert Spark geography with mixed SRID to Iceberg"); + } + @Test public void testPruneGeospatialTypes() { Schema schema = From 6632e21b85e499c4390b2458070a1d057f6cad4c Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Mon, 29 Jun 2026 13:32:47 -0700 Subject: [PATCH 6/9] Spark 4.1: Translate edge algorithm to Spark's type explicitly Replace the inline algorithm != SPHERICAL guard with an explicit EdgeAlgorithm -> EdgeInterpolationAlgorithm translation, so the supported set (only spherical) is self-documenting, and route both geo types through the conversion helpers. --- .../apache/iceberg/spark/SparkTypeToType.java | 9 +-- .../apache/iceberg/spark/TypeToSparkType.java | 58 +++++++++++++------ .../iceberg/spark/TestSparkSchemaUtil.java | 8 +++ 3 files changed, 53 insertions(+), 22 deletions(-) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java index 833ebcda4a23..3301fecf06e5 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java @@ -20,6 +20,7 @@ import java.util.List; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.EdgeAlgorithm; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.spark.sql.types.ArrayType; @@ -177,10 +178,10 @@ public Type atomic(DataType atomic) { throw new UnsupportedOperationException( "Cannot convert Spark geography with mixed SRID to Iceberg"); } - // Spark only supports the spherical edge-interpolation algorithm, which matches the Iceberg - // default, so the Spark algorithm is intentionally not propagated here. Revisit if Spark - // starts supporting additional algorithms. - return Types.GeographyType.of(geography.crs()); + // Propagate the edge algorithm, translating by name since Iceberg and Spark share algorithm + // names. + return Types.GeographyType.of( + geography.crs(), EdgeAlgorithm.fromName(geography.algorithm().toString())); } else if (atomic instanceof NullType) { return Types.UnknownType.get(); } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java index b6fe1a9e2201..a42792b8674e 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java @@ -19,6 +19,7 @@ package org.apache.iceberg.spark; import java.util.List; +import java.util.function.Supplier; import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.Schema; import org.apache.iceberg.relocated.com.google.common.collect.Lists; @@ -35,6 +36,8 @@ import org.apache.spark.sql.types.DateType$; import org.apache.spark.sql.types.DecimalType$; import org.apache.spark.sql.types.DoubleType$; +import org.apache.spark.sql.types.EdgeInterpolationAlgorithm; +import org.apache.spark.sql.types.EdgeInterpolationAlgorithm$; import org.apache.spark.sql.types.FloatType$; import org.apache.spark.sql.types.GeographyType$; import org.apache.spark.sql.types.GeometryType$; @@ -56,6 +59,11 @@ class TypeToSparkType extends TypeUtil.SchemaVisitor { public static final String METADATA_COL_ATTR_KEY = "__metadata_col"; + // Spark's only edge-interpolation algorithm. Spark exposes no Java-accessible constant for it, so + // it is resolved once by name through the companion's case-insensitive parser. + private static final EdgeInterpolationAlgorithm SPARK_SPHERICAL = + EdgeInterpolationAlgorithm$.MODULE$.fromString("SPHERICAL").get(); + @Override public DataType schema(Schema schema, DataType structType) { return structType; @@ -150,15 +158,9 @@ public DataType primitive(Type.PrimitiveType primitive) { case BINARY: return BinaryType$.MODULE$; case GEOMETRY: - Types.GeometryType geometry = (Types.GeometryType) primitive; - return geometryType(geometry.crs()); + return geometryType((Types.GeometryType) primitive); case GEOGRAPHY: - Types.GeographyType geography = (Types.GeographyType) primitive; - if (geography.algorithm() != EdgeAlgorithm.SPHERICAL) { - throw new UnsupportedOperationException( - "Spark does not support geography edge algorithm: " + geography.algorithm()); - } - return geographyType(geography.crs()); + return geographyType((Types.GeographyType) primitive); case DECIMAL: Types.DecimalType decimal = (Types.DecimalType) primitive; return DecimalType$.MODULE$.apply(decimal.precision(), decimal.scale()); @@ -170,21 +172,41 @@ public DataType primitive(Type.PrimitiveType primitive) { } } - private DataType geometryType(String crs) { - // Iceberg allows any non-empty CRS, but Spark only recognizes a fixed set and throws an opaque - // ST_INVALID_CRS_VALUE for the rest; surface it as a clear unsupported-CRS error instead. + private DataType geometryType(Types.GeometryType geometry) { + // The spec lets a geometry CRS be any string identifying a CRS, but Spark recognizes only a + // fixed set; a CRS Spark cannot resolve becomes a clear unsupported-CRS error below. + String crs = geometry.crs(); + return convertAndValidate("geometry", crs, () -> GeometryType$.MODULE$.apply(crs)); + } + + private DataType geographyType(Types.GeographyType geography) { + // The spec requires a geography CRS to be geographic; Spark recognizes only OGC:CRS84, so any + // other CRS becomes a clear unsupported-CRS error below. + EdgeInterpolationAlgorithm algorithm = convertAlgorithm(geography.algorithm()); + String crs = geography.crs(); + return convertAndValidate("geography", crs, () -> GeographyType$.MODULE$.apply(crs, algorithm)); + } + + // Builds a Spark geo type, translating Spark's opaque ST_INVALID_CRS_VALUE for an unrecognized + // CRS into a clear Iceberg error. + private static DataType convertAndValidate( + String kind, String crs, Supplier conversion) { try { - return GeometryType$.MODULE$.apply(crs); + return conversion.get(); } catch (SparkIllegalArgumentException e) { - throw new UnsupportedOperationException("Spark does not support geometry CRS: " + crs, e); + throw new UnsupportedOperationException("Spark does not support " + kind + " CRS: " + crs, e); } } - private DataType geographyType(String crs) { - try { - return GeographyType$.MODULE$.apply(crs); - } catch (SparkIllegalArgumentException e) { - throw new UnsupportedOperationException("Spark does not support geography CRS: " + crs, e); + // Translates Iceberg's edge-interpolation algorithm to Spark's. Spark supports only the spherical + // algorithm (the Iceberg default); every other algorithm is rejected with a clear error. + private static EdgeInterpolationAlgorithm convertAlgorithm(EdgeAlgorithm algorithm) { + switch (algorithm) { + case SPHERICAL: + return SPARK_SPHERICAL; + default: + throw new UnsupportedOperationException( + "Spark does not support geography edge algorithm: " + algorithm); } } diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index 097942f5f8b9..591897c8450a 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -111,6 +111,14 @@ public void testGeospatialTypeConversion() { .isEqualTo(Types.GeometryType.DEFAULT_CRS); assertThat(SparkSchemaUtil.convert(sparkDefaultGeometry)).isEqualTo(defaultGeometry); + // a default-CRS geography round-trips through the null <-> OGC:CRS84 normalization + Types.GeographyType defaultGeography = Types.GeographyType.crs84(); + DataType sparkDefaultGeography = SparkSchemaUtil.convert(defaultGeography); + assertThat(sparkDefaultGeography).isInstanceOf(GeographyType.class); + assertThat(((GeographyType) sparkDefaultGeography).crs()) + .isEqualTo(Types.GeographyType.DEFAULT_CRS); + assertThat(SparkSchemaUtil.convert(sparkDefaultGeography)).isEqualTo(defaultGeography); + Types.GeometryType geometry = Types.GeometryType.of("EPSG:3857"); DataType sparkGeometry = SparkSchemaUtil.convert(geometry); assertThat(sparkGeometry).isInstanceOf(GeometryType.class); From f4ae297ebcab9ac2d9967c46ccc651f621243888 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Thu, 2 Jul 2026 14:35:11 -0700 Subject: [PATCH 7/9] Spark 4.1: Validate geo CRS and edge algorithm during column pruning Mirror the DECIMAL parameter check in PruneColumnsWithoutReordering for geometry and geography: reject a requested Spark geo type whose CRS (both types) or edge algorithm (geography) disagrees with the table's, instead of letting the mismatch pass pruning silently. Spark normally copies the type from the table schema so this is defensive, but it matches how other parameterized primitives are validated. Algorithm compares across type systems (Iceberg EdgeAlgorithm vs Spark EdgeInterpolationAlgorithm), so reuse TypeToSparkType.convertAlgorithm to translate the table's algorithm into Spark's type before comparing; make that helper package-private and add a symmetric reverse helper in SparkTypeToType so both directions fail loudly on an algorithm the other side lacks. Also harden the SPARK_SPHERICAL lookup to fail fast with a clear message if Spark renames the value. --- .../spark/PruneColumnsWithoutReordering.java | 28 +++++++++++++++++++ .../apache/iceberg/spark/SparkTypeToType.java | 21 +++++++++++--- .../apache/iceberg/spark/TypeToSparkType.java | 16 ++++++++--- .../iceberg/spark/TestSparkSchemaUtil.java | 16 +++++++++++ 4 files changed, 73 insertions(+), 8 deletions(-) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java index 003ba218cc85..90fa68594ade 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java @@ -37,6 +37,7 @@ import org.apache.spark.sql.types.DateType$; import org.apache.spark.sql.types.DecimalType; import org.apache.spark.sql.types.DoubleType$; +import org.apache.spark.sql.types.EdgeInterpolationAlgorithm; import org.apache.spark.sql.types.FloatType$; import org.apache.spark.sql.types.GeographyType; import org.apache.spark.sql.types.GeometryType; @@ -226,6 +227,33 @@ public Type primitive(Type.PrimitiveType primitive) { requestedDecimal.precision(), decimal.precision()); break; + case GEOMETRY: + Types.GeometryType geometry = (Types.GeometryType) primitive; + GeometryType requestedGeometry = (GeometryType) current; + Preconditions.checkArgument( + geometry.crs().equalsIgnoreCase(requestedGeometry.crs()), + "Cannot project geometry with incompatible CRS: %s != %s", + requestedGeometry.crs(), + geometry.crs()); + break; + case GEOGRAPHY: + Types.GeographyType geography = (Types.GeographyType) primitive; + GeographyType requestedGeography = (GeographyType) current; + Preconditions.checkArgument( + geography.crs().equalsIgnoreCase(requestedGeography.crs()), + "Cannot project geography with incompatible CRS: %s != %s", + requestedGeography.crs(), + geography.crs()); + // algorithm() is EdgeAlgorithm on Iceberg and EdgeInterpolationAlgorithm on Spark, so + // translate the table's algorithm into Spark's type and compare within one type system. + EdgeInterpolationAlgorithm tableAlgorithm = + TypeToSparkType.convertAlgorithm(geography.algorithm()); + Preconditions.checkArgument( + tableAlgorithm == requestedGeography.algorithm(), + "Cannot project geography with incompatible edge algorithm: %s != %s", + requestedGeography.algorithm(), + tableAlgorithm); + break; default: } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java index 3301fecf06e5..d45b17410690 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java @@ -19,6 +19,7 @@ package org.apache.iceberg.spark; import java.util.List; +import java.util.Locale; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.EdgeAlgorithm; import org.apache.iceberg.types.Type; @@ -32,6 +33,7 @@ import org.apache.spark.sql.types.DateType; import org.apache.spark.sql.types.DecimalType; import org.apache.spark.sql.types.DoubleType; +import org.apache.spark.sql.types.EdgeInterpolationAlgorithm; import org.apache.spark.sql.types.FloatType; import org.apache.spark.sql.types.GeographyType; import org.apache.spark.sql.types.GeometryType; @@ -178,14 +180,25 @@ public Type atomic(DataType atomic) { throw new UnsupportedOperationException( "Cannot convert Spark geography with mixed SRID to Iceberg"); } - // Propagate the edge algorithm, translating by name since Iceberg and Spark share algorithm - // names. - return Types.GeographyType.of( - geography.crs(), EdgeAlgorithm.fromName(geography.algorithm().toString())); + return Types.GeographyType.of(geography.crs(), convertAlgorithm(geography.algorithm())); } else if (atomic instanceof NullType) { return Types.UnknownType.get(); } throw new UnsupportedOperationException("Not a supported type: " + atomic.catalogString()); } + + // Translates Spark's edge-interpolation algorithm to Iceberg's, mirroring the forward direction in + // TypeToSparkType#convertAlgorithm. Spark supports only the spherical algorithm today; anything + // else is rejected loudly rather than silently defaulting, so a new Spark algorithm surfaces here + // instead of being dropped. + private static EdgeAlgorithm convertAlgorithm(EdgeInterpolationAlgorithm algorithm) { + switch (algorithm.toString().toUpperCase(Locale.ROOT)) { + case "SPHERICAL": + return EdgeAlgorithm.SPHERICAL; + default: + throw new UnsupportedOperationException( + "Iceberg does not support Spark geography edge algorithm: " + algorithm); + } + } } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java index a42792b8674e..1ea3d1dca020 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java @@ -60,9 +60,16 @@ class TypeToSparkType extends TypeUtil.SchemaVisitor { public static final String METADATA_COL_ATTR_KEY = "__metadata_col"; // Spark's only edge-interpolation algorithm. Spark exposes no Java-accessible constant for it, so - // it is resolved once by name through the companion's case-insensitive parser. + // it is resolved once by name through the companion's case-insensitive parser. If Spark ever + // renames this value the lookup fails fast at class-load time with a clear message. private static final EdgeInterpolationAlgorithm SPARK_SPHERICAL = - EdgeInterpolationAlgorithm$.MODULE$.fromString("SPHERICAL").get(); + EdgeInterpolationAlgorithm$.MODULE$ + .fromString("SPHERICAL") + .getOrElse( + () -> { + throw new IllegalStateException( + "Spark EdgeInterpolationAlgorithm SPHERICAL not found"); + }); @Override public DataType schema(Schema schema, DataType structType) { @@ -199,8 +206,9 @@ private static DataType convertAndValidate( } // Translates Iceberg's edge-interpolation algorithm to Spark's. Spark supports only the spherical - // algorithm (the Iceberg default); every other algorithm is rejected with a clear error. - private static EdgeInterpolationAlgorithm convertAlgorithm(EdgeAlgorithm algorithm) { + // algorithm (the Iceberg default); every other algorithm is rejected with a clear error. Shared + // with PruneColumnsWithoutReordering, which compares algorithms across the two type systems. + static EdgeInterpolationAlgorithm convertAlgorithm(EdgeAlgorithm algorithm) { switch (algorithm) { case SPHERICAL: return SPARK_SPHERICAL; diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index 591897c8450a..4bfaaf113e8b 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -200,6 +200,22 @@ public void testPruneGeospatialTypeWithIncompatibleRequestedType() { .hasMessageContaining("incompatible type"); } + @Test + public void testPruneGeospatialTypeWithIncompatibleCrs() { + // Spark normally copies the type from the table schema, so this defends against a requested geo + // type whose CRS disagrees with the table's rather than a case reachable through normal reads. + // Both CRS values must be ones Spark recognizes (else construction fails first); OGC:CRS84 and + // EPSG:3857 are both valid and differ, exercising the prune-time CRS check. + Schema schema = new Schema(optional(1, "geom", Types.GeometryType.of("EPSG:3857"))); + + StructType mismatchedCrs = + new StructType().add("geom", GeometryType$.MODULE$.apply("OGC:CRS84"), true); + + assertThatThrownBy(() -> SparkSchemaUtil.prune(schema, mismatchedCrs)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Cannot project geometry with incompatible CRS"); + } + @Test public void testSchemaConversionWithOnlyWriteDefault() { Schema schema = From f5e24be50b4af98d8c5e38cd439b9f2133589e77 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Thu, 2 Jul 2026 14:46:12 -0700 Subject: [PATCH 8/9] Spark 4.1: Fix comment line length in SparkTypeToType --- .../src/main/java/org/apache/iceberg/spark/SparkTypeToType.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java index d45b17410690..6ce085a1b7a1 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java @@ -188,7 +188,7 @@ public Type atomic(DataType atomic) { throw new UnsupportedOperationException("Not a supported type: " + atomic.catalogString()); } - // Translates Spark's edge-interpolation algorithm to Iceberg's, mirroring the forward direction in + // Translates Spark's edge-interpolation algorithm to Iceberg's, mirroring // TypeToSparkType#convertAlgorithm. Spark supports only the spherical algorithm today; anything // else is rejected loudly rather than silently defaulting, so a new Spark algorithm surfaces here // instead of being dropped. From 78941d58579b6672901c25a5bff55603231d9754 Mon Sep 17 00:00:00 2001 From: Xin Huang Date: Thu, 2 Jul 2026 15:22:10 -0700 Subject: [PATCH 9/9] Spark 4.1: Simplify geo CRS handling and use Spark's SPHERICAL constant - Use EdgeInterpolationAlgorithm.SPHERICAL$.MODULE$ directly instead of resolving it by name via fromString(...).getOrElse(...). Spark exposes the case object as a Java-accessible constant (same pattern as BooleanType$ etc.), so the name lookup and its fallback are unnecessary. - Drop the convertAndValidate wrapper that caught SparkIllegalArgumentException and re-threw UnsupportedOperationException. SparkIllegalArgumentException is already an IllegalArgumentException carrying a clear ST_INVALID_CRS_VALUE message, and catching an unchecked type not on apply()'s signature coupled us to a Spark internal. Let it propagate; the error can be refined later if needed. --- .../apache/iceberg/spark/TypeToSparkType.java | 40 ++++--------------- .../iceberg/spark/TestSparkSchemaUtil.java | 10 ++--- 2 files changed, 13 insertions(+), 37 deletions(-) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java index 1ea3d1dca020..dc077937577c 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java @@ -19,7 +19,6 @@ package org.apache.iceberg.spark; import java.util.List; -import java.util.function.Supplier; import org.apache.iceberg.MetadataColumns; import org.apache.iceberg.Schema; import org.apache.iceberg.relocated.com.google.common.collect.Lists; @@ -27,7 +26,6 @@ import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; -import org.apache.spark.SparkIllegalArgumentException; import org.apache.spark.sql.catalyst.expressions.Literal$; import org.apache.spark.sql.types.ArrayType$; import org.apache.spark.sql.types.BinaryType$; @@ -37,7 +35,7 @@ import org.apache.spark.sql.types.DecimalType$; import org.apache.spark.sql.types.DoubleType$; import org.apache.spark.sql.types.EdgeInterpolationAlgorithm; -import org.apache.spark.sql.types.EdgeInterpolationAlgorithm$; +import org.apache.spark.sql.types.EdgeInterpolationAlgorithm.SPHERICAL$; import org.apache.spark.sql.types.FloatType$; import org.apache.spark.sql.types.GeographyType$; import org.apache.spark.sql.types.GeometryType$; @@ -59,17 +57,8 @@ class TypeToSparkType extends TypeUtil.SchemaVisitor { public static final String METADATA_COL_ATTR_KEY = "__metadata_col"; - // Spark's only edge-interpolation algorithm. Spark exposes no Java-accessible constant for it, so - // it is resolved once by name through the companion's case-insensitive parser. If Spark ever - // renames this value the lookup fails fast at class-load time with a clear message. - private static final EdgeInterpolationAlgorithm SPARK_SPHERICAL = - EdgeInterpolationAlgorithm$.MODULE$ - .fromString("SPHERICAL") - .getOrElse( - () -> { - throw new IllegalStateException( - "Spark EdgeInterpolationAlgorithm SPHERICAL not found"); - }); + // Spark's only edge-interpolation algorithm. + private static final EdgeInterpolationAlgorithm SPARK_SPHERICAL = SPHERICAL$.MODULE$; @Override public DataType schema(Schema schema, DataType structType) { @@ -181,28 +170,15 @@ public DataType primitive(Type.PrimitiveType primitive) { private DataType geometryType(Types.GeometryType geometry) { // The spec lets a geometry CRS be any string identifying a CRS, but Spark recognizes only a - // fixed set; a CRS Spark cannot resolve becomes a clear unsupported-CRS error below. - String crs = geometry.crs(); - return convertAndValidate("geometry", crs, () -> GeometryType$.MODULE$.apply(crs)); + // fixed set; a CRS Spark cannot resolve throws SparkIllegalArgumentException (an + // IllegalArgumentException). + return GeometryType$.MODULE$.apply(geometry.crs()); } private DataType geographyType(Types.GeographyType geography) { // The spec requires a geography CRS to be geographic; Spark recognizes only OGC:CRS84, so any - // other CRS becomes a clear unsupported-CRS error below. - EdgeInterpolationAlgorithm algorithm = convertAlgorithm(geography.algorithm()); - String crs = geography.crs(); - return convertAndValidate("geography", crs, () -> GeographyType$.MODULE$.apply(crs, algorithm)); - } - - // Builds a Spark geo type, translating Spark's opaque ST_INVALID_CRS_VALUE for an unrecognized - // CRS into a clear Iceberg error. - private static DataType convertAndValidate( - String kind, String crs, Supplier conversion) { - try { - return conversion.get(); - } catch (SparkIllegalArgumentException e) { - throw new UnsupportedOperationException("Spark does not support " + kind + " CRS: " + crs, e); - } + // other CRS throws SparkIllegalArgumentException (an IllegalArgumentException). + return GeographyType$.MODULE$.apply(geography.crs(), convertAlgorithm(geography.algorithm())); } // Translates Iceberg's edge-interpolation algorithm to Spark's. Spark supports only the spherical diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java index 4bfaaf113e8b..f107167a7a24 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java @@ -146,16 +146,16 @@ public void testGeospatialTypeConversion() { @Test public void testGeospatialCrsUnsupportedBySparkIsRejected() { // Iceberg permits any non-empty CRS, but Spark only recognizes a fixed set; a CRS Spark cannot - // resolve must fail with a clear Iceberg error rather than an opaque Spark exception. + // resolve is rejected by Spark's SparkIllegalArgumentException (an IllegalArgumentException). Types.GeometryType geometry = Types.GeometryType.of("EPSG:4269"); assertThatThrownBy(() -> SparkSchemaUtil.convert(geometry)) - .isInstanceOf(UnsupportedOperationException.class) - .hasMessage("Spark does not support geometry CRS: EPSG:4269"); + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("EPSG:4269"); Types.GeographyType geography = Types.GeographyType.of("EPSG:4269"); assertThatThrownBy(() -> SparkSchemaUtil.convert(geography)) - .isInstanceOf(UnsupportedOperationException.class) - .hasMessage("Spark does not support geography CRS: EPSG:4269"); + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("EPSG:4269"); } @Test