GH-48476: [C++] [Parquet] Add max_row_group_size writer property to limit row groups by compressed byte size

[zhf999]

Rationale for this change

The Parquet writer can currently only limit row groups by row count
(WriterProperties::max_row_group_length()). With wide rows or highly variable
row sizes, a row-count limit produces row groups whose size in bytes varies a
lot. Since query engines and storage systems usually tune around a target row
group size in bytes (HDFS block alignment, reader memory footprint,
parallelism granularity), it is useful to be able to limit row groups by size
as well.

What changes are included in this PR?

New writer property

WriterProperties::max_row_group_size() / Builder::max_row_group_size(int64_t)
specify the maximum size of a row group in bytes. The default is unlimited
(std::numeric_limits<int64_t>::max()), so nothing changes for existing users
unless they opt in.

How the size is estimated

The limit is checked between writes against the data accumulated in the current
row group:

• compressed pages already written to the sink
  (RowGroupWriter::total_compressed_bytes_written()),
• compressed pages still buffered in the column writers
  (RowGroupWriter::total_compressed_bytes()),
• an uncompressed estimate of the values, definition levels, repetition levels
  and dictionary still held by the column encoders
  (RowGroupWriter::estimated_buffered_stats()).

The last group matters: values only become pages once they exceed
data_pagesize, so ignoring the encoder buffers would let a row group
overshoot the limit by up to num_columns * data_pagesize and would make any
limit below that value impossible to honour. Those estimates are uncompressed,
so the total errs on the conservative side and row groups stay at or below the
limit in practice.

Enforcement in both Arrow writer entry points

• WriteRecordBatch starts a new buffered row group whenever the current one
  reaches the row count or the byte size limit. The implementation moved into a
  private WriteRecordBatchBuffered(batch, max_rows_per_row_group) so that the
  row cap can be supplied by the caller.
• WriteTable switches to the buffered path when the limit is set: the table is
  fed through a TableBatchReader in batches of
  min(chunk_size, write_batch_size()) rows, so the same check applies and a
  row group only overshoots by at most one batch. chunk_size is still honoured
  as the maximum number of rows per row group. Without the limit, WriteTable
  keeps using the original non-buffered path unchanged.

A byte limit cannot be enforced on non-buffered row groups because their size
is only known once they have been written, and the row count is fixed before
writing. Row groups created explicitly through
NewRowGroup()/WriteColumnChunk() are therefore not affected; this is
documented on the builder method.

Are these changes tested?

Yes.

• WriterPropertiesTest.RoundTripThroughBuilder covers the new property, with a
  non-default value added to the override_defaults case.
• TestArrowReadWrite.WriteRecordBatchRespectsMaxRowGroupSize and
  TestArrowReadWrite.WriteTableRespectsMaxRowGroupSize check that row groups
  are rolled over once the limit is reached, that every row group stays within
  the limit, and that no row is lost or duplicated. Both use the default 1MB
  data page size, so they also cover the encoder-buffered data.
• TestArrowReadWrite.WriteTableUnlimitedRowGroupSize pins the default
  behaviour, where chunk_size alone decides the row group boundaries.
• TestArrowReadWrite.WriteTableMaxRowGroupSizeRoundTrip verifies that the data
  is unchanged when written through the buffered path.

Are there any user-facing changes?

Yes, a new opt-in writer property. The default is unlimited, so there is no
behaviour change unless it is set. When it is set:

• row groups are buffered in memory until they are flushed, which increases the
  memory footprint of the writer;
• for WriteTable, row group boundaries are decided by both chunk_size and
  the byte limit, and columns are written in parallel if
  ArrowWriterProperties::use_threads() is enabled.

• GitHub Issue: [C++][Parquet] Add configure to limit the row group size #48467
• GitHub Issue: GH-48475: [Ruby] Add support for reading Int32 and UInt32 arrays #48476

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[github-actions]

Thanks for opening a pull request!

This pull request has been automatically converted to a draft because its title doesn't match Arrow's required format.

If this is not a minor PR. Could you open an issue for this pull request on GitHub? https://github.com/apache/arrow/issues/new/choose

Opening GitHub issues ahead of time contributes to the Openness of the Apache Arrow project.

Then could you also rename the pull request title in the following format?

GH-${GITHUB_ISSUE_ID}: [${COMPONENT}] ${SUMMARY}

or

MINOR: [${COMPONENT}] ${SUMMARY}

After updating the title, you can mark the pull request as ready for review.

See also:

• Other pull requests
• Contribution Guidelines - Contributing Overview

[github-actions]

⚠️ GitHub issue #48476 has been automatically assigned in GitHub to PR creator.

[zhf999]

Related PR: #48468

[Reranko05]

GitHub Issue: #48476

Thanks for contributing! Can you edit the PR desc and ref to the issue, it seems a PR is being referenced here.

[wgtmac]

Why only for WriteRecordBatch?

[zhf999]

WriteRecordBatch writes into a buffered row group, which accumulates across calls. That makes the byte size observable between writes, so we can simply stop adding to the current row group once the accumulated compressed size reaches the limit.

While WriteTable uses the non-buffered path, where each call maps to exactly one row group whose row count is fixed up front by the user-supplied chunk_size. Enforcing a byte limit there would require predicting the compressed size before writing, e.g. by estimating an average row size from the previous row group. That's inherently approximate, especially since chunk_size is an explicit contract from the caller.

I wonder if estimation is acceptable in WriteTable?

[wgtmac]

IIRC, WriteTable also splits the table into several row groups if max_row_group_length is reached.

[zhf999]

max_row_group_length can be applied there because row counts are known before writing — it just clamps chunk_size. The byte size simply isn't knowable at that point, which is the asymmetry.

[zhf999]

@wgtmac What about change the WriteTable to buffered path like WriteRecordBatch?

[github-actions]

⚠️ GitHub issue #48476 has been automatically assigned in GitHub to PR creator.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Suppressed comments (3)

cpp/src/parquet/arrow/writer.cc:439

• This changes FileWriter::WriteTable behavior when max_row_group_size is set (it switches to the buffered RecordBatch path and will split row groups by byte size). The PR description currently states WriteTable is intentionally not affected; please update the PR description (and any related docs/release notes) to match the implemented behavior to avoid misleading users.

    // If max_row_group_size is set, use buffered path to write row groups.
    if (this->properties().max_row_group_size() != std::numeric_limits<int64_t>::max()) {
      ::arrow::TableBatchReader reader(table);
      reader.set_chunksize(std::min(chunk_size, this->properties().write_batch_size()));
      while (true) {

cpp/src/parquet/arrow/writer.cc:496

• estimated_row_group_size() calls total_compressed_bytes(), total_compressed_bytes_written(), and estimated_buffered_stats(); each of these does an O(num_columns) scan in RowGroupSerializer. With max_row_group_size enabled, row_group_full() is evaluated after each WriteBatch, so this becomes 2–3 full column scans per batch and can be a noticeable CPU cost for wide schemas.

    // Estimated size of the data accumulated in the current row group.
    auto estimated_row_group_size = [&]() {
      const auto buffered = row_group_writer_->estimated_buffered_stats();
      return row_group_writer_->total_compressed_bytes() +
             row_group_writer_->total_compressed_bytes_written() + buffered.value_bytes +
             buffered.def_level_bytes + buffered.rep_level_bytes + buffered.dict_bytes;
    };

cpp/src/parquet/properties.h:506

• The doc comment claims the size estimate “errs on the conservative side”, but the implementation doesn’t account for per-page overhead (e.g., PageHeader bytes) for data still buffered in encoders, so small pages / low compression can lead to underestimation. Consider softening this guarantee to avoid overpromising the bound.

    /// The limit is checked between writes against an estimate of the data
    /// accumulated in the current row group, which combines the size of the
    /// compressed pages with an uncompressed estimate of the values still
    /// buffered by the column encoders. The estimate is therefore approximate
    /// and errs on the conservative side.

[github-actions]

⚠️ GitHub issue #48476 has been automatically assigned in GitHub to PR creator.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated no new comments.

Suppressed comments (2)

cpp/src/parquet/arrow/writer.cc:503

• row_group_full() always computes estimated_buffered_stats() and calls total_compressed_bytes*() even when max_row_group_size is unlimited. Since WriteRecordBatch() now always goes through WriteRecordBatchBuffered(), this adds an O(num_columns) scan per chunk on the default (unlimited) path and can be a noticeable regression.

    auto row_group_full = [&]() {
      const auto buffered = row_group_writer_->estimated_buffered_stats();
      const int64_t estimated_size = row_group_writer_->total_compressed_bytes() +
                                     row_group_writer_->total_compressed_bytes_written() +
                                     buffered.value_bytes + buffered.def_level_bytes +

cpp/src/parquet/arrow/writer.cc:545

• When max_row_group_size is set, size enforcement is only checked after each WriteBatch() call, but batch_size can be as large as the remaining rows up to max_rows_per_row_group (often 1M). A single large RecordBatch can therefore overshoot the byte limit by a very large margin before the rollover check runs.

      const int64_t batch_size =
          std::min(max_rows_per_row_group - row_group_writer_->num_rows(),
                   batch.num_rows() - offset);
      RETURN_NOT_OK(WriteBatch(offset, batch_size));