Change Data Capture

Change data capture (CDC) turns writes in SlateDB into a durable change stream that external systems can consume. SlateDB exposes native-WAL CDC through slatedb::wal::SlateDbWalReader.

The reader opens a live iterator from the first unconsumed WAL file ID. The iterator owns polling and waits internally when it reaches the current WAL tail; the caller owns durable cursor storage.

When to use WAL-based CDC

Use WAL-based CDC when you need:

• A low-latency stream of inserts, updates, and deletes.
• A durable append-only feed that can be replayed.
• Integration with downstream systems such as Kafka, Pulsar, or data lakes.

WAL-based CDC is not a bootstrap or backfill API. Combine a database backfill with WAL streaming when an initial full copy is required.

Streaming model

Keep a durable cursor containing the last fully consumed WAL file ID. Use zero when starting from the beginning unless you have a more appropriate manifest-derived cursor.

To consume the stream:

1. Compute cursor + 1, checking for overflow, and create one iterator with an unbounded end range beginning at that WAL file ID.
2. Repeatedly call next(). When the iterator reaches the current tail, the call waits and polls internally for the next WAL file.
3. Emit every row in each WalRows batch.
4. After the whole batch succeeds, persist its last_consumed_wal_file_id as the new cursor.
5. After a restart, create a new unbounded iterator beginning at the persisted cursor plus one.

The live iterator does not report exhaustion at the current tail; next() remains pending until a new WAL file is visible or an error occurs. last_wal_file_id remains available when an application needs a point-in-time tail snapshot, but it is not part of the streaming loop.

Batches and cursor progress

One WalRows batch represents one fully consumed WAL file and contains:

• rows: the file’s row entries, in sequence order
• last_consumed_wal_file_id: the durable cursor after that file

Empty fence WAL files produce a batch with no rows. The cursor still advances, so consumers must process the batch rather than flattening batches into rows and losing progress information.

Each row is a RowEntry with:

• key: key bytes
• value: a ValueDeletable value, merge operand, or tombstone
• seq: the globally increasing sequence number
• create_ts and expire_ts: optional timestamps

Deletes appear as tombstones. Merge operands remain raw merge values; downstream consumers must apply the merge logic themselves when materialized values are required.

Visibility and durability

SlateDbWalReader only sees WAL SSTs flushed to object storage. WAL flushes occur when:

• The WAL buffer reaches its size threshold.
• The configured flush interval elapses.
• The caller explicitly requests a WAL or memtable flush.

A memtable flush first forces pending WAL data to durable storage.

The database must have its WAL enabled. WAL support is enabled by default.

Basic live stream

This example creates one unbounded iterator, emits both existing and later writes through it, and advances a durable file cursor.

main.rs

use slatedb::config::{FlushOptions, FlushType};
use slatedb::object_store::{memory::InMemory, path::Path};
use slatedb::wal::{SlateDbWalReaderBuilder, WalReader as _, WalRows};
use slatedb::{Db, RowEntry, ValueDeletable};
use std::sync::Arc;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let object_store = Arc::new(InMemory::new());
    let path = "/change-data-capture-example";

    let db = Db::open(path, object_store.clone()).await?;
    db.put(b"user:1", b"alice").await?;
    db.put(b"user:2", b"bob").await?;
    db.delete(b"user:2").await?;
    flush_wal(&db).await?;

    let wal_reader = SlateDbWalReaderBuilder::new()
        .with_object_store(object_store)
        .with_path(Path::from(path))
        .build()?;
    let mut cursor = 0_u64;
    let start_wal_id = cursor
        .checked_add(1)
        .ok_or_else(|| anyhow::anyhow!("WAL cursor cannot advance"))?;
    let mut iterator = wal_reader.iterator((start_wal_id..).into()).await?;

    // Drain the writes that already exist. Empty fence WALs still advance the
    // cursor, so stop based on emitted rows only after persisting every batch.
    let mut emitted_rows = 0;
    while emitted_rows < 3 {
        let batch = iterator
            .next()
            .await?
            .ok_or_else(|| anyhow::anyhow!("live WAL iterator ended unexpectedly"))?;
        emitted_rows += emit_batch(&batch, &mut cursor);
    }

    // Keep the same iterator alive. Its next call observes this later WAL;
    // callers do not need to discover a new tail or create another iterator.
    db.put(b"user:3", b"carol").await?;
    flush_wal(&db).await?;
    while emitted_rows < 4 {
        let batch = iterator
            .next()
            .await?
            .ok_or_else(|| anyhow::anyhow!("live WAL iterator ended unexpectedly"))?;
        emitted_rows += emit_batch(&batch, &mut cursor);
    }

    db.close().await?;
    Ok(())
}

async fn flush_wal(db: &Db) -> Result<(), slatedb::Error> {
    db.flush_with_options(FlushOptions {
        flush_type: FlushType::Wal,
    })
    .await
}

fn emit_batch(batch: &WalRows, cursor: &mut u64) -> usize {
    for row in &batch.rows {
        emit_row(batch.last_consumed_wal_file_id, row);
    }

    // Persist only after every row in this WAL file has been emitted. This
    // also advances across empty fence WALs.
    *cursor = batch.last_consumed_wal_file_id;
    println!("persist cursor={cursor}");
    batch.rows.len()
}

fn emit_row(wal_id: u64, row: &RowEntry) {
    let key = String::from_utf8_lossy(row.key.as_ref());
    match &row.value {
        ValueDeletable::Value(value) => {
            let value = String::from_utf8_lossy(value.as_ref());
            println!("wal_id={wal_id} seq={} upsert {key}={value}", row.seq);
        }
        ValueDeletable::Merge(value) => {
            let value = String::from_utf8_lossy(value.as_ref());
            println!("wal_id={wal_id} seq={} merge {key}+={value}", row.seq);
        }
        ValueDeletable::Tombstone => {
            println!("wal_id={wal_id} seq={} delete {key}", row.seq);
        }
    }
}

Persisting the cursor only after a complete batch gives at-least-once delivery if a process fails while emitting that WAL file. Make the sink idempotent or transactionally couple sink writes with cursor persistence when duplicates are unacceptable.

Ordering and resumability

WAL file IDs increase monotonically, and entries within each file are returned in sequence order. The minimal cursor is therefore the last fully consumed WAL file ID.

Do not advance the cursor per row. Doing so can lose progress for empty fence WALs and can incorrectly mark a partially emitted WAL file as complete.

If a requested file has already been garbage collected, iteration reports a truncation/data error. The consumer must not silently skip the gap.

Backfills plus streaming

For a full backfill:

1. Begin live WAL streaming and buffer changes.
2. Create a detached database checkpoint.
3. Scan the checkpoint or a clone for existing data.
4. Apply buffered WAL changes in sequence order.
5. Delete the checkpoint.
6. Continue streaming from the persisted WAL cursor.

The sink should tolerate overlap between the backfill and WAL stream.

WAL retention and GC

WAL SSTs are garbage collected according to the configured retention policy. Set WAL GC minimum age long enough for the slowest consumer and monitor cursor lag. A consumer that requests a deleted WAL receives a truncation error instead of silently continuing.

Dedicated WAL object stores

When a database uses a dedicated WAL object store, construct the reader with both stores: the main object store is used for manifest and GC-cutoff state, and the dedicated WAL object store is used for WAL SSTs. In Rust, use SlateDbWalReader::new_for_db_with_wal_object_store.