JS API updated to specify viewport rows/columns when taking snapshots

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

@@ -56,6 +56,9 @@ public static class AddColumnData {
     // Ensure that we clean up only after all copies of the update are released.
     private volatile int refCount = 1;
 
+    // Underlying RecordBatch.length, visible for reading snapshots
+    public long length;
+
     // Field updater for refCount, so we can avoid creating an {@link java.util.concurrent.atomic.AtomicInteger} for
     // each instance.
     private static final AtomicIntegerFieldUpdater<BarrageMessage> REFERENCE_COUNT_UPDATER =

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/util/BarrageStreamReader.java

@@ -182,6 +182,7 @@ public BarrageMessage safelyParseFrom(final StreamReaderOptions options,
                 bodyParsed = true;
                 final int size = decoder.readRawVarint32();
                 final RecordBatch batch = (RecordBatch) header.header(new RecordBatch());
+                msg.length = batch.length();
 
                 // noinspection UnstableApiUsage
                 try (final LittleEndianDataInputStream ois =

java-client/barrage/src/main/java/io/deephaven/client/impl/BarrageSnapshotImpl.java

@@ -118,7 +118,7 @@ public void onNext(final BarrageMessage barrageMessage) {
                 }
 
                 // override server-supplied data and regenerate local rowsets
-                barrageMessage.rowsAdded = RowSetFactory.flat(barrageMessage.addColumnData[0].data.size());
+                barrageMessage.rowsAdded = RowSetFactory.flat(barrageMessage.length);
                 barrageMessage.rowsIncluded = barrageMessage.rowsAdded.copy();
 
                 listener.handleBarrageMessage(barrageMessage);