code: def test_checkpoint_dag_skip_partial(workflow_start_regular_shared):
      outputs = workflow.run(
          checkpoint_dag.options(
              **workflow.options(task_id="checkpoint_dag")
          ).bind(False),
          workflow_id="checkpoint_partial",
      )
      assert np.isclose(outputs, 8388607.5)
      recovered = workflow.resume("checkpoint_partial")
      assert np.isclose(recovered, 8388607.5)
  
      @client_mode_wrap
      def check():
          wf_storage = workflow_storage.WorkflowStorage(
              "checkpoint_partial"
          )
          assert_task_checkpoints(
              wf_storage, "checkpoint_dag", mode="checkpointed"
          )
          assert_task_checkpoints(
              wf_storage, "large_input", mode="output_skipped"
          )
          assert_task_checkpoints(
              wf_storage, "identity", mode="output_skipped"
          )
          assert_task_checkpoints(
              wf_storage, "average", mode="checkpointed"
          )
  
      check()
  
file_path: ray/python/ray/workflow/tests/test_checkpoint.py
imports: import ray
  import pytest
  from ray.tests.conftest import *
  import numpy
  from ray import workflow
  from ray._private.client_mode_hook import client_mode_wrap
  from ray.workflow import workflow_storage
  from ray.workflow.tests.utils import assert_task_checkpoints
label_desc: Use image node_id 3 for calling a global function with example usage: test_checkpoint_dag_skip_partial(workflow_start_regular_shared) without return types