This PR extends PyIceberg geospatial support in three areas:

1. Adds geospatial bounds metric computation from WKB values (geometry + geography).
2. Adds spatial predicate expression/binding support (`st-contains`, `st-intersects`, `st-within`, `st-overlaps`) with conservative evaluator behavior.
3. Improves Arrow/Parquet interoperability for GeoArrow WKB, including explicit handling of geometry vs planar-geography ambiguity at the schema-compatibility boundary.

This increment is compatibility-first and does **not** introduce new runtime dependencies.

Base `geometry`/`geography` types existed, but there were still practical gaps:

- Geospatial columns were not contributing spec-encoded bounds in data-file metrics.
- Spatial predicates were not modeled end-to-end in expression binding/visitor plumbing.
- GeoArrow metadata can be ambiguous for `geometry` vs `geography(..., "planar")`, causing false compatibility failures during import/add-files flows.

- Added pure-Python geospatial utilities in `pyiceberg/utils/geospatial.py`:
  - WKB envelope extraction
  - antimeridian-aware geography envelope merge
  - Iceberg geospatial bound serialization/deserialization
- Added `GeospatialStatsAggregator` and geospatial aggregate helpers in `pyiceberg/io/pyarrow.py`.
- Updated write/import paths to compute geospatial bounds from actual row values (not Parquet binary min/max stats):
  - `write_file(...)`
  - `parquet_file_to_data_file(...)`
- Prevented incorrect partition inference from geospatial envelope bounds.

- Added expression types in `pyiceberg/expressions/__init__.py`:
  - `STContains`, `STIntersects`, `STWithin`, `STOverlaps`
  - bound counterparts and JSON parsing support
- Added visitor dispatch/plumbing in `pyiceberg/expressions/visitors.py`.
- Behavior intentionally conservative in this increment:
  - row-level expression evaluator raises `NotImplementedError`
  - manifest/metrics evaluators return conservative might-match defaults
  - translation paths preserve spatial predicates where possible

- Added GeoArrow WKB decoding helper in `pyiceberg/io/pyarrow.py` to map extension metadata to Iceberg geospatial types.
- Added boundary-only compatibility option in `pyiceberg/schema.py`:
  - `_check_schema_compatible(..., allow_planar_geospatial_equivalence=False)`
- Enabled that option only in `_check_pyarrow_schema_compatible(...)` to allow:
  - `geometry` <-> `geography(..., "planar")` when CRS strings match
  - while still rejecting spherical geography mismatches
- Added one-time warning log when `geoarrow-pyarrow` is unavailable and code falls back to binary.

- Updated user docs: `mkdocs/docs/geospatial.md`
- Added decisions record: `mkdocs/docs/dev/geospatial-types-decisions-v1.md`

Added/updated tests across:

- `tests/utils/test_geospatial.py`
- `tests/io/test_pyarrow_stats.py`
- `tests/io/test_pyarrow.py`
- `tests/expressions/test_spatial_predicates.py`
- `tests/integration/test_geospatial.py`

Coverage includes:

- geospatial bound encoding/decoding (XY/XYZ/XYM/XYZM)
- geography antimeridian behavior
- geospatial metrics generation from write/import paths
- spatial predicate modeling/binding/translation behavior
- planar ambiguity compatibility guardrails
- warning behavior for missing `geoarrow-pyarrow`

- No user-facing API removals.
- New compatibility relaxation is intentionally scoped to Arrow/Parquet schema-compatibility boundary only.
- Core schema/type compatibility remains strict elsewhere.
- No spatial pushdown/row execution implementation in this PR.

- Spatial predicate execution semantics.
- Spatial predicate pushdown/pruning.
- Runtime WKB <-> WKT conversion strategy.

Author: SaymV

mkdocs/docs/geospatial.md

@@ -84,11 +84,11 @@ point_wkb = bytes.fromhex("0101000000000000000000000000000000000000")
 
 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.
+2. **Spatial Predicates Execution**: Spatial predicate APIs (`st-contains`, `st-intersects`, `st-within`, `st-overlaps`) are available in expression trees and binding. Row-level execution and metrics/pushdown evaluation are not implemented yet.
 
-3. **Bounds Metrics**: Geometry/geography columns do not currently contribute to data file bounds metrics.
+3. **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.
 
-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.
+4. **GeoArrow planar ambiguity**: In GeoArrow metadata, `geometry` and `geography(..., 'planar')` can be encoded identically (no explicit edge metadata). PyIceberg resolves this ambiguity at the Arrow/Parquet schema-compatibility boundary by treating them as compatible when CRS matches, while keeping core schema compatibility strict elsewhere.
 
 ## Format Version
 

pyiceberg/expressions/__init__.py

