Add datafusion.execution.soft_max_bytes_per_output_file config parameter by Tpt · Pull Request #24924 · apache/datafusion

Tpt · 2026-09-03T15:49:23Z

Rationale for this change

Allows to roughly limit the size of parquet files when the number of rows is a hard-to-use estimator.
This is especially useful when a column contains potentially large blobs.

What changes are included in this PR?

Adds the config parameter
Add a new AsyncWrite implementation created by ObjectWriterBuilder. It sits between the optional compression and the BufWriter implementation, counting the number of effectively flushed bytes in a Arc<AtomicUsize>.
The Arc<AtomicU64> is used in the demuxer logic to get an estimation of the written size. This estimation lags because writes happens asynchronously but this avoid having to block on encoding and compression.
Use ObjectWriterBuilder when creating the single file AsyncArrowWriter

github-actions · 2026-09-03T16:15:14Z

Thank you for opening this pull request!

Reviewer note: cargo-semver-checks reported the current version number is not SemVer-compatible with the changes in this pull request (compared against the base branch).

Details

     Cloning apache/main
    Building datafusion v55.1.0 (current)
       Built [  55.123s] (current)
     Parsing datafusion v55.1.0 (current)
      Parsed [   0.027s] (current)
    Building datafusion v55.1.0 (baseline)
       Built [  45.924s] (baseline)
     Parsing datafusion v55.1.0 (baseline)
      Parsed [   0.027s] (baseline)
    Checking datafusion v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.770s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 104.300s] datafusion
    Building datafusion-common v55.1.0 (current)
       Built [  26.076s] (current)
     Parsing datafusion-common v55.1.0 (current)
      Parsed [   0.051s] (current)
    Building datafusion-common v55.1.0 (baseline)
       Built [  26.552s] (baseline)
     Parsing datafusion-common v55.1.0 (baseline)
      Parsed [   0.055s] (baseline)
    Checking datafusion-common v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.891s] 223 checks: 222 pass, 1 fail, 0 warn, 31 skip

--- failure constructible_struct_adds_field: struct exhaustively constructible through public API adds field ---

Description:
A pub struct that could be exhaustively constructed with a literal using only public API has a new pub field, breaking existing exhaustive literals.
        ref: https://doc.rust-lang.org/reference/expressions/struct-expr.html
       impl: https://github.com/obi1kenobi/cargo-semver-checks/tree/v0.50.0/src/lints/constructible_struct_adds_field.ron

Failed in:
  field ExecutionOptions.soft_max_bytes_per_output_file in /home/runner/work/datafusion/datafusion/datafusion/common/src/config.rs:894

     Summary semver requires new major version: 1 major and 0 minor checks failed
    Finished [  54.933s] datafusion-common
    Building datafusion-datasource v55.1.0 (current)
       Built [  34.770s] (current)
     Parsing datafusion-datasource v55.1.0 (current)
      Parsed [   0.026s] (current)
    Building datafusion-datasource v55.1.0 (baseline)
       Built [  35.213s] (baseline)
     Parsing datafusion-datasource v55.1.0 (baseline)
      Parsed [   0.025s] (baseline)
    Checking datafusion-datasource v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.327s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  71.735s] datafusion-datasource
    Building datafusion-datasource-arrow v55.1.0 (current)
       Built [  36.054s] (current)
     Parsing datafusion-datasource-arrow v55.1.0 (current)
      Parsed [   0.009s] (current)
    Building datafusion-datasource-arrow v55.1.0 (baseline)
       Built [  35.770s] (baseline)
     Parsing datafusion-datasource-arrow v55.1.0 (baseline)
      Parsed [   0.010s] (baseline)
    Checking datafusion-datasource-arrow v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.085s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  72.946s] datafusion-datasource-arrow
    Building datafusion-datasource-avro v55.1.0 (current)
       Built [  37.627s] (current)
     Parsing datafusion-datasource-avro v55.1.0 (current)
      Parsed [   0.009s] (current)
    Building datafusion-datasource-avro v55.1.0 (baseline)
       Built [  36.254s] (baseline)
     Parsing datafusion-datasource-avro v55.1.0 (baseline)
      Parsed [   0.010s] (baseline)
    Checking datafusion-datasource-avro v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.095s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  75.230s] datafusion-datasource-avro
    Building datafusion-datasource-parquet v55.1.0 (current)
       Built [  40.912s] (current)
     Parsing datafusion-datasource-parquet v55.1.0 (current)
      Parsed [   0.028s] (current)
    Building datafusion-datasource-parquet v55.1.0 (baseline)
       Built [  41.536s] (baseline)
     Parsing datafusion-datasource-parquet v55.1.0 (baseline)
      Parsed [   0.027s] (baseline)
    Checking datafusion-datasource-parquet v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.202s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [  83.973s] datafusion-datasource-parquet
    Building datafusion-sqllogictest v55.1.0 (current)
       Built [  77.831s] (current)
     Parsing datafusion-sqllogictest v55.1.0 (current)
      Parsed [   0.018s] (current)
    Building datafusion-sqllogictest v55.1.0 (baseline)
       Built [  81.632s] (baseline)
     Parsing datafusion-sqllogictest v55.1.0 (baseline)
      Parsed [   0.022s] (baseline)
    Checking datafusion-sqllogictest v55.1.0 -> v55.1.0 (no change; assume patch)
     Checked [   0.113s] 223 checks: 223 pass, 31 skip
     Summary no semver update required
    Finished [ 163.087s] datafusion-sqllogictest

