Support for progressive parquet chunked reading.

[nvdbaranec]

Previously, the parquet chunked reader operated by controlling the size of output chunks only. It would still ingest the entire input file and decompress it, which can take up a considerable amount of memory. With this new 'progressive' support, we also 'chunk' at the input level. Specifically, the user can pass a pass_read_limit value which controls how much memory is used for storing compressed/decompressed data. The reader will make multiple passes over the file, reading as many row groups as it can to attempt to fit within this limit. Within each pass, chunks are emitted as before.

From the external user's perspective, the chunked read mechanism is the same. You call has_next() and read_chunk(). If the user has specified a value for pass_read_limit the set of chunks produced might end up being different (although the concatenation of all of them will still be the same).

The core idea of the code change is to add the idea of the internal pass. Previously we had a file_intermediate_data which held data across read_chunk() calls. There is now a pass_intermediate_data struct which holds information specific to a given pass. Many of the invariant things from the file level before (row groups and chunks to process) are now stored in the pass intermediate data. As we begin each pass, we take the subset of global row groups and chunks that we are going to process for this pass, copy them to out intermediate data, and the remainder of the reader reference this instead of the file-level data.

In order to avoid breaking pre-existing interfaces, there's a new contructor for the chunked_parquet_reader class:

  chunked_parquet_reader(
    std::size_t chunk_read_limit,
    std::size_t pass_read_limit,
    parquet_reader_options const& options,
    rmm::mr::device_memory_resource* mr = rmm::mr::get_current_device_resource());

Checklist

• I am familiar with the Contributing Guidelines.
• New or existing tests cover these changes.
• The documentation is up to date with these changes.

[ttnghia]

Can we move this into the initializer list of this constructor?

[nvdbaranec]

Can't initialize a parent's member as part of your own initialization list.

[ttnghia]

I've removed a lot of member functions in our previous work for chunked reader, because they can be just free functions. So if possible, please remove such declaration and make them free functions. Having them as member functions, you will have to maintain their signatures here.

[nvdbaranec]

If these were made free functions they would all have to be passed large numbers of parameters, which would the code a lot harder to read.

[vuule]

review.flush()

[hyperbolic2346]

Some comments on first pass. I'll make another one tomorrow with a clear head.

[hyperbolic2346]

Above it says decompression limit, but here it says decompress and reading. I am also concerned by the soft limit. This seems like the thing that you have hard limits. Should it explode on over limit or is the OOM the explosion and that is why it is considered a soft limit?

[nvdbaranec]

This is a soft limit because it will attempt to continue even if it can't meet the limit. For example, if the user specified 1 MB and it can't fit even one row group (say 50 MB) into that size, it will still attempt to read/decompress one row group at a time.

[vuule]

another flush, mostly comments coming from difficulty to understand the code.
Still got some parts to dig into.

[vuule]

few more small suggestions, looks good overall.
I think the biggest issue for me is the fragility of the test, which we discussed offline and found a solution for.

[vuule]

/merge