@@ -29,7 +29,7 @@
 from pyiceberg.expressions.literals import AboveMax, BelowMin, Literal, literal
 from pyiceberg.schema import Accessor, Schema
 from pyiceberg.typedef import IcebergBaseModel, IcebergRootModel, L, LiteralValue, StructProtocol
-from pyiceberg.types import DoubleType, FloatType, NestedField
+from pyiceberg.types import DoubleType, FloatType, GeographyType, GeometryType, NestedField
 from pyiceberg.utils.singleton import Singleton
 
 
@@ -48,6 +48,16 @@ def _to_literal(value: L | Literal[L]) -> Literal[L]:
         return literal(value)
 
 
+def _to_bytes(value: bytes | bytearray | memoryview) -> bytes:
+    if isinstance(value, bytes):
+        return value
+    if isinstance(value, bytearray):
+        return bytes(value)
+    if isinstance(value, memoryview):
+        return value.tobytes()
+    raise TypeError(f"Expected bytes-like value, got {type(value)}")
+
+
 class BooleanExpression(IcebergBaseModel, ABC):
     """An expression that evaluates to a boolean."""
 
@@ -109,6 +119,14 @@ def handle_primitive_type(cls, v: Any, handler: ValidatorFunctionWrapHandler) ->
                 return StartsWith(**v)
             elif field_type == "not-starts-with":
                 return NotStartsWith(**v)
+            elif field_type == "st-contains":
+                return STContains(**v)
+            elif field_type == "st-intersects":
+                return STIntersects(**v)
+            elif field_type == "st-within":
+                return STWithin(**v)
+            elif field_type == "st-overlaps":
+                return STOverlaps(**v)
 
             # Set
             elif field_type == "in":
@@ -1106,3 +1124,156 @@ def __invert__(self) -> StartsWith:
     @property
     def as_bound(self) -> type[BoundNotStartsWith]:  # type: ignore
         return BoundNotStartsWith
+
+
+class SpatialPredicate(UnboundPredicate, ABC):
+    type: TypingLiteral["st-contains", "st-intersects", "st-within", "st-overlaps"] = Field(alias="type")
+    term: UnboundTerm
+    value: bytes = Field()
+    model_config = ConfigDict(populate_by_name=True, frozen=True, arbitrary_types_allowed=True)
+
+    def __init__(
+        self,
+        term: str | UnboundTerm,
+        geometry: bytes | bytearray | memoryview | None = None,
+        **kwargs: Any,
+    ) -> None:
+        if geometry is None and "value" in kwargs:
+            geometry = kwargs["value"]
+        if geometry is None:
+            raise TypeError("Spatial predicates require WKB bytes")
+
+        super().__init__(term=_to_unbound_term(term), value=_to_bytes(geometry))
+
+    @property
+    def geometry(self) -> bytes:
+        return self.value
+
+    def bind(self, schema: Schema, case_sensitive: bool = True) -> BoundSpatialPredicate:
+        bound_term = self.term.bind(schema, case_sensitive)
+        if not isinstance(bound_term.ref().field.field_type, (GeometryType, GeographyType)):
+            raise TypeError(
+                f"Spatial predicates can only be bound against geometry/geography fields: {bound_term.ref().field}"
+            )
+        return self.as_bound(bound_term, self.geometry)
+
+    def __eq__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.term == other.term and self.geometry == other.geometry
+        return False
+
+    def __str__(self) -> str:
+        return f"{str(self.__class__.__name__)}(term={repr(self.term)}, geometry={self.geometry!r})"
+
+    def __repr__(self) -> str:
+        return f"{str(self.__class__.__name__)}(term={repr(self.term)}, geometry={self.geometry!r})"
+
+    @property
+    @abstractmethod
+    def as_bound(self) -> type[BoundSpatialPredicate]: ...
+
+
+class BoundSpatialPredicate(BoundPredicate, ABC):
+    value: bytes = Field()
+
+    def __init__(self, term: BoundTerm, geometry: bytes | bytearray | memoryview):
+        super().__init__(term=term, value=_to_bytes(geometry))
+
+    @property
+    def geometry(self) -> bytes:
+        return self.value
+
+    def __eq__(self, other: Any) -> bool:
+        if isinstance(other, self.__class__):
+            return self.term == other.term and self.geometry == other.geometry
+        return False
+
+    def __str__(self) -> str:
+        return f"{self.__class__.__name__}(term={str(self.term)}, geometry={self.geometry!r})"
+
+    def __repr__(self) -> str:
+        return f"{str(self.__class__.__name__)}(term={repr(self.term)}, geometry={self.geometry!r})"
+
+    @property
+    @abstractmethod
+    def as_unbound(self) -> type[SpatialPredicate]: ...
+
+
+class BoundSTContains(BoundSpatialPredicate):
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_unbound(self) -> type[STContains]:
+        return STContains
+
+
+class BoundSTIntersects(BoundSpatialPredicate):
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_unbound(self) -> type[STIntersects]:
+        return STIntersects
+
+
+class BoundSTWithin(BoundSpatialPredicate):
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_unbound(self) -> type[STWithin]:
+        return STWithin
+
+
+class BoundSTOverlaps(BoundSpatialPredicate):
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_unbound(self) -> type[STOverlaps]:
+        return STOverlaps
+
+
+class STContains(SpatialPredicate):
+    type: TypingLiteral["st-contains"] = Field(default="st-contains", alias="type")
+
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_bound(self) -> type[BoundSTContains]:
+        return BoundSTContains
+
+
+class STIntersects(SpatialPredicate):
+    type: TypingLiteral["st-intersects"] = Field(default="st-intersects", alias="type")
+
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_bound(self) -> type[BoundSTIntersects]:
+        return BoundSTIntersects
+
+
+class STWithin(SpatialPredicate):
+    type: TypingLiteral["st-within"] = Field(default="st-within", alias="type")
+
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_bound(self) -> type[BoundSTWithin]:
+        return BoundSTWithin
+
+
+class STOverlaps(SpatialPredicate):
+    type: TypingLiteral["st-overlaps"] = Field(default="st-overlaps", alias="type")
+
+    def __invert__(self) -> BooleanExpression:
+        return Not(child=self)
+
+    @property
+    def as_bound(self) -> type[BoundSTOverlaps]:
+        return BoundSTOverlaps

