Add new dataflow plan nodes for custom offset windows

metricflow-semantics/metricflow_semantics/dag/id_prefix.py

@@ -56,6 +56,8 @@ class StaticIdPrefix(IdPrefix, Enum, metaclass=EnumMetaClassHelper):
     DATAFLOW_NODE_JOIN_CONVERSION_EVENTS_PREFIX = "jce"
     DATAFLOW_NODE_WINDOW_REAGGREGATION_ID_PREFIX = "wr"
     DATAFLOW_NODE_ALIAS_SPECS_ID_PREFIX = "as"
+    DATAFLOW_NODE_CUSTOM_GRANULARITY_BOUNDS_ID_PREFIX = "cgb"
+    DATAFLOW_NODE_OFFSET_BY_CUSTOMG_GRANULARITY_ID_PREFIX = "obcg"
 
     SQL_EXPR_COLUMN_REFERENCE_ID_PREFIX = "cr"
     SQL_EXPR_COMPARISON_ID_PREFIX = "cmp"
@@ -75,6 +77,9 @@ class StaticIdPrefix(IdPrefix, Enum, metaclass=EnumMetaClassHelper):
     SQL_EXPR_BETWEEN_PREFIX = "betw"
     SQL_EXPR_WINDOW_FUNCTION_ID_PREFIX = "wfnc"
     SQL_EXPR_GENERATE_UUID_PREFIX = "uuid"
+    SQL_EXPR_CASE_PREFIX = "case"
+    SQL_EXPR_ARITHMETIC_PREFIX = "arit"
+    SQL_EXPR_INTEGER_PREFIX = "int"
 
     SQL_PLAN_SELECT_STATEMENT_ID_PREFIX = "ss"
     SQL_PLAN_TABLE_FROM_CLAUSE_ID_PREFIX = "tfc"

metricflow-semantics/metricflow_semantics/instances.py

@@ -164,11 +164,7 @@ def with_entity_prefix(
     ) -> TimeDimensionInstance:
         """Returns a new instance with the entity prefix added to the entity links."""
         transformed_spec = self.spec.with_entity_prefix(entity_prefix)
-        return TimeDimensionInstance(
-            associated_columns=(column_association_resolver.resolve_spec(transformed_spec),),
-            defined_from=self.defined_from,
-            spec=transformed_spec,
-        )
+        return self.with_new_spec(transformed_spec, column_association_resolver)
 
     def with_new_defined_from(self, defined_from: Sequence[SemanticModelElementReference]) -> TimeDimensionInstance:
         """Returns a new instance with the defined_from field replaced."""

