code: def test_parquet_read_meta_provider(
      ray_start_regular_shared, fs, data_path
  ):
      df1 = pd.DataFrame({"one": [1, 2, 3], "two": ["a", "b", "c"]})
      table = pa.Table.from_pandas(df1)
      setup_data_path = _unwrap_protocol(data_path)
      path1 = os.path.join(setup_data_path, "test1.parquet")
      pq.write_table(table, path1, filesystem=fs)
      df2 = pd.DataFrame({"one": [4, 5, 6], "two": ["e", "f", "g"]})
      table = pa.Table.from_pandas(df2)
      path2 = os.path.join(setup_data_path, "test2.parquet")
      pq.write_table(table, path2, filesystem=fs)
  
      class TestMetadataProvider(DefaultParquetMetadataProvider):
          def prefetch_file_metadata(
              self, fragments, **ray_remote_args
          ):
              assert (
                  ray_remote_args["num_cpus"]
                  == NUM_CPUS_FOR_META_FETCH_TASK
              )
              assert (
                  ray_remote_args["scheduling_strategy"]
                  == DataContext.get_current().scheduling_strategy
              )
              assert (
                  ray_remote_args["retry_exceptions"]
                  == RETRY_EXCEPTIONS_FOR_META_FETCH_TASK
              )
              return None
  
      ds = ray.data.read_parquet(
          data_path,
          filesystem=fs,
          meta_provider=TestMetadataProvider(),
      )
  
      # Expect precomputed row counts and block sizes to be missing.
      assert ds._meta_count() is None
      assert (
          ds._plan._snapshot_blocks is None
          or ds._plan._snapshot_blocks.size_bytes() == -1
      )
  
      # Expect to lazily compute all metadata correctly.
      check_num_computed(ds, 0, 0)
      assert ds.count() == 6
      assert ds.size_bytes() > 0
      assert ds.schema() is not None
      input_files = ds.input_files()
      assert len(input_files) == 2, input_files
      assert "test1.parquet" in str(input_files)
      assert "test2.parquet" in str(input_files)
      assert (
          str(ds) == "Dataset(num_blocks=2, num_rows=6, "
          "schema={one: int64, two: string})"
      ), ds
      assert (
          repr(ds) == "Dataset(num_blocks=2, num_rows=6, "
          "schema={one: int64, two: string})"
      ), ds
      check_num_computed(ds, 2, 2)
  
      # Forces a data read.
      values = [[s["one"], s["two"]] for s in ds.take()]
      check_num_computed(ds, 2, 2)
      assert sorted(values) == [
          [1, "a"],
          [2, "b"],
          [3, "c"],
          [4, "e"],
          [5, "f"],
          [6, "g"],
      ]
  
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 5 for calling a global function with example usage: test_parquet_read_meta_provider(ray_start_regular_shared, fs, data_path) and returns: None