Update offset implementation (#1826)

When querying offset metrics, we typically apply the offset join before
aggregation. This is because post-aggregation, the grain is often no
longer accessible (e.g., if we have aggregated to the year, month grain
would no longer be available to offset by). This PR updates the logic to
offset post-aggregation when the queried grain matches the offset grain
(a scenario that ensures the offset grain is still available). This
change ensures that the entire period's values are included when
offsetting.

Author: courtneyholcomb

.changes/unreleased/Fixes-20250903-195252.yaml

@@ -0,0 +1,7 @@
+kind: Fixes
+body: Apply offset join after aggregation when queried grain is available post-aggregation.
+  This ensures results match user expectations.
+time: 2025-09-03T19:52:52.4573-07:00
+custom:
+  Author: courtneyholcomb
+  Issue: "1826,1758"

.cursorrules

@@ -0,0 +1,8 @@
+# MetricFlow Project Rules
+
+This project uses **hatch** for dependency management and testing:
+
+- Dependencies: Use `hatch` commands, not `pip install` directly
+- Tests: Use `hatch run dev-env:pytest <path>` instead of `pytest` directly
+- Snapshots: Add `--overwrite-snapshots` for new snapshot tests
+- Common: `make test` or `hatch run dev-env:pytest tests_metricflow/`

metricflow-semantics/metricflow_semantics/specs/measure_spec.py

@@ -105,3 +105,8 @@ def custom_offset_window(self) -> Optional[MetricTimeWindow]:
         if self.offset_window and not self.offset_window.is_standard_granularity:
             return self.offset_window
         return None
+
+    @property
+    def uses_offset(self) -> bool:
+        """Return True if the measure uses an offset."""
+        return self.offset_window is not None or self.offset_to_grain is not None

metricflow-semantics/metricflow_semantics/test_helpers/semantic_manifest_yamls/simple_manifest/metrics.yaml

@@ -933,3 +933,23 @@ metric:
         offset_window: 1 alien_day
         alias: bookings_offset
       - name: bookings
+---
+metric:
+  name: trailing_7_days_bookings
+  description: "trailing 7 days bookings - cumulative metric with window"
+  type: cumulative
+  type_params:
+    measure: bookings
+    cumulative_type_params:
+      window: 7 days
+---
+metric:
+  name: trailing_7_days_bookings_offset_1_week
+  description: "trailing 7 days bookings offset by 1 week - offset_window metric using cumulative input"
+  type: derived
+  type_params:
+    expr: trailing_7_days_bookings_1_week_ago
+    metrics:
+      - name: trailing_7_days_bookings
+        offset_window: 1 week
+        alias: trailing_7_days_bookings_1_week_ago

metricflow/dataflow/builder/dataflow_plan_builder.py

@@ -1366,31 +1366,67 @@ def _build_input_measure_spec(
             ),
         )
 
+        queried_agg_time_dimension_specs = queried_linkable_specs.included_agg_time_dimension_specs_for_measure(
+            measure_reference=measure_spec.reference, semantic_model_lookup=self._semantic_model_lookup
+        )
+        smallest_queried_agg_time_grain: Optional[ExpandedTimeGranularity] = None
         before_aggregation_time_spine_join_description = None
-        # If querying an offset metric, join to time spine.
+        after_aggregation_time_spine_join_description = None
         if child_metric_offset_window is not None or child_metric_offset_to_grain is not None:
