dbt-labs · courtneyholcomb · Jun 20, 2024 · Jun 18, 2024 · Jun 18, 2024 · Jun 19, 2024
diff --git a/.changes/unreleased/Under the Hood-20240618-161241.yaml b/.changes/unreleased/Under the Hood-20240618-161241.yaml
@@ -0,0 +1,7 @@
+kind: Under the Hood
+body: Consume cumulative-specific metric type params from new cumulative_type_params
+  field.
+time: 2024-06-18T16:12:41.132445-07:00
+custom:
+  Author: courtneyholcomb
+  Issue: "1293"
diff --git a/extra-hatch-configuration/requirements.txt b/extra-hatch-configuration/requirements.txt
@@ -1,5 +1,5 @@
 Jinja2>=3.1.3
-dbt-semantic-interfaces==0.6.0.dev2
+dbt-semantic-interfaces==0.6.1
 more-itertools>=8.10.0, <10.2.0
 pydantic>=1.10.0, <1.11.0
 tabulate>=0.8.9

diff --git a/metricflow-semantics/extra-hatch-configuration/requirements.txt b/metricflow-semantics/extra-hatch-configuration/requirements.txt
@@ -1,5 +1,5 @@
 # dbt Cloud depends on metricflow-semantics (dependency set in dbt-mantle), so DSI must always point to a production version here.
-dbt-semantic-interfaces>=0.5.0, <2.0.0
+dbt-semantic-interfaces>=0.6.1, <2.0.0 # Temp: pin to dev release to support cumulative_type_params in MF before they get released in prod.
 graphviz>=0.18.2, <0.21
 python-dateutil>=2.9.0, <2.10.0
 rapidfuzz>=3.0, <4.0
diff --git a/metricflow-semantics/metricflow_semantics/model/dbt_manifest_parser.py b/metricflow-semantics/metricflow_semantics/model/dbt_manifest_parser.py
@@ -10,6 +10,7 @@
 from dbt_semantic_interfaces.transformations.convert_median import (
     ConvertMedianToPercentileRule,
 )
+from dbt_semantic_interfaces.transformations.cumulative_type_params import SetCumulativeTypeParamsRule
 from dbt_semantic_interfaces.transformations.names import LowerCaseNamesRule
 from dbt_semantic_interfaces.transformations.proxy_measure import CreateProxyMeasureRule
 from dbt_semantic_interfaces.transformations.semantic_manifest_transformer import (
@@ -25,6 +26,7 @@ def parse_manifest_from_dbt_generated_manifest(manifest_json_string: str) -> Pyd
     # The serialized object in the dbt project does not have all transformations applied to it at
     # this time, which causes failures with input measure resolution.
     # TODO: remove this transform call once the upstream changes are integrated into our dependency tree
+    # TODO: align rules between DSI, here, and MFS (if possible!)
     rules = (
         # Primary
         (LowerCaseNamesRule(),),
@@ -35,6 +37,7 @@ def parse_manifest_from_dbt_generated_manifest(manifest_json_string: str) -> Pyd
             ConvertCountToSumRule(),
             ConvertMedianToPercentileRule(),
             DedupeMetricInputMeasuresRule(),  # Remove once fix is in core
+            SetCumulativeTypeParamsRule(),
         ),
     )
     model = PydanticSemanticManifestTransformer.transform(raw_model, rules)

diff --git a/metricflow-semantics/metricflow_semantics/model/semantics/linkable_spec_resolver.py b/metricflow-semantics/metricflow_semantics/model/semantics/linkable_spec_resolver.py
@@ -246,7 +246,10 @@ def _metric_requires_metric_time(self, metric: Metric) -> bool:
         metrics_to_check = [metric]
         while metrics_to_check:
             metric_to_check = metrics_to_check.pop()
-            if metric_to_check.type_params.window is not None or metric_to_check.type_params.grain_to_date is not None:
+            if metric_to_check.type_params.cumulative_type_params and (
+                metric_to_check.type_params.cumulative_type_params.window is not None
+                or metric_to_check.type_params.cumulative_type_params.grain_to_date is not None
+            ):
                 return True
             for input_metric in metric_to_check.input_metrics:
                 if input_metric.offset_window is not None or input_metric.offset_to_grain is not None:
