fix(dbt): preserve aggregate expression semantics

converters/dbt/README.md

@@ -130,7 +130,12 @@ 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`
 - `(expr_a) / (expr_b)` → RATIO metric with auto-generated sub-metrics
-- Anything else → SIMPLE metric with the raw expression stored verbatim
+- A root aggregate over a scalar SQL expression (for example,
+  `SUM(CASE WHEN ... THEN amount ELSE 0 END)`) → SIMPLE metric with the
+  scalar expression in `expr` and the root function in `agg`
+- Expressions that cannot be decomposed without changing aggregation semantics,
+  and non-SQL measure expressions such as DAX, are dropped with an
+  `UNSUPPORTED_METRIC_EXPRESSION` issue
 - Time dimensions always receive `TimeGranularity.DAY` (Ossie carries no granularity field)
 
 ## Development

converters/dbt/src/ossie_dbt/cli.py

@@ -29,7 +29,7 @@
 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,27 +40,35 @@
     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.UNSUPPORTED_METRIC_EXPRESSION: "MetricFlow SIMPLE metrics require a scalar expression plus a separate aggregation",
 }
 
 _DROPPED_ISSUE_TYPES = {
     ConverterIssueType.CONVERSION_METRIC_DROPPED,
     ConverterIssueType.PRIVATE_METRIC_DROPPED,
     ConverterIssueType.NATURAL_ENTITY_DROPPED,
+    ConverterIssueType.UNSUPPORTED_METRIC_EXPRESSION,
 }
 
 
+def _print_issues(issues: list[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)
@@ -73,6 +81,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"
+    UNSUPPORTED_METRIC_EXPRESSION = "UNSUPPORTED_METRIC_EXPRESSION"
 
 
 @dataclass(frozen=True)

converters/dbt/src/ossie_dbt/expression_utils.py

@@ -32,17 +32,24 @@ def _col_name(node: exp.Expression) -> str:
     """Return the bare (unqualified) column name from a sqlglot expression node."""
     if isinstance(node, exp.Column):
         return node.name
-    rendered = node.sql()
-    return _strip_qualifier(rendered)
+    return node.sql()
+
+
+def _is_scalar_expression(node: exp.Expression) -> bool:
+    """Return whether a node can safely be placed beneath a MetricFlow aggregation."""
+    return not isinstance(node, exp.Distinct) and not any(
+        isinstance(child, (exp.AggFunc, exp.Window)) for child in node.walk()
+    )
 
 
 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,
+    Returns ``(agg_type, scalar_expr, 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.
+    A simple column expression has its dataset qualifier stripped; compound scalar expressions
+    retain their qualifiers.
     """
     try:
         tree = sqlglot.parse_one(expression.strip())
@@ -52,12 +59,12 @@ def _extract_agg_info(expression: str) -> Optional[Tuple[AggregationType, str, O
     # COUNT(DISTINCT col)
     if isinstance(tree, exp.Count) and isinstance(tree.this, exp.Distinct):
         cols = tree.this.expressions
-        if len(cols) == 1:
+        if len(cols) == 1 and _is_scalar_expression(cols[0]):
             return AggregationType.COUNT_DISTINCT, _col_name(cols[0]), None, False
         return None
 
     # COUNT(col)
-    if isinstance(tree, exp.Count):
+    if isinstance(tree, exp.Count) and _is_scalar_expression(tree.this):
         return AggregationType.COUNT, _col_name(tree.this), None, False
 
     # SUM(CASE WHEN col THEN 1 ELSE 0 END) → SUM_BOOLEAN
@@ -71,21 +78,21 @@ def _extract_agg_info(expression: str) -> Optional[Tuple[AggregationType, str, O
             and default.name == "0"
             and isinstance(ifs[0].args.get("true"), exp.Literal)
             and ifs[0].args["true"].name == "1"
+            and _is_scalar_expression(case)
         ):
             return AggregationType.SUM_BOOLEAN, ifs[0].this.sql(), None, False
-        return None
 
-    # SUM(col)
-    if isinstance(tree, exp.Sum):
+    # SUM(scalar_expr)
+    if isinstance(tree, exp.Sum) and _is_scalar_expression(tree.this):
         return AggregationType.SUM, _col_name(tree.this), None, False
 
-    if isinstance(tree, exp.Avg):
+    if isinstance(tree, exp.Avg) and _is_scalar_expression(tree.this):
         return AggregationType.AVERAGE, _col_name(tree.this), None, False
 
-    if isinstance(tree, exp.Min):
+    if isinstance(tree, exp.Min) and _is_scalar_expression(tree.this):
         return AggregationType.MIN, _col_name(tree.this), None, False
 
-    if isinstance(tree, exp.Max):
+    if isinstance(tree, exp.Max) and _is_scalar_expression(tree.this):
         return AggregationType.MAX, _col_name(tree.this), None, False
 
     # PERCENTILE_CONT(p) WITHIN GROUP (ORDER BY col)
@@ -97,6 +104,7 @@ def _extract_agg_info(expression: str) -> Optional[Tuple[AggregationType, str, O
             isinstance(inner, (exp.PercentileCont, exp.PercentileDisc))
             and isinstance(order, exp.Order)
             and order.expressions
+            and _is_scalar_expression(order.expressions[0])
         ):
             ordered = order.expressions[0]
             col_node = ordered.this if isinstance(ordered, exp.Ordered) else ordered

converters/dbt/src/ossie_dbt/ossie_to_msi.py

@@ -16,17 +16,18 @@
 # under the License.
 
 from dataclasses import dataclass
-from typing import List, Optional, Set
+from typing import List, Optional, Set, Tuple
 
 from ossie import (
     OssieDataset,
     OssieDialect,
+    OssieDialectExpression,
     OssieDocument,
     OssieExpression,
     OssieField,
     OssieSemanticModel,
 )
-from ossie_dbt.converter_issues import ConverterResult
+from ossie_dbt.converter_issues import ConverterIssue, ConverterIssueType, ConverterResult
 from ossie_dbt.expression_utils import (
     _extract_agg_info,
     _get_dataset_qualifier,
@@ -59,7 +60,6 @@
     PydanticSemanticModel,
 )
 from metricflow_semantic_interfaces.type_enums import (
-    AggregationType,
     DimensionType,
     EntityType,
     MetricType,
@@ -91,27 +91,27 @@ class OssieToMSIConverter:
       - single-agg patterns (`SUM(col)`, `COUNT(DISTINCT col)`, …) → SIMPLE
         metric with `metric_aggregation_params` (no measure reference needed)
       - `(expr_a) / (expr_b)` → RATIO (with auto-generated sub-metrics)
-      - anything else → SIMPLE with the raw expression stored in `expr`
+      - expressions that cannot be represented without changing their
+        aggregation semantics are dropped with a ConverterIssue
     """
 
     def __init__(self, dialect: OssieDialect = OssieDialect.ANSI_SQL) -> None:
         self._dialect = dialect
 
     def convert(self, document: OssieDocument) -> ConverterResult[PydanticSemanticManifest]:
         semantic_models: List[PydanticSemanticModel] = []
-        metrics: List[PydanticMetric] = []
 
         for dataset in document.datasets:
             semantic_models.append(self._convert_dataset(dataset, document))
-        metrics.extend(self._convert_metrics(document))
+        metrics, issues = self._convert_metrics(document)
 
         return ConverterResult(
             output=PydanticSemanticManifest(
                 semantic_models=semantic_models,
                 metrics=metrics,
                 project_configuration=PydanticProjectConfiguration(),
             ),
-            issues=[],
+            issues=issues,
         )
 
     # ------------------------------------------------------------------
@@ -267,20 +267,41 @@ def _classify_field(
     # Metric conversion
     # ------------------------------------------------------------------
 
-    def _convert_metrics(self, ossie_sm: OssieSemanticModel) -> List[PydanticMetric]:
+    def _convert_metrics(
+        self, ossie_sm: OssieSemanticModel
+    ) -> Tuple[List[PydanticMetric], List[ConverterIssue]]:
         metrics: List[PydanticMetric] = []
+        issues: List[ConverterIssue] = []
         for metric in ossie_sm.metrics or []:
-            expr_str = self._get_expression(metric.expression)
-            metrics.extend(self._convert_metric(metric.name, expr_str, metric.description, ossie_sm.datasets))
-        return metrics
+            dialect_expr = self._get_dialect_expression(metric.expression)
+            converted = (
+                self._convert_metric(
+                    metric.name,
+                    dialect_expr.expression,
+                    metric.description,
+                    ossie_sm.datasets,
+                )
+                if dialect_expr is not None and dialect_expr.dialect in self._SQL_DIALECTS
+                else None
+            )
+            if converted is None:
+                issues.append(
+                    ConverterIssue(
+                        issue_type=ConverterIssueType.UNSUPPORTED_METRIC_EXPRESSION,
+                        element_name=metric.name,
+                    )
+                )
+            else:
+                metrics.extend(converted)
+        return metrics, issues
 
     def _convert_metric(
         self,
         name: str,
         expr_str: str,
         description: Optional[str],
         datasets: List[OssieDataset],
-    ) -> List[PydanticMetric]:
+    ) -> Optional[List[PydanticMetric]]:
         """Return one or more PydanticMetric objects for the given Ossie expression.
 
         Simple metrics use `metric_aggregation_params` to store aggregation type
@@ -331,6 +352,8 @@ def _convert_metric(
             den_name = f"{name}__denominator"
             num_metrics = self._convert_metric(num_name, num_expr, None, datasets)
             den_metrics = self._convert_metric(den_name, den_expr, None, datasets)
+            if num_metrics is None or den_metrics is None:
+                return None
             ratio_metric = PydanticMetric(
                 name=name,
                 description=description,
@@ -345,30 +368,7 @@ def _convert_metric(
             )
             return [*num_metrics, *den_metrics, ratio_metric]
 
-        # --- Fallback: complex expression that can't be decomposed ---
-        # Store the raw expression in `expr` with a best-guess aggregation type.
-        # The caller is responsible for reviewing and correcting these metrics.
-        fallback_dataset = datasets[0].name if datasets else ""
-        return [
-            PydanticMetric(
-                name=name,
-                description=description,
-                type=MetricType.SIMPLE,
-                type_params=PydanticMetricTypeParams(
-                    expr=expr_str,
-                    metric_aggregation_params=PydanticMetricAggregationParams(
-                        semantic_model=fallback_dataset,
-                        agg=AggregationType.SUM,
-                        agg_params=None,
-                        agg_time_dimension=None,
-                        non_additive_dimension=None,
-                    ),
-                ),
-                filter=None,
-                metadata=None,
-                config=None,
-            )
-        ]
+        return None
 
     # ------------------------------------------------------------------
     # Helpers
@@ -406,10 +406,24 @@ def _find_dataset_for_col(
 
     def _get_expression(self, ossie_expr: OssieExpression) -> str:
         """Return the expression string for the preferred dialect (fallback: first available)."""
+        dialect_expr = self._get_dialect_expression(ossie_expr)
+        return dialect_expr.expression if dialect_expr is not None else ""
+
+    def _get_dialect_expression(
+        self, ossie_expr: OssieExpression
+    ) -> Optional[OssieDialectExpression]:
+        """Return the expression for the preferred dialect (fallback: first available)."""
         for dialect_expr in ossie_expr.dialects:
             if dialect_expr.dialect is self._dialect:
-                return dialect_expr.expression
-        return ossie_expr.dialects[0].expression if ossie_expr.dialects else ""
+                return dialect_expr
+        return ossie_expr.dialects[0] if ossie_expr.dialects else None
+
+    _SQL_DIALECTS = {
+        OssieDialect.ANSI_SQL,
+        OssieDialect.BIGQUERY,
+        OssieDialect.DATABRICKS,
+        OssieDialect.SNOWFLAKE,
+    }
 
     @staticmethod
     def _parse_source(source: str) -> PydanticNodeRelation:

converters/dbt/tests/helpers.py

@@ -194,8 +194,17 @@ def _ossie_dataset(
     )
 
 
-def _ossie_metric(name: str, expression: str, description: str | None = None) -> OssieMetric:
-    return OssieMetric(name=name, expression=_ossie_expr(expression), description=description)
+def _ossie_metric(
+    name: str,
+    expression: str,
+    description: str | None = None,
+    dialect: OssieDialect = OssieDialect.ANSI_SQL,
+) -> OssieMetric:
+    return OssieMetric(
+        name=name,
+        expression=_ossie_expr(expression, dialect=dialect),
+        description=description,
+    )
 
 
 def _ossie_relationship(

converters/dbt/tests/test_cli.py

@@ -59,3 +59,26 @@ def test_ossie_to_msi_writes_valid_manifest_json(tmp_path: Path, monkeypatch: py
     manifest = json.loads(output_path.read_text())
     assert "semantic_models" in manifest
     assert "metrics" in manifest
+
+
+def test_ossie_to_msi_reports_unsupported_metric(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    document = _ossie_doc(
+        datasets=[_ossie_dataset(name="orders")],
+        metrics=[_ossie_metric("gross_margin", "SUM(revenue) - SUM(cost)")],
+    )
+    input_path = tmp_path / "model.yaml"
+    output_path = tmp_path / "semantic_manifest.json"
+    input_path.write_text(document.to_ossie_yaml())
+
+    _run_cli(["ossie-to-msi", "-i", str(input_path), "-o", str(output_path)], monkeypatch)
+
+    manifest = json.loads(output_path.read_text())
+    assert manifest["metrics"] == []
+    assert (
+        "[WARNING] UNSUPPORTED_METRIC_EXPRESSION: gross_margin was dropped during conversion"
+        in capsys.readouterr().err
+    )