Change aggregate operator to use array row data for keys (#303)

koralium · Jan 13, 2024 · 9f7d7f3 · 9f7d7f3
1 parent 0c02e8f
commit 9f7d7f3
Showing 1 changed file with 4 additions and 8 deletions.
diff --git a/src/FlowtideDotNet.Core/Operators/Aggregate/AggregateOperator.cs b/src/FlowtideDotNet.Core/Operators/Aggregate/AggregateOperator.cs
@@ -274,16 +274,12 @@ public override async IAsyncEnumerable<StreamEventBatch> OnRecieve(StreamEventBa
                 RowEvent? key = default;
                 if (groupExpressions != null)
                 {
-                    _flexBufferNewValue.NewObject();
-                    var vectorStart = _flexBufferNewValue.StartVector();
-                    foreach (var groupExpr in groupExpressions)
+                    FlxValue[] newVector = new FlxValue[groupExpressions.Count];
+                    for (int i = 0; i < groupExpressions.Count; i++)
                     {
-                        var result = groupExpr(e);
-                        _flexBufferNewValue.Add(result);
+                        newVector[i] = groupExpressions[i](e);
                     }
-                    _flexBufferNewValue.EndVector(vectorStart, false, false);
-                    var keyBytes = _flexBufferNewValue.Finish();
-                    key = new RowEvent(e.Weight, 0, new CompactRowData(keyBytes));
+                    key = new RowEvent(e.Weight, 0, new ArrayRowData(newVector));
 
                     // Store the key in the temporary tree
                     await _temporaryTree.RMW(key.Value, default, (_, current, exist) =>