feat: add runtime type checking

.vscode/settings.json

@@ -0,0 +1,18 @@
+{
+    "typescript.tsdk": "node_modules/typescript/lib",
+    "typescript.tsserver.experimental.enableProjectDiagnostics": true,
+    "editor.formatOnSave": true,
+    "editor.codeActionsOnSave": {
+        "source.organizeImports": "always"
+    },
+    "editor.defaultFormatter": "esbenp.prettier-vscode",
+    "[typescript]": {
+        "editor.defaultFormatter": "esbenp.prettier-vscode"
+    },
+    "[typescriptreact]": {
+        "editor.defaultFormatter": "esbenp.prettier-vscode"
+    },
+    "[json]": {
+        "editor.defaultFormatter": "esbenp.prettier-vscode"
+    }
+}

packages/backend/package.json

@@ -24,7 +24,7 @@
         "typecheck": "tsc",
         "checks": "yarn vitest run && yarn tsc",
         "publish:minor": "yarn publish --access public --no-git-tag-version",
-        "format": "yarn prettier --config ../../prettier.config.js --write ./src/",
+        "format": "yarn prettier --config ../../.prettierrc.js --write ./src/",
         "benchmarks": "yarn vitest run src/tests/benchmarks/bench.test.ts"
     },
     "dependencies": {

packages/backend/src/QueryEngine.ts

@@ -1,14 +1,14 @@
-import { Pool } from 'pg';
 import { Query, QueryResult, Table } from '@synthql/queries';
-import { composeQuery } from './execution/executors/PgExecutor/composeQuery';
+import { Pool } from 'pg';
 import { QueryPlan, collectLast } from '.';
 import { QueryProvider } from './QueryProvider';
+import { SynthqlError } from './SynthqlError';
 import { execute } from './execution/execute';
-import { QueryExecutor } from './execution/types';
-import { QueryProviderExecutor } from './execution/executors/QueryProviderExecutor';
 import { PgExecutor } from './execution/executors/PgExecutor';
+import { composeQuery } from './execution/executors/PgExecutor/composeQuery';
+import { QueryProviderExecutor } from './execution/executors/QueryProviderExecutor';
+import { QueryExecutor } from './execution/types';
 import { generateLast } from './util/generators/generateLast';
-import { SynthqlError } from './SynthqlError';
 
 export interface QueryEngineProps<DB> {
     /**
@@ -65,13 +65,21 @@ export interface QueryEngineProps<DB> {
      * Whether to log SQL statements or not.
      */
     logging?: boolean;
+
+    /**
+     * The rate at which rows are sampled for runtime type validation.
+     *
+     * Defaults to 0, meaning no validation will be performed.
+     */
+    runtimeValidationSampleRate?: number;
 }
 
 export class QueryEngine<DB> {
     private pool: Pool;
     private schema: string;
     private prependSql?: string;
     private executors: Array<QueryExecutor> = [];
+    private runtimeValidationSampleRate: number;
 
     constructor(config: QueryEngineProps<DB>) {
         this.schema = config.schema ?? 'public';
@@ -94,9 +102,11 @@ export class QueryEngine<DB> {
                 logging: config.logging,
             }),
         ];
+        this.runtimeValidationSampleRate =
+            config.runtimeValidationSampleRate ?? 0;
     }
 
