Statistics-augmented Catalog Service API and DataFusion connector

Cargo.toml

@@ -1,20 +1,11 @@
 [workspace]
 resolver = "2"
-members = [
-    "cli",
-    "connectors/datafusion",
-    "optd/catalog",
-    "optd/core",
-    "optd/storage",
-]
+members = ["cli", "connectors/datafusion", "optd/core", "optd/catalog"]
 
-# By default, only compiles the `optd-core` crate.
 default-members = ["optd/core"]
 
 [workspace.dependencies]
 
-optd-datafusion = { path = "connectors/datafusion" }
-
 tokio = { version = "1.47", features = ["macros", "rt", "sync"] }
 tracing = "0.1"
 

cli/Cargo.toml

@@ -23,3 +23,9 @@ object_store = "0.12.3"
 url = "2.5.4"
 tracing-subscriber = { version = "0.3.19", features = ["env-filter"] }
 tracing = { workspace = true }
+
+futures = "0.3.31"
+optd-catalog = { path = "../optd/catalog", version = "0.1" }
+
+[dev-dependencies]
+tempfile = "3"

cli/smoke_test_cli.sh

@@ -0,0 +1,57 @@
+#!/usr/bin/env bash
+# CLI smoke test - verifies catalog integration is active
+
+set -e  # Exit on error
+
+GREEN='\033[0;32m'
+RED='\033[0;31m'
+RESET='\033[0m'
+
+echo "=== CLI Smoke Test ==="
+
+# Build
+echo "Building..."
+cargo build --package optd-cli --quiet
+if [ ! -f ./target/debug/optd-cli ]; then
+    echo -e "${RED}✗ Build failed${RESET}"
+    exit 1
+fi
+
+CLI=./target/debug/optd-cli
+
+# Test 1: Basic functionality
+echo "Test 1: Basic query execution"
+output=$($CLI -c "SELECT 1 as test;" 2>&1)
+if [ $? -eq 0 ] && echo "$output" | grep -q "OptD catalog"; then
+    echo -e "${GREEN}✓ PASS${RESET} - CLI runs, catalog integration active"
+else
+    echo -e "${RED}✗ FAIL${RESET}"
+    exit 1
+fi
+
+# Test 2: Session persistence (multiple commands)
+echo "Test 2: Session state persistence"
+output=$($CLI -c "CREATE TABLE t (x INT);" -c "INSERT INTO t VALUES (1);" -c "SELECT * FROM t;" 2>&1)
+if [ $? -eq 0 ] && echo "$output" | grep -q "1 row"; then
+    echo -e "${GREEN}✓ PASS${RESET} - Multiple commands work, session persists"
+else
+    echo -e "${RED}✗ FAIL${RESET}"
+    exit 1
+fi
+
+# Test 3: Metadata path configuration
+echo "Test 3: Metadata path environment variable"
+TMPDIR_PATH=$(mktemp -d)
+export OPTD_METADATA_CATALOG_PATH="$TMPDIR_PATH/test.ducklake"
+output=$($CLI -c "SELECT 1;" 2>&1)
+unset OPTD_METADATA_CATALOG_PATH
+rm -rf "$TMPDIR_PATH"
+if echo "$output" | grep -q "Using OptD catalog with metadata path"; then
+    echo -e "${GREEN}✓ PASS${RESET} - Metadata path recognized"
+else
+    echo -e "${RED}✗ FAIL${RESET}"
+    exit 1
+fi
+
+echo ""
+echo -e "${GREEN}✓ All smoke tests passed!${RESET}"

cli/src/lib.rs

@@ -1,11 +1,14 @@
-use std::sync::Arc;
-
 use datafusion::{
+    common::{DataFusionError, Result, exec_err, not_impl_err},
+    datasource::TableProvider,
     execution::{SessionStateBuilder, runtime_env::RuntimeEnv},
+    logical_expr::{CreateExternalTable, LogicalPlanBuilder},
     prelude::{DataFrame, SessionConfig, SessionContext},
+    sql::TableReference,
 };
 use datafusion_cli::cli_context::CliSessionContext;
 use optd_datafusion::{OptdExtensionConfig, SessionStateBuilderOptdExt};
