feat: Simplify Barrage Viewport Table Updates

engine/table/src/main/java/io/deephaven/engine/table/impl/remote/ConstructSnapshot.java

@@ -618,7 +618,7 @@ public static BarrageMessage constructBackplaneSnapshotInPositionSpace(
             final long clockStep =
                     callDataSnapshotFunction(System.identityHashCode(logIdentityObject), control, doSnapshot);
             final BarrageMessage snapshot = snapshotMsg.getValue();
-            snapshot.step = snapshot.firstSeq = snapshot.lastSeq = clockStep;
+            snapshot.firstSeq = snapshot.lastSeq = clockStep;
             return snapshot;
         }
     }

engine/table/src/main/java/io/deephaven/engine/table/impl/util/BarrageMessage.java

@@ -45,7 +45,7 @@ public static class AddColumnData {
 
     public long firstSeq = -1;
     public long lastSeq = -1;
-    public long step = -1;
+    /** The size of the table after this update. -1 if unknown. */
     public long tableSize = -1;
 
     public boolean isSnapshot;

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageStreamGeneratorImpl.java

@@ -1119,9 +1119,12 @@ private int appendModColumns(final RecordBatchMessageView view, final long start
 
     public static abstract class ByteArrayGenerator {
         protected int len;
-        protected byte[] raw;
+        protected volatile byte[] raw;
 
-        protected int addToFlatBuffer(final FlatBufferBuilder builder) {
+        protected abstract void ensureComputed() throws IOException;
+
+        protected int addToFlatBuffer(final FlatBufferBuilder builder) throws IOException {
+            ensureComputed();
             return builder.createByteVector(raw, 0, len);
         }
     }
@@ -1131,22 +1134,31 @@ public static class RowSetGenerator extends ByteArrayGenerator implements SafeCl
 
         public RowSetGenerator(final RowSet rowSet) throws IOException {
             this.original = rowSet.copy();
-            try (final ExposedByteArrayOutputStream baos = new ExposedByteArrayOutputStream();
-                    final LittleEndianDataOutputStream oos = new LittleEndianDataOutputStream(baos)) {
-                ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, rowSet);
-                oos.flush();
-                raw = baos.peekBuffer();
-                len = baos.size();
-            }
         }
 
         @Override
         public void close() {
             original.close();
         }
 
-        public DrainableByteArrayInputStream getInputStream() {
-            return new DrainableByteArrayInputStream(raw, 0, len);
+        protected void ensureComputed() throws IOException {
+            if (raw != null) {
+                return;
+            }
+
+            synchronized (this) {
+                if (raw != null) {
+                    return;
+                }
+
+                try (final ExposedByteArrayOutputStream baos = new ExposedByteArrayOutputStream();
+                     final LittleEndianDataOutputStream oos = new LittleEndianDataOutputStream(baos)) {
+                    ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, original);
+                    oos.flush();
+                    len = baos.size();
+                    raw = baos.peekBuffer();
+                }
+            }
         }
 
         /**
@@ -1158,6 +1170,7 @@ public DrainableByteArrayInputStream getInputStream() {
          */
         protected int addToFlatBuffer(final RowSet viewport, final FlatBufferBuilder builder) throws IOException {
             if (original.subsetOf(viewport)) {
+                ensureComputed();
                 return addToFlatBuffer(builder);
             }
 
@@ -1177,11 +1190,27 @@ protected int addToFlatBuffer(final RowSet viewport, final FlatBufferBuilder bui
     }
 
     public static class BitSetGenerator extends ByteArrayGenerator {
+        private final BitSet original;
+
         public BitSetGenerator(final BitSet bitset) {
-            BitSet original = bitset == null ? new BitSet() : bitset;
-            this.raw = original.toByteArray();
-            final int nBits = original.previousSetBit(Integer.MAX_VALUE - 1) + 1;
-            this.len = (int) ((long) nBits + 7) / 8;
+            original = bitset == null ? new BitSet() : (BitSet) bitset.clone();
+        }
+
+        @Override
+        protected void ensureComputed() {
+            if (raw != null) {
+                return;
+            }
+
+            synchronized (this) {
+                if (raw != null) {
+                    return;
+                }
+
+                final int nBits = original.previousSetBit(Integer.MAX_VALUE - 1) + 1;
+                len = (int) ((long) nBits + 7) / 8;
+                raw = original.toByteArray();
+            }
         }
     }
 
@@ -1190,37 +1219,49 @@ public static class RowSetShiftDataGenerator extends ByteArrayGenerator {
 
         public RowSetShiftDataGenerator(final RowSetShiftData shifted) throws IOException {
             original = shifted;
+        }
 
-            final RowSetBuilderSequential sRangeBuilder = RowSetFactory.builderSequential();
-            final RowSetBuilderSequential eRangeBuilder = RowSetFactory.builderSequential();
-            final RowSetBuilderSequential destBuilder = RowSetFactory.builderSequential();
+        protected void ensureComputed() throws IOException {
+            if (raw != null) {
+                return;
+            }
 
-            if (shifted != null) {
-                for (int i = 0; i < shifted.size(); ++i) {
-                    long s = shifted.getBeginRange(i);
-                    final long dt = shifted.getShiftDelta(i);
+            synchronized (this) {
+                if (raw != null) {
+                    return;
+                }
 
-                    if (dt < 0 && s < -dt) {
-                        s = -dt;
-                    }
+                final RowSetBuilderSequential sRangeBuilder = RowSetFactory.builderSequential();
+                final RowSetBuilderSequential eRangeBuilder = RowSetFactory.builderSequential();
+                final RowSetBuilderSequential destBuilder = RowSetFactory.builderSequential();
+
+                if (original != null) {
+                    for (int i = 0; i < original.size(); ++i) {
+                        long s = original.getBeginRange(i);
+                        final long dt = original.getShiftDelta(i);
+
+                        if (dt < 0 && s < -dt) {
+                            s = -dt;
+                        }
 
-                    sRangeBuilder.appendKey(s);
-                    eRangeBuilder.appendKey(shifted.getEndRange(i));
-                    destBuilder.appendKey(s + dt);
+                        sRangeBuilder.appendKey(s);
+                        eRangeBuilder.appendKey(original.getEndRange(i));
+                        destBuilder.appendKey(s + dt);
+                    }
                 }
-            }
 
-            try (final RowSet sRange = sRangeBuilder.build();
-                    final RowSet eRange = eRangeBuilder.build();
-                    final RowSet dest = destBuilder.build();
-                    final ExposedByteArrayOutputStream baos = new ExposedByteArrayOutputStream();
-                    final LittleEndianDataOutputStream oos = new LittleEndianDataOutputStream(baos)) {
-                ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, sRange);
-                ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, eRange);
-                ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, dest);
-                oos.flush();
-                raw = baos.peekBuffer();
-                len = baos.size();
+                try (final RowSet sRange = sRangeBuilder.build();
+                        final RowSet eRange = eRangeBuilder.build();
+                        final RowSet dest = destBuilder.build();
+                        final ExposedByteArrayOutputStream baos = new ExposedByteArrayOutputStream();
+                        final LittleEndianDataOutputStream oos = new LittleEndianDataOutputStream(baos)) {
+                    ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, sRange);
+                    ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, eRange);
+                    ExternalizableRowSetUtils.writeExternalCompressedDeltas(oos, dest);
+                    oos.flush();
+                    len = baos.size();
+                    raw = baos.peekBuffer();
+                }
             }
         }
     }

