[Data] [11/11] Enable footer-based Parquet reads by default

[goutamvenkat-anyscale]

Why

The last step of the split of #64985: flip Parquet V2 to the footer read path, delete the blind one, and tune the release tests so this can be validated by a real release run.

Everything it depends on has landed — footer types and the bin packer (#65210), the FooterReader actor pool (#65273), the indexer plus count() fix and packer-as-partitioner (#65596), reader IO tuning (#65806), and test env pinning (#65169). The path has been reachable behind RAY_DATA_PARQUET_ENABLE_FOOTER_INDEXER since #65596; this makes it the only path.

Flow

flowchart LR
  RP[read_parquet] --> FF["FooterFileIndexer<br/>+ OnlineBinPacker"]
  RP -.->|deleted| PC["ParquetFileChunker<br/>size-estimated chunks<br/>+ parquet_chunker_target_chunk_size"]
  style PC stroke-dasharray: 4 4

Loading

What changed

The flip — RAY_DATA_PARQUET_ENABLE_FOOTER_INDEXER is gone from both call sites. _get_file_indexer() always returns FooterFileIndexer; get_file_partitioner() always returns OnlineBinPacker.

The deletions — ParquetFileChunker, ParquetFileChunkMetadata, _calculate_row_group_range, _fragments_from_chunk_metadata, and the tests covering only that path.

The config it left behind — DataContext.parquet_chunker_target_chunk_size and DEFAULT_PARQUET_CHUNKER_TARGET_CHUNK_SIZE were read only by ParquetFileChunker.__init__, so they would otherwise stay declared and documented while doing nothing. Verified dead by grep across .py / .yaml / .rst / .md.

Reader pool defaults — RAY_DATA_PARQUET_READER_IO_THREAD_COUNT 128, RAY_DATA_PARQUET_READER_CPU_COUNT 32.

Release tests tuned

Budgets are per-test overrides of RAY_DATA_PARQUET_BIN_PACKING_BYTES (uncompressed bytes of row-group data per read task; default 128 MiB). The shape of the tuning: reads over few large groups want large bins, because small bins shatter the read into short tasks the executor cannot dispatch fast enough and the cluster idles; shuffle-bound reads over many groups want small bins.

Rows whose variants all share the same tuning are collapsed and marked.

release test	bin budget	other tuning
aggregate_groups_fixed_size_hash_shuffle_column02 column14	32 MiB	—
aggregate_groups_fixed_size_hash_shuffle_column08 column13 column14	128 MiB	—
aggregate_groups_fixed_size_shuffle_v2_column02 column14	64 MiB	—
aggregate_groups_fixed_size_shuffle_v2_column08 column13 column14	512 MiB	—
aggregate_groups_fixed_size_sort_shuffle_pull_based_column02 column14	32 MiB	—
aggregate_groups_fixed_size_sort_shuffle_pull_based_column08 column13 column14	256 MiB	—
count_parquet_fixed_size	64 MiB	—
cross_az_map_batches_autoscaling_iptable_failure_injection	1 GiB	—
distributed_training	—	footer actors 20
flat_map	64 MiB	—
iceberg_benchmark_overwrite	64 MiB	—
iter_batches_pyarrow	64 MiB	—
iter_torch_batches	64 MiB	—
joins_{{dataset}}_{{join_type}} *(all 4 variants)*	1.25 GiB	—
map	64 MiB	—
map_batches_fixed_size_{{compute}}_{{format}}_{{repeat_map_batches}} *(all 7 variants)*	1 GiB	scale-up threshold 0.6
map_groups_fixed_size_{{shuffle_strategy}}_{{columns}} *(all 6 variants)*	1.25 GiB	—
mix	—	footer actors 4
random_shuffle_fixed_size	64 MiB	—
read_large_parquet_fixed_size	1 GiB	—
read_parquet_fixed_size	64 MiB	—
read_tfrecords	64 MiB	—
streaming_split	64 MiB	footer batch 50, result batch 50
to_tf	64 MiB	—
tpch_q1_fixed_size_{{shuffle_strategy}} *(all 2 variants)*	192 MiB	—
training_ingest_benchmark	64 MiB	—
wide_schema_pipeline_nested_structs	2 MiB	footer batch 1, footer actors 21, io threads 5000
wide_schema_pipeline_objects	4.2 MiB	footer batch 1, footer actors 9, io threads 5000
wide_schema_pipeline_primitives	256 KiB	footer batch 1, footer actors 27, io threads 5000
wide_schema_pipeline_tensors	39 MiB	footer batch 1, footer actors 22, io threads 5000
write_parquet	1.25 GiB	—

Three matrices are split, because a matrix cannot vary byod.runtime_env per value and their variants want different budgets: wide_schema_pipeline_{{data_type}} into four entries, and one variant each out of iceberg_benchmark_{{mode}} and iter_batches_{{format}}. aggregate_groups keeps its matrix and carries the budget as a matrix dimension with explicit adjustments. Rendered test names are unchanged — 121 before and after — so release baselines still line up.

Not carried over from the original branch: 7 previously-pinned entries no longer exist — the autoscaling variants removed in #65506, plus tpch_q20 and tpch_q22.

Any entry gaining its own runtime_env re-lists the four DEFAULTS guards verbatim: deep_update overwrites lists rather than merging them, so declaring one silently drops the OOM / dead-node / object-store checks. (17 entries on master are already missing guards for this reason; pre-existing, not touched here.)

map_benchmark.py also drops its use_datasource_v2 = False pin, which existed because V2 was spilling.

Budgets originally measured against the pre-decoupling design and refined since. Packing moved between stages in #65596 but the packer and its inputs did not, so bins should be unchanged — a release run against this PR is what confirms that. The wide-schema tests are the first place to look.

Behavior change worth reviewing

Read-task sizing no longer comes from override_num_blocks / read_op_min_num_blocks for Parquet. Bins are sized by RAY_DATA_PARQUET_BIN_PACKING_BYTES, which is also not derived from DataContext.target_max_block_size. That decoupling is deliberate — read units are row-group-accurate rather than size-estimated — but it means the two knobs users reach for first no longer affect Parquet read parallelism. Deriving the budget from target_max_block_size is worth considering as a follow-up; the per-test overrides above are the evidence that a single global default does not fit every workload.

Testing

$ python -m pytest -q python/ray/data/tests/datasource/test_parquet.py
286 passed, 6 skipped in 521.25s

$ python -m pytest -q python/ray/data/tests/datasource/test_read_parquet_v2.py
32 passed

$ python -m pytest -q python/ray/data/tests/unit/datasource_v2/ python/ray/data/_internal/datasource_v2/tests/
181 passed

$ python -m pytest -q python/ray/data/tests/test_predicate_pushdown.py \
                     python/ray/data/tests/test_execution_optimizer_limit_pushdown.py
78 passed

The 286-test parquet suite is the one that matters — it runs the full read surface with the footer path as the default rather than behind a flag. The release YAML is additionally checked for parse validity, unresolved {{...}} in any pin, duplicate byod: keys, and rendered-name stability.

Stack

Final step of an 11-step split of #64985 (43 files / +3234 −719, not reviewable as one unit).

• [Data] [1/11] Extract FileIndexer.list_file_infos #65167 [1] · [Data] [2/11] Push limit into ReadFiles when Limit sits directly on it #65168 [2] · [Data] [3/11] Derive ListFiles pushdown state from the ReadFiles scanner #65214 [3] · [Data] [4/11] Add Parquet footer types and the online bin packer #65210 [4] · [Data] [5/11] Add the Parquet FooterReader actor pool #65273 [5] · [Data] [6/11] Add FooterFileIndexer behind an opt-in flag, with bin packing as a partitioner #65596 [6] · [Data] [8/11] Tune Parquet reader IO threads and Arrow pool sizing #65806 [8] · [Data] [9/11] Pin the Parquet footer actor pool to 1 in tests #65169 [9] — all merged
• [7] (count push-down fix) was absorbed into [Data] [6/11] Add FooterFileIndexer behind an opt-in flag, with bin packing as a partitioner #65596
• [10] (release tuning) is folded into this PR, so a release run exercises the flip and its tuning together
• this PR [11/11]

---

Not a duplicate. File-level overlap check: #63158 none; #65142 overlaps file_indexer.py / context.py; #65406 / #65407 (arrow-rs reader A/B) overlap broadly but are explicitly [DO NOT MERGE] benchmark branches carrying the footer work out-of-tree.
AI assistance was used; every line reviewed by me and tests run locally.

[gemini-code-assist]

Code Review

This pull request removes the legacy size-based Parquet chunking path (including ParquetFileChunker, ParquetFileChunkMetadata, and the NonSamplingFileIndexer fallback) in favor of the footer-based chunking path (FooterFileIndexer and OnlineBinPacker), making it the default and only path for Parquet V2 reads. This cleanup involves removing obsolete configuration options, unit tests, and environment variables, as well as tuning the release benchmarks for the new footer-based bin packing. A review comment suggests adding defensive checks in parquet_file_reader.py to prevent potential KeyError exceptions when accessing chunk_metadata keys.

[gemini-code-assist]

Defensive programming: Accessing 'chunk_metadata["row_group_ids"]' directly without checking if the key exists can raise a 'KeyError' if 'chunk_metadata' is not 'None' but does not contain 'row_group_ids' (e.g., if a custom or legacy chunk metadata format is encountered). Checking for 'row_group_ids' in 'chunk_metadata' and falling back to reading the whole fragment is safer and more robust.

Suggested change

	else:
	fragments.extend(
	_fragments_from_row_group_ids(
	fragment,
	chunk_metadata["row_group_ids"],
	per_row_group_offsets=self._include_row_hash,
	)
	)
	elif "row_group_ids" in chunk_metadata:
	fragments.extend(
	_fragments_from_row_group_ids(
	fragment,
	chunk_metadata["row_group_ids"],
	per_row_group_offsets=self._include_row_hash,
	)
	)
	else:
	fragments.append((fragment, 0))

[ayushk7102]

Do we test integration of the FileIndexer with our online bin packer as the file chunker? I wasn't able to find a test for this, and I think we should test for this end to end

We already unit test online bin packer

[goutamvenkat-anyscale]

Added test_footer_indexer_feeds_online_bin_packer

[ayushk7102]

LGTM