Writing Parquet Row Groups in Parallel: Column Chunks, Dictionary Pages and the Sort-Before-Write Trick

TL;DR: The Parquet write path in DataFusion is a streaming, one-pass pipeline: batches flow in, get fanned out to per-column encoders, get sealed into row groups, and get concatenated into a file that is uploaded as it is produced. Row groups are sized by row count (and optionally by bytes), dictionary encoding and page statistics are configured once and inherited by the reader’s pruning machinery, and — the trick worth stealing — when the input plan is already sorted, the sort order is written into the Parquet footer so the reader gets the ordering for free instead of paying for a sort.

The shape of the write path

There are two entry points, and they are not the same thing.

plan_to_parquet is the simple one — one file per output partition, one writer task per partition, each task pulling batches from its partition’s stream and pushing them into an AsyncArrowWriter (writer.rs:41-72). The buffer it uses, objectstore_writer_buffer_size (10 MiB by default, datafusion/common/src/config.rs:1159), exists mostly so that a very large output file does not hit remote-endpoint limits on the size of a single upload. It is a transport knob, not a format knob.

The interesting path is ParquetSink (sink.rs), which is what COPY ... TO and INSERT INTO actually use. It receives a demultiplexed stream of batches tagged with the file they belong to (hive partitioning splits the output into many files), and it has two writer strategies.

Two writer strategies

The sequential path is parquet::arrow::AsyncArrowWriter: one file, one task, batches handed in order. It is used when allow_single_file_parallelism is false, or when content-defined chunking is enabled — CDC needs the chunker state that lives inside ArrowWriter and persists across row groups (sink.rs:304-308). The parallel path is DataFusion’s own code, and it is the interesting one.

Parallel column encoding inside one file

The idea: a Parquet row group is a set of independent column chunks. There is no reason to encode column a of row group 7 before column b of row group 7. So DataFusion spawns one task per leaf column and feeds them arrays through per-column channels:

// datafusion/datasource-parquet/src/sink.rs:630-645
for writer in col_writers.into_iter() {
    // Buffer size of this channel limits the number of arrays queued up for column level serialization
    let (send_array, receive_array) = mpsc::channel::<ArrowLeafColumn>(max_buffer_size);
    col_array_channels.push(send_array);

    let reservation =
        MemoryConsumer::new("ParquetSink(ArrowColumnWriter)").register(pool);
    let task = SpawnedTask::spawn(column_serializer_task(
        receive_array, writer, reservation, encoding_time.clone(),
    ));
    col_writer_tasks.push(task);
}

The channel depth is maximum_buffered_record_batches_per_stream (default 2, config.rs:1561-1566) — deliberately small, because this is a streaming plan and the memory budget is the scarce resource. The option’s comment says it plainly: the defaults are “tuned for minimum memory usage in a streaming execution plan”, and raising them “is likely worthwhile when writing out already in-memory data”.

A batch is then shredded into leaf arrays and pushed to the right channel:

// datafusion/datasource-parquet/src/sink.rs:684-695
let mut next_channel = 0;
for (array, field) in rb.columns().iter().zip(schema.fields()) {
    for c in compute_leaves(field, array)? {
        if col_array_channels[next_channel].send(c).await.is_err() {
            return Ok(());
        }
        next_channel += 1;
    }
}

compute_leaves is what makes this work for nested schemas: a List<Int32> column is one Arrow column but two Parquet leaves (the offsets and the elements), and each leaf needs its own encoder.

Row groups are sealed concurrently, concatenated in order

When a row group fills up, its column writers are closed and the finished column chunks are handed to a separate task, so the next row group can start encoding without waiting (sink.rs:777-817). The batch that straddles the boundary is split — rb.slice(0, rows_left) — and the remainder is carried into the next row group. That is why the row-group limit can be any number, not just a multiple of the batch size.

The finalizer joins the column tasks, closes each writer, and grows a memory reservation by the encoded size (sink.rs:712-727). The concatenator then writes them into the file, in the order the row groups were started, flushing to the object store when the shared buffer grows past a megabyte (sink.rs:859-876):

// datafusion/datasource-parquet/src/sink.rs:859-876
let mut rg_out = parquet_writer.next_row_group()?;
for chunk in serialized_columns {
    chunk.append_to_row_group(&mut rg_out)?;
    rg_reservation.free();

    let mut buff_to_flush = merged_buff.buffer.try_lock().unwrap();
    if buff_to_flush.len() > BUFFER_FLUSH_BYTES {
        object_store_writer.write_all(buff_to_flush.as_slice()).await?;
        buff_to_flush.clear();
    }
}
rg_out.close()?;

So the shape is: N column encoders running concurrently, R row groups in flight, and one writer that stitches the finished row groups into the file in order and ships bytes to the object store as it goes. The file is never materialized in memory, and the input is never read twice.

Row group sizing

Row group size is the single most consequential write-side decision, because it is the granularity at which the reader can prune. max_row_group_size defaults to 1M rows; max_row_group_bytes is None, and when set, bounds the row group by bytes instead, matching parquet.block.size in parquet-mr (config.rs:1494-1512).

