code: def _split(
      bundle: RefBundle, left_size: int
  ) -> (RefBundle, RefBundle):
      left_blocks, left_meta = [], []
      right_blocks, right_meta = [], []
      acc = 0
      for b, m in bundle.blocks:
          if acc >= left_size:
              right_blocks.append(b)
              right_meta.append(m)
          elif acc + m.num_rows <= left_size:
              left_blocks.append(b)
              left_meta.append(m)
              acc += m.num_rows
          else:
              # Trouble case: split it up.
              lm, rm = _split_meta(m, left_size - acc)
              lb, rb = _split_block(b, left_size - acc)
              left_meta.append(lm)
              right_meta.append(rm)
              left_blocks.append(lb)
              right_blocks.append(rb)
              acc += lm.num_rows
              assert acc == left_size
      left = RefBundle(
          list(zip(left_blocks, left_meta)),
          owns_blocks=bundle.owns_blocks,
      )
      right = RefBundle(
          list(zip(right_blocks, right_meta)),
          owns_blocks=bundle.owns_blocks,
      )
      assert left.num_rows() == left_size
      assert left.num_rows() + right.num_rows() == bundle.num_rows()
      return left, right
  
imports: import math
  from collections import deque
  from typing import Any, Dict, List, Optional
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, NodeIdStr, PhysicalOperator, RefBundle
  from ray.data._internal.execution.util import locality_string
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.stats import StatsDict
  from ray.data.block import Block, BlockAccessor, BlockMetadata
  from ray.types import ObjectRef
label_desc: Use image node_id 1 for calling a global function with example usage: _split(bundle, left_size) and returns: left,  right