[FEA] Allow `read_parquet` with preprocessed `datasources` and `aggregate_reader_metadata`

[wence-]

Is your feature request related to a problem? Please describe.

When performing multi-threaded reads over a set of parquet data sources, this is a typical pattern:

• read metadata to determine the size of the file (e.g. row splits, byte splits)
• in parallel: for all splits, call read_parquet.

One might ask why not use the chunked parquet reader for this setup, there are two reasons:

1. The chunked reader can only deliver outputs on a single stream. Therefore we can't get multi-stream pipelining (even when using multiple threads) from a single chunked reader.
2. You can't drive the chunked reader from multiple host threads because it has too much internal state.

Parquet reading proceeds in two phases:

1. read parquet file metadata
2. read the actual data

When we split the input sources into chunks, we can't save reading the actual data each time, but I would like to avoid reading the metadata multiple times.

Today, to minimise this, I do something like the following:

• read metadata from the first file in the datasource to guesstimate how many total rows (bytes)
• chunk up the files so that a read will produce (hopefully) at least the target number of rows (bytes) per output chunk.
• launch the tasks on those chunks of files

I therefore end up reading metadata for each chunk num_tasks_per_chunk times. And my code is more complicated.

Describe the solution you'd like

Ideally what I would do is:

• read metadata from files
• launch tasks using read metadata

Something like:

auto [datasources, metadata] = prepare_parquet_metadata(source_info, options);

parallel_for (...) { 
    auto chunk_options = parquet_reader_options{options};
    // slice out the piece we want.
    chunk_options.set_num_rows(...);
    chunk_options.set_skip_rows(...);
    auto chunk = read_parquet(chunk_options, metadata, datasources);
    ...
}

This way, I only the pay the cost of reading metadata and preparing the datasources once.

Additional context

I had a look at doing this but it's significantly complicated (to my mind) by the reader_impl storing a vector of unique_ptrs to the datasources, and a unique_ptr to the prepared aggregate_reader_metadata. So it's not "just" as simple as threading everything through.

[JigaoLuo]

I just came across this and had some similar ideas. I’d like to add one point to the pattern and the reason:

• Pattern to add: Reading RowGroup splits. Most Parquet writers generate RowGroups based on row counts. RG semantics also make pruning straightforward.
• Reason to add: Supporting Parquet reads from cloud object storage. Other Parquet readers, such as arrow‑rs, have recently added metadata caching support.

Supporting RowGroup reads is extremely powerful, as you mentioned. I believe that standard Parquet reading each RG with metadata cached is the natural entry point to drive cuDF Parquet reader toward fully multi‑stream pipelining. I have some figures in my paper and will publish them soon.

The above was part of our discussion with @mhaseeb123 a few months ago. Thank you again!

---

My experiment was conducted on my patch of the invasive POC #18891 for metadata caching. However, I believe the performance should be similar, even though we implemented the metadata cache differently.

[mhaseeb123]

Pattern to add: Reading RowGroup splits. Most Parquet writers generate RowGroups based on row counts. RG semantics also make pruning straightforward.

I think you can do this today by caching the metadata and datasource and then launching multiple parquet readers, each with a different split of row groups via options provided to each of them: options.set_row_groups({indices})