Struct DeltaLakeSink 

pub struct DeltaLakeSink { /* private fields */ }

Delta Lake sink connector.

Writes Arrow RecordBatch to Delta Lake tables with ACID transactions, at-least-once delivery (exactly-once opt-in), partitioning, and background compaction.

Lifecycle

new() -> open() -> [begin_epoch() -> write_batch()* -> commit_epoch()] -> close()
                         |                                    |
                         +--- rollback_epoch() (on failure) --+

Exactly-Once Semantics

Each LaminarDB epoch maps to exactly one Delta Lake transaction (version). On recovery, the sink checks _delta_log/ for the last committed epoch via txn (application transaction) metadata. If an epoch was already committed, it is skipped (idempotent commit).

2PC Protocol

pre_commit() coalesces the buffer but does NOT write to Delta. commit_epoch() performs the actual Delta write+commit atomically. rollback_epoch() discards the staged buffer without side effects. This ensures that a rolled-back epoch never leaves data in Delta.

Implementations

impl DeltaLakeSink

pub fn new(config: DeltaLakeSinkConfig) -> Self

Creates a new Delta Lake sink with the given configuration.

pub fn with_schema(config: DeltaLakeSinkConfig, schema: SchemaRef) -> Self

Creates a new Delta Lake sink with an explicit schema.

pub fn state(&self) -> ConnectorState

Returns the current connector state.

pub fn current_epoch(&self) -> u64

Returns the current epoch.

pub fn last_committed_epoch(&self) -> u64

Returns the last committed epoch.

pub fn buffered_rows(&self) -> usize

Returns the number of buffered rows pending flush.

pub fn buffered_bytes(&self) -> u64

Returns the estimated buffered bytes.

pub fn delta_version(&self) -> u64

Returns the current Delta Lake table version.

pub fn sink_metrics(&self) -> &DeltaLakeSinkMetrics

Returns a reference to the sink metrics.

pub fn config(&self) -> &DeltaLakeSinkConfig

Returns the sink configuration.

pub fn should_flush(&self) -> bool

Checks if a buffer flush is needed based on size or time thresholds.

pub fn estimate_batch_size(batch: &RecordBatch) -> u64

Estimates the byte size of a RecordBatch.

pub fn split_changelog_batch( batch: &RecordBatch, ) -> Result<(RecordBatch, RecordBatch), ConnectorError>

Splits a changelog RecordBatch into insert and delete batches.

Uses the _op metadata column:

• "I" (insert), "U" (update-after), "r" (snapshot read) -> insert
• "D" (delete) -> delete

The returned batches exclude metadata columns (those starting with _).

Errors

Returns ConnectorError::ConfigurationError if the _op column is missing or not a string type.

Trait Implementations

impl Debug for DeltaLakeSink

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl SinkConnector for DeltaLakeSink

fn open<'life0, 'life1, 'async_trait>( &'life0 mut self, config: &'life1 ConnectorConfig, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Opens the connector and initializes the connection.

fn write_batch<'life0, 'life1, 'async_trait>( &'life0 mut self, batch: &'life1 RecordBatch, ) -> Pin<Box<dyn Future<Output = Result<WriteResult, ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Writes a batch of records to the external system.

fn schema(&self) -> SchemaRef

Returns the expected input schema for this sink.

fn begin_epoch<'life0, 'async_trait>( &'life0 mut self, epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Begins a new epoch for exactly-once processing.

fn pre_commit<'life0, 'async_trait>( &'life0 mut self, epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Pre-commits the current epoch (phase 1 of two-phase commit).

fn commit_epoch<'life0, 'async_trait>( &'life0 mut self, epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Commits the current epoch (phase 2 of two-phase commit).

fn rollback_epoch<'life0, 'async_trait>( &'life0 mut self, epoch: u64, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Rolls back the current epoch.

fn health_check(&self) -> HealthStatus

Returns the current health status of the connector.

fn metrics(&self) -> ConnectorMetrics

Returns current metrics from the connector.

fn capabilities(&self) -> SinkConnectorCapabilities

Returns the capabilities of this sink connector.

fn flush<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Flushes any buffered data to the external system.

fn close<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectorError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Closes the connector and releases all resources.

fn as_schema_registry_aware(&self) -> Option<&dyn SchemaRegistryAware>

Returns this connector as a SchemaRegistryAware, if supported.

fn as_schema_evolvable(&self) -> Option<&dyn SchemaEvolvable>

Returns this connector as a SchemaEvolvable, if supported.