diff --git a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py index 4f34610f148c..a0e41e037a5f 100644 --- a/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py +++ b/ingestion/src/metadata/ingestion/source/database/dbt/metadata.py @@ -1417,6 +1417,21 @@ def _build_expression_and_related(self, dbt_type, type_params, all_metrics): 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 @@ -1424,6 +1439,11 @@ def _simple_metric_expression(type_params): 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 @@ def _cumulative_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 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})") @@ -1537,6 +1566,23 @@ def _extract_measures(self, metric_node, semantic_models, all_metrics=None) -> l 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]: diff --git a/ingestion/tests/unit/test_dbt.py b/ingestion/tests/unit/test_dbt.py index 282fa4c058a1..0e6bfb1ca622 100644 --- a/ingestion/tests/unit/test_dbt.py +++ b/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