feat: Add Arc<ConfigOptions> to ScalarFunctionArgs, don't copy ConfigOptions on each query (#16970)

* Add `ConfigOptions` to ExecutionProps when execution is started

* Add ConfigOptions to ScalarFunctionArgs, refactor AsyncScalarUDF.invoke_async_with_args to remove ConfigOptions arg.

* Updated OptimizerConfig.options() -> Arc<ConfigOptions> to eliminate clone() calls. Fixed an issue with SimplifyExpressions.rewrite(..) not adding config options to execution_props. Added test to verify it works

* Test update.

* Add note in upgrade guide

* Use Arc all the way down

* start_execution -> mark_start_execution

* Update datafusion/expr/src/execution_props.rs

Co-authored-by: Andrew Lamb <[email protected]>

* Update comments

* Avoid API breakage via #deprecated

* Update upgrade guide for Arc<ConfigOptions> change

* Apply suggestions from code review

* fmt

---------

Co-authored-by: Andrew Lamb <[email protected]>

Author: Omega359

datafusion-examples/examples/async_udf.rs

@@ -29,7 +29,6 @@ use datafusion::common::cast::as_string_view_array;
 use datafusion::common::error::Result;
 use datafusion::common::not_impl_err;
 use datafusion::common::utils::take_function_args;
