code: def test_parquet_read_spread(ray_start_cluster, tmp_path):
      ray.shutdown()
      cluster = ray_start_cluster
      cluster.add_node(
          resources={"bar:1": 100},
          num_cpus=10,
          _system_config={"max_direct_call_object_size": 0},
      )
      cluster.add_node(resources={"bar:2": 100}, num_cpus=10)
      cluster.add_node(resources={"bar:3": 100}, num_cpus=0)
  
      ray.init(cluster.address)
  
      @ray.remote
      def get_node_id():
          return ray.get_runtime_context().get_node_id()
  
      node1_id = ray.get(
          get_node_id.options(resources={"bar:1": 1}).remote()
      )
      node2_id = ray.get(
          get_node_id.options(resources={"bar:2": 1}).remote()
      )
  
      data_path = str(tmp_path)
      df1 = pd.DataFrame(
          {"one": list(range(100)), "two": list(range(100, 200))}
      )
      path1 = os.path.join(data_path, "test1.parquet")
      df1.to_parquet(path1)
      df2 = pd.DataFrame(
          {"one": list(range(300, 400)), "two": list(range(400, 500))}
      )
      path2 = os.path.join(data_path, "test2.parquet")
      df2.to_parquet(path2)
  
      ds = ray.data.read_parquet(data_path)
  
      # Force reads.
      blocks = ds.get_internal_block_refs()
      ray.wait(blocks, num_returns=len(blocks), fetch_local=False)
      location_data = ray.experimental.get_object_locations(blocks)
      locations = []
      for block in blocks:
          locations.extend(location_data[block]["node_ids"])
      assert set(locations) == {node1_id, node2_id}
  
imports: import os
  import shutil
  from typing import Any
  import numpy
  import pandas
  import pyarrow
  import pyarrow.parquet
  import pytest
  from pytest_lazyfixture import lazy_fixture
  import ray
  from ray.data.block import BlockAccessor
  from ray.data.context import DataContext
  from ray.data.datasource import DefaultFileMetadataProvider, DefaultParquetMetadataProvider
  from ray.data.datasource.parquet_base_datasource import ParquetBaseDatasource
  from ray.data.datasource.parquet_datasource import NUM_CPUS_FOR_META_FETCH_TASK, PARALLELIZE_META_FETCH_THRESHOLD, RETRY_EXCEPTIONS_FOR_META_FETCH_TASK, ParquetDatasource, _deserialize_fragments_with_retry, _SerializedFragment
  from ray.data.datasource.path_util import _unwrap_protocol
  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 25 for calling a global function with example usage: test_parquet_read_spread(ray_start_cluster, tmp_path) and returns: ray