From d46f5a569fdbee7b0019197e49044d1f9b10c854 Mon Sep 17 00:00:00 2001 From: Hao Chun Chang Date: Wed, 9 Sep 2026 19:09:12 +0800 Subject: [PATCH 1/2] refactor: drive `datafusion.runtime.*` config from a `ConfigField` schema The eight `datafusion.runtime.*` keys were spelled out in six places: the private `RuntimeConfigValues` struct and its `into_config_entries`, the value derivation in `RuntimeEnv::config_entries`, a hardcoded default table in `RuntimeEnvBuilder::entries`, and a hand-written `match` arm per key in both `SessionContext::set_runtime_variable` and `reset_runtime_variable`. A comment in `set_runtime_variable` asked contributors to remember to update `reset_runtime_variable` when adding an option, and the hand-rolled parse arms are the code that produced the panic fixed in #23316. Add `datafusion_execution::runtime_options::RuntimeOptions`, built with the `config_namespace!` / `config_field!` macros that already back `ConfigOptions`, and make it the single source of the keys, descriptions, defaults and parsing. `SET` and `RESET` now dispatch through `ConfigField`, so adding a runtime option is a one-line struct field. `RuntimeEnvBuilder::entries` reads its defaults from the same constants the builder uses instead of hardcoded strings such as `"100G"`, which removes a silent drift risk. `RuntimeOptions` deliberately stays outside `ConfigOptions`: `datafusion-execution` depends on `datafusion-common`, never the reverse. It is also not a `ConfigExtension`, because extension prefixes may never be `datafusion`. Three per-key listings remain and are now all in one file: the schema, the `from_runtime_env` read of the live resource objects, and the `apply_key` write onto `RuntimeEnvBuilder`. The last two cannot merge into the schema because the builder setters have per-field signatures, and because reported values must come from the live objects rather than from whatever was last requested. Public API is unchanged. `SessionContext::parse_capacity_limit` keeps its signature, its doctest and its behaviour, and now delegates to the shared parser. `RuntimeEnv` and `RuntimeEnvBuilder` keep every public field and setter. Error text changes. `ConfigField::set` is not given the key it is setting, so a leaf parser cannot name the key mid-message the way the old hand-written arms did. Errors now carry the key as a `when setting ''` suffix, which also makes the wording consistent: the previous messages used `for ''` in some arms and `when setting ''` in others. Four assertions are updated to match. `RuntimeEnv::config_entries` output is unchanged, including `unlimited` for an unbounded pool, and the generated `configs.md` runtime table is byte-identical apart from prettier's column padding. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01TpQVFSJ6v2yfMNqwbsfsRe --- datafusion/core/src/execution/context/mod.rs | 219 ++------ datafusion/core/tests/sql/runtime_config.rs | 14 +- datafusion/execution/src/lib.rs | 1 + datafusion/execution/src/runtime_env.rs | 188 +------ datafusion/execution/src/runtime_options.rs | 499 ++++++++++++++++++ .../sqllogictest/test_files/set_variable.slt | 4 +- 6 files changed, 559 insertions(+), 366 deletions(-) create mode 100644 datafusion/execution/src/runtime_options.rs diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs index ff1ad25811440..596c3677a5b63 100644 --- a/datafusion/core/src/execution/context/mod.rs +++ b/datafusion/core/src/execution/context/mod.rs @@ -20,7 +20,6 @@ use std::collections::HashSet; use std::fmt::Debug; use std::sync::{Arc, Weak}; -use std::time::Duration; use super::options::ReadOptions; use crate::datasource::dynamic_file::DynamicListTableFactory; @@ -75,15 +74,9 @@ use datafusion_common::{ tree_node::{TreeNodeRecursion, TreeNodeVisitor}, }; pub use datafusion_execution::TaskContext; -use datafusion_execution::cache::cache_manager::{ - DEFAULT_FILE_STATISTICS_MEMORY_LIMIT, DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT, - DEFAULT_LIST_FILES_CACHE_TTL, DEFAULT_METADATA_CACHE_LIMIT, -}; pub use datafusion_execution::config::SessionConfig; -use datafusion_execution::disk_manager::{ - DEFAULT_MAX_SPILL_MERGE_FAN_IN, DEFAULT_MAX_TEMP_DIRECTORY_SIZE, DiskManagerBuilder, -}; use datafusion_execution::registry::SerializerRegistry; +use datafusion_execution::runtime_options::{self, RuntimeOptions}; use datafusion_expr::HigherOrderUDF; pub use datafusion_expr::execution_props::ExecutionProps; #[cfg(feature = "sql")] @@ -1172,50 +1165,9 @@ impl SessionContext { let mut state = self.state.write(); - let mut builder = RuntimeEnvBuilder::from_runtime_env(state.runtime_env()); - builder = match key { - "memory_limit" => { - let memory_limit = Self::parse_capacity_limit(variable, value)?; - builder.with_memory_limit(memory_limit, 1.0) - } - "max_temp_directory_size" => { - let directory_size = Self::parse_capacity_limit(variable, value)?; - builder.with_max_temp_directory_size(directory_size as u64) - } - "temp_directory" => builder.with_temp_file_path(value), - "metadata_cache_limit" => { - let limit = Self::parse_capacity_limit(variable, value)?; - builder.with_metadata_cache_limit(limit) - } - "list_files_cache_limit" => { - let limit = Self::parse_capacity_limit(variable, value)?; - builder.with_object_list_cache_limit(limit) - } - "list_files_cache_ttl" => { - let duration = Self::parse_duration(variable, value)?; - builder.with_object_list_cache_ttl(Some(duration)) - } - "file_statistics_cache_limit" => { - let limit = Self::parse_capacity_limit(variable, value)?; - builder.with_file_statistics_cache_limit(limit) - } - "max_spill_merge_fan_in" => { - let fan_in = value.parse::().map_err(|e| { - DataFusionError::Plan(format!( - "Failed to parse non-negative integer from '{variable}', value '{value}': {e}" - )) - })?; - builder.with_max_spill_merge_fan_in(fan_in) - } - _ => return plan_err!("Unknown runtime configuration: {variable}"), - // Remember to update `reset_runtime_variable()` when adding new options - }; - - *state = SessionStateBuilder::from(state.clone()) - .with_runtime_env(Arc::new(builder.build()?)) - .build(); - - Ok(()) + let mut options = RuntimeOptions::from_runtime_env(state.runtime_env()); + options.set_entry(key, value)?; + self.apply_runtime_options(&mut state, &options, key) } fn reset_runtime_variable(&self, variable: &str) -> Result<()> { @@ -1223,40 +1175,27 @@ impl SessionContext { let mut state = self.state.write(); - let mut builder = RuntimeEnvBuilder::from_runtime_env(state.runtime_env()); - match key { - "memory_limit" => { - builder.memory_pool = None; - } - "max_temp_directory_size" => { - builder = - builder.with_max_temp_directory_size(DEFAULT_MAX_TEMP_DIRECTORY_SIZE); - } - "temp_directory" => { - builder.disk_manager_builder = Some(DiskManagerBuilder::default()); - } - "metadata_cache_limit" => { - builder = builder.with_metadata_cache_limit(DEFAULT_METADATA_CACHE_LIMIT); - } - "list_files_cache_limit" => { - builder = builder - .with_object_list_cache_limit(DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT); - } - "list_files_cache_ttl" => { - builder = - builder.with_object_list_cache_ttl(DEFAULT_LIST_FILES_CACHE_TTL); - } - "file_statistics_cache_limit" => { - builder = builder.with_file_statistics_cache_limit( - DEFAULT_FILE_STATISTICS_MEMORY_LIMIT, - ); - } - "max_spill_merge_fan_in" => { - builder = - builder.with_max_spill_merge_fan_in(DEFAULT_MAX_SPILL_MERGE_FAN_IN); - } - _ => return plan_err!("Unknown runtime configuration: {variable}"), - } + let mut options = RuntimeOptions::from_runtime_env(state.runtime_env()); + options.reset_entry(key)?; + self.apply_runtime_options(&mut state, &options, key) + } + + /// Rebuild the session's [`RuntimeEnv`] with `key` taken from `options`. + /// + /// Only the named key is applied. Rebuilding every resource on each + /// statement would replace the live memory pool and discard the + /// reservations held against it. + fn apply_runtime_options( + &self, + state: &mut SessionState, + options: &RuntimeOptions, + key: &str, + ) -> Result<()> { + let builder = options.apply_key( + key, + RuntimeEnvBuilder::from_runtime_env(state.runtime_env()), + )?; + *state = SessionStateBuilder::from(state.clone()) .with_runtime_env(Arc::new(builder.build()?)) .build(); @@ -1327,104 +1266,7 @@ impl SessionContext { /// ); /// ``` pub fn parse_capacity_limit(config_name: &str, limit: &str) -> Result { - if limit.trim().is_empty() { - return Err(plan_datafusion_err!( - "Empty limit value found for '{config_name}'" - )); - } - if limit == "0" { - return Ok(0); - } - let (unit_start, unit) = limit.char_indices().next_back().ok_or_else(|| { - plan_datafusion_err!("Empty limit value found for '{config_name}'") - })?; - let number = &limit[..unit_start]; - let number: f64 = number.parse().map_err(|_| { - plan_datafusion_err!( - "Failed to parse number from '{config_name}', limit '{limit}'" - ) - })?; - if number.is_sign_negative() || number.is_infinite() { - return Err(plan_datafusion_err!( - "Limit value should be positive finite number for '{config_name}'" - )); - } - - match unit { - 'K' => Ok((number * 1024.0) as usize), - 'M' => Ok((number * 1024.0 * 1024.0) as usize), - 'G' => Ok((number * 1024.0 * 1024.0 * 1024.0) as usize), - _ => plan_err!( - "Unsupported unit '{unit}' in '{config_name}', limit '{limit}'. \ - Unit must be one of: 'K', 'M', 'G'" - ), - } - } - - fn parse_duration(config_name: &str, duration: &str) -> Result { - if duration.trim().is_empty() { - return Err(plan_datafusion_err!( - "Duration should not be empty or blank for '{config_name}'" - )); - } - - let mut minutes = None; - let mut seconds = None; - - for duration in duration.split_inclusive(&['m', 's']) { - let (unit_start, unit) = - duration.char_indices().next_back().ok_or_else(|| { - plan_datafusion_err!( - "Duration should not be empty or blank for '{config_name}'" - ) - })?; - let number = &duration[..unit_start]; - let number: u64 = number.parse().map_err(|_| { - plan_datafusion_err!("Failed to parse number from duration '{duration}' for '{config_name}'") - })?; - - match unit { - 'm' if minutes.is_none() && seconds.is_none() => minutes = Some(number), - 's' if seconds.is_none() => seconds = Some(number), - other => plan_err!( - "Invalid duration unit: '{other}'. The unit must be either 'm' (minutes), or 's' (seconds), and be in the correct order for '{config_name}'" - )?, - } - } - - let secs = Self::check_overflow(config_name, minutes, 60, seconds)?; - let duration = Duration::from_secs(secs); - - if duration.is_zero() { - return plan_err!( - "Duration must be greater than 0 seconds for '{config_name}'" - ); - } - - Ok(duration) - } - - fn check_overflow( - config_name: &str, - mins: Option, - multiplier: u64, - secs: Option, - ) -> Result { - let first_part_of_secs = mins.unwrap_or_default().checked_mul(multiplier); - if first_part_of_secs.is_none() { - plan_err!( - "Duration has overflowed allowed maximum limit due to 'mins * {multiplier}' when setting '{config_name}'" - )? - } - let second_part_of_secs = first_part_of_secs - .unwrap() - .checked_add(secs.unwrap_or_default()); - if second_part_of_secs.is_none() { - plan_err!( - "Duration has overflowed allowed maximum limit due to 'mins * {multiplier} + secs' when setting '{config_name}'" - )? - } - Ok(second_part_of_secs.unwrap()) + runtime_options::parse_capacity_limit(config_name, limit) } async fn create_custom_table( @@ -2368,6 +2210,7 @@ mod tests { use datafusion_expr::physical_planning_context::PhysicalPlanningContext; use std::error::Error; use std::path::PathBuf; + use std::time::Duration; use datafusion_common::test_util::batches_to_string; use datafusion_common_runtime::SpawnedTask; @@ -2960,7 +2803,7 @@ mod tests { ("1m1s", Duration::from_secs(61)), ] { let have = - SessionContext::parse_duration(LIST_FILES_CACHE_TTL, duration).unwrap(); + runtime_options::parse_duration(LIST_FILES_CACHE_TTL, duration).unwrap(); assert_eq!(want, have); } @@ -2969,7 +2812,7 @@ mod tests { "0s", "0m", "1s0m", "1s1m", "XYZ", "1h", "XYZm2s", "", " ", "-1m", "1m 1s", "1m1s ", " 1m1s", "1\u{b5}", ] { - let have = SessionContext::parse_duration(LIST_FILES_CACHE_TTL, duration); + let have = runtime_options::parse_duration(LIST_FILES_CACHE_TTL, duration); assert!(have.is_err()); assert!( have.unwrap_err() @@ -3008,7 +2851,7 @@ mod tests { ), ] { let have = - SessionContext::parse_duration(LIST_FILES_CACHE_TTL, duration).unwrap(); + runtime_options::parse_duration(LIST_FILES_CACHE_TTL, duration).unwrap(); assert_eq!(want, have); } @@ -3031,7 +2874,7 @@ mod tests { "Duration has overflowed allowed maximum limit due to", ), ] { - let have = SessionContext::parse_duration(LIST_FILES_CACHE_TTL, duration); + let have = runtime_options::parse_duration(LIST_FILES_CACHE_TTL, duration); assert!(have.is_err()); let error_message = have.unwrap_err().message().to_string(); assert!( diff --git a/datafusion/core/tests/sql/runtime_config.rs b/datafusion/core/tests/sql/runtime_config.rs index 1275edc6d8b0c..c0f93ae136290 100644 --- a/datafusion/core/tests/sql/runtime_config.rs +++ b/datafusion/core/tests/sql/runtime_config.rs @@ -158,9 +158,10 @@ async fn test_invalid_memory_limit_when_unit_is_invalid() { assert!(result.is_err()); let error_message = result.unwrap_err().to_string(); assert!( - error_message - .contains("Unsupported unit 'X' in 'datafusion.runtime.memory_limit'") + error_message.contains("Unsupported unit 'X' in limit '100X'") && error_message.contains("Unit must be one of: 'K', 'M', 'G'") + && error_message.contains("when setting 'datafusion.runtime.memory_limit'"), + "{error_message}" ); } @@ -174,9 +175,12 @@ async fn test_invalid_memory_limit_when_limit_is_not_numeric() { assert!(result.is_err()); let error_message = result.unwrap_err().to_string(); - assert!(error_message.contains( - "Failed to parse number from 'datafusion.runtime.memory_limit', limit 'invalid_memory_limit'" - )); + assert!( + error_message + .contains("Failed to parse number from limit 'invalid_memory_limit'") + && error_message.contains("when setting 'datafusion.runtime.memory_limit'"), + "{error_message}" + ); } #[tokio::test] diff --git a/datafusion/execution/src/lib.rs b/datafusion/execution/src/lib.rs index 5af7064f1cb8b..c96cca77c61c2 100644 --- a/datafusion/execution/src/lib.rs +++ b/datafusion/execution/src/lib.rs @@ -36,6 +36,7 @@ pub mod object_store; #[cfg(feature = "parquet_encryption")] pub mod parquet_encryption; pub mod runtime_env; +pub mod runtime_options; pub mod spill_file; mod stream; mod task; diff --git a/datafusion/execution/src/runtime_env.rs b/datafusion/execution/src/runtime_env.rs index fcfe51267e65f..0f3c2b71e411d 100644 --- a/datafusion/execution/src/runtime_env.rs +++ b/datafusion/execution/src/runtime_env.rs @@ -28,8 +28,10 @@ use crate::{ }; use crate::cache::cache_manager::{CacheManager, CacheManagerConfig}; +use crate::memory_pool::MemoryLimit; #[cfg(feature = "parquet_encryption")] use crate::parquet_encryption::{EncryptionFactory, EncryptionFactoryRegistry}; +use crate::runtime_options::{RUNTIME_CONFIG_PREFIX, RuntimeOptions}; use datafusion_common::{Result, config::ConfigEntry}; use object_store::ObjectStore; use std::sync::Arc; @@ -90,79 +92,6 @@ impl Debug for RuntimeEnv { } } -struct RuntimeConfigValues { - memory_limit: Option, - max_temp_directory_size: Option, - max_spill_merge_fan_in: Option, - temp_directory: Option, - metadata_cache_limit: Option, - list_files_cache_limit: Option, - list_files_cache_ttl: Option, - file_statistics_cache_limit: Option, -} - -impl RuntimeConfigValues { - /// Creates runtime configuration entries with the provided values. - /// - /// This defines the structure and metadata for all runtime configuration - /// entries to avoid duplication between `RuntimeEnv::config_entries()` and - /// `RuntimeEnvBuilder::entries()`. - fn into_config_entries(self) -> Vec { - let Self { - memory_limit, - max_temp_directory_size, - max_spill_merge_fan_in, - temp_directory, - metadata_cache_limit, - list_files_cache_limit, - list_files_cache_ttl, - file_statistics_cache_limit, - } = self; - vec![ - ConfigEntry { - key: "datafusion.runtime.memory_limit".to_string(), - value: memory_limit, - description: "Maximum memory limit for query execution. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes.", - }, - ConfigEntry { - key: "datafusion.runtime.max_temp_directory_size".to_string(), - value: max_temp_directory_size, - description: "Maximum temporary file directory size. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes.", - }, - ConfigEntry { - key: "datafusion.runtime.max_spill_merge_fan_in".to_string(), - value: max_spill_merge_fan_in, - description: "Maximum number of spill files opened by one external merge pass. Use 0 for unlimited. Values below 2 still use 2 so a merge can make progress.", - }, - ConfigEntry { - key: "datafusion.runtime.temp_directory".to_string(), - value: temp_directory, - description: "The path to the temporary file directory.", - }, - ConfigEntry { - key: "datafusion.runtime.metadata_cache_limit".to_string(), - value: metadata_cache_limit, - description: "Maximum memory to use for file metadata cache such as Parquet metadata. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes.", - }, - ConfigEntry { - key: "datafusion.runtime.list_files_cache_limit".to_string(), - value: list_files_cache_limit, - description: "Maximum memory to use for list files cache. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes.", - }, - ConfigEntry { - key: "datafusion.runtime.list_files_cache_ttl".to_string(), - value: list_files_cache_ttl, - description: "TTL (time-to-live) of the entries in the list file cache. Supports units m (minutes), and s (seconds). Example: '2m' for 2 minutes.", - }, - ConfigEntry { - key: "datafusion.runtime.file_statistics_cache_limit".to_string(), - value: file_statistics_cache_limit, - description: "Maximum memory to use for file statistics cache. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes.", - }, - ] - } -} - impl RuntimeEnv { /// Registers a custom `ObjectStore` to be used with a specific url. /// This allows DataFusion to create external tables from urls that do not have @@ -252,96 +181,23 @@ impl RuntimeEnv { } /// Returns the current runtime configuration entries + /// + /// Keys, descriptions and ordering come from [`RuntimeOptions`]; the values + /// are read back from the live resource objects, which is why this is not + /// simply `RuntimeOptions::default().entries()`. pub fn config_entries(&self) -> Vec { - use crate::memory_pool::MemoryLimit; - - /// Convert bytes to a human-readable format - fn format_byte_size(size: u64) -> String { - const GB: u64 = 1024 * 1024 * 1024; - const MB: u64 = 1024 * 1024; - const KB: u64 = 1024; - - match size { - s if s >= GB => format!("{}G", s / GB), - s if s >= MB => format!("{}M", s / MB), - s if s >= KB => format!("{}K", s / KB), - s => format!("{s}"), + let mut entries = RuntimeOptions::from_runtime_env(self).entries(); + + // An unbounded pool has no byte value to report, but it is not unset + // either, so `RuntimeOptions` cannot carry it. + if matches!(self.memory_pool.memory_limit(), MemoryLimit::Infinite) { + let key = format!("{RUNTIME_CONFIG_PREFIX}.memory_limit"); + if let Some(entry) = entries.iter_mut().find(|e| e.key == key) { + entry.value = Some("unlimited".to_string()); } } - fn format_duration(duration: Duration) -> String { - let total = duration.as_secs(); - let mins = total / 60; - let secs = total % 60; - - format!("{mins}m{secs}s") - } - - let memory_limit_value = match self.memory_pool.memory_limit() { - MemoryLimit::Finite(size) => Some(format_byte_size( - size.try_into() - .expect("Memory limit size conversion failed"), - )), - MemoryLimit::Infinite => Some("unlimited".to_string()), - MemoryLimit::Unknown => None, - }; - - let max_temp_dir_size = self.disk_manager.max_temp_directory_size(); - let max_temp_dir_value = format_byte_size(max_temp_dir_size); - let max_spill_merge_fan_in = - self.disk_manager.max_spill_merge_fan_in().to_string(); - - let temp_paths = self.disk_manager.temp_dir_paths(); - let temp_dir_value = if temp_paths.is_empty() { - None - } else { - Some( - temp_paths - .iter() - .map(|p| p.display().to_string()) - .collect::>() - .join(","), - ) - }; - - let metadata_cache_limit = self.cache_manager.get_metadata_cache_limit(); - let metadata_cache_value = format_byte_size( - metadata_cache_limit - .try_into() - .expect("Metadata cache size conversion failed"), - ); - - let list_files_cache_limit = self.cache_manager.get_list_files_cache_limit(); - let list_files_cache_value = format_byte_size( - list_files_cache_limit - .try_into() - .expect("List files cache size conversion failed"), - ); - - let list_files_cache_ttl = self - .cache_manager - .get_list_files_cache_ttl() - .map(format_duration); - - let file_statistics_cache_limit = - self.cache_manager.get_file_statistic_cache_limit(); - let file_statistics_cache_value = format_byte_size( - file_statistics_cache_limit - .try_into() - .expect("File statistics cache size conversion failed"), - ); - - RuntimeConfigValues { - memory_limit: memory_limit_value, - max_temp_directory_size: Some(max_temp_dir_value), - max_spill_merge_fan_in: Some(max_spill_merge_fan_in), - temp_directory: temp_dir_value, - metadata_cache_limit: Some(metadata_cache_value), - list_files_cache_limit: Some(list_files_cache_value), - list_files_cache_ttl, - file_statistics_cache_limit: Some(file_statistics_cache_value), - } - .into_config_entries() + entries } } @@ -546,19 +402,9 @@ impl RuntimeEnvBuilder { } } - /// Returns a list of all available runtime configurations with their current values and descriptions + /// Returns a list of all available runtime configurations with their default values and descriptions pub fn entries(&self) -> Vec { - RuntimeConfigValues { - memory_limit: None, - max_temp_directory_size: Some("100G".to_string()), - max_spill_merge_fan_in: Some("0".to_string()), - temp_directory: None, - metadata_cache_limit: Some("50M".to_owned()), - list_files_cache_limit: Some("1M".to_owned()), - list_files_cache_ttl: None, - file_statistics_cache_limit: Some("20M".to_owned()), - } - .into_config_entries() + RuntimeOptions::default().entries() } /// Generate documentation that can be included in the user guide diff --git a/datafusion/execution/src/runtime_options.rs b/datafusion/execution/src/runtime_options.rs new file mode 100644 index 0000000000000..e86229086cc9c --- /dev/null +++ b/datafusion/execution/src/runtime_options.rs @@ -0,0 +1,499 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Typed schema for the `datafusion.runtime.*` configuration namespace. +//! +//! [`RuntimeOptions`] is the single source of truth for the runtime +//! configuration keys, their descriptions, their defaults and how their string +//! values are parsed. It is built with the same [`ConfigField`] machinery that +//! backs [`ConfigOptions`], so `SET`/`RESET` handling no longer needs a +//! hand-written `match` arm per key. +//! +//! [`RuntimeOptions`] deliberately does *not* live inside [`ConfigOptions`]: +//! `datafusion-execution` depends on `datafusion-common`, never the reverse. +//! +//! [`ConfigOptions`]: datafusion_common::config::ConfigOptions + +use std::fmt::{self, Display, Formatter}; +use std::time::Duration; + +use crate::cache::cache_manager::{ + DEFAULT_FILE_STATISTICS_MEMORY_LIMIT, DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT, + DEFAULT_LIST_FILES_CACHE_TTL, DEFAULT_METADATA_CACHE_LIMIT, +}; +use crate::disk_manager::{ + DEFAULT_MAX_SPILL_MERGE_FAN_IN, DEFAULT_MAX_TEMP_DIRECTORY_SIZE, DiskManagerBuilder, +}; +use crate::memory_pool::MemoryLimit; +use crate::runtime_env::{RuntimeEnv, RuntimeEnvBuilder}; + +use datafusion_common::config::{ConfigEntry, ConfigField, Visit}; +use datafusion_common::{ + DataFusionError, Result, config_field, config_namespace, plan_datafusion_err, + plan_err, +}; + +/// Prefix shared by every key in this namespace. +pub const RUNTIME_CONFIG_PREFIX: &str = "datafusion.runtime"; + +/// A byte capacity written as a plain number of bytes (`0`) or a number with a +/// `K`, `M` or `G` suffix (`512K`, `100M`, `1.5G`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct CapacityLimit(pub usize); + +impl Display for CapacityLimit { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + const GB: usize = 1024 * 1024 * 1024; + const MB: usize = 1024 * 1024; + const KB: usize = 1024; + + match self.0 { + s if s >= GB => write!(f, "{}G", s / GB), + s if s >= MB => write!(f, "{}M", s / MB), + s if s >= KB => write!(f, "{}K", s / KB), + s => write!(f, "{s}"), + } + } +} + +config_field!(CapacityLimit, value => CapacityLimit(parse_capacity(value)?)); + +/// A cache time-to-live written as minutes and/or seconds (`90s`, `2m`, `1m30s`). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct CacheTtl(pub Duration); + +impl Display for CacheTtl { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + let total = self.0.as_secs(); + write!(f, "{}m{}s", total / 60, total % 60) + } +} + +config_field!(CacheTtl, value => CacheTtl(parse_ttl(value)?)); + +/// A count that must not be negative. +/// +/// A plain `usize` would parse through [`default_config_transform`], whose +/// message names the Rust type rather than the constraint. +/// +/// [`default_config_transform`]: datafusion_common::config::default_config_transform +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub struct NonNegativeCount(pub usize); + +impl Display for NonNegativeCount { + fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +config_field!(NonNegativeCount, value => NonNegativeCount( + value.parse::().map_err(|e| plan_datafusion_err!( + "Failed to parse non-negative integer from value '{value}': {e}" + ))? +)); + +config_namespace! { + /// Options that configure the [`RuntimeEnv`] shared by a session. + /// + /// Field order below is the order entries are reported in. + pub struct RuntimeOptions { + /// Maximum memory limit for query execution. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes. + pub memory_limit: Option, default = None + + /// Maximum temporary file directory size. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes. + pub max_temp_directory_size: CapacityLimit, default = CapacityLimit(DEFAULT_MAX_TEMP_DIRECTORY_SIZE as usize) + + /// Maximum number of spill files opened by one external merge pass. Use 0 for unlimited. Values below 2 still use 2 so a merge can make progress. + pub max_spill_merge_fan_in: NonNegativeCount, default = NonNegativeCount(DEFAULT_MAX_SPILL_MERGE_FAN_IN) + + /// The path to the temporary file directory. + pub temp_directory: Option, default = None + + /// Maximum memory to use for file metadata cache such as Parquet metadata. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes. + pub metadata_cache_limit: CapacityLimit, default = CapacityLimit(DEFAULT_METADATA_CACHE_LIMIT) + + /// Maximum memory to use for list files cache. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes. + pub list_files_cache_limit: CapacityLimit, default = CapacityLimit(DEFAULT_LIST_FILES_CACHE_MEMORY_LIMIT) + + /// TTL (time-to-live) of the entries in the list file cache. Supports units m (minutes), and s (seconds). Example: '2m' for 2 minutes. + pub list_files_cache_ttl: Option, default = DEFAULT_LIST_FILES_CACHE_TTL.map(CacheTtl) + + /// Maximum memory to use for file statistics cache. Supports suffixes K (kilobytes), M (megabytes), and G (gigabytes) or '0' for 0. Example: '2G' for 2 gigabytes. + pub file_statistics_cache_limit: CapacityLimit, default = CapacityLimit(DEFAULT_FILE_STATISTICS_MEMORY_LIMIT) + } +} + +impl RuntimeOptions { + /// Set `key` (without the `datafusion.runtime.` prefix) from its string form. + pub fn set_entry(&mut self, key: &str, value: &str) -> Result<()> { + ConfigField::set(self, key, value).map_err(|e| qualify(e, key)) + } + + /// Restore `key` (without the `datafusion.runtime.` prefix) to its default. + pub fn reset_entry(&mut self, key: &str) -> Result<()> { + ConfigField::reset(self, key).map_err(|e| qualify(e, key)) + } + + /// Every key in this namespace with its current value and description. + pub fn entries(&self) -> Vec { + struct Visitor(Vec); + + impl Visit for Visitor { + fn some( + &mut self, + key: &str, + value: V, + description: &'static str, + ) { + self.0.push(ConfigEntry { + key: key.to_string(), + value: Some(value.to_string()), + description, + }) + } + + fn none(&mut self, key: &str, description: &'static str) { + self.0.push(ConfigEntry { + key: key.to_string(), + value: None, + description, + }) + } + } + + let mut v = Visitor(vec![]); + self.visit(&mut v, RUNTIME_CONFIG_PREFIX, ""); + v.0 + } + + /// Read the values currently in effect from the live resource objects. + /// + /// A [`MemoryLimit::Infinite`] pool has no representation here; callers that + /// report values handle it separately. + pub fn from_runtime_env(env: &RuntimeEnv) -> Self { + let memory_limit = match env.memory_pool.memory_limit() { + MemoryLimit::Finite(size) => Some(CapacityLimit(size)), + MemoryLimit::Infinite | MemoryLimit::Unknown => None, + }; + + let temp_paths = env.disk_manager.temp_dir_paths(); + let temp_directory = (!temp_paths.is_empty()).then(|| { + temp_paths + .iter() + .map(|p| p.display().to_string()) + .collect::>() + .join(",") + }); + + Self { + memory_limit, + max_temp_directory_size: CapacityLimit( + env.disk_manager.max_temp_directory_size() as usize, + ), + max_spill_merge_fan_in: NonNegativeCount( + env.disk_manager.max_spill_merge_fan_in(), + ), + temp_directory, + metadata_cache_limit: CapacityLimit( + env.cache_manager.get_metadata_cache_limit(), + ), + list_files_cache_limit: CapacityLimit( + env.cache_manager.get_list_files_cache_limit(), + ), + list_files_cache_ttl: env + .cache_manager + .get_list_files_cache_ttl() + .map(CacheTtl), + file_statistics_cache_limit: CapacityLimit( + env.cache_manager.get_file_statistic_cache_limit(), + ), + } + } + + /// Write the value of a single `key` onto `builder`. + /// + /// Only the named key is applied: rebuilding every resource on each `SET` + /// would discard the live memory pool and its outstanding reservations. + pub fn apply_key( + &self, + key: &str, + mut builder: RuntimeEnvBuilder, + ) -> Result { + builder = match key { + "memory_limit" => match self.memory_limit { + Some(limit) => builder.with_memory_limit(limit.0, 1.0), + None => { + builder.memory_pool = None; + builder + } + }, + "max_temp_directory_size" => builder + .with_max_temp_directory_size(self.max_temp_directory_size.0 as u64), + "max_spill_merge_fan_in" => { + builder.with_max_spill_merge_fan_in(self.max_spill_merge_fan_in.0) + } + "temp_directory" => match &self.temp_directory { + Some(path) => builder.with_temp_file_path(path), + None => { + builder.disk_manager_builder = Some(DiskManagerBuilder::default()); + builder + } + }, + "metadata_cache_limit" => { + builder.with_metadata_cache_limit(self.metadata_cache_limit.0) + } + "list_files_cache_limit" => { + builder.with_object_list_cache_limit(self.list_files_cache_limit.0) + } + "list_files_cache_ttl" => builder + .with_object_list_cache_ttl(self.list_files_cache_ttl.map(|ttl| ttl.0)), + "file_statistics_cache_limit" => builder + .with_file_statistics_cache_limit(self.file_statistics_cache_limit.0), + _ => return Err(unknown_key(key)), + }; + Ok(builder) + } +} + +/// `config_namespace!` reports an unknown key as a [`DataFusionError::Configuration`]; +/// parse failures below are [`DataFusionError::Plan`]. Restate the former in this +/// namespace's wording, and name the key on the latter. +fn qualify(e: DataFusionError, key: &str) -> DataFusionError { + match e { + DataFusionError::Configuration(_) => unknown_key(key), + other => name_config(other, &format!("{RUNTIME_CONFIG_PREFIX}.{key}")), + } +} + +/// The parsers below take only the value, because [`ConfigField::set`] is not +/// given the key it is setting. Callers that know the key append it here. +fn name_config(e: DataFusionError, config_name: &str) -> DataFusionError { + match e { + DataFusionError::Plan(msg) => { + plan_datafusion_err!("{msg} when setting '{config_name}'") + } + other => other, + } +} + +fn unknown_key(key: &str) -> DataFusionError { + plan_datafusion_err!("Unknown runtime configuration: {RUNTIME_CONFIG_PREFIX}.{key}") +} + +/// [`parse_capacity`], with `config_name` named in any error. +pub fn parse_capacity_limit(config_name: &str, limit: &str) -> Result { + parse_capacity(limit).map_err(|e| name_config(e, config_name)) +} + +/// [`parse_ttl`], with `config_name` named in any error. +pub fn parse_duration(config_name: &str, duration: &str) -> Result { + parse_ttl(duration).map_err(|e| name_config(e, config_name)) +} + +/// Parse a byte capacity: `0`, or a number with a `K`, `M` or `G` suffix. +pub fn parse_capacity(limit: &str) -> Result { + if limit.trim().is_empty() { + return plan_err!("Empty limit value found"); + } + if limit == "0" { + return Ok(0); + } + let (unit_start, unit) = limit + .char_indices() + .next_back() + .ok_or_else(|| plan_datafusion_err!("Empty limit value found"))?; + let number: f64 = limit[..unit_start].parse().map_err(|_| { + plan_datafusion_err!("Failed to parse number from limit '{limit}'") + })?; + if number.is_sign_negative() || number.is_infinite() { + return plan_err!("Limit value should be positive finite number"); + } + + match unit { + 'K' => Ok((number * 1024.0) as usize), + 'M' => Ok((number * 1024.0 * 1024.0) as usize), + 'G' => Ok((number * 1024.0 * 1024.0 * 1024.0) as usize), + _ => plan_err!( + "Unsupported unit '{unit}' in limit '{limit}'. \ + Unit must be one of: 'K', 'M', 'G'" + ), + } +} + +/// Parse a cache TTL: minutes and/or seconds, in that order (`90s`, `2m`, `1m30s`). +pub fn parse_ttl(ttl: &str) -> Result { + if ttl.trim().is_empty() { + return plan_err!("Duration should not be empty or blank"); + } + + let mut minutes = None; + let mut seconds = None; + + for part in ttl.split_inclusive(&['m', 's']) { + let (unit_start, unit) = part.char_indices().next_back().ok_or_else(|| { + plan_datafusion_err!("Duration should not be empty or blank") + })?; + let number: u64 = part[..unit_start].parse().map_err(|_| { + plan_datafusion_err!("Failed to parse number from duration '{part}'") + })?; + + match unit { + 'm' if minutes.is_none() && seconds.is_none() => minutes = Some(number), + 's' if seconds.is_none() => seconds = Some(number), + other => plan_err!( + "Invalid duration unit: '{other}'. The unit must be either 'm' (minutes), \ + or 's' (seconds), and be in the correct order" + )?, + } + } + + let secs = checked_secs(minutes, seconds)?; + let duration = Duration::from_secs(secs); + if duration.is_zero() { + return plan_err!("Duration must be greater than 0 seconds"); + } + Ok(duration) +} + +fn checked_secs(mins: Option, secs: Option) -> Result { + mins.unwrap_or_default() + .checked_mul(60) + .ok_or_else(|| { + plan_datafusion_err!( + "Duration has overflowed allowed maximum limit due to 'mins * 60'" + ) + })? + .checked_add(secs.unwrap_or_default()) + .ok_or_else(|| { + plan_datafusion_err!( + "Duration has overflowed allowed maximum limit due to 'mins * 60 + secs'" + ) + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn defaults_match_the_runtime_constants() { + let entries = RuntimeOptions::default().entries(); + let value = |key: &str| { + entries + .iter() + .find(|e| e.key == format!("{RUNTIME_CONFIG_PREFIX}.{key}")) + .unwrap_or_else(|| panic!("missing {key}")) + .value + .clone() + }; + + assert_eq!(value("memory_limit"), None); + assert_eq!(value("max_temp_directory_size"), Some("100G".to_string())); + assert_eq!(value("max_spill_merge_fan_in"), Some("0".to_string())); + assert_eq!(value("temp_directory"), None); + assert_eq!(value("metadata_cache_limit"), Some("50M".to_string())); + assert_eq!(value("list_files_cache_limit"), Some("1M".to_string())); + assert_eq!(value("list_files_cache_ttl"), None); + assert_eq!( + value("file_statistics_cache_limit"), + Some("20M".to_string()) + ); + } + + /// `config_entries` reports what is in effect, which is not the same as + /// `RuntimeOptions::default()`: an unbounded pool reports `unlimited`, and a + /// bounded one reports its size. + #[test] + fn config_entries_report_the_live_memory_pool() { + let memory_limit = |env: &RuntimeEnv| { + env.config_entries() + .into_iter() + .find(|e| e.key == format!("{RUNTIME_CONFIG_PREFIX}.memory_limit")) + .expect("missing memory_limit") + .value + }; + + assert_eq!( + memory_limit(&RuntimeEnv::default()), + Some("unlimited".to_string()) + ); + + let bounded = RuntimeEnvBuilder::new() + .with_memory_limit(2 * 1024 * 1024 * 1024, 1.0) + .build() + .unwrap(); + assert_eq!(memory_limit(&bounded), Some("2G".to_string())); + } + + #[test] + fn set_then_reset_round_trips() { + let mut options = RuntimeOptions::default(); + options.set_entry("memory_limit", "1.5G").unwrap(); + assert_eq!( + options.memory_limit, + Some(CapacityLimit((1.5 * 1024.0 * 1024.0 * 1024.0) as usize)) + ); + options.reset_entry("memory_limit").unwrap(); + assert_eq!(options.memory_limit, None); + } + + #[test] + fn unknown_key_is_a_plan_error() { + let err = RuntimeOptions::default() + .set_entry("nope", "1") + .unwrap_err() + .to_string(); + assert!( + err.contains("Unknown runtime configuration: datafusion.runtime.nope"), + "{err}" + ); + } + + #[test] + fn parse_errors_name_the_key() { + let err = RuntimeOptions::default() + .set_entry("memory_limit", "10X") + .unwrap_err() + .to_string(); + assert!(err.contains("Unsupported unit 'X' in limit '10X'"), "{err}"); + assert!( + err.contains("when setting 'datafusion.runtime.memory_limit'"), + "{err}" + ); + } + + #[test] + fn non_ascii_value_does_not_panic() { + // Regression guard for the panic fixed in apache/datafusion#23316. + assert!( + RuntimeOptions::default() + .set_entry("memory_limit", "1️⃣") + .is_err() + ); + } + + #[test] + fn ttl_round_trips_through_display() { + let mut options = RuntimeOptions::default(); + options.set_entry("list_files_cache_ttl", "1m30s").unwrap(); + assert_eq!( + options.list_files_cache_ttl, + Some(CacheTtl(Duration::from_secs(90))) + ); + assert_eq!(options.list_files_cache_ttl.unwrap().to_string(), "1m30s"); + } +} diff --git a/datafusion/sqllogictest/test_files/set_variable.slt b/datafusion/sqllogictest/test_files/set_variable.slt index e36e59bccb66b..4068f554630ee 100644 --- a/datafusion/sqllogictest/test_files/set_variable.slt +++ b/datafusion/sqllogictest/test_files/set_variable.slt @@ -704,13 +704,13 @@ SET datafusion.runtime.memory_limit = NULL statement error DataFusion error: Error during planning: Unsupported value Null SET datafusion.runtime.list_files_cache_ttl = NULL -statement error DataFusion error: Error during planning: Duration should not be empty or blank for 'datafusion.runtime.list_files_cache_ttl' +statement error DataFusion error: Error during planning: Duration should not be empty or blank when setting 'datafusion.runtime.list_files_cache_ttl' SET datafusion.runtime.list_files_cache_ttl = ' ' statement ok SET datafusion.runtime.list_files_cache_ttl = '18446744073709551615s' -statement error DataFusion error: Error during planning: Failed to parse number from duration '18446744073709551616s' for 'datafusion.runtime.list_files_cache_ttl' +statement error DataFusion error: Error during planning: Failed to parse number from duration '18446744073709551616s' when setting 'datafusion.runtime.list_files_cache_ttl' SET datafusion.runtime.list_files_cache_ttl = '18446744073709551616s' statement ok From eb150c41e351f8dd850119dc3f8799b6438934df Mon Sep 17 00:00:00 2001 From: Hao Chun Chang Date: Wed, 9 Sep 2026 19:18:44 +0800 Subject: [PATCH 2/2] refactor: simplify the runtime options plumbing Three cuts, no behaviour change: - `SET`/`RESET` built a `RuntimeOptions` from the live `RuntimeEnv` before writing one key into it. Since only that one key is ever applied, the read was wasted; both paths now start from `RuntimeOptions::default()`. - `RuntimeEnv::config_entries` knew the namespace prefix and the `unlimited` special case. That moves into `RuntimeOptions::env_entries`, so all key knowledge now lives in `runtime_options.rs` and `config_entries` is a one-line delegate. `from_runtime_env` becomes private. - Trimmed the module and type docs. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01TpQVFSJ6v2yfMNqwbsfsRe --- datafusion/core/src/execution/context/mod.rs | 25 +++++------- datafusion/execution/src/runtime_env.rs | 20 +--------- datafusion/execution/src/runtime_options.rs | 41 ++++++++++++-------- 3 files changed, 37 insertions(+), 49 deletions(-) diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs index 596c3677a5b63..6f60eeefb15ba 100644 --- a/datafusion/core/src/execution/context/mod.rs +++ b/datafusion/core/src/execution/context/mod.rs @@ -1163,34 +1163,29 @@ impl SessionContext { fn set_runtime_variable(&self, variable: &str, value: &str) -> Result<()> { let key = variable.strip_prefix("datafusion.runtime.").unwrap(); - let mut state = self.state.write(); - - let mut options = RuntimeOptions::from_runtime_env(state.runtime_env()); + let mut options = RuntimeOptions::default(); options.set_entry(key, value)?; - self.apply_runtime_options(&mut state, &options, key) + self.apply_runtime_option(&options, key) } fn reset_runtime_variable(&self, variable: &str) -> Result<()> { let key = variable.strip_prefix("datafusion.runtime.").unwrap(); - let mut state = self.state.write(); - - let mut options = RuntimeOptions::from_runtime_env(state.runtime_env()); + // Every field of a fresh `RuntimeOptions` already holds its default, so + // this only has to reject an unknown key. + let mut options = RuntimeOptions::default(); options.reset_entry(key)?; - self.apply_runtime_options(&mut state, &options, key) + self.apply_runtime_option(&options, key) } - /// Rebuild the session's [`RuntimeEnv`] with `key` taken from `options`. + /// Rebuild the session's `RuntimeEnv` with `key` taken from `options`. /// /// Only the named key is applied. Rebuilding every resource on each /// statement would replace the live memory pool and discard the /// reservations held against it. - fn apply_runtime_options( - &self, - state: &mut SessionState, - options: &RuntimeOptions, - key: &str, - ) -> Result<()> { + fn apply_runtime_option(&self, options: &RuntimeOptions, key: &str) -> Result<()> { + let mut state = self.state.write(); + let builder = options.apply_key( key, RuntimeEnvBuilder::from_runtime_env(state.runtime_env()), diff --git a/datafusion/execution/src/runtime_env.rs b/datafusion/execution/src/runtime_env.rs index 0f3c2b71e411d..bcca32ff783bc 100644 --- a/datafusion/execution/src/runtime_env.rs +++ b/datafusion/execution/src/runtime_env.rs @@ -28,10 +28,9 @@ use crate::{ }; use crate::cache::cache_manager::{CacheManager, CacheManagerConfig}; -use crate::memory_pool::MemoryLimit; #[cfg(feature = "parquet_encryption")] use crate::parquet_encryption::{EncryptionFactory, EncryptionFactoryRegistry}; -use crate::runtime_options::{RUNTIME_CONFIG_PREFIX, RuntimeOptions}; +use crate::runtime_options::RuntimeOptions; use datafusion_common::{Result, config::ConfigEntry}; use object_store::ObjectStore; use std::sync::Arc; @@ -181,23 +180,8 @@ impl RuntimeEnv { } /// Returns the current runtime configuration entries - /// - /// Keys, descriptions and ordering come from [`RuntimeOptions`]; the values - /// are read back from the live resource objects, which is why this is not - /// simply `RuntimeOptions::default().entries()`. pub fn config_entries(&self) -> Vec { - let mut entries = RuntimeOptions::from_runtime_env(self).entries(); - - // An unbounded pool has no byte value to report, but it is not unset - // either, so `RuntimeOptions` cannot carry it. - if matches!(self.memory_pool.memory_limit(), MemoryLimit::Infinite) { - let key = format!("{RUNTIME_CONFIG_PREFIX}.memory_limit"); - if let Some(entry) = entries.iter_mut().find(|e| e.key == key) { - entry.value = Some("unlimited".to_string()); - } - } - - entries + RuntimeOptions::env_entries(self) } } diff --git a/datafusion/execution/src/runtime_options.rs b/datafusion/execution/src/runtime_options.rs index e86229086cc9c..6bf2617cd623c 100644 --- a/datafusion/execution/src/runtime_options.rs +++ b/datafusion/execution/src/runtime_options.rs @@ -17,16 +17,10 @@ //! Typed schema for the `datafusion.runtime.*` configuration namespace. //! -//! [`RuntimeOptions`] is the single source of truth for the runtime -//! configuration keys, their descriptions, their defaults and how their string -//! values are parsed. It is built with the same [`ConfigField`] machinery that -//! backs [`ConfigOptions`], so `SET`/`RESET` handling no longer needs a -//! hand-written `match` arm per key. -//! -//! [`RuntimeOptions`] deliberately does *not* live inside [`ConfigOptions`]: +//! [`RuntimeOptions`] owns the runtime keys, their descriptions, their defaults +//! and their parsing, using the same [`ConfigField`] machinery that backs +//! `ConfigOptions`. It does not live inside `ConfigOptions` because //! `datafusion-execution` depends on `datafusion-common`, never the reverse. -//! -//! [`ConfigOptions`]: datafusion_common::config::ConfigOptions use std::fmt::{self, Display, Formatter}; use std::time::Duration; @@ -87,10 +81,8 @@ config_field!(CacheTtl, value => CacheTtl(parse_ttl(value)?)); /// A count that must not be negative. /// -/// A plain `usize` would parse through [`default_config_transform`], whose -/// message names the Rust type rather than the constraint. -/// -/// [`default_config_transform`]: datafusion_common::config::default_config_transform +/// A plain `usize` would report `Error parsing '-1' as usize`, naming the Rust +/// type rather than the constraint. #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct NonNegativeCount(pub usize); @@ -180,11 +172,28 @@ impl RuntimeOptions { v.0 } + /// The entries to report for `env`, with values read back from its live + /// resource objects rather than from whatever was last requested. + pub fn env_entries(env: &RuntimeEnv) -> Vec { + let mut entries = Self::from_runtime_env(env).entries(); + + // An unbounded pool is not unset, but has no byte value to report, so + // `memory_limit` cannot carry it. + if matches!(env.memory_pool.memory_limit(), MemoryLimit::Infinite) { + let key = format!("{RUNTIME_CONFIG_PREFIX}.memory_limit"); + if let Some(entry) = entries.iter_mut().find(|e| e.key == key) { + entry.value = Some("unlimited".to_string()); + } + } + + entries + } + /// Read the values currently in effect from the live resource objects. /// - /// A [`MemoryLimit::Infinite`] pool has no representation here; callers that - /// report values handle it separately. - pub fn from_runtime_env(env: &RuntimeEnv) -> Self { + /// A [`MemoryLimit::Infinite`] pool has no representation here, which is why + /// [`Self::env_entries`] handles it separately. + fn from_runtime_env(env: &RuntimeEnv) -> Self { let memory_limit = match env.memory_pool.memory_limit() { MemoryLimit::Finite(size) => Some(CapacityLimit(size)), MemoryLimit::Infinite | MemoryLimit::Unknown => None,