> ## Documentation Index > Fetch the complete documentation index at: https://confect.dev/llms.txt > Use this file to discover all available pages before exploring further. # Streams > Compose, merge, join, and paginate index queries as Effect streams. Stream querying is experimental. The API is usable end to end, but its surface may still change. Query streams let you merge, filter, join, and deduplicate index queries while retaining cursor pagination. A **query stream** is an Effect [`Stream`](https://effect.website/docs/stream/introduction) of decoded documents with stored keys, a shared key layout, and a direction. Use `reader.table(...).stream(...)` for standard indexes; search indexes use [`search`](/server/database/reading#search-indexes). ## A first query After [setting up Confect](/getting-started/quickstart), define a `feed` [paginated query spec](/server/functions#paginated-queries) with the notes document as its `item` schema. Its handler receives `paginationOpts` automatically. This handler merges notes from two roles, removes hidden notes, and returns a page: ```ts confect/notes.impl.ts theme={null} import { FunctionImpl, QueryStream } from "@confect/server"; import * as Effect from "effect/Effect"; import { DatabaseReader } from "./_generated/services"; import databaseSchema from "./_generated/schema"; import notes from "./notes.spec"; const feed = FunctionImpl.make( databaseSchema, notes, "feed", ({ paginationOpts }) => Effect.gen(function* () { const reader = yield* DatabaseReader; const byRole = (role: "admin" | "user") => reader .table("notes") .stream("by_role", (q) => q.eq("author.role", role), "desc"); return yield* QueryStream.merge([byRole("admin"), byRole("user")]).pipe( QueryStream.filter((note) => note.tag !== "hidden"), QueryStream.paginate(paginationOpts), ); }).pipe(Effect.orDie), ); ``` In React, use [`useStreamPaginatedQuery`](/clients/react#usestreampaginatedquery), not `usePaginatedQuery`, for stream-paginated queries. Stream pages aren't tracked by Convex's query journal; the stream hook pins page ranges with `endCursor` to keep loaded pages gap-free as data changes. `reader.table("notes").stream(...)` is different from `reader.table("notes").index(...).stream()` on the [Reading](/server/database/reading#stream) page. The latter is a plain `Stream` over one query; the former is a composable query stream. ## Operations at a glance The last column compares each operation with Effect's `Stream` operations. | Operation | Effect on ordering | In Effect | | - | - | - | | [`stream`](#creating-a-stream) | Keys contain unpinned index values and any implicit ID | A `Stream` that remembers how it is sorted | | [`empty`](#empty-streams) | Uses the supplied layout | `Stream.empty` with a layout | | [`merge`](#merge) | Requires matching layouts; preserves keys | Ordered, unlike `Stream.merge` | | [`filter`/`filterEffect`](#filter-and-map) | Preserves keys and layout | `Stream.filter`, `Stream.filterEffect` | | [`map`/`mapEffect`](#filter-and-map) | Preserves keys and layout | `Stream.map`, `Stream.mapEffect` | | [`distinct`](#distinct) | Preserves keys and layout | `Stream.changes` on a key prefix | | [`narrow`](#narrow-to-a-key-range) | Preserves keys and layout | `dropWhile` and `takeUntil` on the key, pushed into the index | | [`flatMap`](#flat-map) | Concatenates outer and inner keys and layouts | Sequential `Stream.flatMap` | | [`flatMap` with `onEmpty`](#keep-outer-documents-without-inner-documents) | Concatenates outer and inner keys and layouts | `Stream.orElseIfEmpty` on each inner stream | | [`renameKey`](#rename-ordering-labels) | Visible labels replaced; key values kept | No equivalent | | [`reverse`](#reverse) | Preserves keys and layout; flips direction | Reads the index backwards rather than reversing collected values | | [`Stream.runCollect`, `runHead`](#consuming-a-stream) | Consumed | Effect's own consumers | | [`Stream.take`](#consuming-a-stream) | Not retained | A transform returning a plain `Stream` | | [`unique`](#consuming-a-stream) | Consumed | `Stream.runHead` with a uniqueness check | | [`paginate`](#paginating-a-stream) | Consumed | `Stream.take(numItems)` after narrowing to the cursor | ## The ordering contract Every query stream visits its **keys** in its **direction**. Keys are stored separately from emitted values, so mapping a document doesn't change its position. Compatible streams can be merged in key order, and pagination resumes after a saved key rather than an offset. The `KeyLabels` and `OrderDirection` type parameters enforce this contract: TypeScript rejects known label or direction mismatches in `merge`. Runtime checks also require matching directions and [key layouts](#keys-labels-and-layouts), including implicit-ID positions. `flatMap` appends the inner key to the outer key; `renameKey` replaces visible labels without changing key values. `reverse` changes the direction, not the key. ## Example data and diagram notation The examples use these two tables. `n1` and `c1` are readable stand-ins for real Convex document IDs: | `notes` | `text` | `author.role` | `tag` | `_creationTime` | | - | - | - | - | - | | `n1` | `"apple"` | `"admin"` | | 1 | | `n2` | `"banana"` | `"user"` | | 2 | | `n3` | `"apple"` | `"user"` | | 3 | | `n4` | `"cherry"` | `"admin"` | | 4 | | `n5` | `"banana"` | `"admin"` | | 5 | | `n6` | `"date"` | `"user"` | `"hidden"` | 6 | | `comments` | `noteId` | `body` | `_creationTime` | | - | - | - | - | | `c1` | `n1` | `"great"` | 7 | | `c2` | `n1` | `"meh"` | 8 | | `c3` | `n4` | `"nice"` | 9 | `notes` has the indexes `by_text` on `["text"]` and `by_role` on `["author.role"]`; `comments` has `by_note` on `["noteId"]` and `by_body` on `["body"]`. Runtime keys include `_creationTime` and `_id` tiebreakers, so `by_text` orders by text, creation time, then ID. ### Reading the diagrams A stream is drawn as a track, read left to right in **key order**—the way a marble diagram reads left to right in time, with the index's ordering as the axis: Diagram keys are **abbreviated**: `[apple,1]` means `["apple", 1, "n1"]`, with `_id` omitted for space. Joined keys omit both outer and inner IDs: `[1,7]` means `[1, "n1", 7, "c1"]`. Cursor labels such as `cursor: [apple,3]` stand for schematic **strings**, not arrays accepted by the API. Pass returned cursors back unchanged; see [Cursor handling](#cursor-handling). ```text theme={null} by_text ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── n6 ───────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ``` * The stream's name is printed above its track, and `╰` joins the two. * `─ n1 ─` is an element: the document `n1`. Its **key**, the values the stream is ordered by, is printed beneath it. * `─ (n6) ─` is an element that was read but filtered out. It emits nothing, but it still counts toward read budgets, and cursors still advance past it. * A blank stretch of track is key space the stream never read (skipped by a seek). * `┤` is the end of the stream, `╎` a cursor: a position between two keys. * `╞═ … ═╡` is the operation between the input track (above) and its output (below). Tracks in one diagram share columns, so elements that line up vertically hold the same position in the output's order. ## Creating a stream `stream` takes an index name, an optional range callback, and an optional order (`"asc"` by default). It returns a `QueryStream` whose elements are decoded documents, in index order. Creating or composing a stream does not read documents. Reads begin when you run a consuming effect, such as `Stream.runCollect` or `QueryStream.paginate`; each run executes the queries again. ```ts Every note, sorted by text theme={null} reader.table("notes").stream("by_text"); ``` ```ts Every note, sorted by text, descending theme={null} reader.table("notes").stream("by_text", "desc"); ``` ```ts Notes whose text is at least "banana", sorted by text theme={null} reader.table("notes").stream("by_text", (q) => q.gte("text", "banana")); ``` ```ts Notes whose text is exactly "apple", sorted by creation time theme={null} reader.table("notes").stream("by_text", (q) => q.eq("text", "apple")); ``` The four streams over the example data, with each element's key. A bound keeps a field in the key; pinning with `eq` removes it: ```text theme={null} by_text ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── n6 ───────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] by_text desc ╰─ n6 ──────── n4 ──────── n5 ──────── n2 ──────── n3 ──────── n1 ───────┤ [date,6] [cherry,4] [banana,5] [banana,2] [apple,3] [apple,1] gte "banana" ╰───────────────────────── n2 ──────── n5 ──────── n4 ──────── n6 ───────┤ [banana,2] [banana,5] [cherry,4] [date,6] eq "apple" ╰─ n1 ──────── n3 ───────┤ [1] [3] ``` ### Range callbacks The range callback mirrors Convex's index range builder with one addition: it tracks which fields are still *varying* after the range is applied. * `eq` pins the next index field to a value and consumes it—it no longer varies within the stream. * `gt`, `gte`, `lt`, and `lte` bound the next field without consuming it, and must come last (a lower bound may be followed by an upper bound on the same field). Fields must be used in the order the index declares them, and each method only offers the next unused field, so misuse is a type error. ### Keys, labels, and layouts An **key** contains the values that determine an element's position. Every element has its own key values; the stream has one shared **key layout** describing the positions in those keys: their sequence, their labels, and where implicit document-ID tiebreakers occur. For a note ordered by text: | Concept | Example | Meaning | | - | - | - | | Key values | `["apple", 1, "n1"]` | The note's text, creation time, and ID. | | Key labels | `["text", "_creationTime"]` | Names for the positions you can select with `distinct` or relabel with `renameKey`. | | Key layout | `text → _creationTime → implicit ID` | All three positions, including the ID omitted from the labels. | Labels start as index field paths, but `renameKey` can replace them with aliases. Labels can repeat, especially after a join. An **implicit ID** has no label; it still participates in ordering, bounds, and pagination. An explicit ID, such as the `_id` in `by_id`, has a label. Pinning text with `eq` removes its position: the key becomes `[1, "n1"]`, the labels become `["_creationTime"]`, and the layout contains creation time followed by the implicit ID. Range bounds such as `gte` keep the position. A join concatenates the full outer and inner keys, retaining both IDs. For a note joined to its comments, both ordered by creation time: ```text theme={null} Outer labels: ["_creationTime"] Inner labels: ["_creationTime"] Joined labels: ["_creationTime", "_creationTime"] Joined layout: outer time → outer implicit ID → inner time → inner implicit ID One joined key: [1, "n1", 7, "c1"] ``` Labels alone cannot describe the joined layout: the outer ID sits between the two labeled positions. A `distinct` label prefix that reaches into the inner key also includes that intervening ID, so it groups within each outer document. Layouts are **compatible** when they have the same total width, the same labels in the same positions, and the same implicit-ID positions. Direction is stored in `stream.orderDirection` and checked separately. A layout contains no key values, value-type schema, table identity, or equality-pinned values, so different indexes or tables can have compatible layouts. When merging them, choose positions whose values have the same meaning; matching layouts alone does not establish that. Reuse a stream's `keyLayout` when calling `empty` or supplying `flatMap`'s `innerKeyLayout`. Creating a stream to obtain its layout reads no documents. The type is normally inferred, but you can import the layout module directly when you need an annotation: ```ts theme={null} import type * as QueryStreamKeyLabels from "@confect/server/QueryStreamKeyLabels"; import type * as QueryStreamKeyLayout from "@confect/server/QueryStreamKeyLayout"; const keyLayout: QueryStreamKeyLayout.QueryStreamKeyLayout< QueryStreamKeyLabels.QueryStreamKeyLabels<["text", "_creationTime"]> > = reader.table("notes").stream("by_text").keyLayout; ``` The type parameter tracks labels; implicit-ID positions are checked at runtime. For explicit `narrow` bounds, pass key **values**, including any intervening implicit IDs. For `distinct`, pass a prefix of the **labels**; for `renameKey`, pass replacement labels for every labeled position. ### Empty streams `QueryStream.empty` is a query stream with no documents but a known key layout and direction, for the places where a stream is required and there is nothing to put in it. Its usual job is a merge over a list of streams that may turn out empty, since `merge` needs at least one input. ```text theme={null} nothing ╰┤ ``` Pass the document type as a type argument, then reuse a compatible stream's `keyLayout`. Creating that stream does not read documents. The [generated document types](/server/database/schema#document-types) name the document: ```ts theme={null} import { QueryStream } from "@confect/server"; import * as Array from "effect/Array"; import type { NotesDoc } from "./confect/_generated/docs"; const byRole = (role: "admin" | "user") => reader.table("notes").stream("by_role", (q) => q.eq("author.role", role)); const notesByRoles = (roles: ReadonlyArray<"admin" | "user">) => Array.match(roles, { onEmpty: () => QueryStream.empty()(byRole("admin").keyLayout), onNonEmpty: (some) => QueryStream.merge(Array.map(some, byRole)), }); ``` ## Consuming a stream A query stream is a genuine `Stream`, so everything in Effect's `Stream` module applies (`QueryStream.isQueryStream` tells the two apart at runtime): `Stream.runCollect` returns an effect that collects all results; `Stream.runHead` returns an effect producing the first result as an `Option`. `Stream.take(n)` is a transform, not a consumer: it returns a plain stream of at most `n` results, which you can then collect. ```ts theme={null} import * as Effect from "effect/Effect"; import * as Stream from "effect/Stream"; Effect.gen(function* () { const firstTwoTexts = yield* reader .table("notes") .stream("by_text") .pipe( QueryStream.map((note) => note.text), Stream.take(2), Stream.runCollect, ); // ["apple", "apple"] }); ``` Once you apply a plain `Stream` combinator the result is an ordinary `Stream`: it can be consumed, but it no longer retains the key layout and stored keys, so it can't be merged or paginated further. Use the `QueryStream` combinators below when you need to keep that ability. `QueryStream.unique` consumes a stream expected to hold at most one document, returning an `Option` and failing with `NotUniqueError` if there are two or more: ```ts theme={null} import * as Effect from "effect/Effect"; Effect.gen(function* () { const cherry = yield* QueryStream.unique( reader.table("notes").stream("by_text", (q) => q.eq("text", "cherry")), ); }); ``` ```text theme={null} eq "cherry" ╰─ n4 ───────┤ → Some(n4) [4] eq "apple" ╰─ n1 ──────── n3 ───────┤ → NotUniqueError [1] [3] eq "fig" ╰┤ → None ``` ## Composing streams Keep callbacks deterministic and read-only: pagination, seeks, and reversal can reevaluate them for the same document. Effects should be safe to run again, not perform writes or other one-time side effects. ### Merge `QueryStream.merge` interleaves streams with matching key layouts into one ordered stream. Merging is how you query over several index ranges at once, such as the notes by two roles in the example at the top of this page. It is an ordered merge, the step of merge sort that combines sorted runs, always emitting the smallest next key (the largest, descending). It is not `Stream.merge`, which interleaves inputs in arrival order. ```ts theme={null} const admin = reader .table("notes") .stream("by_role", (q) => q.eq("author.role", "admin")); const user = reader .table("notes") .stream("by_role", (q) => q.eq("author.role", "user")); const merged = QueryStream.merge([admin, user]); ``` Pinning `author.role` leaves both streams with visible labels `["_creationTime"]` and a layout containing creation time followed by an implicit ID, so the merge interleaves them by creation time and then ID: ```text theme={null} admin ╰─ n1 ──────────────────────────────── n4 ──────── n5 ───────────────────┤ [1] [4] [5] user ╰───────────── n2 ──────── n3 ──────────────────────────────── n6 ───────┤ [2] [3] [6] ╞═ merge([admin, user]) ═╡ merged ╰─ n1 ──────── n2 ──────── n3 ──────── n4 ──────── n5 ──────── n6 ───────┤ [1] [2] [3] [4] [5] [6] ``` Inputs must have compatible document types and matching key layouts and directions. TypeScript rejects known mismatches; runtime mismatches throw when the streams are combined. For different document shapes or field names, [map and relabel](#rename-ordering-labels) first. Overlapping inputs are not deduplicated: a document matched by two inputs appears twice. ### Filter and map `QueryStream.filter` removes emitted values; `QueryStream.map` transforms them. Both preserve the original stored keys. A rejected document still counts as *read*, so cursors can advance past it. ```ts theme={null} reader .table("notes") .stream("by_text") .pipe( QueryStream.filter((note) => note.tag !== "hidden"), QueryStream.map((note) => note.text), ); ``` The hidden note is read, rejected, and kept in the cursor accounting; the mapped elements keep their keys even though the key's fields are no longer in the elements: ```text theme={null} by_text ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── n6 ───────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ╞═ filter((note) => note.tag !== "hidden") ═╡ filtered ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── (n6) ─────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ╞═ map((note) => note.text) ═╡ mapped ╰─ apple ───── apple ───── banana ──── banana ──── cherry ──── (n6) ─────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ``` A mapper may replace any emitted fields, including fields named in the index, or emit a different shape entirely. It never recomputes the stored key or sorts by the new values. For example, mapping `note.text` to its length still orders by the original text, creation time, and ID—not by length. When the predicate or mapper needs to run an effect—read another table, use a service, or fail with a typed error—use `QueryStream.filterEffect` and `QueryStream.mapEffect` instead. They behave the same way, and the effect's error and requirement types flow into the stream's: ```ts theme={null} // Comments on notes that aren't hidden. reader .table("comments") .stream("by_body") .pipe( QueryStream.filterEffect((comment) => reader .table("notes") .get(comment.noteId) .pipe(Effect.map((note) => note.tag !== "hidden")), ), ); ``` Both run one document's effect at a time by default. Pass `{ concurrency }` to run several at once; elements are still emitted in stream order, so the result is the same query stream, just faster when each effect is a database read: ```ts theme={null} // Each comment with its note attached. reader .table("comments") .stream("by_body") .pipe( QueryStream.mapEffect( (comment) => reader .table("notes") .get(comment.noteId) .pipe(Effect.map((note) => ({ ...comment, note }))), { concurrency: 8 }, ), ); ``` ### Distinct `QueryStream.distinct` keeps the first document for each distinct value of a prefix of the key. After a group's first document, `distinct` seeks past the group with a fresh index read instead of scanning the rest of it. ```ts theme={null} // One note per distinct text. reader .table("notes") .stream("by_text") .pipe(QueryStream.distinct(["text"])); ``` After `n1` is found, the stream seeks straight to the first key past `apple`; `n3` is never read. The same happens after `n2` for `banana`: ```text theme={null} by_text ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── n6 ───────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ╞═ distinct(["text"]) ═╡ distinct ╰─ n1 ──────────────────── n2 ──────────────────── n4 ──────── n6 ───────┤ [apple,1] [banana,2] [cherry,4] [date,6] ``` Pass a prefix of the stream's visible labels, checked by TypeScript and at runtime. The layout resolves those labels to a prefix of the key values. On a `flatMap` result, a prefix reaching into the inner key includes the outer ID, so it groups within each outer document. Implicit IDs after the last selected label are excluded from the grouping prefix. Operation order determines which document represents each group. Apply `filter` (or `filterEffect`) before `distinct` to choose the first matching document; apply it after `distinct` to test the chosen representative, omitting the group if that document fails. [Reversing](#reverse) a distinct stream keeps its representatives and reverses their output order. Reversing the input before applying `distinct` instead chooses the first document from the other direction. ### Narrow to a key range `QueryStream.narrow` restricts a stream to the keys between `start` and `end`. Each endpoint has `keyValues` and a required `inclusive` flag, so you can choose any combination of inclusive and exclusive bounds. Provide at least one endpoint; omit the other to leave that side unbounded. Endpoints follow **stream order**: on an ascending stream, `start` is the lower key; on a descending stream, it is the upper key. Narrowing intersects existing bounds, so repeated calls can only restrict the range further. ```ts theme={null} const between = QueryStream.narrow(reader.table("notes").stream("by_text"), { start: { keyValues: ["banana", 2, "n2"], inclusive: false }, end: { keyValues: ["cherry", 4, "n4"], inclusive: true }, }); ``` Bounds are pushed into underlying index ranges where that preserves the composed query's results. Some compositions, such as `distinct`, may still read outside the output bounds to find the correct representative. For this direct index stream, the bounds become index-range predicates, so the elements outside them are not read at all: ```text theme={null} by_text ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── n6 ───────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ╎ start ╎ end ╞═ narrow: start exclusive, end inclusive ═╡ narrowed ╰───────────────────────────────────── n5 ──────── n4 ───────┤ [banana,5] [cherry,4] ``` A key can be a prefix of the full key. An inclusive prefix includes all keys extending it; an exclusive prefix excludes that whole group. For example, an ascending creation-time stream can select a time window without supplying the trailing `_id`: ```ts theme={null} const startTime = Date.UTC(2026, 0, 1); const endTime = Date.UTC(2026, 0, 2); const duringDay = reader .table("notes") .stream("by_creation_time") .pipe( QueryStream.narrow({ start: { keyValues: [startTime], inclusive: true }, end: { keyValues: [endTime], inclusive: false }, }), ); ``` This includes every note at `startTime` and excludes every note at `endTime`. Adjacent windows can share an endpoint without overlapping. Apply the same bounds to a merged stream when its leading key value is also a creation time. When both endpoints use the same key, the range is empty unless both are inclusive. Two inclusive endpoints select that key, or the entire group for a prefix key. On a `distinct` stream, full-key bounds filter the original representatives exactly; they do not select a replacement from within the narrowed range. Explicit prefix bounds still include or exclude whole groups. Bounds applied before `distinct` constrain its input and can change which document represents a group; bounds applied after it constrain only the output, preserving that original input range and its representatives. To resume pagination, pass the previous page's `continueCursor` directly to `QueryStream.paginate` as `cursor`. ### Flat-map `QueryStream.flatMap` runs an inner stream for each document of an outer stream and concatenates the results, ordered by the outer key and then the inner key. An outer document whose inner stream is empty contributes no elements. Use [`onEmpty`](#keep-outer-documents-without-inner-documents) to emit a placeholder for it. Supply `innerKeyLayout` from a stream whose layout matches every stream returned by the callback. The joined stream needs its layout at construction, before any outer document is available to pass to the callback—even if the outer stream is empty. The layout determines the joined key positions and the width of markers for empty inner streams. Here, pinning `noteId` in `by_note` leaves creation time followed by an implicit ID. A `by_creation_time` stream has the same layout and supplies the template without needing a note ID. Creating that template reads no documents. ```ts theme={null} import { QueryStream } from "@confect/server"; // The comments on admin notes, grouped by note in note order and by // creation time within each note. const commentsOn = (note: NotesDoc) => reader.table("comments").stream("by_note", (q) => q.eq("noteId", note._id)); const innerKeyLayout = reader .table("comments") .stream("by_creation_time").keyLayout; reader .table("notes") .stream("by_role", (q) => q.eq("author.role", "admin")) .pipe(QueryStream.flatMap(commentsOn, { innerKeyLayout })); ``` Each admin note's comment stream runs in turn. The result's key is the outer key followed by the inner key; `n5` has no comments, so it contributes only a filtered marker that keeps cursors moving: ```text theme={null} admin ╰─ n1 ──────────────────── n4 ──────── n5 ───────┤ [1] [4] [5] of n1 ╰─ c1 ──────── c2 ───────┤ of n4 ╰─ c3 ───────┤ of n5 ╰┤ ╞═ flatMap((note) => commentsOn(note), { innerKeyLayout }) ═╡ joined ╰─ c1 ──────── c2 ──────── c3 ──────── (n5) ─────┤ [1,7] [1,8] [4,9] [5,null] ``` Every returned inner stream must match `innerKeyLayout` and run in the outer stream's direction. The template's direction is separate from its layout. TypeScript rejects known label or direction mismatches; runtime mismatches die with `InnerStreamLayoutMismatchError` or `InnerStreamOrderMismatchError` when the join runs. The joined key keeps both IDs: comment `c1` has key `[1, "n1", 7, "c1"]`. For renamed inner streams, apply the same renaming to the template and the streams returned by the callback: ```ts theme={null} const labelCommentTime = QueryStream.renameKey(["commentTime"]); const renamedInnerKeyLayout = reader .table("comments") .stream("by_creation_time") .pipe(labelCommentTime).keyLayout; reader .table("notes") .stream("by_role", (q) => q.eq("author.role", "admin")) .pipe( QueryStream.flatMap((note) => commentsOn(note).pipe(labelCommentTime), { innerKeyLayout: renamedInnerKeyLayout, }), ); ``` The joined labels are now `["_creationTime", "commentTime"]`; both implicit IDs keep their positions. For nested joins, build a template with the same nested composition and reuse its resulting `keyLayout`. A single-index template cannot describe the extra key positions introduced by that composition. ### Keep outer documents without inner documents By default an outer document whose inner stream is empty contributes nothing. Pass `onEmpty` to `flatMap` to keep it: the document is emitted as `onEmpty(outer)`, and the element type widens to include that placeholder. ```ts theme={null} // Every admin note with its comments, and notes without any as a single // entry saying so. reader .table("notes") .stream("by_role", (q) => q.eq("author.role", "admin")) .pipe( QueryStream.flatMap( (note) => commentsOn(note).pipe( QueryStream.map((comment) => ({ note, comment })), ), { innerKeyLayout, onEmpty: (note) => ({ note, comment: undefined }), }, ), ); ``` The placeholder (`∅n5`, that is `onEmpty(n5)`) takes the position the filtered marker had without `onEmpty`, so it sorts first within its outer document and pagination steps past it like any element: ```text theme={null} admin ╰─ n1 ──────────────────── n4 ──────── n5 ───────┤ [1] [4] [5] of n1 ╰─ c1 ──────── c2 ───────┤ of n4 ╰─ c3 ───────┤ of n5 ╰┤ ╞═ flatMap((note) => commentsOn(note), { innerKeyLayout, onEmpty }) ═╡ joined ╰─ c1 ──────── c2 ──────── c3 ──────── ∅n5 ──────┤ [1,7] [1,8] [4,9] [5,null] ``` The elements are the inner stream's documents or the placeholders, so give both a common shape as above. Outer documents filtered out before the join stay absent. Everything else, including the `innerKeyLayout` check and the direction rules, is unchanged. ### Rename ordering labels `QueryStream.renameKey` replaces a stream's visible labels positionally. It does not sort: the elements and their order are untouched. Use it to merge streams from different indexes or tables that sort by the same kind of value under different field names. Since merged streams must also share a document type, map each side to a common shape first: ```ts theme={null} interface Entry { readonly kind: "note" | "comment"; readonly text: string; } // Notes and comments together, alphabetically by their text. const entries = QueryStream.merge([ reader .table("notes") .stream("by_text") // visible ordering labels: ["text", "_creationTime"] .pipe( QueryStream.map((note): Entry => ({ kind: "note", text: note.text })), ), reader .table("comments") .stream("by_body") // visible ordering labels: ["body", "_creationTime"] .pipe( QueryStream.map((comment): Entry => ({ kind: "comment", text: comment.body, })), QueryStream.renameKey(["text", "_creationTime"]), ), ]); ``` Relabeling changes the visible labels; key values and implicit-ID positions stay the same: ```text theme={null} by_body ╰─ c1 ──────── c2 ──────── c3 ───────┤ [great,7] [meh,8] [nice,9] labels: [body, _creationTime] ╞═ renameKey(["text", "_creationTime"]) ═╡ relabeled ╰─ c1 ──────── c2 ──────── c3 ───────┤ [great,7] [meh,8] [nice,9] labels: [text, _creationTime] ``` With matching key labels and implicit-ID positions, the two streams merge by their key values—here, alphabetically: ```text theme={null} notes ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── n6 ───────────────────────────────────────────┤ relabeled ╰───────────────────────────────────────────────────────────────────────── c1 ──────── c2 ──────── c3 ───────┤ ╞═ merge([notes, relabeled]) ═╡ merged ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── n6 ──────── c1 ──────── c2 ──────── c3 ───────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] [great,7] [meh,8] [nice,9] ``` The replacement labels must have exactly as many entries as the original labels, and the values under each position must be comparable—relabeling doesn't reorder anything. A `flatMap` result relabels by its combined key, outer labels first. ### Reverse `QueryStream.reverse` runs a composed stream in the opposite direction. Its typical use is bidirectional pagination: the stream that loads a feed's later pages, reversed, loads its earlier ones. It uses index scans and seeks rather than collecting and reversing the results, and the result is still a query stream. ```ts theme={null} const oldestFirst = QueryStream.merge([admin, user]); // "asc" const newestFirst = QueryStream.reverse(oldestFirst); // "desc" ``` Reversing the merge reverses each of its inputs and merges them the other way: ```text theme={null} merged ╰─ n1 ──────── n2 ──────── n3 ──────── n4 ──────── n5 ──────── n6 ───────┤ [1] [2] [3] [4] [5] [6] ╞═ reverse ═╡ reversed ╰─ n6 ──────── n5 ──────── n4 ──────── n3 ──────── n2 ──────── n1 ───────┤ [6] [5] [4] [3] [2] [1] ``` Merges, filters and maps, flat-maps (their inner streams included), `distinct`, `renameKey`, and `empty` all reverse. `reverse(distinct(q))` keeps the same first representative of each group and flips only their output order. `distinct(reverse(q))` instead chooses each group's representative from the other direction. Preserving representatives can require extra index reads to discover groups and seek their original first documents. ## Paginating a stream `QueryStream.paginate` returns an effect producing one page of a composed stream. Pass the handler's `paginationOpts` as in the [first example](#a-first-query); the options follow Convex's pagination protocol, with key-based page boundaries and read budgets as described below. ### Options | Field | Meaning | | - | - | | `cursor` | Required exclusive start boundary. `null` starts at the beginning of the stream. | | `numItems` | Requested number of emitted values, excluding filtered documents. Ignored as an item limit when `endCursor` is set. | | `endCursor` | Optional inclusive end boundary. Pins the page to a key range regardless of how many values it contains. Use a cursor returned by a previous page. | | `maximumRowsRead` | Optional budget for physical document reads from underlying `QueryStream` index queries, including filtered documents. | | `maximumBytesRead` | Optional budget charging each document's estimated size on every read, not Convex's exact billed bytes. | ### Result | Field | Meaning and action | | - | - | | `page` | Array of emitted values, possibly empty even when the stream made progress past filtered documents. | | `continueCursor` | String identifying the next page's exclusive start. Pass it back unchanged as `cursor` while `isDone` is false. | | `splitCursor` | Optional interior boundary used to subdivide a page. Pin the left page with `endCursor: splitCursor`; start the right page with `cursor: splitCursor`, retaining the original end boundary. | | `pageStatus` | `"SplitRequired"` when a read budget stops the page at a safe boundary; `"SplitRecommended"` when a pinned page has grown too large or a page scans many rows. Both provide a `splitCursor`; otherwise this field may be absent. Stream-aware clients handle splitting. | | `isDone` | `true` confirms the true end of the stream. Reaching a pinned key does not prove completion. A split recommendation can report `false` even at the end; handle splitting before requesting another page. | For a one-shot read inside a handler, consume the result's `page`. For sequential reads, preserve the returned cursor rather than reconstructing it from the last visible document: ```ts theme={null} import { QueryStream } from "@confect/server"; import * as Effect from "effect/Effect"; Effect.gen(function* () { const query = reader.table("notes").stream("by_text"); const first = yield* QueryStream.paginate(query, { cursor: null, numItems: 2, }); const next = first.isDone ? undefined : yield* QueryStream.paginate(query, { cursor: first.continueCursor, numItems: 2, }); }); ``` For reactive clients, return the whole pagination result from the handler so the client can pin and split pages, rather than returning just `page`. ### Page boundaries Three pages of two over the filtered notes follow each prior cursor. In this direct filtered scan, earlier documents aren't reread; the hidden note is read on the last page, counted, and not returned: ```text theme={null} filtered ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── (n6) ─────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ╞═ paginate({ numItems: 2, cursor: null }) ═╡ page 1 ╰─ n1 ──────── n3 ───────╎ continueCursor: [apple,3] ╞═ paginate({ numItems: 2, cursor: [apple,3] }) ═╡ ╎ n2 ──────── n5 ───────╎ continueCursor: [banana,5] ╞═ paginate({ numItems: 2, cursor: [banana,5] }) ═╡ ╎ n4 ──────── (n6) ─────┤ isDone: true ``` With an `endCursor`, a page covers exactly the range between its two cursors, however many documents that range holds as data changes, which is how reactive clients keep adjacent pages gap-free: ```text theme={null} filtered ╰─ n1 ──────── n3 ──────── n2 ──────── n5 ──────── n4 ──────── (n6) ─────┤ [apple,1] [apple,3] [banana,2] [banana,5] [cherry,4] [date,6] ╞═ paginate({ cursor: [apple,3], endCursor: [banana,5], numItems: 2 }) ═╡ ╎ n2 ──────── n5 ───────╎ exactly this range, however many documents it holds ``` ### Read budgets Cursor bounds are pushed into underlying index queries where possible, but compositions may read more documents than they return, including documents read by an earlier page. Both budgets count filtered documents, distinct-group discovery, repeated seeks, merge prefetch, and outer and inner `QueryStream` reads in joins. They do not track arbitrary I/O inside callbacks, such as a separate database lookup in `mapEffect`. A budget-limited page stops at a safe key position so resuming cannot skip an unfinished group or join. If the budget prevents safe progress or a strictly interior split of a pinned page, `paginate` fails instead of returning a non-advancing cursor or repeating the same split. Increase the budget or change the query to reduce the reads needed for progress. Byte accounting happens after each document is read, so the last document can take the total over `maximumBytesRead`. ### Cursor handling Save cursors returned by `QueryStream.paginate` and pass them back unchanged. Use `cursor: null` for the first page, then its `continueCursor` for the next. Do not construct cursors from documents or decode them into `narrow` bounds. Malformed or incompatible cursors fail with a `ConvexError` whose data is `{ paginationError: "InvalidCursor" }`; Confect's pagination clients restart from the first page. Restart pagination when changing the query's filters, ordering, or other semantics, even if an old cursor is still accepted. Stream cursors expose the boundary's runtime labels and key **values**, including document IDs. They are not opaque or signed: clients can read them and craft cursors for chosen keys within the stream's range. Don't paginate publicly over a sensitive indexed field without pinning it with `eq`. ### On the client React's [`useStreamPaginatedQuery`](/clients/react#usestreampaginatedquery) handles page pinning, splitting, and invalid-cursor resets. Foldkit's [`PaginatedQuery`](/clients/foldkit#paginated-queries) machine also works with stream-paginated queries: it pins each page it navigates away from, so going back reloads the same range. ## The QueryStream type TypeScript normally infers these parameters from `stream` and its composition: ```text theme={null} QueryStream ``` | Parameter | Meaning | | - | - | | `Doc` | Each emitted value: initially a decoded document, or a mapped/joined result. `never` means no values, as with an empty stream. | | `KeyLabels` | A branded readonly tuple of visible ordering labels, such as `["text", "_creationTime"]`. Inferred from the index and updated by `eq`, `flatMap`, and `renameKey`; implicit ID tiebreakers are omitted. | | `OrderDirection` | `"asc"` or `"desc"`. The type parameter defaults to their union; the `stream(...)` method defaults to `"asc"`. | | `Error` (`E`) | Typed failures while reading or transforming values. `never` means no typed failures, not no defects. | | `Requirements` (`R`) | Effect services needed to run the stream. `never` means no outstanding service requirements. | For example, the notes `by_text` query carries branded labels for `["text", "_creationTime"]`, emits decoded notes in `"asc"` order, and can fail with `DocumentDecodeError`. For a type alias, prefer `typeof notes` from an existing stream rather than manually supplying its parameters. The `KeyLabels` parameter requires the brand; a plain string tuple is not a labels type. It retains the database access supplied when it was created, so consuming it doesn't require providing `DatabaseReader` again. `mapEffect` and other effectful operations can add errors and service requirements. A `QueryStream` is also a `Stream`; only `QueryStream` carries the visible-label and direction types and the runtime key layout. This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.