feat: Add Geometry & Geography Types (#2859)

Closes #1820

# Rationale for this change

Apache Iceberg v3 introduces native `geometry` and `geography` primitive
types.
This PR adds spec-compliant support for those types in PyIceberg,
including:

- New primitive types with parameter handling and format-version
enforcement
- Schema parsing and round-trip serialization
- Avro mapping using WKB bytes
- PyArrow / Parquet integration with version-aware fallbacks
- Unit test coverage for type behavior and compatibility constraints

A full design and scope discussion is available in the accompanying RFC:
📄 **RFC: [Iceberg v3 Geospatial Primitive
Types](#3004
The RFC documents scope, non-goals, compatibility constraints, and known
limitations.

# Are these changes tested?

Yes.

- Unit tests cover:
  - Type creation (default and custom parameters)
  - `__str__` / `__repr__` round-tripping
  - Equality, hashing, and pickling
  - Format version enforcement (v3-only)
- PyArrow-dependent behavior is version-gated and conditionally tested

# Are there any user-facing changes?

Yes.

- Users may now declare `geometry` and `geography` columns in Iceberg v3
schemas
- Parquet files written with PyArrow ≥ 21 preserve GEO logical types
when available
- Limitations (e.g. no WKB↔WKT conversion, no spatial predicates) are
documented

<!-- Changelog label added -->

Author: SaymV

mkdocs/docs/geospatial.md

@@ -0,0 +1,110 @@
+# Geospatial Types
+
+PyIceberg supports Iceberg v3 geospatial primitive types: `geometry` and `geography`.
+
+## Overview
+
+Iceberg v3 introduces native support for spatial data types:
+
+- **`geometry(C)`**: Represents geometric shapes in a coordinate reference system (CRS)
+- **`geography(C, A)`**: Represents geographic shapes with CRS and calculation algorithm
+
+Both types store values as WKB (Well-Known Binary) bytes.
+
+## Requirements
+
+- Iceberg format version 3 or higher
+- Optional: `geoarrow-pyarrow` for GeoArrow extension type metadata and interoperability. Without it, geometry and geography are written as binary in Parquet while the Iceberg schema still preserves the spatial type. Install with `pip install pyiceberg[geoarrow]`.
+
+## Usage
+
+### Declaring Columns
+
+```python
+from pyiceberg.schema import Schema
+from pyiceberg.types import NestedField, GeometryType, GeographyType
+
+# Schema with geometry and geography columns
+schema = Schema(
+    NestedField(1, "id", IntegerType(), required=True),
+    NestedField(2, "location", GeometryType(), required=True),
+    NestedField(3, "boundary", GeographyType(), required=False),
+)
+```
+
+### Type Parameters
+
+#### GeometryType
+
+```python
+# Default CRS (OGC:CRS84)
+GeometryType()
+
+# Custom CRS
+GeometryType("EPSG:4326")
+```
+
+#### GeographyType
+
+```python
+# Default CRS (OGC:CRS84) and algorithm (spherical)
+GeographyType()
+
+# Custom CRS
+GeographyType("EPSG:4326")
+
+# Custom CRS and algorithm
+GeographyType("EPSG:4326", "planar")
+```
+
+### String Type Syntax
+
+Types can also be specified as strings in schema definitions:
+
+```python
+# Using string type names
+NestedField(1, "point", "geometry", required=True)
+NestedField(2, "region", "geography", required=True)
+
+# With parameters
+NestedField(3, "location", "geometry('EPSG:4326')", required=True)
+NestedField(4, "boundary", "geography('EPSG:4326', 'planar')", required=True)
+```
+
+## Data Representation
+
+Values are represented as WKB (Well-Known Binary) bytes at runtime:
+
+```python
+# Example: Point(0, 0) in WKB format
+point_wkb = bytes.fromhex("0101000000000000000000000000000000000000")
+```
+
+## Current Limitations
+
+1. **WKB/WKT Conversion**: Converting between WKB bytes and WKT strings requires external libraries (like Shapely). PyIceberg does not include this conversion to avoid heavy dependencies.
+
+2. **Spatial Predicates**: Spatial filtering (e.g., ST_Contains, ST_Intersects) is not yet supported for query pushdown.
+
+3. **Bounds Metrics**: Geometry/geography columns do not currently contribute to data file bounds metrics.
+
+4. **Without geoarrow-pyarrow**: When the `geoarrow-pyarrow` package is not installed, geometry and geography columns are stored as binary without GeoArrow extension type metadata. The Iceberg schema preserves type information, but other tools reading the Parquet files directly may not recognize them as spatial types. Install with `pip install pyiceberg[geoarrow]` for full GeoArrow support.
+
+## Format Version
+
+Geometry and geography types require Iceberg format version 3:
+
+```python
+from pyiceberg.table import TableProperties
+
+# Creating a v3 table
+table = catalog.create_table(
+    identifier="db.spatial_table",
+    schema=schema,
+    properties={
+        TableProperties.FORMAT_VERSION: "3"
+    }
+)
+```
+
+Attempting to use these types with format version 1 or 2 will raise a validation error.

pyiceberg/avro/resolver.py

@@ -87,6 +87,8 @@
     DoubleType,
     FixedType,
     FloatType,
+    GeographyType,
+    GeometryType,
     IcebergType,
     IntegerType,
     ListType,
@@ -204,6 +206,14 @@ def visit_binary(self, binary_type: BinaryType) -> Writer:
     def visit_unknown(self, unknown_type: UnknownType) -> Writer:
         return UnknownWriter()
 
+    def visit_geometry(self, geometry_type: "GeometryType") -> Writer:
+        """Geometry is written as WKB bytes in Avro."""
+        return BinaryWriter()
+
+    def visit_geography(self, geography_type: "GeographyType") -> Writer:
+        """Geography is written as WKB bytes in Avro."""
+        return BinaryWriter()
+
 
 CONSTRUCT_WRITER_VISITOR = ConstructWriter()
 
@@ -359,6 +369,14 @@ def visit_binary(self, binary_type: BinaryType, partner: IcebergType | None) ->
     def visit_unknown(self, unknown_type: UnknownType, partner: IcebergType | None) -> Writer:
         return UnknownWriter()
 
+    def visit_geometry(self, geometry_type: "GeometryType", partner: IcebergType | None) -> Writer:
+        """Geometry is written as WKB bytes in Avro."""
+        return BinaryWriter()
+
+    def visit_geography(self, geography_type: "GeographyType", partner: IcebergType | None) -> Writer:
+        """Geography is written as WKB bytes in Avro."""
+        return BinaryWriter()
+
 
 class ReadSchemaResolver(PrimitiveWithPartnerVisitor[IcebergType, Reader]):
     __slots__ = ("read_types", "read_enums", "context")
@@ -498,6 +516,14 @@ def visit_binary(self, binary_type: BinaryType, partner: IcebergType | None) ->
     def visit_unknown(self, unknown_type: UnknownType, partner: IcebergType | None) -> Reader:
         return UnknownReader()
 
+    def visit_geometry(self, geometry_type: "GeometryType", partner: IcebergType | None) -> Reader:
+        """Geometry is read as WKB bytes from Avro."""
+        return BinaryReader()
+
+    def visit_geography(self, geography_type: "GeographyType", partner: IcebergType | None) -> Reader:
+        """Geography is read as WKB bytes from Avro."""
+        return BinaryReader()
+
 
 class SchemaPartnerAccessor(PartnerAccessor[IcebergType]):
     def schema_partner(self, partner: IcebergType | None) -> IcebergType | None:

pyiceberg/conversions.py

@@ -49,6 +49,8 @@
     DoubleType,
     FixedType,
     FloatType,
+    GeographyType,
+    GeometryType,
     IntegerType,
     LongType,
     PrimitiveType,
@@ -182,6 +184,18 @@ def _(type_: UnknownType, _: str) -> None:
     return None
 
 
+@partition_to_py.register(GeometryType)
+@partition_to_py.register(GeographyType)
+@handle_none
+def _(_: PrimitiveType, value_str: str) -> bytes:
+    """Convert a geometry/geography partition string to bytes.
+
+    Note: Partition values for geometry/geography types are expected to be
+    hex-encoded WKB (Well-Known Binary) strings.
+    """
+    return bytes.fromhex(value_str)
+
+
 @singledispatch
 def to_bytes(
     primitive_type: PrimitiveType, _: bool | bytes | Decimal | date | datetime | float | int | str | time | uuid.UUID
@@ -274,6 +288,8 @@ def _(_: UUIDType, value: uuid.UUID | bytes) -> bytes:
 
 @to_bytes.register(BinaryType)
 @to_bytes.register(FixedType)
+@to_bytes.register(GeometryType)
+@to_bytes.register(GeographyType)
 def _(_: PrimitiveType, value: bytes) -> bytes:
     return value
 
@@ -355,6 +371,8 @@ def _(_: StringType, b: bytes) -> str:
 @from_bytes.register(BinaryType)
 @from_bytes.register(FixedType)
 @from_bytes.register(UUIDType)
+@from_bytes.register(GeometryType)
+@from_bytes.register(GeographyType)
 def _(_: PrimitiveType, b: bytes) -> bytes:
     return b
 
@@ -475,6 +493,40 @@ def _(_: UUIDType, val: uuid.UUID) -> str:
     return str(val)
 
 
+@to_json.register(GeometryType)
+def _(_: GeometryType, val: bytes) -> str:
+    """Serialize geometry to WKT string per Iceberg spec.
+
+    Note: This requires WKB to WKT conversion which is not yet implemented.
+    The Iceberg spec requires geometry values to be serialized as WKT strings
+    in JSON, but PyIceberg stores geometry as WKB bytes at runtime.
+
+    Raises:
+        NotImplementedError: WKB to WKT conversion is not yet supported.
+    """
+    raise NotImplementedError(
+        "Geometry JSON serialization requires WKB to WKT conversion, which is not yet implemented. "
+        "See https://iceberg.apache.org/spec/#json-single-value-serialization for spec details."
+    )
+
+
+@to_json.register(GeographyType)
+def _(_: GeographyType, val: bytes) -> str:
+    """Serialize geography to WKT string per Iceberg spec.
+
+    Note: This requires WKB to WKT conversion which is not yet implemented.
+    The Iceberg spec requires geography values to be serialized as WKT strings
+    in JSON, but PyIceberg stores geography as WKB bytes at runtime.
+
+    Raises:
+        NotImplementedError: WKB to WKT conversion is not yet supported.
+    """
+    raise NotImplementedError(
+        "Geography JSON serialization requires WKB to WKT conversion, which is not yet implemented. "
+        "See https://iceberg.apache.org/spec/#json-single-value-serialization for spec details."
+    )
+
+
 @singledispatch  # type: ignore
 def from_json(primitive_type: PrimitiveType, val: Any) -> L:  # type: ignore
     """Convert JSON value types into built-in python values.
@@ -594,3 +646,43 @@ def _(_: UUIDType, val: str | bytes | uuid.UUID) -> uuid.UUID:
         return uuid.UUID(bytes=val)
     else:
         return val
+
+
+@from_json.register(GeometryType)
+def _(_: GeometryType, val: str | bytes) -> bytes:
+    """Convert JSON WKT string into WKB bytes per Iceberg spec.
+
+    Note: This requires WKT to WKB conversion which is not yet implemented.
+    The Iceberg spec requires geometry values to be represented as WKT strings
+    in JSON, but PyIceberg stores geometry as WKB bytes at runtime.
+
+    Raises:
+        NotImplementedError: WKT to WKB conversion is not yet supported.
+    """
+    if isinstance(val, bytes):
+        # Already WKB bytes, return as-is
+        return val
+    raise NotImplementedError(
+        "Geometry JSON deserialization requires WKT to WKB conversion, which is not yet implemented. "
+        "See https://iceberg.apache.org/spec/#json-single-value-serialization for spec details."
+    )
+
+
+@from_json.register(GeographyType)
+def _(_: GeographyType, val: str | bytes) -> bytes:
+    """Convert JSON WKT string into WKB bytes per Iceberg spec.
+
+    Note: This requires WKT to WKB conversion which is not yet implemented.
+    The Iceberg spec requires geography values to be represented as WKT strings
+    in JSON, but PyIceberg stores geography as WKB bytes at runtime.
+
+    Raises:
+        NotImplementedError: WKT to WKB conversion is not yet supported.
+    """
+    if isinstance(val, bytes):
+        # Already WKB bytes, return as-is
+        return val
+    raise NotImplementedError(
+        "Geography JSON deserialization requires WKT to WKB conversion, which is not yet implemented. "
+        "See https://iceberg.apache.org/spec/#json-single-value-serialization for spec details."
+    )

pyiceberg/io/pyarrow.py

@@ -156,6 +156,8 @@
     DoubleType,
     FixedType,
     FloatType,
+    GeographyType,
+    GeometryType,
     IcebergType,
     IntegerType,
     ListType,
@@ -799,6 +801,37 @@ def visit_unknown(self, _: UnknownType) -> pa.DataType:
     def visit_binary(self, _: BinaryType) -> pa.DataType:
         return pa.large_binary()
 
+    def visit_geometry(self, geometry_type: GeometryType) -> pa.DataType:
+        """Convert geometry type to PyArrow type.
+
+        When geoarrow-pyarrow is available, returns a GeoArrow WKB extension type
+        with CRS metadata. Otherwise, falls back to large_binary which stores WKB bytes.
+        """
+        try:
+            import geoarrow.pyarrow as ga
+
+            return ga.wkb().with_crs(geometry_type.crs)
+        except ImportError:
+            return pa.large_binary()
+
+    def visit_geography(self, geography_type: GeographyType) -> pa.DataType:
+        """Convert geography type to PyArrow type.
+
+        When geoarrow-pyarrow is available, returns a GeoArrow WKB extension type
+        with CRS and edge type metadata. Otherwise, falls back to large_binary which stores WKB bytes.
+        """
+        try:
+            import geoarrow.pyarrow as ga
+
+            wkb_type = ga.wkb().with_crs(geography_type.crs)
+            # Map Iceberg algorithm to GeoArrow edge type
+            if geography_type.algorithm == "spherical":
+                wkb_type = wkb_type.with_edge_type(ga.EdgeType.SPHERICAL)
+            # "planar" is the default edge type in GeoArrow, no need to set explicitly
+            return wkb_type
+        except ImportError:
+            return pa.large_binary()
+
 
 def _convert_scalar(value: Any, iceberg_type: IcebergType) -> pa.scalar:
     if not isinstance(iceberg_type, PrimitiveType):
@@ -2130,6 +2163,12 @@ def visit_binary(self, binary_type: BinaryType) -> str:
     def visit_unknown(self, unknown_type: UnknownType) -> str:
         return "UNKNOWN"
 
+    def visit_geometry(self, geometry_type: GeometryType) -> str:
+        return "BYTE_ARRAY"
+
+    def visit_geography(self, geography_type: GeographyType) -> str:
+        return "BYTE_ARRAY"
+
 
 _PRIMITIVE_TO_PHYSICAL_TYPE_VISITOR = PrimitiveToPhysicalType()