class_docstr: Builds a dataflow plan to satisfy a given query.
class_docstr_tok: ["Builds","a","dataflow","plan","to","satisfy","a","given","query","."]
code: def _build_aggregated_measure_from_measure_source_node(
      self,
      metric_input_measure_spec: MetricInputMeasureSpec,
      queried_linkable_specs: LinkableSpecSet,
      time_range_constraint: Optional[TimeRangeConstraint] = None,
      measure_recipe: Optional[DataflowRecipe] = None,
  ) -> BaseOutput:
      measure_spec = metric_input_measure_spec.measure_spec
      cumulative = (
          metric_input_measure_spec.culmination_description is not None
      )
      cumulative_window = (
          metric_input_measure_spec.culmination_description.cumulative_window
          if metric_input_measure_spec.culmination_description
          is not None
          else None
      )
      cumulative_grain_to_date = (
          metric_input_measure_spec.culmination_description.cumulative_grain_to_date
          if metric_input_measure_spec.culmination_description
          else None
      )
      measure_properties = self._build_measure_spec_properties(
          [measure_spec]
      )
      non_additive_dimension_spec = (
          measure_properties.non_additive_dimension_spec
      )
  
      cumulative_metric_adjusted_time_constraint: Optional[
          TimeRangeConstraint
      ] = None
      if cumulative and time_range_constraint is not None:
          logger.info(
              f"Time range constraint before adjustment is {time_range_constraint}"
          )
          granularity: Optional[TimeGranularity] = None
          count = 0
          if cumulative_window is not None:
              granularity = cumulative_window.granularity
              count = cumulative_window.count
          elif cumulative_grain_to_date is not None:
              count = 1
              granularity = cumulative_grain_to_date
  
          cumulative_metric_adjusted_time_constraint = time_range_constraint.adjust_time_constraint_for_cumulative_metric(
              granularity, count
          )
          logger.info(
              f"Adjusted time range constraint {cumulative_metric_adjusted_time_constraint}"
          )
  
      (
          required_linkable_specs,
          extraneous_linkable_specs,
      ) = self.__get_required_and_extraneous_linkable_specs(
          queried_linkable_specs=queried_linkable_specs,
          filter_specs=metric_input_measure_spec.filter_specs,
          non_additive_dimension_spec=non_additive_dimension_spec,
      )
  
      before_aggregation_time_spine_join_description = (
          metric_input_measure_spec.before_aggregation_time_spine_join_description
      )
  
      if measure_recipe is None:
          logger.info(
              f"Looking for a recipe to get:\n"
              f"{pformat_big_objects(measure_specs=[measure_spec], required_linkable_set=required_linkable_specs)}"
          )
          find_recipe_start_time = time.time()
          measure_recipe = self._find_dataflow_recipe(
              measure_spec_properties=measure_properties,
              time_range_constraint=(
                  cumulative_metric_adjusted_time_constraint
                  or time_range_constraint
              )
              if not before_aggregation_time_spine_join_description
              else None,
              linkable_spec_set=required_linkable_specs,
          )
          logger.info(
              f"With {len(self._source_nodes)} source nodes, finding a recipe took "
              f"{time.time() - find_recipe_start_time:.2f}s"
          )
  
      logger.info(
          f"Using recipe:\n{pformat_big_objects(measure_recipe=measure_recipe)}"
      )
  
      if measure_recipe is None:
          # TODO: Improve for better user understandability.
          raise UnableToSatisfyQueryError(
              f"Recipe not found for measure spec: {measure_spec} and linkable specs: {required_linkable_specs}"
          )
  
      # If a cumulative metric is queried with metric_time, join over time range.
      # Otherwise, the measure will be aggregated over all time.
      time_range_node: Optional[JoinOverTimeRangeNode] = None
      if cumulative and queried_linkable_specs.contains_metric_time:
          time_range_node = JoinOverTimeRangeNode(
              parent_node=measure_recipe.source_node,
              window=cumulative_window,
              grain_to_date=cumulative_grain_to_date,
              time_range_constraint=time_range_constraint
              if not before_aggregation_time_spine_join_description
              else None,
          )
  
      # If querying an offset metric, join to time spine before aggregation.
      join_to_time_spine_node: Optional[JoinToTimeSpineNode] = None
      if before_aggregation_time_spine_join_description is not None:
          assert (
              queried_linkable_specs.contains_metric_time
          ), "Joining to time spine requires querying with metric time."
          assert (
              before_aggregation_time_spine_join_description.join_type
              is SqlJoinType.INNER
          ), (
              f"Expected {SqlJoinType.INNER} for joining to time spine before aggregation. Remove this if there's a "
              f"new use case."
          )
          join_to_time_spine_node = JoinToTimeSpineNode(
              parent_node=time_range_node or measure_recipe.source_node,
              requested_metric_time_dimension_specs=list(
                  queried_linkable_specs.metric_time_specs
              ),
              time_range_constraint=time_range_constraint,
              offset_window=before_aggregation_time_spine_join_description.offset_window,
              offset_to_grain=before_aggregation_time_spine_join_description.offset_to_grain,
              join_type=before_aggregation_time_spine_join_description.join_type,
          )
  
      # Only get the required measure and the local linkable instances so that aggregations work correctly.
      filtered_measure_source_node = FilterElementsNode(
          parent_node=join_to_time_spine_node
          or time_range_node
          or measure_recipe.source_node,
          include_specs=InstanceSpecSet(
              measure_specs=(measure_spec,)
          ).merge(
              InstanceSpecSet.from_specs(
                  measure_recipe.required_local_linkable_specs
              ),
          ),
      )
  
      join_targets = measure_recipe.join_targets
      unaggregated_measure_node: BaseOutput
      if len(join_targets) > 0:
          filtered_measures_with_joined_elements = JoinToBaseOutputNode(
              left_node=filtered_measure_source_node,
              join_targets=join_targets,
          )
  
          specs_to_keep_after_join = InstanceSpecSet(
              measure_specs=(measure_spec,)
          ).merge(
              required_linkable_specs.as_spec_set,
          )
  
          after_join_filtered_node = FilterElementsNode(
              parent_node=filtered_measures_with_joined_elements,
              include_specs=specs_to_keep_after_join,
          )
          unaggregated_measure_node = after_join_filtered_node
      else:
          unaggregated_measure_node = filtered_measure_source_node
  
      cumulative_metric_constrained_node: Optional[
          ConstrainTimeRangeNode
      ] = None
      if (
          cumulative_metric_adjusted_time_constraint is not None
          and time_range_constraint is not None
          and queried_linkable_specs.contains_metric_time
      ):
          cumulative_metric_constrained_node = ConstrainTimeRangeNode(
              unaggregated_measure_node, time_range_constraint
          )
  
      pre_aggregate_node: BaseOutput = (
          cumulative_metric_constrained_node
          or unaggregated_measure_node
      )
      if len(metric_input_measure_spec.filter_specs) > 0:
          # Apply where constraint on the node
  
          merged_where_filter_spec = WhereFilterSpec.merge_iterable(
              metric_input_measure_spec.filter_specs
          )
          pre_aggregate_node = WhereConstraintNode(
              parent_node=pre_aggregate_node,
              where_constraint=merged_where_filter_spec,
          )
  
      if non_additive_dimension_spec is not None:
          # Apply semi additive join on the node
          agg_time_dimension = measure_properties.agg_time_dimension
          queried_time_dimension_spec: Optional[
              TimeDimensionSpec
          ] = self._find_non_additive_dimension_in_linkable_specs(
              agg_time_dimension=agg_time_dimension,
              linkable_specs=queried_linkable_specs.as_tuple,
              non_additive_dimension_spec=non_additive_dimension_spec,
          )
          time_dimension_spec = TimeDimensionSpec.from_name(
              non_additive_dimension_spec.name
          )
          window_groupings = tuple(
              LinklessEntitySpec.from_element_name(name)
              for name in non_additive_dimension_spec.window_groupings
          )
          pre_aggregate_node = SemiAdditiveJoinNode(
              parent_node=pre_aggregate_node,
              entity_specs=window_groupings,
              time_dimension_spec=time_dimension_spec,
              agg_by_function=non_additive_dimension_spec.window_choice,
              queried_time_dimension_spec=queried_time_dimension_spec,
          )
  
      if not extraneous_linkable_specs.is_subset_of(
          queried_linkable_specs
      ):
          # At this point, it's the case that there are extraneous specs are not a subset of the queried
          # linkable specs. A filter is needed after, say, a where clause so that the linkable specs in the where clause don't
          # show up in the final result.
          #
          # e.g. for "bookings" by "ds" where "is_instant", "is_instant" should not be in the results.
          pre_aggregate_node = FilterElementsNode(
              parent_node=pre_aggregate_node,
              include_specs=InstanceSpecSet(
                  measure_specs=(measure_spec,)
              ).merge(queried_linkable_specs.as_spec_set),
          )
  
      aggregate_measures_node = AggregateMeasuresNode(
          parent_node=pre_aggregate_node,
          metric_input_measure_specs=(metric_input_measure_spec,),
      )
  
      # Joining to time spine after aggregation is for measures that specify `join_to_timespine`` in the YAML spec.
      after_aggregation_time_spine_join_description = (
          metric_input_measure_spec.after_aggregation_time_spine_join_description
      )
      if after_aggregation_time_spine_join_description is not None:
          assert (
              after_aggregation_time_spine_join_description.join_type
              is SqlJoinType.LEFT_OUTER
          ), (
              f"Expected {SqlJoinType.LEFT_OUTER} for joining to time spine after aggregation. Remove this if "
              f"there's a new use case."
          )
          return JoinToTimeSpineNode(
              parent_node=aggregate_measures_node,
              requested_metric_time_dimension_specs=list(
                  queried_linkable_specs.metric_time_specs
              ),
              join_type=after_aggregation_time_spine_join_description.join_type,
              time_range_constraint=time_range_constraint,
              offset_window=after_aggregation_time_spine_join_description.offset_window,
              offset_to_grain=after_aggregation_time_spine_join_description.offset_to_grain,
          )
      else:
          return aggregate_measures_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 27 for calling the DataflowPlanBuilder obj's underlying member method code with example usage: obj._build_aggregated_measure_from_measure_source_node(metric_input_measure_spec, queried_linkable_specs, time_range_constraint, measure_recipe) and returns: JoinToTimeSpineNode, aggregate_measures_node