server/src/main/java/io/deephaven/server/barrage/BarrageMessageProducer.java

@@ -1413,7 +1413,7 @@ private void updateSubscriptionsSnapshotAndPropagate() {
             }
 
             // prepare updates to propagate
-            final long maxStep = snapshot != null ? snapshot.step : Long.MAX_VALUE;
+            final long maxStep = snapshot != null ? snapshot.firstSeq : Long.MAX_VALUE;
 
             int deltaSplitIdx = pendingDeltas.size();
             for (; deltaSplitIdx > 0; --deltaSplitIdx) {
@@ -2037,6 +2037,7 @@ final class ColumnInfo {
         propagationRowSet.remove(downstream.rowsRemoved);
         downstream.shifted.apply(propagationRowSet);
         propagationRowSet.insert(downstream.rowsAdded);
+        downstream.tableSize = propagationRowSet.size();
 
         return downstream;
     }

server/src/main/java/io/deephaven/server/hierarchicaltable/HierarchicalTableViewSubscription.java

@@ -14,6 +14,7 @@
 import io.deephaven.engine.rowset.RowSet;
 import io.deephaven.engine.rowset.RowSetFactory;
 import io.deephaven.engine.rowset.RowSetShiftData;
+import io.deephaven.engine.rowset.WritableRowSet;
 import io.deephaven.engine.table.ColumnDefinition;
 import io.deephaven.engine.table.TableUpdate;
 import io.deephaven.engine.table.TableUpdateListener;
@@ -92,7 +93,7 @@ HierarchicalTableViewSubscription create(
     // region Guarded by snapshot lock
     private BitSet columns;
     private RowSet rows;
-    private long lastExpandedSize;
+    private final WritableRowSet prevKeyspaceViewportRows = RowSetFactory.empty();
     // endregion Guarded by snapshot lock
 
     private enum State {
@@ -282,16 +283,16 @@ private void process() {
                 return;
             }
             try {
-                lastExpandedSize = buildAndSendSnapshot(streamGeneratorFactory, listener, subscriptionOptions, view,
-                        this::recordSnapshotNanos, this::recordWriteMetrics, columns, rows, lastExpandedSize);
+                buildAndSendSnapshot(streamGeneratorFactory, listener, subscriptionOptions, view,
+                        this::recordSnapshotNanos, this::recordWriteMetrics, columns, rows, prevKeyspaceViewportRows);
             } catch (Exception e) {
                 GrpcUtil.safelyError(listener, errorTransformer.transform(e));
                 state = State.Done;
             }
         }
     }
 
-    private static long buildAndSendSnapshot(
+    private static void buildAndSendSnapshot(
             @NotNull final BarrageStreamGenerator.Factory streamGeneratorFactory,
             @NotNull final StreamObserver<BarrageStreamGenerator.MessageView> listener,
             @NotNull final BarrageSubscriptionOptions subscriptionOptions,
@@ -300,7 +301,7 @@ private static long buildAndSendSnapshot(
             @NotNull final BarragePerformanceLog.WriteMetricsConsumer writeMetricsConsumer,
             @NotNull final BitSet columns,
             @NotNull final RowSet rows,
-            final long lastExpandedSize) {
+            @NotNull final WritableRowSet prevKeyspaceViewportRows) {
         // 1. Grab some schema and snapshot information
         final List<ColumnDefinition<?>> columnDefinitions =
                 view.getHierarchicalTable().getAvailableColumnDefinitions();
@@ -322,18 +323,20 @@ private static long buildAndSendSnapshot(
                 columns, rows, destinations);
         snapshotNanosConsumer.accept(System.nanoTime() - snapshotStartNanos);
 
+        // note that keyspace is identical to position space for HierarchicalTableView snapshots
+        final RowSet snapshotRows = RowSetFactory.flat(expandedSize);
+        final RowSet keyspaceViewportRows = rows.intersect(snapshotRows);
+
         // 4. Make and populate a BarrageMessage
         final BarrageMessage barrageMessage = new BarrageMessage();
+        // We don't populate firstSeq, or lastSeq debugging information; they are not relevant to this use case.
+
         barrageMessage.isSnapshot = true;
-        // We don't populate length, snapshotRowSet, snapshotRowSetIsReversed, or snapshotColumns; they are only set by
-        // the client.
-        // We don't populate step, firstSeq, or lastSeq debugging information; they are not relevant to this use case.
-
-        barrageMessage.rowsAdded = RowSetFactory.flat(expandedSize);
-        barrageMessage.rowsIncluded = RowSetFactory.fromRange(rows.firstRowKey(),
-                Math.min(barrageMessage.rowsAdded.lastRowKey(), rows.lastRowKey()));
-        barrageMessage.rowsRemoved = RowSetFactory.flat(lastExpandedSize);
+        barrageMessage.rowsAdded = snapshotRows;
+        barrageMessage.rowsIncluded = keyspaceViewportRows;
+        barrageMessage.rowsRemoved = RowSetFactory.empty();
         barrageMessage.shifted = RowSetShiftData.EMPTY;
+        barrageMessage.tableSize = expandedSize;
 
         barrageMessage.addColumnData = new BarrageMessage.AddColumnData[numAvailableColumns];
         for (int ci = 0, di = 0; ci < numAvailableColumns; ++ci) {
@@ -357,15 +360,16 @@ private static long buildAndSendSnapshot(
         // 5. Send the BarrageMessage
         try (final BarrageStreamGenerator streamGenerator =
                 streamGeneratorFactory.newGenerator(barrageMessage, writeMetricsConsumer)) {
-            // Note that we're always specifying "isInitialSnapshot=true". This is to provoke the subscription view to
-            // send the added rows on every snapshot, since (1) our added rows are flat, and thus cheap to send, and
-            // (2) we're relying on added rows to signal the full expanded size to the client.
-            GrpcUtil.safelyOnNext(listener,
-                    streamGenerator.getSubView(subscriptionOptions, true, true, rows, false, rows, rows, columns));
+            // initialSnapshot flag is ignored for non-growing viewports
+            final boolean initialSnapshot = false;
+            final boolean isFullSubscription = false;
+            GrpcUtil.safelyOnNext(listener, streamGenerator.getSubView(
+                    subscriptionOptions, initialSnapshot, isFullSubscription, rows, false,
+                    prevKeyspaceViewportRows, keyspaceViewportRows, columns));
         }
 
-        // 6. Let the caller know what the expanded size was
-        return expandedSize;
+        prevKeyspaceViewportRows.clear();
+        prevKeyspaceViewportRows.insert(keyspaceViewportRows);
     }
 
     public void setViewport(