feat(playground): add memory catalog support

crates/integrations/playground/src/catalog.rs

@@ -24,6 +24,7 @@ use anyhow::anyhow;
 use datafusion::catalog::{CatalogProvider, CatalogProviderList};
 use fs_err::read_to_string;
 use iceberg::CatalogBuilder;
+use iceberg::memory::MemoryCatalogBuilder;
 use iceberg_catalog_rest::RestCatalogBuilder;
 use iceberg_datafusion::IcebergCatalogProvider;
 use toml::{Table as TomlTable, Value};
@@ -78,10 +79,6 @@ impl IcebergCatalogList {
             .as_str()
             .ok_or_else(|| anyhow::anyhow!("type is not string"))?;
 
-        if r#type != "rest" {
-            return Err(anyhow::anyhow!("Only rest catalog is supported for now!"));
-        }
-
         let catalog_config = config
             .get("config")
             .ok_or_else(|| anyhow::anyhow!("config not found for catalog {name}"))?
@@ -96,11 +93,21 @@ impl IcebergCatalogList {
                 .ok_or_else(|| anyhow::anyhow!("props {key} is not string"))?;
             props.insert(key.to_string(), value_str.to_string());
         }
-        let catalog = RestCatalogBuilder::default().load(name, props).await?;
+
+        // Create catalog based on type using the appropriate builder
+        let catalog: Arc<dyn iceberg::Catalog> = match r#type {
+            "rest" => Arc::new(RestCatalogBuilder::default().load(name, props).await?),
+            "memory" => Arc::new(MemoryCatalogBuilder::default().load(name, props).await?),
+            _ => {
+                return Err(anyhow::anyhow!(
+                    "Unsupported catalog type: '{type}'. Supported types: rest, memory"
+                ));
+            }
+        };
 
         Ok((
             name.to_string(),
-            Arc::new(IcebergCatalogProvider::try_new(Arc::new(catalog)).await?),
+            Arc::new(IcebergCatalogProvider::try_new(catalog).await?),
         ))
     }
 }
@@ -129,3 +136,74 @@ impl CatalogProviderList for IcebergCatalogList {
             .map(|c| c.clone() as Arc<dyn CatalogProvider>)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_parse_memory_catalog() {
+        let config = r#"
+            [[catalogs]]
+            name = "test_memory"
+            type = "memory"
+            [catalogs.config]
+            warehouse = "/tmp/test-warehouse"
+        "#;
+
+        let toml_table: TomlTable = toml::from_str(config).unwrap();
+        let catalog_list = IcebergCatalogList::parse_table(&toml_table).await.unwrap();
+
+        assert!(
+            catalog_list
+                .catalog_names()
+                .contains(&"test_memory".to_string())
+        );
+        assert!(catalog_list.catalog("test_memory").is_some());
+    }
+
+    #[tokio::test]
+    async fn test_parse_unsupported_catalog_type() {
+        let config = r#"
+            [[catalogs]]
+            name = "test_hive"
+            type = "hive"
+            [catalogs.config]
+            uri = "thrift://localhost:9083"
+        "#;
+
+        let toml_table: TomlTable = toml::from_str(config).unwrap();
+        let result = IcebergCatalogList::parse_table(&toml_table).await;
+
+        assert!(result.is_err());
+        let err_msg = result.unwrap_err().to_string();
+        assert!(err_msg.contains("Unsupported catalog type"));
+        assert!(err_msg.contains("hive"));
+        assert!(err_msg.contains("rest, memory"));
+    }
+
+    #[tokio::test]
+    async fn test_catalog_names() {
+        let config = r#"
+            [[catalogs]]
+            name = "catalog_one"
+            type = "memory"
+            [catalogs.config]
+            warehouse = "/tmp/warehouse1"
+
+            [[catalogs]]
+            name = "catalog_two"
+            type = "memory"
+            [catalogs.config]
+            warehouse = "/tmp/warehouse2"
+        "#;
+
+        let toml_table: TomlTable = toml::from_str(config).unwrap();
+        let catalog_list = IcebergCatalogList::parse_table(&toml_table).await.unwrap();
+
+        let names = catalog_list.catalog_names();
+        assert_eq!(names.len(), 2);
+        assert!(names.contains(&"catalog_one".to_string()));
+        assert!(names.contains(&"catalog_two".to_string()));
+    }
+}