Implement PostgreSQL Metadata Provider (#20)

Author: shefeek-jinnah

Cargo.toml

@@ -26,14 +26,22 @@ futures = "0.3"
 tracing = "0.1"
 thiserror = "2.0"
 
+# Multi-database metadata providers (optional)
+sqlx = { version = "0.8", features = ["runtime-tokio"], optional = true }
+
 [dev-dependencies]
 testcontainers = "0.23"
-testcontainers-modules = { version = "0.11", features = ["minio"] }
+testcontainers-modules = { version = "0.11", features = ["minio", "postgres", "mysql"] }
 tempfile = "3.14"
 anyhow = "1.0"
 aws-sdk-s3 = "1.75"
 aws-credential-types = "1.2"
 
 [features]
 # Allow skipping tests that use docker (in CI, macos doesn't support docker).
-skip-tests-with-docker = []
+skip-tests-with-docker = []
+
+# Metadata provider backends
+metadata-postgres = ["sqlx", "sqlx/postgres", "sqlx/chrono"]
+# Future: metadata-sqlite = ["sqlx", "sqlx/sqlite", "sqlx/chrono"]
+# Future: metadata-mysql = ["sqlx", "sqlx/mysql", "sqlx/chrono"]

examples/basic_query.rs

@@ -1,7 +1,7 @@
 //! Basic DuckLake query example with snapshot isolation
 //!
 //! This example demonstrates how to:
-//! 1. Create a DuckLake catalog from a DuckDB catalog file
+//! 1. Create a DuckLake catalog from DuckDB or PostgreSQL
 //! 2. Bind the catalog to a specific snapshot for query consistency
 //! 3. Register it with DataFusion
 //! 4. Execute a simple SELECT query
@@ -16,14 +16,24 @@
 //! To query data at different points in time, create separate catalogs bound to
 //! different snapshot IDs.
 //!
-//! To run this example, you need:
-//! - A DuckDB database file with DuckLake tables
-//! - Parquet data files referenced by the catalog
+//! ## Usage
 //!
-//! Usage: cargo run --example basic_query <catalog.db> <sql>
+//! With DuckDB catalog:
+//! ```bash
+//! cargo run --example basic_query catalog.db "SELECT * FROM main.users"
+//! ```
+//!
+//! With PostgreSQL catalog (requires --features metadata-postgres):
+//! ```bash
+//! cargo run --example basic_query --features metadata-postgres \
+//!   "postgresql://user:password@localhost:5432/postgres" \
+//!   "SELECT * FROM main.users"
+//! ```
 
 use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::prelude::*;
+#[cfg(feature = "metadata-postgres")]
+use datafusion_ducklake::PostgresMetadataProvider;
 use datafusion_ducklake::{
     DuckLakeCatalog, DuckdbMetadataProvider, MetadataProvider, register_ducklake_functions,
 };
@@ -38,25 +48,51 @@ use url::Url;
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let args: Vec<String> = env::args().collect();
     if args.len() < 3 {
-        eprintln!("Usage: cargo run --example basic_query catalog.db sql");
+        eprintln!("Usage:");
+        eprintln!("  DuckDB:     cargo run --example basic_query catalog.db \"SQL\"");
+        eprintln!(
+            "  PostgreSQL: cargo run --example basic_query --features metadata-postgres \"postgresql://...\" \"SQL\""
+        );
         exit(1);
     }
-    let catalog_path = &args[1];
+    let catalog_source = &args[1];
     let sql = &args[2];
 
-    // // Path to your DuckLake catalog database
-    // let catalog_path = "test_catalog.db";
+    // Detect provider type based on input
+    let is_postgres = catalog_source.starts_with("postgresql://");
 
-    println!("Connecting to DuckLake catalog: {}", catalog_path);
+    if is_postgres {
+        #[cfg(not(feature = "metadata-postgres"))]
+        {
+            eprintln!("Error: PostgreSQL support requires the 'metadata-postgres' feature");
+            eprintln!("Run with: cargo run --example basic_query --features metadata-postgres");
+            exit(1);
+        }
 
-    // Create the metadata provider
-    let provider = Arc::new(DuckdbMetadataProvider::new(catalog_path)?);
+        #[cfg(feature = "metadata-postgres")]
+        {
+            println!("Connecting to PostgreSQL catalog: {}", catalog_source);
+            let provider = Arc::new(PostgresMetadataProvider::new(catalog_source).await?);
+            let snapshot_id = provider.get_current_snapshot()?;
+            println!("Current snapshot ID: {}", snapshot_id);
+            run_query(provider, snapshot_id, sql).await?;
+        }
+    } else {
+        println!("Connecting to DuckDB catalog: {}", catalog_source);
+        let provider = Arc::new(DuckdbMetadataProvider::new(catalog_source)?);
+        let snapshot_id = provider.get_current_snapshot()?;
+        println!("Current snapshot ID: {}", snapshot_id);
+        run_query(provider, snapshot_id, sql).await?;
+    }
 
-    // Get the current snapshot ID
-    // This ensures query consistency - all metadata lookups will use this snapshot
-    let snapshot_id = provider.get_current_snapshot()?;
-    println!("Current snapshot ID: {}", snapshot_id);
+    Ok(())
+}
 
