code: def test_dedupe_serialization_dag(workflow_start_regular_shared):
      from ray.workflow import serialization
      from ray.workflow.tests.utils import skip_client_mode_test
  
      # TODO(suquark): Fix workflow with ObjectRefs as inputs under client mode.
      skip_client_mode_test()
  
      @ray.remote
      def identity(x):
          return x
  
      @ray.remote
      def gather(*args):
          return args
  
      def get_num_uploads():
          manager = serialization.get_or_create_manager()
          stats = ray.get(manager.export_stats.remote())
          return stats.get("num_uploads", 0)
  
      ref = ray.put("hello world 12345")
      list_of_refs = [ref for _ in range(20)]
  
      assert get_num_uploads() == 0
  
      single = identity.bind((ref,))
      double = identity.bind(list_of_refs)
  
      result_ref, result_list = workflow.run(
          gather.bind(single, double)
      )
  
      for result in result_list:
          assert ray.get(*result_ref) == ray.get(result)
  
      # One upload for the initial checkpoint.
      assert get_num_uploads() == 1
  
file_path: ray/python/ray/workflow/tests/test_dag_to_workflow.py
imports: from ray.tests.conftest import *
  import pytest
  import ray
  from ray import workflow
  from ray.dag import InputNode
label_desc: Use image node_id 2 for calling a global function with example usage: test_dedupe_serialization_dag(workflow_start_regular_shared) and returns: x, args, stats