fix(ingestion): handle dbt 1.12+ inline metric spec (type_params.expr fallback)

ingestion/src/metadata/ingestion/source/database/dbt/metadata.py

@@ -1417,13 +1417,33 @@
             related_metrics = self._filter_known_metrics(related_metrics, all_metrics)
         return metric_expression, related_metrics
 
+    @staticmethod
+    def _metric_aggregation(type_params: Any) -> str | None:
+        """Return the aggregation function name from the dbt 1.12+ inline spec.
+
+        In the measure-less spec the aggregation lives on
+        ``type_params.metric_aggregation_params.agg``. The manifest parser keeps that
+        block as an extra attribute, so it may arrive as a dict or an object. Returns
+        ``None`` for the pre-1.12 spec, where the measure itself carries the aggregation.
+        """
+        params = getattr(type_params, "metric_aggregation_params", None)
+        if params is None:
+            return None
+        agg = params.get("agg") if isinstance(params, dict) else getattr(params, "agg", None)
+        return getattr(agg, "value", agg) if agg else None
+
     @staticmethod
     def _simple_metric_expression(type_params):
         expression = None
         measure_ref = getattr(type_params, "measure", None)
         measure_name = getattr(measure_ref, "name", None) if measure_ref else None
         if measure_name:
             expression = MetricExpression(language=Language.SQL, code=measure_name)
+        else:
+            # dbt 1.12+ inline spec: measure ref is absent; use type_params.expr directly
+            expr = getattr(type_params, "expr", None)
+            if expr:
+                expression = MetricExpression(language=Language.SQL, code=expr)
         return expression, None
 
     @staticmethod
@@ -1465,6 +1485,15 @@
         expression = None
         measure_ref = getattr(type_params, "measure", None)
         measure_name = getattr(measure_ref, "name", None) if measure_ref else None
+        if not measure_name:
+            # dbt 1.12+ inline spec: the measure is gone; a cumulative metric now wraps
+            # another metric referenced on cumulative_type_params.metric.
+            cum_params = getattr(type_params, "cumulative_type_params", None)
+            metric_ref = getattr(cum_params, "metric", None) if cum_params else None
+            if isinstance(metric_ref, dict):
+                measure_name = metric_ref.get("name")
+            else:
+                measure_name = getattr(metric_ref, "name", None) if metric_ref else None
         if measure_name:
             window_str = DbtSource._cumulative_window_str(type_params)
             expression = MetricExpression(language=Language.SQL, code=f"cumulative({measure_name}{window_str})")
@@ -1515,7 +1544,7 @@
                 )
         return result
 
     def _extract_measures(self, metric_node, semantic_models, all_metrics=None) -> list[MetricMeasure]:
         result = []
         models = (
             find_semantic_models_transitive(metric_node, semantic_models, all_metrics)
@@ -1537,6 +1566,23 @@
                         expression=getattr(measure, "expr", None),
                     )
                 )
+        # dbt 1.12+ inline spec: semantic model measures list is empty because the aggregation
+        # is defined directly on the metric. Fall back to type_params.expr on the metric node
+        # and carry the aggregation from metric_aggregation_params so the measure matches the
+        # metadata the pre-1.12 spec produced.
+        if not result:
+            type_params = getattr(metric_node, "type_params", None)
+            expr = getattr(type_params, "expr", None) if type_params else None
+            aggregation = self._metric_aggregation(type_params) if type_params else None
+            if expr or aggregation:
+                result.append(
+                    MetricMeasure(
+                        name=getattr(metric_node, "name", ""),
+                        aggregation=aggregation,
+                        description=None,
+                        expression=expr,
+                    )
+                )
         return result
 
     def _extract_filters(self, metric_node) -> list[MetricFilter]:

ingestion/tests/unit/test_dbt.py

@@ -4648,3 +4648,309 @@ def test_source_fqn_follows_the_entity_found_across_services(self):
             source.metadata.add_test_case_results.call_args.kwargs["test_case_fqn"]
             == "actual_svc.RAW_DB.RAW.orders.orders_freshness"
         )
