code: def split_results(results):
      
      from dask.delayed import Delayed
  
      def flatten(results):
          out = []
          if isinstance(results, (list, tuple)):
              for itm in results:
                  out.extend(flatten(itm))
              return out
          return [results]
  
      sources = []
      targets = []
      delayeds = []
  
      for res in flatten(results):
          if isinstance(res, da.Array):
              sources.append(res)
          elif isinstance(res, Delayed):
              delayeds.append(res)
          else:
              targets.append(res)
      return sources, targets, delayeds
docs: Get sources, targets and delayed objects to separate lists from a
      list of results collected from (multiple) writer(s).