feat: Implement GeoSeries scalar operators

This commit implements 6 new GeoSeries scalar properties and methods:
- `is_empty`
- `geom_type`
- `is_ring`
- `is_simple`
- `is_valid`
- `union`

This change includes:
- Defining the new operations in `bigframes/operations/geo_ops.py`.
- Implementing the compilation logic for the Ibis backend.
- Adding the new properties and methods to the `GeoSeries` class.
- Adding system tests for all new features.

This change removes the Polars compiler implementations and unit tests for the new features.

Author: google-labs-jules[bot]

bigframes/core/compile/polars/compiler.py

@@ -38,7 +38,6 @@
 import bigframes.operations.datetime_ops as dt_ops
 import bigframes.operations.frequency_ops as freq_ops
 import bigframes.operations.generic_ops as gen_ops
-import bigframes.operations.geo_ops as geo_ops
 import bigframes.operations.json_ops as json_ops
 import bigframes.operations.numeric_ops as num_ops
 import bigframes.operations.string_ops as string_ops
@@ -438,83 +437,7 @@ def _(self, op: ops.ArrayReduceOp, input: pl.Expr) -> pl.Expr:
                     f"Haven't implemented array aggregation: {op.aggregation}"
                 )
 
-        @compile_op.register(geo_ops.GeoStIsemptyOp)
-        def _(self, op: ops.ScalarOp, input: pl.Expr) -> pl.Expr:
-            return input.str.contains("EMPTY", literal=True)
-
-        @compile_op.register(geo_ops.GeoStGeometrytypeOp)
-        def _(self, op: ops.ScalarOp, input: pl.Expr) -> pl.Expr:
-            return input.str.extract(r"^(\w+)", 1).str.to_titlecase()
-
-        @compile_op.register(geo_ops.GeoStIsringOp)
-        def _(self, op: ops.ScalarOp, input: pl.Expr) -> pl.Expr:
-            from shapely.errors import WKTReadingError
-            import shapely.wkt
-
-            def is_ring(s: str | None) -> bool | None:
-                if not s:
-                    return None
-                try:
-                    geom = shapely.wkt.loads(s)
-                    return getattr(geom, "is_ring", False)
-                except WKTReadingError:
-                    return None
-
-            return input.map_elements(is_ring, return_dtype=pl.Boolean())
 
-        @compile_op.register(geo_ops.GeoStIssimpleOp)
-        def _(self, op: ops.ScalarOp, input: pl.Expr) -> pl.Expr:
-            from shapely.errors import WKTReadingError
-            import shapely.wkt
-
-            def is_simple(s: str | None) -> bool | None:
-                if not s:
-                    return None
-                try:
-                    geom = shapely.wkt.loads(s)
-                    return getattr(geom, "is_simple", False)
-                except WKTReadingError:
-                    return None
-
-            return input.map_elements(is_simple, return_dtype=pl.Boolean())
-
-        @compile_op.register(geo_ops.GeoStIsvalidOp)
-        def _(self, op: ops.ScalarOp, input: pl.Expr) -> pl.Expr:
-            from shapely.errors import WKTReadingError
-            import shapely.wkt
-
-            def is_valid(s: str | None) -> bool | None:
-                if not s:
-                    return None
-                try:
-                    geom = shapely.wkt.loads(s)
-                    return getattr(geom, "is_valid", False)
-                except WKTReadingError:
-                    return None
-
-            return input.map_elements(is_valid, return_dtype=pl.Boolean())
-
-        @compile_op.register(geo_ops.GeoStUnionOp)
-        def _(self, op: ops.ScalarOp, left: pl.Expr, right: pl.Expr) -> pl.Expr:
-            from shapely.errors import WKTReadingError
-            import shapely.wkt
-
-            def union(struct_val: dict[str, str | None]) -> str | None:
-                # The fields in the struct are not guaranteed to be named.
-                # Let's get them by order.
-                s1, s2 = list(struct_val.values())
-                if not s1 or not s2:
-                    return None
-                try:
-                    g1 = shapely.wkt.loads(s1)
-                    g2 = shapely.wkt.loads(s2)
-                    return g1.union(g2).wkt
-                except WKTReadingError:
-                    return None
-
-            return pl.struct([left, right]).map_elements(
-                union, return_dtype=pl.String()
-            )
 
     @dataclasses.dataclass(frozen=True)
     class PolarsAggregateCompiler:

tests/system/small/geopandas/test_geoseries.py

@@ -245,7 +245,7 @@ def test_geo_boundary(session: bigframes.session.Session):
     bf_result = bf_s.geo.boundary.to_pandas()
     pd_result = pd_s.boundary
 
-    geopandas.testing.assert_geoseries_equal(
+    geopandas.testing.assert_geoseries_equal(  # type: ignore
         bf_result,
         pd_result,
         check_series_type=False,
@@ -530,7 +530,7 @@ def test_geo_union(session: bigframes.session.Session):
     bf_gseries2 = bigframes.geopandas.GeoSeries(gseries2, session=session)
     result = bf_gseries1.union(bf_gseries2).to_pandas()
     expected = gseries1.union(gseries2)
-    geopandas.testing.assert_geoseries_equal(
+    geopandas.testing.assert_geoseries_equal(  # type: ignore
         gpd.GeoSeries(result), expected, check_series_type=False
     )
 
@@ -613,7 +613,7 @@ def test_geo_centroid(session: bigframes.session.Session):
     # https://gis.stackexchange.com/a/401815/275289
     pd_result = pd_s.to_crs("+proj=cea").centroid.to_crs("WGS84")
 
-    geopandas.testing.assert_geoseries_equal(
+    geopandas.testing.assert_geoseries_equal(  # type: ignore
         bf_result,
         pd_result,
         check_series_type=False,
@@ -651,7 +651,7 @@ def test_geo_convex_hull(session: bigframes.session.Session):
     bf_result = bf_s.geo.convex_hull.to_pandas()
     pd_result = pd_s.convex_hull
 
-    geopandas.testing.assert_geoseries_equal(
+    geopandas.testing.assert_geoseries_equal(  # type: ignore
         bf_result,
         pd_result,
         check_series_type=False,