This PR reduces the maximum memory usage of writing Parquet files with order-preservation enabled in parallel.

The way the order-preserving parallel Parquet writer works is that it materializes batches of data based on their batch index, then repartitions them into the desired row group sizes, and finally writes the batches out to disk (implemented in #7375).

The problem here is that writing is disconnected from the materialization of batches - and only a single thread can write to the file at the same time. If the other threads materialize data faster than the single thread can write, memory usage will keep on growing, as the other threads are not prevented from accumulating more data in their buffers. In this situation, memory can keep on increasing until the entire Parquet file is read. This is a problem for Parquet files whose uncompressed contents do not fit in memory.

In this PR we address this issue by adding a backpressure mechanism to the batch copy. We use the temporary memory manager to reserve space for the materialized batches (with a maximum of 25% of the memory limit). As threads materialize data, we use the size of the respective ColumnDataCollections (through a new method AllocationSize) to keep track of how much memory we have gathered. If we exceed the available memory, we block the thread from continuing to materialize data. Halted threads may still help with repartitioning and preparing batches through ExecuteTasks, however.

The thread that is processing the minimum batch index can always continue - as the data that that thread is materializing can be written to disk immediately. Note that we do not yet deal with this in a nice way if a single batch index contains loads of data as we currently only repartition & flush after the batch index has been exhausted. However, this scenario is relatively rare in practice and can only really happen when streaming directly from Parquet files that have been written with very large row groups.

Max Threads

This PR adds a new MaxThreads function to the PhysicalSink. This can be used to limit the parallelism of a pipeline at the sink level. This is used for the order-preserving write as processing data with many threads and a low memory footprint is actually counter-productive and will lead to lower performance than using fewer threads, as well as a lower memory footprint. In the batch copy as a heuristic we require at least 4MB per column per thread of available memory. The amount of threads can be scaled down automatically when too little memory is available.

Tests & Benchmarks

Below are some tests and benchmarks:

Lineitem SF10 Parquet (2.5GB compressed, 5.4GB uncompressed, 10GB CSV)

ClickBench Hits Parquet (14GB compressed, 36GB uncompressed, 80GB CSV)