code: def _expand_paths(
      paths: List[str],
      filesystem: "pyarrow.fs.FileSystem",
      partitioning: Optional[Partitioning],
      ignore_missing_paths: bool = False,
  ) -> Iterator[Tuple[str, int]]:
      """Get the file sizes for all provided file paths."""
      from pyarrow.fs import LocalFileSystem
  
      from ray.data.datasource.file_based_datasource import (
          FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD,
      )
      from ray.data.datasource.path_util import _unwrap_protocol
  
      # We break down our processing paths into a few key cases:
      # 1. If len(paths) < threshold, fetch the file info for the individual files/paths
      #    serially.
      # 2. If all paths are contained under the same parent directory (or base directory,
      #    if using partitioning), fetch all file infos at this prefix and filter to the
      #    provided paths on the client; this should be a single file info request.
      # 3. If more than threshold requests required, parallelize them via Ray tasks.
      # 1. Small # of paths case.
      if (
          len(paths) < FILE_SIZE_FETCH_PARALLELIZATION_THRESHOLD
          # Local file systems are very fast to hit.
          or isinstance(filesystem, LocalFileSystem)
      ):
          yield from _get_file_infos_serial(
              paths, filesystem, ignore_missing_paths
          )
      else:
          # 2. Common path prefix case.
          # Get longest common path of all paths.
          common_path = os.path.commonpath(paths)
          # If parent directory (or base directory, if using partitioning) is common to
          # all paths, fetch all file infos at that prefix and filter the response to the
          # provided paths.
          if (
              partitioning is not None
              and common_path == _unwrap_protocol(partitioning.base_dir)
          ) or all(
              str(pathlib.Path(path).parent) == common_path
              for path in paths
          ):
              yield from _get_file_infos_common_path_prefix(
                  paths, common_path, filesystem, ignore_missing_paths
              )
          # 3. Parallelization case.
          else:
              # Parallelize requests via Ray tasks.
              yield from _get_file_infos_parallel(
                  paths, filesystem, ignore_missing_paths
              )
  
imports: import itertools
  import logging
  import os
  import pathlib
  import re
  from typing import TYPE_CHECKING, Any, Callable, Iterator, List, Optional, Tuple, TypeVar, Union
  import numpy
  from ray.data._internal.progress_bar import ProgressBar
  from ray.data._internal.remote_fn import cached_remote_fn
  from ray.data.block import BlockMetadata
  from ray.data.datasource.partitioning import Partitioning
  from ray.util.annotations import DeveloperAPI
label_desc: Use image node_id 2 for calling a global function with example usage: _expand_paths(paths, filesystem, partitioning, ignore_missing_paths) without return types