class_docstr: Helps to evaluate if linkable instances can be obtained using the given node, with joins if necessary.
  
  For example, consider a "start_node" containing the "bookings" measure, "is_instant" dimension, and "listing_id"
  entity with nodes_available_for_joins including a node with the "listing_id" entity, and the "country"
  dimension.
  
  We want to know if we can get "bookings", "is_instant", "listing_id__country" using the start_node. The result
  should be that we know: "is_instant" is available locally (i.e. in the same node), and if we join another node
  containing "listing_id" and "country" by "listing_id", we can get "listing_id__country".
  
  Since it's used on the left side of the join, the items in the "start_node" are sometimes labeled "left".
class_docstr_tok: ["Helps","to","evaluate","if","linkable","instances","can","be","obtained","using","the","given","node",",","with","joins","if","necessary",".","For","example",",","consider","a","``","start_node","''","containing","the","``","bookings","''","measure",",","``","is_instant","''","dimension",",","and","``","listing_id","''","entity","with","nodes_available_for_joins","including","a","node","with","the","``","listing_id","''","entity",",","and","the","``","country","''","dimension",".","We","want","to","know","if","we","can","get","``","bookings","''",",","``","is_instant","''",",","``","listing_id__country","''","using","the","start_node",".","The","result","should","be","that","we","know",":","``","is_instant","''","is","available","locally","(","i.e",".","in","the","same","node",")",",","and","if","we","join","another","node","containing","``","listing_id","''","and","``","country","''","by","``","listing_id","''",",","we","can","get","``","listing_id__country","''",".","Since","it","'s","used","on","the","left","side","of","the","join",",","the","items","in","the","``","start_node","''","are","sometimes","labeled","``","left","''","."]
docstr: Evaluates if the "required_linkable_specs" can be realized by joining the "start_node" with other nodes.
  
  In other words, given the data set associated with "start_node":
  
  * Can all "required_linkable_specs" be retrieved from the start_node? (These would be considered "local").
  * If not, can they be retrieved by joining an available node though a common entity?
  * If so, return all possible ways (by joining different nodes) that can be done.
