datafusion_common/config.rs
1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements. See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership. The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License. You may obtain a copy of the License at
8//
9// http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied. See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Runtime configuration, via [`ConfigOptions`]
19
20use arrow_ipc::CompressionType;
21
22#[cfg(feature = "parquet_encryption")]
23use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
24use crate::error::{_config_datafusion_err, _config_err};
25use crate::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
26use crate::parquet_config::DFParquetWriterVersion;
27use crate::parsers::{CompressionTypeVariant, CsvQuoteStyle};
28use crate::utils::get_available_parallelism;
29use crate::{DataFusionError, Result};
30#[cfg(feature = "parquet_encryption")]
31use hex;
32use std::any::Any;
33use std::collections::{BTreeMap, HashMap};
34use std::error::Error;
35use std::fmt::{self, Display};
36use std::num::NonZeroUsize;
37use std::str::FromStr;
38#[cfg(feature = "parquet_encryption")]
39use std::sync::Arc;
40
41/// A macro that wraps a configuration struct and automatically derives
42/// [`Default`] and [`ConfigField`] for it, allowing it to be used
43/// in the [`ConfigOptions`] configuration tree.
44///
45/// `transform` is used to normalize values before parsing.
46///
47/// For example,
48///
49/// ```ignore
50/// config_namespace! {
51/// /// Amazing config
52/// pub struct MyConfig {
53/// /// Field 1 doc
54/// field1: String, transform = str::to_lowercase, default = "".to_string()
55///
56/// /// Field 2 doc
57/// field2: usize, default = 232
58///
59/// /// Field 3 doc
60/// field3: Option<usize>, default = None
61/// }
62/// }
63/// ```
64///
65/// Will generate
66///
67/// ```ignore
68/// /// Amazing config
69/// #[derive(Debug, Clone)]
70/// #[non_exhaustive]
71/// pub struct MyConfig {
72/// /// Field 1 doc
73/// field1: String,
74/// /// Field 2 doc
75/// field2: usize,
76/// /// Field 3 doc
77/// field3: Option<usize>,
78/// }
79/// impl ConfigField for MyConfig {
80/// fn set(&mut self, key: &str, value: &str) -> Result<()> {
81/// let (key, rem) = key.split_once('.').unwrap_or((key, ""));
82/// match key {
83/// "field1" => {
84/// let value = str::to_lowercase(value);
85/// self.field1.set(rem, value.as_ref())
86/// },
87/// "field2" => self.field2.set(rem, value.as_ref()),
88/// "field3" => self.field3.set(rem, value.as_ref()),
89/// _ => _internal_err!(
90/// "Config value \"{}\" not found on MyConfig",
91/// key
92/// ),
93/// }
94/// }
95///
96/// fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
97/// let key = format!("{}.field1", key_prefix);
98/// let desc = "Field 1 doc";
99/// self.field1.visit(v, key.as_str(), desc);
100/// let key = format!("{}.field2", key_prefix);
101/// let desc = "Field 2 doc";
102/// self.field2.visit(v, key.as_str(), desc);
103/// let key = format!("{}.field3", key_prefix);
104/// let desc = "Field 3 doc";
105/// self.field3.visit(v, key.as_str(), desc);
106/// }
107/// }
108///
109/// impl Default for MyConfig {
110/// fn default() -> Self {
111/// Self {
112/// field1: "".to_string(),
113/// field2: 232,
114/// field3: None,
115/// }
116/// }
117/// }
118/// ```
119///
120/// NB: Misplaced commas may result in nonsensical errors
121#[macro_export]
122macro_rules! config_namespace {
123 (
124 $(#[doc = $struct_d:tt])* // Struct-level documentation attributes
125 $(#[deprecated($($struct_depr:tt)*)])? // Optional struct-level deprecated attribute
126 $(#[allow($($struct_de:tt)*)])?
127 $vis:vis struct $struct_name:ident {
128 $(
129 $(#[doc = $d:tt])* // Field-level documentation attributes
130 $(#[deprecated($($field_depr:tt)*)])? // Optional field-level deprecated attribute
131 $(#[allow($($field_de:tt)*)])?
132 $field_vis:vis $field_name:ident : $field_type:ty,
133 $(warn = $warn:expr,)?
134 $(transform = $transform:expr,)?
135 default = $default:expr
136 )*$(,)*
137 }
138 ) => {
139 $(#[doc = $struct_d])* // Apply struct documentation
140 $(#[deprecated($($struct_depr)*)])? // Apply struct deprecation
141 $(#[allow($($struct_de)*)])?
142 #[derive(Debug, Clone, PartialEq)]
143 $vis struct $struct_name {
144 $(
145 $(#[doc = $d])* // Apply field documentation
146 $(#[deprecated($($field_depr)*)])? // Apply field deprecation
147 $(#[allow($($field_de)*)])?
148 $field_vis $field_name: $field_type,
149 )*
150 }
151
152 impl $crate::config::ConfigField for $struct_name {
153 fn set(&mut self, key: &str, value: &str) -> $crate::error::Result<()> {
154 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
155 match key {
156 $(
157 stringify!($field_name) => {
158 // Safely apply deprecated attribute if present
159 // $(#[allow(deprecated)])?
160 {
161 $(let value = $transform(value);)? // Apply transformation if specified
162 let ret = self.$field_name.set(rem, value.as_ref());
163
164 $(if !$warn.is_empty() {
165 let default: $field_type = $default;
166 if default != self.$field_name {
167 log::warn!($warn);
168 }
169 })? // Log warning if specified, and the value is not the default
170 ret
171 }
172 },
173 )*
174 _ => return $crate::error::_config_err!(
175 "Config value \"{}\" not found on {}", key, stringify!($struct_name)
176 )
177 }
178 }
179
180 fn visit<V: $crate::config::Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
181 $(
182 let key = format!(concat!("{}.", stringify!($field_name)), key_prefix);
183 let desc = concat!($($d),*).trim();
184 self.$field_name.visit(v, key.as_str(), desc);
185 )*
186 }
187
188 fn reset(&mut self, key: &str) -> $crate::error::Result<()> {
189 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
190 match key {
191 $(
192 stringify!($field_name) => {
193 {
194 if rem.is_empty() {
195 let default_value: $field_type = $default;
196 self.$field_name = default_value;
197 Ok(())
198 } else {
199 self.$field_name.reset(rem)
200 }
201 }
202 },
203 )*
204 _ => $crate::error::_config_err!(
205 "Config value \"{}\" not found on {}",
206 key,
207 stringify!($struct_name)
208 ),
209 }
210 }
211 }
212 impl Default for $struct_name {
213 fn default() -> Self {
214 Self {
215 $($field_name: $default),*
216 }
217 }
218 }
219 }
220}
221
222config_namespace! {
223 /// Options related to catalog and directory scanning
224 ///
225 /// See also: [`SessionConfig`]
226 ///
227 /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
228 pub struct CatalogOptions {
229 /// Whether the default catalog and schema should be created automatically.
230 pub create_default_catalog_and_schema: bool, default = true
231
232 /// The default catalog name - this impacts what SQL queries use if not specified
233 pub default_catalog: String, default = "datafusion".to_string()
234
235 /// The default schema name - this impacts what SQL queries use if not specified
236 pub default_schema: String, default = "public".to_string()
237
238 /// Should DataFusion provide access to `information_schema`
239 /// virtual tables for displaying schema information
240 pub information_schema: bool, default = false
241
242 /// Location scanned to load tables for `default` schema
243 pub location: Option<String>, default = None
244
245 /// Type of `TableProvider` to use when loading `default` schema
246 pub format: Option<String>, default = None
247
248 /// Default value for `format.has_header` for `CREATE EXTERNAL TABLE`
249 /// if not specified explicitly in the statement.
250 pub has_header: bool, default = true
251
252 /// Specifies whether newlines in (quoted) CSV values are supported.
253 ///
254 /// This is the default value for `format.newlines_in_values` for `CREATE EXTERNAL TABLE`
255 /// if not specified explicitly in the statement.
256 ///
257 /// Parsing newlines in quoted values may be affected by execution behaviour such as
258 /// parallel file scanning. Setting this to `true` ensures that newlines in values are
259 /// parsed successfully, which may reduce performance.
260 pub newlines_in_values: bool, default = false
261 }
262}
263
264config_namespace! {
265 /// Options related to SQL parser
266 ///
267 /// See also: [`SessionConfig`]
268 ///
269 /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
270 pub struct SqlParserOptions {
271 /// When set to true, SQL parser will parse float as decimal type
272 pub parse_float_as_decimal: bool, default = false
273
274 /// When set to true, SQL parser will normalize ident (convert ident to lowercase when not quoted)
275 pub enable_ident_normalization: bool, default = true
276
277 /// When set to true, SQL parser will normalize options value (convert value to lowercase).
278 /// Note that this option is ignored and will be removed in the future. All case-insensitive values
279 /// are normalized automatically.
280 pub enable_options_value_normalization: bool, warn = "`enable_options_value_normalization` is deprecated and ignored", default = false
281
282 /// Configure the SQL dialect used by DataFusion's parser.
283 /// The configuration reference lists the supported values from [`Dialect::available`].
284 pub dialect: Dialect, default = Dialect::Generic
285 // no need to lowercase because `sqlparser::dialect_from_str`] is case-insensitive
286
287 /// If true, permit lengths for `VARCHAR` such as `VARCHAR(20)`, but
288 /// ignore the length. If false, error if a `VARCHAR` with a length is
289 /// specified. The Arrow type system does not have a notion of maximum
290 /// string length and thus DataFusion can not enforce such limits.
291 pub support_varchar_with_length: bool, default = true
292
293 /// If true, string types (VARCHAR, CHAR, Text, and String) are mapped to `Utf8View` during SQL planning.
294 /// If false, they are mapped to `Utf8`.
295 /// Default is true.
296 pub map_string_types_to_utf8view: bool, default = true
297
298 /// When set to true, the source locations relative to the original SQL
299 /// query (i.e. [`Span`](https://docs.rs/sqlparser/latest/sqlparser/tokenizer/struct.Span.html)) will be collected
300 /// and recorded in the logical plan nodes.
301 pub collect_spans: bool, default = false
302
303 /// Specifies the recursion depth limit when parsing complex SQL Queries
304 pub recursion_limit: ConfigNonZeroUsize, default = non_zero_usize_default(50)
305
306 /// Specifies the default null ordering for query results. There are 4 options:
307 /// - `nulls_max`: Nulls appear last in ascending order.
308 /// - `nulls_min`: Nulls appear first in ascending order.
309 /// - `nulls_first`: Nulls always be first in any order.
310 /// - `nulls_last`: Nulls always be last in any order.
311 ///
312 /// By default, `nulls_max` is used to follow Postgres's behavior.
313 /// postgres rule: <https://www.postgresql.org/docs/current/queries-order.html>
314 pub default_null_ordering: String, default = "nulls_max".to_string()
315
316 /// When set to true, DataFusion may remove `ORDER BY` clauses from
317 /// subqueries or CTEs during SQL planning when their ordering cannot
318 /// affect the result, such as when no `LIMIT` or other
319 /// order-sensitive operator depends on them.
320 ///
321 /// Disable this option to preserve explicit subquery ordering in the
322 /// planned query.
323 pub enable_subquery_sort_elimination: bool, default = true
324 }
325}
326
327/// Metadata for a SQL dialect supported by DataFusion configuration.
328#[derive(Debug, Clone, Copy, PartialEq, Eq)]
329#[non_exhaustive]
330pub struct DialectInfo {
331 pub dialect: Dialect,
332 pub canonical_name: &'static str,
333 pub display_name: &'static str,
334 pub aliases: &'static [&'static str],
335}
336
337// Keep this key in sync with the `SqlParserOptions::dialect` config path.
338const SQL_PARSER_DIALECT_CONFIG_KEY: &str = "datafusion.sql_parser.dialect";
339
340macro_rules! dialect_display_list {
341 ($($display_name:literal),+ $(,)?) => {
342 dialect_display_list!(@acc [] $($display_name),+)
343 };
344 (@acc [$($acc:tt)*] $last:literal) => {
345 concat!($($acc)* $last)
346 };
347 (@acc [$($acc:tt)*] $next:literal, $($rest:literal),+) => {
348 dialect_display_list!(@acc [$($acc)* $next, ", ",] $($rest),+)
349 };
350}
351
352macro_rules! dialect_metadata {
353 (
354 default: $default_variant:ident;
355 $(
356 $variant:ident {
357 canonical_name: $canonical_name:literal,
358 display_name: $display_name:literal,
359 aliases: [$($alias:literal),* $(,)?],
360 }
361 ),+ $(,)?
362 ) => {
363 /// This is the SQL dialect used by DataFusion's parser.
364 /// This mirrors [sqlparser::dialect::Dialect](https://docs.rs/sqlparser/latest/sqlparser/dialect/trait.Dialect.html)
365 /// trait in order to offer an easier API and avoid adding the `sqlparser` dependency
366 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
367 pub enum Dialect {
368 $($variant,)+
369 }
370
371 impl Default for Dialect {
372 fn default() -> Self {
373 Self::$default_variant
374 }
375 }
376
377 const DIALECT_INFOS: &[DialectInfo] = &[
378 $(
379 DialectInfo {
380 dialect: Dialect::$variant,
381 canonical_name: $canonical_name,
382 display_name: $display_name,
383 aliases: &[$($alias),*],
384 },
385 )+
386 ];
387
388 const AVAILABLE_DIALECTS: &str = dialect_display_list!($($display_name),+);
389 const DIALECT_CONFIG_DESCRIPTION: &str = concat!(
390 "Configure the SQL dialect used by DataFusion's parser; supported values include: ",
391 dialect_display_list!($($display_name),+),
392 "."
393 );
394 };
395}
396
397dialect_metadata! {
398 default: Generic;
399 Generic {
400 canonical_name: "generic",
401 display_name: "Generic",
402 aliases: [],
403 },
404 MySQL {
405 canonical_name: "mysql",
406 display_name: "MySQL",
407 aliases: [],
408 },
409 PostgreSQL {
410 canonical_name: "postgresql",
411 display_name: "PostgreSQL",
412 aliases: ["postgres"],
413 },
414 Hive {
415 canonical_name: "hive",
416 display_name: "Hive",
417 aliases: [],
418 },
419 SQLite {
420 canonical_name: "sqlite",
421 display_name: "SQLite",
422 aliases: [],
423 },
424 Snowflake {
425 canonical_name: "snowflake",
426 display_name: "Snowflake",
427 aliases: [],
428 },
429 Redshift {
430 canonical_name: "redshift",
431 display_name: "Redshift",
432 aliases: [],
433 },
434 MsSQL {
435 canonical_name: "mssql",
436 display_name: "MsSQL",
437 aliases: [],
438 },
439 ClickHouse {
440 canonical_name: "clickhouse",
441 display_name: "ClickHouse",
442 aliases: [],
443 },
444 BigQuery {
445 canonical_name: "bigquery",
446 display_name: "BigQuery",
447 aliases: [],
448 },
449 Ansi {
450 canonical_name: "ansi",
451 display_name: "Ansi",
452 aliases: [],
453 },
454 DuckDB {
455 canonical_name: "duckdb",
456 display_name: "DuckDB",
457 aliases: [],
458 },
459 Databricks {
460 canonical_name: "databricks",
461 display_name: "Databricks",
462 aliases: [],
463 },
464 Spark {
465 canonical_name: "spark",
466 display_name: "Spark",
467 aliases: ["sparksql"],
468 },
469}
470
471impl Dialect {
472 /// Return metadata for all supported dialects.
473 pub fn metadata() -> &'static [DialectInfo] {
474 DIALECT_INFOS
475 }
476
477 /// Return all supported dialect names, for use in error messages.
478 pub fn available() -> &'static str {
479 AVAILABLE_DIALECTS
480 }
481
482 fn info(&self) -> &'static DialectInfo {
483 DIALECT_INFOS
484 .iter()
485 .find(|info| info.dialect == *self)
486 .expect("all Dialect variants are listed in DIALECT_INFOS")
487 }
488}
489
490impl AsRef<str> for Dialect {
491 fn as_ref(&self) -> &str {
492 self.info().canonical_name
493 }
494}
495
496impl FromStr for Dialect {
497 type Err = DataFusionError;
498
499 fn from_str(s: &str) -> Result<Self, Self::Err> {
500 for info in DIALECT_INFOS {
501 if info.canonical_name.eq_ignore_ascii_case(s)
502 || info
503 .aliases
504 .iter()
505 .any(|alias| alias.eq_ignore_ascii_case(s))
506 {
507 return Ok(info.dialect);
508 }
509 }
510
511 Err(DataFusionError::Configuration(format!(
512 "Invalid Dialect: {s}. Expected one of: {}",
513 Self::available()
514 )))
515 }
516}
517
518impl ConfigField for Dialect {
519 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
520 let description = if key == SQL_PARSER_DIALECT_CONFIG_KEY {
521 DIALECT_CONFIG_DESCRIPTION
522 } else {
523 description
524 };
525 v.some(key, self, description)
526 }
527
528 fn set(&mut self, _: &str, value: &str) -> Result<()> {
529 *self = Self::from_str(value)?;
530 Ok(())
531 }
532}
533
534impl Display for Dialect {
535 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
536 let str = self.as_ref();
537 write!(f, "{str}")
538 }
539}
540
541#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
542pub enum SpillCompression {
543 Zstd,
544 Lz4Frame,
545 #[default]
546 Uncompressed,
547}
548
549impl FromStr for SpillCompression {
550 type Err = DataFusionError;
551
552 fn from_str(s: &str) -> Result<Self, Self::Err> {
553 match s.to_ascii_lowercase().as_str() {
554 "zstd" => Ok(Self::Zstd),
555 "lz4_frame" => Ok(Self::Lz4Frame),
556 "uncompressed" | "" => Ok(Self::Uncompressed),
557 other => Err(DataFusionError::Configuration(format!(
558 "Invalid Spill file compression type: {other}. Expected one of: zstd, lz4_frame, uncompressed"
559 ))),
560 }
561 }
562}
563
564impl ConfigField for SpillCompression {
565 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
566 v.some(key, self, description)
567 }
568
569 fn set(&mut self, _: &str, value: &str) -> Result<()> {
570 *self = SpillCompression::from_str(value)?;
571 Ok(())
572 }
573}
574
575impl Display for SpillCompression {
576 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
577 let str = match self {
578 Self::Zstd => "zstd",
579 Self::Lz4Frame => "lz4_frame",
580 Self::Uncompressed => "uncompressed",
581 };
582 write!(f, "{str}")
583 }
584}
585
586/// A `usize` configuration value that rejects zero when set from strings.
587///
588/// Use this for options where zero is never a meaningful runtime value.
589/// Invalid values return a configuration error through [`ConfigField`].
590#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
591pub struct ConfigNonZeroUsize(NonZeroUsize);
592
593/// Private helper for hard-coded defaults in `config_namespace!`, which cannot
594/// use `?`. All external construction should use [`ConfigNonZeroUsize::try_new`].
595const fn non_zero_usize_default(value: usize) -> ConfigNonZeroUsize {
596 match NonZeroUsize::new(value) {
597 Some(value) => ConfigNonZeroUsize(value),
598 None => panic!("value must be greater than 0"),
599 }
600}
601
602impl ConfigNonZeroUsize {
603 /// Creates a [`ConfigNonZeroUsize`], returning a configuration error if
604 /// `value` is zero.
605 pub fn try_new(value: usize) -> Result<Self> {
606 NonZeroUsize::new(value)
607 .map(Self)
608 .ok_or_else(|| _config_datafusion_err!("value must be greater than 0"))
609 }
610
611 /// Returns the wrapped `usize`.
612 pub const fn get(self) -> usize {
613 self.0.get()
614 }
615}
616
617impl From<ConfigNonZeroUsize> for usize {
618 fn from(value: ConfigNonZeroUsize) -> Self {
619 value.get()
620 }
621}
622
623impl FromStr for ConfigNonZeroUsize {
624 type Err = DataFusionError;
625
626 fn from_str(s: &str) -> Result<Self, Self::Err> {
627 Self::try_new(default_config_transform(s)?)
628 }
629}
630
631impl ConfigField for ConfigNonZeroUsize {
632 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
633 v.some(key, self, description)
634 }
635
636 fn set(&mut self, key: &str, value: &str) -> Result<()> {
637 if !key.is_empty() {
638 return _config_err!(
639 "Config field batch_size is a scalar ConfigNonZeroUsize and does not have nested field \"{}\"",
640 key
641 );
642 }
643
644 *self = ConfigNonZeroUsize::from_str(value)?;
645 Ok(())
646 }
647
648 fn reset(&mut self, key: &str) -> Result<()> {
649 if key.is_empty() {
650 Ok(())
651 } else {
652 _config_err!(
653 "Config field batch_size is a scalar ConfigNonZeroUsize and does not have nested field \"{}\"",
654 key
655 )
656 }
657 }
658}
659
660impl Display for ConfigNonZeroUsize {
661 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
662 write!(f, "{}", self.get())
663 }
664}
665
666/// A `usize` configuration value that rejects 0 and 1 when set from strings.
667///
668/// Use this for options whose consumer divides the value in half to size an
669/// internal buffer (e.g. a bounded channel capacity): values below 2 would
670/// round down to a zero-capacity buffer and panic. Invalid values return a
671/// configuration error through [`ConfigField`] instead.
672#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
673pub struct ConfigMinTwoUsize(usize);
674
675/// Private helper for hard-coded defaults in `config_namespace!`, which cannot
676/// use `?`. All external construction should use [`ConfigMinTwoUsize::try_new`].
677const fn min_two_usize_default(value: usize) -> ConfigMinTwoUsize {
678 if value >= 2 {
679 ConfigMinTwoUsize(value)
680 } else {
681 panic!("value must be at least 2")
682 }
683}
684
685impl ConfigMinTwoUsize {
686 /// Creates a [`ConfigMinTwoUsize`], returning a configuration error if
687 /// `value` is less than 2.
688 pub fn try_new(value: usize) -> Result<Self> {
689 if value >= 2 {
690 Ok(Self(value))
691 } else {
692 _config_err!("value must be at least 2")
693 }
694 }
695
696 /// Returns the wrapped `usize`.
697 pub const fn get(self) -> usize {
698 self.0
699 }
700}
701
702impl From<ConfigMinTwoUsize> for usize {
703 fn from(value: ConfigMinTwoUsize) -> Self {
704 value.get()
705 }
706}
707
708impl FromStr for ConfigMinTwoUsize {
709 type Err = DataFusionError;
710
711 fn from_str(s: &str) -> Result<Self, Self::Err> {
712 Self::try_new(default_config_transform(s)?)
713 }
714}
715
716impl ConfigField for ConfigMinTwoUsize {
717 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
718 v.some(key, self, description)
719 }
720
721 fn set(&mut self, key: &str, value: &str) -> Result<()> {
722 if !key.is_empty() {
723 return _config_err!(
724 "Config field max_buffered_batches_per_output_file is a scalar ConfigMinTwoUsize and does not have nested field \"{}\"",
725 key
726 );
727 }
728
729 *self = ConfigMinTwoUsize::from_str(value)?;
730 Ok(())
731 }
732
733 fn reset(&mut self, key: &str) -> Result<()> {
734 if key.is_empty() {
735 Ok(())
736 } else {
737 _config_err!(
738 "Config field max_buffered_batches_per_output_file is a scalar ConfigMinTwoUsize and does not have nested field \"{}\"",
739 key
740 )
741 }
742 }
743}
744
745impl Display for ConfigMinTwoUsize {
746 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
747 write!(f, "{}", self.get())
748 }
749}
750
751/// Policy for handling duplicate keys in Spark-compatible map-construction
752/// functions (`map_from_arrays`, `map_from_entries`, `str_to_map`). Mirrors
753/// 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).
754#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
755pub enum MapKeyDedupPolicy {
756 /// Raise `[DUPLICATED_MAP_KEY]` at runtime on any duplicate key.
757 #[default]
758 Exception,
759 /// Keep the last occurrence of each duplicate key.
760 LastWin,
761}
762
763impl FromStr for MapKeyDedupPolicy {
764 type Err = DataFusionError;
765
766 fn from_str(s: &str) -> Result<Self, Self::Err> {
767 match s.to_ascii_uppercase().as_str() {
768 "EXCEPTION" => Ok(Self::Exception),
769 "LAST_WIN" => Ok(Self::LastWin),
770 other => Err(DataFusionError::Configuration(format!(
771 "Invalid MapKeyDedupPolicy: {other}. Expected one of: EXCEPTION, LAST_WIN"
772 ))),
773 }
774 }
775}
776
777impl ConfigField for MapKeyDedupPolicy {
778 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
779 v.some(key, self, description)
780 }
781
782 fn set(&mut self, _: &str, value: &str) -> Result<()> {
783 *self = MapKeyDedupPolicy::from_str(value)?;
784 Ok(())
785 }
786}
787
788impl Display for MapKeyDedupPolicy {
789 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
790 let str = match self {
791 Self::Exception => "EXCEPTION",
792 Self::LastWin => "LAST_WIN",
793 };
794 write!(f, "{str}")
795 }
796}
797
798impl From<SpillCompression> for Option<CompressionType> {
799 fn from(c: SpillCompression) -> Self {
800 match c {
801 SpillCompression::Zstd => Some(CompressionType::ZSTD),
802 SpillCompression::Lz4Frame => Some(CompressionType::LZ4_FRAME),
803 SpillCompression::Uncompressed => None,
804 }
805 }
806}
807
808config_namespace! {
809 /// Options related to query execution
810 ///
811 /// See also: [`SessionConfig`]
812 ///
813 /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
814 pub struct ExecutionOptions {
815 /// Default batch size while creating new batches, it's especially useful for
816 /// buffer-in-memory batches since creating tiny batches would result in too much
817 /// metadata memory consumption
818 pub batch_size: ConfigNonZeroUsize, default = non_zero_usize_default(8192)
819
820 /// A perfect hash join (see `HashJoinExec` for more details) will be considered
821 /// if the range of keys (max - min) on the build side is < this threshold.
822 /// This provides a fast path for joins with very small key ranges,
823 /// bypassing the density check.
824 ///
825 /// Currently only supports cases where build_side.num_rows() < u32::MAX.
826 /// Support for build_side.num_rows() >= u32::MAX will be added in the future.
827 pub perfect_hash_join_small_build_threshold: usize, default = 1024
828
829 /// The minimum required density of join keys on the build side to consider a
830 /// perfect hash join (see `HashJoinExec` for more details). Density is calculated as:
831 /// `(number of rows) / (max_key - min_key + 1)`.
832 /// A perfect hash join may be used if the actual key density > this
833 /// value.
834 ///
835 /// Currently only supports cases where build_side.num_rows() < u32::MAX.
836 /// Support for build_side.num_rows() >= u32::MAX will be added in the future.
837 pub perfect_hash_join_min_key_density: f64, default = 0.15
838
839 /// When set to true, record batches will be examined between each operator and
840 /// small batches will be coalesced into larger batches. This is helpful when there
841 /// are highly selective filters or joins that could produce tiny output batches. The
842 /// target batch size is determined by the configuration setting
843 pub coalesce_batches: bool, default = true
844
845 /// Should DataFusion collect statistics when first creating a table.
846 /// Has no effect after the table is created. Defaults to true.
847 pub collect_statistics: bool, default = true
848
849 /// Number of partitions for query execution. Increasing partitions can increase
850 /// concurrency.
851 ///
852 /// Defaults to the number of CPU cores on the system
853 pub target_partitions: usize, transform = ExecutionOptions::normalized_parallelism, default = get_available_parallelism()
854
855 /// The default time zone
856 ///
857 /// Some functions, e.g. `now` return timestamps in this time zone
858 pub time_zone: Option<String>, default = None
859
860 /// Parquet options
861 pub parquet: ParquetOptions, default = Default::default()
862
863 /// Fan-out during initial physical planning.
864 ///
865 /// This is mostly use to plan `UNION` children in parallel.
866 ///
867 /// Defaults to the number of CPU cores on the system
868 pub planning_concurrency: usize, transform = ExecutionOptions::normalized_parallelism, default = get_available_parallelism()
869
870 /// When set to true, skips verifying that the schema produced by
871 /// planning the input of `LogicalPlan::Aggregate` exactly matches the
872 /// schema of the input plan.
873 ///
874 /// When set to false, if the schema does not match exactly
875 /// (including nullability and metadata), a planning error will be raised.
876 ///
877 /// This is used to workaround bugs in the planner that are now caught by
878 /// the new schema verification step.
879 pub skip_physical_aggregate_schema_check: bool, default = false
880
881 /// Temporary switch for aggregate stream implementations that are being
882 /// migrated from `GroupedHashAggregateStream`.
883 ///
884 /// When set to true, DataFusion tries the migrated implementations when
885 /// their preconditions are satisfied. When set to false, grouped
886 /// aggregation falls back to `GroupedHashAggregateStream`. This option
887 /// will be removed after the migration is finished.
888 ///
889 /// See <https://github.com/apache/datafusion/issues/22710> for details.
890 pub enable_migration_aggregate: bool, default = true
891
892 /// Sets the compression codec used when spilling data to disk.
893 ///
894 /// Since datafusion writes spill files using the Arrow IPC Stream format,
895 /// only codecs supported by the Arrow IPC Stream Writer are allowed.
896 /// Valid values are: uncompressed, lz4_frame, zstd.
897 /// Note: lz4_frame offers faster (de)compression, but typically results in
898 /// larger spill files. In contrast, zstd achieves
899 /// higher compression ratios at the cost of slower (de)compression speed.
900 pub spill_compression: SpillCompression, default = SpillCompression::Uncompressed
901
902 /// Specifies the reserved memory for each spillable sort operation to
903 /// facilitate an in-memory merge.
904 ///
905 /// When a sort operation spills to disk, the in-memory data must be
906 /// sorted and merged before being written to a file. This setting reserves
907 /// a specific amount of memory for that in-memory sort/merge process.
908 ///
909 /// Note: This setting is irrelevant if the sort operation cannot spill
910 /// (i.e., if there's no `DiskManager` configured).
911 pub sort_spill_reservation_bytes: usize, default = 10 * 1024 * 1024
912
913 /// When sorting, below what size should data be concatenated
914 /// and sorted in a single RecordBatch rather than sorted in
915 /// batches and merged.
916 pub sort_in_place_threshold_bytes: usize, default = 1024 * 1024
917
918 /// Maximum buffer capacity (in bytes) per partition for BufferExec
919 /// inserted during sort pushdown optimization.
920 ///
921 /// When PushdownSort eliminates a SortExec under SortPreservingMergeExec,
922 /// a BufferExec is inserted to replace SortExec's buffering role. This
923 /// prevents I/O stalls by allowing the scan to run ahead of the merge.
924 ///
925 /// This uses strictly less memory than the SortExec it replaces (which
926 /// buffers the entire partition). The buffer respects the global memory
927 /// pool limit. Setting this to a large value is safe — actual memory
928 /// usage is bounded by partition size and global memory limits.
929 pub sort_pushdown_buffer_capacity: usize, default = 1024 * 1024 * 1024
930
931 /// Maximum size in bytes for individual spill files before rotating to a new file.
932 ///
933 /// When operators spill data to disk (e.g., RepartitionExec), they write
934 /// multiple batches to the same file until this size limit is reached, then rotate
935 /// to a new file. This reduces syscall overhead compared to one-file-per-batch
936 /// while preventing files from growing too large.
937 ///
938 /// A larger value reduces file creation overhead but may hold more disk space.
939 /// A smaller value creates more files but allows finer-grained space reclamation
940 /// as files can be deleted once fully consumed.
941 ///
942 /// Now only `RepartitionExec` supports this spill file rotation feature, other spilling operators
943 /// may create spill files larger than the limit.
944 ///
945 /// Default: 128 MB
946 pub max_spill_file_size_bytes: ConfigNonZeroUsize, default = non_zero_usize_default(128 * 1024 * 1024)
947
948 /// Number of files to read in parallel when inferring schema and statistics
949 pub meta_fetch_concurrency: ConfigNonZeroUsize, default = non_zero_usize_default(32)
950
951 /// Guarantees a minimum level of output files running in parallel.
952 /// RecordBatches will be distributed in round robin fashion to each
953 /// parallel writer. Each writer is closed and a new file opened once
954 /// soft_max_rows_per_output_file is reached.
955 pub minimum_parallel_output_files: ConfigNonZeroUsize, default = non_zero_usize_default(4)
956
957 /// Target number of rows in output files when writing multiple.
958 /// This is a soft max, so it can be exceeded slightly. There also
959 /// will be one file smaller than the limit if the total
960 /// number of rows written is not roughly divisible by the soft max
961 pub soft_max_rows_per_output_file: ConfigNonZeroUsize, default = non_zero_usize_default(50000000)
962
963 /// This is the maximum number of RecordBatches buffered
964 /// for each output file being worked. Higher values can potentially
965 /// give faster write performance at the cost of higher peak
966 /// memory consumption.
967 ///
968 /// This budget is split evenly between two independent points in the
969 /// write pipeline (see the demuxer diagram in #7791): how many files
970 /// can be in flight from the demuxer to a writer task, and how many
971 /// RecordBatches are buffered for a single file's writer. Must be at
972 /// least 2 so each half gets at least 1 unit of buffering - 0 or 1
973 /// would leave one side with a zero-capacity channel and panic at
974 /// write time.
975 pub max_buffered_batches_per_output_file: ConfigMinTwoUsize, default = min_two_usize_default(2)
976
977 /// Should sub directories be ignored when scanning directories for data
978 /// files. Defaults to true (ignores subdirectories), consistent with
979 /// Hive. Note that this setting does not affect reading partitioned
980 /// tables (e.g. `/table/year=2021/month=01/data.parquet`).
981 pub listing_table_ignore_subdirectory: bool, default = true
982
983 /// Should a `ListingTable` created through the `ListingTableFactory` infer table
984 /// partitions from Hive compliant directories. Defaults to true (partition columns are
985 /// inferred and will be represented in the table schema).
986 pub listing_table_factory_infer_partitions: bool, default = true
987
988 /// Should DataFusion support recursive CTEs
989 pub enable_recursive_ctes: bool, default = true
990
991 /// Attempt to eliminate sorts by packing & sorting files with non-overlapping
992 /// statistics into the same file groups.
993 /// Currently experimental
994 pub split_file_groups_by_statistics: bool, default = false
995
996 /// Should DataFusion keep the columns used for partition_by in the output RecordBatches
997 pub keep_partition_by_columns: bool, default = false
998
999 /// When `true` (the default), DataFusion's built-in file scans
1000 /// dynamically rebalance files across partitions at query execution
1001 /// time: a partition that goes idle reads files (or byte-range morsels)
1002 /// originally assigned to a sibling partition, which keeps all
1003 /// partitions busy in a single process.
1004 ///
1005 /// Executors that depend on the plan-time partition assignment — such as
1006 /// Ballista and datafusion-distributed, which run each partition as an
1007 /// isolated task and never poll the siblings — should set this to
1008 /// `false` so each partition reads only its own file group and no
1009 /// runtime reassignment occurs.
1010 pub enable_file_stream_work_stealing: bool, default = true
1011
1012 /// Aggregation ratio (number of distinct groups / number of input rows)
1013 /// threshold for skipping partial aggregation. If the value is greater
1014 /// then partial aggregation will skip aggregation for further input
1015 pub skip_partial_aggregation_probe_ratio_threshold: f64, default = 0.8
1016
1017 /// Number of input rows partial aggregation partition should process, before
1018 /// aggregation ratio check and trying to switch to skipping aggregation mode
1019 pub skip_partial_aggregation_probe_rows_threshold: usize, default = 100_000
1020
1021 /// Should DataFusion use row number estimates at the input to decide
1022 /// whether increasing parallelism is beneficial or not. By default,
1023 /// only exact row numbers (not estimates) are used for this decision.
1024 /// Setting this flag to `true` will likely produce better plans.
1025 /// if the source of statistics is accurate.
1026 /// We plan to make this the default in the future.
1027 pub use_row_number_estimates_to_optimize_partitioning: bool, default = false
1028
1029 /// Should DataFusion enforce batch size in joins or not. By default,
1030 /// DataFusion will not enforce batch size in joins. Enforcing batch size
1031 /// in joins can reduce memory usage when joining large
1032 /// tables with a highly-selective join filter, but is also slightly slower.
1033 pub enforce_batch_size_in_joins: bool, default = false
1034
1035 /// Size (bytes) of data buffer DataFusion uses when writing output files.
1036 /// This affects the size of the data chunks that are uploaded to remote
1037 /// object stores (e.g. AWS S3). If very large (>= 100 GiB) output files are being
1038 /// written, it may be necessary to increase this size to avoid errors from
1039 /// the remote end point.
1040 pub objectstore_writer_buffer_size: usize, default = 10 * 1024 * 1024
1041
1042 /// Whether to enable ANSI SQL mode.
1043 ///
1044 /// The flag is experimental and relevant only for DataFusion Spark built-in functions
1045 ///
1046 /// When `enable_ansi_mode` is set to `true`, the query engine follows ANSI SQL
1047 /// semantics for expressions, casting, and error handling. This means:
1048 /// - **Strict type coercion rules:** implicit casts between incompatible types are disallowed.
1049 /// - **Standard SQL arithmetic behavior:** operations such as division by zero,
1050 /// numeric overflow, or invalid casts raise runtime errors rather than returning
1051 /// `NULL` or adjusted values.
1052 /// - **Consistent ANSI behavior** for string concatenation, comparisons, and `NULL` handling.
1053 ///
1054 /// When `enable_ansi_mode` is `false` (the default), the engine uses a more permissive,
1055 /// non-ANSI mode designed for user convenience and backward compatibility. In this mode:
1056 /// - Implicit casts between types are allowed (e.g., string to integer when possible).
1057 /// - Arithmetic operations are more lenient — for example, `abs()` on the minimum
1058 /// representable integer value returns the input value instead of raising overflow.
1059 /// - Division by zero or invalid casts may return `NULL` instead of failing.
1060 ///
1061 /// # Default
1062 /// `false` — ANSI SQL mode is disabled by default.
1063 pub enable_ansi_mode: bool, default = false
1064
1065 /// How many bytes to buffer in the probe side of hash joins while the build side is
1066 /// concurrently being built.
1067 ///
1068 /// Without this, hash joins will wait until the full materialization of the build side
1069 /// before polling the probe side. This is useful in scenarios where the query is not
1070 /// completely CPU bounded, allowing to do some early work concurrently and reducing the
1071 /// latency of the query.
1072 ///
1073 /// Note that when hash join buffering is enabled, the probe side will start eagerly
1074 /// polling data, not giving time for the producer side of dynamic filters to produce any
1075 /// meaningful predicate. Queries with dynamic filters might see performance degradation.
1076 ///
1077 /// Disabled by default, set to a number greater than 0 for enabling it.
1078 pub hash_join_buffering_capacity: usize, default = 0
1079 }
1080}
1081
1082config_namespace! {
1083 /// Options for content-defined chunking (CDC) when writing parquet files.
1084 /// Mirrors `parquet::file::properties::CdcOptions`.
1085 ///
1086 /// Carried as a [`ParquetCdcOptions`] in [`ParquetOptions::content_defined_chunking`]
1087 /// with an explicit `enabled` flag, so it can be toggled with dotted config
1088 /// keys (`content_defined_chunking.enabled = true|false`) and the result is
1089 /// independent of the order in which the keys are set.
1090 pub struct ParquetCdcOptions {
1091 /// (writing) EXPERIMENTAL: Enable content-defined chunking (CDC) when writing
1092 /// parquet files. When enabled, parallel writing is automatically disabled
1093 /// since the chunker state must persist across row groups.
1094 pub enabled: bool, default = false
1095
1096 /// Minimum chunk size in bytes. The rolling hash will not trigger a split
1097 /// until this many bytes have been accumulated. Default is 256 KiB.
1098 pub min_chunk_size: usize, default = 256 * 1024
1099
1100 /// Maximum chunk size in bytes. A split is forced when the accumulated
1101 /// size exceeds this value. Default is 1 MiB.
1102 pub max_chunk_size: usize, default = 1024 * 1024
1103
1104 /// Normalization level. Increasing this improves deduplication ratio
1105 /// but increases fragmentation. Recommended range is [-3, 3], default is 0.
1106 pub norm_level: i32, default = 0
1107 }
1108}
1109
1110impl ParquetCdcOptions {
1111 /// Returns enabled CDC options with the default chunking parameters.
1112 ///
1113 /// Shorthand for `ParquetCdcOptions { enabled: true, ..Default::default() }`;
1114 /// combine with struct-update syntax to override parameters, e.g.
1115 /// `ParquetCdcOptions { min_chunk_size: 4096, ..ParquetCdcOptions::enabled() }`.
1116 pub fn enabled() -> Self {
1117 Self {
1118 enabled: true,
1119 ..Default::default()
1120 }
1121 }
1122
1123 /// Returns disabled CDC options (equivalent to [`ParquetCdcOptions::default`]).
1124 pub fn disabled() -> Self {
1125 Self::default()
1126 }
1127}
1128
1129/// Target maximum size of a Parquet row group in bytes.
1130///
1131/// Wraps a `usize` so the "must be greater than zero" constraint (arrow-rs
1132/// panics on a zero byte limit) is validated when the config is set, rather
1133/// than when the writer properties are built.
1134#[derive(Clone, Copy, Debug, PartialEq, Eq)]
1135pub struct MaxRowGroupBytes(usize);
1136
1137impl MaxRowGroupBytes {
1138 /// Creates a `MaxRowGroupBytes`, rejecting zero.
1139 pub fn try_new(value: usize) -> Result<Self> {
1140 if value == 0 {
1141 return Err(DataFusionError::Configuration(
1142 "max_row_group_bytes must be greater than 0".to_string(),
1143 ));
1144 }
1145 Ok(Self(value))
1146 }
1147
1148 /// Returns the configured byte limit.
1149 pub fn get(&self) -> usize {
1150 self.0
1151 }
1152}
1153
1154impl FromStr for MaxRowGroupBytes {
1155 type Err = DataFusionError;
1156
1157 fn from_str(s: &str) -> Result<Self, Self::Err> {
1158 let value = s.parse::<usize>().map_err(|_| {
1159 DataFusionError::Configuration(format!(
1160 "Invalid max_row_group_bytes: '{s}'. Expected a positive integer."
1161 ))
1162 })?;
1163 Self::try_new(value)
1164 }
1165}
1166
1167impl Display for MaxRowGroupBytes {
1168 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1169 write!(f, "{}", self.0)
1170 }
1171}
1172
1173/// `ConfigField` for `Option<MaxRowGroupBytes>`. A custom impl (rather than the
1174/// blanket `Option<F>` one) so an invalid value is rejected without leaving the
1175/// option in an invalid intermediate state on error. `MaxRowGroupBytes`
1176/// deliberately does not implement `Default`, so the blanket impl does not apply.
1177impl ConfigField for Option<MaxRowGroupBytes> {
1178 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
1179 match self {
1180 Some(s) => v.some(key, s, description),
1181 None => v.none(key, description),
1182 }
1183 }
1184
1185 fn set(&mut self, _key: &str, value: &str) -> Result<()> {
1186 *self = Some(MaxRowGroupBytes::from_str(value)?);
1187 Ok(())
1188 }
1189
1190 fn reset(&mut self, _key: &str) -> Result<()> {
1191 *self = None;
1192 Ok(())
1193 }
1194}
1195
1196config_namespace! {
1197 /// Options for reading and writing parquet files
1198 ///
1199 /// See also: [`SessionConfig`]
1200 ///
1201 /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
1202 pub struct ParquetOptions {
1203 // The following options affect reading parquet files
1204
1205 /// (reading) If true, reads the Parquet data page level metadata (the
1206 /// Page Index), if present, to reduce the I/O and number of
1207 /// rows decoded.
1208 pub enable_page_index: bool, default = true
1209
1210 /// (reading) If true, the parquet reader attempts to skip entire row groups based
1211 /// on the predicate in the query and the metadata (min/max values) stored in
1212 /// the parquet file
1213 pub pruning: bool, default = true
1214
1215 /// (reading) If true, the parquet reader skip the optional embedded metadata that may be in
1216 /// the file Schema. This setting can help avoid schema conflicts when querying
1217 /// multiple parquet files with schemas containing compatible types but different metadata
1218 pub skip_metadata: bool, default = true
1219
1220 /// (reading) If specified, the parquet reader will try and fetch the last `size_hint`
1221 /// bytes of the parquet file optimistically. If not specified, two reads are required:
1222 /// One read to fetch the 8-byte parquet footer and
1223 /// another to fetch the metadata length encoded in the footer
1224 /// Default setting to 512 KiB, which should be sufficient for most parquet files,
1225 /// it can reduce one I/O operation per parquet file. If the metadata is larger than
1226 /// the hint, two reads will still be performed.
1227 pub metadata_size_hint: Option<usize>, default = Some(512 * 1024)
1228
1229 /// (reading) If true, filter expressions are be applied during the parquet decoding operation to
1230 /// reduce the number of rows decoded. This optimization is sometimes called "late materialization".
1231 pub pushdown_filters: bool, default = false
1232
1233 /// (reading) If true, filter expressions evaluated during the parquet decoding operation
1234 /// will be reordered heuristically to minimize the cost of evaluation. If false,
1235 /// the filters are applied in the same order as written in the query
1236 pub reorder_filters: bool, default = false
1237
1238 /// (reading) Force the use of RowSelections for filter results, when
1239 /// pushdown_filters is enabled. If false, the reader will automatically
1240 /// choose between a RowSelection and a Bitmap based on the number and
1241 /// pattern of selected rows.
1242 pub force_filter_selections: bool, default = false
1243
1244 /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`,
1245 /// and `Binary/BinaryLarge` with `BinaryView`.
1246 pub schema_force_view_types: bool, default = true
1247
1248 /// (reading) If true, parquet reader will read columns of
1249 /// `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`.
1250 ///
1251 /// Parquet files generated by some legacy writers do not correctly set
1252 /// the UTF8 flag for strings, causing string columns to be loaded as
1253 /// BLOB instead.
1254 pub binary_as_string: bool, default = false
1255
1256 /// (reading) If true, parquet reader will read columns of
1257 /// physical type int96 as originating from a different resolution
1258 /// than nanosecond. This is useful for reading data from systems like Spark
1259 /// which stores microsecond resolution timestamps in an int96 allowing it
1260 /// to write values with a larger date range than 64-bit timestamps with
1261 /// nanosecond resolution.
1262 pub coerce_int96: Option<String>, transform = str::to_lowercase, default = None
1263
1264 /// (reading) Optional timezone applied to INT96 columns when `coerce_int96`
1265 /// is set. When `Some`, INT96 columns coerce to
1266 /// `Timestamp(<coerce_int96>, Some(<tz>))` instead of the default
1267 /// `Timestamp(<coerce_int96>, None)`. Spark and other systems write INT96
1268 /// values as UTC-adjusted instants, so callers that need the resulting
1269 /// Arrow type to be timezone-aware (e.g. for Spark `TimestampType`
1270 /// semantics) should set this to `"UTC"`. No effect when `coerce_int96`
1271 /// is `None`.
1272 pub coerce_int96_tz: Option<String>, default = None
1273
1274 /// (reading) Use any available bloom filters when reading parquet files
1275 pub bloom_filter_on_read: bool, default = true
1276
1277 /// (reading) The maximum predicate cache size, in bytes. When
1278 /// `pushdown_filters` is enabled, sets the maximum memory used to cache
1279 /// the results of predicate evaluation between filter evaluation and
1280 /// output generation. Decreasing this value will reduce memory usage,
1281 /// but may increase IO and CPU usage. None means use the default
1282 /// parquet reader setting. 0 means no caching.
1283 pub max_predicate_cache_size: Option<usize>, default = None
1284
1285 /// Maximum number of values in an `IN (...)` list for which pruning will
1286 /// occur. Longer lists will not be used to prune files, row groups, or
1287 /// data pages.
1288 ///
1289 /// Higher values help in cases such as filtering on a list of
1290 /// ~25-100 identifiers, but also make the predicate more expensive to
1291 /// evaluate. Set to 0 to disable `IN (...)` list pruning entirely.
1292 ///
1293 /// Defaults to 20.
1294 pub max_in_list_size: usize, default = 20
1295
1296 // The following options affect writing to parquet files
1297 // and map to parquet::file::properties::WriterProperties
1298
1299 /// (writing) Sets best effort maximum size of data page in bytes
1300 pub data_pagesize_limit: usize, default = 1024 * 1024
1301
1302 /// (writing) Sets write_batch_size in rows
1303 pub write_batch_size: usize, default = 1024
1304
1305 /// (writing) Sets parquet writer version
1306 /// valid values are "1.0" and "2.0"
1307 pub writer_version: DFParquetWriterVersion, default = DFParquetWriterVersion::default()
1308
1309 /// (writing) Skip encoding the embedded arrow metadata in the KV_meta
1310 ///
1311 /// This is analogous to the `ArrowWriterOptions::with_skip_arrow_metadata`.
1312 /// Refer to <https://docs.rs/parquet/53.3.0/parquet/arrow/arrow_writer/struct.ArrowWriterOptions.html#method.with_skip_arrow_metadata>
1313 pub skip_arrow_metadata: bool, default = false
1314
1315 /// (writing) Sets default parquet compression codec.
1316 /// Valid values are: uncompressed, snappy, gzip(level),
1317 /// brotli(level), lz4, zstd(level), and lz4_raw.
1318 /// These values are not case sensitive. If NULL, uses
1319 /// default parquet writer setting
1320 ///
1321 /// Note that this default setting is not the same as
1322 /// the default parquet writer setting.
1323 pub compression: Option<String>, transform = str::to_lowercase, default = Some("zstd(3)".into())
1324
1325 /// (writing) Sets if dictionary encoding is enabled. If NULL, uses
1326 /// default parquet writer setting
1327 pub dictionary_enabled: Option<bool>, default = Some(true)
1328
1329 /// (writing) Sets best effort maximum dictionary page size, in bytes
1330 pub dictionary_page_size_limit: usize, default = 1024 * 1024
1331
1332 /// (writing) Sets if statistics are enabled for any column
1333 /// Valid values are: "none", "chunk", and "page"
1334 /// These values are not case sensitive. If NULL, uses
1335 /// default parquet writer setting
1336 pub statistics_enabled: Option<String>, transform = str::to_lowercase, default = Some("page".into())
1337
1338 /// (writing) Target maximum number of rows in each row group (defaults to 1M
1339 /// rows). Writing larger row groups requires more memory to write, but
1340 /// can get better compression and be faster to read. When
1341 /// `max_row_group_bytes` is also set, the writer flushes a row group when
1342 /// either limit is reached, whichever comes first.
1343 pub max_row_group_size: usize, default = 1024 * 1024
1344
1345 /// (writing) Target maximum size of each row group in bytes. When set,
1346 /// the writer flushes whenever either this limit or `max_row_group_size`
1347 /// is reached, whichever comes first. Useful for bounding writer memory
1348 /// on wide schemas where a row-count limit can map to very different
1349 /// byte sizes. Matches the behavior of `parquet.block.size` in
1350 /// parquet-mr. If `None` (the default), only the row-count limit
1351 /// applies. Currently only honored when `allow_single_file_parallelism`
1352 /// is `false`; by default the parallel file writer ignores this limit.
1353 pub max_row_group_bytes: Option<MaxRowGroupBytes>, default = None
1354
1355 /// (writing) Sets "created by" property
1356 pub created_by: String, default = concat!("datafusion version ", env!("CARGO_PKG_VERSION")).into()
1357
1358 /// (writing) Sets column index truncate length
1359 pub column_index_truncate_length: Option<usize>, default = Some(64)
1360
1361 /// (writing) Sets statistics truncate length. If NULL, uses
1362 /// default parquet writer setting
1363 pub statistics_truncate_length: Option<usize>, default = Some(64)
1364
1365 /// (writing) Sets best effort maximum number of rows in data page
1366 pub data_page_row_count_limit: usize, default = 20_000
1367
1368 /// (writing) Sets default encoding for any column.
1369 /// Valid values are: plain, plain_dictionary, rle,
1370 /// bit_packed, delta_binary_packed, delta_length_byte_array,
1371 /// delta_byte_array, rle_dictionary, and byte_stream_split.
1372 /// These values are not case sensitive. If NULL, uses
1373 /// default parquet writer setting
1374 pub encoding: Option<String>, transform = str::to_lowercase, default = None
1375
1376 /// (writing) Write bloom filters for all columns when creating parquet files
1377 pub bloom_filter_on_write: bool, default = false
1378
1379 /// (writing) Sets bloom filter false positive probability. If NULL, uses
1380 /// default parquet writer setting
1381 pub bloom_filter_fpp: Option<f64>, default = None
1382
1383 /// (writing) Sets bloom filter number of distinct values. If NULL, uses
1384 /// default parquet writer setting
1385 pub bloom_filter_ndv: Option<u64>, default = None
1386
1387 /// (writing) Controls whether DataFusion will attempt to speed up writing
1388 /// parquet files by serializing them in parallel. Each column
1389 /// in each row group in each output file are serialized in parallel
1390 /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
1391 pub allow_single_file_parallelism: bool, default = true
1392
1393 /// (writing) By default parallel parquet writer is tuned for minimum
1394 /// memory usage in a streaming execution plan. You may see
1395 /// a performance benefit when writing large parquet files
1396 /// by increasing maximum_parallel_row_group_writers and
1397 /// maximum_buffered_record_batches_per_stream if your system
1398 /// has idle cores and can tolerate additional memory usage.
1399 /// Boosting these values is likely worthwhile when
1400 /// writing out already in-memory data, such as from a cached
1401 /// data frame.
1402 pub maximum_parallel_row_group_writers: usize, default = 1
1403
1404 /// (writing) By default parallel parquet writer is tuned for minimum
1405 /// memory usage in a streaming execution plan. You may see
1406 /// a performance benefit when writing large parquet files
1407 /// by increasing maximum_parallel_row_group_writers and
1408 /// maximum_buffered_record_batches_per_stream if your system
1409 /// has idle cores and can tolerate additional memory usage.
1410 /// Boosting these values is likely worthwhile when
1411 /// writing out already in-memory data, such as from a cached
1412 /// data frame.
1413 pub maximum_buffered_record_batches_per_stream: usize, default = 2
1414
1415 /// (writing) EXPERIMENTAL: Content-defined chunking (CDC) options when writing
1416 /// parquet files. Disabled by default; toggle with
1417 /// `content_defined_chunking.enabled = true|false`. The chunking parameters live
1418 /// under the same prefix (e.g. `content_defined_chunking.min_chunk_size`). When
1419 /// enabled, parallel writing is automatically disabled since the chunker state
1420 /// must persist across row groups. Mirrors
1421 /// `parquet::file::properties::WriterProperties::content_defined_chunking`.
1422 pub content_defined_chunking: ParquetCdcOptions, default = Default::default()
1423 }
1424}
1425
1426config_namespace! {
1427 /// Options for configuring Parquet Modular Encryption
1428 ///
1429 /// To use Parquet encryption, you must enable the `parquet_encryption` feature flag, as it is not activated by default.
1430 pub struct ParquetEncryptionOptions {
1431 /// Optional file decryption properties
1432 pub file_decryption: Option<ConfigFileDecryptionProperties>, default = None
1433
1434 /// Optional file encryption properties
1435 pub file_encryption: Option<ConfigFileEncryptionProperties>, default = None
1436
1437 /// Identifier for the encryption factory to use to create file encryption and decryption properties.
1438 /// Encryption factories can be registered in the runtime environment with
1439 /// `RuntimeEnv::register_parquet_encryption_factory`.
1440 pub factory_id: Option<String>, default = None
1441
1442 /// Any encryption factory specific options
1443 pub factory_options: EncryptionFactoryOptions, default = EncryptionFactoryOptions::default()
1444 }
1445}
1446
1447impl ParquetEncryptionOptions {
1448 /// Specify the encryption factory to use for Parquet modular encryption, along with its configuration
1449 pub fn configure_factory(
1450 &mut self,
1451 factory_id: &str,
1452 config: &impl ExtensionOptions,
1453 ) {
1454 self.factory_id = Some(factory_id.to_owned());
1455 self.factory_options.options.clear();
1456 for entry in config.entries() {
1457 if let Some(value) = entry.value {
1458 self.factory_options.options.insert(entry.key, value);
1459 }
1460 }
1461 }
1462}
1463
1464config_namespace! {
1465 /// Options related to query optimization
1466 ///
1467 /// See also: [`SessionConfig`]
1468 ///
1469 /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
1470 pub struct OptimizerOptions {
1471 /// When set to true, the optimizer will push a limit operation into
1472 /// grouped aggregations which have no aggregate expressions, as a soft limit,
1473 /// emitting groups once the limit is reached, before all rows in the group are read.
1474 pub enable_distinct_aggregation_soft_limit: bool, default = true
1475
1476 /// When set to true, the physical plan optimizer will try to add round robin
1477 /// repartitioning to increase parallelism to leverage more CPU cores
1478 pub enable_round_robin_repartition: bool, default = true
1479
1480 /// When set to true, the optimizer will attempt to perform limit operations
1481 /// during aggregations, if possible
1482 pub enable_topk_aggregation: bool, default = true
1483
1484 /// When set to true, the optimizer will attempt to push limit operations
1485 /// past window functions, if possible
1486 pub enable_window_limits: bool, default = true
1487
1488 /// When set to true, the optimizer will replace
1489 /// Filter(rn<=K) → Window(ROW_NUMBER) → Sort patterns with a
1490 /// PartitionedTopKExec that maintains per-partition heaps, avoiding
1491 /// a full sort of the input.
1492 /// When the window partition key has low cardinality, enabling this optimization
1493 /// can improve performance. However, for high cardinality keys, it may
1494 /// cause regressions in both memory usage and runtime.
1495 pub enable_window_topn: bool, default = false
1496
1497 /// When set to true, the optimizer will push TopK (Sort with fetch)
1498 /// below hash repartition when the partition key is a prefix of the
1499 /// sort key, reducing data volume before the shuffle.
1500 pub enable_topk_repartition: bool, default = true
1501
1502 /// When set to true, the optimizer will attempt to push down TopK dynamic filters
1503 /// into the file scan phase.
1504 pub enable_topk_dynamic_filter_pushdown: bool, default = true
1505
1506 /// When set to true, uncorrelated scalar subqueries are
1507 /// left in the logical plan and executed by `ScalarSubqueryExec` during
1508 /// physical execution. When set to false, all scalar subqueries
1509 /// (including uncorrelated ones) are rewritten to left joins by the
1510 /// `ScalarSubqueryToJoin` optimizer rule.
1511 ///
1512 /// Note disabling this option is not recommended. It restores
1513 /// pre <https://github.com/apache/datafusion/pull/21240>
1514 /// behavior, which silently produces incorrect results for
1515 /// multi-row subqueries and does not support scalar subqueries in
1516 /// ORDER BY / JOIN ON / aggregate-function arguments. This option is
1517 /// intended as a temporary escape hatch for distributed execution
1518 /// frameworks and is planned to be removed in a future DataFusion
1519 /// release.
1520 pub enable_physical_uncorrelated_scalar_subquery: bool, default = true
1521
1522 /// When set to true, the optimizer will attempt to push down Join dynamic filters
1523 /// into the file scan phase.
1524 pub enable_join_dynamic_filter_pushdown: bool, default = true
1525
1526 /// When set to true, the optimizer will attempt to push down Aggregate dynamic filters
1527 /// into the file scan phase.
1528 pub enable_aggregate_dynamic_filter_pushdown: bool, default = true
1529
1530 /// When set to true attempts to push down dynamic filters generated by operators (TopK, Join & Aggregate) into the file scan phase.
1531 /// For example, for a query such as `SELECT * FROM t ORDER BY timestamp DESC LIMIT 10`, the optimizer
1532 /// will attempt to push down the current top 10 timestamps that the TopK operator references into the file scans.
1533 /// This means that if we already have 10 timestamps in the year 2025
1534 /// any files that only have timestamps in the year 2024 can be skipped / pruned at various stages in the scan.
1535 /// The config will suppress `enable_join_dynamic_filter_pushdown`, `enable_topk_dynamic_filter_pushdown` & `enable_aggregate_dynamic_filter_pushdown`
1536 /// So if you disable `enable_topk_dynamic_filter_pushdown`, then enable `enable_dynamic_filter_pushdown`, the `enable_topk_dynamic_filter_pushdown` will be overridden.
1537 pub enable_dynamic_filter_pushdown: bool, default = true
1538
1539 /// When set to true, the optimizer will insert filters before a join between
1540 /// a nullable and non-nullable column to filter out nulls on the nullable side. This
1541 /// filter can add additional overhead when the file format does not fully support
1542 /// predicate push down.
1543 pub filter_null_join_keys: bool, default = false
1544
1545 /// Should DataFusion repartition data using the aggregate keys to execute aggregates
1546 /// in parallel using the provided `target_partitions` level
1547 pub repartition_aggregations: bool, default = true
1548
1549 /// Minimum total file size in bytes for file-group byte-range
1550 /// splitting to fire. Files (or merged file groups) smaller than this
1551 /// stay as one partition. Lower values produce more, smaller
1552 /// partitions — better at filling `target_partitions` worth of cores
1553 /// when files are modestly sized, at the cost of slightly more
1554 /// per-partition open / metadata-load overhead.
1555 pub repartition_file_min_size: usize, default = 1024 * 1024
1556
1557 /// Should DataFusion repartition data using the join keys to execute joins in parallel
1558 /// using the provided `target_partitions` level
1559 pub repartition_joins: bool, default = true
1560
1561 /// Should DataFusion allow symmetric hash joins for unbounded data sources even when
1562 /// its inputs do not have any ordering or filtering If the flag is not enabled,
1563 /// the SymmetricHashJoin operator will be unable to prune its internal buffers,
1564 /// resulting in certain join types - such as Full, Left, LeftAnti, LeftSemi, Right,
1565 /// RightAnti, and RightSemi - being produced only at the end of the execution.
1566 /// This is not typical in stream processing. Additionally, without proper design for
1567 /// long runner execution, all types of joins may encounter out-of-memory errors.
1568 pub allow_symmetric_joins_without_pruning: bool, default = true
1569
1570 /// When set to `true`, datasource partitions will be repartitioned to achieve maximum parallelism.
1571 /// This applies to both in-memory partitions and FileSource's file groups (1 group is 1 partition).
1572 ///
1573 /// For FileSources, only Parquet and CSV formats are currently supported.
1574 ///
1575 /// If set to `true` for a FileSource, all files will be repartitioned evenly (i.e., a single large file
1576 /// might be partitioned into smaller chunks) for parallel scanning.
1577 /// If set to `false` for a FileSource, different files will be read in parallel, but repartitioning won't
1578 /// happen within a single file.
1579 ///
1580 /// If set to `true` for an in-memory source, all memtable's partitions will have their batches
1581 /// repartitioned evenly to the desired number of `target_partitions`. Repartitioning can change
1582 /// the total number of partitions and batches per partition, but does not slice the initial
1583 /// record tables provided to the MemTable on creation.
1584 pub repartition_file_scans: bool, default = true
1585
1586 /// Minimum number of distinct partition values required to group files by their
1587 /// Hive partition column values (enabling output partitioning declaration).
1588 ///
1589 /// How the option is used:
1590 /// - preserve_file_partitions=0: Disable it.
1591 /// - preserve_file_partitions=1: Always enable it.
1592 /// - preserve_file_partitions=N, actual file partitions=M: Only enable when M >= N.
1593 /// This threshold preserves I/O parallelism when file partitioning is below it.
1594 ///
1595 /// Note: This may reduce parallelism, rooting from the I/O level, if the number of distinct
1596 /// partitions is less than the target_partitions.
1597 pub preserve_file_partitions: usize, default = 0
1598
1599 /// Should DataFusion repartition data using the partitions keys to execute window
1600 /// functions in parallel using the provided `target_partitions` level
1601 pub repartition_windows: bool, default = true
1602
1603 /// Should DataFusion execute sorts in a per-partition fashion and merge
1604 /// afterwards instead of coalescing first and sorting globally.
1605 /// With this flag is enabled, plans in the form below
1606 ///
1607 /// ```text
1608 /// "SortExec: [a@0 ASC]",
1609 /// " CoalescePartitionsExec",
1610 /// " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
1611 /// ```
1612 /// would turn into the plan below which performs better in multithreaded environments
1613 ///
1614 /// ```text
1615 /// "SortPreservingMergeExec: [a@0 ASC]",
1616 /// " SortExec: [a@0 ASC]",
1617 /// " RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
1618 /// ```
1619 pub repartition_sorts: bool, default = true
1620
1621 /// Partition count threshold for subset satisfaction optimization.
1622 ///
1623 /// When the current partition count is >= this threshold, DataFusion will
1624 /// skip repartitioning if the required partitioning expression is a subset
1625 /// of the current partition expression such as Hash(a) satisfies Hash(a, b).
1626 ///
1627 /// When the current partition count is < this threshold, DataFusion will
1628 /// repartition to increase parallelism even when subset satisfaction applies.
1629 ///
1630 /// Set to 0 to always repartition (disable subset satisfaction optimization).
1631 /// Set to a high value to always use subset satisfaction.
1632 ///
1633 /// Example (subset_repartition_threshold = 4):
1634 /// ```text
1635 /// Hash([a]) satisfies Hash([a, b]) because (Hash([a, b]) is subset of Hash([a])
1636 ///
1637 /// If current partitions (3) < threshold (4), repartition:
1638 /// AggregateExec: mode=FinalPartitioned, gby=[a, b], aggr=[SUM(x)]
1639 /// RepartitionExec: partitioning=Hash([a, b], 8), input_partitions=3
1640 /// AggregateExec: mode=Partial, gby=[a, b], aggr=[SUM(x)]
1641 /// DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 3)
1642 ///
1643 /// If current partitions (8) >= threshold (4), use subset satisfaction:
1644 /// AggregateExec: mode=SinglePartitioned, gby=[a, b], aggr=[SUM(x)]
1645 /// DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 8)
1646 /// ```
1647 pub subset_repartition_threshold: usize, default = 4
1648
1649 /// When true, DataFusion will opportunistically remove sorts when the data is already sorted,
1650 /// (i.e. setting `preserve_order` to true on `RepartitionExec` and
1651 /// using `SortPreservingMergeExec`)
1652 ///
1653 /// When false, DataFusion will maximize plan parallelism using
1654 /// `RepartitionExec` even if this requires subsequently resorting data using a `SortExec`.
1655 pub prefer_existing_sort: bool, default = false
1656
1657 /// When set to true, the logical plan optimizer will produce warning
1658 /// messages if any optimization rules produce errors and then proceed to the next
1659 /// rule. When set to false, any rules that produce errors will cause the query to fail
1660 pub skip_failed_rules: bool, default = false
1661
1662 /// Number of times that the optimizer will attempt to optimize the plan
1663 pub max_passes: usize, default = 3
1664
1665 /// When set to true, the physical plan optimizer will run a top down
1666 /// process to reorder the join keys
1667 pub top_down_join_key_reordering: bool, default = true
1668
1669 /// When set to true, the physical plan optimizer may swap join inputs
1670 /// based on statistics. When set to false, statistics-driven join
1671 /// input reordering is disabled and the original join order in the
1672 /// query is used.
1673 pub join_reordering: bool, default = true
1674
1675 /// When set to true, the physical plan optimizer uses the pluggable
1676 /// `StatisticsRegistry` for statistics propagation across operators.
1677 /// This enables more accurate cardinality estimates compared to each
1678 /// operator's built-in `partition_statistics`.
1679 pub use_statistics_registry: bool, default = false
1680
1681 /// When set to true, the physical plan optimizer will prefer HashJoin over SortMergeJoin.
1682 /// HashJoin can work more efficiently than SortMergeJoin but consumes more memory
1683 pub prefer_hash_join: bool, default = true
1684
1685 /// When set to true, piecewise merge join is enabled. PiecewiseMergeJoin is currently
1686 /// experimental. Physical planner will opt for PiecewiseMergeJoin when there is only
1687 /// one range filter.
1688 pub enable_piecewise_merge_join: bool, default = false
1689
1690 /// The maximum estimated size in bytes for one input side of a HashJoin
1691 /// will be collected into a single partition
1692 pub hash_join_single_partition_threshold: usize, default = 1024 * 1024
1693
1694 /// The maximum estimated size in rows for one input side of a HashJoin
1695 /// will be collected into a single partition
1696 pub hash_join_single_partition_threshold_rows: usize, default = 1024 * 128
1697
1698 /// Maximum size in bytes for the build side of a hash join to be pushed down as an InList expression for dynamic filtering.
1699 /// Build sides larger than this will use hash table lookups instead.
1700 /// Set to 0 to always use hash table lookups.
1701 ///
1702 /// InList pushdown can be more efficient for small build sides because it can result in better
1703 /// statistics pruning as well as use any bloom filters present on the scan side.
1704 /// InList expressions are also more transparent and easier to serialize over the network in distributed uses of DataFusion.
1705 /// On the other hand InList pushdown requires making a copy of the data and thus adds some overhead to the build side and uses more memory.
1706 ///
1707 /// This setting is per-partition, so we may end up using `hash_join_inlist_pushdown_max_size` * `target_partitions` memory.
1708 ///
1709 /// The default is 128kB per partition.
1710 /// This should allow point lookup joins (e.g. joining on a unique primary key) to use InList pushdown in most cases
1711 /// but avoids excessive memory usage or overhead for larger joins.
1712 pub hash_join_inlist_pushdown_max_size: usize, default = 128 * 1024
1713
1714 /// Maximum number of distinct values (rows) in the build side of a hash join to be pushed down as an InList expression for dynamic filtering.
1715 /// Build sides with more rows than this will use hash table lookups instead.
1716 /// Set to 0 to always use hash table lookups.
1717 ///
1718 /// This provides an additional limit beyond `hash_join_inlist_pushdown_max_size` to prevent
1719 /// very large IN lists that might not provide much benefit over hash table lookups.
1720 ///
1721 /// This uses the deduplicated row count once the build side has been evaluated.
1722 ///
1723 /// The default is 150 values per partition.
1724 /// This is inspired by Trino's `max-filter-keys-per-column` setting.
1725 /// See: <https://trino.io/docs/current/admin/dynamic-filtering.html#dynamic-filter-collection-thresholds>
1726 pub hash_join_inlist_pushdown_max_distinct_values: usize, default = 150
1727
1728 /// The default filter selectivity used by Filter Statistics
1729 /// when an exact selectivity cannot be determined. Valid values are
1730 /// between 0 (no selectivity) and 100 (all rows are selected).
1731 pub default_filter_selectivity: u8, default = 20
1732
1733 /// When set to true, the optimizer will not attempt to convert Union to Interleave
1734 pub prefer_existing_union: bool, default = false
1735
1736 /// When set to true, if the returned type is a view type
1737 /// then the output will be coerced to a non-view.
1738 /// Coerces `Utf8View` to `LargeUtf8`, and `BinaryView` to `LargeBinary`.
1739 pub expand_views_at_output: bool, default = false
1740
1741 /// Enable sort pushdown optimization.
1742 /// When enabled, attempts to push sort requirements down to data sources
1743 /// that can natively handle them (e.g., by reversing file/row group read order).
1744 ///
1745 /// Returns **inexact ordering**: Sort operator is kept for correctness,
1746 /// but optimized input enables early termination for TopK queries (ORDER BY ... LIMIT N),
1747 /// providing significant speedup.
1748 ///
1749 /// Memory: No additional overhead (only changes read order).
1750 ///
1751 /// Future: Will add option to detect perfectly sorted data and eliminate Sort completely.
1752 ///
1753 /// Default: true
1754 pub enable_sort_pushdown: bool, default = true
1755
1756 /// When set to true, the optimizer will extract leaf expressions
1757 /// (such as `get_field`) from filter/sort/join nodes into projections
1758 /// closer to the leaf table scans, and push those projections down
1759 /// towards the leaf nodes.
1760 pub enable_leaf_expression_pushdown: bool, default = true
1761
1762 /// When set to true, the logical optimizer will rewrite `UNION DISTINCT` branches that
1763 /// read from the same source and differ only by filter predicates into a single branch
1764 /// with a combined filter. This optimization is conservative and only applies when the
1765 /// branches share the same source and compatible wrapper nodes such as identical
1766 /// projections or aliases.
1767 pub enable_unions_to_filter: bool, default = false
1768 }
1769}
1770
1771config_namespace! {
1772 /// Options controlling explain output
1773 ///
1774 /// See also: [`SessionConfig`]
1775 ///
1776 /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
1777 pub struct ExplainOptions {
1778 /// When set to true, the explain statement will only print logical plans
1779 pub logical_plan_only: bool, default = false
1780
1781 /// When set to true, the explain statement will only print physical plans
1782 pub physical_plan_only: bool, default = false
1783
1784 /// When set to true, the explain statement will print operator statistics
1785 /// for physical plans
1786 pub show_statistics: bool, default = false
1787
1788 /// When set to true, the explain statement will print the partition sizes
1789 pub show_sizes: bool, default = true
1790
1791 /// When set to true, the explain statement will print schema information
1792 pub show_schema: bool, default = false
1793
1794 /// Display format of explain. Default is "indent".
1795 /// When set to "tree", it will print the plan in a tree-rendered format.
1796 pub format: ExplainFormat, default = ExplainFormat::Indent
1797
1798 /// (format=tree only) Maximum total width of the rendered tree.
1799 /// When set to 0, the tree will have no width limit.
1800 pub tree_maximum_render_width: usize, default = 240
1801
1802 /// Verbosity level for "EXPLAIN ANALYZE". Default is "dev"
1803 /// "summary" shows common metrics for high-level insights.
1804 /// "dev" provides deep operator-level introspection for developers.
1805 pub analyze_level: MetricType, default = MetricType::Dev
1806
1807 /// Which metric categories to include in "EXPLAIN ANALYZE" output.
1808 /// Comma-separated list of: "rows", "bytes", "timing", "uncategorized".
1809 /// Use "none" to show plan structure only, or "all" (default) to show everything.
1810 /// Metrics without a declared category are treated as "uncategorized".
1811 pub analyze_categories: ExplainAnalyzeCategories, default = ExplainAnalyzeCategories::All
1812 }
1813}
1814
1815impl ExecutionOptions {
1816 /// Returns the correct parallelism based on the provided `value`.
1817 /// If `value` is `"0"`, returns the default available parallelism, computed with
1818 /// `get_available_parallelism`. Otherwise, returns `value`.
1819 fn normalized_parallelism(value: &str) -> String {
1820 if value.parse::<usize>() == Ok(0) {
1821 get_available_parallelism().to_string()
1822 } else {
1823 value.to_owned()
1824 }
1825 }
1826}
1827
1828config_namespace! {
1829 /// Options controlling the format of output when printing record batches
1830 /// Copies [`arrow::util::display::FormatOptions`]
1831 pub struct FormatOptions {
1832 /// If set to `true` any formatting errors will be written to the output
1833 /// instead of being converted into a [`std::fmt::Error`]
1834 pub safe: bool, default = true
1835 /// Format string for nulls
1836 pub null: String, default = "".into()
1837 /// Date format for date arrays
1838 pub date_format: Option<String>, default = Some("%Y-%m-%d".to_string())
1839 /// Format for DateTime arrays
1840 pub datetime_format: Option<String>, default = Some("%Y-%m-%dT%H:%M:%S%.f".to_string())
1841 /// Timestamp format for timestamp arrays
1842 pub timestamp_format: Option<String>, default = Some("%Y-%m-%dT%H:%M:%S%.f".to_string())
1843 /// Timestamp format for timestamp with timezone arrays. When `None`, ISO 8601 format is used.
1844 pub timestamp_tz_format: Option<String>, default = None
1845 /// Time format for time arrays
1846 pub time_format: Option<String>, default = Some("%H:%M:%S%.f".to_string())
1847 /// Duration format. Can be either `"pretty"` or `"ISO8601"`
1848 pub duration_format: String, transform = str::to_lowercase, default = "pretty".into()
1849 /// Show types in visual representation batches
1850 pub types_info: bool, default = false
1851 }
1852}
1853
1854impl<'a> TryFrom<&'a FormatOptions> for arrow::util::display::FormatOptions<'a> {
1855 type Error = DataFusionError;
1856 fn try_from(options: &'a FormatOptions) -> Result<Self> {
1857 let duration_format = match options.duration_format.as_str() {
1858 "pretty" => arrow::util::display::DurationFormat::Pretty,
1859 "iso8601" => arrow::util::display::DurationFormat::ISO8601,
1860 _ => {
1861 return _config_err!(
1862 "Invalid duration format: {}. Valid values are pretty or iso8601",
1863 options.duration_format
1864 );
1865 }
1866 };
1867
1868 Ok(Self::new()
1869 .with_display_error(options.safe)
1870 .with_null(&options.null)
1871 .with_date_format(options.date_format.as_deref())
1872 .with_datetime_format(options.datetime_format.as_deref())
1873 .with_timestamp_format(options.timestamp_format.as_deref())
1874 .with_timestamp_tz_format(options.timestamp_tz_format.as_deref())
1875 .with_time_format(options.time_format.as_deref())
1876 .with_duration_format(duration_format)
1877 .with_types_info(options.types_info))
1878 }
1879}
1880
1881config_namespace! {
1882 /// Options controlling DataFusion's Spark-compatibility layer (functions
1883 /// under `datafusion/spark`). Keys here mirror their `spark.sql.*`
1884 /// equivalents in Apache Spark.
1885 pub struct SparkOptions {
1886 /// Policy for handling duplicate keys in Spark-compatible map-construction
1887 /// functions (`map_from_arrays`, `map_from_entries`, `str_to_map`).
1888 ///
1889 /// Mirrors Spark's
1890 /// [`spark.sql.mapKeyDedupPolicy`](https://github.com/apache/spark/blob/cf3a34e19dfcf70e2d679217ff1ba21302212472/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L4961):
1891 /// - `EXCEPTION` (default): raise `[DUPLICATED_MAP_KEY]` at runtime on any duplicate key.
1892 /// - `LAST_WIN`: keep the last occurrence of each duplicate key.
1893 ///
1894 /// Values are case-insensitive.
1895 pub map_key_dedup_policy: MapKeyDedupPolicy, default = MapKeyDedupPolicy::Exception
1896 }
1897}
1898
1899/// A key value pair, with a corresponding description
1900#[derive(Debug, Clone, Hash, PartialEq, Eq)]
1901pub struct ConfigEntry {
1902 /// A unique string to identify this config value
1903 pub key: String,
1904
1905 /// The value if any
1906 pub value: Option<String>,
1907
1908 /// A description of this configuration entry
1909 pub description: &'static str,
1910}
1911
1912/// Configuration options struct, able to store both built-in configuration and custom options
1913#[derive(Debug, Clone, Default)]
1914#[non_exhaustive]
1915pub struct ConfigOptions {
1916 /// Catalog options
1917 pub catalog: CatalogOptions,
1918 /// Execution options
1919 pub execution: ExecutionOptions,
1920 /// Optimizer options
1921 pub optimizer: OptimizerOptions,
1922 /// SQL parser options
1923 pub sql_parser: SqlParserOptions,
1924 /// Explain options
1925 pub explain: ExplainOptions,
1926 /// Optional extensions registered using [`Extensions::insert`]
1927 pub extensions: Extensions,
1928 /// Formatting options when printing batches
1929 pub format: FormatOptions,
1930 /// Spark-compatibility options (functions under `datafusion/spark`)
1931 pub spark: SparkOptions,
1932}
1933
1934impl ConfigField for ConfigOptions {
1935 fn visit<V: Visit>(&self, v: &mut V, _key_prefix: &str, _description: &'static str) {
1936 self.catalog.visit(v, "datafusion.catalog", "");
1937 self.execution.visit(v, "datafusion.execution", "");
1938 self.optimizer.visit(v, "datafusion.optimizer", "");
1939 self.explain.visit(v, "datafusion.explain", "");
1940 self.sql_parser.visit(v, "datafusion.sql_parser", "");
1941 self.format.visit(v, "datafusion.format", "");
1942 self.spark.visit(v, "datafusion.spark", "");
1943 }
1944
1945 fn set(&mut self, key: &str, value: &str) -> Result<()> {
1946 // Extensions are handled in the public `ConfigOptions::set`
1947 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
1948 match key {
1949 "catalog" => self.catalog.set(rem, value),
1950 "execution" => self.execution.set(rem, value),
1951 "optimizer" => self.optimizer.set(rem, value),
1952 "explain" => self.explain.set(rem, value),
1953 "sql_parser" => self.sql_parser.set(rem, value),
1954 "format" => self.format.set(rem, value),
1955 "spark" => self.spark.set(rem, value),
1956 _ => _config_err!("Config value \"{key}\" not found on ConfigOptions"),
1957 }
1958 }
1959
1960 /// Reset a configuration option back to its default value
1961 fn reset(&mut self, key: &str) -> Result<()> {
1962 let Some((prefix, rest)) = key.split_once('.') else {
1963 return _config_err!("could not find config namespace for key \"{key}\"");
1964 };
1965
1966 if prefix != "datafusion" {
1967 return _config_err!("Could not find config namespace \"{prefix}\"");
1968 }
1969
1970 let (section, rem) = rest.split_once('.').unwrap_or((rest, ""));
1971 if rem.is_empty() {
1972 return _config_err!("could not find config field for key \"{key}\"");
1973 }
1974
1975 match section {
1976 "catalog" => self.catalog.reset(rem),
1977 "execution" => self.execution.reset(rem),
1978 "optimizer" => {
1979 if rem == "enable_dynamic_filter_pushdown" {
1980 let defaults = OptimizerOptions::default();
1981 self.optimizer.enable_dynamic_filter_pushdown =
1982 defaults.enable_dynamic_filter_pushdown;
1983 self.optimizer.enable_topk_dynamic_filter_pushdown =
1984 defaults.enable_topk_dynamic_filter_pushdown;
1985 self.optimizer.enable_join_dynamic_filter_pushdown =
1986 defaults.enable_join_dynamic_filter_pushdown;
1987 Ok(())
1988 } else {
1989 self.optimizer.reset(rem)
1990 }
1991 }
1992 "explain" => self.explain.reset(rem),
1993 "sql_parser" => self.sql_parser.reset(rem),
1994 "format" => self.format.reset(rem),
1995 "spark" => self.spark.reset(rem),
1996 other => _config_err!("Config value \"{other}\" not found on ConfigOptions"),
1997 }
1998 }
1999}
2000
2001/// This namespace is reserved for interacting with Foreign Function Interface
2002/// (FFI) based configuration extensions.
2003pub const DATAFUSION_FFI_CONFIG_NAMESPACE: &str = "datafusion_ffi";
2004
2005impl ConfigOptions {
2006 /// Creates a new [`ConfigOptions`] with default values
2007 pub fn new() -> Self {
2008 Self::default()
2009 }
2010
2011 /// Set extensions to provided value
2012 pub fn with_extensions(mut self, extensions: Extensions) -> Self {
2013 self.extensions = extensions;
2014 self
2015 }
2016
2017 /// Set a configuration option
2018 pub fn set(&mut self, key: &str, value: &str) -> Result<()> {
2019 let Some((mut prefix, mut inner_key)) = key.split_once('.') else {
2020 return _config_err!("could not find config namespace for key \"{key}\"");
2021 };
2022
2023 if prefix == "datafusion" {
2024 if inner_key == "optimizer.enable_dynamic_filter_pushdown" {
2025 let bool_value = value.parse::<bool>().map_err(|e| {
2026 DataFusionError::Configuration(format!(
2027 "Failed to parse '{value}' as bool: {e}",
2028 ))
2029 })?;
2030
2031 {
2032 self.optimizer.enable_dynamic_filter_pushdown = bool_value;
2033 self.optimizer.enable_topk_dynamic_filter_pushdown = bool_value;
2034 self.optimizer.enable_join_dynamic_filter_pushdown = bool_value;
2035 self.optimizer.enable_aggregate_dynamic_filter_pushdown = bool_value;
2036 }
2037 return Ok(());
2038 }
2039 return ConfigField::set(self, inner_key, value)
2040 .map_err(|e| e.context(format!("Error setting config {key}")));
2041 }
2042
2043 if !self.extensions.0.contains_key(prefix)
2044 && self
2045 .extensions
2046 .0
2047 .contains_key(DATAFUSION_FFI_CONFIG_NAMESPACE)
2048 {
2049 inner_key = key;
2050 prefix = DATAFUSION_FFI_CONFIG_NAMESPACE;
2051 }
2052
2053 let Some(e) = self.extensions.0.get_mut(prefix) else {
2054 return _config_err!("Could not find config namespace \"{prefix}\"");
2055 };
2056 e.0.set(inner_key, value)
2057 }
2058
2059 /// Create new [`ConfigOptions`], taking values from environment variables
2060 /// where possible.
2061 ///
2062 /// For example, to configure `datafusion.execution.batch_size`
2063 /// ([`ExecutionOptions::batch_size`]) you would set the
2064 /// `DATAFUSION_EXECUTION_BATCH_SIZE` environment variable.
2065 ///
2066 /// The name of the environment variable is the option's key, transformed to
2067 /// uppercase and with periods replaced with underscores.
2068 ///
2069 /// Values are parsed according to the [same rules used in casts from
2070 /// Utf8](https://docs.rs/arrow/latest/arrow/compute/kernels/cast/fn.cast.html).
2071 ///
2072 /// If the value in the environment variable cannot be cast to the type of
2073 /// the configuration option, the default value will be used instead and a
2074 /// warning emitted. Environment variables are read when this method is
2075 /// called, and are not re-read later.
2076 pub fn from_env() -> Result<Self> {
2077 struct Visitor(Vec<String>);
2078
2079 impl Visit for Visitor {
2080 fn some<V: Display>(&mut self, key: &str, _: V, _: &'static str) {
2081 self.0.push(key.to_string())
2082 }
2083
2084 fn none(&mut self, key: &str, _: &'static str) {
2085 self.0.push(key.to_string())
2086 }
2087 }
2088
2089 // Extract the names of all fields and then look up the corresponding
2090 // environment variables. This isn't hugely efficient but avoids
2091 // ambiguity between `a.b` and `a_b` which would both correspond
2092 // to an environment variable of `A_B`
2093
2094 let mut keys = Visitor(vec![]);
2095 let mut ret = Self::default();
2096 ret.visit(&mut keys, "datafusion", "");
2097
2098 for key in keys.0 {
2099 let env = key.to_uppercase().replace('.', "_");
2100 if let Some(var) = std::env::var_os(env) {
2101 let value = var.to_string_lossy();
2102 log::info!("Set {key} to {value} from the environment variable");
2103 ret.set(&key, value.as_ref())?;
2104 }
2105 }
2106
2107 Ok(ret)
2108 }
2109
2110 /// Create new ConfigOptions struct, taking values from a string hash map.
2111 ///
2112 /// Only the built-in configurations will be extracted from the hash map
2113 /// and other key value pairs will be ignored.
2114 pub fn from_string_hash_map(settings: &HashMap<String, String>) -> Result<Self> {
2115 struct Visitor(Vec<String>);
2116
2117 impl Visit for Visitor {
2118 fn some<V: Display>(&mut self, key: &str, _: V, _: &'static str) {
2119 self.0.push(key.to_string())
2120 }
2121
2122 fn none(&mut self, key: &str, _: &'static str) {
2123 self.0.push(key.to_string())
2124 }
2125 }
2126
2127 let mut keys = Visitor(vec![]);
2128 let mut ret = Self::default();
2129 ret.visit(&mut keys, "datafusion", "");
2130
2131 for key in keys.0 {
2132 if let Some(var) = settings.get(&key) {
2133 ret.set(&key, var)?;
2134 }
2135 }
2136
2137 Ok(ret)
2138 }
2139
2140 /// Returns the [`ConfigEntry`] stored within this [`ConfigOptions`]
2141 pub fn entries(&self) -> Vec<ConfigEntry> {
2142 struct Visitor(Vec<ConfigEntry>);
2143
2144 impl Visit for Visitor {
2145 fn some<V: Display>(
2146 &mut self,
2147 key: &str,
2148 value: V,
2149 description: &'static str,
2150 ) {
2151 self.0.push(ConfigEntry {
2152 key: key.to_string(),
2153 value: Some(value.to_string()),
2154 description,
2155 })
2156 }
2157
2158 fn none(&mut self, key: &str, description: &'static str) {
2159 self.0.push(ConfigEntry {
2160 key: key.to_string(),
2161 value: None,
2162 description,
2163 })
2164 }
2165 }
2166
2167 let mut v = Visitor(vec![]);
2168 self.visit(&mut v, "datafusion", "");
2169
2170 v.0.extend(self.extensions.0.values().flat_map(|e| e.0.entries()));
2171 v.0
2172 }
2173
2174 /// Generate documentation that can be included in the user guide
2175 pub fn generate_config_markdown() -> String {
2176 use std::fmt::Write as _;
2177
2178 let mut s = Self::default();
2179
2180 // Normalize for display
2181 s.execution.target_partitions = 0;
2182 s.execution.planning_concurrency = 0;
2183
2184 let mut docs = "| key | default | description |\n".to_string();
2185 docs += "|-----|---------|-------------|\n";
2186 let mut entries = s.entries();
2187 entries.sort_unstable_by(|a, b| a.key.cmp(&b.key));
2188
2189 for entry in s.entries() {
2190 let _ = writeln!(
2191 &mut docs,
2192 "| {} | {} | {} |",
2193 entry.key,
2194 entry.value.as_deref().unwrap_or("NULL"),
2195 entry.description
2196 );
2197 }
2198 docs
2199 }
2200}
2201
2202/// [`ConfigExtension`] provides a mechanism to store third-party configuration
2203/// within DataFusion [`ConfigOptions`]
2204///
2205/// This mechanism can be used to pass configuration to user defined functions
2206/// or optimizer passes
2207///
2208/// # Example
2209/// ```
2210/// use datafusion_common::{
2211/// config::ConfigExtension, config::ConfigOptions, extensions_options,
2212/// };
2213/// // Define a new configuration struct using the `extensions_options` macro
2214/// extensions_options! {
2215/// /// My own config options.
2216/// pub struct MyConfig {
2217/// /// Should "foo" be replaced by "bar"?
2218/// pub foo_to_bar: bool, default = true
2219///
2220/// /// How many "baz" should be created?
2221/// pub baz_count: usize, default = 1337
2222/// }
2223/// }
2224///
2225/// impl ConfigExtension for MyConfig {
2226/// const PREFIX: &'static str = "my_config";
2227/// }
2228///
2229/// // set up config struct and register extension
2230/// let mut config = ConfigOptions::default();
2231/// config.extensions.insert(MyConfig::default());
2232///
2233/// // overwrite config default
2234/// config.set("my_config.baz_count", "42").unwrap();
2235///
2236/// // check config state
2237/// let my_config = config.extensions.get::<MyConfig>().unwrap();
2238/// assert!(my_config.foo_to_bar,);
2239/// assert_eq!(my_config.baz_count, 42,);
2240/// ```
2241///
2242/// # Note:
2243/// Unfortunately associated constants are not currently object-safe, and so this
2244/// extends the object-safe [`ExtensionOptions`]
2245pub trait ConfigExtension: ExtensionOptions {
2246 /// Configuration namespace prefix to use
2247 ///
2248 /// All values under this will be prefixed with `$PREFIX + "."`
2249 const PREFIX: &'static str;
2250}
2251
2252/// An object-safe API for storing arbitrary configuration.
2253///
2254/// See [`ConfigExtension`] for user defined configuration
2255pub trait ExtensionOptions: Send + Sync + fmt::Debug + 'static {
2256 /// Return `self` as [`Any`]
2257 ///
2258 /// This is needed until trait upcasting is stabilized
2259 fn as_any(&self) -> &dyn Any;
2260
2261 /// Return `self` as [`Any`]
2262 ///
2263 /// This is needed until trait upcasting is stabilized
2264 fn as_any_mut(&mut self) -> &mut dyn Any;
2265
2266 /// Return a deep clone of this [`ExtensionOptions`]
2267 ///
2268 /// It is important this does not share mutable state to avoid consistency issues
2269 /// with configuration changing whilst queries are executing
2270 fn cloned(&self) -> Box<dyn ExtensionOptions>;
2271
2272 /// Set the given `key`, `value` pair
2273 fn set(&mut self, key: &str, value: &str) -> Result<()>;
2274
2275 /// Returns the [`ConfigEntry`] stored in this [`ExtensionOptions`]
2276 fn entries(&self) -> Vec<ConfigEntry>;
2277}
2278
2279/// A type-safe container for [`ConfigExtension`]
2280#[derive(Debug, Default, Clone)]
2281pub struct Extensions(BTreeMap<&'static str, ExtensionBox>);
2282
2283impl Extensions {
2284 /// Create a new, empty [`Extensions`]
2285 pub fn new() -> Self {
2286 Self(BTreeMap::new())
2287 }
2288
2289 /// Registers a [`ConfigExtension`] with this [`ConfigOptions`]
2290 pub fn insert<T: ConfigExtension>(&mut self, extension: T) {
2291 assert_ne!(T::PREFIX, "datafusion");
2292 let e = ExtensionBox(Box::new(extension));
2293 self.0.insert(T::PREFIX, e);
2294 }
2295
2296 /// Retrieves the extension of the given type if any
2297 pub fn get<T: ConfigExtension>(&self) -> Option<&T> {
2298 self.0.get(T::PREFIX)?.0.as_any().downcast_ref()
2299 }
2300
2301 /// Retrieves the extension of the given type if any
2302 pub fn get_mut<T: ConfigExtension>(&mut self) -> Option<&mut T> {
2303 let e = self.0.get_mut(T::PREFIX)?;
2304 e.0.as_any_mut().downcast_mut()
2305 }
2306
2307 /// Iterates all the config extension entries yielding their prefix and their
2308 /// [ExtensionOptions] implementation.
2309 pub fn iter(
2310 &self,
2311 ) -> impl Iterator<Item = (&'static str, &Box<dyn ExtensionOptions>)> {
2312 self.0.iter().map(|(k, v)| (*k, &v.0))
2313 }
2314}
2315
2316#[derive(Debug)]
2317struct ExtensionBox(Box<dyn ExtensionOptions>);
2318
2319impl Clone for ExtensionBox {
2320 fn clone(&self) -> Self {
2321 Self(self.0.cloned())
2322 }
2323}
2324
2325/// A trait implemented by `config_namespace` and for field types that provides
2326/// the ability to walk and mutate the configuration tree
2327pub trait ConfigField {
2328 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str);
2329
2330 fn set(&mut self, key: &str, value: &str) -> Result<()>;
2331
2332 fn reset(&mut self, key: &str) -> Result<()> {
2333 _config_err!("Reset is not supported for this config field, key: {}", key)
2334 }
2335}
2336
2337impl<F: ConfigField + Default> ConfigField for Option<F> {
2338 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
2339 match self {
2340 Some(s) => s.visit(v, key, description),
2341 None => v.none(key, description),
2342 }
2343 }
2344
2345 fn set(&mut self, key: &str, value: &str) -> Result<()> {
2346 self.get_or_insert_with(Default::default).set(key, value)
2347 }
2348
2349 fn reset(&mut self, key: &str) -> Result<()> {
2350 if key.is_empty() {
2351 *self = Default::default();
2352 Ok(())
2353 } else {
2354 self.get_or_insert_with(Default::default).reset(key)
2355 }
2356 }
2357}
2358
2359/// Default transformation to parse a [`ConfigField`] for a string.
2360///
2361/// This uses [`FromStr`] to parse the data.
2362pub fn default_config_transform<T>(input: &str) -> Result<T>
2363where
2364 T: FromStr,
2365 <T as FromStr>::Err: Sync + Send + Error + 'static,
2366{
2367 input.parse().map_err(|e| {
2368 DataFusionError::Context(
2369 format!(
2370 "Error parsing '{}' as {}",
2371 input,
2372 std::any::type_name::<T>()
2373 ),
2374 Box::new(DataFusionError::External(Box::new(e))),
2375 )
2376 })
2377}
2378
2379/// Macro that generates [`ConfigField`] for a given type.
2380///
2381/// # Usage
2382/// This always requires [`Display`] to be implemented for the given type.
2383///
2384/// There are two ways to invoke this macro. The first one uses
2385/// [`default_config_transform`]/[`FromStr`] to parse the data:
2386///
2387/// ```ignore
2388/// config_field(MyType);
2389/// ```
2390///
2391/// Note that the parsing error MUST implement [`std::error::Error`]!
2392///
2393/// Or you can specify how you want to parse an [`str`] into the type:
2394///
2395/// ```ignore
2396/// fn parse_it(s: &str) -> Result<MyType> {
2397/// ...
2398/// }
2399///
2400/// config_field(
2401/// MyType,
2402/// value => parse_it(value)
2403/// );
2404/// ```
2405#[macro_export]
2406macro_rules! config_field {
2407 ($t:ty) => {
2408 config_field!($t, value => $crate::config::default_config_transform(value)?);
2409 };
2410
2411 ($t:ty, $arg:ident => $transform:expr) => {
2412 impl $crate::config::ConfigField for $t {
2413 fn visit<V: $crate::config::Visit>(&self, v: &mut V, key: &str, description: &'static str) {
2414 v.some(key, self, description)
2415 }
2416
2417 fn set(&mut self, _: &str, $arg: &str) -> $crate::error::Result<()> {
2418 *self = $transform;
2419 Ok(())
2420 }
2421
2422 fn reset(&mut self, key: &str) -> $crate::error::Result<()> {
2423 if key.is_empty() {
2424 *self = <$t as Default>::default();
2425 Ok(())
2426 } else {
2427 $crate::error::_config_err!(
2428 "Config field is a scalar {} and does not have nested field \"{}\"",
2429 stringify!($t),
2430 key
2431 )
2432 }
2433 }
2434 }
2435 };
2436}
2437
2438config_field!(String);
2439config_field!(bool, value => default_config_transform(value.to_lowercase().as_str())?);
2440config_field!(usize);
2441config_field!(f64);
2442config_field!(u64);
2443config_field!(u32);
2444config_field!(i32);
2445
2446impl ConfigField for u8 {
2447 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
2448 v.some(key, self, description)
2449 }
2450
2451 fn set(&mut self, key: &str, value: &str) -> Result<()> {
2452 if value.is_empty() {
2453 return Err(DataFusionError::Configuration(format!(
2454 "Input string for {key} key is empty"
2455 )));
2456 }
2457 // Check if the string is a valid number
2458 if let Ok(num) = value.parse::<u8>() {
2459 // TODO: Let's decide how we treat the numerical strings.
2460 *self = num;
2461 } else {
2462 let bytes = value.as_bytes();
2463 // Check if the first character is ASCII (single byte)
2464 if bytes.len() > 1 || !value.chars().next().unwrap().is_ascii() {
2465 return Err(DataFusionError::Configuration(format!(
2466 "Error parsing {value} as u8. Non-ASCII string provided"
2467 )));
2468 }
2469 *self = bytes[0];
2470 }
2471 Ok(())
2472 }
2473}
2474
2475impl ConfigField for CompressionTypeVariant {
2476 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
2477 v.some(key, self, description)
2478 }
2479
2480 fn set(&mut self, _: &str, value: &str) -> Result<()> {
2481 *self = CompressionTypeVariant::from_str(value)?;
2482 Ok(())
2483 }
2484}
2485
2486impl ConfigField for CsvQuoteStyle {
2487 fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
2488 v.some(key, self, description)
2489 }
2490
2491 fn set(&mut self, _: &str, value: &str) -> Result<()> {
2492 *self = CsvQuoteStyle::from_str(value)?;
2493 Ok(())
2494 }
2495}
2496
2497/// An implementation trait used to recursively walk configuration
2498pub trait Visit {
2499 fn some<V: Display>(&mut self, key: &str, value: V, description: &'static str);
2500
2501 fn none(&mut self, key: &str, description: &'static str);
2502}
2503
2504/// Convenience macro to create [`ExtensionsOptions`].
2505///
2506/// The created structure implements the following traits:
2507///
2508/// - [`Clone`]
2509/// - [`Debug`]
2510/// - [`Default`]
2511/// - [`ExtensionOptions`]
2512///
2513/// # Usage
2514/// The syntax is:
2515///
2516/// ```text
2517/// extensions_options! {
2518/// /// Struct docs (optional).
2519/// [<vis>] struct <StructName> {
2520/// /// Field docs (optional)
2521/// [<vis>] <field_name>: <field_type>, default = <default_value>
2522///
2523/// ... more fields
2524/// }
2525/// }
2526/// ```
2527///
2528/// The placeholders are:
2529/// - `[<vis>]`: Optional visibility modifier like `pub` or `pub(crate)`.
2530/// - `<StructName>`: Struct name like `MyStruct`.
2531/// - `<field_name>`: Field name like `my_field`.
2532/// - `<field_type>`: Field type like `u8`.
2533/// - `<default_value>`: Default value matching the field type like `42`.
2534///
2535/// # Example
2536/// See also a full example on the [`ConfigExtension`] documentation
2537///
2538/// ```
2539/// use datafusion_common::extensions_options;
2540///
2541/// extensions_options! {
2542/// /// My own config options.
2543/// pub struct MyConfig {
2544/// /// Should "foo" be replaced by "bar"?
2545/// pub foo_to_bar: bool, default = true
2546///
2547/// /// How many "baz" should be created?
2548/// pub baz_count: usize, default = 1337
2549/// }
2550/// }
2551/// ```
2552///
2553///
2554/// [`Debug`]: std::fmt::Debug
2555/// [`ExtensionsOptions`]: crate::config::ExtensionOptions
2556#[macro_export]
2557macro_rules! extensions_options {
2558 (
2559 $(#[doc = $struct_d:tt])*
2560 $vis:vis struct $struct_name:ident {
2561 $(
2562 $(#[doc = $d:tt])*
2563 $field_vis:vis $field_name:ident : $field_type:ty, default = $default:expr
2564 )*$(,)*
2565 }
2566 ) => {
2567 $(#[doc = $struct_d])*
2568 #[derive(Debug, Clone)]
2569 #[non_exhaustive]
2570 $vis struct $struct_name{
2571 $(
2572 $(#[doc = $d])*
2573 $field_vis $field_name : $field_type,
2574 )*
2575 }
2576
2577 impl Default for $struct_name {
2578 fn default() -> Self {
2579 Self {
2580 $($field_name: $default),*
2581 }
2582 }
2583 }
2584
2585 impl $crate::config::ExtensionOptions for $struct_name {
2586 fn as_any(&self) -> &dyn ::std::any::Any {
2587 self
2588 }
2589
2590 fn as_any_mut(&mut self) -> &mut dyn ::std::any::Any {
2591 self
2592 }
2593
2594 fn cloned(&self) -> Box<dyn $crate::config::ExtensionOptions> {
2595 Box::new(self.clone())
2596 }
2597
2598 fn set(&mut self, key: &str, value: &str) -> $crate::error::Result<()> {
2599 $crate::config::ConfigField::set(self, key, value)
2600 }
2601
2602 fn entries(&self) -> Vec<$crate::config::ConfigEntry> {
2603 struct Visitor(Vec<$crate::config::ConfigEntry>);
2604
2605 impl $crate::config::Visit for Visitor {
2606 fn some<V: std::fmt::Display>(
2607 &mut self,
2608 key: &str,
2609 value: V,
2610 description: &'static str,
2611 ) {
2612 self.0.push($crate::config::ConfigEntry {
2613 key: key.to_string(),
2614 value: Some(value.to_string()),
2615 description,
2616 })
2617 }
2618
2619 fn none(&mut self, key: &str, description: &'static str) {
2620 self.0.push($crate::config::ConfigEntry {
2621 key: key.to_string(),
2622 value: None,
2623 description,
2624 })
2625 }
2626 }
2627
2628 let mut v = Visitor(vec![]);
2629 // The prefix is not used for extensions.
2630 // The description is generated in ConfigField::visit.
2631 // We can just pass empty strings here.
2632 $crate::config::ConfigField::visit(self, &mut v, "", "");
2633 v.0
2634 }
2635 }
2636
2637 impl $crate::config::ConfigField for $struct_name {
2638 fn set(&mut self, key: &str, value: &str) -> $crate::error::Result<()> {
2639 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
2640 match key {
2641 $(
2642 stringify!($field_name) => {
2643 // Safely apply deprecated attribute if present
2644 // $(#[allow(deprecated)])?
2645 {
2646 self.$field_name.set(rem, value.as_ref())
2647 }
2648 },
2649 )*
2650 _ => return $crate::error::_config_err!(
2651 "Config value \"{}\" not found on {}", key, stringify!($struct_name)
2652 )
2653 }
2654 }
2655
2656 fn visit<V: $crate::config::Visit>(&self, v: &mut V, _key_prefix: &str, _description: &'static str) {
2657 $(
2658 let key = stringify!($field_name).to_string();
2659 let desc = concat!($($d),*).trim();
2660 self.$field_name.visit(v, key.as_str(), desc);
2661 )*
2662 }
2663 }
2664 }
2665}
2666
2667/// These file types have special built in behavior for configuration.
2668/// Use TableOptions::Extensions for configuring other file types.
2669#[derive(Debug, Clone)]
2670pub enum ConfigFileType {
2671 CSV,
2672 #[cfg(feature = "parquet")]
2673 PARQUET,
2674 JSON,
2675}
2676
2677/// Represents the configuration options available for handling different table formats within a data processing application.
2678/// This struct encompasses options for various file formats including CSV, Parquet, and JSON, allowing for flexible configuration
2679/// of parsing and writing behaviors specific to each format. Additionally, it supports extending functionality through custom extensions.
2680#[derive(Debug, Clone, Default)]
2681pub struct TableOptions {
2682 /// Configuration options for CSV file handling. This includes settings like the delimiter,
2683 /// quote character, and whether the first row is considered as headers.
2684 pub csv: CsvOptions,
2685
2686 /// Configuration options for Parquet file handling. This includes settings for compression,
2687 /// encoding, and other Parquet-specific file characteristics.
2688 pub parquet: TableParquetOptions,
2689
2690 /// Configuration options for JSON file handling.
2691 pub json: JsonOptions,
2692
2693 /// The current file format that the table operations should assume. This option allows
2694 /// for dynamic switching between the supported file types (e.g., CSV, Parquet, JSON).
2695 pub current_format: Option<ConfigFileType>,
2696
2697 /// Optional extensions that can be used to extend or customize the behavior of the table
2698 /// options. Extensions can be registered using `Extensions::insert` and might include
2699 /// custom file handling logic, additional configuration parameters, or other enhancements.
2700 pub extensions: Extensions,
2701}
2702
2703impl ConfigField for TableOptions {
2704 /// Visits configuration settings for the current file format, or all formats if none is selected.
2705 ///
2706 /// This method adapts the behavior based on whether a file format is currently selected in `current_format`.
2707 /// If a format is selected, it visits only the settings relevant to that format. Otherwise,
2708 /// it visits all available format settings.
2709 fn visit<V: Visit>(&self, v: &mut V, _key_prefix: &str, _description: &'static str) {
2710 if let Some(file_type) = &self.current_format {
2711 match file_type {
2712 #[cfg(feature = "parquet")]
2713 ConfigFileType::PARQUET => self.parquet.visit(v, "format", ""),
2714 ConfigFileType::CSV => self.csv.visit(v, "format", ""),
2715 ConfigFileType::JSON => self.json.visit(v, "format", ""),
2716 }
2717 } else {
2718 self.csv.visit(v, "csv", "");
2719 self.parquet.visit(v, "parquet", "");
2720 self.json.visit(v, "json", "");
2721 }
2722 }
2723
2724 /// Sets a configuration value for a specific key within `TableOptions`.
2725 ///
2726 /// This method delegates setting configuration values to the specific file format configurations,
2727 /// based on the current format selected. If no format is selected, it returns an error.
2728 ///
2729 /// # Parameters
2730 ///
2731 /// * `key`: The configuration key specifying which setting to adjust, prefixed with the format (e.g., "format.delimiter")
2732 /// for CSV format.
2733 /// * `value`: The value to set for the specified configuration key.
2734 ///
2735 /// # Returns
2736 ///
2737 /// A result indicating success or an error if the key is not recognized, if a format is not specified,
2738 /// or if setting the configuration value fails for the specific format.
2739 fn set(&mut self, key: &str, value: &str) -> Result<()> {
2740 // Extensions are handled in the public `ConfigOptions::set`
2741 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
2742 match key {
2743 "format" => {
2744 let Some(format) = &self.current_format else {
2745 return _config_err!("Specify a format for TableOptions");
2746 };
2747 match format {
2748 #[cfg(feature = "parquet")]
2749 ConfigFileType::PARQUET => self.parquet.set(rem, value),
2750 ConfigFileType::CSV => self.csv.set(rem, value),
2751 ConfigFileType::JSON => self.json.set(rem, value),
2752 }
2753 }
2754 _ => _config_err!("Config value \"{key}\" not found on TableOptions"),
2755 }
2756 }
2757}
2758
2759impl TableOptions {
2760 /// Constructs a new instance of `TableOptions` with default settings.
2761 ///
2762 /// # Returns
2763 ///
2764 /// A new `TableOptions` instance with default configuration values.
2765 pub fn new() -> Self {
2766 Self::default()
2767 }
2768
2769 /// Creates a new `TableOptions` instance initialized with settings from a given session config.
2770 ///
2771 /// # Parameters
2772 ///
2773 /// * `config`: A reference to the session `ConfigOptions` from which to derive initial settings.
2774 ///
2775 /// # Returns
2776 ///
2777 /// A new `TableOptions` instance with settings applied from the session config.
2778 pub fn default_from_session_config(config: &ConfigOptions) -> Self {
2779 let initial = TableOptions::default();
2780 initial.combine_with_session_config(config)
2781 }
2782
2783 /// Updates the current `TableOptions` with settings from a given session config.
2784 ///
2785 /// # Parameters
2786 ///
2787 /// * `config`: A reference to the session `ConfigOptions` whose settings are to be applied.
2788 ///
2789 /// # Returns
2790 ///
2791 /// A new `TableOptions` instance with updated settings from the session config.
2792 #[must_use = "this method returns a new instance"]
2793 pub fn combine_with_session_config(&self, config: &ConfigOptions) -> Self {
2794 let mut clone = self.clone();
2795 clone.parquet.global = config.execution.parquet.clone();
2796 clone
2797 }
2798
2799 /// Sets the file format for the table.
2800 ///
2801 /// # Parameters
2802 ///
2803 /// * `format`: The file format to use (e.g., CSV, Parquet).
2804 pub fn set_config_format(&mut self, format: ConfigFileType) {
2805 self.current_format = Some(format);
2806 }
2807
2808 /// Sets the extensions for this `TableOptions` instance.
2809 ///
2810 /// # Parameters
2811 ///
2812 /// * `extensions`: The `Extensions` instance to set.
2813 ///
2814 /// # Returns
2815 ///
2816 /// A new `TableOptions` instance with the specified extensions applied.
2817 pub fn with_extensions(mut self, extensions: Extensions) -> Self {
2818 self.extensions = extensions;
2819 self
2820 }
2821
2822 /// Sets a specific configuration option.
2823 ///
2824 /// # Parameters
2825 ///
2826 /// * `key`: The configuration key (e.g., "format.delimiter").
2827 /// * `value`: The value to set for the specified key.
2828 ///
2829 /// # Returns
2830 ///
2831 /// A result indicating success or failure in setting the configuration option.
2832 pub fn set(&mut self, key: &str, value: &str) -> Result<()> {
2833 let Some((mut prefix, _)) = key.split_once('.') else {
2834 return _config_err!("could not find config namespace for key \"{key}\"");
2835 };
2836
2837 if prefix == "format" {
2838 return ConfigField::set(self, key, value);
2839 }
2840
2841 if prefix == "execution" {
2842 return Ok(());
2843 }
2844
2845 if !self.extensions.0.contains_key(prefix)
2846 && self
2847 .extensions
2848 .0
2849 .contains_key(DATAFUSION_FFI_CONFIG_NAMESPACE)
2850 {
2851 prefix = DATAFUSION_FFI_CONFIG_NAMESPACE;
2852 }
2853
2854 let Some(e) = self.extensions.0.get_mut(prefix) else {
2855 return _config_err!("Could not find config namespace \"{prefix}\"");
2856 };
2857 e.0.set(key, value)
2858 }
2859
2860 /// Initializes a new `TableOptions` from a hash map of string settings.
2861 ///
2862 /// # Parameters
2863 ///
2864 /// * `settings`: A hash map where each key-value pair represents a configuration setting.
2865 ///
2866 /// # Returns
2867 ///
2868 /// A result containing the new `TableOptions` instance or an error if any setting could not be applied.
2869 pub fn from_string_hash_map(settings: &HashMap<String, String>) -> Result<Self> {
2870 let mut ret = Self::default();
2871 for (k, v) in settings {
2872 ret.set(k, v)?;
2873 }
2874
2875 Ok(ret)
2876 }
2877
2878 /// Modifies the current `TableOptions` instance with settings from a hash map.
2879 ///
2880 /// # Parameters
2881 ///
2882 /// * `settings`: A hash map where each key-value pair represents a configuration setting.
2883 ///
2884 /// # Returns
2885 ///
2886 /// A result indicating success or failure in applying the settings.
2887 pub fn alter_with_string_hash_map(
2888 &mut self,
2889 settings: &HashMap<String, String>,
2890 ) -> Result<()> {
2891 for (k, v) in settings {
2892 self.set(k, v)?;
2893 }
2894 Ok(())
2895 }
2896
2897 /// Retrieves all configuration entries from this `TableOptions`.
2898 ///
2899 /// # Returns
2900 ///
2901 /// A vector of `ConfigEntry` instances, representing all the configuration options within this `TableOptions`.
2902 pub fn entries(&self) -> Vec<ConfigEntry> {
2903 struct Visitor(Vec<ConfigEntry>);
2904
2905 impl Visit for Visitor {
2906 fn some<V: Display>(
2907 &mut self,
2908 key: &str,
2909 value: V,
2910 description: &'static str,
2911 ) {
2912 self.0.push(ConfigEntry {
2913 key: key.to_string(),
2914 value: Some(value.to_string()),
2915 description,
2916 })
2917 }
2918
2919 fn none(&mut self, key: &str, description: &'static str) {
2920 self.0.push(ConfigEntry {
2921 key: key.to_string(),
2922 value: None,
2923 description,
2924 })
2925 }
2926 }
2927
2928 let mut v = Visitor(vec![]);
2929 self.visit(&mut v, "format", "");
2930
2931 v.0.extend(self.extensions.0.values().flat_map(|e| e.0.entries()));
2932 v.0
2933 }
2934}
2935
2936/// Options that control how Parquet files are read, including global options
2937/// that apply to all columns and optional column-specific overrides
2938///
2939/// Closely tied to `ParquetWriterOptions` (see `crate::file_options::parquet_writer::ParquetWriterOptions` when the "parquet" feature is enabled).
2940/// Properties not included in [`TableParquetOptions`] may not be configurable at the external API
2941/// (e.g. sorting_columns).
2942#[derive(Clone, Default, Debug, PartialEq)]
2943pub struct TableParquetOptions {
2944 /// Global Parquet options that propagates to all columns.
2945 pub global: ParquetOptions,
2946 /// Column specific options. Default usage is parquet.XX::column.
2947 pub column_specific_options: HashMap<String, ParquetColumnOptions>,
2948 /// Additional file-level metadata to include. Inserted into the key_value_metadata
2949 /// for the written [`FileMetaData`](https://docs.rs/parquet/latest/parquet/file/metadata/struct.FileMetaData.html).
2950 ///
2951 /// Multiple entries are permitted
2952 /// ```sql
2953 /// OPTIONS (
2954 /// 'format.metadata::key1' '',
2955 /// 'format.metadata::key2' 'value',
2956 /// 'format.metadata::key3' 'value has spaces',
2957 /// 'format.metadata::key4' 'value has special chars :: :',
2958 /// 'format.metadata::key_dupe' 'original will be overwritten',
2959 /// 'format.metadata::key_dupe' 'final'
2960 /// )
2961 /// ```
2962 pub key_value_metadata: HashMap<String, Option<String>>,
2963 /// Options for configuring Parquet modular encryption
2964 ///
2965 /// To use Parquet encryption, you must enable the `parquet_encryption` feature flag, as it is not activated by default.
2966 /// See ConfigFileEncryptionProperties and ConfigFileDecryptionProperties in datafusion/common/src/config.rs
2967 /// These can be set via 'format.crypto', for example:
2968 /// ```sql
2969 /// OPTIONS (
2970 /// 'format.crypto.file_encryption.encrypt_footer' 'true',
2971 /// 'format.crypto.file_encryption.footer_key_as_hex' '30313233343536373839303132333435', -- b"0123456789012345" */
2972 /// 'format.crypto.file_encryption.column_key_as_hex::double_field' '31323334353637383930313233343530', -- b"1234567890123450"
2973 /// 'format.crypto.file_encryption.column_key_as_hex::float_field' '31323334353637383930313233343531', -- b"1234567890123451"
2974 /// -- Same for decryption
2975 /// 'format.crypto.file_decryption.footer_key_as_hex' '30313233343536373839303132333435', -- b"0123456789012345"
2976 /// 'format.crypto.file_decryption.column_key_as_hex::double_field' '31323334353637383930313233343530', -- b"1234567890123450"
2977 /// 'format.crypto.file_decryption.column_key_as_hex::float_field' '31323334353637383930313233343531', -- b"1234567890123451"
2978 /// )
2979 /// ```
2980 /// See datafusion-cli/tests/sql/encrypted_parquet.sql for a more complete example.
2981 /// Note that keys must be provided as in hex format since these are binary strings.
2982 pub crypto: ParquetEncryptionOptions,
2983}
2984
2985impl TableParquetOptions {
2986 /// Return new default TableParquetOptions
2987 pub fn new() -> Self {
2988 Self::default()
2989 }
2990
2991 /// Set whether the encoding of the arrow metadata should occur
2992 /// during the writing of parquet.
2993 ///
2994 /// Default is to encode the arrow schema in the file kv_metadata.
2995 pub fn with_skip_arrow_metadata(self, skip: bool) -> Self {
2996 Self {
2997 global: ParquetOptions {
2998 skip_arrow_metadata: skip,
2999 ..self.global
3000 },
3001 ..self
3002 }
3003 }
3004
3005 /// Retrieves all configuration entries from this `TableParquetOptions`.
3006 ///
3007 /// # Returns
3008 ///
3009 /// A vector of `ConfigEntry` instances, representing all the configuration options within this
3010 pub fn entries(self: &TableParquetOptions) -> Vec<ConfigEntry> {
3011 struct Visitor(Vec<ConfigEntry>);
3012
3013 impl Visit for Visitor {
3014 fn some<V: Display>(
3015 &mut self,
3016 key: &str,
3017 value: V,
3018 description: &'static str,
3019 ) {
3020 self.0.push(ConfigEntry {
3021 key: key[1..].to_string(),
3022 value: Some(value.to_string()),
3023 description,
3024 })
3025 }
3026
3027 fn none(&mut self, key: &str, description: &'static str) {
3028 self.0.push(ConfigEntry {
3029 key: key[1..].to_string(),
3030 value: None,
3031 description,
3032 })
3033 }
3034 }
3035
3036 let mut v = Visitor(vec![]);
3037 self.visit(&mut v, "", "");
3038
3039 v.0
3040 }
3041}
3042
3043impl ConfigField for TableParquetOptions {
3044 fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, description: &'static str) {
3045 self.global.visit(v, key_prefix, description);
3046 self.column_specific_options
3047 .visit(v, key_prefix, description);
3048 self.crypto
3049 .visit(v, &format!("{key_prefix}.crypto"), description);
3050 }
3051
3052 fn set(&mut self, key: &str, value: &str) -> Result<()> {
3053 // Determine if the key is a global, metadata, or column-specific setting
3054 if key.starts_with("metadata::") {
3055 let k = match key.split("::").collect::<Vec<_>>()[..] {
3056 [_meta] | [_meta, ""] => {
3057 return _config_err!(
3058 "Invalid metadata key provided, missing key in metadata::<key>"
3059 );
3060 }
3061 [_meta, k] => k.into(),
3062 _ => {
3063 return _config_err!(
3064 "Invalid metadata key provided, found too many '::' in \"{key}\""
3065 );
3066 }
3067 };
3068 self.key_value_metadata.insert(k, Some(value.into()));
3069 Ok(())
3070 } else if let Some(crypto_feature) = key.strip_prefix("crypto.") {
3071 self.crypto.set(crypto_feature, value)
3072 } else if key.contains("::") {
3073 self.column_specific_options.set(key, value)
3074 } else {
3075 self.global.set(key, value)
3076 }
3077 }
3078}
3079
3080macro_rules! config_namespace_with_hashmap {
3081 (
3082 $(#[doc = $struct_d:tt])*
3083 $(#[deprecated($($struct_depr:tt)*)])? // Optional struct-level deprecated attribute
3084 $vis:vis struct $struct_name:ident {
3085 $(
3086 $(#[doc = $d:tt])*
3087 $(#[deprecated($($field_depr:tt)*)])? // Optional field-level deprecated attribute
3088 $field_vis:vis $field_name:ident : $field_type:ty, $(transform = $transform:expr,)? default = $default:expr
3089 )*$(,)*
3090 }
3091 ) => {
3092
3093 $(#[doc = $struct_d])*
3094 $(#[deprecated($($struct_depr)*)])? // Apply struct deprecation
3095 #[derive(Debug, Clone, PartialEq)]
3096 $vis struct $struct_name{
3097 $(
3098 $(#[doc = $d])*
3099 $(#[deprecated($($field_depr)*)])? // Apply field deprecation
3100 $field_vis $field_name : $field_type,
3101 )*
3102 }
3103
3104 impl ConfigField for $struct_name {
3105 fn set(&mut self, key: &str, value: &str) -> Result<()> {
3106 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
3107 match key {
3108 $(
3109 stringify!($field_name) => {
3110 // Handle deprecated fields
3111 $(let value = $transform(value);)?
3112 self.$field_name.set(rem, value.as_ref())
3113 },
3114 )*
3115 _ => _config_err!(
3116 "Config value \"{}\" not found on {}", key, stringify!($struct_name)
3117 )
3118 }
3119 }
3120
3121 fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
3122 $(
3123 let key = format!(concat!("{}.", stringify!($field_name)), key_prefix);
3124 let desc = concat!($($d),*).trim();
3125 // Handle deprecated fields
3126 self.$field_name.visit(v, key.as_str(), desc);
3127 )*
3128 }
3129 }
3130
3131 impl Default for $struct_name {
3132 fn default() -> Self {
3133 Self {
3134 $($field_name: $default),*
3135 }
3136 }
3137 }
3138
3139 impl ConfigField for HashMap<String,$struct_name> {
3140 fn set(&mut self, key: &str, value: &str) -> Result<()> {
3141 let parts: Vec<&str> = key.splitn(2, "::").collect();
3142 match parts.as_slice() {
3143 [inner_key, hashmap_key] => {
3144 // Get or create the struct for the specified key
3145 let inner_value = self
3146 .entry((*hashmap_key).to_owned())
3147 .or_insert_with($struct_name::default);
3148
3149 inner_value.set(inner_key, value)
3150 }
3151 _ => _config_err!("Unrecognized key '{key}'."),
3152 }
3153 }
3154
3155 fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
3156 for (column_name, col_options) in self {
3157 $(
3158 let key = format!("{}.{field}::{}", key_prefix, column_name, field = stringify!($field_name));
3159 let desc = concat!($($d),*).trim();
3160 col_options.$field_name.visit(v, key.as_str(), desc);
3161 )*
3162 }
3163 }
3164 }
3165 }
3166}
3167
3168config_namespace_with_hashmap! {
3169 /// Options controlling parquet format for individual columns.
3170 ///
3171 /// See [`ParquetOptions`] for more details
3172 pub struct ParquetColumnOptions {
3173 /// Sets if bloom filter is enabled for the column path.
3174 pub bloom_filter_enabled: Option<bool>, default = None
3175
3176 /// Sets encoding for the column path.
3177 /// Valid values are: plain, plain_dictionary, rle,
3178 /// bit_packed, delta_binary_packed, delta_length_byte_array,
3179 /// delta_byte_array, rle_dictionary, and byte_stream_split.
3180 /// These values are not case-sensitive. If NULL, uses
3181 /// default parquet options
3182 pub encoding: Option<String>, default = None
3183
3184 /// Sets if dictionary encoding is enabled for the column path. If NULL, uses
3185 /// default parquet options
3186 pub dictionary_enabled: Option<bool>, default = None
3187
3188 /// Sets default parquet compression codec for the column path.
3189 /// Valid values are: uncompressed, snappy, gzip(level),
3190 /// brotli(level), lz4, zstd(level), and lz4_raw.
3191 /// These values are not case-sensitive. If NULL, uses
3192 /// default parquet options
3193 pub compression: Option<String>, transform = str::to_lowercase, default = None
3194
3195 /// Sets if statistics are enabled for the column
3196 /// Valid values are: "none", "chunk", and "page"
3197 /// These values are not case sensitive. If NULL, uses
3198 /// default parquet options
3199 pub statistics_enabled: Option<String>, default = None
3200
3201 /// Sets bloom filter false positive probability for the column path. If NULL, uses
3202 /// default parquet options
3203 pub bloom_filter_fpp: Option<f64>, default = None
3204
3205 /// Sets bloom filter number of distinct values. If NULL, uses
3206 /// default parquet options
3207 pub bloom_filter_ndv: Option<u64>, default = None
3208 }
3209}
3210
3211#[derive(Clone, Debug, PartialEq)]
3212pub struct ConfigFileEncryptionProperties {
3213 /// Should the parquet footer be encrypted
3214 /// default is true
3215 pub encrypt_footer: bool,
3216 /// Key to use for the parquet footer encoded in hex format
3217 pub footer_key_as_hex: String,
3218 /// Metadata information for footer key
3219 pub footer_key_metadata_as_hex: String,
3220 /// HashMap of column names --> (key in hex format, metadata)
3221 pub column_encryption_properties: HashMap<String, ColumnEncryptionProperties>,
3222 /// AAD prefix string uniquely identifies the file and prevents file swapping
3223 pub aad_prefix_as_hex: String,
3224 /// If true, store the AAD prefix in the file
3225 /// default is false
3226 pub store_aad_prefix: bool,
3227}
3228
3229// Setup to match EncryptionPropertiesBuilder::new()
3230impl Default for ConfigFileEncryptionProperties {
3231 fn default() -> Self {
3232 ConfigFileEncryptionProperties {
3233 encrypt_footer: true,
3234 footer_key_as_hex: String::new(),
3235 footer_key_metadata_as_hex: String::new(),
3236 column_encryption_properties: Default::default(),
3237 aad_prefix_as_hex: String::new(),
3238 store_aad_prefix: false,
3239 }
3240 }
3241}
3242
3243config_namespace_with_hashmap! {
3244 pub struct ColumnEncryptionProperties {
3245 /// Per column encryption key
3246 pub column_key_as_hex: String, default = "".to_string()
3247 /// Per column encryption key metadata
3248 pub column_metadata_as_hex: Option<String>, default = None
3249 }
3250}
3251
3252impl ConfigField for ConfigFileEncryptionProperties {
3253 fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
3254 let key = format!("{key_prefix}.encrypt_footer");
3255 let desc = "Encrypt the footer";
3256 self.encrypt_footer.visit(v, key.as_str(), desc);
3257
3258 let key = format!("{key_prefix}.footer_key_as_hex");
3259 let desc = "Key to use for the parquet footer";
3260 self.footer_key_as_hex.visit(v, key.as_str(), desc);
3261
3262 let key = format!("{key_prefix}.footer_key_metadata_as_hex");
3263 let desc = "Metadata to use for the parquet footer";
3264 self.footer_key_metadata_as_hex.visit(v, key.as_str(), desc);
3265
3266 self.column_encryption_properties.visit(v, key_prefix, desc);
3267
3268 let key = format!("{key_prefix}.aad_prefix_as_hex");
3269 let desc = "AAD prefix to use";
3270 self.aad_prefix_as_hex.visit(v, key.as_str(), desc);
3271
3272 let key = format!("{key_prefix}.store_aad_prefix");
3273 let desc = "If true, store the AAD prefix";
3274 self.store_aad_prefix.visit(v, key.as_str(), desc);
3275
3276 self.aad_prefix_as_hex.visit(v, key.as_str(), desc);
3277 }
3278
3279 fn set(&mut self, key: &str, value: &str) -> Result<()> {
3280 // Any hex encoded values must be pre-encoded using
3281 // hex::encode() before calling set.
3282
3283 if key.contains("::") {
3284 // Handle any column specific properties
3285 return self.column_encryption_properties.set(key, value);
3286 };
3287
3288 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
3289 match key {
3290 "encrypt_footer" => self.encrypt_footer.set(rem, value.as_ref()),
3291 "footer_key_as_hex" => self.footer_key_as_hex.set(rem, value.as_ref()),
3292 "footer_key_metadata_as_hex" => {
3293 self.footer_key_metadata_as_hex.set(rem, value.as_ref())
3294 }
3295 "aad_prefix_as_hex" => self.aad_prefix_as_hex.set(rem, value.as_ref()),
3296 "store_aad_prefix" => self.store_aad_prefix.set(rem, value.as_ref()),
3297 _ => _config_err!(
3298 "Config value \"{}\" not found on ConfigFileEncryptionProperties",
3299 key
3300 ),
3301 }
3302 }
3303}
3304
3305#[cfg(feature = "parquet_encryption")]
3306impl TryFrom<ConfigFileEncryptionProperties> for FileEncryptionProperties {
3307 type Error = DataFusionError;
3308
3309 fn try_from(val: ConfigFileEncryptionProperties) -> Result<Self> {
3310 let mut fep = FileEncryptionProperties::builder(
3311 hex::decode(val.footer_key_as_hex)
3312 .map_err(|e| {
3313 DataFusionError::Configuration(format!("Unable to decode hex footer key from ConfigFileEncryptionProperties: {e}"))
3314 })?,
3315 )
3316 .with_plaintext_footer(!val.encrypt_footer)
3317 .with_aad_prefix_storage(val.store_aad_prefix);
3318
3319 if !val.footer_key_metadata_as_hex.is_empty() {
3320 fep = fep.with_footer_key_metadata(
3321 hex::decode(&val.footer_key_metadata_as_hex)
3322 .map_err(|e| {
3323 DataFusionError::Configuration(format!("Unable to decode hex footer key metadata from ConfigFileEncryptionProperties: {e}"))
3324 })?,
3325 );
3326 }
3327
3328 for (column_name, encryption_props) in val.column_encryption_properties.iter() {
3329 let encryption_key = hex::decode(&encryption_props.column_key_as_hex)
3330 .map_err(|e| {
3331 DataFusionError::Configuration(format!("Unable to decode hex encryption key for column {column_name}: {e}"))
3332 })?;
3333 let key_metadata = encryption_props
3334 .column_metadata_as_hex
3335 .as_ref()
3336 .map(hex::decode)
3337 .transpose()
3338 .map_err(|e| {
3339 DataFusionError::Configuration(format!("Unable to decode hex column metadata for column {column_name}: {e}"))
3340 })?;
3341
3342 match key_metadata {
3343 Some(key_metadata) => {
3344 fep = fep.with_column_key_and_metadata(
3345 column_name,
3346 encryption_key,
3347 key_metadata,
3348 );
3349 }
3350 None => {
3351 fep = fep.with_column_key(column_name, encryption_key);
3352 }
3353 }
3354 }
3355
3356 if !val.aad_prefix_as_hex.is_empty() {
3357 let aad_prefix: Vec<u8> = hex::decode(&val.aad_prefix_as_hex).map_err(|e| {
3358 DataFusionError::Configuration(format!(
3359 "Unable to decode hex AAD prefix from ConfigFileEncryptionProperties: {e}"
3360 ))
3361 })?;
3362 fep = fep.with_aad_prefix(aad_prefix);
3363 }
3364 Ok(Arc::unwrap_or_clone(fep.build().map_err(|e| {
3365 DataFusionError::Configuration(format!(
3366 "Could not build FileEncryptionProperties: {e}"
3367 ))
3368 })?))
3369 }
3370}
3371
3372#[cfg(feature = "parquet_encryption")]
3373impl From<&Arc<FileEncryptionProperties>> for ConfigFileEncryptionProperties {
3374 fn from(f: &Arc<FileEncryptionProperties>) -> Self {
3375 let (column_names_vec, column_keys_vec, column_metas_vec) = f.column_keys();
3376
3377 let mut column_encryption_properties: HashMap<
3378 String,
3379 ColumnEncryptionProperties,
3380 > = HashMap::new();
3381
3382 for (i, column_name) in column_names_vec.iter().enumerate() {
3383 let column_key_as_hex = hex::encode(&column_keys_vec[i]);
3384 let column_metadata_as_hex: Option<String> =
3385 column_metas_vec.get(i).map(hex::encode);
3386 column_encryption_properties.insert(
3387 column_name.clone(),
3388 ColumnEncryptionProperties {
3389 column_key_as_hex,
3390 column_metadata_as_hex,
3391 },
3392 );
3393 }
3394 let aad_prefix = f.aad_prefix().cloned().unwrap_or_default();
3395 ConfigFileEncryptionProperties {
3396 encrypt_footer: f.encrypt_footer(),
3397 footer_key_as_hex: hex::encode(f.footer_key()),
3398 footer_key_metadata_as_hex: f
3399 .footer_key_metadata()
3400 .map(hex::encode)
3401 .unwrap_or_default(),
3402 column_encryption_properties,
3403 aad_prefix_as_hex: hex::encode(aad_prefix),
3404 store_aad_prefix: f.store_aad_prefix(),
3405 }
3406 }
3407}
3408
3409#[derive(Clone, Debug, PartialEq)]
3410pub struct ConfigFileDecryptionProperties {
3411 /// Binary string to use for the parquet footer encoded in hex format
3412 pub footer_key_as_hex: String,
3413 /// HashMap of column names --> key in hex format
3414 pub column_decryption_properties: HashMap<String, ColumnDecryptionProperties>,
3415 /// AAD prefix string uniquely identifies the file and prevents file swapping
3416 pub aad_prefix_as_hex: String,
3417 /// If true, then verify signature for files with plaintext footers.
3418 /// default = true
3419 pub footer_signature_verification: bool,
3420}
3421
3422config_namespace_with_hashmap! {
3423 pub struct ColumnDecryptionProperties {
3424 /// Per column encryption key
3425 pub column_key_as_hex: String, default = "".to_string()
3426 }
3427}
3428
3429// Setup to match DecryptionPropertiesBuilder::new()
3430impl Default for ConfigFileDecryptionProperties {
3431 fn default() -> Self {
3432 ConfigFileDecryptionProperties {
3433 footer_key_as_hex: String::new(),
3434 column_decryption_properties: Default::default(),
3435 aad_prefix_as_hex: String::new(),
3436 footer_signature_verification: true,
3437 }
3438 }
3439}
3440
3441impl ConfigField for ConfigFileDecryptionProperties {
3442 fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
3443 let key = format!("{key_prefix}.footer_key_as_hex");
3444 let desc = "Key to use for the parquet footer";
3445 self.footer_key_as_hex.visit(v, key.as_str(), desc);
3446
3447 let key = format!("{key_prefix}.aad_prefix_as_hex");
3448 let desc = "AAD prefix to use";
3449 self.aad_prefix_as_hex.visit(v, key.as_str(), desc);
3450
3451 let key = format!("{key_prefix}.footer_signature_verification");
3452 let desc = "If true, verify the footer signature";
3453 self.footer_signature_verification
3454 .visit(v, key.as_str(), desc);
3455
3456 self.column_decryption_properties.visit(v, key_prefix, desc);
3457 }
3458
3459 fn set(&mut self, key: &str, value: &str) -> Result<()> {
3460 // Any hex encoded values must be pre-encoded using
3461 // hex::encode() before calling set.
3462
3463 if key.contains("::") {
3464 // Handle any column specific properties
3465 return self.column_decryption_properties.set(key, value);
3466 };
3467
3468 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
3469 match key {
3470 "footer_key_as_hex" => self.footer_key_as_hex.set(rem, value.as_ref()),
3471 "aad_prefix_as_hex" => self.aad_prefix_as_hex.set(rem, value.as_ref()),
3472 "footer_signature_verification" => {
3473 self.footer_signature_verification.set(rem, value.as_ref())
3474 }
3475 _ => _config_err!(
3476 "Config value \"{}\" not found on ConfigFileDecryptionProperties",
3477 key
3478 ),
3479 }
3480 }
3481}
3482
3483#[cfg(feature = "parquet_encryption")]
3484impl TryFrom<ConfigFileDecryptionProperties> for FileDecryptionProperties {
3485 type Error = DataFusionError;
3486
3487 fn try_from(val: ConfigFileDecryptionProperties) -> Result<Self> {
3488 let mut column_names: Vec<&str> = Vec::new();
3489 let mut column_keys: Vec<Vec<u8>> = Vec::new();
3490
3491 for (col_name, decryption_properties) in val.column_decryption_properties.iter() {
3492 let column_key = hex::decode(&decryption_properties.column_key_as_hex).map_err(|e| {
3493 DataFusionError::Configuration(format!
3494 ("Could not decode hex column key from ConfigFileDecryptionProperties for column name {col_name}: {e}."))
3495 })?;
3496 column_names.push(col_name.as_str());
3497 column_keys.push(column_key);
3498 }
3499
3500 let footer_key = hex::decode(val.footer_key_as_hex).map_err(|e| {
3501 DataFusionError::Configuration(format!(
3502 "Could not decode hex footer key from ConfigFileDecryptionProperties: {e}."
3503 ))
3504 })?;
3505
3506 let mut fep = FileDecryptionProperties::builder(footer_key)
3507 .with_column_keys(column_names, column_keys)
3508 .map_err(|e| {
3509 DataFusionError::Configuration(format!(
3510 "Could not set column keys on FileDecryptionPropertiesBuilder: {e}."
3511 ))
3512 })?;
3513
3514 if !val.footer_signature_verification {
3515 fep = fep.disable_footer_signature_verification();
3516 }
3517
3518 if !val.aad_prefix_as_hex.is_empty() {
3519 let aad_prefix = hex::decode(&val.aad_prefix_as_hex).map_err(|e| {
3520 DataFusionError::Configuration(format!(
3521 "Could not decode hex AAD prefix from ConfigFileDecryptionProperties: {e}."
3522 ))
3523 })?;
3524 fep = fep.with_aad_prefix(aad_prefix);
3525 }
3526
3527 Ok(Arc::unwrap_or_clone(fep.build().map_err(|e| {
3528 DataFusionError::Configuration(format!(
3529 "Could not build FileDecryptionProperties: {e}."
3530 ))
3531 })?))
3532 }
3533}
3534
3535#[cfg(feature = "parquet_encryption")]
3536impl TryFrom<&Arc<FileDecryptionProperties>> for ConfigFileDecryptionProperties {
3537 type Error = DataFusionError;
3538
3539 fn try_from(f: &Arc<FileDecryptionProperties>) -> Result<Self> {
3540 let footer_key = f.footer_key(None).map_err(|e| {
3541 DataFusionError::Configuration(format!(
3542 "Could not retrieve footer key from FileDecryptionProperties. \
3543 Note that conversion to ConfigFileDecryptionProperties is not supported \
3544 when using a key retriever: {e}"
3545 ))
3546 })?;
3547
3548 let (column_names_vec, column_keys_vec) = f.column_keys();
3549 let mut column_decryption_properties: HashMap<
3550 String,
3551 ColumnDecryptionProperties,
3552 > = HashMap::new();
3553 for (i, column_name) in column_names_vec.iter().enumerate() {
3554 let props = ColumnDecryptionProperties {
3555 column_key_as_hex: hex::encode(column_keys_vec[i].clone()),
3556 };
3557 column_decryption_properties.insert(column_name.clone(), props);
3558 }
3559
3560 let aad_prefix = f.aad_prefix().cloned().unwrap_or_default();
3561 Ok(ConfigFileDecryptionProperties {
3562 footer_key_as_hex: hex::encode(footer_key.as_ref()),
3563 column_decryption_properties,
3564 aad_prefix_as_hex: hex::encode(aad_prefix),
3565 footer_signature_verification: f.check_plaintext_footer_integrity(),
3566 })
3567 }
3568}
3569
3570/// Holds implementation-specific options for an encryption factory
3571#[derive(Clone, Debug, Default, PartialEq)]
3572pub struct EncryptionFactoryOptions {
3573 pub options: HashMap<String, String>,
3574}
3575
3576impl ConfigField for EncryptionFactoryOptions {
3577 fn visit<V: Visit>(&self, v: &mut V, key: &str, _description: &'static str) {
3578 for (option_key, option_value) in &self.options {
3579 v.some(
3580 &format!("{key}.{option_key}"),
3581 option_value,
3582 "Encryption factory specific option",
3583 );
3584 }
3585 }
3586
3587 fn set(&mut self, key: &str, value: &str) -> Result<()> {
3588 self.options.insert(key.to_owned(), value.to_owned());
3589 Ok(())
3590 }
3591}
3592
3593impl EncryptionFactoryOptions {
3594 /// Convert these encryption factory options to an [`ExtensionOptions`] instance.
3595 pub fn to_extension_options<T: ExtensionOptions + Default>(&self) -> Result<T> {
3596 let mut options = T::default();
3597 for (key, value) in &self.options {
3598 options.set(key, value)?;
3599 }
3600 Ok(options)
3601 }
3602}
3603
3604config_namespace! {
3605 /// Options controlling CSV format
3606 pub struct CsvOptions {
3607 /// Specifies whether there is a CSV header (i.e. the first line
3608 /// consists of is column names). The value `None` indicates that
3609 /// the configuration should be consulted.
3610 pub has_header: Option<bool>, default = None
3611 pub delimiter: u8, default = b','
3612 pub quote: u8, default = b'"'
3613 pub terminator: Option<u8>, default = None
3614 pub escape: Option<u8>, default = None
3615 pub double_quote: Option<bool>, default = None
3616 /// Quote style for CSV writing.
3617 /// One of: "Always", "Necessary", "NonNumeric", "Never"
3618 pub quote_style: CsvQuoteStyle, default = CsvQuoteStyle::Necessary
3619 /// Whether to ignore leading whitespace in string values when writing CSV.
3620 /// Defaults to `false` when `None`.
3621 pub ignore_leading_whitespace: Option<bool>, default = None
3622 /// Whether to ignore trailing whitespace in string values when writing CSV.
3623 /// Defaults to `false` when `None`.
3624 pub ignore_trailing_whitespace: Option<bool>, default = None
3625 /// Specifies whether newlines in (quoted) values are supported.
3626 ///
3627 /// Parsing newlines in quoted values may be affected by execution behaviour such as
3628 /// parallel file scanning. Setting this to `true` ensures that newlines in values are
3629 /// parsed successfully, which may reduce performance.
3630 ///
3631 /// The default behaviour depends on the `datafusion.catalog.newlines_in_values` setting.
3632 pub newlines_in_values: Option<bool>, default = None
3633 pub compression: CompressionTypeVariant, default = CompressionTypeVariant::UNCOMPRESSED
3634 /// Compression level for the output file. The valid range depends on the
3635 /// compression algorithm:
3636 /// - ZSTD: 1 to 22 (default: 3)
3637 /// - GZIP: 0 to 9 (default: 6)
3638 /// - BZIP2: 0 to 9 (default: 6)
3639 /// - XZ: 0 to 9 (default: 6)
3640 /// If not specified, the default level for the compression algorithm is used.
3641 pub compression_level: Option<u32>, default = None
3642 pub schema_infer_max_rec: Option<usize>, default = None
3643 pub date_format: Option<String>, default = None
3644 pub datetime_format: Option<String>, default = None
3645 pub timestamp_format: Option<String>, default = None
3646 pub timestamp_tz_format: Option<String>, default = None
3647 pub time_format: Option<String>, default = None
3648 // The output format for Nulls in the CSV writer.
3649 pub null_value: Option<String>, default = None
3650 // The input regex for Nulls when loading CSVs.
3651 pub null_regex: Option<String>, default = None
3652 pub comment: Option<u8>, default = None
3653 /// Whether to allow truncated rows when parsing, both within a single file and across files.
3654 ///
3655 /// When set to false (default), reading a single CSV file which has rows of different lengths will
3656 /// error; if reading multiple CSV files with different number of columns, it will also fail.
3657 ///
3658 /// When set to true, reading a single CSV file with rows of different lengths will pad the truncated
3659 /// rows with null values for the missing columns; if reading multiple CSV files with different number
3660 /// of columns, it creates a union schema containing all columns found across the files, and will
3661 /// pad any files missing columns with null values for their rows.
3662 pub truncated_rows: Option<bool>, default = None
3663 }
3664}
3665
3666impl CsvOptions {
3667 /// Set a limit in terms of records to scan to infer the schema
3668 /// - default to `DEFAULT_SCHEMA_INFER_MAX_RECORD`
3669 pub fn with_compression(
3670 mut self,
3671 compression_type_variant: CompressionTypeVariant,
3672 ) -> Self {
3673 self.compression = compression_type_variant;
3674 self
3675 }
3676
3677 /// Set a limit in terms of records to scan to infer the schema
3678 /// - default to `DEFAULT_SCHEMA_INFER_MAX_RECORD`
3679 pub fn with_schema_infer_max_rec(mut self, max_rec: usize) -> Self {
3680 self.schema_infer_max_rec = Some(max_rec);
3681 self
3682 }
3683
3684 /// Set true to indicate that the first line is a header.
3685 /// - default to true
3686 pub fn with_has_header(mut self, has_header: bool) -> Self {
3687 self.has_header = Some(has_header);
3688 self
3689 }
3690
3691 /// Returns true if the first line is a header. If format options does not
3692 /// specify whether there is a header, returns `None` (indicating that the
3693 /// configuration should be consulted).
3694 pub fn has_header(&self) -> Option<bool> {
3695 self.has_header
3696 }
3697
3698 /// The character separating values within a row.
3699 /// - default to ','
3700 pub fn with_delimiter(mut self, delimiter: u8) -> Self {
3701 self.delimiter = delimiter;
3702 self
3703 }
3704
3705 /// The quote character in a row.
3706 /// - default to '"'
3707 pub fn with_quote(mut self, quote: u8) -> Self {
3708 self.quote = quote;
3709 self
3710 }
3711
3712 /// The character that terminates a row.
3713 /// - default to None (CRLF)
3714 pub fn with_terminator(mut self, terminator: Option<u8>) -> Self {
3715 self.terminator = terminator;
3716 self
3717 }
3718
3719 /// The escape character in a row.
3720 /// - default is None
3721 pub fn with_escape(mut self, escape: Option<u8>) -> Self {
3722 self.escape = escape;
3723 self
3724 }
3725
3726 /// Set true to indicate that the CSV quotes should be doubled.
3727 /// - default to true
3728 pub fn with_double_quote(mut self, double_quote: bool) -> Self {
3729 self.double_quote = Some(double_quote);
3730 self
3731 }
3732
3733 /// Set the quote style for CSV writing.
3734 pub fn with_quote_style(mut self, quote_style: CsvQuoteStyle) -> Self {
3735 self.quote_style = quote_style;
3736 self
3737 }
3738
3739 /// Set whether to ignore leading whitespace in string values when writing CSV.
3740 pub fn with_ignore_leading_whitespace(
3741 mut self,
3742 ignore_leading_whitespace: bool,
3743 ) -> Self {
3744 self.ignore_leading_whitespace = Some(ignore_leading_whitespace);
3745 self
3746 }
3747
3748 /// Set whether to ignore trailing whitespace in string values when writing CSV.
3749 pub fn with_ignore_trailing_whitespace(
3750 mut self,
3751 ignore_trailing_whitespace: bool,
3752 ) -> Self {
3753 self.ignore_trailing_whitespace = Some(ignore_trailing_whitespace);
3754 self
3755 }
3756
3757 /// Specifies whether newlines in (quoted) values are supported.
3758 ///
3759 /// Parsing newlines in quoted values may be affected by execution behaviour such as
3760 /// parallel file scanning. Setting this to `true` ensures that newlines in values are
3761 /// parsed successfully, which may reduce performance.
3762 ///
3763 /// The default behaviour depends on the `datafusion.catalog.newlines_in_values` setting.
3764 pub fn with_newlines_in_values(mut self, newlines_in_values: bool) -> Self {
3765 self.newlines_in_values = Some(newlines_in_values);
3766 self
3767 }
3768
3769 /// Set a `CompressionTypeVariant` of CSV
3770 /// - defaults to `CompressionTypeVariant::UNCOMPRESSED`
3771 pub fn with_file_compression_type(
3772 mut self,
3773 compression: CompressionTypeVariant,
3774 ) -> Self {
3775 self.compression = compression;
3776 self
3777 }
3778
3779 /// Whether to allow truncated rows when parsing.
3780 /// By default this is set to false and will error if the CSV rows have different lengths.
3781 /// When set to true then it will allow records with less than the expected number of columns and fill the missing columns with nulls.
3782 /// If the record’s schema is not nullable, then it will still return an error.
3783 pub fn with_truncated_rows(mut self, allow: bool) -> Self {
3784 self.truncated_rows = Some(allow);
3785 self
3786 }
3787
3788 /// Set the compression level for the output file.
3789 /// The valid range depends on the compression algorithm.
3790 /// If not specified, the default level for the algorithm is used.
3791 pub fn with_compression_level(mut self, level: u32) -> Self {
3792 self.compression_level = Some(level);
3793 self
3794 }
3795
3796 /// The delimiter character.
3797 pub fn delimiter(&self) -> u8 {
3798 self.delimiter
3799 }
3800
3801 /// The quote character.
3802 pub fn quote(&self) -> u8 {
3803 self.quote
3804 }
3805
3806 /// The terminator character.
3807 pub fn terminator(&self) -> Option<u8> {
3808 self.terminator
3809 }
3810
3811 /// The escape character.
3812 pub fn escape(&self) -> Option<u8> {
3813 self.escape
3814 }
3815}
3816
3817config_namespace! {
3818 /// Options controlling JSON format
3819 pub struct JsonOptions {
3820 pub compression: CompressionTypeVariant, default = CompressionTypeVariant::UNCOMPRESSED
3821 /// Compression level for the output file. The valid range depends on the
3822 /// compression algorithm:
3823 /// - ZSTD: 1 to 22 (default: 3)
3824 /// - GZIP: 0 to 9 (default: 6)
3825 /// - BZIP2: 0 to 9 (default: 6)
3826 /// - XZ: 0 to 9 (default: 6)
3827 /// If not specified, the default level for the compression algorithm is used.
3828 pub compression_level: Option<u32>, default = None
3829 pub schema_infer_max_rec: Option<usize>, default = None
3830 /// The JSON format to use when reading files.
3831 ///
3832 /// When `true` (default), expects newline-delimited JSON (NDJSON):
3833 /// ```text
3834 /// {"key1": 1, "key2": "val"}
3835 /// {"key1": 2, "key2": "vals"}
3836 /// ```
3837 ///
3838 /// When `false`, expects JSON array format:
3839 /// ```text
3840 /// [
3841 /// {"key1": 1, "key2": "val"},
3842 /// {"key1": 2, "key2": "vals"}
3843 /// ]
3844 /// ```
3845 pub newline_delimited: bool, default = true
3846 }
3847}
3848
3849pub trait OutputFormatExt: Display {}
3850
3851#[derive(Debug, Clone, PartialEq)]
3852#[cfg_attr(feature = "parquet", expect(clippy::large_enum_variant))]
3853pub enum OutputFormat {
3854 CSV(CsvOptions),
3855 JSON(JsonOptions),
3856 #[cfg(feature = "parquet")]
3857 PARQUET(TableParquetOptions),
3858 AVRO,
3859 ARROW,
3860}
3861
3862impl Display for OutputFormat {
3863 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
3864 let out = match self {
3865 OutputFormat::CSV(_) => "csv",
3866 OutputFormat::JSON(_) => "json",
3867 #[cfg(feature = "parquet")]
3868 OutputFormat::PARQUET(_) => "parquet",
3869 OutputFormat::AVRO => "avro",
3870 OutputFormat::ARROW => "arrow",
3871 };
3872 write!(f, "{out}")
3873 }
3874}
3875
3876#[cfg(test)]
3877mod tests {
3878 #[cfg(feature = "parquet")]
3879 use crate::assert_contains;
3880 use crate::config::TableParquetOptions;
3881 use crate::config::{
3882 ConfigEntry, ConfigExtension, ConfigField, ConfigFileType, ExtensionOptions,
3883 Extensions, TableOptions,
3884 };
3885 use std::any::Any;
3886 use std::collections::HashMap;
3887
3888 #[derive(Default, Debug, Clone)]
3889 pub struct TestExtensionConfig {
3890 /// Should "foo" be replaced by "bar"?
3891 pub properties: HashMap<String, String>,
3892 }
3893
3894 impl ExtensionOptions for TestExtensionConfig {
3895 fn as_any(&self) -> &dyn Any {
3896 self
3897 }
3898
3899 fn as_any_mut(&mut self) -> &mut dyn Any {
3900 self
3901 }
3902
3903 fn cloned(&self) -> Box<dyn ExtensionOptions> {
3904 Box::new(self.clone())
3905 }
3906
3907 fn set(&mut self, key: &str, value: &str) -> crate::Result<()> {
3908 let (key, rem) = key.split_once('.').unwrap_or((key, ""));
3909 assert_eq!(key, "test");
3910 self.properties.insert(rem.to_owned(), value.to_owned());
3911 Ok(())
3912 }
3913
3914 fn entries(&self) -> Vec<ConfigEntry> {
3915 self.properties
3916 .iter()
3917 .map(|(k, v)| ConfigEntry {
3918 key: k.into(),
3919 value: Some(v.into()),
3920 description: "",
3921 })
3922 .collect()
3923 }
3924 }
3925
3926 impl ConfigExtension for TestExtensionConfig {
3927 const PREFIX: &'static str = "test";
3928 }
3929
3930 #[test]
3931 fn create_table_config() {
3932 let mut extension = Extensions::new();
3933 extension.insert(TestExtensionConfig::default());
3934 let table_config = TableOptions::new().with_extensions(extension);
3935 let kafka_config = table_config.extensions.get::<TestExtensionConfig>();
3936 assert!(kafka_config.is_some())
3937 }
3938
3939 #[test]
3940 fn alter_test_extension_config() {
3941 let mut extension = Extensions::new();
3942 extension.insert(TestExtensionConfig::default());
3943 let mut table_config = TableOptions::new().with_extensions(extension);
3944 table_config.set_config_format(ConfigFileType::CSV);
3945 table_config.set("format.delimiter", ";").unwrap();
3946 assert_eq!(table_config.csv.delimiter, b';');
3947 table_config.set("test.bootstrap.servers", "asd").unwrap();
3948 let kafka_config = table_config
3949 .extensions
3950 .get::<TestExtensionConfig>()
3951 .unwrap();
3952 assert_eq!(
3953 kafka_config.properties.get("bootstrap.servers").unwrap(),
3954 "asd"
3955 );
3956 }
3957
3958 #[test]
3959 fn iter_test_extension_config() {
3960 let mut extension = Extensions::new();
3961 extension.insert(TestExtensionConfig::default());
3962 let table_config = TableOptions::new().with_extensions(extension);
3963 let extensions = table_config.extensions.iter().collect::<Vec<_>>();
3964 assert_eq!(extensions.len(), 1);
3965 assert_eq!(extensions[0].0, TestExtensionConfig::PREFIX);
3966 }
3967
3968 #[test]
3969 fn csv_u8_table_options() {
3970 let mut table_config = TableOptions::new();
3971 table_config.set_config_format(ConfigFileType::CSV);
3972 table_config.set("format.delimiter", ";").unwrap();
3973 assert_eq!(table_config.csv.delimiter as char, ';');
3974 table_config.set("format.escape", "\"").unwrap();
3975 assert_eq!(table_config.csv.escape.unwrap() as char, '"');
3976 table_config.set("format.escape", "\'").unwrap();
3977 assert_eq!(table_config.csv.escape.unwrap() as char, '\'');
3978 }
3979
3980 #[test]
3981 fn warning_only_not_default() {
3982 use std::sync::atomic::AtomicUsize;
3983 static COUNT: AtomicUsize = AtomicUsize::new(0);
3984 use log::{Level, LevelFilter, Metadata, Record};
3985 struct SimpleLogger;
3986 impl log::Log for SimpleLogger {
3987 fn enabled(&self, metadata: &Metadata) -> bool {
3988 metadata.level() <= Level::Info
3989 }
3990
3991 fn log(&self, record: &Record) {
3992 if self.enabled(record.metadata()) {
3993 COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
3994 }
3995 }
3996 fn flush(&self) {}
3997 }
3998 log::set_logger(&SimpleLogger).unwrap();
3999 log::set_max_level(LevelFilter::Info);
4000 let mut sql_parser_options = crate::config::SqlParserOptions::default();
4001 sql_parser_options
4002 .set("enable_options_value_normalization", "false")
4003 .unwrap();
4004 assert_eq!(COUNT.load(std::sync::atomic::Ordering::Relaxed), 0);
4005 sql_parser_options
4006 .set("enable_options_value_normalization", "true")
4007 .unwrap();
4008 assert_eq!(COUNT.load(std::sync::atomic::Ordering::Relaxed), 1);
4009 }
4010
4011 #[test]
4012 fn reset_nested_scalar_reports_helpful_error() {
4013 let mut value = true;
4014 let err = <bool as ConfigField>::reset(&mut value, "nested").unwrap_err();
4015 let message = err.to_string();
4016 assert!(
4017 message.starts_with(
4018 "Invalid or Unsupported Configuration: Config field is a scalar bool and does not have nested field \"nested\""
4019 ),
4020 "unexpected error message: {message}"
4021 );
4022 }
4023
4024 #[cfg(feature = "parquet")]
4025 #[test]
4026 fn parquet_table_options() {
4027 let mut table_config = TableOptions::new();
4028 table_config.set_config_format(ConfigFileType::PARQUET);
4029 table_config
4030 .set("format.bloom_filter_enabled::col1", "true")
4031 .unwrap();
4032 assert_eq!(
4033 table_config.parquet.column_specific_options["col1"].bloom_filter_enabled,
4034 Some(true)
4035 );
4036 }
4037
4038 #[cfg(feature = "parquet_encryption")]
4039 #[test]
4040 fn parquet_table_encryption() {
4041 use crate::config::{
4042 ConfigFileDecryptionProperties, ConfigFileEncryptionProperties,
4043 };
4044 use parquet::encryption::decrypt::FileDecryptionProperties;
4045 use parquet::encryption::encrypt::FileEncryptionProperties;
4046 use std::sync::Arc;
4047
4048 let footer_key = b"0123456789012345".to_vec(); // 128bit/16
4049 let column_names = vec!["double_field", "float_field"];
4050 let column_keys =
4051 vec![b"1234567890123450".to_vec(), b"1234567890123451".to_vec()];
4052
4053 let file_encryption_properties =
4054 FileEncryptionProperties::builder(footer_key.clone())
4055 .with_column_keys(column_names.clone(), column_keys.clone())
4056 .unwrap()
4057 .build()
4058 .unwrap();
4059
4060 let decryption_properties = FileDecryptionProperties::builder(footer_key.clone())
4061 .with_column_keys(column_names.clone(), column_keys.clone())
4062 .unwrap()
4063 .build()
4064 .unwrap();
4065
4066 // Test round-trip
4067 let config_encrypt =
4068 ConfigFileEncryptionProperties::from(&file_encryption_properties);
4069 let encryption_properties_built =
4070 Arc::new(FileEncryptionProperties::try_from(config_encrypt.clone()).unwrap());
4071 assert_eq!(file_encryption_properties, encryption_properties_built);
4072
4073 let config_decrypt =
4074 ConfigFileDecryptionProperties::try_from(&decryption_properties).unwrap();
4075 let decryption_properties_built =
4076 Arc::new(FileDecryptionProperties::try_from(config_decrypt.clone()).unwrap());
4077 assert_eq!(decryption_properties, decryption_properties_built);
4078
4079 ///////////////////////////////////////////////////////////////////////////////////
4080 // Test encryption config
4081
4082 // Display original encryption config
4083 // println!("{:#?}", config_encrypt);
4084
4085 let mut table_config = TableOptions::new();
4086 table_config.set_config_format(ConfigFileType::PARQUET);
4087 table_config
4088 .parquet
4089 .set(
4090 "crypto.file_encryption.encrypt_footer",
4091 config_encrypt.encrypt_footer.to_string().as_str(),
4092 )
4093 .unwrap();
4094 table_config
4095 .parquet
4096 .set(
4097 "crypto.file_encryption.footer_key_as_hex",
4098 config_encrypt.footer_key_as_hex.as_str(),
4099 )
4100 .unwrap();
4101
4102 for (i, col_name) in column_names.iter().enumerate() {
4103 let key = format!("crypto.file_encryption.column_key_as_hex::{col_name}");
4104 let value = hex::encode(column_keys[i].clone());
4105 table_config
4106 .parquet
4107 .set(key.as_str(), value.as_str())
4108 .unwrap();
4109 }
4110
4111 // Print matching final encryption config
4112 // println!("{:#?}", table_config.parquet.crypto.file_encryption);
4113
4114 assert_eq!(
4115 table_config.parquet.crypto.file_encryption,
4116 Some(config_encrypt)
4117 );
4118
4119 ///////////////////////////////////////////////////////////////////////////////////
4120 // Test decryption config
4121
4122 // Display original decryption config
4123 // println!("{:#?}", config_decrypt);
4124
4125 let mut table_config = TableOptions::new();
4126 table_config.set_config_format(ConfigFileType::PARQUET);
4127 table_config
4128 .parquet
4129 .set(
4130 "crypto.file_decryption.footer_key_as_hex",
4131 config_decrypt.footer_key_as_hex.as_str(),
4132 )
4133 .unwrap();
4134
4135 for (i, col_name) in column_names.iter().enumerate() {
4136 let key = format!("crypto.file_decryption.column_key_as_hex::{col_name}");
4137 let value = hex::encode(column_keys[i].clone());
4138 table_config
4139 .parquet
4140 .set(key.as_str(), value.as_str())
4141 .unwrap();
4142 }
4143
4144 // Print matching final decryption config
4145 // println!("{:#?}", table_config.parquet.crypto.file_decryption);
4146
4147 assert_eq!(
4148 table_config.parquet.crypto.file_decryption,
4149 Some(config_decrypt.clone())
4150 );
4151
4152 // Set config directly
4153 let mut table_config = TableOptions::new();
4154 table_config.set_config_format(ConfigFileType::PARQUET);
4155 table_config.parquet.crypto.file_decryption = Some(config_decrypt.clone());
4156 assert_eq!(
4157 table_config.parquet.crypto.file_decryption,
4158 Some(config_decrypt.clone())
4159 );
4160 }
4161
4162 #[cfg(feature = "parquet_encryption")]
4163 #[test]
4164 fn parquet_encryption_invalid_hex_errors_encryption() {
4165 use crate::config::ColumnEncryptionProperties;
4166 use crate::config::ConfigFileEncryptionProperties;
4167 use parquet::encryption::encrypt::FileEncryptionProperties;
4168 use std::collections::HashMap;
4169
4170 let valid_footer_key_as_hex = hex::encode(b"0123456789012345");
4171
4172 let mut enc = ConfigFileEncryptionProperties {
4173 encrypt_footer: true,
4174 footer_key_as_hex: valid_footer_key_as_hex.clone(),
4175 footer_key_metadata_as_hex: String::new(),
4176 column_encryption_properties: HashMap::new(),
4177 aad_prefix_as_hex: String::new(),
4178 store_aad_prefix: false,
4179 };
4180
4181 // Encryption: invalid footer key hex
4182 enc.footer_key_as_hex = "not_hex".to_string();
4183 let err = FileEncryptionProperties::try_from(enc.clone())
4184 .unwrap_err()
4185 .to_string();
4186 assert!(err.contains("Unable to decode hex footer key"));
4187 enc.footer_key_as_hex = valid_footer_key_as_hex.clone();
4188
4189 // Encryption: invalid footer key metadata hex
4190 enc.footer_key_metadata_as_hex = "zz".to_string();
4191 let err = FileEncryptionProperties::try_from(enc.clone())
4192 .unwrap_err()
4193 .to_string();
4194 assert!(err.contains("Unable to decode hex footer key metadata"));
4195 enc.footer_key_metadata_as_hex = String::new();
4196
4197 // Encryption: invalid column key hex
4198 enc.column_encryption_properties.insert(
4199 "col1".to_string(),
4200 ColumnEncryptionProperties {
4201 column_key_as_hex: "bad".to_string(),
4202 column_metadata_as_hex: None,
4203 },
4204 );
4205 let err = FileEncryptionProperties::try_from(enc.clone())
4206 .unwrap_err()
4207 .to_string();
4208 assert!(err.contains("Unable to decode hex encryption key for column col1"));
4209 enc.column_encryption_properties.clear();
4210
4211 // Encryption: invalid column metadata hex
4212 enc.column_encryption_properties.insert(
4213 "col1".to_string(),
4214 ColumnEncryptionProperties {
4215 column_key_as_hex: hex::encode(b"1234567890123450"),
4216 column_metadata_as_hex: Some("zz".to_string()),
4217 },
4218 );
4219 let err = FileEncryptionProperties::try_from(enc.clone())
4220 .unwrap_err()
4221 .to_string();
4222 assert!(err.contains("Unable to decode hex column metadata for column col1"));
4223 enc.column_encryption_properties.clear();
4224
4225 // Encryption: invalid AAD prefix hex
4226 enc.aad_prefix_as_hex = "zz".to_string();
4227 let err = FileEncryptionProperties::try_from(enc.clone())
4228 .unwrap_err()
4229 .to_string();
4230 assert!(err.contains("Unable to decode hex AAD prefix"));
4231 }
4232
4233 #[cfg(feature = "parquet_encryption")]
4234 #[test]
4235 fn parquet_encryption_invalid_hex_errors_decryption() {
4236 use crate::config::ColumnDecryptionProperties;
4237 use crate::config::ConfigFileDecryptionProperties;
4238 use parquet::encryption::decrypt::FileDecryptionProperties;
4239 use std::collections::HashMap;
4240
4241 let valid_footer_key_as_hex = hex::encode(b"0123456789012345");
4242
4243 let mut dec = ConfigFileDecryptionProperties {
4244 footer_key_as_hex: valid_footer_key_as_hex.clone(),
4245 column_decryption_properties: HashMap::new(),
4246 aad_prefix_as_hex: String::new(),
4247 footer_signature_verification: true,
4248 };
4249
4250 // Decryption: invalid column key hex
4251 dec.column_decryption_properties.insert(
4252 "col1".to_string(),
4253 ColumnDecryptionProperties {
4254 column_key_as_hex: "bad".to_string(),
4255 },
4256 );
4257 let err = FileDecryptionProperties::try_from(dec.clone())
4258 .unwrap_err()
4259 .to_string();
4260 assert!(err.contains("Could not decode hex column key"));
4261 assert!(err.contains("col1"));
4262 dec.column_decryption_properties.clear();
4263
4264 // Decryption: invalid footer key hex
4265 dec.footer_key_as_hex = "bad".to_string();
4266 let err = FileDecryptionProperties::try_from(dec.clone())
4267 .unwrap_err()
4268 .to_string();
4269 assert!(err.contains("Could not decode hex footer key"));
4270 dec.footer_key_as_hex = valid_footer_key_as_hex;
4271
4272 // Decryption: invalid AAD prefix hex
4273 dec.aad_prefix_as_hex = "zz".to_string();
4274 let err = FileDecryptionProperties::try_from(dec.clone())
4275 .unwrap_err()
4276 .to_string();
4277 assert!(err.contains("Could not decode hex AAD prefix"));
4278 }
4279
4280 #[cfg(feature = "parquet_encryption")]
4281 #[test]
4282 fn parquet_encryption_factory_config() {
4283 let mut parquet_options = TableParquetOptions::default();
4284
4285 assert_eq!(parquet_options.crypto.factory_id, None);
4286 assert_eq!(parquet_options.crypto.factory_options.options.len(), 0);
4287
4288 let mut input_config = TestExtensionConfig::default();
4289 input_config
4290 .properties
4291 .insert("key1".to_string(), "value 1".to_string());
4292 input_config
4293 .properties
4294 .insert("key2".to_string(), "value 2".to_string());
4295
4296 parquet_options
4297 .crypto
4298 .configure_factory("example_factory", &input_config);
4299
4300 assert_eq!(
4301 parquet_options.crypto.factory_id,
4302 Some("example_factory".to_string())
4303 );
4304 let factory_options = &parquet_options.crypto.factory_options.options;
4305 assert_eq!(factory_options.len(), 2);
4306 assert_eq!(factory_options.get("key1"), Some(&"value 1".to_string()));
4307 assert_eq!(factory_options.get("key2"), Some(&"value 2".to_string()));
4308 }
4309
4310 #[cfg(feature = "parquet_encryption")]
4311 struct ParquetEncryptionKeyRetriever {}
4312
4313 #[cfg(feature = "parquet_encryption")]
4314 impl parquet::encryption::decrypt::KeyRetriever for ParquetEncryptionKeyRetriever {
4315 fn retrieve_key(&self, key_metadata: &[u8]) -> parquet::errors::Result<Vec<u8>> {
4316 if !key_metadata.is_empty() {
4317 Ok(b"1234567890123450".to_vec())
4318 } else {
4319 Err(parquet::errors::ParquetError::General(
4320 "Key metadata not provided".to_string(),
4321 ))
4322 }
4323 }
4324 }
4325
4326 #[cfg(feature = "parquet_encryption")]
4327 #[test]
4328 fn conversion_from_key_retriever_to_config_file_decryption_properties() {
4329 use crate::Result;
4330 use crate::config::ConfigFileDecryptionProperties;
4331 use crate::encryption::FileDecryptionProperties;
4332
4333 let retriever = std::sync::Arc::new(ParquetEncryptionKeyRetriever {});
4334 let decryption_properties =
4335 FileDecryptionProperties::with_key_retriever(retriever)
4336 .build()
4337 .unwrap();
4338 let config_file_decryption_properties: Result<ConfigFileDecryptionProperties> =
4339 (&decryption_properties).try_into();
4340 assert!(config_file_decryption_properties.is_err());
4341 let err = config_file_decryption_properties.unwrap_err().to_string();
4342 assert!(err.contains("key retriever"));
4343 assert!(err.contains("Key metadata not provided"));
4344 }
4345
4346 #[cfg(feature = "parquet")]
4347 #[test]
4348 fn parquet_table_options_config_entry() {
4349 let mut table_config = TableOptions::new();
4350 table_config.set_config_format(ConfigFileType::PARQUET);
4351 table_config
4352 .set("format.bloom_filter_enabled::col1", "true")
4353 .unwrap();
4354 let entries = table_config.entries();
4355 assert!(
4356 entries
4357 .iter()
4358 .any(|item| item.key == "format.bloom_filter_enabled::col1")
4359 )
4360 }
4361
4362 #[cfg(feature = "parquet")]
4363 #[test]
4364 fn parquet_table_parquet_options_config_entry() {
4365 let mut table_parquet_options = TableParquetOptions::new();
4366 table_parquet_options
4367 .set(
4368 "crypto.file_encryption.column_key_as_hex::double_field",
4369 "31323334353637383930313233343530",
4370 )
4371 .unwrap();
4372 let entries = table_parquet_options.entries();
4373 assert!(
4374 entries.iter().any(|item| item.key
4375 == "crypto.file_encryption.column_key_as_hex::double_field")
4376 )
4377 }
4378
4379 #[cfg(feature = "parquet")]
4380 #[test]
4381 fn parquet_table_options_config_metadata_entry() {
4382 let mut table_config = TableOptions::new();
4383 table_config.set_config_format(ConfigFileType::PARQUET);
4384 table_config.set("format.metadata::key1", "").unwrap();
4385 table_config.set("format.metadata::key2", "value2").unwrap();
4386 table_config
4387 .set("format.metadata::key3", "value with spaces ")
4388 .unwrap();
4389 table_config
4390 .set("format.metadata::key4", "value with special chars :: :")
4391 .unwrap();
4392
4393 let parsed_metadata = table_config.parquet.key_value_metadata.clone();
4394 assert_eq!(parsed_metadata.get("should not exist1"), None);
4395 assert_eq!(parsed_metadata.get("key1"), Some(&Some("".into())));
4396 assert_eq!(parsed_metadata.get("key2"), Some(&Some("value2".into())));
4397 assert_eq!(
4398 parsed_metadata.get("key3"),
4399 Some(&Some("value with spaces ".into()))
4400 );
4401 assert_eq!(
4402 parsed_metadata.get("key4"),
4403 Some(&Some("value with special chars :: :".into()))
4404 );
4405
4406 // duplicate keys are overwritten
4407 table_config.set("format.metadata::key_dupe", "A").unwrap();
4408 table_config.set("format.metadata::key_dupe", "B").unwrap();
4409 let parsed_metadata = table_config.parquet.key_value_metadata;
4410 assert_eq!(parsed_metadata.get("key_dupe"), Some(&Some("B".into())));
4411 }
4412 #[cfg(feature = "parquet")]
4413 #[test]
4414 fn test_parquet_writer_version_validation() {
4415 use crate::{config::ConfigOptions, parquet_config::DFParquetWriterVersion};
4416
4417 let mut config = ConfigOptions::default();
4418
4419 // Valid values should work
4420 config
4421 .set("datafusion.execution.parquet.writer_version", "1.0")
4422 .unwrap();
4423 assert_eq!(
4424 config.execution.parquet.writer_version,
4425 DFParquetWriterVersion::V1_0
4426 );
4427
4428 config
4429 .set("datafusion.execution.parquet.writer_version", "2.0")
4430 .unwrap();
4431 assert_eq!(
4432 config.execution.parquet.writer_version,
4433 DFParquetWriterVersion::V2_0
4434 );
4435
4436 // Invalid value should error immediately at SET time
4437 let err = config
4438 .set("datafusion.execution.parquet.writer_version", "3.0")
4439 .unwrap_err();
4440 assert_contains!(
4441 err.to_string(),
4442 "Invalid or Unsupported Configuration: Invalid parquet writer version: 3.0. Expected one of: 1.0, 2.0"
4443 );
4444 }
4445
4446 #[cfg(feature = "parquet")]
4447 #[test]
4448 fn set_cdc_enabled_flag() {
4449 use crate::config::ConfigOptions;
4450
4451 let mut config = ConfigOptions::default();
4452 // CDC is disabled by default.
4453 assert!(!config.execution.parquet.content_defined_chunking.enabled);
4454
4455 // `.enabled = true` enables CDC; parameters keep their defaults.
4456 config
4457 .set(
4458 "datafusion.execution.parquet.content_defined_chunking.enabled",
4459 "true",
4460 )
4461 .unwrap();
4462 let cdc = &config.execution.parquet.content_defined_chunking;
4463 assert!(cdc.enabled);
4464 assert_eq!(cdc.min_chunk_size, 256 * 1024);
4465 assert_eq!(cdc.max_chunk_size, 1024 * 1024);
4466 assert_eq!(cdc.norm_level, 0);
4467
4468 // `.enabled = false` disables CDC.
4469 config
4470 .set(
4471 "datafusion.execution.parquet.content_defined_chunking.enabled",
4472 "false",
4473 )
4474 .unwrap();
4475 assert!(!config.execution.parquet.content_defined_chunking.enabled);
4476 }
4477
4478 #[cfg(feature = "parquet")]
4479 #[test]
4480 fn set_cdc_param_does_not_enable() {
4481 use crate::config::ConfigOptions;
4482
4483 let mut config = ConfigOptions::default();
4484
4485 // Setting a parameter does NOT enable CDC (`enabled` is a distinct field,
4486 // defaulting to false), and the result is independent of key order.
4487 config
4488 .set(
4489 "datafusion.execution.parquet.content_defined_chunking.min_chunk_size",
4490 "1024",
4491 )
4492 .unwrap();
4493 let cdc = &config.execution.parquet.content_defined_chunking;
4494 assert!(!cdc.enabled);
4495 assert_eq!(cdc.min_chunk_size, 1024);
4496 assert_eq!(cdc.max_chunk_size, 1024 * 1024);
4497 assert_eq!(cdc.norm_level, 0);
4498 }
4499
4500 #[test]
4501 fn test_dialect_metadata_roundtrip() {
4502 use crate::config::Dialect;
4503 use std::str::FromStr;
4504
4505 assert_eq!(Dialect::default(), Dialect::Generic);
4506 assert!(!Dialect::metadata().is_empty());
4507
4508 for info in Dialect::metadata() {
4509 let dialect = info.dialect;
4510
4511 assert_eq!(Dialect::from_str(info.canonical_name).unwrap(), dialect);
4512 assert_eq!(
4513 Dialect::from_str(&info.canonical_name.to_ascii_uppercase()).unwrap(),
4514 dialect
4515 );
4516 assert_eq!(dialect.as_ref(), info.canonical_name);
4517 assert_eq!(dialect.to_string(), info.canonical_name);
4518 }
4519 }
4520
4521 #[test]
4522 fn test_dialect_aliases() {
4523 use crate::config::Dialect;
4524 use std::str::FromStr;
4525
4526 for info in Dialect::metadata() {
4527 for alias in info.aliases {
4528 assert_eq!(Dialect::from_str(alias).unwrap(), info.dialect);
4529 assert_eq!(
4530 Dialect::from_str(&alias.to_ascii_uppercase()).unwrap(),
4531 info.dialect
4532 );
4533 }
4534 }
4535 }
4536
4537 #[test]
4538 fn test_available_dialects_includes_each_display_name_once() {
4539 use crate::config::Dialect;
4540 use std::collections::BTreeSet;
4541
4542 let available = Dialect::available();
4543 let listed: Vec<_> = available.split(", ").collect();
4544 let display_names: Vec<_> = Dialect::metadata()
4545 .iter()
4546 .map(|info| info.display_name)
4547 .collect();
4548 let unique_display_names: BTreeSet<_> = display_names.iter().copied().collect();
4549
4550 assert_eq!(display_names.len(), unique_display_names.len());
4551 assert_eq!(listed, display_names);
4552 }
4553
4554 #[test]
4555 fn test_dialect_config_description_uses_metadata() {
4556 use crate::config::{ConfigOptions, Dialect, SQL_PARSER_DIALECT_CONFIG_KEY};
4557
4558 let description = ConfigOptions::default()
4559 .entries()
4560 .into_iter()
4561 .find(|entry| entry.key == SQL_PARSER_DIALECT_CONFIG_KEY)
4562 .unwrap()
4563 .description;
4564
4565 assert!(description.contains(Dialect::available()));
4566 }
4567
4568 #[test]
4569 fn test_invalid_dialect_error_lists_available_dialects() {
4570 use crate::config::Dialect;
4571 use std::str::FromStr;
4572
4573 let error = Dialect::from_str("notadialect").unwrap_err().to_string();
4574
4575 assert!(error.contains("Invalid Dialect: notadialect"));
4576 assert!(error.contains(Dialect::available()));
4577 }
4578
4579 #[test]
4580 fn max_row_group_bytes_rejects_zero() {
4581 use crate::config::MaxRowGroupBytes;
4582 use std::str::FromStr;
4583
4584 assert!(MaxRowGroupBytes::try_new(0).is_err());
4585 assert!(MaxRowGroupBytes::from_str("0").is_err());
4586 assert!(MaxRowGroupBytes::from_str("not_a_number").is_err());
4587 assert_eq!(MaxRowGroupBytes::try_new(128).unwrap().get(), 128);
4588 assert_eq!(MaxRowGroupBytes::from_str("128").unwrap().get(), 128);
4589 }
4590
4591 #[test]
4592 fn parquet_max_row_group_bytes_config_set_rejects_zero() {
4593 use crate::config::ConfigOptions;
4594
4595 let mut options = ConfigOptions::new();
4596 options
4597 .set("datafusion.execution.parquet.max_row_group_bytes", "1024")
4598 .unwrap();
4599 assert_eq!(
4600 options
4601 .execution
4602 .parquet
4603 .max_row_group_bytes
4604 .map(|v| v.get()),
4605 Some(1024)
4606 );
4607
4608 // Zero is rejected at set time, leaving the previous value unchanged.
4609 assert!(
4610 options
4611 .set("datafusion.execution.parquet.max_row_group_bytes", "0")
4612 .is_err()
4613 );
4614 assert_eq!(
4615 options
4616 .execution
4617 .parquet
4618 .max_row_group_bytes
4619 .map(|v| v.get()),
4620 Some(1024)
4621 );
4622 }
4623}