pyiceberg/expressions/visitors.py

@@ -47,7 +47,12 @@
     BoundNotStartsWith,
     BoundPredicate,
     BoundSetPredicate,
+    BoundSpatialPredicate,
     BoundStartsWith,
+    BoundSTContains,
+    BoundSTIntersects,
+    BoundSTOverlaps,
+    BoundSTWithin,
     BoundTerm,
     BoundUnaryPredicate,
     Not,
@@ -326,6 +331,18 @@ def visit_starts_with(self, term: BoundTerm, literal: LiteralValue) -> T:
     def visit_not_starts_with(self, term: BoundTerm, literal: LiteralValue) -> T:
         """Visit bound NotStartsWith predicate."""
 
+    def visit_st_contains(self, term: BoundTerm, geometry: bytes) -> T:
+        raise NotImplementedError(f"{self.__class__.__name__} does not implement st-contains")
+
+    def visit_st_intersects(self, term: BoundTerm, geometry: bytes) -> T:
+        raise NotImplementedError(f"{self.__class__.__name__} does not implement st-intersects")
+
+    def visit_st_within(self, term: BoundTerm, geometry: bytes) -> T:
+        raise NotImplementedError(f"{self.__class__.__name__} does not implement st-within")
+
+    def visit_st_overlaps(self, term: BoundTerm, geometry: bytes) -> T:
+        raise NotImplementedError(f"{self.__class__.__name__} does not implement st-overlaps")
+
     def visit_unbound_predicate(self, predicate: UnboundPredicate) -> T:
         """Visit an unbound predicate.
 
@@ -421,6 +438,26 @@ def _(expr: BoundNotStartsWith, visitor: BoundBooleanExpressionVisitor[T]) -> T:
     return visitor.visit_not_starts_with(term=expr.term, literal=expr.literal)
 
 
+@visit_bound_predicate.register(BoundSTContains)
+def _(expr: BoundSTContains, visitor: BoundBooleanExpressionVisitor[T]) -> T:
+    return visitor.visit_st_contains(term=expr.term, geometry=expr.geometry)
+
+
+@visit_bound_predicate.register(BoundSTIntersects)
+def _(expr: BoundSTIntersects, visitor: BoundBooleanExpressionVisitor[T]) -> T:
+    return visitor.visit_st_intersects(term=expr.term, geometry=expr.geometry)
+
+
+@visit_bound_predicate.register(BoundSTWithin)
+def _(expr: BoundSTWithin, visitor: BoundBooleanExpressionVisitor[T]) -> T:
+    return visitor.visit_st_within(term=expr.term, geometry=expr.geometry)
+
+
+@visit_bound_predicate.register(BoundSTOverlaps)
+def _(expr: BoundSTOverlaps, visitor: BoundBooleanExpressionVisitor[T]) -> T:
+    return visitor.visit_st_overlaps(term=expr.term, geometry=expr.geometry)
+
+
 def rewrite_not(expr: BooleanExpression) -> BooleanExpression:
     return visit(expr, _RewriteNotVisitor())
 
@@ -514,6 +551,18 @@ def visit_starts_with(self, term: BoundTerm, literal: LiteralValue) -> bool:
     def visit_not_starts_with(self, term: BoundTerm, literal: LiteralValue) -> bool:
         return not self.visit_starts_with(term, literal)
 
+    def visit_st_contains(self, term: BoundTerm, geometry: bytes) -> bool:
+        raise NotImplementedError("st-contains row-level evaluation is not implemented")
+
+    def visit_st_intersects(self, term: BoundTerm, geometry: bytes) -> bool:
+        raise NotImplementedError("st-intersects row-level evaluation is not implemented")
+
+    def visit_st_within(self, term: BoundTerm, geometry: bytes) -> bool:
+        raise NotImplementedError("st-within row-level evaluation is not implemented")
+
+    def visit_st_overlaps(self, term: BoundTerm, geometry: bytes) -> bool:
+        raise NotImplementedError("st-overlaps row-level evaluation is not implemented")
+
     def visit_true(self) -> bool:
         return True
 
@@ -762,6 +811,18 @@ def visit_not_starts_with(self, term: BoundTerm, literal: LiteralValue) -> bool:
 
         return ROWS_MIGHT_MATCH
 
+    def visit_st_contains(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
+    def visit_st_intersects(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
+    def visit_st_within(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
+    def visit_st_overlaps(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
     def visit_true(self) -> bool:
         return ROWS_MIGHT_MATCH
 
@@ -905,6 +966,8 @@ def visit_bound_predicate(self, predicate: BoundPredicate) -> BooleanExpression:
                 pred = predicate.as_unbound(field.name, predicate.literal)
             elif isinstance(predicate, BoundSetPredicate):
                 pred = predicate.as_unbound(field.name, predicate.literals)
+            elif isinstance(predicate, BoundSpatialPredicate):
+                raise NotImplementedError("Spatial predicate translation is not supported when source columns are missing")
             else:
                 raise ValueError(f"Unsupported predicate: {predicate}")
 
@@ -926,6 +989,8 @@ def visit_bound_predicate(self, predicate: BoundPredicate) -> BooleanExpression:
             return predicate.as_unbound(file_column_name, predicate.literal)
         elif isinstance(predicate, BoundSetPredicate):
             return predicate.as_unbound(file_column_name, predicate.literals)
+        elif isinstance(predicate, BoundSpatialPredicate):
+            return predicate.as_unbound(file_column_name, predicate.geometry)
         else:
             raise ValueError(f"Unsupported predicate: {predicate}")
 
@@ -1065,6 +1130,18 @@ def visit_starts_with(self, term: BoundTerm, literal: LiteralValue) -> list[tupl
     def visit_not_starts_with(self, term: BoundTerm, literal: LiteralValue) -> list[tuple[str, str, Any]]:
         return []
 
+    def visit_st_contains(self, term: BoundTerm, geometry: bytes) -> list[tuple[str, str, Any]]:
+        return []
+
+    def visit_st_intersects(self, term: BoundTerm, geometry: bytes) -> list[tuple[str, str, Any]]:
+        return []
+
+    def visit_st_within(self, term: BoundTerm, geometry: bytes) -> list[tuple[str, str, Any]]:
+        return []
+
+    def visit_st_overlaps(self, term: BoundTerm, geometry: bytes) -> list[tuple[str, str, Any]]:
+        return []
+
     def visit_true(self) -> list[tuple[str, str, Any]]:
         return []  # Not supported
 
@@ -1153,6 +1230,18 @@ def _is_nan(self, val: Any) -> bool:
             # In the case of None or other non-numeric types
             return False
 
+    def visit_st_contains(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
+    def visit_st_intersects(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
+    def visit_st_within(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
+    def visit_st_overlaps(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_MATCH
+
 
 class _InclusiveMetricsEvaluator(_MetricsEvaluator):
     struct: StructType
@@ -1739,6 +1828,18 @@ def visit_starts_with(self, term: BoundTerm, literal: LiteralValue) -> bool:
     def visit_not_starts_with(self, term: BoundTerm, literal: LiteralValue) -> bool:
         return ROWS_MIGHT_NOT_MATCH
 
+    def visit_st_contains(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_NOT_MATCH
+
+    def visit_st_intersects(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_NOT_MATCH
+
+    def visit_st_within(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_NOT_MATCH
+
+    def visit_st_overlaps(self, term: BoundTerm, geometry: bytes) -> bool:
+        return ROWS_MIGHT_NOT_MATCH
+
     def _get_field(self, field_id: int) -> NestedField:
         field = self.struct.field(field_id=field_id)
         if field is None: