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 crate::error::_config_err;
use crate::parsers::CompressionTypeVariant;
use crate::utils::get_available_parallelism;
use crate::{DataFusionError, Result};
use std::any::Any;
use std::collections::{BTreeMap, HashMap};
use std::error::Error;
use std::fmt::{self, Display};
use std::str::FromStr;

/// 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
                                #[allow(deprecated)]
                                let ret = self.$field_name.set(rem, value.as_ref());

                                $(if !$warn.is_empty() {
                                    let default: $field_type = $default;
                                    #[allow(deprecated)]
                                    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();
                    #[allow(deprecated)]
                    self.$field_name.visit(v, key.as_str(), desc);
                )*
            }
        }
        impl Default for $struct_name {
            fn default() -> Self {
                #[allow(deprecated)]
                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; supported values include: Generic,
        /// MySQL, PostgreSQL, Hive, SQLite, Snowflake, Redshift, MsSQL, ClickHouse, BigQuery, Ansi, DuckDB and Databricks.
        pub dialect: String, default = "generic".to_string()
        // 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, `VARCHAR` is mapped to `Utf8View` during SQL planning.
       /// If false, `VARCHAR` is mapped to `Utf8`  during SQL planning.
       /// Default is false.
        pub map_varchar_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: usize, default = 50
    }
}

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: usize, default = 8192

        /// 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. Applies to the default
        /// `ListingTableProvider` in DataFusion. 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. `EXTRACT(HOUR from SOME_TIME)`, shift the underlying datetime
        /// according to this time zone, and then extract the hour
        pub time_zone: Option<String>, default = Some("+00:00".into())

        /// 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

        /// 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

        /// Number of files to read in parallel when inferring schema and statistics
        pub meta_fetch_concurrency: 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: 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: 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
        pub max_buffered_batches_per_output_file: usize, default = 2

        /// Should sub directories be ignored when scanning directories for data
        /// files. Defaults to true (ignores subdirectories), consistent with