class_docstr: Reduces the number of scans (ReadSqlSourceNodes) in a dataflow plan.
  
  This attempts to reduce the number of scans by combining the parent nodes of CombineAggregatedOutputsNode via the
  ComputeMetricsBranchCombiner.
  
  A plan with a structure similar to
      ...
      <CombineAggregatedOutputsNode>
          <ComputeMetricsNode metrics="[metric0]">
              <AggregateMeasuresNode>
              ...
              </AggregateMeasuresNode>
          </ComputeMetricsNode>
          <ComputeMetricsNode metrics="[metrics1]">
              <AggregateMeasuresNode>
              ...
              </AggregateMeasuresNode>
          </ComputeMetricsNode>
          <ComputeMetricsNode metric="[metrics2]">
              <AggregateMeasuresNode>
              ...
              </AggregateMeasuresNode>
          </ComputeMetricsNode>
      </CombineAggregatedOutputsNode>
      ...
  will be converted to
      ...
      <CombineAggregatedOutputsNode>
          <ComputeMetricsNode metrics="[metric0, metric1]">
              <AggregateMeasuresNode>
              ...
              </AggregateMeasuresNode>
          </ComputeMetricsNode>
          <ComputeMetricsNode metrics="[metric2]">
              <AggregateMeasuresNode>
              ...
              </AggregateMeasuresNode>
          </ComputeMetricsNode>
      </CombineAggregatedOutputsNode>
      ...
  when possible.
  
  In cases where all ComputeMetricsNodes can be combined into a single one, the CombineAggregatedOutputsNode may be removed as
  well.
  
  This traverses the dataflow plan using DFS. When visiting a node (current_node), it first runs the optimization
  process on the parent branches, then tries to create a more optimal version of the current_node with those optimized
  parents.
