martin-g commented on code in PR #25112:
URL: https://github.com/apache/datafusion/pull/25112#discussion_r4071129764
##########
datafusion/core/src/execution/context/mod.rs:
##########
@@ -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();
Review Comment:
Isn't it already uppercased by the parser ?
##########
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] <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.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())
+ )?;
+ }
Review Comment:
Are `or_replace` and `options` intentionally omitted here ?
##########
datafusion/core/src/execution/session_state.rs:
##########
@@ -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<String, Arc<dyn
CatalogProviderFactory>> {
+ &self.inner.catalog_factories
+ }
+
+ /// Get the catalog factories
Review Comment:
```suggestion
/// Get the catalog factories mutably
```
To make the difference against `catalog_factories()` above
##########
datafusion/session/src/catalog.rs:
##########
@@ -233,6 +254,17 @@ impl dyn CatalogProviderList {
}
}
+/// A factory which creates [`CatalogProvider`]s at runtime given a URL.
Review Comment:
A `URL` ? It receives a Session and a CreateExternalCatalog command.
```suggestion
/// A factory which creates [`CatalogProvider`]s at runtime from a
/// [`CreateExternalCatalog`] command (`CREATE EXTERNAL CATALOG ... STORED
AS <TYPE>`).
```
##########
datafusion/core/tests/sql/create_drop.rs:
##########
@@ -89,3 +113,104 @@ 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(())
+}
+
Review Comment:
Add a test for `OR REPLACE`:
```suggestion
#[tokio::test]
async fn create_external_catalog_or_replace() -> 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);
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(())
}
```
##########
docs/source/user-guide/sql/ddl.md:
##########
@@ -35,6 +35,58 @@ CREATE DATABASE [ IF NOT EXISTS ] <i><b>catalog</i></b>
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 ]
+<catalog_name>
+STORED AS <catalog_type>
+[ LOCATION <literal> ]
+[ OPTIONS (<key_value_list>) ]
+
+<key_value_list> := (<literal> <literal>, <literal> <literal>, ...)
+```
+
Review Comment:
```suggestion
`OR REPLACE` and `IF NOT EXISTS` cannot coexist.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
