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..3f693a1cf --- /dev/null +++ b/integration_tests/tests/test_nested_struct_anomalies.py @@ -0,0 +1,271 @@ +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" +# A nested leaf whose final segment is a BigQuery reserved word. Referencing it +# only compiles when each path segment is backtick-quoted (`user_info`.`order`). +RESERVED_WORD_COLUMN = "user_info.order" +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" + # A leaf whose segment is a BigQuery reserved word, so the monitoring + # query only compiles when every path segment is backtick-quoted. + ", cast('paid' as string) as `order`" + ") 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_on_repeated_descendant_rejected( + test_id: str, dbt_project: DbtProject +): + """Leaves under a REPEATED ancestor need UNNEST, so discovery must exclude + them and the test must error at the column lookup rather than generate + invalid SQL.""" + 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="orders.amount", + as_model=True, + ) + assert test_result["status"] == "error" + assert "unable to find column" in test_result["test_results_description"].lower() + + +@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_column_anomalies_on_reserved_word_struct_field( + test_id: str, dbt_project: DbtProject +): + """A nested leaf whose segment is a reserved word (user_info.order) only + compiles when each path segment is backtick-quoted. Used as both the + monitored column and a dimension, it pins the segment-quoting on the column + projection and the dimension select list — drop either and BigQuery rejects + the query as a syntax error rather than these tests staying green.""" + 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": [RESERVED_WORD_COLUMN], + }, + test_column=RESERVED_WORD_COLUMN, + as_model=True, + ) + assert test_result["status"] == "pass" + assert test_result["column_name"].lower() == RESERVED_WORD_COLUMN + + points = _anomaly_test_points(dbt_project, test_id) + assert points, "No metric data points were collected" + # The dimension must resolve to the reserved-word leaf's value, not to nulls. + assert {point["dimension"] for point in points} == {RESERVED_WORD_COLUMN} + assert {point["dimension_value"] for point in points} == {"paid"} + + +@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..f74d42d22 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) %} + + {#- Let the adapter expand nested STRUCT leaves (e.g. user.address.city) into + monitorable columns. No-op on adapters without nested-field support. -#} + {% set column_objects = elementary.flatten_columns_for_monitoring( + column_objects, column_name + ) %} + {% 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..7fc7a8082 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,20 @@ {%- 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.dimension_monitoring_alias(dimension_column) + ) %} {% endfor %} + {#- Ask the adapter how to project and reference the monitored column. For an + ordinary column both are just `column_obj.quoted`; adapters with nested + STRUCT support project a computed expression under a safe alias and + reference that alias in the metric aggregates. -#} + {%- set monitored_column = elementary.monitored_column_projection(column_obj) %} + {%- set monitored_column_projection = monitored_column.projection %} + {%- set monitored_column_expr = monitored_column.expression %} + {% set metric_types = [] %} {% set metric_name_to_type = {} %} {% for metric in column_metrics %} @@ -53,7 +64,7 @@ ), filtered_monitored_table as ( select - {{ column_obj.quoted }}, + {{ monitored_column_projection }}, {%- if dimensions -%} {{ elementary.select_dimensions_columns( @@ -78,7 +89,7 @@ {%- else %} filtered_monitored_table as ( select - {{ column_obj.quoted }}, + {{ monitored_column_projection }}, {%- if dimensions -%} {{ elementary.select_dimensions_columns( @@ -94,7 +105,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 +352,21 @@ {% endif %} {% endmacro %} +{# Renders a dimension select list. When aliasing (`as_prefix`), the dimension + SQL and the alias suffix are resolved per-adapter, so nested struct paths are + handled where supported and everything else stays byte-identical to previous + behaviour. Without a prefix the values are already-built column references and + pass through unchanged. #} {% 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.dimension_monitoring_sql(column) }} + {{- " as " ~ as_prefix ~ "_" ~ elementary.dimension_monitoring_alias(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..ceb80546a 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, "; ") %} + + {# Resolve each dimension to its SQL form. Adapters with nested-field support + segment-quote struct paths (e.g. user.address.city); plain identifiers, + expressions and other adapters pass through unchanged. #} + {% set sql_dimensions = [] %} + {% for dimension in dimensions %} + {% do sql_dimensions.append(elementary.dimension_monitoring_sql(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/edr/data_monitoring/monitors_query/nested_column_support.sql b/macros/edr/data_monitoring/monitors_query/nested_column_support.sql new file mode 100644 index 000000000..ed41d182e --- /dev/null +++ b/macros/edr/data_monitoring/monitors_query/nested_column_support.sql @@ -0,0 +1,118 @@ +{# ---------------------------------------------------------------------- #} +{# Adapter-agnostic hooks for nested-column (STRUCT) monitoring. #} +{# #} +{# The monitoring query macros are cross-warehouse and must not know about #} +{# any specific adapter, so the BigQuery STRUCT handling lives behind these #} +{# dispatched macros. The `default__` implementations are exact no-ops — #} +{# byte-identical to the pre-nested-support behaviour — and the BigQuery #} +{# specifics live in the `bigquery__` overrides, which delegate to the #} +{# helpers in `macros/utils/sql_utils/bigquery_nested_columns.sql`. #} +{# A new warehouse only needs its own `__` overrides; the generic #} +{# macros never change. #} +{# ---------------------------------------------------------------------- #} +{# Expand the columns of a monitored relation before a column is looked up by + name. Lets an adapter surface nested STRUCT leaves (e.g. user.address.city) + as monitorable columns. `column_name` is the requested column so an adapter + can skip the work when the request cannot be a nested path. #} +{% macro flatten_columns_for_monitoring(column_objects, column_name) %} + {{ + return( + adapter.dispatch("flatten_columns_for_monitoring", "elementary")( + column_objects, column_name + ) + ) + }} +{% endmacro %} + +{% macro default__flatten_columns_for_monitoring(column_objects, column_name) %} + {{ return(column_objects) }} +{% endmacro %} + +{% macro bigquery__flatten_columns_for_monitoring(column_objects, column_name) %} + {#- Only a dotted name can refer to a nested STRUCT leaf, so skip the + (potentially wide) flattening pass entirely for ordinary columns. -#} + {%- if "." not in column_name -%} {{ return(column_objects) }} {%- endif -%} + {{ return(elementary.bq_flatten_nested_columns(column_objects)) }} +{% endmacro %} + + +{# How to project a monitored column and how to reference it downstream. + Returns {"projection": , "expression": }. + For an ordinary column both are just the quoted column; an adapter can + override to project a computed expression under an alias and reference the + alias in the metric aggregates. #} +{% macro monitored_column_projection(column_obj) %} + {{ + return( + adapter.dispatch("monitored_column_projection", "elementary")(column_obj) + ) + }} +{% endmacro %} + +{% macro default__monitored_column_projection(column_obj) %} + {{ return({"projection": column_obj.quoted, "expression": column_obj.quoted}) }} +{% endmacro %} + +{% macro bigquery__monitored_column_projection(column_obj) %} + {#- A nested 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 alias = adapter.quote(elementary.bq_safe_alias(column_obj.name)) -%} + {{ + return( + { + "projection": ( + elementary.bq_segment_quote(column_obj.name) ~ " as " ~ alias + ), + "expression": alias, + } + ) + }} + {%- else -%} + {{ + return( + {"projection": column_obj.quoted, "expression": column_obj.quoted} + ) + }} + {%- endif -%} +{% endmacro %} + + +{# SQL form of a dimension for the select list / concat expression. Plain + identifiers and arbitrary SQL expressions pass through unchanged. #} +{% macro dimension_monitoring_sql(dimension) %} + {{ return(adapter.dispatch("dimension_monitoring_sql", "elementary")(dimension)) }} +{% endmacro %} + +{% macro default__dimension_monitoring_sql(dimension) %} + {{ return(dimension) }} +{% endmacro %} + +{% macro bigquery__dimension_monitoring_sql(dimension) %} + {{ return(elementary.bq_segment_quote(dimension)) }} +{% endmacro %} + + +{# Alias-safe form of a dimension, used to build the `dimension_<...>` column + alias. Must be a dot-free identifier; plain identifiers and expressions pass + through unchanged. #} +{% macro dimension_monitoring_alias(dimension) %} + {{ + return( + adapter.dispatch("dimension_monitoring_alias", "elementary")(dimension) + ) + }} +{% endmacro %} + +{% macro default__dimension_monitoring_alias(dimension) %} + {{ return(dimension) }} +{% endmacro %} + +{% macro bigquery__dimension_monitoring_alias(dimension) %} + {{ return(elementary.bq_alias_safe_dimension(dimension)) }} +{% endmacro %} 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..d835e062e --- /dev/null +++ b/macros/utils/sql_utils/bigquery_nested_columns.sql @@ -0,0 +1,129 @@ +{# ---------------------------------------------------------------------- #} +{# 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. + Known limitation: the mapping is not injective (`a.b__c` and `a__b.c` + both become `a__b__c`). Colliding paths used as dimensions on the same + test fail loudly with a duplicate-column error from BigQuery. -#} + {{- 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 %}