Remove ObjectStore from FileScanConfig and ListingTableConfig

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

@@ -18,16 +18,13 @@
 //! Describes the interface and built-in implementations of schemas,
 //! representing collections of named tables.
 
-use parking_lot::{Mutex, RwLock};
+use parking_lot::RwLock;
 use std::any::Any;
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use crate::datasource::listing::{ListingTable, ListingTableConfig, ListingTableUrl};
-use crate::datasource::object_store::ObjectStoreRegistry;
 use crate::datasource::TableProvider;
 use crate::error::{DataFusionError, Result};
-use datafusion_data_access::object_store::ObjectStore;
 
 /// Represents a schema, comprising a number of named tables.
 pub trait SchemaProvider: Sync + Send {
@@ -130,135 +127,19 @@ impl SchemaProvider for MemorySchemaProvider {
     }
 }
 
-/// `ObjectStore` implementation of `SchemaProvider` to enable registering a `ListingTable`
-pub struct ObjectStoreSchemaProvider {
-    tables: RwLock<HashMap<String, Arc<dyn TableProvider>>>,
-    object_store_registry: Arc<Mutex<ObjectStoreRegistry>>,
-}
-
-impl ObjectStoreSchemaProvider {
-    /// Instantiates a new `ObjectStoreSchemaProvider` with an empty collection of tables.
-    pub fn new() -> Self {
-        Self {
-            tables: RwLock::new(HashMap::new()),
-            object_store_registry: Arc::new(Mutex::new(ObjectStoreRegistry::new())),
-        }
-    }
-
-    /// Assign an `ObjectStore` which enables calling `register_listing_table`.
-    pub fn register_object_store(
-        &self,
-        scheme: impl Into<String>,
-        object_store: Arc<dyn ObjectStore>,
-    ) -> Option<Arc<dyn ObjectStore>> {
-        self.object_store_registry
-            .lock()
-            .register_store(scheme.into(), object_store)
-    }
-
-    /// Retrieves a `ObjectStore` instance for a given Url
-    pub fn object_store(
-        &self,
-        url: impl AsRef<url::Url>,
-    ) -> Result<Arc<dyn ObjectStore>> {
-        self.object_store_registry
-            .lock()
-            .get_by_url(url)
-            .map_err(DataFusionError::from)
-    }
-
-    /// If supported by the implementation, adds a new table to this schema by creating a
-    /// `ListingTable` from the provided `url` and a previously registered `ObjectStore`.
-    /// If a table of the same name existed before, it returns "Table already exists" error.
-    pub async fn register_listing_table(
-        &self,
-        name: &str,
-        table_path: ListingTableUrl,
-        config: Option<ListingTableConfig>,
-    ) -> Result<()> {
-        let config = match config {
-            Some(cfg) => cfg,
-            None => {
-                let object_store = self.object_store(&table_path)?;
-                ListingTableConfig::new(object_store, table_path)
-                    .infer()
-                    .await?
-            }
-        };
-
-        let table = Arc::new(ListingTable::try_new(config)?);
-        self.register_table(name.into(), table)?;
-        Ok(())
-    }
-}
-
-impl Default for ObjectStoreSchemaProvider {
-    fn default() -> Self {
-        Self::new()
-    }
-}
-
-impl SchemaProvider for ObjectStoreSchemaProvider {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
-    fn table_names(&self) -> Vec<String> {
-        let tables = self.tables.read();
-        tables.keys().cloned().collect()
-    }
-
-    fn table(&self, name: &str) -> Option<Arc<dyn TableProvider>> {
-        let tables = self.tables.read();
-        tables.get(name).cloned()
-    }
-
-    fn register_table(
-        &self,
-        name: String,
-        table: Arc<dyn TableProvider>,
-    ) -> Result<Option<Arc<dyn TableProvider>>> {
-        if self.table_exist(name.as_str()) {
-            return Err(DataFusionError::Execution(format!(
-                "The table {} already exists",
-                name
-            )));
-        }
-        let mut tables = self.tables.write();
-        Ok(tables.insert(name, table))
-    }
-
-    fn deregister_table(&self, name: &str) -> Result<Option<Arc<dyn TableProvider>>> {
-        let mut tables = self.tables.write();
-        Ok(tables.remove(name))
-    }
-
-    fn table_exist(&self, name: &str) -> bool {
-        let tables = self.tables.read();
-        tables.contains_key(name)
-    }
-}
-
 #[cfg(test)]
 mod tests {
-    use std::ffi::OsStr;
-    use std::path::Path;
     use std::sync::Arc;
 
     use arrow::datatypes::Schema;
+    use datafusion_data_access::object_store::local::LocalFileSystem;
 
     use crate::assert_batches_eq;
-    use crate::catalog::catalog::CatalogProvider;
-    use crate::catalog::catalog::MemoryCatalogProvider;
-    use crate::catalog::schema::{
-        MemorySchemaProvider, ObjectStoreSchemaProvider, SchemaProvider,
-    };
-    use crate::datafusion_data_access::object_store::local::LocalFileSystem;
+    use crate::catalog::catalog::{CatalogProvider, MemoryCatalogProvider};
+    use crate::catalog::schema::{MemorySchemaProvider, SchemaProvider};
     use crate::datasource::empty::EmptyTable;
-    use crate::execution::context::SessionContext;
-
-    use crate::datasource::listing::ListingTableUrl;
-    use futures::StreamExt;
+    use crate::datasource::listing::{ListingTable, ListingTableConfig, ListingTableUrl};
+    use crate::prelude::SessionContext;
 
     #[tokio::test]
     async fn test_mem_provider() {
@@ -282,22 +163,27 @@ mod tests {
     #[tokio::test]
     async fn test_schema_register_listing_table() {
         let testdata = crate::test_util::parquet_test_data();
-        let filename = format!("{}/{}", testdata, "alltypes_plain.parquet");
+        let filename = format!("test:///{}/{}", testdata, "alltypes_plain.parquet");
         let table_path = ListingTableUrl::parse(filename).unwrap();
 
-        let schema = ObjectStoreSchemaProvider::new();
-        let _store = schema.register_object_store("test", Arc::new(LocalFileSystem {}));
+        let catalog = MemoryCatalogProvider::new();
+        let schema = MemorySchemaProvider::new();
 
-        schema
-            .register_listing_table("alltypes_plain", table_path, None)
+        let ctx = SessionContext::new();
+        let store = Arc::new(LocalFileSystem {});
+        ctx.runtime_env().register_object_store("test", store);
+
+        let config = ListingTableConfig::new(table_path)
+            .infer(&ctx.state())
             .await
             .unwrap();
+        let table = ListingTable::try_new(config).unwrap();
 
-        let catalog = MemoryCatalogProvider::new();
-        catalog.register_schema("active", Arc::new(schema)).unwrap();
-
-        let ctx = SessionContext::new();
+        schema
+            .register_table("alltypes_plain".to_string(), Arc::new(table))
+            .unwrap();
 
+        catalog.register_schema("active", Arc::new(schema)).unwrap();
         ctx.register_catalog("cat", Arc::new(catalog));
 
         let df = ctx
@@ -323,61 +209,4 @@ mod tests {
         ];
         assert_batches_eq!(expected, &actual);
     }
-
-    #[tokio::test]
-    async fn test_schema_register_listing_tables() {
-        let testdata = crate::test_util::parquet_test_data();
-
-        let schema = ObjectStoreSchemaProvider::new();
-        let store = schema
-            .register_object_store("file", Arc::new(LocalFileSystem {}))
-            .unwrap();
-
-        let mut files = store.list_file(&testdata).await.unwrap();
-        while let Some(file) = files.next().await {
-            let sized_file = file.unwrap().sized_file;
-            let path = Path::new(&sized_file.path);
-            let file = path.file_name().unwrap();
-            if file == OsStr::new("alltypes_dictionary.parquet")
-                || file == OsStr::new("alltypes_plain.parquet")
-            {
-                let name = path.file_stem().unwrap().to_str().unwrap();
-                let path = ListingTableUrl::parse(&sized_file.path).unwrap();
-                schema
-                    .register_listing_table(name, path, None)
-                    .await
-                    .unwrap();
-            }
-        }
-
-        let tables = vec![
-            String::from("alltypes_dictionary"),
-            String::from("alltypes_plain"),
-        ];
-
-        let mut schema_tables = schema.table_names();
-        schema_tables.sort();
-        assert_eq!(schema_tables, tables);
-    }
-
-    #[tokio::test]
-    #[should_panic(expected = "already exists")]
-    async fn test_schema_register_same_listing_table() {
-        let testdata = crate::test_util::parquet_test_data();
-        let filename = format!("{}/{}", testdata, "alltypes_plain.parquet");
-        let table_path = ListingTableUrl::parse(filename).unwrap();
-
-        let schema = ObjectStoreSchemaProvider::new();
-        let _store = schema.register_object_store("test", Arc::new(LocalFileSystem {}));
-
-        schema
-            .register_listing_table("alltypes_plain", table_path.clone(), None)
-            .await
-            .unwrap();
-
-        schema
-            .register_listing_table("alltypes_plain", table_path, None)
-            .await
-            .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,