Don't combine ComputeMetricsNodes in optimizer if either is a group b…

…y source node

metricflow/dataflow/optimizer/source_scan/cm_branch_combiner.py

@@ -138,9 +138,10 @@ def _log_combine_failure(
         left_node: DataflowPlanNode,
         right_node: DataflowPlanNode,
         combine_failure_reason: str,
+        log_level: Optional[int] = None,
     ) -> None:
         logger.log(
-            level=self._log_level,
+            level=log_level if log_level is not None else self._log_level,
             msg=f"Because {combine_failure_reason}, unable to combine nodes "
             f"left_node={left_node} right_node={right_node}",
         )
@@ -293,6 +294,15 @@ def visit_compute_metrics_node(self, node: ComputeMetricsNode) -> ComputeMetrics
             )
             return ComputeMetricsBranchCombinerResult()
 
+        if self._current_left_node.for_group_by_source_node or current_right_node.for_group_by_source_node:
+            self._log_combine_failure(
+                left_node=self._current_left_node,
+                right_node=current_right_node,
+                combine_failure_reason="at least one node is a group by metric source node, which must be aggregated separately from other nodes",
+                log_level=logging.INFO,  # Temp for debugging in prod
+            )
+            return ComputeMetricsBranchCombinerResult()
+
         assert len(combined_parent_nodes) == 1
         combined_parent_node = combined_parent_nodes[0]
         assert combined_parent_node is not None
@@ -308,6 +318,7 @@ def visit_compute_metrics_node(self, node: ComputeMetricsNode) -> ComputeMetrics
             parent_node=combined_parent_node,
             metric_specs=unique_metric_specs,
             aggregated_to_elements=current_right_node.aggregated_to_elements,
+            for_group_by_source_node=current_right_node.for_group_by_source_node,
         )
         self._log_combine_success(
             left_node=self._current_left_node,