class_docstr: Builds a dataflow plan to satisfy a given query.
class_docstr_tok: ["Builds","a","dataflow","plan","to","satisfy","a","given","query","."]
docstr: Builds a node that contains aggregated values of conversions and opportunities.
docstr_tok: ["Builds","a","node","that","contains","aggregated","values","of","conversions","and","opportunities","."]
code: def _build_aggregated_conversion_node(
      self,
      base_measure_spec: MetricInputMeasureSpec,
      conversion_measure_spec: MetricInputMeasureSpec,
      entity_spec: EntitySpec,
      window: Optional[MetricTimeWindow],
      queried_linkable_specs: LinkableSpecSet,
      time_range_constraint: Optional[TimeRangeConstraint] = None,
      constant_properties: Optional[
          Sequence[ConstantPropertyInput]
      ] = None,
  ) -> BaseOutput:
      """Builds a node that contains aggregated values of conversions and opportunities."""
      # Build measure recipes
      (
          base_required_linkable_specs,
          _,
      ) = self.__get_required_and_extraneous_linkable_specs(
          queried_linkable_specs=queried_linkable_specs,
          filter_specs=base_measure_spec.filter_specs,
      )
      base_measure_recipe = self._find_dataflow_recipe(
          measure_spec_properties=self._build_measure_spec_properties(
              [base_measure_spec.measure_spec]
          ),
          time_range_constraint=time_range_constraint,
          linkable_spec_set=base_required_linkable_specs,
      )
      logger.info(
          f"Recipe for base measure aggregation:\n{pformat_big_objects(measure_recipe=base_measure_recipe)}"
      )
      conversion_measure_recipe = self._find_dataflow_recipe(
          measure_spec_properties=self._build_measure_spec_properties(
              [conversion_measure_spec.measure_spec]
          ),
          linkable_spec_set=LinkableSpecSet(),
      )
      logger.info(
          f"Recipe for conversion measure aggregation:\n{pformat_big_objects(measure_recipe=conversion_measure_recipe)}"
      )
      if base_measure_recipe is None:
          raise UnableToSatisfyQueryError(
              f"Recipe not found for measure spec: {base_measure_spec.measure_spec} and linkable specs: {base_required_linkable_specs}"
          )
      if conversion_measure_recipe is None:
          raise UnableToSatisfyQueryError(
              f"Recipe not found for measure spec: {conversion_measure_spec.measure_spec}"
          )
  
      # Gets the aggregated opportunities
      aggregated_base_measure_node = self.build_aggregated_measure(
          metric_input_measure_spec=base_measure_spec,
          queried_linkable_specs=queried_linkable_specs,
          time_range_constraint=time_range_constraint,
      )
  
      # Build unaggregated conversions source node
      # Generate UUID column for conversion source to uniquely identify each row
      unaggregated_conversion_measure_node = AddGeneratedUuidColumnNode(
          parent_node=conversion_measure_recipe.source_node
      )
  
      # Get the agg time dimension for each measure used for matching conversion time windows
      base_time_dimension_spec = TimeDimensionSpec.from_reference(
          self._semantic_model_lookup.get_agg_time_dimension_for_measure(
              base_measure_spec.measure_spec.reference
          )
      )
      conversion_time_dimension_spec = TimeDimensionSpec.from_reference(
          self._semantic_model_lookup.get_agg_time_dimension_for_measure(
              conversion_measure_spec.measure_spec.reference
          )
      )
  
      # Filter the source nodes with only the required specs needed for the calculation
      constant_property_specs = []
      required_local_specs = [
          base_measure_spec.measure_spec,
          entity_spec,
          base_time_dimension_spec,
      ] + list(base_measure_recipe.required_local_linkable_specs)
      for constant_property in constant_properties or []:
          base_property_spec = (
              self._semantic_model_lookup.get_element_spec_for_name(
                  constant_property.base_property
              )
          )
          conversion_property_spec = (
              self._semantic_model_lookup.get_element_spec_for_name(
                  constant_property.conversion_property
              )
          )
          required_local_specs.append(base_property_spec)
          constant_property_specs.append(
              ConstantPropertySpec(
                  base_spec=base_property_spec,
                  conversion_spec=conversion_property_spec,
              )
          )
  
      # Build the unaggregated base measure node for computing conversions
      unaggregated_base_measure_node = base_measure_recipe.source_node
      if base_measure_recipe.join_targets:
          unaggregated_base_measure_node = JoinToBaseOutputNode(
              left_node=unaggregated_base_measure_node,
              join_targets=base_measure_recipe.join_targets,
          )
      filtered_unaggregated_base_node = FilterElementsNode(
          parent_node=unaggregated_base_measure_node,
          include_specs=InstanceSpecSet.from_specs(required_local_specs)
          .merge(base_required_linkable_specs.as_spec_set)
          .dedupe(),
      )
  
      # Gets the successful conversions using JoinConversionEventsNode
      # The conversion events are joined by the base events which are already time constrained. However, this could
      # be still be constrained, where we adjust the time range to the window size similar to cumulative, but
      # adjusted in the opposite direction.
      join_conversion_node = JoinConversionEventsNode(
          base_node=filtered_unaggregated_base_node,
          base_time_dimension_spec=base_time_dimension_spec,
          conversion_node=unaggregated_conversion_measure_node,
          conversion_measure_spec=conversion_measure_spec.measure_spec,
          conversion_time_dimension_spec=conversion_time_dimension_spec,
          unique_identifier_keys=(
              MetadataSpec.from_name(
                  MetricFlowReservedKeywords.MF_INTERNAL_UUID.value
              ),
          ),
          entity_spec=entity_spec,
          window=window,
          constant_properties=constant_property_specs,
      )
  
      # Aggregate the conversion events with the JoinConversionEventsNode as the source node
      recipe_with_join_conversion_source_node = DataflowRecipe(
          source_node=join_conversion_node,
          required_local_linkable_specs=base_measure_recipe.required_local_linkable_specs,
          join_linkable_instances_recipes=base_measure_recipe.join_linkable_instances_recipes,
      )
      aggregated_conversions_node = self.build_aggregated_measure(
          metric_input_measure_spec=conversion_measure_spec,
          queried_linkable_specs=queried_linkable_specs,
          time_range_constraint=time_range_constraint,
          measure_recipe=recipe_with_join_conversion_source_node,
      )
  
      # Combine the aggregated opportunities and conversion data sets
      return CombineAggregatedOutputsNode(
          parent_nodes=(
              aggregated_base_measure_node,
              aggregated_conversions_node,
          )
      )
  
