code: def _split_leftovers(
      leftovers: BlockList, per_split_needed_rows: List[int]
  ) -> List[BlockPartition]:
      """Split leftover blocks by the num of rows needed."""
      num_splits = len(per_split_needed_rows)
      split_indices = []
      prev = 0
      for i, num_rows_needed in enumerate(per_split_needed_rows):
          split_indices.append(prev + num_rows_needed)
          prev = split_indices[i]
      split_result: Tuple[
          List[List[ObjectRef[Block]]], List[List[BlockMetadata]]
      ] = _split_at_indices(
          leftovers.get_blocks_with_metadata(),
          split_indices,
          leftovers._owned_by_consumer,
      )
      return [
          list(zip(block_refs, meta))
          for block_refs, meta in zip(*split_result)
      ][:num_splits]
  
file_path: ray/python/ray/data/_internal/equalize.py
imports: from typing import List, Tuple
  from ray.data._internal.block_list import BlockList
  from ray.data._internal.split import _calculate_blocks_rows, _split_at_indices
  from ray.data.block import Block, BlockMetadata, BlockPartition
  from ray.types import ObjectRef
label_desc: Use image node_id 4 for calling a global function with example usage: _split_leftovers(leftovers, per_split_needed_rows) and returns: unknown