## Coverage Report

#### Created: 2025-11-22 06:00

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /Users/andrewlamb/Software/datafusion/datafusion/common/src/config.rs

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