datafusion_common/

config.rs

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

//! Runtime configuration, via [`ConfigOptions`]

use arrow_ipc::CompressionType;

#[cfg(feature = "parquet_encryption")]
use crate::encryption::{FileDecryptionProperties, FileEncryptionProperties};
use crate::error::{_config_datafusion_err, _config_err};
use crate::format::{ExplainAnalyzeCategories, ExplainFormat, MetricType};
use crate::parquet_config::DFParquetWriterVersion;
use crate::parsers::{CompressionTypeVariant, CsvQuoteStyle};
use crate::utils::get_available_parallelism;
use crate::{DataFusionError, Result};
#[cfg(feature = "parquet_encryption")]
use hex;
use std::any::Any;
use std::collections::{BTreeMap, HashMap};
use std::error::Error;
use std::fmt::{self, Display};
use std::num::NonZeroUsize;
use std::str::FromStr;
#[cfg(feature = "parquet_encryption")]
use std::sync::Arc;

/// A macro that wraps a configuration struct and automatically derives
/// [`Default`] and [`ConfigField`] for it, allowing it to be used
/// in the [`ConfigOptions`] configuration tree.
///
/// `transform` is used to normalize values before parsing.
///
/// For example,
///
/// ```ignore
/// config_namespace! {
///    /// Amazing config
///    pub struct MyConfig {
///        /// Field 1 doc
///        field1: String, transform = str::to_lowercase, default = "".to_string()
///
///        /// Field 2 doc
///        field2: usize, default = 232
///
///        /// Field 3 doc
///        field3: Option<usize>, default = None
///    }
/// }
/// ```
///
/// Will generate
///
/// ```ignore
/// /// Amazing config
/// #[derive(Debug, Clone)]
/// #[non_exhaustive]
/// pub struct MyConfig {
///     /// Field 1 doc
///     field1: String,
///     /// Field 2 doc
///     field2: usize,
///     /// Field 3 doc
///     field3: Option<usize>,
/// }
/// impl ConfigField for MyConfig {
///     fn set(&mut self, key: &str, value: &str) -> Result<()> {
///         let (key, rem) = key.split_once('.').unwrap_or((key, ""));
///         match key {
///             "field1" => {
///                 let value = str::to_lowercase(value);
///                 self.field1.set(rem, value.as_ref())
///             },
///             "field2" => self.field2.set(rem, value.as_ref()),
///             "field3" => self.field3.set(rem, value.as_ref()),
///             _ => _internal_err!(
///                 "Config value \"{}\" not found on MyConfig",
///                 key
///             ),
///         }
///     }
///
///     fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
///         let key = format!("{}.field1", key_prefix);
///         let desc = "Field 1 doc";
///         self.field1.visit(v, key.as_str(), desc);
///         let key = format!("{}.field2", key_prefix);
///         let desc = "Field 2 doc";
///         self.field2.visit(v, key.as_str(), desc);
///         let key = format!("{}.field3", key_prefix);
///         let desc = "Field 3 doc";
///         self.field3.visit(v, key.as_str(), desc);
///     }
/// }
///
/// impl Default for MyConfig {
///     fn default() -> Self {
///         Self {
///             field1: "".to_string(),
///             field2: 232,
///             field3: None,
///         }
///     }
/// }
/// ```
///
/// NB: Misplaced commas may result in nonsensical errors
#[macro_export]
macro_rules! config_namespace {
    (
        $(#[doc = $struct_d:tt])* // Struct-level documentation attributes
        $(#[deprecated($($struct_depr:tt)*)])? // Optional struct-level deprecated attribute
        $(#[allow($($struct_de:tt)*)])?
        $vis:vis struct $struct_name:ident {
            $(
                $(#[doc = $d:tt])* // Field-level documentation attributes
                $(#[deprecated($($field_depr:tt)*)])? // Optional field-level deprecated attribute
                $(#[allow($($field_de:tt)*)])?
                $field_vis:vis $field_name:ident : $field_type:ty,
                $(warn = $warn:expr,)?
                $(transform = $transform:expr,)?
                default = $default:expr
            )*$(,)*
        }
    ) => {
        $(#[doc = $struct_d])* // Apply struct documentation
        $(#[deprecated($($struct_depr)*)])? // Apply struct deprecation
        $(#[allow($($struct_de)*)])?
        #[derive(Debug, Clone, PartialEq)]
        $vis struct $struct_name {
            $(
                $(#[doc = $d])* // Apply field documentation
                $(#[deprecated($($field_depr)*)])? // Apply field deprecation
                $(#[allow($($field_de)*)])?
                $field_vis $field_name: $field_type,
            )*
        }

        impl $crate::config::ConfigField for $struct_name {
            fn set(&mut self, key: &str, value: &str) -> $crate::error::Result<()> {
                let (key, rem) = key.split_once('.').unwrap_or((key, ""));
                match key {
                    $(
                        stringify!($field_name) => {
                            // Safely apply deprecated attribute if present
                            // $(#[allow(deprecated)])?
                            {
                                $(let value = $transform(value);)? // Apply transformation if specified
                                let ret = self.$field_name.set(rem, value.as_ref());

                                $(if !$warn.is_empty() {
                                    let default: $field_type = $default;
                                    if default != self.$field_name {
                                        log::warn!($warn);
                                    }
                                })? // Log warning if specified, and the value is not the default
                                ret
                            }
                        },
                    )*
                    _ => return $crate::error::_config_err!(
                        "Config value \"{}\" not found on {}", key, stringify!($struct_name)
                    )
                }
            }

            fn visit<V: $crate::config::Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
                $(
                    let key = format!(concat!("{}.", stringify!($field_name)), key_prefix);
                    let desc = concat!($($d),*).trim();
                    self.$field_name.visit(v, key.as_str(), desc);
                )*
            }

            fn reset(&mut self, key: &str) -> $crate::error::Result<()> {
                let (key, rem) = key.split_once('.').unwrap_or((key, ""));
                match key {
                    $(
                        stringify!($field_name) => {
                                    {
                                if rem.is_empty() {
                                    let default_value: $field_type = $default;
                                    self.$field_name = default_value;
                                    Ok(())
                                } else {
                                    self.$field_name.reset(rem)
                                }
                            }
                        },
                    )*
                    _ => $crate::error::_config_err!(
                        "Config value \"{}\" not found on {}",
                        key,
                        stringify!($struct_name)
                    ),
                }
            }
        }
        impl Default for $struct_name {
            fn default() -> Self {
                Self {
                    $($field_name: $default),*
                }
            }
        }
    }
}

config_namespace! {
    /// Options related to catalog and directory scanning
    ///
    /// See also: [`SessionConfig`]
    ///
    /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
    pub struct CatalogOptions {
        /// Whether the default catalog and schema should be created automatically.
        pub create_default_catalog_and_schema: bool, default = true

        /// The default catalog name - this impacts what SQL queries use if not specified
        pub default_catalog: String, default = "datafusion".to_string()

        /// The default schema name - this impacts what SQL queries use if not specified
        pub default_schema: String, default = "public".to_string()

        /// Should DataFusion provide access to `information_schema`
        /// virtual tables for displaying schema information
        pub information_schema: bool, default = false

        /// Location scanned to load tables for `default` schema
        pub location: Option<String>, default = None

        /// Type of `TableProvider` to use when loading `default` schema
        pub format: Option<String>, default = None

        /// Default value for `format.has_header` for `CREATE EXTERNAL TABLE`
        /// if not specified explicitly in the statement.
        pub has_header: bool, default = true

        /// Specifies whether newlines in (quoted) CSV values are supported.
        ///
        /// This is the default value for `format.newlines_in_values` for `CREATE EXTERNAL TABLE`
        /// if not specified explicitly in the statement.
        ///
        /// Parsing newlines in quoted values may be affected by execution behaviour such as
        /// parallel file scanning. Setting this to `true` ensures that newlines in values are
        /// parsed successfully, which may reduce performance.
        pub newlines_in_values: bool, default = false
    }
}

config_namespace! {
    /// Options related to SQL parser
    ///
    /// See also: [`SessionConfig`]
    ///
    /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
    pub struct SqlParserOptions {
        /// When set to true, SQL parser will parse float as decimal type
        pub parse_float_as_decimal: bool, default = false

        /// When set to true, SQL parser will normalize ident (convert ident to lowercase when not quoted)
        pub enable_ident_normalization: bool, default = true

        /// When set to true, SQL parser will normalize options value (convert value to lowercase).
        /// Note that this option is ignored and will be removed in the future. All case-insensitive values
        /// are normalized automatically.
        pub enable_options_value_normalization: bool, warn = "`enable_options_value_normalization` is deprecated and ignored", default = false

        /// Configure the SQL dialect used by DataFusion's parser.
        /// The configuration reference lists the supported values from [`Dialect::available`].
        pub dialect: Dialect, default = Dialect::Generic
        // no need to lowercase because `sqlparser::dialect_from_str`] is case-insensitive

        /// If true, permit lengths for `VARCHAR` such as `VARCHAR(20)`, but
        /// ignore the length. If false, error if a `VARCHAR` with a length is
        /// specified. The Arrow type system does not have a notion of maximum
        /// string length and thus DataFusion can not enforce such limits.
        pub support_varchar_with_length: bool, default = true

        /// If true, string types (VARCHAR, CHAR, Text, and String) are mapped to `Utf8View` during SQL planning.
        /// If false, they are mapped to `Utf8`.
        /// Default is true.
        pub map_string_types_to_utf8view: bool, default = true

        /// When set to true, the source locations relative to the original SQL
        /// query (i.e. [`Span`](https://docs.rs/sqlparser/latest/sqlparser/tokenizer/struct.Span.html)) will be collected
        /// and recorded in the logical plan nodes.
        pub collect_spans: bool, default = false

        /// Specifies the recursion depth limit when parsing complex SQL Queries
        pub recursion_limit: ConfigNonZeroUsize, default = non_zero_usize_default(50)

        /// Specifies the default null ordering for query results. There are 4 options:
        /// - `nulls_max`: Nulls appear last in ascending order.
        /// - `nulls_min`: Nulls appear first in ascending order.
        /// - `nulls_first`: Nulls always be first in any order.
        /// - `nulls_last`: Nulls always be last in any order.
        ///
        /// By default, `nulls_max` is used to follow Postgres's behavior.
        /// postgres rule: <https://www.postgresql.org/docs/current/queries-order.html>
        pub default_null_ordering: String, default = "nulls_max".to_string()

        /// When set to true, DataFusion may remove `ORDER BY` clauses from
        /// subqueries or CTEs during SQL planning when their ordering cannot
        /// affect the result, such as when no `LIMIT` or other
        /// order-sensitive operator depends on them.
        ///
        /// Disable this option to preserve explicit subquery ordering in the
        /// planned query.
        pub enable_subquery_sort_elimination: bool, default = true
    }
}

/// Metadata for a SQL dialect supported by DataFusion configuration.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub struct DialectInfo {
    pub dialect: Dialect,
    pub canonical_name: &'static str,
    pub display_name: &'static str,
    pub aliases: &'static [&'static str],
}

// Keep this key in sync with the `SqlParserOptions::dialect` config path.
const SQL_PARSER_DIALECT_CONFIG_KEY: &str = "datafusion.sql_parser.dialect";

macro_rules! dialect_display_list {
    ($($display_name:literal),+ $(,)?) => {
        dialect_display_list!(@acc [] $($display_name),+)
    };
    (@acc [$($acc:tt)*] $last:literal) => {
        concat!($($acc)* $last)
    };
    (@acc [$($acc:tt)*] $next:literal, $($rest:literal),+) => {
        dialect_display_list!(@acc [$($acc)* $next, ", ",] $($rest),+)
    };
}

macro_rules! dialect_metadata {
    (
        default: $default_variant:ident;
        $(
            $variant:ident {
                canonical_name: $canonical_name:literal,
                display_name: $display_name:literal,
                aliases: [$($alias:literal),* $(,)?],
            }
        ),+ $(,)?
    ) => {
        /// This is the SQL dialect used by DataFusion's parser.
        /// This mirrors [sqlparser::dialect::Dialect](https://docs.rs/sqlparser/latest/sqlparser/dialect/trait.Dialect.html)
        /// trait in order to offer an easier API and avoid adding the `sqlparser` dependency
        #[derive(Debug, Clone, Copy, PartialEq, Eq)]
        pub enum Dialect {
            $($variant,)+
        }

        impl Default for Dialect {
            fn default() -> Self {
                Self::$default_variant
            }
        }

        const DIALECT_INFOS: &[DialectInfo] = &[
            $(
                DialectInfo {
                    dialect: Dialect::$variant,
                    canonical_name: $canonical_name,
                    display_name: $display_name,
                    aliases: &[$($alias),*],
                },
            )+
        ];

        const AVAILABLE_DIALECTS: &str = dialect_display_list!($($display_name),+);
        const DIALECT_CONFIG_DESCRIPTION: &str = concat!(
            "Configure the SQL dialect used by DataFusion's parser; supported values include: ",
            dialect_display_list!($($display_name),+),
            "."
        );
    };
}

dialect_metadata! {
    default: Generic;
    Generic {
        canonical_name: "generic",
        display_name: "Generic",
        aliases: [],
    },
    MySQL {
        canonical_name: "mysql",
        display_name: "MySQL",
        aliases: [],
    },
    PostgreSQL {
        canonical_name: "postgresql",
        display_name: "PostgreSQL",
        aliases: ["postgres"],
    },
    Hive {
        canonical_name: "hive",
        display_name: "Hive",
        aliases: [],
    },
    SQLite {
        canonical_name: "sqlite",
        display_name: "SQLite",
        aliases: [],
    },
    Snowflake {
        canonical_name: "snowflake",
        display_name: "Snowflake",
        aliases: [],
    },
    Redshift {
        canonical_name: "redshift",
        display_name: "Redshift",
        aliases: [],
    },
    MsSQL {
        canonical_name: "mssql",
        display_name: "MsSQL",
        aliases: [],
    },
    ClickHouse {
        canonical_name: "clickhouse",
        display_name: "ClickHouse",
        aliases: [],
    },
    BigQuery {
        canonical_name: "bigquery",
        display_name: "BigQuery",
        aliases: [],
    },
    Ansi {
        canonical_name: "ansi",
        display_name: "Ansi",
        aliases: [],
    },
    DuckDB {
        canonical_name: "duckdb",
        display_name: "DuckDB",
        aliases: [],
    },
    Databricks {
        canonical_name: "databricks",
        display_name: "Databricks",
        aliases: [],
    },
    Spark {
        canonical_name: "spark",
        display_name: "Spark",
        aliases: ["sparksql"],
    },
}

impl Dialect {
    /// Return metadata for all supported dialects.
    pub fn metadata() -> &'static [DialectInfo] {
        DIALECT_INFOS
    }

    /// Return all supported dialect names, for use in error messages.
    pub fn available() -> &'static str {
        AVAILABLE_DIALECTS
    }

    fn info(&self) -> &'static DialectInfo {
        DIALECT_INFOS
            .iter()
            .find(|info| info.dialect == *self)
            .expect("all Dialect variants are listed in DIALECT_INFOS")
    }
}

impl AsRef<str> for Dialect {
    fn as_ref(&self) -> &str {
        self.info().canonical_name
    }
}

impl FromStr for Dialect {
    type Err = DataFusionError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        for info in DIALECT_INFOS {
            if info.canonical_name.eq_ignore_ascii_case(s)
                || info
                    .aliases
                    .iter()
                    .any(|alias| alias.eq_ignore_ascii_case(s))
            {
                return Ok(info.dialect);
            }
        }

        Err(DataFusionError::Configuration(format!(
            "Invalid Dialect: {s}. Expected one of: {}",
            Self::available()
        )))
    }
}

impl ConfigField for Dialect {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        let description = if key == SQL_PARSER_DIALECT_CONFIG_KEY {
            DIALECT_CONFIG_DESCRIPTION
        } else {
            description
        };
        v.some(key, self, description)
    }

    fn set(&mut self, _: &str, value: &str) -> Result<()> {
        *self = Self::from_str(value)?;
        Ok(())
    }
}

impl Display for Dialect {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let str = self.as_ref();
        write!(f, "{str}")
    }
}

#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub enum SpillCompression {
    Zstd,
    Lz4Frame,
    #[default]
    Uncompressed,
}

impl FromStr for SpillCompression {
    type Err = DataFusionError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s.to_ascii_lowercase().as_str() {
            "zstd" => Ok(Self::Zstd),
            "lz4_frame" => Ok(Self::Lz4Frame),
            "uncompressed" | "" => Ok(Self::Uncompressed),
            other => Err(DataFusionError::Configuration(format!(
                "Invalid Spill file compression type: {other}. Expected one of: zstd, lz4_frame, uncompressed"
            ))),
        }
    }
}

impl ConfigField for SpillCompression {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        v.some(key, self, description)
    }

    fn set(&mut self, _: &str, value: &str) -> Result<()> {
        *self = SpillCompression::from_str(value)?;
        Ok(())
    }
}

impl Display for SpillCompression {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let str = match self {
            Self::Zstd => "zstd",
            Self::Lz4Frame => "lz4_frame",
            Self::Uncompressed => "uncompressed",
        };
        write!(f, "{str}")
    }
}

/// A `usize` configuration value that rejects zero when set from strings.
///
/// Use this for options where zero is never a meaningful runtime value.
/// Invalid values return a configuration error through [`ConfigField`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ConfigNonZeroUsize(NonZeroUsize);

/// Private helper for hard-coded defaults in `config_namespace!`, which cannot
/// use `?`. All external construction should use [`ConfigNonZeroUsize::try_new`].
const fn non_zero_usize_default(value: usize) -> ConfigNonZeroUsize {
    match NonZeroUsize::new(value) {
        Some(value) => ConfigNonZeroUsize(value),
        None => panic!("value must be greater than 0"),
    }
}

impl ConfigNonZeroUsize {
    /// Creates a [`ConfigNonZeroUsize`], returning a configuration error if
    /// `value` is zero.
    pub fn try_new(value: usize) -> Result<Self> {
        NonZeroUsize::new(value)
            .map(Self)
            .ok_or_else(|| _config_datafusion_err!("value must be greater than 0"))
    }

    /// Returns the wrapped `usize`.
    pub const fn get(self) -> usize {
        self.0.get()
    }
}

impl From<ConfigNonZeroUsize> for usize {
    fn from(value: ConfigNonZeroUsize) -> Self {
        value.get()
    }
}

impl FromStr for ConfigNonZeroUsize {
    type Err = DataFusionError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        Self::try_new(default_config_transform(s)?)
    }
}

impl ConfigField for ConfigNonZeroUsize {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        v.some(key, self, description)
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        if !key.is_empty() {
            return _config_err!(
                "Config field batch_size is a scalar ConfigNonZeroUsize and does not have nested field \"{}\"",
                key
            );
        }

        *self = ConfigNonZeroUsize::from_str(value)?;
        Ok(())
    }

    fn reset(&mut self, key: &str) -> Result<()> {
        if key.is_empty() {
            Ok(())
        } else {
            _config_err!(
                "Config field batch_size is a scalar ConfigNonZeroUsize and does not have nested field \"{}\"",
                key
            )
        }
    }
}

impl Display for ConfigNonZeroUsize {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.get())
    }
}

/// A `usize` configuration value that rejects 0 and 1 when set from strings.
///
/// Use this for options whose consumer divides the value in half to size an
/// internal buffer (e.g. a bounded channel capacity): values below 2 would
/// round down to a zero-capacity buffer and panic. Invalid values return a
/// configuration error through [`ConfigField`] instead.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ConfigMinTwoUsize(usize);

/// Private helper for hard-coded defaults in `config_namespace!`, which cannot
/// use `?`. All external construction should use [`ConfigMinTwoUsize::try_new`].
const fn min_two_usize_default(value: usize) -> ConfigMinTwoUsize {
    if value >= 2 {
        ConfigMinTwoUsize(value)
    } else {
        panic!("value must be at least 2")
    }
}

impl ConfigMinTwoUsize {
    /// Creates a [`ConfigMinTwoUsize`], returning a configuration error if
    /// `value` is less than 2.
    pub fn try_new(value: usize) -> Result<Self> {
        if value >= 2 {
            Ok(Self(value))
        } else {
            _config_err!("value must be at least 2")
        }
    }

