perf: DH-23398: Improve barrage object serialization/deserialization.

extensions/barrage/benchmark/build.gradle

@@ -41,10 +41,16 @@ def createJmhTask = {
         ]
         jvmArgs.addAll(jvmAddArgs)
         task.jvmArgs jvmArgs
-        task.args cliArgs
+        // cliArgs may be a single benchmark name or a list of JMH command-line arguments
+        task.setArgs(([] + cliArgs).flatten().collect { it.toString() })
 
         return
     })
 }
 
 createJmhTask('jmhRunBarrageRoundTrip', 'BarrageMessageRoundTripBenchmark')
+
+// The String arm runs fewer columns than the primitive default: String cells are wide enough that 5000 columns of
+// them split the snapshot across record batches, which the single-batch reader in this benchmark cannot parse.
+createJmhTask('jmhRunBarrageRoundTripStrings',
+        ['BarrageMessageRoundTripBenchmark', '-p', 'columnType=String', '-p', 'numColumns=1000'])

extensions/barrage/benchmark/src/main/java/io/deephaven/benchmark/barrage/BarrageMessageRoundTripBenchmark.java

@@ -11,6 +11,7 @@
 import io.deephaven.chunk.WritableFloatChunk;
 import io.deephaven.chunk.WritableIntChunk;
 import io.deephaven.chunk.WritableLongChunk;
+import io.deephaven.chunk.WritableObjectChunk;
 import io.deephaven.chunk.WritableShortChunk;
 import io.deephaven.chunk.attributes.Values;
 import io.deephaven.engine.context.ExecutionContext;
@@ -59,7 +60,7 @@
 
 /**
  * Benchmarks the cost of serializing and deserializing a Barrage snapshot message composed of a large number of random
- * primitive columns of a single type.
+ * columns of a single type.
  * <p>
  * The write path mirrors what a Barrage producer does: a {@link BarrageMessage} is handed to a
  * {@link BarrageMessageWriter}, and its full-snapshot {@link BarrageMessageWriter.MessageView} is drained to a byte
@@ -69,6 +70,11 @@
  * <p>
  * The source data is held in array-backed chunks whose {@code close()} is a no-op, so the pre-built message survives
  * repeated serialization and each invocation measures serialization only (no per-invocation data marshalling).
+ * <p>
+ * The whole message must land in a single record batch, since the reader parses one. That bounds
+ * {@code numColumns * numRows * bytes-per-cell}, so the wider {@code String} cells need fewer columns than the
+ * primitives do; {@code jmhRunBarrageRoundTripStrings} runs the {@code String} arm at a shape that fits. Exceed the
+ * bound and setup fails with a message saying so rather than reporting a bogus number.
  */
 @State(Scope.Thread)
 @BenchmarkMode(Mode.AverageTime)
