Skip to main content

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