composable streams of query results #446

ldanilek · 2025-02-08T00:51:39Z

as described in the README, add some helpers for merging and filtering streams of query results.

This extracts the implementations of paginator and streamQuery into new helpers, that allow better composition and more patterns.

First there's reflect which allows you to construct queries with the normal syntax reflect(ctx.db, schema).query(table).withIndex(index, indexRange).order("desc") and then get their internal details, e.g. which index it's looking at. -- the naming is based on https://golangbot.com/reflection/ but it's mostly an internal library.

Then once you have a reflectable query, it can be used as a "stream", which is an async iterable of documents ordered by an index. For this reason, stream(ctx.db, schema) is another way of writing reflect(ctx.db, schema).

Once you have a stream, you can merge them with mergeStreams or filter them with filterStream, generating more streams.

See the README for more details and motivating examples.

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

ianmacartney

I think I get it, but the language around queries, query streams, index streams, and merging is a bit murky even for me after skimming the code. Can you give some bullets of how each term relates to one another in the readme?

I'm starting to wonder if all the query helpers could get bundled into a @convex-dev/query-utils library that you own. It might be a good organization, and more "official-feeling" than helpers. I've already pulled migrations and rate limiting into components, and maybe I should pull validators into validator-utils or something, then have helpers be for developing APIs before crystalizing them into separate packages... one downside of course is discoverability.

ianmacartney · 2025-02-10T19:26:43Z

packages/convex-helpers/README.md

+
+With the `stream` helper, you can construct a stream with the same syntax as
+you would use `DatabaseReader`. Once you have a stream, you can compose them
+to get more streams (still ordered by the same index) with `mergeStreams`, and


what does "get more streams" mean here? reading this first, I'm not sure as a developer (yet)

packages/convex-helpers/README.md

ianmacartney · 2025-02-10T19:30:06Z

packages/convex-helpers/README.md

+    const authorStreams = authors.map(author =>
+      stream(ctx.db, schema).query("messages").withIndex("by_author", q => q.eq("author", author))
+    );
+    const allAuthorsStream = mergeStreams(authorStreams);


I'm not sure what "mergeStreams" means here. each will be concatenated? I'll try to think of other names as I go, but maybe we drop some comments in here that walk through what each step does.

packages/convex-helpers/README.md

ianmacartney · 2025-02-10T19:42:38Z

packages/convex-helpers/server/compare.ts

+  }
+  // Otherwise, it's an POJO.
+  const keys = Object.keys(v).sort();
+  const pojo: Value[] = keys.map((k) => [k, v[k]!]);


Suggested change

const pojo: Value[] = keys.map((k) => [k, v[k]!]);

const pojo: (Value | undefined)[] = keys.map((k) => [k, v[k]]);

i don't think they're the same thing

ianmacartney · 2025-02-10T19:43:38Z

packages/convex-helpers/server/compare.ts

@@ -0,0 +1,86 @@
+import { Value } from "convex/values";


FYI this will make an entrypoint at convex-helpers/server/compare

i thought to do that i had to make an export in package.json
(i don't mind making this an entrypoint, but it shouldn't be necessary for now)

ianmacartney · 2025-02-10T19:52:07Z

packages/convex-helpers/server/stream.ts

