Realtime Subscriptions

The Dart/Flutter SDK supports live queries over WebSocket.

The important ownership boundary is:

• Rust owns the shared connection, reconnect loop, replay filtering, and resume checkpoint tracking.
• Dart owns only the stream wrapper and typed row decoding.

That keeps reconnect behavior consistent with other SDKs and avoids re-implementing replay logic in Flutter apps.

On native Dart targets, subscription traffic uses MessagePack by default on that shared Rust-managed connection, which trims wire size and decode overhead compared with JSON.

Live SQL must follow the strict supported shape: SELECT ... FROM ... WHERE .... Do not include ORDER BY or LIMIT in liveEvents() or materialized live-query SQL.

Use:

1Stream<ChangeEvent> liveEvents(2  String sql, {3  int? batchSize,4  int? lastRows,5  SeqId? from,6  String? subscriptionId,7})

For low-level event handling, liveEvents(...) is the right API.

Unlike query(...), the current live-query APIs take the final SQL string directly and do not expose a separate params argument.

The Dart SDK also exposes materialized live-row helpers when you want the SDK to keep the current row set reconciled for you:

• live<T>(sql, ...)
• liveTable<T>(tableName, ...)

Which API to use

Choose the API based on the shape your app actually needs:

• Use liveEvents(...) when you need low-level protocol events like AckEvent, batched snapshots, insert/update/delete diffs, or exact resume control in your own event loop.
• Use live(...) when your UI wants the latest materialized row list for an arbitrary SQL query.
• Use liveTable(...) when SELECT * FROM namespace.table is enough.

For most widget state and list rendering, prefer the materialized helpers. They remove the need to reconcile InitialDataBatch, InsertEvent, UpdateEvent, and DeleteEvent yourself.

Materialized live rows

live<T>(...) keeps the current row set updated inside the SDK and emits the latest list whenever the query result changes.

1final rowsStream = client.live<Map<String, KalamCellValue>>(2  "SELECT id, room, body, created_at FROM app.messages WHERE room = 'main'",3  lastRows: 20,4  limit: 20,5  onCheckpoint: (checkpoint) {6    persistSeqId(checkpoint.lastSeqId.toString());7  },8);9 10await for (final rows in rowsStream) {11  for (final row in rows) {12    print('${row['id']?.asInt()} ${row['body']?.asString()}');13  }14}

Use liveTable<T>(...) when SELECT * FROM table is enough:

1final taskRows = client.liveTable<Map<String, KalamCellValue>>('app.tasks');2 3await for (final rows in taskRows) {4  print('tasks: ${rows.length}');5}

If your live query uses a natural or composite key instead of a plain id column, pass keyColumns so reconciliation still stays inside the shared Rust core:

1final messages = client.live<Map<String, KalamCellValue>>(2  "SELECT room_id, message_id, body FROM app.messages WHERE room_id = 'main'",3  keyColumns: ['room_id', 'message_id'],4);

You can also cap the in-memory materialized row set with limit, map rows directly into app models with mapRow, and persist a durable resume cursor with onCheckpoint.

Materialized rows vs change events

• liveEvents(...) gives you protocol-level events and incremental changes.
• live(...) and liveTable(...) give you the fully reconciled current row list.
• Materialized helpers are usually the better default for UI code.
• Low-level events are better when you need custom reconciliation, audit logging, or protocol inspection.

Materialized-row limits

The SQL for materialized live rows still follows the same strict supported shape:

• SELECT ... FROM ... WHERE ...
• no ORDER BY
• no LIMIT

The controls apply at different stages:

• batchSize chunks the initial snapshot sent by the server.
• lastRows selects how much history to rewind before live changes begin.
• limit caps the materialized live row set the SDK keeps after startup.

Keep Dart-side presentation logic for sorting and grouping after rows arrive. Use limit only when the materialized live state itself should stay bounded.

Concurrency notes

The Dart SDK test suite covers multi-client fan-out and many simultaneous subscriptions on one shared client connection. That means the supported path is not just single-subscriber demos: concurrent writers and multiple active listeners on the same query are part of the exercised contract.

Basic subscription

1final stream = client.liveEvents(2  "SELECT * FROM app.messages WHERE room = 'main'",3);4 5await for (final event in stream) {6  switch (event) {7    case AckEvent(:final subscriptionId, :final schema, :final totalRows):8      print('ack $subscriptionId: $totalRows rows, columns=${schema.length}');9    case InitialDataBatch(:final rows, :final hasMore):10      print('snapshot batch: ${rows.length}, hasMore=$hasMore');11    case InsertEvent(:final row):12      print('insert: ${row['id']?.asInt()}');13    case UpdateEvent(:final row, :final oldRow):14      print('update: $oldRow -> $row');15    case DeleteEvent(:final row):16      print('delete: $row');17    case SubscriptionError(:final code, :final message):18      print('subscription error [$code]: $message');19  }20}

