From 43d8733300b39b10b255b7e73ccdb02c8410e4d5 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 24 Sep 2026 00:14:56 -0700 Subject: [PATCH] [GH-3182] GeoPandas: Add point-only inner sjoin_nearest Implement distributed point nearest joins using existing ST_KNN behavior, with optional distance output and a maximum-distance result filter. Preserve geometry CRS metadata and document the supported API subset. Add GeoPandas comparisons and regular/broadcast KNN execution coverage. --- docs/setup/release-notes.md | 1 + docs/tutorial/geopandas-api.md | 34 ++ docs/tutorial/geopandas-api.zh.md | 29 ++ .../doc/sedona.spark.geopandas.tools.rst | 5 + python/sedona/spark/geopandas/__init__.py | 2 +- python/sedona/spark/geopandas/geodataframe.py | 31 ++ .../sedona/spark/geopandas/tools/__init__.py | 2 + .../spark/geopandas/tools/sjoin_nearest.py | 261 +++++++++++++++ python/tests/geopandas/test_sjoin_nearest.py | 305 ++++++++++++++++++ 9 files changed, 669 insertions(+), 1 deletion(-) create mode 100644 python/sedona/spark/geopandas/tools/sjoin_nearest.py create mode 100644 python/tests/geopandas/test_sjoin_nearest.py diff --git a/docs/setup/release-notes.md b/docs/setup/release-notes.md index def6fbb77f5..0b8821f5657 100644 --- a/docs/setup/release-notes.md +++ b/docs/setup/release-notes.md @@ -63,6 +63,7 @@ #### GeoPandas API * [GH-2068] - Implement the `limit` parameter for `GeoSeries.fillna` +* [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-3257] - Implement distributed GeoSeries and GeoDataFrame Hilbert-distance spatial ordering * [GH-3260] - Implement distributed GeoSeries and GeoDataFrame identical-geometry equality * [GH-3273] - Implement distributed GeoSeries and GeoDataFrame polygonal coverage validation and invalid-edge diagnostics 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..9e7770fc855 --- /dev/null +++ b/python/sedona/spark/geopandas/tools/sjoin_nearest.py @@ -0,0 +1,261 @@ +# 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. + # Extra-condition loss is tracked in GH-3398. + 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..f3ac326c489 --- /dev/null +++ b/python/tests/geopandas/test_sjoin_nearest.py @@ -0,0 +1,305 @@ +# 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, + ) + + @pytest.mark.parametrize("max_distance", [None, 0.5]) + def test_secondary_geometry_retains_its_crs_on_empty_results(self, max_distance): + left = self.convert(gpd.GeoDataFrame(geometry=[Point(0, 0)], crs=3857)) + # Raw Spark columns carry SRIDs without GeoPandas CRS metadata. An + # empty output must retain the secondary column's source CRS too. + right = sgpd.GeoDataFrame( + self.spark.createDataFrame( + [("POINT (1 0)", "POINT (-73 40)")], ["wkt", "secondary_wkt"] + ).selectExpr( + "ST_SetSRID(ST_GeomFromWKT(wkt), 3857) AS geometry", + "ST_SetSRID(ST_GeomFromWKT(secondary_wkt), 4326) AS secondary", + ), + geometry="geometry", + ) + actual = self.nearest(left, right, max_distance=max_distance) + assert actual.crs.to_epsg() == 3857 + assert actual["secondary"].crs.to_epsg() == 4326 + rows = actual.to_geopandas() + assert rows["secondary"].crs.to_epsg() == 4326 + if max_distance is None: + assert len(rows) == 1 + assert rows["secondary"].iloc[0].equals_exact(Point(-73, 40), 0) + else: + assert rows.empty