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 to compute a metric defined from other metrics.
docstr_tok: ["Builds","a","node","to","compute","a","metric","defined","from","other","metrics","."]
code: def _build_derived_metric_output_node(
      self,
      metric_spec: MetricSpec,
      queried_linkable_specs: LinkableSpecSet,
      filter_spec_factory: WhereSpecFactory,
      time_range_constraint: Optional[TimeRangeConstraint] = None,
  ) -> BaseOutput:
      """Builds a node to compute a metric defined from other metrics."""
      metric = self._metric_lookup.get_metric(metric_spec.reference)
      metric_input_specs = (
          self._build_input_metric_specs_for_derived_metric(
              metric_reference=metric_spec.reference,
              filter_spec_factory=filter_spec_factory,
          )
      )
      logger.info(
          f"For {metric.type} metric: {metric_spec}, needed metrics are:\n"
          f"{pformat_big_objects(metric_input_specs=metric_input_specs)}"
      )
  
      (
          required_linkable_specs,
          extraneous_linkable_specs,
      ) = self.__get_required_and_extraneous_linkable_specs(
          queried_linkable_specs=queried_linkable_specs,
          filter_specs=metric_spec.filter_specs,
      )
  
      parent_nodes: List[BaseOutput] = []
  
      # This is the filter that's defined for the metric in the configs.
      metric_definition_filter_specs = (
          filter_spec_factory.create_from_where_filter_intersection(
              filter_location=WhereFilterLocation.for_metric(
                  metric_spec.reference
              ),
              filter_intersection=metric.filter,
          )
      )
  
      for metric_input_spec in metric_input_specs:
          filter_specs: List[WhereFilterSpec] = []
          filter_specs.extend(metric_definition_filter_specs)
          # These are the filters that's defined as part of the input metric.
          filter_specs.extend(metric_input_spec.filter_specs)
  
          # If metric is offset, we'll apply where constraint after offset to avoid removing values
          # unexpectedly. Time constraint will be applied by INNER JOINing to time spine.
          if not metric_spec.has_time_offset:
              filter_specs.extend(metric_spec.filter_specs)
  
          parent_nodes.append(
              self._build_any_metric_output_node(
                  metric_spec=MetricSpec(
                      element_name=metric_input_spec.element_name,
                      filter_specs=tuple(filter_specs),
                      alias=metric_input_spec.alias,
                      offset_window=metric_input_spec.offset_window,
                      offset_to_grain=metric_input_spec.offset_to_grain,
                  ),
                  queried_linkable_specs=queried_linkable_specs
                  if not metric_spec.has_time_offset
                  else required_linkable_specs,
                  filter_spec_factory=filter_spec_factory,
                  time_range_constraint=time_range_constraint
                  if not metric_spec.has_time_offset
                  else None,
              )
          )
  
      parent_node = (
          parent_nodes[0]
          if len(parent_nodes) == 1
          else CombineAggregatedOutputsNode(parent_nodes=parent_nodes)
      )
      output_node: BaseOutput = ComputeMetricsNode(
          parent_node=parent_node, metric_specs=[metric_spec]
      )
  
      # For nested ratio / derived metrics with time offset, apply offset & where constraint after metric computation.
      if metric_spec.has_time_offset:
          assert (
              queried_linkable_specs.contains_metric_time
          ), "Joining to time spine requires querying with metric_time."
          output_node = JoinToTimeSpineNode(
              parent_node=output_node,
              requested_metric_time_dimension_specs=list(
                  queried_linkable_specs.metric_time_specs
              ),
              time_range_constraint=time_range_constraint,
              offset_window=metric_spec.offset_window,
              offset_to_grain=metric_spec.offset_to_grain,
              join_type=SqlJoinType.INNER,
          )
  
          if len(metric_spec.filter_specs) > 0:
              merged_where_filter = WhereFilterSpec.merge_iterable(
                  metric_spec.filter_specs
              )
              output_node = WhereConstraintNode(
                  parent_node=output_node,
                  where_constraint=merged_where_filter,
              )
          if not extraneous_linkable_specs.is_subset_of(
              queried_linkable_specs
          ):
              output_node = FilterElementsNode(
                  parent_node=output_node,
                  include_specs=InstanceSpecSet(
                      metric_specs=(metric_spec,)
                  ).merge(queried_linkable_specs.as_spec_set),
              )
      return output_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 7 for calling the DataflowPlanBuilder obj's underlying member method code with example usage: obj._build_derived_metric_output_node(metric_spec, queried_linkable_specs, filter_spec_factory, time_range_constraint) and returns: output_node