code: def _generate_per_block_split_indices(
      num_rows_per_block: List[int],
      split_indices: List[int],
  ) -> List[List[int]]:
      """Given num rows per block and valid split indices, generate per block split indices.
  
      Args:
          num_rows_per_block: num of rows per block.
          split_indices: The (global) indices at which to split the blocks.
      Returns:
          Per block split indices indicates each input block's split point(s).
      """
      # for each split index, we iterate though the currnet input block
      # to see if the index falls into this block. if the index
      # falls into this block, we push it back to the current block's
      # split indices. Otherwise, we move on to the next block.
      per_block_split_indices = []
      current_input_block_id = 0
      current_block_split_indices = []
      current_block_global_offset = 0
      current_index_id = 0
  
      while current_index_id < len(split_indices):
          split_index = split_indices[current_index_id]
          current_block_row = num_rows_per_block[current_input_block_id]
          if (
              split_index - current_block_global_offset
              <= current_block_row
          ):
              current_block_split_indices.append(
                  split_index - current_block_global_offset
              )
              current_index_id += 1
              continue
          per_block_split_indices.append(current_block_split_indices)
          current_block_split_indices = []
          current_block_global_offset += num_rows_per_block[
              current_input_block_id
          ]
          current_input_block_id += 1
  
      # we might finished all the indices but there are still blocks left, also
      # current_block_split_indices might not be added yet.
      while len(per_block_split_indices) < len(num_rows_per_block):
          per_block_split_indices.append(current_block_split_indices)
          current_block_split_indices = []
      return per_block_split_indices
  
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 3 for calling a global function with example usage: _generate_per_block_split_indices(num_rows_per_block, split_indices) and returns: per_block_split_indices