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 _find_dataflow_recipe(
      self,
      linkable_spec_set: LinkableSpecSet,
      measure_spec_properties: Optional[MeasureSpecProperties] = None,
      time_range_constraint: Optional[TimeRangeConstraint] = None,
  ) -> Optional[DataflowRecipe]:
      linkable_specs = linkable_spec_set.as_tuple
      if measure_spec_properties:
          source_nodes = self._source_nodes
          potential_source_nodes: Sequence[
              BaseOutput
          ] = self._select_source_nodes_with_measures(
              measure_specs=set(measure_spec_properties.measure_specs),
              source_nodes=source_nodes,
          )
          default_join_type = SqlJoinType.LEFT_OUTER
      else:
          # Only read nodes can be source nodes for queries without measures
          source_nodes = self._read_nodes
          source_nodes_to_linkable_specs = (
              self._select_read_nodes_with_linkable_specs(
                  linkable_specs=linkable_spec_set,
                  read_nodes=source_nodes,
              )
          )
          potential_source_nodes = list(
              source_nodes_to_linkable_specs.keys()
          )
          default_join_type = SqlJoinType.FULL_OUTER
  
      logger.info(
          f"There are {len(potential_source_nodes)} potential source nodes"
      )
  
      logger.info(
          f"Starting search with {len(source_nodes)} source nodes"
      )
      start_time = time.time()
  
      node_processor = PreJoinNodeProcessor(
          semantic_model_lookup=self._semantic_model_lookup,
          node_data_set_resolver=self._node_data_set_resolver,
      )
      if time_range_constraint:
          potential_source_nodes = node_processor.add_time_range_constraint(
              source_nodes=potential_source_nodes,
              metric_time_dimension_reference=self._metric_time_dimension_reference,
              time_range_constraint=time_range_constraint,
          )
  
      nodes_available_for_joins = node_processor.remove_unnecessary_nodes(
          desired_linkable_specs=linkable_specs,
          nodes=source_nodes,
          metric_time_dimension_reference=self._metric_time_dimension_reference,
      )
      logger.info(
          f"After removing unnecessary nodes, there are {len(nodes_available_for_joins)} nodes available for joins"
      )
      if DataflowPlanBuilder._contains_multihop_linkables(
          linkable_specs
      ):
          nodes_available_for_joins = (
              node_processor.add_multi_hop_joins(
                  desired_linkable_specs=linkable_specs,
                  nodes=source_nodes,
                  join_type=default_join_type,
              )
          )
          logger.info(
              f"After adding multi-hop nodes, there are {len(nodes_available_for_joins)} nodes available for joins:\n"
              f"{pformat_big_objects(nodes_available_for_joins)}"
          )
      logger.info(
          f"Processing nodes took: {time.time()-start_time:.2f}s"
      )
  
      node_evaluator = NodeEvaluatorForLinkableInstances(
          semantic_model_lookup=self._semantic_model_lookup,
          nodes_available_for_joins=self._sort_by_suitability(
              nodes_available_for_joins
          ),
          node_data_set_resolver=self._node_data_set_resolver,
      )
  
      # Dict from the node that contains the source node to the evaluation results.
      node_to_evaluation: Dict[
          BaseOutput, LinkableInstanceSatisfiabilityEvaluation
      ] = {}
  
      for node in self._sort_by_suitability(potential_source_nodes):
          data_set = self._node_data_set_resolver.get_output_data_set(
              node
          )
  
          if measure_spec_properties:
              measure_specs = measure_spec_properties.measure_specs
              missing_specs = [
                  spec
                  for spec in measure_specs
                  if spec
                  not in data_set.instance_set.spec_set.measure_specs
              ]
              if missing_specs:
                  logger.debug(
                      f"Skipping evaluation for node since it does not have all of the measure specs {missing_specs}:"
                      f"\n\n{dataflow_dag_as_text(node)}"
                  )
                  continue
  
          logger.debug(
              f"Evaluating source node:\n{pformat_big_objects(source_node=dataflow_dag_as_text(node))}"
          )
  
          start_time = time.time()
          evaluation = node_evaluator.evaluate_node(
              start_node=node,
              required_linkable_specs=list(linkable_specs),
              default_join_type=default_join_type,
          )
          logger.info(
              f"Evaluation of {node} took {time.time() - start_time:.2f}s"
          )
  
          logger.debug(
              f"Evaluation for source node is:\n"
              f"{pformat_big_objects(node=dataflow_dag_as_text(node), evaluation=evaluation)}"
          )
  
          if len(evaluation.unjoinable_linkable_specs) > 0:
              logger.info(
                  f"Skipping {node.node_id} since it contains un-joinable specs: "
                  f"{evaluation.unjoinable_linkable_specs}"
              )
              continue
  
          num_joins_required = len(evaluation.join_recipes)
          logger.info(
              f"Found candidate with node ID '{node.node_id}' with {num_joins_required} joins required."
          )
  
          node_to_evaluation[node] = evaluation
  
          # Since are evaluating nodes with the lowest cost first, if we find one without requiring any joins, then
          # this is going to be the lowest cost solution.
          if len(evaluation.join_recipes) == 0:
              logger.info(
                  "Not evaluating other nodes since we found one that doesn't require joins"
              )
              break
  
      logger.info(
          f"Found {len(node_to_evaluation)} candidate source nodes."
      )
  
      if len(node_to_evaluation) > 0:
          # Find evaluation with lowest number of joins.
          node_with_lowest_cost_plan = min(
              node_to_evaluation,
              key=lambda node: len(
                  node_to_evaluation[node].join_recipes
              ),
          )
          evaluation = node_to_evaluation[node_with_lowest_cost_plan]
          logger.info(
              "Lowest cost plan is:\n"
              + pformat_big_objects(
                  node=dataflow_dag_as_text(node_with_lowest_cost_plan),
                  evaluation=evaluation,
                  joins=len(
                      node_to_evaluation[
                          node_with_lowest_cost_plan
                      ].join_recipes
                  ),
              )
          )
  
          # Nodes containing the linkable instances will be joined to the source node, so these
          # entities will need to be present in the source node.
          required_local_entity_specs = tuple(
              x.join_on_entity for x in evaluation.join_recipes
          )
          # Same thing with partitions.
          required_local_dimension_specs = tuple(
              y.start_node_dimension_spec
              for x in evaluation.join_recipes
              for y in x.join_on_partition_dimensions
          )
          required_local_time_dimension_specs = tuple(
              y.start_node_time_dimension_spec
              for x in evaluation.join_recipes
              for y in x.join_on_partition_time_dimensions
          )
          return DataflowRecipe(
              source_node=node_with_lowest_cost_plan,
              required_local_linkable_specs=(
                  evaluation.local_linkable_specs
                  + required_local_entity_specs
                  + required_local_dimension_specs
                  + required_local_time_dimension_specs
              ),
              join_linkable_instances_recipes=node_to_evaluation[
                  node_with_lowest_cost_plan
              ].join_recipes,
          )
  
      logger.error("No recipe could be constructed.")
      return None
  
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 20 for calling the DataflowPlanBuilder obj's underlying member method code with example usage: obj._find_dataflow_recipe(linkable_spec_set, measure_spec_properties, time_range_constraint) and returns: None, DataflowRecipe