Event sourcing and CQRS

Applies to VAST Cluster 5.5. Examples use the VAST DataBase Python SDK 2.1.0.

Event sourcing keeps every change to your data as an event in an append-only log, and derives the current state from that log. CQRS (command query responsibility segregation) separates the path that writes from the path that reads: writers append events, and readers query a current-state table that one process keeps up to date. This guide shows how to build the pattern on VAST DataBase with the Python SDK. It builds an event table and a current-state table, appends events from two writers at once, and folds them into the current state. It then rebuilds that state by replaying the log.

Run the code blocks in order in one Python session. Below each block is the output it produced on a VAST Cluster 5.5 system.

When to use it

Consider event sourcing and CQRS when:

• Many writers change the same rows. Two open transactions that update or delete the same row conflict, and one of them is rolled back (conflicts between transactions). New rows don't conflict: two transactions that insert into the same table both commit (when two transactions conflict). If every writer only appends, the contention moves to one process that you control.
• You need the history. An audit trail, "what did this order look like at 10:05?", or a replay after a bug in your logic all need the events, not only the latest values.
• Readers want the current state of each entity, without folding events themselves.

It isn't worth it for data that has one writer per row, or where only the latest value matters and nobody asks about the past. Update the rows directly instead (updates and deletes).

The shape

Writers append; one materializer folds; readers read the current state or the log.

1. Commands and events come in, from your services, or through a VAST Event Broker topic.
2. Writers only append to an event table. Each row is one event: an event_id, the aggregate key (the entity the event is about, such as an order), an event type, the time the event happened, the time it was appended, and a payload.
3. One materializer folds new events into a current-state table, with one row per aggregate. It tracks how far it has got with a watermark.
4. Readers query the current state, or read the log for an aggregate's history.

Why it suits VAST DataBase

