diff --git a/integration_tests/tests/test_conditional_result_owners.py b/integration_tests/tests/test_conditional_result_owners.py new file mode 100644 index 000000000..fd53a30cc --- /dev/null +++ b/integration_tests/tests/test_conditional_result_owners.py @@ -0,0 +1,497 @@ +import json +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 + +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], + conditions: list, + test_config: Optional[Dict[str, Any]] = None, +): + return dbt_project.test( + test_id, + "accepted_values", + ACCEPTED_VALUES_ARGS, + test_column=COLUMN_NAME, + data=data, + model_config=MODEL_CONFIG, + test_config={ + **(test_config or {}), + "meta": {"conditional_result_owners": conditions}, + }, + test_vars={"enable_elementary_test_materialization": True}, + ) + + +def test_matching_conditions_union_owners(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: value} for value in ["NL", "JP", "JP", "KR"]] + conditions = [ + {"condition": "value_field = 'JP'", "owners": ["japan@example.com"]}, + { + "condition": "value_field in ('JP', 'KR')", + "owners": "japan@example.com, korea@example.com", + }, + {"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) == ["japan@example.com", "korea@example.com"] + + +def test_no_matching_condition_keeps_default_owners( + test_id: str, dbt_project: DbtProject +): + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + 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"}] + 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] + + +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"] + ] + conditions = [ + {"condition": "dimension_value = 'Superman'", "owners": ["dc@example.com"]}, + { + "condition": "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": {"conditional_result_owners": conditions}}, + ) + 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"}] + 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_conditional_owners(test_id: str, dbt_project: DbtProject): + data = [{COLUMN_NAME: value} for value in ["NL", "JP"]] + conditions = [{"condition": "value_field = 'JP'", "owners": ["japan@example.com"]}] + test_result = _accepted_values_test( + test_id, dbt_project, data, conditions, 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 condition. + data = _dimension_data( + [{"superhero": "Superman"}] * 3 + [{"superhero": "Spiderman"}] * 3, + [{"superhero": "Superman"}, {"superhero": "Spiderman"}], + ) + conditions = [ + {"condition": "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": {"conditional_result_owners": conditions}}, + ) + 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"}, + ], + ) + conditions = [ + { + "condition": "dimension_value = 'DC; Superman'", + "owners": ["dc@example.com"], + }, + { + "condition": "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": {"conditional_result_owners": conditions}}, + ) + 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"}], + ) + conditions = [ + {"condition": "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": {"conditional_result_owners": conditions}}, + 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, [{}]) + conditions = [ + {"condition": "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": {"conditional_result_owners": conditions}}, + ) + assert test_result["status"] == "fail" + 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(1501) + ] + 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"] + + +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_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"]] + 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"]) +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) + + +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/data_monitoring/anomaly_detection/store_anomaly_test_results.sql b/macros/edr/data_monitoring/anomaly_detection/store_anomaly_test_results.sql index 48b349e88..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,5 +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 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/conditional_result_owners.sql b/macros/edr/materializations/test/conditional_result_owners.sql new file mode 100644 index 000000000..8a0361135 --- /dev/null +++ b/macros/edr/materializations/test/conditional_result_owners.sql @@ -0,0 +1,418 @@ +{# + 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: + 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. 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, + 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 %} + {% if conditions is mapping %} {% set conditions = [conditions] %} {% endif %} + + {% set test_unique_id = elementary.insensitive_get_dict_value( + flattened_test, "unique_id" + ) %} + {% if conditions is string or conditions is not iterable %} + {% do exceptions.raise_compiler_error( + "conditional_result_owners of test `{}` must be a list, each item with a `condition` and `owners`.".format( + test_unique_id + ) + ) %} + {% endif %} + {% 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 %} + {% 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( + test_unique_id, item + ) + ) %} + {% endif %} + {% endfor %} + {% 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 = [] %} + {% 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 %} + +{# 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 (" + ~ item.condition + ~ ") then 1 else 0 end) as owners_condition_" + ~ loop.index0 + ) -%} + {%- endfor -%} + {%- do return( + "/* conditional_result_owners of " + ~ test_unique_id + ~ " */ select " + ~ condition_columns + | join(", ") + ) -%} +{% endmacro %} + +{# 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 item in conditions %} + {% set is_match = elementary.insensitive_get_dict_value( + result_row, "owners_condition_" ~ loop.index0 + ) %} + {% if is_match is not none and is_match | int > 0 %} + {% for owner in elementary.normalize_result_owners(item.owners) %} + {% if owner not in matched_owners %} + {% do matched_owners.append(owner) %} + {% endif %} + {% endfor %} + {% endif %} + {% endfor %} + {% 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 + table, so wrapping it in a derived table is safe there too. +#} +{% 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. + 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}) %} + {% 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 }} + ) results + {% endset %} + {% 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(conditions, rows[0])) %} +{% endmacro %} + +{% macro get_anomaly_result_owners_group_key( + full_table_name, column_name, metric_name +) %} + {# 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 %} + +{# + For anomaly tests, the failing rows are the anomalous rows of each result group + (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 rules 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_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 = {} %} + + {% 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 %} + {# 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 %} + {# 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( + 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) %} + {% do owners_by_position.setdefault(position, {}) %} + {# dimension_value shows nulls as NULL #} + {% for value in elementary.get_column_rule_values(rule) %} + {% 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 %} + {% 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 %} + {% 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, + row.column_name, + row.metric_name, + ) + ) %} + {% set owners = owners_by_group.get(group_key) or [] %} + {% for owner in row_owners %} + {% if owner not in owners %} + {% do owners.append(owner) %} + {% endif %} + {% endfor %} + {% do owners_by_group.update( + {group_key: owners | sort | list} + ) %} + {% endif %} + {% 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 %} + {% 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 fdef2c8fc..4fd626206 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( + "conditional_result_owners" + ) %} + {% do elementary.edr_log_warning( + test_unique_id + ~ ": conditional_result_owners is not supported for " + ~ short_name + ~ " tests and will be ignored." + ) %} + {% endif %} {% do return(materialization_macro()) %} {% endif %} {% endif %} @@ -120,6 +132,12 @@ {% 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, result_rows + ) %} + {% if result_owners %} + {% do elementary_test_results_row.update({"owners": result_owners}) %} + {% endif %} {% do elementary.cache_elementary_test_results_rows( [elementary_test_results_row] ) %} diff --git a/macros/edr/tests/test_utils/get_anomaly_query.sql b/macros/edr/tests/test_utils/get_anomaly_query.sql index c59668656..d8169b7f3 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) %} @@ -186,6 +187,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 %}