[GH-3182] GeoPandas: Implement distributed sjoin_nearest

common/src/main/java/org/apache/sedona/common/Predicates.java

@@ -262,4 +262,23 @@ public static boolean knn(
       Geometry leftGeometry, Geometry rightGeometry, int k, boolean useSpheroid) {
     throw new UnsupportedOperationException("KNN predicate is not supported");
   }
+
+  public static boolean knn(
+      Geometry leftGeometry,
+      Geometry rightGeometry,
+      int k,
+      boolean useSpheroid,
+      boolean includeTies) {
+    throw new UnsupportedOperationException("KNN predicate is not supported");
+  }
+
+  public static boolean knn(
+      Geometry leftGeometry,
+      Geometry rightGeometry,
+      int k,
+      boolean useSpheroid,
+      boolean includeTies,
+      boolean exclusive) {
+    throw new UnsupportedOperationException("KNN predicate is not supported");
+  }
 }

docs/api/sql/NearestNeighbourSearching.md

@@ -24,21 +24,43 @@ Sedona supports nearest-neighbour searching on geospatial data by providing a ge
 
 Introduction: join operation to find the k-nearest neighbors of a point or region in a spatial dataset.
 
-Format: `ST_KNN(R: Table, S: Table, k: Integer, use_spheroid: Boolean)`
+Formats:
 
-Where R is the queries side table and S is the object side table, K is the number of neighbors. use_spheroid is a boolean value that determines whether to use the spheroid distance or not.
+```sql
+ST_KNN(query_geometry, object_geometry, k)
+ST_KNN(query_geometry, object_geometry, k, use_spheroid)
+ST_KNN(query_geometry, object_geometry, k, use_spheroid, include_ties)
+ST_KNN(query_geometry, object_geometry, k, use_spheroid, include_ties, exclusive)
+```
+
+`query_geometry` belongs to the query-side table, `object_geometry` belongs to
+the candidate-side table, and `k` is the number of neighbours. `use_spheroid`
+selects spheroid distance instead of planar distance.
+
+The optional `include_ties` argument controls ties for this query. It overrides
+the `spark.sedona.join.knn.includeTieBreakers` session setting when supplied.
+The optional `exclusive` argument excludes candidates that are topologically
+equal to the query geometry before nearest neighbours are ranked. For example,
+the two line strings `LINESTRING (0 0, 1 0)` and
+`LINESTRING (1 0, 0 0)` are equal and are excluded from one another by
+`exclusive = true`.
+
+The three- and four-argument forms retain their existing behavior. To use
+`exclusive`, both query-local Boolean arguments must be supplied.
 
 Queries side table contains geometries that are used to find the k-nearest neighbors in the object side table.
 
 When either queries or objects data contain non-point data (geometries), we take the centroid of each geometry.
 
-In case there are ties in the distance, the result will include all the tied geometries only when the following sedona config is set to true:
+With the three- or four-argument form, tied geometries are included only when
+the following Sedona configuration is set to true:
 
 **Note for Inner Join:**
 
 - The `ST_KNN` join only supports left inner join.
 - It returns only pairs where there is at least one matching neighbor within the k nearest neighbors.
 - If a query point has no valid neighbor (e.g., because k is too large), it is excluded from the result.
+- Null and empty geometries do not participate in a match.
 
 ```
 spark.sedona.join.knn.includeTieBreakers=true

docs/api/sql/NearestNeighbourSearching.zh.md

@@ -24,21 +24,40 @@ Sedona 通过提供地理空间 k 近邻（kNN）连接方法来支持对地理
 
 简介：用于在空间数据集中查找一个点或区域的 k 个最近邻的连接操作。
 
-格式：`ST_KNN(R: Table, S: Table, k: Integer, use_spheroid: Boolean)`
+格式：
 
-其中 R 是查询侧表，S 是对象侧表，K 是邻居数量。use_spheroid 是一个布尔值，决定是否使用椭球距离。
+```sql
+ST_KNN(query_geometry, object_geometry, k)
+ST_KNN(query_geometry, object_geometry, k, use_spheroid)
+ST_KNN(query_geometry, object_geometry, k, use_spheroid, include_ties)
+ST_KNN(query_geometry, object_geometry, k, use_spheroid, include_ties, exclusive)
+```
+
+`query_geometry` 属于查询侧表，`object_geometry` 属于候选对象侧表，`k`
+是邻居数量。`use_spheroid` 用于选择椭球距离而不是平面距离。
+
+可选参数 `include_ties` 仅控制当前查询的并列结果；提供该参数时，它会覆盖
+会话配置 `spark.sedona.join.knn.includeTieBreakers`。可选参数 `exclusive`
+会在近邻排序之前排除与查询几何拓扑相等的候选对象。例如，
+`LINESTRING (0 0, 1 0)` 和 `LINESTRING (1 0, 0 0)` 拓扑相等，因此在
+`exclusive = true` 时会互相排除。
+
+三个和四个参数的形式保持原有行为。若要使用 `exclusive`，必须同时提供两个
+查询级布尔参数。
 
 查询侧表包含用于在对象侧表中查找 k 近邻的几何对象。
 
 当查询数据或对象数据中存在非点几何（其他几何类型）时，会取每个几何对象的质心。
 
-当距离上出现并列时，只有在下面的 sedona 配置被设置为 true 时，结果才会包含所有并列的几何对象：
+使用三个或四个参数的形式时，只有在下面的 Sedona 配置被设置为 true 时，
+结果才会包含所有并列的几何对象：
 
 **关于内连接的说明：**
 
 - `ST_KNN` 连接仅支持 left inner join。
 - 它只返回那些至少存在一个在 k 近邻范围内的匹配邻居的对。
 - 如果某个查询点没有有效邻居（例如 k 设得太大），它会被从结果中排除。
+- 空几何和 null 几何不会参与匹配。
 
 ```
 spark.sedona.join.knn.includeTieBreakers=true

