feat: Simplify Barrage Viewport Table Updates

buildSrc/build.gradle

@@ -10,6 +10,7 @@ java {
 }
 
 repositories {
+    mavenLocal()
     mavenCentral()
     maven {
         url "https://plugins.gradle.org/m2/"

buildSrc/src/main/groovy/io.deephaven.repository-conventions.gradle

@@ -1,5 +1,12 @@
 repositories {
     mavenCentral()
+    mavenLocal()
+    maven {
+        url 'https://oss.sonatype.org/content/repositories/snapshots'
+        content {
+            includeGroup 'com.vertispan.flatbuffers'
+        }
+    }
     maven {
         url 'https://jitpack.io'
         content {

engine/table/src/main/java/io/deephaven/engine/table/impl/FlattenOperation.java

@@ -12,6 +12,7 @@
 import io.deephaven.engine.table.impl.sources.RedirectedColumnSource;
 import io.deephaven.engine.table.impl.util.*;
 import org.apache.commons.lang3.mutable.MutableObject;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.LinkedHashMap;
 import java.util.Map;
@@ -98,6 +99,34 @@ private void onUpdate(final TableUpdate upstream) {
         try (final RowSet prevRowSet = rowSet.copyPrev()) {
             downstream.removed = prevRowSet.invert(upstream.removed());
         }
+
+        if (newSize < prevSize) {
+            resultTable.getRowSet().writableCast().removeRange(newSize, prevSize - 1);
+        } else if (newSize > prevSize) {
+            resultTable.getRowSet().writableCast().insertRange(prevSize, newSize - 1);
+        }
+
+        downstream.shifted = computeFlattenedRowSetShiftData(downstream.removed(), downstream.added(), prevSize);
+        prevSize = newSize;
+        resultTable.notifyListeners(downstream);
+    }
+
+    /**
+     * Compute the shift data for a flattened row set given which rows were removed and which were added.
+     *
+     * @param removed the rows that were removed in the flattened pre-update key-space
+     * @param added the rows that were added in the flattened post-update key-space
+     * @param prevSize the size of the table before the update
+     * @return the shift data
+     */
+    public static RowSetShiftData computeFlattenedRowSetShiftData(
+            @NotNull final RowSet removed,
+            @NotNull final RowSet added,
+            final long prevSize) {
+        if (removed.isEmpty() && added.isEmpty()) {
+            return RowSetShiftData.EMPTY;
+        }
+
         final RowSetShiftData.Builder outShifted = new RowSetShiftData.Builder();
 
         // Helper to ensure that we can prime iterators and still detect the end.
@@ -110,8 +139,8 @@ private void onUpdate(final TableUpdate upstream) {
         };
 
         // Create our range iterators and prime them.
-        final MutableObject<RowSet.RangeIterator> rmIt = new MutableObject<>(downstream.removed().rangeIterator());
-        final MutableObject<RowSet.RangeIterator> addIt = new MutableObject<>(downstream.added().rangeIterator());
+        final MutableObject<RowSet.RangeIterator> rmIt = new MutableObject<>(removed.rangeIterator());
+        final MutableObject<RowSet.RangeIterator> addIt = new MutableObject<>(added.rangeIterator());
         updateIt.accept(rmIt);
         updateIt.accept(addIt);
 
@@ -163,14 +192,6 @@ private void onUpdate(final TableUpdate upstream) {
             outShifted.shiftRange(currMarker, prevSize - 1, currDelta);
         }
 
-        if (newSize < prevSize) {
-            resultTable.getRowSet().writableCast().removeRange(newSize, prevSize - 1);
-        } else if (newSize > prevSize) {
-            resultTable.getRowSet().writableCast().insertRange(prevSize, newSize - 1);
-        }
-
-        downstream.shifted = outShifted.build();
-        prevSize = newSize;
-        resultTable.notifyListeners(downstream);
+        return outShifted.build();
     }
 }

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

@@ -7,6 +7,7 @@
 import io.deephaven.annotations.BuildableStyle;
 import io.deephaven.barrage.flatbuf.BarrageSnapshotRequest;
 import io.deephaven.extensions.barrage.util.StreamReaderOptions;
+import io.deephaven.util.annotations.FinalDefault;
 import org.immutables.value.Value.Default;
 import org.immutables.value.Value.Immutable;
 
@@ -21,12 +22,11 @@ public static BarrageSnapshotOptions of(final io.deephaven.barrage.flatbuf.Barra
         if (options == null) {
             return builder().build();
         }
-        final byte mode = options.columnConversionMode();
         return builder()
                 .useDeephavenNulls(options.useDeephavenNulls())
-                .columnConversionMode(ColumnConversionMode.conversionModeFbToEnum(mode))
                 .batchSize(options.batchSize())
                 .maxMessageSize(options.maxMessageSize())
+                .previewListLengthLimit(options.previewListLengthLimit())
                 .build();
     }
 
@@ -65,27 +65,37 @@ public int maxMessageSize() {
 
     @Override
     @Default
-    public ColumnConversionMode columnConversionMode() {
-        return ColumnConversionMode.Stringify;
+    public int previewListLengthLimit() {
+        return 0;
     }
 
     public int appendTo(FlatBufferBuilder builder) {
         return io.deephaven.barrage.flatbuf.BarrageSnapshotOptions.createBarrageSnapshotOptions(
-                builder, ColumnConversionMode.conversionModeEnumToFb(columnConversionMode()), useDeephavenNulls(),
+                builder, useDeephavenNulls(),
                 batchSize(),
-                maxMessageSize());
+                maxMessageSize(),
+                previewListLengthLimit());
     }
 
     public interface Builder {
 
         Builder useDeephavenNulls(boolean useDeephavenNulls);
 
-        Builder columnConversionMode(ColumnConversionMode columnConversionMode);
+        /**
+         * Deprecated since 0.37.0 and is marked for removal. (our GWT artifacts do not yet support the attributes)
+         */
+        @FinalDefault
+        @Deprecated
+        default Builder columnConversionMode(ColumnConversionMode columnConversionMode) {
+            return this;
+        }
 
         Builder batchSize(int batchSize);
 
         Builder maxMessageSize(int messageSize);
 
+        Builder previewListLengthLimit(int previewListLengthLimit);
+
         BarrageSnapshotOptions build();
     }
 }

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

@@ -59,7 +59,7 @@ BarrageStreamGenerator newGenerator(
      * Obtain a Full-Subscription View of this StreamGenerator that can be sent to a single subscriber.
      *
      * @param options serialization options for this specific view
-     * @param isInitialSnapshot indicates whether or not this is the first snapshot for the listener
+     * @param isInitialSnapshot indicates whether this is the first snapshot for the listener
      * @return a MessageView filtered by the subscription properties that can be sent to that subscriber
      */
     MessageView getSubView(BarrageSubscriptionOptions options, boolean isInitialSnapshot);
@@ -68,15 +68,24 @@ BarrageStreamGenerator newGenerator(
      * Obtain a View of this StreamGenerator that can be sent to a single subscriber.
      *
      * @param options serialization options for this specific view
-     * @param isInitialSnapshot indicates whether or not this is the first snapshot for the listener
+     * @param isInitialSnapshot indicates whether this is the first snapshot for the listener
+     * @param isFullSubscription whether this is a full subscription (possibly a growing viewport)
      * @param viewport is the position-space viewport
      * @param reverseViewport is the viewport reversed (relative to end of table instead of beginning)
+     * @param keyspaceViewportPrev is the key-space viewport in prior to applying the update
      * @param keyspaceViewport is the key-space viewport
      * @param subscribedColumns are the columns subscribed for this view
      * @return a MessageView filtered by the subscription properties that can be sent to that subscriber
      */
-    MessageView getSubView(BarrageSubscriptionOptions options, boolean isInitialSnapshot, @Nullable RowSet viewport,
-            boolean reverseViewport, @Nullable RowSet keyspaceViewport, BitSet subscribedColumns);
+    MessageView getSubView(
+            BarrageSubscriptionOptions options,
+            boolean isInitialSnapshot,
+            boolean isFullSubscription,
+            @Nullable RowSet viewport,
+            boolean reverseViewport,
+            @Nullable RowSet keyspaceViewportPrev,
+            @Nullable RowSet keyspaceViewport,
+            BitSet subscribedColumns);
 
     /**
      * Obtain a Full-Snapshot View of this StreamGenerator that can be sent to a single requestor.