[GH-3180] GeoPandas: Implement distributed cx and clip APIs

docs/tutorial/geopandas-api.md

@@ -181,7 +181,8 @@ nyc_buildings.head()
 
 ### Spatial Filtering
 
-Use spatial indexing and filtering methods. Note that `cx` spatial indexing is not yet implemented in the current version:
+Use `cx` for distributed coordinate-based filtering and `clip` when the
+matching geometries should also be cut to the mask:
 
 ```python
 from shapely.geometry import box
@@ -194,14 +195,15 @@ central_park_bbox = box(
     40.789,  # top-right (longitude, latitude)
 )
 
-# Filter buildings within the bounding box using spatial index
-# Note: This requires collecting data to driver for spatial filtering
-# For large datasets, consider using spatial joins instead
-buildings_sample = nyc_buildings.sample(1000)  # Sample for demonstration
-central_park_buildings = buildings_sample[
-    buildings_sample.geometry.intersects(central_park_bbox)
+# Select features that intersect the coordinate bounds
+central_park_buildings = nyc_buildings.cx[
+    -73.973:-73.951,
+    40.764:40.789,
 ]
 
+# Cut the selected geometries to the exact polygon boundary
+central_park_buildings = central_park_buildings.clip(central_park_bbox)
+
 # Display results
 print(
     central_park_buildings[["BUILD_ID", "PROP_ADDR", "height_val", "geometry"]].head()
@@ -296,7 +298,7 @@ buildings_projected["perimeter"] = buildings_projected.geometry.length
 
 ## Supported Operations
 
-The GeoPandas API for Apache Sedona has implemented **39 GeoSeries functions** and **10 GeoDataFrame functions**, covering the most commonly used GeoPandas operations:
+The GeoPandas API for Apache Sedona implements the most commonly used GeoSeries and GeoDataFrame operations:
 
 ### Data I/O
 
@@ -307,6 +309,8 @@ The GeoPandas API for Apache Sedona has implemented **39 GeoSeries functions** a
 ### Spatial Operations
 
 - `sjoin()` - Spatial joins with various predicates
+- `cx` - Coordinate-based spatial filtering
+- `clip()` - Clip geometries with scalar, rectangular, or distributed masks
 - `buffer()` - Geometric buffering
 - `distance()` - Distance calculations
 - `intersects()`, `contains()`, `within()` - Spatial predicates
@@ -333,6 +337,8 @@ The GeoPandas API for Apache Sedona has implemented **39 GeoSeries functions** a
 - `intersects()`, `contains()`, `within()` - Spatial predicates
 - `intersection()` - Geometric intersection
 - `make_valid()` - Geometry validation and repair
+- `cx` - Coordinate-based spatial filtering
+- `clip()` - Distributed geometry clipping
 - `sindex` - Spatial indexing (limited functionality)
 
 ### Data Conversion

docs/tutorial/geopandas-api.zh.md

@@ -181,7 +181,8 @@ nyc_buildings.head()
 
 ### 空间过滤
 
-使用空间索引与过滤方法。注意：当前版本尚未实现 `cx` 空间索引：
+使用 `cx` 进行分布式坐标范围筛选；如需将相交的几何裁剪到掩膜边界，
+可继续使用 `clip`：
 
 ```python
 from shapely.geometry import box
@@ -194,14 +195,15 @@ central_park_bbox = box(
     40.789,  # 右上角（经度，纬度）
 )
 
