Implement aggregate functions with GROUP BY support - #6
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: dd1e0ed15e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| for key in &binding_keys { | ||
| let batch = bindings.get(key).unwrap(); | ||
| let aligned_columns: Vec<ArrayRef> = batch | ||
| .columns() | ||
| .iter() |
There was a problem hiding this comment.
This loop applies src_take (built from src_var row indices) to every binding in scope, not just the ones correlated with src_var. In queries that have additional bound variables with fewer rows (for example, an extra NodeScan in the same match block), take will receive out-of-range indices and fail the query at runtime. The alignment step should only rewrite correlated bindings (or at minimum guard on compatible row counts).
Useful? React with 👍 / 👎.
| match &proj.expr { | ||
| IRExpr::Aggregate { func, arg } => { | ||
| let (name, col) = evaluate_projection(bindings, arg, params)?; | ||
| let alias = proj.alias.as_deref().unwrap_or(&name); |
There was a problem hiding this comment.
For aggregate projections without an explicit alias, this uses the aggregate argument's name as the output column name (count($p) becomes p) instead of the aggregate expression name (count). That makes unnamed aggregate columns inconsistent and can break downstream lookups/order clauses that expect the default aggregate name unless users manually alias every aggregate.
Useful? React with 👍 / 👎.
| let mut sum = None; | ||
| for &i in group { | ||
| if !arr.is_null(i) { | ||
| *sum.get_or_insert(0) += arr.value(i) as i64; |
There was a problem hiding this comment.
The integer SUM path casts every value to i64, including UInt64. Any input above i64::MAX will wrap/truncate and produce incorrect (potentially negative) sums, so valid unsigned counters can be silently corrupted. UInt64 needs a dedicated accumulation type/path instead of forcing i64.
Useful? React with 👍 / 👎.
| _ => { | ||
| // sum/min/max return null for empty input | ||
| fields.push(Field::new(name, DataType::Int64, true)); | ||
| columns.push(Arc::new(Int64Array::from(vec![None as Option<i64>])) as ArrayRef); |
There was a problem hiding this comment.
The empty-input fallback hardcodes nullable Int64 for the non-COUNT/AVG aggregates, so sum/min/max return a different type when there are zero input rows. This makes result schemas data-dependent (e.g., sum(Float64) yields Int64 only on empty input), which can break typed consumers and schema assumptions.
Useful? React with 👍 / 👎.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
f8ab772 to
f6078d9
Compare
aaltshuler
left a comment
aaltshuler
left a comment
There was a problem hiding this comment.
Three issues from the aggregate patch need attention:
-
Variable projections no longer match their declared result type. The new
IRExpr::Variableprojection path returns the<binding>.idUTF-8 column from the wide batch, but the compiler still resolvesExpr::VariabletoResolvedType::Node(...)and advertises a full struct viaresolved_type_to_field_shape. Queries likereturn { $p }will therefore typecheck as a node-shaped result and execute as a string ID column instead. -
sum(...)output type diverges from the compiler contract.compute_sumnow always materializes aFloat64Array, but projection typing still uses the aggregate argument's type for every aggregate other thancountandavg. A query such asreturn { sum($p.age) }will therefore be typed asInt32/Int64at compile time and returned asFloat64at runtime. -
min/maxon strings are still rejected by typechecking. The executor addsUtf8handling incompute_min_max, but T8 inresolve_expr_typestill groupsminandmaxwithsumandavgunder a numeric-only check. As written,min($n.name)andmax($n.name)still fail in the compiler even though the runtime support was added.
Add runtime support for aggregate functions (count, sum, avg, min, max) with GROUP BY semantics, built on a single wide RecordBatch that eliminates correlation tracking by construction. Execution engine (exec/query.rs): - Replace HashMap<String, RecordBatch> with Option<RecordBatch> where columns are prefixed as <variable>.<property> - NodeScan prefixes columns and cross-joins with existing batch - Expand collects (src_row, dst_id) pairs, takes wide batch rows, appends prefixed destination columns via hconcat - Filter applies single mask to entire wide batch - AntiJoin: fast-path returns BooleanArray mask; slow-path slices one row for inner pipeline execution Projection engine (exec/projection.rs): - aggregate_return groups rows by non-aggregate key columns using length-prefixed string encoding, computes per-group aggregates - SUM accumulates into f64 to avoid integer overflow - MIN/MAX support both numeric and string types - Empty input returns count=0, others=null Compiler (typecheck.rs): - T8: split MIN/MAX from SUM/AVG — allow string arguments - T9: non-aggregate expressions in aggregate queries must be property accesses or variables - SUM type inference returns Float64 (matching runtime) Tests: 8 new integration tests covering grouped count, global count, sum/avg/min/max per company, aggregate+order+limit, string min/max, multi-hop aggregates, and edge cases. https://claude.ai/code/session_019o5NRyYomgETFyd7hpiLey
8b4a24a to
351610d
Compare
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit 351610d. Configure here.
| _ => { | ||
| fields.push(Field::new(name, DataType::Float64, true)); | ||
| columns.push(Arc::new(Float64Array::from(vec![None as Option<f64>])) as ArrayRef); | ||
| } |
There was a problem hiding this comment.
Empty aggregate result uses wrong type for MIN/MAX
Medium Severity
build_empty_aggregate_result unconditionally uses DataType::Float64 for all non-COUNT aggregates (the _ wildcard arm), but MIN/MAX preserve the original column type in both the typecheck (infer_projection_field resolves to the arg type) and execution (compute_min_max builds typed arrays matching the input). This means a global min($p.age) on an empty table returns a Float64 NULL instead of Int32 NULL, and min($p.name) returns Float64 instead of Utf8 — a schema inconsistency that could cause downstream type-mismatch errors or client deserialization failures.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 351610d. Configure here.
…tAdapter scope Addresses the remaining valid review findings (doc/CI accuracy, no behavior change): #2: s3_battery_holds is bucket-gated and lives in the `dst` target, which no CI job ran — so the DST S3 battery never executed despite MATRIX/testing.md marking S3 ✅. Add a 'RustFS DST white-box battery' step to the rustfs_integration job (same grep-guard against a vacuous pass) and add the dst test paths + omnigraph-dst crate to the RustFS change filter so a DST change triggers the job. #6: the MATRIX 'open / recovery sweep' row said unsampled while the same harness ships the Reopen walk op + the dst_recovery failpoint cells (roll-forward + the #296 concurrent-opens cell). Mark it ✅ with the actual coverage. #7: tighten the FaultAdapter wording (module doc + the seeded_op_loop_with_cas_faults comment) — the seam is the StorageAdapter conditional TEXT-OBJECT write (sidecars/schema-staging/cluster-state), NOT the Lance manifest-publish CAS (MergeInsertBuilder), which dst_recovery covers via failpoints. No code change.
3 participants
Summary
This PR adds support for aggregate functions (COUNT, SUM, AVG, MIN, MAX) with GROUP BY semantics to the query execution engine. It also refactors the graph traversal (Expand) operation to maintain row alignment across correlated bindings, which is essential for correct aggregate computation.
Key Changes
Aggregate Function Support
aggregate_return()function to handle queries with aggregate expressionsCOUNT: counts non-null values per groupSUM: sums numeric values (Int32/64, UInt32/64, Float32/64)AVG: computes average as Float64MIN/MAX: finds extrema with support for numeric and string typesRow Alignment for Expand Operations
execute_expand()to mutate bindings in-place and maintain row alignmentResult<RecordBatch>toResult<()>with direct binding updatesiacross all bindings represents one matched tuplesrc_indices,dst_indices) to perform alignedtakeoperations on all correlated bindingsFilter and AntiJoin Alignment
apply_filter()to filter all correlated bindings (same row count) to maintain alignmentexecute_anti_join()to filter all correlated bindings when removing rowsexecute_anti_join()that reconstructs the filter mask for other bindingsCompiler Changes
has_aggregateflag toQueryIRstruct to track if query contains aggregatesAggFuncimport to execution moduleTesting
aggregation.rs) covering:Implementation Details
\x1Funit separator) to handle multi-column grouping\x00NULLto distinguish from empty stringsevaluate_projection()for accessing node IDs in aggregateshttps://claude.ai/code/session_019o5NRyYomgETFyd7hpiLey
Note
Medium Risk
Adds new aggregation execution path and refactors the query pipeline to operate on a single wide
RecordBatchwith prefixed columns, changing traversal/antijoin/filter semantics and ordering resolution; this touches core query execution and could impact result correctness/performance.Overview
Enables aggregate queries (
count,sum,avg,min,max) with implicit GROUP BY semantics by adding an aggregate projection path that groups by non-aggregate return expressions and computes per-group results (including an empty-input global-aggregate case).Refactors execution from per-variable
bindingsto a single wide batch withvar.propertycolumn names: node scans are cross-joined,expandnow maintains row alignment by duplicating source rows per destination and horizontally concatenating prefixed destination columns, and filters/antijoins operate on the wide batch. Ordering is updated to resolve property accesses against either the wide batch or the aggregate result batch.Tightens compiler/typechecking for aggregates: enforces that non-aggregate projections in aggregate queries are valid group keys (only prop accesses or variables) and allows
min/maxover strings; result schema inference updatessum/avgtoFloat64andmin/maxto the argument type. Adds an aggregation-focused test suite and new fixture queries covering grouped/global aggregates plus ordering/limit.Reviewed by Cursor Bugbot for commit 351610d. Bugbot is set up for automated code reviews on this repo. Configure here.