code: def cyclic_join_manifest_dataflow_plan_builder(  # noqa: D
      cyclic_join_semantic_manifest_lookup: SemanticManifestLookup,
      consistent_id_object_repository: ConsistentIdObjectRepository,
  ) -> DataflowPlanBuilder:
      for (
          source_node
      ) in consistent_id_object_repository.cyclic_join_source_nodes:
          logger.error(f"Source node is: {source_node}")
  
      return DataflowPlanBuilder(
          source_nodes=consistent_id_object_repository.cyclic_join_source_nodes,
          read_nodes=list(
              consistent_id_object_repository.cyclic_join_read_nodes.values()
          ),
          semantic_manifest_lookup=cyclic_join_semantic_manifest_lookup,
      )
  
file_path: metricflow/metricflow/test/dataflow/builder/test_cyclic_join.py
imports: from __future__ import annotations
  import logging
  import pytest
  from _pytest.fixtures import FixtureRequest
  from dbt_semantic_interfaces.references import EntityReference
  from metricflow.dataflow.builder.dataflow_plan_builder import DataflowPlanBuilder
  from metricflow.dataflow.dataflow_plan_to_text import dataflow_plan_as_text
  from metricflow.model.semantic_manifest_lookup import SemanticManifestLookup
  from metricflow.specs.specs import DimensionSpec, MetricFlowQuerySpec, MetricSpec
  from metricflow.test.dataflow_plan_to_svg import display_graph_if_requested
  from metricflow.test.fixtures.model_fixtures import ConsistentIdObjectRepository
  from metricflow.test.fixtures.setup_fixtures import MetricFlowTestSessionState
  from metricflow.test.fixtures.sql_client_fixtures import sql_client
  from metricflow.test.snapshot_utils import assert_plan_snapshot_text_equal
label_desc: Use image node_id 1 for calling a global function with example usage: cyclic_join_manifest_dataflow_plan_builder(cyclic_join_semantic_manifest_lookup, consistent_id_object_repository) and returns: DataflowPlanBuilder