-# 使用空间索引筛选边界框内的建筑
-# 注意：该写法需要把数据收集到 driver 才能进行空间过滤
-# 对于大规模数据，建议改用空间连接（spatial join）
-buildings_sample = nyc_buildings.sample(1000)  # 演示用：抽样 1000 行
-central_park_buildings = buildings_sample[
-    buildings_sample.geometry.intersects(central_park_bbox)
+# 筛选与坐标范围相交的要素
+central_park_buildings = nyc_buildings.cx[
+    -73.973:-73.951,
+    40.764:40.789,
 ]
 
+# 将筛选后的几何裁剪到精确的多边形边界
+central_park_buildings = central_park_buildings.clip(central_park_bbox)
+
 # 显示结果
 print(
     central_park_buildings[["BUILD_ID", "PROP_ADDR", "height_val", "geometry"]].head()
@@ -296,7 +298,7 @@ buildings_projected["perimeter"] = buildings_projected.geometry.length
 
 ## 已支持的操作
 
-Apache Sedona 的 GeoPandas API 已实现 **39 个 GeoSeries 函数** 与 **10 个 GeoDataFrame 函数**，覆盖了 GeoPandas 中最常用的操作：
+Apache Sedona 的 GeoPandas API 已实现最常用的 GeoSeries 与 GeoDataFrame 操作：
 
 ### 数据 I/O
 
@@ -307,6 +309,8 @@ Apache Sedona 的 GeoPandas API 已实现 **39 个 GeoSeries 函数** 与 **10 
 ### 空间操作
 
 - `sjoin()` —— 多种谓词的空间连接
+- `cx` —— 基于坐标范围的空间筛选
+- `clip()` —— 使用标量、矩形或分布式掩膜裁剪几何
 - `buffer()` —— 几何缓冲
 - `distance()` —— 距离计算
 - `intersects()`、`contains()`、`within()` —— 空间谓词
@@ -333,6 +337,8 @@ Apache Sedona 的 GeoPandas API 已实现 **39 个 GeoSeries 函数** 与 **10 
 - `intersects()`、`contains()`、`within()` —— 空间谓词
 - `intersection()` —— 几何相交
 - `make_valid()` —— 几何校验与修复
+- `cx` —— 基于坐标范围的空间筛选
+- `clip()` —— 分布式几何裁剪
 - `sindex` —— 空间索引（功能有限）
 
 ### 数据转换

python/pyproject.toml

@@ -71,6 +71,7 @@ dev = [
   "pandas>=2.0.0,<3.0.0",
   "numpy<2",
   "geopandas",
+  "rtree; python_version < '3.9'",
   # https://stackoverflow.com/questions/78949093/how-to-resolve-attributeerror-module-fiona-has-no-attribute-path
   # cannot set geopandas>=0.14.4 since it doesn't support python 3.8, so we pin fiona to <1.10.0
   "fiona<1.10.0",

python/sedona/spark/geopandas/__init__.py

@@ -23,7 +23,7 @@
 from sedona.spark.geopandas.geoseries import GeoSeries
 from sedona.spark.geopandas.geodataframe import GeoDataFrame
 
-from sedona.spark.geopandas.tools import sjoin
+from sedona.spark.geopandas.tools import clip, sjoin
 
 from sedona.spark.geopandas.io import read_file, read_parquet
 

python/sedona/spark/geopandas/base.py

@@ -78,6 +78,47 @@ def sindex(self) -> "SpatialIndex":
         """
         return _delegate_to_geometry_column("sindex", self)
 
+    @property
+    def cx(self):
+        """Select geometries that intersect a coordinate bounding box.
+
+        The indexer accepts an x slice followed by a y slice. Slice bounds are
+        inclusive, may be open-ended, and may be written in either order.
+        Non-``None`` slice steps are ignored with a warning, matching
+        GeoPandas.
+
+        Returns
+        -------
+        _CoordinateIndexer
+            Coordinate indexer returning the same distributed type as this
+            object.
+
+        Examples
+        --------
+        >>> from sedona.spark.geopandas import GeoSeries
+        >>> from shapely.geometry import Point
+        >>> s = GeoSeries([Point(0, 0), Point(1, 1), Point(2, 2)])
+        >>> s.cx[0.5:2, 0.5:1.5]
+        1    POINT (1 1)
+        dtype: geometry
+
+        Open-ended slices are supported:
+
+        >>> s.cx[1:, :]
+        1    POINT (1 1)
+        2    POINT (2 2)
+        dtype: geometry
+
+        Notes
+        -----
+        Selection is evaluated with native Spark and Sedona expressions. Open
+        bounds are derived with a one-row distributed bounds aggregation; no
+        geometry rows are collected to the driver.
+        """
+        from sedona.spark.geopandas.tools.clip import _CoordinateIndexer
+
+        return _CoordinateIndexer(self)
+
     @property
     def has_sindex(self):
         """Check the existence of the spatial index without generating it.
@@ -3796,6 +3837,75 @@ def clip_by_rect(self, xmin, ymin, xmax, ymax):
             "clip_by_rect", self, xmin, ymin, xmax, ymax
         )
 
+    def clip(self, mask, keep_geom_type=False, sort=False):
+        """Clip geometries to a polygonal or rectangular mask.
+
+        Distributed ``GeoSeries`` and ``GeoDataFrame`` masks are dissolved on
+        the cluster before clipping. A four-value list-like mask is interpreted
+        as ``(minx, miny, maxx, maxy)``. Both distributed layers should use the
+        same Coordinate Reference System (CRS); a mismatch emits a warning.
+
+        Parameters
+        ----------
+        mask : GeoDataFrame, GeoSeries, Polygon, MultiPolygon, or list-like
+            Polygonal mask, or four rectangle bounds. Distributed masks are
+            dissolved into one geometry using ``ST_Union_Aggr``.
+        keep_geom_type : bool, default False
+            If True, retain only the input geometry family when clipping
+            creates lower-dimensional geometries.
+        sort : bool, default False
+            If True, return matching rows in their original positional order.
+
+        Returns
+        -------
+        GeoDataFrame or GeoSeries
+            Clipped data of the same type as the caller. Index values,
+            non-geometry columns, active geometry, CRS, and SRID are preserved.
+
+        Examples
+        --------
+        >>> from sedona.spark.geopandas import GeoDataFrame
+        >>> from shapely.geometry import LineString, Point, box
+        >>> gdf = GeoDataFrame(
+        ...     {
+        ...         "name": ["line", "point"],
+        ...         "geometry": [
+        ...             LineString([(0, 0), (2, 2)]),
+        ...             Point(3, 3),
+        ...         ],
+        ...     }
+        ... )
+        >>> gdf.clip(box(0, 0, 1, 1))
+           name               geometry
+        0  line  LINESTRING (0 0, 1 1)
+
+        Rectangle bounds can be supplied directly:
+
+        >>> gdf.geometry.clip((0, 0, 1, 1))
+        0    LINESTRING (0 0, 1 1)
+        Name: geometry, dtype: geometry
+
+        See Also
+        --------
+        GeoSeries.clip_by_rect
+        geopandas.clip
+
+        Notes
+        -----
+        Sedona does not expose a separate fast ``ClipByBox`` operation.
+        Four-value rectangle masks therefore use native ``ST_MakeEnvelope``
+        and ``ST_Intersection``. Boundary-only intersections can consequently
+        differ from GeoPandas' fast, possibly dirty rectangle path.
+        """
+        from sedona.spark.geopandas.tools.clip import clip
+
+        return clip(
+            self,
+            mask,
+            keep_geom_type=keep_geom_type,
+            sort=sort,
+        )
+
     def difference(self, other, align=None):
         """Returns a ``GeoSeries`` of the points in each aligned geometry that
         are not in `other`.

python/sedona/spark/geopandas/geodataframe.py

@@ -396,7 +396,11 @@ def _get_geometry(self) -> sgpd.GeoSeries:
                 )
 
             raise MissingGeometryColumnError(msg)
-        return self[self._geometry_column_name]
+        geometry = self[self._geometry_column_name]
+        empty_crs_source = getattr(self, "_empty_crs_source", None)
+        if empty_crs_source is not None:
+            geometry._empty_crs_source = empty_crs_source
+        return geometry
 
     def _set_geometry(self, col):
         # This check is included in the original geopandas. Note that this prevents assigning a str to the property

python/sedona/spark/geopandas/geoseries.py

@@ -343,6 +343,9 @@ def __init__(
         self._col_label: Label
         self._sindex: SpatialIndex = None
         self._empty_crs_source: typing.Optional["GeoSeries"] = None
+        # Explicit CRS metadata wins over the lineage fallback below when
+        # geometry rows are empty or carry SRID 0.
+        self._empty_crs_value = None
 
         if isinstance(
             data, (GeoDataFrame, GeoSeries, PandasOnSparkSeries, PandasOnSparkDataFrame)
@@ -461,6 +464,10 @@ def crs(self) -> Union["CRS", None]:
         from pyproj import CRS
 
         if self._is_empty():
+            # Empty data has no SRID to inspect, so explicit CRS metadata wins
+            # over the inherited lineage metadata.
+            if self._empty_crs_value is not None:
+                return self._empty_crs_value
             if self._empty_crs_source is not None:
                 return self._empty_crs_source.crs
             return None
@@ -482,7 +489,16 @@ def crs(self) -> Union["CRS", None]:
         srid = 0 if np.isnan(srid) else srid
 
         # Sedona returns 0 if SRID doesn't exist.
-        return CRS.from_user_input(srid) if srid != 0 else None
+        if srid != 0:
+            return CRS.from_user_input(srid)
+        # These fallbacks are metadata rather than a fresh read from geometry
+        # coordinates. Explicit metadata takes precedence over inherited
+        # lineage metadata, including for non-empty geometries with SRID 0.
+        if self._empty_crs_value is not None:
+            return self._empty_crs_value
+        if self._empty_crs_source is not None:
+            return self._empty_crs_source.crs
+        return None
 
     @crs.setter
     def crs(self, value: Union["CRS", None]):
@@ -615,9 +631,15 @@ def set_crs(
 
         spark_col = stf.ST_SetSRID(self.spark.column, new_epsg)
         result = self._query_geometry_column(spark_col, keep_name=True)
+        result._empty_crs_value = crs
+        if crs is None:
+            result._empty_crs_source = None
 
         if inplace:
             self._update_inplace(result, invalidate_sindex=False)
+            self._empty_crs_value = crs
+            if crs is None:
+                self._empty_crs_source = None
             return None
 
         return result
@@ -4033,13 +4055,6 @@ def to_arrow(self, geometry_encoding="WKB", interleaved=True, include_z=None):
             include_z=include_z,
         )
 
-    def clip(self, mask, keep_geom_type: bool = False, sort=False) -> "GeoSeries":
-        raise NotImplementedError(
-            _not_implemented_error(
-                "clip", "Clips geometries to the bounds of a mask geometry."
-            )
-        )
-
     def to_file(
         self,
         path: str,

python/sedona/spark/geopandas/tools/__init__.py

@@ -16,7 +16,9 @@
 # under the License.
 
 from .sjoin import sjoin
+from .clip import clip
 
 __all__ = [
+    "clip",
     "sjoin",
 ]