Support very large tables

engine/chunk/src/main/java/io/deephaven/chunk/ByteChunkToOutputStreamAdapter.java

@@ -0,0 +1,12 @@
+package io.deephaven.chunk;
+
+import io.deephaven.chunk.attributes.Any;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+public class ByteChunkToOutputStreamAdapter {
+    public static void write(OutputStream stream, ByteChunk<? extends Any> chunk, int srcOffset, int length) throws IOException {
+        stream.write(chunk.data, chunk.offset + srcOffset, length);
+    }
+}

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

@@ -8,6 +8,7 @@
 import io.deephaven.base.log.LogOutput;
 import io.deephaven.base.log.LogOutputAppendable;
 import io.deephaven.base.verify.Assert;
+import io.deephaven.chunk.util.pools.ChunkPoolConstants;
 import io.deephaven.configuration.Configuration;
 import io.deephaven.datastructures.util.CollectionUtil;
 import io.deephaven.engine.rowset.*;
@@ -79,6 +80,8 @@ public NoSnapshotAllowedException(String reason) {
     private static final int MAX_CONCURRENT_ATTEMPT_DURATION_MILLIS = Configuration.getInstance()
             .getIntegerWithDefault("ConstructSnapshot.maxConcurrentAttemptDurationMillis", 5000);
 
+    public static final int SNAPSHOT_CHUNK_SIZE = 1 << ChunkPoolConstants.LARGEST_POOLED_CHUNK_LOG2_CAPACITY;
+
     /**
      * Holder for thread-local state.
      */
@@ -1316,8 +1319,6 @@ public static boolean serializeAllTable(final boolean usePrev,
             snapshot.rowsIncluded = snapshot.rowsAdded.copy();
         }
 
-        LongSizedDataStructure.intSize("construct snapshot", snapshot.rowsIncluded.size());
-
         final Map<String, ? extends ColumnSource> sourceMap = table.getColumnSourceMap();
         final String[] columnSources = sourceMap.keySet().toArray(CollectionUtil.ZERO_LENGTH_STRING_ARRAY);
 
@@ -1340,16 +1341,19 @@ public static boolean serializeAllTable(final boolean usePrev,
                 final RowSet rows = columnIsEmpty ? RowSetFactory.empty() : snapshot.rowsIncluded;
                 // Note: cannot use shared context across several calls of differing lengths and no sharing necessary
                 // when empty
-                acd.data = getSnapshotDataAsChunk(columnSource, columnIsEmpty ? null : sharedContext, rows, usePrev);
+                acd.data =
+                        getSnapshotDataAsChunkList(columnSource, columnIsEmpty ? null : sharedContext, rows, usePrev);
                 acd.type = columnSource.getType();
                 acd.componentType = columnSource.getComponentType();
+                acd.chunkType = columnSource.getChunkType();
 
                 final BarrageMessage.ModColumnData mcd = new BarrageMessage.ModColumnData();
                 snapshot.modColumnData[ii] = mcd;
                 mcd.rowsModified = RowSetFactory.empty();
-                mcd.data = getSnapshotDataAsChunk(columnSource, null, RowSetFactory.empty(), usePrev);
+                mcd.data = getSnapshotDataAsChunkList(columnSource, null, RowSetFactory.empty(), usePrev);
                 mcd.type = acd.type;
                 mcd.componentType = acd.componentType;
+                mcd.chunkType = columnSource.getChunkType();
             }
         }
 
@@ -1430,6 +1434,52 @@ private static <T> WritableChunk<Values> getSnapshotDataAsChunk(final ColumnSour
         }
     }
 
