You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/11 07:40:44 UTC
(datafusion) branch gh-readonly-queue/main/pr-24204-a05388e82374ed2943595980b0539b6c1fa629a0 created (now a1f64e94b4)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24204-a05388e82374ed2943595980b0539b6c1fa629a0
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at a1f64e94b4 fix: reject max_buffered_batches_per_output_file values below 2 (#24204)
This branch includes the following new commits:
new a1f64e94b4 fix: reject max_buffered_batches_per_output_file values below 2 (#24204)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: fix: reject max_buffered_batches_per_output_file values below 2 (#24204)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24204-a05388e82374ed2943595980b0539b6c1fa629a0
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a1f64e94b4d4d2eec6d4a8af448cedb86a075f75
Author: DevShiba <11...@users.noreply.github.com>
AuthorDate: Tue Aug 11 04:40:22 2026 -0300
fix: reject max_buffered_batches_per_output_file values below 2 (#24204)
## Which issue does this PR close?
- Part of https://github.com/apache/datafusion/issues/17498
## Rationale for this change
```
DataFusion CLI v54.1.0
> set datafusion.execution.max_buffered_batches_per_output_file = 1;
> COPY (SELECT 1 as a) TO '/tmp/x.parquet';
thread 'tokio-rt-worker' panicked at datafusion/datasource/src/write/demux.rs:287:30:
mpsc bounded channel requires buffer > 0
```
Two call sites (`demux.rs::create_new_file_stream`,
`orchestration.rs::spawn_writer_tasks_and_join`) divide
`max_buffered_batches_per_output_file` in half to size a bounded `mpsc`
channel's capacity. Integer division rounds both 0 *and* 1 down to 0,
and Tokio's `mpsc::channel` panics on a zero capacity. A plain non-zero
check (like the existing `ConfigNonZeroUsize`) would not have been
sufficient here, since 1 also triggers the panic.
A third call site (`demux.rs::hive_style_partitions_demuxer`) uses the
raw value directly without dividing, so it would panic on 0 alone.
## What changes are included in this PR?
Adds `ConfigMinTwoUsize`, mirroring the existing `ConfigNonZeroUsize`
pattern already used for sibling fields (`batch_size`,
`meta_fetch_concurrency`, `minimum_parallel_output_files`, etc.), and
applies it to `max_buffered_batches_per_output_file`. Invalid values are
now rejected with a clear configuration error at set-time instead of
panicking later at write time. Updated the three read sites to call
`.get()`, updated the field doc comment to explain the constraint, and
regenerated `docs/source/user-guide/configs.md` via
`dev/update_config_docs.sh`.
## Are these changes tested?
Yes. Added two `statement error` cases to
`datafusion/sqllogictest/test_files/set_variable.slt` (values 0 and 1),
following the exact pattern already used for the sibling
`ConfigNonZeroUsize` fields in that file. Verified manually with
`datafusion-cli` that 0 and 1 now return a clean error instead of
panicking, and that 2 (the default) and 3 still work correctly. Ran
`cargo test -p datafusion-datasource --lib` (178 passed), the
`set_variable.slt` sqllogictest suite, and `cargo check --workspace
--all-targets` — all clean.
## Are there any user-facing changes?
Yes: setting `datafusion.execution.max_buffered_batches_per_output_file`
to `0` or `1` now returns a configuration error instead of panicking. No
change for any value `>= 2` (including the default of 2).
---
datafusion/common/src/config.rs | 97 +++++++++++++++++++++-
datafusion/datasource/src/write/demux.rs | 5 +-
datafusion/datasource/src/write/orchestration.rs | 2 +-
.../sqllogictest/test_files/information_schema.slt | 2 +-
.../sqllogictest/test_files/set_variable.slt | 19 +++++
docs/source/user-guide/configs.md | 2 +-
6 files changed, 120 insertions(+), 7 deletions(-)
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index f0be10bc6c..f5742f09f9 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -663,6 +663,91 @@ impl Display for ConfigNonZeroUsize {
}
}
+/// A `usize` configuration value that rejects 0 and 1 when set from strings.
+///
+/// Use this for options whose consumer divides the value in half to size an
+/// internal buffer (e.g. a bounded channel capacity): values below 2 would
+/// round down to a zero-capacity buffer and panic. Invalid values return a
+/// configuration error through [`ConfigField`] instead.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ConfigMinTwoUsize(usize);
+
+/// Private helper for hard-coded defaults in `config_namespace!`, which cannot
+/// use `?`. All external construction should use [`ConfigMinTwoUsize::try_new`].
+const fn min_two_usize_default(value: usize) -> ConfigMinTwoUsize {
+ if value >= 2 {
+ ConfigMinTwoUsize(value)
+ } else {
+ panic!("value must be at least 2")
+ }
+}
+
+impl ConfigMinTwoUsize {
+ /// Creates a [`ConfigMinTwoUsize`], returning a configuration error if
+ /// `value` is less than 2.
+ pub fn try_new(value: usize) -> Result<Self> {
+ if value >= 2 {
+ Ok(Self(value))
+ } else {
+ _config_err!("value must be at least 2")
+ }
+ }
+
+ /// Returns the wrapped `usize`.
+ pub const fn get(self) -> usize {
+ self.0
+ }
+}
+
+impl From<ConfigMinTwoUsize> for usize {
+ fn from(value: ConfigMinTwoUsize) -> Self {
+ value.get()
+ }
+}
+
+impl FromStr for ConfigMinTwoUsize {
+ type Err = DataFusionError;
+
+ fn from_str(s: &str) -> Result<Self, Self::Err> {
+ Self::try_new(default_config_transform(s)?)
+ }
+}
+
+impl ConfigField for ConfigMinTwoUsize {
+ fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
+ v.some(key, self, description)
+ }
+
+ fn set(&mut self, key: &str, value: &str) -> Result<()> {
+ if !key.is_empty() {
+ return _config_err!(
+ "Config field max_buffered_batches_per_output_file is a scalar ConfigMinTwoUsize and does not have nested field \"{}\"",
+ key
+ );
+ }
+
+ *self = ConfigMinTwoUsize::from_str(value)?;
+ Ok(())
+ }
+
+ fn reset(&mut self, key: &str) -> Result<()> {
+ if key.is_empty() {
+ Ok(())
+ } else {
+ _config_err!(
+ "Config field max_buffered_batches_per_output_file is a scalar ConfigMinTwoUsize and does not have nested field \"{}\"",
+ key
+ )
+ }
+ }
+}
+
+impl Display for ConfigMinTwoUsize {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ write!(f, "{}", self.get())
+ }
+}
+
/// Policy for handling duplicate keys in Spark-compatible map-construction
/// functions (`map_from_arrays`, `map_from_entries`, `str_to_map`). Mirrors
/// Spark's [`spark.sql.mapKeyDedupPolicy`](https://github.com/apache/spark/blob/cf3a34e19dfcf70e2d679217ff1ba21302212472/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L4961).
@@ -878,8 +963,16 @@ config_namespace! {
/// This is the maximum number of RecordBatches buffered
/// for each output file being worked. Higher values can potentially
/// give faster write performance at the cost of higher peak
- /// memory consumption
- pub max_buffered_batches_per_output_file: usize, default = 2
+ /// memory consumption.
+ ///
+ /// This budget is split evenly between two independent points in the
+ /// write pipeline (see the demuxer diagram in #7791): how many files
+ /// can be in flight from the demuxer to a writer task, and how many
+ /// RecordBatches are buffered for a single file's writer. Must be at
+ /// least 2 so each half gets at least 1 unit of buffering - 0 or 1
+ /// would leave one side with a zero-capacity channel and panic at
+ /// write time.
+ pub max_buffered_batches_per_output_file: ConfigMinTwoUsize, default = min_two_usize_default(2)
/// Should sub directories be ignored when scanning directories for data
/// files. Defaults to true (ignores subdirectories), consistent with
diff --git a/datafusion/datasource/src/write/demux.rs b/datafusion/datasource/src/write/demux.rs
index 6d7de53890..1b3098d309 100644
--- a/datafusion/datasource/src/write/demux.rs
+++ b/datafusion/datasource/src/write/demux.rs
@@ -154,7 +154,7 @@ async fn row_count_demuxer(
let exec_options = &context.session_config().options().execution;
let max_rows_per_file = exec_options.soft_max_rows_per_output_file.get();
- let max_buffered_batches = exec_options.max_buffered_batches_per_output_file;
+ let max_buffered_batches = exec_options.max_buffered_batches_per_output_file.get();
let minimum_parallel_files = exec_options.minimum_parallel_output_files.get();
let mut part_idx = 0;
let write_id = rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 16);
@@ -305,7 +305,8 @@ async fn hive_style_partitions_demuxer(
let write_id = rand::distr::Alphanumeric.sample_string(&mut rand::rng(), 16);
let exec_options = &context.session_config().options().execution;
- let max_buffered_recordbatches = exec_options.max_buffered_batches_per_output_file;
+ let max_buffered_recordbatches =
+ exec_options.max_buffered_batches_per_output_file.get();
// To support non string partition col types, cast the type to &str first
let mut value_map: HashMap<Vec<String>, Sender<RecordBatch>> = HashMap::new();
diff --git a/datafusion/datasource/src/write/orchestration.rs b/datafusion/datasource/src/write/orchestration.rs
index 39c91a1c0d..f75671d950 100644
--- a/datafusion/datasource/src/write/orchestration.rs
+++ b/datafusion/datasource/src/write/orchestration.rs
@@ -259,7 +259,7 @@ pub async fn spawn_writer_tasks_and_join(
.execution
.max_buffered_batches_per_output_file;
- let (tx_file_bundle, rx_file_bundle) = mpsc::channel(rb_buffer_size / 2);
+ let (tx_file_bundle, rx_file_bundle) = mpsc::channel(rb_buffer_size.get() / 2);
let (tx_row_cnt, rx_row_cnt) = tokio::sync::oneshot::channel();
let write_coordinator_task = SpawnedTask::spawn(async move {
stateless_serialize_and_write_files(rx_file_bundle, tx_row_cnt).await
diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt
index 90bb55b0f0..573fb04b34 100644
--- a/datafusion/sqllogictest/test_files/information_schema.slt
+++ b/datafusion/sqllogictest/test_files/information_schema.slt
@@ -386,7 +386,7 @@ datafusion.execution.hash_join_buffering_capacity 0 How many bytes to buffer in
datafusion.execution.keep_partition_by_columns false Should DataFusion keep the columns used for partition_by in the output RecordBatches
datafusion.execution.listing_table_factory_infer_partitions true Should a `ListingTable` created through the `ListingTableFactory` infer table partitions from Hive compliant directories. Defaults to true (partition columns are inferred and will be represented in the table schema).
datafusion.execution.listing_table_ignore_subdirectory true Should sub directories be ignored when scanning directories for data files. Defaults to true (ignores subdirectories), consistent with Hive. Note that this setting does not affect reading partitioned tables (e.g. `/table/year=2021/month=01/data.parquet`).
-datafusion.execution.max_buffered_batches_per_output_file 2 This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption
+datafusion.execution.max_buffered_batches_per_output_file 2 This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption. This budget is split evenly between two independent points in the write pipeline (see the demuxer diagram in #7791): how many files can be in flight from the demuxer to a writer task, and how many RecordBatches are buffered for a single [...]
datafusion.execution.max_spill_file_size_bytes 134217728 Maximum size in bytes for individual spill files before rotating to a new file. When operators spill data to disk (e.g., RepartitionExec), they write multiple batches to the same file until this size limit is reached, then rotate to a new file. This reduces syscall overhead compared to one-file-per-batch while preventing files from growing too large. A larger value reduces file creation overhead but may hold more disk space. A smal [...]
datafusion.execution.meta_fetch_concurrency 32 Number of files to read in parallel when inferring schema and statistics
datafusion.execution.minimum_parallel_output_files 4 Guarantees a minimum level of output files running in parallel. RecordBatches will be distributed in round robin fashion to each parallel writer. Each writer is closed and a new file opened once soft_max_rows_per_output_file is reached.
diff --git a/datafusion/sqllogictest/test_files/set_variable.slt b/datafusion/sqllogictest/test_files/set_variable.slt
index 7da06b2fff..b8db761e79 100644
--- a/datafusion/sqllogictest/test_files/set_variable.slt
+++ b/datafusion/sqllogictest/test_files/set_variable.slt
@@ -759,6 +759,25 @@ caused by
Invalid or Unsupported Configuration: value must be greater than 0
+# max_buffered_batches_per_output_file is halved to size an internal channel
+# capacity, so 0 and 1 both round down to a zero-capacity channel and must be
+# rejected, not just 0.
+statement error
+SET datafusion.execution.max_buffered_batches_per_output_file = 0
+----
+DataFusion error: Error setting config datafusion.execution.max_buffered_batches_per_output_file
+caused by
+Invalid or Unsupported Configuration: value must be at least 2
+
+
+statement error
+SET datafusion.execution.max_buffered_batches_per_output_file = 1
+----
+DataFusion error: Error setting config datafusion.execution.max_buffered_batches_per_output_file
+caused by
+Invalid or Unsupported Configuration: value must be at least 2
+
+
# Config reset
statement ok
RESET datafusion.catalog.create_default_catalog_and_schema
diff --git a/docs/source/user-guide/configs.md b/docs/source/user-guide/configs.md
index 860884e11f..a66ad3edf5 100644
--- a/docs/source/user-guide/configs.md
+++ b/docs/source/user-guide/configs.md
@@ -130,7 +130,7 @@ The following configuration settings are available:
| datafusion.execution.meta_fetch_concurrency | 32 | Number of files to read in parallel when inferring schema and statistics [...]
| datafusion.execution.minimum_parallel_output_files | 4 | Guarantees a minimum level of output files running in parallel. RecordBatches will be distributed in round robin fashion to each parallel writer. Each writer is closed and a new file opened once soft_max_rows_per_output_file is reached. [...]
| datafusion.execution.soft_max_rows_per_output_file | 50000000 | Target number of rows in output files when writing multiple. This is a soft max, so it can be exceeded slightly. There also will be one file smaller than the limit if the total number of rows written is not roughly divisible by the soft max [...]
-| datafusion.execution.max_buffered_batches_per_output_file | 2 | This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption [...]
+| datafusion.execution.max_buffered_batches_per_output_file | 2 | This is the maximum number of RecordBatches buffered for each output file being worked. Higher values can potentially give faster write performance at the cost of higher peak memory consumption. This budget is split evenly between two independent points in the write pipeline (see the demuxer diagram in #7791): how many files can be in flight from the demuxer to a writer task, and how m [...]
| datafusion.execution.listing_table_ignore_subdirectory | true | Should sub directories be ignored when scanning directories for data files. Defaults to true (ignores subdirectories), consistent with Hive. Note that this setting does not affect reading partitioned tables (e.g. `/table/year=2021/month=01/data.parquet`). [...]
| datafusion.execution.listing_table_factory_infer_partitions | true | Should a `ListingTable` created through the `ListingTableFactory` infer table partitions from Hive compliant directories. Defaults to true (partition columns are inferred and will be represented in the table schema). [...]
| datafusion.execution.enable_recursive_ctes | true | Should DataFusion support recursive CTEs [...]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org