Add SessionConfig reference to ScalarFunctionArgs

Author: Omega359

datafusion-examples/examples/composed_extension_codec.rs

@@ -71,8 +71,10 @@ async fn main() {
 
     // deserialize proto back to execution plan
     let runtime = ctx.runtime_env();
+    let state = ctx.state();
+    let config_options = state.config_options();
     let result_exec_plan: Arc<dyn ExecutionPlan> = proto
-        .try_into_physical_plan(&ctx, runtime.deref(), &composed_codec)
+        .try_into_physical_plan(&ctx, config_options, runtime.deref(), &composed_codec)
         .expect("from proto");
 
     // assert that the original and deserialized execution plans are equal

datafusion-examples/examples/expr_api.rs

@@ -26,6 +26,7 @@ use datafusion::common::stats::Precision;
 use datafusion::common::tree_node::{Transformed, TreeNode};
 use datafusion::common::{ColumnStatistics, DFSchema};
 use datafusion::common::{ScalarValue, ToDFSchema};
+use datafusion::config::ConfigOptions;
 use datafusion::error::Result;
 use datafusion::functions_aggregate::first_last::first_value_udaf;
 use datafusion::logical_expr::execution_props::ExecutionProps;
@@ -35,7 +36,9 @@ use datafusion::logical_expr::simplify::SimplifyContext;
 use datafusion::logical_expr::{ColumnarValue, ExprFunctionExt, ExprSchemable, Operator};
 use datafusion::optimizer::analyzer::type_coercion::TypeCoercionRewriter;
 use datafusion::optimizer::simplify_expressions::ExprSimplifier;
-use datafusion::physical_expr::{analyze, AnalysisContext, ExprBoundaries};
+use datafusion::physical_expr::{
+    analyze, create_physical_expr, AnalysisContext, ExprBoundaries,
+};
 use datafusion::prelude::*;
 
 /// This example demonstrates the DataFusion [`Expr`] API.
@@ -176,7 +179,8 @@ fn simplify_demo() -> Result<()> {
     // expressions, such as the current time (to evaluate `now()`
     // correctly)
     let props = ExecutionProps::new();
-    let context = SimplifyContext::new(&props).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:
@@ -191,7 +195,8 @@ 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).with_schema(schema.clone());
+    let context =
+        SimplifyContext::new(&props, config_options).with_schema(schema.clone());
     let simplifier = ExprSimplifier::new(context);
 
     // basic arithmetic simplification
@@ -529,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 physical_expr =
-        datafusion::physical_expr::create_physical_expr(&expr, &df_schema, &props)?;
+    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()
@@ -543,26 +548,21 @@ fn type_coercion_demo() -> Result<()> {
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     // 2. Type coercion with `ExprSimplifier::coerce`.
-    let context = SimplifyContext::new(&props).with_schema(Arc::new(df_schema.clone()));
+    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 = datafusion::physical_expr::create_physical_expr(
-        &coerced_expr,
-        &df_schema,
-        &props,
-    )?;
+    let physical_expr =
+        create_physical_expr(&coerced_expr, &df_schema, &props, config_options)?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     // 3. Type coercion with `TypeCoercionRewriter`.
     let coerced_expr = expr
         .clone()
         .rewrite(&mut TypeCoercionRewriter::new(&df_schema))?
         .data;
-    let physical_expr = datafusion::physical_expr::create_physical_expr(
-        &coerced_expr,
-        &df_schema,
-        &props,
-    )?;
+    let physical_expr =
+        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
@@ -586,11 +586,8 @@ fn type_coercion_demo() -> Result<()> {
             }
         })?
         .data;
-    let physical_expr = datafusion::physical_expr::create_physical_expr(
-        &coerced_expr,
-        &df_schema,
-        &props,
-    )?;
+    let physical_expr =
+        create_physical_expr(&coerced_expr, &df_schema, &props, config_options)?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 
     Ok(())

datafusion-examples/examples/planner_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/pruning.rs

@@ -20,6 +20,7 @@ use std::sync::Arc;
 
 use arrow::array::{ArrayRef, BooleanArray, Int32Array};
 use arrow::datatypes::{DataType, Field, Schema, SchemaRef};
+use datafusion::common::config::ConfigOptions;
 use datafusion::common::{DFSchema, ScalarValue};
 use datafusion::execution::context::ExecutionProps;
 use datafusion::physical_expr::create_physical_expr;
@@ -188,7 +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 physical_expr = create_physical_expr(&expr, &df_schema, &props).unwrap();
+    let config_options = ConfigOptions::default_singleton_arc();
+    let physical_expr =
+        create_physical_expr(&expr, &df_schema, &props, config_options).unwrap();
     PruningPredicate::try_new(physical_expr, schema.clone()).unwrap()
 }
 

datafusion-examples/examples/simple_udtf.rs

@@ -23,6 +23,7 @@ use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::catalog::Session;
 use datafusion::catalog::TableFunctionImpl;
 use datafusion::common::{plan_err, ScalarValue};
+use datafusion::config::ConfigOptions;
 use datafusion::datasource::memory::MemorySourceConfig;
 use datafusion::datasource::TableProvider;
 use datafusion::error::Result;