docstr_tok: ["Evaluates","if","the","``","required_linkable_specs","''","can","be","realized","by","joining","the","``","start_node","''","with","other","nodes",".","In","other","words",",","given","the","data","set","associated","with","``","start_node","''",":","*","Can","all","``","required_linkable_specs","''","be","retrieved","from","the","start_node","?","(","These","would","be","considered","``","local","''",")",".","*","If","not",",","can","they","be","retrieved","by","joining","an","available","node","though","a","common","entity","?","*","If","so",",","return","all","possible","ways","(","by","joining","different","nodes",")","that","can","be","done","."]
code: def evaluate_node(
      self,
      start_node: BaseOutput,
      required_linkable_specs: Sequence[LinkableInstanceSpec],
      default_join_type: SqlJoinType,
  ) -> LinkableInstanceSatisfiabilityEvaluation:
      """Evaluates if the "required_linkable_specs" can be realized by joining the "start_node" with other nodes.
  
      In other words, given the data set associated with "start_node":
  
      * Can all "required_linkable_specs" be retrieved from the start_node? (These would be considered "local").
      * If not, can they be retrieved by joining an available node though a common entity?
      * If so, return all possible ways (by joining different nodes) that can be done.
      """
      candidate_instance_set: InstanceSet = (
          self._node_data_set_resolver.get_output_data_set(
              start_node
          ).instance_set
      )
      candidate_spec_set = candidate_instance_set.spec_set
  
      logger.debug(
          f"Candidate spec set is:\n{pformat_big_objects(candidate_spec_set)}"
      )
  
      data_set_linkable_specs = candidate_spec_set.linkable_specs
  
      # These are linkable specs in the same data set as the measure. Those are considered "local".
      local_linkable_specs = []
  
      # These are linkable specs that aren't in the data set, but they might be able to be joined in.
      possibly_joinable_linkable_specs = []
  
      # Group required_linkable_specs into local / un-joinable / or possibly joinable.
      unjoinable_linkable_specs = []
      for required_linkable_spec in required_linkable_specs:
          is_local = required_linkable_spec in data_set_linkable_specs
          is_unjoinable = (
              len(required_linkable_spec.entity_links) == 0
              or LinklessEntitySpec.from_reference(
                  required_linkable_spec.entity_links[0]
              )
              not in data_set_linkable_specs
          )
          if is_local:
              local_linkable_specs.append(required_linkable_spec)
          elif is_unjoinable:
              unjoinable_linkable_specs.append(required_linkable_spec)
          else:
              possibly_joinable_linkable_specs.append(
                  required_linkable_spec
              )
  
      candidates_for_join = self._find_joinable_candidate_nodes_that_can_satisfy_linkable_specs(
          start_node_instance_set=candidate_instance_set,
          needed_linkable_specs=possibly_joinable_linkable_specs,
          join_type=default_join_type,
      )
      join_candidates: List[JoinLinkableInstancesRecipe] = []
  
      logger.info(
          "Looping over nodes that can be joined to get the required linkable specs"
      )
  
      # Using a greedy approach, try to get the "possibly_joinable_linkable_specs" by iteratively joining nodes with
      # the most matching linkable specs. We try to join nodes with the most matching specs to minimize the number of
      # joins that we have to do to. A knapsack solution is ideal, but punting on that for simplicity.
      while len(possibly_joinable_linkable_specs) > 0:
          logger.info(
              f"Looking for linkable specs:\n{pformat_big_objects(possibly_joinable_linkable_specs)}"
          )
  
          # We've run out of candidate data sets, but there are more linkable specs that we need. That means the
          # rest of the linkable specs can't be joined in, and we're left with unjoinable specs remaining.
          if len(candidates_for_join) == 0:
              logger.info(
                  "There are no more candidate nodes that can be joined, but not all linkable specs have "
                  "been acquired."
              )
              unjoinable_linkable_specs.extend(
                  possibly_joinable_linkable_specs
              )
              break
  
          # Join the best candidate to realize the linkable specs
          next_candidate = candidates_for_join.pop(0)
          logger.info(
              f"The next candidate node to be joined is:\n{pformat_big_objects(next_candidate)}"
          )
          join_candidates.append(next_candidate)
  
          # Update the candidates. Since we'll be joined/ing the previously selected candidate, we no longer need
          # to get those linkable specs from the remaining candidates, and we also need to re-rank based on the new
          # requirements.
          candidates_for_join = self._update_candidates_that_can_satisfy_linkable_specs(
              candidates_for_join=candidates_for_join,
              already_satisfisfied_linkable_specs=next_candidate.satisfiable_linkable_specs,
              join_type=default_join_type,
          )
  
          # The once possibly joinable specs are definitely joinable and no longer need to be searched for.
          # Remove from "possibly_joinable_linkable_specs"
          possibly_joinable_linkable_specs = [
              x
              for x in possibly_joinable_linkable_specs
              if x not in next_candidate.satisfiable_linkable_specs
          ]
  
      logger.info("Done evaluating possible joins")
      return LinkableInstanceSatisfiabilityEvaluation(
          local_linkable_specs=tuple(local_linkable_specs),
          joinable_linkable_specs=tuple(
              itertools.chain.from_iterable(
                  [
                      x.satisfiable_linkable_specs
                      for x in join_candidates
                  ]
              )
          ),
          join_recipes=tuple(join_candidates),
          unjoinable_linkable_specs=tuple(unjoinable_linkable_specs),
      )
  
imports: from __future__ import annotations
  import itertools
  import logging
  from dataclasses import dataclass
  from typing import List, Optional, Sequence, Tuple
  from dbt_semantic_interfaces.pretty_print import pformat_big_objects
  from metricflow.dataflow.builder.node_data_set import DataflowPlanNodeOutputDataSetResolver
  from metricflow.dataflow.builder.partitions import PartitionJoinResolver
  from metricflow.dataflow.dataflow_plan import BaseOutput, FilterElementsNode, JoinDescription, PartitionDimensionJoinDescription, PartitionTimeDimensionJoinDescription, ValidityWindowJoinDescription
  from metricflow.dataset.sql_dataset import SqlDataSet
  from metricflow.instances import InstanceSet
  from metricflow.model.semantics.semantic_model_join_evaluator import SemanticModelJoinEvaluator
  from metricflow.plan_conversion.instance_converters import CreateValidityWindowJoinDescription
  from metricflow.protocols.semantics import SemanticModelAccessor
  from metricflow.specs.specs import InstanceSpecSet, LinkableInstanceSpec, LinklessEntitySpec
  from metricflow.sql.sql_plan import SqlJoinType
label_desc: Use image node_id 4 for calling the NodeEvaluatorForLinkableInstances obj's underlying member method code with example usage: obj.evaluate_node(start_node, required_linkable_specs, default_join_type) and returns: LinkableInstanceSatisfiabilityEvaluation