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..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,7 +37,10 @@ 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; import org.apache.spark.sql.types.IntegerType$; import org.apache.spark.sql.types.LongType$; import org.apache.spark.sql.types.MapType; @@ -224,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: } @@ -244,6 +274,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..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 @@ -19,7 +19,9 @@ 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; import org.apache.iceberg.types.Types; import org.apache.spark.sql.types.ArrayType; @@ -31,7 +33,10 @@ 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; import org.apache.spark.sql.types.IntegerType; import org.apache.spark.sql.types.LongType; import org.apache.spark.sql.types.MapType; @@ -162,10 +167,38 @@ 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) { + 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"); + } + 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 + // 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 09c89bbba813..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 @@ -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; @@ -33,7 +34,11 @@ 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.SPHERICAL$; 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$; @@ -52,6 +57,9 @@ class TypeToSparkType extends TypeUtil.SchemaVisitor { public static final String METADATA_COL_ATTR_KEY = "__metadata_col"; + // Spark's only edge-interpolation algorithm. + private static final EdgeInterpolationAlgorithm SPARK_SPHERICAL = SPHERICAL$.MODULE$; + @Override public DataType schema(Schema schema, DataType structType) { return structType; @@ -145,6 +153,10 @@ public DataType primitive(Type.PrimitiveType primitive) { return BinaryType$.MODULE$; case BINARY: return BinaryType$.MODULE$; + case GEOMETRY: + return geometryType((Types.GeometryType) primitive); + case GEOGRAPHY: + return geographyType((Types.GeographyType) primitive); case DECIMAL: Types.DecimalType decimal = (Types.DecimalType) primitive; return DecimalType$.MODULE$.apply(decimal.precision(), decimal.scale()); @@ -156,6 +168,32 @@ 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 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 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 + // 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; + default: + throw new UnsupportedOperationException( + "Spark does not support geography edge algorithm: " + algorithm); + } + } + 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 d5f407a715ef..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 @@ -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,12 +30,19 @@ 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; 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.DataTypes; +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 +101,121 @@ 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); + + // 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); + assertThat(((GeometryType) sparkGeometry).crs()).isEqualTo("EPSG:3857"); + assertThat(SparkSchemaUtil.convert(sparkGeometry)).isEqualTo(geometry); + + Types.GeographyType geography = Types.GeographyType.of("OGC:CRS84"); + DataType sparkGeography = SparkSchemaUtil.convert(geography); + assertThat(sparkGeography).isInstanceOf(GeographyType.class); + 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("OGC:CRS84"))) + .isEqualTo(geography); + + Types.GeographyType vincentyGeography = + Types.GeographyType.of("OGC:CRS84", EdgeAlgorithm.VINCENTY); + assertThatThrownBy(() -> SparkSchemaUtil.convert(vincentyGeography)) + .isInstanceOf(UnsupportedOperationException.class) + .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 is rejected by Spark's SparkIllegalArgumentException (an IllegalArgumentException). + Types.GeometryType geometry = Types.GeometryType.of("EPSG:4269"); + assertThatThrownBy(() -> SparkSchemaUtil.convert(geometry)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("EPSG:4269"); + + Types.GeographyType geography = Types.GeographyType.of("EPSG:4269"); + assertThatThrownBy(() -> SparkSchemaUtil.convert(geography)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("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 = + new Schema( + optional(1, "geom", Types.GeometryType.of("EPSG:3857")), + optional(2, "geog", Types.GeographyType.of("OGC:CRS84")), + 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 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 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 =