You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sedona.apache.org by ji...@apache.org on 2026/07/29 06:34:27 UTC
(sedona) branch master updated: [GH-3196] GeoPandas: Add distributed sampling, point construction, and serialization (#3197)
This is an automated email from the ASF dual-hosted git repository.
jiayuasu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/sedona.git
The following commit(s) were added to refs/heads/master by this push:
new 7a6ba17d53a [GH-3196] GeoPandas: Add distributed sampling, point construction, and serialization (#3197)
7a6ba17d53a is described below
commit 7a6ba17d53a4f8fb1eb8561f9a70f51b34ea26c8
Author: Jia Yu <ji...@apache.org>
AuthorDate: Tue Jul 28 23:34:21 2026 -0700
[GH-3196] GeoPandas: Add distributed sampling, point construction, and serialization (#3197)
---
docs/tutorial/geopandas-api.md | 20 +-
docs/tutorial/geopandas-api.zh.md | 17 +-
python/sedona/spark/geopandas/__init__.py | 1 +
python/sedona/spark/geopandas/array.py | 272 ++++++++++++++
python/sedona/spark/geopandas/base.py | 94 +++++
python/sedona/spark/geopandas/geodataframe.py | 159 +++++++-
python/sedona/spark/geopandas/geoseries.py | 398 +++++++++++++++++++--
python/tests/geopandas/test_array.py | 221 ++++++++++++
python/tests/geopandas/test_geodataframe.py | 110 ++++++
python/tests/geopandas/test_geoseries.py | 268 ++++++++++++++
.../geopandas/test_match_geopandas_dataframe.py | 41 +++
.../tests/geopandas/test_match_geopandas_series.py | 73 ++++
12 files changed, 1631 insertions(+), 43 deletions(-)
diff --git a/docs/tutorial/geopandas-api.md b/docs/tutorial/geopandas-api.md
index 22b6963771b..2fbeff1894a 100644
--- a/docs/tutorial/geopandas-api.md
+++ b/docs/tutorial/geopandas-api.md
@@ -337,6 +337,8 @@ The GeoPandas API for Apache Sedona implements the most commonly used GeoSeries
- `intersects()`, `contains()`, `within()` - Spatial predicates
- `intersection()` - Geometric intersection
- `make_valid()` - Geometry validation and repair
+- `sample_points()` - Sample polygons by area and lines by length with native
+ distributed expressions
- `cx` - Coordinate-based spatial filtering
- `clip()` - Distributed geometry clipping
- `sindex` - Spatial indexing (limited functionality)
@@ -374,10 +376,24 @@ only aggregate metadata and the API's single geometry result to the driver.
### Data Conversion
- `to_geopandas()` - Convert to traditional GeoPandas
-- `to_wkb()`, `to_wkt()` - Convert to WKB/WKT formats
-- `from_xy()` - Create geometries from coordinates
+- `GeoDataFrame.to_wkb()`, `GeoDataFrame.to_wkt()` - Serialize every geometry
+ column to WKB/WKT in a distributed pandas-on-Spark DataFrame
+- `points_from_xy()`, `GeoSeries.from_xy()` - Create a distributed GeoSeries
+ from coordinate columns without collecting distributed inputs
- `geom_type` - Get geometry types
+Unlike GeoPandas, `GeoDataFrame.to_wkb()` and `GeoDataFrame.to_wkt()` return a
+lazy pandas-on-Spark DataFrame. Call `.to_pandas()` on the result only when a
+local pandas DataFrame is required.
+
+`points_from_xy()` keeps pandas-on-Spark coordinate Series distributed.
+Local lists, NumPy arrays, and pandas objects are first materialized on the
+driver, so use distributed Series for large coordinate columns. Likewise, use
+a distributed `size` Series for large per-row `sample_points()` sizes.
+Sampling output and per-row intermediate work grow with the requested size;
+the current line sampler materializes the row's sampled points before
+collecting them into a MultiPoint.
+
## Complete Workflow Example
```python
diff --git a/docs/tutorial/geopandas-api.zh.md b/docs/tutorial/geopandas-api.zh.md
index 5100ea351c0..d235288bade 100644
--- a/docs/tutorial/geopandas-api.zh.md
+++ b/docs/tutorial/geopandas-api.zh.md
@@ -337,6 +337,7 @@ Apache Sedona 的 GeoPandas API 已实现最常用的 GeoSeries 与 GeoDataFrame
- `intersects()`、`contains()`、`within()` —— 空间谓词
- `intersection()` —— 几何相交
- `make_valid()` —— 几何校验与修复
+- `sample_points()` —— 使用原生分布式表达式按面积对多边形采样、按长度对线采样
- `cx` —— 基于坐标范围的空间筛选
- `clip()` —— 分布式几何裁剪
- `sindex` —— 空间索引（功能有限）
@@ -372,10 +373,22 @@ all_building_parts = collect(buildings.geometry)
### 数据转换
- `to_geopandas()` —— 转换为传统 GeoPandas
-- `to_wkb()`、`to_wkt()` —— 转换为 WKB/WKT
-- `from_xy()` —— 通过坐标创建几何
+- `GeoDataFrame.to_wkb()`、`GeoDataFrame.to_wkt()` —— 在分布式
+ pandas-on-Spark DataFrame 中把所有几何列序列化为 WKB/WKT
+- `points_from_xy()`、`GeoSeries.from_xy()` —— 通过坐标列创建分布式
+ GeoSeries，且不收集分布式输入
- `geom_type` —— 获取几何类型
+与 GeoPandas 不同，`GeoDataFrame.to_wkb()` 和 `GeoDataFrame.to_wkt()` 返回
+惰性的 pandas-on-Spark DataFrame。仅在需要本地 pandas DataFrame 时，才对结果
+调用 `.to_pandas()`。
+
+`points_from_xy()` 会让 pandas-on-Spark 坐标 Series 保持分布式执行。本地
+列表、NumPy 数组和 pandas 对象会先在 driver 上物化，因此大型坐标列应使用
+分布式 Series。类似地，大型逐行 `sample_points()` 数量向量也应使用分布式
+`size` Series。采样输出与逐行中间计算量会随请求的数量增长；当前线采样器会先
+物化该行的采样点，再把它们收集为 MultiPoint。
+
## 完整工作流示例
```python
diff --git a/python/sedona/spark/geopandas/__init__.py b/python/sedona/spark/geopandas/__init__.py
index a89d732f3f8..a5bac9d262a 100644
--- a/python/sedona/spark/geopandas/__init__.py
+++ b/python/sedona/spark/geopandas/__init__.py
@@ -22,6 +22,7 @@
from sedona.spark.geopandas.geoseries import GeoSeries
from sedona.spark.geopandas.geodataframe import GeoDataFrame
+from sedona.spark.geopandas.array import points_from_xy
from sedona.spark.geopandas.tools import clip, sjoin
diff --git a/python/sedona/spark/geopandas/array.py b/python/sedona/spark/geopandas/array.py
new file mode 100644
index 00000000000..b7a875f8464
--- /dev/null
+++ b/python/sedona/spark/geopandas/array.py
@@ -0,0 +1,272 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Distributed geometry-array construction helpers."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+import pandas as pd
+import pyspark.pandas as pspd
+from pyproj import CRS
+from pyspark.pandas.frame import DataFrame as PandasOnSparkDataFrame
+from pyspark.pandas.internal import (
+ InternalField,
+ NATURAL_ORDER_COLUMN_NAME,
+ SPARK_DEFAULT_SERIES_NAME,
+)
+from pyspark.pandas.series import first_series
+from pyspark.pandas.utils import same_anchor, scol_for
+from pyspark.sql import functions as F
+from pyspark.sql.types import (
+ BooleanType,
+ NullType,
+ NumericType,
+ StructField,
+)
+
+from sedona.spark.geopandas._crs import with_crs_metadata
+from sedona.spark.sql import st_constructors as stc
+from sedona.spark.sql import st_functions as stf
+from sedona.spark.sql.types import GeometryType
+
+
+def _is_distributed_series(value: Any) -> bool:
+ return isinstance(value, pspd.Series)
+
+
+def _coordinate_expression(series: pspd.Series, label: str):
+ spark_type = series.spark.data_type
+ if not isinstance(spark_type, (NumericType, BooleanType, NullType)):
+ raise TypeError(
+ f"{label} must be a numeric pandas-on-Spark Series, "
+ f"got {spark_type.simpleString()}"
+ )
+
+ # GeoPandas converts missing coordinates to IEEE NaN before constructing
+ # points. Preserve that distinction from a missing geometry.
+ return F.coalesce(series.spark.column.cast("double"), F.lit(float("nan")))
+
+
+def _coordinate_scalar_expression(value: Any, label: str):
+ try:
+ array = np.asarray(value, dtype="float64")
+ except (TypeError, ValueError, OverflowError) as exc:
+ raise TypeError(
+ f"{label} must be a numeric scalar or pandas-on-Spark Series"
+ ) from exc
+ if array.ndim != 0:
+ raise TypeError(
+ f"{label} must be a numeric scalar when another coordinate is "
+ "distributed"
+ )
+ return F.lit(float(array))
+
+
+def _distributed_points_from_xy(
+ coordinates: list[Any],
+ labels: list[str],
+ crs: Any | None,
+ name: Any | None,
+):
+ distributed_coordinates = [
+ coordinate for coordinate in coordinates if _is_distributed_series(coordinate)
+ ]
+ source = distributed_coordinates[0]
+ if not all(
+ same_anchor(source, coordinate) for coordinate in distributed_coordinates[1:]
+ ):
+ raise ValueError(
+ "pandas-on-Spark coordinate Series must share the same distributed "
+ "frame and index; align them in one DataFrame before calling "
+ "points_from_xy"
+ )
+
+ expressions = [
+ (
+ _coordinate_expression(coordinate, label)
+ if _is_distributed_series(coordinate)
+ else _coordinate_scalar_expression(coordinate, label)
+ )
+ for coordinate, label in zip(coordinates, labels)
+ ]
+ point = (
+ stc.ST_PointZ(*expressions)
+ if len(expressions) == 3
+ else stc.ST_Point(*expressions)
+ )
+ normalized_crs = CRS.from_user_input(crs) if crs is not None else None
+ if normalized_crs is not None:
+ point = stf.ST_SetSRID(point, normalized_crs.to_epsg() or 0)
+
+ source_internal = source._internal
+ source_sdf = source_internal.spark_frame
+ result_name = SPARK_DEFAULT_SERIES_NAME
+ result_template = with_crs_metadata(
+ InternalField(
+ np.dtype("object"),
+ StructField(result_name, GeometryType(), nullable=True),
+ ),
+ normalized_crs,
+ )
+ result_sdf = source_sdf.select(
+ point.alias(result_name, metadata=result_template.metadata),
+ *source_internal.index_spark_columns,
+ scol_for(source_sdf, NATURAL_ORDER_COLUMN_NAME),
+ )
+ result_internal = source_internal.copy(
+ spark_frame=result_sdf,
+ index_spark_columns=[
+ scol_for(result_sdf, column_name)
+ for column_name in source_internal.index_spark_column_names
+ ],
+ column_labels=[(result_name,)],
+ data_spark_columns=[scol_for(result_sdf, result_name)],
+ data_fields=[InternalField(np.dtype("object"), result_sdf.schema[result_name])],
+ column_label_names=[None],
+ )
+
+ from sedona.spark.geopandas.geoseries import GeoSeries
+
+ result = first_series(PandasOnSparkDataFrame(result_internal)).rename(name)
+ return GeoSeries(result)
+
+
+def _local_coordinate_array(value: Any, label: str) -> np.ndarray:
+ try:
+ array = np.asarray(value, dtype="float64")
+ except (TypeError, ValueError, OverflowError) as exc:
+ raise TypeError(f"{label} must be an iterable of numeric values") from exc
+ if array.ndim > 1:
+ raise ValueError(f"{label} must be one-dimensional")
+ return array
+
+
+def _local_points_from_xy(
+ coordinates: list[Any],
+ labels: list[str],
+ crs: Any | None,
+ index,
+ name: Any | None,
+):
+ arrays = [
+ _local_coordinate_array(coordinate, label)
+ for coordinate, label in zip(coordinates, labels)
+ ]
+ try:
+ arrays = list(np.broadcast_arrays(*arrays))
+ except ValueError as exc:
+ lengths = ", ".join(
+ f"{label}={array.size}" for label, array in zip(labels, arrays)
+ )
+ raise ValueError(
+ f"Coordinate lengths are not broadcast-compatible: {lengths}"
+ ) from exc
+ if arrays[0].ndim == 0:
+ raise TypeError("at least one coordinate must be an iterable")
+
+ if index is None and all(isinstance(value, pd.Series) for value in coordinates):
+ first_index = coordinates[0].index
+ if all(value.index.equals(first_index) for value in coordinates[1:]):
+ index = first_index
+
+ frame = pd.DataFrame(
+ {label: array for label, array in zip(labels, arrays)},
+ index=index,
+ )
+ distributed = pspd.from_pandas(frame)
+ return _distributed_points_from_xy(
+ [distributed[label] for label in labels],
+ labels,
+ crs,
+ name,
+ )
+
+
+def _points_from_xy(
+ x,
+ y,
+ z=None,
+ crs=None,
+ *,
+ index=None,
+ name=None,
+):
+ coordinates = [x, y] if z is None else [x, y, z]
+ labels = ["x", "y"] if z is None else ["x", "y", "z"]
+ distributed = [_is_distributed_series(value) for value in coordinates]
+
+ if any(distributed):
+ if index is not None:
+ raise TypeError(
+ "index cannot be supplied with distributed coordinate Series; "
+ "their shared index is preserved automatically"
+ )
+ return _distributed_points_from_xy(coordinates, labels, crs, name)
+
+ return _local_points_from_xy(coordinates, labels, crs, index, name)
+
+
+def points_from_xy(x, y, z=None, crs=None):
+ """
+ Construct a distributed GeoSeries of Point geometries from coordinates.
+
+ Unlike :func:`geopandas.points_from_xy`, which returns an in-memory
+ ``GeometryArray``, this scalable counterpart returns a distributed
+ :class:`GeoSeries` so that its index and CRS metadata can be retained.
+
+ Parameters
+ ----------
+ x, y, z : iterable or pandas-on-Spark Series
+ Coordinate values. Distributed Series must come from the same frame;
+ numeric scalar coordinates are broadcast over that frame.
+ crs : value, optional
+ Coordinate Reference System accepted by
+ :meth:`pyproj.CRS.from_user_input`.
+
+ Returns
+ -------
+ GeoSeries
+ Distributed Point geometries. Shared distributed indexes, including
+ MultiIndexes and duplicate labels, are preserved.
+
+ Examples
+ --------
+ >>> import sedona.spark.geopandas as sgpd
+ >>> points = sgpd.points_from_xy([1, 2], [3, 4], crs="EPSG:4326")
+ >>> points
+ 0 POINT (1 3)
+ 1 POINT (2 4)
+ dtype: geometry
+
+ Notes
+ -----
+ Construction uses native ``ST_Point`` or ``ST_PointZ`` expressions. It
+ does not collect pandas-on-Spark inputs or execute a Python row UDF.
+
+ Local coordinates use NumPy broadcasting and are paired positionally, as
+ in GeoPandas. Local iterables, NumPy arrays, and pandas objects are
+ materialized on the driver and are intended for bounded inputs. Use
+ pandas-on-Spark Series for large coordinate columns.
+
+ Distributed Series are paired within their shared Spark plan and retain
+ that plan's index. Series from unrelated plans are rejected because they
+ have no safe distributed positional relationship.
+ """
+ return _points_from_xy(x, y, z=z, crs=crs)
diff --git a/python/sedona/spark/geopandas/base.py b/python/sedona/spark/geopandas/base.py
index fa62d08be47..a4272fae7b6 100644
--- a/python/sedona/spark/geopandas/base.py
+++ b/python/sedona/spark/geopandas/base.py
@@ -1608,6 +1608,100 @@ class GeoFrame(metaclass=ABCMeta):
"""
return _delegate_to_geometry_column("reverse", self)
+ def sample_points(
+ self,
+ size,
+ method="uniform",
+ seed=None,
+ rng=None,
+ **kwargs,
+ ):
+ """Sample points from each geometry.
+
+ Polygonal geometries are sampled uniformly by area and linear
+ geometries uniformly by length. Unsupported and empty geometries
+ produce an empty ``MultiPoint``. Supported geometries also return
+ ``MultiPoint`` for every size, including zero and one. Sampling is
+ evaluated by native Spark and Sedona expressions and remains
+ distributed.
+
+ Parameters
+ ----------
+ size : int or array-like
+ Number of points to sample from each geometry. An array-like or
+ pandas-on-Spark ``Series`` supplies one positional size per row.
+ Local sequences are materialized on the driver; use a distributed
+ Series for large per-row size vectors.
+ method : str, default "uniform"
+ Sampling method. Sedona currently supports only ``"uniform"``.
+ seed : optional
+ Deprecated alias for ``rng``.
+ rng : optional
+ Any seed accepted by ``numpy.random.default_rng``. Fresh generators
+ initialized with the same seed produce reproducible results. Passing
+ an existing ``Generator`` or ``BitGenerator`` consumes one draw
+ immediately to derive the engine seed.
+ **kwargs
+ Accepted for GeoPandas signature compatibility and ignored for
+ uniform sampling.
+
+ Returns
+ -------
+ GeoSeries
+ Sampled points with the original index and CRS and the name
+ ``"sampled_points"``.
+
+ Examples
+ --------
+ >>> from sedona.spark.geopandas import GeoSeries
+ >>> from shapely.geometry import LineString, Polygon
+ >>> s = GeoSeries(
+ ... [
+ ... Polygon([(0, 0), (1, 0), (1, 1), (0, 1)]),
+ ... LineString([(0, 0), (2, 0)]),
+ ... ]
+ ... )
+ >>> s.sample_points(size=3, rng=7).count_geometries()
+ 0 3
+ 1 3
+ dtype: int64
+
+ Notes
+ -----
+ Sedona and GeoPandas use different random number generators, so the
+ sampled coordinates are not expected to be identical for the same
+ seed. An integer seed is reused for every row, matching GeoPandas's
+ behavior in which congruent geometries receive the same relative sample
+ positions. With ``rng=None``, no reproducibility guarantee is made.
+
+ A stateful NumPy ``Generator`` or ``BitGenerator`` is reduced to one
+ engine seed and then varied deterministically by row position, so its
+ state advancement also differs. Validation of values in a distributed
+ ``size`` Series is lazy and errors are raised when the result is
+ evaluated.
+
+ Sampling work and per-row intermediate memory grow with ``size``. The
+ current line sampler materializes an array of sampled points and costs
+ roughly ``O(size * vertices)`` per geometry before collecting those
+ points into a MultiPoint. Unsupported nonempty geometry types produce
+ empty MultiPoints without GeoPandas's per-row warning, avoiding an eager
+ type scan.
+
+ Sedona follows GeoPandas 1.1.4 and later by always returning a
+ ``MultiPoint``. GeoPandas 1.1.3 and earlier instead return a ``Point``
+ for ``size=1`` and an empty ``GeometryCollection`` for ``size=0``, and
+ may collapse duplicate samples.
+ """
+ return _delegate_to_geometry_column(
+ "sample_points",
+ self,
+ size,
+ method=method,
+ seed=seed,
+ rng=rng,
+ **kwargs,
+ )
+
def segmentize(self, max_segment_length):
"""Returns a ``GeoSeries`` with vertices added to line segments based on
maximum segment length.
diff --git a/python/sedona/spark/geopandas/geodataframe.py b/python/sedona/spark/geopandas/geodataframe.py
index a5968e1892a..23124c95c0b 100644
--- a/python/sedona/spark/geopandas/geodataframe.py
+++ b/python/sedona/spark/geopandas/geodataframe.py
@@ -37,10 +37,13 @@ from pyspark.pandas.internal import (
SPARK_INDEX_NAME_FORMAT,
)
from pyspark.pandas.utils import scol_for
+from pyspark.sql.column import Column
from pyspark.sql import functions as F
from pyspark.sql.types import (
+ BinaryType,
BooleanType,
ByteType,
+ DataType,
DoubleType,
FloatType,
IntegerType,
@@ -57,6 +60,7 @@ from sedona.spark.geopandas.base import GeoFrame
from sedona.spark.sql import st_aggregates as sta
from sedona.spark.sql import st_constructors as stc
from sedona.spark.sql import st_functions as stf
+from sedona.spark.sql.types import GeometryType
from pandas.api.extensions import register_extension_dtype
from geopandas.geodataframe import crs_mismatch_error
@@ -1560,11 +1564,158 @@ class GeoDataFrame(GeoFrame, pspd.DataFrame):
) -> dict:
raise NotImplementedError("to_geo_dict() is not implemented yet.")
- def to_wkb(self, hex: bool = False, **kwargs) -> pd.DataFrame:
- raise NotImplementedError("to_wkb() is not implemented yet.")
+ def _serialize_geometry_columns(
+ self,
+ serializer: Callable[[Column], Column],
+ output_type: DataType,
+ ) -> pspd.DataFrame:
+ """Apply a native serializer to every geometry-typed column."""
+ # Resolve pending index/data expressions onto one Spark frame before
+ # building the serializer projection.
+ internal = self._internal.resolved_copy
+ serialized_columns = []
+ geometry_columns = []
+
+ # Build one projection for the whole frame. Sequential assignment would
+ # align each replacement Series and can multiply duplicate-index rows.
+ for spark_column, spark_name, field in zip(
+ internal.data_spark_columns,
+ internal.data_spark_column_names,
+ internal.data_fields,
+ ):
+ is_geometry = isinstance(field.spark_type, GeometryType)
+ geometry_columns.append(is_geometry)
+ serialized_columns.append(
+ serializer(spark_column).alias(spark_name)
+ if is_geometry
+ else spark_column
+ )
+
+ output_sdf = internal.spark_frame.select(
+ *[
+ scol_for(internal.spark_frame, name)
+ for name in internal.index_spark_column_names
+ ],
+ *serialized_columns,
+ scol_for(internal.spark_frame, NATURAL_ORDER_COLUMN_NAME),
+ )
+ output_schema = output_sdf.schema
+ output_fields = [
+ (
+ field.copy(
+ dtype=np.dtype("object"),
+ spark_type=output_type,
+ nullable=output_schema[spark_name].nullable,
+ metadata={},
+ )
+ if is_geometry
+ else field
+ )
+ for spark_name, field, is_geometry in zip(
+ internal.data_spark_column_names,
+ internal.data_fields,
+ geometry_columns,
+ )
+ ]
+ result_internal = internal.copy(
+ spark_frame=output_sdf,
+ index_spark_columns=[
+ scol_for(output_sdf, name) for name in internal.index_spark_column_names
+ ],
+ data_spark_columns=[
+ scol_for(output_sdf, name) for name in internal.data_spark_column_names
+ ],
+ data_fields=output_fields,
+ )
+ return PandasOnSparkDataFrame(result_internal)
+
+ @staticmethod
+ def _validate_serialization_kwargs(method_name: str, kwargs: dict) -> None:
+ if kwargs:
+ names = ", ".join(sorted(kwargs))
+ raise NotImplementedError(
+ f"GeoDataFrame.{method_name}() does not support Shapely keyword "
+ f"arguments in distributed execution: {names}"
+ )
+
+ def to_wkb(self, hex: bool = False, **kwargs) -> pspd.DataFrame:
+ """
+ Encode all geometry columns in the GeoDataFrame to WKB.
- def to_wkt(self, **kwargs) -> pd.DataFrame:
- raise NotImplementedError("to_wkt() is not implemented yet.")
+ Parameters
+ ----------
+ hex : bool, default False
+ If True, export WKB values as hexadecimal strings. By default,
+ values are returned as binary objects.
+ **kwargs
+ Shapely-specific serialization options are not supported by the
+ distributed implementation and raise ``NotImplementedError``.
+
+ Returns
+ -------
+ pyspark.pandas.DataFrame
+ A distributed DataFrame with every geometry-typed column encoded
+ as WKB and all other columns unchanged.
+
+ Notes
+ -----
+ Duplicate column labels remain subject to pandas-on-Spark's DataFrame
+ construction restrictions.
+
+ Unlike GeoPandas, this method returns a lazy pandas-on-Spark DataFrame.
+ Call ``to_pandas()`` on the result to materialize a local pandas
+ DataFrame.
+
+ See Also
+ --------
+ GeoDataFrame.to_wkt
+ GeoSeries.to_wkb
+ """
+ self._validate_serialization_kwargs("to_wkb", kwargs)
+ serializer, output_type = (
+ (lambda geometry: F.hex(stf.ST_AsBinary(geometry)), StringType())
+ if hex
+ else (stf.ST_AsBinary, BinaryType())
+ )
+ return self._serialize_geometry_columns(serializer, output_type)
+
+ def to_wkt(self, **kwargs) -> pspd.DataFrame:
+ """
+ Encode all geometry columns in the GeoDataFrame to WKT.
+
+ Parameters
+ ----------
+ **kwargs
+ Shapely-specific serialization options are not supported by the
+ distributed implementation and raise ``NotImplementedError``.
+
+ Returns
+ -------
+ pyspark.pandas.DataFrame
+ A distributed DataFrame with every geometry-typed column encoded
+ as WKT and all other columns unchanged.
+
+ Notes
+ -----
+ Duplicate column labels remain subject to pandas-on-Spark's DataFrame
+ construction restrictions.
+
+ Unlike GeoPandas, this method returns a lazy pandas-on-Spark DataFrame.
+ Call ``to_pandas()`` on the result to materialize a local pandas
+ DataFrame.
+
+ Sedona's native WKT writer preserves stored coordinate precision.
+ GeoPandas delegates to Shapely, whose default writer rounds coordinates
+ to six decimal places. Shapely-specific formatting keyword arguments
+ are not supported here.
+
+ See Also
+ --------
+ GeoDataFrame.to_wkb
+ GeoSeries.to_wkt
+ """
+ self._validate_serialization_kwargs("to_wkt", kwargs)
+ return self._serialize_geometry_columns(stf.ST_AsText, StringType())
def to_arrow(
self,
diff --git a/python/sedona/spark/geopandas/geoseries.py b/python/sedona/spark/geopandas/geoseries.py
index 83e3a3e7c3e..f01414b5108 100644
--- a/python/sedona/spark/geopandas/geoseries.py
+++ b/python/sedona/spark/geopandas/geoseries.py
@@ -16,6 +16,7 @@
# under the License.
import sys
+import operator
import typing
import warnings
from typing import Any, Union, Literal, List
@@ -30,8 +31,15 @@ from pyspark.pandas import Series as PandasOnSparkSeries
from pyspark.pandas.frame import DataFrame as PandasOnSparkDataFrame
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
+from pyspark.pandas.utils import same_anchor, scol_for, verify_temp_column_name
+from pyspark.sql.types import (
+ BooleanType,
+ IntegralType,
+ LongType,
+ NullType,
+ StructField,
+ StructType,
+)
from sedona.spark.sql.types import GeometryType
from sedona.spark.sql import st_aggregates as sta
@@ -42,6 +50,7 @@ from sedona.spark.sql import st_predicates as stp
from pyspark.sql import Column as PySparkColumn
from pyspark.sql import functions as F
+
import shapely
from shapely.geometry.base import BaseGeometry
@@ -125,6 +134,104 @@ def _normalize_affine_scalar(value, error_message: str) -> float:
return _normalize_numeric_scalar(value, error_message)
+def _normalize_sample_size_scalar(size) -> int:
+ """Normalize one sample size without accepting boolean values."""
+ if isinstance(size, (bool, np.bool_)):
+ raise TypeError(
+ f"expected a sequence of integers or a single integer, got {size!r}"
+ )
+ try:
+ size = operator.index(size)
+ except TypeError as exc:
+ raise TypeError(
+ f"expected a sequence of integers or a single integer, got {size!r}"
+ ) from exc
+ if size < 0:
+ raise ValueError("negative dimensions are not allowed")
+ if size > 2_147_483_647:
+ raise ValueError("sample size must be at most 2147483647")
+ return size
+
+
+def _normalize_sample_seed(rng) -> int:
+ """Map a NumPy-compatible RNG argument to Sedona's positive seed range."""
+ generator = np.random.default_rng(rng)
+ return int(generator.integers(1, 2_147_483_647))
+
+
+def _attach_ordered_sequence_column(
+ sdf,
+ order_column: PySparkColumn,
+ sequence_column: str,
+):
+ """Attach dense positions without a single-partition window over all rows."""
+ # Use pandas-on-Spark's native implementation so it owns the child
+ # materialization lifecycle and avoids duplicate evaluation.
+ return InternalFrame.attach_distributed_sequence_column(
+ sdf.orderBy(order_column),
+ sequence_column,
+ )
+
+
+def _sample_points_expression(
+ geometry: PySparkColumn,
+ size: PySparkColumn,
+ seed: PySparkColumn,
+ fallback_srid: int,
+) -> PySparkColumn:
+ """Build native expressions for uniform polygon and line sampling."""
+ size = size.cast("long")
+ output_srid = F.coalesce(stf.ST_SRID(geometry), F.lit(fallback_srid))
+ empty_multipoint = stc.ST_GeomFromWKT(F.lit("MULTIPOINT EMPTY"), output_srid)
+ geometry_type = stf.ST_GeometryType(geometry)
+ is_polygonal = geometry_type.isin("ST_Polygon", "ST_MultiPolygon")
+ is_linear = geometry_type.isin("ST_LineString", "ST_MultiLineString")
+
+ polygon_samples = stf.ST_GeneratePoints(
+ geometry, size.cast("int"), seed.cast("int")
+ )
+
+ # Spark's stable hash supplies one uniformly distributed fraction per
+ # ordinal without evaluating Python code on geometry rows.
+ ordinals = F.sequence(F.lit(0), size.cast("int") - F.lit(1))
+ line_samples = F.transform(
+ ordinals,
+ lambda ordinal: stf.ST_LineInterpolatePoint(
+ geometry,
+ F.pmod(
+ F.xxhash64(seed.cast("long"), ordinal.cast("long")),
+ F.lit(9_007_199_254_740_992),
+ ).cast("double")
+ / F.lit(9_007_199_254_740_992.0),
+ ),
+ )
+ collected_line_samples = stf.ST_Collect(line_samples)
+
+ sampled = (
+ F.when(geometry.isNull() | stf.ST_IsEmpty(geometry), empty_multipoint)
+ .when(~(is_polygonal | is_linear), empty_multipoint)
+ .when(size == 0, empty_multipoint)
+ .when(is_polygonal, polygon_samples)
+ .otherwise(collected_line_samples)
+ )
+
+ return (
+ F.when(
+ size.isNull(),
+ F.raise_error(F.lit("sample size values must be integers")),
+ )
+ .when(
+ size < 0,
+ F.raise_error(F.lit("negative dimensions are not allowed")),
+ )
+ .when(
+ size > 2_147_483_647,
+ F.raise_error(F.lit("sample size must be at most 2147483647")),
+ )
+ .otherwise(sampled)
+ )
+
+
def _interpret_origin(geometry: PySparkColumn, origin, with_z: bool):
"""Resolve a local affinity origin to distributed coordinate columns."""
if isinstance(origin, str):
@@ -1503,6 +1610,247 @@ class GeoSeries(GeoFrame, pspd.Series):
returns_geom=True,
)
+ def sample_points(
+ self,
+ size,
+ method="uniform",
+ seed=None,
+ rng=None,
+ **kwargs,
+ ) -> "GeoSeries":
+ if method != "uniform":
+ raise NotImplementedError(
+ "Sedona only supports method='uniform' for distributed sampling"
+ )
+
+ if seed is not None:
+ warnings.warn(
+ "The 'seed' keyword is deprecated. Use 'rng' instead.",
+ FutureWarning,
+ stacklevel=2,
+ )
+ rng = seed
+
+ stateful_rng = isinstance(rng, (np.random.Generator, np.random.BitGenerator))
+ randomize_per_row = rng is None or stateful_rng
+ engine_seed = _normalize_sample_seed(rng)
+
+ has_crs_metadata, metadata_crs = read_crs_metadata(
+ self._internal.data_fields[0]
+ )
+ fallback_srid = (
+ (metadata_crs.to_epsg() or 0)
+ if has_crs_metadata and metadata_crs is not None
+ else 0
+ )
+
+ def seed_column(row_key: PySparkColumn) -> PySparkColumn:
+ if not randomize_per_row:
+ return F.lit(engine_seed)
+ return (
+ F.pmod(
+ F.xxhash64(F.lit(engine_seed), row_key.cast("long")),
+ F.lit(2_147_483_646),
+ )
+ + F.lit(1)
+ ).cast("int")
+
+ def temp_column_name(base: str, reserved_columns) -> str:
+ suffix = 0
+ candidate = f"__sample_points_{base}__"
+ while candidate in reserved_columns:
+ suffix += 1
+ candidate = f"__sample_points_{base}_{suffix}__"
+ reserved_columns.add(candidate)
+ return candidate
+
+ if isinstance(size, PandasOnSparkSeries):
+ size_series = size
+ local_sizes = None
+ elif pd.api.types.is_list_like(size):
+ size_series = None
+ local_sizes = [_normalize_sample_size_scalar(value) for value in size]
+ else:
+ normalized_size = _normalize_sample_size_scalar(size)
+ source_internal = self._internal.resolved_copy
+ source_frame = source_internal.spark_frame
+ geometry_column = source_internal.data_spark_columns[0]
+ row_key = scol_for(source_frame, NATURAL_ORDER_COLUMN_NAME)
+ if stateful_rng:
+ # Natural-order IDs encode Spark partition IDs. Compress them
+ # to row positions so fresh, equivalently seeded stateful RNGs
+ # remain reproducible when the partition layout changes.
+ reserved_columns = set(source_frame.columns)
+ position_col = temp_column_name("position", reserved_columns)
+ source_frame = _attach_ordered_sequence_column(
+ source_frame,
+ scol_for(source_frame, NATURAL_ORDER_COLUMN_NAME),
+ position_col,
+ )
+ row_key = F.col(position_col)
+ spark_expr = _sample_points_expression(
+ geometry_column,
+ F.lit(normalized_size),
+ seed_column(row_key),
+ fallback_srid,
+ )
+ return GeoSeries(
+ self._query_geometry_column(
+ spark_expr,
+ source_frame,
+ returns_geom=True,
+ ).rename("sampled_points")
+ )
+
+ size_type_is_integral = size_series is None or isinstance(
+ size_series.spark.data_type, IntegralType
+ )
+
+ reserved_columns = set(self._internal.spark_frame.columns)
+ if size_series is not None:
+ reserved_columns.update(size_series._internal.spark_frame.columns)
+
+ geometry_value_col = temp_column_name("geometry", reserved_columns)
+ size_value_col = temp_column_name("size", reserved_columns)
+ position_col = temp_column_name("position", reserved_columns)
+ right_present_col = temp_column_name("right_present", reserved_columns)
+
+ if size_series is not None and same_anchor(self, size_series):
+ source_internal = self._internal
+ source_frame = source_internal.spark_frame.select(
+ self.spark.column.alias(geometry_value_col),
+ size_series.spark.column.alias(size_value_col),
+ *[
+ index_col.alias(index_name)
+ for index_col, index_name in zip(
+ source_internal.index_spark_columns,
+ source_internal.index_spark_column_names,
+ )
+ ],
+ scol_for(
+ source_internal.spark_frame,
+ NATURAL_ORDER_COLUMN_NAME,
+ ).alias(NATURAL_ORDER_COLUMN_NAME),
+ )
+ row_key = F.col(NATURAL_ORDER_COLUMN_NAME)
+ if stateful_rng:
+ # A dense position keeps stateful RNG results stable across
+ # equivalent partition layouts. Both values already share one
+ # plan, so only the source needs to be sequenced.
+ source_frame = _attach_ordered_sequence_column(
+ source_frame,
+ row_key,
+ position_col,
+ )
+ row_key = F.col(position_col)
+
+ size_value = F.col(size_value_col)
+ if not size_type_is_integral:
+ size_value = F.raise_error(
+ F.lit("sample size values must be integers")
+ ).cast("long")
+ spark_expr = _sample_points_expression(
+ F.col(geometry_value_col),
+ size_value,
+ seed_column(row_key),
+ fallback_srid,
+ )
+ return GeoSeries(
+ self._query_geometry_column(
+ spark_expr,
+ source_frame,
+ returns_geom=True,
+ ).rename("sampled_points")
+ )
+
+ left_frame = self._internal.spark_frame.select(
+ self.spark.column.alias(geometry_value_col),
+ *[
+ index_col.alias(index_name)
+ for index_col, index_name in zip(
+ self._internal.index_spark_columns,
+ self._internal.index_spark_column_names,
+ )
+ ],
+ scol_for(self._internal.spark_frame, NATURAL_ORDER_COLUMN_NAME).alias(
+ NATURAL_ORDER_COLUMN_NAME
+ ),
+ )
+ left_frame = _attach_ordered_sequence_column(
+ left_frame,
+ F.col(NATURAL_ORDER_COLUMN_NAME),
+ position_col,
+ )
+ if size_series is not None:
+ right_order_col = temp_column_name("right_order", reserved_columns)
+ right_frame = size_series._internal.spark_frame.select(
+ size_series.spark.column.alias(size_value_col),
+ scol_for(
+ size_series._internal.spark_frame,
+ NATURAL_ORDER_COLUMN_NAME,
+ ).alias(right_order_col),
+ F.lit(True).alias(right_present_col),
+ )
+ right_frame = _attach_ordered_sequence_column(
+ right_frame,
+ F.col(right_order_col),
+ position_col,
+ )
+ else:
+ # A local sequence already has a driver-side position. Keep those
+ # positions as Spark rows instead of sorting a temporary
+ # pandas-on-Spark Series or embedding a large Catalyst literal.
+ right_schema = StructType(
+ [
+ StructField(position_col, LongType(), nullable=False),
+ StructField(size_value_col, LongType(), nullable=False),
+ StructField(right_present_col, BooleanType(), nullable=False),
+ ]
+ )
+ # Materialize the validated values once before constructing the
+ # explicitly typed Spark DataFrame, including for an empty input.
+ right_rows = [
+ (position, size_value, True)
+ for position, size_value in enumerate(local_sizes)
+ ]
+ right_frame = self._internal.spark_frame.sparkSession.createDataFrame(
+ right_rows,
+ schema=right_schema,
+ )
+ # GeoPandas zips geometries with array-like sizes: missing sizes are an
+ # error, while extra sizes are ignored (including for an empty source).
+ aligned_frame = left_frame.join(
+ right_frame, on=position_col, how="left"
+ ).orderBy(position_col)
+
+ size_value = F.col(size_value_col)
+ if not size_type_is_integral:
+ # Keep validation lazy so an empty source follows GeoPandas and
+ # returns empty even when the unused size Series has another dtype.
+ size_value = F.raise_error(
+ F.lit("sample size values must be integers")
+ ).cast("long")
+
+ spark_expr = _sample_points_expression(
+ F.col(geometry_value_col),
+ size_value,
+ seed_column(F.col(position_col)),
+ fallback_srid,
+ )
+ spark_expr = F.when(
+ F.col(right_present_col).isNull(),
+ F.raise_error(
+ F.lit("Length of sample sizes does not match length of GeoSeries")
+ ),
+ ).otherwise(spark_expr)
+ return GeoSeries(
+ self._query_geometry_column(
+ spark_expr,
+ aligned_frame,
+ returns_geom=True,
+ ).rename("sampled_points")
+ )
+
def segmentize(self, max_segment_length):
other_series, extended = self._make_series_of_val(max_segment_length)
align = not extended
@@ -3150,41 +3498,21 @@ class GeoSeries(GeoFrame, pspd.Series):
2 POINT (-3 1.5)
dtype: geometry
"""
- from pyspark.sql.types import StructType, StructField, DoubleType
-
- schema = StructType(
- [StructField("x", DoubleType(), True), StructField("y", DoubleType(), True)]
- )
-
- # Spark doesn't automatically cast ints to floats for us
- x = [float(num) for num in x]
- y = [float(num) for num in y]
- z = [float(num) for num in z] if z else None
-
- if z:
- data = list(zip(x, y, z))
- select = stc.ST_PointZ(F.col("x"), F.col("y"), F.col("z"))
- schema.add(StructField("z", DoubleType(), True))
- else:
- data = list(zip(x, y))
- select = stc.ST_Point(F.col("x"), F.col("y"))
-
- geoseries = cls._create_from_select(
- select,
- data,
- schema,
- index,
- crs,
- **kwargs,
+ from sedona.spark.geopandas.array import _points_from_xy
+
+ unsupported = set(kwargs) - {"name"}
+ if unsupported:
+ arguments = ", ".join(sorted(unsupported))
+ raise TypeError(f"Unsupported GeoSeries.from_xy arguments: {arguments}")
+ return _points_from_xy(
+ x,
+ y,
+ z=z,
+ crs=crs,
+ index=index,
+ name=kwargs.get("name"),
)
- if crs:
- from pyproj import CRS
-
- geoseries.crs = CRS.from_user_input(crs).to_epsg()
-
- return geoseries
-
@classmethod
def from_shapely(
cls, data, index=None, crs: Union[Any, None] = None, **kwargs
diff --git a/python/tests/geopandas/test_array.py b/python/tests/geopandas/test_array.py
new file mode 100644
index 00000000000..c7e8ee38f60
--- /dev/null
+++ b/python/tests/geopandas/test_array.py
@@ -0,0 +1,221 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import geopandas as gpd
+import numpy as np
+import pandas as pd
+import pyspark.pandas as ps
+import pytest
+from geopandas.testing import assert_geoseries_equal
+
+import sedona.spark.geopandas as sgpd
+from sedona.spark.geopandas import GeoSeries
+from sedona.spark.sql import st_functions as stf
+from tests.geopandas.test_geopandas_base import TestGeopandasBase
+
+
+def _assert_geoseries_equal_with_nan_xy(actual, expected, nan_xy):
+ """Compare NaN x/y coordinates without relying on undefined GEOS equality."""
+ assert actual.crs == expected.crs
+ nan_xy = np.asarray(nan_xy, dtype=bool)
+ nan_positions = np.flatnonzero(nan_xy)
+ regular_positions = np.flatnonzero(~nan_xy)
+ pd.testing.assert_series_equal(
+ actual.to_wkt().iloc[nan_positions],
+ expected.to_wkt().iloc[nan_positions],
+ )
+ if regular_positions.size:
+ assert_geoseries_equal(
+ actual.iloc[regular_positions],
+ expected.iloc[regular_positions],
+ )
+
+
+class TestPointsFromXY(TestGeopandasBase):
+ def test_points_from_xy_local_matches_geopandas(self):
+ index = pd.Index(["first", "first", "third"], name="feature")
+ x = pd.Series([1, 2, -3.5], index=index, name="longitude")
+ y = pd.Series([4, 5, 6.5], index=index, name="latitude")
+
+ result = sgpd.points_from_xy(x, y, crs="EPSG:4326")
+ expected = gpd.GeoSeries(
+ gpd.points_from_xy(x, y, crs="EPSG:4326"),
+ index=index,
+ crs="EPSG:4326",
+ )
+
+ assert result.name is None
+ assert result.crs == "EPSG:4326"
+ assert_geoseries_equal(result.to_geopandas(), expected)
+ srids = result._internal.spark_frame.select(
+ stf.ST_SRID(result.spark.column).alias("srid")
+ ).collect()
+ assert {row.srid for row in srids} == {4326}
+
+ def test_points_from_xy_local_3d_broadcast_and_empty(self):
+ result = sgpd.points_from_xy([1, 2], [3], [4, 5])
+ # GeoPandas 0.13 does not broadcast its own inputs, so construct the
+ # equivalent oracle from explicitly expanded coordinates.
+ expected = gpd.GeoSeries(gpd.points_from_xy([1, 2], [3, 3], [4, 5]))
+ assert_geoseries_equal(result.to_geopandas(), expected)
+
+ scalar_result = sgpd.points_from_xy(1, [2, 3], 4)
+ scalar_expected = gpd.GeoSeries(gpd.points_from_xy([1, 1], [2, 3], [4, 4]))
+ assert_geoseries_equal(scalar_result.to_geopandas(), scalar_expected)
+
+ empty = sgpd.points_from_xy([], [], crs="EPSG:3857")
+ assert len(empty) == 0
+ assert empty.crs == "EPSG:3857"
+
+ def test_points_from_xy_distributed_preserves_multiindex_and_plan(self):
+ index = pd.MultiIndex.from_tuples(
+ [("a", 1), ("a", 1), ("b", 2)],
+ names=["group", "position"],
+ )
+ pdf = pd.DataFrame(
+ {
+ "x": [1.0, np.nan, 3.0],
+ "y": [4.0, 5.0, 6.0],
+ "z": [7.0, 8.0, np.nan],
+ },
+ index=index,
+ )
+ psdf = ps.from_pandas(pdf)
+
+ result = sgpd.points_from_xy(
+ psdf["x"],
+ psdf["y"],
+ psdf["z"],
+ crs="EPSG:4979",
+ )
+ expected = gpd.GeoSeries(
+ gpd.points_from_xy(pdf["x"], pdf["y"], pdf["z"], crs="EPSG:4979"),
+ index=index,
+ crs="EPSG:4979",
+ )
+
+ assert result.crs == "EPSG:4979"
+ nan_xy = pdf[["x", "y"]].isna().any(axis=1).to_numpy()
+ _assert_geoseries_equal_with_nan_xy(
+ result.to_geopandas(),
+ expected,
+ nan_xy,
+ )
+ srids = result._internal.spark_frame.select(
+ stf.ST_SRID(result.spark.column).alias("srid")
+ ).collect()
+ assert {row.srid for row in srids} == {4979}
+
+ spark_frame = result._internal.spark_frame
+ if hasattr(spark_frame, "_jdf"):
+ plan = spark_frame._jdf.queryExecution().executedPlan().toString()
+ assert "BatchEvalPython" not in plan
+ assert "ArrowEvalPython" not in plan
+ assert "PythonUDF" not in plan
+
+ scalar_result = sgpd.points_from_xy(psdf["x"], 10, crs="EPSG:4326")
+ scalar_expected = gpd.GeoSeries(
+ gpd.points_from_xy(pdf["x"], [10] * len(pdf), crs="EPSG:4326"),
+ index=index,
+ crs="EPSG:4326",
+ )
+ _assert_geoseries_equal_with_nan_xy(
+ scalar_result.to_geopandas(),
+ scalar_expected,
+ pdf["x"].isna().to_numpy(),
+ )
+
+ def test_points_from_xy_preserves_pending_expressions(self):
+ frame = ps.DataFrame(
+ {
+ "x": [1.0, 2.0],
+ "y": [3.0, 4.0],
+ "value": [10, 11],
+ }
+ )
+ frame["computed_index"] = frame["value"] + 100
+ frame = frame.set_index("computed_index")
+
+ result = sgpd.points_from_xy(frame["x"] * 2, frame["y"] + 0.5)
+ actual = result.to_geopandas()
+
+ assert actual.index.equals(pd.Index([110, 111], name="computed_index"))
+ assert actual.to_wkt().tolist() == [
+ "POINT (2 3.5)",
+ "POINT (4 4.5)",
+ ]
+
+ def test_points_from_xy_null_is_nan_coordinate_not_missing_geometry(self):
+ psdf = self.spark.createDataFrame(
+ [(0, None, 2.0), (1, 3.0, None), (2, None, None)],
+ "id long, x double, y double",
+ ).pandas_api(index_col="id")
+
+ result = sgpd.points_from_xy(psdf["x"], psdf["y"])
+
+ assert result.isna().to_pandas().tolist() == [False, False, False]
+ expected = gpd.GeoSeries(
+ gpd.points_from_xy([None, 3.0, None], [2.0, None, None]),
+ index=pd.Index([0, 1, 2], name="id"),
+ )
+ _assert_geoseries_equal_with_nan_xy(
+ result.to_geopandas(),
+ expected,
+ np.ones(len(expected), dtype=bool),
+ )
+
+ def test_geoseries_from_xy_delegates_to_distributed_constructor(self):
+ result = GeoSeries.from_xy(
+ [1, 2],
+ [3, 4],
+ index=pd.Index([10, 20], name="id"),
+ crs="EPSG:4326",
+ name="location",
+ )
+ expected = gpd.GeoSeries(
+ gpd.points_from_xy([1, 2], [3, 4], crs="EPSG:4326"),
+ index=pd.Index([10, 20], name="id"),
+ crs="EPSG:4326",
+ name="location",
+ )
+
+ assert_geoseries_equal(result.to_geopandas(), expected)
+
+ def test_points_from_xy_validates_unsupported_inputs(self):
+ with pytest.raises(TypeError, match="at least one coordinate"):
+ sgpd.points_from_xy(1, 2)
+ with pytest.raises(ValueError, match="broadcast-compatible"):
+ sgpd.points_from_xy([1, 2], [3, 4, 5])
+
+ frame = ps.DataFrame({"x": [1, 2], "y": [3, 4]})
+ with pytest.raises(TypeError, match="numeric scalar"):
+ sgpd.points_from_xy(frame["x"], [3, 4])
+ string_frame = ps.DataFrame({"x": [1, 2], "y": ["a", "b"]})
+ with pytest.raises(
+ TypeError, match="y must be a numeric pandas-on-Spark Series"
+ ):
+ sgpd.points_from_xy(string_frame["x"], string_frame["y"])
+ with pytest.raises(
+ TypeError, match="x must be a numeric pandas-on-Spark Series"
+ ):
+ sgpd.points_from_xy(string_frame["y"], string_frame["x"])
+
+ other_frame = ps.DataFrame({"y": [3, 4]})
+ with pytest.raises(ValueError, match="same distributed frame"):
+ sgpd.points_from_xy(frame["x"], other_frame["y"])
+ with pytest.raises(TypeError, match="index cannot be supplied"):
+ GeoSeries.from_xy(frame["x"], frame["y"], index=[0, 1])
diff --git a/python/tests/geopandas/test_geodataframe.py b/python/tests/geopandas/test_geodataframe.py
index fddd678eef3..e7c51eb7772 100644
--- a/python/tests/geopandas/test_geodataframe.py
+++ b/python/tests/geopandas/test_geodataframe.py
@@ -167,6 +167,116 @@ class TestGeoDataFrame(TestGeopandasBase):
ps_df = ps.DataFrame(data, index=index)
assert_frame_equal(result.to_pandas(), ps_df.to_pandas())
+ def test_to_wkt_and_wkb_distributed(self):
+ index = pd.MultiIndex.from_tuples(
+ [("group-a", 1), ("group-a", 1), ("group-b", 2)],
+ names=["group", "position"],
+ )
+ primary = ("geometry", "primary")
+ secondary = ("geometry", "secondary")
+ value = ("attribute", "value")
+ gpd_df = gpd.GeoDataFrame(
+ {
+ "primary": gpd.GeoSeries(
+ [Point(0, 0), Polygon(), None],
+ index=index,
+ crs="EPSG:4326",
+ ),
+ "secondary": gpd.GeoSeries(
+ [LineString([(0, 0), (1, 1)]), None, Point(2, 2)],
+ index=index,
+ crs="EPSG:4326",
+ ),
+ "value": [10, 11, 12],
+ },
+ index=index,
+ geometry="primary",
+ crs="EPSG:4326",
+ )
+ gpd_df.columns = pd.MultiIndex.from_tuples(
+ [primary, secondary, value], names=["kind", "name"]
+ )
+ gpd_df = gpd_df.set_geometry(primary)
+ sgpd_df = GeoDataFrame(gpd_df)
+
+ wkt_result = sgpd_df.to_wkt()
+ assert isinstance(wkt_result, ps.DataFrame)
+ assert not isinstance(wkt_result, GeoDataFrame)
+ assert_frame_equal(
+ wkt_result.to_pandas(),
+ pd.DataFrame(
+ {
+ primary: ["POINT (0 0)", "POLYGON EMPTY", None],
+ secondary: ["LINESTRING (0 0, 1 1)", None, "POINT (2 2)"],
+ value: [10, 11, 12],
+ },
+ index=index,
+ ).rename_axis(columns=["kind", "name"]),
+ )
+
+ wkb_result = sgpd_df.to_wkb()
+ hex_result = sgpd_df.to_wkb(hex=True)
+ for result in (wkt_result, wkb_result, hex_result):
+ assert isinstance(result, ps.DataFrame)
+ assert not isinstance(result, GeoDataFrame)
+ assert "geometry" not in result.dtypes.astype(str).tolist()
+ for result in (wkb_result, hex_result):
+ result_pdf = result.to_pandas()
+ assert result_pdf.index.equals(index)
+ assert result_pdf.columns.equals(gpd_df.columns)
+ assert result_pdf[value].tolist() == [10, 11, 12]
+ assert sgpd_df.crs == "EPSG:4326"
+
+ binary_pdf = wkb_result.to_pandas()
+ hex_pdf = hex_result.to_pandas()
+ for column in (primary, secondary):
+ for binary, hexadecimal in zip(binary_pdf[column], hex_pdf[column]):
+ if binary is None:
+ assert hexadecimal is None
+ else:
+ assert hexadecimal == binary.hex().upper()
+
+ for result in (wkt_result, wkb_result, hex_result):
+ spark_frame = result._internal.spark_frame
+ if hasattr(spark_frame, "_jdf"):
+ plan = spark_frame._jdf.queryExecution().executedPlan().toString()
+ assert "BatchEvalPython" not in plan
+ assert "ArrowEvalPython" not in plan
+
+ computed_index = GeoDataFrame(
+ {
+ "geometry": [Point(0, 0), Point(1, 1)],
+ "value": [10, 11],
+ }
+ )
+ computed_index["computed_index"] = computed_index["value"] + 100
+ computed_index = GeoDataFrame(computed_index.set_index("computed_index"))
+ computed_wkt = computed_index.to_wkt().to_pandas()
+ computed_wkb = computed_index.to_wkb().to_pandas()
+ expected_index = pd.Index([110, 111], name="computed_index")
+ assert computed_wkt.index.equals(expected_index)
+ assert computed_wkb.index.equals(expected_index)
+ assert computed_wkt["geometry"].tolist() == [
+ "POINT (0 0)",
+ "POINT (1 1)",
+ ]
+ assert [
+ shapely.from_wkb(bytes(value))
+ for value in computed_wkb["geometry"].tolist()
+ ] == [Point(0, 0), Point(1, 1)]
+
+ def test_geometry_serialization_kwargs_are_explicitly_unsupported(self):
+ gdf = GeoDataFrame({"geometry": [Point(1.234567890123, 2.345678901234)]})
+
+ assert gdf.to_wkt().to_pandas()["geometry"].tolist() == [
+ "POINT (1.234567890123 2.345678901234)"
+ ]
+
+ with pytest.raises(NotImplementedError, match="rounding_precision"):
+ gdf.to_wkt(rounding_precision=2)
+ with pytest.raises(NotImplementedError, match="byte_order"):
+ gdf.to_wkb(byte_order=0)
+
def test_getitem(self):
geoms = [Point(x, x) for x in range(3)]
ids = [1, 2, 3]
diff --git a/python/tests/geopandas/test_geoseries.py b/python/tests/geopandas/test_geoseries.py
index 4dadb448cad..2cb0acccfde 100644
--- a/python/tests/geopandas/test_geoseries.py
+++ b/python/tests/geopandas/test_geoseries.py
@@ -22,6 +22,9 @@ import pandas as pd
import geopandas as gpd
import pyspark.pandas as ps
import sedona.spark.geopandas as sgpd
+from pyspark.pandas.internal import InternalFrame, NATURAL_ORDER_COLUMN_NAME
+from pyspark.pandas.utils import scol_for
+from pyspark.sql import functions as F
from sedona.spark.geopandas import GeoSeries, GeoDataFrame
from sedona.spark.geopandas.geoseries import _to_bool
from sedona.spark.sql import st_functions as stf
@@ -2471,6 +2474,271 @@ e": "Feature", "properties": {}, "geometry": {"type": "Point", "coordinates": [3
df_result = s.to_geoframe().reverse()
self.check_sgpd_equals_gpd(df_result, expected)
+ def test_sample_points(self):
+ polygon = box(0, 0, 2, 2)
+ multipolygon = MultiPolygon([box(4, 0, 5, 1), box(7, 0, 9, 2)])
+ line = LineString([(0, 0), (2, 0), (2, 2)])
+ multiline = MultiLineString([[(0, 4), (1, 4)], [(3, 4), (3, 7)]])
+ geometries = [
+ polygon,
+ multipolygon,
+ line,
+ multiline,
+ Point(1, 1),
+ MultiPoint([(0, 0), (1, 1)]),
+ GeometryCollection([Point(0, 0)]),
+ Polygon(),
+ LineString(),
+ None,
+ ]
+ index = pd.Index(
+ [
+ "area",
+ "area",
+ "line",
+ "line",
+ "point",
+ "points",
+ "gc",
+ "ep",
+ "el",
+ "null",
+ ],
+ name="feature_id",
+ )
+ source = GeoSeries(geometries, index=index, crs="EPSG:3857")
+
+ result = source.sample_points(4, rng=0)
+ actual = result.to_geopandas()
+ repeated = source.sample_points(4, rng=0).to_geopandas()
+
+ assert isinstance(result, GeoSeries)
+ assert result.name == "sampled_points"
+ assert result.crs == source.crs
+ assert actual.index.equals(index)
+ assert [geometry.wkb for geometry in actual] == [
+ geometry.wkb for geometry in repeated
+ ]
+ for source_geometry, sampled in zip(geometries[:4], actual.iloc[:4]):
+ assert sampled.geom_type == "MultiPoint"
+ assert len(sampled.geoms) == 4
+ assert all(source_geometry.covers(point) for point in sampled.geoms)
+ for sampled in actual.iloc[4:]:
+ assert sampled.geom_type == "MultiPoint"
+ assert sampled.is_empty
+
+ srids = result._internal.spark_frame.select(
+ stf.ST_SRID(result.spark.column).alias("srid")
+ ).collect()
+ assert {row.srid for row in srids} == {3857}
+
+ if hasattr(result._internal.spark_frame, "_jdf"):
+ plan = (
+ result._internal.spark_frame._jdf.queryExecution()
+ .executedPlan()
+ .toString()
+ )
+ assert "BatchEvalPython" not in plan
+ assert "ArrowEvalPython" not in plan
+ assert "PythonUDF" not in plan
+
+ frame_result = source.to_geoframe().sample_points(4, rng=0)
+ assert isinstance(frame_result, GeoSeries)
+ assert frame_result.name == "sampled_points"
+ assert frame_result.crs == source.crs
+
+ zero = GeoSeries([polygon, line, Point(0, 0)]).sample_points(0, rng=1)
+ zero_types = [geometry.geom_type for geometry in zero.to_geopandas()]
+ assert zero_types == ["MultiPoint", "MultiPoint", "MultiPoint"]
+
+ one = GeoSeries([polygon, line, Point(0, 0)]).sample_points(1, rng=1)
+ one_types = [geometry.geom_type for geometry in one.to_geopandas()]
+ assert one_types == ["MultiPoint", "MultiPoint", "MultiPoint"]
+
+ degenerate_lines = GeoSeries(
+ [
+ LineString([(0, 0), (0, 0)]),
+ MultiLineString([[(1, 1), (1, 1)], [(1, 1), (1, 1)]]),
+ ]
+ ).sample_points(4, rng=1)
+ degenerate_actual = degenerate_lines.to_geopandas()
+ assert [geometry.geom_type for geometry in degenerate_actual] == [
+ "MultiPoint",
+ "MultiPoint",
+ ]
+ assert all(
+ len(geometry.geoms) == 4
+ and all(point.equals(Point(i, i)) for point in geometry.geoms)
+ for i, geometry in enumerate(degenerate_actual)
+ )
+
+ multi_index = pd.MultiIndex.from_tuples(
+ [("a", 1), ("a", 2), ("b", 1), ("b", 2)],
+ names=["group", "position"],
+ )
+ distributed_source = GeoSeries(
+ [polygon, multipolygon, line, multiline],
+ index=multi_index,
+ crs="EPSG:3857",
+ )
+ distributed_result = distributed_source.sample_points(
+ ps.Series([1, 2, 3, 4]), rng=7
+ ).to_geopandas()
+ assert distributed_result.index.equals(multi_index)
+ assert [
+ 1 if geometry.geom_type == "Point" else len(geometry.geoms)
+ for geometry in distributed_result
+ ] == [1, 2, 3, 4]
+
+ extra_sizes = distributed_source.sample_points(
+ [1, 2, 3, 4, 99], rng=7
+ ).to_geopandas()
+ assert [
+ 1 if geometry.geom_type == "Point" else len(geometry.geoms)
+ for geometry in extra_sizes
+ ] == [1, 2, 3, 4]
+
+ same_anchor_index = pd.Index(
+ ["duplicate", "duplicate", "other", "other"],
+ name="feature_id",
+ )
+ same_anchor_frame = GeoDataFrame(
+ gpd.GeoDataFrame(
+ {
+ "geometry": [polygon, multipolygon, line, multiline],
+ "size": [1, 2, 3, 4],
+ },
+ index=same_anchor_index,
+ crs="EPSG:3857",
+ )
+ )
+ same_anchor_result = same_anchor_frame.geometry.sample_points(
+ same_anchor_frame["size"], rng=7
+ )
+ same_anchor_actual = same_anchor_result.to_geopandas()
+ assert same_anchor_actual.index.equals(same_anchor_index)
+ assert [
+ 1 if geometry.geom_type == "Point" else len(geometry.geoms)
+ for geometry in same_anchor_actual
+ ] == [1, 2, 3, 4]
+ if hasattr(same_anchor_result._internal.spark_frame, "_jdf"):
+ plan = (
+ same_anchor_result._internal.spark_frame._jdf.queryExecution()
+ .optimizedPlan()
+ .toString()
+ )
+ assert "Join" not in plan
+ assert "AttachDistributedSequence" not in plan
+
+ # Like GeoPandas, an integer seed restarts the same random stream for
+ # every row, so identical geometries receive identical samples.
+ identical = GeoSeries(
+ [line, line],
+ index=pd.Index(["duplicate", "duplicate"], name="feature_id"),
+ ).sample_points(4, rng=7)
+ identical_actual = identical.to_geopandas()
+ assert identical_actual.iloc[0].wkb == identical_actual.iloc[1].wkb
+
+ for sizes in ([], [1, 2, 3]):
+ empty_result = GeoSeries([], crs="EPSG:3857").sample_points(sizes, rng=7)
+ assert len(empty_result) == 0
+ assert empty_result.name == "sampled_points"
+ assert empty_result.crs == "EPSG:3857"
+
+ empty_float_sizes = GeoSeries([], crs="EPSG:3857").sample_points(
+ ps.Series([], dtype=float), rng=7
+ )
+ assert len(empty_float_sizes) == 0
+ assert empty_float_sizes.name == "sampled_points"
+ assert empty_float_sizes.crs == "EPSG:3857"
+
+ with pytest.raises(TypeError):
+ source.sample_points(True)
+ with pytest.raises(TypeError):
+ source.sample_points(1.5)
+ with pytest.raises(ValueError):
+ source.sample_points(-1)
+ with pytest.raises(NotImplementedError):
+ source.sample_points(1, method="cluster_poisson")
+ with pytest.raises(Exception, match="sample size values must be integers"):
+ source.sample_points(ps.Series([1.0] * len(source))).to_geopandas()
+ for short_sizes in ([], [1]):
+ with pytest.raises(Exception, match="Length of sample sizes"):
+ source.sample_points(short_sizes, rng=1).to_geopandas()
+ with pytest.warns(FutureWarning, match="'seed' keyword is deprecated"):
+ source.sample_points(1, seed=1)
+
+ def test_sample_points_stateful_rng_is_partition_stable(self):
+ line = LineString([(0, 0), (2, 0), (2, 2)])
+ identical_source = GeoSeries(
+ [line, line, line, line],
+ index=pd.Index(["duplicate"] * 4, name="feature_id"),
+ )
+ test_position_col = "__sample_points_test_position__"
+ positioned_frame = InternalFrame.attach_distributed_sequence_column(
+ identical_source._internal.spark_frame.orderBy(NATURAL_ORDER_COLUMN_NAME),
+ test_position_col,
+ )
+ ordered_rows = positioned_frame.select(
+ F.col(test_position_col).alias("position"),
+ scol_for(
+ positioned_frame,
+ identical_source._internal.index_spark_column_names[0],
+ ).alias("feature_id"),
+ scol_for(
+ positioned_frame,
+ identical_source._internal.data_spark_column_names[0],
+ ).alias("geometry"),
+ )
+
+ # These encodings have the same row order but mimic natural-order IDs
+ # generated with one partition and with a boundary before position 2.
+ natural_order_encodings = [
+ F.col("position"),
+ F.when(F.col("position") < 2, F.col("position")).otherwise(
+ F.lit(1 << 33) + F.col("position") - F.lit(2)
+ ),
+ ]
+
+ def source_with_natural_order(encoding):
+ spark_frame = ordered_rows.select(
+ F.col("feature_id"),
+ F.col("geometry"),
+ encoding.cast("long").alias(NATURAL_ORDER_COLUMN_NAME),
+ )
+ return GeoSeries(spark_frame.pandas_api(index_col="feature_id")["geometry"])
+
+ for rng_factory in (
+ lambda: np.random.default_rng(0),
+ lambda: np.random.PCG64(0),
+ ):
+ stateful_results = [
+ source_with_natural_order(encoding).sample_points(4, rng=rng_factory())
+ for encoding in natural_order_encodings
+ ]
+ assert (
+ stateful_results[0]
+ .to_geopandas()
+ .index.equals(pd.Index(["duplicate"] * 4, name="feature_id"))
+ )
+ ordered_wkb = [
+ [
+ row.geometry.wkb
+ for row in result._internal.spark_frame.select(
+ result.spark.column.alias("geometry"),
+ scol_for(
+ result._internal.spark_frame,
+ NATURAL_ORDER_COLUMN_NAME,
+ ).alias("natural_order"),
+ )
+ .orderBy("natural_order")
+ .collect()
+ ]
+ for result in stateful_results
+ ]
+ assert ordered_wkb[0] == ordered_wkb[1]
+ assert len(set(ordered_wkb[0])) == len(ordered_wkb[0])
+
def test_segmentize(self):
s = GeoSeries(
[
diff --git a/python/tests/geopandas/test_match_geopandas_dataframe.py b/python/tests/geopandas/test_match_geopandas_dataframe.py
index 3b0b8eff42d..5659ddf54c8 100644
--- a/python/tests/geopandas/test_match_geopandas_dataframe.py
+++ b/python/tests/geopandas/test_match_geopandas_dataframe.py
@@ -33,8 +33,10 @@ from shapely.geometry import (
from packaging.version import parse as parse_version
from sedona.spark.geopandas import GeoDataFrame, GeoSeries
import geopandas as gpd
+import pandas as pd
from tests.geopandas.test_geopandas_base import TestGeopandasBase
import pyspark.pandas as ps
+from pandas.testing import assert_frame_equal
@pytest.mark.skipif(
@@ -225,6 +227,45 @@ class TestMatchGeopandasDataFrame(TestGeopandasBase):
gpd_result = gpd.GeoDataFrame(data, crs="EPSG:3857").to_json(**kwargs)
assert sgpd_result == gpd_result
+ def test_to_wkt_and_to_wkb(self):
+ index = pd.MultiIndex.from_tuples(
+ [("a", 1), ("a", 1), ("b", 2)],
+ names=["group", "position"],
+ )
+ primary = ("geometry", "primary")
+ secondary = ("geometry", "secondary")
+ value = ("attribute", "value")
+ gpd_df = gpd.GeoDataFrame(
+ {
+ "primary": gpd.GeoSeries(
+ [Point(0, 0), Polygon(), None],
+ index=index,
+ crs="EPSG:4326",
+ ),
+ "secondary": gpd.GeoSeries(
+ [LineString([(0, 0), (1, 1)]), None, Point(2, 2)],
+ index=index,
+ crs="EPSG:4326",
+ ),
+ "value": [1, 2, 3],
+ },
+ index=index,
+ geometry="primary",
+ crs="EPSG:4326",
+ )
+ gpd_df.columns = pd.MultiIndex.from_tuples(
+ [primary, secondary, value], names=["kind", "name"]
+ )
+ gpd_df = gpd_df.set_geometry(primary)
+ sgpd_df = GeoDataFrame(gpd_df)
+
+ assert_frame_equal(sgpd_df.to_wkt().to_pandas(), gpd_df.to_wkt())
+ assert_frame_equal(sgpd_df.to_wkb().to_pandas(), gpd_df.to_wkb())
+ assert_frame_equal(
+ sgpd_df.to_wkb(hex=True).to_pandas(),
+ gpd_df.to_wkb(hex=True),
+ )
+
def test_from_arrow(self):
if parse_version(gpd.__version__) < parse_version("1.0.0"):
return
diff --git a/python/tests/geopandas/test_match_geopandas_series.py b/python/tests/geopandas/test_match_geopandas_series.py
index 085fedca373..4fa9f81d680 100644
--- a/python/tests/geopandas/test_match_geopandas_series.py
+++ b/python/tests/geopandas/test_match_geopandas_series.py
@@ -1276,6 +1276,79 @@ class TestMatchGeopandasSeries(TestGeopandasBase):
gpd_result = gpd.GeoSeries(geom).reverse()
self.check_sgpd_equals_gpd(sgpd_result, gpd_result)
+ @pytest.mark.skipif(
+ parse_version(gpd.__version__) < parse_version("0.14.0"),
+ reason="geopandas sample_points requires version 0.14.0 or higher",
+ )
+ def test_sample_points(self):
+ geometries = [
+ Polygon([(0, 0), (2, 0), (2, 2), (0, 2)]),
+ MultiPolygon(
+ [
+ Polygon([(4, 0), (5, 0), (5, 1), (4, 1)]),
+ Polygon([(7, 0), (9, 0), (9, 2), (7, 2)]),
+ ]
+ ),
+ LineString([(0, 4), (2, 4), (2, 6)]),
+ MultiLineString([[(0, 8), (1, 8)], [(3, 8), (3, 11)]]),
+ Point(0, 0),
+ GeometryCollection([LineString([(0, 0), (1, 1)])]),
+ Polygon(),
+ None,
+ ]
+ index = pd.Index(
+ [
+ "polygon",
+ "multipolygon",
+ "line",
+ "multiline",
+ "point",
+ "gc",
+ "empty",
+ "null",
+ ],
+ name="feature_id",
+ )
+ sgpd_source = GeoSeries(geometries, index=index, crs="EPSG:3857")
+ gpd_source = gpd.GeoSeries(geometries, index=index, crs="EPSG:3857")
+
+ sgpd_result = sgpd_source.sample_points(3, rng=9).to_geopandas()
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore", UserWarning)
+ gpd_result = gpd_source.sample_points(3, rng=9)
+
+ assert sgpd_result.name == gpd_result.name == "sampled_points"
+ assert sgpd_result.crs == gpd_result.crs
+ assert sgpd_result.index.equals(gpd_result.index)
+ assert [geometry.geom_type for geometry in sgpd_result] == [
+ geometry.geom_type for geometry in gpd_result
+ ]
+ assert [len(geometry.geoms) for geometry in sgpd_result] == [
+ len(geometry.geoms) for geometry in gpd_result
+ ]
+ for source_geometry, sampled in zip(geometries[:4], sgpd_result.iloc[:4]):
+ assert all(source_geometry.covers(point) for point in sampled.geoms)
+
+ sizes = [1, 2, 3, 4, 5, 6, 7, 8]
+ sgpd_sized = sgpd_source.sample_points(sizes, rng=9).to_geopandas()
+ with warnings.catch_warnings():
+ warnings.simplefilter("ignore", UserWarning)
+ gpd_sized = gpd_source.sample_points(sizes, rng=9)
+ expected_types = [geometry.geom_type for geometry in gpd_sized]
+ if parse_version(gpd.__version__) < parse_version("1.1.4"):
+ # GeoPandas 1.1.4 stopped applying union_all() to samples and
+ # standardized every supported result, including size 1, as a
+ # MultiPoint. Sedona follows that current return contract.
+ expected_types[0] = "MultiPoint"
+ assert [geometry.geom_type for geometry in sgpd_sized] == expected_types
+ assert [
+ (1 if geometry.geom_type == "Point" else len(geometry.geoms))
+ for geometry in sgpd_sized
+ ] == [
+ (1 if geometry.geom_type == "Point" else len(geometry.geoms))
+ for geometry in gpd_sized
+ ]
+
@pytest.mark.skipif(
parse_version(gpd.__version__) < parse_version("0.14.0"),
reason="geopandas segmentize requires version 0.14.0 or higher",