From 1a39381e23945ed0f28f1df34ca9e37f47a85f9a Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Thu, 1 Oct 2026 11:31:44 +0300 Subject: [PATCH 01/12] Add result_owners: set test result owners from failing rows A test can now route its failures to different owners depending on which rows failed. Rules live under the test's meta; each pairs a SQL expression over the failing rows with a list of owners. Owners of all matching rules are unioned and replace the result's owners. When nothing matches, or the test passes, the result keeps its default owners. Supported for dbt tests (via the Elementary test materialization) and for anomaly tests, where expressions run against the anomalous rows only. Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_result_owners.py | 95 ++++++++++++ .../store_anomaly_test_results.sql | 8 ++ .../materializations/test/result_owners.sql | 135 ++++++++++++++++++ macros/edr/materializations/test/test.sql | 4 + 4 files changed, 242 insertions(+) create mode 100644 integration_tests/tests/test_result_owners.py create mode 100644 macros/edr/materializations/test/result_owners.sql diff --git a/integration_tests/tests/test_result_owners.py b/integration_tests/tests/test_result_owners.py new file mode 100644 index 000000000..b8b8cb058 --- /dev/null +++ b/integration_tests/tests/test_result_owners.py @@ -0,0 +1,95 @@ +import json +from datetime import datetime, timedelta +from typing import Any, Dict, List + +from data_generator import DATE_FORMAT, generate_dates +from dbt_project import DbtProject + +COLUMN_NAME = "country" +DEFAULT_OWNER = "central-team" +MODEL_CONFIG = {"meta": {"owner": DEFAULT_OWNER}} +ACCEPTED_VALUES_ARGS = {"values": ["NL"]} +TIMESTAMP_COLUMN = "updated_at" + + +def _owners(test_result: Dict[str, Any]) -> List[str]: + owners = test_result["owners"] + return json.loads(owners) if isinstance(owners, str) else owners + + +def _accepted_values_test( + test_id: str, dbt_project: DbtProject, data: List[dict], result_owners: list +): + return dbt_project.test( + test_id, + "accepted_values", + ACCEPTED_VALUES_ARGS, + test_column=COLUMN_NAME, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"result_owners": result_owners}}, + test_vars={"enable_elementary_test_materialization": True}, + ) + + +def test_matching_rules_union_owners(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: value} for value in ["NL", "JP", "JP", "KR"]] + result_owners = [ + {"expression": "value_field = 'JP'", "owners": ["japan@example.com"]}, + { + "expression": "value_field in ('JP', 'KR')", + "owners": "japan@example.com, korea@example.com", + }, + {"expression": "value_field = 'TW'", "owners": ["taiwan@example.com"]}, + ] + test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["japan@example.com", "korea@example.com"] + + +def test_no_matching_rule_keeps_default_owners(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + result_owners = [ + {"expression": "value_field = 'TW'", "owners": ["taiwan@example.com"]} + ] + test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + assert test_result["status"] == "fail" + assert _owners(test_result) == [DEFAULT_OWNER] + + +def test_passing_test_keeps_default_owners(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: "NL"}] + result_owners = [{"expression": "1 = 1", "owners": ["everyone@example.com"]}] + test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + assert test_result["status"] == "pass" + assert _owners(test_result) == [DEFAULT_OWNER] + + +def test_dimension_anomalies_result_owners(test_id: str, dbt_project: DbtProject): + utc_today = datetime.utcnow().date() + test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1)) + data: List[Dict[str, Any]] = [ + {TIMESTAMP_COLUMN: test_date.strftime(DATE_FORMAT), "superhero": superhero} + for superhero in ["Superman", "Superman", "Superman", "Spiderman"] + ] + [ + {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT), "superhero": superhero} + for cur_date in training_dates + for superhero in ["Superman", "Spiderman"] + ] + result_owners = [ + {"expression": "dimension_value = 'Superman'", "owners": ["dc@example.com"]}, + { + "expression": "dimension_value = 'Spiderman'", + "owners": ["marvel@example.com"], + }, + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["superhero"]}, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"result_owners": result_owners}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com"] diff --git a/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql b/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql index 48b349e88..cc8c80a7f 100644 --- a/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql +++ b/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql @@ -144,5 +144,13 @@ } %} {% set elementary_test_row = elementary.get_dbt_test_result_row(flattened_test) %} {% do elementary_test_row.update(test_result_dict) %} + {% if failures.data > 0 %} + {% set result_owners = elementary.get_anomaly_test_result_owners( + flattened_test, full_table_name, metric_name, column_name + ) %} + {% if result_owners %} + {% do elementary_test_row.update({"owners": result_owners}) %} + {% endif %} + {% endif %} {% do return(elementary_test_row) %} {% endmacro %} diff --git a/macros/edr/materializations/test/result_owners.sql b/macros/edr/materializations/test/result_owners.sql new file mode 100644 index 000000000..d22fd5a5e --- /dev/null +++ b/macros/edr/materializations/test/result_owners.sql @@ -0,0 +1,135 @@ +{# + Result owners let a single test route its failures to different owners, + depending on which rows failed. Configured on the test's meta: + + meta: + result_owners: + - expression: "dimension_value = 'JAPAN'" + owners: ["japan-dataops@example.com"] + + Each expression is evaluated against the test's failing rows. The owners of + every matching rule are unioned and replace the result's owners. When no rule + matches, the result keeps its default owners. +#} +{% macro get_result_owners_rules(flattened_test) %} + {% set meta = elementary.insensitive_get_dict_value(flattened_test, "meta") or {} %} + {% set rules = meta.get("result_owners") %} + {% if not rules %} {% do return([]) %} {% endif %} + {% if rules is mapping %} {% set rules = [rules] %} {% endif %} + + {% set test_unique_id = elementary.insensitive_get_dict_value( + flattened_test, "unique_id" + ) %} + {% if rules is string or rules is not iterable %} + {% do exceptions.raise_compiler_error( + "result_owners of test `{}` must be a list of rules, each with an `expression` and `owners`.".format( + test_unique_id + ) + ) %} + {% endif %} + {% for rule in rules %} + {% if rule is not mapping or rule.get( + "expression" + ) is not string or not rule.get("expression") or not rule.get("owners") %} + {% do exceptions.raise_compiler_error( + "Invalid result_owners rule in test `{}`: {}. Each rule needs a string `expression` and `owners`.".format( + test_unique_id, rule + ) + ) %} + {% endif %} + {% endfor %} + {% do return(rules) %} +{% endmacro %} + +{# Accepts the same formats as model owners: a string, a comma-separated string, or a list. #} +{% macro normalize_result_owners(owners) %} + {% set normalized = [] %} + {% if owners is string %} {% set owners = owners.split(",") %} + {% elif owners is mapping %} {% set owners = [owners] %} + {% endif %} + {% for owner in owners %} + {% if owner is mapping %} + {% set owner = owner.get("email") or owner.get("name") %} + {% endif %} + {% if owner is string and owner | trim %} + {% do normalized.append(owner | trim) %} + {% endif %} + {% endfor %} + {% do return(normalized) %} +{% endmacro %} + +{% macro get_result_owners_select_clause(rules) %} + {%- set rule_columns = [] -%} + {%- for rule in rules -%} + {%- do rule_columns.append( + "max(case when (" + ~ rule.expression + ~ ") then 1 else 0 end) as result_owners_rule_" + ~ loop.index0 + ) -%} + {%- endfor -%} + {%- do return("select " ~ rule_columns | join(", ")) -%} +{% endmacro %} + +{# Runs a query built from get_result_owners_select_clause and returns the matched owners, or none. #} +{% macro get_matched_result_owners(rules, result_owners_query) %} + {% set rows = elementary.agate_to_dicts( + elementary.run_query(result_owners_query) + ) %} + {% if not rows %} {% do return(none) %} {% endif %} + {% set matched_owners = [] %} + {% for rule in rules %} + {% set is_match = elementary.insensitive_get_dict_value( + rows[0], "result_owners_rule_" ~ loop.index0 + ) %} + {% if is_match is not none and is_match | int > 0 %} + {% for owner in elementary.normalize_result_owners(rule.owners) %} + {% if owner not in matched_owners %} + {% do matched_owners.append(owner) %} + {% endif %} + {% endfor %} + {% endif %} + {% endfor %} + {% do return(matched_owners or none) %} +{% endmacro %} + +{# + For dbt tests, the failing rows are the rows returned by the test's SQL. + On T-SQL (and with tests_use_temp_tables) `sql` already selects from a temp + table, so wrapping it in a derived table is safe there too. +#} +{% macro get_dbt_test_result_owners(flattened_test) %} + {% set rules = elementary.get_result_owners_rules(flattened_test) %} + {% if not rules or elementary.did_test_pass() %} {% do return(none) %} {% endif %} + {% set result_owners_query %} + {{ elementary.get_result_owners_select_clause(rules) }} + from ({{ sql }} + ) results + {% endset %} + {% do return(elementary.get_matched_result_owners(rules, result_owners_query)) %} +{% endmacro %} + +{# + For anomaly tests, the failing rows are the anomalous rows of one result group. + select_override keeps this a single CTE chain, which T-SQL requires. +#} +{% macro get_anomaly_test_result_owners( + flattened_test, full_table_name, metric_name, column_name +) %} + {% set rules = elementary.get_result_owners_rules(flattened_test) %} + {% if not rules %} {% do return(none) %} {% endif %} + {% set group_filter %} + {{ elementary.edr_is_true("is_anomalous") }} + and upper(full_table_name) = upper({{ elementary.const_as_string(full_table_name) }}) + and metric_name = {{ elementary.const_as_string(metric_name) }} + {%- if column_name %} + and upper(column_name) = upper({{ elementary.const_as_string(column_name) }}) + {%- endif %} + {% endset %} + {% set result_owners_query = elementary.get_read_anomaly_scores_query( + flattened_test, + additional_where=group_filter, + select_override=elementary.get_result_owners_select_clause(rules), + ) %} + {% do return(elementary.get_matched_result_owners(rules, result_owners_query)) %} +{% endmacro %} diff --git a/macros/edr/materializations/test/test.sql b/macros/edr/materializations/test/test.sql index 90bb23487..cc87f51a1 100644 --- a/macros/edr/materializations/test/test.sql +++ b/macros/edr/materializations/test/test.sql @@ -120,6 +120,10 @@ {% set elementary_test_results_row = elementary.get_dbt_test_result_row( flattened_test, result_rows ) %} + {% set result_owners = elementary.get_dbt_test_result_owners(flattened_test) %} + {% if result_owners %} + {% do elementary_test_results_row.update({"owners": result_owners}) %} + {% endif %} {% do elementary.cache_elementary_test_results_rows( [elementary_test_results_row] ) %} From 0d9c4bc7feab61509e7542366de3487369b4e76f Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Thu, 1 Oct 2026 11:54:55 +0300 Subject: [PATCH 02/12] Set the source owner under config.meta in result_owners tests dbt Fusion rejects meta placed directly on a source table, so the test errored before writing a result. Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_result_owners.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/integration_tests/tests/test_result_owners.py b/integration_tests/tests/test_result_owners.py index b8b8cb058..bdd177623 100644 --- a/integration_tests/tests/test_result_owners.py +++ b/integration_tests/tests/test_result_owners.py @@ -7,7 +7,7 @@ COLUMN_NAME = "country" DEFAULT_OWNER = "central-team" -MODEL_CONFIG = {"meta": {"owner": DEFAULT_OWNER}} +MODEL_CONFIG = {"config": {"meta": {"owner": DEFAULT_OWNER}}} ACCEPTED_VALUES_ARGS = {"values": ["NL"]} TIMESTAMP_COLUMN = "updated_at" From b9df80b078b826005bc93e6b15c9c66108df38b8 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Sun, 4 Oct 2026 18:05:26 +0300 Subject: [PATCH 03/12] Address review: one grouped anomaly query, sorted owners, more tests - Anomaly tests evaluate result_owners in a single query grouped by full_table_name, column_name and metric_name, instead of one anomaly scores query per result group (adds a group_by option to get_read_anomaly_scores_query). - result_owners rules are validated on every anomaly test run, not only when there are anomalies. - Matched owners are sorted, like the default owners. - Warn when result_owners is set on a test type that ignores it (schema change tests). - Tests: sorted owners, warn status, partial match, multi-column dimension, column anomalies per metric, volume anomalies. Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_result_owners.py | 153 +++++++++++++++++- .../store_anomaly_test_results.sql | 27 ++-- .../materializations/test/result_owners.sql | 83 +++++++--- macros/edr/materializations/test/test.sql | 12 ++ .../tests/test_utils/get_anomaly_query.sql | 4 +- 5 files changed, 240 insertions(+), 39 deletions(-) diff --git a/integration_tests/tests/test_result_owners.py b/integration_tests/tests/test_result_owners.py index bdd177623..beaf42d06 100644 --- a/integration_tests/tests/test_result_owners.py +++ b/integration_tests/tests/test_result_owners.py @@ -1,6 +1,6 @@ import json from datetime import datetime, timedelta -from typing import Any, Dict, List +from typing import Any, Dict, List, Optional from data_generator import DATE_FORMAT, generate_dates from dbt_project import DbtProject @@ -18,7 +18,11 @@ def _owners(test_result: Dict[str, Any]) -> List[str]: def _accepted_values_test( - test_id: str, dbt_project: DbtProject, data: List[dict], result_owners: list + test_id: str, + dbt_project: DbtProject, + data: List[dict], + result_owners: list, + test_config: Optional[Dict[str, Any]] = None, ): return dbt_project.test( test_id, @@ -27,7 +31,7 @@ def _accepted_values_test( test_column=COLUMN_NAME, data=data, model_config=MODEL_CONFIG, - test_config={"meta": {"result_owners": result_owners}}, + test_config={**(test_config or {}), "meta": {"result_owners": result_owners}}, test_vars={"enable_elementary_test_materialization": True}, ) @@ -93,3 +97,146 @@ def test_dimension_anomalies_result_owners(test_id: str, dbt_project: DbtProject ) assert test_result["status"] == "fail" assert _owners(test_result) == ["dc@example.com"] + + +def test_matched_owners_are_sorted(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: "JP"}] + result_owners = [ + {"expression": "1 = 1", "owners": ["b@example.com", "a@example.com"]} + ] + test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + assert _owners(test_result) == ["a@example.com", "b@example.com"] + + +def test_warn_status_gets_result_owners(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + result_owners = [ + {"expression": "value_field = 'JP'", "owners": ["japan@example.com"]} + ] + test_result = _accepted_values_test( + test_id, dbt_project, data, result_owners, test_config={"severity": "warn"} + ) + assert test_result["status"] == "warn" + assert _owners(test_result) == ["japan@example.com"] + + +def _dimension_data( + test_day_rows: List[Dict[str, Any]], training_rows: List[Dict[str, Any]] +) -> List[Dict[str, Any]]: + utc_today = datetime.utcnow().date() + test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1)) + return [ + {TIMESTAMP_COLUMN: test_date.strftime(DATE_FORMAT), **row} + for row in test_day_rows + ] + [ + {TIMESTAMP_COLUMN: cur_date.strftime(DATE_FORMAT), **row} + for cur_date in training_dates + for row in training_rows + ] + + +def test_dimension_anomalies_partial_match(test_id: str, dbt_project: DbtProject): + # Both dimension values are anomalous, but only one has a rule. + data = _dimension_data( + [{"superhero": "Superman"}] * 3 + [{"superhero": "Spiderman"}] * 3, + [{"superhero": "Superman"}, {"superhero": "Spiderman"}], + ) + result_owners = [ + {"expression": "dimension_value = 'Superman'", "owners": ["dc@example.com"]} + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["superhero"]}, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"result_owners": result_owners}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com"] + + +def test_multi_column_dimension_result_owners(test_id: str, dbt_project: DbtProject): + # Multi-column dimension values are joined with "; ". + data = _dimension_data( + [{"universe": "DC", "superhero": "Superman"}] * 3 + + [{"universe": "Marvel", "superhero": "Spiderman"}], + [ + {"universe": "DC", "superhero": "Superman"}, + {"universe": "Marvel", "superhero": "Spiderman"}, + ], + ) + result_owners = [ + { + "expression": "dimension_value = 'DC; Superman'", + "owners": ["dc@example.com"], + }, + { + "expression": "dimension_value = 'Marvel; Spiderman'", + "owners": ["marvel@example.com"], + }, + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + { + "timestamp_column": TIMESTAMP_COLUMN, + "dimensions": ["universe", "superhero"], + }, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"result_owners": result_owners}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com"] + + +def test_column_anomalies_result_owners_per_metric( + test_id: str, dbt_project: DbtProject +): + # One result per metric: each result only gets the owners of its own rows. + data = _dimension_data( + [{"superhero": None}] * 3, + [{"superhero": "Superman"}, {"superhero": "Batman"}], + ) + result_owners = [ + {"expression": "metric_name = 'null_count'", "owners": ["nulls@example.com"]} + ] + test_results = dbt_project.test( + test_id, + "elementary.column_anomalies", + { + "timestamp_column": TIMESTAMP_COLUMN, + "column_anomalies": ["null_count", "null_percent"], + }, + data=data, + test_column="superhero", + model_config=MODEL_CONFIG, + test_config={"meta": {"result_owners": result_owners}}, + multiple_results=True, + ) + owners_by_metric = { + result["test_sub_type"]: _owners(result) for result in test_results + } + assert {result["status"] for result in test_results} == {"fail"} + assert owners_by_metric == { + "null_count": ["nulls@example.com"], + "null_percent": [DEFAULT_OWNER], + } + + +def test_volume_anomalies_result_owners(test_id: str, dbt_project: DbtProject): + data = _dimension_data([{}] * 6, [{}]) + result_owners = [ + {"expression": "metric_name = 'row_count'", "owners": ["volume@example.com"]} + ] + test_result = dbt_project.test( + test_id, + "elementary.volume_anomalies", + {"timestamp_column": TIMESTAMP_COLUMN}, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"result_owners": result_owners}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["volume@example.com"] diff --git a/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql b/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql index cc8c80a7f..4175190ff 100644 --- a/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql +++ b/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql @@ -15,6 +15,10 @@ ) %} {% endfor %} + {% set result_owners_by_group = elementary.get_anomaly_result_owners_by_group( + flattened_test, anomaly_scores_rows + ) %} + {% set elementary_test_results_rows = [] %} {% for anomaly_scores_group, anomaly_scores_rows in anomaly_scores_groups_rows.items() %} {% do elementary.debug_log( @@ -24,14 +28,24 @@ ) %} {% do elementary_test_results_rows.append( elementary.get_anomaly_test_result_row( - flattened_test, anomaly_scores_rows + flattened_test, + anomaly_scores_rows, + result_owners_by_group.get( + elementary.get_anomaly_result_owners_group_key( + anomaly_scores_group[0], + anomaly_scores_group[1], + anomaly_scores_group[2], + ) + ), ) ) %} {% endfor %} {% do elementary.cache_elementary_test_results_rows(elementary_test_results_rows) %} {% endmacro %} -{% macro get_anomaly_test_result_row(flattened_test, anomaly_scores_rows) %} +{% macro get_anomaly_test_result_row( + flattened_test, anomaly_scores_rows, result_owners=none +) %} {%- set latest_row = anomaly_scores_rows[-1] %} {%- set rows_with_score = ( anomaly_scores_rows | rejectattr("anomaly_score", "none") | list @@ -144,13 +158,8 @@ } %} {% set elementary_test_row = elementary.get_dbt_test_result_row(flattened_test) %} {% do elementary_test_row.update(test_result_dict) %} - {% if failures.data > 0 %} - {% set result_owners = elementary.get_anomaly_test_result_owners( - flattened_test, full_table_name, metric_name, column_name - ) %} - {% if result_owners %} - {% do elementary_test_row.update({"owners": result_owners}) %} - {% endif %} + {% if failures.data > 0 and result_owners %} + {% do elementary_test_row.update({"owners": result_owners}) %} {% endif %} {% do return(elementary_test_row) %} {% endmacro %} diff --git a/macros/edr/materializations/test/result_owners.sql b/macros/edr/materializations/test/result_owners.sql index d22fd5a5e..0aab399c0 100644 --- a/macros/edr/materializations/test/result_owners.sql +++ b/macros/edr/materializations/test/result_owners.sql @@ -71,16 +71,12 @@ {%- do return("select " ~ rule_columns | join(", ")) -%} {% endmacro %} -{# Runs a query built from get_result_owners_select_clause and returns the matched owners, or none. #} -{% macro get_matched_result_owners(rules, result_owners_query) %} - {% set rows = elementary.agate_to_dicts( - elementary.run_query(result_owners_query) - ) %} - {% if not rows %} {% do return(none) %} {% endif %} +{# Returns the sorted owners of the rules matched in a row of get_result_owners_select_clause, or none. #} +{% macro get_matched_result_owners(rules, result_row) %} {% set matched_owners = [] %} {% for rule in rules %} {% set is_match = elementary.insensitive_get_dict_value( - rows[0], "result_owners_rule_" ~ loop.index0 + result_row, "result_owners_rule_" ~ loop.index0 ) %} {% if is_match is not none and is_match | int > 0 %} {% for owner in elementary.normalize_result_owners(rule.owners) %} @@ -90,7 +86,7 @@ {% endfor %} {% endif %} {% endfor %} - {% do return(matched_owners or none) %} + {% do return((matched_owners | sort | list) or none) %} {% endmacro %} {# @@ -106,30 +102,65 @@ from ({{ sql }} ) results {% endset %} - {% do return(elementary.get_matched_result_owners(rules, result_owners_query)) %} + {% set rows = elementary.agate_to_dicts( + elementary.run_query(result_owners_query) + ) %} + {% if not rows %} {% do return(none) %} {% endif %} + {% do return(elementary.get_matched_result_owners(rules, rows[0])) %} +{% endmacro %} + +{% macro get_anomaly_result_owners_group_key( + full_table_name, column_name, metric_name +) %} + {% do return( + (full_table_name or "") + | upper ~ "|" ~ (column_name or "") + | upper ~ "|" ~ (metric_name or "") + ) %} {% endmacro %} {# - For anomaly tests, the failing rows are the anomalous rows of one result group. - select_override keeps this a single CTE chain, which T-SQL requires. + For anomaly tests, the failing rows are the anomalous rows of each result group + (full_table_name, column_name, metric_name). All groups are evaluated in one + grouped query; select_override keeps it a single CTE chain, which T-SQL requires. + Returns a dict of group key (see get_anomaly_result_owners_group_key) to owners. + The rules are validated on every run, even when nothing is anomalous. #} -{% macro get_anomaly_test_result_owners( - flattened_test, full_table_name, metric_name, column_name -) %} +{% macro get_anomaly_result_owners_by_group(flattened_test, anomaly_scores_rows) %} {% set rules = elementary.get_result_owners_rules(flattened_test) %} - {% if not rules %} {% do return(none) %} {% endif %} - {% set group_filter %} - {{ elementary.edr_is_true("is_anomalous") }} - and upper(full_table_name) = upper({{ elementary.const_as_string(full_table_name) }}) - and metric_name = {{ elementary.const_as_string(metric_name) }} - {%- if column_name %} - and upper(column_name) = upper({{ elementary.const_as_string(column_name) }}) - {%- endif %} - {% endset %} + {% if not rules %} {% do return({}) %} {% endif %} + {% if not (anomaly_scores_rows | selectattr("is_anomalous") | list) %} + {% do return({}) %} + {% endif %} {% set result_owners_query = elementary.get_read_anomaly_scores_query( flattened_test, - additional_where=group_filter, - select_override=elementary.get_result_owners_select_clause(rules), + additional_where=elementary.edr_is_true("is_anomalous"), + select_override=elementary.get_result_owners_select_clause(rules) + ~ ", full_table_name, column_name, metric_name", + group_by="full_table_name, column_name, metric_name", ) %} - {% do return(elementary.get_matched_result_owners(rules, result_owners_query)) %} + {% set owners_by_group = {} %} + {% for row in elementary.agate_to_dicts( + elementary.run_query(result_owners_query) + ) %} + {% set owners = elementary.get_matched_result_owners(rules, row) %} + {% if owners %} + {% do owners_by_group.update( + { + elementary.get_anomaly_result_owners_group_key( + elementary.insensitive_get_dict_value( + row, "full_table_name" + ), + elementary.insensitive_get_dict_value( + row, "column_name" + ), + elementary.insensitive_get_dict_value( + row, "metric_name" + ), + ): owners + } + ) %} + {% endif %} + {% endfor %} + {% do return(owners_by_group) %} {% endmacro %} diff --git a/macros/edr/materializations/test/test.sql b/macros/edr/materializations/test/test.sql index cc87f51a1..a3bb1bc4d 100644 --- a/macros/edr/materializations/test/test.sql +++ b/macros/edr/materializations/test/test.sql @@ -35,6 +35,18 @@ ) %} {% if elementary_test_type and elementary_test_type != "with_context" %} {# Anomaly detection and schema change tests handle their own result row collection #} + {% if elementary_test_type != "anomaly_detection" and elementary.get_node_meta( + model + ).get( + "result_owners" + ) %} + {% do elementary.edr_log_warning( + test_unique_id + ~ ": result_owners is not supported for " + ~ short_name + ~ " tests and will be ignored." + ) %} + {% endif %} {% do return(materialization_macro()) %} {% endif %} {% endif %} diff --git a/macros/edr/tests/test_utils/get_anomaly_query.sql b/macros/edr/tests/test_utils/get_anomaly_query.sql index 00bc1d2a0..5486e0b54 100644 --- a/macros/edr/tests/test_utils/get_anomaly_query.sql +++ b/macros/edr/tests/test_utils/get_anomaly_query.sql @@ -80,7 +80,8 @@ flattened_test=none, additional_where=none, select_override=none, - extra_cte=none + extra_cte=none, + group_by=none ) %} {% if not flattened_test %} {% set flattened_test = elementary.flatten_test(model) %} @@ -169,6 +170,7 @@ {{ select_override if select_override else 'select * ' }} from final_results where {{ test_configuration.exclude_final_results }} {% if additional_where %} and {{ additional_where }} {% endif %} + {% if group_by %} group by {{ group_by }} {% endif %} {%- endset -%} {{- return(anomaly_query) -}} {% endmacro %} From 7f6d03467599e7d075bfa254fd4173289c6882e6 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Sun, 4 Oct 2026 18:19:34 +0300 Subject: [PATCH 04/12] Compare anomaly result group keys as-is Upper-casing the key could merge quoted columns that differ only in case. Co-Authored-By: Claude Opus 5.5 --- macros/edr/materializations/test/result_owners.sql | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/macros/edr/materializations/test/result_owners.sql b/macros/edr/materializations/test/result_owners.sql index 0aab399c0..7c122ed43 100644 --- a/macros/edr/materializations/test/result_owners.sql +++ b/macros/edr/materializations/test/result_owners.sql @@ -112,11 +112,9 @@ {% macro get_anomaly_result_owners_group_key( full_table_name, column_name, metric_name ) %} - {% do return( - (full_table_name or "") - | upper ~ "|" ~ (column_name or "") - | upper ~ "|" ~ (metric_name or "") - ) %} + {# Both sides of the lookup read the same anomaly scores rows, so values are + compared as-is; quoted identifiers that differ only in case stay distinct. #} + {% do return((full_table_name, column_name, metric_name)) %} {% endmacro %} {# From 978558200ea832a0245547c59fc9ca311282e29f Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Mon, 5 Oct 2026 11:44:19 +0300 Subject: [PATCH 05/12] Rename to conditional_result_owners / condition, cap at 100 Follows the design review: the meta key is conditional_result_owners and each item has a `condition` and `owners`, so the config isn't confused with the existing `owner` key or with alert rules. Configs with more than 100 conditions fail at compile time. The conditions query starts with a comment naming the test, to make a failing condition easy to trace. Co-Authored-By: Claude Opus 5.5 --- ...s.py => test_conditional_result_owners.py} | 93 ++++++++-------- ...ners.sql => conditional_result_owners.sql} | 102 +++++++++++------- macros/edr/materializations/test/test.sql | 4 +- 3 files changed, 114 insertions(+), 85 deletions(-) rename integration_tests/tests/{test_result_owners.py => test_conditional_result_owners.py} (72%) rename macros/edr/materializations/test/{result_owners.sql => conditional_result_owners.sql} (55%) diff --git a/integration_tests/tests/test_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py similarity index 72% rename from integration_tests/tests/test_result_owners.py rename to integration_tests/tests/test_conditional_result_owners.py index beaf42d06..0e754c71e 100644 --- a/integration_tests/tests/test_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -21,7 +21,7 @@ def _accepted_values_test( test_id: str, dbt_project: DbtProject, data: List[dict], - result_owners: list, + conditions: list, test_config: Optional[Dict[str, Any]] = None, ): return dbt_project.test( @@ -31,40 +31,43 @@ def _accepted_values_test( test_column=COLUMN_NAME, data=data, model_config=MODEL_CONFIG, - test_config={**(test_config or {}), "meta": {"result_owners": result_owners}}, + test_config={ + **(test_config or {}), + "meta": {"conditional_result_owners": conditions}, + }, test_vars={"enable_elementary_test_materialization": True}, ) -def test_matching_rules_union_owners(test_id: str, dbt_project: DbtProject): +def test_matching_conditions_union_owners(test_id: str, dbt_project: DbtProject): data = [{COLUMN_NAME: value} for value in ["NL", "JP", "JP", "KR"]] - result_owners = [ - {"expression": "value_field = 'JP'", "owners": ["japan@example.com"]}, + conditions = [ + {"condition": "value_field = 'JP'", "owners": ["japan@example.com"]}, { - "expression": "value_field in ('JP', 'KR')", + "condition": "value_field in ('JP', 'KR')", "owners": "japan@example.com, korea@example.com", }, - {"expression": "value_field = 'TW'", "owners": ["taiwan@example.com"]}, + {"condition": "value_field = 'TW'", "owners": ["taiwan@example.com"]}, ] - test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + test_result = _accepted_values_test(test_id, dbt_project, data, conditions) assert test_result["status"] == "fail" assert _owners(test_result) == ["japan@example.com", "korea@example.com"] -def test_no_matching_rule_keeps_default_owners(test_id: str, dbt_project: DbtProject): +def test_no_matching_condition_keeps_default_owners( + test_id: str, dbt_project: DbtProject +): data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] - result_owners = [ - {"expression": "value_field = 'TW'", "owners": ["taiwan@example.com"]} - ] - test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + conditions = [{"condition": "value_field = 'TW'", "owners": ["taiwan@example.com"]}] + test_result = _accepted_values_test(test_id, dbt_project, data, conditions) assert test_result["status"] == "fail" assert _owners(test_result) == [DEFAULT_OWNER] def test_passing_test_keeps_default_owners(test_id: str, dbt_project: DbtProject): data = [{COLUMN_NAME: "NL"}] - result_owners = [{"expression": "1 = 1", "owners": ["everyone@example.com"]}] - test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + conditions = [{"condition": "1 = 1", "owners": ["everyone@example.com"]}] + test_result = _accepted_values_test(test_id, dbt_project, data, conditions) assert test_result["status"] == "pass" assert _owners(test_result) == [DEFAULT_OWNER] @@ -80,10 +83,10 @@ def test_dimension_anomalies_result_owners(test_id: str, dbt_project: DbtProject for cur_date in training_dates for superhero in ["Superman", "Spiderman"] ] - result_owners = [ - {"expression": "dimension_value = 'Superman'", "owners": ["dc@example.com"]}, + conditions = [ + {"condition": "dimension_value = 'Superman'", "owners": ["dc@example.com"]}, { - "expression": "dimension_value = 'Spiderman'", + "condition": "dimension_value = 'Spiderman'", "owners": ["marvel@example.com"], }, ] @@ -93,7 +96,7 @@ def test_dimension_anomalies_result_owners(test_id: str, dbt_project: DbtProject {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["superhero"]}, data=data, model_config=MODEL_CONFIG, - test_config={"meta": {"result_owners": result_owners}}, + test_config={"meta": {"conditional_result_owners": conditions}}, ) assert test_result["status"] == "fail" assert _owners(test_result) == ["dc@example.com"] @@ -101,20 +104,16 @@ def test_dimension_anomalies_result_owners(test_id: str, dbt_project: DbtProject def test_matched_owners_are_sorted(test_id: str, dbt_project: DbtProject): data = [{COLUMN_NAME: "JP"}] - result_owners = [ - {"expression": "1 = 1", "owners": ["b@example.com", "a@example.com"]} - ] - test_result = _accepted_values_test(test_id, dbt_project, data, result_owners) + conditions = [{"condition": "1 = 1", "owners": ["b@example.com", "a@example.com"]}] + test_result = _accepted_values_test(test_id, dbt_project, data, conditions) assert _owners(test_result) == ["a@example.com", "b@example.com"] -def test_warn_status_gets_result_owners(test_id: str, dbt_project: DbtProject): +def test_warn_status_gets_conditional_owners(test_id: str, dbt_project: DbtProject): data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] - result_owners = [ - {"expression": "value_field = 'JP'", "owners": ["japan@example.com"]} - ] + conditions = [{"condition": "value_field = 'JP'", "owners": ["japan@example.com"]}] test_result = _accepted_values_test( - test_id, dbt_project, data, result_owners, test_config={"severity": "warn"} + test_id, dbt_project, data, conditions, test_config={"severity": "warn"} ) assert test_result["status"] == "warn" assert _owners(test_result) == ["japan@example.com"] @@ -136,13 +135,13 @@ def _dimension_data( def test_dimension_anomalies_partial_match(test_id: str, dbt_project: DbtProject): - # Both dimension values are anomalous, but only one has a rule. + # Both dimension values are anomalous, but only one has a condition. data = _dimension_data( [{"superhero": "Superman"}] * 3 + [{"superhero": "Spiderman"}] * 3, [{"superhero": "Superman"}, {"superhero": "Spiderman"}], ) - result_owners = [ - {"expression": "dimension_value = 'Superman'", "owners": ["dc@example.com"]} + conditions = [ + {"condition": "dimension_value = 'Superman'", "owners": ["dc@example.com"]} ] test_result = dbt_project.test( test_id, @@ -150,7 +149,7 @@ def test_dimension_anomalies_partial_match(test_id: str, dbt_project: DbtProject {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["superhero"]}, data=data, model_config=MODEL_CONFIG, - test_config={"meta": {"result_owners": result_owners}}, + test_config={"meta": {"conditional_result_owners": conditions}}, ) assert test_result["status"] == "fail" assert _owners(test_result) == ["dc@example.com"] @@ -166,13 +165,13 @@ def test_multi_column_dimension_result_owners(test_id: str, dbt_project: DbtProj {"universe": "Marvel", "superhero": "Spiderman"}, ], ) - result_owners = [ + conditions = [ { - "expression": "dimension_value = 'DC; Superman'", + "condition": "dimension_value = 'DC; Superman'", "owners": ["dc@example.com"], }, { - "expression": "dimension_value = 'Marvel; Spiderman'", + "condition": "dimension_value = 'Marvel; Spiderman'", "owners": ["marvel@example.com"], }, ] @@ -185,7 +184,7 @@ def test_multi_column_dimension_result_owners(test_id: str, dbt_project: DbtProj }, data=data, model_config=MODEL_CONFIG, - test_config={"meta": {"result_owners": result_owners}}, + test_config={"meta": {"conditional_result_owners": conditions}}, ) assert test_result["status"] == "fail" assert _owners(test_result) == ["dc@example.com"] @@ -199,8 +198,8 @@ def test_column_anomalies_result_owners_per_metric( [{"superhero": None}] * 3, [{"superhero": "Superman"}, {"superhero": "Batman"}], ) - result_owners = [ - {"expression": "metric_name = 'null_count'", "owners": ["nulls@example.com"]} + conditions = [ + {"condition": "metric_name = 'null_count'", "owners": ["nulls@example.com"]} ] test_results = dbt_project.test( test_id, @@ -212,7 +211,7 @@ def test_column_anomalies_result_owners_per_metric( data=data, test_column="superhero", model_config=MODEL_CONFIG, - test_config={"meta": {"result_owners": result_owners}}, + test_config={"meta": {"conditional_result_owners": conditions}}, multiple_results=True, ) owners_by_metric = { @@ -227,8 +226,8 @@ def test_column_anomalies_result_owners_per_metric( def test_volume_anomalies_result_owners(test_id: str, dbt_project: DbtProject): data = _dimension_data([{}] * 6, [{}]) - result_owners = [ - {"expression": "metric_name = 'row_count'", "owners": ["volume@example.com"]} + conditions = [ + {"condition": "metric_name = 'row_count'", "owners": ["volume@example.com"]} ] test_result = dbt_project.test( test_id, @@ -236,7 +235,17 @@ def test_volume_anomalies_result_owners(test_id: str, dbt_project: DbtProject): {"timestamp_column": TIMESTAMP_COLUMN}, data=data, model_config=MODEL_CONFIG, - test_config={"meta": {"result_owners": result_owners}}, + test_config={"meta": {"conditional_result_owners": conditions}}, ) assert test_result["status"] == "fail" assert _owners(test_result) == ["volume@example.com"] + + +def test_too_many_conditions_errors(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: "JP"}] + conditions = [ + {"condition": "1 = 1", "owners": [f"owner{i}@example.com"]} for i in range(101) + ] + test_result = _accepted_values_test(test_id, dbt_project, data, conditions) + assert test_result["status"] == "error" + assert "Compilation Error" in test_result["test_results_description"] diff --git a/macros/edr/materializations/test/result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql similarity index 55% rename from macros/edr/materializations/test/result_owners.sql rename to macros/edr/materializations/test/conditional_result_owners.sql index 7c122ed43..b6c57b9ea 100644 --- a/macros/edr/materializations/test/result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -1,44 +1,53 @@ {# - Result owners let a single test route its failures to different owners, - depending on which rows failed. Configured on the test's meta: + Conditional result owners let a single test send its failures to different + owners, depending on which rows failed. Configured on the test's meta: meta: - result_owners: - - expression: "dimension_value = 'JAPAN'" + conditional_result_owners: + - condition: "dimension_value = 'JAPAN'" owners: ["japan-dataops@example.com"] - Each expression is evaluated against the test's failing rows. The owners of - every matching rule are unioned and replace the result's owners. When no rule - matches, the result keeps its default owners. + Each condition is SQL evaluated against the test's failing rows. If any + condition matches, the result's owners are the union of the owners of every + matching condition, replacing the default owners. When none matches, the + result keeps its default owners. #} -{% macro get_result_owners_rules(flattened_test) %} +{% macro get_conditional_result_owners(flattened_test) %} + {% set max_conditions = 100 %} {% set meta = elementary.insensitive_get_dict_value(flattened_test, "meta") or {} %} - {% set rules = meta.get("result_owners") %} - {% if not rules %} {% do return([]) %} {% endif %} - {% if rules is mapping %} {% set rules = [rules] %} {% endif %} + {% set conditions = meta.get("conditional_result_owners") %} + {% if not conditions %} {% do return([]) %} {% endif %} + {% if conditions is mapping %} {% set conditions = [conditions] %} {% endif %} {% set test_unique_id = elementary.insensitive_get_dict_value( flattened_test, "unique_id" ) %} - {% if rules is string or rules is not iterable %} + {% if conditions is string or conditions is not iterable %} {% do exceptions.raise_compiler_error( - "result_owners of test `{}` must be a list of rules, each with an `expression` and `owners`.".format( + "conditional_result_owners of test `{}` must be a list, each item with a `condition` and `owners`.".format( test_unique_id ) ) %} {% endif %} - {% for rule in rules %} - {% if rule is not mapping or rule.get( - "expression" - ) is not string or not rule.get("expression") or not rule.get("owners") %} + {% if conditions | length > max_conditions %} + {% do exceptions.raise_compiler_error( + "conditional_result_owners of test `{}` has {} conditions; the maximum is {}.".format( + test_unique_id, conditions | length, max_conditions + ) + ) %} + {% endif %} + {% for item in conditions %} + {% if item is not mapping or item.get( + "condition" + ) is not string or not item.get("condition") or not item.get("owners") %} {% do exceptions.raise_compiler_error( - "Invalid result_owners rule in test `{}`: {}. Each rule needs a string `expression` and `owners`.".format( - test_unique_id, rule + "Invalid conditional_result_owners item in test `{}`: {}. Each item needs a string `condition` and `owners`.".format( + test_unique_id, item ) ) %} {% endif %} {% endfor %} - {% do return(rules) %} + {% do return(conditions) %} {% endmacro %} {# Accepts the same formats as model owners: a string, a comma-separated string, or a list. #} @@ -58,28 +67,35 @@ {% do return(normalized) %} {% endmacro %} -{% macro get_result_owners_select_clause(rules) %} - {%- set rule_columns = [] -%} - {%- for rule in rules -%} - {%- do rule_columns.append( +{# The leading comment names the test, so a failing condition is easy to trace in query history. #} +{% macro get_conditional_result_owners_select_clause(conditions, test_unique_id) %} + {%- set condition_columns = [] -%} + {%- for item in conditions -%} + {%- do condition_columns.append( "max(case when (" - ~ rule.expression - ~ ") then 1 else 0 end) as result_owners_rule_" + ~ item.condition + ~ ") then 1 else 0 end) as owners_condition_" ~ loop.index0 ) -%} {%- endfor -%} - {%- do return("select " ~ rule_columns | join(", ")) -%} + {%- do return( + "/* conditional_result_owners of " + ~ test_unique_id + ~ " */ select " + ~ condition_columns + | join(", ") + ) -%} {% endmacro %} -{# Returns the sorted owners of the rules matched in a row of get_result_owners_select_clause, or none. #} -{% macro get_matched_result_owners(rules, result_row) %} +{# Returns the sorted owners of the conditions matched in a row of get_conditional_result_owners_select_clause, or none. #} +{% macro get_matched_result_owners(conditions, result_row) %} {% set matched_owners = [] %} - {% for rule in rules %} + {% for item in conditions %} {% set is_match = elementary.insensitive_get_dict_value( - result_row, "result_owners_rule_" ~ loop.index0 + result_row, "owners_condition_" ~ loop.index0 ) %} {% if is_match is not none and is_match | int > 0 %} - {% for owner in elementary.normalize_result_owners(rule.owners) %} + {% for owner in elementary.normalize_result_owners(item.owners) %} {% if owner not in matched_owners %} {% do matched_owners.append(owner) %} {% endif %} @@ -95,10 +111,12 @@ table, so wrapping it in a derived table is safe there too. #} {% macro get_dbt_test_result_owners(flattened_test) %} - {% set rules = elementary.get_result_owners_rules(flattened_test) %} - {% if not rules or elementary.did_test_pass() %} {% do return(none) %} {% endif %} + {% set conditions = elementary.get_conditional_result_owners(flattened_test) %} + {% if not conditions or elementary.did_test_pass() %} + {% do return(none) %} + {% endif %} {% set result_owners_query %} - {{ elementary.get_result_owners_select_clause(rules) }} + {{ elementary.get_conditional_result_owners_select_clause(conditions, flattened_test.unique_id) }} from ({{ sql }} ) results {% endset %} @@ -106,7 +124,7 @@ elementary.run_query(result_owners_query) ) %} {% if not rows %} {% do return(none) %} {% endif %} - {% do return(elementary.get_matched_result_owners(rules, rows[0])) %} + {% do return(elementary.get_matched_result_owners(conditions, rows[0])) %} {% endmacro %} {% macro get_anomaly_result_owners_group_key( @@ -122,18 +140,20 @@ (full_table_name, column_name, metric_name). All groups are evaluated in one grouped query; select_override keeps it a single CTE chain, which T-SQL requires. Returns a dict of group key (see get_anomaly_result_owners_group_key) to owners. - The rules are validated on every run, even when nothing is anomalous. + The conditions are validated on every run, even when nothing is anomalous. #} {% macro get_anomaly_result_owners_by_group(flattened_test, anomaly_scores_rows) %} - {% set rules = elementary.get_result_owners_rules(flattened_test) %} - {% if not rules %} {% do return({}) %} {% endif %} + {% set conditions = elementary.get_conditional_result_owners(flattened_test) %} + {% if not conditions %} {% do return({}) %} {% endif %} {% if not (anomaly_scores_rows | selectattr("is_anomalous") | list) %} {% do return({}) %} {% endif %} {% set result_owners_query = elementary.get_read_anomaly_scores_query( flattened_test, additional_where=elementary.edr_is_true("is_anomalous"), - select_override=elementary.get_result_owners_select_clause(rules) + select_override=elementary.get_conditional_result_owners_select_clause( + conditions, flattened_test.unique_id + ) ~ ", full_table_name, column_name, metric_name", group_by="full_table_name, column_name, metric_name", ) %} @@ -141,7 +161,7 @@ {% for row in elementary.agate_to_dicts( elementary.run_query(result_owners_query) ) %} - {% set owners = elementary.get_matched_result_owners(rules, row) %} + {% set owners = elementary.get_matched_result_owners(conditions, row) %} {% if owners %} {% do owners_by_group.update( { diff --git a/macros/edr/materializations/test/test.sql b/macros/edr/materializations/test/test.sql index a3bb1bc4d..01205d654 100644 --- a/macros/edr/materializations/test/test.sql +++ b/macros/edr/materializations/test/test.sql @@ -38,11 +38,11 @@ {% if elementary_test_type != "anomaly_detection" and elementary.get_node_meta( model ).get( - "result_owners" + "conditional_result_owners" ) %} {% do elementary.edr_log_warning( test_unique_id - ~ ": result_owners is not supported for " + ~ ": conditional_result_owners is not supported for " ~ short_name ~ " tests and will be ignored." ) %} From 9fe0fbcd087ac0d63444dc6ee1ccce5ca791e1d5 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Wed, 7 Oct 2026 09:04:59 +0300 Subject: [PATCH 06/12] Test that schema change tests ignore conditional_result_owners Also document that the 100-condition cap is a deliberate guard. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_conditional_result_owners.py | 23 +++++++++++++++++++ .../test/conditional_result_owners.sql | 2 ++ 2 files changed, 25 insertions(+) diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py index 0e754c71e..c5f2bd3ea 100644 --- a/integration_tests/tests/test_conditional_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -2,6 +2,7 @@ from datetime import datetime, timedelta from typing import Any, Dict, List, Optional +import pytest from data_generator import DATE_FORMAT, generate_dates from dbt_project import DbtProject @@ -249,3 +250,25 @@ def test_too_many_conditions_errors(test_id: str, dbt_project: DbtProject): test_result = _accepted_values_test(test_id, dbt_project, data, conditions) assert test_result["status"] == "error" assert "Compilation Error" in test_result["test_results_description"] + + +# Schema change tests don't support conditional_result_owners: the config is +# ignored (with a warning), so results keep the default owners. +@pytest.mark.skip_targets(["databricks", "spark", "athena", "trino"]) +def test_schema_changes_ignore_conditional_owners( + test_id: str, dbt_project: DbtProject +): + conditions = [{"condition": "1 = 1", "owners": ["everyone@example.com"]}] + test_results = dbt_project.test( + test_id, + "elementary.schema_changes_from_baseline", + test_args={"fail_on_added": True}, + columns=[{"name": "id", "data_type": "integer"}], + data=[{"id": 1, "name": "Elon"}], + model_config=MODEL_CONFIG, + test_config={"meta": {"conditional_result_owners": conditions}}, + multiple_results=True, + ) + failed_results = [r for r in test_results if r["status"] == "fail"] + assert failed_results + assert all(_owners(r) == [DEFAULT_OWNER] for r in failed_results) diff --git a/macros/edr/materializations/test/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql index b6c57b9ea..be531831f 100644 --- a/macros/edr/materializations/test/conditional_result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -13,6 +13,8 @@ result keeps its default owners. #} {% macro get_conditional_result_owners(flattened_test) %} + {# A deliberate guard, not a warehouse limit: each condition becomes one + aggregate column in a single query, so this keeps that query a sane size. #} {% set max_conditions = 100 %} {% set meta = elementary.insensitive_get_dict_value(flattened_test, "meta") or {} %} {% set conditions = meta.get("conditional_result_owners") %} From c8d4875af94366aac1f601c800285273bc5167fd Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Wed, 7 Oct 2026 09:12:50 +0300 Subject: [PATCH 07/12] Raise the conditional_result_owners cap to 1500 Generated condition lists can go well past 100. 1500 stays under the strictest per-query column limits (Postgres 1664, Redshift 1600); a new test runs exactly 1500 conditions. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_conditional_result_owners.py | 14 +++++++++++++- .../test/conditional_result_owners.sql | 7 ++++--- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py index c5f2bd3ea..0b50f322a 100644 --- a/integration_tests/tests/test_conditional_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -242,10 +242,22 @@ def test_volume_anomalies_result_owners(test_id: str, dbt_project: DbtProject): assert _owners(test_result) == ["volume@example.com"] +def test_max_conditions_runs(test_id: str, dbt_project: DbtProject): + # 1500 conditions, one column each, must fit in a single query on every warehouse. + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + conditions = [ + {"condition": f"value_field = 'X{i}'", "owners": [f"owner{i}@example.com"]} + for i in range(1499) + ] + [{"condition": "value_field = 'JP'", "owners": ["japan@example.com"]}] + test_result = _accepted_values_test(test_id, dbt_project, data, conditions) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["japan@example.com"] + + def test_too_many_conditions_errors(test_id: str, dbt_project: DbtProject): data = [{COLUMN_NAME: "JP"}] conditions = [ - {"condition": "1 = 1", "owners": [f"owner{i}@example.com"]} for i in range(101) + {"condition": "1 = 1", "owners": [f"owner{i}@example.com"]} for i in range(1501) ] test_result = _accepted_values_test(test_id, dbt_project, data, conditions) assert test_result["status"] == "error" diff --git a/macros/edr/materializations/test/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql index be531831f..745cdaf44 100644 --- a/macros/edr/materializations/test/conditional_result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -13,9 +13,10 @@ result keeps its default owners. #} {% macro get_conditional_result_owners(flattened_test) %} - {# A deliberate guard, not a warehouse limit: each condition becomes one - aggregate column in a single query, so this keeps that query a sane size. #} - {% set max_conditions = 100 %} + {# A deliberate guard: each condition becomes one column in a single query, + and 1500 (plus the anomaly path's group columns) stays under the + strictest warehouse limits on columns per query (Postgres 1664, Redshift 1600). #} + {% set max_conditions = 1500 %} {% set meta = elementary.insensitive_get_dict_value(flattened_test, "meta") or {} %} {% set conditions = meta.get("conditional_result_owners") %} {% if not conditions %} {% do return([]) %} {% endif %} From 385d45e2b1b33ea429461cf0d20c632e5e95d745 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Tue, 13 Oct 2026 13:52:48 +0300 Subject: [PATCH 08/12] Add column rules to conditional_result_owners A rule can name a column of the tested model with the values to match, instead of a SQL condition. For dimension anomalies the column is matched against that dimension; for other tests against that column of the failing rows. A test that doesn't return the column skips the rule instead of erroring, so column rules are safe to set for many tests at once. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_conditional_result_owners.py | 86 +++++++ .../test/conditional_result_owners.sql | 228 ++++++++++++++---- 2 files changed, 265 insertions(+), 49 deletions(-) diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py index 0b50f322a..ece1fb755 100644 --- a/integration_tests/tests/test_conditional_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -264,6 +264,92 @@ def test_too_many_conditions_errors(test_id: str, dbt_project: DbtProject): assert "Compilation Error" in test_result["test_results_description"] + +def test_column_rule_on_dimension(test_id: str, dbt_project: DbtProject): + # A column rule names the dimension column, not dimension_value. + data = _dimension_data( + [{"superhero": "Superman"}] * 3 + [{"superhero": "Spiderman"}], + [{"superhero": "Superman"}, {"superhero": "Spiderman"}], + ) + rules = [ + {"column": "superhero", "values": ["Superman"], "owners": ["dc@example.com"]}, + {"column": "superhero", "values": "Spiderman", "owners": ["marvel@example.com"]}, + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["superhero"]}, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"conditional_result_owners": rules}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com"] + + +def test_column_rule_on_multi_column_dimension(test_id: str, dbt_project: DbtProject): + # Each rule matches its own part of the "; "-joined dimension value. + data = _dimension_data( + [{"universe": "DC", "superhero": "Superman"}] * 3 + + [{"universe": "Marvel", "superhero": "Spiderman"}], + [ + {"universe": "DC", "superhero": "Superman"}, + {"universe": "Marvel", "superhero": "Spiderman"}, + ], + ) + rules = [ + {"column": "Universe", "values": ["DC"], "owners": ["dc@example.com"]}, + {"column": "superhero", "values": ["Superman"], "owners": ["heroes@example.com"]}, + {"column": "universe", "values": ["Marvel"], "owners": ["marvel@example.com"]}, + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + { + "timestamp_column": TIMESTAMP_COLUMN, + "dimensions": ["universe", "superhero"], + }, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"conditional_result_owners": rules}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com", "heroes@example.com"] + + +def test_column_rule_on_returned_column(test_id: str, dbt_project: DbtProject): + # accepted_values returns value_field; column names match case-insensitively. + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + rules = [{"column": "VALUE_FIELD", "values": ["JP"], "owners": ["japan@example.com"]}] + test_result = _accepted_values_test(test_id, dbt_project, data, rules) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["japan@example.com"] + + +def test_column_rule_skipped_when_not_returned(test_id: str, dbt_project: DbtProject): + # accepted_values doesn't return dimension_value, so the rule is skipped + # instead of erroring the test, as a condition on that column would. + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + rules = [{"column": "dimension_value", "values": ["JP"], "owners": ["japan@example.com"]}] + test_result = _accepted_values_test(test_id, dbt_project, data, rules) + assert test_result["status"] == "fail" + assert _owners(test_result) == [DEFAULT_OWNER] + + +def test_rule_with_condition_and_column_errors(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: "JP"}] + rules = [ + { + "condition": "1 = 1", + "column": "value_field", + "values": ["JP"], + "owners": ["japan@example.com"], + } + ] + test_result = _accepted_values_test(test_id, dbt_project, data, rules) + assert test_result["status"] == "error" + assert "Compilation Error" in test_result["test_results_description"] + # Schema change tests don't support conditional_result_owners: the config is # ignored (with a warning), so results keep the default owners. @pytest.mark.skip_targets(["databricks", "spark", "athena", "trino"]) diff --git a/macros/edr/materializations/test/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql index 745cdaf44..a86f59cc3 100644 --- a/macros/edr/materializations/test/conditional_result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -6,11 +6,19 @@ conditional_result_owners: - condition: "dimension_value = 'JAPAN'" owners: ["japan-dataops@example.com"] + - column: country + values: [KOREA, TAIWAN] + owners: ["korea-dataops@example.com"] - Each condition is SQL evaluated against the test's failing rows. If any - condition matches, the result's owners are the union of the owners of every - matching condition, replacing the default owners. When none matches, the - result keeps its default owners. + Each condition is SQL evaluated against the test's failing rows. A column rule + names a column of the tested model instead: for dimension anomalies it is + matched against that dimension, for other tests against that column of the + failing rows, and it is skipped when the test doesn't return that column. That + makes column rules safe to set for many tests at once (for example on a folder), + while a condition errors on tests whose rows don't have the columns it uses. + If any rule matches, the result's owners are the union of the owners of every + matching rule, replacing the default owners. When none matches, the result + keeps its default owners. #} {% macro get_conditional_result_owners(flattened_test) %} {# A deliberate guard: each condition becomes one column in a single query, @@ -40,11 +48,15 @@ ) %} {% endif %} {% for item in conditions %} - {% if item is not mapping or item.get( + {% set is_condition_rule = item is mapping and item.get( "condition" - ) is not string or not item.get("condition") or not item.get("owners") %} + ) is string and item.get("condition") and "column" not in item %} + {% set is_column_rule = item is mapping and item.get( + "column" + ) is string and item.get("column") and "values" in item and "condition" not in item %} + {% if not (is_condition_rule or is_column_rule) or not item.get("owners") %} {% do exceptions.raise_compiler_error( - "Invalid conditional_result_owners item in test `{}`: {}. Each item needs a string `condition` and `owners`.".format( + "Invalid conditional_result_owners item in test `{}`: {}. Each item needs `owners`, and either a string `condition` or a `column` with `values`.".format( test_unique_id, item ) ) %} @@ -53,6 +65,33 @@ {% do return(conditions) %} {% endmacro %} +{# The values of a column rule as a list; a single value may be given without a list. #} +{% macro get_column_rule_values(rule) %} + {% set values = rule.get("values") %} + {% if values is string or values is number or values is none %} + {% do return([values]) %} + {% endif %} + {% do return(values | list) %} +{% endmacro %} + +{# A column rule as a SQL condition on a column of the failing rows. #} +{% macro get_column_rule_condition(quoted_column, rule) %} + {% set literals = [] %} + {% set checks = [] %} + {% for value in elementary.get_column_rule_values(rule) %} + {% if value is none %} + {% do checks.append(quoted_column ~ " is null") %} + {% elif value is number and value is not sameas true and value is not sameas false %} + {% do literals.append(value | string) %} + {% else %} {% do literals.append(elementary.edr_quote(value | string)) %} + {% endif %} + {% endfor %} + {% if literals %} + {% do checks.append(quoted_column ~ " in (" ~ literals | join(", ") ~ ")") %} + {% endif %} + {% do return(checks | join(" or ") or "1 = 0") %} +{% endmacro %} + {# Accepts the same formats as model owners: a string, a comma-separated string, or a list. #} {% macro normalize_result_owners(owners) %} {% set normalized = [] %} @@ -114,10 +153,42 @@ table, so wrapping it in a derived table is safe there too. #} {% macro get_dbt_test_result_owners(flattened_test) %} - {% set conditions = elementary.get_conditional_result_owners(flattened_test) %} - {% if not conditions or elementary.did_test_pass() %} - {% do return(none) %} + {% set rules = elementary.get_conditional_result_owners(flattened_test) %} + {% if not rules or elementary.did_test_pass() %} {% do return(none) %} {% endif %} + {% set conditions = rules | selectattr("condition", "defined") | list %} + {% set column_rules = rules | selectattr("column", "defined") | list %} + {% if column_rules %} + {# Column rules only apply to the columns the test returns, so read them first #} + {% set columns_query %} + select * from ({{ sql }} + ) results where 1 = 0 + {% endset %} + {% set returned_columns = {} %} + {% for column in elementary.run_query(columns_query).column_names %} + {% do returned_columns.update({column | lower: column}) %} + {% endfor %} + {% for rule in column_rules %} + {% set column = returned_columns.get(rule.column | lower) %} + {% if column %} + {% do conditions.append( + { + "condition": elementary.get_column_rule_condition( + adapter.quote(column), rule + ), + "owners": rule.owners, + } + ) %} + {% else %} + {% do elementary.debug_log( + flattened_test.unique_id + ~ ": skipping the conditional_result_owners rule on column " + ~ rule.column + ~ ", which the test doesn't return." + ) %} + {% endif %} + {% endfor %} {% endif %} + {% if not conditions %} {% do return(none) %} {% endif %} {% set result_owners_query %} {{ elementary.get_conditional_result_owners_select_clause(conditions, flattened_test.unique_id) }} from ({{ sql }} @@ -140,48 +211,107 @@ {# For anomaly tests, the failing rows are the anomalous rows of each result group - (full_table_name, column_name, metric_name). All groups are evaluated in one - grouped query; select_override keeps it a single CTE chain, which T-SQL requires. + (full_table_name, column_name, metric_name). Conditions are evaluated for all + groups in one grouped query; select_override keeps it a single CTE chain, which + T-SQL requires. Column rules are matched here against the anomalous rows already + read, by the position of their column in the test's dimensions. Returns a dict of group key (see get_anomaly_result_owners_group_key) to owners. - The conditions are validated on every run, even when nothing is anomalous. + The rules are validated on every run, even when nothing is anomalous. #} {% macro get_anomaly_result_owners_by_group(flattened_test, anomaly_scores_rows) %} - {% set conditions = elementary.get_conditional_result_owners(flattened_test) %} - {% if not conditions %} {% do return({}) %} {% endif %} - {% if not (anomaly_scores_rows | selectattr("is_anomalous") | list) %} - {% do return({}) %} - {% endif %} - {% set result_owners_query = elementary.get_read_anomaly_scores_query( - flattened_test, - additional_where=elementary.edr_is_true("is_anomalous"), - select_override=elementary.get_conditional_result_owners_select_clause( - conditions, flattened_test.unique_id - ) - ~ ", full_table_name, column_name, metric_name", - group_by="full_table_name, column_name, metric_name", - ) %} + {% set rules = elementary.get_conditional_result_owners(flattened_test) %} + {% if not rules %} {% do return({}) %} {% endif %} + {% set anomalous_rows = anomaly_scores_rows | selectattr("is_anomalous") | list %} + {% if not anomalous_rows %} {% do return({}) %} {% endif %} + {% set conditions = rules | selectattr("condition", "defined") | list %} + {% set column_rules = rules | selectattr("column", "defined") | list %} {% set owners_by_group = {} %} - {% for row in elementary.agate_to_dicts( - elementary.run_query(result_owners_query) - ) %} - {% set owners = elementary.get_matched_result_owners(conditions, row) %} - {% if owners %} - {% do owners_by_group.update( - { - elementary.get_anomaly_result_owners_group_key( - elementary.insensitive_get_dict_value( - row, "full_table_name" - ), - elementary.insensitive_get_dict_value( - row, "column_name" - ), - elementary.insensitive_get_dict_value( - row, "metric_name" - ), - ): owners - } - ) %} - {% endif %} - {% endfor %} + + {% if conditions %} + {% set result_owners_query = elementary.get_read_anomaly_scores_query( + flattened_test, + additional_where=elementary.edr_is_true("is_anomalous"), + select_override=elementary.get_conditional_result_owners_select_clause( + conditions, flattened_test.unique_id + ) + ~ ", full_table_name, column_name, metric_name", + group_by="full_table_name, column_name, metric_name", + ) %} + {% for row in elementary.agate_to_dicts( + elementary.run_query(result_owners_query) + ) %} + {% set owners = elementary.get_matched_result_owners(conditions, row) %} + {% if owners %} + {% do owners_by_group.update( + { + elementary.get_anomaly_result_owners_group_key( + elementary.insensitive_get_dict_value( + row, "full_table_name" + ), + elementary.insensitive_get_dict_value( + row, "column_name" + ), + elementary.insensitive_get_dict_value( + row, "metric_name" + ), + ): owners + } + ) %} + {% endif %} + {% endfor %} + {% endif %} + + {% if column_rules %} + {% set dimensions = (flattened_test.get("test_params") or {}).get( + "dimensions" + ) or [] %} + {% if dimensions is string %} {% set dimensions = [dimensions] %} {% endif %} + {% set dimensions = dimensions | map("lower") | list %} + {% for rule in column_rules %} + {% if rule.column | lower not in dimensions %} + {% do elementary.debug_log( + flattened_test.unique_id + ~ ": skipping the conditional_result_owners rule on column " + ~ rule.column + ~ ", which is not a dimension of the test." + ) %} + {% else %} + {% set position = dimensions.index(rule.column | lower) %} + {# dimension_value joins multiple dimensions with "; ", and nulls as NULL #} + {% set rule_values = [] %} + {% for value in elementary.get_column_rule_values(rule) %} + {% do rule_values.append("NULL" if value is none else value | string) %} + {% endfor %} + {% for row in anomalous_rows %} + {% set dimension_value = elementary.insensitive_get_dict_value( + row, "dimension_value" + ) %} + {% set parts = ( + (dimension_value | string).split("; ") + if dimension_value is not none + else [] + ) %} + {% if parts | length == dimensions | length and parts[ + position + ] in rule_values %} + {% set group_key = elementary.get_anomaly_result_owners_group_key( + row.full_table_name, row.column_name, row.metric_name + ) %} + {% set owners = owners_by_group.get(group_key) or [] %} + {% for owner in elementary.normalize_result_owners( + rule.owners + ) %} + {% if owner not in owners %} + {% do owners.append(owner) %} + {% endif %} + {% endfor %} + {% do owners_by_group.update( + {group_key: owners | sort | list} + ) %} + {% endif %} + {% endfor %} + {% endif %} + {% endfor %} + {% endif %} {% do return(owners_by_group) %} {% endmacro %} From f7532698b78530fa5b77c3931c4f8aac803d757d Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Tue, 13 Oct 2026 14:17:26 +0300 Subject: [PATCH 09/12] Fix column rules on Snowflake and ClickHouse; format Column rules read the test's returned column names from an empty result and lowercased them. Snowflake then failed on the quoted lowercase name ("value_field" is not VALUE_FIELD), and ClickHouse returns no column names for an empty result, so the rule was skipped. Read the names in their actual case from one failing row instead (limit 1, top 1 on T-SQL); column rules only run when the test failed or warned, so there is a row. Also applies black and sqlfmt formatting. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_conditional_result_owners.py | 22 ++++-- .../test/conditional_result_owners.sql | 71 ++++++++++++++----- 2 files changed, 72 insertions(+), 21 deletions(-) diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py index ece1fb755..7161e25f2 100644 --- a/integration_tests/tests/test_conditional_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -264,7 +264,6 @@ def test_too_many_conditions_errors(test_id: str, dbt_project: DbtProject): assert "Compilation Error" in test_result["test_results_description"] - def test_column_rule_on_dimension(test_id: str, dbt_project: DbtProject): # A column rule names the dimension column, not dimension_value. data = _dimension_data( @@ -273,7 +272,11 @@ def test_column_rule_on_dimension(test_id: str, dbt_project: DbtProject): ) rules = [ {"column": "superhero", "values": ["Superman"], "owners": ["dc@example.com"]}, - {"column": "superhero", "values": "Spiderman", "owners": ["marvel@example.com"]}, + { + "column": "superhero", + "values": "Spiderman", + "owners": ["marvel@example.com"], + }, ] test_result = dbt_project.test( test_id, @@ -299,7 +302,11 @@ def test_column_rule_on_multi_column_dimension(test_id: str, dbt_project: DbtPro ) rules = [ {"column": "Universe", "values": ["DC"], "owners": ["dc@example.com"]}, - {"column": "superhero", "values": ["Superman"], "owners": ["heroes@example.com"]}, + { + "column": "superhero", + "values": ["Superman"], + "owners": ["heroes@example.com"], + }, {"column": "universe", "values": ["Marvel"], "owners": ["marvel@example.com"]}, ] test_result = dbt_project.test( @@ -320,7 +327,9 @@ def test_column_rule_on_multi_column_dimension(test_id: str, dbt_project: DbtPro def test_column_rule_on_returned_column(test_id: str, dbt_project: DbtProject): # accepted_values returns value_field; column names match case-insensitively. data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] - rules = [{"column": "VALUE_FIELD", "values": ["JP"], "owners": ["japan@example.com"]}] + rules = [ + {"column": "VALUE_FIELD", "values": ["JP"], "owners": ["japan@example.com"]} + ] test_result = _accepted_values_test(test_id, dbt_project, data, rules) assert test_result["status"] == "fail" assert _owners(test_result) == ["japan@example.com"] @@ -330,7 +339,9 @@ def test_column_rule_skipped_when_not_returned(test_id: str, dbt_project: DbtPro # accepted_values doesn't return dimension_value, so the rule is skipped # instead of erroring the test, as a condition on that column would. data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] - rules = [{"column": "dimension_value", "values": ["JP"], "owners": ["japan@example.com"]}] + rules = [ + {"column": "dimension_value", "values": ["JP"], "owners": ["japan@example.com"]} + ] test_result = _accepted_values_test(test_id, dbt_project, data, rules) assert test_result["status"] == "fail" assert _owners(test_result) == [DEFAULT_OWNER] @@ -350,6 +361,7 @@ def test_rule_with_condition_and_column_errors(test_id: str, dbt_project: DbtPro assert test_result["status"] == "error" assert "Compilation Error" in test_result["test_results_description"] + # Schema change tests don't support conditional_result_owners: the config is # ignored (with a warning), so results keep the default owners. @pytest.mark.skip_targets(["databricks", "spark", "athena", "trino"]) diff --git a/macros/edr/materializations/test/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql index a86f59cc3..057b3b019 100644 --- a/macros/edr/materializations/test/conditional_result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -48,12 +48,19 @@ ) %} {% endif %} {% for item in conditions %} - {% set is_condition_rule = item is mapping and item.get( - "condition" - ) is string and item.get("condition") and "column" not in item %} - {% set is_column_rule = item is mapping and item.get( - "column" - ) is string and item.get("column") and "values" in item and "condition" not in item %} + {% set is_condition_rule = ( + item is mapping + and item.get("condition") is string + and item.get("condition") + and "column" not in item + ) %} + {% set is_column_rule = ( + item is mapping + and item.get("column") is string + and item.get("column") + and "values" in item + and "condition" not in item + ) %} {% if not (is_condition_rule or is_column_rule) or not item.get("owners") %} {% do exceptions.raise_compiler_error( "Invalid conditional_result_owners item in test `{}`: {}. Each item needs `owners`, and either a string `condition` or a `column` with `values`.".format( @@ -79,8 +86,7 @@ {% set literals = [] %} {% set checks = [] %} {% for value in elementary.get_column_rule_values(rule) %} - {% if value is none %} - {% do checks.append(quoted_column ~ " is null") %} + {% if value is none %} {% do checks.append(quoted_column ~ " is null") %} {% elif value is number and value is not sameas true and value is not sameas false %} {% do literals.append(value | string) %} {% else %} {% do literals.append(elementary.edr_quote(value | string)) %} @@ -147,6 +153,37 @@ {% do return((matched_owners | sort | list) or none) %} {% endmacro %} +{# + The column names the test's SQL returns, in their actual case so they can be + quoted. Reads one failing row rather than an empty result, because some + adapters (ClickHouse) return no column names for an empty result. Only called + when the test failed or warned, so there is at least one row. + On T-SQL `sql` already selects from a temp table, so it can be a derived table. +#} +{% macro get_test_returned_column_names() %} + {% set columns_query = adapter.dispatch( + "get_test_returned_columns_query", "elementary" + )() %} + {% do return( + elementary.run_query( + columns_query, lowercase_column_names=false + ).column_names + | list + ) %} +{% endmacro %} + +{% macro default__get_test_returned_columns_query() %} + {% do return("select * from (" ~ sql ~ "\n) results limit 1") %} +{% endmacro %} + +{% macro fabric__get_test_returned_columns_query() %} + {% do return("select top 1 * from (" ~ sql ~ "\n) results") %} +{% endmacro %} + +{% macro sqlserver__get_test_returned_columns_query() %} + {% do return(elementary.fabric__get_test_returned_columns_query()) %} +{% endmacro %} + {# For dbt tests, the failing rows are the rows returned by the test's SQL. On T-SQL (and with tests_use_temp_tables) `sql` already selects from a temp @@ -159,12 +196,8 @@ {% set column_rules = rules | selectattr("column", "defined") | list %} {% if column_rules %} {# Column rules only apply to the columns the test returns, so read them first #} - {% set columns_query %} - select * from ({{ sql }} - ) results where 1 = 0 - {% endset %} {% set returned_columns = {} %} - {% for column in elementary.run_query(columns_query).column_names %} + {% for column in elementary.get_test_returned_column_names() %} {% do returned_columns.update({column | lower: column}) %} {% endfor %} {% for rule in column_rules %} @@ -280,7 +313,9 @@ {# dimension_value joins multiple dimensions with "; ", and nulls as NULL #} {% set rule_values = [] %} {% for value in elementary.get_column_rule_values(rule) %} - {% do rule_values.append("NULL" if value is none else value | string) %} + {% do rule_values.append( + "NULL" if value is none else value | string + ) %} {% endfor %} {% for row in anomalous_rows %} {% set dimension_value = elementary.insensitive_get_dict_value( @@ -294,8 +329,12 @@ {% if parts | length == dimensions | length and parts[ position ] in rule_values %} - {% set group_key = elementary.get_anomaly_result_owners_group_key( - row.full_table_name, row.column_name, row.metric_name + {% set group_key = ( + elementary.get_anomaly_result_owners_group_key( + row.full_table_name, + row.column_name, + row.metric_name, + ) ) %} {% set owners = owners_by_group.get(group_key) or [] %} {% for owner in elementary.normalize_result_owners( From 08f374962d68e05597e7fbe747288c2674ec43b6 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Tue, 13 Oct 2026 16:19:38 +0300 Subject: [PATCH 10/12] Fix column rules for inherited dimensions and "; " in values Column rules now read dimensions from the resolved test configuration, so dimensions set in the model config or vars are matched too. A single dimension's value is no longer split on "; ". Co-Authored-By: Claude Opus 5.5 --- .../tests/test_conditional_result_owners.py | 49 +++++++++++++++++++ .../test/conditional_result_owners.sql | 23 ++++++--- 2 files changed, 64 insertions(+), 8 deletions(-) diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py index 7161e25f2..ca6a7ec70 100644 --- a/integration_tests/tests/test_conditional_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -324,6 +324,55 @@ def test_column_rule_on_multi_column_dimension(test_id: str, dbt_project: DbtPro assert _owners(test_result) == ["dc@example.com", "heroes@example.com"] +def test_column_rule_on_dimension_from_test_vars(test_id: str, dbt_project: DbtProject): + # Dimensions resolved from vars, not test arguments, still match column rules. + data = _dimension_data( + [{"superhero": "Superman"}] * 3 + [{"superhero": "Spiderman"}], + [{"superhero": "Superman"}, {"superhero": "Spiderman"}], + ) + rules = [ + {"column": "superhero", "values": ["Superman"], "owners": ["dc@example.com"]} + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + {"timestamp_column": TIMESTAMP_COLUMN}, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"conditional_result_owners": rules}}, + test_vars={"dimensions": ["superhero"]}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com"] + + +def test_column_rule_on_dimension_value_with_separator( + test_id: str, dbt_project: DbtProject +): + # A single dimension's value isn't split on "; ". + data = _dimension_data( + [{"superhero": "Superman; Batman"}] * 3 + [{"superhero": "Spiderman"}], + [{"superhero": "Superman; Batman"}, {"superhero": "Spiderman"}], + ) + rules = [ + { + "column": "superhero", + "values": ["Superman; Batman"], + "owners": ["dc@example.com"], + } + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["superhero"]}, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"conditional_result_owners": rules}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com"] + + def test_column_rule_on_returned_column(test_id: str, dbt_project: DbtProject): # accepted_values returns value_field; column names match case-insensitively. data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] diff --git a/macros/edr/materializations/test/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql index 057b3b019..2a74b7b51 100644 --- a/macros/edr/materializations/test/conditional_result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -295,9 +295,15 @@ {% endif %} {% if column_rules %} - {% set dimensions = (flattened_test.get("test_params") or {}).get( - "dimensions" - ) or [] %} + {# The resolved configuration also has dimensions set in the model config or vars #} + {% set test_configuration = ( + elementary.get_cache(flattened_test.unique_id) or {} + ) %} + {% set dimensions = ( + test_configuration.get("dimensions") + or (flattened_test.get("test_params") or {}).get("dimensions") + or [] + ) %} {% if dimensions is string %} {% set dimensions = [dimensions] %} {% endif %} {% set dimensions = dimensions | map("lower") | list %} {% for rule in column_rules %} @@ -321,11 +327,12 @@ {% set dimension_value = elementary.insensitive_get_dict_value( row, "dimension_value" ) %} - {% set parts = ( - (dimension_value | string).split("; ") - if dimension_value is not none - else [] - ) %} + {# A single dimension's value is never split, since it may contain "; " #} + {% if dimension_value is none %} {% set parts = [] %} + {% elif dimensions | length == 1 %} + {% set parts = [dimension_value | string] %} + {% else %} {% set parts = (dimension_value | string).split("; ") %} + {% endif %} {% if parts | length == dimensions | length and parts[ position ] in rule_values %} From f09434118e043644d833437ba090d540fbfde6fc Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Wed, 21 Oct 2026 17:17:46 +0200 Subject: [PATCH 11/12] Address review: flatter column-rule matching, fewer queries, more tests - Anomaly path: build a value -> owners lookup per dimension position once, then match the anomalous rows in a single pass, instead of looping rows for every rule and value. - Log (debug) how many anomalous rows have a dimension value that can't be split because a value contains "; ", so column rules silently not matching them is visible. - dbt test path: use the sampled failing rows to skip column rules on columns the test doesn't return, without running the column-names query. The query still runs when a rule's column is returned (its real case is needed to quote it) or when samples are disabled. - Tests: condition and column rules unioned in the dbt test and anomaly paths; column rules with samples disabled. Co-Authored-By: Claude Opus 5.5 --- .../tests/test_conditional_result_owners.py | 62 +++++++++++ .../test/conditional_result_owners.sql | 100 +++++++++++++----- macros/edr/materializations/test/test.sql | 4 +- 3 files changed, 141 insertions(+), 25 deletions(-) diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py index ca6a7ec70..27a7829a2 100644 --- a/integration_tests/tests/test_conditional_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -431,3 +431,65 @@ def test_schema_changes_ignore_conditional_owners( failed_results = [r for r in test_results if r["status"] == "fail"] assert failed_results assert all(_owners(r) == [DEFAULT_OWNER] for r in failed_results) + + +def test_condition_and_column_rules_union_on_dbt_test( + test_id: str, dbt_project: DbtProject +): + # Conditions and column rules go through different code; their owners are unioned. + data = [{COLUMN_NAME: value} for value in ["NL", "JP", "KR"]] + rules = [ + {"condition": "value_field = 'JP'", "owners": ["japan@example.com"]}, + {"column": "value_field", "values": ["KR"], "owners": ["korea@example.com"]}, + ] + test_result = _accepted_values_test(test_id, dbt_project, data, rules) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["japan@example.com", "korea@example.com"] + + +def test_condition_and_column_rules_union_on_dimension_anomalies( + test_id: str, dbt_project: DbtProject +): + # Both dimension values are anomalous: one matched by a condition, one by a column rule. + data = _dimension_data( + [{"superhero": "Superman"}] * 3 + [{"superhero": "Spiderman"}] * 3, + [{"superhero": "Superman"}, {"superhero": "Spiderman"}], + ) + rules = [ + {"condition": "dimension_value = 'Superman'", "owners": ["dc@example.com"]}, + { + "column": "superhero", + "values": ["Spiderman"], + "owners": ["marvel@example.com"], + }, + ] + test_result = dbt_project.test( + test_id, + "elementary.dimension_anomalies", + {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": ["superhero"]}, + data=data, + model_config=MODEL_CONFIG, + test_config={"meta": {"conditional_result_owners": rules}}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["dc@example.com", "marvel@example.com"] + + +def test_column_rule_without_samples(test_id: str, dbt_project: DbtProject): + # With samples disabled, the returned columns are read with a separate query. + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + rules = [{"column": "value_field", "values": ["JP"], "owners": ["japan@example.com"]}] + test_result = dbt_project.test( + test_id, + "accepted_values", + ACCEPTED_VALUES_ARGS, + test_column=COLUMN_NAME, + data=data, + model_config=MODEL_CONFIG, + test_config={ + "meta": {"conditional_result_owners": rules, "disable_test_samples": True} + }, + test_vars={"enable_elementary_test_materialization": True}, + ) + assert test_result["status"] == "fail" + assert _owners(test_result) == ["japan@example.com"] diff --git a/macros/edr/materializations/test/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql index 2a74b7b51..12a28f645 100644 --- a/macros/edr/materializations/test/conditional_result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -189,13 +189,33 @@ On T-SQL (and with tests_use_temp_tables) `sql` already selects from a temp table, so wrapping it in a derived table is safe there too. #} -{% macro get_dbt_test_result_owners(flattened_test) %} +{% macro get_dbt_test_result_owners(flattened_test, sample_rows=none) %} {% set rules = elementary.get_conditional_result_owners(flattened_test) %} {% if not rules or elementary.did_test_pass() %} {% do return(none) %} {% endif %} {% set conditions = rules | selectattr("condition", "defined") | list %} {% set column_rules = rules | selectattr("column", "defined") | list %} + {% if column_rules and sample_rows %} + {# The sampled failing rows already show which columns the test returns + (lowercased), so skip rules on columns it doesn't return without a query #} + {% set sampled_columns = sample_rows[0].keys() | map("lower") | list %} + {% set kept_rules = [] %} + {% for rule in column_rules %} + {% if rule.column | lower in sampled_columns %} + {% do kept_rules.append(rule) %} + {% else %} + {% do elementary.debug_log( + flattened_test.unique_id + ~ ": skipping the conditional_result_owners rule on column " + ~ rule.column + ~ ", which the test doesn't return." + ) %} + {% endif %} + {% endfor %} + {% set column_rules = kept_rules %} + {% endif %} {% if column_rules %} - {# Column rules only apply to the columns the test returns, so read them first #} + {# Column rules only apply to the columns the test returns, so read them first. + The real column names (not lowercased) are needed to quote them. #} {% set returned_columns = {} %} {% for column in elementary.get_test_returned_column_names() %} {% do returned_columns.update({column | lower: column}) %} @@ -306,6 +326,9 @@ ) %} {% if dimensions is string %} {% set dimensions = [dimensions] %} {% endif %} {% set dimensions = dimensions | map("lower") | list %} + {# Build a value -> owners lookup per dimension position once, + then match every anomalous row in a single pass. #} + {% set owners_by_position = {} %} {% for rule in column_rules %} {% if rule.column | lower not in dimensions %} {% do elementary.debug_log( @@ -316,26 +339,47 @@ ) %} {% else %} {% set position = dimensions.index(rule.column | lower) %} - {# dimension_value joins multiple dimensions with "; ", and nulls as NULL #} - {% set rule_values = [] %} + {% do owners_by_position.setdefault(position, {}) %} + {# dimension_value shows nulls as NULL #} {% for value in elementary.get_column_rule_values(rule) %} - {% do rule_values.append( - "NULL" if value is none else value | string - ) %} + {% set value_key = "NULL" if value is none else value | string %} + {% do owners_by_position[position].setdefault(value_key, []) %} + {% for owner in elementary.normalize_result_owners(rule.owners) %} + {% if owner not in owners_by_position[position][value_key] %} + {% do owners_by_position[position][value_key].append(owner) %} + {% endif %} + {% endfor %} {% endfor %} - {% for row in anomalous_rows %} - {% set dimension_value = elementary.insensitive_get_dict_value( - row, "dimension_value" - ) %} - {# A single dimension's value is never split, since it may contain "; " #} - {% if dimension_value is none %} {% set parts = [] %} - {% elif dimensions | length == 1 %} - {% set parts = [dimension_value | string] %} - {% else %} {% set parts = (dimension_value | string).split("; ") %} + {% endif %} + {% endfor %} + + {% if owners_by_position %} + {% set unsplittable = namespace(count=0) %} + {% for row in anomalous_rows %} + {% set dimension_value = elementary.insensitive_get_dict_value( + row, "dimension_value" + ) %} + {# dimension_value joins multiple dimensions with "; ". A single + dimension's value is never split, since it may contain "; ". #} + {% if dimension_value is none %} {% set parts = [] %} + {% elif dimensions | length == 1 %} + {% set parts = [dimension_value | string] %} + {% else %} {% set parts = (dimension_value | string).split("; ") %} + {% endif %} + {% if parts | length != dimensions | length %} + {% if dimension_value is not none %} + {% set unsplittable.count = unsplittable.count + 1 %} {% endif %} - {% if parts | length == dimensions | length and parts[ - position - ] in rule_values %} + {% else %} + {% set row_owners = [] %} + {% for position, owners_by_value in owners_by_position.items() %} + {% for owner in owners_by_value.get(parts[position], []) %} + {% if owner not in row_owners %} + {% do row_owners.append(owner) %} + {% endif %} + {% endfor %} + {% endfor %} + {% if row_owners %} {% set group_key = ( elementary.get_anomaly_result_owners_group_key( row.full_table_name, @@ -344,9 +388,7 @@ ) ) %} {% set owners = owners_by_group.get(group_key) or [] %} - {% for owner in elementary.normalize_result_owners( - rule.owners - ) %} + {% for owner in row_owners %} {% if owner not in owners %} {% do owners.append(owner) %} {% endif %} @@ -355,9 +397,19 @@ {group_key: owners | sort | list} ) %} {% endif %} - {% endfor %} + {% endif %} + {% endfor %} + {% if unsplittable.count %} + {% do elementary.debug_log( + flattened_test.unique_id + ~ ": " + ~ unsplittable.count + ~ " anomalous rows have a dimension value containing \"; \", so it can't be split into the test's " + ~ dimensions | length + ~ " dimensions; conditional_result_owners column rules don't match them." + ) %} {% endif %} - {% endfor %} + {% endif %} {% endif %} {% do return(owners_by_group) %} {% endmacro %} diff --git a/macros/edr/materializations/test/test.sql b/macros/edr/materializations/test/test.sql index b10f331b9..4fd626206 100644 --- a/macros/edr/materializations/test/test.sql +++ b/macros/edr/materializations/test/test.sql @@ -132,7 +132,9 @@ {% set elementary_test_results_row = elementary.get_dbt_test_result_row( flattened_test, result_rows ) %} - {% set result_owners = elementary.get_dbt_test_result_owners(flattened_test) %} + {% set result_owners = elementary.get_dbt_test_result_owners( + flattened_test, result_rows + ) %} {% if result_owners %} {% do elementary_test_results_row.update({"owners": result_owners}) %} {% endif %} From 8d77db1a735f467652e9db83bcd346289966ae26 Mon Sep 17 00:00:00 2001 From: Joost Boonzajer Flaes Date: Wed, 21 Oct 2026 17:35:12 +0200 Subject: [PATCH 12/12] Apply black and sqlfmt formatting Co-Authored-By: Claude Opus 5.5 --- integration_tests/tests/test_conditional_result_owners.py | 4 +++- .../materializations/test/conditional_result_owners.sql | 7 +++++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py index 27a7829a2..fd53a30cc 100644 --- a/integration_tests/tests/test_conditional_result_owners.py +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -478,7 +478,9 @@ def test_condition_and_column_rules_union_on_dimension_anomalies( def test_column_rule_without_samples(test_id: str, dbt_project: DbtProject): # With samples disabled, the returned columns are read with a separate query. data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] - rules = [{"column": "value_field", "values": ["JP"], "owners": ["japan@example.com"]}] + rules = [ + {"column": "value_field", "values": ["JP"], "owners": ["japan@example.com"]} + ] test_result = dbt_project.test( test_id, "accepted_values", diff --git a/macros/edr/materializations/test/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql index 12a28f645..8a0361135 100644 --- a/macros/edr/materializations/test/conditional_result_owners.sql +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -346,7 +346,9 @@ {% do owners_by_position[position].setdefault(value_key, []) %} {% for owner in elementary.normalize_result_owners(rule.owners) %} {% if owner not in owners_by_position[position][value_key] %} - {% do owners_by_position[position][value_key].append(owner) %} + {% do owners_by_position[position][value_key].append( + owner + ) %} {% endif %} {% endfor %} {% endfor %} @@ -405,7 +407,8 @@ ~ ": " ~ unsplittable.count ~ " anomalous rows have a dimension value containing \"; \", so it can't be split into the test's " - ~ dimensions | length + ~ dimensions + | length ~ " dimensions; conditional_result_owners column rules don't match them." ) %} {% endif %}