code: def test_parquet_read_partitioned(
      ray_start_regular_shared, fs, data_path
  ):
      df = pd.DataFrame(
          {
              "one": [1, 1, 1, 3, 3, 3],
              "two": ["a", "b", "c", "e", "f", "g"],
          }
      )
      table = pa.Table.from_pandas(df)
      pq.write_to_dataset(
          table,
          root_path=_unwrap_protocol(data_path),
          partition_cols=["one"],
          filesystem=fs,
          use_legacy_dataset=False,
      )
  
      ds = ray.data.read_parquet(data_path, filesystem=fs)
  
      # Test metadata-only parquet ops.
      check_num_computed(ds, 0, 0)
      assert ds.count() == 6
      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
      check_num_computed(ds, 0, 0)
      input_files = ds.input_files()
      assert len(input_files) == 2, input_files
      check_num_computed(ds, 0, 0)
      assert str(ds) == (
          "Dataset(\n"
          "   num_blocks=2,\n"
          "   num_rows=6,\n"
          "   schema={two: string, "
          "one: dictionary<values=int32, indices=int32, ordered=0>}\n"
          ")"
      ), ds
      assert repr(ds) == (
          "Dataset(\n"
          "   num_blocks=2,\n"
          "   num_rows=6,\n"
          "   schema={two: string, "
          "one: dictionary<values=int32, indices=int32, ordered=0>}\n"
          ")"
      ), ds
      check_num_computed(ds, 0, 0)
  
      # Forces a data read.
      values = [[s["one"], s["two"]] for s in ds.take()]
      assert sorted(values) == [
          [1, "a"],
          [1, "b"],
          [1, "c"],
          [3, "e"],
          [3, "f"],
          [3, "g"],
      ]
      check_num_computed(ds, 2, 0)
  
      # Test column selection.
      ds = ray.data.read_parquet(
          data_path, columns=["one"], filesystem=fs
      )
      values = [s["one"] for s in ds.take()]
      assert sorted(values) == [1, 1, 1, 3, 3, 3]
      check_num_computed(ds, 2, 0)
  
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 9 for calling a global function with example usage: test_parquet_read_partitioned(ray_start_regular_shared, fs, data_path) without return types