osi-orionbelt: fix converter round-trip fidelity + validation robustness (#201)

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",

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", "<unnamed>")
+        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", "<unnamed>")
             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", "<unnamed>")
             from_ds = rel.get("from")
             to_ds = rel.get("to")