From 4d7448c601cc39ddd8e2abe48b5d0a26f7539853 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 23 Jul 2026 22:58:16 -0700 Subject: [PATCH 01/11] [GH-3157] Implement distributed GeoSeries explode --- python/sedona/spark/geopandas/geoseries.py | 142 ++++++++++++++++-- python/tests/geopandas/test_geoseries.py | 48 +++++- .../geopandas/test_match_geopandas_series.py | 54 ++++++- 3 files changed, 227 insertions(+), 17 deletions(-) diff --git a/python/sedona/spark/geopandas/geoseries.py b/python/sedona/spark/geopandas/geoseries.py index 6fa633ac622..059913272f1 100644 --- a/python/sedona/spark/geopandas/geoseries.py +++ b/python/sedona/spark/geopandas/geoseries.py @@ -658,17 +658,14 @@ def _query_geometry_column( index_spark_columns = [] index_fields = [] if not is_aggr: - # We always select NATURAL_ORDER_COLUMN_NAME, to avoid having to regenerate it in the result. - # We always select SPARK_DEFAULT_INDEX_NAME, to retain series index info. - - exprs.append(scol_for(df, SPARK_DEFAULT_INDEX_NAME)) - exprs.append(scol_for(df, NATURAL_ORDER_COLUMN_NAME)) - - index_spark_columns = [scol_for(df, SPARK_DEFAULT_INDEX_NAME)] - index_fields = [self._internal.index_fields[0]] + # Preserve every index level and the natural order in the result. + index_spark_columns = [ + scol_for(df, name) for name in self._internal.index_spark_column_names + ] + index_fields = self._internal.index_fields sdf = df.select( col_expr, - scol_for(df, SPARK_DEFAULT_INDEX_NAME), + *index_spark_columns, scol_for(df, NATURAL_ORDER_COLUMN_NAME), ) # Otherwise, if is_aggr, we don't select the index columns. @@ -679,6 +676,7 @@ def _query_geometry_column( spark_frame=sdf, index_fields=index_fields, index_spark_columns=index_spark_columns, + index_names=[None] if is_aggr else self._internal.index_names, data_spark_columns=[scol_for(sdf, rename)], data_fields=[self._internal.data_fields[0].copy(name=rename)], column_label_names=[(rename,)], @@ -2791,12 +2789,130 @@ def fillna( return result def explode(self, ignore_index=False, index_parts=False) -> "GeoSeries": - raise NotImplementedError( - _not_implemented_error( - "explode", - "Explodes multi-part geometries into separate single-part geometries.", + """ + Explode multi-part geometries into multiple single geometries. + + Single rows can become multiple rows. This is analogous to PostGIS + ``ST_Dump``. Geometry collections are expanded by one level, so a + multi-part geometry nested in a collection remains multi-part. + + Parameters + ---------- + ignore_index : bool, default False + If True, the resulting index is labelled 0, 1, ..., n - 1 and + ``index_parts`` is ignored. + index_parts : bool, default False + If True, append a zero-based index level identifying each geometry + produced from an input row. + + Returns + ------- + GeoSeries + Exploded geometries. The original index is repeated by default. + + Examples + -------- + >>> from sedona.spark.geopandas import GeoSeries + >>> from shapely.geometry import MultiPoint + >>> s = GeoSeries( + ... [MultiPoint([(0, 0), (1, 1)]), MultiPoint([(2, 2), (3, 3)])] + ) + >>> s.explode(index_parts=True) + 0 0 POINT (0 0) + 1 POINT (1 1) + 1 0 POINT (2 2) + 1 POINT (3 3) + dtype: geometry + """ + from pyspark.pandas.internal import InternalField + from pyspark.pandas.utils import verify_temp_column_name + + internal = self._internal.resolved_copy + source_sdf = internal.spark_frame + + def temp_column_name(base: str) -> str: + suffix = 0 + candidate = f"__explode_{base}__" + while candidate in source_sdf.columns: + suffix += 1 + candidate = f"__explode_{base}_{suffix}__" + return typing.cast(str, verify_temp_column_name(source_sdf, candidate)) + + parent_order_col = temp_column_name("parent_order") + part_index_col = temp_column_name("part_index") + geometry_col = temp_column_name("geometry") + sequence_col = temp_column_name("sequence") + + exploded_sdf = source_sdf.select( + *internal.index_spark_columns, + scol_for(source_sdf, NATURAL_ORDER_COLUMN_NAME).alias(parent_order_col), + F.posexplode(stf.ST_Dump(internal.data_spark_columns[0])).alias( + part_index_col, geometry_col + ), + ).orderBy(parent_order_col, part_index_col) + + # Use pandas-on-Spark's distributed sequence implementation instead of + # a global row-number window. Besides serving as the ignored index, this + # provides a natural-order column for subsequent operations. + exploded_sdf = InternalFrame.attach_distributed_sequence_column( + exploded_sdf, sequence_col + ) + + data_col = internal.data_spark_column_names[0] + if ignore_index: + output_index_cols = [SPARK_DEFAULT_INDEX_NAME] + index_names = [None] + output_sdf = exploded_sdf.select( + scol_for(exploded_sdf, sequence_col).alias(SPARK_DEFAULT_INDEX_NAME), + scol_for(exploded_sdf, geometry_col).alias(data_col), + scol_for(exploded_sdf, sequence_col).alias(NATURAL_ORDER_COLUMN_NAME), ) + index_fields = [ + InternalField.from_struct_field( + output_sdf.schema[SPARK_DEFAULT_INDEX_NAME] + ) + ] + else: + output_index_cols = list(internal.index_spark_column_names) + index_names = list(internal.index_names) + index_fields = list(internal.index_fields) + index_expressions = [ + scol_for(exploded_sdf, name) for name in output_index_cols + ] + + if index_parts: + output_index_cols.append(part_index_col) + index_names.append(None) + index_expressions.append( + scol_for(exploded_sdf, part_index_col) + .cast("long") + .alias(part_index_col) + ) + + output_sdf = exploded_sdf.select( + *index_expressions, + scol_for(exploded_sdf, geometry_col).alias(data_col), + scol_for(exploded_sdf, sequence_col).alias(NATURAL_ORDER_COLUMN_NAME), + ) + + if index_parts: + index_fields.append( + InternalField.from_struct_field(output_sdf.schema[part_index_col]) + ) + + result_internal = internal.copy( + spark_frame=output_sdf, + index_spark_columns=[ + scol_for(output_sdf, name) for name in output_index_cols + ], + index_names=index_names, + index_fields=index_fields, + data_spark_columns=[scol_for(output_sdf, data_col)], + data_fields=[ + InternalField(np.dtype("object"), output_sdf.schema[data_col]) + ], ) + return GeoSeries(first_series(PandasOnSparkDataFrame(result_internal))) def to_crs( self, crs: Union[Any, None] = None, epsg: Union[int, None] = None diff --git a/python/tests/geopandas/test_geoseries.py b/python/tests/geopandas/test_geoseries.py index 435e0518602..2faf3775919 100644 --- a/python/tests/geopandas/test_geoseries.py +++ b/python/tests/geopandas/test_geoseries.py @@ -404,8 +404,52 @@ def test_fillna(self): expected = gpd.GeoSeries([Point(0, 0), GeometryCollection()], name="geometry") self.check_sgpd_equals_gpd(result, expected) - def test_explode(self): - pass + @pytest.mark.parametrize( + "kwargs", + [ + {}, + {"index_parts": True}, + {"ignore_index": True}, + {"ignore_index": True, "index_parts": True}, + ], + ) + def test_explode(self, kwargs): + from geopandas.testing import assert_geoseries_equal + + geometries = [ + MultiPoint([(0, 0), (1, 1)]), + Point(9, 9), + GeometryCollection([Point(2, 2), MultiPoint([(3, 3), (4, 4)])]), + Point(), + Polygon(), + MultiPoint(), + GeometryCollection(), + None, + ] + index = pd.Index(range(10, 18), name="feature_id") + expected = gpd.GeoSeries( + geometries, + index=index, + name="geometry", + crs="EPSG:4326", + ).explode(**kwargs) + + result = GeoSeries( + geometries, + index=index, + name="geometry", + crs="EPSG:4326", + ).explode(**kwargs) + actual = result.to_geopandas() + + assert_geoseries_equal( + actual, + expected, + check_index_type=False, + check_geom_type=True, + check_crs=True, + ) + pd.testing.assert_index_equal(actual.index, expected.index, exact=False) def test_to_crs(self): from pyproj import CRS diff --git a/python/tests/geopandas/test_match_geopandas_series.py b/python/tests/geopandas/test_match_geopandas_series.py index a46ce56543a..2c48841b93c 100644 --- a/python/tests/geopandas/test_match_geopandas_series.py +++ b/python/tests/geopandas/test_match_geopandas_series.py @@ -401,8 +401,58 @@ def test_fillna(self): gpd_result = gpd.GeoSeries(data).fillna(fill_val) self.check_sgpd_equals_gpd(sgpd_result, gpd_result) - def test_explode(self): - pass + @pytest.mark.parametrize( + "kwargs", + [ + {}, + {"index_parts": True}, + {"ignore_index": True}, + {"ignore_index": True, "index_parts": True}, + ], + ) + def test_explode(self, kwargs): + from geopandas.testing import assert_geoseries_equal + + family_names = [ + "point", + "line", + "polygon", + "multipoint", + "multiline", + "multipolygon", + "collection", + ] + families = [ + self.points, + self.linestrings, + self.polygons, + self.multipoints, + self.multilinestrings, + self.multipolygons, + self.geomcollection, + ] + geometries = [] + index_values = [] + for family_name, family in zip(family_names, families): + geometries.extend(family) + index_values.extend( + (family_name, row_number) for row_number in range(len(family)) + ) + + index = pd.MultiIndex.from_tuples(index_values, names=["family", "row_number"]) + expected = gpd.GeoSeries(geometries, index=index, name="geometry").explode( + **kwargs + ) + result = GeoSeries(geometries, index=index, name="geometry").explode(**kwargs) + actual = result.to_geopandas() + + assert_geoseries_equal( + actual, + expected, + check_index_type=False, + check_geom_type=True, + ) + pd.testing.assert_index_equal(actual.index, expected.index, exact=False) def test_to_crs(self): for geom in self.geoms: From 22e13f5a0184ec9a29aabb834943ac77131cd02b Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 23 Jul 2026 22:58:56 -0700 Subject: [PATCH 02/11] [GH-3156] Implement distributed GeoSeries get_coordinates --- python/sedona/spark/geopandas/base.py | 72 ++++++++++++ python/sedona/spark/geopandas/geoseries.py | 108 +++++++++++++++++- python/tests/geopandas/test_geoseries.py | 94 +++++++++++++++ .../geopandas/test_match_geopandas_series.py | 10 ++ 4 files changed, 283 insertions(+), 1 deletion(-) diff --git a/python/sedona/spark/geopandas/base.py b/python/sedona/spark/geopandas/base.py index cd9d13905ae..a8eaf28bc37 100644 --- a/python/sedona/spark/geopandas/base.py +++ b/python/sedona/spark/geopandas/base.py @@ -363,6 +363,78 @@ def count_coordinates(self): """ return _delegate_to_geometry_column("count_coordinates", self) + def get_coordinates( + self, + include_z=False, + ignore_index=False, + index_parts=False, + *, + include_m=False, + ): + """Get coordinates as a distributed pandas-on-Spark ``DataFrame``. + + The returned frame has ``x`` and ``y`` columns. With + ``include_z=True`` or ``include_m=True``, it also has ``z`` or ``m`` + columns, respectively. Missing optional ordinates are represented by + ``NaN``. + + Parameters + ---------- + include_z : bool, default False + Include Z coordinates. + ignore_index : bool, default False + If True, label the result with a new zero-based sequential index, + ignoring ``index_parts``. + index_parts : bool, default False + If True, append a zero-based coordinate-position level to the + original index. + include_m : bool, default False + Include M coordinates. + + Returns + ------- + pyspark.pandas.DataFrame + + Examples + -------- + >>> from sedona.spark.geopandas import GeoSeries + >>> from shapely.geometry import Point, LineString, Polygon + >>> s = GeoSeries( + ... [ + ... Point(1, 1), + ... LineString([(1, -1), (1, 0)]), + ... Polygon([(3, -1), (4, 0), (3, 1)]), + ... ] + ... ) + >>> s.get_coordinates() + x y + 0 1.0 1.0 + 1 1.0 -1.0 + 1 1.0 0.0 + 2 3.0 -1.0 + 2 4.0 0.0 + 2 3.0 1.0 + 2 3.0 -1.0 + + >>> s.get_coordinates(index_parts=True) + x y + 0 0 1.0 1.0 + 1 0 1.0 -1.0 + 1 1.0 0.0 + 2 0 3.0 -1.0 + 1 4.0 0.0 + 2 3.0 1.0 + 3 3.0 -1.0 + """ + return _delegate_to_geometry_column( + "get_coordinates", + self, + include_z, + ignore_index, + index_parts, + include_m=include_m, + ) + def count_geometries(self): """Return a ``Series`` of ``dtype('int')`` with the number of geometries in each multi-geometry or geometry collection. diff --git a/python/sedona/spark/geopandas/geoseries.py b/python/sedona/spark/geopandas/geoseries.py index 059913272f1..60a4a2930fe 100644 --- a/python/sedona/spark/geopandas/geoseries.py +++ b/python/sedona/spark/geopandas/geoseries.py @@ -29,7 +29,7 @@ from pyspark.pandas.frame import DataFrame as PandasOnSparkDataFrame from pyspark.pandas.internal import InternalFrame from pyspark.pandas.series import first_series -from pyspark.pandas.utils import scol_for +from pyspark.pandas.utils import scol_for, verify_temp_column_name from pyspark.sql.types import NullType from sedona.spark.sql.types import GeometryType @@ -52,6 +52,7 @@ from pyspark.pandas.internal import ( SPARK_DEFAULT_INDEX_NAME, # __index_level_0__ + SPARK_INDEX_NAME_FORMAT, NATURAL_ORDER_COLUMN_NAME, SPARK_DEFAULT_SERIES_NAME, # '0' ) @@ -863,6 +864,111 @@ def count_coordinates(self): returns_geom=False, ) + def get_coordinates( + self, + include_z=False, + ignore_index=False, + index_parts=False, + *, + include_m=False, + ) -> pspd.DataFrame: + source_frame = self._internal.spark_frame + source_order_name = verify_temp_column_name( + source_frame, "__coordinate_source_order__" + ) + position_name = verify_temp_column_name(source_frame, "__coordinate_position__") + point_name = verify_temp_column_name(source_frame, "__coordinate_point__") + + index_column_names = [ + SPARK_INDEX_NAME_FORMAT(i) + for i in range(len(self._internal.index_spark_columns)) + ] + index_columns = [ + column.alias(name) + for column, name in zip( + self._internal.index_spark_columns, index_column_names + ) + ] + + exploded_frame = source_frame.select( + *index_columns, + scol_for(source_frame, NATURAL_ORDER_COLUMN_NAME).alias(source_order_name), + F.posexplode(stf.ST_DumpPoints(self.spark.column)).alias( + position_name, point_name + ), + ).orderBy(source_order_name, position_name) + + sequence_name = verify_temp_column_name(exploded_frame, "__coordinate_order__") + sequenced_frame = InternalFrame.attach_distributed_sequence_column( + exploded_frame, sequence_name + ) + + coordinate_names = ["x", "y"] + coordinate_columns = [ + stf.ST_X(scol_for(sequenced_frame, point_name)).alias("x"), + stf.ST_Y(scol_for(sequenced_frame, point_name)).alias("y"), + ] + if include_z: + coordinate_names.append("z") + coordinate_columns.append( + stf.ST_Z(scol_for(sequenced_frame, point_name)).alias("z") + ) + if include_m: + coordinate_names.append("m") + coordinate_columns.append( + stf.ST_M(scol_for(sequenced_frame, point_name)).alias("m") + ) + + if ignore_index: + result_index_names = [SPARK_DEFAULT_INDEX_NAME] + result_index_labels = [None] + result_index_fields = None + result_index_columns = [ + scol_for(sequenced_frame, sequence_name).alias(SPARK_DEFAULT_INDEX_NAME) + ] + else: + result_index_names = index_column_names + result_index_labels = list(self._internal.index_names) + result_index_fields = [ + field.copy(name=name) + for field, name in zip(self._internal.index_fields, index_column_names) + ] + result_index_columns = [ + scol_for(sequenced_frame, name) for name in index_column_names + ] + + if index_parts: + part_index_name = SPARK_INDEX_NAME_FORMAT(len(result_index_names)) + result_index_names.append(part_index_name) + result_index_labels.append(None) + result_index_fields.append(None) + result_index_columns.append( + scol_for(sequenced_frame, position_name) + .cast("long") + .alias(part_index_name) + ) + + result_frame = sequenced_frame.select( + *result_index_columns, + *coordinate_columns, + scol_for(sequenced_frame, sequence_name).alias(NATURAL_ORDER_COLUMN_NAME), + ) + + internal = InternalFrame( + spark_frame=result_frame, + index_spark_columns=[ + scol_for(result_frame, name) for name in result_index_names + ], + index_names=result_index_labels, + index_fields=result_index_fields, + column_labels=[(name,) for name in coordinate_names], + data_spark_columns=[ + scol_for(result_frame, name) for name in coordinate_names + ], + column_label_names=[None], + ) + return pspd.DataFrame(internal) + def count_geometries(self): spark_expr = stf.ST_NumGeometries(self.spark.column) return self._query_geometry_column( diff --git a/python/tests/geopandas/test_geoseries.py b/python/tests/geopandas/test_geoseries.py index 2faf3775919..9a2ed014380 100644 --- a/python/tests/geopandas/test_geoseries.py +++ b/python/tests/geopandas/test_geoseries.py @@ -876,6 +876,100 @@ def test_count_coordinates(self): df_result = s.to_geoframe().count_coordinates() self.check_pd_series_equal(df_result, expected) + def test_get_coordinates(self): + geometries = [ + Point(1, 1), + LineString([(1, -1), (1, 0)]), + Polygon( + [(3, -1), (4, 0), (3, 1)], + [[(3.2, -0.5), (3.5, 0), (3.2, 0.5)]], + ), + MultiPoint([(5, 5), (6, 6)]), + MultiLineString([[(7, 7), (8, 8)], [(9, 9), (10, 10)]]), + MultiPolygon( + [ + Polygon([(11, 11), (12, 11), (11, 12)]), + Polygon([(13, 13), (14, 13), (13, 14)]), + ] + ), + GeometryCollection([Point(15, 15), LineString([(16, 16), (17, 17)])]), + Point(), + None, + ] + index = pd.Index([90, 80, 70, 60, 50, 40, 30, 20, 10], name="feature_id") + expected_series = gpd.GeoSeries(geometries, index=index) + actual_series = GeoSeries(expected_series) + + options = [ + {}, + {"index_parts": True}, + {"ignore_index": True}, + {"ignore_index": True, "index_parts": True}, + ] + for kwargs in options: + actual = actual_series.get_coordinates(**kwargs) + expected = expected_series.get_coordinates(**kwargs) + assert isinstance(actual, ps.DataFrame) + pd.testing.assert_frame_equal(actual.to_pandas(), expected) + + dataframe_result = actual_series.to_geoframe().get_coordinates() + pd.testing.assert_frame_equal( + dataframe_result.to_pandas(), expected_series.get_coordinates() + ) + + empty_series = gpd.GeoSeries( + [Point(), GeometryCollection(), None], + index=pd.Index([3, 2, 1], name="feature_id"), + ) + actual_empty = GeoSeries(empty_series).get_coordinates(index_parts=True) + expected_empty = empty_series.get_coordinates(index_parts=True) + # Spark keeps a fixed integer schema for the coordinate-position level, + # including when no rows are produced. GeoPandas infers object only for + # that all-empty level. + pd.testing.assert_frame_equal( + actual_empty.to_pandas(), expected_empty, check_index_type=False + ) + + def test_get_coordinates_multi_index(self): + index = pd.MultiIndex.from_tuples( + [("b", 2), ("a", 1), ("b", 1)], names=["group", "feature_id"] + ) + geometries = [ + LineString([(0, 0), (1, 1)]), + Point(2, 2), + Polygon([(3, 3), (4, 3), (3, 4)]), + ] + expected_series = gpd.GeoSeries(geometries, index=index) + actual_series = GeoSeries(expected_series) + + for kwargs in ({}, {"index_parts": True}, {"ignore_index": True}): + actual = actual_series.get_coordinates(**kwargs).to_pandas() + expected = expected_series.get_coordinates(**kwargs) + pd.testing.assert_frame_equal(actual, expected) + + @pytest.mark.skipif( + parse_version(shapely.__version__) < parse_version("2.1.0"), + reason="M coordinates require shapely>=2.1.0", + ) + def test_get_coordinates_zm(self): + geometries_wkt = [ + "POINT (0 1)", + "POINT Z (2 3 4)", + "POINT M (5 6 7)", + "POINT ZM (8 9 10 11)", + ] + expected_series = gpd.GeoSeries.from_wkt(geometries_wkt) + actual_series = GeoSeries.from_wkt(geometries_wkt) + + for kwargs in ( + {"include_z": True}, + {"include_m": True}, + {"include_z": True, "include_m": True}, + ): + actual = actual_series.get_coordinates(**kwargs).to_pandas() + expected = expected_series.get_coordinates(**kwargs) + pd.testing.assert_frame_equal(actual, expected) + def test_count_geometries(self): s = GeoSeries( [ diff --git a/python/tests/geopandas/test_match_geopandas_series.py b/python/tests/geopandas/test_match_geopandas_series.py index 2c48841b93c..c3f60a58b86 100644 --- a/python/tests/geopandas/test_match_geopandas_series.py +++ b/python/tests/geopandas/test_match_geopandas_series.py @@ -631,6 +631,16 @@ def test_count_coordinates(self): gpd_result = gpd.GeoSeries(geom).count_coordinates() self.check_pd_series_equal(sgpd_result, gpd_result) + @pytest.mark.skipif( + parse_version(gpd.__version__) < parse_version("0.13.0"), + reason="geopandas get_coordinates requires version 0.13.0 or higher", + ) + def test_get_coordinates(self): + for geometries in self.geoms: + sgpd_result = GeoSeries(geometries).get_coordinates() + gpd_result = gpd.GeoSeries(geometries).get_coordinates() + pd.testing.assert_frame_equal(sgpd_result.to_pandas(), gpd_result) + def test_count_geometries(self): for geom in self.geoms: sgpd_result = GeoSeries(geom).count_geometries() From 16cb2a3c70ad8292ba0d63325d72ac0e0be50438 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 23 Jul 2026 23:29:27 -0700 Subject: [PATCH 03/11] [GH-3156][GH-3157] Share distributed geometry expansion --- python/sedona/spark/geopandas/geoseries.py | 270 ++++++++++----------- python/tests/geopandas/test_geoseries.py | 27 ++- 2 files changed, 156 insertions(+), 141 deletions(-) diff --git a/python/sedona/spark/geopandas/geoseries.py b/python/sedona/spark/geopandas/geoseries.py index 60a4a2930fe..09bb78e0e63 100644 --- a/python/sedona/spark/geopandas/geoseries.py +++ b/python/sedona/spark/geopandas/geoseries.py @@ -691,6 +691,98 @@ def _query_geometry_column( result = GeoSeries(ps_series) if returns_geom else ps_series return result + def _expand_geometry_array( + self, + array_builder, + ignore_index: bool, + index_parts: bool, + temp_prefix: str, + ): + """Expand a geometry-array expression while preserving index metadata.""" + internal = self._internal.resolved_copy + source_sdf = internal.spark_frame + reserved_names = set(source_sdf.columns) + + def temp_column_name(base: str) -> str: + suffix = 0 + candidate = f"__{temp_prefix}_{base}__" + while candidate in reserved_names: + suffix += 1 + candidate = f"__{temp_prefix}_{base}_{suffix}__" + reserved_names.add(candidate) + return typing.cast(str, verify_temp_column_name(source_sdf, candidate)) + + index_column_names = [ + temp_column_name(f"index_{level}") + for level in range(len(internal.index_spark_columns)) + ] + parent_order_col = temp_column_name("parent_order") + position_col = temp_column_name("position") + value_col = temp_column_name("value") + sequence_col = temp_column_name("sequence") + + expanded_sdf = source_sdf.select( + *[ + column.alias(name) + for column, name in zip( + internal.index_spark_columns, index_column_names + ) + ], + scol_for(source_sdf, NATURAL_ORDER_COLUMN_NAME).alias(parent_order_col), + F.posexplode(array_builder(internal.data_spark_columns[0])).alias( + position_col, value_col + ), + ).orderBy(parent_order_col, position_col) + + # The distributed sequence supplies both ignore_index and a stable + # natural-order column without a single-partition row-number window. + expanded_sdf = InternalFrame.attach_distributed_sequence_column( + expanded_sdf, sequence_col + ) + + if ignore_index: + output_index_cols = [SPARK_DEFAULT_INDEX_NAME] + index_names = [None] + index_fields = None + index_expressions = [ + scol_for(expanded_sdf, sequence_col).alias(SPARK_DEFAULT_INDEX_NAME) + ] + else: + output_index_cols = list(index_column_names) + index_names = list(internal.index_names) + index_fields = [ + field.copy(name=name) + for field, name in zip(internal.index_fields, output_index_cols) + ] + index_expressions = [ + scol_for(expanded_sdf, name) for name in output_index_cols + ] + + if index_parts: + part_index_col = SPARK_INDEX_NAME_FORMAT(len(output_index_cols)) + output_index_cols.append(part_index_col) + index_names.append(None) + index_fields.append(None) + index_expressions.append( + scol_for(expanded_sdf, position_col) + .cast("long") + .alias(part_index_col) + ) + + output_sdf = expanded_sdf.select( + *index_expressions, + scol_for(expanded_sdf, value_col), + scol_for(expanded_sdf, sequence_col).alias(NATURAL_ORDER_COLUMN_NAME), + ) + return ( + internal, + output_sdf, + output_index_cols, + index_names, + index_fields, + value_col, + ) + # ============================================================================ # CONVERSION AND SERIALIZATION METHODS # ============================================================================ @@ -872,95 +964,49 @@ def get_coordinates( *, include_m=False, ) -> pspd.DataFrame: - source_frame = self._internal.spark_frame - source_order_name = verify_temp_column_name( - source_frame, "__coordinate_source_order__" - ) - position_name = verify_temp_column_name(source_frame, "__coordinate_position__") - point_name = verify_temp_column_name(source_frame, "__coordinate_point__") - - index_column_names = [ - SPARK_INDEX_NAME_FORMAT(i) - for i in range(len(self._internal.index_spark_columns)) - ] - index_columns = [ - column.alias(name) - for column, name in zip( - self._internal.index_spark_columns, index_column_names - ) - ] - - exploded_frame = source_frame.select( - *index_columns, - scol_for(source_frame, NATURAL_ORDER_COLUMN_NAME).alias(source_order_name), - F.posexplode(stf.ST_DumpPoints(self.spark.column)).alias( - position_name, point_name - ), - ).orderBy(source_order_name, position_name) - - sequence_name = verify_temp_column_name(exploded_frame, "__coordinate_order__") - sequenced_frame = InternalFrame.attach_distributed_sequence_column( - exploded_frame, sequence_name + ( + _, + expanded_frame, + index_column_names, + index_names, + index_fields, + point_name, + ) = self._expand_geometry_array( + stf.ST_DumpPoints, + ignore_index=ignore_index, + index_parts=index_parts, + temp_prefix="coordinates", ) coordinate_names = ["x", "y"] coordinate_columns = [ - stf.ST_X(scol_for(sequenced_frame, point_name)).alias("x"), - stf.ST_Y(scol_for(sequenced_frame, point_name)).alias("y"), + stf.ST_X(scol_for(expanded_frame, point_name)).alias("x"), + stf.ST_Y(scol_for(expanded_frame, point_name)).alias("y"), ] if include_z: coordinate_names.append("z") coordinate_columns.append( - stf.ST_Z(scol_for(sequenced_frame, point_name)).alias("z") + stf.ST_Z(scol_for(expanded_frame, point_name)).alias("z") ) if include_m: coordinate_names.append("m") coordinate_columns.append( - stf.ST_M(scol_for(sequenced_frame, point_name)).alias("m") + stf.ST_M(scol_for(expanded_frame, point_name)).alias("m") ) - if ignore_index: - result_index_names = [SPARK_DEFAULT_INDEX_NAME] - result_index_labels = [None] - result_index_fields = None - result_index_columns = [ - scol_for(sequenced_frame, sequence_name).alias(SPARK_DEFAULT_INDEX_NAME) - ] - else: - result_index_names = index_column_names - result_index_labels = list(self._internal.index_names) - result_index_fields = [ - field.copy(name=name) - for field, name in zip(self._internal.index_fields, index_column_names) - ] - result_index_columns = [ - scol_for(sequenced_frame, name) for name in index_column_names - ] - - if index_parts: - part_index_name = SPARK_INDEX_NAME_FORMAT(len(result_index_names)) - result_index_names.append(part_index_name) - result_index_labels.append(None) - result_index_fields.append(None) - result_index_columns.append( - scol_for(sequenced_frame, position_name) - .cast("long") - .alias(part_index_name) - ) - - result_frame = sequenced_frame.select( - *result_index_columns, + result_frame = expanded_frame.select( + *[scol_for(expanded_frame, name) for name in index_column_names], *coordinate_columns, - scol_for(sequenced_frame, sequence_name).alias(NATURAL_ORDER_COLUMN_NAME), + scol_for(expanded_frame, NATURAL_ORDER_COLUMN_NAME), ) internal = InternalFrame( spark_frame=result_frame, index_spark_columns=[ - scol_for(result_frame, name) for name in result_index_names + scol_for(result_frame, name) for name in index_column_names ], - index_names=result_index_labels, - index_fields=result_index_fields, + index_names=index_names, + index_fields=index_fields, column_labels=[(name,) for name in coordinate_names], data_spark_columns=[ scol_for(result_frame, name) for name in coordinate_names @@ -2931,80 +2977,26 @@ def explode(self, ignore_index=False, index_parts=False) -> "GeoSeries": dtype: geometry """ from pyspark.pandas.internal import InternalField - from pyspark.pandas.utils import verify_temp_column_name - - internal = self._internal.resolved_copy - source_sdf = internal.spark_frame - - def temp_column_name(base: str) -> str: - suffix = 0 - candidate = f"__explode_{base}__" - while candidate in source_sdf.columns: - suffix += 1 - candidate = f"__explode_{base}_{suffix}__" - return typing.cast(str, verify_temp_column_name(source_sdf, candidate)) - - parent_order_col = temp_column_name("parent_order") - part_index_col = temp_column_name("part_index") - geometry_col = temp_column_name("geometry") - sequence_col = temp_column_name("sequence") - - exploded_sdf = source_sdf.select( - *internal.index_spark_columns, - scol_for(source_sdf, NATURAL_ORDER_COLUMN_NAME).alias(parent_order_col), - F.posexplode(stf.ST_Dump(internal.data_spark_columns[0])).alias( - part_index_col, geometry_col - ), - ).orderBy(parent_order_col, part_index_col) - # Use pandas-on-Spark's distributed sequence implementation instead of - # a global row-number window. Besides serving as the ignored index, this - # provides a natural-order column for subsequent operations. - exploded_sdf = InternalFrame.attach_distributed_sequence_column( - exploded_sdf, sequence_col + ( + internal, + expanded_sdf, + output_index_cols, + index_names, + index_fields, + geometry_col, + ) = self._expand_geometry_array( + stf.ST_Dump, + ignore_index=ignore_index, + index_parts=index_parts, + temp_prefix="explode", ) - data_col = internal.data_spark_column_names[0] - if ignore_index: - output_index_cols = [SPARK_DEFAULT_INDEX_NAME] - index_names = [None] - output_sdf = exploded_sdf.select( - scol_for(exploded_sdf, sequence_col).alias(SPARK_DEFAULT_INDEX_NAME), - scol_for(exploded_sdf, geometry_col).alias(data_col), - scol_for(exploded_sdf, sequence_col).alias(NATURAL_ORDER_COLUMN_NAME), - ) - index_fields = [ - InternalField.from_struct_field( - output_sdf.schema[SPARK_DEFAULT_INDEX_NAME] - ) - ] - else: - output_index_cols = list(internal.index_spark_column_names) - index_names = list(internal.index_names) - index_fields = list(internal.index_fields) - index_expressions = [ - scol_for(exploded_sdf, name) for name in output_index_cols - ] - - if index_parts: - output_index_cols.append(part_index_col) - index_names.append(None) - index_expressions.append( - scol_for(exploded_sdf, part_index_col) - .cast("long") - .alias(part_index_col) - ) - - output_sdf = exploded_sdf.select( - *index_expressions, - scol_for(exploded_sdf, geometry_col).alias(data_col), - scol_for(exploded_sdf, sequence_col).alias(NATURAL_ORDER_COLUMN_NAME), - ) - - if index_parts: - index_fields.append( - InternalField.from_struct_field(output_sdf.schema[part_index_col]) - ) + output_sdf = expanded_sdf.select( + *[scol_for(expanded_sdf, name) for name in output_index_cols], + scol_for(expanded_sdf, geometry_col).alias(data_col), + scol_for(expanded_sdf, NATURAL_ORDER_COLUMN_NAME), + ) result_internal = internal.copy( spark_frame=output_sdf, diff --git a/python/tests/geopandas/test_geoseries.py b/python/tests/geopandas/test_geoseries.py index 9a2ed014380..8a575bc3fa5 100644 --- a/python/tests/geopandas/test_geoseries.py +++ b/python/tests/geopandas/test_geoseries.py @@ -450,6 +450,9 @@ def test_explode(self, kwargs): check_crs=True, ) pd.testing.assert_index_equal(actual.index, expected.index, exact=False) + pd.testing.assert_series_equal( + result.is_empty.to_pandas(), expected.is_empty, check_index_type=False + ) def test_to_crs(self): from pyproj import CRS @@ -917,6 +920,15 @@ def test_get_coordinates(self): dataframe_result.to_pandas(), expected_series.get_coordinates() ) + coordinate_named_index = gpd.GeoSeries( + [Point(1, 2), Point(3, 4)], + index=pd.Index([10, 20], name="x"), + ) + pd.testing.assert_frame_equal( + GeoSeries(coordinate_named_index).get_coordinates().to_pandas(), + coordinate_named_index.get_coordinates(), + ) + empty_series = gpd.GeoSeries( [Point(), GeometryCollection(), None], index=pd.Index([3, 2, 1], name="feature_id"), @@ -947,11 +959,23 @@ def test_get_coordinates_multi_index(self): expected = expected_series.get_coordinates(**kwargs) pd.testing.assert_frame_equal(actual, expected) + def test_get_coordinates_z(self): + geometries_wkt = [ + "POINT (0 1)", + "POINT Z (2 3 4)", + ] + expected_series = gpd.GeoSeries.from_wkt(geometries_wkt) + actual_series = GeoSeries.from_wkt(geometries_wkt) + + actual = actual_series.get_coordinates(include_z=True).to_pandas() + expected = expected_series.get_coordinates(include_z=True) + pd.testing.assert_frame_equal(actual, expected) + @pytest.mark.skipif( parse_version(shapely.__version__) < parse_version("2.1.0"), reason="M coordinates require shapely>=2.1.0", ) - def test_get_coordinates_zm(self): + def test_get_coordinates_m(self): geometries_wkt = [ "POINT (0 1)", "POINT Z (2 3 4)", @@ -962,7 +986,6 @@ def test_get_coordinates_zm(self): actual_series = GeoSeries.from_wkt(geometries_wkt) for kwargs in ( - {"include_z": True}, {"include_m": True}, {"include_z": True, "include_m": True}, ): From fa11d29f9da6f92bf64e68c9528c53949a66f10d Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Thu, 23 Jul 2026 23:32:02 -0700 Subject: [PATCH 04/11] [GH-3157] Preserve CRS for empty explode results --- python/sedona/spark/geopandas/geoseries.py | 10 +++++++++- python/tests/geopandas/test_geoseries.py | 7 +++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/python/sedona/spark/geopandas/geoseries.py b/python/sedona/spark/geopandas/geoseries.py index 09bb78e0e63..fdc71e0610c 100644 --- a/python/sedona/spark/geopandas/geoseries.py +++ b/python/sedona/spark/geopandas/geoseries.py @@ -336,6 +336,7 @@ def __init__( self._anchor: GeoDataFrame self._col_label: Label self._sindex: SpatialIndex = None + self._empty_crs_source: typing.Optional["GeoSeries"] = None if isinstance( data, (GeoDataFrame, GeoSeries, PandasOnSparkSeries, PandasOnSparkDataFrame) @@ -454,6 +455,8 @@ def crs(self) -> Union["CRS", None]: from pyproj import CRS if self._is_empty(): + if self._empty_crs_source is not None: + return self._empty_crs_source.crs return None # F.first is non-deterministic, but it doesn't matter because all non-null values should be the same. @@ -3010,7 +3013,12 @@ def explode(self, ignore_index=False, index_parts=False) -> "GeoSeries": InternalField(np.dtype("object"), output_sdf.schema[data_col]) ], ) - return GeoSeries(first_series(PandasOnSparkDataFrame(result_internal))) + result = GeoSeries(first_series(PandasOnSparkDataFrame(result_internal))) + # An explode can remove every row even though its input carries an + # SRID. Keep a lazy reference to the input so CRS remains available + # without eagerly evaluating or collecting the result. + result._empty_crs_source = self + return result def to_crs( self, crs: Union[Any, None] = None, epsg: Union[int, None] = None diff --git a/python/tests/geopandas/test_geoseries.py b/python/tests/geopandas/test_geoseries.py index 8a575bc3fa5..47deb73fb2e 100644 --- a/python/tests/geopandas/test_geoseries.py +++ b/python/tests/geopandas/test_geoseries.py @@ -454,6 +454,13 @@ def test_explode(self, kwargs): result.is_empty.to_pandas(), expected.is_empty, check_index_type=False ) + all_empty = GeoSeries( + [MultiPoint(), GeometryCollection(), None], + crs="EPSG:4326", + ).explode(**kwargs) + assert all_empty.crs is not None + assert all_empty.crs.to_epsg() == 4326 + def test_to_crs(self): from pyproj import CRS From 71a57939a54636b88172ddb644aa31ba5b051f5a Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 24 Jul 2026 00:02:23 -0700 Subject: [PATCH 05/11] [GH-3158] Implement GeoSeries geom_equals_exact --- .../org/apache/sedona/common/Predicates.java | 5 + .../apache/sedona/common/PredicatesTest.java | 36 ++ docs/api/sql/Geometry-Functions.md | 1 + docs/api/sql/Predicates/ST_EqualsExact.md | 62 +++ python/sedona/spark/geopandas/base.py | 67 +++ python/sedona/spark/geopandas/geoseries.py | 433 +++++++++++++++++- python/sedona/spark/sql/st_predicates.py | 23 + python/tests/geopandas/test_geoseries.py | 284 ++++++++++++ .../geopandas/test_match_geopandas_series.py | 13 + python/tests/sql/test_dataframe_api.py | 14 + python/tests/sql/test_predicate.py | 17 + .../org/apache/sedona/sql/UDF/Catalog.scala | 1 + .../sedona_sql/expressions/Predicates.scala | 13 + .../expressions/st_predicates.scala | 5 + .../sedona/sql/dataFrameAPITestScala.scala | 11 + .../sedona/sql/predicateTestScala.scala | 33 +- 16 files changed, 1014 insertions(+), 4 deletions(-) create mode 100644 docs/api/sql/Predicates/ST_EqualsExact.md diff --git a/common/src/main/java/org/apache/sedona/common/Predicates.java b/common/src/main/java/org/apache/sedona/common/Predicates.java index 916c6204df6..0511f4a2c49 100644 --- a/common/src/main/java/org/apache/sedona/common/Predicates.java +++ b/common/src/main/java/org/apache/sedona/common/Predicates.java @@ -158,6 +158,11 @@ public static boolean equals(Geometry leftGeometry, Geometry rightGeometry) { return leftGeometry.equalsTopo(rightGeometry); } + public static boolean equalsExact( + Geometry leftGeometry, Geometry rightGeometry, double tolerance) { + return leftGeometry.equalsExact(rightGeometry, tolerance); + } + public static boolean disjoint(Geometry leftGeometry, Geometry rightGeometry) { return leftGeometry.disjoint(rightGeometry); } diff --git a/common/src/test/java/org/apache/sedona/common/PredicatesTest.java b/common/src/test/java/org/apache/sedona/common/PredicatesTest.java index 1aac6ee5532..d4214cd9f68 100644 --- a/common/src/test/java/org/apache/sedona/common/PredicatesTest.java +++ b/common/src/test/java/org/apache/sedona/common/PredicatesTest.java @@ -338,6 +338,42 @@ public void testEqualsEmptyGeometries() throws ParseException { assertFalse(Predicates.equals(pointEmpty, point)); } + @Test + public void testEqualsExact() throws ParseException { + Geometry origin = geomFromEWKT("POINT(0 0)"); + Geometry nearby = geomFromEWKT("POINT(0.03 0.04)"); + assertTrue(Predicates.equalsExact(origin, origin, 0.0)); + assertTrue(Predicates.equalsExact(origin, nearby, 0.05)); + assertTrue(Predicates.equalsExact(origin, nearby, 0.051)); + assertFalse(Predicates.equalsExact(origin, nearby, 0.049)); + assertFalse(Predicates.equalsExact(origin, origin, -1.0)); + assertFalse(Predicates.equalsExact(origin, origin, Double.NaN)); + assertTrue(Predicates.equalsExact(origin, nearby, Double.POSITIVE_INFINITY)); + + Geometry line = geomFromEWKT("LINESTRING(0 0, 1 1, 2 0)"); + Geometry reversed = geomFromEWKT("LINESTRING(2 0, 1 1, 0 0)"); + assertFalse(Predicates.equalsExact(line, reversed, 0.0)); + + Geometry pointZ1 = geomFromEWKT("POINT Z (1 2 3)"); + Geometry pointZ2 = geomFromEWKT("POINT Z (1 2 99)"); + assertTrue(Predicates.equalsExact(pointZ1, pointZ2, 0.0)); + + Geometry pointM1 = geomFromEWKT("POINT M (1 2 3)"); + Geometry pointM2 = geomFromEWKT("POINT M (1 2 99)"); + assertTrue(Predicates.equalsExact(pointM1, pointM2, 0.0)); + } + + @Test + public void testEqualsExactEmptyAndCollectionStructure() throws ParseException { + Geometry pointEmpty = geomFromEWKT("POINT EMPTY"); + assertTrue(Predicates.equalsExact(pointEmpty, geomFromEWKT("POINT EMPTY"), 0.0)); + assertFalse(Predicates.equalsExact(pointEmpty, geomFromEWKT("POLYGON EMPTY"), 0.0)); + + Geometry collection = geomFromEWKT("GEOMETRYCOLLECTION(POINT(0 0), LINESTRING(0 0, 1 1))"); + Geometry reordered = geomFromEWKT("GEOMETRYCOLLECTION(LINESTRING(0 0, 1 1), POINT(0 0))"); + assertFalse(Predicates.equalsExact(collection, reordered, 0.0)); + } + @Test public void testRelateBoolean() throws ParseException { Geometry geom1 = geomFromEWKT("POINT(1 2)"); diff --git a/docs/api/sql/Geometry-Functions.md b/docs/api/sql/Geometry-Functions.md index 9e02d382d40..33b4366b403 100644 --- a/docs/api/sql/Geometry-Functions.md +++ b/docs/api/sql/Geometry-Functions.md @@ -168,6 +168,7 @@ These functions test spatial relationships between geometries, returning boolean | [ST_DWithin](Predicates/ST_DWithin.md) | Boolean | Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. | v1.5.1 | | [ST_3DDWithin](Predicates/ST_3DDWithin.md) | Boolean | Returns true if A and B are within a specified 3D Euclidean 'distance'. Accepts Geometry or Box3D inputs. | v1.9.1 | | [ST_Equals](Predicates/ST_Equals.md) | Boolean | Return true if A equals to B | v1.0.0 | +| [ST_EqualsExact](Predicates/ST_EqualsExact.md) | Boolean | Return true if A and B have matching structures and corresponding coordinates within a tolerance | v1.9.1 | | [ST_Intersects](Predicates/ST_Intersects.md) | Boolean | Return true if A intersects B | v1.0.0 | | [ST_OrderingEquals](Predicates/ST_OrderingEquals.md) | Boolean | Returns true if the geometries are equal and the coordinates are in the same order | v1.2.1 | | [ST_Overlaps](Predicates/ST_Overlaps.md) | Boolean | Return true if A overlaps B | v1.0.0 | diff --git a/docs/api/sql/Predicates/ST_EqualsExact.md b/docs/api/sql/Predicates/ST_EqualsExact.md new file mode 100644 index 00000000000..ab274a5f78d --- /dev/null +++ b/docs/api/sql/Predicates/ST_EqualsExact.md @@ -0,0 +1,62 @@ + + +# ST_EqualsExact + +Introduction: Return true if A and B have the same structure and their corresponding coordinates are equal within a tolerance. + +Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores z and m coordinates. + +Format: `ST_EqualsExact (A: Geometry, B: Geometry, tolerance: Double)` + +Return type: `Boolean` + +Since: `v1.9.1` + +SQL Example + +```sql +SELECT ST_EqualsExact( + ST_GeomFromWKT('POINT (0 0)'), + ST_GeomFromWKT('POINT (0.03 0.04)'), + 0.05 +) +``` + +Output: + +``` +true +``` + +The order of coordinates must match: + +```sql +SELECT ST_EqualsExact( + ST_GeomFromWKT('LINESTRING (0 0, 1 1)'), + ST_GeomFromWKT('LINESTRING (1 1, 0 0)'), + 0.0 +) +``` + +Output: + +``` +false +``` diff --git a/python/sedona/spark/geopandas/base.py b/python/sedona/spark/geopandas/base.py index a8eaf28bc37..f85a4eea2c4 100644 --- a/python/sedona/spark/geopandas/base.py +++ b/python/sedona/spark/geopandas/base.py @@ -2948,6 +2948,73 @@ def geom_equals(self, other, align=None): """ return _delegate_to_geometry_column("geom_equals", self, other, align) + def geom_equals_exact(self, other, tolerance, align=None): + """Return ``True`` for geometries that equal aligned `other` to a + given tolerance, otherwise ``False``. + + Equality is structural: geometry types, component ordering, ring + ordering, and vertex ordering must match. Corresponding x and y + coordinates may differ by at most ``tolerance``. Z and M coordinates + are ignored. + + The operation works in a 1-to-1 row-wise manner. + + Parameters + ---------- + other : GeoSeries or geometric object + The GeoSeries (elementwise) or geometric object to compare to. + tolerance : float + Maximum distance allowed between corresponding coordinates. + align : bool | None (default None) + If True, automatically align GeoSeries based on their indices. + If False, compare values in their existing order. None defaults + to True. + + Returns + ------- + Series (bool) + + Examples + -------- + >>> from sedona.spark.geopandas import GeoSeries + >>> from shapely.geometry import Point + >>> s = GeoSeries( + ... [ + ... Point(0, 1.1), + ... Point(0, 1.0), + ... Point(0, 1.2), + ... ] + ... ) + >>> s.geom_equals_exact(Point(0, 1), tolerance=0.1) + 0 False + 1 True + 2 False + dtype: bool + + >>> s.geom_equals_exact(Point(0, 1), tolerance=0.15) + 0 True + 1 True + 2 False + dtype: bool + + Notes + ----- + This method checks geometries row by row; it does not compare each + geometry with every value in `other`. + + As elsewhere in Sedona's GeoPandas compatibility layer, standalone + ``LinearRing`` geometries are serialized as ``LineString`` geometries. + Consequently, this method cannot distinguish those two standalone + input types when their coordinates match. + + See also + -------- + GeoSeries.geom_equals + """ + return _delegate_to_geometry_column( + "geom_equals_exact", self, other, tolerance, align + ) + def interpolate(self, distance, normalized=False): """Return a point at the specified distance along each geometry. diff --git a/python/sedona/spark/geopandas/geoseries.py b/python/sedona/spark/geopandas/geoseries.py index fdc71e0610c..f22d996cc4b 100644 --- a/python/sedona/spark/geopandas/geoseries.py +++ b/python/sedona/spark/geopandas/geoseries.py @@ -17,6 +17,7 @@ import sys import typing +import warnings from typing import Any, Union, Literal, List import numpy as np @@ -27,7 +28,7 @@ import pyspark from pyspark.pandas import Series as PandasOnSparkSeries from pyspark.pandas.frame import DataFrame as PandasOnSparkDataFrame -from pyspark.pandas.internal import InternalFrame +from pyspark.pandas.internal import InternalField, InternalFrame from pyspark.pandas.series import first_series from pyspark.pandas.utils import scol_for, verify_temp_column_name from pyspark.sql.types import NullType @@ -100,8 +101,8 @@ } -def _normalize_affine_scalar(value, error_message: str) -> float: - """Normalize an operation-wide affine parameter to a Python float.""" +def _normalize_numeric_scalar(value, error_message: str) -> float: + """Normalize an operation-wide numeric parameter to a Python float.""" if ( value is None or isinstance(value, (str, bytes, bytearray)) @@ -114,6 +115,11 @@ def _normalize_affine_scalar(value, error_message: str) -> float: raise TypeError(error_message) from exc +def _normalize_affine_scalar(value, error_message: str) -> float: + """Normalize an operation-wide affine parameter to a Python float.""" + return _normalize_numeric_scalar(value, error_message) + + def _interpret_origin(geometry: PySparkColumn, origin, with_z: bool): """Resolve a local affinity origin to distributed coordinate columns.""" if isinstance(origin, str): @@ -1866,6 +1872,40 @@ def geom_equals(self, other, align=None) -> pspd.Series: ) return _to_bool(result) + def geom_equals_exact(self, other, tolerance, align=None) -> pspd.Series: + tolerance = _normalize_numeric_scalar( + tolerance, "'tolerance' must be a numeric scalar" + ) + + if isinstance(other, BaseGeometry): + other_geometry = stc.ST_GeomFromWKB(F.lit(other.wkb)) + spark_expr = stp.ST_EqualsExact( + self.spark.column, other_geometry, tolerance + ) + result = self._boolean_result_preserving_index( + F.coalesce(spark_expr, F.lit(False)), + self._internal.spark_frame, + self._internal.index_spark_columns, + self._internal.index_fields, + self._internal.index_names, + ) + return _to_bool(result) + + if not isinstance(other, (GeoSeries, GeoDataFrame, PandasOnSparkSeries)): + raise TypeError( + "'other' must be a GeoSeries, GeoDataFrame, " + "pandas-on-Spark Series, or geometry" + ) + + other_series, extended = self._make_series_of_val(other) + align = False if extended else align + + return self._geom_equals_exact_series( + other_series, + tolerance, + align, + ) + def interpolate(self, distance, normalized=False) -> "GeoSeries": other_series, extended = self._make_series_of_val(distance) align = not extended @@ -1953,6 +1993,393 @@ def snap(self, other, tolerance, align=None) -> "GeoSeries": ) return result + def _boolean_result_preserving_index( + self, + spark_col: PySparkColumn, + df: pyspark.sql.DataFrame, + index_spark_columns: List[PySparkColumn], + index_fields: List, + index_names: List, + ) -> pspd.Series: + """Build a boolean Series while retaining every source index level.""" + result_index_names = [ + f"__index_level_{level}__" for level in range(len(index_spark_columns)) + ] + result_name = SPARK_DEFAULT_SERIES_NAME + sdf = df.select( + spark_col.alias(result_name), + *[ + index_col.alias(result_index_name) + for index_col, result_index_name in zip( + index_spark_columns, result_index_names + ) + ], + scol_for(df, NATURAL_ORDER_COLUMN_NAME), + ) + + schema_fields = {field.name: field for field in sdf.schema.fields} + result_index_fields = [ + source_field.copy( + name=result_index_name, + spark_type=schema_fields[result_index_name].dataType, + nullable=schema_fields[result_index_name].nullable, + metadata=schema_fields[result_index_name].metadata, + ) + for source_field, result_index_name in zip(index_fields, result_index_names) + ] + result_data_field = InternalField.from_struct_field(schema_fields[result_name]) + internal = InternalFrame( + spark_frame=sdf, + index_spark_columns=[ + scol_for(sdf, result_index_name) + for result_index_name in result_index_names + ], + index_names=index_names, + index_fields=result_index_fields, + column_labels=[(result_name,)], + data_spark_columns=[scol_for(sdf, result_name)], + data_fields=[result_data_field], + column_label_names=[None], + ) + return first_series(PandasOnSparkDataFrame(internal)).rename(None) + + def _geom_equals_exact_series( + self, + other: pspd.Series, + tolerance: float, + align: Union[bool, None], + ) -> pspd.Series: + """Execute exact equality with GeoPandas-compatible row alignment.""" + position_col = "__geom_equals_exact_position__" + left_present_col = "__geom_equals_exact_left_present__" + right_present_col = "__geom_equals_exact_right_present__" + left_order_col = "__geom_equals_exact_left_order__" + right_order_col = "__geom_equals_exact_right_order__" + + left_index_aliases = [ + f"__geom_equals_exact_left_index_{level}__" + for level in range(len(self._internal.index_spark_columns)) + ] + right_index_aliases = [ + f"__geom_equals_exact_right_index_{level}__" + for level in range(len(other._internal.index_spark_columns)) + ] + + left_frame = self._internal.spark_frame.select( + self.spark.column.alias("L"), + *[ + index_col.alias(alias) + for index_col, alias in zip( + self._internal.index_spark_columns, left_index_aliases + ) + ], + scol_for(self._internal.spark_frame, NATURAL_ORDER_COLUMN_NAME).alias( + left_order_col + ), + F.lit(True).alias(left_present_col), + ) + right_frame = other._internal.spark_frame.select( + other.spark.column.alias("R"), + *[ + index_col.alias(alias) + for index_col, alias in zip( + other._internal.index_spark_columns, right_index_aliases + ) + ], + scol_for(other._internal.spark_frame, NATURAL_ORDER_COLUMN_NAME).alias( + right_order_col + ), + F.lit(True).alias(right_present_col), + ) + left_frame = left_frame.orderBy(left_order_col) + right_frame = right_frame.orderBy(right_order_col) + left_frame = InternalFrame.attach_distributed_sequence_column( + left_frame, position_col + ) + right_frame = InternalFrame.attach_distributed_sequence_column( + right_frame, position_col + ) + positional_join = left_frame.join(right_frame, on=position_col, how="outer") + + missing_side = ( + F.col(left_present_col).isNull() | F.col(right_present_col).isNull() + ) + same_index_structure = len(left_index_aliases) == len(right_index_aliases) + if same_index_structure: + index_mismatch = missing_side + for left_index, right_index in zip(left_index_aliases, right_index_aliases): + index_mismatch = index_mismatch | ~F.col(left_index).eqNullSafe( + F.col(right_index) + ) + else: + index_mismatch = F.lit(True) + + status = ( + positional_join.select( + F.col(left_present_col), + F.col(right_present_col), + index_mismatch.alias("__index_mismatch__"), + ) + .agg( + F.count(F.col(left_present_col)).alias("left_count"), + F.count(F.col(right_present_col)).alias("right_count"), + F.max(F.col("__index_mismatch__").cast("int")).alias("index_mismatch"), + ) + .first() + ) + left_count = status["left_count"] + right_count = status["right_count"] + lengths_match = left_count == right_count + indices_match = ( + same_index_structure + and lengths_match + and not bool(status["index_mismatch"] or False) + ) + + if align is False and not lengths_match: + raise ValueError( + "Lengths of inputs do not match. " + f"Left: {left_count}, Right: {right_count}" + ) + + if align is None and not indices_match: + warnings.warn( + "The indices of the left and right GeoSeries' are not equal, " + "and therefore they will be aligned (reordering and/or " + "introducing missing values) before executing the operation. " + "If this alignment is the desired behaviour, you can silence " + "this warning by passing 'align=True'. If you don't want " + "alignment and protect yourself of accidentally aligning, " + "you can pass 'align=False'.", + stacklevel=3, + ) + + if align is False or indices_match: + result_index_columns = [ + f"__index_level_{level}__" for level in range(len(left_index_aliases)) + ] + aligned_frame = positional_join.select( + F.col("L"), + F.col("R"), + *[ + F.col(left_index).alias(result_index) + for left_index, result_index in zip( + left_index_aliases, result_index_columns + ) + ], + F.col(position_col).alias(NATURAL_ORDER_COLUMN_NAME), + ) + result_index_fields = self._internal.index_fields + result_index_names = self._internal.index_names + else: + left_index_names = self._internal.index_names + right_index_names = other._internal.index_names + left_level_count = len(left_index_aliases) + right_level_count = len(right_index_aliases) + + if left_level_count == right_level_count == 1: + join_pairs = [(0, 0)] + output_levels = [(0, 0)] + result_index_names = [ + ( + left_index_names[0] + if left_index_names[0] == right_index_names[0] + else None + ) + ] + join_how = "outer" + preserve_multiindex_order = False + elif ( + left_level_count == right_level_count + and left_index_names == right_index_names + ): + join_pairs = [(level, level) for level in range(left_level_count)] + output_levels = join_pairs + result_index_names = left_index_names + join_how = "outer" + preserve_multiindex_order = False + else: + left_name_positions = {} + right_name_positions = {} + for level, name in enumerate(left_index_names): + if name is not None: + left_name_positions.setdefault(name, []).append(level) + for level, name in enumerate(right_index_names): + if name is not None: + right_name_positions.setdefault(name, []).append(level) + + shared_names = [] + for name in left_index_names: + if ( + name is not None + and name in right_name_positions + and name not in shared_names + ): + shared_names.append(name) + if not shared_names: + raise ValueError("cannot join with no overlapping index names") + + for name in shared_names: + if ( + len(left_name_positions[name]) != 1 + or len(right_name_positions[name]) != 1 + ): + display_name = name[0] if len(name) == 1 else name + raise ValueError( + f"The name {display_name} occurs multiple times, " + "use a level number" + ) + + join_pairs = [ + ( + left_name_positions[name][0], + right_name_positions[name][0], + ) + for name in shared_names + ] + shared_right_levels = {right_level for _, right_level in join_pairs} + + if left_level_count == 1 or right_level_count == 1: + left_is_multiindex = left_level_count > 1 + multiindex_names = ( + left_index_names if left_is_multiindex else right_index_names + ) + output_levels = [] + for level in range(len(multiindex_names)): + if left_is_multiindex: + matching_right = next( + ( + right_level + for left_level, right_level in join_pairs + if left_level == level + ), + None, + ) + output_levels.append((level, matching_right)) + else: + matching_left = next( + ( + left_level + for left_level, right_level in join_pairs + if right_level == level + ), + None, + ) + output_levels.append((matching_left, level)) + result_index_names = multiindex_names + join_how = "left" if left_is_multiindex else "right" + preserve_multiindex_order = True + else: + right_for_left = { + left_level: right_level + for left_level, right_level in join_pairs + } + output_levels = [ + (left_level, right_for_left.get(left_level)) + for left_level in range(left_level_count) + ] + output_levels.extend( + (None, right_level) + for right_level in range(right_level_count) + if right_level not in shared_right_levels + ) + result_index_names = list(left_index_names) + result_index_names.extend( + right_index_names[right_level] + for right_level in range(right_level_count) + if right_level not in shared_right_levels + ) + join_how = "outer" + preserve_multiindex_order = False + + result_index_columns = [ + f"__index_level_{level}__" for level in range(len(output_levels)) + ] + + left_alias = left_frame.alias("left") + right_alias = right_frame.alias("right") + first_left_level, first_right_level = join_pairs[0] + join_condition = left_alias[ + left_index_aliases[first_left_level] + ].eqNullSafe(right_alias[right_index_aliases[first_right_level]]) + for left_level, right_level in join_pairs[1:]: + join_condition = join_condition & left_alias[ + left_index_aliases[left_level] + ].eqNullSafe(right_alias[right_index_aliases[right_level]]) + + joined_by_index = left_alias.join( + right_alias, on=join_condition, how=join_how + ) + result_index_expressions = [] + for (left_level, right_level), result_index in zip( + output_levels, result_index_columns + ): + if left_level is None: + index_expression = right_alias[right_index_aliases[right_level]] + elif right_level is None: + index_expression = left_alias[left_index_aliases[left_level]] + else: + index_expression = F.coalesce( + left_alias[left_index_aliases[left_level]], + right_alias[right_index_aliases[right_level]], + ) + result_index_expressions.append(index_expression.alias(result_index)) + + selected_frame = joined_by_index.select( + left_alias["L"].alias("L"), + right_alias["R"].alias("R"), + *result_index_expressions, + left_alias[position_col].alias(left_order_col), + right_alias[position_col].alias(right_order_col), + ) + if preserve_multiindex_order: + order_columns = ( + [F.col(left_order_col), F.col(right_order_col)] + if left_level_count > 1 + else [F.col(right_order_col), F.col(left_order_col)] + ) + else: + shared_result_levels = [ + output_levels.index(join_pair) for join_pair in join_pairs + ] + remaining_result_levels = [ + level + for level in range(len(result_index_columns)) + if level not in shared_result_levels + ] + order_columns = [ + F.col(result_index_columns[level]).asc_nulls_last() + for level in shared_result_levels + remaining_result_levels + ] + order_columns.extend( + [ + F.col(left_order_col).asc_nulls_last(), + F.col(right_order_col).asc_nulls_last(), + ] + ) + ordered_frame = selected_frame.orderBy(*order_columns) + aligned_frame = InternalFrame.attach_distributed_sequence_column( + ordered_frame.drop(left_order_col, right_order_col), + NATURAL_ORDER_COLUMN_NAME, + ) + aligned_schema = { + field.name: field for field in aligned_frame.schema.fields + } + result_index_fields = [ + InternalField.from_struct_field(aligned_schema[result_index]) + for result_index in result_index_columns + ] + + spark_expr = stp.ST_EqualsExact(F.col("L"), F.col("R"), tolerance) + result = self._boolean_result_preserving_index( + F.coalesce(spark_expr, F.lit(False)), + aligned_frame, + [scol_for(aligned_frame, name) for name in result_index_columns], + result_index_fields, + result_index_names, + ) + return _to_bool(result) + def _row_wise_operation( self, spark_col: PySparkColumn, diff --git a/python/sedona/spark/sql/st_predicates.py b/python/sedona/spark/sql/st_predicates.py index 89b9937d49b..b48cdd75790 100644 --- a/python/sedona/spark/sql/st_predicates.py +++ b/python/sedona/spark/sql/st_predicates.py @@ -91,6 +91,29 @@ def ST_Equals(a: ColumnOrName, b: ColumnOrName) -> Column: return _call_predicate_function("ST_Equals", (a, b)) +@validate_argument_types +def ST_EqualsExact( + a: ColumnOrName, + b: ColumnOrName, + tolerance: Union[ColumnOrName, float, int], +) -> Column: + """Check whether two geometries have the same structure and coordinate ordering + within a tolerance. + + The comparison uses x and y coordinates. Z and M coordinates are ignored. + + :param a: One geometry column to check. + :type a: ColumnOrName + :param b: Other geometry column to check. + :type b: ColumnOrName + :param tolerance: Maximum distance allowed between corresponding coordinates. + :type tolerance: ColumnOrName or float + :return: True if a and b are exactly equal within tolerance, otherwise False. + :rtype: Column + """ + return _call_predicate_function("ST_EqualsExact", (a, b, tolerance)) + + @validate_argument_types def ST_Intersects(a: ColumnOrName, b: ColumnOrName) -> Column: """Check whether a and b intersect. Polymorphic over input type: diff --git a/python/tests/geopandas/test_geoseries.py b/python/tests/geopandas/test_geoseries.py index 47deb73fb2e..5d453507b01 100644 --- a/python/tests/geopandas/test_geoseries.py +++ b/python/tests/geopandas/test_geoseries.py @@ -3785,6 +3785,290 @@ def test_geom_equals(self): expected = pd.Series([True, False, True]) self.check_pd_series_equal(df_result, expected) + def test_geom_equals_exact(self): + s = GeoSeries([Point(0, 1.1), Point(0, 1.0), Point(0, 1.2)]) + + result = s.geom_equals_exact(Point(0, 1), tolerance=0.1) + expected = gpd.GeoSeries( + [Point(0, 1.1), Point(0, 1.0), Point(0, 1.2)] + ).geom_equals_exact(Point(0, 1), tolerance=0.1) + self.check_pd_series_equal(result, expected) + + result = s.geom_equals_exact(Point(0, 1), tolerance=0.15) + expected = gpd.GeoSeries( + [Point(0, 1.1), Point(0, 1.0), Point(0, 1.2)] + ).geom_equals_exact(Point(0, 1), tolerance=0.15) + self.check_pd_series_equal(result, expected) + + df_result = s.to_geoframe().geom_equals_exact(Point(0, 1), tolerance=0.15) + self.check_pd_series_equal(df_result, expected) + + def test_geom_equals_exact_alignment(self): + left_geometries = [Point(0, 0), Point(1, 1), None] + right_geometries = [Point(1, 1), Point(0, 0), Point(9, 9)] + left_index = ["a", "b", "c"] + right_index = ["b", "a", "d"] + + left = GeoSeries(left_geometries, index=left_index) + right = GeoSeries(right_geometries, index=right_index) + expected_left = gpd.GeoSeries(left_geometries, index=left_index) + expected_right = gpd.GeoSeries(right_geometries, index=right_index) + + result = left.geom_equals_exact(right, tolerance=0) + expected = expected_left.geom_equals_exact( + expected_right, tolerance=0, align=True + ) + self.check_pd_series_equal(result, expected) + + result = left.geom_equals_exact(right, tolerance=0, align=True) + self.check_pd_series_equal(result, expected) + + result = left.geom_equals_exact(right, tolerance=0, align=False) + expected = expected_left.geom_equals_exact( + expected_right, tolerance=0, align=False + ) + self.check_pd_series_equal(result, expected) + + def test_geom_equals_exact_duplicate_index_alignment(self): + index = ["a", "a"] + left_geometries = [Point(0, 0), Point(1, 1)] + right_geometries = [Point(0, 0), Point(9, 9)] + + result = GeoSeries(left_geometries, index=index).geom_equals_exact( + GeoSeries(right_geometries, index=index), tolerance=0, align=True + ) + expected = gpd.GeoSeries(left_geometries, index=index).geom_equals_exact( + gpd.GeoSeries(right_geometries, index=index), + tolerance=0, + align=True, + ) + self.check_pd_series_equal(result, expected) + + def test_geom_equals_exact_unequal_duplicate_index_alignment(self): + left_index = ["a", "a", "c"] + right_index = ["a", "a", "b"] + left_geometries = [Point(0, 0), Point(1, 1), Point(2, 2)] + right_geometries = [Point(0, 0), Point(9, 9), Point(3, 3)] + + result = GeoSeries(left_geometries, index=left_index).geom_equals_exact( + GeoSeries(right_geometries, index=right_index), + tolerance=0, + align=True, + ) + expected = gpd.GeoSeries(left_geometries, index=left_index).geom_equals_exact( + gpd.GeoSeries(right_geometries, index=right_index), + tolerance=0, + align=True, + ) + self.check_pd_series_equal(result, expected) + assert len(result) == 6 + + def test_geom_equals_exact_align_false_requires_equal_lengths(self): + left = GeoSeries([Point(0, 0)]) + right = GeoSeries([Point(0, 0), Point(1, 1)]) + + with pytest.raises( + ValueError, + match=r"Lengths of inputs do not match\. Left: 1, Right: 2", + ): + left.geom_equals_exact(right, tolerance=0, align=False) + + def test_geom_equals_exact_preserves_multiindex(self): + left_index = pd.MultiIndex.from_tuples( + [("b", 2), ("a", 1)], names=["group", "row"] + ) + right_index = pd.MultiIndex.from_tuples( + [("a", 1), ("c", 3)], names=["group", "row"] + ) + left_geometries = [Point(2, 2), Point(1, 1)] + right_geometries = [Point(1, 1), Point(3, 3)] + + left = GeoSeries(left_geometries, index=left_index) + right = GeoSeries(right_geometries, index=right_index) + result = left.geom_equals_exact(right, tolerance=0, align=True) + expected = gpd.GeoSeries(left_geometries, index=left_index).geom_equals_exact( + gpd.GeoSeries(right_geometries, index=right_index), + tolerance=0, + align=True, + ) + self.check_pd_series_equal(result, expected) + + positional_result = left.geom_equals_exact(right, tolerance=0, align=False) + positional_expected = gpd.GeoSeries( + left_geometries, index=left_index + ).geom_equals_exact( + gpd.GeoSeries(right_geometries, index=right_index), + tolerance=0, + align=False, + ) + self.check_pd_series_equal(positional_result, positional_expected) + + scalar_result = left.geom_equals_exact(Point(1, 1), tolerance=0) + scalar_expected = gpd.GeoSeries( + left_geometries, index=left_index + ).geom_equals_exact(Point(1, 1), tolerance=0) + self.check_pd_series_equal(scalar_result, scalar_expected) + + duplicate_index = pd.MultiIndex.from_tuples( + [("a", 1), ("a", 1)], names=["group", "row"] + ) + duplicate_result = GeoSeries( + [Point(0, 0), Point(1, 1)], index=duplicate_index + ).geom_equals_exact( + GeoSeries([Point(0, 0), Point(9, 9)], index=duplicate_index), + tolerance=0, + align=True, + ) + duplicate_expected = gpd.GeoSeries( + [Point(0, 0), Point(1, 1)], index=duplicate_index + ).geom_equals_exact( + gpd.GeoSeries([Point(0, 0), Point(9, 9)], index=duplicate_index), + tolerance=0, + align=True, + ) + self.check_pd_series_equal(duplicate_result, duplicate_expected) + + def test_geom_equals_exact_aligns_multiindex_by_name(self): + left_index = pd.MultiIndex.from_tuples( + [("left-2", "b"), ("left-1", "a")], + names=["left_row", "group"], + ) + right_index = pd.MultiIndex.from_tuples( + [("a", "right-3"), ("c", "right-4")], + names=["group", "right_row"], + ) + left_geometries = [Point(2, 2), Point(1, 1)] + right_geometries = [Point(1, 1), Point(3, 3)] + + result = GeoSeries(left_geometries, index=left_index).geom_equals_exact( + GeoSeries(right_geometries, index=right_index), + tolerance=0, + align=True, + ) + expected = gpd.GeoSeries(left_geometries, index=left_index).geom_equals_exact( + gpd.GeoSeries(right_geometries, index=right_index), + tolerance=0, + align=True, + ) + self.check_pd_series_equal(result, expected) + + def test_geom_equals_exact_aligns_different_index_levels(self): + simple_index = pd.Index(["b", "a", "d"], name="group") + multiindex = pd.MultiIndex.from_tuples( + [("b", 2), ("a", 1), ("c", 3)], names=["group", "row"] + ) + simple_geometries = [Point(2, 2), Point(1, 1), Point(4, 4)] + multi_geometries = [Point(2, 2), Point(9, 9), Point(3, 3)] + + result = GeoSeries(simple_geometries, index=simple_index).geom_equals_exact( + GeoSeries(multi_geometries, index=multiindex), + tolerance=0, + align=True, + ) + expected = gpd.GeoSeries( + simple_geometries, index=simple_index + ).geom_equals_exact( + gpd.GeoSeries(multi_geometries, index=multiindex), + tolerance=0, + align=True, + ) + self.check_pd_series_equal(result, expected) + + reverse_result = GeoSeries( + multi_geometries, index=multiindex + ).geom_equals_exact( + GeoSeries(simple_geometries, index=simple_index), + tolerance=0, + align=True, + ) + reverse_expected = gpd.GeoSeries( + multi_geometries, index=multiindex + ).geom_equals_exact( + gpd.GeoSeries(simple_geometries, index=simple_index), + tolerance=0, + align=True, + ) + self.check_pd_series_equal(reverse_result, reverse_expected) + + def test_geom_equals_exact_rejects_unrelated_multiindex_names(self): + left_index = pd.MultiIndex.from_tuples( + [("a", 1)], names=["left_group", "left_row"] + ) + right_index = pd.MultiIndex.from_tuples( + [("b", 2)], names=["right_group", "right_row"] + ) + + with pytest.raises( + ValueError, match="cannot join with no overlapping index names" + ): + GeoSeries([Point(0, 0)], index=left_index).geom_equals_exact( + GeoSeries([Point(0, 0)], index=right_index), + tolerance=0, + align=True, + ) + + def test_geom_equals_exact_linearring_serialization_limitation(self): + ring = LinearRing([(0, 0), (1, 0), (1, 1), (0, 0)]) + line = LineString(ring.coords) + + # Sedona represents standalone LinearRings as LineStrings throughout + # the GeoPandas compatibility layer. + result = GeoSeries([ring]).geom_equals_exact(line, tolerance=0) + self.check_pd_series_equal(result, pd.Series([True])) + + def test_geom_equals_exact_structural_null_and_dimensions(self): + left_geometries = [ + Point(), + LineString(), + Polygon(), + None, + Point(1, 2, 3), + wkt.loads("POINT M (1 2 3)"), + LineString([(0, 0), (1, 1)]), + GeometryCollection([Point(0, 0), LineString([(0, 0), (1, 1)])]), + ] + right_geometries = [ + Point(), + Polygon(), + Polygon(), + None, + Point(1, 2, 99), + wkt.loads("POINT M (1 2 99)"), + LineString([(1, 1), (0, 0)]), + GeometryCollection([LineString([(0, 0), (1, 1)]), Point(0, 0)]), + ] + + result = GeoSeries(left_geometries).geom_equals_exact( + GeoSeries(right_geometries), tolerance=0, align=False + ) + expected = gpd.GeoSeries(left_geometries).geom_equals_exact( + gpd.GeoSeries(right_geometries), tolerance=0, align=False + ) + self.check_pd_series_equal(result, expected) + + @pytest.mark.parametrize("tolerance", [-1.0, np.nan, np.inf]) + def test_geom_equals_exact_special_tolerances(self, tolerance): + geometries = [Point(0, 0), Point(1, 1), None] + result = GeoSeries(geometries).geom_equals_exact( + Point(0, 0), tolerance=tolerance + ) + expected = gpd.GeoSeries(geometries).geom_equals_exact( + Point(0, 0), tolerance=tolerance + ) + self.check_pd_series_equal(result, expected) + + @pytest.mark.parametrize("tolerance", [None, "0.1", [0.1], np.array([0.1])]) + def test_geom_equals_exact_rejects_non_scalar_tolerance(self, tolerance): + s = GeoSeries([Point(0, 0)]) + with pytest.raises(TypeError, match="'tolerance' must be a numeric scalar"): + s.geom_equals_exact(Point(0, 0), tolerance=tolerance) + + @pytest.mark.parametrize("other", [None, 1, "POINT (0 0)", [Point(0, 0)]]) + def test_geom_equals_exact_rejects_non_geometry_other(self, other): + s = GeoSeries([Point(0, 0)]) + with pytest.raises(TypeError, match="'other' must be"): + s.geom_equals_exact(other, tolerance=0) + def test_interpolate(self): s = GeoSeries( [ diff --git a/python/tests/geopandas/test_match_geopandas_series.py b/python/tests/geopandas/test_match_geopandas_series.py index c3f60a58b86..03029a80d90 100644 --- a/python/tests/geopandas/test_match_geopandas_series.py +++ b/python/tests/geopandas/test_match_geopandas_series.py @@ -1760,6 +1760,19 @@ def test_geom_equals(self): ) self.check_pd_series_equal(sgpd_result, gpd_result) + @pytest.mark.parametrize("tolerance", [0.0, 0.25]) + def test_geom_equals_exact(self, tolerance): + geometries = [ + geometry for geometry_family in self.geoms for geometry in geometry_family + ] + sgpd_result = GeoSeries(geometries).geom_equals_exact( + GeoSeries(geometries), tolerance=tolerance, align=False + ) + gpd_result = gpd.GeoSeries(geometries).geom_equals_exact( + gpd.GeoSeries(geometries), tolerance=tolerance, align=False + ) + self.check_pd_series_equal(sgpd_result, gpd_result) + def test_interpolate(self): for geom in [self.linestrings, self.linearrings]: sgpd_result = GeoSeries(geom).interpolate(1.0) diff --git a/python/tests/sql/test_dataframe_api.py b/python/tests/sql/test_dataframe_api.py index f762f73a601..dc9c998d8f5 100644 --- a/python/tests/sql/test_dataframe_api.py +++ b/python/tests/sql/test_dataframe_api.py @@ -1251,6 +1251,17 @@ "", True, ), + ( + stp.ST_EqualsExact, + ( + lambda: f.expr("ST_Point(0.0, 0.0)"), + lambda: f.expr("ST_Point(0.03, 0.04)"), + 0.051, + ), + "triangle_geom", + "", + True, + ), (stp.ST_Intersects, ("a", "b"), "overlapping_polys", "", True), ( stp.ST_OrderingEquals, @@ -1636,6 +1647,9 @@ (stp.ST_Intersects, ("", None)), (stp.ST_OrderingEquals, (None, "")), (stp.ST_OrderingEquals, ("", None)), + (stp.ST_EqualsExact, (None, "", 0.0)), + (stp.ST_EqualsExact, ("", None, 0.0)), + (stp.ST_EqualsExact, ("", "", None)), (stp.ST_Overlaps, (None, "")), (stp.ST_Overlaps, ("", None)), (stp.ST_Touches, (None, "")), diff --git a/python/tests/sql/test_predicate.py b/python/tests/sql/test_predicate.py index a41cf16b3e9..49542c8f62a 100644 --- a/python/tests/sql/test_predicate.py +++ b/python/tests/sql/test_predicate.py @@ -309,6 +309,23 @@ def test_st_ordering_equals_ok(self): assert not not_order_equals_diff_geom.take(1)[0][0] assert not not_order_equals_diff_order.take(1)[0][0] + def test_st_equals_exact(self): + result = self.spark.sql(""" + SELECT + ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.051), + ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.049), + ST_EqualsExact( + ST_GeomFromWKT('LINESTRING(0 0, 1 1)'), + ST_GeomFromWKT('LINESTRING(1 1, 0 0)'), + 0.0 + ), + ST_EqualsExact(NULL, ST_Point(0.0, 0.0), 0.0) + """).first() + assert result[0] is True + assert result[1] is False + assert result[2] is False + assert result[3] is None + def test_st_dwithin(self): test_table = self.spark.sql( "select ST_GeomFromWKT('POINT (0 0)') as origin, ST_GeomFromWKT('POINT (2 0)') as point_1" diff --git a/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala b/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala index ef77a494fb7..7c1ec41f459 100644 --- a/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala +++ b/spark/common/src/main/scala/org/apache/sedona/sql/UDF/Catalog.scala @@ -172,6 +172,7 @@ object Catalog extends AbstractCatalog with Logging { function[ST_DWithin](), function[ST_3DDWithin](), function[ST_Equals](), + function[ST_EqualsExact](), function[ST_Intersects](), function[ST_OrderingEquals](), function[ST_Overlaps](), diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Predicates.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Predicates.scala index d4544207696..0e693f9351c 100644 --- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Predicates.scala +++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/Predicates.scala @@ -262,6 +262,19 @@ private[apache] case class ST_Equals(inputExpressions: Seq[Expression]) } } +/** + * Test if two geometries have the same structure and coordinate ordering within a tolerance. + * + * @param inputExpressions + */ +private[apache] case class ST_EqualsExact(inputExpressions: Seq[Expression]) + extends InferredExpression(inferrableFunction3(Predicates.equalsExact)) { + + protected def withNewChildrenInternal(newChildren: IndexedSeq[Expression]) = { + copy(inputExpressions = newChildren) + } +} + /** * Test if leftGeometry is disjoint from rightGeometry * diff --git a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_predicates.scala b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_predicates.scala index de7416bd2bb..33bc6898c4e 100644 --- a/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_predicates.scala +++ b/spark/common/src/main/scala/org/apache/spark/sql/sedona_sql/expressions/st_predicates.scala @@ -36,6 +36,11 @@ object st_predicates { def ST_Equals(a: Column, b: Column): Column = wrapExpression[ST_Equals](a, b) def ST_Equals(a: String, b: String): Column = wrapExpression[ST_Equals](a, b) + def ST_EqualsExact(a: Column, b: Column, tolerance: Column): Column = + wrapExpression[ST_EqualsExact](a, b, tolerance) + def ST_EqualsExact(a: String, b: String, tolerance: Double): Column = + wrapExpression[ST_EqualsExact](a, b, tolerance) + def ST_Intersects(a: Column, b: Column): Column = wrapExpression[ST_Intersects](a, b) def ST_Intersects(a: String, b: String): Column = wrapExpression[ST_Intersects](a, b) diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala b/spark/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala index 4734a58c29c..813463c5109 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/dataFrameAPITestScala.scala @@ -1891,6 +1891,17 @@ class dataFrameAPITestScala extends TestBaseScala { assert(!actualResult) } + it("Passed ST_EqualsExact") { + val baseDf = sparkSession.sql("SELECT ST_Point(0.0, 0.0) AS a, ST_Point(0.03, 0.04) AS b") + val result = baseDf + .select( + ST_EqualsExact("a", "b", 0.051).alias("within"), + ST_EqualsExact("a", "b", 0.049).alias("outside")) + .first() + assert(result.getBoolean(0)) + assert(!result.getBoolean(1)) + } + it("Passed ST_Covers") { val baseDf = sparkSession.sql( "SELECT ST_GeomFromWKT('POLYGON ((0 0, 1 0, 1 1, 0 0))') AS a, ST_Point(1.0, 0.0) AS b, ST_Point(0.0, 1.0) AS c") diff --git a/spark/common/src/test/scala/org/apache/sedona/sql/predicateTestScala.scala b/spark/common/src/test/scala/org/apache/sedona/sql/predicateTestScala.scala index dae5ba970f4..cfde3541b50 100644 --- a/spark/common/src/test/scala/org/apache/sedona/sql/predicateTestScala.scala +++ b/spark/common/src/test/scala/org/apache/sedona/sql/predicateTestScala.scala @@ -19,7 +19,7 @@ package org.apache.sedona.sql import org.apache.spark.sql.catalyst.expressions.{EmptyRow, Literal} -import org.apache.spark.sql.sedona_sql.expressions.{ST_Contains, ST_CoveredBy, ST_Covers, ST_Crosses, ST_DWithin, ST_Disjoint, ST_Equals, ST_Intersects, ST_OrderingEquals, ST_Overlaps, ST_Point, ST_Touches, ST_Within} +import org.apache.spark.sql.sedona_sql.expressions.{ST_Contains, ST_CoveredBy, ST_Covers, ST_Crosses, ST_DWithin, ST_Disjoint, ST_Equals, ST_EqualsExact, ST_Intersects, ST_OrderingEquals, ST_Overlaps, ST_Point, ST_Touches, ST_Within} class predicateTestScala extends TestBaseScala { @@ -355,6 +355,25 @@ class predicateTestScala extends TestBaseScala { assert(!notOrderEqualsDiffOrder.take(1)(0).get(0).asInstanceOf[Boolean]) } + it("Passed ST_EqualsExact") { + val result = sparkSession + .sql(""" + SELECT + ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.051), + ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.049), + ST_EqualsExact( + ST_GeomFromWKT('LINESTRING(0 0, 1 1)'), + ST_GeomFromWKT('LINESTRING(1 1, 0 0)'), + 0.0), + ST_EqualsExact(NULL, ST_Point(0.0, 0.0), 0.0) + """) + .first() + assert(result.getBoolean(0)) + assert(!result.getBoolean(1)) + assert(!result.getBoolean(2)) + assert(result.isNullAt(3)) + } + it("Passed edge cases of ST_Contains and ST_Covers") { val testtable = sparkSession.sql( "select ST_GeomFromWKT('POLYGON((2 0, 0 2, -2 0, 2 0))') AS a, ST_GeomFromWKT('POINT(2 0)') AS b") @@ -420,5 +439,17 @@ class predicateTestScala extends TestBaseScala { assert(predicate(missing :: missing :: Nil).eval(EmptyRow) == null) } } + + it("Passed null handling in ST_EqualsExact") { + val point = + ST_Point(Literal.create(0.0) :: Literal.create(0.0) :: Literal.create(0.0) :: Nil) + val missing = Literal.create(null) + val tolerance = Literal.create(0.0) + + assert(ST_EqualsExact(point :: point :: tolerance :: Nil).eval(EmptyRow) != null) + assert(ST_EqualsExact(point :: missing :: tolerance :: Nil).eval(EmptyRow) == null) + assert(ST_EqualsExact(missing :: point :: tolerance :: Nil).eval(EmptyRow) == null) + assert(ST_EqualsExact(point :: point :: missing :: Nil).eval(EmptyRow) == null) + } } } From 6d9915512d3155edbf36300dcdcd9eaf6dcd8f1f Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 24 Jul 2026 01:15:35 -0700 Subject: [PATCH 06/11] [GH-3156][GH-3157] Address GeoSeries review feedback --- python/sedona/spark/geopandas/geoseries.py | 34 +++++++++++++++------- python/tests/geopandas/test_geoseries.py | 26 +++++++++++++++++ 2 files changed, 49 insertions(+), 11 deletions(-) diff --git a/python/sedona/spark/geopandas/geoseries.py b/python/sedona/spark/geopandas/geoseries.py index f22d996cc4b..259b2109c60 100644 --- a/python/sedona/spark/geopandas/geoseries.py +++ b/python/sedona/spark/geopandas/geoseries.py @@ -667,12 +667,26 @@ def _query_geometry_column( index_spark_columns = [] index_fields = [] + index_names = [] if not is_aggr: - # Preserve every index level and the natural order in the result. + # Preserve every index level available in the projected frame. + # Some binary-operation helpers intentionally project only the + # first level until they support full MultiIndex alignment. + available_columns = set(df.columns) + index_metadata = [ + (column_name, field, index_name) + for column_name, field, index_name in zip( + self._internal.index_spark_column_names, + self._internal.index_fields, + self._internal.index_names, + ) + if column_name in available_columns + ] index_spark_columns = [ - scol_for(df, name) for name in self._internal.index_spark_column_names + scol_for(df, column_name) for column_name, _, _ in index_metadata ] - index_fields = self._internal.index_fields + index_fields = [field for _, field, _ in index_metadata] + index_names = [index_name for _, _, index_name in index_metadata] sdf = df.select( col_expr, *index_spark_columns, @@ -686,7 +700,8 @@ def _query_geometry_column( spark_frame=sdf, index_fields=index_fields, index_spark_columns=index_spark_columns, - index_names=[None] if is_aggr else self._internal.index_names, + index_names=index_names if index_spark_columns else [None], + column_labels=([(rename,)] if is_aggr else self._internal.column_labels), data_spark_columns=[scol_for(sdf, rename)], data_fields=[self._internal.data_fields[0].copy(name=rename)], column_label_names=[(rename,)], @@ -3398,7 +3413,7 @@ def explode(self, ignore_index=False, index_parts=False) -> "GeoSeries": >>> from shapely.geometry import MultiPoint >>> s = GeoSeries( ... [MultiPoint([(0, 0), (1, 1)]), MultiPoint([(2, 2), (3, 3)])] - ) + ... ) >>> s.explode(index_parts=True) 0 0 POINT (0 0) 1 POINT (1 1) @@ -3406,8 +3421,6 @@ def explode(self, ignore_index=False, index_parts=False) -> "GeoSeries": 1 POINT (3 3) dtype: geometry """ - from pyspark.pandas.internal import InternalField - ( internal, expanded_sdf, @@ -3421,10 +3434,9 @@ def explode(self, ignore_index=False, index_parts=False) -> "GeoSeries": index_parts=index_parts, temp_prefix="explode", ) - data_col = internal.data_spark_column_names[0] output_sdf = expanded_sdf.select( *[scol_for(expanded_sdf, name) for name in output_index_cols], - scol_for(expanded_sdf, geometry_col).alias(data_col), + scol_for(expanded_sdf, geometry_col), scol_for(expanded_sdf, NATURAL_ORDER_COLUMN_NAME), ) @@ -3435,9 +3447,9 @@ def explode(self, ignore_index=False, index_parts=False) -> "GeoSeries": ], index_names=index_names, index_fields=index_fields, - data_spark_columns=[scol_for(output_sdf, data_col)], + data_spark_columns=[scol_for(output_sdf, geometry_col)], data_fields=[ - InternalField(np.dtype("object"), output_sdf.schema[data_col]) + InternalField(np.dtype("object"), output_sdf.schema[geometry_col]) ], ) result = GeoSeries(first_series(PandasOnSparkDataFrame(result_internal))) diff --git a/python/tests/geopandas/test_geoseries.py b/python/tests/geopandas/test_geoseries.py index 5d453507b01..97043e17f1c 100644 --- a/python/tests/geopandas/test_geoseries.py +++ b/python/tests/geopandas/test_geoseries.py @@ -461,6 +461,24 @@ def test_explode(self, kwargs): assert all_empty.crs is not None assert all_empty.crs.to_epsg() == 4326 + def test_explode_docstring_examples_are_syntactically_valid(self): + import doctest + + examples = doctest.DocTestParser().get_examples(GeoSeries.explode.__doc__) + for example in examples: + compile(example.source, "", "single") + + @pytest.mark.parametrize("name", ["__index_level_1__", "__INDEX_LEVEL_1__"]) + def test_explode_internal_name_collision(self, name): + from geopandas.testing import assert_geoseries_equal + + geometries = [MultiPoint([(0, 0), (1, 1)])] + series = GeoSeries(geometries) + series.name = name + result = series.explode(index_parts=True).to_geopandas() + expected = gpd.GeoSeries(geometries, name=name).explode(index_parts=True) + assert_geoseries_equal(result, expected, check_index_type=False) + def test_to_crs(self): from pyproj import CRS @@ -3785,6 +3803,14 @@ def test_geom_equals(self): expected = pd.Series([True, False, True]) self.check_pd_series_equal(df_result, expected) + def test_binary_operation_with_projected_multiindex(self): + index = pd.MultiIndex.from_tuples([("a", 1), ("b", 2)], names=["group", "row"]) + result = GeoSeries([Point(0, 0), Point(1, 1)], index=index).geom_equals( + Point(0, 0) + ) + expected = pd.Series([True, False], index=pd.Index(["a", "b"], name="group")) + self.check_pd_series_equal(result, expected) + def test_geom_equals_exact(self): s = GeoSeries([Point(0, 1.1), Point(0, 1.0), Point(0, 1.2)]) From 12a524f25daa9cbf9e4403ca1a70eb7a9d0fc4c1 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 24 Jul 2026 02:10:04 -0700 Subject: [PATCH 07/11] [FLINK] Add ST_EqualsExact predicate --- docs/api/flink/Geometry-Functions.md | 1 + docs/api/flink/Predicates/ST_EqualsExact.md | 62 +++++++++++++++++++ .../java/org/apache/sedona/flink/Catalog.java | 1 + .../sedona/flink/expressions/Predicates.java | 24 +++++++ .../apache/sedona/flink/PredicateTest.java | 23 +++++++ 5 files changed, 111 insertions(+) create mode 100644 docs/api/flink/Predicates/ST_EqualsExact.md diff --git a/docs/api/flink/Geometry-Functions.md b/docs/api/flink/Geometry-Functions.md index 489ebf11f7b..8c633700cdb 100644 --- a/docs/api/flink/Geometry-Functions.md +++ b/docs/api/flink/Geometry-Functions.md @@ -166,6 +166,7 @@ These functions test spatial relationships between geometries, returning boolean | [ST_Disjoint](Predicates/ST_Disjoint.md) | Boolean | Return true if A and B are disjoint | v1.2.1 | | [ST_DWithin](Predicates/ST_DWithin.md) | Boolean | Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. | v1.5.1 | | [ST_Equals](Predicates/ST_Equals.md) | Boolean | Return true if A equals to B | v1.5.0 | +| [ST_EqualsExact](Predicates/ST_EqualsExact.md) | Boolean | Return true if A and B have matching structures and corresponding coordinates within a tolerance | v1.9.1 | | [ST_Intersects](Predicates/ST_Intersects.md) | Boolean | Return true if A intersects B | v1.2.0 | | [ST_OrderingEquals](Predicates/ST_OrderingEquals.md) | Boolean | Returns true if the geometries are equal and the coordinates are in the same order | v1.2.1 | | [ST_Overlaps](Predicates/ST_Overlaps.md) | Boolean | Return true if A overlaps B | v1.5.0 | diff --git a/docs/api/flink/Predicates/ST_EqualsExact.md b/docs/api/flink/Predicates/ST_EqualsExact.md new file mode 100644 index 00000000000..efb0210eb15 --- /dev/null +++ b/docs/api/flink/Predicates/ST_EqualsExact.md @@ -0,0 +1,62 @@ + + +# ST_EqualsExact + +Introduction: Return true if A and B have the same structure and their corresponding coordinates are equal within a tolerance. + +Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores z and m coordinates. + +Format: `ST_EqualsExact (A: Geometry, B: Geometry, tolerance: Double)` + +Return type: `Boolean` + +Since: `v1.9.1` + +Example: + +```sql +SELECT ST_EqualsExact( + ST_GeomFromWKT('POINT (0 0)'), + ST_GeomFromWKT('POINT (0.03 0.04)'), + 0.05 +) +``` + +Output: + +``` +true +``` + +The order of coordinates must match: + +```sql +SELECT ST_EqualsExact( + ST_GeomFromWKT('LINESTRING (0 0, 1 1)'), + ST_GeomFromWKT('LINESTRING (1 1, 0 0)'), + 0.0 +) +``` + +Output: + +``` +false +``` diff --git a/flink/src/main/java/org/apache/sedona/flink/Catalog.java b/flink/src/main/java/org/apache/sedona/flink/Catalog.java index a97576e5b3d..d1528069878 100644 --- a/flink/src/main/java/org/apache/sedona/flink/Catalog.java +++ b/flink/src/main/java/org/apache/sedona/flink/Catalog.java @@ -269,6 +269,7 @@ public static UserDefinedFunction[] getPredicates() { new Predicates.ST_CoveredBy(), new Predicates.ST_Disjoint(), new Predicates.ST_Equals(), + new Predicates.ST_EqualsExact(), new Predicates.ST_OrderingEquals(), new Predicates.ST_Overlaps(), new Predicates.ST_Touches(), diff --git a/flink/src/main/java/org/apache/sedona/flink/expressions/Predicates.java b/flink/src/main/java/org/apache/sedona/flink/expressions/Predicates.java index 0f6e8c00d6a..6cc90cad297 100644 --- a/flink/src/main/java/org/apache/sedona/flink/expressions/Predicates.java +++ b/flink/src/main/java/org/apache/sedona/flink/expressions/Predicates.java @@ -347,6 +347,30 @@ public Boolean eval( } } + public static class ST_EqualsExact extends ScalarFunction { + + public ST_EqualsExact() {} + + @DataTypeHint("Boolean") + public Boolean eval( + @DataTypeHint( + value = "RAW", + rawSerializer = GeometryTypeSerializer.class, + bridgedTo = Geometry.class) + Object o1, + @DataTypeHint( + value = "RAW", + rawSerializer = GeometryTypeSerializer.class, + bridgedTo = Geometry.class) + Object o2, + @DataTypeHint("Double") Double tolerance) { + if (o1 == null || o2 == null || tolerance == null) return null; + Geometry geom1 = (Geometry) o1; + Geometry geom2 = (Geometry) o2; + return org.apache.sedona.common.Predicates.equalsExact(geom1, geom2, tolerance); + } + } + public static class ST_OrderingEquals extends ScalarFunction { /** Constructor for relation checking without duplicate removal */ diff --git a/flink/src/test/java/org/apache/sedona/flink/PredicateTest.java b/flink/src/test/java/org/apache/sedona/flink/PredicateTest.java index d76a77f7da4..2ef14a163ce 100644 --- a/flink/src/test/java/org/apache/sedona/flink/PredicateTest.java +++ b/flink/src/test/java/org/apache/sedona/flink/PredicateTest.java @@ -206,6 +206,29 @@ public void testEquals() { assertEquals(true, actual); } + @Test + public void testEqualsExact() { + Table table = + tableEnv.sqlQuery( + "SELECT" + + " ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.05)" + + " AS within_tolerance," + + " ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.03, 0.04), 0.049)" + + " AS outside_tolerance," + + " ST_EqualsExact(ST_GeomFromWKT(CAST(NULL AS STRING))," + + " ST_Point(0.0, 0.0), 0.0) AS null_left," + + " ST_EqualsExact(ST_Point(0.0, 0.0)," + + " ST_GeomFromWKT(CAST(NULL AS STRING)), 0.0) AS null_right," + + " ST_EqualsExact(ST_Point(0.0, 0.0), ST_Point(0.0, 0.0)," + + " CAST(NULL AS DOUBLE)) AS null_tolerance"); + org.apache.flink.types.Row row = first(table); + assertEquals(true, row.getField(0)); + assertEquals(false, row.getField(1)); + assertNull(row.getField(2)); + assertNull(row.getField(3)); + assertNull(row.getField(4)); + } + @Test public void testOrderingEquals() { Table lineStringTable = createLineStringTable(testDataSize); From 6abdd53fe1e359ff6eed4a7704e3752744bf46db Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 24 Jul 2026 02:14:13 -0700 Subject: [PATCH 08/11] [GH-3158] Add ST_EqualsExact to Snowflake --- .../vector-data/Geometry-Functions.md | 1 + .../vector-data/Predicates/ST_EqualsExact.md | 60 +++++++++++++++++++ .../snowsql/TestFunctionsGeography.java | 11 ++++ .../snowflake/snowsql/TestPredicates.java | 14 +++++ .../snowflake/snowsql/TestPredicatesV2.java | 14 +++++ .../apache/sedona/snowflake/snowsql/UDFs.java | 9 +++ .../sedona/snowflake/snowsql/UDFsV2.java | 11 ++++ 7 files changed, 120 insertions(+) create mode 100644 docs/api/snowflake/vector-data/Predicates/ST_EqualsExact.md diff --git a/docs/api/snowflake/vector-data/Geometry-Functions.md b/docs/api/snowflake/vector-data/Geometry-Functions.md index 01ecdda07dc..b0f6945c617 100644 --- a/docs/api/snowflake/vector-data/Geometry-Functions.md +++ b/docs/api/snowflake/vector-data/Geometry-Functions.md @@ -158,6 +158,7 @@ These functions test spatial relationships between geometries, returning boolean | [ST_Disjoint](Predicates/ST_Disjoint.md) | Return true if A and B are disjoint | | [ST_DWithin](Predicates/ST_DWithin.md) | Returns true if 'leftGeometry' and 'rightGeometry' are within a specified 'distance'. This function essentially checks if the shortest distance between the envelope of the two geometries is <= the ... | | [ST_Equals](Predicates/ST_Equals.md) | Return true if A equals to B | +| [ST_EqualsExact](Predicates/ST_EqualsExact.md) | Return true if A and B have matching structures and corresponding coordinates within a tolerance | | [ST_Intersects](Predicates/ST_Intersects.md) | Return true if A intersects B | | [ST_OrderingEquals](Predicates/ST_OrderingEquals.md) | Returns true if the geometries are equal and the coordinates are in the same order | | [ST_Overlaps](Predicates/ST_Overlaps.md) | Return true if A overlaps B | diff --git a/docs/api/snowflake/vector-data/Predicates/ST_EqualsExact.md b/docs/api/snowflake/vector-data/Predicates/ST_EqualsExact.md new file mode 100644 index 00000000000..953c1f3c000 --- /dev/null +++ b/docs/api/snowflake/vector-data/Predicates/ST_EqualsExact.md @@ -0,0 +1,60 @@ + + +# ST_EqualsExact + +Introduction: Return true if A and B have the same structure and their corresponding coordinates are equal within a tolerance. + +Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores z and m coordinates. + +Format: `ST_EqualsExact (A: Geometry, B: Geometry, tolerance: Double)` + +Return type: `Boolean` + +SQL Example: + +```sql +SELECT ST_EqualsExact( + ST_GeomFromWKT('POINT (0 0)'), + ST_GeomFromWKT('POINT (0.03 0.04)'), + 0.05 +) +``` + +Output: + +``` +true +``` + +The order of coordinates must match: + +```sql +SELECT ST_EqualsExact( + ST_GeomFromWKT('LINESTRING (0 0, 1 1)'), + ST_GeomFromWKT('LINESTRING (1 1, 0 0)'), + 0.0 +) +``` + +Output: + +``` +false +``` diff --git a/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestFunctionsGeography.java b/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestFunctionsGeography.java index 3d0df7590b1..5f71f9437cc 100644 --- a/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestFunctionsGeography.java +++ b/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestFunctionsGeography.java @@ -28,4 +28,15 @@ public void test_GeometryType() { registerUDFGeography("GeometryType", String.class); verifySqlSingleRes("select sedona.GeometryType(ST_GeographyFromWKT('POINT(1 2)'))", "POINT"); } + + @Test + public void test_ST_EqualsExact() { + registerUDFGeography("ST_EqualsExact", String.class, String.class, double.class); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(ST_GeographyFromWKT('POINT (0 0)'), ST_GeographyFromWKT('POINT (0.03 0.04)'), 0.051)", + true); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(ST_GeographyFromWKT('POINT (0 0)'), ST_GeographyFromWKT('POINT (0.03 0.04)'), 0.049)", + false); + } } diff --git a/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicates.java b/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicates.java index d34d6fa68df..7729933bfb8 100644 --- a/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicates.java +++ b/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicates.java @@ -64,6 +64,20 @@ public void test_ST_Equals() { true); } + @Test + public void test_ST_EqualsExact() { + registerUDF("ST_EqualsExact", byte[].class, byte[].class, double.class); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(SEDONA.ST_GeomFromWKT('POINT (0 0)'), SEDONA.ST_GeomFromWKT('POINT (0.03 0.04)'), 0.051)", + true); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(SEDONA.ST_GeomFromWKT('POINT (0 0)'), SEDONA.ST_GeomFromWKT('POINT (0.03 0.04)'), 0.049)", + false); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(SEDONA.ST_GeomFromWKT('LINESTRING (0 0, 1 1)'), SEDONA.ST_GeomFromWKT('LINESTRING (1 1, 0 0)'), 0.0)", + false); + } + @Test public void test_ST_Intersects() { registerUDF("ST_Intersects", byte[].class, byte[].class); diff --git a/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicatesV2.java b/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicatesV2.java index 3b9486ed47b..2c3cd6c8f0d 100644 --- a/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicatesV2.java +++ b/snowflake-tester/src/test/java/org/apache/sedona/snowflake/snowsql/TestPredicatesV2.java @@ -64,6 +64,20 @@ public void test_ST_Equals() { true); } + @Test + public void test_ST_EqualsExact() { + registerUDFV2("ST_EqualsExact", String.class, String.class, double.class); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(ST_GeometryFromWKT('POINT (0 0)'), ST_GeometryFromWKT('POINT (0.03 0.04)'), 0.051)", + true); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(ST_GeometryFromWKT('POINT (0 0)'), ST_GeometryFromWKT('POINT (0.03 0.04)'), 0.049)", + false); + verifySqlSingleRes( + "SELECT SEDONA.ST_EqualsExact(ST_GeometryFromWKT('LINESTRING (0 0, 1 1)'), ST_GeometryFromWKT('LINESTRING (1 1, 0 0)'), 0.0)", + false); + } + @Test public void test_ST_Intersects() { registerUDFV2("ST_Intersects", String.class, String.class); diff --git a/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFs.java b/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFs.java index f78ad7d3f31..205bc4f0a5e 100644 --- a/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFs.java +++ b/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFs.java @@ -402,6 +402,15 @@ public static boolean ST_Equals(byte[] leftGeometry, byte[] rightGeometry) { GeometrySerde.deserialize(leftGeometry), GeometrySerde.deserialize(rightGeometry)); } + @UDFAnnotations.ParamMeta(argNames = {"leftGeometry", "rightGeometry", "tolerance"}) + public static boolean ST_EqualsExact( + byte[] leftGeometry, byte[] rightGeometry, double tolerance) { + return Predicates.equalsExact( + GeometrySerde.deserialize(leftGeometry), + GeometrySerde.deserialize(rightGeometry), + tolerance); + } + @UDFAnnotations.ParamMeta(argNames = {"geometry"}) public static byte[] ST_ExteriorRing(byte[] geometry) { return GeometrySerde.serialize(Functions.exteriorRing(GeometrySerde.deserialize(geometry))); diff --git a/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFsV2.java b/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFsV2.java index 18cf8a281f8..ce938d5ef7e 100644 --- a/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFsV2.java +++ b/snowflake/src/main/java/org/apache/sedona/snowflake/snowsql/UDFsV2.java @@ -563,6 +563,17 @@ public static boolean ST_Equals(String leftGeometry, String rightGeometry) { GeometrySerde.deserGeoJson(leftGeometry), GeometrySerde.deserGeoJson(rightGeometry)); } + @UDFAnnotations.ParamMeta( + argNames = {"leftGeometry", "rightGeometry", "tolerance"}, + argTypes = {"Geometry", "Geometry", "double"}) + public static boolean ST_EqualsExact( + String leftGeometry, String rightGeometry, double tolerance) { + return Predicates.equalsExact( + GeometrySerde.deserGeoJson(leftGeometry), + GeometrySerde.deserGeoJson(rightGeometry), + tolerance); + } + @UDFAnnotations.ParamMeta( argNames = {"geometry"}, argTypes = {"Geometry"}, From 611fc359ed7fecd99dc9e6a325eb1aca9dc84336 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 24 Jul 2026 08:32:50 -0700 Subject: [PATCH 09/11] [GH-3156] Use valid polygon in explode parity test --- python/tests/geopandas/test_match_geopandas_series.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/python/tests/geopandas/test_match_geopandas_series.py b/python/tests/geopandas/test_match_geopandas_series.py index 03029a80d90..2fb4152d701 100644 --- a/python/tests/geopandas/test_match_geopandas_series.py +++ b/python/tests/geopandas/test_match_geopandas_series.py @@ -85,7 +85,7 @@ def setup_method(self): self.polygons = [ Polygon(), - Polygon([(0, 0), (1, 0), (2, 1), (3, 1)]), + Polygon([(0, 0), (1, 0), (2, 1), (0, 1)]), Polygon([(1, 1), (2, 1), (2, 2), (1, 2)]), ] From 15cc146da02a9a4cd3f53ccd4a3b3c41d67fba97 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 24 Jul 2026 08:56:48 -0700 Subject: [PATCH 10/11] [GH-3158] Address ST_EqualsExact documentation review --- docs/api/sql/Predicates/ST_EqualsExact.md | 4 ++-- python/sedona/spark/sql/st_predicates.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/docs/api/sql/Predicates/ST_EqualsExact.md b/docs/api/sql/Predicates/ST_EqualsExact.md index ab274a5f78d..6cf5c80a6ee 100644 --- a/docs/api/sql/Predicates/ST_EqualsExact.md +++ b/docs/api/sql/Predicates/ST_EqualsExact.md @@ -19,9 +19,9 @@ # ST_EqualsExact -Introduction: Return true if A and B have the same structure and their corresponding coordinates are equal within a tolerance. +Introduction: Returns true if A and B have the same structure and their corresponding coordinates are equal within a tolerance. -Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores z and m coordinates. +Unlike `ST_Equals`, this predicate requires geometry types, component order, ring order, and vertex order to match. The tolerance is the maximum distance allowed between each pair of corresponding coordinates. The comparison uses x and y coordinates and ignores Z and M coordinates. Format: `ST_EqualsExact (A: Geometry, B: Geometry, tolerance: Double)` diff --git a/python/sedona/spark/sql/st_predicates.py b/python/sedona/spark/sql/st_predicates.py index b48cdd75790..dd04b77d1c8 100644 --- a/python/sedona/spark/sql/st_predicates.py +++ b/python/sedona/spark/sql/st_predicates.py @@ -107,7 +107,7 @@ def ST_EqualsExact( :param b: Other geometry column to check. :type b: ColumnOrName :param tolerance: Maximum distance allowed between corresponding coordinates. - :type tolerance: ColumnOrName or float + :type tolerance: ColumnOrName or float or int :return: True if a and b are exactly equal within tolerance, otherwise False. :rtype: Column """ From 2c7b82dc342264630c80701e35861ca7849f3d97 Mon Sep 17 00:00:00 2001 From: Jia Yu Date: Fri, 24 Jul 2026 09:35:14 -0700 Subject: [PATCH 11/11] [GH-3156] Handle invalid geometry in explode parity test --- python/tests/geopandas/test_match_geopandas_series.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/python/tests/geopandas/test_match_geopandas_series.py b/python/tests/geopandas/test_match_geopandas_series.py index 2fb4152d701..1b2210bf05c 100644 --- a/python/tests/geopandas/test_match_geopandas_series.py +++ b/python/tests/geopandas/test_match_geopandas_series.py @@ -85,7 +85,8 @@ def setup_method(self): self.polygons = [ Polygon(), - Polygon([(0, 0), (1, 0), (2, 1), (0, 1)]), + # Keep an invalid polygon to exercise invalid-geometry paths. + Polygon([(0, 0), (1, 0), (2, 1), (3, 1)]), Polygon([(1, 1), (2, 1), (2, 2), (1, 2)]), ] @@ -446,11 +447,14 @@ def test_explode(self, kwargs): result = GeoSeries(geometries, index=index, name="geometry").explode(**kwargs) actual = result.to_geopandas() + # The fixture intentionally includes an invalid polygon. Coordinate-wise + # equality is reliable here while topological equality is not. assert_geoseries_equal( actual, expected, check_index_type=False, check_geom_type=True, + check_less_precise=True, ) pd.testing.assert_index_equal(actual.index, expected.index, exact=False)