Merge remote-tracking branch 'origin/main' into HEAD

# Conflicts:
#	src/databricks/labs/dqx/check_funcs.py

Author: mwojtyczka

docs/dqx/docs/guide/quality_checks_definition.mdx

@@ -342,7 +342,7 @@ Checks defined using declarative syntax must contain the following fields:
    - `arguments`: keyword arguments passed to the check function. Every parameter without a default in the function’s Python signature must appear here.
    - `for_each_column`: (optional) list of column names or expressions for which the same check should be applied. When saving with `DQEngine.save_checks` to Delta or Lakebase tables, DQX stores such checks in compact format (one row per check, `for_each_column` preserved).
 - (optional) `name`: name of the check: autogenerated if not provided.
-- (optional) `filter`: spark expression to filter the rows for which the check is applied (e.g. `"business_unit = 'Finance'"`). The check function will run only on the rows matching the filter condition. The condition can reference any column of the validated dataset, not only the one where you apply the check function. When using dataset-level checks, the filter condition is pushed down as `row_filter` to the check function and applied before aggregation, ensuring that the check operates only on the relevant subset of rows rather than on the aggregated results.
+- (optional) `filter`: spark expression to filter the rows for which the check is applied (e.g. `"business_unit = 'Finance'"`). The check function will run only on the rows matching the filter condition. The condition can reference any column of the validated dataset, not only the one where you apply the check function. When using dataset-level checks, the filter condition is pushed down as `row_filter` to the check function and applied before aggregation, ensuring that the check operates only on the relevant subset of rows rather than on the aggregated results. Subqueries are allowed (e.g. `"customer_id IN (SELECT customer_id FROM main.ref.active_customers)"`), so you can scope a check using a reference table. A filter containing a destructive SQL keyword (e.g. `DELETE`, `DROP`, `INSERT`, `UPDATE`, `TRUNCATE`) is treated as invalid: that check is skipped (reported with `skipped = true` and a message identifying the unsafe filter) while the remaining checks still run.
 - (optional) `user_metadata`: key-value pairs added to the row-level warnings and errors
 
 ## YAML format (declarative approach)
