code: def test_split_read_csv(ray_start_regular_shared, tmp_path):
      ctx = ray.data.context.DataContext.get_current()
  
      def gen(name):
          path = os.path.join(tmp_path, name)
          ray.data.range(1000, parallelism=1).map(
              lambda _: {"out": LARGE_VALUE}
          ).write_csv(path)
          return ray.data.read_csv(path, parallelism=1)
  
      # 20MiB
      ctx.target_max_block_size = 20_000_000
      ds1 = gen("out1")
      assert ds1._block_num_rows() == [1000]
  
      # 3MiB
      ctx.target_max_block_size = 3_000_000
      ds2 = gen("out2")
      nrow = ds2._block_num_rows()
      assert 3 < len(nrow) < 5, nrow
      for x in nrow[:-1]:
          assert 200 < x < 400, (x, nrow)
  
      # 1MiB
      ctx.target_max_block_size = 1_000_000
      ds3 = gen("out3")
      nrow = ds3._block_num_rows()
      assert 8 < len(nrow) < 12, nrow
      for x in nrow[:-1]:
          assert 80 < x < 120, (x, nrow)
  
      # Disabled.
      # Setting infinite block size effectively disables block splitting.
      ctx.target_max_block_size = float("inf")
      ds4 = gen("out4")
      assert ds4._block_num_rows() == [1000]
  
file_path: ray/python/ray/data/tests/test_size_estimation.py
imports: import os
  import uuid
  import pytest
  import ray
  from ray.data._internal.arrow_block import ArrowBlockBuilder
  from ray.tests.conftest import *
label_desc: Use image node_id 5 for calling a global function with example usage: test_split_read_csv(ray_start_regular_shared, tmp_path) and returns: ray