diff --git a/CHANGELOG.md b/CHANGELOG.md index eb1c68d0..f39be996 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `datacontract import` can now import BigQuery type `INTERVAL` (#1367,#1372 @fantastisch) ### Fixed +- `datacontract import` against BigQuery and dbt (BigQuery adapter) now supports compound BigQuery types such as `ARRAY>`, `STRUCT<...>`, `STRING(n)`, and `NUMERIC(p, s)` instead of failing with `Unsupported type`. - Failed business definition IRI lookups now suggest the `ENTROPY_DATA_HOST` value to set when the IRI host does not match the configured entropy-data host. - `datacontract test` no longer reports a physical type mismatch for BigQuery type aliases, such as a `physicalType` of `INTEGER` on an `INT64` column (#1371 @fantastisch) - `datacontract test` no longer fails with `CANNOT_CONVERT_COLUMN_INTO_BOOL` on Databricks when a Spark session is used diff --git a/datacontract/imports/bigquery_importer.py b/datacontract/imports/bigquery_importer.py index ac80b2b2..ccb04097 100644 --- a/datacontract/imports/bigquery_importer.py +++ b/datacontract/imports/bigquery_importer.py @@ -1,6 +1,8 @@ import json import logging -from typing import List +from dataclasses import dataclass +from dataclasses import field as dc_field +from typing import List, Optional, Tuple from open_data_contract_standard.model import OpenDataContractStandard, SchemaProperty @@ -255,8 +257,314 @@ def import_table_fields(table_fields) -> List[SchemaProperty]: return properties +@dataclass +class ParsedBigQueryType: + """A parsed BigQuery type expression. + + Represents both scalar types (``STRING``, ``INT64``, ``NUMERIC(10, 2)``) + and compound types (``ARRAY<...>``, ``STRUCT``). + """ + + base_type: str + """The BigQuery type name in upper case, e.g. ``STRING``, ``ARRAY``, ``STRUCT``.""" + + params: List[str] = dc_field(default_factory=list) + """Positional parameters, e.g. ``["10", "2"]`` for ``NUMERIC(10, 2)``.""" + + element_type: Optional["ParsedBigQueryType"] = None + """For ``ARRAY`` and ``RANGE``: the element type ``T``.""" + + fields: List[Tuple[Optional[str], "ParsedBigQueryType"]] = dc_field(default_factory=list) + """For ``STRUCT``: an ordered list of ``(name, type)`` pairs. + + Field names may be ``None`` for anonymous struct fields. + """ + + +def parse_bigquery_type(type_str: str) -> ParsedBigQueryType: + """Parse a BigQuery type string into a :class:`ParsedBigQueryType`. + + Understands the compound grammar used by ``INFORMATION_SCHEMA.COLUMNS`` and + dbt models: ``STRING``, ``STRING(100)``, ``NUMERIC(10, 2)``, + ``ARRAY``, ``STRUCT``, and arbitrary + nesting such as ``ARRAY>>``. Empty struct + bodies (``STRUCT<>``) are permitted and produce an empty ``fields`` list. + """ + if type_str is None or not str(type_str).strip(): + raise DataContractException( + type="schema", + result="failed", + name="Parse bigquery type", + reason="BigQuery type is empty.", + engine="datacontract", + ) + + parsed, consumed = _parse_bigquery_type(type_str.strip(), 0) + remainder = type_str.strip()[consumed:].strip() + if remainder: + raise DataContractException( + type="schema", + result="failed", + name="Parse bigquery type", + reason=f"Unexpected trailing characters {remainder!r} in bigquery type {type_str!r}.", + engine="datacontract", + ) + return parsed + + +def _parse_bigquery_type(text: str, pos: int) -> Tuple[ParsedBigQueryType, int]: + """Parse a single BigQuery type starting at ``pos``. Returns (parsed, end_pos).""" + pos = _skip_ws(text, pos) + start = pos + while pos < len(text) and (text[pos].isalnum() or text[pos] == "_"): + pos += 1 + if pos == start: + raise DataContractException( + type="schema", + result="failed", + name="Parse bigquery type", + reason=f"Expected a type name at position {start} in {text!r}.", + engine="datacontract", + ) + base_type = text[start:pos].upper() + pos = _skip_ws(text, pos) + + parsed = ParsedBigQueryType(base_type=base_type) + + if pos < len(text) and text[pos] == "<": + pos = _skip_ws(text, pos + 1) + if base_type in ("STRUCT", "RECORD"): + parsed.fields, pos = _parse_struct_fields(text, pos) + else: + # ARRAY, RANGE, and any other angle-bracket container. + if pos < len(text) and text[pos] == ">": + # Empty parameter list, e.g. ARRAY<>. + pos += 1 + else: + element, pos = _parse_bigquery_type(text, pos) + parsed.element_type = element + pos = _skip_ws(text, pos) + if pos >= len(text) or text[pos] != ">": + raise DataContractException( + type="schema", + result="failed", + name="Parse bigquery type", + reason=f"Missing '>' after {base_type} element type in {text!r}.", + engine="datacontract", + ) + pos += 1 + elif pos < len(text) and text[pos] == "(": + parsed.params, pos = _parse_paren_params(text, pos) + + return parsed, pos + + +def _parse_struct_fields(text: str, pos: int) -> Tuple[List[Tuple[Optional[str], ParsedBigQueryType]], int]: + """Parse the body of a ``STRUCT<...>`` starting after the opening ``<``. + + Returns the ordered list of ``(field_name, field_type)`` pairs and the + position just after the closing ``>``. An empty body is allowed. + """ + fields: List[Tuple[Optional[str], ParsedBigQueryType]] = [] + pos = _skip_ws(text, pos) + if pos < len(text) and text[pos] == ">": + return fields, pos + 1 + + while True: + pos = _skip_ws(text, pos) + # A struct field is written as either `` `` or just ````. + # Peek to decide: if we see IDENT followed by whitespace and another + # IDENT-like token (a type name), the first IDENT is the field name. + name_start = pos + while pos < len(text) and (text[pos].isalnum() or text[pos] == "_"): + pos += 1 + first_token = text[name_start:pos] + after_first = _skip_ws(text, pos) + + field_name: Optional[str] = None + if after_first < len(text) and (text[after_first].isalnum() or text[after_first] == "_"): + # Two identifiers in a row — first is the name, second starts the type. + field_name = first_token + pos = after_first + field_type, pos = _parse_bigquery_type(text, pos) + else: + # Only one identifier — treat it as an anonymous type name. + # Re-parse from the beginning of the identifier so parameters + # like ``STRING(100)`` are picked up. + pos = name_start + field_type, pos = _parse_bigquery_type(text, pos) + + fields.append((field_name, field_type)) + pos = _skip_ws(text, pos) + if pos < len(text) and text[pos] == ",": + pos += 1 + continue + if pos < len(text) and text[pos] == ">": + return fields, pos + 1 + raise DataContractException( + type="schema", + result="failed", + name="Parse bigquery type", + reason=f"Expected ',' or '>' in STRUCT body at position {pos} in {text!r}.", + engine="datacontract", + ) + + +def _parse_paren_params(text: str, pos: int) -> Tuple[List[str], int]: + """Parse ``(a, b, c)`` starting at the opening ``(`` and return ``([a, b, c], end)``.""" + assert text[pos] == "(" + end = text.find(")", pos + 1) + if end == -1: + raise DataContractException( + type="schema", + result="failed", + name="Parse bigquery type", + reason=f"Missing ')' in bigquery type {text!r}.", + engine="datacontract", + ) + raw = text[pos + 1 : end] + params = [p.strip() for p in raw.split(",")] if raw.strip() else [] + return params, end + 1 + + +def _skip_ws(text: str, pos: int) -> int: + while pos < len(text) and text[pos].isspace(): + pos += 1 + return pos + + +def build_schema_property_from_bigquery_type( + name: str, + bigquery_type_str: str, + description: Optional[str] = None, + required: Optional[bool] = None, +) -> SchemaProperty: + """Build a :class:`SchemaProperty` for ``name`` from a BigQuery type string. + + Populates nested ``items`` (for ``ARRAY<...>``/``RANGE<...>``) and + ``properties`` (for ``STRUCT<...>``) recursively, and lifts ``STRING(n)`` + length as well as ``NUMERIC(p, s)``/``BIGNUMERIC(p, s)`` precision/scale + onto the resulting property. + """ + parsed = parse_bigquery_type(bigquery_type_str) + return _build_property_from_parsed( + parsed, + name=name, + description=description, + required=required, + ) + + +def _build_property_from_parsed( + parsed: ParsedBigQueryType, + name: str, + description: Optional[str] = None, + required: Optional[bool] = None, +) -> SchemaProperty: + """Recursively convert a :class:`ParsedBigQueryType` into a :class:`SchemaProperty`.""" + base = parsed.base_type + logical_type = _map_base_type_to_logical(base) + physical_type = base + + max_length = _extract_max_length(parsed) + precision, scale = _extract_precision_scale(parsed) + + if base in ("STRUCT", "RECORD"): + nested_properties = [ + _build_property_from_parsed( + field_type, + name=field_name if field_name is not None else f"field_{index + 1}", + ) + for index, (field_name, field_type) in enumerate(parsed.fields) + ] or None + return create_property( + name=name, + logical_type=logical_type, + physical_type=physical_type, + description=description, + required=required, + properties=nested_properties, + ) + + if base in ("ARRAY", "RANGE"): + items_prop: Optional[SchemaProperty] = None + if parsed.element_type is not None: + items_prop = _build_property_from_parsed(parsed.element_type, name="items") + return create_property( + name=name, + logical_type=logical_type, + physical_type=physical_type if base != "ARRAY" else None, + description=description, + required=required, + items=items_prop, + ) + + return create_property( + name=name, + logical_type=logical_type, + physical_type=physical_type, + description=description, + required=required, + max_length=max_length, + precision=precision, + scale=scale, + ) + + +def _extract_max_length(parsed: ParsedBigQueryType) -> Optional[int]: + """Return the ``STRING(n)``/``BYTES(n)`` length parameter, if any.""" + if parsed.base_type in ("STRING", "BYTES") and len(parsed.params) == 1: + try: + return int(parsed.params[0]) + except ValueError: + return None + return None + + +def _extract_precision_scale(parsed: ParsedBigQueryType) -> Tuple[Optional[int], Optional[int]]: + """Return ``(precision, scale)`` for ``NUMERIC``/``BIGNUMERIC``, if provided.""" + if parsed.base_type not in ("NUMERIC", "BIGNUMERIC"): + return None, None + precision: Optional[int] = None + scale: Optional[int] = None + if len(parsed.params) >= 1: + try: + precision = int(parsed.params[0]) + except ValueError: + precision = None + if len(parsed.params) >= 2: + try: + scale = int(parsed.params[1]) + except ValueError: + scale = None + return precision, scale + + def map_type_from_bigquery(bigquery_type_str: str) -> str: - """Map BigQuery type to ODCS logical type.""" + """Map a BigQuery type string to an ODCS logical type. + + Supports both simple types (``STRING``, ``INT64``) and compound types + produced by ``INFORMATION_SCHEMA``/dbt such as ``STRING(100)``, + ``NUMERIC(10, 2)``, ``ARRAY``, ``STRUCT``, and + ``ARRAY>``. Only the outer type drives the + return value; use :func:`build_schema_property_from_bigquery_type` to also + materialize nested ``items``/``properties`` on a :class:`SchemaProperty`. + """ + if bigquery_type_str is None: + raise DataContractException( + type="schema", + result="failed", + name="Map bigquery type to data contract type", + reason="BigQuery type is None.", + engine="datacontract", + ) + + parsed = parse_bigquery_type(bigquery_type_str) + return _map_base_type_to_logical(parsed.base_type) + + +def _map_base_type_to_logical(base_type: str) -> str: + """Map a bare BigQuery base type (no parameters, no children) to an ODCS logical type.""" type_mapping = { "STRING": "string", "BYTES": "array", @@ -274,17 +582,21 @@ def map_type_from_bigquery(bigquery_type_str: str) -> str: "BIGNUMERIC": "number", "GEOGRAPHY": "object", "JSON": "object", + "ARRAY": "array", + "STRUCT": "object", + "RECORD": "object", + "RANGE": "array", "INTERVAL": "string", } - if bigquery_type_str in type_mapping: - return type_mapping[bigquery_type_str] + if base_type in type_mapping: + return type_mapping[base_type] raise DataContractException( type="schema", result="failed", name="Map bigquery type to data contract type", - reason=f"Unsupported type {bigquery_type_str} in bigquery json definition.", + reason=f"Unsupported type {base_type} in bigquery json definition.", engine="datacontract", ) diff --git a/datacontract/imports/dbt_importer.py b/datacontract/imports/dbt_importer.py index 5ed6c4c3..019bb02e 100644 --- a/datacontract/imports/dbt_importer.py +++ b/datacontract/imports/dbt_importer.py @@ -4,7 +4,10 @@ from open_data_contract_standard.model import CustomProperty, OpenDataContractStandard, SchemaProperty -from datacontract.imports.bigquery_importer import map_type_from_bigquery +from datacontract.imports.bigquery_importer import ( + build_schema_property_from_bigquery_type, + map_type_from_bigquery, +) from datacontract.imports.importer import Importer from datacontract.imports.odcs_helper import ( create_odcs, @@ -354,7 +357,6 @@ def create_field( ) -> SchemaProperty: """Create an ODCS SchemaProperty from a dbt column.""" data_type = column.get("data_type") - column_type = convert_data_type_by_adapter_type(data_type, adapter_type) if data_type else "string" all_tests = get_column_tests(manifest, model_unique_id, column.get("name")) @@ -382,6 +384,25 @@ def create_field( if column.get("tags"): custom_props["tags"] = ",".join(column["tags"]) + # BigQuery data types may be compound (``ARRAY>``), so build the + # nested ``items``/``properties`` tree via the BigQuery type parser. Other + # adapters use the flat scalar mapping. + if adapter_type == "bigquery" and data_type: + prop = build_schema_property_from_bigquery_type( + name=column.get("name"), + bigquery_type_str=data_type, + description=column.get("description"), + required=required if required else None, + ) + _apply_column_metadata( + prop, + unique=unique, + is_primary_key=is_primary_key, + custom_props=custom_props, + ) + return prop + + column_type = convert_data_type_by_adapter_type(data_type, adapter_type) if data_type else "string" return create_property( name=column.get("name"), logical_type=column_type, @@ -393,3 +414,22 @@ def create_field( primary_key_position=1 if is_primary_key else None, custom_properties=custom_props if custom_props else None, ) + + +def _apply_column_metadata( + prop: SchemaProperty, + unique: bool, + is_primary_key: bool, + custom_props: dict[str, Any], +) -> None: + """Attach dbt column-level metadata (unique/primary key/custom props) to *prop* in-place.""" + if unique: + prop.unique = True + if is_primary_key: + prop.primaryKey = True + prop.primaryKeyPosition = 1 + if custom_props: + existing = {cp.property: cp.value for cp in (prop.customProperties or [])} + for key, value in custom_props.items(): + existing.setdefault(key, value) + prop.customProperties = [CustomProperty(property=k, value=v) for k, v in existing.items()] diff --git a/tests/test_import_bigquery_type_parser.py b/tests/test_import_bigquery_type_parser.py new file mode 100644 index 00000000..abeb359b --- /dev/null +++ b/tests/test_import_bigquery_type_parser.py @@ -0,0 +1,310 @@ +"""Unit tests for the BigQuery type-string parser and SchemaProperty builder. + +These tests cover the compound BigQuery type grammar (``ARRAY<...>``, +``STRUCT<...>``, ``NUMERIC(p, s)``, arbitrary nesting) that ``dbt`` emits in +model YAML ``data_type`` fields and that BigQuery's +``INFORMATION_SCHEMA.COLUMNS`` returns. +""" + +import pytest + +from datacontract.imports.bigquery_importer import ( + build_schema_property_from_bigquery_type, + map_type_from_bigquery, + parse_bigquery_type, +) +from datacontract.model.exceptions import DataContractException + +# --------------------------------------------------------------------------- +# parse_bigquery_type +# --------------------------------------------------------------------------- + + +class TestParseBigQueryType: + def test_parses_simple_scalar_type(self): + parsed = parse_bigquery_type("STRING") + assert parsed.base_type == "STRING" + assert parsed.params == [] + assert parsed.element_type is None + assert parsed.fields == [] + + def test_lowercases_type_is_normalized_to_upper_case(self): + parsed = parse_bigquery_type("string") + assert parsed.base_type == "STRING" + + def test_parses_string_with_length(self): + parsed = parse_bigquery_type("STRING(100)") + assert parsed.base_type == "STRING" + assert parsed.params == ["100"] + + def test_parses_numeric_with_precision_and_scale(self): + parsed = parse_bigquery_type("NUMERIC(10, 2)") + assert parsed.base_type == "NUMERIC" + assert parsed.params == ["10", "2"] + + def test_parses_array_of_scalar(self): + parsed = parse_bigquery_type("ARRAY") + assert parsed.base_type == "ARRAY" + assert parsed.element_type is not None + assert parsed.element_type.base_type == "INT64" + + def test_parses_empty_struct(self): + parsed = parse_bigquery_type("STRUCT<>") + assert parsed.base_type == "STRUCT" + assert parsed.fields == [] + + def test_parses_array_of_empty_struct(self): + """The exact case that used to crash the dbt importer.""" + parsed = parse_bigquery_type("ARRAY>") + assert parsed.base_type == "ARRAY" + assert parsed.element_type is not None + assert parsed.element_type.base_type == "STRUCT" + assert parsed.element_type.fields == [] + + def test_parses_struct_with_named_fields(self): + parsed = parse_bigquery_type("STRUCT") + assert parsed.base_type == "STRUCT" + assert [name for name, _ in parsed.fields] == ["name", "age"] + assert [ft.base_type for _, ft in parsed.fields] == ["STRING", "INT64"] + + def test_parses_nested_array_of_struct_with_fields(self): + parsed = parse_bigquery_type("ARRAY>") + assert parsed.base_type == "ARRAY" + struct = parsed.element_type + assert struct is not None + assert struct.base_type == "STRUCT" + assert [name for name, _ in struct.fields] == ["name", "age"] + + def test_parses_deeply_nested_types(self): + parsed = parse_bigquery_type("ARRAY>>") + assert parsed.base_type == "ARRAY" + struct = parsed.element_type + assert struct.base_type == "STRUCT" + assert len(struct.fields) == 1 + name, field_type = struct.fields[0] + assert name == "items" + assert field_type.base_type == "ARRAY" + assert field_type.element_type.base_type == "STRING" + + def test_parses_struct_with_parameterized_field_types(self): + parsed = parse_bigquery_type("STRUCT") + code_name, code_type = parsed.fields[0] + amount_name, amount_type = parsed.fields[1] + assert code_name == "code" + assert code_type.base_type == "STRING" + assert code_type.params == ["10"] + assert amount_name == "amount" + assert amount_type.base_type == "NUMERIC" + assert amount_type.params == ["10", "2"] + + def test_ignores_surrounding_whitespace(self): + parsed = parse_bigquery_type(" ARRAY < INT64 > ") + assert parsed.base_type == "ARRAY" + assert parsed.element_type.base_type == "INT64" + + def test_parses_range_of_date(self): + parsed = parse_bigquery_type("RANGE") + assert parsed.base_type == "RANGE" + assert parsed.element_type.base_type == "DATE" + + @pytest.mark.parametrize("bad_input", [None, "", " "]) + def test_rejects_empty_or_none_input(self, bad_input): + with pytest.raises(DataContractException): + parse_bigquery_type(bad_input) + + def test_rejects_unbalanced_angle_brackets(self): + with pytest.raises(DataContractException): + parse_bigquery_type("ARRAY", + "ARRAY>", + "ARRAY>", + ], + ) + def test_array_types_map_to_array_logical_type(self, bigquery_type): + assert map_type_from_bigquery(bigquery_type) == "array" + + @pytest.mark.parametrize( + "bigquery_type", + [ + "STRUCT<>", + "STRUCT", + "RECORD", + ], + ) + def test_struct_types_map_to_object_logical_type(self, bigquery_type): + assert map_type_from_bigquery(bigquery_type) == "object" + + def test_rejects_unsupported_base_type(self): + with pytest.raises(DataContractException) as exc_info: + map_type_from_bigquery("MYSTERY_TYPE") + assert "MYSTERY_TYPE" in str(exc_info.value.reason) + + def test_rejects_none(self): + with pytest.raises(DataContractException): + map_type_from_bigquery(None) + + +# --------------------------------------------------------------------------- +# build_schema_property_from_bigquery_type +# --------------------------------------------------------------------------- + + +class TestBuildSchemaPropertyFromBigQueryType: + def test_builds_scalar_property(self): + prop = build_schema_property_from_bigquery_type("id", "STRING") + assert prop.name == "id" + assert prop.logicalType == "string" + assert prop.physicalType == "STRING" + assert prop.items is None + assert prop.properties is None + + def test_builds_string_property_with_max_length(self): + prop = build_schema_property_from_bigquery_type("code", "STRING(10)") + assert prop.logicalType == "string" + assert prop.physicalType == "STRING" + assert prop.logicalTypeOptions == {"maxLength": 10} + + def test_builds_numeric_property_with_precision_and_scale(self): + prop = build_schema_property_from_bigquery_type("amount", "NUMERIC(10, 2)") + assert prop.logicalType == "number" + assert prop.physicalType == "NUMERIC" + # ODCS v3.1.0 forbids precision/scale in logicalTypeOptions, so the helper + # carries them as customProperties. + custom_props = {cp.property: cp.value for cp in (prop.customProperties or [])} + assert custom_props == {"precision": 10, "scale": 2} + + def test_builds_array_of_scalar_with_items(self): + prop = build_schema_property_from_bigquery_type("tags", "ARRAY") + assert prop.logicalType == "array" + assert prop.items is not None + assert prop.items.name == "items" + assert prop.items.logicalType == "string" + assert prop.items.physicalType == "STRING" + + def test_builds_struct_with_named_field_properties(self): + prop = build_schema_property_from_bigquery_type( + "address", + "STRUCT", + ) + assert prop.logicalType == "object" + assert prop.physicalType == "STRUCT" + assert prop.properties is not None + assert [p.name for p in prop.properties] == ["street", "zip"] + assert [p.logicalType for p in prop.properties] == ["string", "integer"] + + def test_builds_empty_struct_as_object_without_properties(self): + prop = build_schema_property_from_bigquery_type("payload", "STRUCT<>") + assert prop.logicalType == "object" + assert prop.physicalType == "STRUCT" + assert prop.properties is None + + def test_builds_array_of_empty_struct(self): + """The originally reported case: ``ARRAY>``.""" + prop = build_schema_property_from_bigquery_type( + "incomeDetails", + "ARRAY>", + description="Income details of the customer.", + ) + assert prop.name == "incomeDetails" + assert prop.logicalType == "array" + assert prop.description == "Income details of the customer." + assert prop.items is not None + assert prop.items.name == "items" + assert prop.items.logicalType == "object" + assert prop.items.physicalType == "STRUCT" + assert prop.items.properties is None + + def test_builds_array_of_struct_with_named_fields(self): + prop = build_schema_property_from_bigquery_type( + "incomeDetails", + "ARRAY>", + ) + assert prop.logicalType == "array" + items = prop.items + assert items is not None + assert items.logicalType == "object" + assert items.physicalType == "STRUCT" + assert items.properties is not None + assert [p.name for p in items.properties] == ["source", "amount"] + # The nested NUMERIC(10, 2) should carry its precision/scale. + amount = next(p for p in items.properties if p.name == "amount") + amount_custom = {cp.property: cp.value for cp in (amount.customProperties or [])} + assert amount_custom == {"precision": 10, "scale": 2} + + def test_builds_deeply_nested_array_struct_array(self): + prop = build_schema_property_from_bigquery_type( + "records", + "ARRAY>>", + ) + assert prop.logicalType == "array" + struct_items = prop.items + assert struct_items.logicalType == "object" + assert [p.name for p in struct_items.properties] == ["name", "tags"] + tags = next(p for p in struct_items.properties if p.name == "tags") + assert tags.logicalType == "array" + assert tags.items.logicalType == "string" + + def test_preserves_required_and_description(self): + prop = build_schema_property_from_bigquery_type( + "customer_id", + "STRING", + description="Primary customer identifier.", + required=True, + ) + assert prop.required is True + assert prop.description == "Primary customer identifier." + + def test_builds_struct_with_anonymous_fields_as_field_n(self): + """Anonymous struct fields become ``field_1``, ``field_2``, ... in the property.""" + prop = build_schema_property_from_bigquery_type("payload", "STRUCT") + assert prop.logicalType == "object" + assert prop.properties is not None + assert [p.name for p in prop.properties] == ["field_1", "field_2"] + assert [p.logicalType for p in prop.properties] == ["string", "integer"] diff --git a/tests/test_import_dbt.py b/tests/test_import_dbt.py index 431a2b5a..9d0e6004 100644 --- a/tests/test_import_dbt.py +++ b/tests/test_import_dbt.py @@ -3,7 +3,7 @@ from datacontract.cli import app from datacontract.data_contract import DataContract -from datacontract.imports.dbt_importer import read_dbt_manifest +from datacontract.imports.dbt_importer import import_dbt_manifest, read_dbt_manifest # logging.basicConfig(level=logging.DEBUG, force=True) @@ -154,3 +154,115 @@ def test_cli_versioned_filter_v1(): assert result.exit_code == 0 parsed = yaml.safe_load(result.output) assert parsed.get("schema"), "Expected non-empty schema in CLI output for mart_orders.v1" + + +# --- BigQuery compound data_type tests (ARRAY>) --------------------- + + +def _make_bigquery_manifest(columns: dict) -> dict: + """Build a minimal dbt manifest with a single BigQuery model and the given columns.""" + return { + "metadata": { + "project_name": "test_project", + "dbt_version": "1.8.0", + "adapter_type": "bigquery", + }, + "nodes": { + "model.test_project.customers": { + "unique_id": "model.test_project.customers", + "resource_type": "model", + "name": "customers", + "description": "Test customer model", + "config": {"materialized": "table"}, + "columns": columns, + } + }, + "child_map": {"model.test_project.customers": []}, + } + + +def test_import_dbt_bigquery_array_of_empty_struct(): + """Reproduces the original bug: ``ARRAY>`` used to raise ``Unsupported type``.""" + manifest = _make_bigquery_manifest( + { + "incomeDetails": { + "name": "incomeDetails", + "data_type": "ARRAY>", + "description": "Income details of the customer.", + } + } + ) + + result = import_dbt_manifest(manifest, dbt_nodes=[], resource_types=["model"]) + + assert len(result.schema_) == 1 + props = result.schema_[0].properties + assert len(props) == 1 + income = props[0] + assert income.name == "incomeDetails" + assert income.logicalType == "array" + assert income.description == "Income details of the customer." + assert income.items is not None + assert income.items.logicalType == "object" + assert income.items.physicalType == "STRUCT" + assert income.items.properties is None + + +def test_import_dbt_bigquery_array_of_struct_with_named_fields(): + manifest = _make_bigquery_manifest( + { + "incomeDetails": { + "name": "incomeDetails", + "data_type": "ARRAY>", + "description": "Income details of the customer.", + } + } + ) + + result = import_dbt_manifest(manifest, dbt_nodes=[], resource_types=["model"]) + income = result.schema_[0].properties[0] + assert income.logicalType == "array" + items = income.items + assert items.logicalType == "object" + assert items.physicalType == "STRUCT" + assert [p.name for p in items.properties] == ["source", "amount"] + amount = next(p for p in items.properties if p.name == "amount") + amount_custom = {cp.property: cp.value for cp in (amount.customProperties or [])} + assert amount_custom == {"precision": 10, "scale": 2} + + +def test_import_dbt_bigquery_struct_column(): + manifest = _make_bigquery_manifest( + { + "address": { + "name": "address", + "data_type": "STRUCT", + "description": "Postal address.", + } + } + ) + + result = import_dbt_manifest(manifest, dbt_nodes=[], resource_types=["model"]) + address = result.schema_[0].properties[0] + assert address.logicalType == "object" + assert address.physicalType == "STRUCT" + assert [p.name for p in address.properties] == ["street", "zip"] + assert [p.logicalType for p in address.properties] == ["string", "integer"] + + +def test_import_dbt_bigquery_scalar_columns_still_work(): + """Regression guard: simple BigQuery types keep their previous behaviour.""" + manifest = _make_bigquery_manifest( + { + "id": {"name": "id", "data_type": "INT64", "description": "PK"}, + "name": {"name": "name", "data_type": "STRING(50)"}, + } + ) + + result = import_dbt_manifest(manifest, dbt_nodes=[], resource_types=["model"]) + id_prop, name_prop = result.schema_[0].properties + assert id_prop.logicalType == "integer" + assert id_prop.physicalType == "INT64" + assert name_prop.logicalType == "string" + assert name_prop.physicalType == "STRING" + assert name_prop.logicalTypeOptions == {"maxLength": 50}