Stream aggregates with bounded memory instead of buffering all rows#49
Merged
Conversation
Aggregate executors buffered every input row before evaluating, pinning entire column batches (and whole files) in memory. Queries whose aggregates are all COUNT, COUNTIF, SUM, AVG, MIN, or MAX now accumulate incrementally in 4000-row chunks, retaining one detached representative row per group. Finalized values are substituted into select, HAVING, and ORDER BY expressions as literals, keyed by aggregate node identity so same-named aggregates with different FILTERs stay separate. Queries with other aggregates (MEDIAN, ARRAY_AGG, ...) or subqueries fall back to the buffered path. SUM(LENGTH(text)) over a 414MB parquet file drops from 1.3GB to 155MB.
The streaming aggregate path retained each group's first row as a representative, so a high-cardinality GROUP BY still held every input row. Track group key references during planning and substitute their values as literals alongside the aggregates; a representative row is only retained when an expression references columns outside the group keys (or SELECT *). Groups also finalize and yield one at a time when there is no ORDER BY, and FILTERed aggregate arguments are only evaluated for rows that pass the filter, matching buffered semantics. GROUP BY over 495k distinct keys now completes in a 256MB heap.
…terals The structural signature used JSON.stringify, which throws on BigInt literal values, so any aggregate query containing a bigint literal failed during streaming planning. The signature replacer now encodes bigints safely. Group key matching also treated bare and qualified identifiers with the same name as interchangeable. In a join, a bare identifier can resolve to a different column than a qualified group key, so substituting the key value changed results. Matching now requires exact structural equality; mixed qualification retains the representative row and evaluates the identifier against it, preserving buffered semantics.
Aggregates under short-circuited positions (AND/OR right sides, CASE branches past the first WHEN condition) may never be evaluated by the buffered path, so accumulating them eagerly could evaluate expressions the query never asks for. The streaming planner now falls back to buffered aggregation when an aggregate appears in such a position. Grouped ORDER BY previously evaluated every sort term for every group before sorting. Streaming results now go through sortEntriesByTerms, which evaluates later terms only within ties on earlier terms; sort entries carry pre-substituted per-group expressions so finalized aggregate values still replace aggregate calls.
…borts Streaming aggregation evaluates aggregate arguments eagerly, but projection pushdown prunes columns whose output cells are never read, so a subquery whose aggregate output goes unread crashed with ColumnNotFoundError. planStreamingAggregates now receives the child's columns and falls back to buffered aggregation when an aggregate references a column the child does not produce. IN value lists short-circuit once an earlier value matches, and the sorter evaluates later ORDER BY terms only within ties, so aggregates in those positions are now treated as lazy and fall back to buffered aggregation instead of accumulating eagerly. Aborting mid-accumulation now ends the row stream silently, matching the buffered aggregate path and the other executors; group keys over missing columns stay undefined instead of being coerced to null. COUNT/COUNTIF/SUM/AVG/MIN/MAX fold semantics move to a shared accumulator module used by both the streaming path and the scanColumn fast path, so the copies cannot drift. The streaming plan docs no longer overstate the memory bound for COUNT(DISTINCT ...).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
Both aggregate executors buffered every input AsyncRow before evaluating, so a high-cardinality GROUP BY or COUNT(DISTINCT) over ~500k rows exhausted memory even when the result was tiny. This is the engine side of hyparam/hypaware-server#9, where these queries OOM-killed the shared server:
SELECT message_id, count(*) FROM ai_gateway_messages GROUP BY message_idSELECT count(DISTINCT content_text) FROM ai_gateway_messagesSELECT count(DISTINCT session_id), count(DISTINCT conversation_id), ... FROM ai_gateway_messagesApproach
When every aggregate in a query folds into a constant-size accumulator (COUNT, COUNTIF, SUM, AVG, MIN, MAX — including DISTINCT and FILTER), input rows are consumed in 4000-row chunks and discarded:
SELECT *); otherwise memory is bounded by group count alone.Results
Repro of the hypaware-server crashes: 495k-row streaming source, client reads at most 10k rows,
--max-old-space-size=256:GROUP BY message_id(~495k groups)count(DISTINCT content_text)count(DISTINCT ...)The remaining crasher from that issue is plain unbounded
ORDER BY— that needs the caller to pass its row cap asLIMIT(activating the #48 top-k sort) and/or the #42 execution budget.All 1734 tests pass; lint and tsc clean.