[Data] Parallelize ListFiles for the footer indexer; per-test bin budgets

## Listing parallelism

`ListFiles` forced single-task listing whenever the indexer sets
`yields_read_units` -- i.e. always, for Parquet V2 -- on the grounds that
the bin packer "must see the whole file stream to pack globally". But the
V1 partitioner it replaced also runs inside each listing task, so global
packing was never the status quo, and the cost is real: a read that hands
in explicit paths loses listing parallelism entirely.

`distributed_training` passes 200 explicit parquet paths. Master sharded
them across 200 listing tasks (3.00s); the footer path did all 200 footer
reads in one (5.81s), ~65% of that test's +15% regression, with block
count and block size otherwise identical to master.

Drop `yields_read_units` from the `should_parallelize` condition. Only the
shuffle-RNG case still needs a single task. Reads that pass one directory
-- the common case -- shard to one task either way and are unaffected.

The footer reader pool is provisioned per `list_files` call, so it is now
per listing task rather than cluster-wide. Documented as such on
`RAY_DATA_PARQUET_FOOTER_NUM_ACTORS`, and `distributed_training` sets it
to 1 since each of its tasks reads a single footer.

## Per-test bin budgets

The 64 MiB default is right for some datasets and badly wrong for others,
because the useful budget tracks per-file projected bytes -- which ranges
from 0.45 MiB to 3.87 GiB across the suite, and varies with the query's
projection on the same table. Measured from parquet footers and verified
against the V1 partitioner's block counts:

- read_large_parquet         4 GiB    (3.87 GiB/file; 64 MiB -> 5768 tasks)
- write_parquet, joins,      1.25 GiB (~1.17 GiB/file, full schema)
  aggregate_groups, map_groups
- tpch_q20                   192 MiB  (160 MiB/file, 4-column projection)
- streaming_split            36 MiB   (V1 landed on ~31 MB blocks here)

The map_groups value also fixes four outright failures, not just a
slowdown: `map_groups_*_sort_shuffle_pull_based` OOM'd the driver at
64 MiB. Pull-based shuffle holds `input_blocks * output_blocks` object
refs on the driver, so at 2500 blocks that is 17.5 GiB (observed 21.6 GiB
RSS, cgroup kill). At 1.25 GiB it is ~95 blocks and 0.03 GiB.

`wide_schema` gets no budget: those datasets are 10-40 MB total across
9-27 single-row-group files, so every budget from 64 MiB up packs them
into one bin. They need a minimum-block-count floor instead. `mix` gets
no budget either -- it already matches V1 granularity -- only a narrower
reader pool, since it opens 8 datasets over the same 513-file path.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Signed-off-by: Goutam <goutam@anyscale.com>

Author: goutamvenkat-anyscale

python/ray/data/_internal/datasource_v2/listing/file_indexer.py

@@ -42,8 +42,7 @@ def yields_read_units(self) -> bool:
         manifests that still need size-balanced partitioning downstream. An
         indexer that bin-packs internally (e.g. the footer-based Parquet indexer,
         which reads footers and packs row groups into ~one-block manifests)
