code: def test_small_file_split(
      ray_start_10_cpus_shared, restore_data_context
  ):
      last_snapshot = get_initial_core_execution_metrics_snapshot()
  
      ds = ray.data.read_csv("example://iris.csv", parallelism=1)
      materialized_ds = ds.materialize()
      assert materialized_ds.num_blocks() == 1
      last_snapshot = assert_core_execution_metrics_equals(
          CoreExecutionMetrics(
              task_count={
                  "_execute_read_task_split": 1,
                  "_get_datasource_or_legacy_reader": 1,
              },
          ),
          last_snapshot,
      )
  
      materialized_ds = ds.map_batches(lambda x: x).materialize()
      assert materialized_ds.num_blocks() == 1
      last_snapshot = assert_core_execution_metrics_equals(
          CoreExecutionMetrics(
              task_count={
                  "ReadCSV->MapBatches(<lambda>)": 1,
              },
          ),
          last_snapshot,
      )
  
      stats = materialized_ds.stats()
      assert "Operator 1 ReadCSV->MapBatches" in stats, stats
  
      ds = ray.data.read_csv("example://iris.csv", parallelism=10)
      assert ds.num_blocks() == 1
      assert (
          ds.map_batches(lambda x: x).materialize().num_blocks() == 10
      )
      last_snapshot = assert_core_execution_metrics_equals(
          CoreExecutionMetrics(
              task_count={
                  "_get_datasource_or_legacy_reader": 1,
                  "MapBatches(<lambda>)": 10,
                  "ReadCSV->SplitBlocks(10)": 1,
              },
          ),
          last_snapshot,
      )
  
      assert ds.materialize().num_blocks() == 10
      last_snapshot = assert_core_execution_metrics_equals(
          CoreExecutionMetrics(
              task_count={
                  "_execute_read_task_split": 1,
              },
          ),
          last_snapshot,
      )
  
      ds = ray.data.read_csv("example://iris.csv", parallelism=100)
      assert ds.num_blocks() == 1
      assert (
          ds.map_batches(lambda x: x).materialize().num_blocks() == 100
      )
      assert ds.materialize().num_blocks() == 100
  
      ds = ds.map_batches(lambda x: x).materialize()
      stats = ds.stats()
      assert "Operator 1 ReadCSV->SplitBlocks(100)" in stats, stats
      assert "Operator 2 MapBatches" in stats, stats
  
      ctx = ray.data.context.DataContext.get_current()
      # Smaller than a single row.
      ctx.target_max_block_size = 1
      ds = ds.map_batches(lambda x: x).materialize()
      # 150 rows.
      assert ds.num_blocks() == 150
      print(ds.stats())
  
file_path: ray/python/ray/data/tests/test_splitblocks.py
imports: import numpy
  import pytest
  import ray
  from ray.data._internal.execution.operators.map_transformer import _splitrange
  from ray.data.tests.conftest import *
  from ray.data.tests.conftest import CoreExecutionMetrics, assert_core_execution_metrics_equals, get_initial_core_execution_metrics_snapshot
  from ray.tests.conftest import *
label_desc: Use image node_id 2 for calling a global function with example usage: test_small_file_split(ray_start_10_cpus_shared, restore_data_context) without return types