+    private static <T> ArrayList<Chunk<Values>> getSnapshotDataAsChunkList(final ColumnSource<T> columnSource,
+            final SharedContext sharedContext, final RowSet rowSet, final boolean usePrev) {
+        final ColumnSource<?> sourceToUse = ReinterpretUtils.maybeConvertToPrimitive(columnSource);
+        long offset = 0;
+        final long size = rowSet.size();
+        final ArrayList<Chunk<Values>> result = new ArrayList<>();
+
+        if (size == 0) {
+            return result;
+        }
+
+        final int maxChunkSize = (int) Math.min(size, SNAPSHOT_CHUNK_SIZE);
+
+        try (final ColumnSource.FillContext context = sourceToUse.makeFillContext(maxChunkSize, sharedContext);
+                final RowSequence.Iterator it = rowSet.getRowSequenceIterator()) {
+            int chunkSize = maxChunkSize;
+            while (it.hasMore()) {
+                final RowSequence reducedRowSet = it.getNextRowSequenceWithLength(chunkSize);
+                final ChunkType chunkType = sourceToUse.getChunkType();
+
+                // create a new chunk
+                WritableChunk<Values> currentChunk = chunkType.makeWritableChunk(chunkSize);
+
+                if (usePrev) {
+                    sourceToUse.fillPrevChunk(context, currentChunk, reducedRowSet);
+                } else {
+                    sourceToUse.fillChunk(context, currentChunk, reducedRowSet);
+                }
+
+                // add the chunk to the current list
+                result.add(currentChunk);
+
+                // increment the offset for the next chunk (using the actual values written)
+                offset += currentChunk.size();
+
+                // recompute the size of the next chunk
+                if (size - offset > maxChunkSize) {
+                    chunkSize = maxChunkSize;
+                } else {
+                    chunkSize = (int) (size - offset);
+                }
+            }
+        }
+        return result;
+    }
+
     /**
      * Estimate the size of a complete table snapshot in bytes.
      *

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

@@ -5,12 +5,14 @@
 package io.deephaven.engine.table.impl.util;
 
 import io.deephaven.chunk.Chunk;
+import io.deephaven.chunk.ChunkType;
 import io.deephaven.chunk.attributes.Values;
 import io.deephaven.chunk.util.pools.PoolableChunk;
 import io.deephaven.engine.rowset.RowSet;
 import io.deephaven.engine.rowset.RowSetShiftData;
 import io.deephaven.util.SafeCloseable;
 
+import java.util.ArrayList;
 import java.util.BitSet;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
@@ -28,13 +30,15 @@ public static class ModColumnData {
         public RowSet rowsModified;
         public Class<?> type;
         public Class<?> componentType;
-        public Chunk<Values> data;
+        public ArrayList<Chunk<Values>> data;
+        public ChunkType chunkType;
     }
 
     public static class AddColumnData {
         public Class<?> type;
         public Class<?> componentType;
-        public Chunk<Values> data;
+        public ArrayList<Chunk<Values>> data;
+        public ChunkType chunkType;
     }
 
     public long firstSeq = -1;
@@ -94,9 +98,13 @@ public void close() {
                     continue;
                 }
 
-                if (acd.data instanceof PoolableChunk) {
-                    ((PoolableChunk) acd.data).close();
+                for (Chunk<Values> chunk : acd.data) {
+                    if (chunk instanceof PoolableChunk) {
+                        ((PoolableChunk) chunk).close();
+                    }
                 }
+
+                acd.data.clear();
             }
         }
         if (modColumnData != null) {
@@ -105,12 +113,13 @@ public void close() {
                     continue;
                 }
 
-                if (mcd.rowsModified != null) {
-                    mcd.rowsModified.close();
-                }
-                if (mcd.data instanceof PoolableChunk) {
-                    ((PoolableChunk) mcd.data).close();
+                for (Chunk<Values> chunk : mcd.data) {
+                    if (chunk instanceof PoolableChunk) {
+                        ((PoolableChunk) chunk).close();
+                    }
                 }
+
+                mcd.data.clear();
             }
         }
     }

engine/table/src/test/java/io/deephaven/engine/table/impl/QueryTableTest.java

@@ -2404,8 +2404,10 @@ public void testUngroupConstructSnapshotOfBoxedNull() {
 
         try (final BarrageMessage snap = ConstructSnapshot.constructBackplaneSnapshot(this, (BaseTable) ungrouped)) {
             assertEquals(snap.rowsAdded, i(0, 1, 2));
-            assertEquals(snap.addColumnData[0].data.asIntChunk().get(0), io.deephaven.util.QueryConstants.NULL_INT);
-            assertEquals(snap.addColumnData[1].data.asIntChunk().get(2), io.deephaven.util.QueryConstants.NULL_INT);
+            assertEquals(snap.addColumnData[0].data.get(0).asIntChunk().get(0),
+                    io.deephaven.util.QueryConstants.NULL_INT);
+            assertEquals(snap.addColumnData[1].data.get(0).asIntChunk().get(2),
+                    io.deephaven.util.QueryConstants.NULL_INT);
         }
     }