@@ -699,7 +699,7 @@ The table used to store checks has the following structure. You can use `for_eac
 | └─ `function`          | `string`              | Name of the DQX check function to apply.                                                                                                                  |
 | └─ `for_each_column`   | `array<string>`       | (Optional) List of columns for which the same check applies. Stored in compact format (one row per check). Manual tables can use compact or expanded (individual rules instead of for_each_column) format. |
 | └─ `arguments`         | `map<string, string>` | Keyword arguments for the check function. All parameters without a default must be present (or supplied via `for_each_column` for `column` / `columns`).  |
-| `filter`               | `string`              | (Optional) Spark SQL expression to filter rows to which the check is applied, e.g. `"business_unit = 'Finance'"`. The check function will run only on the rows matching the filter condition. The condition can reference any column of the validated dataset, not only the one where you apply the check function. When using dataset-level checks, the filter condition is pushed down as `row_filter` to the check function and applied before aggregation, ensuring that the check operates only on the relevant subset of rows rather than on the aggregated results. |
+| `filter`               | `string`              | (Optional) Spark SQL expression to filter rows to which the check is applied, e.g. `"business_unit = 'Finance'"`. The check function will run only on the rows matching the filter condition. The condition can reference any column of the validated dataset, not only the one where you apply the check function. When using dataset-level checks, the filter condition is pushed down as `row_filter` to the check function and applied before aggregation, ensuring that the check operates only on the relevant subset of rows rather than on the aggregated results. Subqueries are allowed (e.g. `"customer_id IN (SELECT customer_id FROM main.ref.active_customers)"`), so you can scope a check using a reference table. A filter containing a destructive SQL keyword (e.g. `DELETE`, `DROP`, `INSERT`, `UPDATE`, `TRUNCATE`) is treated as invalid: that check is skipped (reported with `skipped = true` and a message identifying the unsafe filter) while the remaining checks still run.|
 | `run_config_name`      | `string`              | Name of the run config name. Could be any string such as input table or job name (use "default" if not provided). Useful for selecting applicable checks. |
 | `user_metadata`        | `map<string, string>` | (Optional) Custom metadata to add to any row-level warnings or errors generated by the check.                                                             |
 | `created_at`           | `timestamp`           | Current local date and time in UTC when the rule set was saved.                                                        |

mcp-server/notebooks/setup.py

@@ -68,6 +68,7 @@
 
 import logging
 import re
+import time
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("dqx-mcp-setup")
@@ -148,20 +149,53 @@ def _validate_principal(value, label):
 
 # Look up the app's service principal
 from databricks.sdk import WorkspaceClient
+from databricks.sdk.errors import NotFound
+
+# App provisioning is asynchronous: `bundle deploy` registers the app, but the app entity and its
+# service principal become queryable via the Apps API only some time later. Under account-wide load
+# (e.g. many CI runs deploying apps concurrently) that window widens, so a single immediate
+# ws.apps.get() right after deploy can race the provisioner and raise NotFound ("App ... does not
+# exist or is deleted"). Poll with backoff until the app exists AND its service_principal_client_id
+# is populated, rather than failing the whole setup on a transient not-ready state.
+_APP_LOOKUP_TIMEOUT_SECONDS = 300
+_APP_LOOKUP_POLL_SECONDS = 10
+
+
+def _get_app_with_sp(ws: WorkspaceClient, name: str):
+    """Return the app once it exists and its service principal client id is populated.
+
+    Retries on NotFound and on a not-yet-assigned SP for up to _APP_LOOKUP_TIMEOUT_SECONDS to
+    tolerate asynchronous app provisioning after `bundle deploy`.
+    """
+    deadline = time.monotonic() + _APP_LOOKUP_TIMEOUT_SECONDS
+    last_reason = "app not found"
+    while True:
+        try:
+            app = ws.apps.get(name)
+            if app.service_principal_client_id:
+                return app
+            last_reason = "service principal not yet assigned"
+        except NotFound:
+            last_reason = "app not found"
+        if time.monotonic() >= deadline:
+            raise TimeoutError(
+                f"App '{name}' was not fully provisioned within {_APP_LOOKUP_TIMEOUT_SECONDS}s "
+                f"({last_reason}). Ensure `bundle deploy`/`bundle run` for the app succeeded before setup."
+            )
+        logger.info(
+            f"Waiting for app '{name}' to be provisioned ({last_reason}); retrying in {_APP_LOOKUP_POLL_SECONDS}s"
+        )
+        time.sleep(_APP_LOOKUP_POLL_SECONDS)
+
 
 ws = WorkspaceClient()
-app = ws.apps.get(app_name)
+app = _get_app_with_sp(ws, app_name)
 
 # UC GRANTs use the SP's application id (client id), not its display_name or numeric SCIM id.
 # The Apps API returns it directly as service_principal_client_id — use that instead of a
 # ws.service_principals.get() SCIM lookup, which requires workspace-admin rights (the SCIM
 # GET /ServicePrincipals/{id} endpoint is admin-only) and would fail for a non-admin deployer.
 sp_principal = app.service_principal_client_id
-if not sp_principal:
-    raise ValueError(
-        f"App '{app_name}' has no service_principal_client_id — cannot resolve the app SP application id "
-        "for UC grants. Ensure the app is fully deployed before running setup."
-    )
 # Interpolated into the GRANT statements below, so validate it (backtick-breakout guard), same as
 # the runner SP application id.
 _validate_sp_id(sp_principal, "app service_principal_client_id")

src/databricks/labs/dqx/anomaly/scoring_utils.py

@@ -13,6 +13,7 @@
 from databricks.labs.dqx.anomaly.anomaly_info_schema import ai_explanation_struct_schema, anomaly_info_struct_schema
 from databricks.labs.dqx.anomaly.segment_utils import canonicalize_segment_values
 from databricks.labs.dqx.errors import InvalidParameterError
+from databricks.labs.dqx.utils import safe_filter_expr
 from databricks.labs.dqx.schema.dq_info_schema import (
     build_dq_info_struct,
     register_dq_info_field,
@@ -248,4 +249,4 @@ def apply_row_filter(df: DataFrame, row_filter: str | None) -> DataFrame:
     row_filter is a SQL expression (e.g. \"region = 'US'\"). Only these rows are run
     through anomaly detection; elsewhere we join results back so output has same row count.
     """
-    return df.filter(F.expr(row_filter)) if row_filter else df
+    return df.filter(safe_filter_expr(row_filter)) if row_filter else df

src/databricks/labs/dqx/check_funcs.py

@@ -18,6 +18,7 @@
 from databricks.labs.dqx.utils import (
     get_column_name_or_alias,
     is_sql_query_safe,
+    safe_filter_expr,
     normalize_col_str,
     get_columns_as_strings,
     to_lowercase,
@@ -430,6 +431,10 @@ def sql_expression(
 
     Args:
         expression: SQL expression. Fail if expression evaluates to False, pass if it evaluates to True.
+            Security note: this parameter accepts arbitrary SQL and is evaluated as-is, so it may
+            include subqueries and run with the permissions of the process executing the checks.
+            Only use check definitions from trusted sources, especially in automated or multi-tenant
+            pipelines.
         msg: optional message of the *Column* type, automatically generated if None
         name: optional name of the resulting column, automatically generated if None
         negate: if the condition should be negated (true) or not. For example, "col is not null" will mark null
@@ -1246,8 +1251,10 @@ def apply(df: DataFrame) -> DataFrame:
                 f"Column '{col_expr_str}' must be of numeric type to perform outlier detection using MAD method, "
                 f"but got type '{column_type.simpleString()}' instead."
             )
-        filter_condition = F.expr(row_filter) if row_filter else F.lit(True)
-        median, mad = _calculate_median_absolute_deviation(df, col_expr_str, row_filter)
+        # Validate and compile the filter (rejecting unsafe SQL); keep None when no filter is given so
+        # the MAD calculation stays a true no-op instead of filtering on F.lit(True).
+        filter_condition = safe_filter_expr(row_filter) if row_filter else None
+        median, mad = _calculate_median_absolute_deviation(df, col_expr_str, filter_condition)
         if median is not None and mad is not None:
             median = float(median)
             mad = float(mad)
@@ -1258,9 +1265,11 @@ def apply(df: DataFrame) -> DataFrame:
             upper_bound_expr = get_limit_expr(upper_bound)
 
             condition = (col_expr < (lower_bound_expr)) | (col_expr > (upper_bound_expr))
+            # Restrict the flagged rows to the filtered subset when a filter is present.
+            outlier_condition = (filter_condition & condition) if filter_condition is not None else condition
 
             # Add outlier detection columns
-            result_df = df.withColumn(condition_col, F.when(filter_condition & condition, True).otherwise(False))
+            result_df = df.withColumn(condition_col, F.when(outlier_condition, True).otherwise(False))
         else:
             # If median or mad could not be calculated, no outliers can be detected
             result_df = df.withColumn(condition_col, F.lit(False))
@@ -1337,7 +1346,7 @@ def apply(df: DataFrame) -> DataFrame:
 
         filter_condition = F.lit(True)
         if row_filter:
-            filter_condition = filter_condition & F.expr(row_filter)
+            filter_condition = filter_condition & safe_filter_expr(row_filter)
 
         if nulls_distinct:
             # All columns must be non-null
@@ -1466,7 +1475,7 @@ def apply(df: DataFrame, spark: SparkSession, ref_dfs: dict[str, DataFrame]) ->
         ref_alias = f"__ref_{col_str_norm}_{unique_str}"
         ref_df_distinct = ref_df.select(ref_col_expr.alias(ref_alias)).distinct()
 
-        filter_expr = F.expr(row_filter) if row_filter else F.lit(True)
+        filter_expr = safe_filter_expr(row_filter)
 
         # col_expr.isNotNull() only filters rows in the single-column non-null-safe path;
         # when col_expr is a struct (composite keys or null_safe=True), the struct is never NULL
@@ -1583,7 +1592,7 @@ def _replace_template(sql: str, replacements: dict[str, str]) -> str:
     def apply(df: DataFrame, spark: SparkSession, ref_dfs: dict[str, DataFrame]) -> DataFrame:
         filtered_df = df
         if row_filter:
-            filtered_df = df.filter(F.expr(row_filter))
+            filtered_df = df.filter(safe_filter_expr(row_filter))
 
         # since the check could be applied multiple times, the views created here must be unique
         filtered_df.createOrReplaceTempView(unique_input_view)
@@ -1956,8 +1965,7 @@ def apply(df: DataFrame) -> DataFrame:
                 f"but got type '{time_col_type.simpleString()}' instead."
             )
 
-        filter_col = F.expr(row_filter) if row_filter else F.lit(True)
-        filtered_expr = F.when(filter_col, aggr_col_expr) if row_filter else aggr_col_expr
+        filtered_expr = F.when(safe_filter_expr(row_filter), aggr_col_expr) if row_filter else aggr_col_expr
         aggr_expr = _build_aggregate_expression(aggr_type, filtered_expr, aggr_params)
 
         group_cols = [F.col(c) if isinstance(c, str) else c for c in (group_by or [])]
@@ -2358,7 +2366,7 @@ def apply(df: DataFrame, spark: SparkSession, ref_dfs: dict[str, DataFrame]) ->
         skipped_columns = [col for col in df.columns if col not in compare_columns and col not in pk_column_names]
 
         # apply filter before aliasing to avoid ambiguity
-        df = df.withColumn(filter_col, F.expr(row_filter) if row_filter else F.lit(True))
+        df = df.withColumn(filter_col, safe_filter_expr(row_filter))
 
         df = df.alias("df")
         ref_df = ref_df.alias("ref_df")
@@ -2465,7 +2473,7 @@ def apply(df: DataFrame) -> DataFrame:
         # Build filter condition
         filter_condition = F.lit(True)
         if row_filter:
-            filter_condition = filter_condition & F.expr(row_filter)
+            filter_condition = filter_condition & safe_filter_expr(row_filter)
 
         # Limit checking to be within the lookback window if needed
         if lookback_windows is not None:
@@ -3584,14 +3592,14 @@ def apply(df: DataFrame) -> DataFrame:
         Returns:
             The DataFrame with additional condition and metric columns for aggregation validation.
         """
-        filter_col = F.expr(row_filter) if row_filter else F.lit(True)
         if row_filter:
             # aggr_col_str == "*" only for count(*) over all rows (the only valid use of column="*").
             # F.expr("*") can't be embedded as the THEN value of a CASE WHEN: Spark's star-expansion
             # resolves it against every column in scope instead of treating it as a placeholder value.
             # count() only cares about nullness, so a non-null literal is a safe stand-in.
+            # safe_filter_expr rejects unsafe SQL in the filter (see #1303).
             then_expr = F.lit(1) if aggr_col_str == "*" else aggr_col_expr
-            filtered_expr = F.when(filter_col, then_expr)
+            filtered_expr = F.when(safe_filter_expr(row_filter), then_expr)
         else:
             filtered_expr = aggr_col_expr
 
@@ -4101,7 +4109,7 @@ def _apply_dataset_level_sql_check(
     # - If row_filter is provided, only rows matching the filter get the condition value
     # - Rows not matching the filter get None (consistent with row-level checks)
     if row_filter:
-        filter_expr = F.expr(row_filter)
+        filter_expr = safe_filter_expr(row_filter)
         result_df = df.withColumn(
             unique_condition_column, F.when(filter_expr, F.lit(condition_value)).otherwise(F.lit(None))
         )
@@ -4143,7 +4151,9 @@ def _validate_sql_query_params(query: str, merge_columns: list[str] | None) -> N
         raise InvalidParameterError("'merge_columns' entries must be non-empty strings.")
 
 
-def _calculate_median_absolute_deviation(df: DataFrame, column: str, filter_condition: str | None) -> tuple[Any, Any]:
+def _calculate_median_absolute_deviation(
+    df: DataFrame, column: str, filter_condition: Column | None
+) -> tuple[Any, Any]:
     """
     Calculate the Median Absolute Deviation (MAD) for a numeric column.
 

src/databricks/labs/dqx/geo/check_funcs.py

@@ -9,6 +9,7 @@
 from databricks.labs.dqx.rule import register_rule, requires_dbr_version
 from databricks.labs.dqx.check_funcs import make_condition, get_normalized_column_and_expr, get_limit_expr
 from databricks.labs.dqx.errors import InvalidParameterError
+from databricks.labs.dqx.utils import safe_filter_expr
 
 POINT_TYPE = "ST_Point"
 LINESTRING_TYPE = "ST_LineString"
@@ -939,7 +940,7 @@ def apply(df: DataFrame) -> DataFrame:
         df_with_hash = df.withColumn(row_id_col, F.xxhash64(F.struct("*")))
 
         # Apply filter after hashing (so hash is available for final join to preserve all rows)
-        filtered_df = df_with_hash.filter(F.expr(row_filter)) if row_filter is not None else df_with_hash
+        filtered_df = df_with_hash.filter(safe_filter_expr(row_filter)) if row_filter is not None else df_with_hash
 
         duplicates_df = filtered_df.groupBy(row_id_col).count().filter("count > 1").select(row_id_col)