python/ray/workflow/step_executor.py
def _resolve_dynamic_workflow_refs(workflow_refs: "List[WorkflowRef]"):
    """Get the output of a workflow step with the step ID at runtime.

    We lookup the output by the following order:
    1. Query cached step output in the workflow manager. Fetch the physical
       output object.
    2. If failed to fetch the physical output object, look into the storage
       to see whether the output is checkpointed. Load the checkpoint.
    3. If failed to load the checkpoint, resume the step and get the output.
    """
    workflow_manager = get_or_create_management_actor()
    context = workflow_context.get_workflow_step_context()
    workflow_id = context.workflow_id
    storage_url = context.storage_url
    workflow_ref_mapping = []
    for workflow_ref in workflow_refs:
        step_ref = ray.get(
            workflow_manager.get_cached_step_output.remote(
                workflow_id, workflow_ref.step_id
            )
        )
        get_cached_step = False
        if step_ref is not None:
            try:
                output = _resolve_static_workflow_ref(step_ref)
                get_cached_step = True
            except Exception:
                get_cached_step = False
        if not get_cached_step:
            wf_store = workflow_storage.get_workflow_storage()
            try:
                output = wf_store.load_step_output(workflow_ref.step_id)
            except Exception:
                current_step_id = workflow_context.get_current_step_id()
                logger.warning(
                    "Failed to get the output of step "
                    f"{workflow_ref.step_id}. Trying to resume it. "
                    f"Current step: '{current_step_id}'"
                )
                step_ref = recovery.resume_workflow_step(
                    workflow_id, workflow_ref.step_id, storage_url, None
                ).persisted_output
                output = _resolve_static_workflow_ref(step_ref)
        workflow_ref_mapping.append(output)
    return workflow_ref_mapping
Get the output of a workflow step with the step ID at runtime.

    We lookup the output by the following order:
    1. Query cached step output in the workflow manager. Fetch the physical
       output object.
    2. If failed to fetch the physical output object, look into the storage
       to see whether the output is checkpointed. Load the checkpoint.
    3. If failed to load the checkpoint, resume the step and get the output.
    
Get the output of a workflow step with the step ID at runtime.
We lookup the output by the following order:
1. Query cached step output in the workflow manager. Fetch the physical
output object.
2. If failed to fetch the physical output object, look into the storage
to see whether the output is checkpointed. Load the checkpoint.
3. If failed to load the checkpoint, resume the step and get the output.
Get the output of a workflow step with the step ID at runtime . We lookup the output by the following order : 1 . Query cached step output in the workflow manager . Fetch the physical output object . 2 . If failed to fetch the physical output object look into the storage to see whether the output is checkpointed . Load the checkpoint . 3 . If failed to load the checkpoint resume the step and get the output .
def _resolve_dynamic_workflow_refs(workflow_refs: "List[WorkflowRef]"):
    workflow_manager = get_or_create_management_actor()
    context = workflow_context.get_workflow_step_context()
    workflow_id = context.workflow_id
    storage_url = context.storage_url
    workflow_ref_mapping = []
    for workflow_ref in workflow_refs:
        step_ref = ray.get(
            workflow_manager.get_cached_step_output.remote(
                workflow_id, workflow_ref.step_id
            )
        )
        get_cached_step = False
        if step_ref is not None:
            try:
                output = _resolve_static_workflow_ref(step_ref)
                get_cached_step = True
            except Exception:
                get_cached_step = False
        if not get_cached_step:
            wf_store = workflow_storage.get_workflow_storage()
            try:
                output = wf_store.load_step_output(workflow_ref.step_id)
            except Exception:
                current_step_id = workflow_context.get_current_step_id()
                logger.warning(
                    "Failed to get the output of step "
                    f"{workflow_ref.step_id}. Trying to resume it. "
                    f"Current step: '{current_step_id}'"
                )
                step_ref = recovery.resume_workflow_step(
                    workflow_id, workflow_ref.step_id, storage_url, None
                ).persisted_output
                output = _resolve_static_workflow_ref(step_ref)
        workflow_ref_mapping.append(output)
    return workflow_ref_mapping
def _resolve_dynamic_workflow_refs ( workflow_refs : "List[WorkflowRef]" ) : workflow_manager = get_or_create_management_actor ( ) context = workflow_context . get_workflow_step_context ( ) workflow_id = context . workflow_id storage_url = context . storage_url workflow_ref_mapping = [ ] for workflow_ref in workflow_refs : step_ref = ray . get ( workflow_manager . get_cached_step_output . remote ( workflow_id , workflow_ref . step_id ) ) get_cached_step = False if step_ref is not None : try : output = _resolve_static_workflow_ref ( step_ref ) get_cached_step = True except Exception : get_cached_step = False if not get_cached_step : wf_store = workflow_storage . get_workflow_storage ( ) try : output = wf_store . load_step_output ( workflow_ref . step_id ) except Exception : current_step_id = workflow_context . get_current_step_id ( ) logger . warning ( "Failed to get the output of step " f"{workflow_ref.step_id}. Trying to resume it. " f"Current step: '{current_step_id}'" ) step_ref = recovery . resume_workflow_step ( workflow_id , workflow_ref . step_id , storage_url , None ) . persisted_output output = _resolve_static_workflow_ref ( step_ref ) workflow_ref_mapping . append ( output ) return workflow_ref_mapping
Get the output of a workflow step with the step ID at runtime.
Get the output of a workflow step with the step ID at runtime .
"""Get the output of a workflow step with the step ID at runtime.

    We lookup the output by the following order:
    1. Query cached step output in the workflow manager. Fetch the physical
       output object.
    2. If failed to fetch the physical output object, look into the storage
       to see whether the output is checkpointed. Load the checkpoint.
    3. If failed to load the checkpoint, resume the step and get the output.
    """
workflow_refs
"List[WorkflowRef]"