Buffer lean rows in ORDER BY and GROUP BY to bound memory

src/backend/dataSource.js

@@ -18,6 +18,74 @@ export function asyncRow(obj, columns) {
   return { columns, cells, resolved: obj }
 }
 
+/**
+ * Reads a column value from a row, preferring the pre-materialized `resolved`
+ * object for exposed columns and falling back to the lazy cell closure. When
+ * present, `resolved` is expected to hold a value for every column in
+ * `row.columns`; it may also carry hidden source fields not visible to SQL.
+ * Returns a value or a Promise; callers await either.
+ *
+ * @param {AsyncRow} row
+ * @param {string} key
+ * @returns {SqlPrimitive | Promise<SqlPrimitive>}
+ */
+export function readCell(row, key) {
+  if (row.resolved != null && row.columns.includes(key)) return row.resolved[key]
+  return row.cells[key]()
+}
+
+/**
+ * Whether a row can read the given key. `resolved` only exposes row.columns;
+ * hidden expression caches may live in cells without appearing in row.columns.
+ *
+ * @param {AsyncRow} row
+ * @param {string} key
+ * @returns {boolean}
+ */
+export function hasCell(row, key) {
+  return row.resolved != null && row.columns.includes(key) ||
+    Object.prototype.hasOwnProperty.call(row.cells, key)
+}
+
+/**
+ * Returns a lazy cell closure for a column, resolve-aware: reuses the row's own
+ * closure when present, otherwise synthesizes one from `resolved` (lean buffered
+ * rows drop their closures to save memory). Returns undefined when the row
+ * exposes neither, so callers can detect an unknown column.
+ *
+ * @param {AsyncRow} row
+ * @param {string} key
+ * @returns {(() => Promise<SqlPrimitive>) | undefined}
+ */
+export function cellThunk(row, key) {
+  const cell = row.cells[key]
+  if (cell) return cell
+  if (row.resolved != null && row.columns.includes(key)) {
+    const { [key]: value } = row.resolved
+    return () => Promise.resolve(value)
+  }
+  return undefined
+}
+
+/**
+ * Resolve-aware view of a row's cells as a complete map. Lean buffered rows
+ * (values in `resolved`, closures dropped to save memory) get their closures
+ * rebuilt from `resolved`, with any hidden non-column cells preserved. Rows
+ * without `resolved` return their own map unchanged so lazy or cached closures
+ * still apply. May return the row's own map, so callers that mutate must copy.
+ *
+ * @param {AsyncRow} row
+ * @returns {AsyncCells}
+ */
+export function rowCells(row) {
+  if (row.resolved == null) return row.cells
+  const { cells } = asyncRow(row.resolved, row.columns)
+  for (const key in row.cells) {
+    if (!(key in cells)) cells[key] = row.cells[key]
+  }
+  return cells
+}
+
 /**
  * Creates an async memory-backed data source from an array of plain objects
  *
@@ -105,7 +173,7 @@ export function cachedDataSource(source) {
             /** @type {AsyncCells} */
             const cells = {}
             for (const key of row.columns) {
-              const cell = row.cells[key]
+              const cell = cellThunk(row, key)
               cells[key] = () => {
                 let value = rowCache.get(key)
                 if (!value) {

src/execute/aggregates.js

@@ -1,3 +1,4 @@
+import { cellThunk, rowCells } from '../backend/dataSource.js'
 import { derivedAlias } from '../expression/alias.js'
 import { evaluateExpr } from '../expression/evaluate.js'
 import { executePlan, selectColumnNames } from './execute.js'
@@ -37,7 +38,9 @@ function projectAggregateColumns(selectColumns, group, context) {
           const dotIndex = key.indexOf('.')
           const outputKey = prefix ? key.substring(prefix.length) : dotIndex >= 0 ? key.substring(dotIndex + 1) : key
           columns.push(outputKey)
-          cells[outputKey] = firstRow.cells[key]
+          // Lean buffered rows (see executeHashAggregate) carry `resolved` but no
+          // cell closures; cellThunk reads from `resolved` when present.
+          cells[outputKey] = cellThunk(firstRow, key)
         }
       }
     } else {
@@ -65,9 +68,11 @@ function projectAggregateColumns(selectColumns, group, context) {
  */
 function aggregateContextRow(group, aggregateRow) {
   const baseRow = group[0] ?? { columns: [], cells: {} }
+  // Lean buffered rows carry `resolved` but no cell closures; rowCells rebuilds
+  // base-column cells from `resolved` for this one row per group (O(groups)).
   return {
     columns: [...baseRow.columns, ...aggregateRow.columns],
-    cells: { ...baseRow.cells, ...aggregateRow.cells },
+    cells: { ...rowCells(baseRow), ...aggregateRow.cells },
   }
 }
 
@@ -84,7 +89,11 @@ export function executeHashAggregate(plan, context) {
     columns: selectColumnNames(plan.columns, child.columns),
     maxRows: child.maxRows,
     async *rows() {
-      // Collect all rows
+      // Collect all rows. GROUP BY buffers its whole input; for already
+      // materialized rows (`resolved` present) keep only the plain object and
+      // drop the O(columns) per-column cell closures, so the buffer holds row
+      // data, not N sets of closures. Aggregates and group keys read these lean
+      // rows straight from `resolved`. Rows without `resolved` are kept as-is.
       /** @type {AsyncRow[]} */
       const allRows = []
       let collectCount = 0
@@ -93,7 +102,7 @@ export function executeHashAggregate(plan, context) {
           await yieldToEventLoop()
           if (context.signal?.aborted) return
         }
-        allRows.push(row)
+        allRows.push(row.resolved ? { columns: row.columns, cells: {}, resolved: row.resolved } : row)
       }
       context.signal?.throwIfAborted()
 
@@ -143,9 +152,13 @@ export function executeHashAggregate(plan, context) {
       /** @type {{ row: AsyncRow, rows: AsyncRow[], outputRow: AsyncRow }[]} */
       const aggregateRows = []
 
+      // The context row (base columns + aggregate aliases) is only needed for
+      // HAVING and grouped ORDER BY; skip building it otherwise so a plain
+      // GROUP BY doesn't retain O(groups) extra rows.
+      const needContextRow = Boolean(plan.having) || Boolean(plan.orderBy?.length)
       for (const group of groups.values()) {
         const asyncRow = projectAggregateColumns(plan.columns, group, context)
-        const contextRow = aggregateContextRow(group, asyncRow)
+        const contextRow = needContextRow ? aggregateContextRow(group, asyncRow) : asyncRow
 
         // Apply HAVING filter
         if (plan.having) {
@@ -222,7 +235,7 @@ export function executeScalarAggregate(plan, context) {
         /** @type {AsyncRow} */
         const havingRow = {
           columns: [...baseRow.columns, ...asyncRow.columns],
-          cells: { ...baseRow.cells, ...asyncRow.cells },
+          cells: { ...rowCells(baseRow), ...asyncRow.cells },
         }
         const passes = await evaluateExpr({
           node: plan.having,

src/execute/execute.js

@@ -1,4 +1,4 @@
-import { memorySource } from '../backend/dataSource.js'
+import { cellThunk, hasCell, memorySource } from '../backend/dataSource.js'
 import { derivedAlias } from '../expression/alias.js'
 import { evaluateExpr } from '../expression/evaluate.js'
 import { parseSql } from '../parse/parse.js'
@@ -531,13 +531,14 @@ function executeProject(plan, context) {
 
         /** @type {AsyncCells} */
         const cells = {}
-        // Only safe to propagate resolved when every output column comes from
-        // the star branch. Derived expressions evaluate lazily and can't be
-        // pre-materialized here, and a partial resolved would make
-        // collect()/downstream identifier fast paths read undefined.
+        // Only safe to propagate resolved when every output column is actually
+        // pre-materialized here. Derived expressions evaluate lazily, and a
+        // partial resolved would make collect()/downstream identifier fast
+        // paths read undefined.
         const source = resolveable ? row.resolved : undefined
         /** @type {Record<string, SqlPrimitive> | undefined} */
         const resolved = source ? {} : undefined
+        let rowResolveable = Boolean(source)
 
         let colIdx = 0
         for (const col of plan.columns) {
@@ -547,7 +548,7 @@ function executeProject(plan, context) {
               if (prefix && !key.startsWith(prefix)) continue
               const dotIndex = key.indexOf('.')
               const outputKey = dotIndex >= 0 ? key.substring(dotIndex + 1) : key
-              cells[outputKey] = row.cells[key]
+              cells[outputKey] = cellThunk(row, key)
               if (resolved && source) resolved[outputKey] = source[key]
               colIdx++
             }
@@ -562,10 +563,15 @@ function executeProject(plan, context) {
             const id = col.expr
             const sourceName = id.prefix ? `${id.prefix}.${id.name}` : id.name
             const alias = columns[colIdx++]
-            if (sourceName in row.cells) {
-              cells[alias] = row.cells[sourceName]
-              if (resolved && source) resolved[alias] = source[sourceName]
+            if (hasCell(row, sourceName)) {
+              cells[alias] = cellThunk(row, sourceName)
+              // Only stay resolveable if the value is actually present in the
+              // source's resolved object; a cells-only key (e.g. a cached sort
+              // key) would otherwise propagate `undefined` into resolved.
+              if (resolved && source && sourceName in source) resolved[alias] = source[sourceName]
+              else rowResolveable = false
             } else {
+              rowResolveable = false
               const { expr } = col
               cells[alias] = () => evaluateExpr({
                 node: expr,
@@ -575,6 +581,7 @@ function executeProject(plan, context) {
               })
             }
           } else {
+            rowResolveable = false
             const alias = columns[colIdx++]
             cells[alias] = () => evaluateExpr({
               node: col.expr,
@@ -585,7 +592,7 @@ function executeProject(plan, context) {
           }
         }
 
-        yield { columns, cells, resolved }
+        yield { columns, cells, resolved: rowResolveable ? resolved : undefined }
       }
     },
   }

src/execute/join.js

@@ -1,3 +1,4 @@
+import { cellThunk, rowCells } from '../backend/dataSource.js'
 import { evaluateExpr } from '../expression/evaluate.js'
 import { keyify, maxBounds } from './utils.js'
 import { executePlan } from './execute.js'
@@ -314,12 +315,16 @@ export function executeHashJoin(plan, context) {
  */
 function mergeOuterRows(outerRow, leftRow, leftTable) {
   const columns = [...outerRow.columns]
+  // The enclosing outer row may be a lean buffered row (empty cells, values in
+  // `resolved`) when the correlated subquery runs under an outer ORDER BY or
+  // GROUP BY. rowCells rehydrates from `resolved` so outer columns stay readable;
+  // copy it since we mutate below.
   /** @type {AsyncCells} */
-  const cells = { ...outerRow.cells }
-  for (const [key, cell] of Object.entries(leftRow.cells)) {
+  const cells = { ...rowCells(outerRow) }
+  for (const key of leftRow.columns) {
     const alias = key.includes('.') ? key : `${leftTable}.${key}`
     if (!(alias in cells)) columns.push(alias)
-    cells[alias] = cell
+    cells[alias] = cellThunk(leftRow, key)
   }
   return { columns, cells }
 }
@@ -370,17 +375,17 @@ function mergeRows(leftRow, rightRow, leftTable, rightTable) {
   const cells = {}
 
   // Add left table columns with prefix
-  for (const [key, cell] of Object.entries(leftRow.cells)) {
+  for (const key of leftRow.columns) {
     const alias = key.includes('.') ? key : `${leftTable}.${key}`
     columns.push(alias)
-    cells[alias] = cell
+    cells[alias] = cellThunk(leftRow, key)
   }
 
   // Add right table columns with prefix
-  for (const [key, cell] of Object.entries(rightRow.cells)) {
+  for (const key of rightRow.columns) {
     const alias = key.includes('.') ? key : `${rightTable}.${key}`
     columns.push(alias)
-    cells[alias] = cell
+    cells[alias] = cellThunk(rightRow, key)
   }
 
   return { columns, cells }

src/execute/sort.js

@@ -1,3 +1,4 @@
+import { rowCells } from '../backend/dataSource.js'
 import { derivedAlias } from '../expression/alias.js'
 import { evaluateExpr } from '../expression/evaluate.js'
 import { executePlan } from './execute.js'
@@ -73,8 +74,13 @@ export async function sortEntriesByTerms({ entries, orderBy, context, cacheValue
           const idx = chunk[i]
           const value = values[i]
           evaluatedValues[idx][orderByIdx] = value
-          if (cacheValues && !(alias in entries[idx].row.cells)) {
-            entries[idx].row.cells[alias] = () => Promise.resolve(value)
+          // Cache the evaluated sort key back onto the row so the output
+          // projection can reuse it instead of recomputing (e.g. an ORDER BY UDF
+          // also referenced in SELECT). Skip real columns (read directly) and
+          // rows with no cells map.
+          const entryRow = entries[idx].row
+          if (cacheValues && entryRow.cells && !entryRow.columns.includes(alias) && !(alias in entryRow.cells)) {
+            entryRow.cells[alias] = () => Promise.resolve(value)
           }
         }
         start += chunk.length
@@ -129,12 +135,22 @@ export function executeSort(plan, context) {
     numRows: child.numRows,
     maxRows: child.maxRows,
     async *rows() {
-      // Buffer all rows
+      // ORDER BY must buffer its whole input before it can emit. For rows that
+      // are already fully materialized (`resolved` present), keep only the plain
+      // object plus a fresh empty cells map, dropping the O(columns) per-column
+      // cell closures: the buffer then holds row data, not N sets of closures.
+      // The empty cells map still lets the sort cache derived sort keys. Rows
+      // without `resolved` (e.g. derived expressions) are kept as-is so their
+      // lazy cells still work.
       /** @type {AsyncRow[]} */
       const rows = []
       for await (const row of child.rows()) {
         if (context.signal?.aborted) return
-        rows.push(row)
+        if (row.resolved) {
+          rows.push({ columns: row.columns, cells: {}, resolved: row.resolved })
+        } else {
+          rows.push(row)
+        }
       }
 
       const sortedRows = await sortEntriesByTerms({
@@ -144,9 +160,17 @@ export function executeSort(plan, context) {
         cacheValues: true,
       })
 
-      // Yield sorted rows
+      // Rebuild full cell closures for lean rows only at emit time, one row at a
+      // time, so downstream consumers get the normal cells interface without the
+      // buffer ever holding N sets of closures. A buffered row is lean exactly
+      // when it carries `resolved` (non-materialized rows are kept as-is above).
+      // Carry over any cached derived sort-key cells added during the sort.
       for (const { row } of sortedRows) {
-        yield row
+        if (!row.resolved) {
+          yield row
+          continue
+        }
+        yield { columns: row.columns, cells: rowCells(row), resolved: row.resolved }
       }
     },
   }

src/execute/utils.js

@@ -1,3 +1,5 @@
+import { readCell } from '../backend/dataSource.js'
+
 /**
  * @import { AsyncRow, OrderByItem, QueryResults, SqlPrimitive } from '../types.js'
  */
@@ -80,7 +82,7 @@ export async function collect(results) {
   }
 
   return Promise.all(rows.map(async asyncRow => {
-    const values = await Promise.all(asyncRow.columns.map(k => asyncRow.cells[k]()))
+    const values = await Promise.all(asyncRow.columns.map(k => readCell(asyncRow, k)))
     /** @type {Record<string, SqlPrimitive>} */
     const item = {}
     for (let i = 0; i < asyncRow.columns.length; i++) {
@@ -185,6 +187,6 @@ export function keyify(...values) {
  * @returns {Promise<string | number | bigint | boolean>}
  */
 export function stableRowKey(row) {
-  return Promise.all(row.columns.map(k => row.cells[k]()))
+  return Promise.all(row.columns.map(k => readCell(row, k)))
     .then(values => keyify(...values))
 }

src/execute/window.js

@@ -1,3 +1,4 @@
+import { rowCells } from '../backend/dataSource.js'
 import { evaluateExpr } from '../expression/evaluate.js'
 import { executePlan } from './execute.js'
 import { compareForTerm, keyify } from './utils.js'
@@ -44,7 +45,7 @@ export function executeWindow(plan, context) {
             await yieldToEventLoop()
             if (context.signal?.aborted) return
           }
-          const cells = { ...row.cells }
+          const cells = { ...rowCells(row) }
           for (const w of plan.windows) {
             const value = i
             cells[w.alias] = () => Promise.resolve(value)
@@ -91,7 +92,7 @@ export function executeWindow(plan, context) {
           if (context.signal?.aborted) return
         }
         const row = rows[i]
-        const cells = { ...row.cells }
+        const cells = { ...rowCells(row) }
         for (let w = 0; w < plan.windows.length; w++) {
           const { alias } = plan.windows[w]
           const value = windowValues[w][i]