alamb commented on code in PR #25112:
URL: https://github.com/apache/datafusion/pull/25112#discussion_r4066042034
##########
datafusion-examples/examples/sql_ops/custom_sql_parser.rs:
##########
@@ -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(&[
Review Comment:
that is much nicer
##########
datafusion/core/src/execution/session_state.rs:
##########
@@ -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<dyn CatalogProviderFactory>,
+ ) -> Self {
+ let mut catalog_factories = self.catalog_factories.unwrap_or_default();
Review Comment:
It would be nice to allow people to pass in `str` here rather than just
String:
```suggestion
pub fn with_catalog_factory(
mut self,
key: impl Into<String>,
catalog_factory: Arc<dyn CatalogProviderFactory>,
) -> Self {
let key = key.into();
let mut catalog_factories =
self.catalog_factories.unwrap_or_default();
```
##########
datafusion/core/tests/sql/create_drop.rs:
##########
@@ -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();
Review Comment:
this is consistent with the other code in this test, so it is all good.
However, it seems like a lot of ceremony. It would be really nice, perhaps a
follow on PR, to use the SessionStateBuilder for this. Something like
```rust
let ctx: SessionContext = SessionStateBuilder::new()
.with_default_features()
.with_catalog_factory("TESTCATALOG", Arc::new(TestCatalogFactory
{}))
.build()
.into();
```
##########
datafusion/sql/src/parser.rs:
##########
@@ -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<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)?;
+
+ #[derive(Default)]
+ struct Builder {
Review Comment:
What value does defining this struct add? It seems like all the fields are
just referenced directly 🤔
##########
parquet-testing:
##########
Review Comment:
I don't think we should change the submodule pins in this PR
##########
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.
Review Comment:
I found this somewhat confusing -- the comments are referring to `CREATE
EXTERNAL CATALOG` but the example seems to use `CREATE FOREIGN CATALOG` 🤔
I think the point is that this example is showing how to support `CREATE
FOREIGN CATALOG` which is similar in functionality to the (about to be) built
in feature of `CREATE EXTERNAL CATALOG`)
```suggestion
//! Note: DataFusion supports `CREATE EXTERNAL CATALOG` out-of-the-box
making use of
//! [`CatalogProviderFactory`](datafusion::catalog::CatalogProviderFactory).
This example
//! implements very similar functionality to demonstrate how to extend the
SQL parser.
```
##########
datafusion/core/tests/sql/create_drop.rs:
##########
@@ -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
Review Comment:
can you also please add a test for `DROP CATALOG cat CASCADE` ?
I would expect it to error with not supported / not implemented
##########
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`
Review Comment:
👍
##########
datafusion/sql/src/parser.rs:
##########
@@ -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";
Review Comment:
can you also please add a test for `DROP CATALOG c CASCADE` ?
##########
datafusion/sql/src/statement.rs:
##########
@@ -814,6 +818,13 @@ impl<S: ContextProvider> 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"
Review Comment:
we should probably update this message to also mention catalog
##########
datafusion/sql/src/statement.rs:
##########
@@ -814,6 +818,13 @@ impl<S: ContextProvider> 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()),
Review Comment:
I think this should also have `CASCADE` here
##########
datafusion/session/src/catalog.rs:
##########
@@ -206,6 +216,17 @@ 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.
+ ///
+ /// By default returns a "Not Implemented" error
+ fn deregister_catalog(&self, name: &str) -> Result<Option<Arc<dyn
CatalogProvider>>> {
Review Comment:
I think this should also take a `cascade` option for consistency with
deregister_schema
https://github.com/apache/datafusion/blob/5efef8ee1337ff9449b062f7eaa7bea0322b3c5b/datafusion/catalog/src/memory/catalog.rs#L108-L129
##########
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())
+ )?;
+ }
+ Ok(())
+ }
+}
+
Review Comment:
As a follow on PR it might be nice to make a builder for
`CreateExternalCatalog` and `DropCatalog`, similar to `CreateTable`
https://github.com/apache/datafusion/blob/5efef8ee1337ff9449b062f7eaa7bea0322b3c5b/datafusion/expr/src/logical_plan/ddl.rs#L283
We can file an issue as a follow on
--
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]
