code: def test_read_binary_snappy_partitioned_with_filter(
      ray_start_regular_shared,
      tmp_path,
      write_base_partitioned_df,
      assert_base_partitioned_ds,
  ):
      def df_to_binary(dataframe, path, **kwargs):
          with open(path, "wb") as f:
              df_string = dataframe.to_string(
                  index=False, header=False, **kwargs
              )
              byte_str = df_string.encode()
              bytes = BytesIO(byte_str)
              snappy.stream_compress(bytes, f)
  
      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_base_partitioned_df(
              partition_keys,
              partition_path_encoder,
              df_to_binary,
          )
          df_to_binary(
              pd.DataFrame({"1": [1]}),
              os.path.join(base_dir, "test.snappy"),
          )
          partition_path_filter = PathPartitionFilter.of(
              style=style,
              base_dir=base_dir,
              field_names=partition_keys,
              filter_fn=skip_unpartitioned,
          )
          ds = ray.data.read_binary_files(
              base_dir,
              partition_filter=partition_path_filter,
              arrow_open_stream_args=dict(compression="snappy"),
          )
          assert_base_partitioned_ds(
              ds,
              count=2,
              num_rows=2,
              schema="{bytes: binary}",
              num_computed=None,
              sorted_values=[b"1 a\n1 b\n1 c", b"3 e\n3 f\n3 g"],
              ds_take_transform_fn=lambda t: extract_values("bytes", t),
          )
          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
  from io import BytesIO
  import pandas
  import pyarrow
  import pytest
  import requests
  import snappy
  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, gen_bin_files
  from ray.tests.conftest import *
label_desc: Use image node_id 9 for calling a global function with example usage: test_read_binary_snappy_partitioned_with_filter(ray_start_regular_shared, tmp_path, write_base_partitioned_df, assert_base_partitioned_ds) and returns: keep