feat: Add Iceberg v3 geospatial primitive types (geometry and geography)

Implement support for Iceberg v3 geospatial types as specified in the
Iceberg specification:

- Add GeometryType(crs) and GeographyType(crs, algorithm) to types.py
- Default CRS is "OGC:CRS84", default algorithm is "spherical"
- Types require format version 3 (minimum_format_version() returns 3)
- Values are stored as WKB (Well-Known Binary) bytes at runtime
- Avro schema conversion maps to "bytes"
- PyArrow conversion maps to large_binary()
- Add type string parsing for geometry('CRS') and geography('CRS', 'algo')
- Add visitor pattern support in schema.py and resolver.py

Note: JSON single-value encoding (WKB<->WKT) raises NotImplementedError
as it requires external libraries (e.g., Shapely) which are not included
to avoid heavy dependencies.

Author: SaymV

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
@@ -273,6 +287,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
 
@@ -354,6 +370,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
 
@@ -474,6 +492,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.
@@ -593,3 +645,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

@@ -157,6 +157,8 @@
     DoubleType,
     FixedType,
     FloatType,
+    GeographyType,
+    GeometryType,
     IcebergType,
     IntegerType,
     ListType,
@@ -798,6 +800,26 @@ 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 binary.
+
+        Note: PyArrow 21.0.0+ supports native GEOMETRY logical type from Arrow GEO.
+        For now, we use large_binary which stores WKB bytes.
+        Future enhancement: detect PyArrow version and use pa.geometry() when available.
+        """
+        # TODO: When PyArrow 21.0.0+ is available, use pa.geometry() with CRS metadata
+        return pa.large_binary()
+
+    def visit_geography(self, geography_type: GeographyType) -> pa.DataType:
+        """Convert geography type to PyArrow binary.
+
+        Note: PyArrow 21.0.0+ supports native GEOGRAPHY logical type from Arrow GEO.
+        For now, we use large_binary which stores WKB bytes.
+        Future enhancement: detect PyArrow version and use pa.geography() when available.
+        """
+        # TODO: When PyArrow 21.0.0+ is available, use pa.geography() with CRS and algorithm metadata
+        return pa.large_binary()
+
 
 def _convert_scalar(value: Any, iceberg_type: IcebergType) -> pa.scalar:
     if not isinstance(iceberg_type, PrimitiveType):
@@ -2111,6 +2133,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()
 

pyiceberg/schema.py

@@ -43,6 +43,8 @@
     DoubleType,
     FixedType,
     FloatType,
+    GeographyType,
+    GeometryType,
     IcebergType,
     IntegerType,
     ListType,
@@ -553,6 +555,10 @@ def primitive(self, primitive: PrimitiveType, primitive_partner: P | None) -> T:
             return self.visit_binary(primitive, primitive_partner)
         elif isinstance(primitive, UnknownType):
             return self.visit_unknown(primitive, primitive_partner)
+        elif isinstance(primitive, GeometryType):
+            return self.visit_geometry(primitive, primitive_partner)
+        elif isinstance(primitive, GeographyType):
+            return self.visit_geography(primitive, primitive_partner)
         else:
             raise ValueError(f"Type not recognized: {primitive}")
 
@@ -624,6 +630,14 @@ def visit_binary(self, binary_type: BinaryType, partner: P | None) -> T:
     def visit_unknown(self, unknown_type: UnknownType, partner: P | None) -> T:
         """Visit a UnknownType."""
 
+    @abstractmethod
+    def visit_geometry(self, geometry_type: GeometryType, partner: P | None) -> T:
+        """Visit a GeometryType."""
+
+    @abstractmethod
+    def visit_geography(self, geography_type: GeographyType, partner: P | None) -> T:
+        """Visit a GeographyType."""
+
 
 class PartnerAccessor(Generic[P], ABC):
     @abstractmethod
@@ -747,6 +761,10 @@ def primitive(self, primitive: PrimitiveType) -> T:
             return self.visit_binary(primitive)
         elif isinstance(primitive, UnknownType):
             return self.visit_unknown(primitive)
+        elif isinstance(primitive, GeometryType):
+            return self.visit_geometry(primitive)
+        elif isinstance(primitive, GeographyType):
+            return self.visit_geography(primitive)
         else:
             raise ValueError(f"Type not recognized: {primitive}")
 
@@ -818,6 +836,14 @@ def visit_binary(self, binary_type: BinaryType) -> T:
     def visit_unknown(self, unknown_type: UnknownType) -> T:
         """Visit a UnknownType."""
 
+    @abstractmethod
+    def visit_geometry(self, geometry_type: GeometryType) -> T:
+        """Visit a GeometryType."""
+
+    @abstractmethod
+    def visit_geography(self, geography_type: GeographyType) -> T:
+        """Visit a GeographyType."""
+
 
 @dataclass(init=True, eq=True, frozen=True)
 class Accessor: