laminar_connectors/lakehouse/delta/
mod.rs

1//! Delta Lake sink connector implementation.
2//!
3//! [`DeltaLakeSink`] implements [`SinkConnector`], writing Arrow `RecordBatch`
4//! data to Delta Lake tables with ACID transactions and at-least-once delivery
5//! (exactly-once is selected once at the pipeline/runtime boundary).
6//!
7//! # Write Strategies
8//!
9//! - **Append mode**: Arrow-to-Parquet zero-copy writes for immutable streams
10//! - **Overwrite mode**: every flush atomically replaces the *whole table* with that
11//!   flush's rows (no partition-scoped replacement)
12//! - **Upsert mode**: CDC MERGE via Z-set changelog integration
13//!
14//! Exactly-once append writes prepare immutable Parquet descriptors during the
15//! checkpoint and publish them with one namespaced, designated Delta commit.
16//! At-least-once append, overwrite, and upsert writes commit through `flush()`.
17//!
18//! # Ring Architecture
19//!
20//! - **Ring 0**: No sink code. Data arrives via SPSC channel (~5ns push).
21//! - **Ring 1**: Batch buffering, Parquet writes, Delta log commits.
22//! - **Ring 2**: Schema management, configuration, health checks.
23
24mod lifecycle;
25mod operations;
26mod publication;
27mod schema_binding;
28
29#[cfg(feature = "delta-lake")]
30use std::future::Future;
31use std::sync::Arc;
32use std::time::{Duration, Instant};
33
34use arrow_array::{Array, RecordBatch};
35#[cfg(feature = "delta-lake")]
36use arrow_schema::DataType;
37use arrow_schema::SchemaRef;
38use async_trait::async_trait;
39use tracing::{debug, info, warn};
40
41#[cfg(feature = "delta-lake")]
42use deltalake::DeltaTable;
43
44#[cfg(feature = "delta-lake")]
45use deltalake::protocol::SaveMode;
46
47use crate::config::{ConnectorConfig, ConnectorState};
48#[cfg(feature = "delta-lake")]
49use crate::connector::{ConnectorTaskGuard, MAX_COORDINATED_COMMIT_PAYLOAD_BYTES};
50use crate::connector::{
51    ConnectorTaskOwner, ConnectorTaskTracker, SinkConnector, SinkConsistency, SinkContract,
52    SinkInputMode, SinkTopology, WriteResult,
53};
54use crate::error::ConnectorError;
55use crate::storage::{StorageCredentialResolver, StorageProvider};
56
57use super::delta_config::{DeltaLakeSinkConfig, DeltaWriteMode};
58use super::delta_metrics::DeltaLakeSinkMetrics;
59use crate::connector::DeliveryGuarantee;
60
61#[cfg(feature = "delta-lake")]
62use publication::{
63    classify_delta_attempt_error, count_collapsed_ops, publish_coordinated_delta_batch,
64    retry_delta_metadata_until, run_tracked_delta_task, DeltaWriteTaskSuccess,
65    UnresolvedDeltaPublication,
66};
67
68/// Delta Lake sink connector.
69///
70/// Writes Arrow `RecordBatch` to Delta Lake tables with ACID transactions,
71/// at-least-once delivery (exactly-once opt-in), partitioning, and
72/// externally managed table maintenance.
73///
74/// # Exactly-Once Semantics
75///
76/// `pre_commit()` materializes immutable Parquet files and returns their Delta
77/// `Add` actions. The runtime durably seals the checkpoint before one
78/// designated committer calls `commit_aggregated()` with every participant's
79/// descriptor. `rollback_epoch()` discards in-memory state; unreferenced files
80/// are reclaimed later by retention-safe vacuum.
81pub struct DeltaLakeSink {
82    /// Sole admission authority for Delta tasks that may outlive a cancelled caller.
83    #[cfg(feature = "delta-lake")]
84    task_owner: ConnectorTaskOwner,
85    /// Stable terminal observer retained by the runtime after this sink is retired.
86    task_tracker: ConnectorTaskTracker,
87    /// Sink configuration.
88    config: DeltaLakeSinkConfig,
89    /// Arrow schema for input batches (set on first write or from existing table).
90    schema: Option<SchemaRef>,
91    query_schema: Option<SchemaRef>,
92    writer_projection: Option<Vec<usize>>,
93    #[cfg(feature = "delta-lake")]
94    schema_binding: Option<crate::schema::resolution::SchemaBinding>,
95    /// Connector lifecycle state.
96    state: ConnectorState,
97    /// Current epoch being written.
98    current_epoch: u64,
99    /// `RecordBatch` buffer for the current epoch.
100    buffer: Vec<RecordBatch>,
101    /// Total rows buffered in current epoch.
102    buffered_rows: usize,
103    /// Total bytes buffered (estimated) in current epoch.
104    buffered_bytes: u64,
105    /// Current Delta Lake table version.
106    delta_version: u64,
107    /// Time when the current buffer started accumulating.
108    buffer_start_time: Option<Instant>,
109    /// Sink metrics.
110    metrics: DeltaLakeSinkMetrics,
111    /// Delta Lake table handle (present when `delta-lake` feature is enabled).
112    #[cfg(feature = "delta-lake")]
113    table: Option<DeltaTable>,
114    /// Staged batches ready for either coordinated descriptor creation or an
115    /// at-least-once flush. This separation lets `rollback_epoch()` discard
116    /// prepared in-memory data without publishing it to the Delta log.
117    staged_batches: Vec<RecordBatch>,
118    /// Rows staged for commit (mirrors `staged_batches`).
119    staged_rows: usize,
120    /// Estimated bytes staged for commit.
121    staged_bytes: u64,
122    /// Uncommitted `Add` actions for immutable Parquet files materialized in
123    /// the current coordinated epoch.
124    #[cfg(feature = "delta-lake")]
125    coordinated_adds: Vec<deltalake::kernel::Add>,
126    /// Canonical table incarnation and write-metadata fingerprint captured
127    /// from the exact immutable snapshot used to create `coordinated_adds`.
128    #[cfg(feature = "delta-lake")]
129    coordinated_binding: Option<super::commit_descriptor::DeltaTableBinding>,
130    /// Exact encoded descriptor size for `coordinated_adds`, or zero when the
131    /// vector is empty.
132    #[cfg(feature = "delta-lake")]
133    coordinated_descriptor_bytes: usize,
134    /// Exact publication that must be retried or observed at its target before
135    /// this instance stages a later cut. Absence cannot resolve a timed-out
136    /// remote catalog mutation because the server may still complete it.
137    #[cfg(feature = "delta-lake")]
138    coordinated_unresolved_publication: Arc<parking_lot::Mutex<Option<UnresolvedDeltaPublication>>>,
139    /// Resolved table path after catalog lookup (may differ from `config.table_path`
140    /// when using Unity/Glue catalogs). Used by `reopen_table()` so retries
141    /// target the same resolved path that `open()` connected to.
142    #[cfg(feature = "delta-lake")]
143    resolved_table_path: String,
144    /// Resolved storage options after catalog lookup.
145    #[cfg(feature = "delta-lake")]
146    resolved_storage_options: std::collections::HashMap<String, String>,
147    /// Expiry of catalog-vended credentials inside `resolved_storage_options`.
148    #[cfg(feature = "delta-lake")]
149    credentials_expire_at: Option<std::time::SystemTime>,
150    /// Pre-built Parquet writer properties for hot-path writes. Built once
151    /// in `init_delta_table()` from `config.parquet`; cloning this is far
152    /// cheaper than rebuilding (string parsing, bloom-filter column setup)
153    /// from scratch on every commit.
154    #[cfg(feature = "delta-lake")]
155    cached_writer_properties: Option<deltalake::parquet::file::properties::WriterProperties>,
156    /// Shared `DataFusion` session for upsert/merge operations. Creating a
157    /// fresh `SessionContext` per merge allocated a runtime env, memory
158    /// pool, and object-store registry each commit; reusing one flattens
159    /// allocator churn under steady-state upsert load.
160    #[cfg(feature = "delta-lake")]
161    merge_session: Option<datafusion::prelude::SessionContext>,
162    /// A dispatched Delta operation lost its result, so this generation cannot
163    /// safely admit another write even after its detached task terminates.
164    #[cfg(feature = "delta-lake")]
165    unresolved_delta_write: bool,
166    /// Test hook that makes coordinated descriptor preparation remain pending.
167    /// This is compiled out of production builds.
168    #[cfg(all(test, feature = "delta-lake"))]
169    stall_descriptor_write: bool,
170}
171
172/// Pre-creates a Unity Catalog external Delta table via the REST API if
173/// `catalog.storage.location` is configured and a schema is available.
174/// Idempotent: treats "already exists" (HTTP 409 / `ALREADY_EXISTS`) as success.
175#[cfg(all(feature = "delta-lake", feature = "delta-lake-unity"))]
176pub(super) async fn ensure_uc_table_exists(
177    config: &DeltaLakeSinkConfig,
178    schema: Option<&SchemaRef>,
179) -> Result<(), ConnectorError> {
180    let super::delta_config::DeltaCatalogType::Unity(ref unity) = config.catalog_type else {
181        return Ok(());
182    };
183    let super::delta_config::UnityAuth::Token(ref access_token) = unity.auth else {
184        return Err(ConnectorError::ConfigurationError(
185            "Unity Catalog table registration requires catalog.access_token".into(),
186        ));
187    };
188    let workspace_url = &unity.workspace_url;
189
190    let Some(ref storage_location) = config.catalog_storage_location else {
191        return Ok(());
192    };
193
194    let Some(arrow_schema) = schema else {
195        warn!(
196            "catalog.storage.location is set but no schema available — \
197             skipping Unity Catalog auto-create"
198        );
199        return Ok(());
200    };
201
202    let catalog = config.catalog_name.as_deref().unwrap_or_default();
203    let schema_name = config.catalog_schema.as_deref().unwrap_or_default();
204    let table_name = config
205        .table_path
206        .strip_prefix("uc://")
207        .and_then(|s| s.rsplit('.').next())
208        .unwrap_or(&config.table_path);
209
210    let columns =
211        super::unity_catalog::arrow_to_uc_columns(arrow_schema, &config.partition_columns)?;
212    super::unity_catalog::create_uc_table(
213        workspace_url,
214        access_token,
215        catalog,
216        schema_name,
217        table_name,
218        storage_location,
219        &columns,
220    )
221    .await
222}
223
224impl std::fmt::Debug for DeltaLakeSink {
225    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
226        f.debug_struct("DeltaLakeSink")
227            .field("state", &self.state)
228            .field("table_path", &"<configured>")
229            .field("mode", &self.config.write_mode)
230            .field("guarantee", &self.config.delivery_guarantee)
231            .field("current_epoch", &self.current_epoch)
232            .field("buffered_rows", &self.buffered_rows)
233            .field("delta_version", &self.delta_version)
234            .finish_non_exhaustive()
235    }
236}
237
238// ── Helper functions ────────────────────────────────────────────────
239
240/// Filters a `RecordBatch` using a boolean mask and projects to the given column indices.
241///
242/// Takes `mask` by value to hand it straight to `BooleanArray::from` without
243/// an intermediate `Vec<bool>` copy. Projects before filtering so the SIMD
244/// kernel only walks user columns, not the dropped metadata columns.
245fn filter_and_project(
246    batch: &RecordBatch,
247    mask: Vec<bool>,
248    col_indices: &[usize],
249) -> Result<RecordBatch, ConnectorError> {
250    use arrow_array::BooleanArray;
251    use arrow_select::filter::filter_record_batch;
252
253    let bool_array = BooleanArray::from(mask);
254
255    let projected = batch
256        .project(col_indices)
257        .map_err(|e| ConnectorError::Internal(format!("batch projection failed: {e}")))?;
258
259    filter_record_batch(&projected, &bool_array)
260        .map_err(|e| ConnectorError::Internal(format!("arrow filter failed: {e}")))
261}
262
263#[cfg(test)]
264#[allow(clippy::cast_possible_wrap)]
265#[allow(clippy::cast_precision_loss)]
266#[allow(clippy::float_cmp)]
267mod tests;
laminar_connectors/lakehouse/delta/mod.rs

laminar_connectors/lakehouse/delta/
mod.rs