+
+
+class TestDbtV12MetricIngest(TestCase):
+    """Unit tests for dbt 1.12+ inline metric spec (measure-less semantic models)."""
+
+    def setUp(self):
+        from metadata.generated.schema.entity.data.metric import (
+            Language,
+            MetricExpression,
+            MetricMeasure,
+        )
+
+        self.Language = Language
+        self.MetricExpression = MetricExpression
+        self.MetricMeasure = MetricMeasure
+
+    # ------------------------------------------------------------------
+    # _simple_metric_expression
+    # ------------------------------------------------------------------
+
+    def test_simple_metric_expression_old_spec_returns_measure_name(self):
+        """Pre-1.12 path: type_params.measure.name is used as the expression."""
+        measure_ref = SimpleNamespace(name="revenue")
+        type_params = SimpleNamespace(measure=measure_ref, expr=None)
+
+        expr, related = DbtSource._simple_metric_expression(type_params)
+
+        assert related is None
+        assert expr is not None
+        assert expr.code == "revenue"
+
+    def test_simple_metric_expression_dbt_v12_falls_back_to_expr(self):
+        """dbt 1.12+: type_params.measure is None; type_params.expr is the column expression."""
+        type_params = SimpleNamespace(measure=None, expr="user_id")
+
+        expr, related = DbtSource._simple_metric_expression(type_params)
+
+        assert related is None
+        assert expr is not None
+        assert expr.code == "user_id"
+        assert expr.language == self.Language.SQL
+
+    def test_simple_metric_expression_dbt_v12_no_measure_no_expr_returns_none(self):
+        """Both measure and expr absent → no expression emitted (no crash)."""
+        type_params = SimpleNamespace(measure=None, expr=None)
+
+        expr, related = DbtSource._simple_metric_expression(type_params)
+
+        assert expr is None
+        assert related is None
+
+    def test_simple_metric_expression_measure_without_name_falls_back_to_expr(self):
+        """measure present but .name is None → fall back to expr."""
+        type_params = SimpleNamespace(measure=SimpleNamespace(name=None), expr="amount")
+
+        expr, _ = DbtSource._simple_metric_expression(type_params)
+
+        assert expr is not None
+        assert expr.code == "amount"
+
+    # ------------------------------------------------------------------
+    # _extract_measures
+    # ------------------------------------------------------------------
+
+    def _make_source(self):
+        """Return a minimally wired DbtSource with mocked metadata."""
+        from metadata.ingestion.source.database.dbt.metadata import DbtSource
+
+        metadata = MagicMock()
+        return DbtSource.__new__(DbtSource), metadata
+
+    def test_extract_measures_from_semantic_model(self):
+        """Pre-1.12: measures come from the semantic model."""
+        source, _ = self._make_source()
+
+        agg = SimpleNamespace(value="count_distinct")
+        sm_measure = SimpleNamespace(name="m_users", agg=agg, description="users", expr="user_id")
+        sm = SimpleNamespace(name="sm1", measures=[sm_measure])
+        metric_node = SimpleNamespace(
+            name="distinct_users",
+            type_params=SimpleNamespace(measure=SimpleNamespace(name="m_users"), expr=None),
+            refs=None,
+            metrics=None,
+        )
+
+        with patch(
+            "metadata.ingestion.source.database.dbt.metadata.find_semantic_models_for_metric",
+            return_value=[sm],
+        ):
+            result = source._extract_measures(metric_node, {})
+
+        assert len(result) == 1
+        assert result[0].name == "m_users"
+        assert result[0].aggregation == "count_distinct"
+        assert result[0].expression == "user_id"
+
+    def test_extract_measures_dbt_v12_empty_sm_measures_uses_type_params_expr(self):
+        """dbt 1.12+: semantic model has no measures → synthetic measure from type_params.expr,
+        carrying the aggregation from metric_aggregation_params (as the pre-1.12 measure did)."""
+        source, _ = self._make_source()
+
+        sm = SimpleNamespace(name="sm1", measures=[])
+        metric_node = SimpleNamespace(
+            name="distinct_users",
+            type_params=SimpleNamespace(
+                measure=None,
+                expr="user_id",
+                metric_aggregation_params={"agg": "count_distinct"},
+            ),
+            refs=None,
+            metrics=None,
+        )
+
+        with patch(
+            "metadata.ingestion.source.database.dbt.metadata.find_semantic_models_for_metric",
+            return_value=[sm],
+        ):
+            result = source._extract_measures(metric_node, {})
+
+        assert len(result) == 1
+        assert result[0].name == "distinct_users"
+        assert result[0].expression == "user_id"
+        assert result[0].aggregation == "count_distinct"
+
+    def test_extract_measures_dbt_v12_no_sm_uses_type_params_expr(self):
+        """dbt 1.12+: no semantic models at all → synthetic measure from type_params.expr."""
+        source, _ = self._make_source()
+
+        metric_node = SimpleNamespace(
+            name="active_users",
+            type_params=SimpleNamespace(measure=None, expr="user_id"),
+            refs=None,
+            metrics=None,
+        )
+
+        with patch(
+            "metadata.ingestion.source.database.dbt.metadata.find_semantic_models_for_metric",
+            return_value=[],
+        ):
+            result = source._extract_measures(metric_node, {})
+
+        assert len(result) == 1
+        assert result[0].name == "active_users"
+        assert result[0].expression == "user_id"
+
+    def test_extract_measures_dbt_v12_no_expr_returns_empty(self):
+        """dbt 1.12+: no semantic model measures and no expr → empty list (no crash)."""
+        source, _ = self._make_source()
+
+        sm = SimpleNamespace(name="sm1", measures=[])
+        metric_node = SimpleNamespace(
+            name="my_metric",
+            type_params=SimpleNamespace(measure=None, expr=None),
+            refs=None,
+            metrics=None,
+        )
+
+        with patch(
+            "metadata.ingestion.source.database.dbt.metadata.find_semantic_models_for_metric",
+            return_value=[sm],
+        ):
+            result = source._extract_measures(metric_node, {})
+
+        assert result == []
+
+
+class TestDbtV12MetricAggregationAndCumulative(TestCase):
+    """dbt 1.12+ inline metrics: aggregation carried onto the measure, and cumulative
+    metrics resolved from cumulative_type_params.metric. The pre-1.12 (measure-based)
+    paths must stay byte-for-byte unchanged, so each new behaviour is guarded on the
+    absence of the old field."""
+
+    def _make_source(self):
+        from metadata.ingestion.source.database.dbt.metadata import DbtSource
+
+        return DbtSource.__new__(DbtSource)
+
+    # ------------------------------------------------------------------
+    # _metric_aggregation
+    # ------------------------------------------------------------------
+
+    def test_metric_aggregation_from_dict(self):
+        tp = SimpleNamespace(metric_aggregation_params={"agg": "count_distinct"})
+        assert DbtSource._metric_aggregation(tp) == "count_distinct"
+
+    def test_metric_aggregation_from_object(self):
+        tp = SimpleNamespace(metric_aggregation_params=SimpleNamespace(agg="sum"))
+        assert DbtSource._metric_aggregation(tp) == "sum"
+
+    def test_metric_aggregation_unwraps_enum_value(self):
+        tp = SimpleNamespace(metric_aggregation_params={"agg": SimpleNamespace(value="median")})
+        assert DbtSource._metric_aggregation(tp) == "median"
+
+    def test_metric_aggregation_none_when_absent(self):
+        assert DbtSource._metric_aggregation(SimpleNamespace(metric_aggregation_params=None)) is None
+        assert DbtSource._metric_aggregation(SimpleNamespace()) is None
+
+    # ------------------------------------------------------------------
+    # _extract_measures aggregation (dbt 1.12+)
+    # ------------------------------------------------------------------
+
+    def test_v12_count_star_no_expr_still_creates_measure(self):
+        """agg with no expr (e.g. count(*)) → measure carrying only the aggregation."""
+        source = self._make_source()
+        metric_node = SimpleNamespace(
+            name="row_count",
+            type_params=SimpleNamespace(measure=None, expr=None, metric_aggregation_params={"agg": "count"}),
+            refs=None,
+            metrics=None,
+        )
+        with patch(
+            "metadata.ingestion.source.database.dbt.metadata.find_semantic_models_for_metric",
+            return_value=[],
+        ):
+            result = source._extract_measures(metric_node, {})
+        assert len(result) == 1
+        assert result[0].name == "row_count"
+        assert result[0].aggregation == "count"
+        assert result[0].expression is None
+
+    def test_v12_no_expr_no_agg_returns_empty(self):
+        source = self._make_source()
+        metric_node = SimpleNamespace(
+            name="empty_metric",
+            type_params=SimpleNamespace(measure=None, expr=None, metric_aggregation_params=None),
+            refs=None,
+            metrics=None,
+        )
+        with patch(
+            "metadata.ingestion.source.database.dbt.metadata.find_semantic_models_for_metric",
+            return_value=[],
+        ):
+            assert source._extract_measures(metric_node, {}) == []
+
+    def test_old_spec_measure_aggregation_unchanged(self):
+        """Pre-1.12: aggregation and name still come straight from the semantic-model measure."""
+        source = self._make_source()
+        measure = SimpleNamespace(
+            name="num_distinct",
+            agg=SimpleNamespace(value="count_distinct"),
+            description="distinct customer count",
+            expr="customer_id",
+        )
+        sm = SimpleNamespace(name="sm1", measures=[measure])
+        metric_node = SimpleNamespace(
+            name="distinct_customers",
+            type_params=SimpleNamespace(
+                measure=SimpleNamespace(name="num_distinct"),
+                expr=None,
+                metric_aggregation_params=None,
+            ),
+            refs=None,
+            metrics=None,
+        )
+        with patch(
+            "metadata.ingestion.source.database.dbt.metadata.find_semantic_models_for_metric",
+            return_value=[sm],
+        ):
+            result = source._extract_measures(metric_node, {})
+        assert result[0].name == "num_distinct"
+        assert result[0].aggregation == "count_distinct"
+        assert result[0].expression == "customer_id"
+        assert result[0].description == "distinct customer count"
+
+    # ------------------------------------------------------------------
+    # _cumulative_metric_expression (dbt 1.12+)
+    # ------------------------------------------------------------------
+
+    def _cumulative_type_params(self, metric):
+        return SimpleNamespace(
+            window=SimpleNamespace(count=7, granularity=SimpleNamespace(value="day")),
+            metric=metric,
+        )
+
+    def test_cumulative_v12_wraps_metric_from_dict(self):
+        type_params = SimpleNamespace(
+            measure=None,
+            cumulative_type_params=self._cumulative_type_params({"name": "distinct_customers"}),
+        )
+        expression, related = DbtSource._cumulative_metric_expression(type_params)
+        assert related is None
+        assert expression.code == "cumulative(distinct_customers over 7 day)"
+
+    def test_cumulative_v12_wraps_metric_from_object(self):
+        type_params = SimpleNamespace(
+            measure=None,
+            cumulative_type_params=self._cumulative_type_params(SimpleNamespace(name="distinct_customers")),
+        )
+        expression, _ = DbtSource._cumulative_metric_expression(type_params)
+        assert expression.code == "cumulative(distinct_customers over 7 day)"
+
+    def test_cumulative_old_spec_wraps_measure_unchanged(self):
+        type_params = SimpleNamespace(
+            measure=SimpleNamespace(name="revenue"),
+            cumulative_type_params=self._cumulative_type_params(None),
+        )
+        expression, _ = DbtSource._cumulative_metric_expression(type_params)
+        assert expression.code == "cumulative(revenue over 7 day)"
+
+    def test_cumulative_v12_no_metric_returns_none(self):
+        type_params = SimpleNamespace(
+            measure=None,
+            cumulative_type_params=SimpleNamespace(window=None, metric=None),
+        )
+        expression, _ = DbtSource._cumulative_metric_expression(type_params)
+        assert expression is None