fix(dbt): stop emitting COUNT(<dataset>.*), SUM(1) is the row count

converters/dbt/README.md

@@ -125,12 +125,15 @@ manifest_json = result.output.model_dump_json(by_alias=True, exclude_none=True,
 | `NATURAL_ENTITY_DROPPED` | Ossie has no natural-key entity type |
 | `CUMULATIVE_SEMANTICS_LOSS` | Window/grain semantics cannot be expressed in an Ossie expression string; the base aggregation is preserved |
 | `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 |
+| `CONSTANT_METRIC_SEMANTIC_MODEL_LOSS` | A metric over a constant, such as a row count (`SUM(1)`), has no column to carry its semantic model; recorded when the manifest has more than one, since converting it back will refuse it |
 
 **Ossie → MSI** reconstructs a best-effort MSI manifest from Ossie's simpler schema. Nothing is dropped for supported inputs, but Ossie carries less structural information than MSI, so the converter makes the following choices:
 
 - 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
+- `SUM(<constant>)` (e.g. `SUM(1)`) keeps its constant, but has no column to place it in a dataset: with more than one dataset it is skipped with `ROW_COUNT_METRIC_DROPPED`
+- `COUNT(DISTINCT *)`, `COUNT(DISTINCT 1)` and the like, anywhere in an expression, are skipped with `ROW_COUNT_METRIC_DROPPED`: they count whether any row exists, not how many
 - `(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

@@ -41,7 +41,15 @@
     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.CONSTANT_METRIC_SEMANTIC_MODEL_LOSS: (
+        "its expression is a constant such as SUM(1), which has no column to say which semantic model it "
+        "counts; with more than one dataset, converting it back to dbt will refuse it"
+    ),
+    ConverterIssueType.ROW_COUNT_METRIC_DROPPED: (
+        "a row count or constant aggregate (COUNT(*), SUM(1), ...) did not identify exactly one dataset "
+        "(qualify a COUNT(*) as COUNT(<dataset>.*)), or has no sensible translation at all, such as "
+        "COUNT(DISTINCT *)"
+    ),
     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"

converters/dbt/src/ossie_dbt/converter_issues.py

@@ -29,6 +29,7 @@ class ConverterIssueType(Enum):
     CUMULATIVE_SEMANTICS_LOSS = "CUMULATIVE_SEMANTICS_LOSS"
     ROW_COUNT_METRIC_DROPPED = "ROW_COUNT_METRIC_DROPPED"
     AMBIGUOUS_REFERENCE_METRIC_DROPPED = "AMBIGUOUS_REFERENCE_METRIC_DROPPED"
+    CONSTANT_METRIC_SEMANTIC_MODEL_LOSS = "CONSTANT_METRIC_SEMANTIC_MODEL_LOSS"
 
 
 @dataclass(frozen=True)

converters/dbt/src/ossie_dbt/expression_utils.py

@@ -54,19 +54,58 @@ def _is_row_count_argument(node: exp.Expression) -> bool:
     return False
 
 
+def _is_constant_expr(expr: str) -> bool:
+    """Return True when ``expr`` is a non-null constant such as ``1``, ``2`` or ``TRUE``, not a column."""
+    try:
+        node = sqlglot.parse_one(expr)
+    except sqlglot.errors.SqlglotError:
+        return False
+    return isinstance(node, exp.Boolean) or (isinstance(node, exp.Literal) and not node.is_string)
+
+
+def _contains_distinct_row_count(expression: str) -> bool:
+    """Return True if ``expression`` contains ``COUNT(DISTINCT <row-count argument>)`` anywhere in its tree.
+
+    ``COUNT(DISTINCT *)`` / ``COUNT(DISTINCT 1)`` and friends parse and run as SQL, but counting distinct
+    values of ``*`` or a constant is not a sensible aggregation for a semantic layer: it answers whether
+    any row exists (0 or 1), not a meaningful total. The caller should drop the metric with an issue
+    rather than fall back to a raw expression, which would wrap this inside another aggregate
+    (``SUM(COUNT(DISTINCT ...))``, not valid for MetricFlow to run) and guess a dataset the way a row
+    count must not.
+
+    Searches the whole tree, not just the top node, so a wrapped or combined form such as
+    ``(COUNT(DISTINCT *))``, ``COUNT(DISTINCT *) * 100`` or ``COALESCE(COUNT(DISTINCT 1), 0)`` is still
+    caught, not only a bare ``COUNT(DISTINCT *)`` as the entire expression. The DISTINCT operand is
+    unnested before the check, so ``COUNT(DISTINCT (*))`` is caught the same way as ``COUNT(DISTINCT *)``.
+    """
+    try:
+        tree = sqlglot.parse_one(expression.strip())
+    except sqlglot.errors.SqlglotError:
+        return False
+    for count in tree.find_all(exp.Count):
+        argument = count.this
+        if count.args.get("expressions") or not isinstance(argument, exp.Distinct):
+            continue
+        operands = argument.expressions
+        if len(operands) == 1 and _is_row_count_argument(operands[0].unnest()):
+            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. ``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``.
+    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;
+    ``SUM`` of a constant returns the constant itself (``SUM(2)`` → ``'2'``). ``COUNT(DISTINCT ...)`` of a
+    row-count argument, and multi-argument ``COUNT``, return ``None``.
     """
     try:
         tree = sqlglot.parse_one(expression.strip())
-    except sqlglot.errors.ParseError:
+    except sqlglot.errors.SqlglotError:
         return None
 
     if isinstance(tree, exp.Count):
@@ -79,8 +118,9 @@ def _extract_agg_info(expression: str) -> Optional[Tuple[AggregationType, str, O
             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
+        if _is_row_count_argument(argument.unnest()):
+            # COUNT(*), COUNT(1), COUNT(TRUE), ... → count all rows; COUNT(DISTINCT ...) of one is not valid SQL.
+            # Unnested so a redundant paren, e.g. COUNT(DISTINCT (*)), is still recognised.
             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
 
@@ -99,8 +139,13 @@ def _extract_agg_info(expression: str) -> Optional[Tuple[AggregationType, str, O
             return AggregationType.SUM_BOOLEAN, ifs[0].this.sql(), None, False
         return None
 
-    # SUM(col)
+    # SUM(col), or SUM(<constant>). A constant keeps its own value (SUM(2) is twice the row count,
+    # not SUM(1)); the caller uses _is_constant_expr to send it through the same dataset check as
+    # COUNT(*), since a constant has no column to place it in a dataset.
     if isinstance(tree, exp.Sum):
+        argument = tree.this.unnest()
+        if _is_constant_expr(argument.sql()):
+            return AggregationType.SUM, argument.sql(), None, False
         return AggregationType.SUM, _col_name(tree.this), None, False
 
     if isinstance(tree, exp.Avg):
@@ -140,7 +185,7 @@ def _try_parse_ratio(expr_str: str) -> Optional[Tuple[str, str]]:
     """Try to parse ``(expr_a) / (expr_b)`` using sqlglot, returning ``(num_expr, den_expr)`` or None."""
     try:
         tree = sqlglot.parse_one(expr_str.strip())
-    except sqlglot.errors.ParseError:
+    except sqlglot.errors.SqlglotError:
         return None
 
     if not isinstance(tree, exp.Div):
@@ -162,7 +207,7 @@ 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:
+    except sqlglot.errors.SqlglotError:
         return None
 
     qualifiers = {

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, FrozenSet, List, Optional, Sequence, Tuple
+from typing import Dict, List, Optional, Sequence, Tuple
 
 from ossie import (
     OssieDataset,
@@ -33,7 +33,7 @@
     OssieRelationship,
 )
 from ossie_dbt.converter_issues import ConverterIssue, ConverterIssueType, ConverterResult
-from ossie_dbt.expression_utils import ROW_COUNT_EXPR
+from ossie_dbt.expression_utils import _is_constant_expr
 from ossie_dbt.filter_utils import _collect_filter_sql, _merge_filter_sqls
 
 from metricflow_semantic_interfaces.enum_extension import assert_values_exhausted
@@ -94,16 +94,6 @@ def __init__(self, dialect: OssieDialect = OssieDialect.ANSI_SQL) -> None:
     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] = []
 
@@ -137,7 +127,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, row_count_metrics)
+                expr = self._resolve_metric_expression(metric, metric_index, expression_cache)
             except AmbiguousDerivedReferenceError:
                 # Every other unsupported shape drops one metric and records an issue;
                 # an ambiguous reference is no reason to fail the whole conversion.
@@ -155,6 +145,16 @@ def convert(
                     description=metric.description,
                 )
             )
+            if len(manifest.semantic_models) > 1 and self._aggregates_a_constant(metric):
+                # SUM(1) and the like carry no column, so the semantic model the metric belonged to
+                # cannot be written into the Ossie expression. Converting it back refuses rather than
+                # guesses (ROW_COUNT_METRIC_DROPPED); record the loss here so it is not silent.
+                issues.append(
+                    ConverterIssue(
+                        issue_type=ConverterIssueType.CONSTANT_METRIC_SEMANTIC_MODEL_LOSS,
+                        element_name=metric.name,
+                    )
+                )
 
         return ConverterResult(
             output=OssieDocument(
@@ -251,7 +251,6 @@ 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."""
@@ -263,13 +262,13 @@ def _resolve_metric_expression(
             return cache[cache_key]
 
         if metric.type is MetricType.SIMPLE:
-            expr = self._resolve_simple(metric, row_count_metrics, combined_filter)
+            expr = self._resolve_simple(metric, combined_filter)
         elif metric.type is MetricType.CUMULATIVE:
-            expr = self._resolve_cumulative(metric, metric_index, cache, row_count_metrics, combined_filter)
+            expr = self._resolve_cumulative(metric, metric_index, cache, combined_filter)
         elif metric.type is MetricType.RATIO:
-            expr = self._resolve_ratio(metric, metric_index, cache, row_count_metrics, combined_filter)
+            expr = self._resolve_ratio(metric, metric_index, cache, combined_filter)
         elif metric.type is MetricType.DERIVED:
-            expr = self._resolve_derived(metric, metric_index, cache, row_count_metrics, combined_filter)
+            expr = self._resolve_derived(metric, metric_index, cache, 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}")
@@ -282,22 +281,38 @@ 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)."""
+        """Resolve a SIMPLE metric using metric_aggregation_params (always set after transformation).
+
+        No special case for a row count: ``SUM`` with ``expr == '1'`` already renders as ``SUM(1)``
+        through the generic path below. That matches ``COUNT(*)`` on any non-empty input (over zero
+        rows ``SUM(1)`` is NULL where ``COUNT(*)`` is 0, as with MetricFlow's own COUNT → SUM rewrite)
+        and is portable across engines, unlike ``COUNT(<dataset>.*)``, which several engines reject or
+        interpret differently.
+        """
         agg_params_obj = metric.type_params.metric_aggregation_params
         if agg_params_obj is None:
             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)
 
+    @staticmethod
+    def _aggregates_a_constant(metric: Metric) -> bool:
+        """Return True for a SIMPLE SUM over a constant expr, e.g. a row count rewritten to SUM(1).
+
+        Only SUM: it is the one aggregation the Ossie → MSI side resolves as a constant (and refuses on
+        ambiguous datasets). MAX(1), AVG(1) and the like go through the ordinary column lookup there.
+        """
+        params = metric.type_params.metric_aggregation_params
+        expr = metric.type_params.expr
+        if metric.type is not MetricType.SIMPLE or params is None or expr is None:
+            return False
+        return params.agg is AggregationType.SUM and _is_constant_expr(expr)
+
     @staticmethod
     def _qualify_col(col: str, semantic_model: str) -> str:
         """Qualify col with semantic_model if it is an unqualified identifier or a COUNT-converted expr."""
@@ -315,7 +330,6 @@ 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.
@@ -333,7 +347,6 @@ def _resolve_cumulative(
             self._lookup_metric(metric_index, sub_input.name, f"CUMULATIVE metric '{metric.name}'"),
             metric_index,
             cache,
-            row_count_metrics,
             sub_filter,
         )
 
@@ -342,7 +355,6 @@ 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."""
@@ -358,14 +370,12 @@ 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})"
@@ -375,7 +385,6 @@ 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.
@@ -405,7 +414,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, row_count_metrics, input_filter)
+            resolved = self._resolve_metric_expression(dep_metric, metric_index, cache, input_filter)
             if dep_metric.type in (MetricType.DERIVED, MetricType.RATIO):
                 resolved = f"({resolved})"
             distinct = resolutions.setdefault(ref, [])