Don't share ConfigOptions (#3886)

benchmarks/src/bin/parquet_filter_pushdown.rs

@@ -20,7 +20,7 @@ use datafusion::common::Result;
 use datafusion::logical_expr::{lit, or, Expr};
 use datafusion::optimizer::utils::disjunction;
 use datafusion::physical_plan::collect;
-use datafusion::prelude::{col, SessionConfig, SessionContext};
+use datafusion::prelude::{col, SessionContext};
 use parquet::file::properties::WriterProperties;
 use parquet_test_utils::{ParquetScanOptions, TestParquetFile};
 use std::path::PathBuf;
@@ -69,9 +69,6 @@ async fn main() -> Result<()> {
     let opt: Opt = Opt::from_args();
     println!("Running benchmarks with the following options: {:?}", opt);
 
-    let config = SessionConfig::new().with_target_partitions(opt.partitions);
-    let mut ctx = SessionContext::with_config(config);
-
     let path = opt.path.join("logs.parquet");
 
     let mut props_builder = WriterProperties::builder();
@@ -88,17 +85,12 @@ async fn main() -> Result<()> {
 
     let test_file = gen_data(path, opt.scale_factor, props_builder.build())?;
 
-    run_benchmarks(&mut ctx, &test_file, opt.iterations, opt.debug).await?;
+    run_benchmarks(opt, &test_file).await?;
 
     Ok(())
 }
 
-async fn run_benchmarks(
-    ctx: &mut SessionContext,
-    test_file: &TestParquetFile,
-    iterations: usize,
-    debug: bool,
-) -> Result<()> {
+async fn run_benchmarks(opt: Opt, test_file: &TestParquetFile) -> Result<()> {
     let scan_options_matrix = vec![
         ParquetScanOptions {
             pushdown_filters: false,
@@ -148,11 +140,14 @@ async fn run_benchmarks(
         println!("Executing with filter '{}'", filter_expr);
         for scan_options in &scan_options_matrix {
             println!("Using scan options {:?}", scan_options);
-            for i in 0..iterations {
+            for i in 0..opt.iterations {
                 let start = Instant::now();
+
+                let config = scan_options.config().with_target_partitions(opt.partitions);
+                let ctx = SessionContext::with_config(config);
+
                 let rows =
-                    exec_scan(ctx, test_file, filter_expr.clone(), *scan_options, debug)
-                        .await?;
+                    exec_scan(&ctx, test_file, filter_expr.clone(), opt.debug).await?;
                 println!(
                     "Iteration {} returned {} rows in {} ms",
                     i,
@@ -170,10 +165,9 @@ async fn exec_scan(
     ctx: &SessionContext,
     test_file: &TestParquetFile,
     filter: Expr,
-    scan_options: ParquetScanOptions,
     debug: bool,
 ) -> Result<usize> {
-    let exec = test_file.create_scan(filter, scan_options).await?;
+    let exec = test_file.create_scan(filter).await?;
 
     let task_ctx = ctx.task_ctx();
     let result = collect(exec, task_ctx).await?;

benchmarks/src/bin/tpch.rs

@@ -395,8 +395,7 @@ async fn get_table(
             }
             "parquet" => {
                 let path = format!("{}/{}", path, table);
-                let format = ParquetFormat::new(state.config_options())
-                    .with_enable_pruning(Some(true));
+                let format = ParquetFormat::default().with_enable_pruning(Some(true));
 
                 (Arc::new(format), path, DEFAULT_PARQUET_EXTENSION)
             }

datafusion-examples/examples/flight_server.rs

@@ -15,14 +15,13 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use std::pin::Pin;
 use std::sync::Arc;
 
 use arrow_flight::SchemaAsIpc;
 use datafusion::arrow::error::ArrowError;
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::listing::{ListingOptions, ListingTableUrl};
-use futures::Stream;
+use futures::stream::BoxStream;
 use tonic::transport::Server;
 use tonic::{Request, Response, Status, Streaming};
 
@@ -39,37 +38,21 @@ pub struct FlightServiceImpl {}
 
 #[tonic::async_trait]
 impl FlightService for FlightServiceImpl {
-    type HandshakeStream = Pin<
-        Box<dyn Stream<Item = Result<HandshakeResponse, Status>> + Send + Sync + 'static>,
-    >;
-    type ListFlightsStream =
-        Pin<Box<dyn Stream<Item = Result<FlightInfo, Status>> + Send + Sync + 'static>>;
-    type DoGetStream =
-        Pin<Box<dyn Stream<Item = Result<FlightData, Status>> + Send + Sync + 'static>>;
-    type DoPutStream =
-        Pin<Box<dyn Stream<Item = Result<PutResult, Status>> + Send + Sync + 'static>>;
-    type DoActionStream = Pin<
-        Box<
-            dyn Stream<Item = Result<arrow_flight::Result, Status>>
-                + Send
-                + Sync
-                + 'static,
-        >,
-    >;
-    type ListActionsStream =
-        Pin<Box<dyn Stream<Item = Result<ActionType, Status>> + Send + Sync + 'static>>;
-    type DoExchangeStream =
-        Pin<Box<dyn Stream<Item = Result<FlightData, Status>> + Send + Sync + 'static>>;
+    type HandshakeStream = BoxStream<'static, Result<HandshakeResponse, Status>>;
+    type ListFlightsStream = BoxStream<'static, Result<FlightInfo, Status>>;
+    type DoGetStream = BoxStream<'static, Result<FlightData, Status>>;
+    type DoPutStream = BoxStream<'static, Result<PutResult, Status>>;
+    type DoActionStream = BoxStream<'static, Result<arrow_flight::Result, Status>>;
+    type ListActionsStream = BoxStream<'static, Result<ActionType, Status>>;
+    type DoExchangeStream = BoxStream<'static, Result<FlightData, Status>>;
 
     async fn get_schema(
         &self,
         request: Request<FlightDescriptor>,
     ) -> Result<Response<SchemaResult>, Status> {
         let request = request.into_inner();
 
-        let config = SessionConfig::new();
-        let listing_options =
-            ListingOptions::new(Arc::new(ParquetFormat::new(config.config_options())));
+        let listing_options = ListingOptions::new(Arc::new(ParquetFormat::default()));
         let table_path =
             ListingTableUrl::parse(&request.path[0]).map_err(to_tonic_err)?;
 
@@ -79,10 +62,10 @@ impl FlightService for FlightServiceImpl {
             .await
             .unwrap();
 
-        let options = datafusion::arrow::ipc::writer::IpcWriteOptions::default();
+        let options = arrow::ipc::writer::IpcWriteOptions::default();
         let schema_result = SchemaAsIpc::new(&schema, &options)
             .try_into()
-            .map_err(|e: ArrowError| tonic::Status::internal(e.to_string()))?;
+            .map_err(|e: ArrowError| Status::internal(e.to_string()))?;
 
         Ok(Response::new(schema_result))
     }

datafusion-examples/examples/parquet_sql_multiple_files.rs

@@ -32,8 +32,7 @@ async fn main() -> Result<()> {
     let testdata = datafusion::test_util::parquet_test_data();
 
     // Configure listing options
-    let file_format =
-        ParquetFormat::new(ctx.config_options()).with_enable_pruning(Some(true));
+    let file_format = ParquetFormat::default().with_enable_pruning(Some(true));
     let listing_options = ListingOptions::new(Arc::new(file_format))
         .with_file_extension(FileType::PARQUET.get_ext());
 

datafusion/core/src/catalog/information_schema.rs

@@ -29,13 +29,13 @@ use arrow::{
     datatypes::{DataType, Field, Schema, SchemaRef},
     record_batch::RecordBatch,
 };
-use parking_lot::RwLock;
 
 use datafusion_common::Result;
 
 use crate::config::ConfigOptions;
 use crate::datasource::streaming::{PartitionStream, StreamingTable};
 use crate::datasource::TableProvider;
+use crate::execution::context::TaskContext;
 use crate::logical_expr::TableType;
 use crate::physical_plan::stream::RecordBatchStreamAdapter;
 use crate::physical_plan::SendableRecordBatchStream;
@@ -55,20 +55,17 @@ const DF_SETTINGS: &str = "df_settings";
 /// schema that can introspect on tables in the catalog_list
 pub(crate) struct CatalogWithInformationSchema {
     catalog_list: Weak<dyn CatalogList>,
-    config_options: Weak<RwLock<ConfigOptions>>,
     /// wrapped provider
     inner: Arc<dyn CatalogProvider>,
 }
 
 impl CatalogWithInformationSchema {
     pub(crate) fn new(
         catalog_list: Weak<dyn CatalogList>,
-        config_options: Weak<RwLock<ConfigOptions>>,
         inner: Arc<dyn CatalogProvider>,
     ) -> Self {
         Self {
             catalog_list,
-            config_options,
             inner,
         }
     }
@@ -89,15 +86,10 @@ impl CatalogProvider for CatalogWithInformationSchema {
 
     fn schema(&self, name: &str) -> Option<Arc<dyn SchemaProvider>> {
         if name.eq_ignore_ascii_case(INFORMATION_SCHEMA) {
-            Weak::upgrade(&self.catalog_list).and_then(|catalog_list| {
-                Weak::upgrade(&self.config_options).map(|config_options| {
-                    Arc::new(InformationSchemaProvider {
-                        config: InformationSchemaConfig {
-                            catalog_list,
-                            config_options,
-                        },
-                    }) as Arc<dyn SchemaProvider>
-                })
+            Weak::upgrade(&self.catalog_list).map(|catalog_list| {
+                Arc::new(InformationSchemaProvider {
+                    config: InformationSchemaConfig { catalog_list },
+                }) as Arc<dyn SchemaProvider>
             })
         } else {
             self.inner.schema(name)
@@ -127,7 +119,6 @@ struct InformationSchemaProvider {
 #[derive(Clone)]
 struct InformationSchemaConfig {
     catalog_list: Arc<dyn CatalogList>,
-    config_options: Arc<RwLock<ConfigOptions>>,
 }
 
 impl InformationSchemaConfig {
@@ -220,8 +211,12 @@ impl InformationSchemaConfig {
     }
 
     /// Construct the `information_schema.df_settings` virtual table
-    fn make_df_settings(&self, builder: &mut InformationSchemaDfSettingsBuilder) {
-        for (name, setting) in self.config_options.read().options() {
+    fn make_df_settings(
+        &self,
+        config_options: &ConfigOptions,
+        builder: &mut InformationSchemaDfSettingsBuilder,
+    ) {
+        for (name, setting) in config_options.options() {
             builder.add_setting(name, setting.to_string());
         }
     }
@@ -298,7 +293,7 @@ impl PartitionStream for InformationSchemaTables {
         &self.schema
     }
 
-    fn execute(&self) -> SendableRecordBatchStream {
+    fn execute(&self, _ctx: Arc<TaskContext>) -> SendableRecordBatchStream {
         let mut builder = self.builder();
         let config = self.config.clone();
         Box::pin(RecordBatchStreamAdapter::new(
@@ -389,7 +384,7 @@ impl PartitionStream for InformationSchemaViews {
         &self.schema
     }
 
-    fn execute(&self) -> SendableRecordBatchStream {
+    fn execute(&self, _ctx: Arc<TaskContext>) -> SendableRecordBatchStream {
         let mut builder = self.builder();
         let config = self.config.clone();
         Box::pin(RecordBatchStreamAdapter::new(
@@ -503,7 +498,7 @@ impl PartitionStream for InformationSchemaColumns {
         &self.schema
     }
 
-    fn execute(&self) -> SendableRecordBatchStream {
+    fn execute(&self, _ctx: Arc<TaskContext>) -> SendableRecordBatchStream {
         let mut builder = self.builder();
         let config = self.config.clone();
         Box::pin(RecordBatchStreamAdapter::new(
@@ -690,15 +685,18 @@ impl PartitionStream for InformationSchemaDfSettings {
         &self.schema
     }
 
-    fn execute(&self) -> SendableRecordBatchStream {
-        let mut builder = self.builder();
+    fn execute(&self, ctx: Arc<TaskContext>) -> SendableRecordBatchStream {
         let config = self.config.clone();
+        let mut builder = self.builder();
         Box::pin(RecordBatchStreamAdapter::new(
             self.schema.clone(),
             // TODO: Stream this
             futures::stream::once(async move {
                 // create a mem table with the names of tables
-                config.make_df_settings(&mut builder);
+                config.make_df_settings(
+                    ctx.session_config().config_options(),
+                    &mut builder,
+                );
                 Ok(builder.finish())
             }),
         ))

datafusion/core/src/config.rs

@@ -21,11 +21,9 @@ use arrow::datatypes::DataType;
 use datafusion_common::ScalarValue;
 use itertools::Itertools;
 use log::warn;
-use parking_lot::RwLock;
 use std::collections::{BTreeMap, HashMap};
 use std::env;
 use std::fmt::{Debug, Formatter};
-use std::sync::Arc;
 
 /*-************************************
 *  Catalog related
@@ -484,11 +482,6 @@ impl ConfigOptions {
         Self { options }
     }
 
-    /// Create a new [`ConfigOptions`] wrapped in an RwLock and Arc
-    pub fn into_shareable(self) -> Arc<RwLock<Self>> {
-        Arc::new(RwLock::new(self))
-    }
-
     /// Create new ConfigOptions struct, taking values from
     /// environment variables where possible.
     ///

datafusion/core/src/datasource/file_format/mod.rs

@@ -128,7 +128,6 @@ pub(crate) mod test_util {
                     projection,
                     limit,
                     table_partition_cols: vec![],
-                    config_options: state.config_options(),
                     output_ordering: None,
                 },
                 &[],