diff --git a/Cargo.lock b/Cargo.lock
index 767e85947..82e3520ce 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1951,7 +1951,7 @@ dependencies = [
  "serde",
  "serde_json",
  "snafu",
- "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78)",
+ "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0)",
  "strum",
  "time",
  "tokio",
@@ -2246,7 +2246,7 @@ dependencies = [
 [[package]]
 name = "datafusion"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "arrow-ipc 55.2.0",
@@ -2288,7 +2288,7 @@ dependencies = [
  "parquet 55.2.0",
  "rand 0.8.5",
  "regex",
- "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78)",
+ "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0)",
  "tempfile",
  "tokio",
  "url",
@@ -2300,7 +2300,7 @@ dependencies = [
 [[package]]
 name = "datafusion-catalog"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2325,7 +2325,7 @@ dependencies = [
 [[package]]
 name = "datafusion-catalog-listing"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2347,7 +2347,7 @@ dependencies = [
 [[package]]
 name = "datafusion-common"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "ahash 0.8.12",
  "arrow",
@@ -2362,7 +2362,7 @@ dependencies = [
  "parquet 55.2.0",
  "paste",
  "recursive",
- "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78)",
+ "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0)",
  "tokio",
  "web-time",
 ]
@@ -2370,7 +2370,7 @@ dependencies = [
 [[package]]
 name = "datafusion-common-runtime"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "futures",
  "log",
@@ -2380,7 +2380,7 @@ dependencies = [
 [[package]]
 name = "datafusion-datasource"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-compression",
@@ -2415,7 +2415,7 @@ dependencies = [
 [[package]]
 name = "datafusion-datasource-csv"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2439,7 +2439,7 @@ dependencies = [
 [[package]]
 name = "datafusion-datasource-json"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2463,7 +2463,7 @@ dependencies = [
 [[package]]
 name = "datafusion-datasource-parquet"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2493,12 +2493,12 @@ dependencies = [
 [[package]]
 name = "datafusion-doc"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 
 [[package]]
 name = "datafusion-execution"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "dashmap",
@@ -2516,7 +2516,7 @@ dependencies = [
 [[package]]
 name = "datafusion-expr"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "chrono",
@@ -2530,13 +2530,13 @@ dependencies = [
  "paste",
  "recursive",
  "serde_json",
- "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78)",
+ "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0)",
 ]
 
 [[package]]
 name = "datafusion-expr-common"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "datafusion-common",
@@ -2548,7 +2548,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "arrow-buffer 55.2.0",
@@ -2576,7 +2576,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-aggregate"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "ahash 0.8.12",
  "arrow",
@@ -2596,7 +2596,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-aggregate-common"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "ahash 0.8.12",
  "arrow",
@@ -2619,7 +2619,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-nested"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "arrow-ord 55.2.0",
@@ -2639,7 +2639,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-table"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2654,7 +2654,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-window"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "datafusion-common",
  "datafusion-doc",
@@ -2670,7 +2670,7 @@ dependencies = [
 [[package]]
 name = "datafusion-functions-window-common"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "datafusion-common",
  "datafusion-physical-expr-common",
@@ -2679,7 +2679,7 @@ dependencies = [
 [[package]]
 name = "datafusion-macros"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "datafusion-expr",
  "quote",
@@ -2689,7 +2689,7 @@ dependencies = [
 [[package]]
 name = "datafusion-optimizer"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "chrono",
@@ -2707,7 +2707,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-expr"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "ahash 0.8.12",
  "arrow",
@@ -2728,7 +2728,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-expr-common"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "ahash 0.8.12",
  "arrow",
@@ -2741,7 +2741,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-optimizer"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "datafusion-common",
@@ -2759,7 +2759,7 @@ dependencies = [
 [[package]]
 name = "datafusion-physical-plan"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "ahash 0.8.12",
  "arrow",
@@ -2788,7 +2788,7 @@ dependencies = [
 [[package]]
 name = "datafusion-session"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "async-trait",
@@ -2811,7 +2811,7 @@ dependencies = [
 [[package]]
 name = "datafusion-sql"
 version = "47.0.0"
-source = "git+https://github.com/Embucket/datafusion.git?rev=330a88732a63376af59d48a57ca15072ce3b336a#330a88732a63376af59d48a57ca15072ce3b336a"
+source = "git+https://github.com/Embucket/datafusion.git?rev=087ebef56368ce471532b6d6b69eaab94cddd651#087ebef56368ce471532b6d6b69eaab94cddd651"
 dependencies = [
  "arrow",
  "bigdecimal",
@@ -2821,7 +2821,7 @@ dependencies = [
  "log",
  "recursive",
  "regex",
- "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78)",
+ "sqlparser 0.55.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0)",
 ]
 
 [[package]]
@@ -7327,11 +7327,11 @@ checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67"
 [[package]]
 name = "sqlparser"
 version = "0.55.0"
-source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78#97c03c04230fdf789c49c64eaaa5c0e77ffc8c78"
+source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0#12655c2be19d4796236154f8826e23f84d2978b0"
 dependencies = [
  "log",
  "recursive",
- "sqlparser_derive 0.3.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78)",
+ "sqlparser_derive 0.3.0 (git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0)",
 ]
 
 [[package]]
@@ -7347,7 +7347,7 @@ dependencies = [
 [[package]]
 name = "sqlparser_derive"
 version = "0.3.0"
-source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=97c03c04230fdf789c49c64eaaa5c0e77ffc8c78#97c03c04230fdf789c49c64eaaa5c0e77ffc8c78"
+source = "git+https://github.com/Embucket/datafusion-sqlparser-rs.git?rev=12655c2be19d4796236154f8826e23f84d2978b0#12655c2be19d4796236154f8826e23f84d2978b0"
 dependencies = [
  "proc-macro2",
  "quote",
diff --git a/Cargo.toml b/Cargo.toml
index 93dd4fd12..48799a3ec 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -89,12 +89,12 @@ uuid = { version = "1.10.0", features = ["v4", "serde"] }
 validator = { version = "0.20.0", features = ["derive"] }
 
 [patch.crates-io]
-datafusion = { git = "https://github.com/Embucket/datafusion.git", rev = "330a88732a63376af59d48a57ca15072ce3b336a" }
-datafusion-common = { git = "https://github.com/Embucket/datafusion.git", rev = "330a88732a63376af59d48a57ca15072ce3b336a" }
-datafusion-expr = { git = "https://github.com/Embucket/datafusion.git", rev = "330a88732a63376af59d48a57ca15072ce3b336a" }
-datafusion-physical-plan = { git = "https://github.com/Embucket/datafusion.git", rev = "330a88732a63376af59d48a57ca15072ce3b336a" }
-datafusion-doc = { git = "https://github.com/Embucket/datafusion.git", rev = "330a88732a63376af59d48a57ca15072ce3b336a" }
-datafusion-macros = { git = "https://github.com/Embucket/datafusion.git", rev = "330a88732a63376af59d48a57ca15072ce3b336a" }
+datafusion = { git = "https://github.com/Embucket/datafusion.git", rev = "087ebef56368ce471532b6d6b69eaab94cddd651" }
+datafusion-common = { git = "https://github.com/Embucket/datafusion.git", rev = "087ebef56368ce471532b6d6b69eaab94cddd651" }
+datafusion-expr = { git = "https://github.com/Embucket/datafusion.git", rev = "087ebef56368ce471532b6d6b69eaab94cddd651" }
+datafusion-physical-plan = { git = "https://github.com/Embucket/datafusion.git", rev = "087ebef56368ce471532b6d6b69eaab94cddd651" }
+datafusion-doc = { git = "https://github.com/Embucket/datafusion.git", rev = "087ebef56368ce471532b6d6b69eaab94cddd651" }
+datafusion-macros = { git = "https://github.com/Embucket/datafusion.git", rev = "087ebef56368ce471532b6d6b69eaab94cddd651" }
 
 [workspace.lints.clippy]
 all = { level = "deny", priority = -1 }
diff --git a/crates/api-iceberg-rest/src/error.rs b/crates/api-iceberg-rest/src/error.rs
index 1ea5eec3f..a09e12353 100644
--- a/crates/api-iceberg-rest/src/error.rs
+++ b/crates/api-iceberg-rest/src/error.rs
@@ -70,7 +70,8 @@ impl IntoResponse for Error {
             | core_metastore::Error::DatabaseAlreadyExists { .. }
             | core_metastore::Error::SchemaAlreadyExists { .. }
             | core_metastore::Error::TableAlreadyExists { .. }
-            | core_metastore::Error::VolumeInUse { .. } => http::StatusCode::CONFLICT,
+            | core_metastore::Error::VolumeInUse { .. }
+            | core_metastore::Error::DatabaseInUse { .. } => http::StatusCode::CONFLICT,
             core_metastore::Error::TableRequirementFailed { .. } => {
                 http::StatusCode::UNPROCESSABLE_ENTITY
             }
diff --git a/crates/api-internal-rest/src/error.rs b/crates/api-internal-rest/src/error.rs
index c04b81fe6..94dc0908a 100644
--- a/crates/api-internal-rest/src/error.rs
+++ b/crates/api-internal-rest/src/error.rs
@@ -109,7 +109,8 @@ impl IntoResponse for Error {
                 | core_metastore::Error::DatabaseAlreadyExists { .. }
                 | core_metastore::Error::SchemaAlreadyExists { .. }
                 | core_metastore::Error::TableAlreadyExists { .. }
-                | core_metastore::Error::VolumeInUse { .. } => http::StatusCode::CONFLICT,
+                | core_metastore::Error::VolumeInUse { .. }
+                | core_metastore::Error::DatabaseInUse { .. } => http::StatusCode::CONFLICT,
                 core_metastore::Error::TableRequirementFailed { .. } => {
                     http::StatusCode::UNPROCESSABLE_ENTITY
                 }
diff --git a/crates/core-executor/Cargo.toml b/crates/core-executor/Cargo.toml
index acca533c0..9dd1a2116 100644
--- a/crates/core-executor/Cargo.toml
+++ b/crates/core-executor/Cargo.toml
@@ -24,7 +24,7 @@ datafusion-functions-json = { workspace = true }
 datafusion-physical-plan = { workspace = true }
 datafusion_iceberg = { workspace = true }
 futures = { workspace = true }
-sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "97c03c04230fdf789c49c64eaaa5c0e77ffc8c78", features = [
+sqlparser = { git = "https://github.com/Embucket/datafusion-sqlparser-rs.git", rev = "12655c2be19d4796236154f8826e23f84d2978b0", features = [
   "visitor",
 ] }
 iceberg-rust = { workspace = true }
diff --git a/crates/core-executor/src/error.rs b/crates/core-executor/src/error.rs
index 5c710d644..002042978 100644
--- a/crates/core-executor/src/error.rs
+++ b/crates/core-executor/src/error.rs
@@ -1,5 +1,3 @@
-use std::backtrace::Backtrace;
-
 use datafusion_common::DataFusionError;
 use df_catalog::error::Error as CatalogError;
 use error_stack_trace;
@@ -7,6 +5,8 @@ use iceberg_rust::error::Error as IcebergError;
 use iceberg_s3tables_catalog::error::Error as S3tablesError;
 use snafu::Location;
 use snafu::prelude::*;
+use std::backtrace::Backtrace;
+use std::fmt::Display;
 
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -187,9 +187,9 @@ pub enum Error {
         location: Location,
     },
 
-    #[snafu(display("Object of type {type_name} with name {name} already exists"))]
+    #[snafu(display("Object of type {type:?} with name {name} already exists"))]
     ObjectAlreadyExists {
-        type_name: String,
+        r#type: ObjectType,
         name: String,
         #[snafu(implicit)]
         location: Location,
@@ -277,9 +277,23 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
+    #[snafu(display("Failed to create database '{name}' without external volume"))]
+    ExternalVolumeRequiredForCreateDatabase {
+        name: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
+    #[snafu(display("Failed to drop database: {source}"))]
+    DropDatabase {
+        #[snafu(source(from(CatalogError, Box::new)))]
+        source: Box<CatalogError>,
+        #[snafu(implicit)]
+        location: Location,
+    },
 
-    #[snafu(display("Failed to drop catalog: {source}"))]
-    DropCatalog {
+    #[snafu(display("Failed to create database: {source}"))]
+    CreateDatabase {
         #[snafu(source(from(CatalogError, Box::new)))]
         source: Box<CatalogError>,
         #[snafu(implicit)]
@@ -304,11 +318,6 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-    #[snafu(display("Only CREATE TABLE/CREATE SCHEMA statements are supported"))]
-    OnlyTableSchemaCreateStatements {
-        #[snafu(implicit)]
-        location: Location,
-    },
     #[snafu(display("Only DROP statements are supported"))]
     OnlyDropStatements {
         #[snafu(implicit)]
@@ -450,3 +459,20 @@ pub enum Error {
         location: Location,
     },
 }
+
+#[derive(Debug)]
+pub enum ObjectType {
+    Database,
+    Schema,
+    Table,
+}
+
+impl Display for ObjectType {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Database => write!(f, "database"),
+            Self::Schema => write!(f, "schema"),
+            Self::Table => write!(f, "table"),
+        }
+    }
+}
diff --git a/crates/core-executor/src/query.rs b/crates/core-executor/src/query.rs
index d413f533f..38c3d154e 100644
--- a/crates/core-executor/src/query.rs
+++ b/crates/core-executor/src/query.rs
@@ -5,7 +5,9 @@ use super::catalog::{
     catalog_list::EmbucketCatalogList, catalogs::embucket::catalog::EmbucketCatalog,
 };
 use super::datafusion::planner::ExtendedSqlToRel;
-use super::error::{self as ex_error, Error, RefreshCatalogListSnafu, Result};
+use super::error::{
+    self as ex_error, Error, ObjectType as ExistingObjectType, RefreshCatalogListSnafu, Result,
+};
 use super::session::UserSession;
 use super::utils::{NormalizedIdent, is_logical_plan_effectively_empty};
 use crate::datafusion::logical_plan::merge::MergeIntoCOWSink;
@@ -61,7 +63,6 @@ use datafusion_iceberg::catalog::schema::IcebergSchema;
 use datafusion_iceberg::table::DataFusionTableConfigBuilder;
 use df_catalog::catalog::CachingCatalog;
 use df_catalog::catalog_list::CachedEntity;
-use df_catalog::error::Error as CatalogError;
 use df_catalog::information_schema::session_params::SessionProperty;
 use df_catalog::table::CachingTable;
 use embucket_functions::semi_structured::variant::visitors::visit_all;
@@ -179,24 +180,16 @@ impl UserQuery {
             .as_any()
             .downcast_ref::<EmbucketCatalogList>()
         {
-            let res = catalog_list_impl
-                .invalidate_cache(entity)
+            catalog_list_impl
+                .invalidate_cache(entity.normalized())
                 .await
-                .context(RefreshCatalogListSnafu);
-
-            // Not sure we need this, but just in case
-            // TODO: Remove following block when refresh_catalog removed
-            if let Err(Error::RefreshCatalogList { source, .. }) = res {
-                // refresh entire catalog if cache error happen
-                if let CatalogError::InvalidCache { .. } = *source {
-                    self.refresh_catalog().await?;
-                }
-            }
+                .context(RefreshCatalogListSnafu)?;
         }
         Ok(())
     }
 
-    async fn refresh_catalog(&self) -> Result<()> {
+    #[instrument(name = "UserQuery::drop_catalog", level = "debug", skip(self), err)]
+    async fn drop_catalog(&self, catalog: &str, cascade: bool) -> Result<()> {
         if let Some(catalog_list_impl) = self
             .session
             .ctx
@@ -206,14 +199,15 @@ impl UserQuery {
             .downcast_ref::<EmbucketCatalogList>()
         {
             catalog_list_impl
-                .refresh()
+                .drop_catalog(catalog, cascade)
                 .await
-                .context(RefreshCatalogListSnafu)?;
+                .context(ex_error::DropDatabaseSnafu)?;
         }
         Ok(())
     }
 
-    async fn drop_catalog(&self, catalog: &str, cascade: bool) -> Result<()> {
+    #[instrument(name = "UserQuery::create_catalog", level = "debug", skip(self), err)]
+    async fn create_catalog(&self, catalog: &str, volume: &str) -> Result<()> {
         if let Some(catalog_list_impl) = self
             .session
             .ctx
@@ -223,9 +217,9 @@ impl UserQuery {
             .downcast_ref::<EmbucketCatalogList>()
         {
             catalog_list_impl
-                .drop_catalog(catalog, cascade)
+                .create_catalog(catalog, volume)
                 .await
-                .context(ex_error::DropCatalogSnafu)?;
+                .context(ex_error::CreateDatabaseSnafu)?;
         }
         Ok(())
     }
@@ -360,17 +354,17 @@ impl UserQuery {
                 Statement::CreateTable { .. } => {
                     return Box::pin(self.create_table_query(*s)).await;
                 }
-                Statement::CreateDatabase { .. } => {
-                    // TODO: Databases are only able to be created through the
-                    // metastore API. We need to add Snowflake volume syntax to CREATE DATABASE query
-                    return ex_error::OnlyTableSchemaCreateStatementsSnafu.fail();
-                    /*return self
-                    .create_database(db_name, if_not_exists)
-                    .await;*/
-                }
-                Statement::CreateSchema { .. } => {
-                    return Box::pin(self.create_schema(*s)).await;
+                Statement::CreateDatabase {
+                    db_name,
+                    if_not_exists,
+                    external_volume,
+                    ..
+                } => {
+                    return self
+                        .create_database(db_name, if_not_exists, external_volume)
+                        .await;
                 }
+                Statement::CreateSchema { .. } => return Box::pin(self.create_schema(*s)).await,
                 Statement::CreateStage { .. } => {
                     // We support only CSV uploads for now
                     return Box::pin(self.create_stage_query(*s)).await;
@@ -394,9 +388,7 @@ impl UserQuery {
                 | Statement::ShowFunctions { .. }
                 | Statement::ShowObjects { .. }
                 | Statement::ShowVariables { .. }
-                | Statement::ShowVariable { .. } => {
-                    return Box::pin(self.show_query(*s)).await;
-                }
+                | Statement::ShowVariable { .. } => return Box::pin(self.show_query(*s)).await,
                 Statement::Truncate { table_names, .. } => {
                     return Box::pin(self.truncate_table(table_names)).await;
                 }
@@ -404,12 +396,8 @@ impl UserQuery {
                     self.traverse_and_update_query(subquery.as_mut()).await;
                     return Box::pin(self.execute_with_custom_plan(&subquery.to_string())).await;
                 }
-                Statement::Drop { .. } => {
-                    return Box::pin(self.drop_query(*s)).await;
-                }
-                Statement::Merge { .. } => {
-                    return Box::pin(self.merge_query(*s)).await;
-                }
+                Statement::Drop { .. } => return Box::pin(self.drop_query(*s)).await,
+                Statement::Merge { .. } => return Box::pin(self.merge_query(*s)).await,
                 _ => {}
             }
         } else if let DFStatement::CreateExternalTable(cetable) = statement {
@@ -741,7 +729,7 @@ impl UserQuery {
                     .context(ex_error::IcebergSnafu)?;
             } else {
                 return ex_error::ObjectAlreadyExistsSnafu {
-                    type_name: "table".to_string(),
+                    r#type: ExistingObjectType::Table,
                     name: ident.to_string(),
                 }
                 .fail();
@@ -1100,7 +1088,7 @@ impl UserQuery {
 
                 Ok((source_plan, None))
             }
-            _ => ex_error::MergeSourceNotSupportedSnafu.fail(),
+            _ => MergeSourceNotSupportedSnafu.fail(),
         }?;
 
         let source_schema = source_plan.schema().clone();
@@ -1203,6 +1191,34 @@ impl UserQuery {
         .await
     }
 
+    #[instrument(name = "UserQuery::create_database", level = "trace", skip(self), err)]
+    pub async fn create_database(
+        &self,
+        db_name: ObjectName,
+        if_not_exists: bool,
+        external_volume: Option<String>,
+    ) -> Result<QueryResult> {
+        let catalog_name = object_name_to_string(&db_name);
+        if external_volume.is_none() {
+            return ex_error::ExternalVolumeRequiredForCreateDatabaseSnafu { name: catalog_name }
+                .fail();
+        }
+        let catalog_exist = self.get_catalog(&catalog_name).is_ok();
+        if catalog_exist {
+            if if_not_exists {
+                return self.created_entity_response();
+            }
+            return ex_error::ObjectAlreadyExistsSnafu {
+                r#type: ExistingObjectType::Database,
+                name: catalog_name,
+            }
+            .fail();
+        }
+        self.create_catalog(&catalog_name, &external_volume.unwrap_or_default())
+            .await?;
+        self.created_entity_response()
+    }
+
     #[instrument(name = "UserQuery::create_schema", level = "trace", skip(self), err)]
     pub async fn create_schema(&self, statement: Statement) -> Result<QueryResult> {
         let Statement::CreateSchema {
@@ -1243,7 +1259,7 @@ impl UserQuery {
                 return self.created_entity_response();
             }
             return ex_error::ObjectAlreadyExistsSnafu {
-                type_name: "schema".to_string(),
+                r#type: ExistingObjectType::Schema,
                 name: ident.schema,
             }
             .fail();
@@ -1273,8 +1289,8 @@ impl UserQuery {
     #[allow(clippy::too_many_lines)]
     pub async fn show_query(&self, statement: Statement) -> Result<QueryResult> {
         let query = match statement {
-            Statement::ShowDatabases { .. } => {
-                format!(
+            Statement::ShowDatabases { show_options, .. } => {
+                let sql = format!(
                     "SELECT
                         NULL as created_on,
                         database_name as name,
@@ -1283,7 +1299,14 @@ impl UserQuery {
                         NULL as schema_name
                     FROM {}.information_schema.databases",
                     self.current_database()
-                )
+                );
+                let mut filters = Vec::new();
+                if let Some(filter) =
+                    build_starts_with_filter(show_options.starts_with, "database_name")
+                {
+                    filters.push(filter);
+                }
+                apply_show_filters(sql, &filters)
             }
             Statement::ShowSchemas { show_options, .. } => {
                 let reference = self.resolve_show_in_name(show_options.show_in, false);
diff --git a/crates/core-executor/src/snowflake_error.rs b/crates/core-executor/src/snowflake_error.rs
index f001bc71a..074d3a0e9 100644
--- a/crates/core-executor/src/snowflake_error.rs
+++ b/crates/core-executor/src/snowflake_error.rs
@@ -164,6 +164,7 @@ impl From<Error> for SnowflakeError {
                     MetastoreError::TableNotFound { .. } => Self::Custom { message },
                     MetastoreError::TableObjectStoreNotFound { .. } => Self::Custom { message },
                     MetastoreError::VolumeInUse { .. } => Self::Custom { message },
+                    MetastoreError::DatabaseInUse { .. } => Self::Custom { message },
                     MetastoreError::Iceberg { .. } => Self::Custom { message },
                     MetastoreError::TableMetadataBuilder { .. } => Self::Custom { message },
                     MetastoreError::Serde { .. } => Self::Custom { message },
@@ -174,7 +175,8 @@ impl From<Error> for SnowflakeError {
             Error::InvalidDatabaseIdentifier { .. } => Self::Custom { message },
             Error::InvalidTableIdentifier { .. } => Self::Custom { message },
             Error::InvalidSchemaIdentifier { .. } => Self::Custom { message },
-            Error::DropCatalog { .. } => Self::Custom { message },
+            Error::DropDatabase { .. } => Self::Custom { message },
+            Error::CreateDatabase { .. } => Self::Custom { message },
             Error::InvalidFilePath { .. } => Self::Custom { message },
             Error::InvalidBucketIdentifier { .. } => Self::Custom { message },
             Error::Arrow { .. } => Self::Custom { message },
@@ -198,10 +200,10 @@ impl From<Error> for SnowflakeError {
             Error::UploadFailed { .. } => Self::Custom { message },
             Error::CatalogListDowncast { .. } => Self::Custom { message },
             Error::RegisterCatalog { .. } => Self::Custom { message },
+            Error::ExternalVolumeRequiredForCreateDatabase { .. } => Self::Custom { message },
             Error::SerdeParse { .. } => Self::Custom { message },
             Error::OnyUseWithVariables { .. } => Self::Custom { message },
             Error::OnlyPrimitiveStatements { .. } => Self::Custom { message },
-            Error::OnlyTableSchemaCreateStatements { .. } => Self::Custom { message },
             Error::OnlyDropStatements { .. } => Self::Custom { message },
             Error::OnlyDropTableViewStatements { .. } => Self::Custom { message },
             Error::OnlyCreateTableStatements { .. } => Self::Custom { message },
diff --git a/crates/core-executor/src/tests/query.rs b/crates/core-executor/src/tests/query.rs
index 72a332e20..78da71ed8 100644
--- a/crates/core-executor/src/tests/query.rs
+++ b/crates/core-executor/src/tests/query.rs
@@ -214,7 +214,19 @@ macro_rules! test_query {
     };
 }
 
-test_query!(drop_database, "DROP DATABASE embucket");
+test_query!(drop_database_error_in_use, "DROP DATABASE embucket");
+
+test_query!(
+    drop_database,
+    "DROP DATABASE embucket",
+    setup_queries = ["DROP SCHEMA embucket.public"]
+);
+
+test_query!(
+    create_database,
+    "SHOW DATABASES STARTS WITH 'db_test'",
+    setup_queries = ["CREATE DATABASE db_test external_volume = 'test_volume'",]
+);
 
 // CREATE SCHEMA
 test_query!(
@@ -353,6 +365,7 @@ test_query!(
     sort_all = true,
     snapshot_path = "session"
 );
+test_query!(show_databases_filter, "SHOW DATABASES STARTS WITH 'em'");
 
 // SHOW SCHEMAS
 test_query!(
diff --git a/crates/core-executor/src/tests/snapshots/query_create_database.snap b/crates/core-executor/src/tests/snapshots/query_create_database.snap
new file mode 100644
index 000000000..434c1f37b
--- /dev/null
+++ b/crates/core-executor/src/tests/snapshots/query_create_database.snap
@@ -0,0 +1,14 @@
+---
+source: crates/core-executor/src/tests/query.rs
+description: "\"SHOW DATABASES STARTS WITH 'db_test'\""
+info: "Setup queries: CREATE DATABASE db_test external_volume = 'test_volume'"
+---
+Ok(
+    [
+        "+------------+---------+----------+---------------+-------------+",
+        "| created_on | name    | kind     | database_name | schema_name |",
+        "+------------+---------+----------+---------------+-------------+",
+        "|            | db_test | STANDARD |               |             |",
+        "+------------+---------+----------+---------------+-------------+",
+    ],
+)
diff --git a/crates/core-executor/src/tests/snapshots/query_drop_database_error_in_use.snap b/crates/core-executor/src/tests/snapshots/query_drop_database_error_in_use.snap
new file mode 100644
index 000000000..0d2c460b1
--- /dev/null
+++ b/crates/core-executor/src/tests/snapshots/query_drop_database_error_in_use.snap
@@ -0,0 +1,7 @@
+---
+source: crates/core-executor/src/tests/query.rs
+description: "\"DROP DATABASE embucket\""
+---
+Err(
+    "Error: Failed to drop database: Metastore error: Database embucket in use by schema(s): public",
+)
diff --git a/crates/core-executor/src/tests/snapshots/query_show_databases_filter.snap b/crates/core-executor/src/tests/snapshots/query_show_databases_filter.snap
new file mode 100644
index 000000000..2f8ed7f20
--- /dev/null
+++ b/crates/core-executor/src/tests/snapshots/query_show_databases_filter.snap
@@ -0,0 +1,13 @@
+---
+source: crates/core-executor/src/tests/query.rs
+description: "\"SHOW DATABASES STARTS WITH 'em'\""
+---
+Ok(
+    [
+        "+------------+----------+----------+---------------+-------------+",
+        "| created_on | name     | kind     | database_name | schema_name |",
+        "+------------+----------+----------+---------------+-------------+",
+        "|            | embucket | STANDARD |               |             |",
+        "+------------+----------+----------+---------------+-------------+",
+    ],
+)
diff --git a/crates/core-metastore/src/error.rs b/crates/core-metastore/src/error.rs
index 2fbf3825e..d94b0ad6c 100644
--- a/crates/core-metastore/src/error.rs
+++ b/crates/core-metastore/src/error.rs
@@ -182,6 +182,14 @@ pub enum Error {
         location: Location,
     },
 
+    #[snafu(display("Database {database} in use by schema(s): {schema}"))]
+    DatabaseInUse {
+        database: String,
+        schema: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+
     #[snafu(display("Iceberg error: {error}"))]
     Iceberg {
         #[snafu(source(from(IcebergError, Box::new)))]
diff --git a/crates/core-metastore/src/metastore.rs b/crates/core-metastore/src/metastore.rs
index 304929e79..2d6e8a746 100644
--- a/crates/core-metastore/src/metastore.rs
+++ b/crates/core-metastore/src/metastore.rs
@@ -412,6 +412,16 @@ impl Metastore for SlateDBMetastore {
                 .map(|schema| self.delete_schema(&schema.ident, cascade))
                 .collect::<Vec<_>>();
             futures::future::try_join_all(futures).await?;
+        } else if !schemas.is_empty() {
+            return Err(metastore_error::DatabaseInUseSnafu {
+                database: name,
+                schema: schemas
+                    .iter()
+                    .map(|s| s.ident.schema.clone())
+                    .collect::<Vec<_>>()
+                    .join(", "),
+            }
+            .build());
         }
         let key = format!("{KEY_DATABASE}/{name}");
         self.delete_object(&key).await
diff --git a/crates/core-metastore/src/models/schema.rs b/crates/core-metastore/src/models/schema.rs
index 849128e60..6639948be 100644
--- a/crates/core-metastore/src/models/schema.rs
+++ b/crates/core-metastore/src/models/schema.rs
@@ -22,6 +22,14 @@ impl SchemaIdent {
     pub const fn new(database: DatabaseIdent, schema: String) -> Self {
         Self { schema, database }
     }
+
+    #[must_use]
+    pub fn normalized(&self) -> Self {
+        Self {
+            schema: self.schema.to_ascii_lowercase(),
+            database: self.database.to_ascii_lowercase(),
+        }
+    }
 }
 
 impl std::fmt::Display for SchemaIdent {
diff --git a/crates/core-metastore/src/models/table.rs b/crates/core-metastore/src/models/table.rs
index 9eafa8e01..0b49c2ae1 100644
--- a/crates/core-metastore/src/models/table.rs
+++ b/crates/core-metastore/src/models/table.rs
@@ -41,6 +41,15 @@ impl TableIdent {
         let namespace = vec![self.schema.clone()];
         Identifier::new(&namespace, &self.table)
     }
+
+    #[must_use]
+    pub fn normalized(&self) -> Self {
+        Self {
+            table: self.table.to_ascii_lowercase(),
+            schema: self.schema.to_ascii_lowercase(),
+            database: self.database.to_ascii_lowercase(),
+        }
+    }
 }
 
 impl From<TableIdent> for SchemaIdent {
diff --git a/crates/df-catalog/src/catalog.rs b/crates/df-catalog/src/catalog.rs
index 4bddb29db..beccf3ccb 100644
--- a/crates/df-catalog/src/catalog.rs
+++ b/crates/df-catalog/src/catalog.rs
@@ -15,7 +15,7 @@ pub struct CachingCatalog {
 
 #[derive(Clone, Debug)]
 pub enum CatalogType {
-    Internal,
+    Embucket,
     Memory,
     S3tables,
 }
@@ -28,7 +28,7 @@ impl CachingCatalog {
             should_refresh: false,
             enable_information_schema: true,
             name,
-            catalog_type: CatalogType::Internal,
+            catalog_type: CatalogType::Embucket,
         }
     }
     #[must_use]
diff --git a/crates/df-catalog/src/catalog_list.rs b/crates/df-catalog/src/catalog_list.rs
index d22688b53..78447e338 100644
--- a/crates/df-catalog/src/catalog_list.rs
+++ b/crates/df-catalog/src/catalog_list.rs
@@ -4,7 +4,8 @@ use crate::catalog::{CachingCatalog, CatalogType};
 use crate::catalogs::slatedb::catalog::{SLATEDB_CATALOG, SlateDBCatalog};
 use crate::df_error;
 use crate::error::{
-    self as df_catalog_error, InvalidCacheSnafu, MetastoreSnafu, NotImplementedSnafu, Result,
+    self as df_catalog_error, InvalidCacheSnafu, MetastoreSnafu, MissingVolumeSnafu,
+    NotImplementedSnafu, Result, UnsupportedFeature,
 };
 use crate::schema::CachingSchema;
 use crate::table::CachingTable;
@@ -12,10 +13,13 @@ use aws_config::{BehaviorVersion, Region, SdkConfig};
 use aws_credential_types::Credentials;
 use aws_credential_types::provider::SharedCredentialsProvider;
 use core_history::HistoryStore;
-use core_metastore::{AwsCredentials, Metastore, VolumeType as MetastoreVolumeType};
+use core_metastore::{
+    AwsCredentials, Database, Metastore, VolumeType as MetastoreVolumeType, VolumeType,
+};
 use core_metastore::{SchemaIdent, TableIdent};
 use core_utils::scan_iterator::ScanIterator;
 use dashmap::DashMap;
+use datafusion::catalog::MemoryCatalogProvider;
 use datafusion::{
     catalog::{CatalogProvider, CatalogProviderList},
     execution::object_store::ObjectStoreRegistry,
@@ -40,6 +44,16 @@ pub enum CachedEntity {
     Table(TableIdent),
 }
 
+impl CachedEntity {
+    #[must_use]
+    pub fn normalized(&self) -> Self {
+        match self {
+            Self::Schema(ident) => Self::Schema(ident.normalized()),
+            Self::Table(ident) => Self::Table(ident.normalized()),
+        }
+    }
+}
+
 pub struct EmbucketCatalogList {
     pub metastore: Arc<dyn Metastore>,
     pub history_store: Arc<dyn HistoryStore>,
@@ -74,22 +88,77 @@ impl EmbucketCatalogList {
             .fail();
         };
         match catalog.catalog_type {
-            CatalogType::Internal => {
-                // Set cascade to true to delete all tables in the database
+            CatalogType::Embucket => {
                 self.metastore
-                    .delete_database(&name.to_string(), true)
+                    .delete_database(&name.to_string(), cascade)
                     .await
                     .context(MetastoreSnafu)?;
                 Ok(())
             }
             CatalogType::Memory => Ok(()),
             CatalogType::S3tables => NotImplementedSnafu {
+                feature: UnsupportedFeature::DropS3TablesDatabase,
                 details: "Dropping S3 tables catalogs is not supported",
             }
             .fail(),
         }
     }
 
+    #[tracing::instrument(
+        name = "EmbucketCatalogList::create_catalog",
+        level = "debug",
+        skip(self),
+        err
+    )]
+    pub async fn create_catalog(&self, catalog_name: &str, volume_ident: &str) -> Result<()> {
+        let volume = self
+            .metastore
+            .get_volume(&volume_ident.to_string())
+            .await
+            .context(MetastoreSnafu)?;
+
+        let Some(volume) = volume else {
+            return MissingVolumeSnafu {
+                name: volume_ident.to_string(),
+            }
+            .fail();
+        };
+
+        match volume.volume {
+            VolumeType::S3(_) | VolumeType::File(_) => {
+                let database = Database {
+                    ident: catalog_name.to_owned(),
+                    volume: volume_ident.to_owned(),
+                    properties: None,
+                };
+                self.metastore
+                    .create_database(&catalog_name.to_owned(), database)
+                    .await
+                    .context(MetastoreSnafu)?;
+                self.catalogs.insert(
+                    catalog_name.to_owned(),
+                    Arc::new(self.get_embucket_catalog(catalog_name)?),
+                );
+            }
+            VolumeType::Memory => {
+                let provider = MemoryCatalogProvider::new();
+                let catalog = CachingCatalog::new(Arc::new(provider), catalog_name.to_owned())
+                    .with_refresh(true)
+                    .with_catalog_type(CatalogType::Memory);
+                self.catalogs
+                    .insert(catalog_name.to_owned(), Arc::new(catalog));
+            }
+            VolumeType::S3Tables(_) => {
+                return NotImplementedSnafu {
+                    feature: UnsupportedFeature::CreateS3TablesDatabase,
+                    details: "Creating an S3 table catalog is not supported",
+                }
+                .fail();
+            }
+        }
+        Ok(())
+    }
+
     /// Discovers and registers all available catalogs into the catalog registry.
     ///
     /// This method performs the following steps:
@@ -137,20 +206,22 @@ impl EmbucketCatalogList {
             .await
             .context(df_catalog_error::CoreSnafu)?
             .into_iter()
-            .map(|db| {
-                let iceberg_catalog =
-                    EmbucketIcebergCatalog::new(self.metastore.clone(), db.ident.clone())
-                        .context(MetastoreSnafu)?;
-                let catalog: Arc<dyn CatalogProvider> = Arc::new(EmbucketCatalog::new(
-                    db.ident.clone(),
-                    self.metastore.clone(),
-                    Arc::new(iceberg_catalog),
-                ));
-                Ok(CachingCatalog::new(catalog, db.ident.clone()).with_refresh(true))
-            })
+            .map(|db| self.get_embucket_catalog(&db.ident.clone()))
             .collect()
     }
 
+    fn get_embucket_catalog(&self, database_name: &str) -> Result<CachingCatalog> {
+        let iceberg_catalog =
+            EmbucketIcebergCatalog::new(self.metastore.clone(), database_name.to_owned())
+                .context(MetastoreSnafu)?;
+        let catalog: Arc<dyn CatalogProvider> = Arc::new(EmbucketCatalog::new(
+            database_name.to_owned(),
+            self.metastore.clone(),
+            Arc::new(iceberg_catalog),
+        ));
+        Ok(CachingCatalog::new(catalog, database_name.to_owned()).with_refresh(true))
+    }
+
     #[must_use]
     #[tracing::instrument(
         name = "EmbucketCatalogList::slatedb_catalog",
diff --git a/crates/df-catalog/src/error.rs b/crates/df-catalog/src/error.rs
index 7aab6a129..4244d68ce 100644
--- a/crates/df-catalog/src/error.rs
+++ b/crates/df-catalog/src/error.rs
@@ -79,10 +79,23 @@ pub enum Error {
         #[snafu(implicit)]
         location: Location,
     },
-    #[snafu(display("Feature not implemented: '{details:?}'"))]
+    #[snafu(display("Feature not implemented: {feature:?}, {details:?}"))]
     NotImplemented {
+        feature: UnsupportedFeature,
         details: String,
         #[snafu(implicit)]
         location: Location,
     },
+    #[snafu(display("Missing volume with ident: '{name:?}'"))]
+    MissingVolume {
+        name: String,
+        #[snafu(implicit)]
+        location: Location,
+    },
+}
+
+#[derive(Debug)]
+pub enum UnsupportedFeature {
+    DropS3TablesDatabase,
+    CreateS3TablesDatabase,
 }