code: def test_streaming_split_e2e(ray_start_10_cpus_shared):
      def get_lengths(*iterators, use_iter_batches=True):
          lengths = []
  
          class Runner(threading.Thread):
              def __init__(self, it):
                  self.it = it
                  super().__init__()
  
              def run(self):
                  it = self.it
                  x = 0
                  if use_iter_batches:
                      for batch in it.iter_batches():
                          for arr in batch.values():
                              x += arr.size
                  else:
                      for _ in it.iter_rows():
                          x += 1
                  lengths.append(x)
  
          runners = [Runner(it) for it in iterators]
          for r in runners:
              r.start()
          for r in runners:
              r.join()
  
          lengths.sort()
          return lengths
  
      ds = ray.data.range(1000)
      (
          i1,
          i2,
      ) = ds.streaming_split(2, equal=True)
      for _ in range(2):
          lengths = get_lengths(i1, i2)
          assert lengths == [500, 500], lengths
  
      ds = ray.data.range(1)
      (
          i1,
          i2,
      ) = ds.streaming_split(2, equal=True)
      for _ in range(2):
          lengths = get_lengths(i1, i2)
          assert lengths == [0, 0], lengths
  
      ds = ray.data.range(1)
      (
          i1,
          i2,
      ) = ds.streaming_split(2, equal=False)
      for _ in range(2):
          lengths = get_lengths(i1, i2)
          assert lengths == [0, 1], lengths
  
      ds = ray.data.range(1000, parallelism=10)
      for equal_split, use_iter_batches in itertools.product(
          [True, False], [True, False]
      ):
          i1, i2, i3 = ds.streaming_split(3, equal=equal_split)
          for _ in range(2):
              lengths = get_lengths(
                  i1, i2, i3, use_iter_batches=use_iter_batches
              )
              if equal_split:
                  assert lengths == [333, 333, 333], lengths
              else:
                  assert lengths == [300, 300, 400], lengths
  
imports: import itertools
  import random
  import threading
  import time
  from typing import Any, List
  import pandas
  import pytest
  import ray
  from ray import cloudpickle
  from ray._private.test_utils import wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, RefBundle
  from ray.data._internal.execution.operators.base_physical_operator import AllToAllOperator
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.operators.output_splitter import OutputSplitter
  from ray.data._internal.execution.streaming_executor import StreamingExecutor
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.context import DataContext
  from ray.data.tests.conftest import *
  from ray.data.tests.util import extract_values
label_desc: Use image node_id 6 for calling a global function with example usage: test_streaming_split_e2e(ray_start_10_cpus_shared) and returns: lengths