    /// Returns the wrapped `usize`.
    pub const fn get(self) -> usize {
        self.0
    }
}

impl From<ConfigMinTwoUsize> for usize {
    fn from(value: ConfigMinTwoUsize) -> Self {
        value.get()
    }
}

impl FromStr for ConfigMinTwoUsize {
    type Err = DataFusionError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        Self::try_new(default_config_transform(s)?)
    }
}

impl ConfigField for ConfigMinTwoUsize {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        v.some(key, self, description)
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        if !key.is_empty() {
            return _config_err!(
                "Config field max_buffered_batches_per_output_file is a scalar ConfigMinTwoUsize and does not have nested field \"{}\"",
                key
            );
        }

        *self = ConfigMinTwoUsize::from_str(value)?;
        Ok(())
    }

    fn reset(&mut self, key: &str) -> Result<()> {
        if key.is_empty() {
            Ok(())
        } else {
            _config_err!(
                "Config field max_buffered_batches_per_output_file is a scalar ConfigMinTwoUsize and does not have nested field \"{}\"",
                key
            )
        }
    }
}

impl Display for ConfigMinTwoUsize {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.get())
    }
}

/// Policy for handling duplicate keys in Spark-compatible map-construction
/// functions (`map_from_arrays`, `map_from_entries`, `str_to_map`). Mirrors
/// Spark's [`spark.sql.mapKeyDedupPolicy`](https://github.com/apache/spark/blob/cf3a34e19dfcf70e2d679217ff1ba21302212472/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L4961).
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
pub enum MapKeyDedupPolicy {
    /// Raise `[DUPLICATED_MAP_KEY]` at runtime on any duplicate key.
    #[default]
    Exception,
    /// Keep the last occurrence of each duplicate key.
    LastWin,
}

impl FromStr for MapKeyDedupPolicy {
    type Err = DataFusionError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        match s.to_ascii_uppercase().as_str() {
            "EXCEPTION" => Ok(Self::Exception),
            "LAST_WIN" => Ok(Self::LastWin),
            other => Err(DataFusionError::Configuration(format!(
                "Invalid MapKeyDedupPolicy: {other}. Expected one of: EXCEPTION, LAST_WIN"
            ))),
        }
    }
}

impl ConfigField for MapKeyDedupPolicy {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        v.some(key, self, description)
    }

    fn set(&mut self, _: &str, value: &str) -> Result<()> {
        *self = MapKeyDedupPolicy::from_str(value)?;
        Ok(())
    }
}

impl Display for MapKeyDedupPolicy {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let str = match self {
            Self::Exception => "EXCEPTION",
            Self::LastWin => "LAST_WIN",
        };
        write!(f, "{str}")
    }
}

impl From<SpillCompression> for Option<CompressionType> {
    fn from(c: SpillCompression) -> Self {
        match c {
            SpillCompression::Zstd => Some(CompressionType::ZSTD),
            SpillCompression::Lz4Frame => Some(CompressionType::LZ4_FRAME),
            SpillCompression::Uncompressed => None,
        }
    }
}

config_namespace! {
    /// Options related to query execution
    ///
    /// See also: [`SessionConfig`]
    ///
    /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
    pub struct ExecutionOptions {
        /// Default batch size while creating new batches, it's especially useful for
        /// buffer-in-memory batches since creating tiny batches would result in too much
        /// metadata memory consumption
        pub batch_size: ConfigNonZeroUsize, default = non_zero_usize_default(8192)

        /// A perfect hash join (see `HashJoinExec` for more details) will be considered
        /// if the range of keys (max - min) on the build side is < this threshold.
        /// This provides a fast path for joins with very small key ranges,
        /// bypassing the density check.
        ///
        /// Currently only supports cases where build_side.num_rows() < u32::MAX.
        /// Support for build_side.num_rows() >= u32::MAX will be added in the future.
        pub perfect_hash_join_small_build_threshold: usize, default = 1024

        /// The minimum required density of join keys on the build side to consider a
        /// perfect hash join (see `HashJoinExec` for more details). Density is calculated as:
        /// `(number of rows) / (max_key - min_key + 1)`.
        /// A perfect hash join may be used if the actual key density > this
        /// value.
        ///
        /// Currently only supports cases where build_side.num_rows() < u32::MAX.
        /// Support for build_side.num_rows() >= u32::MAX will be added in the future.
        pub perfect_hash_join_min_key_density: f64, default = 0.15

        /// When set to true, record batches will be examined between each operator and
        /// small batches will be coalesced into larger batches. This is helpful when there
        /// are highly selective filters or joins that could produce tiny output batches. The
        /// target batch size is determined by the configuration setting
        pub coalesce_batches: bool, default = true

        /// Should DataFusion collect statistics when first creating a table.
        /// Has no effect after the table is created. Defaults to true.
        pub collect_statistics: bool, default = true

        /// Number of partitions for query execution. Increasing partitions can increase
        /// concurrency.
        ///
        /// Defaults to the number of CPU cores on the system
        pub target_partitions: usize, transform = ExecutionOptions::normalized_parallelism, default = get_available_parallelism()

        /// The default time zone
        ///
        /// Some functions, e.g. `now` return timestamps in this time zone
        pub time_zone: Option<String>, default = None

        /// Parquet options
        pub parquet: ParquetOptions, default = Default::default()

        /// Fan-out during initial physical planning.
        ///
        /// This is mostly use to plan `UNION` children in parallel.
        ///
        /// Defaults to the number of CPU cores on the system
        pub planning_concurrency: usize, transform = ExecutionOptions::normalized_parallelism, default = get_available_parallelism()

        /// When set to true, skips verifying that the schema produced by
        /// planning the input of `LogicalPlan::Aggregate` exactly matches the
        /// schema of the input plan.
        ///
        /// When set to false, if the schema does not match exactly
        /// (including nullability and metadata), a planning error will be raised.
        ///
        /// This is used to workaround bugs in the planner that are now caught by
        /// the new schema verification step.
        pub skip_physical_aggregate_schema_check: bool, default = false

        /// Temporary switch for aggregate stream implementations that are being
        /// migrated from `GroupedHashAggregateStream`.
        ///
        /// When set to true, DataFusion tries the migrated implementations when
        /// their preconditions are satisfied. When set to false, grouped
        /// aggregation falls back to `GroupedHashAggregateStream`. This option
        /// will be removed after the migration is finished.
        ///
        /// See <https://github.com/apache/datafusion/issues/22710> for details.
        pub enable_migration_aggregate: bool, default = true

        /// Sets the compression codec used when spilling data to disk.
        ///
        /// Since datafusion writes spill files using the Arrow IPC Stream format,
        /// only codecs supported by the Arrow IPC Stream Writer are allowed.
        /// Valid values are: uncompressed, lz4_frame, zstd.
        /// Note: lz4_frame offers faster (de)compression, but typically results in
        /// larger spill files. In contrast, zstd achieves
        /// higher compression ratios at the cost of slower (de)compression speed.
        pub spill_compression: SpillCompression, default = SpillCompression::Uncompressed

        /// Specifies the reserved memory for each spillable sort operation to
        /// facilitate an in-memory merge.
        ///
        /// When a sort operation spills to disk, the in-memory data must be
        /// sorted and merged before being written to a file. This setting reserves
        /// a specific amount of memory for that in-memory sort/merge process.
        ///
        /// Note: This setting is irrelevant if the sort operation cannot spill
        /// (i.e., if there's no `DiskManager` configured).
        pub sort_spill_reservation_bytes: usize, default = 10 * 1024 * 1024

        /// When sorting, below what size should data be concatenated
        /// and sorted in a single RecordBatch rather than sorted in
        /// batches and merged.
        pub sort_in_place_threshold_bytes: usize, default = 1024 * 1024

        /// Maximum buffer capacity (in bytes) per partition for BufferExec
        /// inserted during sort pushdown optimization.
        ///
        /// When PushdownSort eliminates a SortExec under SortPreservingMergeExec,
        /// a BufferExec is inserted to replace SortExec's buffering role. This
        /// prevents I/O stalls by allowing the scan to run ahead of the merge.
        ///
        /// This uses strictly less memory than the SortExec it replaces (which
        /// buffers the entire partition). The buffer respects the global memory
        /// pool limit. Setting this to a large value is safe — actual memory
        /// usage is bounded by partition size and global memory limits.
        pub sort_pushdown_buffer_capacity: usize, default = 1024 * 1024 * 1024

        /// Maximum size in bytes for individual spill files before rotating to a new file.
        ///
        /// When operators spill data to disk (e.g., RepartitionExec), they write
        /// multiple batches to the same file until this size limit is reached, then rotate
        /// to a new file. This reduces syscall overhead compared to one-file-per-batch
        /// while preventing files from growing too large.
        ///
        /// A larger value reduces file creation overhead but may hold more disk space.
        /// A smaller value creates more files but allows finer-grained space reclamation
        /// as files can be deleted once fully consumed.
        ///
        /// Now only `RepartitionExec` supports this spill file rotation feature, other spilling operators
        /// may create spill files larger than the limit.
        ///
        /// Default: 128 MB
        pub max_spill_file_size_bytes: ConfigNonZeroUsize, default = non_zero_usize_default(128 * 1024 * 1024)

        /// Number of files to read in parallel when inferring schema and statistics
        pub meta_fetch_concurrency: ConfigNonZeroUsize, default = non_zero_usize_default(32)

        /// Guarantees a minimum level of output files running in parallel.
        /// RecordBatches will be distributed in round robin fashion to each
        /// parallel writer. Each writer is closed and a new file opened once
        /// soft_max_rows_per_output_file is reached.
        pub minimum_parallel_output_files: ConfigNonZeroUsize, default = non_zero_usize_default(4)

        /// Target number of rows in output files when writing multiple.
        /// This is a soft max, so it can be exceeded slightly. There also
        /// will be one file smaller than the limit if the total
        /// number of rows written is not roughly divisible by the soft max
        pub soft_max_rows_per_output_file: ConfigNonZeroUsize, default = non_zero_usize_default(50000000)

        /// This is the maximum number of RecordBatches buffered
        /// for each output file being worked. Higher values can potentially
        /// give faster write performance at the cost of higher peak
        /// memory consumption.
        ///
        /// This budget is split evenly between two independent points in the
        /// write pipeline (see the demuxer diagram in #7791): how many files
        /// can be in flight from the demuxer to a writer task, and how many
        /// RecordBatches are buffered for a single file's writer. Must be at
        /// least 2 so each half gets at least 1 unit of buffering - 0 or 1
        /// would leave one side with a zero-capacity channel and panic at
        /// write time.
        pub max_buffered_batches_per_output_file: ConfigMinTwoUsize, default = min_two_usize_default(2)

        /// Should sub directories be ignored when scanning directories for data
        /// files. Defaults to true (ignores subdirectories), consistent with
        /// Hive. Note that this setting does not affect reading partitioned
        /// tables (e.g. `/table/year=2021/month=01/data.parquet`).
        pub listing_table_ignore_subdirectory: bool, default = true

        /// Should a `ListingTable` created through the `ListingTableFactory` infer table
        /// partitions from Hive compliant directories. Defaults to true (partition columns are
        /// inferred and will be represented in the table schema).
        pub listing_table_factory_infer_partitions: bool, default = true

        /// Should DataFusion support recursive CTEs
        pub enable_recursive_ctes: bool, default = true

        /// Attempt to eliminate sorts by packing & sorting files with non-overlapping
        /// statistics into the same file groups.
        /// Currently experimental
        pub split_file_groups_by_statistics: bool, default = false

        /// Should DataFusion keep the columns used for partition_by in the output RecordBatches
        pub keep_partition_by_columns: bool, default = false

        /// When `true` (the default), DataFusion's built-in file scans
        /// dynamically rebalance files across partitions at query execution
        /// time: a partition that goes idle reads files (or byte-range morsels)
        /// originally assigned to a sibling partition, which keeps all
        /// partitions busy in a single process.
        ///
        /// Executors that depend on the plan-time partition assignment — such as
        /// Ballista and datafusion-distributed, which run each partition as an
        /// isolated task and never poll the siblings — should set this to
        /// `false` so each partition reads only its own file group and no
        /// runtime reassignment occurs.
        pub enable_file_stream_work_stealing: bool, default = true

        /// Aggregation ratio (number of distinct groups / number of input rows)
        /// threshold for skipping partial aggregation. If the value is greater
        /// then partial aggregation will skip aggregation for further input
        pub skip_partial_aggregation_probe_ratio_threshold: f64, default = 0.8

        /// Number of input rows partial aggregation partition should process, before
        /// aggregation ratio check and trying to switch to skipping aggregation mode
        pub skip_partial_aggregation_probe_rows_threshold: usize, default = 100_000

        /// Should DataFusion use row number estimates at the input to decide
        /// whether increasing parallelism is beneficial or not. By default,
        /// only exact row numbers (not estimates) are used for this decision.
        /// Setting this flag to `true` will likely produce better plans.
        /// if the source of statistics is accurate.
        /// We plan to make this the default in the future.
        pub use_row_number_estimates_to_optimize_partitioning: bool, default = false

        /// Should DataFusion enforce batch size in joins or not. By default,
        /// DataFusion will not enforce batch size in joins. Enforcing batch size
        /// in joins can reduce memory usage when joining large
        /// tables with a highly-selective join filter, but is also slightly slower.
        pub enforce_batch_size_in_joins: bool, default = false

        /// Size (bytes) of data buffer DataFusion uses when writing output files.
        /// This affects the size of the data chunks that are uploaded to remote
        /// object stores (e.g. AWS S3). If very large (>= 100 GiB) output files are being
        /// written, it may be necessary to increase this size to avoid errors from
        /// the remote end point.
        pub objectstore_writer_buffer_size: usize, default = 10 * 1024 * 1024

        /// Whether to enable ANSI SQL mode.
        ///
        /// The flag is experimental and relevant only for DataFusion Spark built-in functions
        ///
        /// When `enable_ansi_mode` is set to `true`, the query engine follows ANSI SQL
        /// semantics for expressions, casting, and error handling. This means:
        /// - **Strict type coercion rules:** implicit casts between incompatible types are disallowed.
        /// - **Standard SQL arithmetic behavior:** operations such as division by zero,
        ///   numeric overflow, or invalid casts raise runtime errors rather than returning
        ///   `NULL` or adjusted values.
        /// - **Consistent ANSI behavior** for string concatenation, comparisons, and `NULL` handling.
        ///
        /// When `enable_ansi_mode` is `false` (the default), the engine uses a more permissive,
        /// non-ANSI mode designed for user convenience and backward compatibility. In this mode:
        /// - Implicit casts between types are allowed (e.g., string to integer when possible).
        /// - Arithmetic operations are more lenient — for example, `abs()` on the minimum
        ///   representable integer value returns the input value instead of raising overflow.
        /// - Division by zero or invalid casts may return `NULL` instead of failing.
        ///
        /// # Default
        /// `false` — ANSI SQL mode is disabled by default.
        pub enable_ansi_mode: bool, default = false

        /// How many bytes to buffer in the probe side of hash joins while the build side is
        /// concurrently being built.
        ///
        /// Without this, hash joins will wait until the full materialization of the build side
        /// before polling the probe side. This is useful in scenarios where the query is not
        /// completely CPU bounded, allowing to do some early work concurrently and reducing the
        /// latency of the query.
        ///
        /// Note that when hash join buffering is enabled, the probe side will start eagerly
        /// polling data, not giving time for the producer side of dynamic filters to produce any
        /// meaningful predicate. Queries with dynamic filters might see performance degradation.
        ///
        /// Disabled by default, set to a number greater than 0 for enabling it.
        pub hash_join_buffering_capacity: usize, default = 0
    }
}

config_namespace! {
    /// Options for content-defined chunking (CDC) when writing parquet files.
    /// Mirrors `parquet::file::properties::CdcOptions`.
    ///
    /// Carried as a [`ParquetCdcOptions`] in [`ParquetOptions::content_defined_chunking`]
    /// with an explicit `enabled` flag, so it can be toggled with dotted config
    /// keys (`content_defined_chunking.enabled = true|false`) and the result is
    /// independent of the order in which the keys are set.
    pub struct ParquetCdcOptions {
        /// (writing) EXPERIMENTAL: Enable content-defined chunking (CDC) when writing
        /// parquet files. When enabled, parallel writing is automatically disabled
        /// since the chunker state must persist across row groups.
        pub enabled: bool, default = false

        /// Minimum chunk size in bytes. The rolling hash will not trigger a split
        /// until this many bytes have been accumulated. Default is 256 KiB.
        pub min_chunk_size: usize, default = 256 * 1024

        /// Maximum chunk size in bytes. A split is forced when the accumulated
        /// size exceeds this value. Default is 1 MiB.
        pub max_chunk_size: usize, default = 1024 * 1024

        /// Normalization level. Increasing this improves deduplication ratio
        /// but increases fragmentation. Recommended range is [-3, 3], default is 0.
        pub norm_level: i32, default = 0
    }
}

impl ParquetCdcOptions {
    /// Returns enabled CDC options with the default chunking parameters.
    ///
    /// Shorthand for `ParquetCdcOptions { enabled: true, ..Default::default() }`;
    /// combine with struct-update syntax to override parameters, e.g.
    /// `ParquetCdcOptions { min_chunk_size: 4096, ..ParquetCdcOptions::enabled() }`.
    pub fn enabled() -> Self {
        Self {
            enabled: true,
            ..Default::default()
        }
    }

    /// Returns disabled CDC options (equivalent to [`ParquetCdcOptions::default`]).
    pub fn disabled() -> Self {
        Self::default()
    }
}

/// Target maximum size of a Parquet row group in bytes.
///
/// Wraps a `usize` so the "must be greater than zero" constraint (arrow-rs
/// panics on a zero byte limit) is validated when the config is set, rather
/// than when the writer properties are built.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MaxRowGroupBytes(usize);

impl MaxRowGroupBytes {
    /// Creates a `MaxRowGroupBytes`, rejecting zero.
    pub fn try_new(value: usize) -> Result<Self> {
        if value == 0 {
            return Err(DataFusionError::Configuration(
                "max_row_group_bytes must be greater than 0".to_string(),
            ));
        }
        Ok(Self(value))
    }

    /// Returns the configured byte limit.
    pub fn get(&self) -> usize {
        self.0
    }
}

impl FromStr for MaxRowGroupBytes {
    type Err = DataFusionError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        let value = s.parse::<usize>().map_err(|_| {
            DataFusionError::Configuration(format!(
                "Invalid max_row_group_bytes: '{s}'. Expected a positive integer."
            ))
        })?;
        Self::try_new(value)
    }
}

impl Display for MaxRowGroupBytes {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "{}", self.0)
    }
}

/// `ConfigField` for `Option<MaxRowGroupBytes>`. A custom impl (rather than the
/// blanket `Option<F>` one) so an invalid value is rejected without leaving the
/// option in an invalid intermediate state on error. `MaxRowGroupBytes`
/// deliberately does not implement `Default`, so the blanket impl does not apply.
impl ConfigField for Option<MaxRowGroupBytes> {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        match self {
            Some(s) => v.some(key, s, description),
            None => v.none(key, description),
        }
    }

    fn set(&mut self, _key: &str, value: &str) -> Result<()> {
        *self = Some(MaxRowGroupBytes::from_str(value)?);
        Ok(())
    }

    fn reset(&mut self, _key: &str) -> Result<()> {
        *self = None;
        Ok(())
    }
}