codecov-commenter · 2026-09-03T19:13:13Z

Codecov Report

❌ Patch coverage is 89.87342% with 32 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.42%. Comparing base (140c7c5) to head (1554c47).
⚠️ Report is 82 commits behind head on main.

Files with missing lines	Patch %	Lines
datafusion/datasource/src/write/mod.rs	70.00%	14 Missing and 1 partial ⚠️
datafusion/core/src/datasource/listing/table.rs	94.27%	0 Missing and 11 partials ⚠️
datafusion/datasource/src/write/demux.rs	93.61%	2 Missing and 1 partial ⚠️
datafusion/datasource-parquet/src/sink.rs	90.00%	0 Missing and 1 partial ⚠️
datafusion/datasource/src/file_compression_type.rs	50.00%	1 Missing ⚠️
datafusion/datasource/src/write/orchestration.rs	93.33%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24924      +/-   ##
==========================================
+ Coverage   81.93%   82.42%   +0.49%     
==========================================
  Files        1136     1139       +3     
  Lines      429152   435636    +6484     
  Branches   429152   435636    +6484     
==========================================
+ Hits       351633   359082    +7449     
+ Misses      56475    54836    -1639     
- Partials    21044    21718     +674

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

kosiew

@Tpt,

Thanks for working on this. I like the goal of making output file sizing more practical when row counts are a poor proxy for size, especially for blob-heavy datasets.

I found one issue that I think needs to be addressed before merging. The new byte limit is currently based on Arrow RecordBatch memory size rather than the number of serialized bytes written to the output file. For Parquet, compression can make those two values very different, so the current behavior can rotate files much earlier than the documented output-file byte target suggests.

I also left one small test coverage suggestion for the new config option.

kosiew · 2026-09-08T10:15:41Z

            part_idx += 1;
        }
        row_counts[next_send_steam] += rb.num_rows();
+        bytes_counts[next_send_steam] += rb.get_array_memory_size();


I think this needs to measure serialized output bytes rather than the Arrow allocation size. get_array_memory_size() gives us the pre-serialization RecordBatch size, but Parquet defaults to zstd(3), so a highly compressible blob batch could be hundreds of MiB in memory and only a small fraction of that on disk.

In that case, the demuxer would rotate before the next batch and could create much smaller files than the configured soft_max_bytes_per_output_file suggests. That also seems particularly relevant to the blob use case this option is intended to help with.

Could we move the accounting to a point where serialized or emitted bytes are known, and feed that back to the demuxer if needed? If the intent is instead to use input memory size as an estimate, I think the option and docs should be narrowed to make that explicit.

It would also be useful to add a regression test using a large, highly compressible binary column with Parquet compression enabled. The current parameterized test uses UNCOMPRESSED, so it would not catch this difference.

