code: def test_filter(ray_start_regular_shared, target_max_block_size):
      # Test 10 tasks, each task returning 10 blocks, each block has 1 row and each
      # row has 1024 bytes.
      num_blocks_per_task = 10
      block_size = 1024
  
      ds = ray.data.read_datasource(
          RandomBytesDatasource(),
          parallelism=1,
          num_batches_per_task=num_blocks_per_task,
          row_size=block_size,
      )
  
      ds = ds.filter(lambda _: True)
      ds = ds.materialize()
      assert ds.count() == num_blocks_per_task
      assert ds.num_blocks() == num_blocks_per_task
  
      ds = ds.filter(lambda _: False)
      ds = ds.materialize()
      assert ds.count() == 0
      assert ds.num_blocks() == num_blocks_per_task
  
imports: import os
  import time
  from dataclasses import astuple, dataclass
  import numpy
  import pandas
  import pyarrow
  import pytest
  import ray
  from ray.data import Dataset
  from ray.data._internal.arrow_block import ArrowBlockAccessor
  from ray.data._internal.lazy_block_list import LazyBlockList
  from ray.data.block import BlockMetadata
  from ray.data.datasource import Datasource
  from ray.data.datasource.csv_datasource import CSVDatasource
  from ray.data.datasource.datasource import Reader, ReadTask
  from ray.data.tests.conftest import CoreExecutionMetrics, assert_blocks_expected_in_plasma, assert_core_execution_metrics_equals, get_initial_core_execution_metrics_snapshot
  from ray.tests.conftest import *
label_desc: Use image node_id 3 for calling a global function with example usage: test_filter(ray_start_regular_shared, target_max_block_size) without return types