You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/24 18:56:08 UTC
(datafusion) 01/01: Add an extension point to allow SQL based registration of external catalogs (#25112)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-25112-e640ae03b423d8ec3a2122e5383f5c74ec2861a4
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 975c0f0a48fd9706dc49c79eba3fdb4e7f44f806
Author: Pepijn Van Eeckhoudt <pe...@vaneeckhoudt.net>
AuthorDate: Thu Sep 24 18:55:42 2026 +0000
Add an extension point to allow SQL based registration of external catalogs (#25112)
## Which issue does this PR close?
- Closes #25111.
## Rationale for this change
Adding a `CatalogProviderFactory` trait, similar to
`TableProviderFactory`, makes it possible for extensions to register
support for external catalogs that can then be used by consumers without
requiring additional code. This is quite convenient when working with
open table formats via catalogs.
## What changes are included in this PR?
- Adds a `CatalogProviderFactory` trait similar to
`TableProviderFactory`
- Adds `CREATE EXTERNAL CATALOG` and `DROP CATALOG` support
## What is the testing strategy for this PR?
- Covered by unit and integration tests
## Are there any user-facing changes?
Yes, the supported DDL has been extended. Documentation has been updated
to reflect this.
---------
Co-authored-by: Andrew Lamb <an...@nerdnetworks.org>
---
datafusion-cli/src/catalog.rs | 8 +
datafusion-examples/examples/data_io/catalog.rs | 15 +
.../examples/sql_ops/custom_sql_parser.rs | 87 +++--
datafusion/catalog/src/catalog.rs | 4 +-
datafusion/catalog/src/dynamic_file/catalog.rs | 8 +
datafusion/catalog/src/memory/catalog.rs | 38 ++-
datafusion/core/src/execution/context/mod.rs | 94 +++++-
datafusion/core/src/execution/session_state.rs | 67 +++-
datafusion/core/tests/sql/create_drop.rs | 167 +++++++++-
datafusion/expr/src/logical_plan/ddl.rs | 110 +++++++
datafusion/expr/src/logical_plan/mod.rs | 7 +-
datafusion/expr/src/logical_plan/tree_node.rs | 2 +
datafusion/optimizer/src/optimizer.rs | 2 +
datafusion/proto/src/logical_plan/mod.rs | 6 +
datafusion/session/src/catalog.rs | 50 ++-
datafusion/session/src/lib.rs | 3 +-
datafusion/sql/src/parser.rs | 355 ++++++++++++++++++++-
datafusion/sql/src/resolve.rs | 2 +
datafusion/sql/src/statement.rs | 72 ++++-
datafusion/sqllogictest/test_files/ddl.slt | 46 ++-
.../sqllogictest/test_files/information_schema.slt | 2 +-
.../information_schema_multiple_catalogs.slt | 4 +-
docs/source/library-user-guide/catalogs.md | 68 ++++
docs/source/user-guide/sql/ddl.md | 60 +++-
24 files changed, 1176 insertions(+), 101 deletions(-)
diff --git a/datafusion-cli/src/catalog.rs b/datafusion-cli/src/catalog.rs
index ca24da7873..f583651b5f 100644
--- a/datafusion-cli/src/catalog.rs
+++ b/datafusion-cli/src/catalog.rs
@@ -57,6 +57,14 @@ impl CatalogProviderList for DynamicObjectStoreCatalog {
self.inner.register_catalog(name, catalog)
}
+ fn deregister_catalog(
+ &self,
+ name: &str,
+ cascade: bool,
+ ) -> Result<Option<Arc<dyn CatalogProvider>>> {
+ self.inner.deregister_catalog(name, cascade)
+ }
+
fn catalog_names(&self) -> Vec<String> {
self.inner.catalog_names()
}
diff --git a/datafusion-examples/examples/data_io/catalog.rs b/datafusion-examples/examples/data_io/catalog.rs
index 7e5cc5a4cf..acb4798c59 100644
--- a/datafusion-examples/examples/data_io/catalog.rs
+++ b/datafusion-examples/examples/data_io/catalog.rs
@@ -279,6 +279,21 @@ impl CatalogProviderList for CustomCatalogProviderList {
Some(catalog)
}
+ fn deregister_catalog(
+ &self,
+ name: &str,
+ cascade: bool,
+ ) -> Result<Option<Arc<dyn CatalogProvider>>> {
+ let mut cats = self.catalogs.write().unwrap();
+ let Some(cat) = cats.get_mut(name) else {
+ return Ok(None);
+ };
+
+ cat.prepare_deregister_catalog(cascade)?;
+
+ Ok(cats.remove(name))
+ }
+
/// Retrieves the list of available catalog names
fn catalog_names(&self) -> Vec<String> {
let cats = self.catalogs.read().unwrap();
diff --git a/datafusion-examples/examples/sql_ops/custom_sql_parser.rs b/datafusion-examples/examples/sql_ops/custom_sql_parser.rs
index 308a0de62a..ccf511917a 100644
--- a/datafusion-examples/examples/sql_ops/custom_sql_parser.rs
+++ b/datafusion-examples/examples/sql_ops/custom_sql_parser.rs
@@ -16,11 +16,16 @@
// under the License.
//! This example demonstrates extending the DataFusion SQL parser to support
-//! custom DDL statements, specifically `CREATE EXTERNAL CATALOG`.
+//! custom DDL statements, specifically `CREATE FOREIGN CATALOG`.
+//!
+//! Note: DataFusion supports `CREATE EXTERNAL CATALOG` out-of-the-box making use of
+//! [`CatalogProviderFactory`](datafusion::catalog::CatalogProviderFactory). This example
+//! is a partial reimplementation of the existing functionality to demonstrate how to extend the
+//! SQL parser.
//!
//! ### Custom Syntax
//! ```sql
-//! CREATE EXTERNAL CATALOG my_catalog
+//! CREATE FOREIGN CATALOG my_catalog
//! STORED AS ICEBERG
//! LOCATION 's3://my-bucket/warehouse/'
//! OPTIONS (
@@ -59,7 +64,7 @@ use object_store::local::LocalFileSystem;
/// Entry point for the example.
pub async fn custom_sql_parser() -> Result<()> {
- // Use standard Parquet testing data as our "external" source.
+ // Use standard Parquet testing data as our "foreign" source.
let base_path = datafusion::common::test_util::parquet_test_data();
let base_path = std::path::Path::new(&base_path).canonicalize()?;
@@ -71,7 +76,7 @@ pub async fn custom_sql_parser() -> Result<()> {
.unwrap_or_else(|_| base_path.to_string_lossy().to_string());
let create_catalog_sql = format!(
- "CREATE EXTERNAL CATALOG parquet_testing
+ "CREATE FOREIGN CATALOG parquet_testing
STORED AS parquet
LOCATION 'local://workspace/{location}'
OPTIONS (
@@ -90,10 +95,10 @@ pub async fn custom_sql_parser() -> Result<()> {
let err = ctx_standard
.sql(&create_catalog_sql)
.await
- .expect_err("Expected the standard parser to reject CREATE EXTERNAL CATALOG (custom DDL syntax)");
+ .expect_err("Expected the standard parser to reject CREATE FOREIGN CATALOG (custom DDL syntax)");
println!("Error: {err}\n");
- assert_snapshot!(err.to_string(), @r#"SQL error: ParserError("Expected: TABLE, found: CATALOG at Line: 1, Column: 17")"#);
+ assert_snapshot!(err.to_string(), @r#"SQL error: ParserError("Expected: an object type after CREATE, found: FOREIGN at Line: 1, Column: 8")"#);
// =========================================================================
// Part 2: Custom parser handles the statement
@@ -106,11 +111,11 @@ pub async fn custom_sql_parser() -> Result<()> {
let mut parser = CustomParser::new(&create_catalog_sql)?;
let statement = parser.parse_statement()?;
match statement {
- CustomStatement::CreateExternalCatalog(stmt) => {
- handle_create_external_catalog(&ctx, stmt).await?;
+ CustomStatement::CreateForeignCatalog(stmt) => {
+ handle_create_foreign_catalog(&ctx, stmt).await?;
}
CustomStatement::DFStatement(_) => {
- panic!("Expected CreateExternalCatalog statement");
+ panic!("Expected CreateForeignCatalog statement");
}
}
@@ -141,10 +146,10 @@ async fn execute_sql(ctx: &SessionContext, sql: &str) -> Result<String> {
Ok(arrow::util::pretty::pretty_format_batches(&batches)?.to_string())
}
-/// Custom handler for the `CREATE EXTERNAL CATALOG` statement.
-async fn handle_create_external_catalog(
+/// Custom handler for the `CREATE FOREIGN CATALOG` statement.
+async fn handle_create_foreign_catalog(
ctx: &SessionContext,
- stmt: CreateExternalCatalog,
+ stmt: CreateForeignCatalog,
) -> Result<()> {
let factory = ListingTableFactory::new();
let catalog = Arc::new(MemoryCatalogProvider::new());
@@ -240,13 +245,13 @@ async fn handle_create_external_catalog(
pub enum CustomStatement {
/// Standard DataFusion statement
DFStatement(Box<Statement>),
- /// Custom `CREATE EXTERNAL CATALOG` statement
- CreateExternalCatalog(CreateExternalCatalog),
+ /// Custom `CREATE FOREIGN CATALOG` statement
+ CreateForeignCatalog(CreateForeignCatalog),
}
-/// Data structure for `CREATE EXTERNAL CATALOG`.
+/// Data structure for `CREATE FOREIGN CATALOG`.
#[derive(Debug, Clone)]
-pub struct CreateExternalCatalog {
+pub struct CreateForeignCatalog {
pub name: ObjectName,
pub catalog_type: String,
pub location: String,
@@ -257,16 +262,16 @@ impl Display for CustomStatement {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::DFStatement(s) => write!(f, "{s}"),
- Self::CreateExternalCatalog(s) => write!(f, "{s}"),
+ Self::CreateForeignCatalog(s) => write!(f, "{s}"),
}
}
}
-impl Display for CreateExternalCatalog {
+impl Display for CreateForeignCatalog {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
- "CREATE EXTERNAL CATALOG {} STORED AS {} LOCATION '{}'",
+ "CREATE FOREIGN CATALOG {} STORED AS {} LOCATION '{}'",
self.name, self.catalog_type, self.location
)?;
if !self.options.is_empty() {
@@ -296,28 +301,23 @@ impl<'a> CustomParser<'a> {
}
pub fn parse_statement(&mut self) -> Result<CustomStatement> {
- if self.is_create_external_catalog() {
- return self.parse_create_external_catalog();
+ if let Some(cfc) = self.parse_create_foreign_catalog()? {
+ return Ok(CustomStatement::CreateForeignCatalog(cfc));
}
+
Ok(CustomStatement::DFStatement(Box::new(
self.df_parser.parse_statement()?,
)))
}
- fn is_create_external_catalog(&self) -> bool {
- let t1 = &self.df_parser.parser.peek_nth_token(0).token;
- let t2 = &self.df_parser.parser.peek_nth_token(1).token;
- let t3 = &self.df_parser.parser.peek_nth_token(2).token;
-
- matches!(t1, Token::Word(w) if w.keyword == Keyword::CREATE)
- && matches!(t2, Token::Word(w) if w.keyword == Keyword::EXTERNAL)
- && matches!(t3, Token::Word(w) if w.value.to_uppercase() == "CATALOG")
- }
-
- fn parse_create_external_catalog(&mut self) -> Result<CustomStatement> {
- // Consume prefix tokens: CREATE EXTERNAL CATALOG
- for _ in 0..3 {
- self.df_parser.parser.next_token();
+ /// Parses a `CREATE FOREIGN CATALOG` statement.
+ fn parse_create_foreign_catalog(&mut self) -> Result<Option<CreateForeignCatalog>> {
+ if !self.df_parser.parser.parse_keywords(&[
+ Keyword::CREATE,
+ Keyword::FOREIGN,
+ Keyword::CATALOG,
+ ]) {
+ return Ok(None);
}
let name = self
@@ -373,16 +373,13 @@ impl<'a> CustomParser<'a> {
}
}
- Ok(CustomStatement::CreateExternalCatalog(
- CreateExternalCatalog {
- name,
- catalog_type: catalog_type
- .ok_or_else(|| plan_datafusion_err!("Missing STORED AS"))?,
- location: location
- .ok_or_else(|| plan_datafusion_err!("Missing LOCATION"))?,
- options,
- },
- ))
+ Ok(Some(CreateForeignCatalog {
+ name,
+ catalog_type: catalog_type
+ .ok_or_else(|| plan_datafusion_err!("Missing STORED AS"))?,
+ location: location.ok_or_else(|| plan_datafusion_err!("Missing LOCATION"))?,
+ options,
+ }))
}
/// Parse options in the form: (key [=] value, key [=] value, ...)
diff --git a/datafusion/catalog/src/catalog.rs b/datafusion/catalog/src/catalog.rs
index 07da1293a7..2552ee8676 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 4437d99667..b008c347d6 100644
--- a/datafusion/catalog/src/dynamic_file/catalog.rs
+++ b/datafusion/catalog/src/dynamic_file/catalog.rs
@@ -49,6 +49,14 @@ impl CatalogProviderList for DynamicFileCatalog {
self.inner.register_catalog(name, catalog)
}
+ fn deregister_catalog(
+ &self,
+ name: &str,
+ cascade: bool,
+ ) -> datafusion_common::Result<Option<Arc<dyn CatalogProvider>>> {
+ self.inner.deregister_catalog(name, cascade)
+ }
+
fn catalog_names(&self) -> Vec<String> {
self.inner.catalog_names()
}
diff --git a/datafusion/catalog/src/memory/catalog.rs b/datafusion/catalog/src/memory/catalog.rs
index ebe6b9dfa0..4793e24ceb 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,24 @@ impl CatalogProviderList for MemoryCatalogProviderList {
self.catalogs.insert(name, catalog)
}
+ fn deregister_catalog(
+ &self,
+ name: &str,
+ cascade: bool,
+ ) -> Result<Option<Arc<dyn CatalogProvider>>> {
+ let Some(mut entry) = self.catalogs.get_mut(name) else {
+ return Ok(None);
+ };
+
+ let catalog = entry.value_mut();
+ catalog.prepare_deregister_catalog(cascade)?;
+
+ // Drop entry to release its lock
+ drop(entry);
+
+ Ok(self.catalogs.remove(name).map(|(_, catalog)| catalog))
+ }
+
fn catalog_names(&self) -> Vec<String> {
self.catalogs.iter().map(|c| c.key().clone()).collect()
}
@@ -97,7 +115,7 @@ impl CatalogProvider for MemoryCatalogProvider {
&self,
name: &str,
schema: Arc<dyn SchemaProvider>,
- ) -> datafusion_common::Result<Option<Arc<dyn SchemaProvider>>> {
+ ) -> Result<Option<Arc<dyn SchemaProvider>>> {
Ok(self.schemas.insert(name.into(), schema))
}
@@ -105,7 +123,7 @@ impl CatalogProvider for MemoryCatalogProvider {
&self,
name: &str,
cascade: bool,
- ) -> datafusion_common::Result<Option<Arc<dyn SchemaProvider>>> {
+ ) -> Result<Option<Arc<dyn SchemaProvider>>> {
if let Some(schema) = self.schema(name) {
let table_names = schema.table_names();
match (table_names.is_empty(), cascade) {
@@ -114,7 +132,7 @@ impl CatalogProvider for MemoryCatalogProvider {
Ok(Some(removed))
}
(false, false) => exec_err!(
- "Cannot drop schema {} because other tables depend on it: {}",
+ "Cannot drop schema {} containing tables: {}",
name,
itertools::join(table_names.iter(), ", ")
),
@@ -123,4 +141,16 @@ impl CatalogProvider for MemoryCatalogProvider {
Ok(None)
}
}
+
+ fn prepare_deregister_catalog(&self, cascade: bool) -> Result<()> {
+ if !cascade && !self.schemas.is_empty() {
+ return exec_err!(
+ "Cannot drop catalog containing schemas: {}",
+ itertools::join(self.schema_names().iter(), ", ")
+ );
+ }
+
+ self.schemas.clear();
+ Ok(())
+ }
}
diff --git a/datafusion/core/src/execution/context/mod.rs b/datafusion/core/src/execution/context/mod.rs
index ff1ad25811..1b1174e535 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<Arc<dyn CatalogProviderFactory>> {
+ 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<Box<dyn Future<Output = _> + 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<DataFrame> {
+ 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<Arc<dyn CatalogProvider>> {
+ 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<DataFrame> {
let DropTable {
name, if_exists, ..
@@ -1106,6 +1169,25 @@ impl SessionContext {
exec_err!("Schema '{schema_ref}' doesn't exist.")
}
+ fn drop_catalog(&self, cmd: DropCatalog) -> Result<DataFrame> {
+ let DropCatalog {
+ name,
+ if_exists,
+ cascade,
+ ..
+ } = cmd;
+ let dereg = self
+ .state
+ .write()
+ .catalog_list()
+ .deregister_catalog(&name, cascade)?;
+ 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;
@@ -2768,7 +2850,7 @@ mod tests {
);
// Create catalog
- ctx.sql("CREATE DATABASE test").await?.collect().await?;
+ ctx.sql("CREATE CATALOG test").await?.collect().await?;
// Create schema
ctx.sql("CREATE SCHEMA test.abc").await?.collect().await?;
diff --git a/datafusion/core/src/execution/session_state.rs b/datafusion/core/src/execution/session_state.rs
index aa8ba4c3b7..040a90a244 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<String, Arc<dyn TableProviderFactory>>,
+ /// 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 <TYPE>` statement, for catalogs
+ /// backed by an external implementation.
+ ///
+ /// [`CatalogProvider`]: crate::catalog::CatalogProvider
+ catalog_factories: HashMap<String, Arc<dyn CatalogProviderFactory>>,
/// Runtime environment
runtime_env: Arc<RuntimeEnv>,
/// [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);
@@ -487,13 +500,25 @@ impl SessionState {
&self.inner.table_factories
}
- /// Get the table factories
+ /// Get the table factories mutably
pub fn table_factories_mut(
&mut self,
) -> &mut HashMap<String, Arc<dyn TableProviderFactory>> {
&mut Arc::make_mut(&mut self.inner).table_factories
}
+ /// Get the catalog factories
+ pub fn catalog_factories(&self) -> &HashMap<String, Arc<dyn CatalogProviderFactory>> {
+ &self.inner.catalog_factories
+ }
+
+ /// Get the catalog factories mutably
+ pub fn catalog_factories_mut(
+ &mut self,
+ ) -> &mut HashMap<String, Arc<dyn CatalogProviderFactory>> {
+ &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<TableOptions>,
execution_props: Option<ExecutionProps>,
table_factories: Option<HashMap<String, Arc<dyn TableProviderFactory>>>,
+ catalog_factories: Option<HashMap<String, Arc<dyn CatalogProviderFactory>>>,
runtime_env: Option<Arc<RuntimeEnv>>,
function_factory: Option<Arc<dyn FunctionFactory>>,
cache_factory: Option<Arc<dyn CacheFactory>>,
@@ -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,
@@ -1538,11 +1566,11 @@ impl SessionStateBuilder {
/// Add a [`TableProviderFactory`] to the map of factories
pub fn with_table_factory(
mut self,
- key: String,
+ key: impl Into<String>,
table_factory: Arc<dyn TableProviderFactory>,
) -> Self {
let mut table_factories = self.table_factories.unwrap_or_default();
- table_factories.insert(key, table_factory);
+ table_factories.insert(key.into(), table_factory);
self.table_factories = Some(table_factories);
self
}
@@ -1556,6 +1584,27 @@ impl SessionStateBuilder {
self
}
+ /// Add a [`CatalogProviderFactory`] to the map of factories
+ pub fn with_catalog_factory(
+ mut self,
+ key: impl Into<String>,
+ catalog_factory: Arc<dyn CatalogProviderFactory>,
+ ) -> Self {
+ let mut catalog_factories = self.catalog_factories.unwrap_or_default();
+ catalog_factories.insert(key.into(), 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<String, Arc<dyn CatalogProviderFactory>>,
+ ) -> Self {
+ self.catalog_factories = Some(catalog_factories);
+ self
+ }
+
/// Set the [`RuntimeEnv`]
pub fn with_runtime_env(mut self, runtime_env: Arc<RuntimeEnv>) -> 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<HashMap<String, Arc<dyn CatalogProviderFactory>>> {
+ &mut self.catalog_factories
+ }
+
/// Returns the current runtime_env value
pub fn runtime_env(&mut self) -> &mut Option<Arc<RuntimeEnv>> {
&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 4a60a79ff5..d04ca0f071 100644
--- a/datafusion/core/tests/sql/create_drop.rs
+++ b/datafusion/core/tests/sql/create_drop.rs
@@ -15,18 +15,42 @@
// 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<Arc<dyn CatalogProvider>> {
+ 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();
- state
- .table_factories_mut()
- .insert("DELTATABLE".to_string(), Arc::new(TestTableFactory {}));
- let ctx = SessionContext::new_with_state(state);
+ let ctx: SessionContext = SessionStateBuilder::new()
+ .with_default_features()
+ .with_table_factory("DELTATABLE", Arc::new(TestTableFactory {}))
+ .build()
+ .into();
let sql = "CREATE EXTERNAL TABLE dt STORED AS DELTATABLE LOCATION 's3://bucket/schema/table';";
ctx.sql(sql).await.unwrap();
@@ -41,11 +65,11 @@ async fn create_custom_table() -> Result<()> {
#[tokio::test]
async fn create_external_table_with_ddl() -> Result<()> {
- let mut state = SessionStateBuilder::new().with_default_features().build();
- state
- .table_factories_mut()
- .insert("MOCKTABLE".to_string(), Arc::new(TestTableFactory {}));
- let ctx = SessionContext::new_with_state(state);
+ let ctx: SessionContext = SessionStateBuilder::new()
+ .with_default_features()
+ .with_table_factory("MOCKTABLE", Arc::new(TestTableFactory {}))
+ .build()
+ .into();
let sql = "CREATE EXTERNAL TABLE dt (a_id integer, a_str string, a_bool boolean) STORED AS MOCKTABLE LOCATION 'mockprotocol://path/to/table';";
ctx.sql(sql).await.unwrap();
@@ -89,3 +113,126 @@ async fn create_drop_table() -> Result<()> {
Ok(())
}
+
+#[tokio::test]
+async fn create_external_catalog_with_factory() -> Result<()> {
+ let ctx: SessionContext = SessionStateBuilder::new()
+ .with_default_features()
+ .with_catalog_factory("TESTCATALOG", Arc::new(TestCatalogFactory {}))
+ .build()
+ .into();
+ 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 ctx: SessionContext = SessionStateBuilder::new()
+ .with_default_features()
+ .with_catalog_factory("TESTCATALOG", Arc::new(TestCatalogFactory {}))
+ .build()
+ .into();
+
+ 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 ctx: SessionContext = SessionStateBuilder::new()
+ .with_default_features()
+ .with_catalog_factory("TESTCATALOG", Arc::new(TestCatalogFactory {}))
+ .build()
+ .into();
+
+ 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_external_catalog_or_replace() -> Result<()> {
+ let ctx: SessionContext = SessionStateBuilder::new()
+ .with_default_features()
+ .with_catalog_factory("TESTCATALOG", Arc::new(TestCatalogFactory {}))
+ .build()
+ .into();
+
+ ctx.sql("CREATE EXTERNAL CATALOG cat STORED AS TESTCATALOG LOCATION 's3://x'")
+ .await?;
+ let original = ctx.catalog("cat").unwrap();
+
+ ctx.sql(
+ "CREATE OR REPLACE EXTERNAL CATALOG cat STORED AS TESTCATALOG LOCATION 's3://x'",
+ )
+ .await?;
+ let replacement = ctx.catalog("cat").unwrap();
+
+ assert!(!Arc::ptr_eq(&original, &replacement));
+ Ok(())
+}
+
+#[tokio::test]
+async fn create_drop_catalog() -> Result<()> {
+ let ctx: SessionContext = SessionStateBuilder::new()
+ .with_default_features()
+ .with_catalog_factory("TESTCATALOG", Arc::new(TestCatalogFactory {}))
+ .build()
+ .into();
+
+ 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 51d88e43c1..16c46d31c2 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<CreateExternalCatalog>),
/// 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 (aka "Database").
+ DropCatalog(DropCatalog),
/// Create function statement. Boxed for the same reason as
/// [`Self::CreateExternalTable`] (~288 bytes).
CreateFunction(Box<CreateFunction>),
@@ -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<String>,
+ /// 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<String, String>,
+ /// Dummy schema
+ pub schema: DFSchemaRef,
+}
+
+// Hashing refers to a subset of fields considered in PartialEq.
+impl Hash for CreateExternalCatalog {
+ fn hash<H: Hasher>(&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<Ordering> {
+ #[derive(PartialEq, PartialOrd)]
+ struct ComparableCreateExternalCatalog<'a> {
+ pub catalog_name: &'a String,
+ pub catalog_type: &'a String,
+ pub location: &'a Option<String>,
+ 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,31 @@ 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,
+ /// Whether drop should cascade
+ pub cascade: 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<Ordering> {
+ 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 98113d12c1..7f2208ae12 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 ee43666736..4431f2b719 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 6bf68c7d6d..5973483b25 100644
--- a/datafusion/optimizer/src/optimizer.rs
+++ b/datafusion/optimizer/src/optimizer.rs
@@ -478,10 +478,12 @@ fn map_children_mut<F: FnMut(&mut LogicalPlan) -> Result<bool>>(
| 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 37b42fc9a3..171043c233 100644
--- a/datafusion/proto/src/logical_plan/mod.rs
+++ b/datafusion/proto/src/logical_plan/mod.rs
@@ -2262,6 +2262,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 bd9eb781ab..233cf45998 100644
--- a/datafusion/session/src/catalog.rs
+++ b/datafusion/session/src/catalog.rs
@@ -20,12 +20,15 @@ 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.
///
-/// [`Session`](crate::Session) implementations that do not provide catalog
+/// [`Session`] implementations that do not provide catalog
/// access can return this list explicitly.
#[derive(Debug, Default)]
pub struct EmptyCatalogProviderList;
@@ -46,6 +49,14 @@ impl CatalogProviderList for EmptyCatalogProviderList {
fn catalog(&self, _name: &str) -> Option<Arc<dyn CatalogProvider>> {
None
}
+
+ fn deregister_catalog(
+ &self,
+ _name: &str,
+ _cascade: bool,
+ ) -> Result<Option<Arc<dyn CatalogProvider>>> {
+ Ok(None)
+ }
}
/// Represents a catalog, comprising a number of named schemas.
@@ -171,6 +182,14 @@ pub trait CatalogProvider: Any + Debug + Sync + Send {
) -> Result<Option<Arc<dyn SchemaProvider>>> {
not_impl_err!("Deregistering new schemas is not supported")
}
+
+ /// Called when a catalog is being dropped. Implementations can use this hook
+ /// to prevent a catalog from being dropped if it is not empty.
+ ///
+ /// By default returns a "Not Implemented" error
+ fn prepare_deregister_catalog(&self, _cascade: bool) -> Result<()> {
+ not_impl_err!("Deregistering all schemas is not supported")
+ }
}
impl dyn CatalogProvider {
@@ -206,6 +225,24 @@ pub trait CatalogProviderList: Any + Debug + Sync + Send {
catalog: Arc<dyn CatalogProvider>,
) -> Option<Arc<dyn CatalogProvider>>;
+ /// 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.
+ ///
+ /// If `cascade` is `false`, implementations should return an error if the
+ /// catalog is not empty.
+ ///
+ /// By default returns a "Not Implemented" error
+ fn deregister_catalog(
+ &self,
+ name: &str,
+ cascade: bool,
+ ) -> Result<Option<Arc<dyn CatalogProvider>>> {
+ let _ = (name, cascade);
+ not_impl_err!("Deregistering catalogs is not supported")
+ }
+
/// Retrieves the list of available catalog names
fn catalog_names(&self) -> Vec<String>;
@@ -233,6 +270,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<Arc<dyn CatalogProvider>>;
+}
+
#[cfg(test)]
mod tests {
use super::{CatalogProviderList, EmptyCatalogProviderList};
diff --git a/datafusion/session/src/lib.rs b/datafusion/session/src/lib.rs
index 6f7cfb7792..26c675c20f 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 12c2569740..4435126586 100644
--- a/datafusion/sql/src/parser.rs
+++ b/datafusion/sql/src/parser.rs
@@ -24,7 +24,7 @@ use datafusion_common::DataFusionError;
use datafusion_common::config::{ConfigNonZeroUsize, SqlParserOptions};
use datafusion_common::format::{ExplainFormat, ExplainStatementOptions};
use datafusion_common::{Diagnostic, Span, sql_err};
-use sqlparser::ast::{ExprWithAlias, Ident, OrderByOptions, OrderBySort};
+use sqlparser::ast::{ExprWithAlias, Ident, ObjectType, OrderByOptions, OrderBySort};
use sqlparser::tokenizer::TokenWithSpan;
use sqlparser::{
ast::{
@@ -271,7 +271,11 @@ pub struct CreateExternalTable {
impl fmt::Display for CreateExternalTable {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- write!(f, "CREATE EXTERNAL TABLE ")?;
+ write!(f, "CREATE ")?;
+ if self.or_replace {
+ write!(f, "OR REPLACE ")?;
+ }
+ write!(f, "EXTERNAL TABLE ")?;
if self.if_not_exists {
write!(f, "IF NOT EXISTS ")?;
}
@@ -309,6 +313,76 @@ impl fmt::Display for CreateExternalTable {
}
}
+/// DataFusion extension `CREATE EXTERNAL CATALOG` statement.
+///
+/// ```sql
+/// CREATE [OR REPLACE] EXTERNAL CATALOG [IF NOT EXISTS] <catalog_name>
+/// STORED AS <catalog_type>
+/// [ LOCATION <literal> ]
+/// [ OPTIONS (<key_value_list>) ]
+///
+/// <key_value_list> := (<literal> <literal>, <literal> <literal>, ...)
+/// ```
+#[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<String>,
+ /// 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.or_replace {
+ write!(f, "OR REPLACE ")?;
+ }
+ write!(f, "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,
+ /// Whether drop should cascade
+ pub cascade: 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 +411,8 @@ pub enum Statement {
Statement(Box<SQLStatement>),
/// Extension: `CREATE EXTERNAL TABLE`
CreateExternalTable(CreateExternalTable),
+ /// Extension: `CREATE EXTERNAL CATALOG`
+ CreateExternalCatalog(CreateExternalCatalog),
/// Extension: `COPY TO`
CopyTo(CopyToStatement),
/// EXPLAIN for extensions
@@ -350,6 +426,7 @@ 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::CopyTo(stmt) => write!(f, "{stmt}"),
Statement::Explain(stmt) => write!(f, "{stmt}"),
Statement::Reset(stmt) => write!(f, "{stmt}"),
@@ -639,6 +716,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()
@@ -936,26 +1020,51 @@ 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 if self.parser.parse_keywords(&[Keyword::CATALOG]) {
+ Ok(Statement::Statement(Box::from(
+ self.parser.parse_create_database()?,
+ )))
} 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<Statement, DataFusionError> {
+ 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<Vec<String>, DataFusionError> {
let mut partitions: Vec<String> = vec![];
if !self.parser.consume_token(&Token::LParen)
@@ -1252,6 +1361,103 @@ 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<Statement, DataFusionError> {
+ 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)?;
+ let mut catalog_type: Option<String> = None;
+ let mut location: Option<String> = None;
+ let mut options: Option<Vec<(String, Value)>> = None;
+
+ 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(catalog_type.as_ref(), "STORED AS")?;
+ catalog_type = Some(self.parse_file_format()?);
+ }
+ Keyword::LOCATION => {
+ ensure_not_set(location.as_ref(), "LOCATION")?;
+ location = Some(self.parser.parse_literal_string()?);
+ }
+ Keyword::OPTIONS => {
+ ensure_not_set(options.as_ref(), "OPTIONS")?;
+ 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) = 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,
+ if_not_exists,
+ or_replace,
+ options: options.unwrap_or_default(),
+ }))
+ }
+
+ /// Parses a `DROP CATALOG` statement, with `DROP CATALOG` already consumed.
+ fn parse_drop_catalog(&mut self) -> Result<Statement, DataFusionError> {
+ let if_exists = self.parser.parse_keywords(&[Keyword::IF, Keyword::EXISTS]);
+ let name = self.parser.parse_object_name(true)?;
+
+ let loc = self.parser.peek_token_ref().span.start;
+ let cascade = self.parser.parse_keyword(Keyword::CASCADE);
+ let restrict = self.parser.parse_keyword(Keyword::RESTRICT);
+ if cascade && restrict {
+ return parser_err!(format!(
+ "Cannot specify both CASCADE and RESTRICT in DROP DATABASE {loc}"
+ ));
+ }
+
+ Ok(Statement::Statement(Box::new(
+ sqlparser::ast::Statement::Drop {
+ object_type: ObjectType::Database,
+ if_exists,
+ names: vec![name],
+ cascade,
+ restrict,
+ purge: false,
+ temporary: false,
+ table: None,
+ },
+ )))
+ }
+
/// Parses one or more external table locations.
fn parse_locations(&mut self) -> Result<Vec<String>, DataFusionError> {
if !self.parser.consume_token(&Token::LParen) {
@@ -1843,6 +2049,143 @@ 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::Statement(Box::new(sqlparser::ast::Statement::Drop {
+ object_type: ObjectType::Database,
+ if_exists: false,
+ names: vec![ObjectName::from(vec![Ident::from("c")])],
+ cascade: false,
+ restrict: false,
+ purge: false,
+ temporary: false,
+ table: None,
+ }));
+ expect_parse_ok(sql, expected)?;
+
+ let sql = "DROP CATALOG IF EXISTS c";
+ let expected = Statement::Statement(Box::new(sqlparser::ast::Statement::Drop {
+ object_type: ObjectType::Database,
+ if_exists: true,
+ names: vec![ObjectName::from(vec![Ident::from("c")])],
+ cascade: false,
+ restrict: false,
+ purge: false,
+ temporary: false,
+ table: None,
+ }));
+ expect_parse_ok(sql, expected)?;
+
+ let sql = "DROP CATALOG c CASCADE";
+ let expected = Statement::Statement(Box::new(sqlparser::ast::Statement::Drop {
+ object_type: ObjectType::Database,
+ if_exists: false,
+ names: vec![ObjectName::from(vec![Ident::from("c")])],
+ cascade: true,
+ restrict: false,
+ purge: false,
+ temporary: false,
+ table: None,
+ }));
+ expect_parse_ok(sql, expected)?;
+
+ let sql = "DROP CATALOG IF EXISTS c CASCADE";
+ let expected = Statement::Statement(Box::new(sqlparser::ast::Statement::Drop {
+ object_type: ObjectType::Database,
+ if_exists: true,
+ names: vec![ObjectName::from(vec![Ident::from("c")])],
+ cascade: true,
+ restrict: false,
+ purge: false,
+ temporary: false,
+ table: None,
+ }));
+ expect_parse_ok(sql, expected)?;
+
+ 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 d1c172502f..b6b1a74849 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(_) => {}
}
Ok(())
}
diff --git a/datafusion/sql/src/statement.rs b/datafusion/sql/src/statement.rs
index cdf58327e1..fe0b359839 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,
+ ExplainStatement, LexOrdering, ResetStatement, Statement as DFStatement,
};
use crate::planner::{
ContextProvider, PlannerContext, SqlToRel, object_name_to_qualifier,
@@ -48,6 +48,7 @@ use datafusion_expr::logical_plan::builder::project;
use datafusion_expr::utils::expr_to_columns;
use datafusion_expr::{
Analyze, Cast, CreateCatalog, CreateCatalogSchema,
+ CreateExternalCatalog as PlanCreateExternalCatalog,
CreateExternalTable as PlanCreateExternalTable, CreateFunction, CreateFunctionBody,
CreateIndex as PlanCreateIndex, CreateMemoryTable, CreateView, Deallocate,
DescribeTable, DmlStatement, DropCatalogSchema, DropFunction, DropTable, DropView,
@@ -234,6 +235,7 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
pub fn statement_to_plan(&self, statement: DFStatement) -> Result<LogicalPlan> {
let plan = match statement {
DFStatement::CreateExternalTable(s) => self.external_table_to_plan(s)?,
+ DFStatement::CreateExternalCatalog(s) => self.external_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 }) => {
@@ -788,31 +790,32 @@ impl<S: ContextProvider> 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 })
}
@@ -839,8 +842,16 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
},
)))
}
+ ObjectType::Database => Ok(LogicalPlan::Ddl(
+ DdlStatement::DropCatalog(datafusion_expr::DropCatalog {
+ name: object_name_to_string(&name),
+ if_exists,
+ cascade,
+ schema: DFSchemaRef::new(DFSchema::empty()),
+ }),
+ )),
_ => not_impl_err!(
- "Only `DROP TABLE/VIEW/SCHEMA ...` statement is supported currently"
+ "Only `DROP TABLE/VIEW/SCHEMA/CATALOG/DATABASE ...` statement is supported currently"
),
}
}
@@ -1919,6 +1930,43 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
)))
}
+ fn external_catalog_to_plan(
+ &self,
+ statement: CreateExternalCatalog,
+ ) -> Result<LogicalPlan> {
+ 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()),
+ }),
+ )))
+ }
+
/// 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/datafusion/sqllogictest/test_files/ddl.slt b/datafusion/sqllogictest/test_files/ddl.slt
index 21c195c3d2..968fcf40a4 100644
--- a/datafusion/sqllogictest/test_files/ddl.slt
+++ b/datafusion/sqllogictest/test_files/ddl.slt
@@ -769,7 +769,7 @@ drop table t;
# Dropping schemas
##########
-statement error DataFusion error: Execution error: Cannot drop schema foo_schema because other tables depend on it: bar
+statement error DataFusion error: Execution error: Cannot drop schema foo_schema containing tables: bar
DROP SCHEMA foo_schema;
statement ok
@@ -1312,3 +1312,47 @@ insert into t values (map{null: 1}, null);
statement ok
drop table t;
+
+# Test create/drop catalog
+statement ok
+create catalog cat;
+
+statement ok
+create catalog if not exists cat;
+
+statement ok
+create schema cat.schema;
+
+statement ok
+create table cat.schema.table (id int);
+
+statement error DataFusion error: Execution error: Cannot drop catalog containing schemas: schema
+drop catalog cat;
+
+statement ok
+drop catalog cat cascade;
+
+statement error DataFusion error: Execution error: Catalog 'cat' doesn't exist\.
+drop catalog cat;
+
+statement ok
+drop catalog if exists cat;
+
+# Test that catalog and database are interchangeable
+statement ok
+create database cat;
+
+statement error DataFusion error: Execution error: Catalog 'cat' already exists
+create catalog cat;
+
+statement ok
+drop catalog cat;
+
+statement ok
+create catalog cat;
+
+statement error DataFusion error: Execution error: Catalog 'cat' already exists
+create database cat;
+
+statement ok
+drop database cat;
diff --git a/datafusion/sqllogictest/test_files/information_schema.slt b/datafusion/sqllogictest/test_files/information_schema.slt
index 924883da42..0a8f800350 100644
--- a/datafusion/sqllogictest/test_files/information_schema.slt
+++ b/datafusion/sqllogictest/test_files/information_schema.slt
@@ -764,7 +764,7 @@ SHOW CREATE TABLE test.xyz
----
datafusion test xyz CREATE VIEW test.xyz AS SELECT * FROM abc
-statement error DataFusion error: Execution error: Cannot drop schema test because other tables depend on it: xyz
+statement error DataFusion error: Execution error: Cannot drop schema test containing tables: xyz
DROP SCHEMA test;
statement ok
diff --git a/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt b/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt
index e2d26740e3..a7f7120888 100644
--- a/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt
+++ b/datafusion/sqllogictest/test_files/information_schema_multiple_catalogs.slt
@@ -44,7 +44,7 @@ datafusion information_schema views VIEW
# Create multiple catalogs
###########
statement ok
-create database my_catalog;
+create catalog my_catalog;
statement ok
create schema my_catalog.my_schema;
@@ -62,7 +62,7 @@ statement ok
create table t2 as values(1);
statement ok
-create database my_other_catalog;
+create catalog my_other_catalog;
statement ok
create schema my_other_catalog.my_other_schema;
diff --git a/docs/source/library-user-guide/catalogs.md b/docs/source/library-user-guide/catalogs.md
index 3c8238b73e..dea4e413e2 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 <catalog_type> ...` 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<Arc<dyn CatalogProvider>> {
+ // `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 0d76775bcc..8382795e74 100644
--- a/docs/source/user-guide/sql/ddl.md
+++ b/docs/source/user-guide/sql/ddl.md
@@ -22,17 +22,69 @@
DDL stands for "Data Definition Language" and relates to creating and
modifying catalog objects such as Tables.
-## CREATE DATABASE
+## CREATE CATALOG / CREATE DATABASE
-Create catalog with specified name.
+Create catalog with specified name. In DataFusion, `CREATE DATABASE` is an alias for `CREATE CATALOG`.
<pre>
-CREATE DATABASE [ IF NOT EXISTS ] <i><b>catalog</i></b>
+CREATE CATALOG [ IF NOT EXISTS ] <i><b>catalog</i></b>
</pre>
```sql
-- create catalog cat
-CREATE DATABASE cat;
+CREATE CATALOG 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 ]
+<catalog_name>
+STORED AS <catalog_type>
+[ LOCATION <literal> ]
+[ OPTIONS (<key_value_list>) ]
+
+<key_value_list> := (<literal> <literal>, <literal> <literal>, ...)
+```
+
+`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.
+
+<pre>
+DROP CATALOG [ IF EXISTS ] <b><i>catalog_name</i></b>;
+</pre>
+
+```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
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org