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

converters/dbt/README.md

@@ -130,6 +130,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(*)`, or a qualifier that matches no dataset, is skipped with a `ROW_COUNT_METRIC_DROPPED` warning
 - `(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/cli.py

@@ -25,11 +25,12 @@
 import argparse
 import sys
 from pathlib import Path
+from typing import Sequence
 
 import yaml
 
 from ossie import OssieDocument
-from ossie_dbt.converter_issues import ConverterIssueType
+from ossie_dbt.converter_issues import ConverterIssue, ConverterIssueType
 from ossie_dbt.msi_to_ossie import MSIToOssieConverter
 from ossie_dbt.ossie_to_msi import OssieToMSIConverter
 
@@ -40,6 +41,7 @@
     ConverterIssueType.PRIVATE_METRIC_DROPPED: "Ossie has no visibility modifiers",
     ConverterIssueType.NATURAL_ENTITY_DROPPED: "Ossie has no natural-key entity type",
     ConverterIssueType.CUMULATIVE_SEMANTICS_LOSS: "Ossie expressions cannot represent window or grain semantics; the base aggregation was preserved",
+    ConverterIssueType.ROW_COUNT_METRIC_DROPPED: "COUNT(*) does not identify exactly one dataset to count rows of; write it as COUNT(<dataset>.*)",
     ConverterIssueType.AMBIGUOUS_REFERENCE_METRIC_DROPPED: (
         "an input metric is listed more than once under one reference with differing filters, "
         "so the expression reference is ambiguous; give each occurrence a distinct alias"
@@ -50,22 +52,26 @@
     ConverterIssueType.CONVERSION_METRIC_DROPPED,
     ConverterIssueType.PRIVATE_METRIC_DROPPED,
     ConverterIssueType.NATURAL_ENTITY_DROPPED,
+    ConverterIssueType.ROW_COUNT_METRIC_DROPPED,
     ConverterIssueType.AMBIGUOUS_REFERENCE_METRIC_DROPPED,
 }
 
 
+def _print_issues(issues: Sequence[ConverterIssue]) -> None:
+    for issue in issues:
+        verb = "was dropped" if issue.issue_type in _DROPPED_ISSUE_TYPES else "was converted with loss"
+        reason = _ISSUE_REASON[issue.issue_type]
+        print(f"[WARNING] {issue.issue_type.value}: {issue.element_name} {verb} during conversion because {reason}", file=sys.stderr)
+
+
 def _cmd_msi_to_ossie(args: argparse.Namespace) -> None:
     input_path = Path(args.input)
     output_path = Path(args.output)
 
     manifest = parse_manifest_from_dbt_generated_manifest(input_path.read_text())
     result = MSIToOssieConverter().convert(manifest, ossie_model_name=args.model_name)
 
-    if result.issues:
-        for issue in result.issues:
-            verb = "was dropped" if issue.issue_type in _DROPPED_ISSUE_TYPES else "was converted with loss"
-            reason = _ISSUE_REASON[issue.issue_type]
-            print(f"[WARNING] {issue.issue_type.value}: {issue.element_name} {verb} during conversion because {reason}", file=sys.stderr)
+    _print_issues(result.issues)
 
     output_path.write_text(result.output.to_ossie_yaml())
     print(f"Written to {output_path}", file=sys.stderr)
@@ -78,6 +84,7 @@ def _cmd_ossie_to_msi(args: argparse.Namespace) -> None:
     raw = yaml.safe_load(input_path.read_text())
     document = OssieDocument.model_validate(raw)
     result = OssieToMSIConverter().convert(document)
+    _print_issues(result.issues)
 
     # PydanticSemanticManifest subclasses pydantic.v1.BaseModel, whose JSON
     # serializer is .json(), not the pydantic v2 .model_dump_json().

converters/dbt/src/ossie_dbt/converter_issues.py

@@ -27,6 +27,7 @@ class ConverterIssueType(Enum):
     PRIVATE_METRIC_DROPPED = "PRIVATE_METRIC_DROPPED"
     NATURAL_ENTITY_DROPPED = "NATURAL_ENTITY_DROPPED"
     CUMULATIVE_SEMANTICS_LOSS = "CUMULATIVE_SEMANTICS_LOSS"
+    ROW_COUNT_METRIC_DROPPED = "ROW_COUNT_METRIC_DROPPED"
     AMBIGUOUS_REFERENCE_METRIC_DROPPED = "AMBIGUOUS_REFERENCE_METRIC_DROPPED"
 
 

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,50 @@ def _col_name(node: exp.Expression) -> str:
     return _strip_qualifier(rendered)
 
 
+def _is_row_count_argument(node: exp.Expression) -> bool:
+    """Return True for ``*`` (bare or qualified) and for any non-null constant.
+
+    None of these can ever be NULL, so ``COUNT()`` of one counts every row. A string literal is left
+    alone, since ``COUNT('x')`` is not a row-count idiom anyone writes on purpose.
+    """
+    if isinstance(node, exp.Star) or (isinstance(node, exp.Column) and isinstance(node.this, exp.Star)):
+        return True
+    if isinstance(node, exp.Boolean):
+        return True
+    if isinstance(node, exp.Literal) and not node.is_string:
+        return True
+    return False
+
+
 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`` of ``*`` or of any non-null constant
+    (``COUNT(1)``, ``COUNT(TRUE)``, ...) returns ``ROW_COUNT_EXPR`` instead of a column name;
+    ``COUNT(DISTINCT ...)`` of one of those, 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_row_count_argument(argument):
+            # COUNT(*), COUNT(1), COUNT(TRUE), ... → count all rows; COUNT(DISTINCT ...) of one 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
@@ -81,14 +82,28 @@ class AmbiguousDerivedReferenceError(Exception):
 
 
 class MSIToOssieConverter:
-    """Converts an MSI SemanticManifest into an Ossie Document."""
+    """Converts an MSI SemanticManifest into an Ossie Document.
+
+    Holds no per-call state on the instance, so one instance may be reused, including concurrently
+    from multiple threads, across any number of ``convert()`` calls.
+    """
 
     def __init__(self, dialect: OssieDialect = OssieDialect.ANSI_SQL) -> None:
         self._dialect = dialect
 
     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>.*).
