diff --git a/packages/osi-orionbelt/src/osi_orionbelt/osi_to_obml.py b/packages/osi-orionbelt/src/osi_orionbelt/osi_to_obml.py index c55743f8..c8e0322a 100644 --- a/packages/osi-orionbelt/src/osi_orionbelt/osi_to_obml.py +++ b/packages/osi-orionbelt/src/osi_orionbelt/osi_to_obml.py @@ -20,6 +20,12 @@ OSI_TO_OBML_TYPE, ) +# A dataset/column identifier in a resolved metric expression: either a bare SQL +# word or a bracket-quoted token. ``_resolve_column_refs`` bracket-quotes any +# canonical name that is not a bare word (e.g. a display name with spaces) so the +# downstream ``dataset.column`` parsers below never split it on whitespace. +_RESOLVED_IDENT = r"(?:\w+|\[[^\]]+\])" + class OSItoOBML: """Convert an OSI semantic model YAML to OBML format.""" @@ -578,7 +584,23 @@ def _extract_dimensions(self, datasets: list) -> dict: except (json.JSONDecodeError, TypeError): pass break - dimensions[field_name] = dim_def + # Dimension names must be unique across the model. When the same + # field name occurs in more than one data object (e.g. Orders.date + # and Invoices.date), qualify the later one with its data object + # instead of silently overwriting the earlier dimension. + key = field_name + if key in dimensions and dimensions[key].get("dataObject") != ds_name: + key = f"{ds_name} {field_name}" + suffix = 2 + while key in dimensions: + key = f"{ds_name} {field_name} {suffix}" + suffix += 1 + self.warnings.append( + f"Dimension name '{field_name}' occurs in multiple data " + f"objects; emitted '{ds_name}.{field_name}' as dimension " + f"'{key}' to avoid a collision." + ) + dimensions[key] = dim_def return dimensions def _convert_metrics(self, osi_metrics: list, ds_map: dict) -> tuple[dict, dict]: @@ -600,16 +622,35 @@ def _convert_metrics(self, osi_metrics: list, ds_map: dict) -> tuple[dict, dict] # Case-insensitive dataset/field index for resolving SQL identifiers # back to their canonical OSI names (Snowflake/Databricks expressions - # commonly upper-case or quote them). - ds_lc = {name.lower(): name for name in ds_map} - fields_lc = { - name: { - f["name"].lower(): f["name"] - for f in ds.get("fields", []) or [] - if isinstance(f, dict) and f.get("name") - } - for name, ds in ds_map.items() - } + # commonly upper-case or quote them). Identifiers resolve by BOTH the + # OSI name and the physical code (source-table code / bare-identifier + # field expression): our own OBML -> OSI emitter writes metric SQL + # against the physical code (e.g. SUM(fact_orders.amount)), so resolving + # names only would drop such metrics on the return trip. Names take + # precedence over codes on any collision. + ds_lc: dict[str, str] = {} + for ds_name in ds_map: + ds_lc.setdefault(ds_name.lower(), ds_name) + for ds_name, ds in ds_map.items(): + # Unquote so a quoted source table (Snowflake/Databricks style, + # e.g. WH.PUBLIC."fact_orders") is indexed by its bare code. + table_code = self._unquote_identifier(self._parse_source(ds.get("source", ds_name))[2]) + if table_code: + ds_lc.setdefault(table_code.lower(), ds_name) + + fields_lc: dict[str, dict[str, str]] = {} + for ds_name, ds in ds_map.items(): + fmap: dict[str, str] = {} + osi_fields = ds.get("fields", []) or [] + for f in osi_fields: + if isinstance(f, dict) and f.get("name"): + fmap.setdefault(f["name"].lower(), f["name"]) + for f in osi_fields: + if isinstance(f, dict) and f.get("name"): + code = self._field_expr_identifier(f) + if code: + fmap.setdefault(code.lower(), f["name"]) + fields_lc[ds_name] = fmap for m in osi_metrics: name = m["name"] @@ -973,6 +1014,33 @@ def _unquote_identifier(ident: str) -> str: return ident[1:-1] return ident + @staticmethod + def _q_ident(name: str) -> str: + """Bracket-quote a canonical name that is not a bare SQL word, so the + downstream ``dataset.column`` parsers (which split on ``\\w+``) do not + break on display names containing spaces or other punctuation.""" + return name if re.fullmatch(r"\w+", name) else f"[{name}]" + + @staticmethod + def _field_expr_identifier(field: dict) -> str | None: + """Physical column code of a field when its expression is a single + (optionally quoted) identifier, so code-based metric references (e.g. + ``fact_orders.amount`` or a Snowflake ``"net_amount"``) resolve back to + the field. Returns ``None`` for computed expressions with no single + column code. + """ + expr = field.get("expression") + if not isinstance(expr, dict): + return None + for dialect in expr.get("dialects", []) or []: + if isinstance(dialect, dict): + text = dialect.get("expression") + if isinstance(text, str): + candidate = OSItoOBML._unquote_identifier(text) + if re.fullmatch(r"[A-Za-z_]\w*", candidate): + return candidate + return None + def _resolve_column_refs( self, expr: str, ds_lc: dict[str, str], fields_lc: dict[str, dict[str, str]] ) -> str | None: @@ -999,7 +1067,9 @@ def repl(match: re.Match[str]) -> str: if col_real is None: unresolved = True return match.group(0) - return f"{ds_real}.{col_real}" + # Bracket-quote names that are not bare words so the downstream + # dataset.column parsers keep multi-word display names intact. + return f"{self._q_ident(ds_real)}.{self._q_ident(col_real)}" rewritten = _COLUMN_REF_RE.sub(repl, expr) return None if unresolved else rewritten @@ -1027,13 +1097,13 @@ def _parse_simple_agg(self, expr: str) -> tuple | None: import re expr = expr.strip() - pattern = r"^(\w+)\(\s*(DISTINCT\s+)?(\w+)\.(\w+)\s*\)$" + pattern = rf"^(\w+)\(\s*(DISTINCT\s+)?({_RESOLVED_IDENT})\.({_RESOLVED_IDENT})\s*\)$" match = re.match(pattern, expr, re.IGNORECASE) if match: agg = match.group(1) is_distinct = match.group(2) is not None - dataset = match.group(3) - column = match.group(4) + dataset = self._unquote_identifier(match.group(3)) + column = self._unquote_identifier(match.group(4)) return agg, dataset, column, is_distinct return None @@ -1070,10 +1140,10 @@ def _parse_expr_agg(self, expr: str) -> tuple | None: return None # Unmatched open paren # Must contain dataset.column references - if not re.search(r"\w+\.\w+", inner): + if not re.search(rf"{_RESOLVED_IDENT}\.{_RESOLVED_IDENT}", inner): return None # Must NOT be a simple dataset.column (already handled by _parse_simple_agg) - if re.match(r"^(DISTINCT\s+)?\w+\.\w+$", inner, re.IGNORECASE): + if re.match(rf"^(DISTINCT\s+)?{_RESOLVED_IDENT}\.{_RESOLVED_IDENT}$", inner, re.IGNORECASE): return None # Must NOT contain nested aggregation calls (those are complex metrics) if re.search(r"\b(" + "|".join(agg_funcs) + r")\s*\(", inner, re.IGNORECASE): @@ -1089,7 +1159,14 @@ def _sql_refs_to_obml(self, sql_expr: str) -> str: (resolved upstream), so the bare-identifier branch is what fires. """ return _COLUMN_REF_RE.sub( - lambda m: "{[" + m.group("ds") + "].[" + m.group("col") + "]}", sql_expr + lambda m: ( + "{[" + + self._unquote_identifier(m.group("ds")) + + "].[" + + self._unquote_identifier(m.group("col")) + + "]}" + ), + sql_expr, ) def _decompose_complex_metric(self, name: str, expr: str) -> tuple[str, dict]: @@ -1146,12 +1223,13 @@ def _decompose_complex_metric(self, name: str, expr: str) -> tuple[str, dict]: inner_clean = inner[dm.end() :].strip() # Is it a simple dataset.column? - simple = re.match(r"^(\w+)\.(\w+)$", inner_clean) + simple = re.match(rf"^({_RESOLVED_IDENT})\.({_RESOLVED_IDENT})$", inner_clean) if simple: - dataset = simple.group(1) - column = simple.group(2) + dataset = self._unquote_identifier(simple.group(1)) + column = self._unquote_identifier(simple.group(2)) suffix = "_distinct" if is_distinct else "" - measure_key = f"_{dataset}_{column}_{agg.lower()}{suffix}" + key_stub = re.sub(r"\W+", "_", f"{dataset}_{column}") + measure_key = f"_{key_stub}_{agg.lower()}{suffix}" measure_def: dict[str, Any] = { "columns": [{"dataObject": dataset, "column": column}], "resultType": "float", diff --git a/packages/osi-orionbelt/src/osi_orionbelt/validation.py b/packages/osi-orionbelt/src/osi_orionbelt/validation.py index f26d2b7b..53f47d30 100644 --- a/packages/osi-orionbelt/src/osi_orionbelt/validation.py +++ b/packages/osi-orionbelt/src/osi_orionbelt/validation.py @@ -221,13 +221,22 @@ def validate_osi(osi_dict: dict[str, Any], schema_path: Path | None = None) -> V # 1. JSON Schema validation (OSI uses Draft 2020-12) _validate_json_schema(osi_dict, schema_path or _OSI_SCHEMA_PATH, result, draft="draft2020") + # The semantic checks below assume a well-formed structure (lists of dicts). + # JSON Schema validation above already reports structural errors, so guard + # every level here rather than raising on malformed input. + def _as_dict_list(value: Any) -> list[dict[str, Any]]: + return [item for item in value if isinstance(item, dict)] if isinstance(value, list) else [] + + models = _as_dict_list(osi_dict.get("semantic_model", [])) + # 2. Unique name checks - for model in osi_dict.get("semantic_model", []): + for model in models: model_name = model.get("name", "") + datasets = _as_dict_list(model.get("datasets", [])) # Unique dataset names dataset_names: list[str] = [] - for ds in model.get("datasets", []): + for ds in datasets: name = ds.get("name", "") if name in dataset_names: result.semantic_errors.append( @@ -236,10 +245,10 @@ def validate_osi(osi_dict: dict[str, Any], schema_path: Path | None = None) -> V dataset_names.append(name) # Unique field names within each dataset - for ds in model.get("datasets", []): + for ds in datasets: ds_name = ds.get("name", "") field_names: list[str] = [] - for field in ds.get("fields", []): + for field in _as_dict_list(ds.get("fields", [])): fname = field.get("name", "") if fname in field_names: result.semantic_errors.append( @@ -249,7 +258,7 @@ def validate_osi(osi_dict: dict[str, Any], schema_path: Path | None = None) -> V # Unique metric names metric_names: list[str] = [] - for m in model.get("metrics", []): + for m in _as_dict_list(model.get("metrics", [])): mname = m.get("name", "") if mname in metric_names: result.semantic_errors.append( @@ -259,7 +268,7 @@ def validate_osi(osi_dict: dict[str, Any], schema_path: Path | None = None) -> V # Unique relationship names rel_names: list[str] = [] - for r in model.get("relationships", []): + for r in _as_dict_list(model.get("relationships", [])): rname = r.get("name", "") if rname in rel_names: result.semantic_errors.append( @@ -269,9 +278,10 @@ def validate_osi(osi_dict: dict[str, Any], schema_path: Path | None = None) -> V rel_names.append(rname) # 3. Reference checks — relationships reference existing datasets - for model in osi_dict.get("semantic_model", []): - ds_name_set = {ds.get("name") for ds in model.get("datasets", []) if ds.get("name")} - for rel in model.get("relationships", []): + for model in models: + datasets = _as_dict_list(model.get("datasets", [])) + ds_name_set = {ds.get("name") for ds in datasets if ds.get("name")} + for rel in _as_dict_list(model.get("relationships", [])): rel_name = rel.get("name", "") from_ds = rel.get("from") to_ds = rel.get("to") diff --git a/packages/osi-orionbelt/tests/test_osi_converter_roundtrip_robustness.py b/packages/osi-orionbelt/tests/test_osi_converter_roundtrip_robustness.py new file mode 100644 index 00000000..c3cf992c --- /dev/null +++ b/packages/osi-orionbelt/tests/test_osi_converter_roundtrip_robustness.py @@ -0,0 +1,296 @@ +"""Regression tests for issue #201: OSI<->OBML converter round-trip fidelity +and validation robustness. + +Covers three bugs found during the apache/ossie#153 review: + +- P1: metric references broke the OBML -> OSI -> OBML round trip when a data + object's physical ``code`` differs from its display name (the emitter writes + the physical code, the importer resolved only by dataset name). +- P2: dimensions extracted from OSI collided when the same field name occurred + in more than one dataset (the second silently overwrote the first). +- P2: ``validate_osi`` raised ``AttributeError`` on malformed input instead of + returning a result with schema errors. +""" + +from __future__ import annotations + +import json +from typing import Any + +import osi_orionbelt.converter as conv + +# --------------------------------------------------------------------------- +# P1: metric round-trip when dataObject.code != display name +# --------------------------------------------------------------------------- + + +class TestMetricRoundTripCodeVsName: + _OBML: dict[str, Any] = { + "version": 1.0, + "dataObjects": { + "Orders": { + "code": "fact_orders", # physical table code differs from name + "database": "WH", + "schema": "PUBLIC", + "columns": {"Amount": {"code": "amount", "abstractType": "float"}}, + } + }, + "measures": { + "Revenue": { + "columns": [{"dataObject": "Orders", "column": "Amount"}], + "resultType": "float", + "aggregation": "sum", + } + }, + } + + def test_emitter_uses_physical_code(self) -> None: + osi = conv.OBMLtoOSI(self._OBML).convert() + sql = osi["semantic_model"][0]["metrics"][0]["expression"]["dialects"][0]["expression"] + assert "fact_orders" in sql # confirms the emit side uses the code + + def test_measure_survives_round_trip(self) -> None: + osi = conv.OBMLtoOSI(self._OBML).convert() + obml = conv.OSItoOBML(osi).convert() + + # The measure must come back as a real measure, not be stashed as an + # unconverted metric. + assert "Revenue" in obml.get("measures", {}), obml.get("measures") + rev = obml["measures"]["Revenue"] + assert rev["aggregation"] == "sum" + assert rev["columns"] == [{"dataObject": "Orders", "column": "amount"}] + + # Nothing about Revenue should have leaked into an unconverted-metric stash. + stashed = json.dumps(obml.get("customExtensions", [])) + assert "Revenue" not in stashed + + def test_third_party_field_name_differs_from_code(self) -> None: + # Snowflake-style OSI: field display name != physical column code, and + # the metric references the physical code. Must resolve, not drop. + osi = { + "version": "0.2.0.dev0", + "semantic_model": [ + { + "name": "sales", + "datasets": [ + { + "name": "Orders", + "source": "WH.PUBLIC.fact_orders", + "fields": [ + { + "name": "Amount", + "expression": { + "dialects": [ + {"dialect": "ANSI_SQL", "expression": "amount"} + ] + }, + "data_type": "number", + } + ], + } + ], + "metrics": [ + { + "name": "Revenue", + "expression": { + "dialects": [ + {"dialect": "ANSI_SQL", "expression": "SUM(fact_orders.amount)"} + ] + }, + } + ], + } + ], + } + obml = conv.OSItoOBML(osi).convert() + assert "Revenue" in obml.get("measures", {}), obml.get("measures") + assert obml["measures"]["Revenue"]["columns"] == [ + {"dataObject": "Orders", "column": "Amount"} + ] + + def test_physical_code_maps_to_field_name_with_space(self) -> None: + # OSI field display name contains a space; the metric references the + # physical column code. Resolution must produce a valid OBML measure + # (column "Net Amount"), not a broken "{[Orders].[Net]} Amount" that + # fails query resolution. + osi = { + "version": "0.2.0.dev0", + "semantic_model": [ + { + "name": "sales", + "datasets": [ + { + "name": "Orders", + "source": "WH.PUBLIC.fact_orders", + "fields": [ + { + "name": "Net Amount", + "expression": { + "dialects": [ + {"dialect": "ANSI_SQL", "expression": "net_amount"} + ] + }, + "data_type": "number", + } + ], + } + ], + "metrics": [ + { + "name": "Net Revenue", + "expression": { + "dialects": [ + { + "dialect": "ANSI_SQL", + "expression": "SUM(fact_orders.net_amount)", + } + ] + }, + } + ], + } + ], + } + obml = conv.OSItoOBML(osi).convert() + assert "Net Revenue" in obml.get("measures", {}), obml.get("measures") + m = obml["measures"]["Net Revenue"] + assert m["aggregation"] == "sum" + assert m["columns"] == [{"dataObject": "Orders", "column": "Net Amount"}] + # It resolved to a clean single-column measure, not a dangling expression. + assert "expression" not in m + + def test_quoted_physical_identifiers_resolve(self) -> None: + # Snowflake/Databricks style: the source table and the field expression + # are quoted identifiers. A metric referencing the bare physical code + # must still resolve to a queryable measure, not fall through to LOSSY. + osi = { + "version": "0.2.0.dev0", + "semantic_model": [ + { + "name": "sales", + "datasets": [ + { + "name": "Orders", + "source": 'WH.PUBLIC."fact_orders"', + "fields": [ + { + "name": "Amount", + "expression": { + "dialects": [ + {"dialect": "ANSI_SQL", "expression": '"net_amount"'} + ] + }, + "data_type": "number", + } + ], + } + ], + "metrics": [ + { + "name": "Revenue", + "expression": { + "dialects": [ + { + "dialect": "ANSI_SQL", + "expression": "SUM(fact_orders.net_amount)", + } + ] + }, + } + ], + } + ], + } + obml = conv.OSItoOBML(osi).convert() + assert "Revenue" in obml.get("measures", {}), obml.get("measures") + assert obml["measures"]["Revenue"]["columns"] == [ + {"dataObject": "Orders", "column": "Amount"} + ] + + +# --------------------------------------------------------------------------- +# P2: same field name in two datasets must not collide into one dimension +# --------------------------------------------------------------------------- + + +def _dim_dataset(name: str, source: str) -> dict[str, Any]: + return { + "name": name, + "source": source, + "fields": [ + { + "name": "order_date", + "expression": {"dialects": [{"dialect": "ANSI_SQL", "expression": "order_date"}]}, + "data_type": "date", + "dimension": {}, + } + ], + } + + +class TestDimensionNameCollision: + def test_both_dimensions_survive(self) -> None: + osi = { + "version": "0.2.0.dev0", + "semantic_model": [ + { + "name": "sales", + "datasets": [ + _dim_dataset("Orders", "WH.PUBLIC.orders"), + _dim_dataset("Invoices", "WH.PUBLIC.invoices"), + ], + } + ], + } + c = conv.OSItoOBML(osi) + obml = c.convert() + dims = obml.get("dimensions", {}) + + # Two distinct dimensions, one per data object — neither dropped. + assert len(dims) == 2, dims + assert {d["dataObject"] for d in dims.values()} == {"Orders", "Invoices"} + # The collision was disambiguated, not silent. + assert any("multiple data objects" in w for w in c.warnings) + + +# --------------------------------------------------------------------------- +# P2: validate_osi robustness on malformed input +# --------------------------------------------------------------------------- + + +class TestValidateOsiRobustness: + def test_malformed_datasets_does_not_raise(self) -> None: + # datasets is a string, not a list — must return a result, not raise. + r = conv.validate_osi( + {"version": "0.1.1", "semantic_model": [{"name": "x", "datasets": "not-an-array"}]} + ) + assert r is not None + # No garbage semantic errors from iterating a string char-by-char. + assert all("DUPLICATE" not in e for e in r.semantic_errors) + + def test_malformed_fields_does_not_raise(self) -> None: + r = conv.validate_osi( + { + "version": "0.2.0.dev0", + "semantic_model": [{"name": "x", "datasets": [{"name": "D", "fields": "nope"}]}], + } + ) + assert r is not None + + def test_still_flags_duplicate_datasets(self) -> None: + # The robustness guards must not suppress real semantic errors. + r = conv.validate_osi( + { + "version": "0.2.0.dev0", + "semantic_model": [ + { + "name": "m", + "datasets": [ + {"name": "D", "source": "a.b.d", "fields": []}, + {"name": "D", "source": "a.b.d2", "fields": []}, + ], + } + ], + } + ) + assert any("DUPLICATE_DATASET" in e for e in r.semantic_errors)