Updates.

Author: Omega359

Cargo.lock

@@ -179,8 +179,8 @@ fn simplify_demo() -> Result<()> {
     // expressions, such as the current time (to evaluate `now()`
     // correctly)
     let props = ExecutionProps::new();
-    let config_options = ConfigOptions::default();
-    let context = SimplifyContext::new(&props, &config_options).with_schema(schema);
+    let config_options = ConfigOptions::default_singleton_arc();
+    let context = SimplifyContext::new(&props, config_options).with_schema(schema);
     let simplifier = ExprSimplifier::new(context);
 
     // And then call the simplify_expr function:
@@ -196,7 +196,7 @@ fn simplify_demo() -> Result<()> {
     // here are some other examples of what DataFusion is capable of
     let schema = Schema::new(vec![make_field("i", DataType::Int64)]).to_dfschema_ref()?;
     let context =
-        SimplifyContext::new(&props, &config_options).with_schema(schema.clone());
+        SimplifyContext::new(&props, config_options).with_schema(schema.clone());
     let simplifier = ExprSimplifier::new(context);
 
     // basic arithmetic simplification
@@ -534,8 +534,8 @@ fn type_coercion_demo() -> Result<()> {
 
     // Evaluation with an expression that has not been type coerced cannot succeed.
     let props = ExecutionProps::default();
-    let config_options = ConfigOptions::default();
-    let physical_expr = create_physical_expr(&expr, &df_schema, &props, &config_options)?;
+    let config_options = ConfigOptions::default_singleton_arc();
+    let physical_expr = create_physical_expr(&expr, &df_schema, &props, config_options)?;
     let e = physical_expr.evaluate(&batch).unwrap_err();
     assert!(e
         .find_root()
@@ -548,12 +548,12 @@ fn type_coercion_demo() -> Result<()> {
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     // 2. Type coercion with `ExprSimplifier::coerce`.
-    let context = SimplifyContext::new(&props, &config_options)
+    let context = SimplifyContext::new(&props, config_options)
         .with_schema(Arc::new(df_schema.clone()));
     let simplifier = ExprSimplifier::new(context);
     let coerced_expr = simplifier.coerce(expr.clone(), &df_schema)?;
     let physical_expr =
-        create_physical_expr(&coerced_expr, &df_schema, &props, &config_options)?;
+        create_physical_expr(&coerced_expr, &df_schema, &props, config_options)?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     // 3. Type coercion with `TypeCoercionRewriter`.
@@ -562,7 +562,7 @@ fn type_coercion_demo() -> Result<()> {
         .rewrite(&mut TypeCoercionRewriter::new(&df_schema))?
         .data;
     let physical_expr =
-        create_physical_expr(&coerced_expr, &df_schema, &props, &config_options)?;
+        create_physical_expr(&coerced_expr, &df_schema, &props, config_options)?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     // 4. Apply explicit type coercion by manually rewriting the expression
@@ -587,7 +587,7 @@ fn type_coercion_demo() -> Result<()> {
         })?
         .data;
     let physical_expr =
-        create_physical_expr(&coerced_expr, &df_schema, &props, &config_options)?;
+        create_physical_expr(&coerced_expr, &df_schema, &props, config_options)?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     Ok(())

datafusion-examples/examples/expr_api.rs

@@ -16,6 +16,7 @@
 // under the License.
 
 use datafusion::error::Result;
+use datafusion::execution::session_state::SessionStateOptimizerConfig;
 use datafusion::logical_expr::{LogicalPlan, PlanType};
 use datafusion::physical_plan::{displayable, DisplayFormatType};
 use datafusion::physical_planner::DefaultPhysicalPlanner;
@@ -97,17 +98,19 @@ async fn to_physical_plan_step_by_step_demo(
     ctx: &SessionContext,
 ) -> Result<()> {
     // First analyze the logical plan
-    let analyzed_logical_plan = ctx.state().analyzer().execute_and_check(
+    let session_state = ctx.state();
+    let analyzed_logical_plan = session_state.analyzer().execute_and_check(
         input,
-        ctx.state().config_options(),
+        session_state.config_options(),
         |_, _| (),
     )?;
     println!("Analyzed logical plan:\n\n{:?}\n\n", analyzed_logical_plan);
 
     // Optimize the analyzed logical plan
-    let optimized_logical_plan = ctx.state().optimizer().optimize(
+    let session_optimizer_config = SessionStateOptimizerConfig::new(&session_state);
+    let optimized_logical_plan = session_state.optimizer().optimize(
         analyzed_logical_plan,
-        &ctx.state(),
+        &session_optimizer_config,
         |_, _| (),
     )?;
     println!(
@@ -116,10 +119,9 @@ async fn to_physical_plan_step_by_step_demo(
     );
 
     // Create the physical plan
-    let physical_plan = ctx
-        .state()
+    let physical_plan = session_state
         .query_planner()
-        .create_physical_plan(&optimized_logical_plan, &ctx.state())
+        .create_physical_plan(&optimized_logical_plan, &session_state)
         .await?;
     println!(
         "Final physical plan:\n\n{}\n\n",
@@ -139,7 +141,7 @@ async fn to_physical_plan_step_by_step_demo(
     // on DefaultPhysicalPlanner. Not all planners will provide this feature.
     let planner = DefaultPhysicalPlanner::default();
     let physical_plan =
-        planner.optimize_physical_plan(physical_plan, &ctx.state(), |_, _| {})?;
+        planner.optimize_physical_plan(physical_plan, &session_state, |_, _| {})?;
     println!(
         "Optimized physical plan:\n\n{}\n\n",
         displayable(physical_plan.as_ref())

datafusion-examples/examples/planner_api.rs

@@ -189,9 +189,9 @@ impl PruningStatistics for MyCatalog {
 fn create_pruning_predicate(expr: Expr, schema: &SchemaRef) -> PruningPredicate {
     let df_schema = DFSchema::try_from(schema.as_ref().clone()).unwrap();
     let props = ExecutionProps::new();
-    let config_options = ConfigOptions::default();
+    let config_options = ConfigOptions::default_singleton_arc();
     let physical_expr =
-        create_physical_expr(&expr, &df_schema, &props, &config_options).unwrap();
+        create_physical_expr(&expr, &df_schema, &props, config_options).unwrap();
     PruningPredicate::try_new(physical_expr, schema.clone()).unwrap()
 }
 

datafusion-examples/examples/pruning.rs

@@ -147,7 +147,7 @@ impl TableFunctionImpl for LocalCsvTableFunc {
             .map(|expr| {
                 // try to simplify the expression, so 1+2 becomes 3, for example
                 let execution_props = ExecutionProps::new();
-                let config_options = Arc::unwrap_or_clone(self.config_options.clone());
+                let config_options = Arc::clone(&self.config_options);
                 let info = SimplifyContext::new(&execution_props, &config_options);
                 let expr = ExprSimplifier::new(info).simplify(expr.clone())?;
 

datafusion-examples/examples/simple_udtf.rs

@@ -38,9 +38,9 @@ use futures::stream::FuturesUnordered;
 use futures::{stream::BoxStream, StreamExt, TryStreamExt};
 use log::{debug, trace};
 
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{TreeNode, TreeNodeRecursion};
 use datafusion_common::{Column, DFSchema, DataFusionError};
-use datafusion_execution::config::SessionConfig;
 use datafusion_expr::execution_props::ExecutionProps;
 use datafusion_expr::{Expr, Volatility};
 use datafusion_physical_expr::create_physical_expr;
@@ -244,7 +244,7 @@ async fn prune_partitions(
     partitions: Vec<Partition>,
     filters: &[Expr],
     partition_cols: &[(String, DataType)],
-    config: &SessionConfig,
+    config_options: &Arc<ConfigOptions>,
 ) -> Result<Vec<Partition>> {
     if filters.is_empty() {
         return Ok(partitions);
@@ -291,7 +291,6 @@ async fn prune_partitions(
 
     let batch = RecordBatch::try_new(schema, arrays)?;
     let props = ExecutionProps::new();
-    let config_options = config.options();
 
     // Applies `filter` to `batch` returning `None` on error
     let do_filter = |filter| -> Result<ArrayRef> {
@@ -414,6 +413,7 @@ pub async fn pruned_partition_list<'a>(
     filters: &'a [Expr],
     file_extension: &'a str,
     partition_cols: &'a [(String, DataType)],
+    config_options: &Arc<ConfigOptions>,
 ) -> Result<BoxStream<'a, Result<PartitionedFile>>> {
     // if no partition col => simply list all the files
     if partition_cols.is_empty() {
@@ -443,7 +443,7 @@ pub async fn pruned_partition_list<'a>(
         partitions,
         filters,
         partition_cols,
-        ctx.config(),
+        config_options,
     )
     .await?;
 
@@ -613,6 +613,7 @@ mod tests {
             &[filter],
             ".parquet",
             &[(String::from("mypartition"), DataType::Utf8)],
+            &Arc::clone(ConfigOptions::default_singleton_arc()),
         )
         .await
         .expect("partition pruning failed")
@@ -638,6 +639,7 @@ mod tests {
             &[filter],
             ".parquet",
             &[(String::from("mypartition"), DataType::Utf8)],
+            &Arc::clone(ConfigOptions::default_singleton_arc()),
         )
         .await
         .expect("partition pruning failed")
@@ -681,6 +683,7 @@ mod tests {
                 (String::from("part1"), DataType::Utf8),
                 (String::from("part2"), DataType::Utf8),
             ],
+            &Arc::clone(ConfigOptions::default_singleton_arc()),
         )
         .await
         .expect("partition pruning failed")

datafusion/catalog-listing/src/helpers.rs

@@ -17,16 +17,16 @@
 
 //! 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;
-
-use crate::error::_config_err;
-use crate::parsers::CompressionTypeVariant;
-use crate::utils::get_available_parallelism;
-use crate::{DataFusionError, Result};
+use std::sync::{Arc, LazyLock};
 
 /// A macro that wraps a configuration struct and automatically derives
 /// [`Default`] and [`ConfigField`] for it, allowing it to be used
@@ -772,7 +772,20 @@ impl ConfigField for ConfigOptions {
     }
 }
 
+static CONFIG_OPTIONS_SINGLETON: LazyLock<Arc<ConfigOptions>> =
+    LazyLock::new(|| Arc::new(ConfigOptions::default()));
+
 impl ConfigOptions {
+    /// this is a static singleton to be used for testing only where the default values are sufficient
+    pub fn default_singleton() -> &'static ConfigOptions {
+        CONFIG_OPTIONS_SINGLETON.as_ref()
+    }
+
+    /// this is a static singleton to be used for testing only where the default values are sufficient
+    pub fn default_singleton_arc() -> &'static Arc<ConfigOptions> {
+        &CONFIG_OPTIONS_SINGLETON
+    }
+
     /// Creates a new [`ConfigOptions`] with default values
     pub fn new() -> Self {
         Self::default()

datafusion/common/src/config.rs

@@ -891,6 +891,8 @@ impl TableProvider for ListingTable {
         }
 
         let output_ordering = self.try_create_output_ordering()?;
+        let config_options = Arc::new(session_state.config_options().clone());
+
         match state
             .config_options()
             .execution
@@ -924,7 +926,7 @@ impl TableProvider for ListingTable {
                     &expr,
                     &table_df_schema,
                     state.execution_props(),
-                    session_state.config_options(),
+                    &config_options,
                 )?;
                 Some(filters)
             }
@@ -1022,13 +1024,15 @@ impl TableProvider for ListingTable {
 
         // TODO (https://github.com/apache/datafusion/issues/11600) remove downcast_ref from here?
         let session_state = state.as_any().downcast_ref::<SessionState>().unwrap();
+        let config_options = Arc::new(state.config_options().clone());
         let file_list_stream = pruned_partition_list(
             session_state,
             store.as_ref(),
             table_path,
             &[],
             &self.options.file_extension,
             &self.options.table_partition_cols,
+            &config_options,
         )
         .await?;
 
@@ -1095,6 +1099,7 @@ impl ListingTable {
             return Ok((vec![], Statistics::new_unknown(&self.file_schema)));
         };
         // list files (with partitions)
+        let config_options = Arc::new(ctx.config_options().clone());
         let file_list = future::try_join_all(self.table_paths.iter().map(|table_path| {
             pruned_partition_list(
                 ctx,
@@ -1103,6 +1108,7 @@ impl ListingTable {
                 filters,
                 &self.options.file_extension,
                 &self.options.table_partition_cols,
+                &config_options,
             )
         }))
         .await?;

datafusion/core/src/datasource/listing/table.rs

@@ -236,6 +236,8 @@ impl TableProvider for MemTable {
 
         // add sort information if present
         let sort_order = self.sort_order.lock();
+        let config_options = Arc::new(state.config_options().clone());
+
         if !sort_order.is_empty() {
             let df_schema = DFSchema::try_from(self.schema.as_ref().clone())?;
 
@@ -246,7 +248,7 @@ impl TableProvider for MemTable {
                         sort_exprs,
                         &df_schema,
                         state.execution_props(),
-                        state.config_options(),
+                        &config_options,
                     )
                 })
                 .collect::<Result<Vec<_>>>()?;

datafusion/core/src/datasource/memory.rs

@@ -1829,6 +1829,7 @@ mod tests {
     use crate::execution::session_state::SessionStateBuilder;
     use crate::physical_planner::PhysicalPlanner;
     use async_trait::async_trait;
+    use datafusion_common::config::ConfigOptions;
     use datafusion_expr::planner::TypePlanner;
     use sqlparser::ast;
     use tempfile::TempDir;
@@ -2281,6 +2282,7 @@ mod tests {
             _expr: &Expr,
             _input_dfschema: &DFSchema,
             _session_state: &SessionState,
+            _config_options: &Arc<ConfigOptions>,
         ) -> Result<Arc<dyn PhysicalExpr>> {
             unimplemented!()
         }