+        row_count_metrics: FrozenSet[str] = 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] = []
 
@@ -118,7 +133,7 @@ def convert(
                     ConverterIssue(issue_type=ConverterIssueType.CUMULATIVE_SEMANTICS_LOSS, element_name=metric.name)
                 )
             try:
-                expr = self._resolve_metric_expression(metric, metric_index, expression_cache)
+                expr = self._resolve_metric_expression(metric, metric_index, expression_cache, row_count_metrics)
             except AmbiguousDerivedReferenceError:
                 # Every other unsupported shape drops one metric and records an issue;
                 # an ambiguous reference is no reason to fail the whole conversion.
@@ -232,6 +247,7 @@ def _resolve_metric_expression(
         metric: Metric,
         metric_index: Dict[str, Metric],
         cache: Dict[Tuple[str, Optional[str]], str],
+        row_count_metrics: FrozenSet[str],
         parent_filter: Optional[str] = None,
     ) -> str:
         """Recursively resolve a metric to a fully-inlined SQL expression string."""
@@ -243,13 +259,13 @@ def _resolve_metric_expression(
             return cache[cache_key]
 
         if metric.type is MetricType.SIMPLE:
-            expr = self._resolve_simple(metric, combined_filter)
+            expr = self._resolve_simple(metric, row_count_metrics, combined_filter)
         elif metric.type is MetricType.CUMULATIVE:
-            expr = self._resolve_cumulative(metric, metric_index, cache, combined_filter)
+            expr = self._resolve_cumulative(metric, metric_index, cache, row_count_metrics, combined_filter)
         elif metric.type is MetricType.RATIO:
-            expr = self._resolve_ratio(metric, metric_index, cache, combined_filter)
+            expr = self._resolve_ratio(metric, metric_index, cache, row_count_metrics, combined_filter)
         elif metric.type is MetricType.DERIVED:
-            expr = self._resolve_derived(metric, metric_index, cache, combined_filter)
+            expr = self._resolve_derived(metric, metric_index, cache, row_count_metrics, combined_filter)
         elif metric.type is MetricType.CONVERSION:
             # CONVERSION metrics are skipped in convert(); this branch should never be reached.
             raise RuntimeError(f"Unexpected CONVERSION metric in expression resolver: metric_name={metric.name!r}")
@@ -262,6 +278,7 @@ def _resolve_metric_expression(
     def _resolve_simple(
         self,
         metric: Metric,
+        row_count_metrics: FrozenSet[str],
         filter_sql: Optional[str] = None,
     ) -> str:
         """Resolve a SIMPLE metric using metric_aggregation_params (always set after transformation)."""
@@ -270,6 +287,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 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)
@@ -291,6 +311,7 @@ def _resolve_cumulative(
         metric: Metric,
         metric_index: Dict[str, Metric],
         cache: Dict[Tuple[str, Optional[str]], str],
+        row_count_metrics: FrozenSet[str],
         filter_sql: Optional[str] = None,
     ) -> str:
         """Resolve a CUMULATIVE metric to its base aggregation expression.
@@ -308,6 +329,7 @@ def _resolve_cumulative(
             self._lookup_metric(metric_index, sub_input.name, f"CUMULATIVE metric '{metric.name}'"),
             metric_index,
             cache,
+            row_count_metrics,
             sub_filter,
         )
 
@@ -316,6 +338,7 @@ def _resolve_ratio(
         metric: Metric,
         metric_index: Dict[str, Metric],
         cache: Dict[Tuple[str, Optional[str]], str],
+        row_count_metrics: FrozenSet[str],
         filter_sql: Optional[str] = None,
     ) -> str:
         """Resolve a RATIO metric as (numerator) / (denominator), both fully inlined."""
@@ -331,12 +354,14 @@ def _resolve_ratio(
             self._lookup_metric(metric_index, num_input.name, f"RATIO metric '{metric.name}' numerator"),
             metric_index,
             cache,
+            row_count_metrics,
             num_filter,
         )
         den_expr = self._resolve_metric_expression(
             self._lookup_metric(metric_index, den_input.name, f"RATIO metric '{metric.name}' denominator"),
             metric_index,
             cache,
+            row_count_metrics,
             den_filter,
         )
         return f"({num_expr}) / ({den_expr})"
@@ -346,6 +371,7 @@ def _resolve_derived(
         metric: Metric,
         metric_index: Dict[str, Metric],
         cache: Dict[Tuple[str, Optional[str]], str],
+        row_count_metrics: FrozenSet[str],
         filter_sql: Optional[str] = None,
     ) -> str:
         """Resolve a DERIVED metric by substituting each input metric's expression into the expr string.
@@ -375,7 +401,7 @@ def _resolve_derived(
             ref = input_metric.alias if input_metric.alias else input_metric.name
             dep_metric = self._lookup_metric(metric_index, input_metric.name, f"DERIVED metric '{metric.name}'")
             input_filter = _merge_filter_sqls(filter_sql, _collect_filter_sql(input_metric.filter))
-            resolved = self._resolve_metric_expression(dep_metric, metric_index, cache, input_filter)
+            resolved = self._resolve_metric_expression(dep_metric, metric_index, cache, row_count_metrics, input_filter)
             if dep_metric.type in (MetricType.DERIVED, MetricType.RATIO):
                 resolved = f"({resolved})"
             distinct = resolutions.setdefault(ref, [])