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_plan_for_distinct_values(
      self, query_spec: MetricFlowQuerySpec
  ) -> DataflowPlan:
      assert (
          not query_spec.metric_specs
      ), "Can't build distinct values plan with metrics."
      query_level_filter_specs: Sequence[WhereFilterSpec] = ()
      if (
          query_spec.filter_intersection is not None
          and len(query_spec.filter_intersection.where_filters) > 0
      ):
          filter_spec_factory = WhereSpecFactory(
              column_association_resolver=self._column_association_resolver,
              spec_resolution_lookup=query_spec.filter_spec_resolution_lookup
              or FilterSpecResolutionLookUp.empty_instance(),
          )
  
          query_level_filter_specs = (
              filter_spec_factory.create_from_where_filter_intersection(
                  filter_location=WhereFilterLocation.for_query(
                      tuple(
                          metric_spec.reference
                          for metric_spec in query_spec.metric_specs
                      )
                  ),
                  filter_intersection=query_spec.filter_intersection,
              )
          )
  
      (
          required_linkable_specs,
          _,
      ) = self.__get_required_and_extraneous_linkable_specs(
          queried_linkable_specs=query_spec.linkable_specs,
          filter_specs=query_level_filter_specs,
      )
      dataflow_recipe = self._find_dataflow_recipe(
          linkable_spec_set=required_linkable_specs
      )
      if not dataflow_recipe:
          raise UnableToSatisfyQueryError(
              f"Recipe not found for linkable specs: {required_linkable_specs}"
          )
  
      joined_node: Optional[JoinToBaseOutputNode] = None
      if dataflow_recipe.join_targets:
          joined_node = JoinToBaseOutputNode(
              left_node=dataflow_recipe.source_node,
              join_targets=dataflow_recipe.join_targets,
          )
  
      where_constraint_node: Optional[WhereConstraintNode] = None
      if len(query_level_filter_specs) > 0:
          where_constraint_node = WhereConstraintNode(
              parent_node=joined_node or dataflow_recipe.source_node,
              where_constraint=WhereFilterSpec.merge_iterable(
                  query_level_filter_specs
              ),
          )
  
      distinct_values_node = FilterElementsNode(
          parent_node=where_constraint_node
          or joined_node
          or dataflow_recipe.source_node,
          include_specs=query_spec.linkable_specs.as_spec_set,
          distinct=True,
      )
  
      min_max_node: Optional[MinMaxNode] = None
      if query_spec.min_max_only:
          min_max_node = MinMaxNode(parent_node=distinct_values_node)
  
      sink_node = self.build_sink_node(
          parent_node=min_max_node or distinct_values_node,
          order_by_specs=query_spec.order_by_specs,
          limit=query_spec.limit,
      )
  
      plan_id = IdGeneratorRegistry.for_class(
          DataflowPlanBuilder
      ).create_id(DATAFLOW_PLAN_PREFIX)
  
      return DataflowPlan(
          plan_id=plan_id, sink_output_nodes=[sink_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 11 for calling the DataflowPlanBuilder obj's underlying member method code with example usage: obj._build_plan_for_distinct_values(query_spec) and returns: DataflowPlan