feat(sqllogictest): Add support for iceberg datafusion sqllogictest integration

crates/sqllogictest/Cargo.toml

@@ -30,13 +30,26 @@ async-trait = { workspace = true }
 datafusion = { workspace = true }
 datafusion-sqllogictest = { workspace = true }
 enum-ordinalize = { workspace = true }
+env_logger = "0.11.8"
+iceberg = { workspace = true }
+iceberg-datafusion = { workspace = true }
 indicatif = { workspace = true }
+log = "0.4.28"
 sqllogictest = { workspace = true }
 toml = { workspace = true }
 serde = { workspace = true }
+tempfile = { workspace = true }
 tracing = { workspace = true }
 tokio = { workspace = true }
 
+[dev-dependencies]
+libtest-mimic = "0.8.1"
+
+[[test]]
+harness = false
+name = "sqllogictests"
+path = "tests/sqllogictests.rs"
+
 [package.metadata.cargo-machete]
 # These dependencies are added to ensure minimal dependency version
 ignored = ["enum-ordinalize"]

crates/sqllogictest/src/engine/datafusion.rs

@@ -15,53 +15,76 @@
 // specific language governing permissions and limitations
 // under the License.
 
+use std::collections::HashMap;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-use anyhow::{Context, anyhow};
 use datafusion::catalog::CatalogProvider;
 use datafusion::prelude::{SessionConfig, SessionContext};
 use datafusion_sqllogictest::DataFusion;
+use iceberg::CatalogBuilder;
+use iceberg::memory::{MEMORY_CATALOG_WAREHOUSE, MemoryCatalogBuilder};
+use iceberg_datafusion::IcebergCatalogProvider;
 use indicatif::ProgressBar;
-use sqllogictest::runner::AsyncDB;
+use tempfile::TempDir;
 use toml::Table as TomlTable;
 
-use crate::engine::EngineRunner;
+use crate::engine::{EngineRunner, run_slt_with_runner};
 use crate::error::Result;
 
 pub struct DataFusionEngine {
-    datafusion: DataFusion,
+    test_data_path: PathBuf,
+    session_context: SessionContext,
 }
 
 #[async_trait::async_trait]
 impl EngineRunner for DataFusionEngine {
     async fn run_slt_file(&mut self, path: &Path) -> Result<()> {
-        let content = std::fs::read_to_string(path)
-            .with_context(|| format!("Failed to read slt file {:?}", path))
-            .map_err(|e| anyhow!(e))?;
+        let ctx = self.session_context.clone();
+        let testdata = self.test_data_path.clone();
 
-        self.datafusion
-            .run(content.as_str())
-            .await
-            .with_context(|| format!("Failed to run slt file {:?}", path))
-            .map_err(|e| anyhow!(e))?;
+        let runner = sqllogictest::Runner::new({
+            move || {
+                let ctx = ctx.clone();
+                let testdata = testdata.clone();
+                async move {
+                    // Everything here is owned; no `self` capture.
+                    Ok(DataFusion::new(ctx, testdata, ProgressBar::new(100)))
+                }
+            }
+        });
 
-        Ok(())
+        run_slt_with_runner(runner, path).await
     }
 }
 
 impl DataFusionEngine {
     pub async fn new(config: TomlTable) -> Result<Self> {
-        let session_config = SessionConfig::new().with_target_partitions(4);
+        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?);
 
         Ok(Self {
-            datafusion: DataFusion::new(ctx, PathBuf::from("testdata"), ProgressBar::new(100)),
+            test_data_path: PathBuf::from("testdata"),
+            session_context: ctx,
         })
     }
 
     async fn create_catalog(_: &TomlTable) -> anyhow::Result<Arc<dyn CatalogProvider>> {
-        todo!()
+        let temp_dir = TempDir::new()?;
+        let path_str = temp_dir.path().to_str().unwrap().to_string();
+
+        let catalog = MemoryCatalogBuilder::default()
+            .load(
+                "memory",
+                HashMap::from([(MEMORY_CATALOG_WAREHOUSE.to_string(), path_str)]),
+            )
+            .await?;
+
+        Ok(Arc::new(
+            IcebergCatalogProvider::try_new(Arc::new(catalog)).await?,
+        ))
     }
 }

crates/sqllogictest/src/engine/mod.rs

@@ -19,70 +19,94 @@ mod datafusion;
 
 use std::path::Path;
 
+use anyhow::anyhow;
+use sqllogictest::{AsyncDB, MakeConnection, Runner, parse_file};
 use toml::Table as TomlTable;
 
 use crate::engine::datafusion::DataFusionEngine;
-use crate::error::Result;
+use crate::error::{Error, Result};
 
-const KEY_TYPE: &str = "type";
 const TYPE_DATAFUSION: &str = "datafusion";
+const ERRS_PER_FILE_LIMIT: usize = 10;
 
 #[async_trait::async_trait]
-pub trait EngineRunner: Sized {
+pub trait EngineRunner: Send {
     async fn run_slt_file(&mut self, path: &Path) -> Result<()>;
 }
 
-pub enum Engine {
-    DataFusion(DataFusionEngine),
+pub async fn load_engine_runner(
+    engine_type: &str,
+    cfg: TomlTable,
+) -> Result<Box<dyn EngineRunner>> {
+    match engine_type {
+        TYPE_DATAFUSION => Ok(Box::new(DataFusionEngine::new(cfg).await?)),
+        _ => Err(anyhow::anyhow!("Unsupported engine type: {}", engine_type).into()),
+    }
 }
 
-impl Engine {
-    pub async fn new(config: TomlTable) -> Result<Self> {
-        let engine_type = config
-            .get(KEY_TYPE)
-            .ok_or_else(|| anyhow::anyhow!("Missing required key: {KEY_TYPE}"))?
-            .as_str()
-            .ok_or_else(|| anyhow::anyhow!("Config value for {KEY_TYPE} must be a string"))?;
-
-        match engine_type {
-            TYPE_DATAFUSION => {
-                let engine = DataFusionEngine::new(config).await?;
-                Ok(Engine::DataFusion(engine))
-            }
-            _ => Err(anyhow::anyhow!("Unsupported engine type: {engine_type}").into()),
+pub async fn run_slt_with_runner<D, M>(
+    mut runner: Runner<D, M>,
+    step_slt_file: impl AsRef<Path>,
+) -> Result<()>
+where
+    D: AsyncDB + Send + 'static,
+    M: MakeConnection<Conn = D> + Send + 'static,
+{
+    let path = step_slt_file.as_ref().canonicalize()?;
+
+    let records = parse_file(&path).map_err(|e| Error(anyhow!("parsing SLT file failed: {e}")))?;
+
+    let mut errs = vec![];
+    for record in records {
+        if let Err(err) = runner.run_async(record).await {
+            errs.push(format!("{err}"));
         }
     }
 
-    pub async fn run_slt_file(&mut self, path: &Path) -> Result<()> {
-        match self {
-            Engine::DataFusion(engine) => engine.run_slt_file(path).await,
+    if !errs.is_empty() {
+        let mut msg = format!("{} errors in file {}\n\n", errs.len(), path.display());
+        for (i, err) in errs.iter().enumerate() {
+            if i >= ERRS_PER_FILE_LIMIT {
+                msg.push_str(&format!(
+                    "... other {} errors in {} not shown ...\n\n",
+                    errs.len() - ERRS_PER_FILE_LIMIT,
+                    path.display()
+                ));
+                break;
+            }
+            msg.push_str(&format!("{}. {err}\n\n", i + 1));
         }
+        return Err(Error(anyhow!(msg)));
     }
+
+    Ok(())
 }
 
 #[cfg(test)]
 mod tests {
-    use toml::Table as TomlTable;
-
-    use crate::engine::Engine;
+    use crate::engine::{TYPE_DATAFUSION, load_engine_runner};
 
     #[tokio::test]
-    async fn test_engine_new_missing_type_key() {
-        let config = TomlTable::new();
-        let result = Engine::new(config).await;
+    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;
 
         assert!(result.is_err());
     }
 
     #[tokio::test]
-    async fn test_engine_invalid_type() {
+    async fn test_load_datafusion() {
         let input = r#"
             [engines]
-            random = { type = "random_engine", url = "http://localhost:8181" }
+            df = { type = "datafusion" }
         "#;
         let tbl = toml::from_str(input).unwrap();
-        let result = Engine::new(tbl).await;
+        let result = load_engine_runner(TYPE_DATAFUSION, tbl).await;
 
-        assert!(result.is_err());
+        assert!(result.is_ok());
     }
 }

crates/sqllogictest/src/error.rs

@@ -44,3 +44,9 @@ impl From<anyhow::Error> for Error {
         Self(value)
     }
 }
+
+impl From<std::io::Error> for Error {
+    fn from(value: std::io::Error) -> Self {
+        Self(value.into())
+    }
+}

crates/sqllogictest/src/lib.rs

@@ -18,9 +18,6 @@
 // This lib contains codes copied from
 // [Apache Datafusion](https://github.com/apache/datafusion/tree/main/datafusion/sqllogictest)
 
-#[allow(dead_code)]
 mod engine;
-#[allow(dead_code)]
-mod error;
-#[allow(dead_code)]
-mod schedule;
+pub mod error;
+pub mod schedule;

crates/sqllogictest/src/schedule.rs

@@ -24,11 +24,11 @@ use serde::{Deserialize, Serialize};
 use toml::{Table as TomlTable, Value};
 use tracing::info;
 
-use crate::engine::Engine;
+use crate::engine::{EngineRunner, load_engine_runner};
 
 pub struct Schedule {
     /// Engine names to engine instances
-    engines: HashMap<String, Engine>,
+    engines: HashMap<String, Box<dyn EngineRunner>>,
     /// List of test steps to run
     steps: Vec<Step>,
     /// Path of the schedule file
@@ -44,7 +44,11 @@ pub struct Step {
 }
 
 impl Schedule {
-    pub fn new(engines: HashMap<String, Engine>, steps: Vec<Step>, schedule_file: String) -> Self {
+    pub fn new(
+        engines: HashMap<String, Box<dyn EngineRunner>>,
+        steps: Vec<Step>,
+        schedule_file: String,
+    ) -> Self {
         Self {
             engines,
             steps,
@@ -102,7 +106,9 @@ impl Schedule {
         Ok(())
     }
 
-    async fn parse_engines(table: &TomlTable) -> anyhow::Result<HashMap<String, Engine>> {
+    async fn parse_engines(
+        table: &TomlTable,
+    ) -> anyhow::Result<HashMap<String, Box<dyn EngineRunner>>> {
         let engines_tbl = table
             .get("engines")
             .with_context(|| "Schedule file must have an 'engines' table")?
@@ -117,9 +123,13 @@ impl Schedule {
                 .ok_or_else(|| anyhow!("Config of engine '{name}' is not a table"))?
                 .clone();
 
-            let engine = Engine::new(cfg_tbl)
-                .await
-                .with_context(|| format!("Failed to construct engine '{name}'"))?;
+            let engine_type = cfg_tbl
+                .get("type")
+                .ok_or_else(|| anyhow::anyhow!("Engine {name} doesn't have a 'type' field"))?
+                .as_str()
+                .ok_or_else(|| anyhow::anyhow!("Engine {name} type must be a string"))?;
+
+            let engine = load_engine_runner(engine_type, cfg_tbl.clone()).await?;
 
             if engines.insert(name.clone(), engine).is_some() {
                 return Err(anyhow!("Duplicate engine '{name}'"));

crates/sqllogictest/testdata/schedules/df_test.toml

@@ -0,0 +1,26 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[engines]
+df = { type = "datafusion" }
+
+[catalog]
+memory = { type = "in-memory" }
+
+[[steps]]
+engine = "df"
+slt = "df_test/show_tables.slt"

crates/sqllogictest/testdata/slts/df_test/show_tables.slt

@@ -0,0 +1,34 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+query TTTT rowsort
+SHOW TABLES
+----
+datafusion information_schema columns VIEW
+datafusion information_schema df_settings VIEW
+datafusion information_schema parameters VIEW
+datafusion information_schema routines VIEW
+datafusion information_schema schemata VIEW
+datafusion information_schema tables VIEW
+datafusion information_schema views VIEW
+default information_schema columns VIEW
+default information_schema df_settings VIEW
+default information_schema parameters VIEW
+default information_schema routines VIEW
+default information_schema schemata VIEW
+default information_schema tables VIEW
+default information_schema views VIEW