def METHOD_NAME(  # noqa: D
    request: FixtureRequest,
    mf_test_session_state: MetricFlowTestSessionState,
    consistent_id_object_repository: ConsistentIdObjectRepository,
    simple_semantic_manifest_lookup: SemanticManifestLookup,
    time_spine_source: TimeSpineSource,
) -> None:
    """Tests getting the data set from a dataflow plan with a join."""
    resolver: DataflowPlanNodeOutputDataSetResolver = DataflowPlanNodeOutputDataSetResolver(
        column_association_resolver=DunderColumnAssociationResolver(simple_semantic_manifest_lookup),
        semantic_manifest_lookup=simple_semantic_manifest_lookup,
    )
    # Join "revenue" with "users_latest" to get "user__home_state_latest"
    revenue_node = consistent_id_object_repository.simple_model_read_nodes["revenue"]
    users_node = consistent_id_object_repository.simple_model_read_nodes["users_latest"]
    join_node = JoinToBaseOutputNode(
        left_node=revenue_node,
        join_targets=[
            JoinDescription(
                join_node=users_node,
                join_on_entity=LinklessEntitySpec.from_element_name("user"),
                join_on_partition_dimensions=(),
                join_on_partition_time_dimensions=(),
            )
        ],
    )
    join_node_output_data_set = resolver.get_output_data_set(join_node)
    assert_spec_set_snapshot_equal(
        request=request,
        mf_test_session_state=mf_test_session_state,
        set_id="result0",
        spec_set=join_node_output_data_set.instance_set.spec_set,
    )