Add SqlExecutionBudget propagation through executor

Plumbs an optional SqlExecutionBudget through the executor context so
materializing operators can detect runaway resource use and abort with a
structured error indicating which limit was hit.

Budget fields: maxRowsToMaterialize (global row counter across operators),
maxHeapBytes (global byte counter), maxIntermediateBytes (per-operator byte
counter), timeoutMs (wall-clock deadline), allowDerivedColumnScan (opts out
of the scalar aggregate scanColumn fast path).

Operators that materialize now call context.budget?.operator(name).addRow()
before pinning each row: Sort, HashAggregate, ScalarAggregate slow path,
HashJoin and NestedLoopJoin (right side), PositionalJoin (both sides),
Distinct, UNION/INTERSECT/EXCEPT, and Window's non-streaming path. Streaming
loops (filterRows, scanColumnAggregate chunks) check the timeout deadline so
long-running pure-stream queries also abort.

Errors throw a new SqlBudgetError with limit/value/max/operator fields.
Budget passes through Subquery, LATERAL, and correlated subquery contexts.

28 new tests cover each limit reaching threshold, the structured error
shape, fast-path opt-out, and confirm budget-free queries are unchanged.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: philcunliffe

src/execute/aggregates.js

@@ -80,11 +80,13 @@ export function executeHashAggregate(plan, context) {
     columns: selectColumnNames(plan.columns, child.columns),
     maxRows: child.maxRows,
     async *rows() {
+      const op = context.budget?.operator('HashAggregate')
       // Collect all rows
       /** @type {AsyncRow[]} */
       const allRows = []
       for await (const row of child.rows()) {
         if (context.signal?.aborted) return
+        op?.addRow()
         allRows.push(row)
       }
 
@@ -147,7 +149,8 @@ export function executeHashAggregate(plan, context) {
  */
 export function executeScalarAggregate(plan, context) {
   // Fast path: use scanColumn when available
-  const fast = tryColumnScanAggregate(plan, context)
+  const allowFast = context.budget ? context.budget.allowDerivedColumnScan : true
+  const fast = allowFast ? tryColumnScanAggregate(plan, context) : undefined
   if (fast) {
     return {
       columns: selectColumnNames(plan.columns, []),
@@ -163,11 +166,13 @@ export function executeScalarAggregate(plan, context) {
     numRows: plan.having ? undefined : 1,
     maxRows: 1,
     async *rows() {
+      const op = context.budget?.operator('ScalarAggregate')
       // Collect all rows into single group
       /** @type {AsyncRow[]} */
       const group = []
       for await (const row of child.rows()) {
         if (context.signal?.aborted) return
+        op?.addRow()
         group.push(row)
       }
 
@@ -212,7 +217,7 @@ export function executeScalarAggregate(plan, context) {
  * @param {ExecuteContext} context
  * @returns {(() => AsyncGenerator<AsyncRow>) | undefined}
  */
-function tryColumnScanAggregate(plan, { tables, signal }) {
+function tryColumnScanAggregate(plan, { tables, signal, budget }) {
   // No HAVING support in fast path
   if (plan.having) return
   // Child must be a direct table scan
@@ -243,7 +248,7 @@ function tryColumnScanAggregate(plan, { tables, signal }) {
 
     for (const spec of specs) {
       columns.push(spec.alias)
-      cells[spec.alias] = () => scanColumnAggregate({ table, spec, limit, offset, signal })
+      cells[spec.alias] = () => scanColumnAggregate({ table, spec, limit, offset, signal, budget })
     }
 
     yield { columns, cells }
@@ -283,15 +288,17 @@ function extractColumnAggSpec({ expr, alias }) {
  * @param {number} [options.limit]
  * @param {number} [options.offset]
  * @param {AbortSignal} [options.signal]
+ * @param {import('../types.js').BudgetTracker} [options.budget]
  * @returns {Promise<SqlPrimitive>}
  */
-async function scanColumnAggregate({ table, spec, limit, offset, signal }) {
+async function scanColumnAggregate({ table, spec, limit, offset, signal, budget }) {
   const values = table.scanColumn({ column: spec.column, limit, offset, signal })
 
   if (spec.funcName === 'COUNT' && spec.distinct) {
     const seen = new Set()
     for await (const chunk of values) {
       if (signal?.aborted) return
+      budget?.checkTimeout()
       for (let i = 0; i < chunk.length; i++) {
         const v = chunk[i]
         if (v == null) continue
@@ -305,6 +312,7 @@ async function scanColumnAggregate({ table, spec, limit, offset, signal }) {
     let count = 0
     for await (const chunk of values) {
       if (signal?.aborted) return
+      budget?.checkTimeout()
       for (let i = 0; i < chunk.length; i++) {
         if (chunk[i] != null) count++
       }
@@ -322,6 +330,7 @@ async function scanColumnAggregate({ table, spec, limit, offset, signal }) {
 
   for await (const chunk of values) {
     if (signal?.aborted) return
+    budget?.checkTimeout()
     for (let i = 0; i < chunk.length; i++) {
       const v = chunk[i]
       if (v == null) continue

src/execute/budget.js

@@ -0,0 +1,109 @@
+/**
+ * @import { BudgetOperator, BudgetTracker, SqlExecutionBudget } from '../types.js'
+ */
+
+const DEFAULT_ROW_BYTES = 64
+
+/**
+ * Structured error thrown when a SQL execution budget is exceeded. The
+ * `limit` field identifies which budget field was breached so callers can
+ * decide whether to abort, fall back to a different strategy, or surface a
+ * user-facing message.
+ */
+export class SqlBudgetError extends Error {
+  /**
+   * @param {Object} options
+   * @param {'maxRowsToMaterialize' | 'maxHeapBytes' | 'maxIntermediateBytes' | 'timeoutMs'} options.limit
+   * @param {number} options.value - actual measured value at the time of abort
+   * @param {number} options.max - configured budget limit
+   * @param {string} [options.operator] - operator name where the limit was hit
+   */
+  constructor({ limit, value, max, operator }) {
+    const where = operator ? ` (operator=${operator})` : ''
+    const message = `SQL execution budget exceeded: ${limit}=${value} > ${max}${where}`
+    super(message)
+    this.name = 'SqlBudgetError'
+    this.limit = limit
+    this.value = value
+    this.max = max
+    if (operator !== undefined) this.operator = operator
+  }
+}
+
+/**
+ * Creates a budget tracker for a single query execution. Returns undefined
+ * when no budget is provided so callers can use `tracker?.operator(...)`
+ * without conditional plumbing through every operator.
+ *
+ * @param {SqlExecutionBudget} [budget]
+ * @returns {BudgetTracker | undefined}
+ */
+export function createBudget(budget) {
+  if (!budget) return undefined
+  const startTime = Date.now()
+  const { timeoutMs } = budget
+  const deadline = timeoutMs !== undefined ? startTime + timeoutMs : undefined
+
+  let totalRows = 0
+  let totalHeapBytes = 0
+
+  /**
+   * @param {string} [operator]
+   */
+  function timeoutCheck(operator) {
+    if (deadline === undefined || timeoutMs === undefined) return
+    if (Date.now() > deadline) {
+      throw new SqlBudgetError({
+        limit: 'timeoutMs',
+        value: Date.now() - startTime,
+        max: timeoutMs,
+        operator,
+      })
+    }
+  }
+
+  return {
+    budget,
+    allowDerivedColumnScan: budget.allowDerivedColumnScan !== false,
+    checkTimeout() { timeoutCheck() },
+    operator(name) {
+      let opBytes = 0
+      /** @type {BudgetOperator} */
+      const handle = {
+        addRow(approxBytes) {
+          const bytes = approxBytes ?? DEFAULT_ROW_BYTES
+          totalRows++
+          if (budget.maxRowsToMaterialize !== undefined && totalRows > budget.maxRowsToMaterialize) {
+            throw new SqlBudgetError({
+              limit: 'maxRowsToMaterialize',
+              value: totalRows,
+              max: budget.maxRowsToMaterialize,
+              operator: name,
+            })
+          }
+          opBytes += bytes
+          if (budget.maxIntermediateBytes !== undefined && opBytes > budget.maxIntermediateBytes) {
+            throw new SqlBudgetError({
+              limit: 'maxIntermediateBytes',
+              value: opBytes,
+              max: budget.maxIntermediateBytes,
+              operator: name,
+            })
+          }
+          totalHeapBytes += bytes
+          if (budget.maxHeapBytes !== undefined && totalHeapBytes > budget.maxHeapBytes) {
+            throw new SqlBudgetError({
+              limit: 'maxHeapBytes',
+              value: totalHeapBytes,
+              max: budget.maxHeapBytes,
+              operator: name,
+            })
+          }
+          timeoutCheck(name)
+        },
+        checkTimeout() { timeoutCheck(name) },
+      }
+      return handle
+    },
+  }
+}

src/execute/execute.js

@@ -8,6 +8,7 @@ import { statementScope } from '../plan/columns.js'
 import { validateScan, validateTable } from '../validation/tables.js'
 import { executeHashAggregate, executeScalarAggregate } from './aggregates.js'
 import { filterBatches, limitBatches, projectBatchesSimple } from './batchOps.js'
+import { createBudget } from './budget.js'
 import { executeHashJoin, executeNestedLoopJoin, executePositionalJoin } from './join.js'
 import { executeSort } from './sort.js'
 import { addBounds, minBounds, stableRowKey } from './utils.js'
@@ -24,7 +25,7 @@ import { executeWindow } from './window.js'
  * @param {ExecuteSqlOptions} options
  * @returns {QueryResults}
  */
-export function executeSql({ tables, query, functions, signal }) {
+export function executeSql({ tables, query, functions, signal, budget }) {
   const parsed = typeof query === 'string' ? parseSql({ query, functions }) : query
 
   // Normalize tables: convert arrays to AsyncDataSource
@@ -47,7 +48,7 @@ export function executeSql({ tables, query, functions, signal }) {
   const ctePlans = new Map()
   /** @type {Map<string, string[]>} */
   const cteColumns = new Map()
-  const context = { tables: normalizedTables, functions, signal, scope, ctePlans, cteColumns }
+  const context = { tables: normalizedTables, functions, signal, scope, ctePlans, cteColumns, budget: createBudget(budget) }
   const plan = planSql({ query: parsed, functions, tables: normalizedTables, ctePlans, cteColumns })
   return executePlan({ plan, context })
 }
@@ -425,6 +426,7 @@ async function* filterRows(rows, condition, context, limit) {
     buffer.push({ row, rowIndex })
 
     if (buffer.length >= chunkSize) {
+      context.budget?.checkTimeout()
       const results = await Promise.all(buffer.map(b =>
         evaluateExpr({ node: condition, row: b.row, rowIndex: b.rowIndex, context })
       ))
@@ -626,6 +628,7 @@ function executeDistinct(plan, context) {
     maxRows: child.maxRows,
     async *rows() {
       const { signal } = context
+      const op = context.budget?.operator('Distinct')
       const MAX_CHUNK = 256
 
       const seen = new Set()
@@ -642,6 +645,7 @@ function executeDistinct(plan, context) {
           for (let i = 0; i < buffer.length; i++) {
             const key = await keys[i]
             if (!seen.has(key)) {
+              op?.addRow()
               seen.add(key)
               yield buffer[i]
             }
@@ -656,6 +660,7 @@ function executeDistinct(plan, context) {
         for (let i = 0; i < buffer.length; i++) {
           const key = await keys[i]
           if (!seen.has(key)) {
+            op?.addRow()
             seen.add(key)
             yield buffer[i]
           }
@@ -714,11 +719,13 @@ function executeSetOperation(plan, context) {
         maxRows: addBounds(left.maxRows, right.maxRows),
         async *rows() {
           // UNION: yield deduplicated rows from both sides
+          const op = context.budget?.operator('Union')
           const seen = new Set()
           for await (const row of left.rows()) {
             if (signal?.aborted) return
             const key = await stableRowKey(row)
             if (!seen.has(key)) {
+              op?.addRow()
               seen.add(key)
               yield row
             }
@@ -727,6 +734,7 @@ function executeSetOperation(plan, context) {
             if (signal?.aborted) return
             const key = await stableRowKey(row)
             if (!seen.has(key)) {
+              op?.addRow()
               seen.add(key)
               yield row
             }
@@ -741,12 +749,14 @@ function executeSetOperation(plan, context) {
       columns: left.columns,
       maxRows: minBounds(left.maxRows, right.maxRows),
       async *rows() {
+        const op = context.budget?.operator('Intersect')
         // Materialize right side keys
         /** @type {Map<any, number>} */
         const rightKeys = new Map()
         for await (const row of right.rows()) {
           if (signal?.aborted) return
           const key = await stableRowKey(row)
+          if (!rightKeys.has(key)) op?.addRow()
           rightKeys.set(key, (rightKeys.get(key) ?? 0) + 1)
         }
 
@@ -768,6 +778,7 @@ function executeSetOperation(plan, context) {
             if (signal?.aborted) return
             const key = await stableRowKey(row)
             if (rightKeys.has(key) && !seen.has(key)) {
+              op?.addRow()
               seen.add(key)
               yield row
             }
@@ -783,12 +794,14 @@ function executeSetOperation(plan, context) {
       columns: left.columns,
       maxRows: left.maxRows,
       async *rows() {
+        const op = context.budget?.operator('Except')
         // Materialize right side keys
         /** @type {Map<any, number>} */
         const rightKeys = new Map()
         for await (const row of right.rows()) {
           if (signal?.aborted) return
           const key = await stableRowKey(row)
+          if (!rightKeys.has(key)) op?.addRow()
           rightKeys.set(key, (rightKeys.get(key) ?? 0) + 1)
         }
 
@@ -811,6 +824,7 @@ function executeSetOperation(plan, context) {
             if (signal?.aborted) return
             const key = await stableRowKey(row)
             if (!rightKeys.has(key) && !seen.has(key)) {
+              op?.addRow()
               seen.add(key)
               yield row
             }

src/execute/join.js

@@ -23,6 +23,7 @@ export function executeNestedLoopJoin(plan, context) {
   return {
     columns: mergeColumnNames(left.columns, right.columns, plan.leftAlias, plan.rightAlias),
     async *rows() {
+      const op = context.budget?.operator('NestedLoopJoin')
       const leftTable = plan.leftAlias
       const rightTable = plan.rightAlias
 
@@ -31,6 +32,7 @@ export function executeNestedLoopJoin(plan, context) {
       const rightRows = []
       for await (const row of right.rows()) {
         if (context.signal?.aborted) return
+        op?.addRow()
         rightRows.push(row)
       }
 
@@ -153,6 +155,7 @@ export function executePositionalJoin(plan, context) {
     maxRows: maxBounds(left.maxRows, right.maxRows),
     async *rows() {
       const { signal } = context
+      const op = context.budget?.operator('PositionalJoin')
       const leftTable = plan.leftAlias
       const rightTable = plan.rightAlias
 
@@ -161,13 +164,15 @@ export function executePositionalJoin(plan, context) {
       const leftRows = []
       for await (const row of left.rows()) {
         if (signal?.aborted) return
+        op?.addRow()
         leftRows.push(row)
       }
 
       /** @type {AsyncRow[]} */
       const rightRows = []
       for await (const row of right.rows()) {
         if (signal?.aborted) return
+        op?.addRow()
         rightRows.push(row)
       }
 
@@ -198,6 +203,7 @@ export function executeHashJoin(plan, context) {
   return {
     columns: mergeColumnNames(left.columns, right.columns, plan.leftAlias, plan.rightAlias),
     async *rows() {
+      const op = context.budget?.operator('HashJoin')
       const leftTable = plan.leftAlias
       const rightTable = plan.rightAlias
       const { leftKeys, rightKeys, residual } = plan
@@ -207,6 +213,7 @@ export function executeHashJoin(plan, context) {
       const rightRows = []
       for await (const row of right.rows()) {
         if (context.signal?.aborted) return
+        op?.addRow()
         rightRows.push(row)
       }
 

src/execute/sort.js

@@ -355,10 +355,12 @@ export function executeSort(plan, context) {
       }
 
       // Full sort path: buffer all rows, then sort.
+      const op = context.budget?.operator('Sort')
       /** @type {AsyncRow[]} */
       const rows = []
       for await (const row of child.rows()) {
         if (context.signal?.aborted) return
+        op?.addRow()
         rows.push(row)
       }