code: def _read_fragments(
      block_udf,
      to_batches_kwargs,
      default_read_batch_size_rows,
      columns,
      schema,
      serialized_fragments: List[_SerializedFragment],
      include_paths: bool,
  ) -> Iterator["pyarrow.Table"]:
      # This import is necessary to load the tensor extension type.
      from ray.data.extensions.tensor_extension import (
          ArrowTensorType,
      )  # noqa
  
      # Deserialize after loading the filesystem class.
      fragments: List[
          "pyarrow._dataset.ParquetFileFragment"
      ] = _deserialize_fragments_with_retry(serialized_fragments)
  
      # Ensure that we're reading at least one dataset fragment.
      assert len(fragments) > 0
  
      import pyarrow as pa
      from pyarrow.dataset import _get_partition_keys
  
      logger.debug(f"Reading {len(fragments)} parquet fragments")
      use_threads = to_batches_kwargs.pop("use_threads", False)
      batch_size = to_batches_kwargs.pop(
          "batch_size", default_read_batch_size_rows
      )
      for fragment in fragments:
          part = _get_partition_keys(fragment.partition_expression)
          batches = fragment.to_batches(
              use_threads=use_threads,
              columns=columns,
              schema=schema,
              batch_size=batch_size,
              **to_batches_kwargs,
          )
          for batch in batches:
              table = pa.Table.from_batches([batch], schema=schema)
              if part:
                  for col, value in part.items():
                      if columns and col not in columns:
                          continue
                      table = table.set_column(
                          table.schema.get_field_index(col),
                          col,
                          pa.array([value] * len(table)),
                      )
              if include_paths:
                  table = table.append_column(
                      "path", [[fragment.path]] * len(table)
                  )
              # If the table is empty, drop it.
              if table.num_rows > 0:
                  if block_udf is not None:
                      yield block_udf(table)
                  else:
                      yield table
  
imports: import logging
  from typing import TYPE_CHECKING, Any, Callable, Dict, Iterator, List, Literal, Optional, Union
  import numpy
  import ray
  import ray.cloudpickle
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data._internal.util import _check_pyarrow_version, _is_local_scheme
  from ray.data.block import Block
  from ray.data.context import DataContext
  from ray.data.datasource import Datasource
  from ray.data.datasource._default_metadata_providers import get_generic_metadata_provider
  from ray.data.datasource.datasource import ReadTask
  from ray.data.datasource.file_meta_provider import DefaultParquetMetadataProvider, ParquetMetadataProvider, _handle_read_os_error
  from ray.data.datasource.partitioning import PathPartitionFilter
  from ray.data.datasource.path_util import _has_file_extension, _resolve_paths_and_filesystem
  from ray.util.annotations import PublicAPI
label_desc: Use image node_id 3 for calling a global function with example usage: _read_fragments(block_udf, to_batches_kwargs, default_read_batch_size_rows, columns, schema, serialized_fragments, include_paths) without return types