Add bounded live query cursor continuation

This commit is contained in:
Timothy J. Aveni
2026-09-22 17:20:00 -07:00
parent b8571d054b
commit ca83dbd1a7
4 changed files with 51 additions and 8 deletions
+23 -6
View File
@@ -382,12 +382,20 @@ export function compileExecutionPlan(input: WorkspaceRevision): ExecutionPlan {
"Local and replica queries must use the same root contract", "Local and replica queries must use the same root contract",
); );
const traversals = (linked: LinkedQuery) => { const traversals = (linked: LinkedQuery) => {
const paths = new Map<string, boolean>(); const paths = new Map<string, { complete: boolean; requiresComplete: boolean; windows: Set<string> }>();
const entry = (key: string) => {
let value = paths.get(key);
if (!value) {
value = { complete: false, requiresComplete: false, windows: new Set() };
paths.set(key, value);
}
return value;
};
const visit = (selections: LinkedQuery["checked"]["selection"], parents: string[]) => { const visit = (selections: LinkedQuery["checked"]["selection"], parents: string[]) => {
for (const selection of selections) { for (const selection of selections) {
for (const stage of selection.relational?.stages ?? []) { for (const stage of selection.relational?.stages ?? []) {
if (stage.operation.kind === "source") if (stage.operation.kind === "source")
paths.set(JSON.stringify([...parents, stage.operation.memberId]), false); entry(JSON.stringify([...parents, stage.operation.memberId])).requiresComplete = true;
} }
const member = selection.interfaceRevisionId const member = selection.interfaceRevisionId
? interfaces.get(selection.interfaceRevisionId)?.members.find((m) => m.id === selection.memberId) ? interfaces.get(selection.interfaceRevisionId)?.members.find((m) => m.id === selection.memberId)
@@ -397,7 +405,11 @@ export function compileExecutionPlan(input: WorkspaceRevision): ExecutionPlan {
const many = member.cardinality === "many" || member.cardinality === "many-unique"; const many = member.cardinality === "many" || member.cardinality === "many-unique";
const complete = !selection.predicate && (!many || !!selection.arguments.all); const complete = !selection.predicate && (!many || !!selection.arguments.all);
const key = JSON.stringify(path); const key = JSON.stringify(path);
paths.set(key, complete || paths.get(key) === true); const value = entry(key);
value.complete ||= complete;
if (selection.arguments.first)
value.windows.add(canonicalJson({ arguments: selection.arguments, conditions: selection.conditions }));
else value.requiresComplete = true;
} }
visit(selection.selection, path); visit(selection.selection, path);
} }
@@ -406,18 +418,23 @@ export function compileExecutionPlan(input: WorkspaceRevision): ExecutionPlan {
return paths; return paths;
}; };
const admittedTraversals = traversals(source.query); const admittedTraversals = traversals(source.query);
for (const traversal of traversals(query.query).keys()) { for (const [traversal, needed] of traversals(query.query)) {
if (!admittedTraversals.has(traversal)) if (!admittedTraversals.has(traversal))
throw new ExecutionPlanError( throw new ExecutionPlanError(
"REPLICA_TRAVERSAL_MISSING", "REPLICA_TRAVERSAL_MISSING",
query.id, query.id,
`Replica source does not admit relationship path ${traversal}`, `Replica source does not admit relationship path ${traversal}`,
); );
if (!admittedTraversals.get(traversal)) if (
!admittedTraversals.get(traversal)!.complete &&
(needed.requiresComplete ||
!needed.windows.size ||
[...needed.windows].some((window) => !admittedTraversals.get(traversal)!.windows.has(window)))
)
throw new ExecutionPlanError( throw new ExecutionPlanError(
"REPLICA_RELATION_INCOMPLETE", "REPLICA_RELATION_INCOMPLETE",
query.id, query.id,
`Local relationship ${traversal} needs an unfiltered bounded all selection in its replica source; a filtered/windowed source cannot supply unseen matches`, `Local relationship ${traversal} needs either complete source candidates or the identical confirmed first window; a window cannot supply unseen matches`,
); );
} }
// The admission executor follows these canonical dependencies on the admitted roots. // The admission executor follows these canonical dependencies on the admitted roots.
-2
View File
@@ -22,8 +22,6 @@ export function unsupportedQueryOperator(query: CheckedQuery): string | undefine
selections.some((selection) => selection.conditions.length > 0 || conditional(selection.selection)); selections.some((selection) => selection.conditions.length > 0 || conditional(selection.selection));
const walk = (selections: readonly QuerySelection[]): string | undefined => { const walk = (selections: readonly QuerySelection[]): string | undefined => {
for (const selection of selections) { for (const selection of selections) {
const after = selection.arguments.after;
if (after && !(after.kind === "literal" && after.value === null)) return "cursor continuation";
if (selection.predicate) { if (selection.predicate) {
const reason = predicate(selection.predicate); const reason = predicate(selection.predicate);
if (reason) return reason; if (reason) return reason;
+8
View File
@@ -112,6 +112,14 @@ test("typed reference operands and scalar membership tests reject unbounded lite
test("new runtime operator matrix accepts direct count and rejects unsupported relational plans", async () => { test("new runtime operator matrix accepts direct count and rejects unsupported relational plans", async () => {
const { unsupportedQueryOperator } = await import("../src/execution-world/query-operators.js"); const { unsupportedQueryOperator } = await import("../src/execution-world/query-operators.js");
assert.equal(unsupportedQueryOperator(await compile("aggregate { count }")), undefined); assert.equal(unsupportedQueryOperator(await compile("aggregate { count }")), undefined);
assert.equal(
unsupportedQueryOperator(
await compileAggregationDocument(
`query Report($after: Cursor) {root {tasks(first: 30, after: $after, orderBy: [{cost: ASC}]) {entries {cursor node {cost}} pageInfo {hasNextPage endCursor}}}}`,
),
),
undefined,
);
for (const expression of [ for (const expression of [
"aggregate { sum { estimatedHours } }", "aggregate { sum { estimatedHours } }",
"filter(where: {estimatedHours: {gt: 0}}) { aggregate { count } }", "filter(where: {estimatedHours: {gt: 0}}) { aggregate { count } }",
+20
View File
@@ -0,0 +1,20 @@
import assert from "node:assert/strict";
import { test } from "node:test";
import { loadReaderWorkspace } from "./reader-fixture.js";
import { compileExecutionPlan } from "../src/execution-world/index.js";
test("local relationship pages may reuse exactly the confirmed source window", async () => {
const workspace = await loadReaderWorkspace((name, text) => {
if (name === "queries/replica-inputs.graphql")
return text.replace("highlights(all: 1000)", "highlights(first: 10)");
if (name === "queries/reader-screen.graphql")
return text
.replace("ReaderScreen($color: String!)", "ReaderScreen")
.replace(
"highlights(all: 1000, where: {color: {eq: $color}}, orderBy: [{range: {start: ASC}}])",
"highlights(first: 10)",
)
.replace("_camino { relations { highlights { aggregate { count } } } }", "");
return text;
});
assert.doesNotThrow(() => compileExecutionPlan(workspace));
});