diff --git a/datafusion-cli/src/catalog.rs b/datafusion-cli/src/catalog.rs index ca24da7873bc1..b4c7c1c56e2ca 100644 --- a/datafusion-cli/src/catalog.rs +++ b/datafusion-cli/src/catalog.rs @@ -57,6 +57,10 @@ impl CatalogProviderList for DynamicObjectStoreCatalog { self.inner.register_catalog(name, catalog) } + fn deregister_catalog(&self, name: &str) -> Result>> { + self.inner.deregister_catalog(name) + } + fn catalog_names(&self) -> Vec { self.inner.catalog_names() } diff --git a/datafusion-examples/examples/data_io/catalog.rs b/datafusion-examples/examples/data_io/catalog.rs index 7e5cc5a4cfc05..ee879083d104a 100644 --- a/datafusion-examples/examples/data_io/catalog.rs +++ b/datafusion-examples/examples/data_io/catalog.rs @@ -279,6 +279,11 @@ impl CatalogProviderList for CustomCatalogProviderList { Some(catalog) } + fn deregister_catalog(&self, name: &str) -> Result>> { + let mut cats = self.catalogs.write().unwrap(); + Ok(cats.remove(name)) + } + /// Retrieves the list of available catalog names fn catalog_names(&self) -> Vec { let cats = self.catalogs.read().unwrap(); diff --git a/datafusion/catalog/src/catalog.rs b/datafusion/catalog/src/catalog.rs index 07da1293a781d..2552ee8676560 100644 --- a/datafusion/catalog/src/catalog.rs +++ b/datafusion/catalog/src/catalog.rs @@ -16,7 +16,9 @@ // under the License. // Re-export from this module for backwards compatibility. -pub use datafusion_session::{CatalogProvider, CatalogProviderList}; +pub use datafusion_session::{ + CatalogProvider, CatalogProviderFactory, CatalogProviderList, +}; // Re-export so users can access this type through `datafusion_catalog` and // `datafusion::catalog` without depending directly on `datafusion_session`. pub use datafusion_session::EmptyCatalogProviderList; diff --git a/datafusion/catalog/src/dynamic_file/catalog.rs b/datafusion/catalog/src/dynamic_file/catalog.rs index 4437d99667547..81f084d345d40 100644 --- a/datafusion/catalog/src/dynamic_file/catalog.rs +++ b/datafusion/catalog/src/dynamic_file/catalog.rs @@ -49,6 +49,13 @@ impl CatalogProviderList for DynamicFileCatalog { self.inner.register_catalog(name, catalog) } + fn deregister_catalog( + &self, + name: &str, + ) -> datafusion_common::Result>> { + self.inner.deregister_catalog(name) + } + fn catalog_names(&self) -> Vec { self.inner.catalog_names() } diff --git a/datafusion/catalog/src/memory/catalog.rs b/datafusion/catalog/src/memory/catalog.rs index ebe6b9dfa0ebc..717e319651bf6 100644 --- a/datafusion/catalog/src/memory/catalog.rs +++ b/datafusion/catalog/src/memory/catalog.rs @@ -20,7 +20,7 @@ use crate::{CatalogProvider, CatalogProviderList, SchemaProvider}; use dashmap::DashMap; -use datafusion_common::exec_err; +use datafusion_common::{Result, exec_err}; use std::sync::Arc; /// Simple in-memory list of catalogs @@ -54,6 +54,10 @@ impl CatalogProviderList for MemoryCatalogProviderList { self.catalogs.insert(name, catalog) } + fn deregister_catalog(&self, name: &str) -> Result>> { + Ok(self.catalogs.remove(name).map(|(_, catalog)| catalog)) + } + fn catalog_names(&self) -> Vec { self.catalogs.iter().map(|c| c.key().clone()).collect() } @@ -97,7 +101,7 @@ impl CatalogProvider for MemoryCatalogProvider { &self, name: &str, schema: Arc, - ) -> datafusion_common::Result>> { + ) -> Result>> { Ok(self.schemas.insert(name.into(), schema)) } @@ -105,7 +109,7 @@ impl CatalogProvider for MemoryCatalogProvider { &self, name: &str, cascade: bool, - ) -> datafusion_common::Result>> { + ) -> Result>> { if let Some(schema) = self.schema(name) { let table_names = schema.table_names(); match (table_names.is_empty(), cascade) { diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs index ff1ad25811440..bf5e6e3b449c9 100644 --- a/datafusion/core/src/execution/context/mod.rs +++ b/datafusion/core/src/execution/context/mod.rs @@ -28,7 +28,8 @@ use crate::execution::session_state::SessionStateBuilder; use crate::{ catalog::listing_schema::ListingSchemaProvider, catalog::{ - CatalogProvider, CatalogProviderList, TableProvider, TableProviderFactory, + CatalogProvider, CatalogProviderFactory, CatalogProviderList, TableProvider, + TableProviderFactory, }, dataframe::DataFrame, datasource::listing::{ @@ -44,10 +45,10 @@ use crate::{ logical_expr::AggregateUDF, logical_expr::ScalarUDF, logical_expr::{ - CreateCatalog, CreateCatalogSchema, CreateExternalTable, CreateFunction, - CreateMemoryTable, CreateView, DropCatalogSchema, DropFunction, DropTable, - DropView, Execute, LogicalPlan, LogicalPlanBuilder, Prepare, ResetVariable, - SetVariable, TableType, UNNAMED_TABLE, + CreateCatalog, CreateCatalogSchema, CreateExternalCatalog, CreateExternalTable, + CreateFunction, CreateMemoryTable, CreateView, DropCatalog, DropCatalogSchema, + DropFunction, DropTable, DropView, Execute, LogicalPlan, LogicalPlanBuilder, + Prepare, ResetVariable, SetVariable, TableType, UNNAMED_TABLE, }, physical_expr::PhysicalExpr, physical_plan::ExecutionPlan, @@ -561,6 +562,19 @@ impl SessionContext { self.state.read().table_factories().get(file_type).cloned() } + /// Return the [`CatalogProviderFactory`] that is registered for the + /// specified catalog type, if any. + pub fn catalog_factory( + &self, + catalog_type: &str, + ) -> Option> { + self.state + .read() + .catalog_factories() + .get(catalog_type) + .cloned() + } + /// Return the `enable_ident_normalization` of this Session pub fn enable_ident_normalization(&self) -> bool { self.state @@ -705,9 +719,15 @@ impl SessionContext { self.create_catalog_schema(cmd) } DdlStatement::CreateCatalog(cmd) => self.create_catalog(cmd), + DdlStatement::CreateExternalCatalog(cmd) => { + (Box::pin(async move { self.create_external_catalog(&cmd).await }) + as std::pin::Pin + Send>>) + .await + } DdlStatement::DropTable(cmd) => Box::pin(self.drop_table(cmd)).await, DdlStatement::DropView(cmd) => Box::pin(self.drop_view(cmd)).await, DdlStatement::DropCatalogSchema(cmd) => self.drop_schema(cmd), + DdlStatement::DropCatalog(cmd) => self.drop_catalog(cmd), DdlStatement::CreateFunction(cmd) => { Box::pin(self.create_function(*cmd)).await } @@ -1043,6 +1063,49 @@ impl SessionContext { } } + async fn create_external_catalog( + &self, + cmd: &CreateExternalCatalog, + ) -> Result { + let exists = self.catalog(cmd.catalog_name.as_str()).is_some(); + + match (cmd.if_not_exists, cmd.or_replace, exists) { + (true, false, true) => self.return_empty_dataframe(), + (true, true, true) => { + exec_err!("'IF NOT EXISTS' cannot coexist with 'REPLACE'") + } + (false, false, true) => { + exec_err!("External catalog '{}' already exists", cmd.catalog_name) + } + (_, _, _) => { + let new_catalog = self.create_custom_catalog(cmd).await?; + self.state + .write() + .catalog_list() + .register_catalog(cmd.catalog_name.clone(), new_catalog); + self.return_empty_dataframe() + } + } + } + + async fn create_custom_catalog( + &self, + cmd: &CreateExternalCatalog, + ) -> Result> { + let state = self.state.read().clone(); + let catalog_type = cmd.catalog_type.to_uppercase(); + let factory = state + .catalog_factories() + .get(catalog_type.as_str()) + .ok_or_else(|| { + exec_datafusion_err!( + "Unable to find catalog factory for {}", + cmd.catalog_type + ) + })?; + factory.create(&state, cmd).await + } + async fn drop_table(&self, cmd: DropTable) -> Result { let DropTable { name, if_exists, .. @@ -1106,6 +1169,22 @@ impl SessionContext { exec_err!("Schema '{schema_ref}' doesn't exist.") } + fn drop_catalog(&self, cmd: DropCatalog) -> Result { + let DropCatalog { + name, if_exists, .. + } = cmd; + let dereg = self + .state + .write() + .catalog_list() + .deregister_catalog(&name)?; + match (dereg, if_exists) { + (Some(_), _) => self.return_empty_dataframe(), + (None, true) => self.return_empty_dataframe(), + (None, false) => exec_err!("Catalog '{name}' doesn't exist."), + } + } + fn set_variable(&self, stmt: SetVariable) -> Result<()> { let SetVariable { variable, value } = stmt; diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs index aa8ba4c3b733b..e0811827d3034 100644 --- a/datafusion/core/src/execution/session_state.rs +++ b/datafusion/core/src/execution/session_state.rs @@ -23,7 +23,9 @@ use std::collections::{HashMap, HashSet}; use std::fmt::Debug; use std::sync::Arc; -use crate::catalog::{CatalogProviderList, SchemaProvider, TableProviderFactory}; +use crate::catalog::{ + CatalogProviderFactory, CatalogProviderList, SchemaProvider, TableProviderFactory, +}; use crate::datasource::file_format::FileFormatFactory; #[cfg(feature = "sql")] use crate::datasource::provider_as_source; @@ -202,6 +204,16 @@ struct SessionStateInner { /// /// [`TableProvider`]: crate::catalog::TableProvider table_factories: HashMap>, + /// CatalogProviderFactories for different catalog implementations. + /// + /// Maps strings like "ICEBERG" to an instance of [`CatalogProviderFactory`] + /// + /// This is used to create [`CatalogProvider`] instances for the + /// `CREATE EXTERNAL CATALOG ... STORED AS ` statement, for catalogs + /// backed by an external implementation. + /// + /// [`CatalogProvider`]: crate::catalog::CatalogProvider + catalog_factories: HashMap>, /// Runtime environment runtime_env: Arc, /// [FunctionFactory] to support pluggable user defined function handler. @@ -246,6 +258,7 @@ impl Debug for SessionState { .field("execution_props", &self.execution_props) .field("table_options", &self.inner.table_options) .field("table_factories", &self.inner.table_factories) + .field("catalog_factories", &self.inner.catalog_factories) .field("function_factory", &self.inner.function_factory) .field("cache_factory", &self.inner.cache_factory) .field("expr_planners", &self.inner.expr_planners); @@ -494,6 +507,18 @@ impl SessionState { &mut Arc::make_mut(&mut self.inner).table_factories } + /// Get the catalog factories + pub fn catalog_factories(&self) -> &HashMap> { + &self.inner.catalog_factories + } + + /// Get the catalog factories + pub fn catalog_factories_mut( + &mut self, + ) -> &mut HashMap> { + &mut Arc::make_mut(&mut self.inner).catalog_factories + } + /// Parse an SQL string into an DataFusion specific AST /// [`Statement`]. See [`SessionContext::sql`] for running queries. /// @@ -1134,6 +1159,7 @@ pub struct SessionStateBuilder { table_options: Option, execution_props: Option, table_factories: Option>>, + catalog_factories: Option>>, runtime_env: Option>, function_factory: Option>, cache_factory: Option>, @@ -1177,6 +1203,7 @@ impl SessionStateBuilder { config: None, execution_props: None, table_factories: None, + catalog_factories: None, runtime_env: None, function_factory: None, cache_factory: None, @@ -1242,6 +1269,7 @@ impl SessionStateBuilder { table_options: Some(existing.table_options), execution_props: Some(execution_props), table_factories: Some(existing.table_factories), + catalog_factories: Some(existing.catalog_factories), runtime_env: Some(existing.runtime_env), function_factory: existing.function_factory, cache_factory: existing.cache_factory, @@ -1556,6 +1584,27 @@ impl SessionStateBuilder { self } + /// Add a [`CatalogProviderFactory`] to the map of factories + pub fn with_catalog_factory( + mut self, + key: String, + catalog_factory: Arc, + ) -> Self { + let mut catalog_factories = self.catalog_factories.unwrap_or_default(); + catalog_factories.insert(key, catalog_factory); + self.catalog_factories = Some(catalog_factories); + self + } + + /// Set the map of [`CatalogProviderFactory`]s + pub fn with_catalog_factories( + mut self, + catalog_factories: HashMap>, + ) -> Self { + self.catalog_factories = Some(catalog_factories); + self + } + /// Set the [`RuntimeEnv`] pub fn with_runtime_env(mut self, runtime_env: Arc) -> Self { self.runtime_env = Some(runtime_env); @@ -1655,6 +1704,7 @@ impl SessionStateBuilder { config, execution_props, table_factories, + catalog_factories, runtime_env, function_factory, cache_factory, @@ -1696,6 +1746,7 @@ impl SessionStateBuilder { }), config, table_factories: table_factories.unwrap_or_default(), + catalog_factories: catalog_factories.unwrap_or_default(), runtime_env, function_factory, cache_factory, @@ -1937,6 +1988,13 @@ impl SessionStateBuilder { &mut self.table_factories } + /// Returns the current catalog_factories value + pub fn catalog_factories( + &mut self, + ) -> &mut Option>> { + &mut self.catalog_factories + } + /// Returns the current runtime_env value pub fn runtime_env(&mut self) -> &mut Option> { &mut self.runtime_env @@ -1989,6 +2047,7 @@ impl Debug for SessionStateBuilder { .field("execution_props", &self.execution_props) .field("table_options", &self.table_options) .field("table_factories", &self.table_factories) + .field("catalog_factories", &self.catalog_factories) .field("function_factory", &self.function_factory) .field("cache_factory", &self.cache_factory) .field("expr_planners", &self.expr_planners); diff --git a/datafusion/core/tests/sql/create_drop.rs b/datafusion/core/tests/sql/create_drop.rs index 4a60a79ff5de3..f98685f6a3f7a 100644 --- a/datafusion/core/tests/sql/create_drop.rs +++ b/datafusion/core/tests/sql/create_drop.rs @@ -15,11 +15,35 @@ // specific language governing permissions and limitations // under the License. +use async_trait::async_trait; +use datafusion::catalog::{ + CatalogProvider, CatalogProviderFactory, MemoryCatalogProvider, +}; use datafusion::execution::session_state::SessionStateBuilder; +use datafusion::logical_expr::CreateExternalCatalog; use datafusion::test_util::TestTableFactory; +use datafusion_catalog::Session; +use datafusion_common::exec_err; use super::*; +#[derive(Debug)] +struct TestCatalogFactory {} + +#[async_trait] +impl CatalogProviderFactory for TestCatalogFactory { + async fn create( + &self, + _state: &dyn Session, + cmd: &CreateExternalCatalog, + ) -> Result> { + if cmd.options.contains_key("fail") { + return exec_err!("catalog factory configured to fail"); + } + Ok(Arc::new(MemoryCatalogProvider::new())) + } +} + #[tokio::test] async fn create_custom_table() -> Result<()> { let mut state = SessionStateBuilder::new().with_default_features().build(); @@ -89,3 +113,105 @@ async fn create_drop_table() -> Result<()> { Ok(()) } + +#[tokio::test] +async fn create_external_catalog_with_factory() -> Result<()> { + let mut state = SessionStateBuilder::new().with_default_features().build(); + state + .catalog_factories_mut() + .insert("TESTCATALOG".to_string(), Arc::new(TestCatalogFactory {})); + let ctx = SessionContext::new_with_state(state); + + let sql = "CREATE EXTERNAL CATALOG cat STORED AS TESTCATALOG LOCATION 's3://bucket/warehouse' OPTIONS ('warehouse' 'cat')"; + ctx.sql(sql).await?; + + assert!( + ctx.catalog("cat").is_some(), + "Catalog should have been created!" + ); + + Ok(()) +} + +#[tokio::test] +async fn create_external_catalog_unknown_factory() -> Result<()> { + let ctx = SessionContext::new(); + + let sql = "CREATE EXTERNAL CATALOG cat STORED AS TESTCATALOG LOCATION 's3://bucket/warehouse'"; + let err = ctx.sql(sql).await.unwrap_err(); + assert_contains!( + err.to_string(), + "Unable to find catalog factory for TESTCATALOG" + ); + + Ok(()) +} + +#[tokio::test] +async fn create_external_catalog_factory_error_not_registered() -> Result<()> { + let mut state = SessionStateBuilder::new().with_default_features().build(); + state + .catalog_factories_mut() + .insert("TESTCATALOG".to_string(), Arc::new(TestCatalogFactory {})); + let ctx = SessionContext::new_with_state(state); + + let sql = "CREATE EXTERNAL CATALOG cat STORED AS TESTCATALOG LOCATION 's3://x' OPTIONS ('fail' 'true')"; + let err = ctx.sql(sql).await.unwrap_err(); + assert_contains!(err.to_string(), "catalog factory configured to fail"); + assert!( + ctx.catalog("cat").is_none(), + "Catalog should not have been registered when the factory errors" + ); + + Ok(()) +} + +#[tokio::test] +async fn create_external_catalog_if_not_exists() -> Result<()> { + let mut state = SessionStateBuilder::new().with_default_features().build(); + state + .catalog_factories_mut() + .insert("TESTCATALOG".to_string(), Arc::new(TestCatalogFactory {})); + let ctx = SessionContext::new_with_state(state); + + let sql = "CREATE EXTERNAL CATALOG cat STORED AS TESTCATALOG LOCATION 's3://x'"; + ctx.sql(sql).await?; + + // creating it again without IF NOT EXISTS should fail + let err = ctx.sql(sql).await.unwrap_err(); + assert_contains!(err.to_string(), "already exists"); + + // ... but should succeed with IF NOT EXISTS + let sql = "CREATE EXTERNAL CATALOG IF NOT EXISTS cat STORED AS TESTCATALOG LOCATION 's3://x'"; + ctx.sql(sql).await?; + + Ok(()) +} + +#[tokio::test] +async fn create_drop_catalog() -> Result<()> { + let mut state = SessionStateBuilder::new().with_default_features().build(); + state + .catalog_factories_mut() + .insert("TESTCATALOG".to_string(), Arc::new(TestCatalogFactory {})); + let ctx = SessionContext::new_with_state(state); + + let sql = "CREATE EXTERNAL CATALOG cat STORED AS TESTCATALOG LOCATION 's3://x'"; + ctx.sql(sql).await?; + assert!(ctx.catalog("cat").is_some()); + + ctx.sql("DROP CATALOG cat").await?; + assert!( + ctx.catalog("cat").is_none(), + "Catalog should have been dropped!" + ); + + // dropping again should fail without IF EXISTS + let err = ctx.sql("DROP CATALOG cat").await.unwrap_err(); + assert_contains!(err.to_string(), "doesn't exist"); + + // ... but should succeed with IF EXISTS + ctx.sql("DROP CATALOG IF EXISTS cat").await?; + + Ok(()) +} diff --git a/datafusion/expr/src/logical_plan/ddl.rs b/datafusion/expr/src/logical_plan/ddl.rs index 51d88e43c1576..afd5241c5a68b 100644 --- a/datafusion/expr/src/logical_plan/ddl.rs +++ b/datafusion/expr/src/logical_plan/ddl.rs @@ -50,6 +50,8 @@ pub enum DdlStatement { CreateCatalogSchema(CreateCatalogSchema), /// Creates a new catalog (aka "Database"). CreateCatalog(CreateCatalog), + /// Creates a new catalog by invoking a registered `CatalogProviderFactory`. + CreateExternalCatalog(Box), /// Creates a new index. CreateIndex(CreateIndex), /// Drops a table. @@ -58,6 +60,8 @@ pub enum DdlStatement { DropView(DropView), /// Drops a catalog schema DropCatalogSchema(DropCatalogSchema), + /// Drops a catalog previously created with `CREATE EXTERNAL CATALOG`. + DropCatalog(DropCatalog), /// Create function statement. Boxed for the same reason as /// [`Self::CreateExternalTable`] (~288 bytes). CreateFunction(Box), @@ -76,10 +80,12 @@ impl DdlStatement { schema } DdlStatement::CreateCatalog(CreateCatalog { schema, .. }) => schema, + DdlStatement::CreateExternalCatalog(ce) => &ce.schema, DdlStatement::CreateIndex(CreateIndex { schema, .. }) => schema, DdlStatement::DropTable(DropTable { schema, .. }) => schema, DdlStatement::DropView(DropView { schema, .. }) => schema, DdlStatement::DropCatalogSchema(DropCatalogSchema { schema, .. }) => schema, + DdlStatement::DropCatalog(DropCatalog { schema, .. }) => schema, DdlStatement::CreateFunction(cf) => &cf.schema, DdlStatement::DropFunction(DropFunction { schema, .. }) => schema, } @@ -94,10 +100,12 @@ impl DdlStatement { DdlStatement::CreateView(_) => "CreateView", DdlStatement::CreateCatalogSchema(_) => "CreateCatalogSchema", DdlStatement::CreateCatalog(_) => "CreateCatalog", + DdlStatement::CreateExternalCatalog(_) => "CreateExternalCatalog", DdlStatement::CreateIndex(_) => "CreateIndex", DdlStatement::DropTable(_) => "DropTable", DdlStatement::DropView(_) => "DropView", DdlStatement::DropCatalogSchema(_) => "DropCatalogSchema", + DdlStatement::DropCatalog(_) => "DropCatalog", DdlStatement::CreateFunction(_) => "CreateFunction", DdlStatement::DropFunction(_) => "DropFunction", } @@ -109,6 +117,7 @@ impl DdlStatement { DdlStatement::CreateExternalTable(_) => vec![], DdlStatement::CreateCatalogSchema(_) => vec![], DdlStatement::CreateCatalog(_) => vec![], + DdlStatement::CreateExternalCatalog(_) => vec![], DdlStatement::CreateMemoryTable(CreateMemoryTable { input, .. }) => { vec![input] } @@ -117,6 +126,7 @@ impl DdlStatement { DdlStatement::DropTable(_) => vec![], DdlStatement::DropView(_) => vec![], DdlStatement::DropCatalogSchema(_) => vec![], + DdlStatement::DropCatalog(_) => vec![], DdlStatement::CreateFunction(_) => vec![], DdlStatement::DropFunction(_) => vec![], } @@ -166,6 +176,9 @@ impl DdlStatement { }) => { write!(f, "CreateCatalog: {catalog_name:?}") } + DdlStatement::CreateExternalCatalog(ce) => { + write!(f, "CreateExternalCatalog: {:?}", ce.catalog_name) + } DdlStatement::CreateIndex(CreateIndex { name, .. }) => { write!(f, "CreateIndex: {name:?}") } @@ -190,6 +203,11 @@ impl DdlStatement { "DropCatalogSchema: {name:?} if not exist:={if_exists} cascade:={cascade}" ) } + DdlStatement::DropCatalog(DropCatalog { + name, if_exists, .. + }) => { + write!(f, "DropCatalog: {name:?} if not exist:={if_exists}") + } DdlStatement::CreateFunction(cf) => { let name = &cf.name; write!(f, "CreateFunction: name {name:?}") @@ -530,6 +548,73 @@ impl PartialOrd for CreateCatalog { } } +/// Creates a catalog by invoking a registered `CatalogProviderFactory`. +/// +/// This mirrors [`CreateExternalTable`], which creates a table by invoking a +/// registered `TableProviderFactory`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CreateExternalCatalog { + /// The catalog name + pub catalog_name: String, + /// The key used to look up the `CatalogProviderFactory` (the `STORED AS` clause) + pub catalog_type: String, + /// The physical location of the catalog, if applicable + pub location: Option, + /// Do nothing (except issuing a notice) if a catalog with the same name already exists + pub if_not_exists: bool, + /// Option to replace the catalog if it already exists + pub or_replace: bool, + /// Catalog(provider) specific options + pub options: HashMap, + /// Dummy schema + pub schema: DFSchemaRef, +} + +// Hashing refers to a subset of fields considered in PartialEq. +impl Hash for CreateExternalCatalog { + fn hash(&self, state: &mut H) { + self.catalog_name.hash(state); + self.catalog_type.hash(state); + self.location.hash(state); + self.if_not_exists.hash(state); + self.or_replace.hash(state); + self.options.len().hash(state); // HashMap is not hashable + } +} + +// Manual implementation needed because of `schema` and `options` fields. +// Comparison excludes these fields. +impl PartialOrd for CreateExternalCatalog { + fn partial_cmp(&self, other: &Self) -> Option { + #[derive(PartialEq, PartialOrd)] + struct ComparableCreateExternalCatalog<'a> { + pub catalog_name: &'a String, + pub catalog_type: &'a String, + pub location: &'a Option, + pub if_not_exists: &'a bool, + pub or_replace: &'a bool, + } + let comparable_self = ComparableCreateExternalCatalog { + catalog_name: &self.catalog_name, + catalog_type: &self.catalog_type, + location: &self.location, + if_not_exists: &self.if_not_exists, + or_replace: &self.or_replace, + }; + let comparable_other = ComparableCreateExternalCatalog { + catalog_name: &other.catalog_name, + catalog_type: &other.catalog_type, + location: &other.location, + if_not_exists: &other.if_not_exists, + or_replace: &other.or_replace, + }; + comparable_self + .partial_cmp(&comparable_other) + // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields + .filter(|cmp| *cmp != Ordering::Equal || self == other) + } +} + /// Creates a schema. #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct CreateCatalogSchema { @@ -627,6 +712,29 @@ impl PartialOrd for DropCatalogSchema { } } +/// Drops a catalog previously created with `CREATE EXTERNAL CATALOG`. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct DropCatalog { + /// The catalog name + pub name: String, + /// If the catalog exists + pub if_exists: bool, + /// Dummy schema + pub schema: DFSchemaRef, +} + +// Manual implementation needed because of `schema` field. Comparison excludes this field. +impl PartialOrd for DropCatalog { + fn partial_cmp(&self, other: &Self) -> Option { + match self.name.partial_cmp(&other.name) { + Some(Ordering::Equal) => self.if_exists.partial_cmp(&other.if_exists), + cmp => cmp, + } + // TODO (https://github.com/apache/datafusion/issues/17477) avoid recomparing all fields + .filter(|cmp| *cmp != Ordering::Equal || self == other) + } +} + /// Arguments passed to the `CREATE FUNCTION` statement /// /// These statements are turned into executable functions using [`FunctionFactory`] diff --git a/datafusion/expr/src/logical_plan/mod.rs b/datafusion/expr/src/logical_plan/mod.rs index 98113d12c1b4a..7f2208ae12938 100644 --- a/datafusion/expr/src/logical_plan/mod.rs +++ b/datafusion/expr/src/logical_plan/mod.rs @@ -32,9 +32,10 @@ pub use builder::{ union, wrap_projection_for_join_if_necessary, }; pub use ddl::{ - CreateCatalog, CreateCatalogSchema, CreateExternalTable, CreateFunction, - CreateFunctionBody, CreateIndex, CreateMemoryTable, CreateView, DdlStatement, - DropCatalogSchema, DropFunction, DropTable, DropView, OperateFunctionArg, + CreateCatalog, CreateCatalogSchema, CreateExternalCatalog, CreateExternalTable, + CreateFunction, CreateFunctionBody, CreateIndex, CreateMemoryTable, CreateView, + DdlStatement, DropCatalog, DropCatalogSchema, DropFunction, DropTable, DropView, + OperateFunctionArg, }; pub use dml::{ DmlStatement, MergeIntoAction, MergeIntoClause, MergeIntoClauseKind, MergeIntoOp, diff --git a/datafusion/expr/src/logical_plan/tree_node.rs b/datafusion/expr/src/logical_plan/tree_node.rs index ee43666736fe6..4431f2b719b80 100644 --- a/datafusion/expr/src/logical_plan/tree_node.rs +++ b/datafusion/expr/src/logical_plan/tree_node.rs @@ -325,10 +325,12 @@ impl TreeNode for LogicalPlan { DdlStatement::CreateExternalTable(_) | DdlStatement::CreateCatalogSchema(_) | DdlStatement::CreateCatalog(_) + | DdlStatement::CreateExternalCatalog(_) | DdlStatement::CreateIndex(_) | DdlStatement::DropTable(_) | DdlStatement::DropView(_) | DdlStatement::DropCatalogSchema(_) + | DdlStatement::DropCatalog(_) | DdlStatement::CreateFunction(_) | DdlStatement::DropFunction(_) => Transformed::no(ddl), } diff --git a/datafusion/optimizer/src/optimizer.rs b/datafusion/optimizer/src/optimizer.rs index ca4a6688b50c5..12a030c1b2081 100644 --- a/datafusion/optimizer/src/optimizer.rs +++ b/datafusion/optimizer/src/optimizer.rs @@ -478,10 +478,12 @@ fn map_children_mut Result>( | LogicalPlan::Ddl(DdlStatement::CreateExternalTable(_)) | LogicalPlan::Ddl(DdlStatement::CreateCatalogSchema(_)) | LogicalPlan::Ddl(DdlStatement::CreateCatalog(_)) + | LogicalPlan::Ddl(DdlStatement::CreateExternalCatalog(_)) | LogicalPlan::Ddl(DdlStatement::CreateIndex(_)) | LogicalPlan::Ddl(DdlStatement::DropTable(_)) | LogicalPlan::Ddl(DdlStatement::DropView(_)) | LogicalPlan::Ddl(DdlStatement::DropCatalogSchema(_)) + | LogicalPlan::Ddl(DdlStatement::DropCatalog(_)) | LogicalPlan::Ddl(DdlStatement::CreateFunction(_)) | LogicalPlan::Ddl(DdlStatement::DropFunction(_)) | LogicalPlan::Statement(_) => false, diff --git a/datafusion/proto/src/logical_plan/mod.rs b/datafusion/proto/src/logical_plan/mod.rs index 647bceeea15cf..1d3f946d5671c 100644 --- a/datafusion/proto/src/logical_plan/mod.rs +++ b/datafusion/proto/src/logical_plan/mod.rs @@ -2137,6 +2137,12 @@ impl AsLogicalPlan for LogicalPlanNode { LogicalPlan::Ddl(DdlStatement::DropCatalogSchema(_)) => Err(proto_error( "LogicalPlan serde is not yet implemented for DropCatalogSchema", )), + LogicalPlan::Ddl(DdlStatement::CreateExternalCatalog(_)) => Err(proto_error( + "LogicalPlan serde is not yet implemented for CreateExternalCatalog", + )), + LogicalPlan::Ddl(DdlStatement::DropCatalog(_)) => Err(proto_error( + "LogicalPlan serde is not yet implemented for DropCatalog", + )), LogicalPlan::Ddl(DdlStatement::CreateFunction(_)) => Err(proto_error( "LogicalPlan serde is not yet implemented for CreateFunction", )), diff --git a/datafusion/session/src/catalog.rs b/datafusion/session/src/catalog.rs index bd9eb781abe77..b78005aa188c5 100644 --- a/datafusion/session/src/catalog.rs +++ b/datafusion/session/src/catalog.rs @@ -20,8 +20,11 @@ use std::fmt::Debug; use std::sync::Arc; pub use crate::schema::SchemaProvider; +use crate::session::Session; +use async_trait::async_trait; use datafusion_common::Result; use datafusion_common::not_impl_err; +use datafusion_expr::CreateExternalCatalog; /// A catalog list that contains no catalogs. /// @@ -46,6 +49,13 @@ impl CatalogProviderList for EmptyCatalogProviderList { fn catalog(&self, _name: &str) -> Option> { None } + + fn deregister_catalog( + &self, + _name: &str, + ) -> Result>> { + Ok(None) + } } /// Represents a catalog, comprising a number of named schemas. @@ -206,6 +216,17 @@ pub trait CatalogProviderList: Any + Debug + Sync + Send { catalog: Arc, ) -> Option>; + /// Removes a catalog from this list, returning it if it existed. + /// + /// Implementations of this method should return `Ok(None)` if no catalog + /// with `name` exists. + /// + /// By default returns a "Not Implemented" error + fn deregister_catalog(&self, name: &str) -> Result>> { + let _ = name; + not_impl_err!("Deregistering catalogs is not supported") + } + /// Retrieves the list of available catalog names fn catalog_names(&self) -> Vec; @@ -233,6 +254,17 @@ impl dyn CatalogProviderList { } } +/// A factory which creates [`CatalogProvider`]s at runtime given a URL. +#[async_trait] +pub trait CatalogProviderFactory: Debug + Sync + Send { + /// Create a [`CatalogProvider`] using the given `cmd` + async fn create( + &self, + state: &dyn Session, + cmd: &CreateExternalCatalog, + ) -> Result>; +} + #[cfg(test)] mod tests { use super::{CatalogProviderList, EmptyCatalogProviderList}; diff --git a/datafusion/session/src/lib.rs b/datafusion/session/src/lib.rs index 6f7cfb7792c73..26c675c20f928 100644 --- a/datafusion/session/src/lib.rs +++ b/datafusion/session/src/lib.rs @@ -53,7 +53,8 @@ pub mod session; pub mod table; pub use crate::catalog::{ - CatalogProvider, CatalogProviderList, EmptyCatalogProviderList, + CatalogProvider, CatalogProviderFactory, CatalogProviderList, + EmptyCatalogProviderList, }; pub use crate::physical_optimizer::{PhysicalOptimizerContext, PhysicalOptimizerRule}; pub use crate::planner::{ diff --git a/datafusion/sql/src/parser.rs b/datafusion/sql/src/parser.rs index fcf4708f1bf94..675304adc111a 100644 --- a/datafusion/sql/src/parser.rs +++ b/datafusion/sql/src/parser.rs @@ -309,6 +309,70 @@ impl fmt::Display for CreateExternalTable { } } +/// DataFusion extension `CREATE EXTERNAL CATALOG` statement. +/// +/// ```sql +/// CREATE [OR REPLACE] EXTERNAL CATALOG [IF NOT EXISTS] +/// STORED AS +/// [ LOCATION ] +/// [ OPTIONS () ] +/// +/// := ( , , ...) +/// ``` +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CreateExternalCatalog { + /// Catalog name + pub catalog_name: ObjectName, + /// The key used to look up the registered `CatalogProviderFactory` + pub catalog_type: String, + /// The physical location of the catalog, if applicable + pub location: Option, + /// Option to not error if catalog already exists + pub if_not_exists: bool, + /// Option to replace the catalog if it already exists + pub or_replace: bool, + /// Catalog(provider) specific options + pub options: Vec<(String, Value)>, +} + +impl fmt::Display for CreateExternalCatalog { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "CREATE EXTERNAL CATALOG ")?; + if self.if_not_exists { + write!(f, "IF NOT EXISTS ")?; + } + write!(f, "{} ", self.catalog_name)?; + write!(f, "STORED AS {}", self.catalog_type)?; + if let Some(location) = &self.location { + write!( + f, + " LOCATION {}", + Value::SingleQuotedString(location.clone()) + )?; + } + Ok(()) + } +} + +/// DataFusion extension `DROP CATALOG` statement. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct DropCatalog { + /// Catalog name + pub name: ObjectName, + /// Option to not error if the catalog does not exist + pub if_exists: bool, +} + +impl fmt::Display for DropCatalog { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "DROP CATALOG ")?; + if self.if_exists { + write!(f, "IF EXISTS ")?; + } + write!(f, "{}", self.name) + } +} + /// DataFusion extension for `RESET` #[derive(Debug, Clone, PartialEq, Eq)] pub enum ResetStatement { @@ -337,6 +401,10 @@ pub enum Statement { Statement(Box), /// Extension: `CREATE EXTERNAL TABLE` CreateExternalTable(CreateExternalTable), + /// Extension: `CREATE EXTERNAL CATALOG` + CreateExternalCatalog(CreateExternalCatalog), + /// Extension: `DROP CATALOG` + DropCatalog(DropCatalog), /// Extension: `COPY TO` CopyTo(CopyToStatement), /// EXPLAIN for extensions @@ -350,6 +418,8 @@ impl fmt::Display for Statement { match self { Statement::Statement(stmt) => write!(f, "{stmt}"), Statement::CreateExternalTable(stmt) => write!(f, "{stmt}"), + Statement::CreateExternalCatalog(stmt) => write!(f, "{stmt}"), + Statement::DropCatalog(stmt) => write!(f, "{stmt}"), Statement::CopyTo(stmt) => write!(f, "{stmt}"), Statement::Explain(stmt) => write!(f, "{stmt}"), Statement::Reset(stmt) => write!(f, "{stmt}"), @@ -635,6 +705,13 @@ impl<'a> DFParser<'a> { self.parser.next_token(); // RESET self.parse_reset() } + Keyword::DROP + if self + .parser + .parse_keywords(&[Keyword::DROP, Keyword::CATALOG]) => + { + self.parse_drop_catalog() + } _ => { // use sqlparser-rs parser self.parse_and_handle_statement() @@ -932,26 +1009,47 @@ impl<'a> DFParser<'a> { .parser .parse_keywords(&[Keyword::OR, Keyword::REPLACE, Keyword::EXTERNAL]) { - self.parse_create_external_table(false, true) + self.parse_create_external(false, true) } else if self.parser.parse_keywords(&[ Keyword::OR, Keyword::REPLACE, Keyword::UNBOUNDED, Keyword::EXTERNAL, ]) { - self.parse_create_external_table(true, true) + self.parse_create_external(true, true) } else if self.parser.parse_keyword(Keyword::EXTERNAL) { - self.parse_create_external_table(false, false) + self.parse_create_external(false, false) } else if self .parser .parse_keywords(&[Keyword::UNBOUNDED, Keyword::EXTERNAL]) { - self.parse_create_external_table(true, false) + self.parse_create_external(true, false) } else { Ok(Statement::Statement(Box::from(self.parser.parse_create()?))) } } + /// Dispatches `CREATE [OR REPLACE] [UNBOUNDED] EXTERNAL ...` (with the + /// leading keywords already consumed) to either `CREATE EXTERNAL TABLE` + /// or `CREATE EXTERNAL CATALOG`, based on the keyword that follows + /// `EXTERNAL`. + fn parse_create_external( + &mut self, + unbounded: bool, + or_replace: bool, + ) -> Result { + if self.parser.parse_keyword(Keyword::CATALOG) { + if unbounded { + return parser_err!( + "UNBOUNDED is not supported for CREATE EXTERNAL CATALOG" + ); + } + self.parse_create_external_catalog(or_replace) + } else { + self.parse_create_external_table(unbounded, or_replace) + } + } + fn parse_partitions(&mut self) -> Result, DataFusionError> { let mut partitions: Vec = vec![]; if !self.parser.consume_token(&Token::LParen) @@ -1248,6 +1346,87 @@ impl<'a> DFParser<'a> { Ok(Statement::CreateExternalTable(create)) } + /// Parses a `CREATE EXTERNAL CATALOG` statement, with `CREATE [OR + /// REPLACE] EXTERNAL CATALOG` already consumed. + fn parse_create_external_catalog( + &mut self, + or_replace: bool, + ) -> Result { + let if_not_exists = + self.parser + .parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]); + + if if_not_exists && or_replace { + return parser_err!("'IF NOT EXISTS' cannot coexist with 'REPLACE'"); + } + + let catalog_name = self.parser.parse_object_name(true)?; + + #[derive(Default)] + struct Builder { + catalog_type: Option, + location: Option, + options: Option>, + } + let mut builder = Builder::default(); + + loop { + if let Some(keyword) = self.parser.parse_one_of_keywords(&[ + Keyword::STORED, + Keyword::LOCATION, + Keyword::OPTIONS, + ]) { + match keyword { + Keyword::STORED => { + self.parser.expect_keyword(Keyword::AS)?; + ensure_not_set(builder.catalog_type.as_ref(), "STORED AS")?; + builder.catalog_type = Some(self.parse_file_format()?); + } + Keyword::LOCATION => { + ensure_not_set(builder.location.as_ref(), "LOCATION")?; + builder.location = Some(self.parser.parse_literal_string()?); + } + Keyword::OPTIONS => { + ensure_not_set(builder.options.as_ref(), "OPTIONS")?; + builder.options = Some(self.parse_value_options()?); + } + _ => { + unreachable!() + } + } + } else { + let token = self.parser.peek_token(); + if token == Token::EOF || token == Token::SemiColon { + break; + } else { + return self.expected("end of statement or ;", &token)?; + } + } + } + + let Some(catalog_type) = builder.catalog_type else { + return sql_err!(ParserError::ParserError( + "Missing STORED AS clause in CREATE EXTERNAL CATALOG statement".into(), + )); + }; + + Ok(Statement::CreateExternalCatalog(CreateExternalCatalog { + catalog_name, + catalog_type, + location: builder.location, + if_not_exists, + or_replace, + options: builder.options.unwrap_or_default(), + })) + } + + /// Parses a `DROP CATALOG` statement, with `DROP CATALOG` already consumed. + fn parse_drop_catalog(&mut self) -> Result { + let if_exists = self.parser.parse_keywords(&[Keyword::IF, Keyword::EXISTS]); + let name = self.parser.parse_object_name(true)?; + Ok(Statement::DropCatalog(DropCatalog { name, if_exists })) + } + /// Parses one or more external table locations. fn parse_locations(&mut self) -> Result, DataFusionError> { if !self.parser.consume_token(&Token::LParen) { @@ -1830,6 +2009,113 @@ mod tests { Ok(()) } + fn make_create_external_catalog(catalog_type: &str) -> CreateExternalCatalog { + CreateExternalCatalog { + catalog_name: ObjectName::from(vec![Ident::from("c")]), + catalog_type: catalog_type.to_string(), + location: None, + if_not_exists: false, + or_replace: false, + options: vec![], + } + } + + #[test] + fn create_external_catalog() -> Result<(), DataFusionError> { + // minimal: just STORED AS + let sql = "CREATE EXTERNAL CATALOG c STORED AS ICEBERG"; + let expected = Statement::CreateExternalCatalog(CreateExternalCatalog { + catalog_type: "ICEBERG".to_string(), + ..make_create_external_catalog("ICEBERG") + }); + expect_parse_ok(sql, expected)?; + + // with LOCATION + let sql = "CREATE EXTERNAL CATALOG c STORED AS ICEBERG LOCATION 's3://bucket/warehouse'"; + let expected = Statement::CreateExternalCatalog(CreateExternalCatalog { + location: Some("s3://bucket/warehouse".to_string()), + ..make_create_external_catalog("ICEBERG") + }); + expect_parse_ok(sql, expected)?; + + // with OPTIONS + let sql = "CREATE EXTERNAL CATALOG c STORED AS ICEBERG OPTIONS ('catalog.uri' 'http://rest:8181', 'warehouse' 'c')"; + let expected = Statement::CreateExternalCatalog(CreateExternalCatalog { + options: vec![ + ( + "catalog.uri".into(), + Value::SingleQuotedString("http://rest:8181".into()), + ), + ("warehouse".into(), Value::SingleQuotedString("c".into())), + ], + ..make_create_external_catalog("ICEBERG") + }); + expect_parse_ok(sql, expected)?; + + // IF NOT EXISTS + let sql = "CREATE EXTERNAL CATALOG IF NOT EXISTS c STORED AS ICEBERG"; + let expected = Statement::CreateExternalCatalog(CreateExternalCatalog { + if_not_exists: true, + ..make_create_external_catalog("ICEBERG") + }); + expect_parse_ok(sql, expected)?; + + // OR REPLACE + let sql = "CREATE OR REPLACE EXTERNAL CATALOG c STORED AS ICEBERG"; + let expected = Statement::CreateExternalCatalog(CreateExternalCatalog { + or_replace: true, + ..make_create_external_catalog("ICEBERG") + }); + expect_parse_ok(sql, expected)?; + + // IF NOT EXISTS and OR REPLACE cannot coexist + expect_parse_error( + "CREATE OR REPLACE EXTERNAL CATALOG IF NOT EXISTS c STORED AS ICEBERG", + "'IF NOT EXISTS' cannot coexist with 'REPLACE'", + ); + + // missing STORED AS + expect_parse_error( + "CREATE EXTERNAL CATALOG c", + "Missing STORED AS clause in CREATE EXTERNAL CATALOG statement", + ); + + // UNBOUNDED is not applicable to catalogs + expect_parse_error( + "CREATE UNBOUNDED EXTERNAL CATALOG c STORED AS ICEBERG", + "UNBOUNDED is not supported for CREATE EXTERNAL CATALOG", + ); + + Ok(()) + } + + #[test] + fn drop_catalog() -> Result<(), DataFusionError> { + let sql = "DROP CATALOG c"; + let expected = Statement::DropCatalog(DropCatalog { + name: ObjectName::from(vec![Ident::from("c")]), + if_exists: false, + }); + expect_parse_ok(sql, expected)?; + + let sql = "DROP CATALOG IF EXISTS c"; + let expected = Statement::DropCatalog(DropCatalog { + name: ObjectName::from(vec![Ident::from("c")]), + if_exists: true, + }); + expect_parse_ok(sql, expected)?; + + // DROP TABLE / VIEW / SCHEMA are unaffected by the DROP EXTERNAL + // CATALOG dispatch and continue to use the native parser. + let sql = "DROP TABLE t"; + assert!(matches!( + DFParser::parse_sql(sql)?.pop_front(), + Some(Statement::Statement(_)) + )); + + Ok(()) + } + #[test] fn copy_to_table_to_table() -> Result<(), DataFusionError> { // positive case diff --git a/datafusion/sql/src/resolve.rs b/datafusion/sql/src/resolve.rs index d1c172502ff11..f1e8d117d6582 100644 --- a/datafusion/sql/src/resolve.rs +++ b/datafusion/sql/src/resolve.rs @@ -186,6 +186,8 @@ fn visit_statement(statement: &DFStatement, visitor: &mut RelationVisitor) -> Re visit_statement(&explain.statement, visitor)?; } DFStatement::Reset(_) => {} + // Catalogs are not tables, so there is nothing to resolve here. + DFStatement::CreateExternalCatalog(_) | DFStatement::DropCatalog(_) => {} } Ok(()) } diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs index 1a9072212f2f3..dd5e47d3c61d4 100644 --- a/datafusion/sql/src/statement.rs +++ b/datafusion/sql/src/statement.rs @@ -21,8 +21,8 @@ use std::str::FromStr; use std::sync::Arc; use crate::parser::{ - CopyToSource, CopyToStatement, CreateExternalTable, DFParser, ExplainStatement, - LexOrdering, ResetStatement, Statement as DFStatement, + CopyToSource, CopyToStatement, CreateExternalCatalog, CreateExternalTable, DFParser, + DropCatalog, ExplainStatement, LexOrdering, ResetStatement, Statement as DFStatement, }; use crate::planner::{ ContextProvider, PlannerContext, SqlToRel, object_name_to_qualifier, @@ -49,13 +49,14 @@ use datafusion_expr::logical_plan::builder::project; use datafusion_expr::utils::expr_to_columns; use datafusion_expr::{ Analyze, CreateCatalog, CreateCatalogSchema, + CreateExternalCatalog as PlanCreateExternalCatalog, CreateExternalTable as PlanCreateExternalTable, CreateFunction, CreateFunctionBody, CreateIndex as PlanCreateIndex, CreateMemoryTable, CreateView, Deallocate, - DescribeTable, DmlStatement, DropCatalogSchema, DropFunction, DropTable, DropView, - EmptyRelation, Execute, Explain, ExplainFormat, Expr, ExprSchemable, Filter, - LogicalPlan, LogicalPlanBuilder, OperateFunctionArg, PlanType, Prepare, - ResetVariable, SetVariable, SortExpr, Statement as PlanStatement, ToStringifiedPlan, - TransactionAccessMode, TransactionConclusion, TransactionEnd, + DescribeTable, DmlStatement, DropCatalog as PlanDropCatalog, DropCatalogSchema, + DropFunction, DropTable, DropView, EmptyRelation, Execute, Explain, ExplainFormat, + Expr, ExprSchemable, Filter, LogicalPlan, LogicalPlanBuilder, OperateFunctionArg, + PlanType, Prepare, ResetVariable, SetVariable, SortExpr, Statement as PlanStatement, + ToStringifiedPlan, TransactionAccessMode, TransactionConclusion, TransactionEnd, TransactionIsolationLevel, TransactionStart, Volatility, WriteOp, cast, }; use sqlparser::ast::{ @@ -229,6 +230,8 @@ impl SqlToRel<'_, S> { pub fn statement_to_plan(&self, statement: DFStatement) -> Result { match statement { DFStatement::CreateExternalTable(s) => self.external_table_to_plan(s), + DFStatement::CreateExternalCatalog(s) => self.external_catalog_to_plan(s), + DFStatement::DropCatalog(s) => self.drop_catalog_to_plan(s), DFStatement::Statement(s) => self.sql_statement_to_plan(*s), DFStatement::CopyTo(s) => self.copy_to_plan(s), DFStatement::Explain(ExplainStatement { options, statement }) => { @@ -761,31 +764,32 @@ impl SqlToRel<'_, S> { // We don't support cascade and purge for now. // nor do we support multiple object names let name = match names.len() { - 0 => Err(ParserError("Missing table name.".to_string()).into()), - 1 => self.object_name_to_table_reference(names.pop().unwrap()), - _ => { - Err(ParserError("Multiple objects not supported".to_string()) - .into()) - } + 0 => Err::<_, DataFusionError>( + ParserError("Missing table name.".to_string()).into(), + ), + 1 => Ok(names.pop().unwrap()), + _ => Err::<_, DataFusionError>( + ParserError("Multiple objects not supported".to_string()).into(), + ), }?; match object_type { ObjectType::Table => { Ok(LogicalPlan::Ddl(DdlStatement::DropTable(DropTable { - name, + name: self.object_name_to_table_reference(name)?, if_exists, schema: DFSchemaRef::new(DFSchema::empty()), }))) } ObjectType::View => { Ok(LogicalPlan::Ddl(DdlStatement::DropView(DropView { - name, + name: self.object_name_to_table_reference(name)?, if_exists, schema: DFSchemaRef::new(DFSchema::empty()), }))) } ObjectType::Schema => { - let name = match name { + let name = match self.object_name_to_table_reference(name)? { TableReference::Bare { table } => { Ok(SchemaReference::Bare { schema: table }) } @@ -812,6 +816,13 @@ impl SqlToRel<'_, S> { }, ))) } + ObjectType::Database => Ok(LogicalPlan::Ddl( + DdlStatement::DropCatalog(datafusion_expr::DropCatalog { + name: object_name_to_string(&name), + if_exists, + schema: DFSchemaRef::new(DFSchema::empty()), + }), + )), _ => not_impl_err!( "Only `DROP TABLE/VIEW/SCHEMA ...` statement is supported currently" ), @@ -1884,6 +1895,54 @@ impl SqlToRel<'_, S> { ))) } + fn external_catalog_to_plan( + &self, + statement: CreateExternalCatalog, + ) -> Result { + let CreateExternalCatalog { + catalog_name, + catalog_type, + location, + if_not_exists, + or_replace, + options, + } = statement; + + let mut options_map = HashMap::with_capacity(options.len()); + for (key, value) in options { + if options_map.contains_key(&key) { + return plan_err!("Option {key} is specified multiple times"); + } + let Some(value_string) = crate::utils::value_to_string(&value) else { + return plan_err!("Unsupported Value {}", value); + }; + options_map.insert(key, value_string); + } + + Ok(LogicalPlan::Ddl(DdlStatement::CreateExternalCatalog( + Box::new(PlanCreateExternalCatalog { + catalog_name: object_name_to_string(&catalog_name), + catalog_type, + location, + if_not_exists, + or_replace, + options: options_map, + schema: Arc::new(DFSchema::empty()), + }), + ))) + } + + fn drop_catalog_to_plan(&self, statement: DropCatalog) -> Result { + let DropCatalog { name, if_exists } = statement; + Ok(LogicalPlan::Ddl(DdlStatement::DropCatalog( + PlanDropCatalog { + name: object_name_to_string(&name), + if_exists, + schema: DFSchemaRef::new(DFSchema::empty()), + }, + ))) + } + /// Get the indices of the constraint columns in the schema. /// If any column is not found, return an error. fn get_constraint_column_indices( diff --git a/docs/source/library-user-guide/catalogs.md b/docs/source/library-user-guide/catalogs.md index 3c8238b73edbb..dea4e413e2089 100644 --- a/docs/source/library-user-guide/catalogs.md +++ b/docs/source/library-user-guide/catalogs.md @@ -305,6 +305,74 @@ impl CatalogProviderList for MemoryCatalogProviderList { Like other traits, it also maintains the mapping of the Catalog's name to the CatalogProvider. +## Catalog Provider Factories + +The catalogs above are all registered programmatically, ahead of time, +before a query ever runs. Sometimes it is useful to let users attach a +catalog dynamically from SQL instead — for example, a catalog backed by a +remote catalog service such as an Iceberg REST catalog. This is exactly +analogous to how [`TableProviderFactory`] lets `CREATE EXTERNAL TABLE` +create a `TableProvider` "on the fly": `CatalogProviderFactory` lets +`CREATE EXTERNAL CATALOG ... STORED AS ...` create a +`CatalogProvider` on the fly. + +To support `CREATE EXTERNAL CATALOG`, implement `CatalogProviderFactory`: + +```rust +use std::sync::Arc; +use async_trait::async_trait; +use datafusion::catalog::{CatalogProvider, CatalogProviderFactory, MemoryCatalogProvider, Session}; +use datafusion::common::Result; +use datafusion::logical_expr::CreateExternalCatalog; + +#[derive(Debug)] +struct MyCatalogProviderFactory {} + +#[async_trait] +impl CatalogProviderFactory for MyCatalogProviderFactory { + async fn create( + &self, + _state: &dyn Session, + cmd: &CreateExternalCatalog, + ) -> Result> { + // `cmd.location` and `cmd.options` carry whatever was supplied in the + // `LOCATION` and `OPTIONS` clauses of the SQL statement; a real + // implementation would use them to connect to the remote catalog + // service. This example just returns an empty in-memory catalog. + Ok(Arc::new(MemoryCatalogProvider::new())) + } +} +``` + +Then register it on the `SessionState`, keyed by the `STORED AS` value that +should resolve to it: + +```rust,ignore +use datafusion::execution::session_state::SessionStateBuilder; + +let state = SessionStateBuilder::new() + .with_default_features() + .with_catalog_factory("MY_CATALOG_TYPE".to_string(), Arc::new(MyCatalogProviderFactory {})) + .build(); +``` + +With the factory registered, users can attach and detach the catalog from +SQL: + +```sql +CREATE EXTERNAL CATALOG my_catalog +STORED AS MY_CATALOG_TYPE +LOCATION 's3://bucket/warehouse' +OPTIONS ('warehouse' 'my_catalog'); + +DROP CATALOG my_catalog; +``` + +See [DDL: `CREATE EXTERNAL CATALOG`](../user-guide/sql/ddl.md#create-external-catalog) +for the full SQL syntax. + +[`tableproviderfactory`]: https://docs.rs/datafusion/latest/datafusion/catalog/trait.TableProviderFactory.html + ## Recap To recap, you need to: diff --git a/docs/source/user-guide/sql/ddl.md b/docs/source/user-guide/sql/ddl.md index 0d76775bcc1c6..9b5345416eb60 100644 --- a/docs/source/user-guide/sql/ddl.md +++ b/docs/source/user-guide/sql/ddl.md @@ -35,6 +35,58 @@ CREATE DATABASE [ IF NOT EXISTS ] catalog CREATE DATABASE cat; ``` +## CREATE EXTERNAL CATALOG + +`CREATE EXTERNAL CATALOG` registers a catalog built by a registered +[`CatalogProviderFactory`], such as a catalog backed by a remote catalog +service (for example, an Iceberg REST catalog), so that it can be queried +alongside DataFusion's built-in catalogs. A `CatalogProviderFactory` must +first be registered on the `SessionState` with a key matching the +`STORED AS` clause below — see the [Catalog Provider Factories] section of +the Library User Guide for how to implement and register one. + +The supported syntax is: + +```sql +CREATE [OR REPLACE] EXTERNAL CATALOG +[ IF NOT EXISTS ] + +STORED AS +[ LOCATION ] +[ OPTIONS () ] + + := ( , , ...) +``` + +`catalog_type` identifies which registered `CatalogProviderFactory` to +invoke; it is looked up the same way `file_type` is for +[`CREATE EXTERNAL TABLE`](#create-external-table). + +```sql +CREATE EXTERNAL CATALOG my_catalog +STORED AS ICEBERG +LOCATION 's3://bucket/warehouse' +OPTIONS ('catalog.uri' 'http://rest-catalog:8181', 'warehouse' 'my_catalog'); +``` + +[`catalogproviderfactory`]: https://docs.rs/datafusion/latest/datafusion/catalog/trait.CatalogProviderFactory.html +[catalog provider factories]: ../../library-user-guide/catalogs.md#catalog-provider-factories + +## DROP CATALOG + +Removes a catalog previously registered with `CREATE EXTERNAL CATALOG` from +DataFusion's catalog list. + +
+DROP CATALOG [ IF EXISTS ] catalog_name;
+
+ +```sql +DROP CATALOG my_catalog; +-- or use 'if exists' to silently ignore if the catalog doesn't exist +DROP CATALOG IF EXISTS nonexistent_catalog; +``` + ## CREATE SCHEMA Create schema under specified catalog, or the default DataFusion catalog if not specified. diff --git a/parquet-testing b/parquet-testing index 107b36603e051..4b1ce4502afff 160000 --- a/parquet-testing +++ b/parquet-testing @@ -1 +1 @@ -Subproject commit 107b36603e051aee26bd93e04b871034f6c756c0 +Subproject commit 4b1ce4502afff8d20c9b4bb08d07e04e21cdeff3 diff --git a/testing b/testing index 7df2b70baf4f0..9ff285c88565f 160000 --- a/testing +++ b/testing @@ -1 +1 @@ -Subproject commit 7df2b70baf4f081ebf8e0c6bd22745cf3cbfd824 +Subproject commit 9ff285c88565f0f6abc855918c6a342e70e4909c