feat(catalog): impl builder for SqlCatalog

Cargo.toml

@@ -78,6 +78,7 @@ http = "1.2"
 iceberg = { version = "0.7.0", path = "./crates/iceberg" }
 iceberg-catalog-glue = { version = "0.7.0", path = "./crates/catalog/glue" }
 iceberg-catalog-hms = { version = "0.7.0", path = "./crates/catalog/hms" }
+iceberg-catalog-sql = { version = "0.7.0", path = "./crates/catalog/sql" }
 iceberg-catalog-rest = { version = "0.7.0", path = "./crates/catalog/rest" }
 iceberg-catalog-s3tables = { version = "0.7.0", path = "./crates/catalog/s3tables" }
 iceberg-datafusion = { version = "0.7.0", path = "./crates/integrations/datafusion" }
@@ -111,6 +112,7 @@ serde_repr = "0.1.16"
 serde_with = "3.4"
 smol = "2.0.2"
 sqllogictest = "0.28.3"
+sqlx = { version = "0.8.1", default-features = false }
 stacker = "0.1.20"
 strum = "0.27.2"
 tempfile = "3.18"

crates/catalog/loader/Cargo.toml

@@ -34,5 +34,10 @@ iceberg-catalog-rest = { workspace = true }
 iceberg-catalog-glue = { workspace = true }
 iceberg-catalog-s3tables = { workspace = true }
 iceberg-catalog-hms = { workspace = true }
+iceberg-catalog-sql = { workspace = true }
 tokio = { workspace = true }
 async-trait = { workspace = true }
+
+[dev-dependencies]
+sqlx = { workspace = true, features = ["runtime-tokio", "sqlite", "migrate"] }
+tempfile = { workspace = true }

crates/catalog/loader/src/lib.rs

@@ -24,6 +24,7 @@ use iceberg_catalog_glue::GlueCatalogBuilder;
 use iceberg_catalog_hms::HmsCatalogBuilder;
 use iceberg_catalog_rest::RestCatalogBuilder;
 use iceberg_catalog_s3tables::S3TablesCatalogBuilder;
+use iceberg_catalog_sql::SqlCatalogBuilder;
 
 /// A CatalogBuilderFactory creating a new catalog builder.
 type CatalogBuilderFactory = fn() -> Box<dyn BoxedCatalogBuilder>;
@@ -34,6 +35,7 @@ static CATALOG_REGISTRY: &[(&str, CatalogBuilderFactory)] = &[
     ("glue", || Box::new(GlueCatalogBuilder::default())),
     ("s3tables", || Box::new(S3TablesCatalogBuilder::default())),
     ("hms", || Box::new(HmsCatalogBuilder::default())),
+    ("sql", || Box::new(SqlCatalogBuilder::default())),
 ];
 
 /// Return the list of supported catalog types.
@@ -108,6 +110,9 @@ impl CatalogLoader<'_> {
 mod tests {
     use std::collections::HashMap;
 
+    use sqlx::migrate::MigrateDatabase;
+    use tempfile::TempDir;
+
     use crate::{CatalogLoader, load};
 
     #[tokio::test]
@@ -220,6 +225,35 @@ mod tests {
         assert!(catalog.is_ok());
     }
 
+    fn temp_path() -> String {
+        let temp_dir = TempDir::new().unwrap();
+        temp_dir.path().to_str().unwrap().to_string()
+    }
+
+    #[tokio::test]
+    async fn test_catalog_loader_pattern_sql_catalog() {
+        use iceberg_catalog_sql::{SQL_CATALOG_PROP_URI, SQL_CATALOG_PROP_WAREHOUSE};
+
+        let uri = format!("sqlite:{}", temp_path());
+        sqlx::Sqlite::create_database(&uri).await.unwrap();
+
+        let catalog_loader = load("sql").unwrap();
+        let catalog = catalog_loader
+            .load(
+                "sql".to_string(),
+                HashMap::from([
+                    (SQL_CATALOG_PROP_URI.to_string(), uri),
+                    (
+                        SQL_CATALOG_PROP_WAREHOUSE.to_string(),
+                        "s3://warehouse".to_string(),
+                    ),
+                ]),
+            )
+            .await;
+
+        assert!(catalog.is_ok());
+    }
+
     #[tokio::test]
     async fn test_error_message_includes_supported_types() {
         let err = match load("does-not-exist") {

crates/catalog/sql/Cargo.toml

@@ -32,10 +32,9 @@ repository = { workspace = true }
 async-trait = { workspace = true }
 iceberg = { workspace = true }
 sqlx = { version = "0.8.1", features = ["any"], default-features = false }
-typed-builder = { workspace = true }
+strum = { workspace = true }
 
 [dev-dependencies]
-iceberg_test_utils = { path = "../../test_utils", features = ["tests"] }
 itertools = { workspace = true }
 regex = "1.10.5"
 sqlx = { version = "0.8.1", features = [