code: def test_parquet_read_partitioned_with_filter(
      ray_start_regular_shared, tmp_path
  ):
      df = pd.DataFrame(
          {
              "one": [1, 1, 1, 3, 3, 3],
              "two": ["a", "a", "b", "b", "c", "c"],
          }
      )
      table = pa.Table.from_pandas(df)
      pq.write_to_dataset(
          table,
          root_path=str(tmp_path),
          partition_cols=["one"],
          use_legacy_dataset=False,
      )
  
      # 2 partitions, 1 empty partition, 1 block/read task
  
      ds = ray.data.read_parquet(
          str(tmp_path),
          parallelism=1,
          filter=(pa.dataset.field("two") == "a"),
      )
  
      values = [[s["one"], s["two"]] for s in ds.take()]
      check_num_computed(ds, 1, 0)
      assert sorted(values) == [[1, "a"], [1, "a"]]
      assert ds.count() == 2
  
      # 2 partitions, 1 empty partition, 2 block/read tasks, 1 empty block
  
      ds = ray.data.read_parquet(
          str(tmp_path),
          parallelism=2,
          filter=(pa.dataset.field("two") == "a"),
      )
  
      values = [[s["one"], s["two"]] for s in ds.take()]
      check_num_computed(ds, 2, 0)
      assert sorted(values) == [[1, "a"], [1, "a"]]
      assert ds.count() == 2
  
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 10 for calling a global function with example usage: test_parquet_read_partitioned_with_filter(ray_start_regular_shared, tmp_path) without return types