metricflow-semantics/metricflow_semantics/specs/dunder_column_association_resolver.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-from metricflow_semantics.model.semantic_manifest_lookup import SemanticManifestLookup
 from metricflow_semantics.naming.linkable_spec_name import DUNDER
 from metricflow_semantics.specs.column_assoc import (
     ColumnAssociation,
@@ -28,8 +27,8 @@ class DunderColumnAssociationResolver(ColumnAssociationResolver):
     listing__country
     """
 
-    def __init__(self, semantic_manifest_lookup: SemanticManifestLookup) -> None:  # noqa: D107
-        self._visitor_helper = DunderColumnAssociationResolverVisitor(semantic_manifest_lookup)
+    def __init__(self) -> None:  # noqa: D107
+        self._visitor_helper = DunderColumnAssociationResolverVisitor()
 
     def resolve_spec(self, spec: InstanceSpec) -> ColumnAssociation:  # noqa: D102
         return spec.accept(self._visitor_helper)
@@ -38,9 +37,6 @@ def resolve_spec(self, spec: InstanceSpec) -> ColumnAssociation:  # noqa: D102
 class DunderColumnAssociationResolverVisitor(InstanceSpecVisitor[ColumnAssociation]):
     """Visitor helper class for DefaultColumnAssociationResolver2."""
 
-    def __init__(self, semantic_manifest_lookup: SemanticManifestLookup) -> None:  # noqa: D107
-        self._semantic_manifest_lookup = semantic_manifest_lookup
-
     def visit_metric_spec(self, metric_spec: MetricSpec) -> ColumnAssociation:  # noqa: D102
         return ColumnAssociation(metric_spec.element_name if metric_spec.alias is None else metric_spec.alias)
 
@@ -58,6 +54,11 @@ def visit_time_dimension_spec(self, time_dimension_spec: TimeDimensionSpec) -> C
                 if time_dimension_spec.aggregation_state
                 else ""
             )
+            + (
+                f"{DUNDER}{time_dimension_spec.window_function.value.lower()}"
+                if time_dimension_spec.window_function
+                else ""
+            )
         )
 
     def visit_entity_spec(self, entity_spec: EntitySpec) -> ColumnAssociation:  # noqa: D102

metricflow-semantics/metricflow_semantics/specs/time_dimension_spec.py

@@ -15,6 +15,7 @@
 from metricflow_semantics.naming.linkable_spec_name import StructuredLinkableSpecName
 from metricflow_semantics.specs.dimension_spec import DimensionSpec
 from metricflow_semantics.specs.instance_spec import InstanceSpecVisitor
+from metricflow_semantics.sql.sql_exprs import SqlWindowFunction
 from metricflow_semantics.time.granularity import ExpandedTimeGranularity
 from metricflow_semantics.visitor import VisitorOutputT
 
@@ -91,6 +92,8 @@ class TimeDimensionSpec(DimensionSpec):  # noqa: D101
     # Used for semi-additive joins. Some more thought is needed, but this may be useful in InstanceSpec.
     aggregation_state: Optional[AggregationState] = None
 
+    window_function: Optional[SqlWindowFunction] = None
+
     @property
     def without_first_entity_link(self) -> TimeDimensionSpec:  # noqa: D102
         assert len(self.entity_links) > 0, f"Spec does not have any entity links: {self}"
@@ -99,6 +102,8 @@ def without_first_entity_link(self) -> TimeDimensionSpec:  # noqa: D102
             entity_links=self.entity_links[1:],
             time_granularity=self.time_granularity,
             date_part=self.date_part,
+            aggregation_state=self.aggregation_state,
+            window_function=self.window_function,
         )
 
     @property
@@ -108,6 +113,8 @@ def without_entity_links(self) -> TimeDimensionSpec:  # noqa: D102
             time_granularity=self.time_granularity,
             date_part=self.date_part,
             entity_links=(),
+            aggregation_state=self.aggregation_state,
+            window_function=self.window_function,
         )
 
     @property
@@ -153,6 +160,7 @@ def with_grain(self, time_granularity: ExpandedTimeGranularity) -> TimeDimension
             time_granularity=time_granularity,
             date_part=self.date_part,
             aggregation_state=self.aggregation_state,
+            window_function=self.window_function,
         )
 
     def with_base_grain(self) -> TimeDimensionSpec:  # noqa: D102
@@ -162,6 +170,7 @@ def with_base_grain(self) -> TimeDimensionSpec:  # noqa: D102
             time_granularity=ExpandedTimeGranularity.from_time_granularity(self.time_granularity.base_granularity),
             date_part=self.date_part,
             aggregation_state=self.aggregation_state,
+            window_function=self.window_function,
         )
 
     def with_grain_and_date_part(  # noqa: D102
@@ -173,6 +182,7 @@ def with_grain_and_date_part(  # noqa: D102
             time_granularity=time_granularity,
             date_part=date_part,
             aggregation_state=self.aggregation_state,
+            window_function=self.window_function,
         )
 
     def with_aggregation_state(self, aggregation_state: AggregationState) -> TimeDimensionSpec:  # noqa: D102
@@ -182,6 +192,17 @@ def with_aggregation_state(self, aggregation_state: AggregationState) -> TimeDim
             time_granularity=self.time_granularity,
             date_part=self.date_part,
             aggregation_state=aggregation_state,
+            window_function=self.window_function,
+        )
+
+    def with_window_function(self, window_function: SqlWindowFunction) -> TimeDimensionSpec:  # noqa: D102
+        return TimeDimensionSpec(
+            element_name=self.element_name,
+            entity_links=self.entity_links,
+            time_granularity=self.time_granularity,
+            date_part=self.date_part,
+            aggregation_state=self.aggregation_state,
+            window_function=window_function,
         )
 
     def comparison_key(self, exclude_fields: Sequence[TimeDimensionSpecField] = ()) -> TimeDimensionSpecComparisonKey:
@@ -243,6 +264,7 @@ def with_entity_prefix(self, entity_prefix: EntityReference) -> TimeDimensionSpe
             time_granularity=self.time_granularity,
             date_part=self.date_part,
             aggregation_state=self.aggregation_state,
+            window_function=self.window_function,
         )
 
     @staticmethod