feat: Add ConfigOptions to ScalarFunctionArgs

datafusion-examples/examples/composed_extension_codec.rs

@@ -71,8 +71,14 @@ async fn main() {
 
     // deserialize proto back to execution plan
     let runtime = ctx.runtime_env();
+    let config_options = ctx.state().config_options().clone();
     let result_exec_plan: Arc<dyn ExecutionPlan> = proto
-        .try_into_physical_plan(&ctx, runtime.deref(), &composed_codec)
+        .try_into_physical_plan(
+            &ctx,
+            Arc::new(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

@@ -28,6 +28,7 @@ use datafusion::functions_aggregate::first_last::first_value_udaf;
 use datafusion::optimizer::simplify_expressions::ExprSimplifier;
 use datafusion::physical_expr::{analyze, AnalysisContext, ExprBoundaries};
 use datafusion::prelude::*;
+use datafusion_common::config::ConfigOptions;
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_common::{ScalarValue, ToDFSchema};
 use datafusion_expr::execution_props::ExecutionProps;
@@ -356,8 +357,13 @@ 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 = Arc::new(ConfigOptions::default());
+    let physical_expr = datafusion_physical_expr::create_physical_expr(
+        &expr,
+        &df_schema,
+        &props,
+        Arc::clone(&config_options),
+    )?;
     let e = physical_expr.evaluate(&batch).unwrap_err();
     assert!(e
         .find_root()
@@ -377,6 +383,7 @@ fn type_coercion_demo() -> Result<()> {
         &coerced_expr,
         &df_schema,
         &props,
+        Arc::clone(&config_options),
     )?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 
@@ -389,6 +396,7 @@ fn type_coercion_demo() -> Result<()> {
         &coerced_expr,
         &df_schema,
         &props,
+        Arc::clone(&config_options),
     )?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 
@@ -417,6 +425,7 @@ fn type_coercion_demo() -> Result<()> {
         &coerced_expr,
         &df_schema,
         &props,
+        Arc::clone(&config_options),
     )?;
     assert!(physical_expr.evaluate(&batch).is_ok());
 

datafusion-examples/examples/pruning.rs

@@ -22,6 +22,7 @@ use datafusion::execution::context::ExecutionProps;
 use datafusion::physical_expr::create_physical_expr;
 use datafusion::physical_optimizer::pruning::{PruningPredicate, PruningStatistics};
 use datafusion::prelude::*;
+use datafusion_common::config::ConfigOptions;
 use std::collections::HashSet;
 use std::sync::Arc;
 
@@ -187,7 +188,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 = Arc::new(ConfigOptions::default());
+    let physical_expr =
+        create_physical_expr(&expr, &df_schema, &props, config_options).unwrap();
     PruningPredicate::try_new(physical_expr, schema.clone()).unwrap()
 }
 

datafusion/common/src/config.rs

@@ -665,7 +665,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,

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

@@ -39,6 +39,7 @@ 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::{Expr, Volatility};
@@ -283,10 +284,16 @@ async fn prune_partitions(
 
     // TODO: Plumb this down
     let props = ExecutionProps::new();
+    let config_options = Arc::new(ConfigOptions::default());
 
     // 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,
+            Arc::clone(&config_options),
+        )?;
         expr.evaluate(&batch)?.into_array(partitions.len())
     };
 

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

@@ -887,6 +887,7 @@ impl TableProvider for ListingTable {
                     &expr,
                     &table_df_schema,
                     state.execution_props(),
+                    Arc::new(state.config_options().clone()),
                 )?;
                 Some(filters)
             }

datafusion/core/src/datasource/memory.rs

@@ -238,6 +238,7 @@ impl TableProvider for MemTable {
                         sort_exprs,
                         &df_schema,
                         state.execution_props(),
+                        Arc::new(state.config_options().clone()),
                     )
                 })
                 .collect::<Result<Vec<_>>>()?;

datafusion/core/src/datasource/physical_plan/file_scan_config.rs

@@ -617,10 +617,10 @@ fn create_output_array(
 
 #[cfg(test)]
 mod tests {
-    use arrow_array::Int32Array;
-
     use super::*;
     use crate::{test::columns, test_util::aggr_test_schema};
+    use arrow_array::Int32Array;
+    use datafusion_common::config::ConfigOptions;
 
     #[test]
     fn physical_plan_config_no_projection() {
@@ -1121,6 +1121,7 @@ mod tests {
                             &expr,
                             &DFSchema::try_from(table_schema.as_ref().clone())?,
                             &ExecutionProps::default(),
+                            Arc::new(ConfigOptions::default()),
                         )
                     })
                     .collect::<Result<Vec<_>>>()?,

datafusion/core/src/execution/session_state.rs

@@ -756,7 +756,12 @@ impl SessionState {
                 .transform_up(|expr| rewrite.rewrite(expr, df_schema, config_options))?
                 .data;
         }
-        create_physical_expr(&expr, df_schema, self.execution_props())
+        create_physical_expr(
+            &expr,
+            df_schema,
+            self.execution_props(),
+            Arc::new(config_options.clone()),
+        )
     }
 
     /// Return the session ID

datafusion/core/src/physical_optimizer/projection_pushdown.rs

@@ -1417,6 +1417,7 @@ mod tests {
                     )),
                 ],
                 DataType::Int32,
+                Arc::new(ConfigOptions::default()),
             )),
             Arc::new(CaseExpr::try_new(
                 Some(Arc::new(Column::new("d", 2))),
@@ -1482,6 +1483,7 @@ mod tests {
                     )),
                 ],
                 DataType::Int32,
+                Arc::new(ConfigOptions::default()),
             )),
             Arc::new(CaseExpr::try_new(
                 Some(Arc::new(Column::new("d", 3))),
@@ -1550,6 +1552,7 @@ mod tests {
                     )),
                 ],
                 DataType::Int32,
+                Arc::new(ConfigOptions::default()),
             )),
             Arc::new(CaseExpr::try_new(
                 Some(Arc::new(Column::new("d", 2))),
@@ -1615,6 +1618,7 @@ mod tests {
                     )),
                 ],
                 DataType::Int32,
+                Arc::new(ConfigOptions::default()),
             )),
             Arc::new(CaseExpr::try_new(
                 Some(Arc::new(Column::new("d_new", 3))),