-use datafusion::config::ConfigOptions;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::logical_expr::async_udf::{AsyncScalarUDF, AsyncScalarUDFImpl};
 use datafusion::logical_expr::{
@@ -195,11 +194,7 @@ impl AsyncScalarUDFImpl for AskLLM {
     /// is processing the query, so you may wish to make actual network requests
     /// on a different `Runtime`, as explained in the `thread_pools.rs` example
     /// in this directory.
-    async fn invoke_async_with_args(
-        &self,
-        args: ScalarFunctionArgs,
-        _option: &ConfigOptions,
-    ) -> Result<ArrayRef> {
+    async fn invoke_async_with_args(&self, args: ScalarFunctionArgs) -> Result<ArrayRef> {
         // in a real UDF you would likely want to special case constant
         // arguments to improve performance, but this example converts the
         // arguments to arrays for simplicity.

datafusion-examples/examples/sql_frontend.rs

@@ -83,7 +83,7 @@ pub fn main() -> Result<()> {
     let config = OptimizerContext::default().with_skip_failing_rules(false);
     let analyzed_plan = Analyzer::new().execute_and_check(
         logical_plan,
-        config.options(),
+        &config.options(),
         observe_analyzer,
     )?;
     // Note that the Analyzer has added a CAST to the plan to align the types

datafusion/common/src/config.rs

@@ -923,7 +923,7 @@ impl<'a> TryInto<arrow::util::display::FormatOptions<'a>> for &'a FormatOptions
 }
 
 /// 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,

datafusion/core/src/execution/context/mod.rs

@@ -1646,7 +1646,7 @@ impl SessionContext {
     /// [`ConfigOptions`]: crate::config::ConfigOptions
     pub fn state(&self) -> SessionState {
         let mut state = self.state.read().clone();
-        state.execution_props_mut().start_execution();
+        state.mark_start_execution();
         state
     }
 

datafusion/core/src/execution/session_state.rs

@@ -574,7 +574,7 @@ impl SessionState {
             // analyze & capture output of each rule
             let analyzer_result = self.analyzer.execute_and_check(
                 e.plan.as_ref().clone(),
-                self.options(),
+                &self.options(),
                 |analyzed_plan, analyzer| {
                     let analyzer_name = analyzer.name().to_string();
                     let plan_type = PlanType::AnalyzedLogicalPlan { analyzer_name };
@@ -636,7 +636,7 @@ impl SessionState {
         } else {
             let analyzed_plan = self.analyzer.execute_and_check(
                 plan.clone(),
-                self.options(),
+                &self.options(),
                 |_, _| {},
             )?;
             self.optimizer.optimize(analyzed_plan, self, |_, _| {})
@@ -738,10 +738,16 @@ impl SessionState {
     }
 
     /// return the configuration options
-    pub fn config_options(&self) -> &ConfigOptions {
+    pub fn config_options(&self) -> &Arc<ConfigOptions> {
         self.config.options()
     }
 
+    /// Mark the start of the execution
+    pub fn mark_start_execution(&mut self) {
+        let config = Arc::clone(self.config.options());
+        self.execution_props.mark_start_execution(config);
+    }
+
     /// Return the table options
     pub fn table_options(&self) -> &TableOptions {
         &self.table_options
@@ -1891,8 +1897,8 @@ impl OptimizerConfig for SessionState {
         &self.execution_props.alias_generator
     }
 
-    fn options(&self) -> &ConfigOptions {
-        self.config_options()
+    fn options(&self) -> Arc<ConfigOptions> {
+        Arc::clone(self.config.options())
     }
 
     fn function_registry(&self) -> Option<&dyn FunctionRegistry> {

datafusion/core/tests/fuzz_cases/equivalence/ordering.rs

@@ -21,6 +21,7 @@ use crate::fuzz_cases::equivalence::utils::{
     is_table_same_after_sort, TestScalarUDF,
 };
 use arrow::compute::SortOptions;
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::Result;
 use datafusion_expr::{Operator, ScalarUDF};
 use datafusion_physical_expr::equivalence::{
@@ -110,6 +111,7 @@ fn test_ordering_satisfy_with_equivalence_complex_random() -> Result<()> {
             Arc::clone(&test_fun),
             vec![col_a],
             &test_schema,
+            Arc::new(ConfigOptions::default()),
         )?);
         let a_plus_b = Arc::new(BinaryExpr::new(
             col("a", &test_schema)?,

datafusion/core/tests/fuzz_cases/equivalence/projection.rs

@@ -20,6 +20,7 @@ use crate::fuzz_cases::equivalence::utils::{
     is_table_same_after_sort, TestScalarUDF,
 };
 use arrow::compute::SortOptions;
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::Result;
 use datafusion_expr::{Operator, ScalarUDF};
 use datafusion_physical_expr::equivalence::ProjectionMapping;
@@ -49,6 +50,7 @@ fn project_orderings_random() -> Result<()> {
             Arc::clone(&test_fun),
             vec![col_a],
             &test_schema,
+            Arc::new(ConfigOptions::default()),
         )?);
         // a + b
         let a_plus_b = Arc::new(BinaryExpr::new(
@@ -122,6 +124,7 @@ fn ordering_satisfy_after_projection_random() -> Result<()> {
             Arc::clone(&test_fun),
             vec![col_a],
             &test_schema,
+            Arc::new(ConfigOptions::default()),
         )?) as PhysicalExprRef;
         // a + b
         let a_plus_b = Arc::new(BinaryExpr::new(

datafusion/core/tests/fuzz_cases/equivalence/properties.rs

@@ -28,6 +28,7 @@ use datafusion_physical_expr::expressions::{col, BinaryExpr};
 use datafusion_physical_expr::{LexOrdering, ScalarFunctionExpr};
 use datafusion_physical_expr_common::sort_expr::PhysicalSortExpr;
 
+use datafusion_common::config::ConfigOptions;
 use itertools::Itertools;
 
 #[test]
@@ -49,6 +50,7 @@ fn test_find_longest_permutation_random() -> Result<()> {
             Arc::clone(&test_fun),
             vec![col_a],
             &test_schema,
+            Arc::new(ConfigOptions::default()),
         )?) as _;
 
         let a_plus_b = Arc::new(BinaryExpr::new(

datafusion/core/tests/optimizer/mod.rs

@@ -160,7 +160,7 @@ fn test_sql(sql: &str) -> Result<LogicalPlan> {
     let analyzer = Analyzer::new();
     let optimizer = Optimizer::new();
     // analyze and optimize the logical plan
-    let plan = analyzer.execute_and_check(plan, config.options(), |_, _| {})?;
+    let plan = analyzer.execute_and_check(plan, &config.options(), |_, _| {})?;
     optimizer.optimize(plan, &config, |_, _| {})
 }
 

datafusion/core/tests/parquet/file_statistics.rs

@@ -38,6 +38,7 @@ use datafusion_execution::runtime_env::RuntimeEnvBuilder;
 use datafusion_expr::{col, lit, Expr};
 
 use datafusion::datasource::physical_plan::FileScanConfig;
+use datafusion_common::config::ConfigOptions;
 use datafusion_physical_optimizer::filter_pushdown::FilterPushdown;
 use datafusion_physical_optimizer::PhysicalOptimizerRule;
 use datafusion_physical_plan::filter::FilterExec;
@@ -55,7 +56,7 @@ async fn check_stats_precision_with_filter_pushdown() {
     let table = get_listing_table(&table_path, None, &opt).await;
 
     let (_, _, state) = get_cache_runtime_state();
-    let mut options = state.config().options().clone();
+    let mut options: ConfigOptions = state.config().options().as_ref().clone();
     options.execution.parquet.pushdown_filters = true;
 
     // Scan without filter, stats are exact