code: def fuse_splits_into_multiple_return(dsk, keys):
      if not isinstance(dsk, HighLevelGraph):
          dsk = HighLevelGraph.from_collections(
              id(dsk), dsk, dependencies=()
          )
      else:
          dsk = dsk.copy()
      dependencies = dsk.dependencies.copy()
      for k, v in dsk.items():
          if istask(v) and v[0] == shuffle_group:
              task_deps = dependencies[k]
              # Only rewrite shuffle group split if all downstream dependencies
              # are splits.
              if all(
                  istask(dsk[dep]) and dsk[dep][0] == operator.getitem
                  for dep in task_deps
              ):
                  for dep in task_deps:
                      # Rewrite split
                      pass
  
file_path: ray/python/ray/util/dask/optimizations.py
imports: import operator
  import warnings
  import dask
  from dask import core
  from dask.core import istask
  from dask.dataframe.core import _concat
  from dask.dataframe.optimize import optimize
  from dask.dataframe.shuffle import shuffle_group
  from dask.highlevelgraph import HighLevelGraph
  from .scheduler import MultipleReturnFunc, multiple_return_get
label_desc: Use image node_id 1 for calling a global function with example usage: fuse_splits_into_multiple_return(dsk, keys) without return types