@@ -142,7 +143,8 @@ impl TableFunctionImpl for LocalCsvTableFunc {
             .map(|expr| {
                 // try to simplify the expression, so 1+2 becomes 3, for example
                 let execution_props = ExecutionProps::new();
-                let info = SimplifyContext::new(&execution_props);
+                let config_options = ConfigOptions::default_singleton_arc();
+                let info = SimplifyContext::new(&execution_props, config_options);
                 let expr = ExprSimplifier::new(info).simplify(expr.clone())?;
 
                 if let Expr::Literal(ScalarValue::Int64(Some(limit))) = expr {

datafusion/catalog-listing/src/helpers.rs

@@ -33,13 +33,15 @@ use arrow::{
     datatypes::{DataType, Field, Fields, Schema},
     record_batch::RecordBatch,
 };
-use datafusion_expr::execution_props::ExecutionProps;
+
 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_expr::execution_props::ExecutionProps;
 use datafusion_expr::{Expr, Volatility};
 use datafusion_physical_expr::create_physical_expr;
 use object_store::path::Path;
@@ -242,6 +244,7 @@ async fn prune_partitions(
     partitions: Vec<Partition>,
     filters: &[Expr],
     partition_cols: &[(String, DataType)],
+    config_options: &Arc<ConfigOptions>,
 ) -> Result<Vec<Partition>> {
     if filters.is_empty() {
         return Ok(partitions);
@@ -293,7 +296,7 @@ async fn prune_partitions(
 
     // Applies `filter` to `batch` returning `None` on error
     let do_filter = |filter| -> Result<ArrayRef> {
-        let expr = create_physical_expr(filter, &df_schema, &props)?;
+        let expr = create_physical_expr(filter, &df_schema, &props, config_options)?;
         expr.evaluate(&batch)?.into_array(partitions.len())
     };
 
@@ -412,6 +415,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() {
@@ -436,8 +440,14 @@ pub async fn pruned_partition_list<'a>(
             .await?;
     debug!("Listed {} partitions", partitions.len());
 
-    let pruned =
-        prune_partitions(table_path, partitions, filters, partition_cols).await?;
+    let pruned = prune_partitions(
+        table_path,
+        partitions,
+        filters,
+        partition_cols,
+        config_options,
+    )
+    .await?;
 
     debug!("Pruning yielded {} partitions", pruned.len());
 
@@ -605,6 +615,7 @@ mod tests {
             &[filter],
             ".parquet",
             &[(String::from("mypartition"), DataType::Utf8)],
+            &Arc::clone(ConfigOptions::default_singleton_arc()),
         )
         .await
         .expect("partition pruning failed")
@@ -630,6 +641,7 @@ mod tests {
             &[filter],
             ".parquet",
             &[(String::from("mypartition"), DataType::Utf8)],
+            &Arc::clone(ConfigOptions::default_singleton_arc()),
         )
         .await
         .expect("partition pruning failed")
@@ -673,6 +685,7 @@ mod tests {
                 (String::from("part1"), DataType::Utf8),
                 (String::from("part2"), DataType::Utf8),
             ],
+            &Arc::clone(ConfigOptions::default_singleton_arc()),
         )
         .await
         .expect("partition pruning failed")
@@ -1016,10 +1029,17 @@ mod tests {
                 .unwrap();
         }
 
-        (Arc::new(memory), Arc::new(MockSession {}))
+        (
+            Arc::new(memory),
+            Arc::new(MockSession {
+                config: SessionConfig::new(),
+            }),
+        )
     }
 
-    struct MockSession {}
+    struct MockSession {
+        config: SessionConfig,
+    }
 
     #[async_trait]
     impl Session for MockSession {
@@ -1028,7 +1048,7 @@ mod tests {
         }
 
         fn config(&self) -> &SessionConfig {
-            unimplemented!()
+            &self.config
         }
 
         async fn create_physical_plan(

datafusion/common/src/config.rs

@@ -22,6 +22,7 @@ use std::collections::{BTreeMap, HashMap};
 use std::error::Error;
 use std::fmt::{self, Display};
 use std::str::FromStr;
+use std::sync::{Arc, LazyLock};
 
 use crate::error::_config_err;
 use crate::parsers::CompressionTypeVariant;
@@ -724,7 +725,7 @@ config_namespace! {
 }
 
 /// A key value pair, with a corresponding description
-#[derive(Debug)]
+#[derive(Debug, Hash, PartialEq, Eq)]
 pub struct ConfigEntry {
     /// A unique string to identify this config value
     pub key: String,
@@ -777,7 +778,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/core/Cargo.toml

@@ -144,7 +144,7 @@ zstd = { version = "0.13", optional = true, default-features = false }
 
 [dev-dependencies]
 async-trait = { workspace = true }
-criterion = { workspace = true, features = ["async_tokio"] }
+criterion = { workspace = true, features = ["async_tokio", "async_futures"] }
 ctor = { workspace = true }
 dashmap = "6.1.0"
 datafusion-doc = { workspace = true }