-        returns ``True``; ``ListFiles`` then skips the partitioner and runs
-        listing as a single task so packing sees the whole file stream.
+        returns ``True``; ``ListFiles`` then skips the partitioner.
         """
         return False
 

python/ray/data/_internal/datasource_v2/listing/footer_file_indexer.py

@@ -28,9 +28,12 @@
 
 logger = logging.getLogger(__name__)
 
-# A pool of footer-reading actors spread across the cluster. Footer reads are
-# network-bound, so several actors each driving many concurrent reads keeps IO
-# from bottlenecking on a single node.
+# Footer-reading actors spread across the cluster, provisioned **per ListFiles
+# task**. Footer reads are network-bound, so several actors each driving many
+# concurrent reads keeps IO from bottlenecking on a single node. ``ListFiles``
+# shards user-supplied paths across parallel listing tasks, so a read that hands
+# in many explicit paths stands up this many actors per shard -- lower it for
+# those, rather than assuming it is a cluster-wide total.
 _DEFAULT_NUM_ACTORS = env_integer("RAY_DATA_PARQUET_FOOTER_NUM_ACTORS", 32)
 _DEFAULT_IO_CONCURRENCY = env_integer("RAY_DATA_PARQUET_FOOTER_IO_CONCURRENCY", 128)
 # Files per ``read_footers`` call. Small footers -> batch several per task to
@@ -104,7 +107,7 @@ def __init__(
     @property
     def yields_read_units(self) -> bool:
         # list_files already emits bin-packed read units, so ListFiles skips the
-        # partitioner and lists in a single task (global packing + one pool).
+        # partitioner. Packing is per listing task, not global.
         return True
 
     def list_files(

python/ray/data/_internal/planner/plan_list_files_op.py

@@ -71,10 +71,18 @@ def plan_list_files_op(
     shuffle_config = op.shuffle_config_factory()
 
     # Some indexers (e.g. the footer-based Parquet indexer) already emit
-    # bin-packed read units from ``list_files`` -- they need the whole file
-    # stream on one task to pack globally, and there's nothing left to partition.
+    # bin-packed read units from ``list_files``, so there's nothing left for a
+    # partitioner to do. They still list in parallel: each task bin-packs the
+    # slice of the path list it was handed, which is the same per-task locality
+    # a partitioner gets when listing is split.
     yields_read_units = indexer.yields_read_units
 
+    # One task is required only when shuffle needs a single global RNG over the
+    # whole listing. Note that per-call indexer resources -- the footer indexer's
+    # reader-actor pool -- are provisioned per listing task, so a read handing in
+    # many explicit paths should size that pool accordingly.
+    should_parallelize = shuffle_config is None
+
     transform_fns: List[MapTransformFn] = [
         BlockMapTransformFn(
             partial(
@@ -120,12 +128,7 @@ def plan_list_files_op(
     map_op = MapOperator.create(
         map_transformer,
         _create_input_data_buffer(
-            op,
-            data_context,
-            # A single task is required when shuffle needs one global RNG over
-            # the full listing, or when the indexer bin-packs read units itself
-            # (it must see the whole file stream to pack globally).
-            should_parallelize=shuffle_config is None and not yields_read_units,
+            op, data_context, should_parallelize=should_parallelize
         ),
         data_context,
         name="ListFiles",

release/release_data_tests.yaml

@@ -46,6 +46,13 @@
   python: "3.10"
   cluster:
     anyscale_sdk_2026: true
+    byod:
+      runtime_env:
+        # Files here are ~3.9 GiB uncompressed with ~69 MiB row groups. The
+        # default bin budget can't fit two row groups, so every one becomes its
+        # own read task (~5.8k tasks for 103 files). 4 GiB packs a file per bin,
+        # matching the whole-file granularity the V1 partitioner produced.
+        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=4294967296
     cluster_compute: "{{scaling}}_cpu_compute.yaml"
 
   matrix:
@@ -113,6 +120,13 @@
   python: "3.10"
   cluster:
     anyscale_sdk_2026: true
+    byod:
+      runtime_env:
+        # sf1000 lineitem is read with its full 17-column schema: ~1.16 GiB
+        # uncompressed per file over 1000 files. 1.25 GiB packs ~one file per
+        # bin (~943 read tasks vs ~1000 blocks under V1); the default budget
+        # would produce ~18k tasks.
+        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=1342177280
   run:
     timeout: 3600
     script: >
@@ -201,6 +215,12 @@
         - RAYTEST_FAIL_ON_WORKER_OOM=1
         - RAYTEST_FAIL_ON_DEAD_NODES=1
         - RAYTEST_FAIL_ON_SPILLING=0
+        # ``groupby_benchmark`` reads sf100 lineitem with its full 17-column
+        # schema: ~1.17 GiB uncompressed per file over 100 files. 1.25 GiB packs
+        # ~one file per bin (~95 read tasks), matching the whole-file
+        # granularity of the V1 partitioner. The default budget yields ~2500
+        # read tasks, which starves the downstream shuffle.
+        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=1342177280
     cluster_compute: "{{scaling}}_all_to_all_compute.yaml"
 
   run:
@@ -228,6 +248,9 @@
         - RAYTEST_FAIL_ON_WORKER_OOM=1
         - RAYTEST_FAIL_ON_DEAD_NODES=1
         - RAYTEST_FAIL_ON_SPILLING=0
+        # Same sf100 lineitem full-schema read as the aggregate variants; see
+        # the note there for why 1.25 GiB.
+        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=1342177280
     cluster_compute: "{{scaling}}_all_to_all_compute.yaml"
 
   run:
@@ -269,6 +292,10 @@
         - RAYTEST_FAIL_ON_WORKER_OOM=1
         - RAYTEST_FAIL_ON_DEAD_NODES=1
         - RAYTEST_FAIL_ON_SPILLING=0
+        # Both sides are read with their full schema: ~1.17 GiB uncompressed
+        # per lineitem file. 1.25 GiB packs ~one file per bin (~95 read tasks
+        # vs 100 blocks under V1).
+        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=1342177280
     cluster_compute: fixed_size_100_cpu_compute.yaml
 
   matrix:
@@ -304,7 +331,16 @@
         - RAYTEST_FAIL_ON_SPILLING=1
         # S3 tensor data was written by Ray 2.49-2.54 using cloudpickle.
         - RAY_DATA_AUTOLOAD_CLOUDPICKLE_TENSOR_METADATA=1
-        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=39000000
+        # No bin budget override here. These datasets are 10-40 MB in total
+        # across 9-27 single-row-group files, so every budget from 64 MiB up
+        # packs them into a single bin -- except `tensors`, whose 37.25 MiB row
+        # groups already exceed the 64 MiB default's two-per-bin threshold and
+        # so land one per bin on their own. Splitting `primitives` (0.45 MiB
+        # per file), `nested_structs` (1.3 MiB) and `objects` (4.2 MiB) needs a
+        # minimum-block-count floor, not a byte budget.
+        #
+        # These datasets are small enough that the footer-reader pool is pure
+        # startup cost; keep it narrow and spread the few files across it.
         - RAY_DATA_PARQUET_FOOTER_NUM_ACTORS=22
         - RAY_DATA_PARQUET_FOOTER_BATCH_SIZE=1
         - RAY_DATA_PARQUET_READER_IO_THREAD_COUNT=5000
@@ -328,6 +364,17 @@
   python: "3.10"
   cluster:
     anyscale_sdk_2026: true
+    byod:
+      runtime_env:
+        # imagenet parquet: 7416 single-row-group files of ~8.15 MiB. The V1
+        # partitioner sized buckets off a 5x encoding-ratio estimate, which
+        # over-counts for already-compressed image bytes, so it landed on ~4
+        # files (~31 MB) per block. The footer path measures the real bytes, so
+        # the default budget produces ~66 MB blocks -- 2x coarser. With
+        # ``equal=True`` the splitter buffers whole blocks, so coarser blocks
+        # mean more read-ahead is thrown away when the consumers stop. 36 MiB
+        # reproduces the ~4-files-per-block granularity.
+        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=37748736
   run:
     timeout: 300
     wait_for_nodes:
@@ -359,6 +406,17 @@
   python: "3.10"
   cluster:
     anyscale_sdk_2026: true
+    byod:
+      runtime_env:
+        # No bin budget override: at the default this dataset (513 single-row-
+        # group files of ~88 MB) already packs ~one file per bin, matching the
+        # granularity the V1 partitioner produced.
+        #
+        # This benchmark opens 8 datasets over the same path, so the footer
+        # path costs 8 x 513 footer reads and 8 x 32 reader actors -- work V1
+        # never did, and enough to dominate the ~40s `random_mix` variants.
+        # Narrow the pool; the per-actor IO concurrency still covers 513 files.
+        - RAY_DATA_PARQUET_FOOTER_NUM_ACTORS=4
     cluster_compute: dataset_mixing/compute_8_cpu.yaml
   run:
     timeout: 600
@@ -404,6 +462,12 @@
     anyscale_sdk_2026: true
     byod:
       post_build_script: byod_install_mosaicml.sh
+      runtime_env:
+        # This benchmark hands in 200 explicit parquet paths, so ListFiles
+        # shards them into 200 tasks of one path each. The footer reader pool is
+        # provisioned per listing task, so the default would stand up 32 actors
+        # to read a single footer. One actor per task is all this shape needs.
+        - RAY_DATA_PARQUET_FOOTER_NUM_ACTORS=1
     cluster_compute: dataset/multi_node_train_16_workers.yaml
 
   run:
@@ -1443,6 +1507,15 @@
 
   cluster:
     anyscale_sdk_2026: true
+    byod:
+      runtime_env:
+        # q20 projects 4 of lineitem's 17 columns -> ~160 MiB uncompressed per
+        # file. At the default budget that splits each file ~4 ways (400 blocks
+        # vs the 100 the V1 partitioner produced), and the extra blocks become
+        # extra shuffle-map tasks feeding HashAggregate. 192 MiB packs one file
+        # per bin. Per-query value: the same table under q21's 2-column
+        # projection is only ~92 MiB per file.
+        - RAY_DATA_PARQUET_BIN_PACKING_BYTES=201326592
     cluster_compute: "{{scaling}}_all_to_all_compute.yaml"
 
   run: