feat(sqllogictest): support dynamic catalog configuration

This PR adds support for configuring the catalog type in sqllogictest
schedule files, addressing issue #1780.

Changes:
- Add [catalog] section parsing in schedule files
- Support 'memory' (default) and all catalog types from iceberg-catalog-loader
- Update DataFusionEngine to accept catalog from schedule configuration
- Add comprehensive tests for catalog configuration parsing

The catalog can now be configured in schedule files like:

[catalog]
type = "rest"

[catalog.properties]
uri = "http://localhost:8181"

If no catalog section is specified, defaults to memory catalog.

Closes #1780

Author: AndreaBozzo

Cargo.lock

@@ -81,6 +81,7 @@ http = "1.2"
 iceberg = { version = "0.8.0", path = "./crates/iceberg" }
 iceberg-catalog-glue = { version = "0.8.0", path = "./crates/catalog/glue" }
 iceberg-catalog-hms = { version = "0.8.0", path = "./crates/catalog/hms" }
+iceberg-catalog-loader = { version = "0.8.0", path = "./crates/catalog/loader" }
 iceberg-catalog-rest = { version = "0.8.0", path = "./crates/catalog/rest" }
 iceberg-catalog-s3tables = { version = "0.8.0", path = "./crates/catalog/s3tables" }
 iceberg-catalog-sql = { version = "0.8.0", path = "./crates/catalog/sql" }

Cargo.toml

@@ -32,6 +32,7 @@ datafusion-sqllogictest = { workspace = true }
 enum-ordinalize = { workspace = true }
 env_logger = { workspace = true }
 iceberg = { workspace = true }
+iceberg-catalog-loader = { workspace = true }
 iceberg-datafusion = { workspace = true }
 indicatif = { workspace = true }
 log = { workspace = true }

crates/sqllogictest/Cargo.toml

@@ -19,12 +19,10 @@ use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-use datafusion::catalog::CatalogProvider;
 use datafusion::prelude::{SessionConfig, SessionContext};
 use datafusion_sqllogictest::DataFusion;
-use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
 use iceberg::spec::{NestedField, PrimitiveType, Schema, Transform, Type, UnboundPartitionSpec};
-use iceberg::{Catalog, CatalogBuilder, NamespaceIdent, TableCreation};
+use iceberg::{Catalog, NamespaceIdent, TableCreation};
 use iceberg_datafusion::IcebergCatalogProvider;
 use indicatif::ProgressBar;
 use toml::Table as TomlTable;
@@ -59,49 +57,54 @@ impl EngineRunner for DataFusionEngine {
 }
 
 impl DataFusionEngine {
-    pub async fn new(config: TomlTable) -> Result<Self> {
+    pub async fn new(_config: TomlTable, catalog: Arc<dyn Catalog>) -> Result<Self> {
         let session_config = SessionConfig::new()
             .with_target_partitions(4)
             .with_information_schema(true);
         let ctx = SessionContext::new_with_config(session_config);
-        ctx.register_catalog("default", Self::create_catalog(&config).await?);
+
+        // Create test namespace and tables in the catalog
+        Self::setup_test_data(&catalog).await?;
+
+        // Register the catalog with DataFusion
+        let catalog_provider = IcebergCatalogProvider::try_new(catalog)
+            .await
+            .map_err(|e| {
+                crate::error::Error(anyhow::anyhow!("Failed to create catalog provider: {e}"))
+            })?;
+        ctx.register_catalog("default", Arc::new(catalog_provider));
 
         Ok(Self {
             test_data_path: PathBuf::from("testdata"),
             session_context: ctx,
         })
     }
 
-    async fn create_catalog(_: &TomlTable) -> anyhow::Result<Arc<dyn CatalogProvider>> {
-        // TODO: support dynamic catalog configuration
-        //  See: https://github.com/apache/iceberg-rust/issues/1780
-        let catalog = MemoryCatalogBuilder::default()
-            .load(
-                "memory",
-                HashMap::from([(
-                    MEMORY_CATALOG_WAREHOUSE.to_string(),
-                    "memory://".to_string(),
-                )]),
-            )
-            .await?;
-
+    /// Set up the test namespace and tables in the catalog.
+    async fn setup_test_data(catalog: &Arc<dyn Catalog>) -> anyhow::Result<()> {
         // Create a test namespace for INSERT INTO tests
         let namespace = NamespaceIdent::new("default".to_string());
-        catalog.create_namespace(&namespace, HashMap::new()).await?;
 
-        // Create test tables
-        Self::create_unpartitioned_table(&catalog, &namespace).await?;
-        Self::create_partitioned_table(&catalog, &namespace).await?;
+        // Try to create the namespace, ignore if it already exists
+        if catalog
+            .create_namespace(&namespace, HashMap::new())
+            .await
+            .is_err()
+        {
+            // Namespace might already exist, that's ok
+        }
+
+        // Create test tables (ignore errors if they already exist)
+        let _ = Self::create_unpartitioned_table(catalog, &namespace).await;
+        let _ = Self::create_partitioned_table(catalog, &namespace).await;
 
-        Ok(Arc::new(
-            IcebergCatalogProvider::try_new(Arc::new(catalog)).await?,
-        ))
+        Ok(())
     }
 
     /// Create an unpartitioned test table with id and name columns
     /// TODO: this can be removed when we support CREATE TABLE
     async fn create_unpartitioned_table(
-        catalog: &impl Catalog,
+        catalog: &Arc<dyn Catalog>,
         namespace: &NamespaceIdent,
     ) -> anyhow::Result<()> {
         let schema = Schema::builder()
@@ -128,7 +131,7 @@ impl DataFusionEngine {
     /// Partitioned by category using identity transform
     /// TODO: this can be removed when we support CREATE TABLE
     async fn create_partitioned_table(
-        catalog: &impl Catalog,
+        catalog: &Arc<dyn Catalog>,
         namespace: &NamespaceIdent,
     ) -> anyhow::Result<()> {
         let schema = Schema::builder()

crates/sqllogictest/src/engine/datafusion.rs

@@ -18,8 +18,10 @@
 mod datafusion;
 
 use std::path::Path;
+use std::sync::Arc;
 
 use anyhow::anyhow;
+use iceberg::Catalog;
 use sqllogictest::{AsyncDB, MakeConnection, Runner, parse_file};
 use toml::Table as TomlTable;
 
@@ -36,9 +38,10 @@ pub trait EngineRunner: Send {
 pub async fn load_engine_runner(
     engine_type: &str,
     cfg: TomlTable,
+    catalog: Arc<dyn Catalog>,
 ) -> Result<Box<dyn EngineRunner>> {
     match engine_type {
-        TYPE_DATAFUSION => Ok(Box::new(DataFusionEngine::new(cfg).await?)),
+        TYPE_DATAFUSION => Ok(Box::new(DataFusionEngine::new(cfg, catalog).await?)),
         _ => Err(anyhow::anyhow!("Unsupported engine type: {engine_type}").into()),
     }
 }
@@ -65,16 +68,38 @@ where
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
+    use std::sync::Arc;
+
+    use iceberg::CatalogBuilder;
+    use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
+
     use crate::engine::{TYPE_DATAFUSION, load_engine_runner};
 
+    async fn create_test_catalog() -> Arc<dyn iceberg::Catalog> {
+        Arc::new(
+            MemoryCatalogBuilder::default()
+                .load(
+                    "test",
+                    HashMap::from([(
+                        MEMORY_CATALOG_WAREHOUSE.to_string(),
+                        "memory://".to_string(),
+                    )]),
+                )
+                .await
+                .unwrap(),
+        )
+    }
+
     #[tokio::test]
     async fn test_engine_invalid_type() {
         let input = r#"
             [engines]
             random = { type = "random_engine", url = "http://localhost:8181" }
         "#;
         let tbl = toml::from_str(input).unwrap();
-        let result = load_engine_runner("random_engine", tbl).await;
+        let catalog = create_test_catalog().await;
+        let result = load_engine_runner("random_engine", tbl, catalog).await;
 
         assert!(result.is_err());
     }
@@ -86,7 +111,8 @@ mod tests {
             df = { type = "datafusion" }
         "#;
         let tbl = toml::from_str(input).unwrap();
-        let result = load_engine_runner(TYPE_DATAFUSION, tbl).await;
+        let catalog = create_test_catalog().await;
+        let result = load_engine_runner(TYPE_DATAFUSION, tbl, catalog).await;
 
         assert!(result.is_ok());
     }