Support metric aliases in queries

metricflow-semantics/metricflow_semantics/specs/metric_spec.py

@@ -50,7 +50,7 @@ def without_offset(self) -> MetricSpec:
         """Represents the metric spec with any time offsets removed."""
         return MetricSpec(element_name=self.element_name, filter_spec_set=self.filter_spec_set, alias=self.alias)
 
-    def with_alias(self, alias: str) -> MetricSpec:
+    def with_alias(self, alias: Optional[str]) -> MetricSpec:
         """Add the alias to the metric spec."""
         return MetricSpec(
             element_name=self.element_name,

metricflow/dataflow/builder/dataflow_plan_builder.py

@@ -156,12 +156,14 @@ def _build_query_output_node(
         self, query_spec: MetricFlowQuerySpec, for_group_by_source_node: bool = False
     ) -> DataflowPlanNode:
         """Build SQL output node from query inputs. May be used to build query DFP or source node."""
+        metric_specs: Tuple[MetricSpec, ...] = ()
         for metric_spec in query_spec.metric_specs:
             if (
                 len(metric_spec.filter_spec_set.all_filter_specs) > 0
                 or metric_spec.offset_to_grain is not None
                 or metric_spec.offset_window is not None
             ):
+                metric_specs += (metric_spec.with_alias(None),) if metric_spec.alias else (metric_spec,)
                 raise ValueError(
                     f"The metric specs in the query spec should not contain any metric modifiers. Got: {metric_spec}"
                 )

metricflow/dataset/sql_dataset.py

@@ -164,6 +164,22 @@ def instance_for_spec(self, spec: InstanceSpec) -> MdoInstance:
             str(LazyFormat("Did not find instance matching spec in dataset.", spec=spec, instances=instances))
         )
 
+    def instance_for_column_name(self, column_name: str) -> MdoInstance:
+        """Given a spec, return the instance associated with it in the data set."""
+        instances = self.instance_set.as_tuple
+        for instance in instances:
+            if instance.associated_column.column_name == column_name:
+                return instance
+        raise RuntimeError(
+            str(
+                LazyFormat(
+                    "Did not find instance matching column name in dataset.",
+                    column_name=column_name,
+                    instances=instances,
+                )
+            )
+        )
+
     def instance_from_time_dimension_grain_and_date_part(
         self, time_dimension_spec: TimeDimensionSpec
     ) -> TimeDimensionInstance:

metricflow/plan_conversion/dataflow_to_sql.py

@@ -1537,7 +1537,9 @@ def visit_alias_specs_node(self, node: AliasSpecsNode) -> SqlDataSet:  # noqa: D
             new_spec = spec_to_alias.output_spec
 
             # Find the instance in the parent data set with matching grain & date part.
-            old_instance = parent_data_set.instance_for_spec(old_spec)
+            old_instance = parent_data_set.instance_for_column_name(
+                self._column_association_resolver.resolve_spec(old_spec).column_name
+            )
 
             # Build new instance & select column to match requested spec.
             new_instance = old_instance.with_new_spec(