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/06/22 00:11:30 UTC
(datafusion) branch gh-readonly-queue/main/pr-23054-a27f030d5829a4460e87dbb8d2a6382c8f9ecd4f created (now 4e54312e43)
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-23054-a27f030d5829a4460e87dbb8d2a6382c8f9ecd4f
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at 4e54312e43 minor: Validate `batch_size` configuration when setting it (#23054)
This branch includes the following new commits:
new 4e54312e43 minor: Validate `batch_size` configuration when setting it (#23054)
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: minor: Validate `batch_size` configuration when setting it (#23054)
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-23054-a27f030d5829a4460e87dbb8d2a6382c8f9ecd4f
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 4e54312e43f8a402145bfbc8c6b4dbbd9ad871ae
Author: Yongting You <20...@gmail.com>
AuthorDate: Mon Jun 22 08:11:07 2026 +0800
minor: Validate `batch_size` configuration when setting it (#23054)
## Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax. For example
`Closes #123` indicates that this PR will close issue #123.
-->
- Part of https://github.com/apache/datafusion/issues/17498
## Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
In datafusion-cli:
```
DataFusion CLI v53.1.0
> set datafusion.execution.batch_size=0;
0 row(s) fetched.
Elapsed 0.001 seconds.
```
A '0' batch_size is a invalid config value, and it's widely used in most
operators to control behavior. If operators don't validate `batch_size`,
the 0 value might cause bugs/panics. This PR rejects it when setting the
config.
I triggered a bug on my first attempt:
```
> set datafusion.execution.batch_size=0;
0 row(s) fetched.
Elapsed 0.000 seconds.
> select * from generate_series(100);
+-------+
| value |
+-------+
+-------+
0 row(s) fetched.
Elapsed 0.004 seconds.
```
## What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
Implement a custom type for non-zero usize configuration value on
`batch_size`, and rejects '0' value on initialization. Rust native
`NonZero<u32>` is not used, because custom type allows to provide better
error message on errors, and the implementation complexity is similar.
Note other invalid inputs like '-1' 'a' are already rejected and covered
by existing tests.
## Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
3. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
Yes, sqllogictest
## Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
-->
<!--
If there are any breaking changes to public APIs, please add the `api
change` label.
-->
No
---
datafusion-cli/src/main.rs | 3 +-
datafusion/common/src/config.rs | 85 +++++++++++++++++++++-
datafusion/core/tests/config_from_env.rs | 4 +-
.../core/tests/execution/datasource_split.rs | 3 +-
datafusion/core/tests/parquet/filter_pushdown.rs | 3 +-
.../physical_optimizer/enforce_distribution.rs | 4 +-
datafusion/core/tests/sql/runtime_config.rs | 2 +-
datafusion/execution/src/config.rs | 26 ++++---
datafusion/functions-table/src/generate_series.rs | 2 +-
.../ensure_requirements/enforce_distribution.rs | 2 +-
datafusion/physical-plan/src/async_func.rs | 2 +-
.../src/sorts/sort_preserving_merge.rs | 3 +-
.../sqllogictest/test_files/set_variable.slt | 10 ++-
13 files changed, 121 insertions(+), 28 deletions(-)
diff --git a/datafusion-cli/src/main.rs b/datafusion-cli/src/main.rs
index a86896c41d..cf77e5415d 100644
--- a/datafusion-cli/src/main.rs
+++ b/datafusion-cli/src/main.rs
@@ -335,7 +335,8 @@ fn get_session_config(args: &Args) -> Result<SessionConfig> {
if batch_size == 0 {
return config_err!("batch_size must be greater than 0");
}
- config_options.execution.batch_size = batch_size;
+ config_options.execution.batch_size =
+ datafusion_common::config::ConfigNonZeroUsize::try_new(batch_size)?;
};
// use easier to understand "tree" mode by default
diff --git a/datafusion/common/src/config.rs b/datafusion/common/src/config.rs
index 536afbfed4..cc263dfe3e 100644
--- a/datafusion/common/src/config.rs
+++ b/datafusion/common/src/config.rs
@@ -21,7 +21,7 @@ use arrow_ipc::CompressionType;
#[cfg(feature = "parquet_encryption")]
use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
-use crate::error::_config_err;
+use crate::error::{_config_datafusion_err, _config_err};
use crate::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
use crate::parquet_config::DFParquetWriterVersion;
use crate::parsers::{CompressionTypeVariant, CsvQuoteStyle};
@@ -33,6 +33,7 @@ use std::any::Any;
use std::collections::{BTreeMap, HashMap};
use std::error::Error;
use std::fmt::{self, Display};
+use std::num::NonZeroUsize;
use std::str::FromStr;
#[cfg(feature = "parquet_encryption")]
use std::sync::Arc;
@@ -582,6 +583,86 @@ impl Display for SpillCompression {
}
}
+/// A `usize` configuration value that rejects zero when set from strings.
+///
+/// Use this for options where zero is never a meaningful runtime value.
+/// Invalid values return a configuration error through [`ConfigField`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct ConfigNonZeroUsize(NonZeroUsize);
+
+/// Private helper for hard-coded defaults in `config_namespace!`, which cannot
+/// use `?`. All external construction should use [`ConfigNonZeroUsize::try_new`].
+const fn non_zero_usize_default(value: usize) -> ConfigNonZeroUsize {
+ match NonZeroUsize::new(value) {
+ Some(value) => ConfigNonZeroUsize(value),
+ None => panic!("value must be greater than 0"),
+ }
+}
+
+impl ConfigNonZeroUsize {
+ /// Creates a [`ConfigNonZeroUsize`], returning a configuration error if
+ /// `value` is zero.
+ pub fn try_new(value: usize) -> Result<Self> {
+ NonZeroUsize::new(value)
+ .map(Self)
+ .ok_or_else(|| _config_datafusion_err!("value must be greater than 0"))
+ }
+
+ /// Returns the wrapped `usize`.
+ pub const fn get(self) -> usize {
+ self.0.get()
+ }
+}
+
+impl From<ConfigNonZeroUsize> for usize {
+ fn from(value: ConfigNonZeroUsize) -> Self {
+ value.get()
+ }
+}
+
+impl FromStr for ConfigNonZeroUsize {
+ type Err = DataFusionError;
+
+ fn from_str(s: &str) -> Result<Self, Self::Err> {
+ Self::try_new(default_config_transform(s)?)
+ }
+}
+
+impl ConfigField for ConfigNonZeroUsize {
+ 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 batch_size is a scalar ConfigNonZeroUsize and does not have nested field \"{}\"",
+ key
+ );
+ }
+
+ *self = ConfigNonZeroUsize::from_str(value)?;
+ Ok(())
+ }
+
+ fn reset(&mut self, key: &str) -> Result<()> {
+ if key.is_empty() {
+ Ok(())
+ } else {
+ _config_err!(
+ "Config field batch_size is a scalar ConfigNonZeroUsize and does not have nested field \"{}\"",
+ key
+ )
+ }
+ }
+}
+
+impl Display for ConfigNonZeroUsize {
+ 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).
@@ -649,7 +730,7 @@ config_namespace! {
/// Default batch size while creating new batches, it's especially useful for
/// buffer-in-memory batches since creating tiny batches would result in too much
/// metadata memory consumption
- pub batch_size: usize, default = 8192
+ pub batch_size: ConfigNonZeroUsize, default = non_zero_usize_default(8192)
/// A perfect hash join (see `HashJoinExec` for more details) will be considered
/// if the range of keys (max - min) on the build side is < this threshold.
diff --git a/datafusion/core/tests/config_from_env.rs b/datafusion/core/tests/config_from_env.rs
index 6375d4e25d..6b09a6367d 100644
--- a/datafusion/core/tests/config_from_env.rs
+++ b/datafusion/core/tests/config_from_env.rs
@@ -45,7 +45,7 @@ fn from_env() {
// for valid testing
env::set_var(env_key, "4096");
let config = ConfigOptions::from_env().unwrap();
- assert_eq!(config.execution.batch_size, 4096);
+ assert_eq!(config.execution.batch_size.get(), 4096);
// for invalid testing
env::set_var(env_key, "abc");
@@ -57,6 +57,6 @@ fn from_env() {
env::remove_var(env_key);
let config = ConfigOptions::from_env().unwrap();
- assert_eq!(config.execution.batch_size, 8192); // set to its default value
+ assert_eq!(config.execution.batch_size.get(), 8192); // set to its default value
}
}
diff --git a/datafusion/core/tests/execution/datasource_split.rs b/datafusion/core/tests/execution/datasource_split.rs
index 370249cd80..171e873649 100644
--- a/datafusion/core/tests/execution/datasource_split.rs
+++ b/datafusion/core/tests/execution/datasource_split.rs
@@ -61,6 +61,7 @@ async fn datasource_splits_large_batches() -> datafusion_common::Result<()> {
.options()
.execution
.batch_size
+ .get()
);
let total: usize = batches.iter().map(|b| b.num_rows()).sum();
assert_eq!(total, batch_size);
@@ -70,7 +71,7 @@ async fn datasource_splits_large_batches() -> datafusion_common::Result<()> {
#[tokio::test]
async fn datasource_exact_batch_size_no_split() -> datafusion_common::Result<()> {
let session_config = datafusion_execution::config::SessionConfig::new();
- let configured_batch_size = session_config.options().execution.batch_size;
+ let configured_batch_size = session_config.options().execution.batch_size.get();
let batches = create_and_collect_batches(configured_batch_size).await?;
diff --git a/datafusion/core/tests/parquet/filter_pushdown.rs b/datafusion/core/tests/parquet/filter_pushdown.rs
index e6266b2c08..5dfcd50c01 100644
--- a/datafusion/core/tests/parquet/filter_pushdown.rs
+++ b/datafusion/core/tests/parquet/filter_pushdown.rs
@@ -648,7 +648,8 @@ async fn predicate_cache_stats_issue_19561() -> datafusion_common::Result<()> {
let mut config = SessionConfig::new();
config.options_mut().execution.parquet.pushdown_filters = true;
// force to get multiple batches to trigger repeated metric compound bug
- config.options_mut().execution.batch_size = 1;
+ config.options_mut().execution.batch_size =
+ datafusion_common::config::ConfigNonZeroUsize::try_new(1)?;
let ctx = SessionContext::new_with_config(config);
// The cache is on by default, and used when filter pushdown is enabled
PredicateCacheTest {
diff --git a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
index 426e1fa745..9424322396 100644
--- a/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
+++ b/datafusion/core/tests/physical_optimizer/enforce_distribution.rs
@@ -570,7 +570,9 @@ fn test_suite_default_config_options() -> ConfigOptions {
config.execution.target_partitions = 10;
// Use a small batch size, to trigger RoundRobin in tests
- config.execution.batch_size = 1;
+ config.execution.batch_size =
+ datafusion_common::config::ConfigNonZeroUsize::try_new(1)
+ .expect("test batch size must be greater than zero");
config
}
diff --git a/datafusion/core/tests/sql/runtime_config.rs b/datafusion/core/tests/sql/runtime_config.rs
index a9f57a0793..604d137540 100644
--- a/datafusion/core/tests/sql/runtime_config.rs
+++ b/datafusion/core/tests/sql/runtime_config.rs
@@ -115,7 +115,7 @@ async fn test_multiple_configs() {
assert!(result.is_ok(), "Should not fail due to memory limit");
let state = ctx.state();
- let batch_size = state.config().options().execution.batch_size;
+ let batch_size = state.config().options().execution.batch_size.get();
assert_eq!(batch_size, 2048);
}
diff --git a/datafusion/execution/src/config.rs b/datafusion/execution/src/config.rs
index b2917a4583..0a2a98eab6 100644
--- a/datafusion/execution/src/config.rs
+++ b/datafusion/execution/src/config.rs
@@ -19,7 +19,7 @@ use std::{collections::HashMap, sync::Arc};
use datafusion_common::{
Result, ScalarValue,
- config::{ConfigExtension, ConfigOptions, SpillCompression},
+ config::{ConfigExtension, ConfigNonZeroUsize, ConfigOptions, SpillCompression},
extensions::Extensions,
};
@@ -51,7 +51,7 @@ use datafusion_common::{
/// .set_bool("datafusion.execution.parquet.pushdown_filters", true);
///
/// assert_eq!(config.batch_size(), 1234);
-/// assert_eq!(config.options().execution.batch_size, 1234);
+/// assert_eq!(config.options().execution.batch_size.get(), 1234);
/// assert_eq!(config.options().execution.parquet.pushdown_filters, true);
/// ```
///
@@ -60,15 +60,16 @@ use datafusion_common::{
///
/// ```
/// # use datafusion_execution::config::SessionConfig;
-/// # use datafusion_common::ScalarValue;
+/// # use datafusion_common::config::ConfigNonZeroUsize;
/// #
/// let mut config = SessionConfig::new();
-/// config.options_mut().execution.batch_size = 1234;
+/// config.options_mut().execution.batch_size = ConfigNonZeroUsize::try_new(1234)?;
/// config.options_mut().execution.parquet.pushdown_filters = true;
/// #
/// # assert_eq!(config.batch_size(), 1234);
-/// # assert_eq!(config.options().execution.batch_size, 1234);
+/// # assert_eq!(config.options().execution.batch_size.get(), 1234);
/// # assert_eq!(config.options().execution.parquet.pushdown_filters, true);
+/// # datafusion_common::Result::<()>::Ok(())
/// ```
///
/// ## Built-in options
@@ -137,7 +138,7 @@ impl SessionConfig {
/// use datafusion_execution::config::SessionConfig;
///
/// let config = SessionConfig::new();
- /// assert!(config.options().execution.batch_size > 0);
+ /// assert!(config.options().execution.batch_size.get() > 0);
/// ```
pub fn options(&self) -> &Arc<ConfigOptions> {
&self.options
@@ -148,11 +149,13 @@ impl SessionConfig {
/// Can be used to set configuration options.
///
/// ```
+ /// use datafusion_common::config::ConfigNonZeroUsize;
/// use datafusion_execution::config::SessionConfig;
///
/// let mut config = SessionConfig::new();
- /// config.options_mut().execution.batch_size = 1024;
- /// assert_eq!(config.options().execution.batch_size, 1024);
+ /// config.options_mut().execution.batch_size = ConfigNonZeroUsize::try_new(1024)?;
+ /// assert_eq!(config.options().execution.batch_size.get(), 1024);
+ /// # datafusion_common::Result::<()>::Ok(())
/// ```
pub fn options_mut(&mut self) -> &mut ConfigOptions {
Arc::make_mut(&mut self.options)
@@ -186,9 +189,8 @@ impl SessionConfig {
/// Customize batch size
pub fn with_batch_size(mut self, n: usize) -> Self {
- // batch size must be greater than zero
- assert!(n > 0);
- self.options_mut().execution.batch_size = n;
+ self.options_mut().execution.batch_size =
+ ConfigNonZeroUsize::try_new(n).expect("batch size must be greater than zero");
self
}
@@ -391,7 +393,7 @@ impl SessionConfig {
/// Get the currently configured batch size
pub fn batch_size(&self) -> usize {
- self.options.execution.batch_size
+ self.options.execution.batch_size.get()
}
/// Enables or disables the coalescence of small batches into larger batches
diff --git a/datafusion/functions-table/src/generate_series.rs b/datafusion/functions-table/src/generate_series.rs
index 52baa7e6cf..0e8eca6bc2 100644
--- a/datafusion/functions-table/src/generate_series.rs
+++ b/datafusion/functions-table/src/generate_series.rs
@@ -484,7 +484,7 @@ impl TableProvider for GenerateSeriesTable {
_filters: &[Expr],
_limit: Option<usize>,
) -> Result<Arc<dyn ExecutionPlan>> {
- let batch_size = state.config_options().execution.batch_size;
+ let batch_size = state.config_options().execution.batch_size.get();
let generator = self.as_generator(batch_size)?;
let mut exec = LazyMemoryExec::try_new(self.schema(), vec![generator])?
.with_projection(projection.cloned());
diff --git a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
index 6d9550fa50..76cb59a305 100644
--- a/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
+++ b/datafusion/physical-optimizer/src/ensure_requirements/enforce_distribution.rs
@@ -1080,7 +1080,7 @@ pub fn ensure_distribution(
// When `false`, round robin repartition will not be added to increase parallelism
let enable_round_robin = config.optimizer.enable_round_robin_repartition;
let repartition_file_scans = config.optimizer.repartition_file_scans;
- let batch_size = config.execution.batch_size;
+ let batch_size = config.execution.batch_size.get();
let should_use_estimates = config
.execution
.use_row_number_estimates_to_optimize_partitioning;
diff --git a/datafusion/physical-plan/src/async_func.rs b/datafusion/physical-plan/src/async_func.rs
index 94860e54ca..efac83bbbe 100644
--- a/datafusion/physical-plan/src/async_func.rs
+++ b/datafusion/physical-plan/src/async_func.rs
@@ -208,7 +208,7 @@ impl ExecutionPlan for AsyncFuncExec {
input_stream,
batch_coalescer: LimitedBatchCoalescer::new(
Arc::clone(&self.input.schema()),
- config_options_ref.execution.batch_size,
+ config_options_ref.execution.batch_size.get(),
None,
),
};
diff --git a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
index dcf3a7baad..77a7d8f8f2 100644
--- a/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
+++ b/datafusion/physical-plan/src/sorts/sort_preserving_merge.rs
@@ -468,7 +468,8 @@ mod tests {
.with_memory_limit(20_000_000, 1.0)
.build_arc()?;
let mut config = SessionConfig::new();
- config.options_mut().execution.batch_size = target_batch_size;
+ config.options_mut().execution.batch_size =
+ datafusion_common::config::ConfigNonZeroUsize::try_new(target_batch_size)?;
let task_ctx = TaskContext::default()
.with_runtime(runtime)
.with_session_config(config);
diff --git a/datafusion/sqllogictest/test_files/set_variable.slt b/datafusion/sqllogictest/test_files/set_variable.slt
index 6f58e5fb31..0514deba28 100644
--- a/datafusion/sqllogictest/test_files/set_variable.slt
+++ b/datafusion/sqllogictest/test_files/set_variable.slt
@@ -104,12 +104,12 @@ statement ok
set datafusion.catalog.information_schema = true
statement ok
-SET datafusion.execution.batch_size to 0
+SET datafusion.execution.batch_size to 310104
query TT
SHOW datafusion.execution.batch_size
----
-datafusion.execution.batch_size 0
+datafusion.execution.batch_size 310104
statement ok
SET datafusion.execution.batch_size to '1'
@@ -382,7 +382,7 @@ statement error DataFusion error: Invalid or Unsupported Configuration: Config v
RESET datafusion.execution.batches_size
# reset invalid variable - extra suffix on valid field
-statement error DataFusion error: Invalid or Unsupported Configuration: Config field is a scalar usize and does not have nested field "bar"
+statement error DataFusion error: Invalid or Unsupported Configuration: Config field batch_size is a scalar ConfigNonZeroUsize and does not have nested field "bar"
RESET datafusion.execution.batch_size.bar
#############################################
@@ -707,6 +707,10 @@ SET datafusion.runtime.list_files_cache_ttl = '1m18446744073709551555s'
statement error DataFusion error: Error during planning: Duration has overflowed allowed maximum limit due to 'mins \* 60 \+ secs' when setting 'datafusion\.runtime\.list_files_cache_ttl'
SET datafusion.runtime.list_files_cache_ttl = '1m18446744073709551556s'
+# Set invalid value and ensures error
+statement error DataFusion error: Invalid or Unsupported Configuration: value must be greater than 0
+SET datafusion.execution.batch_size = 0
+
# Config reset
statement ok
RESET datafusion.catalog.create_default_catalog_and_schema
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org