add pagination options support for maximumRowsRead

Author: ianmacartney

convex/pagination.test.ts

@@ -2,6 +2,7 @@ import { expect, test } from "vitest";
 import { convexTest } from "../index";
 import { api } from "./_generated/api";
 import schema from "./schema";
+import type { PaginationResult } from "convex/server";
 
 test("paginate", async () => {
   const t = convexTest(schema);
@@ -55,3 +56,139 @@ test("paginate", async () => {
   expect(page3).toMatchObject([]);
   expect(isDone3).toEqual(true);
 });
+
+test("paginate with maximumRowsRead", async () => {
+  const t = convexTest(schema);
+  await t.run(async (ctx) => {
+    for (let i = 0; i < 10; i++) {
+      await ctx.db.insert("messages", {
+        author: "sarah",
+        body: `msg${i}`,
+      });
+    }
+  });
+
+  // With maximumRowsRead=3, we should get at most 3 docs and SplitRequired
+  const result = (await t.query(api.pagination.listAll, {
+    paginationOptions: {
+      cursor: null,
+      numItems: 10,
+      maximumRowsRead: 3,
+    },
+  })) as PaginationResult<any>;
+
+  expect(result.page.length).toBeLessThanOrEqual(3);
+  expect(result.pageStatus).toEqual("SplitRequired");
+  expect(result.splitCursor).toBeTruthy();
+  expect(result.isDone).toEqual(false);
+
+  // Continue from the continueCursor
+  const result2 = (await t.query(api.pagination.listAll, {
+    paginationOptions: {
+      cursor: result.continueCursor,
+      numItems: 10,
+    },
+  })) as PaginationResult<any>;
+
+  // Combined pages should cover all 10 docs
+  expect(result.page.length + result2.page.length).toEqual(10);
+  expect(result2.isDone).toEqual(true);
+});
+
+test("paginate with maximumBytesRead", async () => {
+  const t = convexTest(schema);
+  await t.run(async (ctx) => {
+    for (let i = 0; i < 5; i++) {
+      await ctx.db.insert("messages", {
+        author: "sarah",
+        body: "x".repeat(100),
+      });
+    }
+  });
+
+  // Use a very small byte limit to force early termination
+  const result = (await t.query(api.pagination.listAll, {
+    paginationOptions: {
+      cursor: null,
+      numItems: 10,
+      maximumBytesRead: 1, // Very small, should stop after first doc
+    },
+  })) as PaginationResult<any>;
+
+  expect(result.page.length).toBeLessThanOrEqual(1);
+  expect(result.pageStatus).toEqual("SplitRequired");
+  expect(result.isDone).toEqual(false);
+});
+
+test("paginate with filter and maximumRowsRead", async () => {
+  const t = convexTest(schema);
+  await t.run(async (ctx) => {
+    // Insert many docs, only some match the filter
+    for (let i = 0; i < 10; i++) {
+      await ctx.db.insert("messages", {
+        author: i % 3 === 0 ? "sarah" : "michal",
+        body: `msg${i}`,
+      });
+    }
+  });
+
+  // Filter for sarah (4 docs: 0, 3, 6, 9), but maximumRowsRead=5
+  // means we scan at most 5 rows from the pipeline
+  const result = (await t.query(api.pagination.list, {
+    author: "sarah",
+    paginationOptions: {
+      cursor: null,
+      numItems: 10,
+      maximumRowsRead: 5,
+    },
+  })) as PaginationResult<any>;
+
+  // Should have scanned 5 rows, getting some sarah docs but not all
+  expect(result.pageStatus).toEqual("SplitRequired");
+  expect(result.isDone).toEqual(false);
+});
+
+test("paginate with endCursor", async () => {
+  const t = convexTest(schema);
+  await t.run(async (ctx) => {
+    for (let i = 0; i < 5; i++) {
+      await ctx.db.insert("messages", {
+        author: "sarah",
+        body: `msg${i}`,
+      });
+    }
+  });
+
+  // First get a page to get a cursor
+  const result1 = (await t.query(api.pagination.listAll, {
+    paginationOptions: {
+      cursor: null,
+      numItems: 2,
+    },
+  })) as PaginationResult<any>;
+
+  expect(result1.page.length).toEqual(2);
+  expect(result1.isDone).toEqual(false);
+
+  // Now get the next page
+  const result2 = (await t.query(api.pagination.listAll, {
+    paginationOptions: {
+      cursor: result1.continueCursor,
+      numItems: 2,
+    },
+  })) as PaginationResult<any>;
+
+  expect(result2.page.length).toEqual(2);
+
+  // Use endCursor to re-fetch the second page bounded
+  const result3 = (await t.query(api.pagination.listAll, {
+    paginationOptions: {
+      cursor: result1.continueCursor,
+      numItems: 100, // Large numItems, but bounded by endCursor
+      endCursor: result2.continueCursor,
+    },
+  })) as PaginationResult<any>;
+
+  // Should get the same docs as result2
+  expect(result3.page.length).toEqual(result2.page.length);
+});

convex/pagination.ts

@@ -16,3 +16,12 @@ export const list = query({
       .paginate(args.paginationOptions);
   },
 });
+
+export const listAll = query({
+  args: {
+    paginationOptions: paginationOptsValidator,
+  },
+  handler: async (ctx, args) => {
+    return await ctx.db.query("messages").paginate(args.paginationOptions);
+  },
+});

index.ts

@@ -33,6 +33,7 @@ import {
   JSONValue,
   Value,
   convexToJson,
+  getDocumentSize,
   jsonToConvex,
 } from "convex/values";
 import { compareValues } from "./compare.js";
@@ -421,41 +422,108 @@ class DatabaseFake {
   paginate({
     query,
     cursor,
+    endCursor,
     pageSize,
+    maximumRowsRead,
+    maximumBytesRead,
   }: {
     query: SerializedQuery;
     cursor: string | null;
+    endCursor?: string | null;
     pageSize: number;
+    maximumRowsRead?: number | null;
+    maximumBytesRead?: number | null;
   }) {
-    const queryId = this.startQuery(query);
-    const page = [];
+    const { sortedDocs, filterFn } = this._resolveQuerySource(query);
+
+    const page: GenericDocument[] = [];
     let isInPage = cursor === null;
     let isDone = false;
-    let continueCursor = null;
-    for (;;) {
-      const { value, done } = this.queryNext(queryId);
-      if (done) {
-        isDone = true;
-        // We have reached the end of the query. Return a cursor that indicates
-        // "end query", which we can do with any string that isn't a valid _id.
-        continueCursor = "_end_cursor";
-        break;
-      }
-      if (isInPage) {
-        page.push(value);
-        if (page.length >= pageSize) {
-          continueCursor = value!._id;
+    let continueCursor: string | null = null;
+    let splitCursor: string | null = null;
+    let pageStatus: "SplitRecommended" | "SplitRequired" | null = null;
+    let rowsRead = 0;
+    let bytesRead = 0;
+    const readDocIds: string[] = [];
+
+    for (const doc of sortedDocs) {
+      if (!isInPage) {
+        if ((doc._id as string) === cursor) {
+          isInPage = true;
+        }
+        continue;
+      }
+
+      // endCursor: stop when we reach this document (inclusive boundary)
+      if (endCursor && endCursor !== "_end_cursor") {
+        if ((doc._id as string) === endCursor) {
+          // Include this doc if it passes filter, then stop
+          rowsRead += 1;
+          bytesRead += getDocumentSize(doc);
+          readDocIds.push(doc._id as string);
+          if (filterFn(doc)) {
+            page.push(doc);
+          }
+          continueCursor = doc._id as string;
           break;
         }
       }
-      if (value!._id === cursor) {
-        isInPage = true;
+
+      rowsRead += 1;
+      bytesRead += getDocumentSize(doc);
+      readDocIds.push(doc._id as string);
+
+      // Check bandwidth limits
+      let hitLimit = false;
+      if (maximumRowsRead && rowsRead >= maximumRowsRead) {
+        hitLimit = true;
+      }
+      if (maximumBytesRead && bytesRead >= maximumBytesRead) {
+        hitLimit = true;
+      }
+
+      if (filterFn(doc)) {
+        page.push(doc);
+      }
+
+      if (hitLimit) {
+        pageStatus = "SplitRequired";
+        continueCursor = doc._id as string;
+        break;
+      }
+
+      if (!endCursor && page.length >= pageSize) {
+        continueCursor = doc._id as string;
+        break;
       }
     }
+
+    if (continueCursor === null) {
+      isDone = true;
+      continueCursor = "_end_cursor";
+    }
+
+    // Compute splitCursor at midpoint when limits are hit
+    if (pageStatus === "SplitRequired" && readDocIds.length >= 2) {
+      const midIdx = Math.floor((readDocIds.length - 1) / 2);
+      splitCursor = readDocIds[midIdx];
+    } else if (
+      pageStatus === null &&
+      rowsRead > pageSize + 1 &&
+      readDocIds.length >= 2
+    ) {
+      // Recommend split when we had to scan significantly more rows than pageSize
+      pageStatus = "SplitRecommended";
+      const midIdx = Math.floor((readDocIds.length - 1) / 2);
+      splitCursor = readDocIds[midIdx];
+    }
+
     return {
       page,
       isDone,
       continueCursor,
+      splitCursor,
+      pageStatus,
     };
   }
 
@@ -478,7 +546,18 @@ class DatabaseFake {
     }
   }
 
-  private _evaluateQuery(query: SerializedQuery): Array<GenericDocument> {
+  /**
+   * Resolves the query source: loads documents matching the source (table scan,
+   * index range, or search), sorts them, and returns along with the compiled
+   * operator filter function. This separates "rows entering the pipeline"
+   * (sortedDocs) from "rows exiting" (after filterFn), which is needed for
+   * maximumRowsRead tracking in paginate().
+   */
+  private _resolveQuerySource(query: SerializedQuery): {
+    sortedDocs: GenericDocument[];
+    filterFn: (doc: GenericDocument) => boolean;
+    limit: number | null;
+  } {
     const source = query.source;
     let results: GenericDocument[] = [];
     let fieldPathsToSortBy: string[];
@@ -543,6 +622,7 @@ class DatabaseFake {
         break;
       }
     }
+
     const filters = query.operators
       .filter(
         (operator): operator is { filter: FilterJson } => "filter" in operator,
@@ -554,8 +634,6 @@ class DatabaseFake {
         (operator): operator is { limit: number } => "limit" in operator,
       )[0] ?? null;
 
-    results = results.filter((v) => filters.every((f) => evaluateFilter(v, f)));
-
     results.sort((a, b) => {
       const orderMultiplier = order === "asc" ? 1 : -1;
       let v = 0;
@@ -568,10 +646,22 @@ class DatabaseFake {
       return v * orderMultiplier;
     });
 
+    const filterFn = (doc: GenericDocument) =>
+      filters.every((f) => evaluateFilter(doc, f));
+
+    return {
+      sortedDocs: results,
+      filterFn,
+      limit: limit?.limit ?? null,
+    };
+  }
+
+  private _evaluateQuery(query: SerializedQuery): Array<GenericDocument> {
+    const { sortedDocs, filterFn, limit } = this._resolveQuerySource(query);
+    let results = sortedDocs.filter(filterFn);
     if (limit !== null) {
-      return results.slice(0, limit.limit);
+      results = results.slice(0, limit);
     }
-
     return results;
   }
 
@@ -1128,13 +1218,32 @@ function asyncSyscallImpl() {
         return JSON.stringify(convexToJson({ value, done }));
       }
       case "1.0/queryPage": {
-        const { query, cursor, pageSize } = args;
-        const { page, isDone, continueCursor } = db.paginate({
+        const {
           query,
           cursor,
+          endCursor,
           pageSize,
-        });
-        return JSON.stringify(convexToJson({ page, isDone, continueCursor }));
+          maximumRowsRead,
+          maximumBytesRead,
+        } = args;
+        const { page, isDone, continueCursor, splitCursor, pageStatus } =
+          db.paginate({
+            query,
+            cursor,
+            endCursor,
+            pageSize,
+            maximumRowsRead,
+            maximumBytesRead,
+          });
+        return JSON.stringify(
+          convexToJson({
+            page,
+            isDone,
+            continueCursor,
+            splitCursor,
+            pageStatus,
+          }),
+        );
       }
       case "1.0/insert": {
         const _id = db.insert(args.table, jsonToConvex(args.value));