[owenowenisme]

Description

We previously used fragment.subset to construct Parquet sub-fragments. However, it eagerly fetches footers during construction, which causes performance issues when many small files are packed into a single task. Because sub-fragment construction runs sequentially, this eagerly fetched metadata becomes a major bottleneck.

This PR switches to fragment.format.make_fragment, deferring the footer fetching to the scanning stage. Since scanning runs across multiple threads, the footer retrieval is now fully parallelized.

Mirco benchmark

Reading sf1000 supplier table on my laptop

	baseline	this PR
read tasks	2	2
largest task wall	392.45 s	44.25 s
e2e wall	413.1 s	64.8 s
rows	10,000,000	10,000,000

Related issues

Link related issues: "Fixes #1234", "Closes #1234", or "Related to #1234".

Additional information

Optional: Add implementation details, API changes, usage examples, screenshots, etc.

[gemini-code-assist]

Code Review

This pull request updates the Parquet file chunking utilities in Ray Data. Specifically, in parquet_file_chunking_utils.py, when per_row_group_offsets is not present, it now lazily creates a fragment using fragment.format.make_fragment instead of recursively calling _subset. There are no review comments, and we have no feedback to provide.

[Yin-1022]

I've verified the behavior locally with a Parquet file containing multiple row groups.

• A fragment rebuilt with make_fragment(..., row_groups=[0, 2]) preserved the selected non-contiguous row groups, partition expression, and produced the same table as fragment.subset(row_group_ids=[0, 2]).

• After deleting the underlying file, fragment.subset(...) failed immediately during fragment construction, while make_fragment(...) still constructed successfully and only failed when the fragment was scanned. This confirms that the footer/file access is deferred as intended.

The change looks correct to me. LGTM.