code: def test_dynamic_block_split(ray_start_regular_shared):
      ctx = ray.data.context.DataContext.get_current()
      target_max_block_size = ctx.target_max_block_size
      # Reduce target max block size to trigger block splitting on small input.
      # Otherwise we have to generate big input files, which is unnecessary.
      ctx.target_max_block_size = 1
      try:
          root = "example://image-datasets/simple"
          ds = ray.data.read_images(root, parallelism=1)
          assert ds.num_blocks() == 1
          ds = ds.materialize()
          # Verify dynamic block splitting taking effect to generate more blocks.
          assert ds.num_blocks() == 3
  
          # Test union of same datasets
          union_ds = ds.union(ds, ds, ds).materialize()
          assert union_ds.num_blocks() == 12
      finally:
          ctx.target_max_block_size = target_max_block_size
  
imports: import os
  import tempfile
  from typing import Dict
  import numpy
  import pyarrow
  import pytest
  from fsspec.implementations.local import LocalFileSystem
  from PIL import Image
  import ray
  from ray.data.datasource import Partitioning
  from ray.data.datasource.file_meta_provider import FastFileMetadataProvider
  from ray.data.datasource.image_datasource import ImageDatasource, _ImageFileMetadataProvider
  from ray.data.extensions import ArrowTensorType
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.tests.conftest import *
label_desc: Use image node_id 15 for calling the TestReadImages obj's underlying member method code with example usage: obj.test_dynamic_block_split(ray_start_regular_shared) without return types