class_docstr: Builds a dataflow plan to satisfy a given query.
class_docstr_tok: ["Builds","a","dataflow","plan","to","satisfy","a","given","query","."]
docstr: Return the input measure spec required to compute the base metric.
  
  "child" refers to the derived metric that uses the metric specified by metric_reference in the definition.
  descendent_filter_specs includes all filter specs required to compute the metric in the query. This includes the
  filters in the query and any filter in the definition of metrics in between.
docstr_tok: ["Return","the","input","measure","spec","required","to","compute","the","base","metric",".","``","child","''","refers","to","the","derived","metric","that","uses","the","metric","specified","by","metric_reference","in","the","definition",".","descendent_filter_specs","includes","all","filter","specs","required","to","compute","the","metric","in","the","query",".","This","includes","the","filters","in","the","query","and","any","filter","in","the","definition","of","metrics","in","between","."]
code: def _build_input_measure_spec_for_base_metric(
      self,
      filter_spec_factory: WhereSpecFactory,
      metric_reference: MetricReference,
      child_metric_offset_window: Optional[MetricTimeWindow],
      child_metric_offset_to_grain: Optional[TimeGranularity],
      descendent_filter_specs: Sequence[WhereFilterSpec],
      query_contains_metric_time: bool,
      culmination_description: Optional[CumulativeMeasureDescription],
  ) -> MetricInputMeasureSpec:
      """Return the input measure spec required to compute the base metric.
  
      "child" refers to the derived metric that uses the metric specified by metric_reference in the definition.
      descendent_filter_specs includes all filter specs required to compute the metric in the query. This includes the
      filters in the query and any filter in the definition of metrics in between.
      """
      metric = self._metric_lookup.get_metric(metric_reference)
  
      if (
          metric.type is MetricType.SIMPLE
          or metric.type is MetricType.CUMULATIVE
      ):
          pass
      elif (
          metric.type is MetricType.RATIO
          or metric.type is MetricType.DERIVED
          or metric.type is MetricType.CONVERSION
      ):
          raise ValueError(
              "This should only be called for base metrics."
          )
      else:
          assert_values_exhausted(metric.type)
  
      assert (
          len(metric.input_measures) == 1
      ), f"A base metric should not have multiple measures. Got{metric.input_measures}"
  
      input_measure = metric.input_measures[0]
  
      measure_spec = MeasureSpec(
          element_name=input_measure.name,
          non_additive_dimension_spec=self._semantic_model_lookup.non_additive_dimension_specs_by_measure.get(
              input_measure.measure_reference
          ),
      )
  
      before_aggregation_time_spine_join_description = None
      # If querying an offset metric, join to time spine.
      if (
          child_metric_offset_window is not None
          or child_metric_offset_to_grain is not None
      ):
          before_aggregation_time_spine_join_description = (
              JoinToTimeSpineDescription(
                  join_type=SqlJoinType.INNER,
                  offset_window=child_metric_offset_window,
                  offset_to_grain=child_metric_offset_to_grain,
              )
          )
  
      # Even if the measure is configured to join to time spine, if there's no metric_time in the query,
      # there's no need to join to the time spine since all metric_time will be aggregated.
      after_aggregation_time_spine_join_description = None
      if input_measure.join_to_timespine and query_contains_metric_time:
          after_aggregation_time_spine_join_description = (
              JoinToTimeSpineDescription(
                  join_type=SqlJoinType.LEFT_OUTER,
                  offset_window=None,
                  offset_to_grain=None,
              )
          )
  
      filter_specs: List[WhereFilterSpec] = []
      filter_specs.extend(
          filter_spec_factory.create_from_where_filter_intersection(
              filter_location=WhereFilterLocation.for_metric(
                  metric_reference
              ),
              filter_intersection=input_measure.filter,
          )
      )
      filter_specs.extend(
          filter_spec_factory.create_from_where_filter_intersection(
              filter_location=WhereFilterLocation.for_metric(
                  metric_reference
              ),
              filter_intersection=metric.filter,
          )
      )
      filter_specs.extend(descendent_filter_specs)
  
      return MetricInputMeasureSpec(
          measure_spec=measure_spec,
          fill_nulls_with=input_measure.fill_nulls_with,
          offset_window=child_metric_offset_window,
          offset_to_grain=child_metric_offset_to_grain,
          culmination_description=culmination_description,
          filter_specs=tuple(filter_specs),
          alias=input_measure.alias,
          before_aggregation_time_spine_join_description=before_aggregation_time_spine_join_description,
          after_aggregation_time_spine_join_description=after_aggregation_time_spine_join_description,
      )
  
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 23 for calling the DataflowPlanBuilder obj's underlying member method code with example usage: obj._build_input_measure_spec_for_base_metric(filter_spec_factory, metric_reference, child_metric_offset_window, child_metric_offset_to_grain, descendent_filter_specs, query_contains_metric_time, culmination_description) and returns: MetricInputMeasureSpec