diff --git a/docs/setup/release-notes.md b/docs/setup/release-notes.md
index def6fbb77f5..a3359f3378a 100644
--- a/docs/setup/release-notes.md
+++ b/docs/setup/release-notes.md
@@ -62,6 +62,7 @@
#### GeoPandas API
+* [GH-3182] - Add distributed point-only inner `sjoin_nearest` joins using existing Sedona KNN tie semantics, with optional distance output and a maximum-distance result filter
* [GH-2068] - Implement the `limit` parameter for `GeoSeries.fillna`
* [GH-3257] - Implement distributed GeoSeries and GeoDataFrame Hilbert-distance spatial ordering
* [GH-3260] - Implement distributed GeoSeries and GeoDataFrame identical-geometry equality
diff --git a/docs/tutorial/geopandas-api.md b/docs/tutorial/geopandas-api.md
index b6186e9c245..b0164c713fe 100644
--- a/docs/tutorial/geopandas-api.md
+++ b/docs/tutorial/geopandas-api.md
@@ -239,6 +239,40 @@ intersects_result = left_df.sjoin(right_df, predicate="intersects")
contains_result = left_df.sjoin(right_df, predicate="contains")
```
+### Nearest point joins
+
+`GeoDataFrame.sjoin_nearest`, `sedona.spark.geopandas.sjoin_nearest`, and
+`sedona.spark.geopandas.tools.sjoin_nearest` use Sedona's distributed KNN join.
+This initial implementation supports **point inputs and inner joins**:
+
+```python
+nearest = left_points.sjoin_nearest(
+ right_points,
+ distance_col="distance",
+ max_distance=100,
+)
+```
+
+The result retains the left geometry, CRS and index, the right index, and both
+frames' attributes. Overlapping attributes receive `_left` and `_right`
+suffixes. Null and empty geometries do not match. Distances are planar, use
+CRS units, and ignore Z; use a projected CRS for meaningful distance units.
+The optional positive finite `max_distance` filters nearest results after the
+search, rather than reducing the search radius. `distance_col` must be a new
+column. Output row order is unspecified.
+
+**Ties follow Sedona configuration, not GeoPandas' all-ties guarantee.**
+`spark.sedona.join.knn.includeTieBreakers` defaults to `false`, choosing one
+nearest candidate with unspecified tie selection. Setting it to `true`
+includes equidistant candidates, but distinct rows with identical candidate
+geometries can still be omitted. This API does not change the session setting.
+
+Only `how="inner"`, `exclusive=False`, default suffixes, unique flat string
+columns and single-level indexes are supported. Duplicate index values are
+allowed. Unsupported options, non-point non-empty geometries and ambiguous
+output column names raise explicit errors. Input validation evaluates scalar
+summaries on Spark; geometry rows remain distributed.
+
### Coordinate Reference System Operations
Transform geometries between different coordinate reference systems:
diff --git a/docs/tutorial/geopandas-api.zh.md b/docs/tutorial/geopandas-api.zh.md
index d4ffdc81d54..5cfe2b00c66 100644
--- a/docs/tutorial/geopandas-api.zh.md
+++ b/docs/tutorial/geopandas-api.zh.md
@@ -239,6 +239,35 @@ intersects_result = left_df.sjoin(right_df, predicate="intersects")
contains_result = left_df.sjoin(right_df, predicate="contains")
```
+### 最近邻点连接
+
+`GeoDataFrame.sjoin_nearest`、`sedona.spark.geopandas.sjoin_nearest` 和
+`sedona.spark.geopandas.tools.sjoin_nearest` 使用 Sedona 的分布式 KNN 连接。
+当前实现支持 **点数据的内连接**：
+
+```python
+nearest = left_points.sjoin_nearest(
+ right_points,
+ distance_col="distance",
+ max_distance=100,
+)
+```
+
+结果保留左侧几何、CRS 和索引，以及右侧索引和双方属性。重名属性使用
+`_left` 和 `_right` 后缀。缺失或空几何不参与匹配。距离按平面计算，使用 CRS
+单位并忽略 Z；请使用投影 CRS 获取有意义的距离单位。可选的 `max_distance`
+必须是有限正数，仅在最近邻搜索后过滤结果，不会缩小搜索范围。
+`distance_col` 必须是新列，输出行的顺序不保证。
+
+**等距候选遵循 Sedona 配置，不保证像 GeoPandas 一样返回所有等距行。**
+`spark.sedona.join.knn.includeTieBreakers` 默认为 `false`，在等距候选中选择
+一行，选择顺序不保证。设为 `true` 会返回等距候选，但候选几何完全相同的
+不同记录仍可能被省略。本 API 不会修改此会话配置。
+
+目前仅支持 `how="inner"`、`exclusive=False`、默认后缀、唯一的单层字符串列名
+和单层索引（允许重复索引值）。不支持的选项、非空的非点几何及歧义输出列名
+会明确报错。输入验证在 Spark 上计算标量汇总，几何数据始终保持分布式。
+
### 坐标参考系操作
在不同坐标参考系（CRS）之间转换几何对象：
diff --git a/python/sedona/doc/sedona.spark.geopandas.tools.rst b/python/sedona/doc/sedona.spark.geopandas.tools.rst
index fcfb52ab87f..92b8e3971db 100644
--- a/python/sedona/doc/sedona.spark.geopandas.tools.rst
+++ b/python/sedona/doc/sedona.spark.geopandas.tools.rst
@@ -5,3 +5,8 @@ sedona.spark.geopandas.tools
:members:
:show-inheritance:
:undoc-members:
+
+.. automodule:: sedona.spark.geopandas.tools.sjoin_nearest
+ :members:
+ :show-inheritance:
+ :undoc-members:
diff --git a/python/sedona/spark/geopandas/__init__.py b/python/sedona/spark/geopandas/__init__.py
index a15cb4f9809..2b60008d6bf 100644
--- a/python/sedona/spark/geopandas/__init__.py
+++ b/python/sedona/spark/geopandas/__init__.py
@@ -24,7 +24,7 @@
from sedona.spark.geopandas.geodataframe import GeoDataFrame
from sedona.spark.geopandas.array import points_from_xy
-from sedona.spark.geopandas.tools import clip, overlay, sjoin
+from sedona.spark.geopandas.tools import clip, overlay, sjoin, sjoin_nearest
from sedona.spark.geopandas.io import list_layers, read_file, read_parquet
diff --git a/python/sedona/spark/geopandas/geodataframe.py b/python/sedona/spark/geopandas/geodataframe.py
index f0d53e0d3f3..1d99a6256ba 100644
--- a/python/sedona/spark/geopandas/geodataframe.py
+++ b/python/sedona/spark/geopandas/geodataframe.py
@@ -2994,6 +2994,37 @@ def overlay(
make_valid=make_valid,
)
+ def sjoin_nearest(
+ self,
+ right,
+ how="inner",
+ max_distance=None,
+ lsuffix="left",
+ rsuffix="right",
+ distance_col=None,
+ exclusive=False,
+ ):
+ """Join points to their nearest neighbors using Sedona KNN semantics.
+
+ Supports inner joins, default suffixes, ``exclusive=False``, an optional
+ positive ``max_distance`` result filter, and a new ``distance_col``.
+ Ties follow ``spark.sedona.join.knn.includeTieBreakers``; all GeoPandas
+ equidistant rows are not guaranteed. See
+ :func:`sedona.spark.geopandas.sjoin_nearest` for the full restrictions.
+ """
+ from sedona.spark.geopandas.tools.sjoin_nearest import sjoin_nearest
+
+ return sjoin_nearest(
+ self,
+ right,
+ how=how,
+ max_distance=max_distance,
+ lsuffix=lsuffix,
+ rsuffix=rsuffix,
+ distance_col=distance_col,
+ exclusive=exclusive,
+ )
+
def sjoin(
self,
other,
diff --git a/python/sedona/spark/geopandas/tools/__init__.py b/python/sedona/spark/geopandas/tools/__init__.py
index 2666029ec93..9741319ec10 100644
--- a/python/sedona/spark/geopandas/tools/__init__.py
+++ b/python/sedona/spark/geopandas/tools/__init__.py
@@ -19,10 +19,12 @@
from .collect import collect
from .overlay import overlay
from .sjoin import sjoin
+from .sjoin_nearest import sjoin_nearest
__all__ = [
"clip",
"collect",
"overlay",
"sjoin",
+ "sjoin_nearest",
]
diff --git a/python/sedona/spark/geopandas/tools/sjoin_nearest.py b/python/sedona/spark/geopandas/tools/sjoin_nearest.py
new file mode 100644
index 00000000000..663585dac19
--- /dev/null
+++ b/python/sedona/spark/geopandas/tools/sjoin_nearest.py
@@ -0,0 +1,260 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Point nearest joins using Sedona's existing KNN execution strategies."""
+
+import math
+import numbers
+import warnings
+
+import pyspark.pandas as ps
+from pyspark.pandas.internal import InternalField, InternalFrame
+from pyspark.pandas.utils import scol_for
+from pyspark.sql import functions as F
+
+from sedona.spark.geopandas import GeoDataFrame
+from sedona.spark.geopandas._crs import warn_crs_mismatch, with_crs_metadata
+from sedona.spark.sql import st_functions as stf
+from sedona.spark.sql.types import GeometryType
+
+
+def _prepare_side(frame, prefix):
+ """Project physical aliases and check point inputs without collecting geometries."""
+ internal = frame._internal
+ position = frame._active_geometry_position()
+ geometry_name = f"{prefix}_{position}"
+ columns = []
+ for i, (column, field) in enumerate(
+ zip(internal.data_spark_columns, internal.data_fields)
+ ):
+ # Resolve source CRS before filtering: an empty result cannot infer it from SRIDs.
+ if isinstance(field.spark_type, GeometryType) or i == position:
+ field = with_crs_metadata(field, frame[internal.column_labels[i][0]].crs)
+ columns.append(column.alias(f"{prefix}_{i}", metadata=field.metadata))
+ projected = internal.spark_frame.select(
+ internal.index_spark_columns[0].alias(f"{prefix}_index"), *columns
+ )
+ geometry = F.col(geometry_name)
+ usable = geometry.isNotNull() & ~stf.ST_IsEmpty(geometry)
+ stats = projected.agg(
+ F.max(F.when(usable, 1).otherwise(0)).alias("has_points"),
+ F.max(
+ F.when(usable & (stf.ST_GeometryType(geometry) != "ST_Point"), 1).otherwise(
+ 0
+ )
+ ).alias("nonpoint"),
+ ).first()
+ if stats.nonpoint:
+ raise NotImplementedError(
+ "sjoin_nearest currently supports only Point geometries"
+ )
+ return projected.filter(usable), geometry_name, bool(stats.has_points)
+
+
+def sjoin_nearest(
+ left_df,
+ right_df,
+ how="inner",
+ max_distance=None,
+ lsuffix="left",
+ rsuffix="right",
+ distance_col=None,
+ exclusive=False,
+):
+ """Join each left point to its nearest right point using planar distance.
+
+ This initial implementation supports point inputs, ``how='inner'``,
+ ``exclusive=False``, default suffixes, unique string column names and
+ single-level indexes (including duplicate index values). Null and empty
+ geometries do not match. The left geometry, CRS and index are retained.
+
+ Parameters
+ ----------
+ left_df, right_df : GeoDataFrame
+ Distributed point frames to join.
+ how : str, default 'inner'
+ Only inner joins are supported.
+ max_distance : float, optional
+ Positive finite distance in CRS units. Filters nearest results; it
+ does not reduce the KNN search radius.
+ lsuffix, rsuffix : str, default 'left', 'right'
+ Only the default suffixes are supported for overlapping columns.
+ distance_col : str, optional
+ New output column for the distance. Must not replace an input column.
+ exclusive : bool, default False
+ Only False is supported; equal points can match at distance zero.
+
+ Returns
+ -------
+ GeoDataFrame
+ Nearest matches, with no guaranteed row order.
+
+ Notes
+ -----
+ Ties follow ``spark.sedona.join.knn.includeTieBreakers`` (default False).
+ False chooses one nearest candidate with unspecified tie selection. True
+ includes ties, but Sedona may omit distinct rows with identical candidate
+ geometries. Unlike GeoPandas, all equidistant rows are not guaranteed.
+ This function does not change that session setting.
+
+ Distances are planar and ignore Z. Use a projected CRS for meaningful
+ distance units. Scalar validation summaries are evaluated eagerly on
+ Spark; input geometry rows remain distributed.
+ """
+ for name, frame in (("left_df", left_df), ("right_df", right_df)):
+ if not isinstance(frame, GeoDataFrame):
+ raise ValueError(f"'{name}' should be a GeoDataFrame")
+ internal = frame._internal
+ if len(internal.index_spark_columns) != 1:
+ raise NotImplementedError("sjoin_nearest requires a single-level index")
+ labels = internal.column_labels
+ if any(len(label) != 1 or not isinstance(label[0], str) for label in labels):
+ raise NotImplementedError("sjoin_nearest requires flat string columns")
+ if len(set(labels)) != len(labels):
+ raise NotImplementedError("sjoin_nearest requires unique columns")
+ if frame._active_geometry_position() is None:
+ raise ValueError(f"'{name}' has no active geometry column")
+ if how != "inner":
+ raise NotImplementedError("sjoin_nearest currently supports only how='inner'")
+ if exclusive is not False:
+ raise NotImplementedError(
+ "sjoin_nearest currently supports only exclusive=False"
+ )
+ if lsuffix != "left" or rsuffix != "right":
+ raise NotImplementedError(
+ "sjoin_nearest currently supports only default suffixes"
+ )
+ if max_distance is not None and (
+ isinstance(max_distance, bool)
+ or not isinstance(max_distance, numbers.Real)
+ or not math.isfinite(max_distance)
+ or max_distance <= 0
+ ):
+ raise ValueError("max_distance must be a positive finite number")
+ if distance_col is not None and (
+ not isinstance(distance_col, str)
+ or distance_col in left_df.columns
+ or distance_col in right_df.columns
+ ):
+ raise ValueError("distance_col must be a new string column name")
+
+ left_labels = list(left_df.columns)
+ right_labels = [
+ label for label in right_df.columns if label != right_df.active_geometry_name
+ ]
+ right_index_name = right_df._internal.index_names[0]
+ right_index_label = (
+ right_index_name[0] if right_index_name is not None else "index_right"
+ )
+ if not isinstance(right_index_label, str):
+ raise NotImplementedError(
+ "sjoin_nearest requires a string or unnamed right index"
+ )
+ if right_index_label in right_labels or (
+ right_index_name is None and right_index_label in left_labels
+ ):
+ raise ValueError(f"'{right_index_label}' conflicts with a join column")
+ right_labels.insert(0, right_index_label)
+ left_index_name = left_df._internal.index_names[0]
+ left_index_label = (
+ left_index_name[0] if left_index_name is not None else "index_left"
+ )
+ if left_index_label in left_labels or (
+ left_index_name is None and left_index_label in right_labels
+ ):
+ raise ValueError(f"'{left_index_label}' conflicts with a join column")
+ overlap = (set(left_labels) | {left_index_label}).intersection(right_labels)
+ result_index_name = (
+ (f"{left_index_label}_left",)
+ if left_index_label in overlap and left_index_name is not None
+ else left_index_name
+ )
+ left_output = [
+ (
+ label + "_left"
+ if label in overlap and label != left_df.active_geometry_name
+ else label
+ )
+ for label in left_labels
+ ]
+ right_output = [
+ label + "_right" if label in overlap else label for label in right_labels
+ ]
+ output_labels = left_output + right_output
+ if len(set(output_labels)) != len(output_labels):
+ raise ValueError("sjoin_nearest suffixes create duplicate output columns")
+ if distance_col in output_labels:
+ raise ValueError("distance_col conflicts with a join output column")
+
+ left_crs, right_crs = left_df.crs, right_df.crs
+ warn_crs_mismatch(left_crs, right_crs)
+ if any(crs is not None and crs.is_geographic for crs in (left_crs, right_crs)):
+ warnings.warn(
+ "sjoin_nearest uses planar distances in a geographic CRS; "
+ "use to_crs() to project the geometries before joining.",
+ UserWarning,
+ stacklevel=2,
+ )
+ left, left_geometry, left_has_points = _prepare_side(left_df, "__nearest_l")
+ right, right_geometry, right_has_points = _prepare_side(right_df, "__nearest_r")
+ if left_has_points and right_has_points:
+ joined = left.join(
+ right,
+ F.expr(f"ST_KNN({left_geometry}, {right_geometry}, 1, false)"),
+ "inner",
+ )
+ else:
+ # Existing KNN partitioning needs a non-empty extent on both sides.
+ # A false join builds the correct empty schema without invoking KNN.
+ joined = left.limit(0).join(right.limit(0), F.lit(False), "inner")
+ if distance_col is not None or max_distance is not None:
+ distance = stf.ST_Distance(F.col(left_geometry), F.col(right_geometry))
+ joined = joined.withColumn("__nearest_distance", distance)
+ if max_distance is not None:
+ # A non-deterministic guard keeps this filter above the join.
+ # Partition IDs are always nonnegative. Without the guard Spark
+ # pushes the bound into the join: Sedona may choose a distance
+ # join instead of KNN, or discard it in a broadcast KNN plan.
+ joined = joined.filter(
+ F.when(
+ F.spark_partition_id() >= 0,
+ F.col("__nearest_distance") <= float(max_distance),
+ ).otherwise(False)
+ )
+
+ output_names = [f"__nearest_l_{i}" for i in range(len(left_labels))]
+ output_names += ["__nearest_r_index"] + [
+ f"__nearest_r_{i}"
+ for i, label in enumerate(right_df.columns)
+ if label != right_df.active_geometry_name
+ ]
+ if distance_col is not None:
+ output_names.append("__nearest_distance")
+ output_labels.append(distance_col)
+ result = joined.select("__nearest_l_index", *output_names)
+ internal = InternalFrame(
+ spark_frame=result,
+ index_spark_columns=[scol_for(result, "__nearest_l_index")],
+ index_names=[result_index_name],
+ column_labels=[(label,) for label in output_labels],
+ data_spark_columns=[scol_for(result, name) for name in output_names],
+ data_fields=[
+ InternalField.from_struct_field(result.schema[name])
+ for name in output_names
+ ],
+ )
+ return GeoDataFrame(ps.DataFrame(internal), geometry=left_df.active_geometry_name)
diff --git a/python/tests/geopandas/test_sjoin_nearest.py b/python/tests/geopandas/test_sjoin_nearest.py
new file mode 100644
index 00000000000..6292f122514
--- /dev/null
+++ b/python/tests/geopandas/test_sjoin_nearest.py
@@ -0,0 +1,280 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import geopandas as gpd
+import pandas as pd
+import pytest
+import pyspark.pandas as ps
+import shapely
+from packaging.version import parse as parse_version
+from geopandas.testing import assert_geodataframe_equal
+from shapely.geometry import LineString, Point
+
+from sedona.spark import geopandas as sgpd
+from tests.geopandas.test_geopandas_base import TestGeopandasBase
+
+pytestmark = pytest.mark.skipif(
+ parse_version(shapely.__version__) < parse_version("2.0.0"),
+ reason="GeoPandas nearest comparisons require Shapely >= 2.0",
+)
+
+
+class TestSpatialJoinNearest(TestGeopandasBase):
+ @staticmethod
+ def convert(frame):
+ with ps.option_context("compute.ops_on_diff_frames", True):
+ return sgpd.GeoDataFrame(frame)
+
+ def frames(self):
+ left = gpd.GeoDataFrame(
+ {"name": ["a", "b"], "geometry": [Point(0, 0), Point(10, 0)]},
+ index=pd.Index([5, 5], name="left_id"),
+ crs=3857,
+ )
+ right = gpd.GeoDataFrame(
+ {"name": ["near", "far"], "geometry": [Point(1, 0), Point(30, 0)]},
+ index=pd.Index([8, 9], name="right_id"),
+ crs=3857,
+ )
+ return left, right
+
+ def nearest(self, left, right, **kwargs):
+ join = getattr(sgpd, "sjoin_nearest", None)
+ assert callable(join), "The public sjoin_nearest API is missing"
+ return join(left, right, **kwargs)
+
+ @pytest.mark.parametrize("max_distance", [None, 1.0, 0.5])
+ def test_nearest_matches_geopandas_and_keeps_duplicate_indexes(self, max_distance):
+ left, right = self.frames()
+ actual = self.nearest(
+ self.convert(left),
+ self.convert(right),
+ max_distance=max_distance,
+ distance_col="distance",
+ )
+ expected = left.sjoin_nearest(
+ right, max_distance=max_distance, distance_col="distance"
+ )
+ assert_geodataframe_equal(
+ actual.to_geopandas().sort_values("name_left"),
+ expected.sort_values("name_left"),
+ check_dtype=False,
+ )
+ assert actual.crs == left.crs
+
+ @pytest.mark.parametrize("strategy", ["regular", "left", "right"])
+ @pytest.mark.parametrize("max_distance", [None, 1.0, 0.5])
+ @pytest.mark.parametrize("include_ties, expected", [(False, 1), (True, 2)])
+ def test_honors_session_ties_without_changing_configuration(
+ self, include_ties, expected, strategy, max_distance
+ ):
+ key = "spark.sedona.join.knn.includeTieBreakers"
+ previous = self.spark.conf.get(key, "false")
+ self.spark.conf.set(key, str(include_ties).lower())
+ try:
+ left = self.convert(gpd.GeoDataFrame(geometry=[Point(0, 0)], crs=3857))
+ right = self.convert(
+ gpd.GeoDataFrame(
+ {"value": ["west", "east", "far"]},
+ geometry=[Point(-1, 0), Point(1, 0), Point(3, 0)],
+ crs=3857,
+ )
+ )
+ if strategy != "regular":
+ frame = left if strategy == "left" else right
+ hinted = sgpd.GeoDataFrame(
+ ps.DataFrame(
+ frame._internal.copy(
+ spark_frame=frame._internal.spark_frame.hint("broadcast")
+ )
+ ),
+ geometry="geometry",
+ )
+ if strategy == "left":
+ left = hinted
+ else:
+ right = hinted
+ actual = left.sjoin_nearest(
+ right, distance_col="distance", max_distance=max_distance
+ )
+ rows = actual.to_geopandas()
+ assert len(rows) == (0 if max_distance == 0.5 else expected)
+ assert set(rows["distance"]) == (set() if max_distance == 0.5 else {1.0})
+ assert set(rows["value"]).issubset({"west", "east"})
+ assert self.spark.conf.get(key) == str(include_ties).lower()
+ plan = (
+ actual._internal.spark_frame._jdf.queryExecution()
+ .executedPlan()
+ .toString()
+ )
+ expected_plan = {
+ "regular": "KNNJoin",
+ "left": "BroadcastQuerySideKNNJoin",
+ "right": "BroadcastObjectSideKNNJoin",
+ }[strategy]
+ assert expected_plan in plan
+ assert "CartesianProduct" not in plan
+ assert "PythonUDF" not in plan
+ finally:
+ self.spark.conf.set(key, previous)
+
+ @pytest.mark.parametrize(
+ "side, geometries",
+ [
+ ("left", []),
+ ("right", []),
+ ("left", [Point(), None]),
+ ("right", [Point(), None]),
+ ],
+ )
+ def test_empty_inputs_keep_schema_and_crs(self, side, geometries):
+ left, right = self.frames()
+ # Spark 3.5 infers a UDT from the first value, so put the empty Point
+ # before None when constructing the fixture. Both still cannot match.
+ empty = gpd.GeoDataFrame(
+ {"name": ["x"] * len(geometries)}, geometry=geometries, crs=3857
+ )
+ empty.index.name = f"{side}_id"
+ if side == "left":
+ left = empty
+ else:
+ right = empty
+ actual = self.nearest(
+ self.convert(left), self.convert(right), distance_col="distance"
+ )
+ expected = left.sjoin_nearest(right, distance_col="distance")
+ assert_geodataframe_equal(
+ actual.to_geopandas(), expected, check_dtype=False, check_index_type=False
+ )
+ assert actual.crs == left.crs
+
+ def test_null_empty_and_special_column_names(self):
+ left = gpd.GeoDataFrame(
+ {
+ "a.b": ["keep", "null", "empty"],
+ "geom left": [Point(0, 0), None, Point()],
+ },
+ geometry="geom left",
+ crs=3857,
+ )
+ right = gpd.GeoDataFrame(
+ {
+ "a.b": ["empty", "null", "near"],
+ "geom right": [Point(), None, Point(1, 0)],
+ },
+ geometry="geom right",
+ crs=3857,
+ )
+ actual = self.nearest(
+ self.convert(left), self.convert(right), distance_col="d.m`"
+ )
+ expected = left.sjoin_nearest(right, distance_col="d.m`")
+ assert_geodataframe_equal(actual.to_geopandas(), expected, check_dtype=False)
+
+ @pytest.mark.parametrize(
+ "kwargs",
+ [
+ {"how": "left"},
+ {"how": "right"},
+ {"exclusive": True},
+ {"lsuffix": "l"},
+ {"rsuffix": "r"},
+ ],
+ )
+ def test_unsupported_options_raise(self, kwargs):
+ left, right = self.frames()
+ with pytest.raises(NotImplementedError):
+ self.nearest(self.convert(left), self.convert(right), **kwargs)
+
+ @pytest.mark.parametrize("distance", [0, -1, float("nan"), float("inf"), "1", True])
+ def test_invalid_max_distance_raises(self, distance):
+ left, right = self.frames()
+ with pytest.raises(ValueError):
+ self.nearest(self.convert(left), self.convert(right), max_distance=distance)
+
+ @pytest.mark.parametrize("side", ["left", "right"])
+ def test_nonpoint_geometries_raise(self, side):
+ left, right = self.frames()
+ frame = left if side == "left" else right
+ frame.loc[frame.index[0], "geometry"] = LineString([(0, 0), (1, 1)])
+ with pytest.raises(NotImplementedError, match="Point"):
+ self.nearest(self.convert(left), self.convert(right))
+
+ def test_rejects_multiindex_and_existing_distance_column(self):
+ left, right = self.frames()
+ with pytest.raises(ValueError, match="distance_col"):
+ self.nearest(self.convert(left), self.convert(right), distance_col="name")
+ left.index = pd.MultiIndex.from_tuples([(1, "a"), (2, "b")])
+ with pytest.raises(NotImplementedError, match="index"):
+ self.nearest(self.convert(left), self.convert(right))
+
+ @pytest.mark.parametrize("explicit_none", [False, True])
+ def test_empty_result_preserves_source_crs_provenance(self, explicit_none):
+ from sedona.spark.geopandas._crs import with_crs_metadata
+
+ raw = self.spark.sql("SELECT ST_SetSRID(ST_Point(0D, 0D), 3857) AS geometry")
+ left = sgpd.GeoDataFrame(raw, geometry="geometry")
+ if explicit_none:
+ internal = left._internal
+ field = with_crs_metadata(internal.data_fields[0], None)
+ left = sgpd.GeoDataFrame(
+ ps.DataFrame(internal.copy(data_fields=[field])), geometry="geometry"
+ )
+ right = sgpd.GeoDataFrame(
+ self.spark.sql("SELECT ST_SetSRID(ST_Point(2D, 0D), 3857) AS geometry"),
+ geometry="geometry",
+ )
+ if explicit_none:
+ with pytest.warns(UserWarning, match="CRS mismatch"):
+ actual = self.nearest(left, right, max_distance=1)
+ else:
+ actual = self.nearest(left, right, max_distance=1)
+ assert actual.to_geopandas().empty
+ assert actual.crs == left.crs
+ assert (actual.crs is None) == explicit_none
+
+ def test_duplicate_query_geometries_remain_distinct_rows(self):
+ left = self.convert(
+ gpd.GeoDataFrame(
+ {"row": ["a", "b"]}, geometry=[Point(0, 0)] * 2, index=[7, 7], crs=3857
+ )
+ )
+ right = self.convert(
+ gpd.GeoDataFrame(geometry=[Point(0, 0), Point(2, 0)], crs=3857)
+ )
+ actual = self.nearest(left, right, distance_col="distance").to_geopandas()
+ assert sorted(actual["row"]) == ["a", "b"]
+ assert list(actual.index) == [7, 7]
+ assert set(actual["distance"]) == {0.0}
+
+ def test_rejects_output_label_collisions(self):
+ left, right = self.frames()
+ left["name_left"] = "existing"
+ with pytest.raises(ValueError, match="duplicate output"):
+ self.nearest(self.convert(left), self.convert(right))
+
+ def test_named_index_overlap_matches_geopandas(self):
+ left, right = self.frames()
+ left.index.name = "id"
+ right.index.name = "id"
+ actual = self.nearest(self.convert(left), self.convert(right)).to_geopandas()
+ expected = left.sjoin_nearest(right)
+ assert_geodataframe_equal(
+ actual.sort_values("name_left"),
+ expected.sort_values("name_left"),
+ check_dtype=False,
+ )