@@ -497,9 +500,11 @@ def _get_metric_time_elements(self, measure_reference: Optional[MeasureReference
                         dimension_type=DimensionType.TIME,
                         entity_links=(),
                         join_path=SemanticModelJoinPath(
-                            left_semantic_model_reference=measure_semantic_model.reference
-                            if measure_semantic_model
-                            else SemanticModelDerivation.VIRTUAL_SEMANTIC_MODEL_REFERENCE,
+                            left_semantic_model_reference=(
+                                measure_semantic_model.reference
+                                if measure_semantic_model
+                                else SemanticModelDerivation.VIRTUAL_SEMANTIC_MODEL_REFERENCE
+                            ),
                         ),
                         # Anything that's not at the base time granularity of the measure's aggregation time dimension
                         # should be considered derived.

diff --git a/metricflow-semantics/metricflow_semantics/query/validation_rules/metric_time_requirements.py b/metricflow-semantics/metricflow_semantics/query/validation_rules/metric_time_requirements.py
@@ -79,7 +79,11 @@ def validate_metric_in_resolution_dag(
         elif metric.type is MetricType.CUMULATIVE:
             if (
                 metric.type_params is not None
-                and (metric.type_params.window is not None or metric.type_params.grain_to_date is not None)
+                and metric.type_params.cumulative_type_params is not None
+                and (
+                    metric.type_params.cumulative_type_params.window is not None
+                    or metric.type_params.cumulative_type_params.grain_to_date is not None
+                )
                 and not query_includes_metric_time_or_agg_time_dimension
             ):
                 return MetricFlowQueryResolutionIssueSet.from_issue(

diff --git a/...semantics/test_helpers/semantic_manifest_yamls/ambiguous_resolution_manifest/metrics.yaml b/...semantics/test_helpers/semantic_manifest_yamls/ambiguous_resolution_manifest/metrics.yaml
@@ -70,7 +70,8 @@ metric:
   type: cumulative
   type_params:
     measure: monthly_measure_0
-    window: 2 months
+    cumulative_type_params:
+      window: 2 months
 ---
 metric:
   name: derived_metric_with_common_filtered_metric_0

diff --git a/...t_helpers/semantic_manifest_yamls/extended_date_manifest/metrics/bookings_cumulative.yaml b/...t_helpers/semantic_manifest_yamls/extended_date_manifest/metrics/bookings_cumulative.yaml
@@ -4,4 +4,5 @@ metric:
   type: cumulative
   type_params:
     measure: bookings
-    window: 7 days
+    cumulative_type_params:
+      window: 7 days
diff --git a/...s/semantic_manifest_yamls/extended_date_manifest/metrics/bookings_monthly_cumulative.yaml b/...s/semantic_manifest_yamls/extended_date_manifest/metrics/bookings_monthly_cumulative.yaml
@@ -4,4 +4,5 @@ metric:
   type: cumulative
   type_params:
     measure: bookings_monthly
-    window: 3 month
+    cumulative_type_params:
+      window: 3 month
diff --git a/...cs/metricflow_semantics/test_helpers/semantic_manifest_yamls/simple_manifest/metrics.yaml b/...cs/metricflow_semantics/test_helpers/semantic_manifest_yamls/simple_manifest/metrics.yaml
@@ -135,14 +135,18 @@ metric:
   type: cumulative
   type_params:
     measure: txn_revenue
-    window: 2 month
+    cumulative_type_params:
+      window: 2 month
+      period_agg: average
 ---
 metric:
   name: "revenue_all_time"
   description: "revenue_all_time"
   type: cumulative
   type_params:
     measure: txn_revenue
+    cumulative_type_params:
+      period_agg: last
 ---
 metric:
   name: "every_two_days_bookers"
@@ -158,7 +162,8 @@ metric:
   type: cumulative
   type_params:
     measure: txn_revenue
-    grain_to_date: month
+    cumulative_type_params:
+      grain_to_date: month
 ---
 metric:
   name: booking_fees
@@ -629,7 +634,8 @@ metric:
       name: bookers
       join_to_timespine: true
       fill_nulls_with: 0
-    window: 2 days
+    cumulative_type_params:
+      window: 2 days
 ---
 metric:
   name: bookings_growth_2_weeks_fill_nulls_with_0

diff --git a/metricflow-semantics/tests_metricflow_semantics/query/test_query_parser.py b/metricflow-semantics/tests_metricflow_semantics/query/test_query_parser.py
@@ -130,7 +130,8 @@
       type: cumulative
       type_params:
         measure: revenue
-        window: 7 days
+        cumulative_type_params:
+          window: 7 days
     ---
     metric:
       name: revenue_sub_10

diff --git a/metricflow/dataflow/builder/dataflow_plan_builder.py b/metricflow/dataflow/builder/dataflow_plan_builder.py
@@ -495,8 +495,16 @@ def _build_base_metric_output_node(
             child_metric_offset_to_grain=metric_spec.offset_to_grain,
             cumulative_description=(
                 CumulativeMeasureDescription(
-                    cumulative_window=metric.type_params.window,
-                    cumulative_grain_to_date=metric.type_params.grain_to_date,
+                    cumulative_window=(
+                        metric.type_params.cumulative_type_params.window
+                        if metric.type_params.cumulative_type_params
+                        else None
+                    ),
+                    cumulative_grain_to_date=(
+                        metric.type_params.cumulative_type_params.grain_to_date
+                        if metric.type_params.cumulative_type_params
+                        else None
+                    ),
                 )
                 if metric.type is MetricType.CUMULATIVE
                 else None

diff --git a/metricflow/plan_conversion/dataflow_to_sql.py b/metricflow/plan_conversion/dataflow_to_sql.py
@@ -10,6 +10,7 @@
 from dbt_semantic_interfaces.references import EntityReference, MetricModelReference
 from dbt_semantic_interfaces.type_enums.aggregation_type import AggregationType
 from dbt_semantic_interfaces.type_enums.conversion_calculation_type import ConversionCalculationType
+from dbt_semantic_interfaces.type_enums.period_agg import PeriodAggregation
 from dbt_semantic_interfaces.validations.unique_valid_name import MetricFlowReservedKeywords
 from metricflow_semantics.aggregation_properties import AggregationState
 from metricflow_semantics.dag.id_prefix import StaticIdPrefix
@@ -1635,13 +1636,14 @@ def visit_window_reaggregation_node(self, node: WindowReaggregationNode) -> SqlD
             f"specs: {expected_specs}. Got: {parent_instance_set.as_tuple}."
         )
 
-        # Pending DSI upgrade:
-        # sql_window_function = SqlWindowFunction[
-        #     self._metric_lookup.get_metric(
-        #         metric_instance.spec.reference
-        #     ).type_params.cumulative_type_params.period_agg.name
-        # ]
-        sql_window_function = SqlWindowFunction.FIRST_VALUE  # placeholder for now
+        cumulative_type_params = self._metric_lookup.get_metric(
+            metric_instance.spec.reference
+        ).type_params.cumulative_type_params
+        sql_window_function = SqlWindowFunction.get_window_function_for_period_agg(
+            cumulative_type_params.period_agg
+            if cumulative_type_params and cumulative_type_params.period_agg
+            else PeriodAggregation.FIRST
+        )
         order_by_args = []
         if sql_window_function.requires_ordering:
             order_by_args.append(

diff --git a/metricflow/sql/sql_exprs.py b/metricflow/sql/sql_exprs.py
@@ -13,6 +13,7 @@
 from dbt_semantic_interfaces.protocols.measure import MeasureAggregationParameters
 from dbt_semantic_interfaces.type_enums.aggregation_type import AggregationType
 from dbt_semantic_interfaces.type_enums.date_part import DatePart
+from dbt_semantic_interfaces.type_enums.period_agg import PeriodAggregation
 from dbt_semantic_interfaces.type_enums.time_granularity import TimeGranularity
 from metricflow_semantics.dag.id_prefix import IdPrefix, StaticIdPrefix
 from metricflow_semantics.dag.mf_dag import DagNode, DisplayedProperty, NodeId
@@ -967,6 +968,18 @@ def requires_ordering(self) -> bool:
         else:
             assert_values_exhausted(self)
 
+    @classmethod
+    def get_window_function_for_period_agg(cls, period_agg: PeriodAggregation) -> SqlWindowFunction:
+        """Get the window function to use for given period agg option."""
+        if period_agg is PeriodAggregation.FIRST:
+            return cls.FIRST_VALUE
+        elif period_agg is PeriodAggregation.LAST:
+            return cls.LAST_VALUE
+        elif period_agg is PeriodAggregation.AVERAGE:
+            return cls.AVERAGE
+        else:
+            assert_values_exhausted(period_agg)
+
 
 @dataclass(frozen=True)
 class SqlWindowOrderByArgument:

diff --git a/tests_metricflow/integration/test_cases/itest_cumulative_metric.yaml b/tests_metricflow/integration/test_cases/itest_cumulative_metric.yaml
@@ -428,7 +428,7 @@ integration_test:
       SELECT
         metric_time__week
         , metric_time__quarter
-        , FIRST_VALUE(revenue_all_time) OVER (
+        , LAST_VALUE(revenue_all_time) OVER (
           PARTITION BY metric_time__week, metric_time__quarter
           ORDER BY metric_time__day
           ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
@@ -467,7 +467,7 @@ integration_test:
       SELECT
         revenue_instance__ds__month
         , metric_time__week
-        , FIRST_VALUE(trailing_2_months_revenue) OVER (
+        , AVG(trailing_2_months_revenue) OVER (
           PARTITION BY revenue_instance__ds__month, metric_time__week
           ORDER BY metric_time__day
           ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
@@ -562,7 +562,7 @@ integration_test:
         SELECT
           metric_time__month
           , metric_time__year
-          , FIRST_VALUE(t2mr) OVER (
+          , AVG(t2mr) OVER (
             PARTITION BY metric_time__month, metric_time__year
             ORDER BY metric_time__day
             ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING

diff --git a/tests_metricflow/query_rendering/test_cumulative_metric_rendering.py b/tests_metricflow/query_rendering/test_cumulative_metric_rendering.py
@@ -623,7 +623,6 @@ def test_derived_cumulative_metric_with_non_default_grains(
 
 
 # TODO: write the following tests when unblocked
-# - Render each of the allowed period_aggs (both set in YAML & default)
 # - Query cumulative metric with non-day default_grain (using default grain and non-default grain)
 # - Query 2 metrics with different default_grains using metric_time (no grain specified)
 # - If default grain is WEEK, query with a higher grain (check that we still get correct values)

diff --git a/...endering.py/SqlQueryPlan/BigQuery/test_all_time_metric_with_non_default_grains__plan0.sql b/...endering.py/SqlQueryPlan/BigQuery/test_all_time_metric_with_non_default_grains__plan0.sql
@@ -8,7 +8,7 @@ FROM (
   SELECT
     subq_7.metric_time__week
     , subq_7.metric_time__quarter
-    , FIRST_VALUE(subq_7.revenue_all_time) OVER (
+    , LAST_VALUE(subq_7.revenue_all_time) OVER (
       PARTITION BY
         subq_7.metric_time__week
         , subq_7.metric_time__quarter

diff --git a/...y/SqlQueryPlan/BigQuery/test_all_time_metric_with_non_default_grains__plan0_optimized.sql b/...y/SqlQueryPlan/BigQuery/test_all_time_metric_with_non_default_grains__plan0_optimized.sql
@@ -8,7 +8,7 @@ FROM (
   SELECT
     metric_time__week
     , metric_time__quarter
-    , FIRST_VALUE(revenue_all_time) OVER (
+    , LAST_VALUE(revenue_all_time) OVER (
       PARTITION BY
         metric_time__week
         , metric_time__quarter

diff --git a/...ndering.py/SqlQueryPlan/BigQuery/test_cumulative_metric_with_non_default_grain__plan0.sql b/...ndering.py/SqlQueryPlan/BigQuery/test_cumulative_metric_with_non_default_grain__plan0.sql
@@ -6,7 +6,7 @@ FROM (
   -- Window Function for Metric Re-aggregation
   SELECT
     subq_7.metric_time__week
-    , FIRST_VALUE(subq_7.revenue_all_time) OVER (
+    , LAST_VALUE(subq_7.revenue_all_time) OVER (
       PARTITION BY subq_7.metric_time__week
       ORDER BY subq_7.metric_time__day
       ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING

diff --git a/.../SqlQueryPlan/BigQuery/test_cumulative_metric_with_non_default_grain__plan0_optimized.sql b/.../SqlQueryPlan/BigQuery/test_cumulative_metric_with_non_default_grain__plan0_optimized.sql
@@ -6,7 +6,7 @@ FROM (
   -- Window Function for Metric Re-aggregation
   SELECT
     metric_time__week
-    , FIRST_VALUE(revenue_all_time) OVER (
+    , LAST_VALUE(revenue_all_time) OVER (
       PARTITION BY metric_time__week
       ORDER BY metric_time__day
       ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING

diff --git a/...y/SqlQueryPlan/BigQuery/test_derived_cumulative_metric_with_non_default_grains__plan0.sql b/...y/SqlQueryPlan/BigQuery/test_derived_cumulative_metric_with_non_default_grains__plan0.sql
@@ -11,11 +11,7 @@ FROM (
     -- Window Function for Metric Re-aggregation
     SELECT
       subq_7.metric_time__week
-      , FIRST_VALUE(subq_7.t2mr) OVER (
-        PARTITION BY subq_7.metric_time__week
-        ORDER BY subq_7.metric_time__day
-        ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
-      ) AS t2mr
+      , AVG(subq_7.t2mr) OVER (PARTITION BY subq_7.metric_time__week) AS t2mr
     FROM (
       -- Compute Metrics via Expressions
       SELECT

diff --git a/...Plan/BigQuery/test_derived_cumulative_metric_with_non_default_grains__plan0_optimized.sql b/...Plan/BigQuery/test_derived_cumulative_metric_with_non_default_grains__plan0_optimized.sql
@@ -11,19 +11,14 @@ FROM (
     -- Window Function for Metric Re-aggregation
     SELECT
       metric_time__week
-      , FIRST_VALUE(t2mr) OVER (
-        PARTITION BY metric_time__week
-        ORDER BY metric_time__day
-        ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
-      ) AS t2mr
+      , AVG(t2mr) OVER (PARTITION BY metric_time__week) AS t2mr
     FROM (
       -- Join Self Over Time Range
       -- Pass Only Elements: ['txn_revenue', 'metric_time__week', 'metric_time__day']
       -- Aggregate Measures
       -- Compute Metrics via Expressions
       SELECT
-        subq_12.metric_time__day AS metric_time__day
-        , subq_12.metric_time__week AS metric_time__week
+        subq_12.metric_time__week AS metric_time__week
         , SUM(revenue_src_28000.revenue) AS t2mr
       FROM (
         -- Time Spine

diff --git a/...c_rendering.py/SqlQueryPlan/BigQuery/test_window_metric_with_non_default_grain__plan0.sql b/...c_rendering.py/SqlQueryPlan/BigQuery/test_window_metric_with_non_default_grain__plan0.sql
@@ -6,11 +6,7 @@ FROM (
   -- Window Function for Metric Re-aggregation
   SELECT
     subq_7.metric_time__year
-    , FIRST_VALUE(subq_7.trailing_2_months_revenue) OVER (
-      PARTITION BY subq_7.metric_time__year
-      ORDER BY subq_7.metric_time__day
-      ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
-    ) AS trailing_2_months_revenue
+    , AVG(subq_7.trailing_2_months_revenue) OVER (PARTITION BY subq_7.metric_time__year) AS trailing_2_months_revenue
   FROM (
     -- Compute Metrics via Expressions
     SELECT

diff --git a/...g.py/SqlQueryPlan/BigQuery/test_window_metric_with_non_default_grain__plan0_optimized.sql b/...g.py/SqlQueryPlan/BigQuery/test_window_metric_with_non_default_grain__plan0_optimized.sql
@@ -6,19 +6,14 @@ FROM (
   -- Window Function for Metric Re-aggregation
   SELECT
     metric_time__year
-    , FIRST_VALUE(trailing_2_months_revenue) OVER (
-      PARTITION BY metric_time__year
-      ORDER BY metric_time__day
-      ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
-    ) AS trailing_2_months_revenue
+    , AVG(trailing_2_months_revenue) OVER (PARTITION BY metric_time__year) AS trailing_2_months_revenue
   FROM (
     -- Join Self Over Time Range
     -- Pass Only Elements: ['txn_revenue', 'metric_time__year', 'metric_time__day']
     -- Aggregate Measures
     -- Compute Metrics via Expressions
     SELECT
-      subq_11.metric_time__day AS metric_time__day
-      , subq_11.metric_time__year AS metric_time__year
+      subq_11.metric_time__year AS metric_time__year
       , SUM(revenue_src_28000.revenue) AS trailing_2_months_revenue
     FROM (
       -- Time Spine

diff --git a/...dering.py/SqlQueryPlan/Databricks/test_all_time_metric_with_non_default_grains__plan0.sql b/...dering.py/SqlQueryPlan/Databricks/test_all_time_metric_with_non_default_grains__plan0.sql
@@ -8,7 +8,7 @@ FROM (
   SELECT
     subq_7.metric_time__week
     , subq_7.metric_time__quarter
-    , FIRST_VALUE(subq_7.revenue_all_time) OVER (
+    , LAST_VALUE(subq_7.revenue_all_time) OVER (
       PARTITION BY
         subq_7.metric_time__week
         , subq_7.metric_time__quarter

diff --git a/...SqlQueryPlan/Databricks/test_all_time_metric_with_non_default_grains__plan0_optimized.sql b/...SqlQueryPlan/Databricks/test_all_time_metric_with_non_default_grains__plan0_optimized.sql
@@ -8,7 +8,7 @@ FROM (
   SELECT
     metric_time__week
     , metric_time__quarter
-    , FIRST_VALUE(revenue_all_time) OVER (
+    , LAST_VALUE(revenue_all_time) OVER (
       PARTITION BY
         metric_time__week
         , metric_time__quarter