Update to convert metric nodes to cte.

Author: plypaul

metricflow/dataflow/dataflow_plan_analyzer.py

@@ -41,7 +41,7 @@ def find_common_branches(dataflow_plan: DataflowPlan) -> Sequence[DataflowPlanNo
 
     @staticmethod
     def group_nodes_by_type(dataflow_plan: DataflowPlan) -> DataflowPlanNodeSet:
-        """Grouops dataflow plan nodes by type."""
+        """Groups dataflow plan nodes by type."""
         grouping_visitor = _GroupNodesByTypeVisitor()
         return dataflow_plan.sink_node.accept(grouping_visitor)
 
@@ -98,12 +98,14 @@ class DataflowPlanNodeSet(Mergeable):
 
     compute_metric_nodes: FrozenSet[ComputeMetricsNode]
 
+    @override
     def merge(self, other: DataflowPlanNodeSet) -> DataflowPlanNodeSet:
         return DataflowPlanNodeSet(
             compute_metric_nodes=self.compute_metric_nodes.union(other.compute_metric_nodes),
         )
 
     @classmethod
+    @override
     def empty_instance(cls) -> DataflowPlanNodeSet:
         return DataflowPlanNodeSet(
             compute_metric_nodes=frozenset(),

metricflow/plan_conversion/dataflow_to_sql.py

@@ -273,7 +273,10 @@ def _get_nodes_to_convert_to_cte(
         """Handles logic for selecting which nodes to convert to CTEs based on the request."""
         dataflow_plan = dataflow_plan_node.as_plan()
         nodes_to_convert_to_cte: Set[DataflowPlanNode] = set(DataflowPlanAnalyzer.find_common_branches(dataflow_plan))
-        # Additional nodes will be added later.
+
+        compute_metric_nodes = DataflowPlanAnalyzer.group_nodes_by_type(dataflow_plan).compute_metric_nodes
+        if len(compute_metric_nodes) > 1:
+            nodes_to_convert_to_cte.update(compute_metric_nodes)
 
         return frozenset(nodes_to_convert_to_cte)