+use std::sync::Arc;
 
 pub struct OptdCliSessionContext {
     inner: SessionContext,
@@ -39,10 +42,62 @@ impl OptdCliSessionContext {
         &self.inner
     }
 
-    pub fn return_empty_dataframe(&self) -> datafusion::common::Result<DataFrame> {
-        let plan = datafusion::logical_expr::LogicalPlanBuilder::empty(false).build()?;
+    pub fn return_empty_dataframe(&self) -> Result<DataFrame> {
+        let plan = LogicalPlanBuilder::empty(false).build()?;
         Ok(DataFrame::new(self.inner.state(), plan))
     }
+
+    async fn create_external_table(&self, cmd: &CreateExternalTable) -> Result<DataFrame> {
+        let exist = self.inner.table_exist(cmd.name.clone())?;
+
+        if cmd.temporary {
+            return not_impl_err!("Temporary tables not supported");
+        }
+
+        if exist {
+            match cmd.if_not_exists {
+                true => return self.return_empty_dataframe(),
+                false => {
+                    return exec_err!("Table '{}' already exists", cmd.name);
+                }
+            }
+        }
+
+        let table_provider: Arc<dyn TableProvider> = self.create_custom_table(cmd).await?;
+        self.register_table(cmd.name.clone(), table_provider)?;
+
+        self.return_empty_dataframe()
+    }
+
+    async fn create_custom_table(
+        &self,
+        cmd: &CreateExternalTable,
+    ) -> Result<Arc<dyn TableProvider>> {
+        let state = self.inner.state_ref().read().clone();
+        let file_type = cmd.file_type.to_uppercase();
+        let factory = state
+            .table_factories()
+            .get(file_type.as_str())
+            .ok_or_else(|| {
+                DataFusionError::Execution(format!("Unable to find factory for {}", cmd.file_type))
+            })?;
+        let table = (*factory).create(&state, cmd).await?;
+        Ok(table)
+    }
+
+    pub fn register_table(
+        &self,
+        table_ref: impl Into<TableReference>,
+        provider: Arc<dyn TableProvider>,
+    ) -> Result<Option<Arc<dyn TableProvider>>> {
+        let table_ref: TableReference = table_ref.into();
+        let table = table_ref.table().to_owned();
+        self.inner
+            .state_ref()
+            .read()
+            .schema_for_ref(table_ref)?
+            .register_table(table, provider)
+    }
 }
 
 impl CliSessionContext for OptdCliSessionContext {
@@ -72,12 +127,8 @@ impl CliSessionContext for OptdCliSessionContext {
         plan: datafusion::logical_expr::LogicalPlan,
     ) -> ::core::pin::Pin<
         Box<
-            dyn ::core::future::Future<
-                    Output = Result<
-                        datafusion::prelude::DataFrame,
-                        datafusion::common::DataFusionError,
-                    >,
-                > + ::core::marker::Send
+            dyn ::core::future::Future<Output = Result<DataFrame, DataFusionError>>
+                + ::core::marker::Send
                 + 'async_trait,
         >,
     >
@@ -102,8 +153,14 @@ impl CliSessionContext for OptdCliSessionContext {
                     }
                     _ => (),
                 }
+            } else if let datafusion::logical_expr::LogicalPlan::Ddl(ddl) = &plan {
+                match ddl {
+                    datafusion::logical_expr::DdlStatement::CreateExternalTable(create_table) => {
+                        return self.create_external_table(&create_table).await;
+                    }
+                    _ => (),
+                }
             }
-
             self.inner.execute_logical_plan(plan).await
         };
 

cli/src/main.rs

@@ -44,7 +44,9 @@ use datafusion::common::config_err;
 use datafusion::config::ConfigOptions;
 use datafusion::execution::disk_manager::{DiskManagerBuilder, DiskManagerMode};
 
+use optd_catalog::{CatalogService, DuckLakeCatalog};
 use optd_cli::OptdCliSessionContext;
+use optd_datafusion::OptdCatalogProviderList;
 
 #[derive(Debug, Parser, PartialEq)]
 #[clap(author, version, about, long_about= None)]
@@ -214,11 +216,35 @@ async fn main_inner() -> Result<()> {
     let cli_ctx = cli_ctx.enable_url_table();
     let ctx = cli_ctx.inner();
 
+    // Initialize catalog with optional DuckLake catalog service
+    let catalog_handle = if let Ok(metadata_path) = env::var("OPTD_METADATA_CATALOG_PATH") {
+        if !args.quiet {
+            println!("Using OptD catalog with metadata path: {}", metadata_path);
+        }
+        let ducklake_catalog = DuckLakeCatalog::try_new(None, Some(&metadata_path))
+            .map_err(|e| DataFusionError::External(Box::new(e)))?;
+        let (service, handle) = CatalogService::new(ducklake_catalog);
+        tokio::spawn(async move { service.run().await });
+        Some(handle)
+    } else {
+        if !args.quiet {
+            println!("OptD catalog integration enabled (no persistent metadata)");
+        }
+        None
+    };
+
+    // Wrap the catalog list with OptdCatalogProviderList
+    let original_catalog_list = ctx.state().catalog_list().clone();
+    let optd_catalog_list =
+        OptdCatalogProviderList::new(original_catalog_list.clone(), catalog_handle);
+
     // install dynamic catalog provider that can register required object stores
-    ctx.register_catalog_list(Arc::new(DynamicObjectStoreCatalog::new(
-        ctx.state().catalog_list().clone(),
+    // and wrap it with OptD catalog provider
+    let dynamic_catalog = Arc::new(DynamicObjectStoreCatalog::new(
+        Arc::new(optd_catalog_list),
         ctx.state_weak_ref(),
-    )));
+    ));
+    ctx.register_catalog_list(dynamic_catalog);
 
     // register `parquet_metadata` table function to get metadata from parquet files
     ctx.register_udtf("parquet_metadata", Arc::new(ParquetMetadataFunc {}));