code: def test_parquet_read_with_udf(ray_start_regular_shared, tmp_path):
      one_data = list(range(6))
      df = pd.DataFrame(
          {"one": one_data, "two": 2 * ["a"] + 2 * ["b"] + 2 * ["c"]}
      )
      table = pa.Table.from_pandas(df)
      pq.write_to_dataset(
          table,
          root_path=str(tmp_path),
          partition_cols=["two"],
          use_legacy_dataset=False,
      )
  
      def _block_udf(block: pa.Table):
          df = block.to_pandas()
          df["one"] += 1
          return pa.Table.from_pandas(df)
  
      # 1 block/read task
  
      ds = ray.data.read_parquet(
          str(tmp_path), parallelism=1, _block_udf=_block_udf
      )
  
      ones, twos = zip(*[[s["one"], s["two"]] for s in ds.take()])
      check_num_computed(ds, 1, 0)
      np.testing.assert_array_equal(
          sorted(ones), np.array(one_data) + 1
      )
  
      # 2 blocks/read tasks
  
      ds = ray.data.read_parquet(
          str(tmp_path), parallelism=2, _block_udf=_block_udf
      )
  
      ones, twos = zip(*[[s["one"], s["two"]] for s in ds.take()])
      check_num_computed(ds, 2, 0)
      np.testing.assert_array_equal(
          sorted(ones), np.array(one_data) + 1
      )
  
      # 2 blocks/read tasks, 1 empty block
  
      ds = ray.data.read_parquet(
          str(tmp_path),
          parallelism=2,
          filter=(pa.dataset.field("two") == "a"),
          _block_udf=_block_udf,
      )
  
      ones, twos = zip(*[[s["one"], s["two"]] for s in ds.take()])
      check_num_computed(ds, 2, 0)
      np.testing.assert_array_equal(
          sorted(ones), np.array(one_data[:2]) + 1
      )
  
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 14 for calling a global function with example usage: test_parquet_read_with_udf(ray_start_regular_shared, tmp_path) and returns: pa