imports: from __future__ import annotations
  import logging
  import time
  from dataclasses import dataclass
  from typing import Dict, List, Optional, Sequence, Set, Tuple, Union
  from dbt_semantic_interfaces.enum_extension import assert_values_exhausted
  from dbt_semantic_interfaces.implementations.metric import PydanticMetricTimeWindow
  from dbt_semantic_interfaces.pretty_print import pformat_big_objects
  from dbt_semantic_interfaces.protocols.metric import ConstantPropertyInput, ConversionTypeParams, MetricInputMeasure, MetricTimeWindow, MetricType
  from dbt_semantic_interfaces.references import MeasureReference, MetricReference, TimeDimensionReference
  from dbt_semantic_interfaces.type_enums.time_granularity import TimeGranularity
  from dbt_semantic_interfaces.validations.unique_valid_name import MetricFlowReservedKeywords
  from metricflow.dag.id_generation import DATAFLOW_PLAN_PREFIX, IdGeneratorRegistry
  from metricflow.dataflow.builder.node_data_set import DataflowPlanNodeOutputDataSetResolver
  from metricflow.dataflow.builder.node_evaluator import JoinLinkableInstancesRecipe, LinkableInstanceSatisfiabilityEvaluation, NodeEvaluatorForLinkableInstances
  from metricflow.dataflow.dataflow_plan import AddGeneratedUuidColumnNode, AggregateMeasuresNode, BaseOutput, CombineAggregatedOutputsNode, ComputeMetricsNode, ConstrainTimeRangeNode, DataflowPlan, FilterElementsNode, JoinConversionEventsNode, JoinDescription, JoinOverTimeRangeNode, JoinToBaseOutputNode, JoinToTimeSpineNode, 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.sql_table import SqlTable
  from metricflow.dataset.dataset import DataSet
  from metricflow.errors.errors import UnableToSatisfyQueryError
  from metricflow.filters.time_constraint import TimeRangeConstraint
  from metricflow.mf_logging.pretty_print import mf_pformat
  from metricflow.mf_logging.runtime import log_runtime
  from metricflow.model.semantic_manifest_lookup import SemanticManifestLookup
  from metricflow.plan_conversion.column_resolver import DunderColumnAssociationResolver
  from metricflow.plan_conversion.node_processor import PreJoinNodeProcessor
  from metricflow.query.group_by_item.filter_spec_resolution.filter_location import WhereFilterLocation
  from metricflow.query.group_by_item.filter_spec_resolution.filter_spec_lookup import FilterSpecResolutionLookUp
  from metricflow.specs.column_assoc import ColumnAssociationResolver
  from metricflow.specs.specs import ConstantPropertySpec, CumulativeMeasureDescription, EntitySpec, InstanceSpecSet, JoinToTimeSpineDescription, LinkableInstanceSpec, LinkableSpecSet, LinklessEntitySpec, MeasureSpec, MetadataSpec, MetricFlowQuerySpec, MetricInputMeasureSpec, MetricSpec, NonAdditiveDimensionSpec, OrderBySpec, TimeDimensionSpec, WhereFilterSpec
  from metricflow.specs.where_filter_transform import WhereSpecFactory
  from metricflow.sql.sql_plan import SqlJoinType
label_desc: Use image node_id 4 for calling the DataflowPlanBuilder obj's underlying member method code with example usage: obj._build_aggregated_conversion_node(base_measure_spec, conversion_measure_spec, entity_spec, window, queried_linkable_specs, time_range_constraint, constant_properties) and returns: CombineAggregatedOutputsNode