+async fn run_query(
+    provider: Arc<dyn MetadataProvider>,
+    snapshot_id: i64,
+    sql: &str,
+) -> Result<(), Box<dyn std::error::Error>> {
     // Create runtime and register object stores
     // For MinIO or S3, register the object store with the runtime
     let runtime = Arc::new(RuntimeEnv::default());
@@ -77,11 +113,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Create the DuckLake catalog bound to the snapshot
     // This ensures all queries through this catalog see consistent data
     // from this specific snapshot, even if the underlying data changes
-    let ducklake_catalog = DuckLakeCatalog::with_snapshot(provider, snapshot_id)?;
-
-    // Alternative: Use the backward-compatible constructor that automatically
-    // binds to the current snapshot:
-    // let ducklake_catalog = DuckLakeCatalog::new(DuckdbMetadataProvider::new(catalog_path)?)?;
+    let ducklake_catalog = DuckLakeCatalog::with_snapshot(provider.clone(), snapshot_id)?;
 
     println!("✓ Connected to DuckLake catalog");
 
@@ -90,9 +122,6 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Create DataFusion session context
     let ctx = SessionContext::new_with_config_rt(config, runtime.clone());
 
-    // Get the provider before moving the catalog
-    let provider = ducklake_catalog.provider();
-
     // Register the DuckLake catalog (standard DataFusion pattern)
     ctx.register_catalog("ducklake", Arc::new(ducklake_catalog));
 
@@ -121,17 +150,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     }
 
-    // Example query (adjust schema and table names to match your data)
-    // Uncomment and modify this once you have actual DuckLake data:
-
+    // Execute the query
     println!("\nExecuting query...");
     let df = ctx.sql(sql).await?;
 
     // Show the query results
     df.show().await?;
 
     println!("\n✓ Example completed successfully!");
-    println!("\nTo run a query, create a DuckLake database and uncomment the query section.");
 
     Ok(())
 }

src/error.rs

@@ -17,6 +17,11 @@ pub enum DuckLakeError {
     #[error("DuckDB error: {0}")]
     DuckDb(#[from] duckdb::Error),
 
+    /// sqlx database error (for PostgreSQL metadata provider)
+    #[cfg(feature = "metadata-postgres")]
+    #[error("Database error: {0}")]
+    Sqlx(#[from] sqlx::Error),
+
     /// Catalog not found
     #[error("Catalog not found: {0}")]
     CatalogNotFound(String),

src/lib.rs

@@ -47,6 +47,10 @@ pub mod table;
 pub mod table_functions;
 pub mod types;
 
+// Multi-database metadata providers (optional, feature-gated)
+#[cfg(feature = "metadata-postgres")]
+pub mod metadata_provider_postgres;
+
 // Result type for DuckLake operations
 pub type Result<T> = std::result::Result<T, DuckLakeError>;
 
@@ -58,3 +62,7 @@ pub use metadata_provider_duckdb::DuckdbMetadataProvider;
 pub use schema::DuckLakeSchema;
 pub use table::DuckLakeTable;
 pub use table_functions::register_ducklake_functions;
+
+// Re-export multi-database metadata providers (feature-gated)
+#[cfg(feature = "metadata-postgres")]
+pub use metadata_provider_postgres::PostgresMetadataProvider;

src/metadata_provider.rs

@@ -324,3 +324,12 @@ pub trait MetadataProvider: Send + Sync + std::fmt::Debug {
     /// List all files across all tables for a snapshot
     fn list_all_files(&self, snapshot_id: i64) -> Result<Vec<FileWithTable>>;
 }
+
+#[cfg(feature = "metadata-postgres")]
+/// Helper function to bridge async sqlx operations to sync MetadataProvider trait
+pub(crate) fn block_on<F, T>(f: F) -> T
+where
+    F: std::future::Future<Output = T>,
+{
+    tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(f))
+}