Remove ObjectStore from FileScanConfig and ListingTableConfig (#2668)

* Remove ObjectStore from FileScanConfig and ListingTableConfig

* Update ballista pin

Author: tustvold

benchmarks/src/bin/tpch.rs

@@ -44,7 +44,6 @@ use datafusion::{
 };
 use datafusion::{
     arrow::util::pretty,
-    datafusion_data_access::object_store::local::LocalFileSystem,
     datasource::listing::{ListingOptions, ListingTable, ListingTableConfig},
 };
 
@@ -427,7 +426,7 @@ fn get_table(
     };
 
     let table_path = ListingTableUrl::parse(path)?;
-    let config = ListingTableConfig::new(Arc::new(LocalFileSystem {}), table_path)
+    let config = ListingTableConfig::new(table_path)
         .with_listing_options(options)
         .with_schema(schema);
 

datafusion-examples/examples/flight_server.rs

@@ -19,7 +19,6 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use arrow_flight::SchemaAsIpc;
-use datafusion::datafusion_data_access::object_store::local::LocalFileSystem;
 use datafusion::datasource::file_format::parquet::ParquetFormat;
 use datafusion::datasource::listing::{ListingOptions, ListingTableUrl};
 use futures::Stream;
@@ -71,8 +70,9 @@ impl FlightService for FlightServiceImpl {
         let table_path =
             ListingTableUrl::parse(&request.path[0]).map_err(to_tonic_err)?;
 
+        let ctx = SessionContext::new();
         let schema = listing_options
-            .infer_schema(Arc::new(LocalFileSystem {}), &table_path)
+            .infer_schema(&ctx.state(), &table_path)
             .await
             .unwrap();
 

datafusion/core/benches/sort_limit_query_sql.rs

@@ -18,7 +18,6 @@
 #[macro_use]
 extern crate criterion;
 use criterion::Criterion;
-use datafusion::datafusion_data_access::object_store::local::LocalFileSystem;
 use datafusion::datasource::file_format::csv::CsvFormat;
 use datafusion::datasource::listing::{
     ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl,
@@ -71,7 +70,7 @@ fn create_context() -> Arc<Mutex<SessionContext>> {
     // create CSV data source
     let listing_options = ListingOptions::new(Arc::new(CsvFormat::default()));
 
-    let config = ListingTableConfig::new(Arc::new(LocalFileSystem {}), table_path)
+    let config = ListingTableConfig::new(table_path)
         .with_listing_options(listing_options)
         .with_schema(schema);
 

datafusion/core/src/catalog/schema.rs

@@ -171,11 +171,10 @@ mod tests {
 
         let ctx = SessionContext::new();
         let store = Arc::new(LocalFileSystem {});
-        ctx.runtime_env()
-            .register_object_store("file", store.clone());
+        ctx.runtime_env().register_object_store("file", store);
 
-        let config = ListingTableConfig::new(store, table_path)
-            .infer()
+        let config = ListingTableConfig::new(table_path)
+            .infer(&ctx.state())
             .await
             .unwrap();
         let table = ListingTable::try_new(config).unwrap();

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

@@ -115,7 +115,6 @@ pub(crate) mod test_util {
         let exec = format
             .create_physical_plan(
                 FileScanConfig {
-                    object_store: store,
                     object_store_url: ObjectStoreUrl::local_filesystem(),
                     file_schema,
                     file_groups,