fix(dbt): emit expr '1' for COUNT(*) metrics

converters/dbt/README.md

@@ -129,6 +129,7 @@ manifest_json = result.output.model_dump_json(by_alias=True, exclude_none=True,
 
 - Composite primary and unique keys are rejected because MSI entities cannot preserve grouped key semantics
 - Single aggregations (`SUM(col)`, `COUNT(DISTINCT col)`, etc.) → SIMPLE metric with `metric_aggregation_params`
+- `COUNT(*)` / `COUNT(<dataset>.*)` → `count` SIMPLE metric with `expr: '1'`, because MetricFlow cannot render a bare `*` inside a count. The counted dataset comes from the qualifier, so with more than one dataset write `COUNT(orders.*)`; a bare `COUNT(*)` is rejected as ambiguous
 - `(expr_a) / (expr_b)` → RATIO metric with auto-generated sub-metrics
 - Anything else → SIMPLE metric with the raw expression stored verbatim
 - Time dimensions always receive `TimeGranularity.DAY` (Ossie carries no granularity field)

converters/dbt/src/ossie_dbt/expression_utils.py

@@ -22,6 +22,9 @@
 
 from metricflow_semantic_interfaces.type_enums import AggregationType
 
+# expr for "count all rows": MetricFlow wraps a count's expr in CASE WHEN, where a bare * is invalid
+ROW_COUNT_EXPR = "1"
+
 
 def _strip_qualifier(col: str) -> str:
     """Strip a leading dataset qualifier, e.g. 'orders.amount' → 'amount'."""
@@ -36,29 +39,39 @@ def _col_name(node: exp.Expression) -> str:
     return _strip_qualifier(rendered)
 
 
+def _is_star(node: exp.Expression) -> bool:
+    """Return True for ``*`` and for a qualified ``dataset.*``."""
+    return isinstance(node, exp.Star) or (isinstance(node, exp.Column) and isinstance(node.this, exp.Star))
+
+
 def _extract_agg_info(expression: str) -> Optional[Tuple[AggregationType, str, Optional[float], bool]]:
     """Parse a SQL aggregation expression using sqlglot.
 
     Returns ``(agg_type, bare_col, percentile, use_discrete_percentile)`` for recognised patterns,
     ``None`` otherwise. ``percentile`` is only set for ``PERCENTILE`` aggregations; it is ``None``
     for all others. ``use_discrete_percentile`` is ``True`` only for ``PERCENTILE_DISC``.
-    The returned column name has any dataset qualifier stripped.
+    The returned column name has any dataset qualifier stripped. ``COUNT(*)`` and ``COUNT(dataset.*)`` return
+    ``ROW_COUNT_EXPR`` instead of a column name; ``COUNT(DISTINCT *)`` and multi-argument ``COUNT`` return ``None``.
     """
     try:
         tree = sqlglot.parse_one(expression.strip())
     except sqlglot.errors.ParseError:
         return None
 
-    # COUNT(DISTINCT col)
-    if isinstance(tree, exp.Count) and isinstance(tree.this, exp.Distinct):
-        cols = tree.this.expressions
-        if len(cols) == 1:
-            return AggregationType.COUNT_DISTINCT, _col_name(cols[0]), None, False
-        return None
-
-    # COUNT(col)
     if isinstance(tree, exp.Count):
-        return AggregationType.COUNT, _col_name(tree.this), None, False
+        # COUNT(a, b) has no single-column equivalent
+        if tree.args.get("expressions"):
+            return None
+        argument, distinct = tree.this, False
+        if isinstance(argument, exp.Distinct):
+            operands = argument.expressions
+            if len(operands) != 1:
+                return None
+            argument, distinct = operands[0], True
+        if _is_star(argument):
+            # COUNT(*) → count all rows; COUNT(DISTINCT *) is not valid SQL
+            return None if distinct else (AggregationType.COUNT, ROW_COUNT_EXPR, None, False)
+        return (AggregationType.COUNT_DISTINCT if distinct else AggregationType.COUNT), _col_name(argument), None, False
 
     # SUM(CASE WHEN col THEN 1 ELSE 0 END) → SUM_BOOLEAN
     if isinstance(tree, exp.Sum) and isinstance(tree.this, exp.Case):

converters/dbt/src/ossie_dbt/msi_to_ossie.py

@@ -19,7 +19,7 @@
 from collections import defaultdict
 from dataclasses import dataclass
 from itertools import combinations
-from typing import Dict, List, Optional, Sequence, Tuple
+from typing import Dict, FrozenSet, List, Optional, Sequence, Tuple
 
 from ossie import (
     OssieDataset,
@@ -33,6 +33,7 @@
     OssieRelationship,
 )
 from ossie_dbt.converter_issues import ConverterIssue, ConverterIssueType, ConverterResult
+from ossie_dbt.expression_utils import ROW_COUNT_EXPR
 from ossie_dbt.filter_utils import _collect_filter_sql, _merge_filter_sqls
 
 from metricflow_semantic_interfaces.enum_extension import assert_values_exhausted
@@ -77,10 +78,21 @@ class MSIToOssieConverter:
 
     def __init__(self, dialect: OssieDialect = OssieDialect.ANSI_SQL) -> None:
         self._dialect = dialect
+        self._row_count_metrics: FrozenSet[str] = frozenset()
 
     def convert(
         self, manifest: PydanticSemanticManifest, ossie_model_name: str = "semantic_model"
     ) -> ConverterResult[OssieDocument]:
+        # The transformer rewrites COUNT to SUM (leaving expr '1' as SUM(1)), which loses the dataset a row
+        # count belongs to. Remember these metrics so they come back as COUNT(<dataset>.*).
+        self._row_count_metrics = frozenset(
+            metric.name
+            for metric in manifest.metrics
+            if metric.type is MetricType.SIMPLE
+            and metric.type_params.metric_aggregation_params is not None
+            and metric.type_params.metric_aggregation_params.agg is AggregationType.COUNT
+            and metric.type_params.expr == ROW_COUNT_EXPR
+        )
         manifest = PydanticSemanticManifestTransformer.transform(manifest)
         issues: List[ConverterIssue] = []
 
@@ -251,6 +263,9 @@ def _resolve_simple(
             raise ValueError(
                 f"SIMPLE metric has no metric_aggregation_params after transformation: metric_name={metric.name!r}"
             )
+        # With a filter the count is emitted as SUM(CASE WHEN <filter> THEN 1 END), which has no `dataset.*` form.
+        if metric.name in self._row_count_metrics and not filter_sql:
+            return f"COUNT({agg_params_obj.semantic_model}.*)"
         col = metric.type_params.expr if metric.type_params.expr is not None else metric.name
         col = self._qualify_col(col, agg_params_obj.semantic_model)
         return self._build_agg_expression(agg_params_obj.agg, col, agg_params_obj.agg_params, filter_sql)

converters/dbt/src/ossie_dbt/ossie_to_msi.py

@@ -28,6 +28,7 @@
 )
 from ossie_dbt.converter_issues import ConverterResult
 from ossie_dbt.expression_utils import (
+    ROW_COUNT_EXPR,
     _extract_agg_info,
     _get_dataset_qualifier,
     _strip_qualifier,
@@ -293,7 +294,11 @@ def _convert_metric(
         agg_result = _extract_agg_info(expr_str)
         if agg_result is not None:
             agg, col, percentile, use_discrete = agg_result
-            semantic_model_name = self._find_dataset_for_col(expr_str, col, datasets)
+            if agg is AggregationType.COUNT and col == ROW_COUNT_EXPR:
+                # A constant, not a column: it must not go through the column → dataset lookup.
+                semantic_model_name = self._find_dataset_for_row_count(name, expr_str, datasets)
+            else:
+                semantic_model_name = self._find_dataset_for_col(expr_str, col, datasets)
             agg_params = (
                 PydanticMeasureAggregationParameters(
                     percentile=percentile,
@@ -404,6 +409,38 @@ def _find_dataset_for_col(
 
         return datasets[0].name if datasets else ""
 
+    @staticmethod
+    def _find_dataset_for_row_count(
+        metric_name: str,
+        raw_expr_str: str,
+        datasets: List[OssieDataset],
+    ) -> str:
+        """Determine which dataset a ``COUNT(*)`` counts the rows of.
+
+        Only the qualifier decides: ``COUNT(orders.*)`` names ``orders`` (a schema prefix such as
+        ``db.orders.*`` is matched on its last segment). A bare ``COUNT(*)`` is only unambiguous when the
+        document has a single dataset; otherwise guessing would count the rows of an unrelated table.
+        """
+        dataset_names = [dataset.name for dataset in datasets]
+        qualifier = _get_dataset_qualifier(raw_expr_str)
+        if qualifier is None:
+            if len(dataset_names) == 1:
+                return dataset_names[0]
+            raise ValueError(
+                f"Metric {metric_name!r}: 'COUNT(*)' is ambiguous with {len(dataset_names)} datasets "
+                f"({', '.join(dataset_names)}); qualify it as 'COUNT(<dataset>.*)'"
+            )
+
+        if qualifier in dataset_names:
+            return qualifier
+        by_last_segment = [name for name in dataset_names if _strip_qualifier(name) == _strip_qualifier(qualifier)]
+        if len(by_last_segment) == 1:
+            return by_last_segment[0]
+        raise ValueError(
+            f"Metric {metric_name!r}: 'COUNT({qualifier}.*)' does not match exactly one dataset "
+            f"(datasets: {', '.join(dataset_names)})"
+        )
+
     def _get_expression(self, ossie_expr: OssieExpression) -> str:
         """Return the expression string for the preferred dialect (fallback: first available)."""
         for dialect_expr in ossie_expr.dialects:

converters/dbt/tests/test_msi_to_ossie.py

@@ -81,6 +81,34 @@ def _ossie_metrics(result: OssieDocument) -> list:
     return metrics
 
 
+def _metric_with_agg(
+    name: str,
+    agg: AggregationType,
+    expr: str,
+    semantic_model: str,
+    filter_sql: Optional[str] = None,
+) -> PydanticMetric:
+    """A SIMPLE metric that carries its aggregation in metric_aggregation_params."""
+    return PydanticMetric(
+        name=name,
+        description=None,
+        type=MetricType.SIMPLE,
+        type_params=PydanticMetricTypeParams(
+            expr=expr,
+            metric_aggregation_params=PydanticMetricAggregationParams(
+                semantic_model=semantic_model,
+                agg=agg,
+                agg_params=None,
+                agg_time_dimension=None,
+                non_additive_dimension=None,
+            ),
+        ),
+        filter=_filter(filter_sql) if filter_sql else None,
+        metadata=default_meta(),
+        config=None,
+    )
+
+
 # ---------------------------------------------------------------------------
 # Tests
 # ---------------------------------------------------------------------------
@@ -584,6 +612,30 @@ def test_simple_metric_with_metric_aggregation_params(self) -> None:
 
         assert _ossie_metrics(result)[0].expression.dialects[0].expression == "AVG(orders.price)"
 
+    def test_count_of_all_rows_keeps_its_semantic_model(self) -> None:
+        customers = semantic_model_with_guaranteed_meta(name="customers")
+        orders = semantic_model_with_guaranteed_meta(name="orders")
+        metric = _metric_with_agg("order_count", AggregationType.COUNT, "1", "orders")
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[customers, orders], metrics=[metric])).output
+
+        assert _ossie_metrics(result)[0].expression.dialects[0].expression == "COUNT(orders.*)"
+
+    def test_sum_of_constant_one_stays_a_sum(self) -> None:
+        orders = semantic_model_with_guaranteed_meta(name="orders")
+        metric = _metric_with_agg("row_total", AggregationType.SUM, "1", "orders")
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[orders], metrics=[metric])).output
+
+        assert _ossie_metrics(result)[0].expression.dialects[0].expression == "SUM(1)"
+
+    def test_filtered_count_of_all_rows_keeps_the_filter(self) -> None:
+        orders = semantic_model_with_guaranteed_meta(name="orders")
+        metric = _metric_with_agg("paid_orders", AggregationType.COUNT, "1", "orders", filter_sql="status = 'paid'")
+        result = MSIToOssieConverter().convert(_manifest(semantic_models=[orders], metrics=[metric])).output
+
+        assert (
+            _ossie_metrics(result)[0].expression.dialects[0].expression == "SUM(CASE WHEN status = 'paid' THEN 1 END)"
+        )
+
     # --- RATIO ---
 
     def test_ratio_metric_inlines_sub_expressions(self, snapshot: SnapshotAssertion) -> None:

converters/dbt/tests/test_ossie_to_msi.py

@@ -319,6 +319,115 @@ def test_count_distinct_expression(self) -> None:
         sm = result.semantic_models[0]
         assert len(sm.measures) == 0
 
+    @pytest.mark.parametrize("expression", ["COUNT(*)", "COUNT(orders.*)"])
+    def test_count_star_uses_constant_expr(self, expression: str) -> None:
+        doc = _ossie_doc(
+            datasets=[_ossie_dataset("orders", fields=[_ossie_field("order_id")])],
+            metrics=[_ossie_metric("order_count", expression)],
+        )
+        result = OssieToMSIConverter().convert(doc).output
+
+        m = result.metrics[0]
+        assert m.type_params.metric_aggregation_params is not None
+        assert m.type_params.metric_aggregation_params.agg == AggregationType.COUNT
+        assert m.type_params.expr == "1"
+
+    def test_qualified_count_star_uses_dataset_qualifier(self) -> None:
+        doc = _ossie_doc(
+            datasets=[
+                _ossie_dataset("customers", fields=[_ossie_field("customer_id")]),
+                _ossie_dataset("orders", fields=[_ossie_field("order_id")]),
+            ],
+            metrics=[_ossie_metric("order_count", "COUNT(orders.*)")],
+        )
+        result = OssieToMSIConverter().convert(doc).output
+
+        m = result.metrics[0]
+        assert m.type_params.expr == "1"
+        assert m.type_params.metric_aggregation_params is not None
+        assert m.type_params.metric_aggregation_params.semantic_model == "orders"
+
+    @staticmethod
+    def _customers_and_orders() -> list:
+        return [
+            _ossie_dataset("customers", fields=[_ossie_field("customer_id")]),
+            _ossie_dataset("orders", fields=[_ossie_field("order_id"), _ossie_field("amount")]),
+        ]
+
+    @pytest.mark.parametrize("expression", ["COUNT(*)", "COUNT(1)"])
+    def test_bare_row_count_with_multiple_datasets_is_ambiguous(self, expression: str) -> None:
+        doc = _ossie_doc(datasets=self._customers_and_orders(), metrics=[_ossie_metric("order_count", expression)])
+
+        with pytest.raises(ValueError, match="ambiguous with 2 datasets"):
+            OssieToMSIConverter().convert(doc)
+
+    def test_bare_count_star_in_ratio_with_multiple_datasets_is_ambiguous(self) -> None:
+        doc = _ossie_doc(
+            datasets=self._customers_and_orders(),
+            metrics=[_ossie_metric("avg_order_value", "(SUM(amount)) / (COUNT(*))")],
+        )
+
+        with pytest.raises(ValueError, match="ambiguous"):
+            OssieToMSIConverter().convert(doc)
+
+    def test_qualified_count_star_in_ratio_binds_both_sides_to_the_same_dataset(self) -> None:
+        doc = _ossie_doc(
+            datasets=self._customers_and_orders(),
+            metrics=[_ossie_metric("avg_order_value", "(SUM(orders.amount)) / (COUNT(orders.*))")],
+        )
+        result = OssieToMSIConverter().convert(doc).output
+
+        by_name = {m.name: m for m in result.metrics}
+        for sub_metric in ("avg_order_value__numerator", "avg_order_value__denominator"):
+            params = by_name[sub_metric].type_params.metric_aggregation_params
+            assert params is not None
+            assert params.semantic_model == "orders"
+
+    def test_schema_qualified_count_star_matches_dataset_by_last_segment(self) -> None:
+        doc = _ossie_doc(
+            datasets=self._customers_and_orders(),
+            metrics=[_ossie_metric("order_count", "COUNT(db.orders.*)")],
+        )
+        result = OssieToMSIConverter().convert(doc).output
+
+        params = result.metrics[0].type_params.metric_aggregation_params
+        assert params is not None
+        assert params.semantic_model == "orders"
+
+    def test_count_star_of_unknown_dataset_is_rejected(self) -> None:
+        doc = _ossie_doc(
+            datasets=self._customers_and_orders(),
+            metrics=[_ossie_metric("order_count", "COUNT(nope.*)")],
+        )
+
+        with pytest.raises(ValueError, match="does not match exactly one dataset"):
+            OssieToMSIConverter().convert(doc)
+
+    def test_count_star_matching_several_datasets_by_last_segment_is_rejected(self) -> None:
+        doc = _ossie_doc(
+            datasets=[
+                _ossie_dataset("a.orders", fields=[_ossie_field("order_id")]),
+                _ossie_dataset("b.orders", fields=[_ossie_field("order_id")]),
+            ],
+            metrics=[_ossie_metric("order_count", "COUNT(orders.*)")],
+        )
+
+        with pytest.raises(ValueError, match="does not match exactly one dataset"):
+            OssieToMSIConverter().convert(doc)
+
+    @pytest.mark.parametrize(
+        "expression",
+        ["COUNT(DISTINCT *)", "COUNT(orders.*, amount)", "COUNT(db.orders, *)"],
+    )
+    def test_unsupported_count_star_forms_fall_back_to_the_raw_expression(self, expression: str) -> None:
+        doc = _ossie_doc(
+            datasets=[_ossie_dataset("orders", fields=[_ossie_field("order_id"), _ossie_field("amount")])],
+            metrics=[_ossie_metric("odd_count", expression)],
+        )
+        result = OssieToMSIConverter().convert(doc).output
+
+        assert result.metrics[0].type_params.expr == expression
+
     def test_ratio_expression_produces_ratio_metric(self) -> None:
         doc = _ossie_doc(
             datasets=[
@@ -496,6 +605,36 @@ def test_ossie_to_msi_to_ossie_preserves_structure(self, snapshot: SnapshotAsser
         assert metrics[0].expression.dialects[0].expression == "SUM(orders.amount)"
         assert ossie_doc.to_ossie_yaml() == snapshot
 
+    def test_count_star_keeps_its_dataset_across_round_trip(self) -> None:
+        """COUNT(orders.*) must not drift to another dataset (or to SUM) on Ossie → MSI → Ossie → MSI."""
+        original = _ossie_doc(
+            datasets=[
+                _ossie_dataset("customers", fields=[_ossie_field("customer_id")]),
+                _ossie_dataset("orders", fields=[_ossie_field("order_id"), _ossie_field("amount")]),
+            ],
+            metrics=[
+                _ossie_metric("order_count", "COUNT(orders.*)"),
+                _ossie_metric("avg_order_value", "(SUM(orders.amount)) / (COUNT(orders.*))"),
+            ],
+        )
+
+        msi = OssieToMSIConverter().convert(original).output
+        ossie_doc = MSIToOssieConverter().convert(msi).output
+
+        expressions = {m.name: m.expression.dialects[0].expression for m in ossie_doc.metrics or []}
+        assert expressions["order_count"] == "COUNT(orders.*)"
+        assert "COUNT(orders.*)" in expressions["avg_order_value"]
+
+        again = OssieToMSIConverter().convert(ossie_doc).output
+        order_count = next(m for m in again.metrics if m.name == "order_count")
+        params = order_count.type_params.metric_aggregation_params
+        assert params is not None
+        assert (params.agg, order_count.type_params.expr, params.semantic_model) == (
+            AggregationType.COUNT,
+            "1",
+            "orders",
+        )
+
     def test_discrete_percentile_survives_round_trip(self) -> None:
         """A PERCENTILE_DISC metric keeps use_discrete_percentile through MSI -> Ossie -> MSI."""
         orders = semantic_model_with_guaranteed_meta(