fix(dbt): resolve SUM_BOOLEAN semantic model for qualified columns

converters/dbt/src/ossie_dbt/expression_utils.py

@@ -133,22 +133,16 @@ def _try_parse_ratio(expr_str: str) -> Optional[Tuple[str, str]]:
     return num.sql(), den.sql()
 
 
-def _get_raw_inner_col(expression: str) -> Optional[str]:
-    """Extract the raw column reference from inside a simple aggregation, before stripping qualifiers."""
+def _get_dataset_qualifier(expression: str) -> Optional[str]:
+    """Return the sole dataset qualifier referenced by an expression, if present."""
     try:
         tree = sqlglot.parse_one(expression.strip())
     except sqlglot.errors.ParseError:
         return None
 
-    if not isinstance(tree, exp.AggFunc):
-        return None
-
-    inner = tree.this
-    if inner is None:
-        return None
-
-    # For COUNT(DISTINCT col), unwrap the Distinct node
-    if isinstance(inner, exp.Distinct) and inner.expressions:
-        return inner.expressions[0].sql()
-
-    return inner.sql()
+    qualifiers = {
+        ".".join(part.sql() for part in column.parts[:-1])
+        for column in tree.find_all(exp.Column)
+        if len(column.parts) > 1
+    }
+    return qualifiers.pop() if len(qualifiers) == 1 else None

converters/dbt/src/ossie_dbt/osi_to_msi.py

@@ -29,7 +29,7 @@
 from ossie_dbt.converter_issues import ConverterResult
 from ossie_dbt.expression_utils import (
     _extract_agg_info,
-    _get_raw_inner_col,
+    _get_dataset_qualifier,
     _strip_qualifier,
     _try_parse_ratio,
 )
@@ -369,11 +369,12 @@ def _find_dataset_for_col(
         For unqualified references the datasets are scanned for a matching field name.
         Falls back to the first dataset's name if no match is found.
         """
-        # Check for a dataset qualifier in the raw expression (e.g. "orders.amount")
-        raw_inner = _get_raw_inner_col(raw_expr_str)
-        if raw_inner and "." in raw_inner:
-            ds_name, _ = raw_inner.rsplit(".", 1)
-            return ds_name
+        # Check for a dataset qualifier in the raw expression (e.g. "orders.amount").
+        # Parse column references instead of splitting the rendered inner expression,
+        # which may be a compound CASE expression for SUM_BOOLEAN metrics.
+        dataset_qualifier = _get_dataset_qualifier(raw_expr_str)
+        if dataset_qualifier:
+            return dataset_qualifier
 
         # Scan datasets for a field whose name or expression matches the bare column
         for dataset in datasets:

converters/dbt/tests/test_osi_to_msi.py

@@ -342,6 +342,42 @@ def test_dataset_qualified_column_reference(self) -> None:
         assert m.type_params.metric_aggregation_params.semantic_model == "orders"
         assert m.type_params.expr == "amount"
 
+    def test_sum_boolean_qualified_column_resolves_semantic_model(self) -> None:
+        doc = _osi_doc(
+            datasets=[
+                _osi_dataset("customers", fields=[_osi_field("customer_id")]),
+                _osi_dataset("orders", fields=[_osi_field("order_id")]),
+            ],
+            metrics=[
+                _osi_metric(
+                    "has_order",
+                    "SUM(CASE WHEN orders.order_id IS NOT NULL THEN 1 ELSE 0 END)",
+                )
+            ],
+        )
+        result = OSIToMSIConverter().convert(doc).output
+
+        metric = result.metrics[0]
+        assert metric.type_params.metric_aggregation_params is not None
+        assert metric.type_params.metric_aggregation_params.agg == AggregationType.SUM_BOOLEAN
+        assert metric.type_params.metric_aggregation_params.semantic_model == "orders"
+
+    def test_fully_qualified_column_preserves_dataset_name(self) -> None:
+        doc = _osi_doc(
+            datasets=[
+                _osi_dataset(
+                    "analytics.orders",
+                    fields=[_osi_field("order_id")],
+                )
+            ],
+            metrics=[_osi_metric("order_count", "COUNT(analytics.orders.order_id)")],
+        )
+        result = OSIToMSIConverter().convert(doc).output
+
+        metric = result.metrics[0]
+        assert metric.type_params.metric_aggregation_params is not None
+        assert metric.type_params.metric_aggregation_params.semantic_model == "analytics.orders"
+
     def test_percentile_cont_0_5_produces_median(self) -> None:
         doc = _osi_doc(
             datasets=[_osi_dataset("orders", fields=[_osi_field("amount")])],