class_docstr_tok: ["Reduces","the","number","of","scans","(","ReadSqlSourceNodes",")","in","a","dataflow","plan",".","This","attempts","to","reduce","the","number","of","scans","by","combining","the","parent","nodes","of","CombineAggregatedOutputsNode","via","the","ComputeMetricsBranchCombiner",".","A","plan","with","a","structure","similar","to","...","<","CombineAggregatedOutputsNode",">","<","ComputeMetricsNode","metrics=","''","[","metric0","]","''",">","<","AggregateMeasuresNode",">","...","<","\/AggregateMeasuresNode",">","<","\/ComputeMetricsNode",">","<","ComputeMetricsNode","metrics=","''","[","metrics1","]","''",">","<","AggregateMeasuresNode",">","...","<","\/AggregateMeasuresNode",">","<","\/ComputeMetricsNode",">","<","ComputeMetricsNode","metric=","''","[","metrics2","]","''",">","<","AggregateMeasuresNode",">","...","<","\/AggregateMeasuresNode",">","<","\/ComputeMetricsNode",">","<","\/CombineAggregatedOutputsNode",">","...","will","be","converted","to","...","<","CombineAggregatedOutputsNode",">","<","ComputeMetricsNode","metrics=","''","[","metric0",",","metric1","]","''",">","<","AggregateMeasuresNode",">","...","<","\/AggregateMeasuresNode",">","<","\/ComputeMetricsNode",">","<","ComputeMetricsNode","metrics=","''","[","metric2","]","''",">","<","AggregateMeasuresNode",">","...","<","\/AggregateMeasuresNode",">","<","\/ComputeMetricsNode",">","<","\/CombineAggregatedOutputsNode",">","...","when","possible",".","In","cases","where","all","ComputeMetricsNodes","can","be","combined","into","a","single","one",",","the","CombineAggregatedOutputsNode","may","be","removed","as","well",".","This","traverses","the","dataflow","plan","using","DFS",".","When","visiting","a","node","(","current_node",")",",","it","first","runs","the","optimization","process","on","the","parent","branches",",","then","tries","to","create","a","more","optimal","version","of","the","current_node","with","those","optimized","parents","."]
code: def visit_combine_aggregated_outputs_node(  # noqa: D
      self, node: CombineAggregatedOutputsNode
  ) -> OptimizeBranchResult:
      self._log_visit_node_type(node)
      # The parent node of the CombineAggregatedOutputsNode can be either ComputeMetricsNodes or CombineAggregatedOutputsNodes
  
      # Stores the result of running this optimizer on each parent branch separately.
      optimized_parent_branches = []
      logger.log(
          level=self._log_level,
          msg=f"{node} has {len(node.parent_nodes)} parent branches",
      )
  
      # Run the optimizer on the parent branch to handle derived metrics, which are defined recursively in the DAG.
      for parent_branch in node.parent_nodes:
          result: OptimizeBranchResult = parent_branch.accept(self)
  
          assert result.sink_node is None, (
              f"Traversing the parents of of {node.__class__.__name__} should not have produced any "
              f"{SinkOutput.__class__.__name__} nodes"
          )
  
          assert (
              result.base_output_node is not None
          ), f"Traversing the parents of a CombineAggregatedOutputsNode should always produce a BaseOutput. Got: {result}"
          optimized_parent_branches.append(result.base_output_node)
  
      # Try to combine (using ComputeMetricsBranchCombiner) as many parent branches as possible in a
      # greedy N^2 approach. The optimality of this approach needs more thought to prove conclusively, but given
      # the seemingly transitive properties of the combination operation, this seems reasonable.
      combined_parent_branches: List[BaseOutput] = []
      for optimized_parent_branch in optimized_parent_branches:
          combination_results = SourceScanOptimizer._combine_branches(
              left_branches=combined_parent_branches,
              right_branch=optimized_parent_branch,
          )
  
          # If optimized_parent_branch couldn't be combined with any of the existing ones, add it to the list.
          if not any(
              x.combined_branch is not None for x in combination_results
          ):
              combined_parent_branches.append(optimized_parent_branch)
          # Otherwise, replaced the branch with the one that was combined in combined_parent_branches
          else:
              combined_parent_branches = [
                  branch_combination_result.left_branch
                  if branch_combination_result.combined_branch is None
                  else branch_combination_result.combined_branch
                  for branch_combination_result in combination_results
              ]
  
      logger.log(
          level=self._log_level,
          msg=f"Got {len(combined_parent_branches)} branches after combination",
      )
      assert len(combined_parent_branches) > 0
  
      # If we were able to reduce the parent branches of the CombineAggregatedOutputsNode into a single one, there's no need
      # for a CombineAggregatedOutputsNode.
      if len(combined_parent_branches) == 1:
          return OptimizeBranchResult(
              base_output_node=combined_parent_branches[0]
          )
  
      return OptimizeBranchResult(
          base_output_node=CombineAggregatedOutputsNode(
              parent_nodes=combined_parent_branches
          )
      )
  
imports: from __future__ import annotations
  import logging
  from dataclasses import dataclass
  from typing import List, Optional, Sequence
  from metricflow.dag.id_generation import OPTIMIZED_DATAFLOW_PLAN_PREFIX, IdGeneratorRegistry
  from metricflow.dataflow.dataflow_plan import AddGeneratedUuidColumnNode, AggregateMeasuresNode, BaseOutput, CombineAggregatedOutputsNode, ComputeMetricsNode, ConstrainTimeRangeNode, DataflowPlan, DataflowPlanNode, DataflowPlanNodeVisitor, FilterElementsNode, JoinConversionEventsNode, JoinOverTimeRangeNode, JoinToBaseOutputNode, JoinToTimeSpineNode, MetricTimeDimensionTransformNode, MinMaxNode, OrderByLimitNode, ReadSqlSourceNode, SemiAdditiveJoinNode, SinkOutput, WhereConstraintNode, WriteToResultDataframeNode, WriteToResultTableNode
  from metricflow.dataflow.dataflow_plan_to_text import dataflow_dag_as_text
  from metricflow.dataflow.optimizer.dataflow_plan_optimizer import DataflowPlanOptimizer
  from metricflow.dataflow.optimizer.source_scan.cm_branch_combiner import ComputeMetricsBranchCombiner, ComputeMetricsBranchCombinerResult
label_desc: Use image node_id 15 for calling the SourceScanOptimizer obj's underlying member method code with example usage: obj.visit_combine_aggregated_outputs_node(node) and returns: OptimizeBranchResult, OptimizeBranchResult