Polars Writes Parquet From a Streaming Pipeline

TL;DR: A Parquet file is a sequence of row groups, and a row group is a sequence of pages, and pages are compressed. That means a Parquet writer cannot just start writing bytes as rows arrive — it has to decide where the row group boundaries are, and it cannot know that until it has seen enough rows. Polars solves this with a resize pipeline between the streaming engine and the writer: morsels arrive at whatever size the upstream produced them, get re-cut into row groups of a target size, and are then encoded column-by-column in parallel while a single task writes the pages to the file in order.

The hard part of streaming Parquet is not the encoding. It is that a row group boundary is a decision, and the decision depends on data you have not seen yet. If you flush a row group too eagerly you get a million tiny row groups and a file that is slow to read. If you wait too long you hold the whole thing in memory.

Let me walk through how the pieces fit, starting from the Python API and going down to the byte writes.

The API

The entry point is sink_parquet on a LazyFrame (crates/polars-python/src/lazyframe/general.rs:796):

rust
    #[cfg(feature = "parquet")]
    #[pyo3(signature = (
        target, sink_options, compression, compression_level, statistics, row_group_size, data_page_size,
        metadata, arrow_schema
    ))]
    fn sink_parquet(
        &self,
        py: Python<'_>,
        target: PyFileSinkDestination,
        sink_options: PySinkOptions,
        compression: &str,
        compression_level: Option<i32>,
        statistics: Wrap<StatisticsOptions>,
        row_group_size: Option<usize>,
        data_page_size: Option<usize>,
        metadata: Wrap<Option<KeyValueMetadata>>,
        arrow_schema: Option<Wrap<ArrowSchema>>,
    ) -> PyResult<PyLazyFrame> {

It builds a ParquetWriteOptions and hands it to LazyFrame::sink (crates/polars-lazy/src/frame/mod.rs:807):

rust
    pub fn sink(
        mut self,
        sink_type: SinkDestination,
        file_format: FileWriteFormat,
        unified_sink_args: UnifiedSinkArgs,
    ) -> PolarsResult<Self> {
        polars_ensure!(
            !matches!(self.logical_plan, DslPlan::Sink { .. }),
            InvalidOperation: "cannot create a sink on top of another sink"
        );

        self.logical_plan = DslPlan::Sink {

Note that sink is a plan operation, not an execution one. The returned LazyFrame is a plan that ends in a sink; it is executed when you collect() it. This is why sink_parquet returns a PyLazyFrame and not a path.

The options themselves (crates/polars-io/src/parquet/write/options.rs:24):

rust
pub struct ParquetWriteOptions {
    /// Data page compression
    pub compression: ParquetCompression,
    /// Compute and write column statistics.
    pub statistics: StatisticsOptions,
    /// If `None` will be all written to a single row group.
    pub row_group_size: Option<usize>,
    /// if `None` will be 1024^2 bytes
    pub data_page_size: Option<usize>,
    /// Custom file-level key value metadata
    pub key_value_metadata: Option<KeyValueMetadata>,
    pub arrow_schema: Option<ArrowSchemaRef>,
    #[cfg_attr(feature = "serde", serde(default))]
    pub compat_level: Option<CompatLevel>,
}

row_group_size: None does not mean “one row group” in the streaming engine — the comment is about the old, non-streaming path. The streaming path has its own default, which we will meet shortly.

The sink node

The sink is a node in the streaming graph (crates/polars-stream/src/nodes/io_sinks/mod.rs:30):

rust
pub struct IOSinkNode {
    name: PlSmallStr,
    state: IOSinkNodeState,
    io_metrics: Option<Arc<IOMetrics>>,
    verbose: bool,
    partitioned_by_key: bool,
}

It has a peculiar lifecycle: it does not do any work during the query. It waits for the input pipe to report Done, and only then does it initialise the writer pipeline and wait for it (crates/polars-stream/src/nodes/io_sinks/mod.rs:84):

rust
        recv[0] = if recv[0] == PortState::Done {
            // Ensure initialize / writes empty file for empty output.
            self.state
                .initialize(&self.name, execution_state, self.io_metrics.clone())?;

            match std::mem::replace(&mut self.state, IOSinkNodeState::Finished) {
                IOSinkNodeState::Initialized {
                    phase_channel_tx,
                    task_handle,
                } => {

Wait — that reads as if the sink receives no morsels at all during the query. It does receive them; they just go somewhere else. The sink takes a serial receive port (crates/polars-stream/src/nodes/io_sinks/mod.rs:141):

rust
        let phase_morsel_rx = recv_ports[0].take().unwrap().serial();

and hands it to a “phase” channel. The point of the phase mechanism is that the sink may be executed in several phases by the graph scheduler, and the writer pipeline must survive across those phases. The phase task relays morsels from each phase into a single long-lived channel (crates/polars-stream/src/nodes/io_sinks/mod.rs:222):

rust
        executor::spawn(TaskPriority::High, task_metrics, async move {
            let mut morsel_seq: u64 = 1;

            while let Ok(mut phase_rx) = phase_channel_rx.recv().await {
                while let Ok(mut morsel) = phase_rx.recv().await {
                    morsel.set_seq(MorselSeq::new(morsel_seq));
                    morsel_seq = morsel_seq.saturating_add(1);

                    if multi_phase_tx.send(morsel).await.is_err() {
                        break;
                    }
                }
            }
        });

Note the sequence numbers are renumbered here. The sink does not care about the upstream’s ordering guarantees; it cares about producing a file, and for that it just needs a total order over the morsels it receives. Renumbering makes the writer’s ordering logic trivial.

The sink also always sends one empty morsel at the start (crates/polars-stream/src/nodes/io_sinks/mod.rs:216):

rust
        let _ = multi_phase_tx.try_send(Morsel::new_unregistered(
            DataFrame::empty_with_arc_schema(config.input_schema.clone()),
            MorselSeq::new(0),
            SourceToken::default(),
        ));

That is how an empty result still produces a valid, empty Parquet file rather than nothing at all.

The resize pipeline: morsels are the wrong size

Now the interesting part. The upstream produces morsels of about POLARS_IDEAL_MORSEL_SIZE rows (100,000 by default). A Parquet row group wants a different number of rows, and the two are not related: the user may have asked for row_group_size = 1_000_000, or for something byte-driven, or left it alone.

So between the engine and the writer there is a stage that re-cuts the morsels. That stage is MorselResizePipeline (crates/polars-stream/src/nodes/io_sinks/components/morsel_resize_pipeline.rs:22):

rust
/// Splits large morsels / combines small morsels.
pub struct MorselResizePipeline {
    pub schema: Arc<Schema>,
    pub target_sink_morsel_size: TargetSinkMorselSize,
    pub inflight_morsel_semaphore: Arc<tokio::sync::Semaphore>,
    pub morsel_rx: connector::Receiver<Morsel>,
    pub morsel_tx: connector::Sender<SinkMorsel>,
}

It is not a trivial pass-through. It maintains a buffer of received dataframes, and it decides, for each outgoing chunk, how many rows to take from the buffer (crates/polars-stream/src/nodes/io_sinks/components/morsel_resize_pipeline.rs:54):

rust
        loop {
            let df_to_send = if let Some(num_rows_to_take) = num_rows_to_take_iter.next() {
                take_n_rows_from_buffered(Arc::clone(&schema), &mut buffered_rows, num_rows_to_take)
            } else if let Some(n_rows) = next_chunk_lengths.pop() {
                let next_chunk_size = logical_received_size
                    .calc_delta(n_rows as IdxSize, physical_received_size)
                    .unwrap();

                let logical_buffered_size = logical_received_size.checked_sub(sent_size).unwrap();

                let send_buffered_as_one_df;

                (send_buffered_as_one_df, num_rows_to_take_iter) = target_sink_morsel_size
                    .calc_next_splits(logical_buffered_size, next_chunk_size);

                logical_received_size = logical_received_size.add(next_chunk_size)?;

                if send_buffered_as_one_df {
                    take_n_rows_from_buffered(
                        Arc::clone(&schema),
                        &mut buffered_rows,
                        idxsize_to_u64(logical_buffered_size.num_rows),
                    )
                } else {
                    continue;
                }
            } else if let Ok(morsel) = {

Three cases, and they are worth reading carefully:

There is a pending split plan. Take that many rows from the buffer and send them.
There is a new chunk of rows. Ask the target size how to split the combination of what is buffered and what just arrived. If the answer is “flush the buffer as one chunk”, do that; otherwise adopt the new split plan and loop.
No more rows. Flush whatever is left as one final chunk.

The logical_received_size / physical_received_size distinction is there because a morsel can have several Arrow chunks, and the split logic works on chunk boundaries where possible. The sent_size tracks what has already gone out.

Taking rows out of the buffer is a stack (crates/polars-stream/src/nodes/io_sinks/components/morsel_resize_pipeline.rs:188):

rust
fn take_n_rows_from_buffered(
    schema: Arc<Schema>,
    buffered_rows: &mut VecDeque<DfWithOffset>,
    mut n_rows: u64,
) -> DataFrame {
    if n_rows == 0 {
        return DataFrame::empty_with_arc_schema(schema);
    }

    let mut stacked_df: Option<DataFrame> = None;

    while n_rows != 0 {
        let next_df = buffered_rows.front_mut().unwrap();

        let next_df = if next_df.height() as u64 > n_rows {
            next_df.split_off_front(n_rows as usize)
        } else {
            buffered_rows.pop_front().unwrap().into_df()
        };

        if next_df.height() as u64 == n_rows && stacked_df.is_none() {
            return next_df;
        }
        ...

If the requested row count happens to land exactly on a frame boundary, the frame is returned as-is — no copy, no vertical stack. Only when the boundary falls in the middle of a frame does the frame get split and the pieces stacked.

Deciding the row group size

The decision logic is TargetSinkMorselSize::calc_next_splits (crates/polars-stream/src/nodes/io_sinks/components/size.rs:211):

rust
impl TargetSinkMorselSize {
    pub fn calc_next_splits(
        &self,
        buffered_size: RowCountAndSize,
        incoming_size: RowCountAndSize,
    ) -> (bool, PartSizesIter) {
        let combined_size = buffered_size.checked_add(incoming_size).unwrap();

        let mut flush_buffered_as_one_split = false;
        let (mut part_sizes_iter, mut limited_by) = self.build_part_sizes_iter(combined_size);

and build_part_sizes_iter is where the two limits — rows and bytes — are arbitrated (crates/polars-stream/src/nodes/io_sinks/components/size.rs:274):

rust
    fn build_part_sizes_iter(&self, size: RowCountAndSize) -> (PartSizesIter, LimitedBy) {
        if size.num_rows == 0 {
            return (PartSizesIter::default(), LimitedBy::default());
        }

        let n_parts_by_num_rows = if self.target_num_rows_mode == SplitMode::Exact {
            u64::max(1, idxsize_to_u64(size.num_rows / self.target_num_rows))
        } else {
            calc_n_parts(
                idxsize_to_u64(size.num_rows),
                NonZeroU64::new(idxsize_to_u64(self.target_num_rows.get())).unwrap(),
            )
        };

        let mut max_parts_by_num_bytes = 0;
        let mut n_parts_by_num_bytes = 0;

        if self.target_num_bytes.get() != u64::MAX {
            max_parts_by_num_bytes = idxsize_to_u64(size.num_rows / self.target_num_bytes_min_rows);
            n_parts_by_num_bytes = calc_n_parts(size.num_bytes, self.target_num_bytes);
        };

The SplitMode distinction is the interesting bit:

Exact means “every row group must have exactly target_num_rows rows”. This is what you want when the user set row_group_size explicitly: the file must have the row group layout they asked for.
Approximate means “get close to target_num_rows”. This is the default when the user did not ask for anything, and it lets the pipeline roll a row group as soon as it has roughly enough rows, which keeps the pipeline moving instead of waiting for an exact count.

The calc_n_parts helper is the shared arithmetic (crates/polars-utils/src/calc_morsel_split.rs:69):

rust
/// Number of parts to split `size` to minimize the average part size difference to `target_part_size`.
pub fn calc_n_parts(size: u64, target_part_size: NonZeroU64) -> u64 {
    if size <= target_part_size.get() {
        return if size == 0 { 0 } else { 1 };
    }

    let n_parts = size / target_part_size;

    (n_parts..=n_parts.saturating_add(1))
        .min_by_key(|n_parts| (size / *n_parts).abs_diff(target_part_size.get()))
        .unwrap()
}

Only two candidates are considered: n and n+1 parts. The winner is the one whose resulting part size is closest to the target. That is enough, because size / n and size / (n+1) straddle the target.

The part sizes themselves come from PartSizesIter (crates/polars-utils/src/calc_morsel_split.rs:45):

rust
impl Iterator for PartSizesIter {
    type Item = u64;

    fn next(&mut self) -> Option<Self::Item> {
        self.remaining_parts = self.remaining_parts.checked_sub(1)?;
        Some(self.base_part_size + (self.remaining_parts >= self.remainder_cutoff) as u64)
    }

It is a counting-sort-style distribution of the remainder: the first remainder_cutoff parts get one extra row, the rest do not. The result is that sum(part_sizes) == total exactly, with the parts differing by at most one row.

There is one more subtlety in calc_next_splits (crates/polars-stream/src/nodes/io_sinks/components/size.rs:236):

rust
        if limited_by == LimitedBy::Rows
            && part_sizes_iter.len() <= 1
            && self.target_num_rows_mode != SplitMode::Exact
            && part_sizes_iter.base_part_size().checked_mul(2).is_some_and(
                // base_part_size < (4/3) * target_num_rows
                |double_base_part_size| {
                    u64::abs_diff(
                        double_base_part_size,
                        idxsize_to_u64(self.target_num_rows.get()),
                    ) < u64::abs_diff(
                        part_sizes_iter.base_part_size(),
                        idxsize_to_u64(self.target_num_rows.get()),
                    )
                },
            )
        {
            // Wait for more data to have a fuller chunk.
            part_sizes_iter = PartSizesIter::default()
        }

In Approximate mode, if the accumulated data is already close to target_num_rows, the pipeline waits for more rather than flushing a slightly-short row group. The 4/3 heuristic: if doubling the current size would land closer to the target than the current size does, then the current size is “under” the target by more than a factor of 4/3, so it is worth waiting.

The defaults

ParquetWriterStarter decides the target (crates/polars-stream/src/nodes/io_sinks/writers/parquet/mod.rs:52):

rust
    fn target_sink_morsel_size(&self) -> TargetSinkMorselSize {
        let target_num_bytes_min_rows = const { NonZeroIdxSize::new(16384).unwrap() };

        if let Some(row_group_size) = self.row_group_size
            && row_group_size > 0
        {
            TargetSinkMorselSize {
                target_num_rows: row_group_size.try_into().unwrap(),
                target_num_bytes: NonZeroU64::MAX,
                target_num_bytes_min_rows,
                target_num_rows_mode: SplitMode::Exact,
            }
        } else {
            let (env_num_rows, env_num_bytes) = ideal_sink_morsel_size_env();

            TargetSinkMorselSize {
                target_num_rows: env_num_rows
                    .unwrap_or(const { NonZeroIdxSize::new(122_880).unwrap() }),
                target_num_bytes: env_num_bytes.unwrap_or(NonZeroU64::MAX),
                target_num_bytes_min_rows,
                target_num_rows_mode: SplitMode::Approximate,
            }
        }
    }

So:

If the user set row_group_size, the mode is Exact and the byte limit is disabled. The file will have exactly the row group layout requested.
Otherwise the default is 122,880 rows in Approximate mode. That number is 120 × 1024, and it is a deliberate choice: Parquet readers and the ecosystem generally cope better with row groups in the low hundreds of thousands of rows.
The byte limit is u64::MAX (disabled) unless POLARS_IDEAL_SINK_MORSEL_SIZE_BYTES is set, in which case the target_num_bytes_min_rows of 16,384 acts as a floor: a row group never gets smaller than 16k rows even if the byte budget says so.

Three tasks: resize, encode, write

The single-file sink pipeline is assembled in crates/polars-stream/src/nodes/io_sinks/pipeline_initialization/single_file.rs:136:

rust
    let (writer_tx, writer_rx) = connector::connector();
    let writer_handle = file_writer_starter.start_file_writer(
        writer_rx,
        file_open_task,
        num_pipelines_per_sink,
        execution_state.task_metrics.clone(),
    )?;

    let schema = Arc::clone(&file_schema);
    let inflight_morsel_semaphore =
        Arc::new(tokio::sync::Semaphore::new(inflight_morsel_limit.get()));

    let resize_pipeline = MorselResizePipeline {
        schema,
        target_sink_morsel_size,
        inflight_morsel_semaphore,
        morsel_rx,
        morsel_tx: writer_tx,
    };

The semaphore is the memory bound. Every outgoing sink morsel holds a permit, and the permit is released only when the data has actually been written (crates/polars-stream/src/nodes/io_sinks/components/sink_morsel.rs:3):

rust
pub type SinkMorselPermit = tokio::sync::OwnedSemaphorePermit;

/// In-flight morsel in the IO sink. Holds a permit against a semaphore that restricts
/// the total number of sink morsels in memory.
pub struct SinkMorsel {
    df: DataFrame,
    /// Should only be dropped once the data associated with this morsel has been dropped from memory.
    permit: SinkMorselPermit,
}

The limit is num_pipelines + 4 (crates/polars-stream/src/nodes/io_sinks/config.rs:30):

rust
    pub fn inflight_morsel_limit(&self, num_pipelines: NonZeroUsize) -> NonZeroUsize {
        ...
        NonZeroUsize::saturating_add(
            num_pipelines,
            // Additional buffer to accommodate head-of-line blocking
            4,
        )
    }

The extra four is for head-of-line blocking: if one row group takes longer to encode than the others, the other permits let the pipeline keep feeding the encoder.

Now the writer itself. ParquetWriterStarter::start_file_writer (crates/polars-stream/src/nodes/io_sinks/writers/parquet/mod.rs:102) sets up three things: a channel of encoded row groups, a writer task and an encoder task.

rust
        let (encoded_row_group_tx, encoded_row_group_rx) = tokio::sync::mpsc::channel::<
            executor::AbortOnDropHandle<PolarsResult<EncodedRowGroup>>,
        >(num_pipelines.get());

The channel’s capacity is the number of pipelines, so up to that many row groups can be in flight — being encoded, or encoded and waiting to be written.

The WriteOptions are built once (crates/polars-stream/src/nodes/io_sinks/writers/parquet/mod.rs:106):

rust
        let key_value_metadata = self.options.key_value_metadata.clone();
        let write_options = WriteOptions {
            statistics: self.options.statistics,
            compression: self.options.compression.into(),
            version: Version::V1,
            data_page_size: self.options.data_page_size,
        };

Version::V1 is hardcoded here — the streaming writer does not emit Parquet v2 files even though the option type allows it. compression is converted from the user-facing ParquetCompression enum (crates/polars-io/src/parquet/write/options.rs:55):

rust
impl From<ParquetCompression> for CompressionOptions {
    fn from(value: ParquetCompression) -> Self {
        use ParquetCompression::*;
        match value {
            Uncompressed => CompressionOptions::Uncompressed,
            Snappy => CompressionOptions::Snappy,
            Gzip(level) => CompressionOptions::Gzip(level),
            Brotli(level) => CompressionOptions::Brotli(level),
            Lz4Raw => CompressionOptions::Lz4Raw,
            Zstd(level) => CompressionOptions::Zstd(level),
        }
    }
}

and the default is Zstd (crates/polars-io/src/parquet/write/options.rs:49):

rust
impl Default for ParquetCompression {
    fn default() -> Self {
        Self::Zstd(None)
    }
}

The encoder: parallel across columns

The encoder task (crates/polars-stream/src/nodes/io_sinks/writers/parquet/row_group_encoder.rs:52) receives sink morsels and turns each into an encoded row group:

rust
        while let Ok(morsel) = morsel_rx.recv().await {
            let arrow_schema = Arc::clone(&arrow_schema);
            let schema_descriptor = Arc::clone(&schema_descriptor);
            let encodings = Buffer::clone(&encodings);
            let column_task_metrics = task_metrics.clone();

            let row_group_encode_handle = executor::AbortOnDropHandle::new(executor::spawn(
                TaskPriority::High,
                task_metrics.as_deref(),
                async move {
                    let (df, morsel_permit) = morsel.into_inner();
                    let num_rows = df.height();

                    let mut data: Vec<Vec<CompressedPage>> = Vec::with_capacity(num_leaf_columns);

The whole row group is encoded in a spawned task, and the handle is sent to the writer rather than the result. That is the key to the pipelining: the writer does not wait for the encoding to finish before it starts the next one.

Inside the task, the columns are encoded in parallel (crates/polars-stream/src/nodes/io_sinks/writers/parquet/row_group_encoder.rs:73):

rust
                    for fut in parallelize_first_to_local(
                        TaskPriority::High,
                        column_task_metrics.as_deref(),
                        df.into_columns().into_iter().enumerate().map(|(i, c)| {
                            let arrow_schema = Arc::clone(&arrow_schema);
                            let schema_descriptor = Arc::clone(&schema_descriptor);
                            let encodings = Buffer::clone(&encodings);

                            async move {
                                let parquet_type = &schema_descriptor.fields()[i];
                                let encodings = encodings[i].as_slice();
                                let array =
                                    c.as_materialized_series().rechunk().to_arrow_with_field(
                                        0,
                                        Cow::Borrowed(arrow_schema.get_at_index(i).unwrap().1),
                                        true,
                                    )?;

                                let mut data: UnitVec<Vec<CompressedPage>> =
                                    UnitVec::with_capacity(num_leaf_columns);

                                for encode_page_iter in array_to_columns(
                                    array,
                                    parquet_type.clone(),
                                    write_options,
                                    encodings,
                                )? {
                                    let compressed_pages: Vec<CompressedPage> =
                                        Compressor::new_from_vec(
                                            encode_page_iter.map(|result| {
                                                result.map_err(|e| {
                                                    ParquetError::FeatureNotSupported(format!(
                                                        "reraised in polars: {e}",
                                                    ))
                                                })
                                            }),
                                            write_options.compression,
                                            vec![],
                                        )
                                        .collect::<ParquetResult<_>>()?;

                                    data.push(compressed_pages)
                                }

                                PolarsResult::Ok(data)
                            }
                        }),
                    ) {
                        data.extend(fut.await?);
                    }

parallelize_first_to_local keeps the first column on the current thread and spawns the rest (crates/polars-async/src/primitives/opt_spawned_future.rs:59):

rust
/// Parallelizes futures across the computational async runtime.
///
/// As an optimization for cache access, the first future is kept on the current thread. If there
/// is only 1 future, then all data is kept on the current thread and spawn is not called at all.

That is a cache-locality optimisation: the first column’s data is already in this core’s cache, so encoding it here avoids a cross-core transfer. The remaining columns are spawned, so a wide table is encoded on all cores at once.

Note the order of operations inside the column future: rechunk() to a single Arrow chunk, convert to Arrow with the field’s metadata attached, then array_to_columns to produce the encoded (but uncompressed) pages, then Compressor to compress them. The compression happens per page, inside the same parallel task, so the compressed bytes never leave the core that produced them.

The writer: one task, one file

The writer task is deliberately on the I/O runtime, not the compute one (crates/polars-stream/src/nodes/io_sinks/writers/parquet/mod.rs:117):

rust
        let io_handle = tokio_handle_ext::AbortOnDropHandle(
            ASYNC.spawn(
                io_writer::IOWriter {
                    file,
                    encoded_row_group_rx,
                    arrow_schema,
                    schema_descriptor: Arc::clone(&schema_descriptor),
                    write_options,
                    encodings: Buffer::clone(&encodings),
                    key_value_metadata,
                    num_leaf_columns,
                }
                .run(),
            ),
        );

and its body is short (crates/polars-stream/src/nodes/io_sinks/writers/parquet/io_writer.rs:57):

rust
        while let Some(handle) = encoded_row_group_rx.recv().await {
            let EncodedRowGroup {
                num_rows,
                data,
                morsel_permit,
            } = handle.await?;
            assert_eq!(data.len(), num_leaf_columns);
            parquet_writer.write_row_group(num_rows as u64, &data)?;
            drop(data);
            drop(morsel_permit);
        }

handle.await? is where the writer waits for the row group to finish encoding — but by then the next row group is already being encoded on another core. So the pipeline is:

text

  resize task            encoder task(s)              writer task
  (compute)              (compute, N in flight)        (IO runtime)
 ┌───────────┐  morsel  ┌──────────────┐  pages  ┌──────────────┐
 │ cut rows  │ ───────► │ encode cols  │ ──────► │ write bytes  │
 │ to target │          │ + compress   │  (chan) │ to file      │
 └───────────┘          └──────────────┘         └──────────────┘

The morsel_permit is dropped after the write, so the semaphore that bounds in-flight memory is only released once the data is on disk. That is what keeps the memory bound honest.

The BatchedWriter wraps the actual Parquet file writer (crates/polars-io/src/parquet/write/batched_writer.rs:29):

rust
pub struct BatchedWriter<W: Write> {
    // A mutex so that streaming engine can get concurrent read access to
    // compress pages.
    //
    // @TODO: Remove mutex when old streaming engine is removed
    pub(super) writer: Mutex<FileWriter<W>>,
    // @TODO: Remove when old streaming engine is removed
    pub(super) parquet_schema: SchemaDescriptor,
    pub(super) encodings: Buffer<Vec<Encoding>>,
    pub(super) options: WriteOptions,
    pub(super) parallel: bool,
    pub(super) key_value_metadata: Option<KeyValueMetadata>,
}

and write_row_group hands the already-compressed pages straight through (crates/polars-io/src/parquet/write/batched_writer.rs:104):

rust
    /// Note: `num_rows` can be passed as `u64::MAX` to infer `num_rows` from the encoded data.
    pub fn write_row_group(
        &mut self,
        num_rows: u64,
        rg: &[Vec<CompressedPage>],
    ) -> PolarsResult<()> {
        let writer = self.writer.get_mut().unwrap();
        let rg = DynIter::new(rg.iter().map(|col_pages| {
            Ok(DynStreamingIterator::new(
                fallible_streaming_iterator::convert(col_pages.iter().map(PolarsResult::Ok)),
            ))
        }));
        writer.write(num_rows, rg)?;
        Ok(())
    }

No re-encoding, no re-compression — the pages arrive ready. The underlying FileWriter::write (crates/polars-parquet/src/parquet/write/file.rs:167) is where the bytes actually hit the file:

rust
    /// Writes a row group to the file.
    ///
    /// This call is IO-bounded
    pub fn write<E>(
        &mut self,
        num_rows: u64,
        row_group: RowGroupIterColumns<'_, E>,
    ) -> ParquetResult<()>
    where
        ParquetError: From<E>,
        E: std::error::Error,
    {
        if self.offset == 0 {
            self.start()?;
        }
        let ordinal = self.row_groups.len();
        let (group, specs, size) = write_row_group(
            &mut self.writer,
            num_rows,
            self.offset,
            self.schema.columns(),
            row_group,
            ordinal,
        )?;
        self.offset += size;
        self.row_groups.push(group);
        self.page_specs.push(specs);
        Ok(())
    }

The row group metadata is accumulated in memory (row_groups, page_specs) and written once at the end. That is the one piece of state that grows with the number of row groups, but it is tiny compared to the data.

Finally the footer (crates/polars-stream/src/nodes/io_sinks/writers/parquet/io_writer.rs:73):

rust
        parquet_writer.finish()?;
        let parquet_metadata = {
            let writer = parquet_writer.get_writer().lock().unwrap();
            let mut out = Vec::new();
            write_metadata_sidecar(&mut out, writer.metadata().unwrap())?;
            out
        };
        drop(parquet_writer);
        buffered_file.flush()?;
        drop(buffered_file);

        file.set_parquet_metadata(parquet_metadata);
        file.close(sync_on_close).await?;

The metadata sidecar is written to a separate buffer rather than into the file — that is what makes the partitioned sink possible, where several files’ row groups are described by one metadata blob.

What is parallel and what is serialised

Putting it all in one table:

stage	where	parallel?
resize (re-cut morsels)	compute executor	no, one task
encode + compress a row group	compute executor	yes, one task per column (first column kept local)
several row groups at once	compute executor	yes, up to `num_pipelines` in flight
write bytes to the file	I/O runtime	no, one task owns the file
file open / close	I/O runtime	no

The invariant is: the file itself is written by exactly one task, in row group order. Everything else is parallel. That is the whole design of the Parquet sink: parallelism is pushed as far upstream as possible, and the serial part is reduced to “append these bytes to this file”.

The price is that the writer is a bottleneck if the encoding is fast. With a wide table and cheap compression, the encoder can produce row groups faster than the disk can absorb them — and then the channel between them fills up and the encoder blocks. That is correct behaviour, not a bug: the backpressure just moves the bottleneck to where it actually is.

Partitioned sinks

The same machinery drives the partitioned sink, with one extra stage. start_partition_sink_pipeline (crates/polars-stream/src/nodes/io_sinks/pipeline_initialization/partition_by.rs:29) inserts a partitioner between the engine and the resize stage, and a distributor that opens files on demand.

Each partition gets its own writer, its own resize pipeline and its own row group sequence, so a partitioned write produces several independent Parquet files, each with its own row groups. The number of simultaneously open files is bounded (crates/polars-stream/src/nodes/io_sinks/config.rs:46):

rust
    pub fn max_open_sinks(&self) -> NonZeroUsize {
        ...
        if self.target.is_cloud_location() {
            const { NonZeroUsize::new(512).unwrap() }
        } else {
            const { NonZeroUsize::new(128).unwrap() }
        }
    }

128 local files, 512 in the cloud. Beyond that the sink has to close and reopen files, which is why partitioned writes to object storage have a per-file size limit option (crates/polars-stream/src/nodes/io_sinks/pipeline_initialization/partition_by.rs:112):

rust
    if let Some(file_size_limit) = file_size_limit {
        target_sink_morsel_size.target_num_rows = NonZeroIdxSize::min(
            target_sink_morsel_size.target_num_rows,
            file_size_limit.num_rows,
        );
        target_sink_morsel_size.target_num_bytes = NonZeroU64::min(
            target_sink_morsel_size.target_num_bytes,
            file_size_limit.num_bytes,
        );
        target_sink_morsel_size.target_num_rows_mode = SplitMode::Exact;
    }

Note that the file size limit also forces SplitMode::Exact: if you said “at most 1M rows per file”, the row groups have to add up to exactly that, not approximately.

The old, non-streaming path

For comparison, the classic path is much simpler (crates/polars-io/src/parquet/write/writer.rs:147):

rust
    /// Write the given DataFrame in the writer `W`.
    /// Returns the total size of the file.
    pub fn finish(self, df: &mut DataFrame) -> PolarsResult<u64> {
        let chunked_df = chunk_df_for_writing(df, self.row_group_size.unwrap_or(512 * 512))?;
        let mut batched = self.batched(chunked_df.schema())?;
        batched.write_batch(&chunked_df)?;
        batched.finish()
    }

chunk_df_for_writing splits the frame into row groups of row_group_size rows (default 512 × 512 = 262,144), and write_batch encodes and compresses them eagerly (crates/polars-io/src/parquet/write/batched_writer.rs:251):

rust
/// This serializer encodes and compresses all eagerly in memory.
/// Used for separating compute from IO.
fn create_eager_serializer(

Note “eagerly in memory” — the old path materialises the entire encoded row group before writing it. The streaming path is the same code with the eagerness removed: the row group is encoded in a spawned task, and the handle is what travels down the channel.

The bottom line

The design of the Parquet sink comes down to four decisions:

Row group boundaries are decided by a separate stage. The streaming engine’s morsels are not row groups. A resize pipeline re-cuts them, using PartSizesIter to distribute the remainder exactly, and using calc_n_parts to pick a split count that lands close to the target.
The mode matters. Exact when the user asked for a specific row_group_size, Approximate otherwise. In approximate mode the writer will even wait for more rows rather than flush a short row group.
Encoding is parallel, writing is serial. Column encoding and compression run on the compute executor, in parallel across columns and across row groups. Exactly one task owns the file and writes the pages in order.
The memory bound is a semaphore, not a guess. Every in-flight sink morsel holds a permit, released only after the data is written.

The knobs, if you want to experiment:

POLARS_IDEAL_SINK_MORSEL_SIZE_ROWS — the default row group size when row_group_size is not set (default 122,880).
POLARS_IDEAL_SINK_MORSEL_SIZE_BYTES — a byte-based limit on the row group, with a 16,384-row floor.
POLARS_INFLIGHT_SINK_MORSEL_LIMIT — how many sink morsels may be in memory at once (default num_pipelines + 4).
POLARS_MAX_OPEN_SINKS — how many files a partitioned sink may hold open at once.