code: def test_split_read_parquet(ray_start_regular_shared, tmp_path):
      ctx = ray.data.context.DataContext.get_current()
  
      def gen(name):
          path = os.path.join(tmp_path, name)
          ds = (
              ray.data.range(200000, parallelism=1)
              .map(lambda _: {"out": uuid.uuid4().hex})
              .materialize()
          )
          # Fully execute the operations prior to write, because with
          # parallelism=1, there is only one task; so the write operator
          # will only write to one file, even though there are multiple
          # blocks created by block splitting.
          ds.write_parquet(path)
          return ray.data.read_parquet(path, parallelism=1)
  
      # 20MiB
      ctx.target_max_block_size = 20_000_000
      ds1 = gen("out1")
      assert ds1._block_num_rows() == [200000]
  
      # 3MiB
      ctx.target_max_block_size = 3_000_000
      ds2 = gen("out2")
      nrow = ds2._block_num_rows()
      assert 2 < len(nrow) < 5, nrow
      for x in nrow[:-1]:
          assert 50000 < x < 95000, (x, nrow)
  
      # 1MiB
      ctx.target_max_block_size = 1_000_000
      ds3 = gen("out3")
      nrow = ds3._block_num_rows()
      assert 6 < len(nrow) < 12, nrow
      for x in nrow[:-1]:
          assert 20000 < x < 35000, (x, nrow)
  
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 6 for calling a global function with example usage: test_split_read_parquet(ray_start_regular_shared, tmp_path) and returns: ray