Add conditional_result_owners: set test result owners from failing rows

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 = {"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"]

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 %}

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 %}

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]
     ) %}