[GH-3051] Implement distributed GeoPandas translate

python/sedona/spark/geopandas/base.py

@@ -1571,6 +1571,62 @@ def skew(self, xs=0.0, ys=0.0, origin="center", use_radians=False):
         """
         return _delegate_to_geometry_column("skew", self, xs, ys, origin, use_radians)
 
+    def translate(self, xoff=0.0, yoff=0.0, zoff=0.0):
+        """Return a ``GeoSeries`` with translated geometries.
+
+        Each geometry is shifted by constant offsets along its coordinate
+        dimensions.
+
+        Parameters
+        ----------
+        xoff : float, default 0.0
+            Offset along the x dimension.
+        yoff : float, default 0.0
+            Offset along the y dimension.
+        zoff : float, default 0.0
+            Offset along the z dimension for geometries that have z
+            coordinates.
+
+        Returns
+        -------
+        GeoSeries
+            The translated geometries.
+
+        Notes
+        -----
+        Two-dimensional geometries remain two-dimensional. Results for mixed
+        2D/3D ``GeometryCollection`` objects, M or ZM ordinates, NaN z
+        coordinates, and non-finite offsets may differ from GeoPandas because
+        Sedona uses JTS while GeoPandas uses Shapely. This method applies
+        Sedona's distributed semantics and does not materialize geometries
+        locally to emulate Shapely.
+
+        Examples
+        --------
+        >>> from shapely.geometry import Point, LineString, Polygon
+        >>> from sedona.spark.geopandas import GeoSeries
+        >>> s = GeoSeries(
+        ...     [
+        ...         Point(1, 1),
+        ...         LineString([(1, -1), (1, 0)]),
+        ...         Polygon([(3, -1), (4, 0), (3, 1)]),
+        ...     ]
+        ... )
+        >>> s
+        0                         POINT (1 1)
+        1              LINESTRING (1 -1, 1 0)
+        2    POLYGON ((3 -1, 4 0, 3 1, 3 -1))
+        dtype: geometry
+
+        >>> s.translate(2, 3)
+        0                       POINT (3 4)
+        1            LINESTRING (3 2, 3 3)
+        2    POLYGON ((5 2, 6 3, 5 4, 5 2))
+        dtype: geometry
+
+        """
+        return _delegate_to_geometry_column("translate", self, xoff, yoff, zoff)
+
     def force_2d(self):
         """Force the dimensionality of a geometry to 2D.
 

python/sedona/spark/geopandas/geoseries.py

@@ -1389,6 +1389,13 @@ def skew(self, xs=0.0, ys=0.0, origin="center", use_radians=False) -> "GeoSeries
         spark_expr = F.when(stf.ST_IsEmpty(geometry), geometry).otherwise(skewed)
         return self._query_geometry_column(spark_expr, returns_geom=True)
 
+    def translate(self, xoff=0.0, yoff=0.0, zoff=0.0) -> "GeoSeries":
+        xoff = _normalize_affine_scalar(xoff, "'xoff' must be a numeric scalar")
+        yoff = _normalize_affine_scalar(yoff, "'yoff' must be a numeric scalar")
+        zoff = _normalize_affine_scalar(zoff, "'zoff' must be a numeric scalar")
+        spark_expr = stf.ST_Translate(self.spark.column, xoff, yoff, zoff)
+        return self._query_geometry_column(spark_expr, returns_geom=True)
+
     def force_2d(self) -> "GeoSeries":
         spark_expr = stf.ST_Force_2D(self.spark.column)
         return self._query_geometry_column(spark_expr, returns_geom=True)

python/tests/geopandas/test_geoseries.py

@@ -2490,6 +2490,132 @@ def test_skew_validates_angles_units_and_origin(self):
         with pytest.raises(ValueError, match="origin must be"):
             empty_source.skew(origin="invalid")
 
