code: def test_numpy_read_partitioned_with_filter(
      ray_start_regular_shared,
      tmp_path,
      write_partitioned_df,
      assert_base_partitioned_ds,
  ):
      def df_to_np(dataframe, path, **kwargs):
          np.save(
              path,
              dataframe.to_numpy(dtype=np.dtype(np.int8)),
              **kwargs,
          )
  
      df = pd.DataFrame(
          {"one": [1, 1, 1, 3, 3, 3], "two": [0, 1, 2, 3, 4, 5]}
      )
      partition_keys = ["one"]
      kept_file_counter = Counter.remote()
      skipped_file_counter = Counter.remote()
  
      def skip_unpartitioned(kv_dict):
          keep = bool(kv_dict)
          counter = kept_file_counter if keep else skipped_file_counter
          ray.get(counter.increment.remote())
          return keep
  
      for style in [PartitionStyle.HIVE, PartitionStyle.DIRECTORY]:
          base_dir = os.path.join(tmp_path, style.value)
          partition_path_encoder = PathPartitionEncoder.of(
              style=style,
              base_dir=base_dir,
              field_names=partition_keys,
          )
          write_partitioned_df(
              df,
              partition_keys,
              partition_path_encoder,
              df_to_np,
          )
          df_to_np(df, os.path.join(base_dir, "test.npy"))
          partition_path_filter = PathPartitionFilter.of(
              style=style,
              base_dir=base_dir,
              field_names=partition_keys,
              filter_fn=skip_unpartitioned,
          )
          ds = ray.data.read_numpy(
              base_dir, partition_filter=partition_path_filter
          )
  
          vals = [[1, 0], [1, 1], [1, 2], [3, 3], [3, 4], [3, 5]]
          val_str = "".join(f"array({v}, dtype=int8), " for v in vals)[
              :-2
          ]
          assert_base_partitioned_ds(
              ds,
              schema="{data: numpy.ndarray(shape=(2,), dtype=int8)}",
              sorted_values=f"[[{val_str}]]",
              ds_take_transform_fn=lambda taken: [
                  extract_values("data", taken)
              ],
              sorted_values_transform_fn=lambda sorted_values: str(
                  sorted_values
              ),
          )
          assert ray.get(kept_file_counter.get.remote()) == 2
          assert ray.get(skipped_file_counter.get.remote()) == 1
          ray.get(kept_file_counter.reset.remote())
          ray.get(skipped_file_counter.reset.remote())
  
imports: import os
  import numpy
  import pandas
  import pyarrow
  import pytest
  from pytest_lazyfixture import lazy_fixture
  import ray
  from ray.data.datasource import BaseFileMetadataProvider, FastFileMetadataProvider, Partitioning, PartitionStyle, PathPartitionFilter
  from ray.data.tests.conftest import *
  from ray.data.tests.mock_http_server import *
  from ray.data.tests.test_partitioning import PathPartitionEncoder
  from ray.data.tests.util import Counter, extract_values
  from ray.tests.conftest import *
label_desc: Use image node_id 9 for calling a global function with example usage: test_numpy_read_partitioned_with_filter(ray_start_regular_shared, tmp_path, write_partitioned_df, assert_base_partitioned_ds) and returns: keep