Use builders for GroupBy initialization and updates

engine/table/src/main/java/io/deephaven/engine/table/impl/by/AggregationContext.java

@@ -234,10 +234,11 @@ ModifiedColumnSet[] getInputModifiedColumnSets(QueryTable input) {
      * initialization.
      *
      * @param resultTable The result {@link QueryTable} after initialization
+     * @param maxDestination The maximum destination slot created during initialization
      */
-    void propagateInitialStateToOperators(@NotNull final QueryTable resultTable) {
+    void propagateInitialStateToOperators(@NotNull final QueryTable resultTable, final int maxDestination) {
         for (final IterativeChunkedAggregationOperator operator : operators) {
-            operator.propagateInitialState(resultTable);
+            operator.propagateInitialState(resultTable, maxDestination);
         }
     }
 
@@ -266,7 +267,7 @@ UnaryOperator<ModifiedColumnSet>[] initializeRefreshing(@NotNull final QueryTabl
     /**
      * Allow all operators to reset any per-step internal state. Note that the arguments to this method should not be
      * mutated in any way.
-     *
+     * 
      * @param upstream The upstream {@link TableUpdateImpl}
      * @param startingDestinationsCount The number of used destinations at the beginning of this step
      */
@@ -280,7 +281,7 @@ void resetOperatorsForStep(@NotNull final TableUpdate upstream, final int starti
      * Allow all operators to perform any internal state keeping needed for destinations that were added (went from 0
      * keys to &gt 0), removed (went from &gt 0 keys to 0), or modified (keys added or removed, or keys modified) by
      * this iteration. Note that the arguments to this method should not be mutated in any way.
-     *
+     * 
      * @param downstream The downstream {@link TableUpdate} (which does <em>not</em> have its {@link ModifiedColumnSet}
      *        finalized yet)
      * @param newDestinations New destinations added on this update

engine/table/src/main/java/io/deephaven/engine/table/impl/by/ByteStreamSortedFirstOrLastChunkedOperator.java

@@ -138,7 +138,7 @@ private boolean addChunk(@NotNull final ByteChunk<? extends Values> values,
     }
 
     @Override
-    public void propagateInitialState(@NotNull final QueryTable resultTable) {
+    public void propagateInitialState(@NotNull final QueryTable resultTable, int startingDestinationsCount) {
         copyStreamToResult(resultTable.getRowSet());
         redirections = null;
     }

engine/table/src/main/java/io/deephaven/engine/table/impl/by/CharStreamSortedFirstOrLastChunkedOperator.java

@@ -133,7 +133,7 @@ private boolean addChunk(@NotNull final CharChunk<? extends Values> values,
     }
 
     @Override
-    public void propagateInitialState(@NotNull final QueryTable resultTable) {
+    public void propagateInitialState(@NotNull final QueryTable resultTable, int startingDestinationsCount) {
         copyStreamToResult(resultTable.getRowSet());
         redirections = null;
     }

engine/table/src/main/java/io/deephaven/engine/table/impl/by/ChunkedOperatorAggregationHelper.java

@@ -219,7 +219,7 @@ private static QueryTable aggregation(
 
         // Construct the result table
         final QueryTable result = new QueryTable(resultRowSet, resultColumnSourceMap);
-        ac.propagateInitialStateToOperators(result);
+        ac.propagateInitialStateToOperators(result, outputPosition.intValue());
 
         if (input.isRefreshing()) {
             assert keyColumnsCopied != null;
@@ -1593,7 +1593,7 @@ private static QueryTable staticGroupedAggregation(QueryTable withView, String k
 
         final QueryTable result = new QueryTable(RowSetFactory.flat(responsiveGroups).toTracking(),
                 resultColumnSourceMap);
-        ac.propagateInitialStateToOperators(result);
+        ac.propagateInitialStateToOperators(result, responsiveGroups);
 
         final ReverseLookupListener rll = ReverseLookupListener.makeReverseLookupListenerWithSnapshot(result, keyName);
         ac.setReverseLookupFunction(k -> (int) rll.get(k));
@@ -1957,7 +1957,8 @@ private static QueryTable noKeyAggregation(
 
         final QueryTable result = new QueryTable(RowSetFactory.flat(initialResultSize).toTracking(),
                 resultColumnSourceMap);
-        ac.propagateInitialStateToOperators(result);
+        // always will create one result for zerokey
+        ac.propagateInitialStateToOperators(result, 1);
 
         if (table.isRefreshing()) {
             ac.startTrackingPrevValues();
@@ -2412,7 +2413,7 @@ public void addRange(final long firstRowKey, final long lastRowKey) {
      * We also know that we will only modify the rows that existed when we start, so that we can clamp the maximum key
      * for the builder to the maximum output position without loss of fidelity.
      */
-    private static class BitmapRandomBuilder implements RowSetBuilderRandom {
+    public static class BitmapRandomBuilder implements RowSetBuilderRandom {
 
         /**
          * An upper bound on {@code lastUsed}. That is, the highest bit index that may be used in {@code bitset}.
@@ -2434,7 +2435,7 @@ private static class BitmapRandomBuilder implements RowSetBuilderRandom {
          */
         long[] bitset;
 
-        private BitmapRandomBuilder(int maxKey) {
+        public BitmapRandomBuilder(int maxKey) {
             this.maxKey = maxKey;
         }
 

engine/table/src/main/java/io/deephaven/engine/table/impl/by/DoubleStreamSortedFirstOrLastChunkedOperator.java

@@ -138,7 +138,7 @@ private boolean addChunk(@NotNull final DoubleChunk<? extends Values> values,
     }
 
     @Override
-    public void propagateInitialState(@NotNull final QueryTable resultTable) {
+    public void propagateInitialState(@NotNull final QueryTable resultTable, int startingDestinationsCount) {
         copyStreamToResult(resultTable.getRowSet());
         redirections = null;
     }

engine/table/src/main/java/io/deephaven/engine/table/impl/by/FloatStreamSortedFirstOrLastChunkedOperator.java

@@ -138,7 +138,7 @@ private boolean addChunk(@NotNull final FloatChunk<? extends Values> values,
     }
 
     @Override
-    public void propagateInitialState(@NotNull final QueryTable resultTable) {
+    public void propagateInitialState(@NotNull final QueryTable resultTable, int startingDestinationsCount) {
         copyStreamToResult(resultTable.getRowSet());
         redirections = null;
     }

engine/table/src/main/java/io/deephaven/engine/table/impl/by/FormulaChunkedOperator.java

@@ -28,15 +28,14 @@
 import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.function.LongConsumer;
 import java.util.function.UnaryOperator;
 
 import static io.deephaven.engine.table.impl.sources.ArrayBackedColumnSource.BLOCK_SIZE;
 
 /**
  * An {@link IterativeChunkedAggregationOperator} used in the implementation of {@link Table#applyToAllBy}.
  */
-class FormulaChunkedOperator implements StateChangeRecorder, IterativeChunkedAggregationOperator {
+class FormulaChunkedOperator implements IterativeChunkedAggregationOperator {
 
     private final GroupByChunkedOperator groupBy;
     private final boolean delegateToBy;
@@ -101,16 +100,6 @@ class FormulaChunkedOperator implements StateChangeRecorder, IterativeChunkedAgg
         }
     }
 
-    @Override
-    public void startRecording(LongConsumer reincarnatedDestinationCallback, LongConsumer emptiedDestinationCallback) {
-        groupBy.startRecording(reincarnatedDestinationCallback, emptiedDestinationCallback);
-    }
-
-    @Override
-    public void finishRecording() {
-        groupBy.finishRecording();
-    }
-
     @Override
     public void addChunk(final BucketedContext bucketedContext, final Chunk<? extends Values> values,
             @NotNull final LongChunk<? extends RowKeys> inputRowKeys,
@@ -257,9 +246,9 @@ public void ensureCapacity(final long tableSize) {
     }
 
     @Override
-    public void propagateInitialState(@NotNull final QueryTable resultTable) {
+    public void propagateInitialState(@NotNull final QueryTable resultTable, int startingDestinationsCount) {
         if (delegateToBy) {
-            groupBy.propagateInitialState(resultTable);
+            groupBy.propagateInitialState(resultTable, startingDestinationsCount);
         }
 
         final Map<String, ? extends ColumnSource<?>> byResultColumns = groupBy.getResultColumns();