docs/tutorial/geopandas-api.md

@@ -237,6 +237,36 @@ intersects_result = left_df.sjoin(right_df, predicate="intersects")
 contains_result = left_df.sjoin(right_df, predicate="contains")
 ```
 
+### Nearest-Neighbor Joins
+
+`sjoin_nearest()` uses Sedona's distributed KNN join plan, so geometry rows are
+not collected to the driver. It supports `inner`, `left`, and `right` results;
+left and right joins retain unmatched rows from their respective side while
+nearest matching remains distributed.
+
+```python
+nearest = left_df.sjoin_nearest(
+    right_df,
+    how="left",
+    max_distance=500,
+    distance_col="distance",
+    exclusive=True,
+)
+```
+
+Every candidate tied at the nearest distance is returned. `max_distance`
+limits eligible matches, but it is currently applied after nearest candidates
+are found and does not prune the distributed KNN search. `exclusive=True`
+excludes candidates topologically equal to the query geometry before ranking.
+Distances are planar and use CRS units; geographic CRS inputs emit a warning
+because their distance results may be inaccurate.
+
+For inputs with MultiIndex columns or tuple-valued index names, the result uses
+a padded MultiIndex when needed. This is the pandas-on-Spark representation of
+GeoPandas' mixed tuple-and-string object column index, which pandas-on-Spark
+cannot represent directly. The operation rejects suffixing or padding
+combinations that would create duplicate labels.
+
 ### Coordinate Reference System Operations
 
 Transform geometries between different coordinate reference systems:
@@ -296,7 +326,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 +337,7 @@ The GeoPandas API for Apache Sedona has implemented **39 GeoSeries functions** a
 ### Spatial Operations
 
 - `sjoin()` - Spatial joins with various predicates
+- `sjoin_nearest()` - Distributed nearest-neighbor joins
 - `buffer()` - Geometric buffering
 - `distance()` - Distance calculations
 - `intersects()`, `contains()`, `within()` - Spatial predicates

docs/tutorial/geopandas-api.zh.md

@@ -237,6 +237,32 @@ intersects_result = left_df.sjoin(right_df, predicate="intersects")
 contains_result = left_df.sjoin(right_df, predicate="contains")
 ```
 
+### 最近邻连接
+
+`sjoin_nearest()` 使用 Sedona 的分布式 KNN 连接计划，不会把几何数据行收集到
+driver。它支持 `inner`、`left` 和 `right`；left 与 right 连接会保留对应一侧
+未匹配的行，而最近邻匹配本身仍以分布式方式执行。
+
+```python
+nearest = left_df.sjoin_nearest(
+    right_df,
+    how="left",
+    max_distance=500,
+    distance_col="distance",
+    exclusive=True,
+)
+```
+
+所有在最近距离上并列的候选对象都会返回。`max_distance` 用于限制可匹配的
+距离，但目前是在找到最近候选对象后才应用，因此不会缩小分布式 KNN 搜索范围。
+`exclusive=True` 会在排序之前排除与查询几何拓扑相等的候选对象。距离按平面
+计算并使用 CRS 的单位；输入为地理坐标系时会发出警告，因为距离结果可能不准确。
+
+对于使用 MultiIndex 列或元组型索引名称的输入，结果会在必要时使用补齐后的
+MultiIndex。这是 pandas-on-Spark 对 GeoPandas 混合元组与字符串对象列索引的表示
+方式；后者无法由 pandas-on-Spark 直接表达。如果后缀处理或补齐会产生重复标签，
+该操作将拒绝执行。
+
 ### 坐标参考系操作
 
 在不同坐标参考系（CRS）之间转换几何对象：
@@ -296,7 +322,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 +333,7 @@ Apache Sedona 的 GeoPandas API 已实现 **39 个 GeoSeries 函数** 与 **10 
 ### 空间操作
 
 - `sjoin()` —— 多种谓词的空间连接
