diff --git a/docs/dqx/docs/reference/quality_checks.mdx b/docs/dqx/docs/reference/quality_checks.mdx index a385a22c5..0c74889e4 100644 --- a/docs/dqx/docs/reference/quality_checks.mdx +++ b/docs/dqx/docs/reference/quality_checks.mdx @@ -100,6 +100,7 @@ You can also define your own custom checks in Python (see [Creating custom check | `is_num_points_not_greater_than` | Checks whether the values in the input column are geometries with number of coordinate pairs greater than the specified limit. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `value`: number of points value to compare against (can be a number, column name, or SQL expression) | | `is_num_points_equal_to` | Checks whether the values in the input column are geometries with number of coordinate pairs equal to the specified limit. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `value`: number of points value to compare against (can be a number, column name, or SQL expression) | | `is_num_points_not_equal_to` | Checks whether the values in the input column are geometries with number of coordinate pairs not equal to the specified limit. This function requires Databricks serverless compute or runtime >= 17.1. | `column`: column to check (can be a string column name or a column expression); `value`: number of points value to compare against (can be a number, column name, or SQL expression) | +| `is_geo_within_distance` | Checks whether the values in the input column are within a geodesic distance, in meters, of a reference geography using `st_distance`. Distances are measured along the WGS 84 ellipsoid, so the check is meaningful for global data where planar `GEOMETRY` distances are not. A row is reported when the shortest distance to the reference is strictly greater than `distance`. When a convert flag is set to `True`, `try_to_geography` is applied to parse the input from WKT, WKB, EWKT, EWKB or GeoJSON. Null values, and rows where `distance` evaluates to null, are skipped. Requires Databricks runtime 17.1 or above. | `column`: column to check (can be a string column name or a column expression); `reference_geometry`: reference geography as a literal WKT/EWKT/GeoJSON string or WKB/EWKB bytes value, or a `Column` expression (e.g. `F.col('col_name')`) — a plain string is always treated as a literal, not a column name; `distance`: maximum allowed distance in meters as a non-negative number, a `Column` expression, or a string SQL expression; `convert_column`: when `True`, applies `try_to_geography` to convert the column values to GEOGRAPHY (default `False`); `convert_reference_geometry`: when `True`, applies `try_to_geography` to convert the reference geography to GEOGRAPHY (default `False`) | @@ -914,6 +915,18 @@ For brevity, the `name` field in the examples is omitted and it will be auto-gen arguments: column: polygon_geom value: 1 + +# is_geo_within_distance check (geo, geodesic distance in meters, requires runtime 17.1+) +# location point must lie within 1 km of the reference point +- criticality: error + check: + function: is_geo_within_distance + arguments: + column: location + reference_geometry: "POINT(4.90 52.37)" + distance: 1000 + convert_column: true + convert_reference_geometry: true ``` @@ -1649,6 +1662,20 @@ checks = [ check_func_kwargs={"value": 1} ), + # is_geo_within_distance check (geo, geodesic distance in meters, requires runtime 17.1+) + # location point must lie within 1 km of the reference point + DQRowRule( + criticality="error", + check_func=geo_check_funcs.is_geo_within_distance, + column="location", # or as expr: F.col("location") + check_func_kwargs={ + "reference_geometry": "POINT(4.90 52.37)", + "distance": 1000, + "convert_column": True, + "convert_reference_geometry": True, + } + ), + # sql_expression check DQRowRule( criticality="error", @@ -2681,7 +2708,7 @@ When checks are loaded, the `__decimal__` format is automatically converted back
**Checks defined programmatically using DQX classes** ```python -from databricks.labs.dqx.rule import DQDatasetRule, DQForEachColRule +from databricks.labs.dqx.rule import DQRowRule, DQDatasetRule, DQForEachColRule from pyspark.sql.types import StructType, StructField, StringType, IntegerType from databricks.labs.dqx import check_funcs from databricks.labs.dqx.geo import check_funcs as geo_check_funcs @@ -3132,7 +3159,7 @@ checks = [ # is_geo_contains check (geo, precise only, requires runtime 17.1+) # reference polygon must contain each location point (uses st_contains) - DQDatasetRule( + DQRowRule( criticality="error", check_func=geo_check_funcs.is_geo_contains, column="location", # or as expr: F.col("location") @@ -3145,7 +3172,7 @@ checks = [ # is_geo_covers check — approximate mode, # H3 resolution 7 (~5 km² cells); default when precise=False - DQDatasetRule( + DQRowRule( criticality="error", check_func=geo_check_funcs.is_geo_covers, column="location", # or as expr: F.col("location") @@ -3157,7 +3184,7 @@ checks = [ # is_geo_covers check — precise mode (geo, requires runtime 17.1+) # uses st_covers; includes boundary points unlike st_contains - DQDatasetRule( + DQRowRule( criticality="error", check_func=geo_check_funcs.is_geo_covers, column="location", # or as expr: F.col("location") @@ -3171,7 +3198,7 @@ checks = [ # is_geo_intersects check — approximate mode (geo, requires runtime 17.1+) # at least one shared H3 cell between location and the reference polygon - DQDatasetRule( + DQRowRule( criticality="error", check_func=geo_check_funcs.is_geo_intersects, column="location", # or as expr: F.col("location") @@ -3183,7 +3210,7 @@ checks = [ # is_geo_intersects check — precise mode (geo, requires runtime 17.1+) # uses st_intersects for exact computation - DQDatasetRule( + DQRowRule( criticality="error", check_func=geo_check_funcs.is_geo_intersects, column="location", # or as expr: F.col("location") @@ -3197,7 +3224,7 @@ checks = [ # is_geo_touches check (geo, precise only, requires runtime 17.1+) # location point must touch (share a boundary point with) the reference polygon - DQDatasetRule( + DQRowRule( criticality="error", check_func=geo_check_funcs.is_geo_touches, column="location", # or as expr: F.col("location") @@ -3210,7 +3237,7 @@ checks = [ # is_geo_within check (geo, precise only, requires runtime 17.1+) # reference geometry must be within the column geometry (converse of is_geo_contains) - DQDatasetRule( + DQRowRule( criticality="error", check_func=geo_check_funcs.is_geo_within, column="region", # or as expr: F.col("region") diff --git a/src/databricks/labs/dqx/geo/check_funcs.py b/src/databricks/labs/dqx/geo/check_funcs.py index 757a5e8ae..d24fa99de 100644 --- a/src/databricks/labs/dqx/geo/check_funcs.py +++ b/src/databricks/labs/dqx/geo/check_funcs.py @@ -1,4 +1,5 @@ from collections.abc import Callable +import math import operator as py_operator import uuid from typing import Literal @@ -1340,3 +1341,104 @@ def is_geo_within( return _has_topological_relationship_precise( column, reference_geometry, convert_column, convert_reference_geometry, "WITHIN" ) + + +@requires_dbr_version("17.1") +@register_rule("row") +def is_geo_within_distance( + column: str | Column, + reference_geometry: str | bytes | Column, + distance: int | float | str | Column, + convert_column: bool = False, + convert_reference_geometry: bool = False, +) -> Column: + """Checks if the column geography is within a geodesic distance of the reference geography using `st_distance`. + + The distance is measured in meters along the WGS 84 ellipsoid, so the check is meaningful for + global data where planar `GEOMETRY` distances are not. A value is reported when the shortest + distance between it and the reference geography is strictly greater than *distance*. + + Both the target column and the reference geometry are always handled as `GEOGRAPHY`. + When conversion is requested (*convert_column* or *convert_reference_geometry* set to True), + *try_to_geography* is applied to parse the value from any supported format (WKT, WKB, EWKT, EWKB, + GeoJSON). + See https://docs.databricks.com/aws/en/sql/language-manual/functions/try_to_geography for details. + When conversion is not requested, the input is assumed to already hold a native `GEOGRAPHY` value. + + Args: + column: Column to check. Null values are skipped for validation. + reference_geometry: Reference geography as a literal WKT/EWKT/GeoJSON string or WKB/EWKB bytes + value, or a Column expression (e.g. *F.col('col_name')*) to reference another column. A + plain string is always treated as a literal, not a column name. + distance: Maximum allowed distance in meters. Accepts a non-negative number, a Column + expression (e.g. *F.col('radius_m')*), or a string SQL expression evaluated against the + input DataFrame. Rows where the distance expression evaluates to null are skipped. + convert_column: When True, *try_to_geography* is applied to convert column values to GEOGRAPHY. + When False (default), the column is assumed to already hold a native GEOGRAPHY value. + convert_reference_geometry: When True, *try_to_geography* is applied to convert the reference + geometry to GEOGRAPHY. When False (default), the reference geometry is assumed to already + hold a native GEOGRAPHY value. + + Returns: + Column object indicating whether values in the input column are farther than *distance* meters + from the reference geography. + + Raises: + InvalidParameterError: If *distance* is a boolean, or a numeric literal that is negative, + NaN or infinite. + + Note: + This function requires Databricks serverless compute or runtime 17.1 or above. + """ + # `bool` is a subclass of `int`, so it would otherwise slip through as a 0/1 metre radius. + if isinstance(distance, bool) or ( + isinstance(distance, (int, float)) and not (math.isfinite(distance) and distance >= 0) + ): + raise InvalidParameterError(f"'distance' must be a finite, non-negative number of meters, got {distance!r}.") + + col_str_norm, col_expr_str, col_expr = get_normalized_column_and_expr(column) + + ref_col = reference_geometry if isinstance(reference_geometry, Column) else F.lit(reference_geometry) + col_geog = F.call_function("try_to_geography", col_expr) if convert_column else col_expr + ref_geog = F.call_function("try_to_geography", ref_col) if convert_reference_geometry else ref_col + distance_expr = get_limit_expr(distance) + + # `try_to_geography` yields NULL for values that fail to parse. The column and the reference are + # reported separately so the error message points at the value the user has to fix. Null input + # values are skipped before these are evaluated, so a NULL here always means "unparseable". + col_invalid = col_geog.isNull() + ref_invalid = ref_geog.isNull() + is_too_far = F.call_function("st_distance", col_geog, ref_geog) > distance_expr + + condition = F.when(col_expr.isNull(), F.lit(None)).otherwise(col_invalid | ref_invalid | is_too_far) + + # How the offending value is rendered depends on the input contract. When the column is converted, + # it holds a WKT/WKB-style value that casts to string losslessly, and the raw text is what the user + # needs to see - `st_astext` would be NULL on exactly the unparseable values the message is about. + # When conversion is off the column is already a native GEOGRAPHY, which has no string cast, so the + # value has to be rendered with `st_astext`; a NULL there is already excluded by the null guard. + text_value_col = col_expr.cast("string") if convert_column else F.call_function("st_astext", col_geog) + + invalid_column_message = F.concat_ws( + "", + F.lit("value `"), + text_value_col, + F.lit(f"` in column `{col_expr_str}` is not a valid geography"), + ) + invalid_reference_message = F.lit(f"reference geometry for column `{col_expr_str}` is not a valid geography") + too_far_message = F.concat_ws( + "", + F.lit("value `"), + text_value_col, + F.lit(f"` in column `{col_expr_str}` is farther than "), + distance_expr.cast("string"), + F.lit(" meters from the reference geometry"), + ) + + return make_condition( + condition, + F.when(col_invalid, invalid_column_message) + .when(ref_invalid, invalid_reference_message) + .otherwise(too_far_message), + alias=f"{col_str_norm}_is_not_within_distance_from_reference_geometry", + ) diff --git a/tests/integration/test_apply_checks.py b/tests/integration/test_apply_checks.py index 008f08257..b9b067819 100755 --- a/tests/integration/test_apply_checks.py +++ b/tests/integration/test_apply_checks.py @@ -7742,6 +7742,29 @@ def test_apply_checks_all_geo_checks_using_classes(skip_if_runtime_not_geo_compa column=F.col("polygon_geom"), check_func_kwargs={"value": 2}, ), + # is_geo_within_distance check (geo, geodesic distance in meters, requires runtime 17.1+) + DQRowRule( + criticality="error", + check_func=geo_check_funcs.is_geo_within_distance, + column="point_geom", + check_func_kwargs={ + "reference_geometry": "POINT(1 1)", + "distance": 1000, + "convert_column": True, + "convert_reference_geometry": True, + }, + ), + DQRowRule( + criticality="error", + check_func=geo_check_funcs.is_geo_within_distance, + column=F.col("point_geom"), + check_func_kwargs={ + "reference_geometry": "POINT(1 1)", + "distance": 1000, + "convert_column": True, + "convert_reference_geometry": True, + }, + ), ] dq_engine = DQEngine(ws) diff --git a/tests/integration/test_row_checks_geo.py b/tests/integration/test_row_checks_geo.py index 0354a84b0..f5853dc28 100644 --- a/tests/integration/test_row_checks_geo.py +++ b/tests/integration/test_row_checks_geo.py @@ -1,3 +1,4 @@ +import pyspark.sql.functions as F from pyspark.testing.utils import assertDataFrameEqual from databricks.labs.dqx.geo.check_funcs import ( is_area_equal_to, @@ -30,6 +31,7 @@ is_geo_intersects, is_geo_touches, is_geo_within, + is_geo_within_distance, ) _POINT_INSIDE = "POINT(4.9 52.37)" @@ -47,6 +49,8 @@ _INTERSECTS_APPROXIMATE_SCHEMA = "geom: string, geom_does_not_intersect_reference_geometry_approximately: string" _TOUCHES_SCHEMA = "geom: string, geom_does_not_touch_reference_geometry: string" _WITHIN_SCHEMA = "geom: string, geom_does_not_contain_reference_geometry: string" +_WITHIN_DISTANCE_SCHEMA = "geom: string, geom_is_not_within_distance_from_reference_geometry: string" +_WITHIN_DISTANCE_CONDITION_SCHEMA = "geom_is_not_within_distance_from_reference_geometry: string" def _contains_violation(value: str) -> str: @@ -77,6 +81,10 @@ def _within_violation(value: str) -> str: return f"value `{value}` in column `geom` does not contain reference geometry" +def _within_distance_violation(value: str, distance: int) -> str: + return f"value `{value}` in column `geom` is farther than {distance} meters from the reference geometry" + + def test_is_geometry(skip_if_runtime_not_geo_compatible, spark): input_schema = "geom_string: string, geom_binary: binary, geom_int: int" test_df = spark.createDataFrame( @@ -1162,3 +1170,121 @@ def test_is_geo_within_exterior_reference_violation(skip_if_runtime_not_geo_comp [[column_polygon, _within_violation(column_polygon)], [None, None]], _WITHIN_SCHEMA ) assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_inside_radius_no_violation(skip_if_runtime_not_geo_compatible, spark): + """A point roughly 68 m from the reference is inside the 1 km radius — no violation.""" + point = "POINT(4.901 52.37)" + test_df = spark.createDataFrame([[point], [None]], _GEO_SCHEMA) + condition = is_geo_within_distance( + "geom", _POINT_INSIDE, 1000, convert_column=True, convert_reference_geometry=True + ) + actual = test_df.select("geom", condition) + expected = spark.createDataFrame([[point, None], [None, None]], _WITHIN_DISTANCE_SCHEMA) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_on_reference_no_violation(skip_if_runtime_not_geo_compatible, spark): + """A point identical to the reference is at distance 0, which passes even a zero radius.""" + test_df = spark.createDataFrame([[_POINT_INSIDE], [None]], _GEO_SCHEMA) + condition = is_geo_within_distance("geom", _POINT_INSIDE, 0, convert_column=True, convert_reference_geometry=True) + actual = test_df.select("geom", condition) + expected = spark.createDataFrame([[_POINT_INSIDE, None], [None, None]], _WITHIN_DISTANCE_SCHEMA) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_outside_radius_violation(skip_if_runtime_not_geo_compatible, spark): + """A point roughly 10 km from the reference is outside the 1 km radius — violation.""" + point = "POINT(5.05 52.37)" + test_df = spark.createDataFrame([[point], [None]], _GEO_SCHEMA) + condition = is_geo_within_distance( + "geom", _POINT_INSIDE, 1000, convert_column=True, convert_reference_geometry=True + ) + actual = test_df.select("geom", condition) + expected = spark.createDataFrame( + [[point, _within_distance_violation(point, 1000)], [None, None]], _WITHIN_DISTANCE_SCHEMA + ) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_with_column_distance(skip_if_runtime_not_geo_compatible, spark): + """The radius can vary per row when supplied as a column expression.""" + near, far = "POINT(4.901 52.37)", "POINT(5.05 52.37)" + test_df = spark.createDataFrame([[near, 1000], [far, 1000], [far, 20000]], "geom: string, radius_m: int") + condition = is_geo_within_distance( + "geom", _POINT_INSIDE, F.col("radius_m"), convert_column=True, convert_reference_geometry=True + ) + actual = test_df.select("geom", condition) + expected = spark.createDataFrame( + [[near, None], [far, _within_distance_violation(far, 1000)], [far, None]], _WITHIN_DISTANCE_SCHEMA + ) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_null_distance_is_skipped(skip_if_runtime_not_geo_compatible, spark): + """A null radius makes the comparison unknown, so the row is skipped rather than flagged.""" + far = "POINT(5.05 52.37)" + test_df = spark.createDataFrame([[far, None]], "geom: string, radius_m: int") + condition = is_geo_within_distance( + "geom", _POINT_INSIDE, F.col("radius_m"), convert_column=True, convert_reference_geometry=True + ) + actual = test_df.select("geom", condition) + expected = spark.createDataFrame([[far, None]], _WITHIN_DISTANCE_SCHEMA) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_invalid_column_violation(skip_if_runtime_not_geo_compatible, spark): + """An unparseable column value is reported as an invalid geography, not as a distance violation.""" + test_df = spark.createDataFrame([[_POINT_INVALID], [None]], _GEO_SCHEMA) + condition = is_geo_within_distance( + "geom", _POINT_INSIDE, 1000, convert_column=True, convert_reference_geometry=True + ) + actual = test_df.select("geom", condition) + expected = spark.createDataFrame( + [ + [_POINT_INVALID, f"value `{_POINT_INVALID}` in column `geom` is not a valid geography"], + [None, None], + ], + _WITHIN_DISTANCE_SCHEMA, + ) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_native_geography_no_violation(skip_if_runtime_not_geo_compatible, spark): + """With convert_column=False both inputs are already native GEOGRAPHY values.""" + test_df = spark.createDataFrame([["POINT(4.901 52.37)"], [None]], _GEO_SCHEMA).select( + F.call_function("try_to_geography", F.col("geom")).alias("geom") + ) + condition = is_geo_within_distance("geom", F.call_function("try_to_geography", F.lit(_POINT_INSIDE)), 1000) + actual = test_df.select(condition) + expected = spark.createDataFrame([[None], [None]], _WITHIN_DISTANCE_CONDITION_SCHEMA) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_native_geography_violation(skip_if_runtime_not_geo_compatible, spark): + """A native GEOGRAPHY value outside the radius is flagged, with the value rendered via st_astext.""" + point = "POINT(5.05 52.37)" + test_df = spark.createDataFrame([[point]], _GEO_SCHEMA).select( + F.call_function("try_to_geography", F.col("geom")).alias("geom") + ) + condition = is_geo_within_distance("geom", F.call_function("try_to_geography", F.lit(_POINT_INSIDE)), 1000) + actual = test_df.select(condition) + expected = spark.createDataFrame([[_within_distance_violation(point, 1000)]], _WITHIN_DISTANCE_CONDITION_SCHEMA) + assertDataFrameEqual(actual, expected, checkRowOrder=False) + + +def test_is_geo_within_distance_invalid_reference_violation(skip_if_runtime_not_geo_compatible, spark): + """An unparseable reference is reported separately from an unparseable column value.""" + test_df = spark.createDataFrame([[_POINT_INSIDE], [None]], _GEO_SCHEMA) + condition = is_geo_within_distance( + "geom", _REF_POLYGON_INVALID, 1000, convert_column=True, convert_reference_geometry=True + ) + actual = test_df.select("geom", condition) + expected = spark.createDataFrame( + [ + [_POINT_INSIDE, "reference geometry for column `geom` is not a valid geography"], + [None, None], + ], + _WITHIN_DISTANCE_SCHEMA, + ) + assertDataFrameEqual(actual, expected, checkRowOrder=False) diff --git a/tests/perf/test_apply_checks.py b/tests/perf/test_apply_checks.py index c1874c392..e0098f2d6 100644 --- a/tests/perf/test_apply_checks.py +++ b/tests/perf/test_apply_checks.py @@ -2177,6 +2177,31 @@ def test_benchmark_is_geo_covers_precise(benchmark, ws, generated_df): assert actual_count == EXPECTED_ROWS +def test_benchmark_is_geo_within_distance(benchmark, ws, generated_df): + """Benchmark `is_geo_within_distance`. + + Uses col_geo_point against a reference point with a 1 km radius to benchmark the geodesic + `st_distance` path on GEOGRAPHY values. + """ + dq_engine = DQEngine(workspace_client=ws, extra_params=EXTRA_PARAMS) + checks = [ + DQRowRule( + criticality="warn", + check_func=geo_check_funcs.is_geo_within_distance, + column="col_geo_point", + check_func_kwargs={ + "reference_geometry": "POINT(4.90 52.37)", + "distance": 1000, + "convert_column": True, + "convert_reference_geometry": True, + }, + ) + ] + checked = dq_engine.apply_checks(generated_df, checks) + actual_count = benchmark(lambda: checked.count()) + assert actual_count == EXPECTED_ROWS + + def test_benchmark_is_geo_covers_approximate(benchmark, ws, generated_df): """Benchmark `is_geo_covers` approximate version. diff --git a/tests/resources/all_row_geo_checks.yaml b/tests/resources/all_row_geo_checks.yaml index f3f4da64c..904c67b50 100644 --- a/tests/resources/all_row_geo_checks.yaml +++ b/tests/resources/all_row_geo_checks.yaml @@ -259,3 +259,14 @@ convert_column: true convert_reference_geometry: true +# is_geo_within_distance check (geo, geodesic distance in meters, requires runtime 17.1+) +# location point must lie within 1 km of the reference point +- criticality: error + check: + function: is_geo_within_distance + arguments: + column: point_geom + reference_geometry: "POINT(1 1)" + distance: 1000 + convert_column: true + convert_reference_geometry: true diff --git a/tests/unit/test_check_func_signatures.py b/tests/unit/test_check_func_signatures.py index fc6160ad7..d256323d5 100644 --- a/tests/unit/test_check_func_signatures.py +++ b/tests/unit/test_check_func_signatures.py @@ -187,6 +187,13 @@ ), "is_geo_touches": ("column", "reference_geometry", "convert_column", "convert_reference_geometry"), "is_geo_within": ("column", "reference_geometry", "convert_column", "convert_reference_geometry"), + "is_geo_within_distance": ( + "column", + "reference_geometry", + "distance", + "convert_column", + "convert_reference_geometry", + ), "does_not_contain_pii": ("column", "language", "threshold", "entities", "nlp_engine_config"), } diff --git a/tests/unit/test_geo_check_funcs.py b/tests/unit/test_geo_check_funcs.py index 86454a5df..6e3d2170d 100644 --- a/tests/unit/test_geo_check_funcs.py +++ b/tests/unit/test_geo_check_funcs.py @@ -1,4 +1,5 @@ import pytest +import pyspark.sql.functions as F from databricks.labs.dqx.errors import InvalidParameterError from databricks.labs.dqx.geo.check_funcs import ( @@ -7,9 +8,12 @@ is_geo_intersects, is_geo_touches, is_geo_within, + is_geo_within_distance, ) _REFERENCE_GEOMETRY_WKT = "POLYGON((0 0, 10 0, 10 10, 0 10, 0 0))" +_REFERENCE_POINT_WKT = "POINT(4.90 52.37)" +_REFERENCE_POINT_WKB = bytes.fromhex("0101000000B81E85EB51981340F6285C8FC2354A40") def test_is_geo_contains_does_not_raise(): @@ -36,6 +40,57 @@ def test_is_geo_within_with_conversion_does_not_raise(): is_geo_within("location", _REFERENCE_GEOMETRY_WKT, convert_column=True, convert_reference_geometry=True) +def test_is_geo_within_distance_does_not_raise(): + is_geo_within_distance("location", _REFERENCE_POINT_WKT, 1000) + + +def test_is_geo_within_distance_with_conversion_does_not_raise(): + is_geo_within_distance("location", _REFERENCE_POINT_WKT, 1000, convert_column=True, convert_reference_geometry=True) + + +def test_is_geo_within_distance_with_column_reference_does_not_raise(): + is_geo_within_distance("location", F.col("reference_location"), 1000) + + +def test_is_geo_within_distance_with_bytes_reference_does_not_raise(): + is_geo_within_distance("location", _REFERENCE_POINT_WKB, 1000, convert_reference_geometry=True) + + +def test_is_geo_within_distance_with_column_distance_does_not_raise(): + is_geo_within_distance("location", _REFERENCE_POINT_WKT, F.col("radius_m")) + + +def test_is_geo_within_distance_with_expression_distance_does_not_raise(): + is_geo_within_distance("location", _REFERENCE_POINT_WKT, "radius_m * 2") + + +def test_is_geo_within_distance_accepts_zero_distance(): + is_geo_within_distance("location", _REFERENCE_POINT_WKT, 0) + + +@pytest.mark.parametrize("distance", [-1, -0.5, float("nan"), float("inf"), float("-inf"), True, False]) +def test_is_geo_within_distance_rejects_invalid_distance(distance): + with pytest.raises(InvalidParameterError, match="finite, non-negative"): + is_geo_within_distance("location", _REFERENCE_GEOMETRY_WKT, distance) + + +def test_is_geo_within_distance_without_conversion_has_proper_alias(): + """The native GEOGRAPHY path renders values with st_astext, but must keep the same alias.""" + column = is_geo_within_distance("location", F.col("reference_location"), 1000) + column_str = _column_expression_clean(column) + assert column_str.endswith( + "location_is_not_within_distance_from_reference_geometry" + ), f'{column_str} has incorrect alias suffix' + + +def test_is_geo_within_distance_has_proper_alias(): + column = is_geo_within_distance("location", _REFERENCE_POINT_WKT, 1000) + column_str = _column_expression_clean(column) + assert column_str.endswith( + "location_is_not_within_distance_from_reference_geometry" + ), f'{column_str} has incorrect alias suffix' + + def test_is_geo_covers_precise_does_not_raise(): is_geo_covers("location", _REFERENCE_GEOMETRY_WKT, precise=True)