diff --git a/packages/2-sql/5-runtime/src/codecs/decoding.ts b/packages/2-sql/5-runtime/src/codecs/decoding.ts index 2daec9663e7a..50324f230073 100644 --- a/packages/2-sql/5-runtime/src/codecs/decoding.ts +++ b/packages/2-sql/5-runtime/src/codecs/decoding.ts @@ -9,14 +9,39 @@ import type { ContractCodecRegistry, ProjectionItem, RawQueryAst, + RawQueryColumn, SqlCodecCallContext, } from '@internal/sql-relational-core/ast'; +import { blindCast } from '@internal/utils/casts'; import { isStructuredError } from '@internal/utils/structured-error'; type ColumnRef = { table: string; column: string }; +type IncludeAggregateValue = object | string | number | boolean | null; + +interface DecodeFieldPlan { + readonly alias: string; + readonly codec: Codec | undefined; + readonly ref: ColumnRef | undefined; + readonly callColumn: SqlCodecCallContext['column']; + readonly include: boolean; + readonly many: boolean; +} + +interface CompiledRowDecoder { + readonly createTasks: ( + row: Record, + rowCtx: SqlCodecCallContext, + ) => Promise[]; + readonly createResult: ( + row: Record, + settled: unknown[], + ) => Record; +} export interface DecodeContext { readonly aliases: ReadonlyArray | undefined; + readonly fields: ReadonlyArray | undefined; + readonly compiled: CompiledRowDecoder | undefined; readonly codecs: ReadonlyMap; readonly columnRefs: ReadonlyMap; readonly includeAliases: ReadonlySet; @@ -33,15 +58,6 @@ export interface DecodeContext { const WIRE_PREVIEW_LIMIT = 100; const EMPTY_INCLUDE_ALIASES: ReadonlySet = new Set(); -function projectionListFromAst( - ast: Exclude, -): ReadonlyArray | undefined { - if (ast.kind === 'select') { - return ast.projection; - } - return ast.returning; -} - function resolveProjectionCodec( item: ProjectionItem, contractCodecs: ContractCodecRegistry | undefined, @@ -57,6 +73,8 @@ const EMPTY_MANY_ALIASES: ReadonlySet = new Set(); function undecodedContext(): DecodeContext { return { aliases: undefined, + fields: undefined, + compiled: undefined, codecs: new Map(), columnRefs: new Map(), includeAliases: EMPTY_INCLUDE_ALIASES, @@ -77,22 +95,35 @@ function undecodedContext(): DecodeContext { function rawQueryDecodeContext( ast: RawQueryAst, contractCodecs: ContractCodecRegistry | undefined, + options: BuildDecodeContextOptions, ): DecodeContext { if (ast.result.kind === 'affected-count') { return undecodedContext(); } const aliases: string[] = []; + const fields: DecodeFieldPlan[] = []; const codecs = new Map(); - for (const [name, column] of Object.entries(ast.result.columns)) { + for (const [name, column] of Object.entries(ast.result.columns)) { aliases.push(name); - if (contractCodecs) { - codecs.set(name, contractCodecs.forCodecRef({ codecId: column.codecId })); + const codec = contractCodecs?.forCodecRef({ codecId: column.codecId }); + if (codec) { + codecs.set(name, codec); } + fields.push({ + alias: name, + codec, + ref: undefined, + callColumn: undefined, + include: false, + many: false, + }); } return { aliases, + fields, + compiled: options.reusable ? compileRowDecoder(fields) : undefined, codecs, columnRefs: new Map(), includeAliases: EMPTY_INCLUDE_ALIASES, @@ -101,20 +132,26 @@ function rawQueryDecodeContext( }; } +export interface BuildDecodeContextOptions { + readonly reusable?: boolean; +} + export function buildDecodeContext( ast: AnyQueryAst, contractCodecs: ContractCodecRegistry | undefined, + options: BuildDecodeContextOptions = {}, ): DecodeContext { if (ast.kind === 'raw-query') { - return rawQueryDecodeContext(ast, contractCodecs); + return rawQueryDecodeContext(ast, contractCodecs, options); } - const projection = projectionListFromAst(ast); + const projection = ast.kind === 'select' ? ast.projection : ast.returning; if (!projection || projection.length === 0) { return undecodedContext(); } const aliases: string[] = []; + const fields: DecodeFieldPlan[] = []; const codecs = new Map(); const columnRefs = new Map(); const includeAliases = new Set(); @@ -128,21 +165,34 @@ export function buildDecodeContext( codecs.set(item.alias, codec); } - if (item.codec?.many) { + const many = item.codec?.many === true; + if (many) { manyAliases.add(item.alias); } + let ref: ColumnRef | undefined; + let include = false; if (item.expr.kind === 'column-ref') { - columnRefs.set(item.alias, { - table: item.expr.table, - column: item.expr.column, - }); + ref = { table: item.expr.table, column: item.expr.column }; + columnRefs.set(item.alias, ref); } else if (item.expr.kind === 'subquery' || item.expr.kind === 'json-array-agg') { + include = true; includeAliases.add(item.alias); } + const callColumn = ref ? { table: ref.table, name: ref.column } : undefined; + fields.push({ alias: item.alias, codec, ref, callColumn, include, many }); } - return { aliases, codecs, columnRefs, includeAliases, manyAliases, aliasSource: 'projection' }; + return { + aliases, + fields, + compiled: options.reusable ? compileRowDecoder(fields) : undefined, + codecs, + columnRefs, + includeAliases, + manyAliases, + aliasSource: 'projection', + }; } function previewWireValue(wireValue: unknown): string { @@ -190,7 +240,7 @@ function wrapIncludeAggregateFailure(error: unknown, alias: string, wireValue: u throw wrapped; } -function decodeIncludeAggregate(alias: string, wireValue: unknown): unknown { +function decodeIncludeAggregate(alias: string, wireValue: unknown): IncludeAggregateValue { if (wireValue === null || wireValue === undefined) { return []; } @@ -222,72 +272,113 @@ function decodeIncludeAggregate(alias: string, wireValue: unknown): unknown { * * For `many`-flagged aliases the driver has already parsed the wire form into a JS array; this function maps the element codec over that array element-by-element, passing `null` elements through unchanged. Element-level failures surface through the existing `RUNTIME.DECODE_FAILED` envelope with the column/codec context from the parent cell. */ -async function decodeField( - alias: string, +async function decodeManyField( + field: DecodeFieldPlan, + wireValue: unknown, + codec: Codec, + cellCtx: SqlCodecCallContext, +): Promise { + const { alias, ref } = field; + if (!Array.isArray(wireValue)) { + wrapDecodeFailure( + new TypeError( + `expected an array from the driver for many-typed column, got ${typeof wireValue}`, + ), + alias, + ref, + codec, + wireValue, + ); + } + const decoded: unknown[] = []; + for (const elem of wireValue) { + if (elem === null || elem === undefined) { + decoded.push(null); + continue; + } + try { + decoded.push(await codec.decode(elem, cellCtx)); + } catch (error) { + if (isStructuredError(error)) throw error; + wrapDecodeFailure(error, alias, ref, codec, elem); + } + } + return decoded; +} + +function decodeField( + field: DecodeFieldPlan, wireValue: unknown, - decodeCtx: DecodeContext, rowCtx: SqlCodecCallContext, ): Promise { if (wireValue === null) { - return null; + return Promise.resolve(null); } - const codec = decodeCtx.codecs.get(alias); + const { alias, codec, ref, callColumn } = field; if (!codec) { - return wireValue; + return Promise.resolve(wireValue); } - const ref = decodeCtx.columnRefs.get(alias); - + const signal = rowCtx.signal; let cellCtx: SqlCodecCallContext; - if (ref) { - cellCtx = { ...rowCtx, column: { table: ref.table, name: ref.column } }; + if (callColumn) { + cellCtx = signal === undefined ? { column: callColumn } : { signal, column: callColumn }; } else { - const { column: _drop, ...rowCtxWithoutColumn } = rowCtx; - cellCtx = rowCtxWithoutColumn; + cellCtx = signal === undefined ? {} : { signal }; } - if (decodeCtx.manyAliases.has(alias)) { - if (!Array.isArray(wireValue)) { - wrapDecodeFailure( - new TypeError( - `expected an array from the driver for many-typed column, got ${typeof wireValue}`, - ), - alias, - ref, - codec, - wireValue, - ); - } - const decoded: unknown[] = []; - for (const elem of wireValue) { - if (elem === null || elem === undefined) { - decoded.push(null); - continue; - } - try { - decoded.push(await codec.decode(elem, cellCtx)); - } catch (error) { - if (isStructuredError(error)) throw error; - wrapDecodeFailure(error, alias, ref, codec, elem); - } - } - return decoded; + if (field.many) { + return decodeManyField(field, wireValue, codec, cellCtx); } - try { - return await codec.decode(wireValue, cellCtx); - } catch (error) { - // Any structured envelope (dotted `code` per `isStructuredError`) is - // stable by construction — let it pass through unchanged. This covers - // every `runtimeError`-built envelope and plain `structuredError` - // envelopes from extension codecs (e.g. a codec-authored - // `RUNTIME.DECODE_FAILED` — no double wrap). Symmetric with the - // encode-side guard. + const wrapFailure = (error: unknown): never => { if (isStructuredError(error)) { throw error; } wrapDecodeFailure(error, alias, ref, codec, wireValue); + }; + + try { + const decoded = codec.decode(wireValue, cellCtx); + return decoded instanceof Promise + ? decoded.catch(wrapFailure) + : Promise.resolve(decoded).catch(wrapFailure); + } catch (error) { + return Promise.reject(error).catch(wrapFailure); + } +} + +function compileRowDecoder(fields: ReadonlyArray): CompiledRowDecoder | undefined { + const taskExpressions = fields.map((field, index) => + field.include + ? 'Promise.resolve(undefined)' + : `decodeField(fields[${index}], row[${JSON.stringify(field.alias)}], rowCtx)`, + ); + const resultProperties = fields.map((field, index) => { + const alias = JSON.stringify(field.alias); + const value = field.include + ? `decodeIncludeAggregate(${alias}, row[${alias}])` + : `settled[${index}]`; + return `[${alias}]: ${value}`; + }); + try { + const create = new Function( + 'decodeField', + 'decodeIncludeAggregate', + 'fields', + `"use strict"; +return { + createTasks(row, rowCtx) { return [${taskExpressions.join(',')}]; }, + createResult(row, settled) { return {${resultProperties.join(',')}}; } +};`, + ); + return blindCast< + CompiledRowDecoder, + 'new Function is generated exclusively from JSON-escaped aliases and numeric field indices' + >(create(decodeField, decodeIncludeAggregate, fields)); + } catch { + return undefined; } } @@ -332,31 +423,50 @@ export async function decodeRow( } } - const tasks: Promise[] = []; + const compiled = decodeCtx.compiled; + let tasks: Promise[]; const includeIndices: { index: number; alias: string; value: unknown }[] = []; - for (let i = 0; i < aliases.length; i++) { - const alias = aliases[i] as string; - const wireValue = row[alias]; - - if (decodeCtx.includeAliases.has(alias)) { - includeIndices.push({ index: i, alias, value: wireValue }); - tasks.push(Promise.resolve(undefined)); - continue; + if (compiled) { + tasks = compiled.createTasks(row, rowCtx); + } else { + tasks = new Array>(aliases.length); + const fields = decodeCtx.fields; + if (fields === undefined) { + let index = 0; + for (const alias of aliases) { + tasks[index++] = Promise.resolve(row[alias]); + } + } else { + let index = 0; + for (const field of fields) { + const wireValue = row[field.alias]; + if (field.include) { + includeIndices.push({ index, alias: field.alias, value: wireValue }); + tasks[index++] = Promise.resolve(undefined); + continue; + } + tasks[index++] = decodeField(field, wireValue, rowCtx); + } } - - tasks.push(decodeField(alias, wireValue, decodeCtx, rowCtx)); } - const settled = await raceAgainstAbort(Promise.all(tasks), signal, 'decode'); + const allTasks = Promise.all(tasks); + const settled = + signal === undefined ? await allTasks : await raceAgainstAbort(allTasks, signal, 'decode'); + + if (compiled) { + return compiled.createResult(row, settled); + } for (const entry of includeIndices) { settled[entry.index] = decodeIncludeAggregate(entry.alias, entry.value); } const decoded: Record = {}; - for (let i = 0; i < aliases.length; i++) { - decoded[aliases[i] as string] = settled[i]; + let index = 0; + for (const alias of aliases) { + decoded[alias] = settled[index++]; } return decoded; } diff --git a/packages/2-sql/5-runtime/src/sql-runtime.ts b/packages/2-sql/5-runtime/src/sql-runtime.ts index 874b9c8e4f8b..3e337d1eacdf 100644 --- a/packages/2-sql/5-runtime/src/sql-runtime.ts +++ b/packages/2-sql/5-runtime/src/sql-runtime.ts @@ -578,7 +578,9 @@ export abstract class SqlRuntimeBase = Co params: orderedRefs.map((r) => (r.kind === 'param-ref' ? r.value : undefined)), }); - const decodeContext = buildDecodeContext(finalPlan.ast, this.contractCodecs); + const decodeContext = buildDecodeContext(finalPlan.ast, this.contractCodecs, { + reusable: true, + }); const paramMetadata = deriveParamMetadata(finalPlan.ast); const internals: PreparedStatementInternals = Object.freeze({ diff --git a/packages/2-sql/5-runtime/test/codec-async.test.ts b/packages/2-sql/5-runtime/test/codec-async.test.ts index 287d190f6707..d2f6b86ae4d2 100644 --- a/packages/2-sql/5-runtime/test/codec-async.test.ts +++ b/packages/2-sql/5-runtime/test/codec-async.test.ts @@ -14,7 +14,7 @@ import { } from '@internal/sql-relational-core/ast'; import type { SqlExecutionPlan } from '@internal/sql-relational-core/plan'; import { timeouts } from '@repo/test-utils'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { buildDecodeContext, decodeRow } from '../src/codecs/decoding'; import { encodeParams } from '../src/codecs/encoding'; import { createAsyncSecretCodec, decryptSecret, encryptSecret } from './seeded-secret-codec'; @@ -373,6 +373,49 @@ describe('decodeRow — async, concurrent per-cell dispatch', () => { expect(result).toEqual({ a: 'A:A-DEC', b: 'B:B-DEC', n: 42 }); }); + it('dispatches later cells after an earlier codec throws synchronously', async () => { + const cause = new Error('sync failure'); + let laterCalls = 0; + const syncThrowCodec: Codec = { + ...defineTestCodec({ + typeId: 'test/sync-throw@1', + targetTypes: ['text'], + encode: (value: string) => value, + decode: (wire: string) => wire, + }), + decode: () => { + throw cause; + }, + }; + const registry: Codec[] = [ + syncThrowCodec, + defineTestCodec({ + typeId: 'test/later@1', + targetTypes: ['text'], + encode: (value: string) => value, + decode: (wire: string) => { + laterCalls++; + return wire; + }, + }), + ]; + const plan = buildAstPlan({ + projections: [ + { alias: 'first', codecId: 'test/sync-throw@1' }, + { alias: 'later', codecId: 'test/later@1' }, + ], + }); + + await expect( + decodeRow( + { first: 'first', later: 'later' }, + buildDecodeContext(plan.ast, buildTestContractCodecs(registry)), + {}, + ), + ).rejects.toMatchObject({ code: 'RUNTIME.DECODE_FAILED', cause }); + expect(laterCalls).toBe(1); + }); + it('always awaits codec.decode and yields plain values (no Promise leaks)', async () => { const registry = [ defineTestCodec({ @@ -488,6 +531,67 @@ describe('decodeRow — async, concurrent per-cell dispatch', () => { }); }); + it('decodes mixed codec and passthrough fields without an abort signal', async () => { + const registry = [ + defineTestCodec({ + typeId: 'test/uppercase@1', + targetTypes: ['text'], + encode: (value: string) => value, + decode: async (wire: string) => wire.toUpperCase(), + }), + ]; + const plan = buildAstPlan({ + projections: [ + { alias: 'before' }, + { alias: 'name', codecId: 'test/uppercase@1' }, + { alias: 'nullable', codecId: 'test/uppercase@1' }, + { alias: 'after' }, + ], + }); + + const result = await decodeRow( + { before: 1, name: 'alice', nullable: null, after: 3 }, + buildDecodeContext(plan.ast, buildTestContractCodecs(registry)), + {}, + ); + + expect(result).toEqual({ before: 1, name: 'ALICE', nullable: null, after: 3 }); + }); + + it('uses the generic decoder for one-shot contexts', async () => { + const plan = buildAstPlan({ projections: [{ alias: 'value' }] }); + const context = buildDecodeContext(plan.ast, buildTestContractCodecs([])); + + expect(context.compiled).toBeUndefined(); + await expect(decodeRow({ value: 'safe' }, context, {})).resolves.toEqual({ value: 'safe' }); + }); + + it('treats projection aliases as data when compiling a reusable row shape', async () => { + const alias = 'field"];\nthrow new Error("injected") //'; + const plan = buildAstPlan({ projections: [{ alias }] }); + const context = buildDecodeContext(plan.ast, buildTestContractCodecs([]), { reusable: true }); + + expect(context.compiled).toBeDefined(); + await expect(decodeRow({ [alias]: 'safe' }, context, {})).resolves.toEqual({ [alias]: 'safe' }); + }); + + it('falls back to the generic decoder when reusable context compilation is unavailable', async () => { + const plan = buildAstPlan({ projections: [{ alias: 'value' }] }); + vi.stubGlobal('Function', function unavailableFunctionConstructor(): never { + throw new EvalError('unsafe-eval is disabled'); + }); + + try { + const context = buildDecodeContext(plan.ast, buildTestContractCodecs([]), { + reusable: true, + }); + expect(context.compiled).toBeUndefined(); + await expect(decodeRow({ value: 'safe' }, context, {})).resolves.toEqual({ value: 'safe' }); + } finally { + vi.unstubAllGlobals(); + } + }); + it('passes wire values through for raw plans (no AST, no codec decoding)', async () => { const registry = [ defineTestCodec({ @@ -527,6 +631,34 @@ describe('decodeRow — async, concurrent per-cell dispatch', () => { }); }); + it('rejects an inherited projection alias as missing', async () => { + const plan = buildAstPlan({ projections: [{ alias: 'toString' }] }); + + await expect( + decodeRow({}, buildDecodeContext(plan.ast, buildTestContractCodecs([])), {}), + ).rejects.toMatchObject({ + code: 'RUNTIME.DECODE_FAILED', + details: { + alias: 'toString', + expectedAliases: ['toString'], + presentKeys: [], + }, + }); + }); + + it('preserves an own undefined value as distinct from a missing projection alias', async () => { + const plan = buildAstPlan({ projections: [{ alias: 'value' }] }); + + const result = await decodeRow( + { value: undefined }, + buildDecodeContext(plan.ast, buildTestContractCodecs([])), + {}, + ); + + expect(Object.hasOwn(result, 'value')).toBe(true); + expect(result['value']).toBeUndefined(); + }); + it('preserves wire null for AST-backed plans (distinct from missing alias)', async () => { const registry = [ defineTestCodec({ diff --git a/packages/2-sql/5-runtime/test/codec-decode-ctx.test.ts b/packages/2-sql/5-runtime/test/codec-decode-ctx.test.ts index 177dc98722e5..564ca736a539 100644 --- a/packages/2-sql/5-runtime/test/codec-decode-ctx.test.ts +++ b/packages/2-sql/5-runtime/test/codec-decode-ctx.test.ts @@ -117,6 +117,32 @@ describe('decodeRow — SqlCodecCallContext threading', () => { ]); }); + it('populates ctx.column from a default empty row context', async () => { + let observed: SqlCodecCallContext | undefined; + const registry = [ + defineTestCodec({ + typeId: 'test/observe-empty-ctx@1', + targetTypes: ['text'], + encode: (value: string) => value, + decode: (wire: string, ctx?: SqlCodecCallContext) => { + observed = ctx; + return wire; + }, + }), + ]; + const plan = buildPlan([ + columnProjection('email', 'users', 'email', 'test/observe-empty-ctx@1'), + ]); + + await decodeRow( + { email: 'user@example.com' }, + buildDecodeContext(plan.ast, buildTestContractCodecs(registry)), + {}, + ); + + expect(observed).toEqual({ column: { table: 'users', name: 'email' } }); + }); + it('populates ctx.column when the projection points at a different table.column than the alias', async () => { let observed: SqlCodecCallContext | undefined; const registry = [ diff --git a/test/bench/bench/decode-row.ts b/test/bench/bench/decode-row.ts index cec22300cd54..f9c20343c51b 100644 --- a/test/bench/bench/decode-row.ts +++ b/test/bench/bench/decode-row.ts @@ -63,7 +63,11 @@ function createBenchmarkCases( if (!rows) { throw new Error(`No fixture rows for query "${name}"`); } - return { name, rows, decodeCtx: buildDecodeContext(plan.ast, contractCodecs) }; + return { + name, + rows, + decodeCtx: buildDecodeContext(plan.ast, contractCodecs, { reusable: true }), + }; }); }