+- `sjoin_nearest()` —— 分布式最近邻连接
 - `buffer()` —— 几何缓冲
 - `distance()` —— 距离计算
 - `intersects()`、`contains()`、`within()` —— 空间谓词

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 sjoin, sjoin_nearest
 
 from sedona.spark.geopandas.io import read_file, read_parquet
 

python/sedona/spark/geopandas/geodataframe.py

@@ -33,6 +33,7 @@
 
 from sedona.spark.geopandas._typing import Label
 from sedona.spark.geopandas.base import GeoFrame
+from sedona.spark.sql import st_functions as stf
 
 from pandas.api.extensions import register_extension_dtype
 from geopandas.geodataframe import crs_mismatch_error
@@ -338,7 +339,39 @@ def __init__(
                 raise ValueError(crs_mismatch_error)
 
         if geometry:
-            self.set_geometry(geometry, inplace=True, crs=crs)
+            existing_geometry = None
+            if crs is not None and pd.api.types.is_hashable(geometry):
+                try:
+                    candidate = self[geometry]
+                except (KeyError, TypeError, ValueError):
+                    pass
+                else:
+                    if isinstance(candidate, sgpd.GeoSeries):
+                        existing_geometry = candidate
+
+            if existing_geometry is None:
+                self.set_geometry(geometry, inplace=True, crs=crs)
+            else:
+                # Updating the existing Spark column directly avoids index
+                # alignment, which would multiply rows for duplicate indexes.
+                from pyproj import CRS
+
+                normalized_crs = CRS.from_user_input(crs)
+                new_epsg = normalized_crs.to_epsg() or 0
+                self._update_internal_frame(
+                    self._internal.with_new_spark_column(
+                        existing_geometry._column_label,
+                        stf.ST_SetSRID(existing_geometry.spark.column, new_epsg),
+                    )
+                )
+                self._geometry_column_name = geometry
+                empty_crs_source = self[geometry]
+                empty_crs_source._empty_crs_value = normalized_crs
+                object.__setattr__(
+                    self,
+                    "_empty_crs_source",
+                    empty_crs_source,
+                )
 
         if geometry is None and "geometry" in self.columns:
 
@@ -396,7 +429,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
@@ -591,6 +628,7 @@ def set_geometry(
         if new_series:
             # Note: This casts GeoSeries back into pspd.Series, so we lose any metadata that's not serialized.
             frame[geo_column_name] = level
+            object.__setattr__(frame, "_empty_crs_source", level)
 
         if not inplace:
             return frame
@@ -710,7 +748,14 @@ def _to_geopandas(self) -> gpd.GeoDataFrame:
             else:
                 pd_df[col_name] = series._to_pandas()
 
-        return gpd.GeoDataFrame(pd_df, geometry=self._geometry_column_name)
+        result = gpd.GeoDataFrame(
+            pd_df,
+            geometry=self._geometry_column_name,
+            crs=self.crs if self._geometry_column_name is not None else None,
+        )
+        if self._geometry_column_name is None:
+            result._geometry_column_name = None
+        return result
 
     def to_spark_pandas(self) -> pspd.DataFrame:
         """
@@ -1450,6 +1495,59 @@ def sjoin(
             **kwargs,
         )
 
+    def sjoin_nearest(
+        self,
+        other,
+        how="inner",
+        max_distance=None,
+        lsuffix="left",
+        rsuffix="right",
+        distance_col=None,
+        exclusive=False,
+    ):
+        """Join another GeoDataFrame using distributed nearest neighbours.
+
+        Parameters
+        ----------
+        other : GeoDataFrame
+            Right-hand frame.
+        how : {"inner", "left", "right"}, default "inner"
+            Join mode and side whose index and active geometry are retained.
+        max_distance : float, optional
+            Maximum planar distance for a nearest match. This is applied after
+            nearest candidates are found and does not currently prune the KNN
+            search.
+        lsuffix, rsuffix : str
+            Suffixes for overlapping columns.
+        distance_col : str, optional
+            Name of a column containing match distances.
+        exclusive : bool, default False
+            Exclude topologically equal candidates.
+
+        Returns
+        -------
+        GeoDataFrame
+            Distributed nearest-join result.
+
+        See Also
+        --------
+        sedona.spark.geopandas.sjoin_nearest
+        """
+        from sedona.spark.geopandas.tools.sjoin import (
+            sjoin_nearest as sjoin_nearest_tool,
+        )
+
+        return sjoin_nearest_tool(
+            self,
+            other,
+            how=how,
+            max_distance=max_distance,
+            lsuffix=lsuffix,
+            rsuffix=rsuffix,
+            distance_col=distance_col,
+            exclusive=exclusive,
+        )
+
     # ============================================================================
     # I/O OPERATIONS
     # ============================================================================