The byte limit is the interesting one, and the reason it is None by default is a deliberate trade-off: the parallel writer ignores it. From the option’s own documentation: “Currently only honored when allow_single_file_parallelism is false; by default the parallel file writer ignores this limit.” The parallel writer computes its row-group boundary from max_row_group_row_count() alone (sink.rs:747-750), because bounding bytes in a design where row groups are encoded concurrently and asynchronously would require knowing the encoded size before the row group is finished — which is exactly the information that only exists after encoding.

Dictionary encoding and statistics: write-side decisions, read-side wins

The writer’s defaults come from ParquetOptions and are translated into WriterProperties in one place (datafusion/common/src/file_options/parquet_writer.rs:259-276): page size limit, write batch size, writer version, dictionary page size limit, statistics level, max row group rows and bytes, created_by, column index and statistics truncation lengths, data page row count limit, bloom filter switch, and the NDV-per-row-group switch.

The defaults that matter for a reader:

That last row is the sharpest example of a write-time decision with a read-time payoff. DataFusion’s reader can prune row groups with bloom filters (row_group_filter.rs:460-508), but a bloom filter has to be in the file first, and bloom_filter_on_write defaults to false. Turning it on costs space and write time and buys row-group skipping for equality predicates — a trade that is worth it for selective high-cardinality columns and not worth it for low-cardinality ones (where the min/max statistics already do the job).

Similarly, statistics_truncate_length = 64 is the reason that a string column’s row-group bounds are “Alp”..“Alq” rather than exact, and the reason full-match inference can fail on long strings (row_group_filter.rs:189-203).

The sort-before-write trick

Here is the trick worth stealing. Parquet has a sorting_columns field in its metadata: it declares that the file’s rows are sorted by these columns. DataFusion writes it when the input plan is already ordered, and reads it back to skip the sort.

Writing it (file_format.rs:645-668):

// Convert ordering requirements to Parquet SortingColumns for file metadata
let sorting_columns = if let Some(ref requirements) = order_requirements {
    let ordering: LexOrdering = requirements.clone().into();
    let writer_schema = get_writer_schema(&conf);
    // In cases like `COPY (... ORDER BY ...) TO ...` the ORDER BY clause
    // may not be compatible with Parquet sorting columns (e.g. ordering on `random()`).
    // So if we cannot create a Parquet sorting column from the ordering requirement,
    // we skip setting sorting columns on the Parquet sink.
    lex_ordering_to_sorting_columns(&ordering, conf.output_schema(), &writer_schema)
        .ok()
        .filter(|columns| !columns.is_empty())
} else {
    None
};

The order_requirements are not invented by the writer — they come from the plan: input_exec.properties().output_ordering() (datafusion/core/src/physical_planner.rs:930-940).

So: if the plan that feeds the sink already has an ordering — because the user wrote COPY (SELECT ... ORDER BY x) TO ..., or because an upstream operator produces sorted output — the writer records it in the footer. It does not impose an order; it only documents one it was given. That distinction is what makes it safe: the metadata is a claim the writer can make because it saw the data in that order. The conversion is conservative in the right way: only plain column references can become sorting columns (metadata.rs:1145-1151).

Reading it back, ordering_from_parquet_metadata (metadata.rs:1209-1236) turns the footer’s sorting columns into a LexOrdering, which the source advertises as its output ordering — and that is what lets the optimizer delete the SortExec above the scan.

The read side has a second, related trick for files that are not declared sorted: reorder the row groups by their min statistics, so a TopK query finds good row groups first (access_plan.rs:727-749). Row groups are sorted lexicographically by the min values of the longest prefix of the sort order made of plain columns present in the file schema, with secondary keys breaking ties when the leading column’s min is equal across row groups — without them the reorder is a no-op on such files and the TopK dynamic filter converges only as fast as disk order allows.

This is the “sort before write” idea from the other direction: if you cannot afford to sort at write time, you can at least approximate the order at read time using statistics you already paid for, and let the TopK’s dynamic filter prune the rest.

The catch

• The parallel writer bypasses ArrowWriter. Content-defined chunking is therefore disabled with it — the chunker state lives in ArrowWriter and persists across row groups (sink.rs:304-308). CDC forces the sequential path.
• max_row_group_bytes does not work with parallel writing. Setting it and leaving allow_single_file_parallelism = true gives you a row group limit that is silently not applied. This is documented in the option, but it is the kind of footgun that costs an afternoon.
• Ordering metadata is a promise, not a check. If the input plan claims an ordering it does not have, the footer will lie, and the reader will happily skip a sort it should have performed. DataFusion mitigates this by deriving the ordering from the plan rather than from the data, so the promise inherits the plan’s correctness.
• Bloom filters cost space and write time for a benefit that only materialises when the reader also has bloom pruning enabled (datafusion.execution.parquet.bloom_filter_on_read).
• Statistics truncation is a silent precision loss. 64 bytes is enough for most keys and not enough for long strings; the effect is that pruning gets weaker, never wrong — but “never wrong” here comes from the untrusted-stats machinery on the read side, which is a separate piece of machinery you now depend on.

The general lesson of the write path is that Parquet’s read performance is decided at write time. Row group size, dictionary encoding, statistics granularity, bloom filters, and sorting metadata are all write-side choices that the reader either cashes in or does not. DataFusion exposes all of them as configuration, which is the minimum you owe your future queries.