+  }
+  lt(field: string, value: Value) {
+    if (!this.canUpperBound(field)) {
+      throw new Error(`Cannot use lt on field '${field}'`);


what's an example of this?

this is like if you have an index on ["a", "b"] and you do q=>q.lt("b", 1), or q=>q.lt("a", 1).lt("a", 1) or q=>q.eq("a", 1).lt("a", 1). there are tests :)

ianmacartney · 2025-02-10T19:53:45Z

packages/convex-helpers/server/stream.ts

+}
+
+/**
+ * Merge multiple streams, provided in any order, into a single stream.


Document what order the data will be returned in, and add an example like you do for concatStream

ianmacartney · 2025-02-10T20:02:48Z

packages/convex-helpers/server/stream.ts

+/**
+ * Apply a filter to a stream.
+ * 
+ * Watch out for sparse filters, as they may read unbounded amounts of data.


Could we pass in a rowReadLimit or something so folks can bound - like "give me 10 non-deleted users' messages, but cap the search at 1k users. Would that end up with surprising results later? where pagination might think it's done? Maybe throw a catch-able error with the results so far? Or do you have other ideas for how to gracefully fail there? Maybe it's "usually" totally fine, but there was a burst of user deletions, and now a query is stuck in a failing state?

ianmacartney · 2025-02-10T20:07:36Z

tbh I hit some review fatigue when I got to the huge stream.ts and compare.ts logic. I can look more closely at whatever feels most important to you, or go deep into it once I have self-hosting things under control

Co-authored-by: Ian Macartney <ian@convex.dev>

…convex-helpers into lee/paginator-stream

ldanilek · 2025-02-14T00:45:04Z

packages/convex-helpers/server/stream.ts

+  DataModel extends GenericDataModel,
+  T extends TableNamesInDataModel<DataModel>,
+> {
+  iterWithKeys(): AsyncIterable<[DocumentByName<DataModel, T> | null, IndexKey]>;


TODO: IndexKey can contain undefined, make sure we're not serializing it as a convex value

ldanilek · 2025-02-14T01:01:26Z

packages/convex-helpers/README.md

+  - e.g. `stream(ctx.db, schema).query("messages").withIndex("by_author", (q) => q.eq("author", "user1"))`
+- `mergeStreams` combines multiple streams into a new stream, ordered by the same index.
+- `filterStream` filters out documents from a stream based on a TypeScript predicate.
+- `queryStream` converts a stream into a query, so you can call `.first()`, `.collect()`, `.paginate()`, etc.


ldanilek · 2025-02-14T01:01:59Z

packages/convex-helpers/README.md

+## Composable streams of query results
+
+These are helper functions for constructing and composing streams of query results.
+


add a motivating example here

ldanilek · 2025-02-14T01:07:56Z

packages/convex-helpers/README.md

+      stream(ctx.db, schema)
+        .query("messages")
+        .withIndex("by_author", (q) =>
+          q.eq("author", author).eq("unread", unread),


does this order by author,unread still? or by _creationTime?

ldanilek · 2025-02-14T01:08:42Z

packages/convex-helpers/README.md

+    );
+    // Merge the two streams into a single stream of all messages authored by
+    // `args.author`, ordered by _creationTime descending.
+    const allMessagesByCreationTime = mergeStreams(...messagesForUnreadStatus);


separate cursor for each mergeStream?

ldanilek added 8 commits February 7, 2025 11:54

streams

7955f1f

tests

b7988ab

tests

c5c388e

readme for stream

f1f374b

.

1994737

.

0145b26

add package

81cccd8

examples

e832bbb

ldanilek requested a review from ianmacartney February 8, 2025 00:51

lint

e015d2e

ianmacartney reviewed Feb 10, 2025

View reviewed changes

ldanilek and others added 10 commits February 10, 2025 14:06

Update packages/convex-helpers/README.md

7a2ae59

Co-authored-by: Ian Macartney <ian@convex.dev>

Update packages/convex-helpers/README.md

4ba07ac

Co-authored-by: Ian Macartney <ian@convex.dev>

Update packages/convex-helpers/README.md

f57be08

Co-authored-by: Ian Macartney <ian@convex.dev>

Update packages/convex-helpers/README.md

4f9b619

Co-authored-by: Ian Macartney <ian@convex.dev>

Merge branch 'main' into lee/paginator-stream

b3f399d

merge

1daf64b

Merge branch 'lee/paginator-stream' of https://github.com/get-convex/…

9309324

…convex-helpers into lee/paginator-stream

document mergeStreams better

29cdbc6

readme

d812680

implement maximumRowsRead

caefef9

ldanilek commented Feb 14, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

composable streams of query results #446

composable streams of query results #446

ldanilek commented Feb 8, 2025

ianmacartney left a comment

ianmacartney Feb 10, 2025

ianmacartney Feb 10, 2025

ianmacartney Feb 10, 2025

ldanilek Feb 12, 2025

ianmacartney Feb 10, 2025

ldanilek Feb 12, 2025

ianmacartney Feb 10, 2025

ldanilek Feb 12, 2025

ianmacartney Feb 10, 2025

ianmacartney Feb 10, 2025

ianmacartney commented Feb 10, 2025

ldanilek Feb 14, 2025

ldanilek Feb 14, 2025

ldanilek Feb 14, 2025

ldanilek Feb 14, 2025

ldanilek Feb 14, 2025

	const pojo: Value[] = keys.map((k) => [k, v[k]!]);
	const pojo: (Value \| undefined)[] = keys.map((k) => [k, v[k]]);

		## Composable streams of query results

		These are helper functions for constructing and composing streams of query results.

composable streams of query results #446

Are you sure you want to change the base?

composable streams of query results #446

Conversation

ldanilek commented Feb 8, 2025

ianmacartney left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ianmacartney commented Feb 10, 2025

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment