Made merge columns optional in sql query check (#945)

### PR summary:

1. Made sql_query support **merge_columns be optional/empty** and
documenting/demoing the two modes (row-level with merge_columns vs.
dataset-level). Added notebooks/docs/tests to prove the new behaviour,
including row-filters, metrics observers, ref DF support, negation, and
performance comparisons.

1. **Follow-up polish:** centralised the dataset-level logic into
_apply_dataset_level_sql_check, reused the new path in integration
tests, and tightened unit expectations so helpers (like DQDatasetRule)
stay type-stable.

1. **Hardening + helper cleanup**: reintroduced merge-column validation
to fail fast on bad inputs, forced dataset-level queries to error if
they emit more than one row, and added the shared
build_quality_violation helper so all integration expectations reuse the
same metadata factory. Also refreshed formatting/linting.

****TL;DR:** sql_query now covers both dataset-level and row-level
scenarios with clear validation and deterministic outputs, plus
comprehensive tests and docs to back it up.**

## Changes
<!-- Summary of your changes that are easy to understand. Add
screenshots when necessary -->

### Linked issues
<!-- DOC: Link issue with a keyword: close, closes, closed, fix, fixes,
fixed, resolve, resolves, resolved. See
https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword
-->

Resolves #938 

### Tests
<!-- How is this tested? Please see the checklist below and also
describe any other relevant tests -->

- [x] manually tested
- [x] added unit tests
- [x] added integration tests
- [ ] added end-to-end tests
- [ ] added performance tests

---------

Co-authored-by: Greg Hansen <[email protected]>
Co-authored-by: Marcin Wojtyczka <[email protected]>
Co-authored-by: Copilot <[email protected]>
Co-authored-by: mwojtyczka <[email protected]>

Author: 4 people

demos/dqx_demo_library.py

@@ -943,11 +943,15 @@ def not_ends_with(column: str, suffix: str) -> Column:
 # COMMAND ----------
 
 # MAGIC %md
-# MAGIC #### Using `sql_query` check
+# MAGIC #### Using `sql_query` check - Row-level validation
+# MAGIC 
+# MAGIC The `sql_query` check supports two modes:
+# MAGIC - **Row-level validation** (with `merge_columns`): Query results are joined back to mark specific rows
+# MAGIC - **Dataset-level validation** (without `merge_columns`): Check result applies to all rows
 
 # COMMAND ----------
 
-# using DQX classes
+# Row-level validation example: Check each sensor against its threshold
 from databricks.labs.dqx.rule import DQDatasetRule
 from databricks.labs.dqx.check_funcs import sql_query
 
@@ -973,7 +977,7 @@ def not_ends_with(column: str, suffix: str) -> Column:
         check_func=sql_query,
         check_func_kwargs={
             "query": query,
-            "merge_columns": ["sensor_id"],
+            "merge_columns": ["sensor_id"],  # Results joined back by sensor_id
             "condition_column": "condition",  # the check fails if this column evaluates to True
             "msg": "one of the sensor reading is greater than limit",
             "name": "sensor_reading_check",
@@ -990,6 +994,41 @@ def not_ends_with(column: str, suffix: str) -> Column:
 
 # COMMAND ----------
 
+# MAGIC %md
+# MAGIC #### Using `sql_query` check - Dataset-level validation
+# MAGIC 
+# MAGIC When `merge_columns` is not provided, the check applies to all rows (all pass or all fail together).
+# MAGIC This is useful for dataset-level aggregate validations.
+
+# COMMAND ----------
+
+# Dataset-level validation example: Check total sensor count
+dataset_query = """
+    SELECT COUNT(DISTINCT sensor_id) < 1 AS condition
+    FROM {{ sensor }}
+"""
+
+checks = [
+    DQDatasetRule(
+        criticality="warn",
+        check_func=sql_query,
+        check_func_kwargs={
+            "query": dataset_query,
+            # No merge_columns = dataset-level check (all rows get same result)
+            "condition_column": "condition",
+            "msg": "Dataset has no sensors",
+            "name": "dataset_has_sensors",
+            "input_placeholder": "sensor",
+        },
+    ),
+]
+
+ref_dfs = {"sensor_specs": sensor_specs_df}
+valid_and_quarantine_df = dq_engine.apply_checks(sensor_df, checks, ref_dfs=ref_dfs)
+display(valid_and_quarantine_df)
+
+# COMMAND ----------
+
 # using YAML declarative approach
 checks = yaml.safe_load(
     """
@@ -1028,6 +1067,30 @@ def not_ends_with(column: str, suffix: str) -> Column:
 
 # COMMAND ----------
 
+# YAML example for dataset-level validation (without merge_columns)
+checks_dataset_level = yaml.safe_load(
+    """
+    - criticality: warn
+      check:
+        function: sql_query
+        arguments:
+          # No merge_columns = dataset-level validation
+          condition_column: condition
+          msg: Dataset has no sensors
+          name: dataset_has_sensors
+          input_placeholder: sensor
+          query: |
+            SELECT COUNT(DISTINCT sensor_id) < 1 AS condition
+            FROM {{ sensor }}
+    """
+)
+
+ref_dfs = {"sensor_specs": sensor_specs_df}
+valid_and_quarantine_df = dq_engine.apply_checks_by_metadata(sensor_df, checks_dataset_level, ref_dfs=ref_dfs)
+display(valid_and_quarantine_df)
+
+# COMMAND ----------
+
 # MAGIC %md
 # MAGIC #### Defining custom python dataset-level check
 

docs/dqx/docs/reference/benchmarks.mdx

@@ -23,6 +23,7 @@ sidebar_position: 13
 | test_benchmark_compare_datasets | 3.598445 | 3.556993 | 3.430710 | 3.793938 | 0.158157 | 0.280218 | 3.466942 | 3.747160 | 5 | 0 | 2 | 0.28 |
 | test_benchmark_foreach_compare_datasets[n_rows_100000000_n_columns_5] | 25.879615 | 25.919933 | 25.536855 | 26.071184 | 0.217230 | 0.307223 | 25.748681 | 26.055904 | 5 | 0 | 1 | 0.04 |
 | test_benchmark_foreach_foreign_key[n_rows_100000000_n_columns_5] | 24.264873 | 22.893218 | 20.587308 | 29.037093 | 4.062789 | 7.705522 | 20.652819 | 28.358341 | 5 | 0 | 1 | 0.04 |
+| test_benchmark_foreach_has_no_outliers[n_rows_100000000_n_columns_5] | 22.524313 | 22.347593 | 22.104944 | 22.924248 | 0.374170 | 0.646915 | 22.271984 | 22.918899 | 5 | 0 | 3 | 0.04 |
 | test_benchmark_foreach_has_valid_schema[n_rows_100000000_n_columns_5] | 1.068582 | 1.050490 | 0.979350 | 1.219259 | 0.092674 | 0.112164 | 1.003924 | 1.116088 | 5 | 0 | 1 | 0.94 |
 | test_benchmark_foreach_is_aggr_equal[n_rows_100000000_n_columns_5] | 1.239298 | 1.213153 | 1.192442 | 1.341836 | 0.060654 | 0.068928 | 1.200719 | 1.269646 | 5 | 0 | 1 | 0.81 |
 | test_benchmark_foreach_is_aggr_not_equal[n_rows_100000000_n_columns_5] | 1.264898 | 1.250273 | 1.218577 | 1.345211 | 0.051090 | 0.071957 | 1.225905 | 1.297862 | 5 | 0 | 1 | 0.79 |
@@ -54,6 +55,7 @@ sidebar_position: 13
 | test_benchmark_foreach_sql_query[n_rows_100000000_n_columns_5] | 4.578799 | 4.602143 | 4.442396 | 4.644892 | 0.083901 | 0.113694 | 4.530776 | 4.644470 | 5 | 0 | 1 | 0.22 |
 | test_benchmark_foreign_key | 31.784272 | 31.787610 | 31.414708 | 32.123221 | 0.269713 | 0.386951 | 31.597198 | 31.984149 | 5 | 0 | 2 | 0.03 |
 | test_benchmark_has_dimension | 0.215338 | 0.213285 | 0.210530 | 0.223131 | 0.005056 | 0.007086 | 0.211819 | 0.218905 | 5 | 0 | 1 | 4.64 |
+| test_benchmark_has_no_outliers | 0.234952 | 0.228169 | 0.224165 | 0.257274 | 0.013649 | 0.017354 | 0.225936 | 0.243290 | 5 | 0 | 1 | 4.26 |
 | test_benchmark_has_valid_schema | 0.172078 | 0.172141 | 0.163793 | 0.181081 | 0.006715 | 0.009295 | 0.167010 | 0.176305 | 6 | 0 | 2 | 5.81 |
 | test_benchmark_has_x_coordinate_between | 0.217192 | 0.213656 | 0.209310 | 0.236233 | 0.011150 | 0.012638 | 0.209410 | 0.222048 | 5 | 0 | 1 | 4.60 |
 | test_benchmark_has_y_coordinate_between | 0.218497 | 0.219630 | 0.209352 | 0.234111 | 0.010103 | 0.013743 | 0.209584 | 0.223327 | 5 | 0 | 1 | 4.58 |