config_namespace! {
    /// Options for reading and writing parquet files
    ///
    /// See also: [`SessionConfig`]
    ///
    /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
    pub struct ParquetOptions {
        // The following options affect reading parquet files

        /// (reading) If true, reads the Parquet data page level metadata (the
        /// Page Index), if present, to reduce the I/O and number of
        /// rows decoded.
        pub enable_page_index: bool, default = true

        /// (reading) If true, the parquet reader attempts to skip entire row groups based
        /// on the predicate in the query and the metadata (min/max values) stored in
        /// the parquet file
        pub pruning: bool, default = true

        /// (reading) If true, the parquet reader skip the optional embedded metadata that may be in
        /// the file Schema. This setting can help avoid schema conflicts when querying
        /// multiple parquet files with schemas containing compatible types but different metadata
        pub skip_metadata: bool, default = true

        /// (reading) If specified, the parquet reader will try and fetch the last `size_hint`
        /// bytes of the parquet file optimistically. If not specified, two reads are required:
        /// One read to fetch the 8-byte parquet footer and
        /// another to fetch the metadata length encoded in the footer
        /// Default setting to 512 KiB, which should be sufficient for most parquet files,
        /// it can reduce one I/O operation per parquet file. If the metadata is larger than
        /// the hint, two reads will still be performed.
        pub metadata_size_hint: Option<usize>, default = Some(512 * 1024)

        /// (reading) If true, filter expressions are be applied during the parquet decoding operation to
        /// reduce the number of rows decoded. This optimization is sometimes called "late materialization".
        pub pushdown_filters: bool, default = false

        /// (reading) If true, filter expressions evaluated during the parquet decoding operation
        /// will be reordered heuristically to minimize the cost of evaluation. If false,
        /// the filters are applied in the same order as written in the query
        pub reorder_filters: bool, default = false

        /// (reading) Force the use of RowSelections for filter results, when
        /// pushdown_filters is enabled. If false, the reader will automatically
        /// choose between a RowSelection and a Bitmap based on the number and
        /// pattern of selected rows.
        pub force_filter_selections: bool, default = false

        /// (reading) If true, parquet reader will read columns of `Utf8/Utf8Large` with `Utf8View`,
        /// and `Binary/BinaryLarge` with `BinaryView`.
        pub schema_force_view_types: bool, default = true

        /// (reading) If true, parquet reader will read columns of
        /// `Binary/LargeBinary` with `Utf8`, and `BinaryView` with `Utf8View`.
        ///
        /// Parquet files generated by some legacy writers do not correctly set
        /// the UTF8 flag for strings, causing string columns to be loaded as
        /// BLOB instead.
        pub binary_as_string: bool, default = false

        /// (reading) If true, parquet reader will read columns of
        /// physical type int96 as originating from a different resolution
        /// than nanosecond. This is useful for reading data from systems like Spark
        /// which stores microsecond resolution timestamps in an int96 allowing it
        /// to write values with a larger date range than 64-bit timestamps with
        /// nanosecond resolution.
        pub coerce_int96: Option<String>, transform = str::to_lowercase, default = None

        /// (reading) Optional timezone applied to INT96 columns when `coerce_int96`
        /// is set. When `Some`, INT96 columns coerce to
        /// `Timestamp(<coerce_int96>, Some(<tz>))` instead of the default
        /// `Timestamp(<coerce_int96>, None)`. Spark and other systems write INT96
        /// values as UTC-adjusted instants, so callers that need the resulting
        /// Arrow type to be timezone-aware (e.g. for Spark `TimestampType`
        /// semantics) should set this to `"UTC"`. No effect when `coerce_int96`
        /// is `None`.
        pub coerce_int96_tz: Option<String>, default = None

        /// (reading) Use any available bloom filters when reading parquet files
        pub bloom_filter_on_read: bool, default = true

        /// (reading) The maximum predicate cache size, in bytes. When
        /// `pushdown_filters` is enabled, sets the maximum memory used to cache
        /// the results of predicate evaluation between filter evaluation and
        /// output generation. Decreasing this value will reduce memory usage,
        /// but may increase IO and CPU usage. None means use the default
        /// parquet reader setting. 0 means no caching.
        pub max_predicate_cache_size: Option<usize>, default = None

        /// Maximum number of values in an `IN (...)` list for which pruning will
        /// occur. Longer lists will not be used to prune files, row groups, or
        /// data pages.
        ///
        /// Higher values help in cases such as filtering on a list of
        /// ~25-100 identifiers, but also make the predicate more expensive to
        /// evaluate. Set to 0 to disable `IN (...)` list pruning entirely.
        ///
        /// Defaults to 20.
        pub max_in_list_size: usize, default = 20

        // The following options affect writing to parquet files
        // and map to parquet::file::properties::WriterProperties

        /// (writing) Sets best effort maximum size of data page in bytes
        pub data_pagesize_limit: usize, default = 1024 * 1024

        /// (writing) Sets write_batch_size in rows
        pub write_batch_size: usize, default = 1024

        /// (writing) Sets parquet writer version
        /// valid values are "1.0" and "2.0"
        pub writer_version: DFParquetWriterVersion, default = DFParquetWriterVersion::default()

        /// (writing) Skip encoding the embedded arrow metadata in the KV_meta
        ///
        /// This is analogous to the `ArrowWriterOptions::with_skip_arrow_metadata`.
        /// Refer to <https://docs.rs/parquet/53.3.0/parquet/arrow/arrow_writer/struct.ArrowWriterOptions.html#method.with_skip_arrow_metadata>
        pub skip_arrow_metadata: bool, default = false

        /// (writing) Sets default parquet compression codec.
        /// Valid values are: uncompressed, snappy, gzip(level),
        /// brotli(level), lz4, zstd(level), and lz4_raw.
        /// These values are not case sensitive. If NULL, uses
        /// default parquet writer setting
        ///
        /// Note that this default setting is not the same as
        /// the default parquet writer setting.
        pub compression: Option<String>, transform = str::to_lowercase, default = Some("zstd(3)".into())

        /// (writing) Sets if dictionary encoding is enabled. If NULL, uses
        /// default parquet writer setting
        pub dictionary_enabled: Option<bool>, default = Some(true)

        /// (writing) Sets best effort maximum dictionary page size, in bytes
        pub dictionary_page_size_limit: usize, default = 1024 * 1024

        /// (writing) Sets if statistics are enabled for any column
        /// Valid values are: "none", "chunk", and "page"
        /// These values are not case sensitive. If NULL, uses
        /// default parquet writer setting
        pub statistics_enabled: Option<String>, transform = str::to_lowercase, default = Some("page".into())

        /// (writing) Target maximum number of rows in each row group (defaults to 1M
        /// rows). Writing larger row groups requires more memory to write, but
        /// can get better compression and be faster to read. When
        /// `max_row_group_bytes` is also set, the writer flushes a row group when
        /// either limit is reached, whichever comes first.
        pub max_row_group_size: usize, default =  1024 * 1024

        /// (writing) Target maximum size of each row group in bytes. When set,
        /// the writer flushes whenever either this limit or `max_row_group_size`
        /// is reached, whichever comes first. Useful for bounding writer memory
        /// on wide schemas where a row-count limit can map to very different
        /// byte sizes. Matches the behavior of `parquet.block.size` in
        /// parquet-mr. If `None` (the default), only the row-count limit
        /// applies. Currently only honored when `allow_single_file_parallelism`
        /// is `false`; by default the parallel file writer ignores this limit.
        pub max_row_group_bytes: Option<MaxRowGroupBytes>, default = None

        /// (writing) Sets "created by" property
        pub created_by: String, default = concat!("datafusion version ", env!("CARGO_PKG_VERSION")).into()

        /// (writing) Sets column index truncate length
        pub column_index_truncate_length: Option<usize>, default = Some(64)

        /// (writing) Sets statistics truncate length. If NULL, uses
        /// default parquet writer setting
        pub statistics_truncate_length: Option<usize>, default = Some(64)

        /// (writing) Sets best effort maximum number of rows in data page
        pub data_page_row_count_limit: usize, default = 20_000

        /// (writing)  Sets default encoding for any column.
        /// Valid values are: plain, plain_dictionary, rle,
        /// bit_packed, delta_binary_packed, delta_length_byte_array,
        /// delta_byte_array, rle_dictionary, and byte_stream_split.
        /// These values are not case sensitive. If NULL, uses
        /// default parquet writer setting
        pub encoding: Option<String>, transform = str::to_lowercase, default = None

        /// (writing) Write bloom filters for all columns when creating parquet files
        pub bloom_filter_on_write: bool, default = false

        /// (writing) Sets bloom filter false positive probability. If NULL, uses
        /// default parquet writer setting
        pub bloom_filter_fpp: Option<f64>, default = None

        /// (writing) Sets bloom filter number of distinct values. If NULL, uses
        /// default parquet writer setting
        pub bloom_filter_ndv: Option<u64>, default = None

        /// (writing) Controls whether DataFusion will attempt to speed up writing
        /// parquet files by serializing them in parallel. Each column
        /// in each row group in each output file are serialized in parallel
        /// leveraging a maximum possible core count of n_files*n_row_groups*n_columns.
        pub allow_single_file_parallelism: bool, default = true

        /// (writing) By default parallel parquet writer is tuned for minimum
        /// memory usage in a streaming execution plan. You may see
        /// a performance benefit when writing large parquet files
        /// by increasing maximum_parallel_row_group_writers and
        /// maximum_buffered_record_batches_per_stream if your system
        /// has idle cores and can tolerate additional memory usage.
        /// Boosting these values is likely worthwhile when
        /// writing out already in-memory data, such as from a cached
        /// data frame.
        pub maximum_parallel_row_group_writers: usize, default = 1

        /// (writing) By default parallel parquet writer is tuned for minimum
        /// memory usage in a streaming execution plan. You may see
        /// a performance benefit when writing large parquet files
        /// by increasing maximum_parallel_row_group_writers and
        /// maximum_buffered_record_batches_per_stream if your system
        /// has idle cores and can tolerate additional memory usage.
        /// Boosting these values is likely worthwhile when
        /// writing out already in-memory data, such as from a cached
        /// data frame.
        pub maximum_buffered_record_batches_per_stream: usize, default = 2

        /// (writing) EXPERIMENTAL: Content-defined chunking (CDC) options when writing
        /// parquet files. Disabled by default; toggle with
        /// `content_defined_chunking.enabled = true|false`. The chunking parameters live
        /// under the same prefix (e.g. `content_defined_chunking.min_chunk_size`). When
        /// enabled, parallel writing is automatically disabled since the chunker state
        /// must persist across row groups. Mirrors
        /// `parquet::file::properties::WriterProperties::content_defined_chunking`.
        pub content_defined_chunking: ParquetCdcOptions, default = Default::default()
    }
}

config_namespace! {
    /// Options for configuring Parquet Modular Encryption
    ///
    /// To use Parquet encryption, you must enable the `parquet_encryption` feature flag, as it is not activated by default.
    pub struct ParquetEncryptionOptions {
        /// Optional file decryption properties
        pub file_decryption: Option<ConfigFileDecryptionProperties>, default = None

        /// Optional file encryption properties
        pub file_encryption: Option<ConfigFileEncryptionProperties>, default = None

        /// Identifier for the encryption factory to use to create file encryption and decryption properties.
        /// Encryption factories can be registered in the runtime environment with
        /// `RuntimeEnv::register_parquet_encryption_factory`.
        pub factory_id: Option<String>, default = None

        /// Any encryption factory specific options
        pub factory_options: EncryptionFactoryOptions, default = EncryptionFactoryOptions::default()
    }
}

impl ParquetEncryptionOptions {
    /// Specify the encryption factory to use for Parquet modular encryption, along with its configuration
    pub fn configure_factory(
        &mut self,
        factory_id: &str,
        config: &impl ExtensionOptions,
    ) {
        self.factory_id = Some(factory_id.to_owned());
        self.factory_options.options.clear();
        for entry in config.entries() {
            if let Some(value) = entry.value {
                self.factory_options.options.insert(entry.key, value);
            }
        }
    }
}

config_namespace! {
    /// Options related to query optimization
    ///
    /// See also: [`SessionConfig`]
    ///
    /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
    pub struct OptimizerOptions {
        /// When set to true, the optimizer will push a limit operation into
        /// grouped aggregations which have no aggregate expressions, as a soft limit,
        /// emitting groups once the limit is reached, before all rows in the group are read.
        pub enable_distinct_aggregation_soft_limit: bool, default = true

        /// When set to true, the physical plan optimizer will try to add round robin
        /// repartitioning to increase parallelism to leverage more CPU cores
        pub enable_round_robin_repartition: bool, default = true

        /// When set to true, the optimizer will attempt to perform limit operations
        /// during aggregations, if possible
        pub enable_topk_aggregation: bool, default = true

        /// When set to true, the optimizer will attempt to push limit operations
        /// past window functions, if possible
        pub enable_window_limits: bool, default = true

        /// When set to true, the optimizer will replace
        /// Filter(rn<=K) → Window(ROW_NUMBER) → Sort patterns with a
        /// PartitionedTopKExec that maintains per-partition heaps, avoiding
        /// a full sort of the input.
        /// When the window partition key has low cardinality, enabling this optimization
        /// can improve performance. However, for high cardinality keys, it may
        /// cause regressions in both memory usage and runtime.
        pub enable_window_topn: bool, default = false

        /// When set to true, the optimizer will push TopK (Sort with fetch)
        /// below hash repartition when the partition key is a prefix of the
        /// sort key, reducing data volume before the shuffle.
        pub enable_topk_repartition: bool, default = true

        /// When set to true, the optimizer will attempt to push down TopK dynamic filters
        /// into the file scan phase.
        pub enable_topk_dynamic_filter_pushdown: bool, default = true

        /// When set to true, uncorrelated scalar subqueries are
        /// left in the logical plan and executed by `ScalarSubqueryExec` during
        /// physical execution. When set to false, all scalar subqueries
        /// (including uncorrelated ones) are rewritten to left joins by the
        /// `ScalarSubqueryToJoin` optimizer rule.
        ///
        /// Note disabling this option is not recommended. It restores
        /// pre <https://github.com/apache/datafusion/pull/21240>
        /// behavior, which silently produces incorrect results for
        /// multi-row subqueries and does not support scalar subqueries in
        /// ORDER BY / JOIN ON / aggregate-function arguments. This option is
        /// intended as a temporary escape hatch for distributed execution
        /// frameworks and is planned to be removed in a future DataFusion
        /// release.
        pub enable_physical_uncorrelated_scalar_subquery: bool, default = true

        /// When set to true, the optimizer will attempt to push down Join dynamic filters
        /// into the file scan phase.
        pub enable_join_dynamic_filter_pushdown: bool, default = true

        /// When set to true, the optimizer will attempt to push down Aggregate dynamic filters
        /// into the file scan phase.
        pub enable_aggregate_dynamic_filter_pushdown: bool, default = true

        /// When set to true attempts to push down dynamic filters generated by operators (TopK, Join & Aggregate) into the file scan phase.
        /// For example, for a query such as `SELECT * FROM t ORDER BY timestamp DESC LIMIT 10`, the optimizer
        /// will attempt to push down the current top 10 timestamps that the TopK operator references into the file scans.
        /// This means that if we already have 10 timestamps in the year 2025
        /// any files that only have timestamps in the year 2024 can be skipped / pruned at various stages in the scan.
        /// The config will suppress `enable_join_dynamic_filter_pushdown`, `enable_topk_dynamic_filter_pushdown` & `enable_aggregate_dynamic_filter_pushdown`
        /// So if you disable `enable_topk_dynamic_filter_pushdown`, then enable `enable_dynamic_filter_pushdown`, the `enable_topk_dynamic_filter_pushdown` will be overridden.
        pub enable_dynamic_filter_pushdown: bool, default = true

        /// When set to true, the optimizer will insert filters before a join between
        /// a nullable and non-nullable column to filter out nulls on the nullable side. This
        /// filter can add additional overhead when the file format does not fully support
        /// predicate push down.
        pub filter_null_join_keys: bool, default = false

        /// Should DataFusion repartition data using the aggregate keys to execute aggregates
        /// in parallel using the provided `target_partitions` level
        pub repartition_aggregations: bool, default = true

        /// Minimum total file size in bytes for file-group byte-range
        /// splitting to fire. Files (or merged file groups) smaller than this
        /// stay as one partition. Lower values produce more, smaller
        /// partitions — better at filling `target_partitions` worth of cores
        /// when files are modestly sized, at the cost of slightly more
        /// per-partition open / metadata-load overhead.
        pub repartition_file_min_size: usize, default = 1024 * 1024

        /// Should DataFusion repartition data using the join keys to execute joins in parallel
        /// using the provided `target_partitions` level
        pub repartition_joins: bool, default = true

        /// Should DataFusion allow symmetric hash joins for unbounded data sources even when
        /// its inputs do not have any ordering or filtering If the flag is not enabled,
        /// the SymmetricHashJoin operator will be unable to prune its internal buffers,
        /// resulting in certain join types - such as Full, Left, LeftAnti, LeftSemi, Right,
        /// RightAnti, and RightSemi - being produced only at the end of the execution.
        /// This is not typical in stream processing. Additionally, without proper design for
        /// long runner execution, all types of joins may encounter out-of-memory errors.
        pub allow_symmetric_joins_without_pruning: bool, default = true

        /// When set to `true`, datasource partitions will be repartitioned to achieve maximum parallelism.
        /// This applies to both in-memory partitions and FileSource's file groups (1 group is 1 partition).
        ///
        /// For FileSources, only Parquet and CSV formats are currently supported.
        ///
        /// If set to `true` for a FileSource, all files will be repartitioned evenly (i.e., a single large file
        /// might be partitioned into smaller chunks) for parallel scanning.
        /// If set to `false` for a FileSource, different files will be read in parallel, but repartitioning won't
        /// happen within a single file.
        ///
        /// If set to `true` for an in-memory source, all memtable's partitions will have their batches
        /// repartitioned evenly to the desired number of `target_partitions`. Repartitioning can change
        /// the total number of partitions and batches per partition, but does not slice the initial
        /// record tables provided to the MemTable on creation.
        pub repartition_file_scans: bool, default = true

        /// Minimum number of distinct partition values required to group files by their
        /// Hive partition column values (enabling output partitioning declaration).
        ///
        /// How the option is used:
        ///     - preserve_file_partitions=0: Disable it.
        ///     - preserve_file_partitions=1: Always enable it.
        ///     - preserve_file_partitions=N, actual file partitions=M: Only enable when M >= N.
        ///     This threshold preserves I/O parallelism when file partitioning is below it.
        ///
        /// Note: This may reduce parallelism, rooting from the I/O level, if the number of distinct
        /// partitions is less than the target_partitions.
        pub preserve_file_partitions: usize, default = 0

        /// Should DataFusion repartition data using the partitions keys to execute window
        /// functions in parallel using the provided `target_partitions` level
        pub repartition_windows: bool, default = true

        /// Should DataFusion execute sorts in a per-partition fashion and merge
        /// afterwards instead of coalescing first and sorting globally.
        /// With this flag is enabled, plans in the form below
        ///
        /// ```text
        ///      "SortExec: [a@0 ASC]",
        ///      "  CoalescePartitionsExec",
        ///      "    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
        /// ```
        /// would turn into the plan below which performs better in multithreaded environments
        ///
        /// ```text
        ///      "SortPreservingMergeExec: [a@0 ASC]",
        ///      "  SortExec: [a@0 ASC]",
        ///      "    RepartitionExec: partitioning=RoundRobinBatch(8), input_partitions=1",
        /// ```
        pub repartition_sorts: bool, default = true

        /// Partition count threshold for subset satisfaction optimization.
        ///
        /// When the current partition count is >= this threshold, DataFusion will
        /// skip repartitioning if the required partitioning expression is a subset
        /// of the current partition expression such as Hash(a) satisfies Hash(a, b).
        ///
        /// When the current partition count is < this threshold, DataFusion will
        /// repartition to increase parallelism even when subset satisfaction applies.
        ///
        /// Set to 0 to always repartition (disable subset satisfaction optimization).
        /// Set to a high value to always use subset satisfaction.
        ///
        /// Example (subset_repartition_threshold = 4):
        /// ```text
        ///     Hash([a]) satisfies Hash([a, b]) because (Hash([a, b]) is subset of Hash([a])
        ///
        ///     If current partitions (3) < threshold (4), repartition:
        ///     AggregateExec: mode=FinalPartitioned, gby=[a, b], aggr=[SUM(x)]
        ///       RepartitionExec: partitioning=Hash([a, b], 8), input_partitions=3
        ///         AggregateExec: mode=Partial, gby=[a, b], aggr=[SUM(x)]
        ///           DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 3)
        ///
        ///     If current partitions (8) >= threshold (4), use subset satisfaction:
        ///     AggregateExec: mode=SinglePartitioned, gby=[a, b], aggr=[SUM(x)]
        ///       DataSourceExec: file_groups={...}, output_partitioning=Hash([a], 8)
        /// ```
        pub subset_repartition_threshold: usize, default = 4

        /// When true, DataFusion will opportunistically remove sorts when the data is already sorted,
        /// (i.e. setting `preserve_order` to true on `RepartitionExec`  and
        /// using `SortPreservingMergeExec`)
        ///
        /// When false, DataFusion will maximize plan parallelism using
        /// `RepartitionExec` even if this requires subsequently resorting data using a `SortExec`.
        pub prefer_existing_sort: bool, default = false

        /// When set to true, the logical plan optimizer will produce warning
        /// messages if any optimization rules produce errors and then proceed to the next
        /// rule. When set to false, any rules that produce errors will cause the query to fail
        pub skip_failed_rules: bool, default = false

        /// Number of times that the optimizer will attempt to optimize the plan
        pub max_passes: usize, default = 3

        /// When set to true, the physical plan optimizer will run a top down
        /// process to reorder the join keys
        pub top_down_join_key_reordering: bool, default = true

        /// When set to true, the physical plan optimizer may swap join inputs
        /// based on statistics. When set to false, statistics-driven join
        /// input reordering is disabled and the original join order in the
        /// query is used.
        pub join_reordering: bool, default = true

        /// When set to true, the physical plan optimizer uses the pluggable
        /// `StatisticsRegistry` for statistics propagation across operators.
        /// This enables more accurate cardinality estimates compared to each
        /// operator's built-in `partition_statistics`.
        pub use_statistics_registry: bool, default = false

        /// When set to true, the physical plan optimizer will prefer HashJoin over SortMergeJoin.
        /// HashJoin can work more efficiently than SortMergeJoin but consumes more memory
        pub prefer_hash_join: bool, default = true

        /// When set to true, piecewise merge join is enabled. PiecewiseMergeJoin is currently
        /// experimental. Physical planner will opt for PiecewiseMergeJoin when there is only
        /// one range filter.
        pub enable_piecewise_merge_join: bool, default = false

        /// The maximum estimated size in bytes for one input side of a HashJoin
        /// will be collected into a single partition
        pub hash_join_single_partition_threshold: usize, default = 1024 * 1024

        /// The maximum estimated size in rows for one input side of a HashJoin
        /// will be collected into a single partition
        pub hash_join_single_partition_threshold_rows: usize, default = 1024 * 128

        /// Maximum size in bytes for the build side of a hash join to be pushed down as an InList expression for dynamic filtering.
        /// Build sides larger than this will use hash table lookups instead.
        /// Set to 0 to always use hash table lookups.
        ///
        /// InList pushdown can be more efficient for small build sides because it can result in better
        /// statistics pruning as well as use any bloom filters present on the scan side.
        /// InList expressions are also more transparent and easier to serialize over the network in distributed uses of DataFusion.
        /// On the other hand InList pushdown requires making a copy of the data and thus adds some overhead to the build side and uses more memory.
        ///
        /// This setting is per-partition, so we may end up using `hash_join_inlist_pushdown_max_size` * `target_partitions` memory.
        ///
        /// The default is 128kB per partition.
        /// This should allow point lookup joins (e.g. joining on a unique primary key) to use InList pushdown in most cases
        /// but avoids excessive memory usage or overhead for larger joins.
        pub hash_join_inlist_pushdown_max_size: usize, default = 128 * 1024

        /// Maximum number of distinct values (rows) in the build side of a hash join to be pushed down as an InList expression for dynamic filtering.
        /// Build sides with more rows than this will use hash table lookups instead.
        /// Set to 0 to always use hash table lookups.
        ///
        /// This provides an additional limit beyond `hash_join_inlist_pushdown_max_size` to prevent
        /// very large IN lists that might not provide much benefit over hash table lookups.
        ///
        /// This uses the deduplicated row count once the build side has been evaluated.
        ///
        /// The default is 150 values per partition.
        /// This is inspired by Trino's `max-filter-keys-per-column` setting.
        /// See: <https://trino.io/docs/current/admin/dynamic-filtering.html#dynamic-filter-collection-thresholds>
        pub hash_join_inlist_pushdown_max_distinct_values: usize, default = 150

        /// The default filter selectivity used by Filter Statistics
        /// when an exact selectivity cannot be determined. Valid values are
        /// between 0 (no selectivity) and 100 (all rows are selected).
        pub default_filter_selectivity: u8, default = 20

        /// When set to true, the optimizer will not attempt to convert Union to Interleave
        pub prefer_existing_union: bool, default = false

        /// When set to true, if the returned type is a view type
        /// then the output will be coerced to a non-view.
        /// Coerces `Utf8View` to `LargeUtf8`, and `BinaryView` to `LargeBinary`.
        pub expand_views_at_output: bool, default = false

        /// Enable sort pushdown optimization.
        /// When enabled, attempts to push sort requirements down to data sources
        /// that can natively handle them (e.g., by reversing file/row group read order).
        ///
        /// Returns **inexact ordering**: Sort operator is kept for correctness,
        /// but optimized input enables early termination for TopK queries (ORDER BY ... LIMIT N),
        /// providing significant speedup.
        ///
        /// Memory: No additional overhead (only changes read order).
        ///
        /// Future: Will add option to detect perfectly sorted data and eliminate Sort completely.
        ///
        /// Default: true
        pub enable_sort_pushdown: bool, default = true

        /// When set to true, the optimizer will extract leaf expressions
        /// (such as `get_field`) from filter/sort/join nodes into projections
        /// closer to the leaf table scans, and push those projections down
        /// towards the leaf nodes.
        pub enable_leaf_expression_pushdown: bool, default = true

        /// When set to true, the logical optimizer will rewrite `UNION DISTINCT` branches that
        /// read from the same source and differ only by filter predicates into a single branch
        /// with a combined filter. This optimization is conservative and only applies when the
        /// branches share the same source and compatible wrapper nodes such as identical
        /// projections or aliases.
        pub enable_unions_to_filter: bool, default = false
    }
}

config_namespace! {
    /// Options controlling explain output
    ///
    /// See also: [`SessionConfig`]
    ///
    /// [`SessionConfig`]: https://docs.rs/datafusion/latest/datafusion/prelude/struct.SessionConfig.html
    pub struct ExplainOptions {
        /// When set to true, the explain statement will only print logical plans
        pub logical_plan_only: bool, default = false

        /// When set to true, the explain statement will only print physical plans
        pub physical_plan_only: bool, default = false

        /// When set to true, the explain statement will print operator statistics
        /// for physical plans
        pub show_statistics: bool, default = false

        /// When set to true, the explain statement will print the partition sizes
        pub show_sizes: bool, default = true

        /// When set to true, the explain statement will print schema information
        pub show_schema: bool, default = false

        /// Display format of explain. Default is "indent".
        /// When set to "tree", it will print the plan in a tree-rendered format.
        pub format: ExplainFormat, default = ExplainFormat::Indent

        /// (format=tree only) Maximum total width of the rendered tree.
        /// When set to 0, the tree will have no width limit.
        pub tree_maximum_render_width: usize, default = 240

        /// Verbosity level for "EXPLAIN ANALYZE". Default is "dev"
        /// "summary" shows common metrics for high-level insights.
        /// "dev" provides deep operator-level introspection for developers.
        pub analyze_level: MetricType, default = MetricType::Dev

        /// Which metric categories to include in "EXPLAIN ANALYZE" output.
        /// Comma-separated list of: "rows", "bytes", "timing", "uncategorized".
        /// Use "none" to show plan structure only, or "all" (default) to show everything.
        /// Metrics without a declared category are treated as "uncategorized".
        pub analyze_categories: ExplainAnalyzeCategories, default = ExplainAnalyzeCategories::All
    }
}

impl ExecutionOptions {
    /// Returns the correct parallelism based on the provided `value`.
    /// If `value` is `"0"`, returns the default available parallelism, computed with
    /// `get_available_parallelism`. Otherwise, returns `value`.
    fn normalized_parallelism(value: &str) -> String {
        if value.parse::<usize>() == Ok(0) {
            get_available_parallelism().to_string()
        } else {
            value.to_owned()
        }
    }
}

config_namespace! {
    /// Options controlling the format of output when printing record batches
    /// Copies [`arrow::util::display::FormatOptions`]
    pub struct FormatOptions {
        /// If set to `true` any formatting errors will be written to the output
        /// instead of being converted into a [`std::fmt::Error`]
        pub safe: bool, default = true
        /// Format string for nulls
        pub null: String, default = "".into()
        /// Date format for date arrays
        pub date_format: Option<String>, default = Some("%Y-%m-%d".to_string())
        /// Format for DateTime arrays
        pub datetime_format: Option<String>, default = Some("%Y-%m-%dT%H:%M:%S%.f".to_string())
        /// Timestamp format for timestamp arrays
        pub timestamp_format: Option<String>, default = Some("%Y-%m-%dT%H:%M:%S%.f".to_string())
        /// Timestamp format for timestamp with timezone arrays. When `None`, ISO 8601 format is used.
        pub timestamp_tz_format: Option<String>, default = None
        /// Time format for time arrays
        pub time_format: Option<String>, default = Some("%H:%M:%S%.f".to_string())
        /// Duration format. Can be either `"pretty"` or `"ISO8601"`
        pub duration_format: String, transform = str::to_lowercase, default = "pretty".into()
        /// Show types in visual representation batches
        pub types_info: bool, default = false
    }
}

impl<'a> TryFrom<&'a FormatOptions> for arrow::util::display::FormatOptions<'a> {
    type Error = DataFusionError;
    fn try_from(options: &'a FormatOptions) -> Result<Self> {
        let duration_format = match options.duration_format.as_str() {
            "pretty" => arrow::util::display::DurationFormat::Pretty,
            "iso8601" => arrow::util::display::DurationFormat::ISO8601,
            _ => {
                return _config_err!(
                    "Invalid duration format: {}. Valid values are pretty or iso8601",
                    options.duration_format
                );
            }
        };

        Ok(Self::new()
            .with_display_error(options.safe)
            .with_null(&options.null)
            .with_date_format(options.date_format.as_deref())
            .with_datetime_format(options.datetime_format.as_deref())
            .with_timestamp_format(options.timestamp_format.as_deref())
            .with_timestamp_tz_format(options.timestamp_tz_format.as_deref())
            .with_time_format(options.time_format.as_deref())
            .with_duration_format(duration_format)
            .with_types_info(options.types_info))
    }
}

config_namespace! {
    /// Options controlling DataFusion's Spark-compatibility layer (functions
    /// under `datafusion/spark`). Keys here mirror their `spark.sql.*`
    /// equivalents in Apache Spark.
    pub struct SparkOptions {
        /// Policy for handling duplicate keys in Spark-compatible map-construction
        /// functions (`map_from_arrays`, `map_from_entries`, `str_to_map`).
        ///
        /// Mirrors Spark's
        /// [`spark.sql.mapKeyDedupPolicy`](https://github.com/apache/spark/blob/cf3a34e19dfcf70e2d679217ff1ba21302212472/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala#L4961):
        /// - `EXCEPTION` (default): raise `[DUPLICATED_MAP_KEY]` at runtime on any duplicate key.
        /// - `LAST_WIN`: keep the last occurrence of each duplicate key.
        ///
        /// Values are case-insensitive.
        pub map_key_dedup_policy: MapKeyDedupPolicy, default = MapKeyDedupPolicy::Exception
    }
}

/// A key value pair, with a corresponding description
#[derive(Debug, Clone, Hash, PartialEq, Eq)]
pub struct ConfigEntry {
    /// A unique string to identify this config value
    pub key: String,

    /// The value if any
    pub value: Option<String>,

    /// A description of this configuration entry
    pub description: &'static str,
}

/// Configuration options struct, able to store both built-in configuration and custom options
#[derive(Debug, Clone, Default)]
#[non_exhaustive]
pub struct ConfigOptions {
    /// Catalog options
    pub catalog: CatalogOptions,
    /// Execution options
    pub execution: ExecutionOptions,
    /// Optimizer options
    pub optimizer: OptimizerOptions,
    /// SQL parser options
    pub sql_parser: SqlParserOptions,
    /// Explain options
    pub explain: ExplainOptions,
    /// Optional extensions registered using [`Extensions::insert`]
    pub extensions: Extensions,
    /// Formatting options when printing batches
    pub format: FormatOptions,
    /// Spark-compatibility options (functions under `datafusion/spark`)
    pub spark: SparkOptions,
}

impl ConfigField for ConfigOptions {
    fn visit<V: Visit>(&self, v: &mut V, _key_prefix: &str, _description: &'static str) {
        self.catalog.visit(v, "datafusion.catalog", "");
        self.execution.visit(v, "datafusion.execution", "");
        self.optimizer.visit(v, "datafusion.optimizer", "");
        self.explain.visit(v, "datafusion.explain", "");
        self.sql_parser.visit(v, "datafusion.sql_parser", "");
        self.format.visit(v, "datafusion.format", "");
        self.spark.visit(v, "datafusion.spark", "");
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        // Extensions are handled in the public `ConfigOptions::set`
        let (key, rem) = key.split_once('.').unwrap_or((key, ""));
        match key {
            "catalog" => self.catalog.set(rem, value),
            "execution" => self.execution.set(rem, value),
            "optimizer" => self.optimizer.set(rem, value),
            "explain" => self.explain.set(rem, value),
            "sql_parser" => self.sql_parser.set(rem, value),
            "format" => self.format.set(rem, value),
            "spark" => self.spark.set(rem, value),
            _ => _config_err!("Config value \"{key}\" not found on ConfigOptions"),
        }
    }

    /// Reset a configuration option back to its default value
    fn reset(&mut self, key: &str) -> Result<()> {
        let Some((prefix, rest)) = key.split_once('.') else {
            return _config_err!("could not find config namespace for key \"{key}\"");
        };

        if prefix != "datafusion" {
            return _config_err!("Could not find config namespace \"{prefix}\"");
        }

        let (section, rem) = rest.split_once('.').unwrap_or((rest, ""));
        if rem.is_empty() {
            return _config_err!("could not find config field for key \"{key}\"");
        }

        match section {
            "catalog" => self.catalog.reset(rem),
            "execution" => self.execution.reset(rem),
            "optimizer" => {
                if rem == "enable_dynamic_filter_pushdown" {
                    let defaults = OptimizerOptions::default();
                    self.optimizer.enable_dynamic_filter_pushdown =
                        defaults.enable_dynamic_filter_pushdown;
                    self.optimizer.enable_topk_dynamic_filter_pushdown =
                        defaults.enable_topk_dynamic_filter_pushdown;
                    self.optimizer.enable_join_dynamic_filter_pushdown =
                        defaults.enable_join_dynamic_filter_pushdown;
                    Ok(())
                } else {
                    self.optimizer.reset(rem)
                }
            }
            "explain" => self.explain.reset(rem),
            "sql_parser" => self.sql_parser.reset(rem),
            "format" => self.format.reset(rem),
            "spark" => self.spark.reset(rem),
            other => _config_err!("Config value \"{other}\" not found on ConfigOptions"),
        }
    }
}

/// This namespace is reserved for interacting with Foreign Function Interface
/// (FFI) based configuration extensions.
pub const DATAFUSION_FFI_CONFIG_NAMESPACE: &str = "datafusion_ffi";

impl ConfigOptions {
    /// Creates a new [`ConfigOptions`] with default values
    pub fn new() -> Self {
        Self::default()
    }

    /// Set extensions to provided value
    pub fn with_extensions(mut self, extensions: Extensions) -> Self {
        self.extensions = extensions;
        self
    }

    /// Set a configuration option
    pub fn set(&mut self, key: &str, value: &str) -> Result<()> {
        let Some((mut prefix, mut inner_key)) = key.split_once('.') else {
            return _config_err!("could not find config namespace for key \"{key}\"");
        };

        if prefix == "datafusion" {
            if inner_key == "optimizer.enable_dynamic_filter_pushdown" {
                let bool_value = value.parse::<bool>().map_err(|e| {
                    DataFusionError::Configuration(format!(
                        "Failed to parse '{value}' as bool: {e}",
                    ))
                })?;

                {
                    self.optimizer.enable_dynamic_filter_pushdown = bool_value;
                    self.optimizer.enable_topk_dynamic_filter_pushdown = bool_value;
                    self.optimizer.enable_join_dynamic_filter_pushdown = bool_value;
                    self.optimizer.enable_aggregate_dynamic_filter_pushdown = bool_value;
                }
                return Ok(());
            }
            return ConfigField::set(self, inner_key, value)
                .map_err(|e| e.context(format!("Error setting config {key}")));
        }

        if !self.extensions.0.contains_key(prefix)
            && self
                .extensions
                .0
                .contains_key(DATAFUSION_FFI_CONFIG_NAMESPACE)
        {
            inner_key = key;
            prefix = DATAFUSION_FFI_CONFIG_NAMESPACE;
        }

        let Some(e) = self.extensions.0.get_mut(prefix) else {
            return _config_err!("Could not find config namespace \"{prefix}\"");
        };
        e.0.set(inner_key, value)
    }

    /// Create new [`ConfigOptions`], taking values from environment variables
    /// where possible.
    ///
    /// For example, to configure `datafusion.execution.batch_size`
    /// ([`ExecutionOptions::batch_size`]) you would set the
    /// `DATAFUSION_EXECUTION_BATCH_SIZE` environment variable.
    ///
    /// The name of the environment variable is the option's key, transformed to
    /// uppercase and with periods replaced with underscores.
    ///
    /// Values are parsed according to the [same rules used in casts from
    /// Utf8](https://docs.rs/arrow/latest/arrow/compute/kernels/cast/fn.cast.html).
    ///
    /// If the value in the environment variable cannot be cast to the type of
    /// the configuration option, the default value will be used instead and a
    /// warning emitted. Environment variables are read when this method is
    /// called, and are not re-read later.
    pub fn from_env() -> Result<Self> {
        struct Visitor(Vec<String>);

        impl Visit for Visitor {
            fn some<V: Display>(&mut self, key: &str, _: V, _: &'static str) {
                self.0.push(key.to_string())
            }

            fn none(&mut self, key: &str, _: &'static str) {
                self.0.push(key.to_string())
            }
        }

        // Extract the names of all fields and then look up the corresponding
        // environment variables. This isn't hugely efficient but avoids
        // ambiguity between `a.b` and `a_b` which would both correspond
        // to an environment variable of `A_B`

        let mut keys = Visitor(vec![]);
        let mut ret = Self::default();
        ret.visit(&mut keys, "datafusion", "");

        for key in keys.0 {
            let env = key.to_uppercase().replace('.', "_");
            if let Some(var) = std::env::var_os(env) {
                let value = var.to_string_lossy();
                log::info!("Set {key} to {value} from the environment variable");
                ret.set(&key, value.as_ref())?;
            }
        }

        Ok(ret)
    }

    /// Create new ConfigOptions struct, taking values from a string hash map.
    ///
    /// Only the built-in configurations will be extracted from the hash map
    /// and other key value pairs will be ignored.
    pub fn from_string_hash_map(settings: &HashMap<String, String>) -> Result<Self> {
        struct Visitor(Vec<String>);

        impl Visit for Visitor {
            fn some<V: Display>(&mut self, key: &str, _: V, _: &'static str) {
                self.0.push(key.to_string())
            }

            fn none(&mut self, key: &str, _: &'static str) {
                self.0.push(key.to_string())
            }
        }

        let mut keys = Visitor(vec![]);
        let mut ret = Self::default();
        ret.visit(&mut keys, "datafusion", "");

        for key in keys.0 {
            if let Some(var) = settings.get(&key) {
                ret.set(&key, var)?;
            }
        }

        Ok(ret)
    }

    /// Returns the [`ConfigEntry`] stored within this [`ConfigOptions`]
    pub fn entries(&self) -> Vec<ConfigEntry> {
        struct Visitor(Vec<ConfigEntry>);

        impl Visit for Visitor {
            fn some<V: Display>(
                &mut self,
                key: &str,
                value: V,
                description: &'static str,
            ) {
                self.0.push(ConfigEntry {
                    key: key.to_string(),
                    value: Some(value.to_string()),
                    description,
                })
            }

            fn none(&mut self, key: &str, description: &'static str) {
                self.0.push(ConfigEntry {
                    key: key.to_string(),
                    value: None,
                    description,
                })
            }
        }

        let mut v = Visitor(vec![]);
        self.visit(&mut v, "datafusion", "");

        v.0.extend(self.extensions.0.values().flat_map(|e| e.0.entries()));
        v.0
    }

    /// Generate documentation that can be included in the user guide
    pub fn generate_config_markdown() -> String {
        use std::fmt::Write as _;

        let mut s = Self::default();

        // Normalize for display
        s.execution.target_partitions = 0;
        s.execution.planning_concurrency = 0;

        let mut docs = "| key | default | description |\n".to_string();
        docs += "|-----|---------|-------------|\n";
        let mut entries = s.entries();
        entries.sort_unstable_by(|a, b| a.key.cmp(&b.key));

        for entry in s.entries() {
            let _ = writeln!(
                &mut docs,
                "| {} | {} | {} |",
                entry.key,
                entry.value.as_deref().unwrap_or("NULL"),
                entry.description
            );
        }
        docs
    }
}

/// [`ConfigExtension`] provides a mechanism to store third-party configuration
/// within DataFusion [`ConfigOptions`]
///
/// This mechanism can be used to pass configuration to user defined functions
/// or optimizer passes
///
/// # Example
/// ```
/// use datafusion_common::{
///     config::ConfigExtension, config::ConfigOptions, extensions_options,
/// };
/// // Define a new configuration struct using the `extensions_options` macro
/// extensions_options! {
///    /// My own config options.
///    pub struct MyConfig {
///        /// Should "foo" be replaced by "bar"?
///        pub foo_to_bar: bool, default = true
///
///        /// How many "baz" should be created?
///        pub baz_count: usize, default = 1337
///    }
/// }
///
/// impl ConfigExtension for MyConfig {
///     const PREFIX: &'static str = "my_config";
/// }
///
/// // set up config struct and register extension
/// let mut config = ConfigOptions::default();
/// config.extensions.insert(MyConfig::default());
///
/// // overwrite config default
/// config.set("my_config.baz_count", "42").unwrap();
///
/// // check config state
/// let my_config = config.extensions.get::<MyConfig>().unwrap();
/// assert!(my_config.foo_to_bar,);
/// assert_eq!(my_config.baz_count, 42,);
/// ```
///
/// # Note:
/// Unfortunately associated constants are not currently object-safe, and so this
/// extends the object-safe [`ExtensionOptions`]
pub trait ConfigExtension: ExtensionOptions {
    /// Configuration namespace prefix to use
    ///
    /// All values under this will be prefixed with `$PREFIX + "."`
    const PREFIX: &'static str;
}

/// An object-safe API for storing arbitrary configuration.
///
/// See [`ConfigExtension`] for user defined configuration
pub trait ExtensionOptions: Send + Sync + fmt::Debug + 'static {
    /// Return `self` as [`Any`]
    ///
    /// This is needed until trait upcasting is stabilized
    fn as_any(&self) -> &dyn Any;

    /// Return `self` as [`Any`]
    ///
    /// This is needed until trait upcasting is stabilized
    fn as_any_mut(&mut self) -> &mut dyn Any;

    /// Return a deep clone of this [`ExtensionOptions`]
    ///
    /// It is important this does not share mutable state to avoid consistency issues
    /// with configuration changing whilst queries are executing
    fn cloned(&self) -> Box<dyn ExtensionOptions>;

    /// Set the given `key`, `value` pair
    fn set(&mut self, key: &str, value: &str) -> Result<()>;

    /// Returns the [`ConfigEntry`] stored in this [`ExtensionOptions`]
    fn entries(&self) -> Vec<ConfigEntry>;
}

/// A type-safe container for [`ConfigExtension`]
#[derive(Debug, Default, Clone)]
pub struct Extensions(BTreeMap<&'static str, ExtensionBox>);

impl Extensions {
    /// Create a new, empty [`Extensions`]
    pub fn new() -> Self {
        Self(BTreeMap::new())
    }

    /// Registers a [`ConfigExtension`] with this [`ConfigOptions`]
    pub fn insert<T: ConfigExtension>(&mut self, extension: T) {
        assert_ne!(T::PREFIX, "datafusion");
        let e = ExtensionBox(Box::new(extension));
        self.0.insert(T::PREFIX, e);
    }

    /// Retrieves the extension of the given type if any
    pub fn get<T: ConfigExtension>(&self) -> Option<&T> {
        self.0.get(T::PREFIX)?.0.as_any().downcast_ref()
    }

    /// Retrieves the extension of the given type if any
    pub fn get_mut<T: ConfigExtension>(&mut self) -> Option<&mut T> {
        let e = self.0.get_mut(T::PREFIX)?;
        e.0.as_any_mut().downcast_mut()
    }

    /// Iterates all the config extension entries yielding their prefix and their
    /// [ExtensionOptions] implementation.
    pub fn iter(
        &self,
    ) -> impl Iterator<Item = (&'static str, &Box<dyn ExtensionOptions>)> {
        self.0.iter().map(|(k, v)| (*k, &v.0))
    }
}

#[derive(Debug)]
struct ExtensionBox(Box<dyn ExtensionOptions>);

impl Clone for ExtensionBox {
    fn clone(&self) -> Self {
        Self(self.0.cloned())
    }
}

/// A trait implemented by `config_namespace` and for field types that provides
/// the ability to walk and mutate the configuration tree
pub trait ConfigField {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str);

    fn set(&mut self, key: &str, value: &str) -> Result<()>;

    fn reset(&mut self, key: &str) -> Result<()> {
        _config_err!("Reset is not supported for this config field, key: {}", key)
    }
}

impl<F: ConfigField + Default> ConfigField for Option<F> {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        match self {
            Some(s) => s.visit(v, key, description),
            None => v.none(key, description),
        }
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        self.get_or_insert_with(Default::default).set(key, value)
    }

    fn reset(&mut self, key: &str) -> Result<()> {
        if key.is_empty() {
            *self = Default::default();
            Ok(())
        } else {
            self.get_or_insert_with(Default::default).reset(key)
        }
    }
}

/// Default transformation to parse a [`ConfigField`] for a string.
///
/// This uses [`FromStr`] to parse the data.
pub fn default_config_transform<T>(input: &str) -> Result<T>
where
    T: FromStr,
    <T as FromStr>::Err: Sync + Send + Error + 'static,
{
    input.parse().map_err(|e| {
        DataFusionError::Context(
            format!(
                "Error parsing '{}' as {}",
                input,
                std::any::type_name::<T>()
            ),
            Box::new(DataFusionError::External(Box::new(e))),
        )
    })
}

/// Macro that generates [`ConfigField`] for a given type.
///
/// # Usage
/// This always requires [`Display`] to be implemented for the given type.
///
/// There are two ways to invoke this macro. The first one uses
/// [`default_config_transform`]/[`FromStr`] to parse the data:
///
/// ```ignore
/// config_field(MyType);
/// ```
///
/// Note that the parsing error MUST implement [`std::error::Error`]!
///
/// Or you can specify how you want to parse an [`str`] into the type:
///
/// ```ignore
/// fn parse_it(s: &str) -> Result<MyType> {
///     ...
/// }
///
/// config_field(
///     MyType,
///     value => parse_it(value)
/// );
/// ```
#[macro_export]
macro_rules! config_field {
    ($t:ty) => {
        config_field!($t, value => $crate::config::default_config_transform(value)?);
    };

    ($t:ty, $arg:ident => $transform:expr) => {
        impl $crate::config::ConfigField for $t {
            fn visit<V: $crate::config::Visit>(&self, v: &mut V, key: &str, description: &'static str) {
                v.some(key, self, description)
            }

            fn set(&mut self, _: &str, $arg: &str) -> $crate::error::Result<()> {
                *self = $transform;
                Ok(())
            }

            fn reset(&mut self, key: &str) -> $crate::error::Result<()> {
                if key.is_empty() {
                    *self = <$t as Default>::default();
                    Ok(())
                } else {
                    $crate::error::_config_err!(
                        "Config field is a scalar {} and does not have nested field \"{}\"",
                        stringify!($t),
                        key
                    )
                }
            }
        }
    };
}

config_field!(String);
config_field!(bool, value => default_config_transform(value.to_lowercase().as_str())?);
config_field!(usize);
config_field!(f64);
config_field!(u64);
config_field!(u32);
config_field!(i32);

impl ConfigField for u8 {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        v.some(key, self, description)
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        if value.is_empty() {
            return Err(DataFusionError::Configuration(format!(
                "Input string for {key} key is empty"
            )));
        }
        // Check if the string is a valid number
        if let Ok(num) = value.parse::<u8>() {
            // TODO: Let's decide how we treat the numerical strings.
            *self = num;
        } else {
            let bytes = value.as_bytes();
            // Check if the first character is ASCII (single byte)
            if bytes.len() > 1 || !value.chars().next().unwrap().is_ascii() {
                return Err(DataFusionError::Configuration(format!(
                    "Error parsing {value} as u8. Non-ASCII string provided"
                )));
            }
            *self = bytes[0];
        }
        Ok(())
    }
}

impl ConfigField for CompressionTypeVariant {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        v.some(key, self, description)
    }

    fn set(&mut self, _: &str, value: &str) -> Result<()> {
        *self = CompressionTypeVariant::from_str(value)?;
        Ok(())
    }
}

impl ConfigField for CsvQuoteStyle {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, description: &'static str) {
        v.some(key, self, description)
    }

    fn set(&mut self, _: &str, value: &str) -> Result<()> {
        *self = CsvQuoteStyle::from_str(value)?;
        Ok(())
    }
}