-            before_aggregation_time_spine_join_description = JoinToTimeSpineDescription(
+            if child_metric_offset_window is not None:
+                offset_grain_name = child_metric_offset_window.granularity
+                if ExpandedTimeGranularity.is_standard_granularity_name(offset_grain_name):
+                    offset_grain = ExpandedTimeGranularity.from_time_granularity(TimeGranularity(offset_grain_name))
+                else:
+                    offset_grain = ExpandedTimeGranularity(
+                        name=offset_grain_name,
+                        base_granularity=self._get_base_grain_for_custom_grain(offset_grain_name),
+                    )
+            else:
+                assert (
+                    child_metric_offset_to_grain is not None
+                ), "Offset to grain must be specified if no offset window is specified."
+                offset_grain = ExpandedTimeGranularity.from_time_granularity(child_metric_offset_to_grain)
+
+            # Determine the smallest queried agg time dimension grain (this is the grain we'll aggregate to)
+            if len(queried_agg_time_dimension_specs) == 0:
+                raise ValueError(
+                    "No agg_time_dimension requested in offset metric query. This should have been validated earlier."
+                )
+            smallest_queried_agg_time_grain = self._sort_by_base_granularity(queried_agg_time_dimension_specs)[
+                0
+            ].time_granularity
+
+            # If the smallest queried grain is equal to the offset grain, join after aggregation. Otherwise, the grain
+            # needed for offset will not be available anymore, so join before aggregation.
+            join_to_time_spine_description = JoinToTimeSpineDescription(
                 join_type=SqlJoinType.INNER,
                 offset_window=child_metric_offset_window,
                 offset_to_grain=child_metric_offset_to_grain,
             )
-
-        # Even if the measure is configured to join to time spine, if there's no agg_time_dimension in the query,
-        # there's no need to join to the time spine since all time will be aggregated.
-        after_aggregation_time_spine_join_description = None
-        if input_measure.join_to_timespine:
             if (
-                len(
-                    queried_linkable_specs.included_agg_time_dimension_specs_for_measure(
-                        measure_reference=measure_spec.reference, semantic_model_lookup=self._semantic_model_lookup
-                    )
-                )
-                > 0
+                offset_grain
+                and smallest_queried_agg_time_grain == offset_grain
+                and not offset_grain.is_custom_granularity  # custom offset window has special logic handled later
             ):
+                after_aggregation_time_spine_join_description = join_to_time_spine_description
+            else:
+                before_aggregation_time_spine_join_description = join_to_time_spine_description
+
+        # Measures configured to join to time spine will join to time spine after aggregation using LEFT OUTER JOIN.
+        # If there's no agg_time_dimension in the query, skip time spine join since all time will be aggregated.
+        # If we already need to join to time spine after aggregation due to offset, and the measure is also configured
+        # to join to time spine, update to use LEFT OUTER JOIN.
+        if input_measure.join_to_timespine and (len(queried_agg_time_dimension_specs) > 0):
+            if after_aggregation_time_spine_join_description is not None:
                 after_aggregation_time_spine_join_description = JoinToTimeSpineDescription(
                     join_type=SqlJoinType.LEFT_OUTER,
-                    offset_window=None,
-                    offset_to_grain=None,
+                    offset_window=after_aggregation_time_spine_join_description.offset_window,
+                    offset_to_grain=after_aggregation_time_spine_join_description.offset_to_grain,
+                )
+            else:
+                after_aggregation_time_spine_join_description = JoinToTimeSpineDescription(
+                    join_type=SqlJoinType.LEFT_OUTER, offset_window=None, offset_to_grain=None
                 )
 
         return MetricInputMeasureSpec(
@@ -1544,7 +1580,7 @@ def __get_required_linkable_specs(
 
         return required_linkable_specs
 
-    def _build_time_spine_join_node_for_measure_config(
+    def _build_time_spine_join_node_for_after_aggregation(
         self,
         join_description: JoinToTimeSpineDescription,
         measure_reference: MeasureReference,
@@ -1554,11 +1590,6 @@ def _build_time_spine_join_node_for_measure_config(
         after_aggregation_where_filter_specs: Sequence[WhereFilterSpec],
         time_range_constraint: Optional[TimeRangeConstraint],
     ) -> DataflowPlanNode:
-        assert join_description.join_type is SqlJoinType.LEFT_OUTER, (
-            f"Expected {SqlJoinType.LEFT_OUTER} for joining to time spine after aggregation. Remove this if "
-            f"there's a new use case."
-        )
-
         # Find filters that contain only metric_time or agg_time_dimension. They will be applied to the time spine table.
         agg_time_only_filters: List[WhereFilterSpec] = []
         non_agg_time_filters: List[WhereFilterSpec] = []
@@ -1584,6 +1615,8 @@ def _build_time_spine_join_node_for_measure_config(
             requested_agg_time_dimension_specs=queried_agg_time_dimension_specs,
             join_on_time_dimension_spec=join_spec,
             join_type=join_description.join_type,
+            standard_offset_window=join_description.standard_offset_window,
+            offset_to_grain=join_description.offset_to_grain,
         )
 
         # Since new rows might have been added due to time spine join, re-apply constraints here. Only re-apply filters
@@ -1654,15 +1687,14 @@ def _build_time_spine_join_node_for_nested_offset(
             )
         return output_node
 
-    def _build_time_spine_join_node_for_offset(
+    def _build_time_spine_join_node_for_before_aggregation(
         self,
         join_description: JoinToTimeSpineDescription,
         measure_properties: MeasureSpecProperties,
         queried_agg_time_dimension_specs: Tuple[TimeDimensionSpec, ...],
         metric_source_node: DataflowPlanNode,
         use_offset_custom_granularity_node: bool,
     ) -> DataflowPlanNode:
-        """Build a node to join to the time spine for a measure with the `join_to_timespine: true` YAML config."""
         assert join_description.join_type is SqlJoinType.INNER, (
             f"Expected {SqlJoinType.INNER} for joining to time spine before aggregation. Remove this if there's a "
             f"new use case."
@@ -1761,6 +1793,15 @@ def _build_aggregated_measure_from_measure_source_node(
         before_aggregation_time_spine_join_description = (
             metric_input_measure_spec.before_aggregation_time_spine_join_description
         )
+        after_aggregation_time_spine_join_description = (
+            metric_input_measure_spec.after_aggregation_time_spine_join_description
+        )
+        uses_offset = (
+            before_aggregation_time_spine_join_description
+            and before_aggregation_time_spine_join_description.uses_offset
+        ) or (
+            after_aggregation_time_spine_join_description and after_aggregation_time_spine_join_description.uses_offset
+        )
 
         if measure_recipe is None:
             logger.debug(
@@ -1772,8 +1813,7 @@ def _build_aggregated_measure_from_measure_source_node(
             )
             measure_time_constraint = (
                 (cumulative_metric_adjusted_time_constraint or predicate_pushdown_state.time_range_constraint)
-                # If joining to time spine for time offset, constraints will be applied after that join.
-                if not before_aggregation_time_spine_join_description
+                if not uses_offset  # Time constraints will be applied after offset
                 else None
             )
             if measure_time_constraint is None:
@@ -1805,10 +1845,14 @@ def _build_aggregated_measure_from_measure_source_node(
                 f"Unable to join all items in request. Measure: {measure_spec.element_name}; Specs to join: {required_linkable_specs}"
             )
 
-        queried_agg_time_dimension_specs = queried_linkable_specs.included_agg_time_dimension_specs_for_measure(
-            measure_reference=measure_spec.reference, semantic_model_lookup=self._semantic_model_lookup
+        queried_agg_time_dimension_specs = tuple(
+            queried_linkable_specs.included_agg_time_dimension_specs_for_measure(
+                measure_reference=measure_spec.reference, semantic_model_lookup=self._semantic_model_lookup
+            )
+        )
+        base_queried_agg_time_dimension_specs = tuple(
+            TimeDimensionSpec.with_base_grains(queried_agg_time_dimension_specs)
         )
-        base_queried_agg_time_dimension_specs = TimeDimensionSpec.with_base_grains(queried_agg_time_dimension_specs)
 
         # If a cumulative metric is queried with metric_time / agg_time_dimension, join over time range.
         # Otherwise, the measure will be aggregated over all time.
@@ -1821,11 +1865,7 @@ def _build_aggregated_measure_from_measure_source_node(
                 grain_to_date=cumulative_grain_to_date,
                 # Note: we use the original constraint here because the JoinOverTimeRangeNode will eventually get
                 # rendered with an interval that expands the join window
-                time_range_constraint=(
-                    predicate_pushdown_state.time_range_constraint
-                    if not before_aggregation_time_spine_join_description
-                    else None
-                ),
+                time_range_constraint=(predicate_pushdown_state.time_range_constraint if not uses_offset else None),
             )
 
         # If querying an offset metric, join to time spine before aggregation.
@@ -1836,10 +1876,10 @@ def _build_aggregated_measure_from_measure_source_node(
             == {before_aggregation_time_spine_join_description.custom_offset_window.granularity}
         )
         if before_aggregation_time_spine_join_description and queried_agg_time_dimension_specs:
-            unaggregated_measure_node = self._build_time_spine_join_node_for_offset(
+            unaggregated_measure_node = self._build_time_spine_join_node_for_before_aggregation(
                 join_description=before_aggregation_time_spine_join_description,
                 measure_properties=measure_properties,
-                queried_agg_time_dimension_specs=tuple(queried_agg_time_dimension_specs),
+                queried_agg_time_dimension_specs=queried_agg_time_dimension_specs,
                 metric_source_node=unaggregated_measure_node,
                 use_offset_custom_granularity_node=use_offset_custom_granularity_node,
             )
@@ -1874,21 +1914,16 @@ def _build_aggregated_measure_from_measure_source_node(
             parent_node=unaggregated_measure_node, metric_input_measure_specs=(metric_input_measure_spec,)
         )
 
-        # Joining to time spine after aggregation is for measures that specify `join_to_timespine: true` in the YAML spec.
-        after_aggregation_time_spine_join_description = (
-            metric_input_measure_spec.after_aggregation_time_spine_join_description
-        )
-        queried_agg_time_dimension_specs = queried_linkable_specs.included_agg_time_dimension_specs_for_measure(
-            measure_reference=measure_spec.reference, semantic_model_lookup=self._semantic_model_lookup
-        )
         if after_aggregation_time_spine_join_description and queried_agg_time_dimension_specs:
-            return self._build_time_spine_join_node_for_measure_config(
+            return self._build_time_spine_join_node_for_after_aggregation(
                 join_description=after_aggregation_time_spine_join_description,
                 measure_reference=measure_spec.reference,
                 measure_source_node=aggregate_measures_node,
-                queried_agg_time_dimension_specs=tuple(queried_agg_time_dimension_specs),
+                queried_agg_time_dimension_specs=queried_agg_time_dimension_specs,
                 queried_linkable_specs=queried_linkable_specs.as_tuple,
-                after_aggregation_where_filter_specs=metric_input_measure_spec.filter_spec_set.after_measure_aggregation_filter_specs,
+                after_aggregation_where_filter_specs=(
+                    metric_input_measure_spec.filter_spec_set.after_measure_aggregation_filter_specs
+                ),
                 time_range_constraint=predicate_pushdown_state.time_range_constraint,
             )
 

tests_metricflow/generate_snapshots.py

@@ -81,34 +81,34 @@ class MetricFlowTestCredentialSetForAllEngines(FrozenBaseModel):  # noqa: D101
     @property
     def as_configurations(self) -> Sequence[MetricFlowEngineConfiguration]:  # noqa: D102
         return (
-            MetricFlowEngineConfiguration(
-                engine=SqlEngine.DUCKDB,
-                credential_set=self.duck_db,
-            ),
-            MetricFlowEngineConfiguration(
-                engine=SqlEngine.REDSHIFT,
-                credential_set=self.redshift,
-            ),
-            MetricFlowEngineConfiguration(
-                engine=SqlEngine.SNOWFLAKE,
-                credential_set=self.snowflake,
-            ),
-            MetricFlowEngineConfiguration(
-                engine=SqlEngine.BIGQUERY,
-                credential_set=self.big_query,
-            ),
+            # MetricFlowEngineConfiguration(
+            #     engine=SqlEngine.DUCKDB,
+            #     credential_set=self.duck_db,
+            # ),
+            # MetricFlowEngineConfiguration(
+            #     engine=SqlEngine.REDSHIFT,
+            #     credential_set=self.redshift,
+            # ),
+            # MetricFlowEngineConfiguration(
+            #     engine=SqlEngine.SNOWFLAKE,
+            #     credential_set=self.snowflake,
+            # ),
+            # MetricFlowEngineConfiguration(
+            #     engine=SqlEngine.BIGQUERY,
+            #     credential_set=self.big_query,
+            # ),
             MetricFlowEngineConfiguration(
                 engine=SqlEngine.DATABRICKS,
                 credential_set=self.databricks,
             ),
-            MetricFlowEngineConfiguration(
-                engine=SqlEngine.POSTGRES,
-                credential_set=self.postgres,
-            ),
-            MetricFlowEngineConfiguration(
-                engine=SqlEngine.TRINO,
-                credential_set=self.trino,
-            ),
+            # MetricFlowEngineConfiguration(
+            #     engine=SqlEngine.POSTGRES,
+            #     credential_set=self.postgres,
+            # ),
+            # MetricFlowEngineConfiguration(
+            #     engine=SqlEngine.TRINO,
+            #     credential_set=self.trino,
+            # ),
         )