-    execute<TTable extends Table<DB>, TQuery extends Query<DB, TTable>>(
+    execute<TQuery extends Query>(
         query: TQuery,
         opts?: {
             /**
@@ -112,11 +122,12 @@ export class QueryEngine<DB> {
              */
             returnLastOnly?: boolean;
         },
-    ): AsyncGenerator<QueryResult<DB, TQuery>> {
-        const gen = execute<DB, TQuery>(query, {
+    ): AsyncGenerator<QueryResult<TQuery>> {
+        const gen = execute<TQuery>(query, {
             executors: this.executors,
             defaultSchema: opts?.schema ?? this.schema,
             prependSql: this.prependSql,
+            runtimeValidationSampleRate: this.runtimeValidationSampleRate,
         });
 
         if (opts?.returnLastOnly) {
@@ -126,10 +137,7 @@ export class QueryEngine<DB> {
         return gen;
     }
 
-    async executeAndWait<
-        TTable extends Table<DB>,
-        TQuery extends Query<DB, TTable>,
-    >(
+    async executeAndWait<TTable extends Table<DB>, TQuery extends Query>(
         query: TQuery,
         opts?: {
             /**
@@ -140,10 +148,10 @@ export class QueryEngine<DB> {
              */
             schema?: string;
         },
-    ): Promise<QueryResult<DB, TQuery>> {
+    ): Promise<QueryResult<TQuery>> {
         return await collectLast(
             generateLast(
-                execute<DB, TQuery>(query, {
+                execute<TQuery>(query, {
                     executors: this.executors,
                     defaultSchema: opts?.schema ?? this.schema,
                     prependSql: this.prependSql,
@@ -152,7 +160,9 @@ export class QueryEngine<DB> {
         );
     }
 
-    compile<T>(query: T extends Query<DB, infer TTable> ? T : never): {
+    compile<T extends Query>(
+        query: T,
+    ): {
         sql: string;
         params: any[];
     } {
@@ -164,9 +174,7 @@ export class QueryEngine<DB> {
         return sqlBuilder.build();
     }
 
-    async explain<TTable extends Table<DB>>(
-        query: Query<DB, TTable>,
-    ): Promise<QueryPlan> {
+    async explain<TTable extends Table<DB>>(query: Query): Promise<QueryPlan> {
         const { sqlBuilder } = composeQuery({
             defaultSchema: this.schema,
             query,

packages/backend/src/execution/composeExecutionResults.ts

@@ -1,22 +1,24 @@
-import { AnyDB, AnyTable, QueryResult } from '@synthql/queries';
+import { Query, QueryResult } from '@synthql/queries';
 import { applyCardinality } from '../query/applyCardinality';
 import { assertHasKey } from '../util/asserts/assertHasKey';
 import { setIn } from '../util/tree/setIn';
 import { ExecResultNode, ExecResultTree, ResultRow } from './types';
 
 export function composeExecutionResults(
     tree: ExecResultTree,
-): QueryResult<AnyDB, AnyTable> {
+): QueryResult<Query> {
     const queryResult: ResultRow[] = tree.root.result;
 
     for (const node of tree.root.children) {
         composeExecutionResultsRecursively(node, queryResult);
     }
 
-    return applyCardinality(
+    const result = applyCardinality(
         queryResult,
         tree.root.inputQuery.cardinality ?? 'many',
-    ) as QueryResult<AnyDB, AnyTable>;
+    ) as QueryResult<Query>;
+
+    return result;
 }
 
 function composeExecutionResultsRecursively(

packages/backend/src/execution/execute.test.ts

@@ -1,8 +1,8 @@
-import { col, query } from '@synthql/queries';
+import { query } from '@synthql/queries';
 import { describe, expect, test } from 'vitest';
 import { collectLast } from '..';
 import { QueryProvider } from '../QueryProvider';
-import { DB, schema } from '../tests/generated';
+import { DB, ref, schema } from '../tests/generated';
 import { PgCatalogInt4, PgCatalogText } from '../tests/generated/db';
 import { execute } from './execute';
 import { QueryProviderExecutor } from './executors/QueryProviderExecutor';
@@ -217,7 +217,7 @@ describe('execute', () => {
             .one();
 
         const result = await collectLast(
-            execute<DbWithVirtualTables, typeof q>(q, {
+            execute(q, {
                 executors: [new QueryProviderExecutor([actorProvider])],
                 defaultSchema,
             }),
@@ -238,7 +238,7 @@ describe('execute', () => {
                 .include({
                     rating: from('film_rating')
                         .columns('rating')
-                        .where({ film_id: col('film.film_id') })
+                        .where({ film_id: ref('film.film_id') })
                         .one(),
                 })
                 .one();
@@ -247,7 +247,7 @@ describe('execute', () => {
         const q = findFilmWithRating(1);
 
         const result = await collectLast(
-            execute<DbWithVirtualTables, typeof q>(q, {
+            execute(q, {
                 executors: [
                     new QueryProviderExecutor([
                         filmProvider,
@@ -269,7 +269,7 @@ describe('execute', () => {
         const q2 = findFilmWithRating(2);
 
         const result2 = await collectLast(
-            execute<DbWithVirtualTables, typeof q2>(q2, {
+            execute(q2, {
                 executors: [
                     new QueryProviderExecutor([
                         filmProvider,

packages/backend/src/execution/execute.ts

@@ -1,13 +1,16 @@
 import { Query, QueryResult } from '@synthql/queries';
 import { composeExecutionResults } from './composeExecutionResults';
-import { createExecutionPlan } from './planning/createExecutionPlan';
 import { executePlan } from './execution/executePlan';
+import { runSampledValidation } from './execution/runSampledValidation';
+import { shouldYield } from './execution/shouldYield';
+import { createExecutionPlan } from './planning/createExecutionPlan';
 import { QueryExecutor } from './types';
 
 export interface ExecuteProps {
     executors: Array<QueryExecutor>;
     defaultSchema: string;
     prependSql?: string;
+    runtimeValidationSampleRate?: number;
 }
 
 /**
@@ -33,13 +36,25 @@ export interface ExecuteProps {
  * We need to compose them into a single result.
  *
  */
-export async function* execute<DB, TQuery extends Query<DB>>(
+export async function* execute<TQuery extends Query>(
     query: TQuery,
     props: ExecuteProps,
-): AsyncGenerator<QueryResult<DB, TQuery>> {
+): AsyncGenerator<QueryResult<TQuery>> {
     const plan = createExecutionPlan(query, props);
 
     for await (const resultTree of executePlan(plan, props)) {
-        yield composeExecutionResults(resultTree);
+        if (shouldYield(resultTree)) {
+            // TODO(fhur) see if we can avoid this cast
+            const results = composeExecutionResults(
+                resultTree,
+            ) as QueryResult<TQuery>;
+
+            runSampledValidation({
+                rows: results,
+                schema: query.schema,
+                sampleRate: props.runtimeValidationSampleRate,
+            });
+            yield results;
+        }
     }
 }

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/runSampledValidation.ts

@@ -0,0 +1,33 @@
+import { TSchema } from '@sinclair/typebox';
+import { Value, ValueError } from '@sinclair/typebox/value';
+
+export function runSampledValidation({
+    schema,
+    sampleRate,
+    rows,
+}: {
+    schema?: TSchema;
+    sampleRate?: number;
+    rows: unknown;
+}) {
+    if (sampleRate === undefined || schema === undefined) {
+        return;
+    }
+
+    const shouldSample = Math.random() < sampleRate;
+    if (!shouldSample) {
+        return;
+    }
+    const error = Value.Errors(schema, rows).First();
+    if (error) {
+        throw new Error(formatError(error));
+    }
+}
+
+function formatError(error: ValueError) {
+    return [
+        `Validation error at ${error.path}: ${error.message}`,
+        `Actual: ${error.value}`,
+        `Expected schema: ${error.schema}`,
+    ].join('\n');
+}

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

@@ -0,0 +1,41 @@
+import { ExecResultNode, ExecResultTree } from '../types';
+
+/**
+ * Looks at the execution tree and determines if we should yield results to the client.
+ *
+ * The current implementation should only yield once - when all planning nodes have been
+ * executed.
+ */
+export function shouldYield(tree: ExecResultTree): boolean {
+    return shouldYieldNode(tree.root);
+}
+
+/**
+ * Recursively checks if we should yield for a given node.
+ *
+ * As a reminder, an `ExecResultNode` is by definition, the execution of the planned node
+ * (i.e. the node that was initially planned).
+ *
+ * An ExecResultNode might have planned children, but no executed children. This indicates that
+ * execution has not finished for the children.
+ */
+function shouldYieldNode(node: ExecResultNode): boolean {
+    const plannedChildren = node.planNode.children;
+    const executedChildren = node.children;
+
+    // If the number of executed children is less than the number of planned children,
+    // we should not yield yet
+    if (executedChildren.length < plannedChildren.length) {
+        return false;
+    }
+
+    // Recursively check all executed children
+    for (const child of executedChildren) {
+        if (!shouldYieldNode(child)) {
+            return false;
+        }
+    }
+
+    // If we've reached this point, all children (and their descendants) are fully executed
+    return true;
+}

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

@@ -1,7 +1,6 @@
-import { col } from '@synthql/queries';
 import { describe, expect, it } from 'vitest';
 import { PgExecutor } from '.';
-import { from } from '../../../tests/generated';
+import { col, from } from '../../../tests/generated';
 import { pool } from '../../../tests/queryEngine';
 import { QueryProviderExecutor } from '../QueryProviderExecutor';
 
@@ -66,11 +65,11 @@ describe('PgExecutor', () => {
         ]);
     });
 
-    const q2 = from('actor')
-        .columns('actor_id', 'first_name', 'last_name')
-        .take(2);
-
     it('Actor table SynthQL query executes to expected result', async () => {
+        const q2 = from('actor')
+            .columns('actor_id', 'first_name', 'last_name')
+            .take(2);
+
         const result = await executor.execute(q2, executeProps);
 
         expect(result).toEqual([