skip to content

In a Flutter app using Drift, how does a query's watch() stream know to re-emit after a write, and where does that mechanism break down?

level: middleimportance: must knowfreq 55%

answer

  1. table-level invalidation, not row-level
  2. writes through drift APIs notify
  3. initial snapshot on listen
  4. customStatement does not notify
  5. commit first, then streams re-run

basics

~20 s

Drift records which tables each watched query reads; every insert, update or delete made through drift's APIs marks those tables changed, and affected queries re-run and emit. Writes that bypass drift's notifications, such as customStatement or another connection, trigger nothing.

solid answer

~40 s

Any `Selectable` (a `select`, a join, a manager query or a `customSelect` with `readsFrom`) offers `watch()`, `watchSingle()` and `watchSingleOrNull()`. Drift's stream-query store remembers which tables each active stream reads. When a write goes through drift's `insert`, `update`, `delete`, `customUpdate` or `customInsert`, drift dispatches a table update and re-runs every stream on those tables, then emits the fresh result. A new listener always gets a current snapshot first. The mechanism is a table-level heuristic: it may re-emit when the rows you care about did not change, and it misses writes it cannot see, such as `customStatement` (call `markTablesUpdated` afterwards), a native SQLite client, or a second independent database instance. Inside a `transaction`, streams created outside it update once, after a successful commit.

code

dart · 25 lines
dart
import 'package:drift/drift.dart';

extension StreakQueries on AppDatabase {
  Stream<List<CheckIn>> watchCheckIns(int habitId) {
    final query = select(checkIns)
      ..where((c) => c.habit.equals(habitId))
      ..orderBy([(c) => OrderingTerm.desc(c.day)]);
    return query.watch();
  }

  Future<void> checkIn(int habitId) async {
    // Goes through drift, so watchCheckIns re-emits after the insert.
    await into(checkIns).insert(
      CheckInsCompanion.insert(habit: habitId, day: DateTime.now()),
    );
  }

  Future<void> purgeOrphans() async {
    // customStatement does not notify stream queries on its own.
    await customStatement(
      'DELETE FROM check_ins WHERE habit NOT IN (SELECT id FROM habits)',
    );
    markTablesUpdated({checkIns});
  }
}

go deeper

for a junior

Recall that watch() returns a Stream that emits immediately and again after writes, and that a StreamBuilder can render it.

for a middle

Explain the table-set bookkeeping, why customSelect needs readsFrom, and why customStatement needs markTablesUpdated.

for a senior

Discuss over-emission cost, transaction timing, independent instances across isolates and keeping watched queries cheap on large tables.

for a principal

Judge when table-level reactive queries are enough and when a feature needs explicit change events or a narrower read model.

## The problem watch() solves A habit tracker has a streak screen that shows how many days in a row a habit was checked in. The user taps "done" on another screen, a check-in row is inserted, and the streak must update without the screen polling the database or being told explicitly. **Drift** solves this with **stream queries**: any query can be turned into a Dart `Stream` that emits a new result whenever the data it depends on may have changed. In Flutter you consume it with a `StreamBuilder` or a stream-based provider. ## The API surface Every runnable query in drift implements `Selectable<T>`, which pairs one-shot and streaming methods: | One-shot | Streaming | Emits | |---|---|---| | `get()` | `watch()` | `List<T>` of all rows | | `getSingle()` | `watchSingle()` | exactly one row, an error otherwise | | `getSingleOrNull()` | `watchSingleOrNull()` | one row or `null` | That includes `select(checkIns)`, a join built with `.join([...])`, the generated manager API and `customSelect(...)`. A custom query must pass `readsFrom: {checkIns}` because drift cannot infer from raw SQL which tables to watch. ## How re-emission actually works 1. When a stream is listened to, drift runs the query once and emits the result, so you never need a separate `get()` before `watch()`. 2. Drift's **stream-query store** records the set of tables that stream reads. 3. Every write issued through drift's APIs (`into(t).insert`, `update(t)`, `delete(t)`, `batch`, `customInsert`, `customUpdate` with its `updates:` set) ends by dispatching a **table update** for the tables it touched. 4. Each active stream whose table set intersects the update is re-run and emits its new result. Two practical details follow. Streams are keyed by their SQL and variables, so two widgets watching the identical query share one underlying query. And after the last listener cancels, drift keeps the cached stream until the next event-loop turn (`Timer.run`) so a `StreamBuilder` that resubscribes during a rebuild does not re-query; that pending timer is why widget tests should `await db.close()`. ## Where it breaks down - **It is table-level.** Drift cannot tell which rows changed, so inserting a check-in for any habit re-runs every stream on `check_ins`, and a stream may emit a result identical to the previous one. Keep watched queries small and cheap. - **Writes drift does not see.** `customStatement` explicitly does not update stream queries; follow it with `markTablesUpdated({checkIns})` or use `customUpdate(..., updates: {checkIns})`. A native SQLite client or plugin writing to the same file is invisible too. - **Independent instances.** Two separately opened drift databases on the same file, for example one in a background isolate, do not share a stream store. Sharing one logical database across isolates (drift's isolate APIs or `shareAcrossIsolates`) fixes that. - **Transactions.** A stream created outside a `transaction` block updates once, after the transaction commits; if it rolls back, the stream does not update. A stream created inside the block sees each write and closes when the transaction ends. ## One check-in, traced end to end 1. The streak screen's state object calls `db.watchCheckIns(habitId)` once and hands the stream to a `StreamBuilder`. 2. Drift runs the `SELECT` and emits the current list; the builder renders a streak of, say, six days. 3. On another screen the user taps "done", and `into(checkIns).insert(...)` completes. 4. Drift dispatches an insert update for `check_ins`; the stream-query store finds the streak stream in that table's watcher set. 5. The query re-runs and emits seven rows; the builder rebuilds with the new streak, without either screen knowing about the other. Because the screens are decoupled through the database, the same mechanism covers writes from a notification action or a sync job, as long as those writes go through the same drift database instance. ## Using it well in Flutter - Create the stream once, in `initState`, a provider or a bloc, not inside `build`, so a rebuild does not create and re-run a fresh query every frame. - Map rows to view models with `.map(...)` on the query before `watch()` rather than doing heavy work in the widget. - For a derived value such as a streak count, prefer a query that returns just that value over watching every check-in and counting in Dart.

  • In Drift, why would a customSelect stream never update even though rows are being inserted into the table it reads?
    Drift does not infer at runtime which tables a `customSelect`'s SQL reads, so it needs `readsFrom: {checkIns}`. Without it the stream has an empty table set, no table update intersects it, and it only ever emits its initial snapshot.
  • In Drift, what does a watched query created outside a transaction see while the transaction is still running?
    Nothing new. Drift holds the table updates until the transaction completes successfully and then re-runs affected streams once; if the transaction throws and rolls back, those streams are not re-run at all.
  • In a Flutter widget test that uses a Drift database, why do you close the database at the end?
    Drift keeps a cancelled stream cached until the next event-loop turn using a timer, and the test binding reports pending timers as a failure. Awaiting `db.close()` shuts the stream store down cleanly.

It works like a building's fire panel that knows only which floor an alarm came from, not which room: every watcher on that floor gets re-checked, even if their own room was fine, and a fire started by someone who bypassed the panel is never announced.

saying these in an interview costs you the question

  • Drift streams update only the rows that actually changed.
  • You must call get() first because watch() waits for the next write.
  • Any SQL run on the database file triggers drift streams.
  • customStatement writes refresh watched queries automatically.
  • Streams see uncommitted writes from another transaction immediately.