code: def test_block_slicing(
      ray_start_regular_shared,
      restore_data_context,
      target_max_block_size,
      batch_size,
      num_batches,
      expected_num_blocks,
  ):
      ctx = ray.data.context.DataContext.get_current()
      ctx.target_max_block_size = target_max_block_size
  
      # Row sizes smaller than this seem to add significant amounts of per-row
      # metadata overhead.
      row_size = 128
      num_rows_per_batch = int(batch_size / row_size)
      num_tasks = 1
  
      ds = ray.data.read_datasource(
          RandomBytesDatasource(),
          parallelism=num_tasks,
          num_batches_per_task=num_batches,
          num_rows_per_batch=num_rows_per_batch,
          row_size=row_size,
          use_bytes=False,
          use_arrow=True,
      ).materialize()
      assert ds.num_blocks() == expected_num_blocks
  
      block_sizes = []
      num_rows = 0
      for batch in ds.iter_batches(
          batch_size=None, batch_format="numpy"
      ):
          block_sizes.append(batch["one"].size)
          num_rows += len(batch["one"])
      assert num_rows == num_rows_per_batch * num_batches
      for size in block_sizes:
          # Blocks are not too big.
          assert (
              size
              <= target_max_block_size
              * ray.data.context.MAX_SAFE_BLOCK_SIZE_FACTOR
          )
          # Blocks are not too small.
          assert size >= target_max_block_size / 2
  
imports: import os
  import time
  from dataclasses import astuple, dataclass
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data import Dataset
  from ray.data._internal.arrow_block import ArrowBlockAccessor
  from ray.data._internal.lazy_block_list import LazyBlockList
  from ray.data.block import BlockMetadata
  from ray.data.datasource import Datasource
  from ray.data.datasource.csv_datasource import CSVDatasource
  from ray.data.datasource.datasource import Reader, ReadTask
  from ray.data.tests.conftest import CoreExecutionMetrics, assert_blocks_expected_in_plasma, assert_core_execution_metrics_equals, get_initial_core_execution_metrics_snapshot
  from ray.tests.conftest import *
label_desc: Use image node_id 13 for calling a global function with example usage: test_block_slicing(ray_start_regular_shared, restore_data_context, target_max_block_size, batch_size, num_batches, expected_num_blocks) without return types