Function _read_datasource_v2

python/ray/data/read_api.py:426–588

Internal entry point for ``DataSourceV2`` reads. Wires a ``ListFiles → ReadFiles`` logical chain: - :class:`ListFiles` owns listing (via the datasource's ``FileIndexer``), optional global shuffle (``FileShuffleConfig``), and size-balanced bucketing (``RoundRobinPartitioner``).

(
    datasource,
    *,
    parallelism: int = -1,
    num_cpus: Optional[float] = None,
    num_gpus: Optional[float] = None,
    memory: Optional[float] = None,
    ray_remote_args: Optional[Dict[str, Any]] = None,
    concurrency: Optional[int] = None,
    compute: Optional[ComputeStrategy] = None,
    partition_filter: Optional[PathPartitionFilter] = None,
    block_udf: Optional[Callable[[Any], Any]] = None,
)

@wrap_auto_init
def _read_datasource_v2(
    datasource,
    *,
    parallelism: int = -1,
    num_cpus: Optional[float] = None,
    num_gpus: Optional[float] = None,
    memory: Optional[float] = None,
    ray_remote_args: Optional[Dict[str, Any]] = None,
    concurrency: Optional[int] = None,
    compute: Optional[ComputeStrategy] = None,
    partition_filter: Optional[PathPartitionFilter] = None,
    block_udf: Optional[Callable[[Any], Any]] = None,
) -> Dataset:
    """Internal entry point for ``DataSourceV2`` reads.

    Wires a ``ListFiles → ReadFiles`` logical chain:

    - :class:`ListFiles` owns listing (via the datasource's ``FileIndexer``),
      optional global shuffle (``FileShuffleConfig``), and size-balanced
      bucketing (``RoundRobinPartitioner``). Its physical planner
      parallelizes listing across path shards and emits manifest blocks.
    - :class:`ReadFiles` consumes the manifest blocks and reads each bucket
      via ``scanner.create_reader().read(manifest)``.

    Schema inference happens once on the driver by sampling the first
    file — no caching layer needed.
    """
    import time

    from ray.data._internal.datasource_v2.listing.listing_utils import (
        _build_pruners,
        sample_files,
    )
    from ray.data._internal.datasource_v2.partitioners.round_robin_partitioner import (
        RoundRobinPartitioner,
    )
    from ray.data.datasource.file_based_datasource import FileShuffleConfig

    ctx = DataContext.get_current()

    if not datasource.supports_distributed_reads:
        import ray.util.client

        if ray.util.client.ray.is_connected():
            raise ValueError(
                "Because you're using Ray Client, read tasks scheduled on the "
                "Ray cluster can't access your local files. To fix this issue, "
                "store files in cloud storage or a distributed filesystem like "
                "NFS."
            )

    ray_remote_args = _resolve_read_remote_args(
        datasource,
        ray_remote_args,
        num_cpus,
        num_gpus,
        memory,
        ctx,