faucet_common_delta/
connection.rs

1//! The shared Delta connection block: `table_uri` + `credentials` +
2//! `storage_options`, and the open-table / handler-registration helpers both
3//! the source and sink rely on.
4
5use std::collections::HashMap;
6
7use faucet_core::FaucetError;
8use schemars::JsonSchema;
9use serde::{Deserialize, Serialize};
10
11use crate::credentials::DeltaCredentials;
12
13/// Location + credentials for a Delta table. Flattened (`#[serde(flatten)]`)
14/// into both the source and sink config so the same three keys appear at the
15/// connector-config top level.
16#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
17pub struct DeltaConnection {
18    /// Delta table location URI: `file:///abs/path`, `s3://bucket/key`,
19    /// `abfss://…`, `gs://bucket/key`. A bare local path is accepted and
20    /// treated as `file://`.
21    pub table_uri: String,
22
23    /// Object-store credentials. Defaults to the ambient provider chain.
24    #[serde(default)]
25    pub credentials: DeltaCredentials,
26
27    /// Extra `storage_options` passed verbatim to delta-rs (e.g. a MinIO
28    /// endpoint override, or any object-store key the [`DeltaCredentials`]
29    /// enum does not model). Keys set here **win** over credential-derived
30    /// keys.
31    #[serde(default)]
32    pub storage_options: HashMap<String, String>,
33}
34
35impl DeltaConnection {
36    /// Build the effective `storage_options` map: credential-derived keys with
37    /// the user's explicit `storage_options` layered on top (explicit wins).
38    pub fn merged_storage_options(&self) -> HashMap<String, String> {
39        let mut opts = self.storage_options.clone();
40        self.credentials.apply(&mut opts);
41        opts
42    }
43
44    /// The table URI with any embedded credentials scrubbed — safe for logs,
45    /// lineage, and `dataset_uri()`.
46    pub fn redacted_uri(&self) -> String {
47        faucet_core::util::redact_uri_credentials(&self.table_uri)
48    }
49
50    /// Register the object-store handlers delta-rs needs for this URI's scheme.
51    ///
52    /// delta-rs does not auto-register the cloud object stores; the relevant
53    /// `register_handlers` must run once before opening/creating an `s3://` /
54    /// `abfss://` / `gs://` table. Idempotent and cheap — safe to call before
55    /// every open.
56    pub fn register_handlers(&self) {
57        register_all_handlers();
58    }
59
60    /// Resolve `table_uri` to a `Url`, turning bare local paths into `file://`
61    /// (delta-rs's `open_table*` functions take a `Url`, not a string).
62    fn table_url(&self) -> Result<url::Url, FaucetError> {
63        deltalake::ensure_table_uri(&self.table_uri).map_err(|e| {
64            FaucetError::Config(format!(
65                "delta: invalid table_uri '{}': {e}",
66                self.redacted_uri()
67            ))
68        })
69    }
70
71    /// The resolved table location as a string — for delta-rs builders that
72    /// take a location string (e.g. `CreateBuilder::with_location`).
73    pub fn location_string(&self) -> Result<String, FaucetError> {
74        Ok(self.table_url()?.to_string())
75    }
76
77    /// Open the table at its latest version. Registers handlers first.
78    pub async fn open(&self) -> Result<deltalake::DeltaTable, FaucetError> {
79        self.register_handlers();
80        let url = self.table_url()?;
81        deltalake::open_table_with_storage_options(url, self.merged_storage_options())
82            .await
83            .map_err(|e| self.map_open_error(e))
84    }
85
86    /// Open the table if it exists; return `Ok(None)` when the URI is not (yet)
87    /// a Delta table. Other errors (auth, I/O) propagate. Used by the sink's
88    /// `create_if_not_missing` path.
89    pub async fn open_optional(&self) -> Result<Option<deltalake::DeltaTable>, FaucetError> {
90        self.register_handlers();
91        let url = self.table_url()?;
92        match deltalake::open_table_with_storage_options(url, self.merged_storage_options()).await {
93            Ok(t) => Ok(Some(t)),
94            Err(e) if is_missing_table(&e) => Ok(None),
95            Err(e) => Err(self.map_open_error(e)),
96        }
97    }
98
99    /// Open the table pinned to a specific version (time travel).
100    pub async fn open_at_version(
101        &self,
102        version: u64,
103    ) -> Result<deltalake::DeltaTable, FaucetError> {
104        let mut table = self.open().await?;
105        table
106            .load_version(version)
107            .await
108            .map_err(|e| self.map_open_error(e))?;
109        Ok(table)
110    }
111
112    /// Open the table as of a timestamp (RFC 3339). Time travel.
113    pub async fn open_at_timestamp(
114        &self,
115        timestamp: &str,
116    ) -> Result<deltalake::DeltaTable, FaucetError> {
117        let ts = chrono::DateTime::parse_from_rfc3339(timestamp)
118            .map_err(|e| {
119                FaucetError::Config(format!(
120                    "delta: invalid `timestamp` '{timestamp}' (expected RFC 3339): {e}"
121                ))
122            })?
123            .with_timezone(&chrono::Utc);
124        let mut table = self.open().await?;
125        table
126            .load_with_datetime(ts)
127            .await
128            .map_err(|e| self.map_open_error(e))?;
129        Ok(table)
130    }
131
132    fn map_open_error(&self, e: deltalake::DeltaTableError) -> FaucetError {
133        FaucetError::Source(format!(
134            "delta: failed to open table '{}': {e}",
135            self.redacted_uri()
136        ))
137    }
138}
139
140/// Whether a delta-rs error means "there is no Delta table here (yet)" —
141/// either an explicit `NotATable` or an underlying object-store "not found".
142fn is_missing_table(e: &deltalake::DeltaTableError) -> bool {
143    if matches!(e, deltalake::DeltaTableError::NotATable(_)) {
144        return true;
145    }
146    // A brand-new location surfaces as an object-store NotFound wrapped in a
147    // generic error; match on the rendered message so the create-if-missing
148    // path doesn't treat "empty prefix" as a hard failure.
149    let msg = e.to_string().to_lowercase();
150    msg.contains("not found") || msg.contains("no such file") || msg.contains("does not exist")
151}
152
153/// Register every compiled-in cloud object-store handler exactly once.
154///
155/// Each backend is gated on its cargo feature; with no cloud feature this is a
156/// no-op (local filesystem needs no registration).
157fn register_all_handlers() {
158    #[cfg(feature = "s3")]
159    {
160        use std::sync::Once;
161        static S3: Once = Once::new();
162        S3.call_once(|| deltalake::aws::register_handlers(None));
163    }
164    #[cfg(feature = "azure")]
165    {
166        use std::sync::Once;
167        static AZURE: Once = Once::new();
168        AZURE.call_once(|| deltalake::azure::register_handlers(None));
169    }
170    #[cfg(feature = "gcs")]
171    {
172        use std::sync::Once;
173        static GCS: Once = Once::new();
174        GCS.call_once(|| deltalake::gcp::register_handlers(None));
175    }
176}
177
178#[cfg(test)]
179mod tests {
180    use super::*;
181    use serde_json::json;
182
183    fn conn(uri: &str) -> DeltaConnection {
184        DeltaConnection {
185            table_uri: uri.into(),
186            credentials: DeltaCredentials::default(),
187            storage_options: HashMap::new(),
188        }
189    }
190
191    #[test]
192    fn flatten_shape_round_trips() {
193        let v = json!({
194            "table_uri": "s3://bucket/tbl",
195            "credentials": { "type": "aws", "config": { "region": "us-east-1" } },
196            "storage_options": { "AWS_ALLOW_HTTP": "true" }
197        });
198        let c: DeltaConnection = serde_json::from_value(v).unwrap();
199        assert_eq!(c.table_uri, "s3://bucket/tbl");
200        let opts = c.merged_storage_options();
201        assert_eq!(opts["AWS_REGION"], "us-east-1");
202        assert_eq!(opts["AWS_ALLOW_HTTP"], "true");
203    }
204
205    #[test]
206    fn explicit_storage_option_overrides_credential() {
207        let v = json!({
208            "table_uri": "s3://bucket/tbl",
209            "credentials": { "type": "aws", "config": { "region": "us-east-1" } },
210            "storage_options": { "AWS_REGION": "eu-west-1" }
211        });
212        let c: DeltaConnection = serde_json::from_value(v).unwrap();
213        assert_eq!(c.merged_storage_options()["AWS_REGION"], "eu-west-1");
214    }
215
216    #[test]
217    fn defaults_omit_credentials_and_options() {
218        let c: DeltaConnection =
219            serde_json::from_value(json!({ "table_uri": "file:///tmp/t" })).unwrap();
220        assert_eq!(c.credentials, DeltaCredentials::Default);
221        assert!(c.merged_storage_options().is_empty());
222    }
223
224    #[test]
225    fn redacted_uri_passes_plain_uri_through() {
226        assert_eq!(conn("s3://bucket/tbl").redacted_uri(), "s3://bucket/tbl");
227    }
228
229    #[test]
230    fn register_handlers_is_idempotent() {
231        // No cloud feature in the default test build → pure no-op, but must not
232        // panic when called repeatedly.
233        let c = conn("file:///tmp/t");
234        c.register_handlers();
235        c.register_handlers();
236    }
237}
faucet_common_delta/connection.rs

faucet_common_delta/
connection.rs
