class_docstr: Processes source nodes before other nodes are joined.
  
  Generally, the source nodes will be combined with other dataflow plan nodes to produce a new set of nodes to realize
  a condition of the query. For example, to realize a time range constraint, a ConstrainTimeRangeNode will be added
  to the source nodes.
  
  e.g.
  
  <SomeDataflowPlanNode/>
  
  ->
  
  <ConstrainTimeRangeNode>
      <SomeDataflowPlanNode/>
  </ConstrainTimeRangeNode>
docstr: Assemble nodes representing all possible one-hop joins.
docstr_tok: ["Assemble","nodes","representing","all","possible","one-hop","joins","."]
code: def _get_candidates_nodes_for_multi_hop(
      self,
      desired_linkable_spec: LinkableInstanceSpec,
      nodes: Sequence[BaseOutput],
      join_type: SqlJoinType,
  ) -> Sequence[MultiHopJoinCandidate]:
      """Assemble nodes representing all possible one-hop joins."""
      if len(desired_linkable_spec.entity_links) > MAX_JOIN_HOPS:
          raise NotImplementedError(
              f"Multi-hop joins with more than {MAX_JOIN_HOPS} entity links not yet supported. "
              f"Got: {desired_linkable_spec}"
          )
      if len(desired_linkable_spec.entity_links) != 2:
          return ()
  
      multi_hop_join_candidates: List[MultiHopJoinCandidate] = []
      logger.info(f"Creating nodes for {desired_linkable_spec}")
  
      for first_node_that_could_be_joined in nodes:
          data_set_of_first_node_that_could_be_joined = (
              self._node_data_set_resolver.get_output_data_set(
                  first_node_that_could_be_joined
              )
          )
  
          # When joining on the entity, the first node needs the first and second entity links.
          if not (
              self._node_contains_entity(
                  node=first_node_that_could_be_joined,
                  entity_reference=desired_linkable_spec.entity_links[
                      0
                  ],
              )
              and self._node_contains_entity(
                  node=first_node_that_could_be_joined,
                  entity_reference=desired_linkable_spec.entity_links[
                      1
                  ],
              )
          ):
              continue
  
          for second_node_that_could_be_joined in nodes:
              if not (
                  self._node_contains_entity(
                      node=second_node_that_could_be_joined,
                      entity_reference=desired_linkable_spec.entity_links[
                          1
                      ],
                  )
              ):
                  continue
  
              # Avoid loops between the same semantic models.
              if (
                  second_node_that_could_be_joined.node_id
                  == first_node_that_could_be_joined.node_id
              ):
                  continue
  
              data_set_of_second_node_that_can_be_joined = (
                  self._node_data_set_resolver.get_output_data_set(
                      second_node_that_could_be_joined
                  )
              )
  
              # If the element name of the linkable spec doesn't exist in the joined data set, then it can't be
              # useful for obtaining that linkable spec.
              element_names_in_data_set = ToElementNameSet().transform(
                  data_set_of_second_node_that_can_be_joined.instance_set.spec_set
              )
  
              if (
                  desired_linkable_spec.element_name
                  not in element_names_in_data_set
              ):
                  continue
  
              # The first and second nodes are joined by this entity
              entity_reference_to_join_first_and_second_nodes = (
                  desired_linkable_spec.entity_links[1]
              )
  
              if not self._join_evaluator.is_valid_instance_set_join(
                  left_instance_set=data_set_of_first_node_that_could_be_joined.instance_set,
                  right_instance_set=data_set_of_second_node_that_can_be_joined.instance_set,
                  on_entity_reference=entity_reference_to_join_first_and_second_nodes,
              ):
                  continue
  
              # filter measures out of joinable_node
              specs = (
                  data_set_of_second_node_that_can_be_joined.instance_set.spec_set
              )
              filtered_joinable_node = FilterElementsNode(
                  parent_node=second_node_that_could_be_joined,
                  include_specs=InstanceSpecSet.from_specs(
                      specs.dimension_specs
                      + specs.entity_specs
                      + specs.time_dimension_specs
                  ),
              )
  
              join_on_partition_dimensions = self._partition_resolver.resolve_partition_dimension_joins(
                  start_node_spec_set=data_set_of_first_node_that_could_be_joined.instance_set.spec_set,
                  node_to_join_spec_set=data_set_of_second_node_that_can_be_joined.instance_set.spec_set,
              )
              join_on_partition_time_dimensions = self._partition_resolver.resolve_partition_time_dimension_joins(
                  start_node_spec_set=data_set_of_first_node_that_could_be_joined.instance_set.spec_set,
                  node_to_join_spec_set=data_set_of_second_node_that_can_be_joined.instance_set.spec_set,
              )
  
              multi_hop_join_candidates.append(
                  MultiHopJoinCandidate(
                      node_with_multi_hop_elements=JoinToBaseOutputNode(
                          left_node=first_node_that_could_be_joined,
                          join_targets=[
                              JoinDescription(
                                  join_node=filtered_joinable_node,
                                  join_on_entity=LinklessEntitySpec.from_reference(
                                      desired_linkable_spec.entity_links[
                                          1
                                      ]
                                  ),
                                  join_on_partition_dimensions=join_on_partition_dimensions,
                                  join_on_partition_time_dimensions=join_on_partition_time_dimensions,
                                  join_type=join_type,
                              )
                          ],
                      ),
                      lineage=MultiHopJoinCandidateLineage(
                          first_node_to_join=first_node_that_could_be_joined,
                          second_node_to_join=second_node_that_could_be_joined,
                          # entity_spec_in_first_node should already not have entity links since we checked
                          # for that, but using this method for type checking.
                          join_second_node_by_entity=LinklessEntitySpec.from_reference(
                              desired_linkable_spec.entity_links[1]
                          ),
                      ),
                  )
              )
  
      for multi_hop_join_candidate in multi_hop_join_candidates:
          output_data_set = (
              self._node_data_set_resolver.get_output_data_set(
                  multi_hop_join_candidate.node_with_multi_hop_elements
              )
          )
          logger.debug(
              f"Node {multi_hop_join_candidate.node_with_multi_hop_elements} has spec set:\n"
              f"{pformat_big_objects(output_data_set.instance_set.spec_set)}"
          )
  
      return multi_hop_join_candidates
  
imports: from __future__ import annotations
  import logging
  from dataclasses import dataclass
  from typing import List, Optional, Sequence, Set
  from dbt_semantic_interfaces.pretty_print import pformat_big_objects
  from dbt_semantic_interfaces.references import EntityReference, TimeDimensionReference
  from metricflow.dataflow.builder.node_data_set import DataflowPlanNodeOutputDataSetResolver
  from metricflow.dataflow.builder.partitions import PartitionJoinResolver
  from metricflow.dataflow.dataflow_plan import BaseOutput, ConstrainTimeRangeNode, FilterElementsNode, JoinDescription, JoinToBaseOutputNode
  from metricflow.filters.time_constraint import TimeRangeConstraint
  from metricflow.model.semantics.semantic_model_join_evaluator import MAX_JOIN_HOPS, SemanticModelJoinEvaluator
  from metricflow.protocols.semantics import SemanticModelAccessor
  from metricflow.specs.spec_set_transforms import ToElementNameSet
  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 PreJoinNodeProcessor obj's underlying member method code with example usage: obj._get_candidates_nodes_for_multi_hop(desired_linkable_spec, nodes, join_type) and returns: multi_hop_join_candidates, 