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}