Support BigQuery nested STRUCT fields in anomaly tests

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)

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

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

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(