laminar_connectors/lakehouse/
mod.rs

1//! Lakehouse connectors (Delta Lake, Apache Iceberg).
2
3// Versioned envelope for Delta coordinated-commit descriptors.
4#[cfg(feature = "delta-lake")]
5mod commit_descriptor;
6
7// Delta Lake modules
8pub mod delta;
9pub mod delta_config;
10#[cfg(feature = "delta-lake")]
11pub mod delta_io;
12#[cfg(feature = "delta-lake")]
13pub mod delta_lookup;
14pub mod delta_metrics;
15#[cfg(feature = "delta-lake")]
16pub mod delta_reference;
17pub mod delta_source;
18pub mod delta_source_config;
19#[cfg(feature = "delta-lake")]
20pub mod delta_table_provider;
21#[cfg(feature = "delta-lake-unity")]
22pub(crate) mod unity_catalog;
23
24// Apache Iceberg modules
25pub mod iceberg;
26pub mod iceberg_config;
27#[cfg(feature = "iceberg-core")]
28pub mod iceberg_io;
29#[cfg(feature = "iceberg-core")]
30pub mod iceberg_lookup;
31#[cfg(feature = "iceberg-core")]
32pub mod iceberg_reference;
33#[cfg(feature = "iceberg-core")]
34mod iceberg_scan;
35pub mod iceberg_source;
36
37#[cfg(any(feature = "delta-lake", feature = "iceberg-core"))]
38mod schema_resolution;
39
40// Common metrics
41pub mod metrics;
42#[cfg(any(test, feature = "delta-lake", feature = "iceberg-core"))]
43mod snapshot_schema;
44
45// Re-export Delta Lake types at module level.
46pub use delta::DeltaLakeSink;
47pub use delta_config::{DeltaCatalogType, DeltaLakeSinkConfig, DeltaWriteMode};
48#[cfg(feature = "delta-lake")]
49pub use delta_lookup::{DeltaLookupSource, DeltaLookupSourceConfig};
50pub use delta_metrics::DeltaLakeSinkMetrics;
51#[cfg(feature = "delta-lake")]
52pub use delta_reference::DeltaReferenceTableSource;
53pub use delta_source::DeltaSource;
54pub use delta_source_config::DeltaSourceConfig;
55pub use metrics::LakehouseSinkMetrics;
56
57// Re-export Iceberg types at module level.
58pub use iceberg::IcebergSink;
59use iceberg_config::{
60    sink_config_keys as iceberg_sink_config_keys, source_config_keys as iceberg_source_config_keys,
61};
62pub use iceberg_config::{
63    IcebergCatalogAuthType, IcebergCatalogConfig, IcebergCatalogType, IcebergNullOrder,
64    IcebergPartitionField, IcebergReadBootstrap, IcebergReadMode, IcebergSchemaEvolutionMode,
65    IcebergSinkConfig, IcebergSortDirection, IcebergSortField, IcebergSourceConfig,
66    IcebergStorageConfig, IcebergStorageEncryption, IcebergStorageType, IcebergTransform,
67    IcebergWriteDistributionMode, IcebergWriteMode,
68};
69#[cfg(feature = "iceberg-core")]
70pub use iceberg_lookup::{IcebergLookupSource, IcebergLookupSourceConfig};
71#[cfg(feature = "iceberg-core")]
72pub use iceberg_reference::IcebergReferenceTableSource;
73pub use iceberg_source::IcebergSource;
74
75use std::sync::Arc;
76
77use crate::config::{ConfigKeySpec, ConnectorInfo};
78use crate::registry::ConnectorRegistry;
79
80/// Registers the Delta Lake sink connector with the given registry.
81///
82/// # Errors
83///
84/// Returns the registry error when a name is already registered or the registry is frozen.
85pub fn register_delta_lake_sink(
86    registry: &ConnectorRegistry,
87) -> Result<(), crate::error::ConnectorError> {
88    let info = ConnectorInfo {
89        schema_capabilities: crate::schema::resolution::SchemaCapabilities::metadata(
90            &[],
91            true,
92            crate::schema::resolution::SchemaPreparation::ExplicitTableCreation,
93        ),
94        name: "delta-lake".to_string(),
95        display_name: "Delta Lake Sink".to_string(),
96        version: env!("CARGO_PKG_VERSION").to_string(),
97        is_source: false,
98        is_sink: true,
99        config_keys: delta_lake_config_keys(),
100    };
101
102    registry.register_sink(
103        "delta-lake",
104        info,
105        Arc::new(|config, registry: Option<&Arc<prometheus::Registry>>| {
106            Ok(Box::new(DeltaLakeSink::new(
107                DeltaLakeSinkConfig::from_config(config)?,
108                registry.map(Arc::as_ref),
109            )))
110        }),
111    )
112}
113
114/// Registers the Delta Lake source connector with the given registry.
115///
116/// # Errors
117///
118/// Returns the registry error when a name is already registered or the registry is frozen.
119pub fn register_delta_lake_source(
120    registry: &ConnectorRegistry,
121) -> Result<(), crate::error::ConnectorError> {
122    let info = ConnectorInfo {
123        schema_capabilities: crate::schema::resolution::SchemaCapabilities::metadata(
124            &[],
125            false,
126            crate::schema::resolution::SchemaPreparation::None,
127        ),
128        name: "delta-lake".to_string(),
129        display_name: "Delta Lake Source".to_string(),
130        version: env!("CARGO_PKG_VERSION").to_string(),
131        is_source: true,
132        is_sink: false,
133        config_keys: delta_lake_source_config_keys(),
134    };
135
136    registry.register_source(
137        "delta-lake",
138        info.clone(),
139        Arc::new(|registry: Option<&Arc<prometheus::Registry>>| {
140            Ok(Box::new(DeltaSource::new(
141                DeltaSourceConfig::default(),
142                registry.map(Arc::as_ref),
143            )))
144        }),
145    )?;
146
147    // Register finite startup snapshots for replicated reference tables.
148    #[cfg(feature = "delta-lake")]
149    registry.register_table_source(
150        "delta-lake",
151        info.clone(),
152        Arc::new(|config, declared_schema| {
153            Ok(Box::new(DeltaReferenceTableSource::from_connector_config(
154                config,
155                declared_schema,
156            )?))
157        }),
158    )?;
159
160    // Register lookup source factory for on-demand/partial cache mode.
161    #[cfg(feature = "delta-lake")]
162    registry.register_lookup_source("delta-lake", info, Arc::new(DeltaLookupFactory))?;
163
164    Ok(())
165}
166
167#[cfg(feature = "delta-lake")]
168struct DeltaLookupFactory;
169
170#[cfg(feature = "delta-lake")]
171#[async_trait::async_trait]
172impl crate::registry::LookupSourceFactory for DeltaLookupFactory {
173    async fn build(
174        &self,
175        config: crate::config::ConnectorConfig,
176        _declared_schema: Option<arrow_schema::SchemaRef>,
177    ) -> Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, crate::error::ConnectorError>
178    {
179        use crate::lakehouse::delta_source_config::DeltaSourceConfig;
180        let pk_columns: Vec<String> = config
181            .get("_primary_key_columns")
182            .unwrap_or("")
183            .split(',')
184            .map(|s| s.trim().to_string())
185            .filter(|s| !s.is_empty())
186            .collect();
187
188        if pk_columns.is_empty() {
189            return Err(crate::error::ConnectorError::ConfigurationError(
190                "delta-lake lookup source requires primary key columns".into(),
191            ));
192        }
193
194        let src_config = DeltaSourceConfig::from_config(&config)?;
195
196        let resolved = crate::lakehouse::delta_io::resolve_catalog_options(
197            &src_config.catalog_type,
198            src_config.catalog_database.as_deref(),
199            src_config.catalog_name.as_deref(),
200            src_config.catalog_schema.as_deref(),
201            &src_config.table_path,
202            &src_config.storage_options,
203            crate::lakehouse::delta_io::CatalogAccess::Read,
204        )
205        .await?;
206        if resolved.credentials_expire_at.is_some() {
207            return Err(crate::error::ConnectorError::ConfigurationError(
208                "on-demand Delta lookups keep the storage credentials they open with and cannot \
209                 renew Unity Catalog vended credentials; use a reference table, or \
210                 catalog.storage.credentials=configured"
211                    .into(),
212            ));
213        }
214
215        let lookup_config = DeltaLookupSourceConfig {
216            table_path: resolved.location,
217            storage_options: resolved.storage_options,
218            primary_key_columns: pk_columns,
219            table_name: "delta_lookup".to_string(),
220        };
221
222        // `From<LookupError>` preserves transient/non-transient class.
223        let source = DeltaLookupSource::open(lookup_config).await?;
224
225        Ok(Arc::new(source) as Arc<dyn laminar_core::lookup::source::LookupSourceDyn>)
226    }
227}
228
229/// Registers the Iceberg sink connector with the given registry.
230///
231/// # Errors
232///
233/// Returns the registry error when a name is already registered or the registry is frozen.
234pub fn register_iceberg_sink(
235    registry: &ConnectorRegistry,
236) -> Result<(), crate::error::ConnectorError> {
237    let info = ConnectorInfo {
238        schema_capabilities: crate::schema::resolution::SchemaCapabilities::metadata(
239            &[],
240            true,
241            crate::schema::resolution::SchemaPreparation::ExplicitTableCreation,
242        ),
243        name: "iceberg".to_string(),
244        display_name: "Apache Iceberg Sink".to_string(),
245        version: env!("CARGO_PKG_VERSION").to_string(),
246        is_source: false,
247        is_sink: true,
248        config_keys: iceberg_sink_config_keys(),
249    };
250
251    registry.register_sink(
252        "iceberg",
253        info,
254        Arc::new(|config, registry: Option<&Arc<prometheus::Registry>>| {
255            Ok(Box::new(IcebergSink::new(
256                IcebergSinkConfig::from_config(config)?,
257                registry.map(Arc::as_ref),
258            )))
259        }),
260    )
261}
262
263/// Registers the Iceberg source connector with the given registry.
264///
265/// # Errors
266///
267/// Returns the registry error when a name is already registered or the registry is frozen.
268pub fn register_iceberg_source(
269    registry: &ConnectorRegistry,
270) -> Result<(), crate::error::ConnectorError> {
271    let info = ConnectorInfo {
272        schema_capabilities: crate::schema::resolution::SchemaCapabilities::metadata(
273            &[],
274            false,
275            crate::schema::resolution::SchemaPreparation::None,
276        ),
277        name: "iceberg".to_string(),
278        display_name: "Apache Iceberg Source".to_string(),
279        version: env!("CARGO_PKG_VERSION").to_string(),
280        is_source: true,
281        is_sink: false,
282        config_keys: iceberg_source_config_keys(),
283    };
284
285    registry.register_source(
286        "iceberg",
287        info.clone(),
288        Arc::new(|registry: Option<&Arc<prometheus::Registry>>| {
289            let mut config = crate::config::ConnectorConfig::new("iceberg");
290            config.set("catalog.uri", "http://localhost:8181");
291            config.set("catalog.warehouse", "s3://default/wh");
292            config.set("namespace", "default");
293            config.set("table.name", "default");
294            Ok(Box::new(IcebergSource::new(
295                IcebergSourceConfig::from_config(&config)?,
296                registry.map(Arc::as_ref),
297            )))
298        }),
299    )?;
300
301    // Register finite startup snapshots for replicated reference tables.
302    #[cfg(feature = "iceberg-core")]
303    registry.register_table_source(
304        "iceberg",
305        info.clone(),
306        Arc::new(|config, declared_schema| {
307            Ok(Box::new(
308                IcebergReferenceTableSource::from_connector_config(config, declared_schema)?,
309            ))
310        }),
311    )?;
312
313    // Register lookup source factory for on-demand/partial cache mode.
314    #[cfg(feature = "iceberg-core")]
315    registry.register_lookup_source("iceberg", info, Arc::new(IcebergLookupFactory))?;
316
317    Ok(())
318}
319
320#[cfg(feature = "iceberg-core")]
321struct IcebergLookupFactory;
322
323#[cfg(feature = "iceberg-core")]
324#[async_trait::async_trait]
325impl crate::registry::LookupSourceFactory for IcebergLookupFactory {
326    async fn build(
327        &self,
328        config: crate::config::ConnectorConfig,
329        _declared_schema: Option<arrow_schema::SchemaRef>,
330    ) -> Result<Arc<dyn laminar_core::lookup::source::LookupSourceDyn>, crate::error::ConnectorError>
331    {
332        use crate::lakehouse::iceberg_config::{IcebergCatalogConfig, IcebergStorageConfig};
333        let pk_columns: Vec<String> = config
334            .get("_primary_key_columns")
335            .unwrap_or("")
336            .split(',')
337            .map(|s| s.trim().to_string())
338            .filter(|s| !s.is_empty())
339            .collect();
340
341        if pk_columns.is_empty() {
342            return Err(crate::error::ConnectorError::ConfigurationError(
343                "iceberg lookup source requires primary key columns".into(),
344            ));
345        }
346
347        let catalog = IcebergCatalogConfig::from_config(&config)?;
348        let storage = IcebergStorageConfig::from_config(&config)?;
349        let lookup_config = IcebergLookupSourceConfig {
350            catalog,
351            storage,
352            primary_key_columns: pk_columns,
353        };
354
355        let source = IcebergLookupSource::open(lookup_config).await?;
356        Ok(Arc::new(source) as Arc<dyn laminar_core::lookup::source::LookupSourceDyn>)
357    }
358}
359
360/// Registers all lakehouse sink connectors (Delta Lake, Iceberg).
361///
362/// # Errors
363///
364/// Returns the first registry error.
365pub fn register_lakehouse_sinks(
366    registry: &ConnectorRegistry,
367) -> Result<(), crate::error::ConnectorError> {
368    register_delta_lake_sink(registry)?;
369    register_iceberg_sink(registry)
370}
371
372/// Registers all lakehouse source connectors.
373///
374/// # Errors
375///
376/// Returns the first registry error.
377pub fn register_lakehouse_sources(
378    registry: &ConnectorRegistry,
379) -> Result<(), crate::error::ConnectorError> {
380    register_delta_lake_source(registry)?;
381    register_iceberg_source(registry)
382}
383
384fn delta_lake_config_keys() -> Vec<ConfigKeySpec> {
385    let mut keys = vec![
386        ConfigKeySpec::required(
387            "table.path",
388            "Path to Delta Lake table (local, s3://, az://, gs://)",
389        ),
390        ConfigKeySpec::optional(
391            "partition.columns",
392            "Comma-separated partition column names",
393            "",
394        ),
395        ConfigKeySpec::optional(
396            "target.file.size",
397            "Target Parquet file size in bytes",
398            "134217728",
399        ),
400        ConfigKeySpec::optional(
401            "max.buffer.records",
402            "Maximum records to buffer before flushing",
403            "100000",
404        ),
405        ConfigKeySpec::optional(
406            "max.buffer.duration.ms",
407            "Maximum time to buffer before flushing (ms)",
408            "60000",
409        ),
410        ConfigKeySpec::optional(
411            "schema.evolution",
412            "Enable automatic schema evolution (additive columns)",
413            "false",
414        ),
415        ConfigKeySpec::optional(
416            "write.mode",
417            "Write mode: append, overwrite (each flush replaces the whole table), upsert",
418            "append",
419        ),
420        ConfigKeySpec::optional(
421            "merge.key.columns",
422            "Key columns for upsert MERGE (required for upsert mode)",
423            "",
424        ),
425    ];
426    keys.extend(delta_catalog_config_keys());
427    keys.extend([
428        ConfigKeySpec::optional(
429            "catalog.storage.location",
430            "Storage location for auto-created UC external tables (e.g. s3://bucket/path)",
431            "",
432        ),
433        // ── LogStore configuration ──
434        ConfigKeySpec::optional(
435            "storage.s3_locking_provider",
436            "S3 locking provider: 'dynamodb' for DynamoDB-backed log store",
437            "",
438        ),
439        ConfigKeySpec::optional(
440            "storage.dynamodb_table_name",
441            "DynamoDB table name for S3 locking (default: delta_log)",
442            "",
443        ),
444        // ── Cloud storage credentials (resolved via StorageCredentialResolver) ──
445        ConfigKeySpec::optional(
446            "storage.aws_access_key_id",
447            "AWS access key ID (falls back to AWS_ACCESS_KEY_ID env var)",
448            "",
449        ),
450        ConfigKeySpec::optional(
451            "storage.aws_secret_access_key",
452            "AWS secret access key (falls back to AWS_SECRET_ACCESS_KEY env var)",
453            "",
454        ),
455        ConfigKeySpec::optional(
456            "storage.aws_region",
457            "AWS region for S3 paths (falls back to AWS_REGION env var)",
458            "",
459        ),
460        ConfigKeySpec::optional(
461            "storage.aws_session_token",
462            "AWS session token for temporary credentials (falls back to AWS_SESSION_TOKEN)",
463            "",
464        ),
465        ConfigKeySpec::optional(
466            "storage.aws_endpoint",
467            "Custom S3 endpoint (MinIO, LocalStack; falls back to AWS_ENDPOINT_URL)",
468            "",
469        ),
470        ConfigKeySpec::optional(
471            "storage.aws_profile",
472            "AWS profile name (falls back to AWS_PROFILE env var)",
473            "",
474        ),
475        ConfigKeySpec::optional(
476            "storage.azure_storage_account_name",
477            "Azure storage account name (falls back to AZURE_STORAGE_ACCOUNT_NAME)",
478            "",
479        ),
480        ConfigKeySpec::optional(
481            "storage.azure_storage_account_key",
482            "Azure storage account key (falls back to AZURE_STORAGE_ACCOUNT_KEY)",
483            "",
484        ),
485        ConfigKeySpec::optional(
486            "storage.azure_storage_sas_token",
487            "Azure SAS token (falls back to AZURE_STORAGE_SAS_TOKEN)",
488            "",
489        ),
490        ConfigKeySpec::optional(
491            "storage.azure_storage_client_id",
492            "Azure client ID for service principal auth (falls back to AZURE_CLIENT_ID)",
493            "",
494        ),
495        ConfigKeySpec::optional(
496            "storage.google_service_account_path",
497            "Path to GCS service account JSON (falls back to GOOGLE_APPLICATION_CREDENTIALS)",
498            "",
499        ),
500        ConfigKeySpec::optional(
501            "storage.google_service_account_key",
502            "Inline GCS service account JSON (falls back to GOOGLE_SERVICE_ACCOUNT_KEY)",
503            "",
504        ),
505    ]);
506    keys
507}
508
509/// Catalog keys shared by the Delta source and sink.
510fn delta_catalog_config_keys() -> Vec<ConfigKeySpec> {
511    vec![
512        ConfigKeySpec::optional("catalog.type", "Catalog type: none, glue, unity", "none"),
513        ConfigKeySpec::optional(
514            "catalog.database",
515            "Catalog database name (required for Glue)",
516            "",
517        ),
518        ConfigKeySpec::optional("catalog.name", "Catalog name (required for Unity)", ""),
519        ConfigKeySpec::optional(
520            "catalog.schema",
521            "Catalog schema name (required for Unity)",
522            "",
523        ),
524        ConfigKeySpec::optional(
525            "catalog.workspace_url",
526            "Databricks workspace URL (required for Unity)",
527            "",
528        ),
529        ConfigKeySpec::optional(
530            "catalog.access_token",
531            "Unity Catalog token (PAT or service-principal token); or use client_id/client_secret",
532            "",
533        ),
534        ConfigKeySpec::optional(
535            "catalog.client_id",
536            "Unity Catalog OAuth M2M service-principal client id",
537            "",
538        ),
539        ConfigKeySpec::optional(
540            "catalog.client_secret",
541            "Unity Catalog OAuth M2M service-principal secret",
542            "",
543        ),
544        ConfigKeySpec::optional(
545            "catalog.storage.credentials",
546            "Unity storage credentials: vended (table-scoped, renewed) or configured (storage.*)",
547            "vended",
548        ),
549    ]
550}
551
552fn delta_lake_source_config_keys() -> Vec<ConfigKeySpec> {
553    let mut keys = vec![
554        ConfigKeySpec::required(
555            "table.path",
556            "Path to Delta Lake table (local, s3://, az://, gs://)",
557        ),
558        ConfigKeySpec::optional(
559            "read.mode",
560            "changes (snapshot/history then the change data feed) or snapshot (one snapshot, then complete)",
561            "changes",
562        ),
563        ConfigKeySpec::optional(
564            "starting.version",
565            "Changes mode: first change version to read (inclusive), or 'latest' for commits after \
566             first start; unset emits the latest snapshot, then its later changes",
567            "",
568        ),
569        ConfigKeySpec::optional(
570            "poll.interval.ms",
571            "How often to poll for new versions (ms)",
572            "1000",
573        ),
574    ];
575    keys.extend(delta_catalog_config_keys());
576    keys.extend([
577        // ── LogStore configuration ──
578        ConfigKeySpec::optional(
579            "storage.s3_locking_provider",
580            "S3 locking provider: 'dynamodb' for DynamoDB-backed log store",
581            "",
582        ),
583        ConfigKeySpec::optional(
584            "storage.dynamodb_table_name",
585            "DynamoDB table name for S3 locking (default: delta_log)",
586            "",
587        ),
588        // ── Cloud storage credentials ──
589        ConfigKeySpec::optional("storage.aws_access_key_id", "AWS access key ID", ""),
590        ConfigKeySpec::optional("storage.aws_secret_access_key", "AWS secret access key", ""),
591        ConfigKeySpec::optional("storage.aws_region", "AWS region for S3 paths", ""),
592        ConfigKeySpec::optional(
593            "storage.azure_storage_account_name",
594            "Azure storage account name",
595            "",
596        ),
597        ConfigKeySpec::optional(
598            "storage.azure_storage_account_key",
599            "Azure storage account key",
600            "",
601        ),
602        ConfigKeySpec::optional(
603            "storage.google_service_account_path",
604            "Path to GCS service account JSON",
605            "",
606        ),
607    ]);
608    keys
609}
610
611#[cfg(test)]
612mod tests;
laminar_connectors/lakehouse/mod.rs

laminar_connectors/lakehouse/
mod.rs
