Skip to content

Spark 4.1: Map geo Spark types #16851

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
szehon-ho merged 9 commits into from
Jul 3, 2026
Merged

Spark 4.1: Map geo Spark types #16851

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
b9e112b
Spark 4.1: Map geo Spark types
Jun 17, 2026
129153d
Spark 4.1: Reject non-spherical geography mapping
Jun 18, 2026
582c759
Spark 4.1: Use OGC:CRS84 for geography type conversion tests
Jun 24, 2026
55af6ce
Spark 4.1: Cover default-CRS geo round-trip and incompatible prune pr…
Jun 25, 2026
5b01562
Spark 4.1: Reject unsupported CRS and mixed SRID at the Spark boundary
Jun 29, 2026
6632e21
Spark 4.1: Translate edge algorithm to Spark's type explicitly
Jun 29, 2026
f4ae297
Spark 4.1: Validate geo CRS and edge algorithm during column pruning
Jul 2, 2026
f5e24be
Spark 4.1: Fix comment line length in SparkTypeToType
Jul 2, 2026
78941d5
Spark 4.1: Simplify geo CRS handling and use Spark's SPHERICAL constant
Jul 2, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 32 additions & 0 deletions spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/PruneColumnsWithoutReordering.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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:
}

Expand All @@ -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))

@szehon-ho szehon-ho Jun 17, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unlike DECIMAL, geo types get no additional CRS/SRID compatibility check in primitive(). Since primitive() returns the table's own type there's no corruption, but a requested geo column with a mismatched CRS passes pruning silently. Is deferring SRID validation intentional (consistent with UUID->String), or worth a follow-up?

@huan233usc huan233usc Jun 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point. I am leaving CRS/SRID validation out of this split for now: pruning only validates the requested Spark type class and returns the table Iceberg type, so it cannot change the table CRS. The CRS/SRID compatibility check belongs with the read/write path where Spark values are materialized.

@szehon-ho szehon-ho Jul 2, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Revisiting this — I think it's worth adding CRS/algorithm checks here now, mirroring DECIMAL. Unlikely in normal reads (Spark copies types from the table schema), but cheap defensive validation.

Example for primitive():

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",
      geometry.crs(),
      requestedGeometry.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",
      geography.crs(),
      requestedGeography.crs());
  Preconditions.checkArgument(
      geography.algorithm() == requestedGeography.algorithm(),
      "Cannot project geography with incompatible edge algorithm: %s != %s",
      geography.algorithm(),
      requestedGeography.algorithm());
  break;

And a test, e.g. table geometry(EPSG:3857) with requested GeometryType(EPSG:4326) → Cannot project geometry with incompatible CRS.

@huan233usc huan233usc Jul 2, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done, added the CRS/algorithm checks plus a test. The algorithm compare is cross-type (Iceberg vs Spark enum), so I reused convertAlgorithm to translate first instead of ==. Thanks!

Comment thread
huan233usc marked this conversation as resolved.
.put(TypeID.UNKNOWN, ImmutableSet.of(NullType$.class))
.buildOrThrow();
}
33 changes: 33 additions & 0 deletions spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/SparkTypeToType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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);
}
}
}
38 changes: 38 additions & 0 deletions spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/TypeToSparkType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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$;
Expand All @@ -52,6 +57,9 @@ class TypeToSparkType extends TypeUtil.SchemaVisitor<DataType> {

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;
Expand Down Expand Up @@ -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());
Expand All @@ -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();
Expand Down
123 changes: 123 additions & 0 deletions spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/TestSparkSchemaUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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");

@szehon-ho szehon-ho Jun 25, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: this only exercises explicit CRS values. Adding a default-CRS geometry round-trip (Types.GeometryType.crs84()) would lock in the null <-> OGC:CRS84 normalization, which is the subtlest part of the mapping.

@huan233usc huan233usc Jun 29, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added above

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);
Comment thread
huan233usc marked this conversation as resolved.
Comment thread
huan233usc marked this conversation as resolved.

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

@szehon-ho szehon-ho Jun 25, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit: this covers only the happy path. A negative case (requesting an incompatible Spark type for a geo column, triggering the Cannot project ... incompatible type precondition) would round out coverage, though it mirrors how other primitives are tested so it's optional.

}

@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 =
Expand Down