Add row-level geo distance check and fix documentation examples

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`) |
 </details>
 
 <Admonition type="warning" title="Applicability">
@@ -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
 ```
 
 <Admonition type="info" title="Decimal Values in Metadata Format">
@@ -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
 <details style={{ backgroundColor: 'transparent', color: 'neutral' }}>
 <summary>**Checks defined programmatically using DQX classes**</summary>
 ```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")

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",
+    )

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)