• Appends don't conflict. Conflicts are checked per cell, when you write. Inserting new rows into the same table from several transactions doesn't conflict, while updates and deletes of the same row do (when two transactions conflict). An append-only log turns many contended writers into many uncontended ones. Eight processes appending batches to one sorted table at once, with overlapping aggregate keys, committed every transaction with no conflicts and no retries.
• Inserts are the cheap write. A delete doesn't remove data: it writes a tombstone, and a background job frees the space later. An update writes new values at the row's existing position, and older versions are cleaned up in the background (write path). An event log does no updates or deletes until you prune it.
• One transaction covers the whole fold. A transaction can read and write several tables, and its changes become visible all at once when it commits (commit and rollback, when changes become visible). The materializer can read new events, replace current-state rows and advance its watermark atomically. A reader never sees half a pass.
• A sort key on the aggregate key makes history reads cheap. A sorted table helps queries that filter on its sort columns (sorted tables). Sorting the event table on the aggregate key, then event time, serves "all events for order 101". Sort columns can't be updated, which doesn't matter for a log you never update (columns you can't update).
• Readers don't block writers. A transaction reads a snapshot taken when it began; readers don't wait for writers, and writers don't wait for readers (what a transaction sees).

Design choices

Idempotency and duplicates

Expect the same event more than once. A Kafka producer with idempotence off delivers at least once (delivery guarantees), and a consumer that crashes before it commits its offsets reads the same events again (consume in a group). Your own retries of a whole transaction can append twice if the first attempt committed and only the reply was lost (retries the SDK does for you).

So give every event an event_id that the producer assigns once, for example a UUID, and keep it the same on every retry. Don't de-duplicate on write: that would need a read before each append, and appends would no longer be blind. Keep duplicates in the log and drop them when you fold, by event_id. If your events have no ID, (aggregate key, event time) can stand in, if two distinct events for one aggregate never share a time.

Make the fold a pure function of the events: the same set of events, in any order and with any duplicates, gives the same state. Sort by event time, then by event_id to break ties, and apply each event_id once. Then reprocessing an event is harmless, and replay gives the same answer as the incremental passes.

Duplicates stay in the log; the fold drops them by event_id.

One materializer

Run exactly one materializer at a time. As the only writer to the current-state table, it never conflicts with anyone. Your application picks that one process, for example with a lease, leader election or a single scheduled job.

As a backstop, have every pass update the same watermark row in the same transaction as its fold. If two passes overlap, the second one's update of that row fails at once with vastdb.errors.Conflict (TabularWwc, "Write write conflict"), and its transaction is rolled back as a whole; committing it then fails with TabularTxAlreadyRollbacked. The first pass's fold and watermark stand (when two transactions conflict).

If two passes overlap, the watermark row makes the second one fail.

Transactions use snapshot isolation: the cluster checks whether two writes overlap, not whether one transaction read data that another changed (what this means for your code). With one materializer folding each aggregate, that's all this pattern needs.

A watermark for incremental passes

Each pass reads only the events appended since the last one. Store the watermark, the append time up to which events have been folded, in a small table, and advance it in the same transaction as the fold. If the pass fails, nothing commits: the watermark stays where it was, and the next pass reads the same events again.

Use the append time, not the event time, for the watermark. An event can arrive long after it happened, and a watermark on event time would skip it.

Choose how passes find new events. The filter on appended_at runs in the cluster, so only the new events come back to the materializer. How much of the log the cluster reads to find them depends on how you lay out the event table, which you choose when you create it (decisions you can't undo):

• Sorted on the aggregate key, as in the example: one aggregate's history is a cheap read, and each pass reads through the log to find the new events, since appended_at isn't a sort column. The cost of a pass grows with the size of the log, so pair this layout with pruning.
• Sorted on appended_at: each pass reads only the recent range, and a history read for one aggregate becomes a filter on order_id instead of a range on the sort key.
• A short log: prune folded events after a retention period and keep a checkpoint, so every pass reads a small table whatever the layout.

The event table's layout decides how much of the log a pass reads.

Leave a grace period. A pass sees only transactions that committed before it began (what a transaction sees). A writer that stamped appended_at and then committed after the pass began isn't in that pass. If the pass then moved the watermark past that writer's appended_at, the event would never be folded. So fold only up to a cutoff some time in the past, and set the watermark to that cutoff. Make the grace period longer than the longest time between a writer stamping appended_at and its commit, with a margin: a few times that. A writer that waited before committing, while a pass with a cutoff of now ran, lost its event for good; with a cutoff of now minus a grace period longer than the wait, the next pass folded it. Keep writer transactions short: append, then commit. The cluster rolls back transactions that stay open or idle too long (transactions that are left open), but it doesn't publish the limits, so don't size the grace period from them.

A cutoff of now can lose an event that commits late; a grace period keeps it.

Keep each pass small as well: if a backlog builds up, fold it in several passes, each with a later cutoff.

Rebuild by replay

The log is the source of truth; the current-state table can always be rebuilt from it. Rebuild after you fix a bug in the fold, or to add a column to the current state. Fold the whole log into a new table, check it, and then switch your readers to it (Table.rename() renames a table; see the SDK reference). The worked example does this.

Retention and compaction of the log

A log you never prune keeps growing. To prune it, delete events older than the watermark minus a retention period. Select them with internal_row_id=True and delete() them, in a transaction (delete rows). Deletes are tombstones, so space comes back only after background cleanup, and the row count in get_stats() drops only then (write path).

Once you prune, replay can only start from the oldest event you kept. Before you delete events, save a checkpoint: the current state as of the watermark, in its own table. Rebuilding then starts from the checkpoint and folds only the events after it.

Save a checkpoint before you prune; a rebuild starts from it.

A table that's both partitioned and sorted can drop a whole day of events at once with delete_partitions(), which is significantly faster than deleting rows one by one. But delete_partitions() isn't transactional, and changing partitions while other writes target the table can leave the data in an indeterminate state (delete whole partitions). An event log usually has writers all the time, so prefer row deletes unless you can pause them. Partitioning must be chosen when you create the table (partitioned and sorted).

Consistent reads and snapshots

Within one transaction, every read sees the same snapshot. A reader that reads the current state and the watermark, or the current state and the log tail, in one transaction gets a consistent view (what a transaction sees).

For "as of" questions over a longer period, ask your platform team for snapshots of the database's path. A bucket snapshot holds the event log and the current state as they were when it was taken. It's also a safety net before you prune the log or switch to a rebuilt table. Snapshots are read-only: read from a snapshot in one transaction and write to the live tables in another (snapshots are read-only).

Worked example

The example models orders. A checkout service and a fulfilment service append events; one materializer folds them into the current state of each order. Timestamps and IDs are fixed so the output is the same on every run. In your own code, writers stamp appended_at with the current time when they append, and the materializer's cutoff is the current time minus the grace period.

Set up

Connect the same way as in the quickstart, with AWS_S3_ENDPOINT_URL, AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY set, and VASTDB_BUCKET naming your bucket. This block creates the schema docs_es_cqrs with three tables:

• events, the log, sorted on order_id and then event_time;
• current_state, one row per order;
• watermarks, one row holding the materializer's watermark.

import json
import os
from datetime import datetime

import pyarrow as pa
import vastdb

session = vastdb.connect()
BUCKET = os.environ["VASTDB_BUCKET"]
SCHEMA = "docs_es_cqrs"

EVENTS = pa.schema([
    ("event_id", pa.utf8()),
    ("order_id", pa.int64()),
    ("event_type", pa.utf8()),
    ("event_time", pa.timestamp("us")),
    ("appended_at", pa.timestamp("us")),
    ("payload", pa.utf8()),
])
CURRENT = pa.schema([
    ("order_id", pa.int64()),
    ("status", pa.utf8()),
    ("amount_cents", pa.int64()),
    ("events_applied", pa.int64()),
    ("last_event_id", pa.utf8()),
    ("last_event_time", pa.timestamp("us")),
])
WATERMARKS = pa.schema([("name", pa.utf8()), ("appended_at", pa.timestamp("us"))])

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).create_schema(SCHEMA, fail_if_exists=False)
    schema.create_table("events", EVENTS, sorting_key=["order_id", "event_time"], fail_if_exists=False)
    schema.create_table("current_state", CURRENT, fail_if_exists=False)
    marks = schema.create_table("watermarks", WATERMARKS, fail_if_exists=False)
    if marks.select().read_all().num_rows == 0:
        marks.insert(pa.table({"name": ["orders"], "appended_at": [datetime(1970, 1, 1)]},
                              schema=WATERMARKS))

def table(tx, name):
    return tx.bucket(BUCKET).schema(SCHEMA).table(name)

print("Ready")

Ready

The aggregate key, order_id, is an integer. Sorting uses only the first eight bytes of a sort column, so string keys with a long shared prefix, such as order-000101, don't sort well (choosing sort columns).

Append events from two writers

event() builds one event row, and append() inserts a batch of them. This block opens a transaction for each writer and appends from both while both are open, as two services would. The fulfilment service also appends event e-05 twice, as a redelivered message would:

def event(event_id, order_id, event_type, event_time, appended_at, **payload):
    return {"event_id": event_id, "order_id": order_id, "event_type": event_type,
            "event_time": event_time, "appended_at": appended_at,
            "payload": json.dumps(payload, sort_keys=True)}

def append(tx, events):
    table(tx, "events").insert(pa.Table.from_pylist(events, schema=EVENTS))

def at(hour, minute, second=0):
    return datetime(2026, 10, 11, hour, minute, second)

with session.transaction() as checkout:
    with session.transaction() as fulfilment:
        append(checkout, [
            event("e-01", 101, "created", at(10, 0), at(10, 0, 1), amount_cents=2500),
            event("e-02", 102, "created", at(10, 0, 5), at(10, 0, 6), amount_cents=1200),
            event("e-03", 101, "paid", at(10, 1), at(10, 1, 1)),
        ])
        append(fulfilment, [
            event("e-04", 102, "cancelled", at(10, 3), at(10, 3, 1)),
            event("e-05", 101, "shipped", at(10, 5), at(10, 5, 2)),
            event("e-05", 101, "shipped", at(10, 5), at(10, 5, 3)),  # redelivered
        ])
    print("fulfilment committed")
print("checkout committed")

with session.transaction() as tx:
    print("rows in the log:", table(tx, "events").select(columns=["event_id"]).read_all().num_rows)

fulfilment committed
checkout committed
rows in the log: 6

Both writers inserted into the same table while both transactions were open, and both committed. In production, append in batches that are as large as you reasonably can: many small inserts leave many small pieces that queries read until background compaction rewrites them (small writes).

Run a materializer pass

fold() turns events into current-state rows. It's a pure function: it sorts by event time and event_id, applies each event_id once, and ignores the order and duplicates of its input.

materialize(cutoff) is one pass, in one transaction. It finds the orders that have events appended after the watermark and up to cutoff. It re-folds each of those orders from its history in the log, which a sorted read on order_id serves. Then it replaces their current-state rows and moves the watermark to cutoff:

STATUS = {"created": "open", "paid": "paid", "shipped": "shipped",
          "delivered": "delivered", "cancelled": "cancelled"}

def fold(events):
    """Fold events into one current-state row per order. Order and duplicates don't matter."""
    state, seen = {}, set()
    for e in sorted(events.to_pylist(), key=lambda e: (e["event_time"], e["event_id"])):
        if e["event_id"] in seen:
            continue
        seen.add(e["event_id"])
        s = state.setdefault(e["order_id"], {"order_id": e["order_id"], "amount_cents": None,
                                             "events_applied": 0})
        s["status"] = STATUS[e["event_type"]]
        s["amount_cents"] = json.loads(e["payload"]).get("amount_cents", s["amount_cents"])
        s["events_applied"] += 1
        s["last_event_id"], s["last_event_time"] = e["event_id"], e["event_time"]
    return pa.Table.from_pylist([state[k] for k in sorted(state)], schema=CURRENT)

def materialize(cutoff):
    with session.transaction() as tx:
        events, current, marks = (table(tx, n) for n in ("events", "current_state", "watermarks"))
        mark = marks.select(columns=["appended_at"], internal_row_id=True).read_all()
        since = mark["appended_at"][0].as_py()

        new = events.select(columns=["order_id"], predicate=(events["appended_at"] > since)
                            & (events["appended_at"] <= cutoff)).read_all()
        orders = sorted(set(new["order_id"].to_pylist()))
        if orders:
            history = events.select(predicate=events["order_id"].isin(orders)
                                    & (events["appended_at"] <= cutoff)).read_all()
            old = current.select(columns=["order_id"], predicate=current["order_id"].isin(orders),
                                 internal_row_id=True).read_all()
            if old.num_rows:
                current.delete(old)
            current.insert(fold(history))

        marks.update(pa.table({"$row_id": mark["$row_id"],
                               "appended_at": pa.array([cutoff], pa.timestamp("us"))}),
                     columns=["appended_at"])
    print(f"folded {new.num_rows} new events for orders {orders}; watermark {cutoff}")

materialize(cutoff=at(10, 6))

folded 6 new events for orders [101, 102]; watermark 2026-10-11 10:06:00

The new rows include both copies of e-05; the fold applies it once. The delete, the insert and the watermark update commit together, so a reader sees either the state before the pass or the state after it.

Re-folding an aggregate from its history keeps the fold simple and makes each pass idempotent. It reads more as an aggregate's history grows; once you prune the log, fold from a checkpoint instead.

More events arrive, and the next pass picks up only those. A third pass, with no new events, changes nothing:

with session.transaction() as tx:
    append(tx, [
        event("e-06", 103, "created", at(10, 7), at(10, 7, 1), amount_cents=900),
        event("e-07", 101, "delivered", at(10, 8), at(10, 8, 1)),
    ])

materialize(cutoff=at(10, 10))
materialize(cutoff=at(10, 10))

folded 2 new events for orders [101, 103]; watermark 2026-10-11 10:10:00
folded 0 new events for orders []; watermark 2026-10-11 10:10:00

Query the current state and the history

Readers query current_state like any table:

with session.transaction() as tx:
    rows = table(tx, "current_state").select().read_all().sort_by("order_id")
for r in rows.to_pylist():
    print(r["order_id"], r["status"], r["amount_cents"], r["events_applied"], r["last_event_id"])

101 delivered 2500 4 e-07
102 cancelled 1200 2 e-04
103 open 900 1 e-06

For one order's history, filter the log on the sort key. Folding that history gives the order's state including events the materializer hasn't reached yet, which suits a detail view that must be fresh:

with session.transaction() as tx:
    log = table(tx, "events")
    history = log.select(predicate=log["order_id"] == 101).read_all()

for r in history.sort_by([("event_time", "ascending"), ("event_id", "ascending")]).to_pylist():
    print(r["event_time"], r["event_id"], r["event_type"])
print("state from the log:", fold(history).to_pylist()[0]["status"])

2026-10-11 10:00:00 e-01 created
2026-10-11 10:01:00 e-03 paid
2026-10-11 10:05:00 e-05 shipped
2026-10-11 10:05:00 e-05 shipped
2026-10-11 10:08:00 e-07 delivered
state from the log: delivered

The log keeps the redelivered e-05 twice; the fold applies it once.

Replay to rebuild the current state

Fold the whole log into a new table, and compare it with the table the incremental passes built:

with session.transaction() as tx:
    everything = table(tx, "events").select().read_all()
    rebuilt = fold(everything)
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    schema.create_table("current_state_rebuilt", CURRENT, fail_if_exists=False).insert(rebuilt)
    live = table(tx, "current_state").select().read_all().sort_by("order_id")

print("events replayed:", everything.num_rows)
print("rebuilt rows:", rebuilt.num_rows)
print("same as current_state:", rebuilt.equals(live.select(rebuilt.column_names)))

events replayed: 8
rebuilt rows: 3
same as current_state: True

On a large log, replay in chunks of aggregates or of append time, each in its own short transaction, rather than reading everything into memory at once.

Clean up

with session.transaction() as tx:
    schema = tx.bucket(BUCKET).schema(SCHEMA)
    for t in schema.tables():
        t.drop()
    schema.drop()
print("Dropped", SCHEMA)

Dropped docs_es_cqrs

Optional: Event Broker ingress

If your commands and events already flow through Kafka, put a VAST Event Broker topic in front of the log. A consumer in a group reads a batch of events, appends it in one transaction, and commits its offsets only after that transaction has committed. A crash between the two makes the group read the batch again (consume in a group), and the fold drops the duplicates by event_id. This fragment uses kafka_conf() from the consuming guide's setup and event() and append() from above:

from datetime import timezone
from confluent_kafka import Consumer, KafkaException

consumer = Consumer(kafka_conf({"group.id": "orders-event-log",
                                "auto.offset.reset": "earliest", "enable.auto.commit": False}))
consumer.subscribe(["orders"])

while True:
    messages = consumer.consume(num_messages=500, timeout=1.0)
    if not messages:
        continue
    now = datetime.now(timezone.utc).replace(tzinfo=None)
    rows = []
    for m in messages:
        if m.error():
            raise KafkaException(m.error())
        e = json.loads(m.value())
        rows.append(event(e["event_id"], e["order_id"], e["event_type"],
                          datetime.fromisoformat(e["event_time"]), now, **e.get("payload", {})))
    with session.transaction() as tx:
        append(tx, rows)
    consumer.commit(asynchronous=False)

Things to know on this path:

• Commit offsets per batch, not per message. A synchronous commit is a round trip to the broker. Committing after every message cut consumption throughput sharply, while one synchronous commit per batch of 500 cost much less than the append transaction before it. That's no reason to turn on enable.auto.commit: auto-commit commits on a timer, whether or not the append has committed, so a failed append loses the batch.

• The topic is already a table, in the kafka_topics schema, and you can query it with the SDK, Trino or Spark. But retention and compaction remove its rows in the background, and you can't write to it through the VAST DataBase API (query topics as tables). Keep your own event table as the permanent log.

• Blob expansion writes the JSON fields of each produced event into typed columns of a table, in the same transaction as the event, with no consumer code. Check its rules on target tables before you use it as the log.

• Committed offsets aren't replicated. After a failover, a group starts again from its auto.offset.reset setting (delete a consumer group). With earliest, that means duplicates, which the fold absorbs.

A VAST DataEngine function can also be the writer: each event that triggers it can append a row (VAST DataBase from a function).

Things to watch

• The newest events are sorted in the background. In a sorted table, new rows land unsorted and the cluster sorts them in blocks of 512K rows (write path: sorted tables). Reads are always correct; for the newest block the sort key just helps less.
• Don't measure the backlog with get_stats(). Its row counts are kept by background work and can lag, especially on sorted tables. Count the rows after the watermark with a query instead (table statistics).
• Keep materializer passes short. The cluster rolls back transactions that stay open too long (transactions that are left open). A rolled-back pass leaves nothing behind; the next one starts from the same watermark.
• Retry whole transactions, for writers and the materializer alike, and only on conflicts (retrying a whole transaction). A writer's retry after an ambiguous connection error can append twice; the event_id makes that harmless.

Before you create the tables, also check the general table rules that apply to any design: which table structures can be combined and replicated (limits), how snapshots hold on to deleted rows, and dropping a table that's being written to.

Trino and Spark

Writers can append with INSERT, and a materializer can run its fold as one transaction, with START TRANSACTION and COMMIT in Trino or ndb.create_tx() and ndb.commit_tx() in Spark (transactions in Trino, Spark and the VAST Query Engine, update and delete with Trino, Spark or the VAST Query Engine).

See also

• Transactions in practice and transactions: conflicts, isolation and retries.
• Updates and deletes and write path: what updates and deletes cost.
• Data modeling: sort keys and the other table structures.
• Snapshots and replication
• Consume events and query topics as tables