diff --git a/integration_tests/tests/test_nested_struct_anomalies.py b/integration_tests/tests/test_nested_struct_anomalies.py new file mode 100644 index 000000000..47ac8bbd3 --- /dev/null +++ b/integration_tests/tests/test_nested_struct_anomalies.py @@ -0,0 +1,211 @@ +import json +from datetime import date, datetime, timedelta +from typing import List, Optional, Sequence, Tuple, Union + +import pytest +from data_generator import DATE_FORMAT, generate_dates +from dbt_project import DbtProject + +TIMESTAMP_COLUMN = "updated_at" +NESTED_COLUMN = "user_info.address.city" +PLAIN_COLUMN = "superhero" +COLUMN_TEST_NAME = "elementary.column_anomalies" +DIMENSION_TEST_NAME = "elementary.dimension_anomalies" + +# Nested STRUCT leaves are only supported on BigQuery. +SUPPORTED_TARGETS = ["bigquery"] + +# (updated_at, superhero, user_info.address.city) +Row = Tuple[Union[date, datetime], str, Optional[str]] + + +def _row_sql(updated_at: Union[date, datetime], superhero: str, city: Optional[str]): + city_sql = "cast(null as string)" if city is None else f"cast('{city}' as string)" + return ( + f"select timestamp '{updated_at.strftime(DATE_FORMAT)}' as {TIMESTAMP_COLUMN}" + f", cast('{superhero}' as string) as {PLAIN_COLUMN}" + ", struct(" + f"struct({city_sql} as city, cast('US' as string) as country) as address" + ", cast('hero' as string) as name" + ") as user_info" + # A REPEATED leaf and a REPEATED ancestor, so that nested-column + # discovery has to skip fields that would require UNNEST rather than + # generating invalid SQL for them. + ", [struct(cast(1 as int64) as amount)] as orders" + ", ['tag'] as tags" + ) + + +def _create_struct_model(dbt_project: DbtProject, test_id: str, rows: Sequence[Row]): + """Materialize a table with nested STRUCT columns, then leave it in place. + + ``DbtProject.test(as_model=True)`` re-creates a dummy model file with the + same name so the node exists in the manifest; the physical table built here + is what the test actually reads. + """ + query = "\nunion all\n".join(_row_sql(*row) for row in rows) + with dbt_project.create_temp_model_for_existing_table( + test_id, materialization="table", raw_code=query + ) as model_path: + assert dbt_project.dbt_runner.run( + select=str(model_path) + ), "Failed to build the nested STRUCT model" + + +def _stable_rows(base_date) -> List[Row]: + return [ + (cur_date, superhero, city) + for cur_date in generate_dates(base_date=base_date) + for superhero, city in [("Superman", "Metropolis"), ("Batman", "Gotham")] + ] + + +def _anomaly_test_points(dbt_project: DbtProject, test_id: str): + results = dbt_project.run_query(dbt_project.samples_query(test_id)) + return [json.loads(result["result_row"]) for result in results] + + +@pytest.mark.only_on_targets(SUPPORTED_TARGETS) +def test_anomalyless_column_anomalies_on_struct_field( + test_id: str, dbt_project: DbtProject +): + utc_today = datetime.utcnow().date() + _create_struct_model(dbt_project, test_id, _stable_rows(utc_today - timedelta(1))) + + test_result = dbt_project.test( + test_id, + COLUMN_TEST_NAME, + {"timestamp_column": TIMESTAMP_COLUMN, "column_anomalies": ["null_count"]}, + test_column=NESTED_COLUMN, + as_model=True, + ) + assert test_result["status"] == "pass" + # The dotted path is what alerts display, so it must survive into the results. + assert test_result["column_name"].lower() == NESTED_COLUMN + + +@pytest.mark.only_on_targets(SUPPORTED_TARGETS) +def test_anomalous_column_anomalies_on_struct_field( + test_id: str, dbt_project: DbtProject +): + utc_today = datetime.utcnow().date() + test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1)) + + rows: List[Row] = [(test_date, "Superman", None) for _ in range(3)] + rows += [ + (cur_date, superhero, city) + for cur_date in training_dates + for superhero, city in [("Superman", "Metropolis"), ("Batman", "Gotham")] + ] + _create_struct_model(dbt_project, test_id, rows) + + test_result = dbt_project.test( + test_id, + COLUMN_TEST_NAME, + {"timestamp_column": TIMESTAMP_COLUMN, "column_anomalies": ["null_count"]}, + test_column=NESTED_COLUMN, + as_model=True, + ) + assert test_result["status"] == "fail" + + +@pytest.mark.only_on_targets(SUPPORTED_TARGETS) +def test_column_anomalies_with_struct_dimension(test_id: str, dbt_project: DbtProject): + """Plain monitored column, nested STRUCT leaf as the dimension.""" + utc_today = datetime.utcnow().date() + _create_struct_model(dbt_project, test_id, _stable_rows(utc_today - timedelta(1))) + + test_result = dbt_project.test( + test_id, + COLUMN_TEST_NAME, + { + "timestamp_column": TIMESTAMP_COLUMN, + "column_anomalies": ["null_count"], + "dimensions": [NESTED_COLUMN], + }, + test_column=PLAIN_COLUMN, + as_model=True, + ) + assert test_result["status"] == "pass" + + points = _anomaly_test_points(dbt_project, test_id) + assert points, "No metric data points were collected" + # The dimension must resolve to the STRUCT leaf's values, not to nulls. + assert {point["dimension"] for point in points} == {NESTED_COLUMN} + assert {point["dimension_value"] for point in points} == {"Metropolis", "Gotham"} + + +@pytest.mark.only_on_targets(SUPPORTED_TARGETS) +def test_column_anomalies_on_struct_field_with_struct_dimension( + test_id: str, dbt_project: DbtProject +): + """Both the monitored column and the dimension are nested STRUCT leaves.""" + utc_today = datetime.utcnow().date() + _create_struct_model(dbt_project, test_id, _stable_rows(utc_today - timedelta(1))) + + test_result = dbt_project.test( + test_id, + COLUMN_TEST_NAME, + { + "timestamp_column": TIMESTAMP_COLUMN, + "column_anomalies": ["null_count"], + "dimensions": [NESTED_COLUMN], + }, + test_column=NESTED_COLUMN, + as_model=True, + ) + assert test_result["status"] == "pass" + assert test_result["column_name"].lower() == NESTED_COLUMN + + +@pytest.mark.only_on_targets(SUPPORTED_TARGETS) +def test_anomalyless_dimension_anomalies_on_struct_field( + test_id: str, dbt_project: DbtProject +): + utc_today = datetime.utcnow().date() + _create_struct_model(dbt_project, test_id, _stable_rows(utc_today - timedelta(1))) + + test_result = dbt_project.test( + test_id, + DIMENSION_TEST_NAME, + {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": [NESTED_COLUMN]}, + as_model=True, + ) + assert test_result["status"] == "pass" + + +@pytest.mark.only_on_targets(SUPPORTED_TARGETS) +def test_anomalous_dimension_anomalies_on_struct_field( + test_id: str, dbt_project: DbtProject +): + utc_today = datetime.utcnow().date() + test_date, *training_dates = generate_dates(base_date=utc_today - timedelta(1)) + + rows: List[Row] = [ + (test_date, superhero, city) + for superhero, city in [ + ("Superman", "Metropolis"), + ("Superman", "Metropolis"), + ("Superman", "Metropolis"), + ("Batman", "Gotham"), + ] + ] + rows += [ + (cur_date, superhero, city) + for cur_date in training_dates + for superhero, city in [("Superman", "Metropolis"), ("Batman", "Gotham")] + ] + _create_struct_model(dbt_project, test_id, rows) + + test_result = dbt_project.test( + test_id, + DIMENSION_TEST_NAME, + {"timestamp_column": TIMESTAMP_COLUMN, "dimensions": [NESTED_COLUMN]}, + as_model=True, + ) + assert test_result["status"] == "fail" + + points = _anomaly_test_points(dbt_project, test_id) + # Only anomalous dimension values are stored for dimension anomalies. + assert {point["dimension_value"] for point in points} == {"Metropolis"} + assert any(point["is_anomalous"] for point in points) diff --git a/macros/edr/data_monitoring/data_monitors_configuration/get_column_monitors.sql b/macros/edr/data_monitoring/data_monitors_configuration/get_column_monitors.sql index c32dcf520..9eed8d6f2 100644 --- a/macros/edr/data_monitoring/data_monitors_configuration/get_column_monitors.sql +++ b/macros/edr/data_monitoring/data_monitors_configuration/get_column_monitors.sql @@ -1,7 +1,13 @@ {% macro get_column_obj_and_monitors(model_relation, column_name, monitors=none) %} - {% set column_obj_and_monitors = [] %} {% set column_objects = adapter.get_columns_in_relation(model_relation) %} + + {#- Only a dotted name can refer to a nested STRUCT leaf, so skip the + (potentially wide) flattening pass entirely for ordinary columns. -#} + {% if "." in column_name %} + {% set column_objects = elementary.bq_flatten_nested_columns(column_objects) %} + {% endif %} + {% for column_obj in column_objects %} {% if column_obj.name.strip('"') | lower == column_name.strip('"') | lower %} {% set column_monitors = elementary.column_monitors_by_type( @@ -21,6 +27,9 @@ {% set column_obj_and_monitors = [] %} {% set column_objects = adapter.get_columns_in_relation(model_relation) %} + {#- Nested STRUCT leaves are intentionally not expanded here: auto-monitoring + every leaf would balloon the test surface on wide STRUCT schemas. Users + opt in per column via `column_anomalies` with a dotted `column_name`. -#} {% for column_obj in column_objects %} {% set column_monitors = elementary.column_monitors_by_type( elementary.get_column_data_type(column_obj), monitors diff --git a/macros/edr/data_monitoring/monitors_query/column_monitoring_query.sql b/macros/edr/data_monitoring/monitors_query/column_monitoring_query.sql index ecd1d3564..2b1e026b9 100644 --- a/macros/edr/data_monitoring/monitors_query/column_monitoring_query.sql +++ b/macros/edr/data_monitoring/monitors_query/column_monitoring_query.sql @@ -15,9 +15,33 @@ {%- set timestamp_column = metric_properties.timestamp_column %} {% set prefixed_dimensions = [] %} {% for dimension_column in dimensions %} - {% do prefixed_dimensions.append("dimension_" ~ dimension_column) %} + {% do prefixed_dimensions.append( + "dimension_" ~ elementary.bq_alias_safe_dimension(dimension_column) + ) %} {% endfor %} + {#- A nested BigQuery struct leaf (user.address.city) cannot be referenced via + `column_obj.quoted` — that wraps the whole dotted name in one pair of + backticks — and projecting it into a CTE unaliased would collapse the path + to its last segment. Project it segment-quoted under a dot-free alias and + have the metric aggregates reference that alias instead. Non-nested + columns keep using `column_obj.quoted`, so identifier quoting (reserved + words, case-sensitive names) is never lost. -#} + {%- if elementary.bq_is_nested_identifier(column_obj.name) %} + {%- set nested_alias = adapter.quote( + elementary.bq_safe_alias(column_obj.name) + ) %} + {%- set monitored_column_projection = ( + elementary.bq_segment_quote(column_obj.name) + ~ " as " + ~ nested_alias + ) %} + {%- set monitored_column_expr = nested_alias %} + {%- else %} + {%- set monitored_column_projection = column_obj.quoted %} + {%- set monitored_column_expr = column_obj.quoted %} + {%- endif %} + {% set metric_types = [] %} {% set metric_name_to_type = {} %} {% for metric in column_metrics %} @@ -53,7 +77,7 @@ ), filtered_monitored_table as ( select - {{ column_obj.quoted }}, + {{ monitored_column_projection }}, {%- if dimensions -%} {{ elementary.select_dimensions_columns( @@ -78,7 +102,7 @@ {%- else %} filtered_monitored_table as ( select - {{ column_obj.quoted }}, + {{ monitored_column_projection }}, {%- if dimensions -%} {{ elementary.select_dimensions_columns( @@ -94,7 +118,7 @@ column_metrics as ( {%- if column_metrics %} - {%- set column = column_obj.quoted -%} + {%- set column = monitored_column_expr -%} select {%- if timestamp_column %} edr_bucket_start as bucket_start, edr_bucket_end as bucket_end, @@ -341,16 +365,20 @@ {% endif %} {% endmacro %} +{# Segment-quotes nested BigQuery struct dimensions and sanitises the alias + suffix. Both helpers are no-ops for plain identifiers, SQL expressions and + non-BigQuery adapters, so this stays byte-identical to previous behaviour + outside of nested struct references. #} {% macro select_dimensions_columns(dimension_columns, as_prefix="") %} {% set select_statements %} {%- for column in dimension_columns -%} - {{ column }} {%- if as_prefix -%} - {{ " as " ~ as_prefix ~ "_" ~ column }} - {%- endif -%} - {%- if not loop.last -%} - {{ ", " }} + {{ elementary.bq_segment_quote(column) }} + {{- " as " ~ as_prefix ~ "_" ~ elementary.bq_alias_safe_dimension(column) -}} + {%- else -%} + {{ column }} {%- endif -%} + {%- if not loop.last -%}{{ ", " }}{%- endif -%} {%- endfor -%} {% endset %} {{ return(select_statements) }} diff --git a/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql b/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql index 3aa20fe19..30119655a 100644 --- a/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql +++ b/macros/edr/data_monitoring/monitors_query/dimension_monitoring_query.sql @@ -12,8 +12,16 @@ elementary.relation_to_full_name(monitored_table_relation) ) %} {% set dimensions_string = elementary.join_list(dimensions, "; ") %} + + {# Segment-quote nested struct paths (e.g. user.address.city) for BigQuery so + they compile correctly. Plain identifiers, expressions and non-BigQuery + adapters pass through unchanged. #} + {% set sql_dimensions = [] %} + {% for dimension in dimensions %} + {% do sql_dimensions.append(elementary.bq_segment_quote(dimension)) %} + {% endfor %} {% set concat_dimensions_sql_expression = elementary.list_concat_with_separator( - dimensions, "; " + sql_dimensions, "; " ) %} {% set timestamp_column = metric_properties.timestamp_column %} {%- set data_monitoring_metrics_relation = elementary.get_elementary_relation( diff --git a/macros/utils/sql_utils/bigquery_nested_columns.sql b/macros/utils/sql_utils/bigquery_nested_columns.sql new file mode 100644 index 000000000..add1686da --- /dev/null +++ b/macros/utils/sql_utils/bigquery_nested_columns.sql @@ -0,0 +1,126 @@ +{# ---------------------------------------------------------------------- #} +{# BigQuery STRUCT nested-field helpers. #} +{# #} +{# All of these are no-ops outside BigQuery and for anything that is not a #} +{# plain dotted identifier path, so callers can apply them unconditionally #} +{# without changing behaviour on other adapters. #} +{# ---------------------------------------------------------------------- #} +{% macro bq_is_nested_identifier(name) %} + {#- True only on BigQuery and only when `name` is a plain dotted identifier + path (e.g. user.address.city) — i.e. an actual nested STRUCT reference. + Returns false for plain identifiers, SQL expressions (dimensions are + documented as accepting arbitrary expressions, which must pass through + untouched) and non-BigQuery adapters. + Every segment must start with a letter or underscore, so numeric + literals used as dimensions (0.99) are not mistaken for paths. -#} + {%- if target.type != "bigquery" or name is not string -%} + {{ return(false) }} + {%- endif -%} + {{ + return( + modules.re.match("^[A-Za-z_]\\w*(\\.[A-Za-z_]\\w*)+$", name) is not none + ) + }} +{% endmacro %} + +{% macro bq_segment_quote(name) %} + {#- Segment-quote a nested identifier path for BigQuery: + user.address.city -> `user`.`address`.`city`. + `BigQueryColumn.quoted` cannot be used here — it wraps the whole string + in a single pair of backticks, which BigQuery reads as one column + literally named "user.address.city". + Anything that is not a nested identifier path is returned unchanged. -#} + {%- if elementary.bq_is_nested_identifier(name) -%} + {%- set parts = [] -%} + {%- for seg in name.split(".") -%} + {%- do parts.append("`" ~ seg ~ "`") -%} + {%- endfor -%} + {{ parts | join(".") }} + {%- else -%} {{ name }} + {%- endif -%} +{% endmacro %} + +{% macro bq_safe_alias(name) %} + {#- Convert a dotted identifier path into a dot-free SQL identifier. + Projecting `select user.address.city from t` into a CTE without an alias + names the resulting column `city`, losing the path, so nested columns + must be aliased on the way in. Only call this for names that satisfy + `bq_is_nested_identifier` — on arbitrary SQL expressions it produces + nonsense. -#} + {{- name | replace(".", "__") -}} +{% endmacro %} + +{% macro bq_alias_safe_dimension(dimension) %} + {#- Alias-safe form of a dimension: dot-free for nested BigQuery struct + paths, unchanged for plain identifiers and SQL expressions. -#} + {%- if elementary.bq_is_nested_identifier(dimension) -%} + {{- elementary.bq_safe_alias(dimension) -}} + {%- else -%} {{- dimension -}} + {%- endif -%} +{% endmacro %} + +{% macro bq_flatten_nested_columns(column_objects) %} + {#- Expand BigQuery STRUCT columns into their monitorable leaves, keeping the + top-level STRUCT alongside them so that `column_name=user` keeps working. + Leaves under a REPEATED ancestor are excluded — reaching them requires + UNNEST. Returns `column_objects` unchanged on non-BigQuery adapters. -#} + {%- if target.type != "bigquery" -%} {{ return(column_objects) }} {%- endif -%} + {%- set expanded = [] -%} + {%- for column_obj in column_objects -%} + {%- do expanded.append(column_obj) -%} + {%- if column_obj.fields | length > 0 -%} + {#- `BigQueryColumn.flatten()` discards ancestor modes, so a NULLABLE + leaf under a REPEATED ancestor still satisfies + `leaf.mode != 'REPEATED'`. Build the set of safe leaf names via an + ancestor-aware walker and filter `flatten()` against it. -#} + {%- set safe_names = elementary.bq_safe_leaf_names(column_obj) -%} + {%- for leaf in column_obj.flatten() -%} + {%- if leaf.name in safe_names -%} + {%- do expanded.append(leaf) -%} + {%- endif -%} + {%- endfor -%} + {%- endif -%} + {%- endfor -%} + {{ return(expanded) }} +{% endmacro %} + +{% macro bq_safe_leaf_names(column_obj) %} + {#- Walk a BigQuery STRUCT tree and collect dotted leaf names that are safe to + monitor without UNNEST — i.e. no REPEATED ancestor anywhere in the path, + and the leaf itself is not REPEATED. `BigQueryColumn.flatten()` returns + leaf columns with the leaf's own mode but discards ancestor modes, so this + walker is the source of truth for "which leaves can we project directly?". + Names are built the same way `flatten()` builds them (from `.column`), so + the two are directly comparable. -#} + {%- set safe_names = [] -%} + {%- if column_obj.mode != "REPEATED" and column_obj.fields is defined and column_obj.fields | length > 0 -%} + {%- for child in column_obj.fields -%} + {%- do elementary._bq_walk_collect( + child, [column_obj.column], false, safe_names + ) -%} + {%- endfor -%} + {%- endif -%} + {{ return(safe_names) }} +{% endmacro %} + +{% macro _bq_walk_collect(field, prefix, has_repeated_ancestor, safe_names) %} + {#- Recursive helper for `bq_safe_leaf_names`. `field` is a `BigQueryColumn` + (`BigQueryColumn.fields` wraps its subfields via `wrap_subfields`), so it + exposes `.name`, `.mode` and `.fields`. Propagates whether any ancestor + was REPEATED and appends safe leaf names to `safe_names`. -#} + {%- set new_prefix = prefix + [field.name] -%} + {%- if field.fields | length == 0 -%} + {%- if not has_repeated_ancestor and field.mode != "REPEATED" -%} + {%- do safe_names.append(new_prefix | join(".")) -%} + {%- endif -%} + {%- else -%} + {%- set new_has_repeated = has_repeated_ancestor or ( + field.mode == "REPEATED" + ) -%} + {%- for child in field.fields -%} + {%- do elementary._bq_walk_collect( + child, new_prefix, new_has_repeated, safe_names + ) -%} + {%- endfor -%} + {%- endif -%} +{% endmacro %}