Sorry for the slow reply, got trapped by other things and thank you so much for the review

This is a great point. Done. I used a Arc<AtomicUsize> to record and report back the output size without blocking the code. An other approach might be to have an other channel giving back the written size but:

either the demuxer block on it and we lose must of the concurrency and pipelining advantage (I guess very much a no go)

we don't block and we get the same behavior of Arc<AtomicUsize> with more overhead and code

The downside of Arc<AtomicUsize> is indeed that the reporting might lag quite a lot. I have updated the documentation to highlight this.

Allows to roughly limit the size of parquet files when the number of rows is a hard-to-use estimator

kosiew

@Tpt,

Thanks for working on this. I like the direction of adding a byte-based soft limit, especially for workloads where row count is a poor proxy for file size.

I found one blocking issue with how the byte count is measured for Parquet. The current accounting is based on Arrow memory size rather than the encoded output size, which can cause highly compressible data to rotate files far too early. I also left one small suggestion to add SQL-path coverage for rejecting zero.

Once the byte accounting reflects serialized or emitted bytes, this should line up much better with the option's documented behavior.

Tpt · 2026-09-17T12:43:57Z

@kosiew thank you for having a look again at this MR. I am a bit confused, your new comments seems to repeat the ones from your previous review and skip the changes pushed yesterday cb03a02

But maybe I am missing something, sorry if it's the case

kosiew

@Tpt,

Thanks for the follow-up. I reviewed commit cb03a02189 (Use the actual written file size). The Parquet side of the Arrow-memory versus encoded-size issue looks fixed now, including the compressed binary regression coverage. The SQL coverage for rejecting soft_max_bytes_per_output_file = 0 is also in place.

There is still one remaining issue with compressed CSV/JSON output. The current accounting uses the serializer byte count before compression, so the soft limit can still be based on uncompressed bytes for gzip, bzip2, xz, and zstd output.

Because of that, the claim that this now uses the actual written file size is still a bit broader than the implementation. Please either move the accounting to the compressed writer/object-store boundary and add a compressed stateless-format regression test, or explicitly restrict this option to uncompressed stateless writers.

