code: def test_iter_batches_e2e(
      ray_start_regular_shared, batch_size, drop_last, prefetch_batches
  ):
      def collate_fn(batch: pd.DataFrame):
          return batch + 1
  
      block_refs_iter = itertools.starmap(
          lambda block, metadata: (ray.put(block), metadata),
          block_generator(num_blocks=4, num_rows=2),
      )
  
      output_batches = iter_batches(
          block_refs_iter,
          dataset_tag="dataset",
          batch_size=batch_size,
          prefetch_batches=prefetch_batches,
          batch_format="pandas",
          collate_fn=collate_fn,
          drop_last=drop_last,
      )
  
      output_batches = list(output_batches)
  
      assert len(output_batches) > 0
      for df in output_batches:
          # Check batch formatting.
          assert isinstance(df, pd.DataFrame)
          # Check batch size.
          if batch_size == 3 and not drop_last:
              assert len(df) in {2, 3}
          else:
              assert len(df) == batch_size
  
      concat_df = pd.concat(output_batches)
      # Test that collate_fn is applied.
      assert concat_df["foo"].iloc[0] == 1
      # Make sure order is preserved.
      for i in range(len(concat_df) - 1):
          assert (
              concat_df["foo"].iloc[i + 1] >= concat_df["foo"].iloc[i]
          )
  
file_path: ray/python/ray/data/tests/block_batching/test_iter_batches.py
imports: import itertools
  import queue
  import threading
  import time
  from typing import Iterator, List, Tuple
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data._internal.block_batching.interfaces import Batch, BlockPrefetcher
  from ray.data._internal.block_batching.iter_batches import iter_batches, prefetch_batches_locally, restore_original_order
  from ray.data.block import Block, BlockMetadata
label_desc: Use image node_id 5 for calling a global function with example usage: test_iter_batches_e2e(ray_start_regular_shared, batch_size, drop_last, prefetch_batches) and returns: unknown