@@ -80,7 +86,13 @@ public class BarrageMessageRoundTripBenchmark {
 
     private static final BarrageMessageWriter.Factory WRITER_FACTORY = new BarrageMessageWriterImpl.Factory();
 
-    /** The primitive element type of every column. */
+    /** Number of distinct values a {@code String} column draws from; see {@link #makeStringPool(Random)}. */
+    private static final int STRING_POOL_SIZE = 1024;
+
+    /**
+     * The element type of every column. {@code String} is reachable but not in the default set, because it needs a
+     * smaller {@code numColumns} to stay within one record batch; run it via {@code jmhRunBarrageRoundTripStrings}.
+     */
     @Param({"double", "float", "long", "int", "short", "byte"})
     private String columnType;
 
@@ -92,6 +104,14 @@ public class BarrageMessageRoundTripBenchmark {
     @Param({"2000"})
     private int numRows;
 
+    /**
+     * Length of each generated {@code String} value; ignored by the primitive types. The offsets buffer costs four
+     * bytes per row regardless of this, so it governs how much of the encode cost is payload rather than bookkeeping.
+     * Single-valued so that it does not multiply the primitive runs — override with {@code -p stringLength=8,64}.
+     */
+    @Param({"16"})
+    private int stringLength;
+
     /** Fraction of cells that are null; exercises the validity-buffer / DH-null encoding paths. */
     @Param({"0.0", "0.1"})
     private double nullFraction;
@@ -110,6 +130,9 @@ public class BarrageMessageRoundTripBenchmark {
     private Class<?>[] wireTypes;
     private Class<?>[] wireComponentTypes;
 
+    /** The distinct values drawn from for {@code String} columns; null for the primitive types. */
+    private String[] stringPool;
+
     // the reusable, pre-built message and its pre-serialized bytes
     private BarrageMessage message;
     private ChunkWriter<Chunk<Values>>[] chunkWriters;
@@ -125,7 +148,7 @@ public void setup() {
                 .build();
         reader = new BarrageMessageReaderImpl();
 
-        final Class<?> dataType = primitiveClass(columnType);
+        final Class<?> dataType = elementClass(columnType);
         final ChunkType chunkType = ChunkType.fromElementType(dataType);
 
         // Build the flat table schema of numColumns columns of the chosen type.
@@ -152,6 +175,7 @@ public void setup() {
         // Generate the random column data once. The chunks are array-backed and have a no-op close(), so the message
         // can be handed to a writer repeatedly without losing its data.
         final Random random = new Random(0xB33FCAFEL);
+        stringPool = columnType.equals("String") ? makeStringPool(random) : null;
         final BarrageMessage.AddColumnData[] addColumnData = new BarrageMessage.AddColumnData[numColumns];
         for (int ci = 0; ci < numColumns; ++ci) {
             final BarrageMessage.AddColumnData acd = new BarrageMessage.AddColumnData();
@@ -193,7 +217,7 @@ public void tearDown() {
         executionContext.close();
     }
 
-    private static Class<?> primitiveClass(final String type) {
+    private static Class<?> elementClass(final String type) {
         switch (type) {
             case "double":
                 return double.class;
@@ -207,6 +231,8 @@ private static Class<?> primitiveClass(final String type) {
                 return short.class;
             case "byte":
                 return byte.class;
+            case "String":
+                return String.class;
             default:
                 throw new IllegalArgumentException("Unknown column type: " + type);
         }
@@ -216,6 +242,26 @@ private boolean nextIsNull(final Random random) {
         return nullFraction > 0.0 && random.nextDouble() < nullFraction;
     }
 
+    /**
+     * Build the pool of distinct values that {@code String} cells are drawn from.
+     * <p>
+     * Cells share these instances rather than each holding a unique {@code String}, which both keeps
+     * {@code numColumns * numRows} cells within a reasonable heap and matches how string columns actually look in
+     * practice — symbols and identifiers repeat. Sharing does not shortcut any of the measured work: the writer still
+     * encodes every cell to UTF-8 and the reader still constructs a fresh {@code String} per cell.
+     */
+    private String[] makeStringPool(final Random random) {
+        final char[] chars = new char[stringLength];
+        final String[] pool = new String[STRING_POOL_SIZE];
+        for (int pi = 0; pi < pool.length; ++pi) {
+            for (int ci = 0; ci < stringLength; ++ci) {
+                chars[ci] = (char) ('a' + random.nextInt(26));
+            }
+            pool[pi] = new String(chars);
+        }
+        return pool;
+    }
+
     private WritableChunk<Values> makeColumn(final Random random) {
         switch (columnType) {
             case "double": {
@@ -260,6 +306,13 @@ private WritableChunk<Values> makeColumn(final Random random) {
                 }
                 return WritableByteChunk.writableChunkWrap(values);
             }
+            case "String": {
+                final String[] values = new String[numRows];
+                for (int ri = 0; ri < numRows; ++ri) {
+                    values[ri] = nextIsNull(random) ? null : stringPool[random.nextInt(stringPool.length)];
+                }
+                return WritableObjectChunk.writableChunkWrap(values);
+            }
             default:
                 throw new IllegalArgumentException("Unknown column type: " + columnType);
         }

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkReader.java

@@ -6,6 +6,7 @@
 import io.deephaven.chunk.Chunk;
 import io.deephaven.chunk.ChunkType;
 import io.deephaven.chunk.WritableChunk;
+import io.deephaven.chunk.WritableIntChunk;
 import io.deephaven.chunk.WritableLongChunk;
 import io.deephaven.chunk.attributes.Any;
 import io.deephaven.chunk.attributes.Values;
@@ -29,6 +30,33 @@ public abstract class BaseChunkReader<READ_CHUNK_TYPE extends WritableChunk<Valu
     protected static final int BULK_READ_BUFFER_BYTES = Configuration.getInstance()
             .getIntegerForClassWithDefault(BaseChunkReader.class, "bulkReadBufferBytes", 4096);
 
+    /** Number of {@code int}s decoded per bulk-read window (see {@link #BULK_READ_BUFFER_BYTES}). */
+    private static final int BULK_READ_INTS = Math.max(1, BULK_READ_BUFFER_BYTES / Integer.BYTES);
+
+    /**
+     * Read {@code numElements} little-endian {@code int}s — an Arrow offsets or lengths buffer — into {@code dest},
+     * pulling the payload in bounded windows into a reused buffer rather than making one {@link DataInput} call, i.e.
+     * four individual byte reads, per value.
+     *
+     * @param is the input to read from
+     * @param dest the chunk to populate, starting at position zero
+     * @param numElements the number of values to read
+     */
+    protected static void readIntBuffer(
+            @NotNull final DataInput is,
+            @NotNull final WritableIntChunk<?> dest,
+            final int numElements) throws IOException {
+        final byte[] buffer = new byte[Math.min(numElements, BULK_READ_INTS) * Integer.BYTES];
+        for (int ei = 0; ei < numElements;) {
+            final int n = Math.min(BULK_READ_INTS, numElements - ei);
+            is.readFully(buffer, 0, n * Integer.BYTES);
+            for (int jj = 0; jj < n; ++jj) {
+                dest.set(ei + jj, LittleEndianCodec.getInt(buffer, jj * Integer.BYTES));
+            }
+            ei += n;
+        }
+    }
+
     @FunctionalInterface
     public interface ChunkTransformer<READ_CHUNK_TYPE extends Chunk<Values>, DEST_CHUNK_TYPE extends WritableChunk<Values>> {
         void transform(READ_CHUNK_TYPE source, DEST_CHUNK_TYPE dest, int destOffset);