Struct CdcSyncEngine 

pub struct CdcSyncEngine<C, O> { /* private fields */ }

CDC-based sync engine that replicates changes from OLTP to OLAP.

Polls the CDC consumer for new events, converts them to DML, and applies them to the OLAP engine.

Features:

• Batch INSERT: Consecutive INSERT events for the same table are grouped into a single multi-row INSERT for better OLAP performance.
• CDC pruning: Optionally prunes processed CDC events after successful sync to prevent unbounded growth of the _rhei_cdc_log table.
• Transaction safety: Each sync cycle is wrapped in a BEGIN/COMMIT on the OLAP side with ROLLBACK on failure. Note: the actual transactional guarantee is backend-dependent — DuckDB supports real transactions, while DataFusion treats BEGIN/COMMIT/ROLLBACK as no-ops, so a mid-cycle failure may leave OLAP partially updated.

Implementations

impl<C, O> CdcSyncEngine<C, O>

where C: CdcConsumer, O: OlapEngine,

pub fn new( cdc: C, olap: O, schema_registry: SchemaRegistry, batch_size: u32, ) -> Self

Create a new CdcSyncEngine.

• cdc — CDC consumer that provides change events (e.g. SQLite trigger log).
• olap — OLAP engine that receives the converted DML.
• schema_registry — registry of table schemas used to build WHERE clauses and Arrow batches.
• batch_size — maximum number of CDC events to fetch per sync_once call.

CDC pruning is enabled by default (prune_after_sync = true). Sync mode defaults to rhei_core::types::SyncMode::Destructive.

pub fn with_prune_after_sync(self, prune: bool) -> Self

Set whether to prune processed CDC events after sync.

pub fn with_sync_mode(self, mode: SyncMode) -> Self

Set the sync mode: Destructive (default) or Temporal (SCD Type 2).

Trait Implementations

impl<C, O> SyncEngine for CdcSyncEngine<C, O>

where C: CdcConsumer, O: OlapEngine,

type Error = SyncError

Engine-specific error type returned by all fallible methods.

async fn sync_once(&self) -> Result<SyncResult, Self::Error>

Poll the CDC log and apply any new events to the OLAP engine.

async fn status(&self) -> Result<SyncStatus, Self::Error>

Return a snapshot of the current replication state.