code: def test_parquet_read_parallel_meta_fetch(
      ray_start_regular_shared, fs, data_path
  ):
      setup_data_path = _unwrap_protocol(data_path)
      num_dfs = PARALLELIZE_META_FETCH_THRESHOLD + 1
      for idx in range(num_dfs):
          df = pd.DataFrame(
              {"one": list(range(3 * idx, 3 * (idx + 1)))}
          )
          table = pa.Table.from_pandas(df)
          path = os.path.join(setup_data_path, f"test_{idx}.parquet")
          pq.write_table(table, path, filesystem=fs)
  
      parallelism = 8
      ds = ray.data.read_parquet(
          data_path, filesystem=fs, parallelism=parallelism
      )
  
      # Test metadata-only parquet ops.
      check_num_computed(ds, 0, 0)
      assert ds.count() == num_dfs * 3
      assert ds.size_bytes() > 0
      # Schema information and input files are available from Parquet metadata,
      # so we do not need to compute the first block.
      assert ds.schema() is not None
      input_files = ds.input_files()
      assert len(input_files) == num_dfs, input_files
      check_num_computed(ds, 0, 0)
  
      # Forces a data read.
      values = [s["one"] for s in ds.take(limit=3 * num_dfs)]
      check_num_computed(ds, parallelism, 0)
      assert sorted(values) == list(range(3 * num_dfs))
  
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 15 for calling a global function with example usage: test_parquet_read_parallel_meta_fetch(ray_start_regular_shared, fs, data_path) without return types