code: def _generate_global_split_results(
      all_blocks_split_results: Iterable[
          Tuple[ObjectRef[Block], BlockMetadata]
      ],
      global_split_sizes: List[int],
  ) -> Tuple[List[List[ObjectRef[Block]]], List[List[BlockMetadata]]]:
      """Reassemble per block's split result into final split result."""
      result_blocks = []
      result_metas = []
  
      current_blocks = []
      current_meta = []
      current_split_size = 0
      current_split_id = 0
  
      while current_split_id < len(global_split_sizes):
          if current_split_size >= global_split_sizes[current_split_id]:
              assert (
                  current_split_size
                  == global_split_sizes[current_split_id]
              )
              result_blocks.append(current_blocks)
              result_metas.append(current_meta)
  
              current_blocks = []
              current_meta = []
              current_split_size = 0
              current_split_id += 1
          else:
              (block_ref, meta) = next(all_blocks_split_results)
              current_blocks.append(block_ref)
              current_meta.append(meta)
              current_split_size += meta.num_rows
  
      return result_blocks, result_metas
  
file_path: ray/python/ray/data/_internal/split.py
imports: import itertools
  import logging
  from typing import Iterable, List, Tuple, Union
  import ray
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.memory_tracing import trace_deallocation
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data.block import Block, BlockAccessor, BlockExecStats, BlockMetadata, BlockPartition
  from ray.types import ObjectRef
label_desc: Use image node_id 7 for calling a global function with example usage: _generate_global_split_results(all_blocks_split_results, global_split_sizes) and returns: result_blocks,  result_metas