kosiew · 2026-09-18T02:41:25Z

    while let Some(task) = rx.recv().await {
        match task.join().await {
            Ok(Ok((cnt, bytes))) => {
+                serialized_bytes += bytes.len();


serialized_bytes is counting the bytes returned by the CSV/JSON serializer before they go through the compression wrapper installed by ObjectWriterBuilder.

That means COPY ... TO with gzip, bzip2, xz, or zstd can still rotate based on the uncompressed size. For highly compressible data, this reproduces the original issue outside Parquet.

Could we account for bytes at the compressed writer/object-store boundary and add a regression test for a compressed stateless format? Alternatively, if that is not intended to be supported, I think the option should be explicitly restricted to uncompressed stateless writers.

Thank you! I indeed missed that!

Could we account for bytes at the compressed writer/object-store boundary

Indeed. I have updated ObjectWriterBuilder to count the number of bytes flushed to the underlying writer. It is done with a new intermediate AsyncWrite implementation that sits between object store and the compression implementation

kosiew

@Tpt,

Thanks for the updates. The implementation now counts bytes at the right layer, below whole-file compression, and the SQL zero-value validation is covered as well.

I still see two things that should be addressed before merge: regression coverage for compressed output accounting, and the existing SemVer issue from adding the public config field.

kosiew · 2026-09-21T09:04:19Z

+            let config = SessionConfig::from_string_hash_map(&config_map)?;
+            let ctx = SessionContext::new_with_config(config);
+
+            // A deterministic high-entropy payload avoids compression reducing


Could we add a regression case that specifically verifies compressed-size accounting here?

The gzip/zstd cases use high-entropy payloads and only check that files.len() > 1. The previous implementation based on RecordBatch::get_array_memory_size() would also rotate for this data, so these cases would still pass with the behavior we are trying to prevent.

A deterministic, highly compressible CSV or JSON case would cover this better. Ideally it would assert an exact or bounded file count, or otherwise verify emitted size, so that it fails with the previous batch-memory accounting but passes when rotation is based on the compressed bytes actually written.

Good point. Thanks! Done 1554c47

kosiew · 2026-09-21T09:04:19Z

+        /// process RecordBatches. The final file size may exceed this limit due
+        /// to batches buffered before the limit is observed, the size of a batch,
+        /// and file metadata written when the file is finalized.
+        pub soft_max_bytes_per_output_file: ConfigNonZeroUsize, default = non_zero_usize_default(4294967295)


Adding this field breaks downstream struct literals.

Could we either follow the project's approved breaking API process, including any required upgrade or release documentation, or use a compatible design?

This MR is for the main branch, that is targeting next DataFusion breaking release. I guess config changes are fine in them

Agreed this can land on main for the next major release. The concern is process/documentation, not that the change is prohibited.

Per the API health policy, please:

add an entry to the next-major Upgrade Guide explaining affected ExecutionOptions / ConfigOptions struct literals and migration (..Default::default() or initialize the new field).

github-actions Bot added documentation Improvements or additions to documentation core Core DataFusion crate common Related to common crate datasource Changes to the datasource crate labels Sep 3, 2026

github-actions Bot added the auto detected api change Auto detected API change label Sep 3, 2026

Tpt force-pushed the tpt/soft_max_bytes_per_output_bytes branch 2 times, most recently from 20cc5e6 to 94e622b Compare September 3, 2026 18:48

github-actions Bot added the sqllogictest SQL Logic Tests (.slt) label Sep 3, 2026

kosiew requested changes Sep 8, 2026

View reviewed changes

Add datafusion.execution.soft_max_bytes_per_output_file config parameter

d67a0ce

Allows to roughly limit the size of parquet files when the number of rows is a hard-to-use estimator

Tpt force-pushed the tpt/soft_max_bytes_per_output_bytes branch from 94e622b to f9054bc Compare September 16, 2026 17:52

Use the actual written file size

cb03a02

Tpt force-pushed the tpt/soft_max_bytes_per_output_bytes branch from f9054bc to cb03a02 Compare September 16, 2026 18:04

kosiew requested changes Sep 17, 2026

View reviewed changes

Comment thread datafusion/datasource/src/write/demux.rs

Comment thread datafusion/common/src/config.rs

kosiew requested changes Sep 18, 2026

View reviewed changes

Tpt force-pushed the tpt/soft_max_bytes_per_output_bytes branch from 702e0f3 to ae3ee4c Compare September 18, 2026 19:22

Tpt added 2 commits September 18, 2026 21:26

Get the written size from the AsyncRead used

591d9c3

Add a test

1adf75a

Tpt force-pushed the tpt/soft_max_bytes_per_output_bytes branch 2 times, most recently from ae3ee4c to 1adf75a Compare September 18, 2026 19:48

kosiew requested changes Sep 21, 2026

View reviewed changes

kosiew added the api change Changes the API exposed to users of the crate label Sep 21, 2026

Cover compression case

1554c47

Conversation

Tpt commented Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Rationale for this change

What changes are included in this PR?

Uh oh!

github-actions Bot commented Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

codecov-commenter commented Sep 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

kosiew left a comment

Choose a reason for hiding this comment

Uh oh!

kosiew Sep 8, 2026

Choose a reason for hiding this comment

Uh oh!

Tpt Sep 16, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

kosiew left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Tpt commented Sep 17, 2026

Uh oh!

kosiew left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kosiew Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

Tpt Sep 18, 2026

Choose a reason for hiding this comment

Uh oh!

kosiew left a comment

Choose a reason for hiding this comment

Uh oh!

kosiew Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

Tpt Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

kosiew Sep 21, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Tpt Sep 21, 2026

Choose a reason for hiding this comment

Uh oh!

kosiew Sep 22, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Tpt commented Sep 3, 2026 •

edited

Loading

github-actions Bot commented Sep 3, 2026 •

edited

Loading

codecov-commenter commented Sep 3, 2026 •

edited

Loading

Tpt Sep 16, 2026 •

edited

Loading

kosiew left a comment •

edited

Loading

kosiew Sep 21, 2026 •

edited

Loading