laminar_connectors/lakehouse/delta_table_provider/
mod.rs

1//! Delta Lake table provider integration with `DataFusion`.
2//!
3//! This module provides a thin helper to open a Delta Lake table and
4//! register it as a `TableProvider` in a `SessionContext`.
5//!
6//! # Usage
7//!
8//! ```rust,ignore
9//! use laminar_connectors::lakehouse::delta_table_provider::register_delta_table;
10//! use datafusion::prelude::SessionContext;
11//! use std::collections::HashMap;
12//!
13//! let ctx = SessionContext::new();
14//! register_delta_table(&ctx, "my_table", "/path/to/delta/table", HashMap::new()).await?;
15//!
16//! // Now query it:
17//! let df = ctx.sql("SELECT * FROM my_table").await?;
18//! ```
19
20#[cfg(feature = "delta-lake")]
21use std::collections::HashMap;
22
23#[cfg(feature = "delta-lake")]
24use std::sync::Arc;
25
26#[cfg(feature = "delta-lake")]
27use datafusion::prelude::SessionContext;
28
29#[cfg(feature = "delta-lake")]
30use tracing::info;
31
32#[cfg(feature = "delta-lake")]
33use crate::error::ConnectorError;
34
35/// Opens a Delta Lake table and registers it as a table provider in the
36/// given `DataFusion` `SessionContext`.
37///
38/// # Arguments
39///
40/// * `ctx` - The `DataFusion` session context to register in
41/// * `name` - The SQL table name (e.g., `"trades"`)
42/// * `table_uri` - Path to the Delta Lake table (local, `s3://`, `az://`, `gs://`)
43/// * `storage_options` - Storage credentials and configuration
44///
45/// # Errors
46///
47/// Returns `ConnectorError::ConnectionFailed` if the table cannot be opened,
48/// or `ConnectorError::Internal` if registration fails.
49#[cfg(feature = "delta-lake")]
50#[allow(clippy::implicit_hasher)]
51pub async fn register_delta_table(
52    ctx: &SessionContext,
53    name: &str,
54    table_uri: &str,
55    storage_options: HashMap<String, String>,
56) -> Result<(), ConnectorError> {
57    use super::delta_io;
58
59    // The URI is deliberately excluded: validation below rejects signed queries, but this log
60    // occurs before validation and must never publish one.
61    info!(name, "registering Delta Lake table as TableProvider");
62
63    // Open the existing table.
64    let table = delta_io::open_or_create_table(table_uri, storage_options, None).await?;
65
66    // Register the table's object store with the session so scans can resolve
67    // non-local URLs (s3://, az://, gs://); without it, reading an
68    // object-store-backed table fails with "No suitable object store found".
69    // (Local-filesystem tables use DataFusion's built-in store.)
70    table
71        .update_datafusion_session(&ctx.state())
72        .map_err(|e| ConnectorError::Internal(format!("register Delta object store: {e}")))?;
73
74    // Build a DeltaTableProvider (which implements TableProvider) from the table.
75    let provider =
76        table.table_provider().build().await.map_err(|e| {
77            ConnectorError::Internal(format!("failed to build table provider: {e}"))
78        })?;
79
80    ctx.register_table(
81        datafusion::common::TableReference::bare(name),
82        Arc::new(provider),
83    )
84    .map_err(|e| {
85        ConnectorError::Internal(format!("failed to register Delta table '{name}': {e}"))
86    })?;
87
88    info!(name, "Delta Lake table registered successfully");
89
90    Ok(())
91}
92
93#[cfg(all(test, feature = "delta-lake"))]
94mod tests;
laminar_connectors/lakehouse/delta_table_provider/mod.rs

laminar_connectors/lakehouse/delta_table_provider/
mod.rs