+    def test_translate_documented_example_and_delegation(self):
+        geoms = [
+            Point(1, 1),
+            LineString([(1, -1), (1, 0)]),
+            Polygon([(3, -1), (4, 0), (3, 1)]),
+            None,
+        ]
+        s = GeoSeries(geoms)
+
+        # Docstring example: translate(2, 3)
+        result = s.translate(2, 3)
+        expected = gpd.GeoSeries(
+            [
+                Point(3, 4),
+                LineString([(3, 2), (3, 3)]),
+                Polygon([(5, 2), (6, 3), (5, 4), (5, 2)]),
+                None,
+            ]
+        )
+        self.check_sgpd_equals_gpd(result, expected)
+
+        df_result = s.to_geoframe().translate(2, 3)
+        assert isinstance(df_result, GeoSeries)
+        self.check_sgpd_equals_gpd(df_result, expected)
+
+    @pytest.mark.parametrize(
+        "offsets",
+        [
+            pytest.param((), id="defaults"),
+            pytest.param((2, -3, 0), id="integer-xy"),
+            pytest.param((-1.5, 0.25, -2.0), id="negative-fractional"),
+            pytest.param(
+                (np.float64(2), np.int64(3), np.float32(5)),
+                id="numpy-scalars",
+            ),
+        ],
+    )
+    def test_translate_offsets_and_dimensions(self, offsets):
+        geoms = [
+            Point(1, 2),
+            LineString([(0, 0), (2, 4)]),
+            Point(1, 2, 3),
+            LineString([(0, 0, 1), (2, 4, 7)]),
+        ]
+        source = GeoSeries(geoms)
+        expected = gpd.GeoSeries(geoms).translate(*offsets)
+
+        result = source.translate(*offsets)
+
+        self.check_sgpd_equals_gpd(result, expected)
+        actual = result.to_geopandas().sort_index()
+        assert not actual.iloc[0].has_z
+        assert not actual.iloc[1].has_z
+        assert actual.iloc[2].has_z
+        assert actual.iloc[3].has_z
+
+    def test_translate_preserves_metadata_empty_and_null(self):
+        geoms = [
+            Point(1, 2),
+            Point(),
+            LineString(),
+            Polygon(),
+            None,
+        ]
+        index = pd.Index(
+            ["point", "empty-point", "empty-line", "empty-polygon", "null"],
+            name="feature_id",
+        )
+        source = GeoSeries(geoms, index=index, crs="EPSG:3857", name="geometry")
+        expected = gpd.GeoSeries(
+            geoms, index=index, crs="EPSG:3857", name="geometry"
+        ).translate(2, -3, 5)
+
+        result = source.translate(2, -3, 5)
+
+        self.check_sgpd_equals_gpd(result, expected)
+        assert result.name is None
+        assert result.crs == source.crs == expected.crs
+        actual = result.to_geopandas()
+        assert actual.loc["empty-point"].is_empty
+        assert actual.loc["empty-point"].geom_type == "Point"
+        assert actual.loc["empty-line"].is_empty
+        assert actual.loc["empty-line"].geom_type == "LineString"
+        assert actual.loc["empty-polygon"].is_empty
+        assert actual.loc["empty-polygon"].geom_type == "Polygon"
+        assert actual.loc["null"] is None
+
+        srids = result._internal.spark_frame.select(
+            stf.ST_SRID(result.spark.column).alias("srid")
+        ).collect()
+        assert {row.srid for row in srids if row.srid is not None} == {3857}
+
+        frame_result = source.to_geoframe().translate(2, -3, 5)
+        assert isinstance(frame_result, GeoSeries)
+        self.check_sgpd_equals_gpd(frame_result, expected)
+        assert frame_result.crs == source.crs
+
+    def test_translate_validates_offsets(self):
+        source = GeoSeries([Point(1, 2)])
+        invalid_offsets = [
+            ("xoff", None),
+            ("yoff", "2"),
+            ("zoff", np.array([2])),
+            ("xoff", np.array(2.0)),
+            ("yoff", [2]),
+            ("zoff", 2 + 1j),
+            ("xoff", pd.Series([2.0])),
+        ]
+        for name, value in invalid_offsets:
+            with pytest.raises(TypeError, match=rf"'{name}' must be a numeric scalar"):
+                source.translate(**{name: value})
+
+        for value in (ps.Series([2.0]), source.spark.column):
+            for name in ("xoff", "yoff", "zoff"):
+                with pytest.raises(
+                    TypeError, match=rf"'{name}' must be a numeric scalar"
+                ):
+                    source.translate(**{name: value})
+
+        # Operation-wide arguments are validated even when there is no
+        # non-empty geometry on which to apply the transformation.
+        empty_source = GeoSeries([Point(), LineString(), Polygon(), None])
+        for name in ("xoff", "yoff", "zoff"):
+            with pytest.raises(TypeError, match=rf"'{name}' must be a numeric scalar"):
+                empty_source.translate(**{name: None})
+
     def test_force_2d(self):
         s = sgpd.GeoSeries(
             [

python/tests/geopandas/test_match_geopandas_series.py

@@ -1207,6 +1207,52 @@ def test_skew_3d_preserves_z(self):
         expected_z = shapely.get_coordinates(geoms, include_z=True)[:, 2]
         np.testing.assert_array_equal(actual_z, expected_z)
 
+    @pytest.mark.parametrize(
+        "xoff,yoff,zoff",
+        [
+            (2, 3, 0),
+            (0, 0, 0),
+            (1.5, -2, 0),
+            (0, 0, 5),
+        ],
+        ids=["positive", "identity", "mixed", "z-only"],
+    )
+    def test_translate_2d(self, xoff, yoff, zoff):
+        for geom in self.geoms:
+            # Filter empty geometries within each group
+            non_empty = [g for g in geom if g is not None and not g.is_empty]
+            if not non_empty:
+                continue
+
+            sgpd_result = GeoSeries(non_empty).translate(xoff, yoff, zoff)
+            gpd_result = gpd.GeoSeries(non_empty).translate(xoff, yoff, zoff)
+            self.check_sgpd_equals_gpd(sgpd_result, gpd_result)
+
+    def test_translate_3d(self):
+        polygon = Polygon([(0, 0, 1), (3, 0, 2), (1, 2, 4), (0, 0, 1)])
+        geoms = [
+            Point(1, 2, 3),
+            LineString([(0, 0, 1), (2, 1, 4)]),
+            polygon,
+            MultiPoint([(0, 0, 1), (1, 2, 3)]),
+            MultiLineString([[(0, 0, 1), (2, 1, 3)], [(1, -1, 2), (3, 2, 4)]]),
+            MultiPolygon([polygon]),
+            GeometryCollection(
+                [Point(1, 2, 3), LineString([(0, 0, 1), (2, 1, 4)]), polygon]
+            ),
+        ]
+        translate_kwargs = {"xoff": -2.5, "yoff": 3.25, "zoff": 5.0}
+
+        sgpd_result = GeoSeries(geoms).translate(**translate_kwargs)
+        gpd_result = gpd.GeoSeries(geoms).translate(**translate_kwargs)
+
+        self.check_sgpd_equals_gpd(sgpd_result, gpd_result)
+        actual_z = shapely.get_coordinates(
+            sgpd_result.to_geopandas().sort_index().array, include_z=True
+        )[:, 2]
+        source_z = shapely.get_coordinates(geoms, include_z=True)[:, 2]
+        np.testing.assert_allclose(actual_z, source_z + translate_kwargs["zoff"])
+
     def test_force_2d(self):
         # force_2d was added from geopandas 1.0.0
         if parse_version(gpd.__version__) < parse_version("1.0.0"):