Snapshot vs live changes

Most subscriptions follow this pattern:

1. AckEvent confirms the subscription and includes schema metadata.
2. One or more InitialDataBatch events deliver the initial snapshot.
3. InsertEvent / UpdateEvent / DeleteEvent deliver live changes.

If you only care about live changes, you can ignore InitialDataBatch after initial UI hydration.

Controlling the initial snapshot

batchSize

If the initial query returns many rows, the server can send it in batches. batchSize controls the maximum rows per snapshot batch.

1final stream = client.liveEvents(2  'SELECT * FROM app.large_table',3  batchSize: 100,4);

You will receive one or more InitialDataBatch events with hasMore=true, followed by hasMore=false and then live changes.

lastRows

lastRows asks the server to “rewind” and include the last N rows before live changes begin. This is useful for chat timelines or activity feeds.

1final stream = client.liveEvents(2  'SELECT * FROM chat.messages WHERE room_id = $1',3  lastRows: 50,4);

subscriptionId

Provide an explicit subscriptionId if you want stable IDs for client-side bookkeeping:

1final stream = client.liveEvents(2  "SELECT * FROM app.messages WHERE room = 'main'",3  subscriptionId: 'messages-feed',4);

from

Use from to resume a subscription from a known sequence ID.

from is the only resume cursor exposed by the Dart API. Commit ordering and snapshot boundaries stay on the server side.

This is most useful when you:

• persist a checkpoint (last processed seq id) across app restarts
• want to re-subscribe without re-processing older events

1final stream = client.liveEvents(2  "SELECT * FROM app.messages WHERE room = 'main'",3  from: SeqId.parse('123'),4);

For materialized live rows, prefer onCheckpoint so each applied snapshot gives you the latest SeqId directly. getSubscriptions() remains available when you need to inspect the whole shared connection state.

When you resume with from, the intended contract is strict resume:

• rows with _seq <= from must not be replayed
• rows written while the connection was down must still arrive after reconnect
• new live rows after reconnect continue on the same stream

The Dart reconnect and resume test suite covers these cases with duplicate _seq assertions so the SDK keeps a strict “no replay, no missed change” boundary.

Listing active subscriptions

Use getSubscriptions() to inspect all subscriptions on the shared connection.

1final subs = await client.getSubscriptions();2for (final sub in subs) {3  print('${sub.id}: ${sub.query} (lastSeqId=${sub.lastSeqId}, closed=${sub.closed})');4}

For server-side observability while debugging, query SELECT * FROM system.live. That view is node-local and reflects the subscriptions currently held in memory on the node you query. It is not persisted to RocksDB and it is not replicated through Raft.

Cancelling a subscription

The SDK returns a Stream<ChangeEvent>. Cancel by cancelling your StreamSubscription.

1final sub = client.liveEvents('SELECT * FROM app.messages').listen((_) {});2 3// later4await sub.cancel();

Cancellation stays lightweight on the Dart side. The SDK cancels the Rust subscription handle on the shared connection, waits briefly for the pull loop to unblock, and then closes the stream. That avoids leaving a hanging Dart-side pull loop after a widget or provider disposes.

Data types

All subscription events expose rows as Map<String, KalamCellValue> for type-safe access:

• InitialDataBatch.rows → List<Map<String, KalamCellValue>>
• InsertEvent.row / InsertEvent.rows
• UpdateEvent.columnIndexes / UpdateEvent.row / UpdateEvent.oldRow / UpdateEvent.oldRows
• DeleteEvent.row / DeleteEvent.oldRows

UPDATE event behavior

For UpdateEvent, row and rows are sparse typed maps reconstructed from the Ack schema. They contain only the changed columns plus the primary key columns and _seq.

oldRow contains the previous values for that same sparse column set.

columnIndexes exposes the original positional update layout from the wire protocol. It lines up with the values sent on the Rust bridge raw event if you need exact sparse-column ordering.

1case UpdateEvent(:final row, :final oldRow):2  // row has ALL non-null columns — no need to re-query3  final id = row['id']?.asString();4  final updateType = row['update_type']?.asString();5  final targetId = row['target_id']?.asString();6  print('Trigger: $updateType on $targetId');7 8  // oldRow has only the changed columns9  for (final key in oldRow.keys) {10    if (!key.startsWith('_') && key != 'id') {11      print('  changed: $key: ${oldRow[key]} → ${row[key]}');12    }13  }

Use the typed accessors to extract values:

1case InsertEvent(:final row):2  final id    = row['id']?.asInt();3  final name  = row['name']?.asString();4  final score = row['score']?.asDouble();5  final flag  = row['active']?.asBool();

These event accessors are already decoded from JSON row payloads; no manual List<List<dynamic>> parsing is needed.

See Cell Values for the full KalamCellValue API.

Next

• Client Lifecycle
• API: WebSocket Protocol
• Architecture: Live Query