Hard time trying to understand the traceability of tasks on the flow graph

[hugocatlas]

Hello Prefect team,

I'm having a hard time trying to understand how to "force" tasks to be linked together on the flow graph.

Take a look at the following flow snippet:

@flow(
    log_prints=True,
    task_runner=RayTaskRunner(),
    on_failure=[WorkflowTools.notify_slack, WorkflowTools.notify_mail],
)
def merge_parquets_on_s3(workflow: Dict[str, Any]) -> None:
    workflow_tools = quote(WorkflowTools(workflow=workflow))

    (
        contents2merge,
        contents_already_merged,
        contents_successes,
    ) = DataGovernanceTasks.get_content_from_s3(workflow_tools=workflow_tools)

    if not contents2merge + contents_already_merged + contents_successes:
        workflow_tools[0].log_info(message="Nothing to do. Leaving...")

    if queued_items := DataGovernanceTasks.get_items_awaiting_merging(
        workflow_tools=workflow_tools, bucket_contents=contents2merge
    ):
        statuses = [
            DataGovernanceTasks.merge_parquet_files_in_path.submit(
                workflow_tools=workflow_tools,
                queued_item=queued_item,
            )
            for queued_item in queued_items
        ]
...

The DataGovernanceTasks.get_content_from_s3 task returns a tuple containing three lists that are further consumed by the DataGovernanceTasks.get_items_awaiting_merging task. The latter returns a dictionary that is further consumed by DataGovernanceTasks.merge_parquet_files_in_path task.

On the flow graph, I have the following:

As you can see, get_items_awaiting_merging is correctly displayed and linked to the downstream tasks. However, this doesn't happen to get_content_from_s3 task.

In other situations, I have found my way around "forcing" these connections. But in this case, I really don't understand what is blocking.

Could you please clarify what it takes so that the flow graph correctly displays the tasks dependencies?

Cheers,

Hugo