feat: Minimal support for deferred queries (#52)

packages/backend/src/execution/composeExecutionResults.ts

@@ -1,4 +1,10 @@
-import { AnyDB, AnyTable, QueryResult } from '@synthql/queries';
+import {
+    AnyDB,
+    AnyQuery,
+    AnyTable,
+    DeferredResult,
+    QueryResult,
+} from '@synthql/queries';
 import { applyCardinality } from '../query/applyCardinality';
 import { assertHasKey } from '../util/asserts/assertHasKey';
 import { setIn } from '../util/tree/setIn';
@@ -13,9 +19,9 @@ export function composeExecutionResults(
         composeExecutionResultsRecursively(node, queryResult);
     }
 
-    return applyCardinality(
+    return applyCardinalityAndDeferredResult(
         queryResult,
-        tree.root.inputQuery.cardinality ?? 'many',
+        tree.root.inputQuery,
     ) as QueryResult<AnyDB, AnyTable>;
 }
 
@@ -40,13 +46,30 @@ function composeExecutionResultsRecursively(
             return true;
         };
         const rows = result.filter((row) => predicate(row));
-        return applyCardinality(rows, inputQuery.cardinality ?? 'many', {
-            query: inputQuery,
-            row: rows,
-        });
+
+        return applyCardinalityAndDeferredResult(rows, inputQuery);
     });
 
     for (const child of node.children) {
         composeExecutionResultsRecursively(child, queryResult);
     }
 }
+
+function applyDeferredQueryResult<T>(
+    result: T,
+    defer: boolean = false,
+): DeferredResult<T> | T {
+    if (!defer) {
+        return result;
+    }
+    return { status: 'done', data: result };
+}
+
+function applyCardinalityAndDeferredResult<T>(rows: T[], inputQuery: AnyQuery) {
+    const withCardinality = applyCardinality(rows, inputQuery.cardinality ?? 'many', {
+        query: inputQuery,
+        row: rows,
+    });
+
+    return applyDeferredQueryResult(withCardinality, inputQuery.lazy);
+}

packages/backend/src/execution/execute.ts

@@ -3,6 +3,7 @@ import { composeExecutionResults } from './composeExecutionResults';
 import { createExecutionPlan } from './planning/createExecutionPlan';
 import { executePlan } from './execution/executePlan';
 import { QueryExecutor } from './types';
+import { shouldYield } from './execution/shouldYield';
 
 export interface ExecuteProps {
     executors: Array<QueryExecutor>;
@@ -40,6 +41,8 @@ export async function* execute<DB, TQuery extends Query<DB>>(
     const plan = createExecutionPlan(query, props);
 
     for await (const resultTree of executePlan(plan, props)) {
-        yield composeExecutionResults(resultTree);
+        if (shouldYield(resultTree)) {
+            yield composeExecutionResults(resultTree);
+        }
     }
 }

packages/backend/src/execution/execution/executePlan.ts

@@ -52,6 +52,7 @@ export function executePlan(
                 filters: createFilters(planNode),
                 inputQuery: planNode.inputQuery,
                 result: rows,
+                planNode,
                 children: [],
             };
         },

packages/backend/src/execution/execution/shouldYield.ts

@@ -0,0 +1,25 @@
+import { ExecResultNode, ExecResultTree } from '../types';
+
+export function shouldYield(tree: ExecResultTree): boolean {
+    return shouldYieldNode(tree.root);
+}
+
+function shouldYieldNode(node: ExecResultNode): boolean {
+    const plannedChildren = node.planNode.children;
+    const executedChildren = node.children;
+
+    const allChildrenExecuted =
+        plannedChildren.length === executedChildren.length;
+
+    if (allChildrenExecuted) {
+        return true;
+    }
+
+    for (const child of executedChildren) {
+        if (!shouldYieldNode(child)) {
+            return false;
+        }
+    }
+
+    return false;
+}

packages/backend/src/execution/executors/PgExecutor/PgExecutor.ts

@@ -131,7 +131,10 @@ export class PgExecutor implements QueryExecutor<PgQueryResult> {
 
         try {
             if (this.props.logging) {
-                console.log(format(sql, { language: 'postgresql' }));
+                console.log(
+                    `[${query.name}]:`,
+                    format(sql, { language: 'postgresql' }),
+                );
             }
 
             const queryResult = await client.query(sql, params);

packages/backend/src/execution/types.d.ts

@@ -158,6 +158,7 @@ export interface ExecResultNode {
      * The original query that was executed.
      */
     inputQuery: AnyQuery;
+    planNode: ExecutionPlanNode;
     children: ExecResultNode[];
 }
 

packages/backend/src/query/iterateQuery.test.ts

@@ -194,13 +194,15 @@ describe('iterateQuery', () => {
                 where: {
                     film_id: 1,
                 },
+                cardinality: 'many',
             },
             expected: [
                 {
                     query: {
                         from: 'film',
                         select: {},
                         where: { film_id: 1 },
+                        cardinality: 'many',
                     },
                     insertionPath: [],
                 },

packages/backend/src/tests/e2e/nxm.test.ts

@@ -41,7 +41,6 @@ describe('n x m', () => {
         executors: [
             new PgExecutor({
                 defaultSchema: 'public',
-                logging: true,
                 pool,
             }),
         ],

packages/backend/src/tests/e2e/payments.test.ts

@@ -17,7 +17,6 @@ describe('e2e', () => {
     test.skip(describeQuery(q), async () => {
         const pgExecutor = new PgExecutor({
             defaultSchema: 'public',
-            logging: true,
             pool,
         });
 

packages/backend/src/tests/e2e/select.test.ts

@@ -164,4 +164,43 @@ describe('select', () => {
             Array.from(expected).sort(comparator),
         );
     });
+
+    test('defer', async () => {
+        const limit = 20;
+        const language = from('language')
+            .columns('name')
+            .where({ language_id: col('film.language_id') })
+            .defer()
+            .first();
+
+        const query = from('film')
+            .columns('title')
+            .include({ language })
+            .limit(limit)
+            .all();
+
+        const languageNonDeferred = from('language')
+            .columns('name')
+            .where({ language_id: col('film.language_id') })
+            .first();
+
+        const queryNonDeferred = from('film')
+            .columns('title')
+            .include({ language: languageNonDeferred })
+            .limit(limit)
+            .all();
+
+        const result = await run(query);
+        const resultNonDeferred = await run(queryNonDeferred);
+
+        expect(
+            result.map(
+                (r) =>
+                    r.language.status === 'done' && {
+                        ...r,
+                        language: r.language.data,
+                    },
+            ),
+        ).toEqual(resultNonDeferred);
+    });
 });

packages/backend/src/tests/e2e/store-with-customers.test.ts

@@ -36,7 +36,6 @@ describe('e2e', () => {
 
     const pgExecutor = new PgExecutor({
         defaultSchema: 'public',
-        logging: true,
         pool,
     });
     const execProps = {

packages/backend/src/tests/e2e/store-with-films.test.ts

@@ -40,7 +40,6 @@ describe('e2e', () => {
 
     const pgExecutor = new PgExecutor({
         defaultSchema: 'public',
-        logging: true,
         pool,
     });
     const execProps = {