/// An implementation trait used to recursively walk configuration
pub trait Visit {
    fn some<V: Display>(&mut self, key: &str, value: V, description: &'static str);

    fn none(&mut self, key: &str, description: &'static str);
}

/// Convenience macro to create [`ExtensionsOptions`].
///
/// The created structure implements the following traits:
///
/// - [`Clone`]
/// - [`Debug`]
/// - [`Default`]
/// - [`ExtensionOptions`]
///
/// # Usage
/// The syntax is:
///
/// ```text
/// extensions_options! {
///      /// Struct docs (optional).
///     [<vis>] struct <StructName> {
///         /// Field docs (optional)
///         [<vis>] <field_name>: <field_type>, default = <default_value>
///
///         ... more fields
///     }
/// }
/// ```
///
/// The placeholders are:
/// - `[<vis>]`: Optional visibility modifier like `pub` or `pub(crate)`.
/// - `<StructName>`: Struct name like `MyStruct`.
/// - `<field_name>`: Field name like `my_field`.
/// - `<field_type>`: Field type like `u8`.
/// - `<default_value>`: Default value matching the field type like `42`.
///
/// # Example
/// See also a full example on the [`ConfigExtension`] documentation
///
/// ```
/// use datafusion_common::extensions_options;
///
/// extensions_options! {
///     /// My own config options.
///     pub struct MyConfig {
///         /// Should "foo" be replaced by "bar"?
///         pub foo_to_bar: bool, default = true
///
///         /// How many "baz" should be created?
///         pub baz_count: usize, default = 1337
///     }
/// }
/// ```
///
///
/// [`Debug`]: std::fmt::Debug
/// [`ExtensionsOptions`]: crate::config::ExtensionOptions
#[macro_export]
macro_rules! extensions_options {
    (
     $(#[doc = $struct_d:tt])*
     $vis:vis struct $struct_name:ident {
        $(
        $(#[doc = $d:tt])*
        $field_vis:vis $field_name:ident : $field_type:ty, default = $default:expr
        )*$(,)*
    }
    ) => {
        $(#[doc = $struct_d])*
        #[derive(Debug, Clone)]
        #[non_exhaustive]
        $vis struct $struct_name{
            $(
            $(#[doc = $d])*
            $field_vis $field_name : $field_type,
            )*
        }

        impl Default for $struct_name {
            fn default() -> Self {
                Self {
                    $($field_name: $default),*
                }
            }
        }

        impl $crate::config::ExtensionOptions for $struct_name {
            fn as_any(&self) -> &dyn ::std::any::Any {
                self
            }

            fn as_any_mut(&mut self) -> &mut dyn ::std::any::Any {
                self
            }

            fn cloned(&self) -> Box<dyn $crate::config::ExtensionOptions> {
                Box::new(self.clone())
            }

            fn set(&mut self, key: &str, value: &str) -> $crate::error::Result<()> {
                $crate::config::ConfigField::set(self, key, value)
            }

            fn entries(&self) -> Vec<$crate::config::ConfigEntry> {
                struct Visitor(Vec<$crate::config::ConfigEntry>);

                impl $crate::config::Visit for Visitor {
                    fn some<V: std::fmt::Display>(
                        &mut self,
                        key: &str,
                        value: V,
                        description: &'static str,
                    ) {
                        self.0.push($crate::config::ConfigEntry {
                            key: key.to_string(),
                            value: Some(value.to_string()),
                            description,
                        })
                    }

                    fn none(&mut self, key: &str, description: &'static str) {
                        self.0.push($crate::config::ConfigEntry {
                            key: key.to_string(),
                            value: None,
                            description,
                        })
                    }
                }

                let mut v = Visitor(vec![]);
                // The prefix is not used for extensions.
                // The description is generated in ConfigField::visit.
                // We can just pass empty strings here.
                $crate::config::ConfigField::visit(self, &mut v, "", "");
                v.0
            }
        }

        impl $crate::config::ConfigField for $struct_name {
            fn set(&mut self, key: &str, value: &str) -> $crate::error::Result<()> {
                let (key, rem) = key.split_once('.').unwrap_or((key, ""));
                match key {
                    $(
                        stringify!($field_name) => {
                            // Safely apply deprecated attribute if present
                            // $(#[allow(deprecated)])?
                            {
                                            self.$field_name.set(rem, value.as_ref())
                            }
                        },
                    )*
                    _ => return $crate::error::_config_err!(
                        "Config value \"{}\" not found on {}", key, stringify!($struct_name)
                    )
                }
            }

            fn visit<V: $crate::config::Visit>(&self, v: &mut V, _key_prefix: &str, _description: &'static str) {
                $(
                    let key = stringify!($field_name).to_string();
                    let desc = concat!($($d),*).trim();
                    self.$field_name.visit(v, key.as_str(), desc);
                )*
            }
        }
    }
}

/// These file types have special built in behavior for configuration.
/// Use TableOptions::Extensions for configuring other file types.
#[derive(Debug, Clone)]
pub enum ConfigFileType {
    CSV,
    #[cfg(feature = "parquet")]
    PARQUET,
    JSON,
}

/// Represents the configuration options available for handling different table formats within a data processing application.
/// This struct encompasses options for various file formats including CSV, Parquet, and JSON, allowing for flexible configuration
/// of parsing and writing behaviors specific to each format. Additionally, it supports extending functionality through custom extensions.
#[derive(Debug, Clone, Default)]
pub struct TableOptions {
    /// Configuration options for CSV file handling. This includes settings like the delimiter,
    /// quote character, and whether the first row is considered as headers.
    pub csv: CsvOptions,

    /// Configuration options for Parquet file handling. This includes settings for compression,
    /// encoding, and other Parquet-specific file characteristics.
    pub parquet: TableParquetOptions,

    /// Configuration options for JSON file handling.
    pub json: JsonOptions,

    /// The current file format that the table operations should assume. This option allows
    /// for dynamic switching between the supported file types (e.g., CSV, Parquet, JSON).
    pub current_format: Option<ConfigFileType>,

    /// Optional extensions that can be used to extend or customize the behavior of the table
    /// options. Extensions can be registered using `Extensions::insert` and might include
    /// custom file handling logic, additional configuration parameters, or other enhancements.
    pub extensions: Extensions,
}

impl ConfigField for TableOptions {
    /// Visits configuration settings for the current file format, or all formats if none is selected.
    ///
    /// This method adapts the behavior based on whether a file format is currently selected in `current_format`.
    /// If a format is selected, it visits only the settings relevant to that format. Otherwise,
    /// it visits all available format settings.
    fn visit<V: Visit>(&self, v: &mut V, _key_prefix: &str, _description: &'static str) {
        if let Some(file_type) = &self.current_format {
            match file_type {
                #[cfg(feature = "parquet")]
                ConfigFileType::PARQUET => self.parquet.visit(v, "format", ""),
                ConfigFileType::CSV => self.csv.visit(v, "format", ""),
                ConfigFileType::JSON => self.json.visit(v, "format", ""),
            }
        } else {
            self.csv.visit(v, "csv", "");
            self.parquet.visit(v, "parquet", "");
            self.json.visit(v, "json", "");
        }
    }

    /// Sets a configuration value for a specific key within `TableOptions`.
    ///
    /// This method delegates setting configuration values to the specific file format configurations,
    /// based on the current format selected. If no format is selected, it returns an error.
    ///
    /// # Parameters
    ///
    /// * `key`: The configuration key specifying which setting to adjust, prefixed with the format (e.g., "format.delimiter")
    ///   for CSV format.
    /// * `value`: The value to set for the specified configuration key.
    ///
    /// # Returns
    ///
    /// A result indicating success or an error if the key is not recognized, if a format is not specified,
    /// or if setting the configuration value fails for the specific format.
    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        // Extensions are handled in the public `ConfigOptions::set`
        let (key, rem) = key.split_once('.').unwrap_or((key, ""));
        match key {
            "format" => {
                let Some(format) = &self.current_format else {
                    return _config_err!("Specify a format for TableOptions");
                };
                match format {
                    #[cfg(feature = "parquet")]
                    ConfigFileType::PARQUET => self.parquet.set(rem, value),
                    ConfigFileType::CSV => self.csv.set(rem, value),
                    ConfigFileType::JSON => self.json.set(rem, value),
                }
            }
            _ => _config_err!("Config value \"{key}\" not found on TableOptions"),
        }
    }
}

impl TableOptions {
    /// Constructs a new instance of `TableOptions` with default settings.
    ///
    /// # Returns
    ///
    /// A new `TableOptions` instance with default configuration values.
    pub fn new() -> Self {
        Self::default()
    }

    /// Creates a new `TableOptions` instance initialized with settings from a given session config.
    ///
    /// # Parameters
    ///
    /// * `config`: A reference to the session `ConfigOptions` from which to derive initial settings.
    ///
    /// # Returns
    ///
    /// A new `TableOptions` instance with settings applied from the session config.
    pub fn default_from_session_config(config: &ConfigOptions) -> Self {
        let initial = TableOptions::default();
        initial.combine_with_session_config(config)
    }

    /// Updates the current `TableOptions` with settings from a given session config.
    ///
    /// # Parameters
    ///
    /// * `config`: A reference to the session `ConfigOptions` whose settings are to be applied.
    ///
    /// # Returns
    ///
    /// A new `TableOptions` instance with updated settings from the session config.
    #[must_use = "this method returns a new instance"]
    pub fn combine_with_session_config(&self, config: &ConfigOptions) -> Self {
        let mut clone = self.clone();
        clone.parquet.global = config.execution.parquet.clone();
        clone
    }

    /// Sets the file format for the table.
    ///
    /// # Parameters
    ///
    /// * `format`: The file format to use (e.g., CSV, Parquet).
    pub fn set_config_format(&mut self, format: ConfigFileType) {
        self.current_format = Some(format);
    }

    /// Sets the extensions for this `TableOptions` instance.
    ///
    /// # Parameters
    ///
    /// * `extensions`: The `Extensions` instance to set.
    ///
    /// # Returns
    ///
    /// A new `TableOptions` instance with the specified extensions applied.
    pub fn with_extensions(mut self, extensions: Extensions) -> Self {
        self.extensions = extensions;
        self
    }

    /// Sets a specific configuration option.
    ///
    /// # Parameters
    ///
    /// * `key`: The configuration key (e.g., "format.delimiter").
    /// * `value`: The value to set for the specified key.
    ///
    /// # Returns
    ///
    /// A result indicating success or failure in setting the configuration option.
    pub fn set(&mut self, key: &str, value: &str) -> Result<()> {
        let Some((mut prefix, _)) = key.split_once('.') else {
            return _config_err!("could not find config namespace for key \"{key}\"");
        };

        if prefix == "format" {
            return ConfigField::set(self, key, value);
        }

        if prefix == "execution" {
            return Ok(());
        }

        if !self.extensions.0.contains_key(prefix)
            && self
                .extensions
                .0
                .contains_key(DATAFUSION_FFI_CONFIG_NAMESPACE)
        {
            prefix = DATAFUSION_FFI_CONFIG_NAMESPACE;
        }

        let Some(e) = self.extensions.0.get_mut(prefix) else {
            return _config_err!("Could not find config namespace \"{prefix}\"");
        };
        e.0.set(key, value)
    }

    /// Initializes a new `TableOptions` from a hash map of string settings.
    ///
    /// # Parameters
    ///
    /// * `settings`: A hash map where each key-value pair represents a configuration setting.
    ///
    /// # Returns
    ///
    /// A result containing the new `TableOptions` instance or an error if any setting could not be applied.
    pub fn from_string_hash_map(settings: &HashMap<String, String>) -> Result<Self> {
        let mut ret = Self::default();
        for (k, v) in settings {
            ret.set(k, v)?;
        }

        Ok(ret)
    }

    /// Modifies the current `TableOptions` instance with settings from a hash map.
    ///
    /// # Parameters
    ///
    /// * `settings`: A hash map where each key-value pair represents a configuration setting.
    ///
    /// # Returns
    ///
    /// A result indicating success or failure in applying the settings.
    pub fn alter_with_string_hash_map(
        &mut self,
        settings: &HashMap<String, String>,
    ) -> Result<()> {
        for (k, v) in settings {
            self.set(k, v)?;
        }
        Ok(())
    }

    /// Retrieves all configuration entries from this `TableOptions`.
    ///
    /// # Returns
    ///
    /// A vector of `ConfigEntry` instances, representing all the configuration options within this `TableOptions`.
    pub fn entries(&self) -> Vec<ConfigEntry> {
        struct Visitor(Vec<ConfigEntry>);

        impl Visit for Visitor {
            fn some<V: Display>(
                &mut self,
                key: &str,
                value: V,
                description: &'static str,
            ) {
                self.0.push(ConfigEntry {
                    key: key.to_string(),
                    value: Some(value.to_string()),
                    description,
                })
            }

            fn none(&mut self, key: &str, description: &'static str) {
                self.0.push(ConfigEntry {
                    key: key.to_string(),
                    value: None,
                    description,
                })
            }
        }

        let mut v = Visitor(vec![]);
        self.visit(&mut v, "format", "");

        v.0.extend(self.extensions.0.values().flat_map(|e| e.0.entries()));
        v.0
    }
}

/// Options that control how Parquet files are read, including global options
/// that apply to all columns and optional column-specific overrides
///
/// Closely tied to `ParquetWriterOptions` (see `crate::file_options::parquet_writer::ParquetWriterOptions` when the "parquet" feature is enabled).
/// Properties not included in [`TableParquetOptions`] may not be configurable at the external API
/// (e.g. sorting_columns).
#[derive(Clone, Default, Debug, PartialEq)]
pub struct TableParquetOptions {
    /// Global Parquet options that propagates to all columns.
    pub global: ParquetOptions,
    /// Column specific options. Default usage is parquet.XX::column.
    pub column_specific_options: HashMap<String, ParquetColumnOptions>,
    /// Additional file-level metadata to include. Inserted into the key_value_metadata
    /// for the written [`FileMetaData`](https://docs.rs/parquet/latest/parquet/file/metadata/struct.FileMetaData.html).
    ///
    /// Multiple entries are permitted
    /// ```sql
    /// OPTIONS (
    ///    'format.metadata::key1' '',
    ///    'format.metadata::key2' 'value',
    ///    'format.metadata::key3' 'value has spaces',
    ///    'format.metadata::key4' 'value has special chars :: :',
    ///    'format.metadata::key_dupe' 'original will be overwritten',
    ///    'format.metadata::key_dupe' 'final'
    /// )
    /// ```
    pub key_value_metadata: HashMap<String, Option<String>>,
    /// Options for configuring Parquet modular encryption
    ///
    /// To use Parquet encryption, you must enable the `parquet_encryption` feature flag, as it is not activated by default.
    /// See ConfigFileEncryptionProperties and ConfigFileDecryptionProperties in datafusion/common/src/config.rs
    /// These can be set via 'format.crypto', for example:
    /// ```sql
    /// OPTIONS (
    ///    'format.crypto.file_encryption.encrypt_footer' 'true',
    ///    'format.crypto.file_encryption.footer_key_as_hex' '30313233343536373839303132333435',  -- b"0123456789012345" */
    ///    'format.crypto.file_encryption.column_key_as_hex::double_field' '31323334353637383930313233343530', -- b"1234567890123450"
    ///    'format.crypto.file_encryption.column_key_as_hex::float_field' '31323334353637383930313233343531', -- b"1234567890123451"
    ///     -- Same for decryption
    ///    'format.crypto.file_decryption.footer_key_as_hex' '30313233343536373839303132333435', -- b"0123456789012345"
    ///    'format.crypto.file_decryption.column_key_as_hex::double_field' '31323334353637383930313233343530', -- b"1234567890123450"
    ///    'format.crypto.file_decryption.column_key_as_hex::float_field' '31323334353637383930313233343531', -- b"1234567890123451"
    /// )
    /// ```
    /// See datafusion-cli/tests/sql/encrypted_parquet.sql for a more complete example.
    /// Note that keys must be provided as in hex format since these are binary strings.
    pub crypto: ParquetEncryptionOptions,
}

impl TableParquetOptions {
    /// Return new default TableParquetOptions
    pub fn new() -> Self {
        Self::default()
    }

    /// Set whether the encoding of the arrow metadata should occur
    /// during the writing of parquet.
    ///
    /// Default is to encode the arrow schema in the file kv_metadata.
    pub fn with_skip_arrow_metadata(self, skip: bool) -> Self {
        Self {
            global: ParquetOptions {
                skip_arrow_metadata: skip,
                ..self.global
            },
            ..self
        }
    }

    /// Retrieves all configuration entries from this `TableParquetOptions`.
    ///
    /// # Returns
    ///
    /// A vector of `ConfigEntry` instances, representing all the configuration options within this
    pub fn entries(self: &TableParquetOptions) -> Vec<ConfigEntry> {
        struct Visitor(Vec<ConfigEntry>);

        impl Visit for Visitor {
            fn some<V: Display>(
                &mut self,
                key: &str,
                value: V,
                description: &'static str,
            ) {
                self.0.push(ConfigEntry {
                    key: key[1..].to_string(),
                    value: Some(value.to_string()),
                    description,
                })
            }

            fn none(&mut self, key: &str, description: &'static str) {
                self.0.push(ConfigEntry {
                    key: key[1..].to_string(),
                    value: None,
                    description,
                })
            }
        }

        let mut v = Visitor(vec![]);
        self.visit(&mut v, "", "");

        v.0
    }
}

impl ConfigField for TableParquetOptions {
    fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, description: &'static str) {
        self.global.visit(v, key_prefix, description);
        self.column_specific_options
            .visit(v, key_prefix, description);
        self.crypto
            .visit(v, &format!("{key_prefix}.crypto"), description);
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        // Determine if the key is a global, metadata, or column-specific setting
        if key.starts_with("metadata::") {
            let k = match key.split("::").collect::<Vec<_>>()[..] {
                [_meta] | [_meta, ""] => {
                    return _config_err!(
                        "Invalid metadata key provided, missing key in metadata::<key>"
                    );
                }
                [_meta, k] => k.into(),
                _ => {
                    return _config_err!(
                        "Invalid metadata key provided, found too many '::' in \"{key}\""
                    );
                }
            };
            self.key_value_metadata.insert(k, Some(value.into()));
            Ok(())
        } else if let Some(crypto_feature) = key.strip_prefix("crypto.") {
            self.crypto.set(crypto_feature, value)
        } else if key.contains("::") {
            self.column_specific_options.set(key, value)
        } else {
            self.global.set(key, value)
        }
    }
}

macro_rules! config_namespace_with_hashmap {
    (
     $(#[doc = $struct_d:tt])*
     $(#[deprecated($($struct_depr:tt)*)])?  // Optional struct-level deprecated attribute
     $vis:vis struct $struct_name:ident {
        $(
        $(#[doc = $d:tt])*
        $(#[deprecated($($field_depr:tt)*)])? // Optional field-level deprecated attribute
        $field_vis:vis $field_name:ident : $field_type:ty, $(transform = $transform:expr,)? default = $default:expr
        )*$(,)*
    }
    ) => {

        $(#[doc = $struct_d])*
        $(#[deprecated($($struct_depr)*)])?  // Apply struct deprecation
        #[derive(Debug, Clone, PartialEq)]
        $vis struct $struct_name{
            $(
            $(#[doc = $d])*
            $(#[deprecated($($field_depr)*)])? // Apply field deprecation
            $field_vis $field_name : $field_type,
            )*
        }

        impl ConfigField for $struct_name {
            fn set(&mut self, key: &str, value: &str) -> Result<()> {
                let (key, rem) = key.split_once('.').unwrap_or((key, ""));
                match key {
                    $(
                       stringify!($field_name) => {
                           // Handle deprecated fields
                           $(let value = $transform(value);)?
                           self.$field_name.set(rem, value.as_ref())
                       },
                    )*
                    _ => _config_err!(
                        "Config value \"{}\" not found on {}", key, stringify!($struct_name)
                    )
                }
            }

            fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
                $(
                let key = format!(concat!("{}.", stringify!($field_name)), key_prefix);
                let desc = concat!($($d),*).trim();
                // Handle deprecated fields
                self.$field_name.visit(v, key.as_str(), desc);
                )*
            }
        }

        impl Default for $struct_name {
            fn default() -> Self {
                Self {
                    $($field_name: $default),*
                }
            }
        }

        impl ConfigField for HashMap<String,$struct_name> {
            fn set(&mut self, key: &str, value: &str) -> Result<()> {
                let parts: Vec<&str> = key.splitn(2, "::").collect();
                match parts.as_slice() {
                    [inner_key, hashmap_key] => {
                        // Get or create the struct for the specified key
                        let inner_value = self
                            .entry((*hashmap_key).to_owned())
                            .or_insert_with($struct_name::default);

                        inner_value.set(inner_key, value)
                    }
                    _ => _config_err!("Unrecognized key '{key}'."),
                }
            }

            fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
                for (column_name, col_options) in self {
                    $(
                    let key = format!("{}.{field}::{}", key_prefix, column_name, field = stringify!($field_name));
                    let desc = concat!($($d),*).trim();
                    col_options.$field_name.visit(v, key.as_str(), desc);
                    )*
                }
            }
        }
    }
}

config_namespace_with_hashmap! {
    /// Options controlling parquet format for individual columns.
    ///
    /// See [`ParquetOptions`] for more details
    pub struct ParquetColumnOptions {
        /// Sets if bloom filter is enabled for the column path.
        pub bloom_filter_enabled: Option<bool>, default = None

        /// Sets encoding for the column path.
        /// Valid values are: plain, plain_dictionary, rle,
        /// bit_packed, delta_binary_packed, delta_length_byte_array,
        /// delta_byte_array, rle_dictionary, and byte_stream_split.
        /// These values are not case-sensitive. If NULL, uses
        /// default parquet options
        pub encoding: Option<String>, default = None

        /// Sets if dictionary encoding is enabled for the column path. If NULL, uses
        /// default parquet options
        pub dictionary_enabled: Option<bool>, default = None

        /// Sets default parquet compression codec for the column path.
        /// Valid values are: uncompressed, snappy, gzip(level),
        /// brotli(level), lz4, zstd(level), and lz4_raw.
        /// These values are not case-sensitive. If NULL, uses
        /// default parquet options
        pub compression: Option<String>, transform = str::to_lowercase, default = None

        /// Sets if statistics are enabled for the column
        /// Valid values are: "none", "chunk", and "page"
        /// These values are not case sensitive. If NULL, uses
        /// default parquet options
        pub statistics_enabled: Option<String>, default = None

        /// Sets bloom filter false positive probability for the column path. If NULL, uses
        /// default parquet options
        pub bloom_filter_fpp: Option<f64>, default = None

        /// Sets bloom filter number of distinct values. If NULL, uses
        /// default parquet options
        pub bloom_filter_ndv: Option<u64>, default = None
    }
}

#[derive(Clone, Debug, PartialEq)]
pub struct ConfigFileEncryptionProperties {
    /// Should the parquet footer be encrypted
    /// default is true
    pub encrypt_footer: bool,
    /// Key to use for the parquet footer encoded in hex format
    pub footer_key_as_hex: String,
    /// Metadata information for footer key
    pub footer_key_metadata_as_hex: String,
    /// HashMap of column names --> (key in hex format, metadata)
    pub column_encryption_properties: HashMap<String, ColumnEncryptionProperties>,
    /// AAD prefix string uniquely identifies the file and prevents file swapping
    pub aad_prefix_as_hex: String,
    /// If true, store the AAD prefix in the file
    /// default is false
    pub store_aad_prefix: bool,
}

// Setup to match EncryptionPropertiesBuilder::new()
impl Default for ConfigFileEncryptionProperties {
    fn default() -> Self {
        ConfigFileEncryptionProperties {
            encrypt_footer: true,
            footer_key_as_hex: String::new(),
            footer_key_metadata_as_hex: String::new(),
            column_encryption_properties: Default::default(),
            aad_prefix_as_hex: String::new(),
            store_aad_prefix: false,
        }
    }
}

config_namespace_with_hashmap! {
    pub struct ColumnEncryptionProperties {
        /// Per column encryption key
        pub column_key_as_hex: String, default = "".to_string()
        /// Per column encryption key metadata
        pub column_metadata_as_hex: Option<String>, default = None
    }
}

impl ConfigField for ConfigFileEncryptionProperties {
    fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
        let key = format!("{key_prefix}.encrypt_footer");
        let desc = "Encrypt the footer";
        self.encrypt_footer.visit(v, key.as_str(), desc);

        let key = format!("{key_prefix}.footer_key_as_hex");
        let desc = "Key to use for the parquet footer";
        self.footer_key_as_hex.visit(v, key.as_str(), desc);

        let key = format!("{key_prefix}.footer_key_metadata_as_hex");
        let desc = "Metadata to use for the parquet footer";
        self.footer_key_metadata_as_hex.visit(v, key.as_str(), desc);

        self.column_encryption_properties.visit(v, key_prefix, desc);

        let key = format!("{key_prefix}.aad_prefix_as_hex");
        let desc = "AAD prefix to use";
        self.aad_prefix_as_hex.visit(v, key.as_str(), desc);

        let key = format!("{key_prefix}.store_aad_prefix");
        let desc = "If true, store the AAD prefix";
        self.store_aad_prefix.visit(v, key.as_str(), desc);

        self.aad_prefix_as_hex.visit(v, key.as_str(), desc);
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        // Any hex encoded values must be pre-encoded using
        // hex::encode() before calling set.

        if key.contains("::") {
            // Handle any column specific properties
            return self.column_encryption_properties.set(key, value);
        };

        let (key, rem) = key.split_once('.').unwrap_or((key, ""));
        match key {
            "encrypt_footer" => self.encrypt_footer.set(rem, value.as_ref()),
            "footer_key_as_hex" => self.footer_key_as_hex.set(rem, value.as_ref()),
            "footer_key_metadata_as_hex" => {
                self.footer_key_metadata_as_hex.set(rem, value.as_ref())
            }
            "aad_prefix_as_hex" => self.aad_prefix_as_hex.set(rem, value.as_ref()),
            "store_aad_prefix" => self.store_aad_prefix.set(rem, value.as_ref()),
            _ => _config_err!(
                "Config value \"{}\" not found on ConfigFileEncryptionProperties",
                key
            ),
        }
    }
}

#[cfg(feature = "parquet_encryption")]
impl TryFrom<ConfigFileEncryptionProperties> for FileEncryptionProperties {
    type Error = DataFusionError;

    fn try_from(val: ConfigFileEncryptionProperties) -> Result<Self> {
        let mut fep = FileEncryptionProperties::builder(
            hex::decode(val.footer_key_as_hex)
            .map_err(|e| {
                DataFusionError::Configuration(format!("Unable to decode hex footer key from ConfigFileEncryptionProperties: {e}"))
            })?,
        )
        .with_plaintext_footer(!val.encrypt_footer)
        .with_aad_prefix_storage(val.store_aad_prefix);

        if !val.footer_key_metadata_as_hex.is_empty() {
            fep = fep.with_footer_key_metadata(
                hex::decode(&val.footer_key_metadata_as_hex)
                    .map_err(|e| {
                        DataFusionError::Configuration(format!("Unable to decode hex footer key metadata from ConfigFileEncryptionProperties: {e}"))
                    })?,
            );
        }

        for (column_name, encryption_props) in val.column_encryption_properties.iter() {
            let encryption_key = hex::decode(&encryption_props.column_key_as_hex)
                .map_err(|e| {
                    DataFusionError::Configuration(format!("Unable to decode hex encryption key for column {column_name}: {e}"))
                })?;
            let key_metadata = encryption_props
                .column_metadata_as_hex
                .as_ref()
                .map(hex::decode)
                .transpose()
                .map_err(|e| {
                    DataFusionError::Configuration(format!("Unable to decode hex column metadata for column {column_name}: {e}"))
                })?;

            match key_metadata {
                Some(key_metadata) => {
                    fep = fep.with_column_key_and_metadata(
                        column_name,
                        encryption_key,
                        key_metadata,
                    );
                }
                None => {
                    fep = fep.with_column_key(column_name, encryption_key);
                }
            }
        }

        if !val.aad_prefix_as_hex.is_empty() {
            let aad_prefix: Vec<u8> = hex::decode(&val.aad_prefix_as_hex).map_err(|e| {
                DataFusionError::Configuration(format!(
                    "Unable to decode hex AAD prefix from ConfigFileEncryptionProperties: {e}"
                ))
            })?;
            fep = fep.with_aad_prefix(aad_prefix);
        }
        Ok(Arc::unwrap_or_clone(fep.build().map_err(|e| {
            DataFusionError::Configuration(format!(
                "Could not build FileEncryptionProperties: {e}"
            ))
        })?))
    }
}

#[cfg(feature = "parquet_encryption")]
impl From<&Arc<FileEncryptionProperties>> for ConfigFileEncryptionProperties {
    fn from(f: &Arc<FileEncryptionProperties>) -> Self {
        let (column_names_vec, column_keys_vec, column_metas_vec) = f.column_keys();

        let mut column_encryption_properties: HashMap<
            String,
            ColumnEncryptionProperties,
        > = HashMap::new();

        for (i, column_name) in column_names_vec.iter().enumerate() {
            let column_key_as_hex = hex::encode(&column_keys_vec[i]);
            let column_metadata_as_hex: Option<String> =
                column_metas_vec.get(i).map(hex::encode);
            column_encryption_properties.insert(
                column_name.clone(),
                ColumnEncryptionProperties {
                    column_key_as_hex,
                    column_metadata_as_hex,
                },
            );
        }
        let aad_prefix = f.aad_prefix().cloned().unwrap_or_default();
        ConfigFileEncryptionProperties {
            encrypt_footer: f.encrypt_footer(),
            footer_key_as_hex: hex::encode(f.footer_key()),
            footer_key_metadata_as_hex: f
                .footer_key_metadata()
                .map(hex::encode)
                .unwrap_or_default(),
            column_encryption_properties,
            aad_prefix_as_hex: hex::encode(aad_prefix),
            store_aad_prefix: f.store_aad_prefix(),
        }
    }
}

#[derive(Clone, Debug, PartialEq)]
pub struct ConfigFileDecryptionProperties {
    /// Binary string to use for the parquet footer encoded in hex format
    pub footer_key_as_hex: String,
    /// HashMap of column names --> key in hex format
    pub column_decryption_properties: HashMap<String, ColumnDecryptionProperties>,
    /// AAD prefix string uniquely identifies the file and prevents file swapping
    pub aad_prefix_as_hex: String,
    /// If true, then verify signature for files with plaintext footers.
    /// default = true
    pub footer_signature_verification: bool,
}

config_namespace_with_hashmap! {
    pub struct ColumnDecryptionProperties {
        /// Per column encryption key
        pub column_key_as_hex: String, default = "".to_string()
    }
}

// Setup to match DecryptionPropertiesBuilder::new()
impl Default for ConfigFileDecryptionProperties {
    fn default() -> Self {
        ConfigFileDecryptionProperties {
            footer_key_as_hex: String::new(),
            column_decryption_properties: Default::default(),
            aad_prefix_as_hex: String::new(),
            footer_signature_verification: true,
        }
    }
}

impl ConfigField for ConfigFileDecryptionProperties {
    fn visit<V: Visit>(&self, v: &mut V, key_prefix: &str, _description: &'static str) {
        let key = format!("{key_prefix}.footer_key_as_hex");
        let desc = "Key to use for the parquet footer";
        self.footer_key_as_hex.visit(v, key.as_str(), desc);

        let key = format!("{key_prefix}.aad_prefix_as_hex");
        let desc = "AAD prefix to use";
        self.aad_prefix_as_hex.visit(v, key.as_str(), desc);

        let key = format!("{key_prefix}.footer_signature_verification");
        let desc = "If true, verify the footer signature";
        self.footer_signature_verification
            .visit(v, key.as_str(), desc);

        self.column_decryption_properties.visit(v, key_prefix, desc);
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        // Any hex encoded values must be pre-encoded using
        // hex::encode() before calling set.

        if key.contains("::") {
            // Handle any column specific properties
            return self.column_decryption_properties.set(key, value);
        };

        let (key, rem) = key.split_once('.').unwrap_or((key, ""));
        match key {
            "footer_key_as_hex" => self.footer_key_as_hex.set(rem, value.as_ref()),
            "aad_prefix_as_hex" => self.aad_prefix_as_hex.set(rem, value.as_ref()),
            "footer_signature_verification" => {
                self.footer_signature_verification.set(rem, value.as_ref())
            }
            _ => _config_err!(
                "Config value \"{}\" not found on ConfigFileDecryptionProperties",
                key
            ),
        }
    }
}

#[cfg(feature = "parquet_encryption")]
impl TryFrom<ConfigFileDecryptionProperties> for FileDecryptionProperties {
    type Error = DataFusionError;

    fn try_from(val: ConfigFileDecryptionProperties) -> Result<Self> {
        let mut column_names: Vec<&str> = Vec::new();
        let mut column_keys: Vec<Vec<u8>> = Vec::new();

        for (col_name, decryption_properties) in val.column_decryption_properties.iter() {
            let column_key = hex::decode(&decryption_properties.column_key_as_hex).map_err(|e| {
                        DataFusionError::Configuration(format!
                            ("Could not decode hex column key from ConfigFileDecryptionProperties for column name {col_name}: {e}."))
                    })?;
            column_names.push(col_name.as_str());
            column_keys.push(column_key);
        }

        let footer_key = hex::decode(val.footer_key_as_hex).map_err(|e| {
            DataFusionError::Configuration(format!(
                "Could not decode hex footer key from ConfigFileDecryptionProperties: {e}."
            ))
        })?;

        let mut fep = FileDecryptionProperties::builder(footer_key)
            .with_column_keys(column_names, column_keys)
            .map_err(|e| {
                DataFusionError::Configuration(format!(
                    "Could not set column keys on FileDecryptionPropertiesBuilder: {e}."
                ))
            })?;

        if !val.footer_signature_verification {
            fep = fep.disable_footer_signature_verification();
        }

        if !val.aad_prefix_as_hex.is_empty() {
            let aad_prefix = hex::decode(&val.aad_prefix_as_hex).map_err(|e| {
                DataFusionError::Configuration(format!(
                    "Could not decode hex AAD prefix from ConfigFileDecryptionProperties: {e}."
                ))
            })?;
            fep = fep.with_aad_prefix(aad_prefix);
        }

        Ok(Arc::unwrap_or_clone(fep.build().map_err(|e| {
            DataFusionError::Configuration(format!(
                "Could not build FileDecryptionProperties: {e}."
            ))
        })?))
    }
}

#[cfg(feature = "parquet_encryption")]
impl TryFrom<&Arc<FileDecryptionProperties>> for ConfigFileDecryptionProperties {
    type Error = DataFusionError;

    fn try_from(f: &Arc<FileDecryptionProperties>) -> Result<Self> {
        let footer_key = f.footer_key(None).map_err(|e| {
            DataFusionError::Configuration(format!(
                "Could not retrieve footer key from FileDecryptionProperties. \
                Note that conversion to ConfigFileDecryptionProperties is not supported \
                when using a key retriever: {e}"
            ))
        })?;

        let (column_names_vec, column_keys_vec) = f.column_keys();
        let mut column_decryption_properties: HashMap<
            String,
            ColumnDecryptionProperties,
        > = HashMap::new();
        for (i, column_name) in column_names_vec.iter().enumerate() {
            let props = ColumnDecryptionProperties {
                column_key_as_hex: hex::encode(column_keys_vec[i].clone()),
            };
            column_decryption_properties.insert(column_name.clone(), props);
        }

        let aad_prefix = f.aad_prefix().cloned().unwrap_or_default();
        Ok(ConfigFileDecryptionProperties {
            footer_key_as_hex: hex::encode(footer_key.as_ref()),
            column_decryption_properties,
            aad_prefix_as_hex: hex::encode(aad_prefix),
            footer_signature_verification: f.check_plaintext_footer_integrity(),
        })
    }
}

/// Holds implementation-specific options for an encryption factory
#[derive(Clone, Debug, Default, PartialEq)]
pub struct EncryptionFactoryOptions {
    pub options: HashMap<String, String>,
}

impl ConfigField for EncryptionFactoryOptions {
    fn visit<V: Visit>(&self, v: &mut V, key: &str, _description: &'static str) {
        for (option_key, option_value) in &self.options {
            v.some(
                &format!("{key}.{option_key}"),
                option_value,
                "Encryption factory specific option",
            );
        }
    }

    fn set(&mut self, key: &str, value: &str) -> Result<()> {
        self.options.insert(key.to_owned(), value.to_owned());
        Ok(())
    }
}

impl EncryptionFactoryOptions {
    /// Convert these encryption factory options to an [`ExtensionOptions`] instance.
    pub fn to_extension_options<T: ExtensionOptions + Default>(&self) -> Result<T> {
        let mut options = T::default();
        for (key, value) in &self.options {
            options.set(key, value)?;
        }
        Ok(options)
    }
}

config_namespace! {
    /// Options controlling CSV format
    pub struct CsvOptions {
        /// Specifies whether there is a CSV header (i.e. the first line
        /// consists of is column names). The value `None` indicates that
        /// the configuration should be consulted.
        pub has_header: Option<bool>, default = None
        pub delimiter: u8, default = b','
        pub quote: u8, default = b'"'
        pub terminator: Option<u8>, default = None
        pub escape: Option<u8>, default = None
        pub double_quote: Option<bool>, default = None
        /// Quote style for CSV writing.
        /// One of: "Always", "Necessary", "NonNumeric", "Never"
        pub quote_style: CsvQuoteStyle, default = CsvQuoteStyle::Necessary
        /// Whether to ignore leading whitespace in string values when writing CSV.
        /// Defaults to `false` when `None`.
        pub ignore_leading_whitespace: Option<bool>, default = None
        /// Whether to ignore trailing whitespace in string values when writing CSV.
        /// Defaults to `false` when `None`.
        pub ignore_trailing_whitespace: Option<bool>, default = None
        /// Specifies whether newlines in (quoted) values are supported.
        ///
        /// Parsing newlines in quoted values may be affected by execution behaviour such as
        /// parallel file scanning. Setting this to `true` ensures that newlines in values are
        /// parsed successfully, which may reduce performance.
        ///
        /// The default behaviour depends on the `datafusion.catalog.newlines_in_values` setting.
        pub newlines_in_values: Option<bool>, default = None
        pub compression: CompressionTypeVariant, default = CompressionTypeVariant::UNCOMPRESSED
        /// Compression level for the output file. The valid range depends on the
        /// compression algorithm:
        /// - ZSTD: 1 to 22 (default: 3)
        /// - GZIP: 0 to 9 (default: 6)
        /// - BZIP2: 0 to 9 (default: 6)
        /// - XZ: 0 to 9 (default: 6)
        /// If not specified, the default level for the compression algorithm is used.
        pub compression_level: Option<u32>, default = None
        pub schema_infer_max_rec: Option<usize>, default = None
        pub date_format: Option<String>, default = None
        pub datetime_format: Option<String>, default = None
        pub timestamp_format: Option<String>, default = None
        pub timestamp_tz_format: Option<String>, default = None
        pub time_format: Option<String>, default = None
        // The output format for Nulls in the CSV writer.
        pub null_value: Option<String>, default = None
        // The input regex for Nulls when loading CSVs.
        pub null_regex: Option<String>, default = None
        pub comment: Option<u8>, default = None
        /// Whether to allow truncated rows when parsing, both within a single file and across files.
        ///
        /// When set to false (default), reading a single CSV file which has rows of different lengths will
        /// error; if reading multiple CSV files with different number of columns, it will also fail.
        ///
        /// When set to true, reading a single CSV file with rows of different lengths will pad the truncated
        /// rows with null values for the missing columns; if reading multiple CSV files with different number
        /// of columns, it creates a union schema containing all columns found across the files, and will
        /// pad any files missing columns with null values for their rows.
        pub truncated_rows: Option<bool>, default = None
    }
}

impl CsvOptions {
    /// Set a limit in terms of records to scan to infer the schema
    /// - default to `DEFAULT_SCHEMA_INFER_MAX_RECORD`
    pub fn with_compression(
        mut self,
        compression_type_variant: CompressionTypeVariant,
    ) -> Self {
        self.compression = compression_type_variant;
        self
    }

    /// Set a limit in terms of records to scan to infer the schema
    /// - default to `DEFAULT_SCHEMA_INFER_MAX_RECORD`
    pub fn with_schema_infer_max_rec(mut self, max_rec: usize) -> Self {
        self.schema_infer_max_rec = Some(max_rec);
        self
    }

    /// Set true to indicate that the first line is a header.
    /// - default to true
    pub fn with_has_header(mut self, has_header: bool) -> Self {
        self.has_header = Some(has_header);
        self
    }

    /// Returns true if the first line is a header. If format options does not
    /// specify whether there is a header, returns `None` (indicating that the
    /// configuration should be consulted).
    pub fn has_header(&self) -> Option<bool> {
        self.has_header
    }

    /// The character separating values within a row.
    /// - default to ','
    pub fn with_delimiter(mut self, delimiter: u8) -> Self {
        self.delimiter = delimiter;
        self
    }

    /// The quote character in a row.
    /// - default to '"'
    pub fn with_quote(mut self, quote: u8) -> Self {
        self.quote = quote;
        self
    }

    /// The character that terminates a row.
    /// - default to None (CRLF)
    pub fn with_terminator(mut self, terminator: Option<u8>) -> Self {
        self.terminator = terminator;
        self
    }

    /// The escape character in a row.
    /// - default is None
    pub fn with_escape(mut self, escape: Option<u8>) -> Self {
        self.escape = escape;
        self
    }

    /// Set true to indicate that the CSV quotes should be doubled.
    /// - default to true
    pub fn with_double_quote(mut self, double_quote: bool) -> Self {
        self.double_quote = Some(double_quote);
        self
    }

    /// Set the quote style for CSV writing.
    pub fn with_quote_style(mut self, quote_style: CsvQuoteStyle) -> Self {
        self.quote_style = quote_style;
        self
    }

    /// Set whether to ignore leading whitespace in string values when writing CSV.
    pub fn with_ignore_leading_whitespace(
        mut self,
        ignore_leading_whitespace: bool,
    ) -> Self {
        self.ignore_leading_whitespace = Some(ignore_leading_whitespace);
        self
    }

    /// Set whether to ignore trailing whitespace in string values when writing CSV.
    pub fn with_ignore_trailing_whitespace(
        mut self,
        ignore_trailing_whitespace: bool,
    ) -> Self {
        self.ignore_trailing_whitespace = Some(ignore_trailing_whitespace);
        self
    }

    /// Specifies whether newlines in (quoted) values are supported.
    ///
    /// Parsing newlines in quoted values may be affected by execution behaviour such as
    /// parallel file scanning. Setting this to `true` ensures that newlines in values are
    /// parsed successfully, which may reduce performance.
    ///
    /// The default behaviour depends on the `datafusion.catalog.newlines_in_values` setting.
    pub fn with_newlines_in_values(mut self, newlines_in_values: bool) -> Self {
        self.newlines_in_values = Some(newlines_in_values);
        self
    }

    /// Set a `CompressionTypeVariant` of CSV
    /// - defaults to `CompressionTypeVariant::UNCOMPRESSED`
    pub fn with_file_compression_type(
        mut self,
        compression: CompressionTypeVariant,
    ) -> Self {
        self.compression = compression;
        self
    }

    /// Whether to allow truncated rows when parsing.
    /// By default this is set to false and will error if the CSV rows have different lengths.
    /// When set to true then it will allow records with less than the expected number of columns and fill the missing columns with nulls.
    /// If the record’s schema is not nullable, then it will still return an error.
    pub fn with_truncated_rows(mut self, allow: bool) -> Self {
        self.truncated_rows = Some(allow);
        self
    }

    /// Set the compression level for the output file.
    /// The valid range depends on the compression algorithm.
    /// If not specified, the default level for the algorithm is used.
    pub fn with_compression_level(mut self, level: u32) -> Self {
        self.compression_level = Some(level);
        self
    }

    /// The delimiter character.
    pub fn delimiter(&self) -> u8 {
        self.delimiter
    }

    /// The quote character.
    pub fn quote(&self) -> u8 {
        self.quote
    }

    /// The terminator character.
    pub fn terminator(&self) -> Option<u8> {
        self.terminator
    }

    /// The escape character.
    pub fn escape(&self) -> Option<u8> {
        self.escape
    }
}

config_namespace! {
    /// Options controlling JSON format
    pub struct JsonOptions {
        pub compression: CompressionTypeVariant, default = CompressionTypeVariant::UNCOMPRESSED
        /// Compression level for the output file. The valid range depends on the
        /// compression algorithm:
        /// - ZSTD: 1 to 22 (default: 3)
        /// - GZIP: 0 to 9 (default: 6)
        /// - BZIP2: 0 to 9 (default: 6)
        /// - XZ: 0 to 9 (default: 6)
        /// If not specified, the default level for the compression algorithm is used.
        pub compression_level: Option<u32>, default = None
        pub schema_infer_max_rec: Option<usize>, default = None
       /// The JSON format to use when reading files.
       ///
       /// When `true` (default), expects newline-delimited JSON (NDJSON):
       /// ```text
       /// {"key1": 1, "key2": "val"}
       /// {"key1": 2, "key2": "vals"}
       /// ```
       ///
       /// When `false`, expects JSON array format:
       /// ```text
       /// [
       ///   {"key1": 1, "key2": "val"},
       ///   {"key1": 2, "key2": "vals"}
       /// ]
       /// ```
       pub newline_delimited: bool, default = true
    }
}

pub trait OutputFormatExt: Display {}

#[derive(Debug, Clone, PartialEq)]
#[cfg_attr(feature = "parquet", expect(clippy::large_enum_variant))]
pub enum OutputFormat {
    CSV(CsvOptions),
    JSON(JsonOptions),
    #[cfg(feature = "parquet")]
    PARQUET(TableParquetOptions),
    AVRO,
    ARROW,
}

impl Display for OutputFormat {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let out = match self {
            OutputFormat::CSV(_) => "csv",
            OutputFormat::JSON(_) => "json",
            #[cfg(feature = "parquet")]
            OutputFormat::PARQUET(_) => "parquet",
            OutputFormat::AVRO => "avro",
            OutputFormat::ARROW => "arrow",
        };
        write!(f, "{out}")
    }
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "parquet")]
    use crate::assert_contains;
    use crate::config::TableParquetOptions;
    use crate::config::{
        ConfigEntry, ConfigExtension, ConfigField, ConfigFileType, ExtensionOptions,
        Extensions, TableOptions,
    };
    use std::any::Any;
    use std::collections::HashMap;

    #[derive(Default, Debug, Clone)]
    pub struct TestExtensionConfig {
        /// Should "foo" be replaced by "bar"?
        pub properties: HashMap<String, String>,
    }

    impl ExtensionOptions for TestExtensionConfig {
        fn as_any(&self) -> &dyn Any {
            self
        }

        fn as_any_mut(&mut self) -> &mut dyn Any {
            self
        }

        fn cloned(&self) -> Box<dyn ExtensionOptions> {
            Box::new(self.clone())
        }

        fn set(&mut self, key: &str, value: &str) -> crate::Result<()> {
            let (key, rem) = key.split_once('.').unwrap_or((key, ""));
            assert_eq!(key, "test");
            self.properties.insert(rem.to_owned(), value.to_owned());
            Ok(())
        }

        fn entries(&self) -> Vec<ConfigEntry> {
            self.properties
                .iter()
                .map(|(k, v)| ConfigEntry {
                    key: k.into(),
                    value: Some(v.into()),
                    description: "",
                })
                .collect()
        }
    }

    impl ConfigExtension for TestExtensionConfig {
        const PREFIX: &'static str = "test";
    }

    #[test]
    fn create_table_config() {
        let mut extension = Extensions::new();
        extension.insert(TestExtensionConfig::default());
        let table_config = TableOptions::new().with_extensions(extension);
        let kafka_config = table_config.extensions.get::<TestExtensionConfig>();
        assert!(kafka_config.is_some())
    }

    #[test]
    fn alter_test_extension_config() {
        let mut extension = Extensions::new();
        extension.insert(TestExtensionConfig::default());
        let mut table_config = TableOptions::new().with_extensions(extension);
        table_config.set_config_format(ConfigFileType::CSV);
        table_config.set("format.delimiter", ";").unwrap();
        assert_eq!(table_config.csv.delimiter, b';');
        table_config.set("test.bootstrap.servers", "asd").unwrap();
        let kafka_config = table_config
            .extensions
            .get::<TestExtensionConfig>()
            .unwrap();
        assert_eq!(
            kafka_config.properties.get("bootstrap.servers").unwrap(),
            "asd"
        );
    }

    #[test]
    fn iter_test_extension_config() {
        let mut extension = Extensions::new();
        extension.insert(TestExtensionConfig::default());
        let table_config = TableOptions::new().with_extensions(extension);
        let extensions = table_config.extensions.iter().collect::<Vec<_>>();
        assert_eq!(extensions.len(), 1);
        assert_eq!(extensions[0].0, TestExtensionConfig::PREFIX);
    }

    #[test]
    fn csv_u8_table_options() {
        let mut table_config = TableOptions::new();
        table_config.set_config_format(ConfigFileType::CSV);
        table_config.set("format.delimiter", ";").unwrap();
        assert_eq!(table_config.csv.delimiter as char, ';');
        table_config.set("format.escape", "\"").unwrap();
        assert_eq!(table_config.csv.escape.unwrap() as char, '"');
        table_config.set("format.escape", "\'").unwrap();
        assert_eq!(table_config.csv.escape.unwrap() as char, '\'');
    }

    #[test]
    fn warning_only_not_default() {
        use std::sync::atomic::AtomicUsize;
        static COUNT: AtomicUsize = AtomicUsize::new(0);
        use log::{Level, LevelFilter, Metadata, Record};
        struct SimpleLogger;
        impl log::Log for SimpleLogger {
            fn enabled(&self, metadata: &Metadata) -> bool {
                metadata.level() <= Level::Info
            }

            fn log(&self, record: &Record) {
                if self.enabled(record.metadata()) {
                    COUNT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
            }
            fn flush(&self) {}
        }
        log::set_logger(&SimpleLogger).unwrap();
        log::set_max_level(LevelFilter::Info);
        let mut sql_parser_options = crate::config::SqlParserOptions::default();
        sql_parser_options
            .set("enable_options_value_normalization", "false")
            .unwrap();
        assert_eq!(COUNT.load(std::sync::atomic::Ordering::Relaxed), 0);
        sql_parser_options
            .set("enable_options_value_normalization", "true")
            .unwrap();
        assert_eq!(COUNT.load(std::sync::atomic::Ordering::Relaxed), 1);
    }

    #[test]
    fn reset_nested_scalar_reports_helpful_error() {
        let mut value = true;
        let err = <bool as ConfigField>::reset(&mut value, "nested").unwrap_err();
        let message = err.to_string();
        assert!(
            message.starts_with(
                "Invalid or Unsupported Configuration: Config field is a scalar bool and does not have nested field \"nested\""
            ),
            "unexpected error message: {message}"
        );
    }

    #[cfg(feature = "parquet")]
    #[test]
    fn parquet_table_options() {
        let mut table_config = TableOptions::new();
        table_config.set_config_format(ConfigFileType::PARQUET);
        table_config
            .set("format.bloom_filter_enabled::col1", "true")
            .unwrap();
        assert_eq!(
            table_config.parquet.column_specific_options["col1"].bloom_filter_enabled,
            Some(true)
        );
    }

    #[cfg(feature = "parquet_encryption")]
    #[test]
    fn parquet_table_encryption() {
        use crate::config::{
            ConfigFileDecryptionProperties, ConfigFileEncryptionProperties,
        };
        use parquet::encryption::decrypt::FileDecryptionProperties;
        use parquet::encryption::encrypt::FileEncryptionProperties;
        use std::sync::Arc;

        let footer_key = b"0123456789012345".to_vec(); // 128bit/16
        let column_names = vec!["double_field", "float_field"];
        let column_keys =
            vec![b"1234567890123450".to_vec(), b"1234567890123451".to_vec()];

        let file_encryption_properties =
            FileEncryptionProperties::builder(footer_key.clone())
                .with_column_keys(column_names.clone(), column_keys.clone())
                .unwrap()
                .build()
                .unwrap();

        let decryption_properties = FileDecryptionProperties::builder(footer_key.clone())
            .with_column_keys(column_names.clone(), column_keys.clone())
            .unwrap()
            .build()
            .unwrap();

        // Test round-trip
        let config_encrypt =
            ConfigFileEncryptionProperties::from(&file_encryption_properties);
        let encryption_properties_built =
            Arc::new(FileEncryptionProperties::try_from(config_encrypt.clone()).unwrap());
        assert_eq!(file_encryption_properties, encryption_properties_built);

        let config_decrypt =
            ConfigFileDecryptionProperties::try_from(&decryption_properties).unwrap();
        let decryption_properties_built =
            Arc::new(FileDecryptionProperties::try_from(config_decrypt.clone()).unwrap());
        assert_eq!(decryption_properties, decryption_properties_built);

        ///////////////////////////////////////////////////////////////////////////////////
        // Test encryption config

        // Display original encryption config
        // println!("{:#?}", config_encrypt);

        let mut table_config = TableOptions::new();
        table_config.set_config_format(ConfigFileType::PARQUET);
        table_config
            .parquet
            .set(
                "crypto.file_encryption.encrypt_footer",
                config_encrypt.encrypt_footer.to_string().as_str(),
            )
            .unwrap();
        table_config
            .parquet
            .set(
                "crypto.file_encryption.footer_key_as_hex",
                config_encrypt.footer_key_as_hex.as_str(),
            )
            .unwrap();

        for (i, col_name) in column_names.iter().enumerate() {
            let key = format!("crypto.file_encryption.column_key_as_hex::{col_name}");
            let value = hex::encode(column_keys[i].clone());
            table_config
                .parquet
                .set(key.as_str(), value.as_str())
                .unwrap();
        }

        // Print matching final encryption config
        // println!("{:#?}", table_config.parquet.crypto.file_encryption);

        assert_eq!(
            table_config.parquet.crypto.file_encryption,
            Some(config_encrypt)
        );

        ///////////////////////////////////////////////////////////////////////////////////
        // Test decryption config

        // Display original decryption config
        // println!("{:#?}", config_decrypt);

        let mut table_config = TableOptions::new();
        table_config.set_config_format(ConfigFileType::PARQUET);
        table_config
            .parquet
            .set(
                "crypto.file_decryption.footer_key_as_hex",
                config_decrypt.footer_key_as_hex.as_str(),
            )
            .unwrap();

        for (i, col_name) in column_names.iter().enumerate() {
            let key = format!("crypto.file_decryption.column_key_as_hex::{col_name}");
            let value = hex::encode(column_keys[i].clone());
            table_config
                .parquet
                .set(key.as_str(), value.as_str())
                .unwrap();
        }

        // Print matching final decryption config
        // println!("{:#?}", table_config.parquet.crypto.file_decryption);

        assert_eq!(
            table_config.parquet.crypto.file_decryption,
            Some(config_decrypt.clone())
        );

        // Set config directly
        let mut table_config = TableOptions::new();
        table_config.set_config_format(ConfigFileType::PARQUET);
        table_config.parquet.crypto.file_decryption = Some(config_decrypt.clone());
        assert_eq!(
            table_config.parquet.crypto.file_decryption,
            Some(config_decrypt.clone())
        );
    }

    #[cfg(feature = "parquet_encryption")]
    #[test]
    fn parquet_encryption_invalid_hex_errors_encryption() {
        use crate::config::ColumnEncryptionProperties;
        use crate::config::ConfigFileEncryptionProperties;
        use parquet::encryption::encrypt::FileEncryptionProperties;
        use std::collections::HashMap;

        let valid_footer_key_as_hex = hex::encode(b"0123456789012345");

        let mut enc = ConfigFileEncryptionProperties {
            encrypt_footer: true,
            footer_key_as_hex: valid_footer_key_as_hex.clone(),
            footer_key_metadata_as_hex: String::new(),
            column_encryption_properties: HashMap::new(),
            aad_prefix_as_hex: String::new(),
            store_aad_prefix: false,
        };

        // Encryption: invalid footer key hex
        enc.footer_key_as_hex = "not_hex".to_string();
        let err = FileEncryptionProperties::try_from(enc.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Unable to decode hex footer key"));
        enc.footer_key_as_hex = valid_footer_key_as_hex.clone();

        // Encryption: invalid footer key metadata hex
        enc.footer_key_metadata_as_hex = "zz".to_string();
        let err = FileEncryptionProperties::try_from(enc.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Unable to decode hex footer key metadata"));
        enc.footer_key_metadata_as_hex = String::new();

        // Encryption: invalid column key hex
        enc.column_encryption_properties.insert(
            "col1".to_string(),
            ColumnEncryptionProperties {
                column_key_as_hex: "bad".to_string(),
                column_metadata_as_hex: None,
            },
        );
        let err = FileEncryptionProperties::try_from(enc.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Unable to decode hex encryption key for column col1"));
        enc.column_encryption_properties.clear();

        // Encryption: invalid column metadata hex
        enc.column_encryption_properties.insert(
            "col1".to_string(),
            ColumnEncryptionProperties {
                column_key_as_hex: hex::encode(b"1234567890123450"),
                column_metadata_as_hex: Some("zz".to_string()),
            },
        );
        let err = FileEncryptionProperties::try_from(enc.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Unable to decode hex column metadata for column col1"));
        enc.column_encryption_properties.clear();

        // Encryption: invalid AAD prefix hex
        enc.aad_prefix_as_hex = "zz".to_string();
        let err = FileEncryptionProperties::try_from(enc.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Unable to decode hex AAD prefix"));
    }

    #[cfg(feature = "parquet_encryption")]
    #[test]
    fn parquet_encryption_invalid_hex_errors_decryption() {
        use crate::config::ColumnDecryptionProperties;
        use crate::config::ConfigFileDecryptionProperties;
        use parquet::encryption::decrypt::FileDecryptionProperties;
        use std::collections::HashMap;

        let valid_footer_key_as_hex = hex::encode(b"0123456789012345");

        let mut dec = ConfigFileDecryptionProperties {
            footer_key_as_hex: valid_footer_key_as_hex.clone(),
            column_decryption_properties: HashMap::new(),
            aad_prefix_as_hex: String::new(),
            footer_signature_verification: true,
        };

        // Decryption: invalid column key hex
        dec.column_decryption_properties.insert(
            "col1".to_string(),
            ColumnDecryptionProperties {
                column_key_as_hex: "bad".to_string(),
            },
        );
        let err = FileDecryptionProperties::try_from(dec.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Could not decode hex column key"));
        assert!(err.contains("col1"));
        dec.column_decryption_properties.clear();

        // Decryption: invalid footer key hex
        dec.footer_key_as_hex = "bad".to_string();
        let err = FileDecryptionProperties::try_from(dec.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Could not decode hex footer key"));
        dec.footer_key_as_hex = valid_footer_key_as_hex;

        // Decryption: invalid AAD prefix hex
        dec.aad_prefix_as_hex = "zz".to_string();
        let err = FileDecryptionProperties::try_from(dec.clone())
            .unwrap_err()
            .to_string();
        assert!(err.contains("Could not decode hex AAD prefix"));
    }

    #[cfg(feature = "parquet_encryption")]
    #[test]
    fn parquet_encryption_factory_config() {
        let mut parquet_options = TableParquetOptions::default();

        assert_eq!(parquet_options.crypto.factory_id, None);
        assert_eq!(parquet_options.crypto.factory_options.options.len(), 0);

        let mut input_config = TestExtensionConfig::default();
        input_config
            .properties
            .insert("key1".to_string(), "value 1".to_string());
        input_config
            .properties
            .insert("key2".to_string(), "value 2".to_string());

        parquet_options
            .crypto
            .configure_factory("example_factory", &input_config);

        assert_eq!(
            parquet_options.crypto.factory_id,
            Some("example_factory".to_string())
        );
        let factory_options = &parquet_options.crypto.factory_options.options;
        assert_eq!(factory_options.len(), 2);
        assert_eq!(factory_options.get("key1"), Some(&"value 1".to_string()));
        assert_eq!(factory_options.get("key2"), Some(&"value 2".to_string()));
    }

    #[cfg(feature = "parquet_encryption")]
    struct ParquetEncryptionKeyRetriever {}

    #[cfg(feature = "parquet_encryption")]
    impl parquet::encryption::decrypt::KeyRetriever for ParquetEncryptionKeyRetriever {
        fn retrieve_key(&self, key_metadata: &[u8]) -> parquet::errors::Result<Vec<u8>> {
            if !key_metadata.is_empty() {
                Ok(b"1234567890123450".to_vec())
            } else {
                Err(parquet::errors::ParquetError::General(
                    "Key metadata not provided".to_string(),
                ))
            }
        }
    }

    #[cfg(feature = "parquet_encryption")]
    #[test]
    fn conversion_from_key_retriever_to_config_file_decryption_properties() {
        use crate::Result;
        use crate::config::ConfigFileDecryptionProperties;
        use crate::encryption::FileDecryptionProperties;

        let retriever = std::sync::Arc::new(ParquetEncryptionKeyRetriever {});
        let decryption_properties =
            FileDecryptionProperties::with_key_retriever(retriever)
                .build()
                .unwrap();
        let config_file_decryption_properties: Result<ConfigFileDecryptionProperties> =
            (&decryption_properties).try_into();
        assert!(config_file_decryption_properties.is_err());
        let err = config_file_decryption_properties.unwrap_err().to_string();
        assert!(err.contains("key retriever"));
        assert!(err.contains("Key metadata not provided"));
    }

    #[cfg(feature = "parquet")]
    #[test]
    fn parquet_table_options_config_entry() {
        let mut table_config = TableOptions::new();
        table_config.set_config_format(ConfigFileType::PARQUET);
        table_config
            .set("format.bloom_filter_enabled::col1", "true")
            .unwrap();
        let entries = table_config.entries();
        assert!(
            entries
                .iter()
                .any(|item| item.key == "format.bloom_filter_enabled::col1")
        )
    }

    #[cfg(feature = "parquet")]
    #[test]
    fn parquet_table_parquet_options_config_entry() {
        let mut table_parquet_options = TableParquetOptions::new();
        table_parquet_options
            .set(
                "crypto.file_encryption.column_key_as_hex::double_field",
                "31323334353637383930313233343530",
            )
            .unwrap();
        let entries = table_parquet_options.entries();
        assert!(
            entries.iter().any(|item| item.key
                == "crypto.file_encryption.column_key_as_hex::double_field")
        )
    }

    #[cfg(feature = "parquet")]
    #[test]
    fn parquet_table_options_config_metadata_entry() {
        let mut table_config = TableOptions::new();
        table_config.set_config_format(ConfigFileType::PARQUET);
        table_config.set("format.metadata::key1", "").unwrap();
        table_config.set("format.metadata::key2", "value2").unwrap();
        table_config
            .set("format.metadata::key3", "value with spaces ")
            .unwrap();
        table_config
            .set("format.metadata::key4", "value with special chars :: :")
            .unwrap();

        let parsed_metadata = table_config.parquet.key_value_metadata.clone();
        assert_eq!(parsed_metadata.get("should not exist1"), None);
        assert_eq!(parsed_metadata.get("key1"), Some(&Some("".into())));
        assert_eq!(parsed_metadata.get("key2"), Some(&Some("value2".into())));
        assert_eq!(
            parsed_metadata.get("key3"),
            Some(&Some("value with spaces ".into()))
        );
        assert_eq!(
            parsed_metadata.get("key4"),
            Some(&Some("value with special chars :: :".into()))
        );

        // duplicate keys are overwritten
        table_config.set("format.metadata::key_dupe", "A").unwrap();
        table_config.set("format.metadata::key_dupe", "B").unwrap();
        let parsed_metadata = table_config.parquet.key_value_metadata;
        assert_eq!(parsed_metadata.get("key_dupe"), Some(&Some("B".into())));
    }
    #[cfg(feature = "parquet")]
    #[test]
    fn test_parquet_writer_version_validation() {
        use crate::{config::ConfigOptions, parquet_config::DFParquetWriterVersion};

        let mut config = ConfigOptions::default();

        // Valid values should work
        config
            .set("datafusion.execution.parquet.writer_version", "1.0")
            .unwrap();
        assert_eq!(
            config.execution.parquet.writer_version,
            DFParquetWriterVersion::V1_0
        );

        config
            .set("datafusion.execution.parquet.writer_version", "2.0")
            .unwrap();
        assert_eq!(
            config.execution.parquet.writer_version,
            DFParquetWriterVersion::V2_0
        );

        // Invalid value should error immediately at SET time
        let err = config
            .set("datafusion.execution.parquet.writer_version", "3.0")
            .unwrap_err();
        assert_contains!(
            err.to_string(),
            "Invalid or Unsupported Configuration: Invalid parquet writer version: 3.0. Expected one of: 1.0, 2.0"
        );
    }

    #[cfg(feature = "parquet")]
    #[test]
    fn set_cdc_enabled_flag() {
        use crate::config::ConfigOptions;

        let mut config = ConfigOptions::default();
        // CDC is disabled by default.
        assert!(!config.execution.parquet.content_defined_chunking.enabled);

        // `.enabled = true` enables CDC; parameters keep their defaults.
        config
            .set(
                "datafusion.execution.parquet.content_defined_chunking.enabled",
                "true",
            )
            .unwrap();
        let cdc = &config.execution.parquet.content_defined_chunking;
        assert!(cdc.enabled);
        assert_eq!(cdc.min_chunk_size, 256 * 1024);
        assert_eq!(cdc.max_chunk_size, 1024 * 1024);
        assert_eq!(cdc.norm_level, 0);

        // `.enabled = false` disables CDC.
        config
            .set(
                "datafusion.execution.parquet.content_defined_chunking.enabled",
                "false",
            )
            .unwrap();
        assert!(!config.execution.parquet.content_defined_chunking.enabled);
    }

    #[cfg(feature = "parquet")]
    #[test]
    fn set_cdc_param_does_not_enable() {
        use crate::config::ConfigOptions;

        let mut config = ConfigOptions::default();

        // Setting a parameter does NOT enable CDC (`enabled` is a distinct field,
        // defaulting to false), and the result is independent of key order.
        config
            .set(
                "datafusion.execution.parquet.content_defined_chunking.min_chunk_size",
                "1024",
            )
            .unwrap();
        let cdc = &config.execution.parquet.content_defined_chunking;
        assert!(!cdc.enabled);
        assert_eq!(cdc.min_chunk_size, 1024);
        assert_eq!(cdc.max_chunk_size, 1024 * 1024);
        assert_eq!(cdc.norm_level, 0);
    }

    #[test]
    fn test_dialect_metadata_roundtrip() {
        use crate::config::Dialect;
        use std::str::FromStr;

        assert_eq!(Dialect::default(), Dialect::Generic);
        assert!(!Dialect::metadata().is_empty());

        for info in Dialect::metadata() {
            let dialect = info.dialect;

            assert_eq!(Dialect::from_str(info.canonical_name).unwrap(), dialect);
            assert_eq!(
                Dialect::from_str(&info.canonical_name.to_ascii_uppercase()).unwrap(),
                dialect
            );
            assert_eq!(dialect.as_ref(), info.canonical_name);
            assert_eq!(dialect.to_string(), info.canonical_name);
        }
    }

    #[test]
    fn test_dialect_aliases() {
        use crate::config::Dialect;
        use std::str::FromStr;

        for info in Dialect::metadata() {
            for alias in info.aliases {
                assert_eq!(Dialect::from_str(alias).unwrap(), info.dialect);
                assert_eq!(
                    Dialect::from_str(&alias.to_ascii_uppercase()).unwrap(),
                    info.dialect
                );
            }
        }
    }

    #[test]
    fn test_available_dialects_includes_each_display_name_once() {
        use crate::config::Dialect;
        use std::collections::BTreeSet;

        let available = Dialect::available();
        let listed: Vec<_> = available.split(", ").collect();
        let display_names: Vec<_> = Dialect::metadata()
            .iter()
            .map(|info| info.display_name)
            .collect();
        let unique_display_names: BTreeSet<_> = display_names.iter().copied().collect();

        assert_eq!(display_names.len(), unique_display_names.len());
        assert_eq!(listed, display_names);
    }

    #[test]
    fn test_dialect_config_description_uses_metadata() {
        use crate::config::{ConfigOptions, Dialect, SQL_PARSER_DIALECT_CONFIG_KEY};

        let description = ConfigOptions::default()
            .entries()
            .into_iter()
            .find(|entry| entry.key == SQL_PARSER_DIALECT_CONFIG_KEY)
            .unwrap()
            .description;

        assert!(description.contains(Dialect::available()));
    }

    #[test]
    fn test_invalid_dialect_error_lists_available_dialects() {
        use crate::config::Dialect;
        use std::str::FromStr;

        let error = Dialect::from_str("notadialect").unwrap_err().to_string();

        assert!(error.contains("Invalid Dialect: notadialect"));
        assert!(error.contains(Dialect::available()));
    }

    #[test]
    fn max_row_group_bytes_rejects_zero() {
        use crate::config::MaxRowGroupBytes;
        use std::str::FromStr;

        assert!(MaxRowGroupBytes::try_new(0).is_err());
        assert!(MaxRowGroupBytes::from_str("0").is_err());
        assert!(MaxRowGroupBytes::from_str("not_a_number").is_err());
        assert_eq!(MaxRowGroupBytes::try_new(128).unwrap().get(), 128);
        assert_eq!(MaxRowGroupBytes::from_str("128").unwrap().get(), 128);
    }

    #[test]
    fn parquet_max_row_group_bytes_config_set_rejects_zero() {
        use crate::config::ConfigOptions;

        let mut options = ConfigOptions::new();
        options
            .set("datafusion.execution.parquet.max_row_group_bytes", "1024")
            .unwrap();
        assert_eq!(
            options
                .execution
                .parquet
                .max_row_group_bytes
                .map(|v| v.get()),
            Some(1024)
        );

        // Zero is rejected at set time, leaving the previous value unchanged.
        assert!(
            options
                .set("datafusion.execution.parquet.max_row_group_bytes", "0")
                .is_err()
        );
        assert_eq!(
            options
                .execution
                .parquet
                .max_row_group_bytes
                .map(|v| v.get()),
            Some(1024)
        );
    }
}