diff --git a/coverage/build.gradle b/coverage/build.gradle index d484280a0c4..1529fd8b84d 100644 --- a/coverage/build.gradle +++ b/coverage/build.gradle @@ -16,12 +16,7 @@ tasks.register("jacoco-merge", JacocoReport) { } additionalSourceDirs = files(jprojects.sourceSets.main.allSource.srcDirs) sourceDirectories = files(jprojects.sourceSets.main.allSource.srcDirs) - classDirectories = files(jprojects.sourceSets.main.output).asFileTree.matching { - // DH-23193 added a GWT supersource LittleEndianCodec that compiles into web-client-api - // under the same package as the real class in extensions-barrage, causing a JaCoCo - // "Can't add different class with same name" error. Exclude only the supersource copy. - exclude 'io/deephaven/extensions/barrage/chunk/LittleEndianCodec.class' - } + classDirectories = files(jprojects.sourceSets.main.output) reports { html.required = true html.outputLocation = layout.buildDirectory.dir('./reports/java/java-coverage.html') diff --git a/extensions/barrage/benchmark/build.gradle b/extensions/barrage/benchmark/build.gradle index fac3e1aa637..dd582fa923a 100644 --- a/extensions/barrage/benchmark/build.gradle +++ b/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']) diff --git a/extensions/barrage/benchmark/src/main/java/io/deephaven/benchmark/barrage/BarrageMessageRoundTripBenchmark.java b/extensions/barrage/benchmark/src/main/java/io/deephaven/benchmark/barrage/BarrageMessageRoundTripBenchmark.java index 95c46d78285..93a43d9709b 100644 --- a/extensions/barrage/benchmark/src/main/java/io/deephaven/benchmark/barrage/BarrageMessageRoundTripBenchmark.java +++ b/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. *

* 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 @@ *

* 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). + *

+ * 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>[] 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. + *

+ * 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 makeColumn(final Random random) { switch (columnType) { case "double": { @@ -260,6 +306,13 @@ private WritableChunk 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); } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkReader.java index a85b477ae70..31c86055e51 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkReader.java +++ b/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 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, DEST_CHUNK_TYPE extends WritableChunk> { void transform(READ_CHUNK_TYPE source, DEST_CHUNK_TYPE dest, int destOffset); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkWriter.java index 2892a1395c2..18f3264c8a2 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BaseChunkWriter.java @@ -19,6 +19,8 @@ import java.io.DataOutput; import java.io.IOException; +import java.io.OutputStream; +import java.util.Arrays; import java.util.function.Supplier; public abstract class BaseChunkWriter> @@ -40,6 +42,9 @@ public interface ChunkTransformer> { protected static final int BULK_WRITE_BUFFER_BYTES = Configuration.getInstance() .getIntegerForClassWithDefault(BaseChunkWriter.class, "bulkWriteBufferBytes", 4096); + /** Number of {@code int}s buffered per bulk-write window (see {@link #BULK_WRITE_BUFFER_BYTES}). */ + private static final int BULK_WRITE_INTS = Math.max(1, BULK_WRITE_BUFFER_BYTES / Integer.BYTES); + private final ChunkTransformer transformer; private final Supplier emptyChunkSupplier; /** the size of each element in bytes if fixed */ @@ -103,27 +108,19 @@ public boolean isFieldNullable() { } /** - * Compute the number of nulls in the subset. + * Report the nullness of each row of the subset, in order, to {@code validity}. + *

+ * This is invoked at most once per {@link BaseChunkInputStream}: the null count carried by the field node and the + * bytes of the validity buffer are both derived from this single traversal of the row data. * * @param context the context for the chunk * @param subset the subset of rows to consider - * @return the number of nulls in the subset + * @param validity the validity buffer to populate */ - protected abstract int computeNullCount( - @NotNull Context context, - @NotNull RowSequence subset); - - /** - * Update the validity buffer for the subset. - * - * @param context the context for the chunk - * @param subset the subset of rows to consider - * @param serContext the serialization context - */ - protected abstract void writeValidityBufferInternal( + protected abstract void computeValidity( @NotNull Context context, @NotNull RowSequence subset, - @NotNull SerContext serContext); + @NotNull ValidityBuffer validity); abstract class BaseChunkInputStream extends DrainableColumn { protected final CONTEXT_TYPE context; @@ -132,6 +129,8 @@ abstract class BaseChunkInputStream extends Draina protected boolean hasBeenRead = false; private final int nullCount; + /** The bitmap to drain, retained only when we will actually send a validity buffer. */ + private final ValidityBuffer validityBuffer; BaseChunkInputStream( @NotNull final CONTEXT_TYPE context, @@ -152,10 +151,18 @@ abstract class BaseChunkInputStream extends Draina "Subset " + this.subset + " is out of bounds for context of size " + context.size()); } - if (dhNullable && options.useDeephavenNulls()) { + // A non-nullable field never reports nulls (see nullCount()) and a dh-nullable field encodes them in the + // payload itself, so in both cases the traversal below would produce a bitmap nobody reads. + if (!fieldNullable || (dhNullable && options.useDeephavenNulls())) { nullCount = 0; + validityBuffer = null; } else { - nullCount = computeNullCount(context, this.subset); + final ValidityBuffer validity = new ValidityBuffer(this.subset.intSize()); + computeValidity(context, this.subset, validity); + nullCount = validity.nullCount(); + // Retain the bitmap only if we will send it; a batch of null-free columns would otherwise hold one + // bitmap per column until it drains. + validityBuffer = nullCount == 0 ? null : validity; } } @@ -205,8 +212,13 @@ protected long writeValidityBuffer(final DataOutput dos) { return 0; } - try (final SerContext serContext = new SerContext(dos)) { - writeValidityBufferInternal(context, subset, serContext); + // the bitmap was packed into little-endian bytes when it was computed; emit it with a single bulk write + final byte[] bytes = validityBuffer.bytes(); + try { + dos.write(bytes, 0, bytes.length); + } catch (final IOException e) { + throw new UncheckedDeephavenException( + "Unexpected exception while draining data to OutputStream: ", e); } return getValidityMapSerializationSizeFor(subset.intSize()); @@ -262,38 +274,178 @@ protected static int getNumLongsForBitPackOfSize(final int numElements) { return ((numElements + 63) / 64); } - protected static final class SerContext implements SafeCloseable { - private final DataOutput dos; + /** + * A bit per element, packed LSB-first into little-endian 64-bit words, as Arrow encodes a validity buffer: a set + * bit marks a valid (non-null) element. Nulls are counted as the bits are appended, so a single traversal of the + * row data yields both the field node's null count and the bytes of the validity buffer. + *

+ * {@link BooleanChunkWriter} also uses this to pack its payload, which has the same shape, via + * {@link #packed(int)}. + */ + protected static final class ValidityBuffer { + private final int numElements; + + /** + * Allocated on the first null. A column with no nulls sends no validity buffer at all, and that is the common + * case, so until a null appears the only state worth maintaining is {@link #count} — this keeps the traversal + * as cheap as the bare null count it replaced. The bits skipped along the way are all set, so the buffer can + * still be reconstructed exactly whenever a null does turn up. + */ + private byte[] bytes; + /** Number of elements appended so far; the packed bit position when {@link #bytes} is non-null. */ + private int count = 0; private long accumulator = 0; - private long count = 0; + private int byteOffset = 0; + private int nullCount = 0; + private boolean sealed = false; + + /** + * A validity bitmap, materialized only once a null is appended. A column with no nulls sends no validity buffer + * at all, so asking such a buffer for its {@link #bytes()} is a caller bug and throws. + */ + public ValidityBuffer(final int numElements) { + this.numElements = numElements; + } + + /** + * A bit-packed buffer that is materialized up front, for a caller that needs the bits whatever the values turn + * out to be. {@link BooleanChunkWriter} packs its payload this way, where a set bit means TRUE rather than + * non-null and an all-TRUE column must still emit a full buffer. + * + * @param numElements the number of elements to be appended + * @return a buffer whose {@link #bytes()} is always available + */ + public static ValidityBuffer packed(final int numElements) { + final ValidityBuffer buffer = new ValidityBuffer(numElements); + buffer.allocate(); + return buffer; + } + + public void setNextIsNull(final boolean isNull) { + if (bytes != null) { + appendPacked(isNull); + } else if (isNull) { + allocate(); + appendPacked(true); + } else { + ++count; + } + } + + /** + * Append {@code numElements} null entries; equivalent to that many {@code setNextIsNull(true)} calls, but + * without visiting each element. + * + * @param numElements the number of null entries to append + */ + public void setNextAreNull(final int numElements) { + if (numElements == 0) { + return; + } + allocate(); + nullCount += numElements; + for (int remaining = numElements; remaining > 0;) { + final int inThisWord = Math.min(remaining, 64 - (count & 63)); + count += inThisWord; + if ((count & 63) == 0) { + flushWord(); + } + remaining -= inThisWord; + } + } - public SerContext(@NotNull final DataOutput dos) { - this.dos = dos; + private void appendPacked(final boolean isNull) { + if (isNull) { + ++nullCount; + } else { + accumulator |= 1L << (count & 63); + } + if ((++count & 63) == 0) { + flushWord(); + } } - public void setNextIsNull(boolean isNull) { - if (!isNull) { - accumulator |= 1L << count; + /** + * Materialize the buffer at the first null. Every element visited so far was valid, so the words already passed + * are all-ones, as are the bits of the word in progress. + */ + private void allocate() { + if (bytes != null) { + return; } - if (++count == 64) { + bytes = new byte[getValidityMapSerializationSizeFor(numElements)]; + byteOffset = (count >>> 6) * Long.BYTES; + Arrays.fill(bytes, 0, byteOffset, (byte) 0xFF); + accumulator = (1L << (count & 63)) - 1; + } + + public int nullCount() { + return nullCount; + } + + /** + * Finalize and return the packed bytes; exactly {@code getValidityMapSerializationSizeFor(numElements)} of + * them. No element may be appended afterwards. + * + * @throws IllegalStateException if nothing was ever materialized, i.e. no null was appended and the buffer was + * not created by {@link #packed(int)} + */ + public byte[] bytes() { + if (bytes == null) { + throw new IllegalStateException("No bits have been packed: a validity buffer is only written when " + + "nullCount() is non-zero; a caller that needs the bytes regardless must use packed()"); + } + if (!sealed) { + sealed = true; + if ((count & 63) != 0) { + flushWord(); + } + } + return bytes; + } + + private void flushWord() { + LittleEndianCodec.putLong(bytes, byteOffset, accumulator); + byteOffset += Long.BYTES; + accumulator = 0; + } + } + + /** + * Buffers little-endian {@code int} values — an Arrow offset or lengths buffer — and flushes them in windows of + * {@link #BULK_WRITE_BUFFER_BYTES}, rather than making one {@link DataOutput} call, i.e. four individual byte + * writes, per value. + */ + protected static final class BulkIntWriter implements SafeCloseable { + private final OutputStream outputStream; + private final byte[] buffer; + private int bufferPos = 0; + + public BulkIntWriter(@NotNull final OutputStream outputStream) { + this.outputStream = outputStream; + this.buffer = new byte[BULK_WRITE_INTS * Integer.BYTES]; + } + + public void write(final int value) { + LittleEndianCodec.putInt(buffer, bufferPos, value); + bufferPos += Integer.BYTES; + if (bufferPos == buffer.length) { flush(); } } private void flush() { - if (count == 0) { + if (bufferPos == 0) { return; } - try { - dos.writeLong(accumulator); + outputStream.write(buffer, 0, bufferPos); } catch (final IOException e) { throw new UncheckedDeephavenException( "Unexpected exception while draining data to OutputStream: ", e); } - accumulator = 0; - count = 0; + bufferPos = 0; } @Override diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BigDecimalChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BigDecimalChunkWriter.java index d87c6c80a62..64f9636f540 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BigDecimalChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BigDecimalChunkWriter.java @@ -9,7 +9,6 @@ import io.deephaven.chunk.ObjectChunk; import io.deephaven.chunk.attributes.Values; import io.deephaven.engine.rowset.RowSequence; -import io.deephaven.util.mutable.MutableInt; import org.apache.arrow.vector.types.pojo.ArrowType; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -40,26 +39,12 @@ public BigDecimalChunkWriter( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> { - if (objectChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(objectChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(objectChunk.isNull((int) row))); } @Override diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BooleanChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BooleanChunkWriter.java index e6aa0e703af..138cd31d552 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BooleanChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/BooleanChunkWriter.java @@ -12,7 +12,6 @@ import io.deephaven.extensions.barrage.BarrageOptions; import io.deephaven.util.BooleanUtils; import io.deephaven.util.datastructures.LongSizedDataStructure; -import io.deephaven.util.mutable.MutableInt; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -52,22 +51,10 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount(@NotNull Context context, @NotNull RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); + protected void computeValidity(@NotNull Context context, @NotNull RowSequence subset, + @NotNull ValidityBuffer validity) { final ByteChunk byteChunk = context.getChunk().asByteChunk(); - subset.forAllRowKeys(row -> { - if (BooleanUtils.isNull(byteChunk.get((int) row))) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal(@NotNull Context context, @NotNull RowSequence subset, - @NotNull SerContext serContext) { - final ByteChunk byteChunk = context.getChunk().asByteChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(BooleanUtils.isNull(byteChunk.get((int) row)))); + subset.forAllRowKeys(row -> validity.setNextIsNull(BooleanUtils.isNull(byteChunk.get((int) row)))); } private class BooleanChunkInputStream extends BaseChunkInputStream { @@ -116,12 +103,14 @@ public int drainTo(final OutputStream outputStream) throws IOException { bytesWritten += writeValidityBuffer(dos); // write the payload buffer - // we cheat and re-use validity buffer serialization code - try (final SerContext serContext = new SerContext(dos)) { - final ByteChunk byteChunk = context.getChunk().asByteChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull( - BooleanUtils.byteAsBoolean(byteChunk.get((int) row)) != Boolean.TRUE)); - } + // we cheat and re-use validity buffer bit-packing; a set bit is TRUE rather than non-null. It is packed up + // front because an all-TRUE column appends no "null" and still has to emit a full buffer. + final ValidityBuffer payload = ValidityBuffer.packed(subset.intSize(DEBUG_NAME)); + final ByteChunk byteChunk = context.getChunk().asByteChunk(); + subset.forAllRowKeys(row -> payload.setNextIsNull( + BooleanUtils.byteAsBoolean(byteChunk.get((int) row)) != Boolean.TRUE)); + final byte[] payloadBytes = payload.bytes(); + dos.write(payloadBytes, 0, payloadBytes.length); bytesWritten += getNumLongsForBitPackOfSize(subset.intSize(DEBUG_NAME)) * (long) Long.BYTES; return LongSizedDataStructure.intSize(DEBUG_NAME, bytesWritten); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkReader.java index c166f77db98..bae414866a0 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkReader.java @@ -25,6 +25,9 @@ public class ByteChunkReader extends BaseChunkReader> { private static final String DEBUG_NAME = "ByteChunkReader"; + // Number of elements decoded per bounded bulk-read window (see BaseChunkReader#BULK_READ_BUFFER_BYTES). + private static final int BULK_READ_ELEMENTS = Math.max(1, BULK_READ_BUFFER_BYTES / Byte.BYTES); + public static , T extends ChunkReader> ChunkReader> transformFrom( final T wireReader, final ChunkTransformer> wireTransform) { @@ -92,13 +95,16 @@ private static void useDeephavenNulls( final ChunkWriter.FieldNodeInfo nodeInfo, final WritableByteChunk chunk, final int offset) throws IOException { - // Bytes have no endianness, so transfer the payload directly into the chunk's backing array in bounded windows - // rather than one DataInput#readByte call per element (and without materializing the whole payload at once). - for (int ei = 0; ei < nodeInfo.numElements;) { - final int length = Math.min(Math.max(1, BULK_READ_BUFFER_BYTES), nodeInfo.numElements - ei); + final int numElements = nodeInfo.numElements; + // region PayloadDhNulls + // Bytes have no endianness, so transfer the payload straight into the chunk's backing array in + // bounded windows rather than decoding element by element through a staging buffer. + for (int ei = 0; ei < numElements;) { + final int length = Math.min(BULK_READ_ELEMENTS, numElements - ei); is.readFully(chunk.array(), chunk.arrayOffset() + offset + ei, length); ei += length; } + // endregion PayloadDhNulls } private static void useValidityBuffer( @@ -110,13 +116,15 @@ private static void useValidityBuffer( final int numElements = nodeInfo.numElements; final int numValidityWords = (numElements + 63) / 64; - // The payload carries a value slot for every element, including nulls; transfer it directly into the chunk's - // backing array in bounded windows, then overwrite the invalid positions with the null value. + // region PayloadValidityBuffer + // The payload carries a value slot for every element, including nulls; transfer it straight into + // the chunk's backing array in bounded windows, then overwrite the invalid positions with null. for (int ei = 0; ei < numElements;) { - final int length = Math.min(Math.max(1, BULK_READ_BUFFER_BYTES), numElements - ei); + final int length = Math.min(BULK_READ_ELEMENTS, numElements - ei); is.readFully(chunk.array(), chunk.arrayOffset() + offset + ei, length); ei += length; } + // endregion PayloadValidityBuffer int ei = 0; for (int vi = 0; vi < numValidityWords; ++vi) { @@ -124,7 +132,7 @@ private static void useValidityBuffer( long validityWord = isValid.get(vi); do { if ((validityWord & 1) == 1) { - // Skip the run of valid slots (already read) to the next null. + // Skip the run of valid slots (already decoded) to the next null. final int valids = Math.min(Long.numberOfTrailingZeros(~validityWord), bitsLeftInThisWord); ei += valids; validityWord >>= valids; diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkWriter.java index 5f2a2006736..60eac7e8c96 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ByteChunkWriter.java @@ -92,26 +92,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final ByteChunk byteChunk = context.getChunk().asByteChunk(); - subset.forAllRowKeys(row -> { - if (byteChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final ByteChunk byteChunk = context.getChunk().asByteChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(byteChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(byteChunk.isNull((int) row))); } private class ByteChunkInputStream extends BaseChunkInputStream { @@ -148,13 +134,14 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity buffer bytesWritten += writeValidityBuffer(dos); - // write the payload buffer in bounded windows, gathering values into a byte[] and flushing a full window - // with a single bulk write rather than one DataOutput value at a time + // write the payload buffer in bounded windows, encoding each value into little-endian bytes (via + // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value, + // i.e. one individual byte write per byte of the value, at a time. final ByteChunk byteChunk = context.getChunk().asByteChunk(); final byte[] buffer = new byte[BULK_WRITE_ELEMENTS * Byte.BYTES]; final MutableInt bufferPos = new MutableInt(0); subset.forAllRowKeys(row -> { - buffer[bufferPos.get()] = byteChunk.get((int) row); + LittleEndianCodec.putByte(buffer, bufferPos.get(), byteChunk.get((int) row)); bufferPos.add(Byte.BYTES); if (bufferPos.get() == buffer.length) { try { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkReader.java index 4ddcb395178..3e53b7845d1 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkReader.java @@ -21,6 +21,9 @@ public class CharChunkReader extends BaseChunkReader> { private static final String DEBUG_NAME = "CharChunkReader"; + // Number of elements decoded per bounded bulk-read window (see BaseChunkReader#BULK_READ_BUFFER_BYTES). + private static final int BULK_READ_ELEMENTS = Math.max(1, BULK_READ_BUFFER_BYTES / Character.BYTES); + public static , T extends ChunkReader> ChunkReader> transformFrom( final T wireReader, final ChunkTransformer> wireTransform) { @@ -88,9 +91,20 @@ private static void useDeephavenNulls( final ChunkWriter.FieldNodeInfo nodeInfo, final WritableCharChunk chunk, final int offset) throws IOException { - for (int ii = 0; ii < nodeInfo.numElements; ++ii) { - chunk.set(offset + ii, is.readChar()); + final int numElements = nodeInfo.numElements; + // region PayloadDhNulls + // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes + // via LittleEndianCodec (VarHandle on the JVM, GWT-safe arithmetic in the web client's super-source). + final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Character.BYTES]; + for (int ei = 0; ei < numElements;) { + final int n = Math.min(BULK_READ_ELEMENTS, numElements - ei); + is.readFully(buffer, 0, n * Character.BYTES); + for (int jj = 0; jj < n; ++jj) { + chunk.set(offset + ei + jj, LittleEndianCodec.getChar(buffer, jj * Character.BYTES)); + } + ei += n; } + // endregion PayloadDhNulls } private static void useValidityBuffer( @@ -102,35 +116,39 @@ private static void useValidityBuffer( final int numElements = nodeInfo.numElements; final int numValidityWords = (numElements + 63) / 64; - int ei = 0; - int pendingSkips = 0; + // region PayloadValidityBuffer + // The payload carries a value slot for every element, including nulls; read it in bounded windows into a + // reused buffer and decode each value, then overwrite the invalid positions with the null value. + final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Character.BYTES]; + for (int ei = 0; ei < numElements;) { + final int n = Math.min(BULK_READ_ELEMENTS, numElements - ei); + is.readFully(buffer, 0, n * Character.BYTES); + for (int jj = 0; jj < n; ++jj) { + chunk.set(offset + ei + jj, LittleEndianCodec.getChar(buffer, jj * Character.BYTES)); + } + ei += n; + } + // endregion PayloadValidityBuffer + int ei = 0; for (int vi = 0; vi < numValidityWords; ++vi) { int bitsLeftInThisWord = Math.min(64, numElements - vi * 64); long validityWord = isValid.get(vi); do { if ((validityWord & 1) == 1) { - if (pendingSkips > 0) { - is.skipBytes(pendingSkips * Character.BYTES); - chunk.fillWithNullValue(offset + ei, pendingSkips); - ei += pendingSkips; - pendingSkips = 0; - } - chunk.set(offset + ei++, is.readChar()); - validityWord >>= 1; - bitsLeftInThisWord--; + // Skip the run of valid slots (already decoded) to the next null. + final int valids = Math.min(Long.numberOfTrailingZeros(~validityWord), bitsLeftInThisWord); + ei += valids; + validityWord >>= valids; + bitsLeftInThisWord -= valids; } else { - final int skips = Math.min(Long.numberOfTrailingZeros(validityWord), bitsLeftInThisWord); - pendingSkips += skips; - validityWord >>= skips; - bitsLeftInThisWord -= skips; + final int nulls = Math.min(Long.numberOfTrailingZeros(validityWord), bitsLeftInThisWord); + chunk.fillWithNullValue(offset + ei, nulls); + ei += nulls; + validityWord >>= nulls; + bitsLeftInThisWord -= nulls; } } while (bitsLeftInThisWord > 0); } - - if (pendingSkips > 0) { - is.skipBytes(pendingSkips * Character.BYTES); - chunk.fillWithNullValue(offset + ei, pendingSkips); - } } } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkWriter.java index 5a4c22c23f8..b0e65c67c29 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/CharChunkWriter.java @@ -25,6 +25,9 @@ public class CharChunkWriter> extends BaseChunkWriter { private static final String DEBUG_NAME = "CharChunkWriter"; + + // Number of elements encoded per bounded bulk-write window (see BaseChunkWriter#BULK_WRITE_BUFFER_BYTES). + private static final int BULK_WRITE_ELEMENTS = Math.max(1, BULK_WRITE_BUFFER_BYTES / Character.BYTES); private static final CharChunkWriter> NULLABLE_IDENTITY_INSTANCE = new CharChunkWriter<>( null, CharChunk::getEmptyChunk, true); private static final CharChunkWriter> NON_NULLABLE_IDENTITY_INSTANCE = new CharChunkWriter<>( @@ -85,26 +88,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final CharChunk charChunk = context.getChunk().asCharChunk(); - subset.forAllRowKeys(row -> { - if (charChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final CharChunk charChunk = context.getChunk().asCharChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(charChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(charChunk.isNull((int) row))); } private class CharChunkInputStream extends BaseChunkInputStream { @@ -141,16 +130,28 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity buffer bytesWritten += writeValidityBuffer(dos); - // write the payload buffer + // write the payload buffer in bounded windows, encoding each value into little-endian bytes (via + // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value, + // i.e. one individual byte write per byte of the value, at a time. final CharChunk charChunk = context.getChunk().asCharChunk(); + final byte[] buffer = new byte[BULK_WRITE_ELEMENTS * Character.BYTES]; + final MutableInt bufferPos = new MutableInt(0); subset.forAllRowKeys(row -> { - try { - dos.writeChar(charChunk.get((int) row)); - } catch (final IOException e) { - throw new UncheckedDeephavenException( - "Unexpected exception while draining data to OutputStream: ", e); + LittleEndianCodec.putChar(buffer, bufferPos.get(), charChunk.get((int) row)); + bufferPos.add(Character.BYTES); + if (bufferPos.get() == buffer.length) { + try { + outputStream.write(buffer, 0, buffer.length); + } catch (final IOException e) { + throw new UncheckedDeephavenException( + "Unexpected exception while draining data to OutputStream: ", e); + } + bufferPos.set(0); } }); + if (bufferPos.get() > 0) { + outputStream.write(buffer, 0, bufferPos.get()); + } bytesWritten += elementSize * subset.size(); bytesWritten += writePadBuffer(dos, bytesWritten); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DictionaryChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DictionaryChunkWriter.java index d9954cbd398..559ccb4a26e 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DictionaryChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DictionaryChunkWriter.java @@ -72,18 +72,10 @@ public ChunkType getValuesChunkType() { } @Override - protected int computeNullCount( - @NotNull final ChunkWriter.Context context, - @NotNull final RowSequence subset) { - // Not called — we override getInputStream completely. - throw new UnsupportedOperationException(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final ChunkWriter.Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { // Not called — we override getInputStream completely. throw new UnsupportedOperationException(); } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkReader.java index 83e07cae5f1..5aee0e65e53 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkReader.java @@ -96,6 +96,7 @@ private static void useDeephavenNulls( final WritableDoubleChunk chunk, final int offset) throws IOException { final int numElements = nodeInfo.numElements; + // region PayloadDhNulls // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes // via LittleEndianCodec (VarHandle on the JVM, GWT-safe arithmetic in the web client's super-source). final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Double.BYTES]; @@ -107,6 +108,7 @@ private static void useDeephavenNulls( } ei += n; } + // endregion PayloadDhNulls } private static void useValidityBuffer( @@ -118,6 +120,7 @@ private static void useValidityBuffer( final int numElements = nodeInfo.numElements; final int numValidityWords = (numElements + 63) / 64; + // region PayloadValidityBuffer // The payload carries a value slot for every element, including nulls; read it in bounded windows into a // reused buffer and decode each value, then overwrite the invalid positions with the null value. final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Double.BYTES]; @@ -129,6 +132,7 @@ private static void useValidityBuffer( } ei += n; } + // endregion PayloadValidityBuffer int ei = 0; for (int vi = 0; vi < numValidityWords; ++vi) { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkWriter.java index ee18ee1cb9c..dcb973ac6ee 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/DoubleChunkWriter.java @@ -92,26 +92,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final DoubleChunk doubleChunk = context.getChunk().asDoubleChunk(); - subset.forAllRowKeys(row -> { - if (doubleChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final DoubleChunk doubleChunk = context.getChunk().asDoubleChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(doubleChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(doubleChunk.isNull((int) row))); } private class DoubleChunkInputStream extends BaseChunkInputStream { @@ -149,8 +135,8 @@ public int drainTo(final OutputStream outputStream) throws IOException { bytesWritten += writeValidityBuffer(dos); // write the payload buffer in bounded windows, encoding each value into little-endian bytes (via - // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value - // (eight bytes) at a time. + // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value, + // i.e. one individual byte write per byte of the value, at a time. final DoubleChunk doubleChunk = context.getChunk().asDoubleChunk(); final byte[] buffer = new byte[BULK_WRITE_ELEMENTS * Double.BYTES]; final MutableInt bufferPos = new MutableInt(0); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FixedWidthObjectChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FixedWidthObjectChunkWriter.java index 3035e8c9053..c1e5b7b4b78 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FixedWidthObjectChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FixedWidthObjectChunkWriter.java @@ -6,7 +6,6 @@ import io.deephaven.chunk.ObjectChunk; import io.deephaven.chunk.attributes.Values; import io.deephaven.engine.rowset.RowSequence; -import io.deephaven.util.mutable.MutableInt; import org.jetbrains.annotations.NotNull; public abstract class FixedWidthObjectChunkWriter extends FixedWidthChunkWriter> { @@ -19,25 +18,11 @@ public FixedWidthObjectChunkWriter( } @Override - protected int computeNullCount( - @NotNull final BaseChunkWriter.Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> { - if (objectChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final BaseChunkWriter.Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(objectChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(objectChunk.isNull((int) row))); } } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkReader.java index bcdc6283727..cd4b5e91aba 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkReader.java @@ -96,7 +96,9 @@ private static void useDeephavenNulls( final WritableFloatChunk chunk, final int offset) throws IOException { final int numElements = nodeInfo.numElements; - // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes. + // region PayloadDhNulls + // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes + // via LittleEndianCodec (VarHandle on the JVM, GWT-safe arithmetic in the web client's super-source). final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Float.BYTES]; for (int ei = 0; ei < numElements;) { final int n = Math.min(BULK_READ_ELEMENTS, numElements - ei); @@ -106,6 +108,7 @@ private static void useDeephavenNulls( } ei += n; } + // endregion PayloadDhNulls } private static void useValidityBuffer( @@ -117,6 +120,7 @@ private static void useValidityBuffer( final int numElements = nodeInfo.numElements; final int numValidityWords = (numElements + 63) / 64; + // region PayloadValidityBuffer // The payload carries a value slot for every element, including nulls; read it in bounded windows into a // reused buffer and decode each value, then overwrite the invalid positions with the null value. final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Float.BYTES]; @@ -128,6 +132,7 @@ private static void useValidityBuffer( } ei += n; } + // endregion PayloadValidityBuffer int ei = 0; for (int vi = 0; vi < numValidityWords; ++vi) { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkWriter.java index 1da56b24d41..4f7a8eab20e 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/FloatChunkWriter.java @@ -92,26 +92,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final FloatChunk floatChunk = context.getChunk().asFloatChunk(); - subset.forAllRowKeys(row -> { - if (floatChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final FloatChunk floatChunk = context.getChunk().asFloatChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(floatChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(floatChunk.isNull((int) row))); } private class FloatChunkInputStream extends BaseChunkInputStream { @@ -148,8 +134,9 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity buffer bytesWritten += writeValidityBuffer(dos); - // write the payload buffer in bounded windows, encoding each value into little-endian bytes and flushing a - // full window with a single bulk write rather than one DataOutput value (four bytes) at a time + // write the payload buffer in bounded windows, encoding each value into little-endian bytes (via + // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value, + // i.e. one individual byte write per byte of the value, at a time. final FloatChunk floatChunk = context.getChunk().asFloatChunk(); final byte[] buffer = new byte[BULK_WRITE_ELEMENTS * Float.BYTES]; final MutableInt bufferPos = new MutableInt(0); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkReader.java index 8f060b8f0e9..186f169b292 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkReader.java @@ -96,7 +96,9 @@ private static void useDeephavenNulls( final WritableIntChunk chunk, final int offset) throws IOException { final int numElements = nodeInfo.numElements; - // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes. + // region PayloadDhNulls + // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes + // via LittleEndianCodec (VarHandle on the JVM, GWT-safe arithmetic in the web client's super-source). final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Integer.BYTES]; for (int ei = 0; ei < numElements;) { final int n = Math.min(BULK_READ_ELEMENTS, numElements - ei); @@ -106,6 +108,7 @@ private static void useDeephavenNulls( } ei += n; } + // endregion PayloadDhNulls } private static void useValidityBuffer( @@ -117,6 +120,7 @@ private static void useValidityBuffer( final int numElements = nodeInfo.numElements; final int numValidityWords = (numElements + 63) / 64; + // region PayloadValidityBuffer // The payload carries a value slot for every element, including nulls; read it in bounded windows into a // reused buffer and decode each value, then overwrite the invalid positions with the null value. final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Integer.BYTES]; @@ -128,6 +132,7 @@ private static void useValidityBuffer( } ei += n; } + // endregion PayloadValidityBuffer int ei = 0; for (int vi = 0; vi < numValidityWords; ++vi) { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkWriter.java index 4c5049cb27f..cf97db9b738 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/IntChunkWriter.java @@ -92,26 +92,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final IntChunk intChunk = context.getChunk().asIntChunk(); - subset.forAllRowKeys(row -> { - if (intChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final IntChunk intChunk = context.getChunk().asIntChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(intChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(intChunk.isNull((int) row))); } private class IntChunkInputStream extends BaseChunkInputStream { @@ -148,8 +134,9 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity buffer bytesWritten += writeValidityBuffer(dos); - // write the payload buffer in bounded windows, encoding each value into little-endian bytes and flushing a - // full window with a single bulk write rather than one DataOutput value (four bytes) at a time + // write the payload buffer in bounded windows, encoding each value into little-endian bytes (via + // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value, + // i.e. one individual byte write per byte of the value, at a time. final IntChunk intChunk = context.getChunk().asIntChunk(); final byte[] buffer = new byte[BULK_WRITE_ELEMENTS * Integer.BYTES]; final MutableInt bufferPos = new MutableInt(0); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkReader.java index 445a0e33db5..a841934a50e 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkReader.java @@ -94,9 +94,7 @@ public WritableObjectChunk readChunk( throw new IllegalStateException( "list offset buffer is too short for the expected number of elements"); } - for (int ii = 0; ii < numOffsets; ++ii) { - offsets.set(ii, is.readInt()); - } + readIntBuffer(is, offsets, numOffsets); if (offBufRead < offsetsBufferLength) { is.skipBytes(LongSizedDataStructure.intSize(DEBUG_NAME, offsetsBufferLength - offBufRead)); } @@ -109,9 +107,7 @@ public WritableObjectChunk readChunk( throw new IllegalStateException( "list sizes buffer is too short for the expected number of elements"); } - for (int ii = 0; ii < nodeInfo.numElements; ++ii) { - lengths.set(ii, is.readInt()); - } + readIntBuffer(is, lengths, nodeInfo.numElements); if (lenBufRead < lengthsBufferLength) { is.skipBytes(LongSizedDataStructure.intSize(DEBUG_NAME, lengthsBufferLength - lenBufRead)); } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkWriter.java index 4f1fca1a457..eeab1acdf77 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ListChunkWriter.java @@ -15,7 +15,6 @@ import io.deephaven.engine.rowset.RowSetFactory; import io.deephaven.extensions.barrage.BarrageOptions; import io.deephaven.util.datastructures.LongSizedDataStructure; -import io.deephaven.util.mutable.MutableInt; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -45,26 +44,12 @@ public ListChunkWriter( } @Override - protected int computeNullCount( - @NotNull final ChunkWriter.Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> { - if (objectChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final ChunkWriter.Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(objectChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(objectChunk.isNull((int) row))); } @Override @@ -277,12 +262,15 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity array with LSB indexing bytesWritten += writeValidityBuffer(dos); - // write offsets array + // Write the offsets array in bounded windows, flushing a full window with a single bulk write rather than + // one DataOutput int — i.e. four individual byte writes — per element. if (mode == ListChunkReader.Mode.VARIABLE) { // write down only offset (+1) buffer final WritableIntChunk offsetsToUse = myOffsets == null ? context.offsets : myOffsets; - for (int i = 0; i < offsetsToUse.size(); ++i) { - dos.writeInt(offsetsToUse.get(i)); + try (final BulkIntWriter offsetWriter = new BulkIntWriter(outputStream)) { + for (int i = 0; i < offsetsToUse.size(); ++i) { + offsetWriter.write(offsetsToUse.get(i)); + } } bytesWritten += ((long) offsetsToUse.size()) * Integer.BYTES; bytesWritten += writePadBuffer(dos, bytesWritten); @@ -291,15 +279,19 @@ public int drainTo(final OutputStream outputStream) throws IOException { final WritableIntChunk offsetsToUse = myOffsets == null ? context.offsets : myOffsets; // note that we have one extra offset because we keep dense offsets internally - for (int i = 0; i < offsetsToUse.size() - 1; ++i) { - dos.writeInt(offsetsToUse.get(i)); + try (final BulkIntWriter offsetWriter = new BulkIntWriter(outputStream)) { + for (int i = 0; i < offsetsToUse.size() - 1; ++i) { + offsetWriter.write(offsetsToUse.get(i)); + } } bytesWritten += ((long) offsetsToUse.size() - 1) * Integer.BYTES; bytesWritten += writePadBuffer(dos, bytesWritten); // write down length buffer - for (int i = 0; i < offsetsToUse.size() - 1; ++i) { - dos.writeInt(offsetsToUse.get(i + 1) - offsetsToUse.get(i)); + try (final BulkIntWriter lengthWriter = new BulkIntWriter(outputStream)) { + for (int i = 0; i < offsetsToUse.size() - 1; ++i) { + lengthWriter.write(offsetsToUse.get(i + 1) - offsetsToUse.get(i)); + } } bytesWritten += ((long) offsetsToUse.size() - 1) * Integer.BYTES; bytesWritten += writePadBuffer(dos, bytesWritten); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java index 82a1b74eab0..71959ef4b33 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java @@ -25,6 +25,8 @@ private LittleEndianCodec() {} MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.LITTLE_ENDIAN); private static final VarHandle SHORT = MethodHandles.byteArrayViewVarHandle(short[].class, ByteOrder.LITTLE_ENDIAN); + private static final VarHandle CHAR = + MethodHandles.byteArrayViewVarHandle(char[].class, ByteOrder.LITTLE_ENDIAN); private static final VarHandle DOUBLE = MethodHandles.byteArrayViewVarHandle(double[].class, ByteOrder.LITTLE_ENDIAN); private static final VarHandle FLOAT = @@ -42,6 +44,15 @@ static short getShort(final byte[] b, final int o) { return (short) SHORT.get(b, o); } + static char getChar(final byte[] b, final int o) { + return (char) CHAR.get(b, o); + } + + /** A single byte has no byte order; present so the replicated readers/writers can share one shape. */ + static byte getByte(final byte[] b, final int o) { + return b[o]; + } + static double getDouble(final byte[] b, final int o) { return (double) DOUBLE.get(b, o); } @@ -62,6 +73,15 @@ static void putShort(final byte[] b, final int o, final short v) { SHORT.set(b, o, v); } + static void putChar(final byte[] b, final int o, final char v) { + CHAR.set(b, o, v); + } + + /** A single byte has no byte order; present so the replicated readers/writers can share one shape. */ + static void putByte(final byte[] b, final int o, final byte v) { + b[o] = v; + } + static void putDouble(final byte[] b, final int o, final double v) { DOUBLE.set(b, o, v); } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkReader.java index 3dd1e968706..ee694ae7b28 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkReader.java @@ -95,7 +95,9 @@ private static void useDeephavenNulls( final WritableLongChunk chunk, final int offset) throws IOException { final int numElements = nodeInfo.numElements; - // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes. + // region PayloadDhNulls + // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes + // via LittleEndianCodec (VarHandle on the JVM, GWT-safe arithmetic in the web client's super-source). final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Long.BYTES]; for (int ei = 0; ei < numElements;) { final int n = Math.min(BULK_READ_ELEMENTS, numElements - ei); @@ -105,6 +107,7 @@ private static void useDeephavenNulls( } ei += n; } + // endregion PayloadDhNulls } private static void useValidityBuffer( @@ -116,6 +119,7 @@ private static void useValidityBuffer( final int numElements = nodeInfo.numElements; final int numValidityWords = (numElements + 63) / 64; + // region PayloadValidityBuffer // The payload carries a value slot for every element, including nulls; read it in bounded windows into a // reused buffer and decode each value, then overwrite the invalid positions with the null value. final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Long.BYTES]; @@ -127,6 +131,7 @@ private static void useValidityBuffer( } ei += n; } + // endregion PayloadValidityBuffer int ei = 0; for (int vi = 0; vi < numValidityWords; ++vi) { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkWriter.java index 879e5406c8c..53d01efe0e7 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/LongChunkWriter.java @@ -92,26 +92,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final LongChunk longChunk = context.getChunk().asLongChunk(); - subset.forAllRowKeys(row -> { - if (longChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final LongChunk longChunk = context.getChunk().asLongChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(longChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(longChunk.isNull((int) row))); } private class LongChunkInputStream extends BaseChunkInputStream { @@ -148,8 +134,9 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity buffer bytesWritten += writeValidityBuffer(dos); - // write the payload buffer in bounded windows, encoding each value into little-endian bytes and flushing a - // full window with a single bulk write rather than one DataOutput value (eight bytes) at a time + // write the payload buffer in bounded windows, encoding each value into little-endian bytes (via + // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value, + // i.e. one individual byte write per byte of the value, at a time. final LongChunk longChunk = context.getChunk().asLongChunk(); final byte[] buffer = new byte[BULK_WRITE_ELEMENTS * Long.BYTES]; final MutableInt bufferPos = new MutableInt(0); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkReader.java index 8c33c0d0c00..7b354ed42d0 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkReader.java @@ -83,9 +83,7 @@ public WritableObjectChunk readChunk( throw new IllegalStateException( "map offset buffer is too short for the expected number of elements"); } - for (int ii = 0; ii < numOffsets; ++ii) { - offsets.set(ii, is.readInt()); - } + readIntBuffer(is, offsets, numOffsets); if (offBufRead < offsetsBufferLength) { is.skipBytes(LongSizedDataStructure.intSize(DEBUG_NAME, offsetsBufferLength - offBufRead)); } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkWriter.java index 6ac2d54fa89..fe9d0ced308 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/MapChunkWriter.java @@ -19,7 +19,6 @@ import io.deephaven.engine.table.impl.util.unboxer.ChunkUnboxer; import io.deephaven.extensions.barrage.BarrageOptions; import io.deephaven.util.datastructures.LongSizedDataStructure; -import io.deephaven.util.mutable.MutableInt; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; @@ -57,26 +56,12 @@ public Context makeContext( } @Override - protected int computeNullCount( - @NotNull final BaseChunkWriter.Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> { - if (objectChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final BaseChunkWriter.Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(objectChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(objectChunk.isNull((int) row))); } public final class Context extends ChunkWriter.Context { @@ -305,10 +290,13 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity array with LSB indexing bytesWritten += writeValidityBuffer(dos); - // write offsets array + // Write the offsets array in bounded windows, flushing a full window with a single bulk write rather than + // one DataOutput int — i.e. four individual byte writes — per element. final WritableIntChunk offsetsToUse = myOffsets == null ? context.offsets : myOffsets; - for (int i = 0; i < offsetsToUse.size(); ++i) { - dos.writeInt(offsetsToUse.get(i)); + try (final BulkIntWriter offsetWriter = new BulkIntWriter(outputStream)) { + for (int i = 0; i < offsetsToUse.size(); ++i) { + offsetWriter.write(offsetsToUse.get(i)); + } } bytesWritten += ((long) offsetsToUse.size()) * Integer.BYTES; bytesWritten += writePadBuffer(dos, bytesWritten); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/NullChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/NullChunkWriter.java index 9c4468e205d..f3b5a0012f8 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/NullChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/NullChunkWriter.java @@ -35,16 +35,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount(@NotNull final Context context, @NotNull final RowSequence subset) { - return subset.intSize("NullChunkWriter"); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { - // nothing to do; this is a null column + @NotNull final ValidityBuffer validity) { + // every element of a null column is null + validity.setNextAreNull(subset.intSize(DEBUG_NAME)); } public static class NullDrainableColumn extends DrainableColumn { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/RunEndEncodedChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/RunEndEncodedChunkWriter.java index 6e84e2d63e5..a28ec8ed924 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/RunEndEncodedChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/RunEndEncodedChunkWriter.java @@ -62,19 +62,12 @@ public RunEndEncodedChunkWriter( } @Override - protected int computeNullCount( - @NotNull final ChunkWriter.Context context, - @NotNull final RowSequence subset) { - // The REE parent array never contains nulls; nullability lives in the values child. - return 0; - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final ChunkWriter.Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { - // REE parent has no validity buffer. + @NotNull final ValidityBuffer validity) { + // The REE parent array never contains nulls (so it has no validity buffer); nullability lives in the values + // child. Leaving the bitmap empty reports a null count of zero, which suppresses the buffer. } @Override diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkReader.java index 7938337d62a..458e7b9cb86 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkReader.java @@ -96,7 +96,9 @@ private static void useDeephavenNulls( final WritableShortChunk chunk, final int offset) throws IOException { final int numElements = nodeInfo.numElements; - // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes. + // region PayloadDhNulls + // Read the payload in bounded windows into a reused buffer and decode each value from its little-endian bytes + // via LittleEndianCodec (VarHandle on the JVM, GWT-safe arithmetic in the web client's super-source). final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Short.BYTES]; for (int ei = 0; ei < numElements;) { final int n = Math.min(BULK_READ_ELEMENTS, numElements - ei); @@ -106,6 +108,7 @@ private static void useDeephavenNulls( } ei += n; } + // endregion PayloadDhNulls } private static void useValidityBuffer( @@ -117,6 +120,7 @@ private static void useValidityBuffer( final int numElements = nodeInfo.numElements; final int numValidityWords = (numElements + 63) / 64; + // region PayloadValidityBuffer // The payload carries a value slot for every element, including nulls; read it in bounded windows into a // reused buffer and decode each value, then overwrite the invalid positions with the null value. final byte[] buffer = new byte[Math.min(numElements, BULK_READ_ELEMENTS) * Short.BYTES]; @@ -128,6 +132,7 @@ private static void useValidityBuffer( } ei += n; } + // endregion PayloadValidityBuffer int ei = 0; for (int vi = 0; vi < numValidityWords; ++vi) { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkWriter.java index a82f69ab248..d6ef01430e7 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ShortChunkWriter.java @@ -92,26 +92,12 @@ public DrainableColumn getInputStream( } @Override - protected int computeNullCount( - @NotNull final Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final ShortChunk shortChunk = context.getChunk().asShortChunk(); - subset.forAllRowKeys(row -> { - if (shortChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final ShortChunk shortChunk = context.getChunk().asShortChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(shortChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(shortChunk.isNull((int) row))); } private class ShortChunkInputStream extends BaseChunkInputStream { @@ -148,8 +134,9 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity buffer bytesWritten += writeValidityBuffer(dos); - // write the payload buffer in bounded windows, encoding each value into little-endian bytes and flushing a - // full window with a single bulk write rather than one DataOutput value (two bytes) at a time + // write the payload buffer in bounded windows, encoding each value into little-endian bytes (via + // LittleEndianCodec) and flushing a full window with a single bulk write rather than one DataOutput value, + // i.e. one individual byte write per byte of the value, at a time. final ShortChunk shortChunk = context.getChunk().asShortChunk(); final byte[] buffer = new byte[BULK_WRITE_ELEMENTS * Short.BYTES]; final MutableInt bufferPos = new MutableInt(0); diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/UnionChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/UnionChunkWriter.java index 25a35919ef3..933da197c5d 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/UnionChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/UnionChunkWriter.java @@ -62,26 +62,12 @@ public Context makeContext( } @Override - protected int computeNullCount( - @NotNull final ChunkWriter.Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); - final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> { - if (objectChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal( + protected void computeValidity( @NotNull final ChunkWriter.Context context, @NotNull final RowSequence subset, - @NotNull final SerContext serContext) { + @NotNull final ValidityBuffer validity) { final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(objectChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(objectChunk.isNull((int) row))); } public final class Context extends ChunkWriter.Context { diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkReader.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkReader.java index 2ed74b27960..66becbb81f3 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkReader.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkReader.java @@ -78,9 +78,7 @@ public WritableObjectChunk readChunk( if (offsetsBuffer < offBufRead) { throw new IllegalStateException("offset buffer is too short for the expected number of elements"); } - for (int i = 0; i < numElements + 1; ++i) { - offsets.set(i, is.readInt()); - } + readIntBuffer(is, offsets, numElements + 1); if (offBufRead < offsetsBuffer) { is.skipBytes(LongSizedDataStructure.intSize(DEBUG_NAME, offsetsBuffer - offBufRead)); } diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkWriter.java index 69697a1a618..356e1baf7aa 100644 --- a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkWriter.java +++ b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/VarBinaryChunkWriter.java @@ -57,24 +57,12 @@ public Context makeContext( } @Override - protected int computeNullCount( + protected void computeValidity( @NotNull final ChunkWriter.Context context, - @NotNull final RowSequence subset) { - final MutableInt nullCount = new MutableInt(0); + @NotNull final RowSequence subset, + @NotNull final ValidityBuffer validity) { final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> { - if (objectChunk.isNull((int) row)) { - nullCount.increment(); - } - }); - return nullCount.get(); - } - - @Override - protected void writeValidityBufferInternal(ChunkWriter.@NotNull Context context, @NotNull RowSequence subset, - @NotNull SerContext serContext) { - final ObjectChunk objectChunk = context.getChunk().asObjectChunk(); - subset.forAllRowKeys(row -> serContext.setNextIsNull(objectChunk.isNull((int) row))); + subset.forAllRowKeys(row -> validity.setNextIsNull(objectChunk.isNull((int) row))); } public final class Context extends ChunkWriter.Context { @@ -190,31 +178,30 @@ public int drainTo(final OutputStream outputStream) throws IOException { // write the validity buffer bytesWritten.add(writeValidityBuffer(dos)); - // write offsets array - if (!subset.isEmpty()) { - dos.writeInt(0); - } + // Write the offsets array in bounded windows, flushing a full window with a single bulk write rather than + // one DataOutput int — i.e. four individual byte writes — per row. + long numRows = subset.size(); + try (final BulkIntWriter offsetWriter = new BulkIntWriter(outputStream)) { + if (!subset.isEmpty()) { + offsetWriter.write(0); + } - final MutableInt logicalSize = new MutableInt(); - subset.forAllRowKeys((idx) -> { - try { + final MutableInt logicalSize = new MutableInt(); + subset.forAllRowKeys((idx) -> { logicalSize.add(LongSizedDataStructure.intSize("int cast", context.byteStorage.getPayloadSize((int) idx, (int) idx))); - dos.writeInt(logicalSize.get()); - } catch (final IOException e) { - throw new UncheckedDeephavenException("couldn't drain data to OutputStream", e); + offsetWriter.write(logicalSize.get()); + }); + if (numRows > 0) { + numRows += 1; } - }); - long numRows = subset.size(); - if (numRows > 0) { - numRows += 1; - } - bytesWritten.add(Integer.BYTES * numRows); + bytesWritten.add(Integer.BYTES * numRows); - if (!subset.isEmpty() && (subset.size() & 0x1) == 0) { - // then we must pad to align next buffer - dos.writeInt(0); - bytesWritten.add(Integer.BYTES); + if (!subset.isEmpty() && (subset.size() & 0x1) == 0) { + // then we must pad to align next buffer + offsetWriter.write(0); + bytesWritten.add(Integer.BYTES); + } } subset.forAllRowKeyRanges((s, e) -> { diff --git a/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/BarrageColumnRoundTripTest.java b/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/BarrageColumnRoundTripTest.java index 74cb760ca39..6844974fbe3 100644 --- a/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/BarrageColumnRoundTripTest.java +++ b/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/BarrageColumnRoundTripTest.java @@ -74,6 +74,7 @@ import java.util.Random; import java.util.function.Consumer; import java.util.function.IntFunction; +import java.util.function.IntPredicate; import java.util.stream.LongStream; public class BarrageColumnRoundTripTest extends RefreshingTableTestCase { @@ -84,6 +85,14 @@ public class BarrageColumnRoundTripTest extends RefreshingTableTestCase { private static final int FIXED_LIST_LEN = 4; private static final int MAX_LIST_LEN = 10; + /** + * Rows per round trip. Deliberately more than two 64-element validity words, and not a multiple of 64: the validity + * bitmap is packed a word at a time on the write side and walked in runs of valid/null slots on the read side, so a + * chunk narrower than a single word would leave every word-boundary transition and the partial trailing word + * untested. + */ + private static final int ROUND_TRIP_NUM_ROWS = 133; + private static final BarrageSubscriptionOptions OPT_DEFAULT = BarrageSubscriptionOptions.builder() .build(); private static final BarrageSubscriptionOptions OPT_DH_NULLS = @@ -637,6 +646,130 @@ public void testZDTAsLongChunkSerialization() throws IOException { } } + /** + * Positions at which a run of nulls or of valid values starting there straddles a 64-element validity word + * boundary. {@link #ROUND_TRIP_NUM_ROWS} is wide enough for all of them. + */ + private static final int[] BOUNDARY_PIVOTS = {0, 1, 61, 62, 63, 64, 65, 66, 127, 128, 129}; + + /** + * Round trips a column for each null pattern that turns over on a validity word boundary, plus the all-null and + * all-valid extremes. The writer packs the bitmap a word at a time and the reader walks it in runs of valid and + * null slots, so these are the transitions where either side can drop or misplace an element; the type-specific + * tests above use patterns like {@code i % 7} that do not land on them deliberately. + */ + private static void roundTripEachBoundaryPattern(final BoundaryPatternRunner runner) throws IOException { + runner.run("all valid", index -> false); + runner.run("all null", index -> true); + for (final int pivot : BOUNDARY_PIVOTS) { + runner.run("first null at " + pivot, index -> index >= pivot); + runner.run("first valid at " + pivot, index -> index < pivot); + runner.run("lone null at " + pivot, index -> index == pivot); + runner.run("lone valid at " + pivot, index -> index != pivot); + } + } + + private interface BoundaryPatternRunner { + void run(String description, IntPredicate isNull) throws IOException; + } + + public void testIntValidityWordBoundaries() throws IOException { + for (final BarrageSubscriptionOptions opts : OPTIONS) { + roundTripEachBoundaryPattern((description, isNull) -> testRoundTripSerialization( + SpecialMode.NONE, opts, int.class, + (utO) -> { + final WritableIntChunk chunk = utO.asWritableIntChunk(); + for (int i = 0; i < chunk.size(); ++i) { + chunk.set(i, isNull.test(i) ? QueryConstants.NULL_INT : i * 31 + 7); + } + }, + (utO, utC, subset, offset) -> { + final WritableIntChunk original = utO.asWritableIntChunk(); + final WritableIntChunk computed = utC.asWritableIntChunk(); + final RowSequence rows = + subset == null ? RowSetFactory.flat(original.size()) : subset; + final MutableInt off = new MutableInt(); + rows.forAllRowKeys(key -> Assert.equals(original.get((int) key), description, + computed.get(offset + off.getAndIncrement()), "computed")); + })); + } + } + + public void testObjectValidityWordBoundaries() throws IOException { + // the var-binary path keeps its own offsets buffer alongside the validity bitmap + for (final BarrageSubscriptionOptions opts : new BarrageSubscriptionOptions[] {OPT_DEFAULT, OPT_DH_NULLS}) { + roundTripEachBoundaryPattern((description, isNull) -> testRoundTripSerialization( + SpecialMode.NONE, opts, Object.class, + (utO) -> { + final WritableObjectChunk chunk = utO.asWritableObjectChunk(); + for (int i = 0; i < chunk.size(); ++i) { + chunk.set(i, isNull.test(i) ? null : Integer.toString(i)); + } + }, + new ObjectIdentityValidator())); + } + } + + /** + * The fixed-width payload writers gather values into a {@code BaseChunkWriter#BULK_WRITE_BUFFER_BYTES} window and + * flush it with a single write. At the default 4096 bytes that takes 512 doubles, 1024 ints or 4096 bytes, so + * {@link #ROUND_TRIP_NUM_ROWS} never fills one and the flush-a-full-window branch goes untaken; real messages fill + * it constantly. This covers a full window plus the partial remainder that follows, for the narrowest and widest + * elements and one in between. + */ + public void testBulkWritePayloadWindowFlush() throws IOException { + final int wideRows = 5000; + for (final BarrageSubscriptionOptions opts : new BarrageSubscriptionOptions[] {OPT_DEFAULT, OPT_DH_NULLS}) { + testRoundTripSerialization(SpecialMode.NONE, opts, int.class, wideRows, + (utO) -> { + final WritableIntChunk chunk = utO.asWritableIntChunk(); + for (int i = 0; i < chunk.size(); ++i) { + chunk.set(i, i % 97 == 0 ? QueryConstants.NULL_INT : i * 31 + 7); + } + }, + (utO, utC, subset, offset) -> { + final WritableIntChunk original = utO.asWritableIntChunk(); + final WritableIntChunk computed = utC.asWritableIntChunk(); + final RowSequence rows = subset == null ? RowSetFactory.flat(original.size()) : subset; + final MutableInt off = new MutableInt(); + rows.forAllRowKeys(key -> Assert.equals(original.get((int) key), "original", + computed.get(offset + off.getAndIncrement()), "computed")); + }); + + testRoundTripSerialization(SpecialMode.NONE, opts, byte.class, wideRows, + (utO) -> { + final WritableByteChunk chunk = utO.asWritableByteChunk(); + for (int i = 0; i < chunk.size(); ++i) { + chunk.set(i, i % 97 == 0 ? QueryConstants.NULL_BYTE : (byte) (i * 31 + 7)); + } + }, + (utO, utC, subset, offset) -> { + final WritableByteChunk original = utO.asWritableByteChunk(); + final WritableByteChunk computed = utC.asWritableByteChunk(); + final RowSequence rows = subset == null ? RowSetFactory.flat(original.size()) : subset; + final MutableInt off = new MutableInt(); + rows.forAllRowKeys(key -> Assert.equals(original.get((int) key), "original", + computed.get(offset + off.getAndIncrement()), "computed")); + }); + + testRoundTripSerialization(SpecialMode.NONE, opts, double.class, wideRows, + (utO) -> { + final WritableDoubleChunk chunk = utO.asWritableDoubleChunk(); + for (int i = 0; i < chunk.size(); ++i) { + chunk.set(i, i % 97 == 0 ? QueryConstants.NULL_DOUBLE : i * 2.25); + } + }, + (utO, utC, subset, offset) -> { + final WritableDoubleChunk original = utO.asWritableDoubleChunk(); + final WritableDoubleChunk computed = utC.asWritableDoubleChunk(); + final RowSequence rows = subset == null ? RowSetFactory.flat(original.size()) : subset; + final MutableInt off = new MutableInt(); + rows.forAllRowKeys(key -> Assert.equals(original.get((int) key), "original", + computed.get(offset + off.getAndIncrement()), "computed")); + }); + } + } + public void testObjectSerialization() throws IOException { testRoundTripSerialization(SpecialMode.NONE, OPT_DEFAULT, Object.class, initObjectChunk(Integer::toString), new ObjectIdentityValidator<>()); @@ -1814,7 +1947,17 @@ private static void testRoundTripSerialization( Class type, final Consumer> initData, final Validator validator) throws IOException { - final int NUM_ROWS = 8; + testRoundTripSerialization(mode, options, type, ROUND_TRIP_NUM_ROWS, initData, validator); + } + + private static void testRoundTripSerialization( + final SpecialMode mode, + final BarrageSubscriptionOptions options, + Class type, + final int chunkRows, + final Consumer> initData, + final Validator validator) throws IOException { + final int NUM_ROWS = chunkRows; final ChunkType chunkType; final Class readType; if (type == ZonedDateTime.class) { diff --git a/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/BooleanChunkWriterTest.java b/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/BooleanChunkWriterTest.java new file mode 100644 index 00000000000..1811a083ecc --- /dev/null +++ b/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/BooleanChunkWriterTest.java @@ -0,0 +1,109 @@ +// +// Copyright (c) 2016-2026 Deephaven Data Labs and Patent Pending +// +package io.deephaven.extensions.barrage.chunk; + +import io.deephaven.chunk.ByteChunk; +import io.deephaven.chunk.WritableByteChunk; +import io.deephaven.chunk.attributes.Values; +import io.deephaven.extensions.barrage.BarrageSubscriptionOptions; +import io.deephaven.util.BooleanUtils; +import org.junit.Test; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.function.IntFunction; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Tests the wire bytes {@link BooleanChunkWriter} produces. + *

+ * A boolean column is two bit-packed buffers: a validity bitmap (set bit = non-null, omitted entirely when there are no + * nulls) followed by the payload (set bit = TRUE). The writer packs the payload with the same + * {@link BaseChunkWriter.ValidityBuffer} it uses for validity, so an all-TRUE column reaches + * {@link BaseChunkWriter.ValidityBuffer#bytes()} without ever having appended a "null" — the case that has to be + * materialized up front rather than refused. + */ +public class BooleanChunkWriterTest { + + private static final BarrageSubscriptionOptions OPTS = BarrageSubscriptionOptions.builder().build(); + + /** Not a multiple of 64, so the bitmaps carry a partial trailing word. */ + private static final int NUM_ROWS = 133; + + private static final int BITMAP_BYTES = ((NUM_ROWS + 63) / 64) * 8; + + /** Bit {@code ii} set when {@code predicate} holds; bits past the last element left clear. */ + private static byte[] expectedBitmap(final IntFunction predicate) { + final byte[] expected = new byte[BITMAP_BYTES]; + for (int ii = 0; ii < NUM_ROWS; ++ii) { + if (predicate.apply(ii)) { + expected[ii / 8] |= (byte) (1 << (ii & 7)); + } + } + return expected; + } + + /** Drains a column of {@code values} and returns the bytes, asserting the reported null count along the way. */ + private static byte[] drain(final IntFunction values, final int expectedNullCount) throws IOException { + final BooleanChunkWriter> writer = BooleanChunkWriter.getIdentity(true); + final WritableByteChunk chunk = WritableByteChunk.writableChunkWrap(new byte[NUM_ROWS]); + for (int ii = 0; ii < NUM_ROWS; ++ii) { + chunk.set(ii, BooleanUtils.booleanAsByte(values.apply(ii))); + } + final ByteArrayOutputStream out = new ByteArrayOutputStream(); + try (final ChunkWriter.Context context = writer.makeContext(chunk, 0); + final ChunkWriter.DrainableColumn column = writer.getInputStream(context, null, OPTS)) { + assertThat(column.nullCount()).as("nullCount").isEqualTo(expectedNullCount); + column.drainTo(out); + } + return out.toByteArray(); + } + + @Test + public void allTrueOmitsValidityAndPacksEveryBit() throws IOException { + // No null, so no validity buffer is sent and the payload is the only thing on the wire. This is the case that + // asks the packer for bytes without ever having appended a "null". + final byte[] bytes = drain(index -> Boolean.TRUE, 0); + assertThat(bytes).hasSize(BITMAP_BYTES); + assertThat(bytes).containsExactly(expectedBitmap(index -> true)); + } + + @Test + public void allFalseOmitsValidityAndPacksNoBits() throws IOException { + final byte[] bytes = drain(index -> Boolean.FALSE, 0); + assertThat(bytes).hasSize(BITMAP_BYTES); + assertThat(bytes).containsExactly(new byte[BITMAP_BYTES]); + } + + @Test + public void allNullSendsAnEmptyValidityAndNoPayloadBits() throws IOException { + final byte[] bytes = drain(index -> null, NUM_ROWS); + assertThat(bytes).hasSize(2 * BITMAP_BYTES); + // validity: every element null, so no bit set; payload: nothing is TRUE + assertThat(bytes).containsExactly(new byte[2 * BITMAP_BYTES]); + } + + @Test + public void mixedValuesPackValidityThenPayload() throws IOException { + // null every 5th, TRUE on even indices; transitions land either side of the 64-element word boundaries + final IntFunction values = index -> index % 5 == 0 ? null : (index % 2 == 0); + int expectedNulls = 0; + for (int ii = 0; ii < NUM_ROWS; ++ii) { + if (values.apply(ii) == null) { + ++expectedNulls; + } + } + + final byte[] bytes = drain(values, expectedNulls); + assertThat(bytes).hasSize(2 * BITMAP_BYTES); + + final byte[] expectedValidity = expectedBitmap(index -> values.apply(index) != null); + final byte[] expectedPayload = expectedBitmap(index -> values.apply(index) == Boolean.TRUE); + final byte[] expected = new byte[2 * BITMAP_BYTES]; + System.arraycopy(expectedValidity, 0, expected, 0, BITMAP_BYTES); + System.arraycopy(expectedPayload, 0, expected, BITMAP_BYTES, BITMAP_BYTES); + assertThat(bytes).containsExactly(expected); + } +} diff --git a/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/ValidityBufferTest.java b/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/ValidityBufferTest.java new file mode 100644 index 00000000000..b9a63cb453c --- /dev/null +++ b/extensions/barrage/src/test/java/io/deephaven/extensions/barrage/chunk/ValidityBufferTest.java @@ -0,0 +1,283 @@ +// +// Copyright (c) 2016-2026 Deephaven Data Labs and Patent Pending +// +package io.deephaven.extensions.barrage.chunk; + +import org.junit.Test; + +import java.util.Arrays; +import java.util.Random; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Tests {@link BaseChunkWriter.ValidityBuffer}, which packs the Arrow validity bitmap and counts nulls in a single + * traversal. + *

+ * The buffer only materializes its bytes once a null appears, back-filling the all-valid words it skipped, so its + * behavior turns over at 64-element word boundaries. Everything here is therefore driven across those boundaries + * explicitly; the expected bitmaps come from a deliberately independent byte-oriented reference + * ({@link #expectedBitmap}) rather than from the word-oriented little-endian packing the implementation uses. + */ +public class ValidityBufferTest { + + /** + * Arrow validity bitmap for {@code isNull}, computed a byte at a time: bit {@code ii} is set when element + * {@code ii} is valid, and bits past the last element stay clear. + */ + private static byte[] expectedBitmap(final boolean[] isNull) { + final byte[] expected = new byte[((isNull.length + 63) / 64) * 8]; + for (int ii = 0; ii < isNull.length; ++ii) { + if (!isNull[ii]) { + expected[ii / 8] |= (byte) (1 << (ii & 7)); + } + } + return expected; + } + + private static int countNulls(final boolean[] isNull) { + int nulls = 0; + for (final boolean elementIsNull : isNull) { + if (elementIsNull) { + ++nulls; + } + } + return nulls; + } + + private static BaseChunkWriter.ValidityBuffer fill( + final BaseChunkWriter.ValidityBuffer buffer, + final boolean[] isNull) { + for (final boolean elementIsNull : isNull) { + buffer.setNextIsNull(elementIsNull); + } + return buffer; + } + + /** + * Feed {@code isNull} through both flavors of buffer and assert their outputs. + *

+ * The lazily materialized buffer only has bytes once a null has been appended; with no nulls the column sends no + * validity buffer at all, so asking for the bytes is a caller bug and must throw rather than quietly produce an + * all-valid bitmap. The eagerly {@link BaseChunkWriter.ValidityBuffer#packed packed} buffer always has bytes, and + * they must agree with the lazy one wherever both are legal. The null count is read before the bytes, matching the + * order {@code BaseChunkInputStream} uses: it needs the count for the field node before draining. + */ + private static void assertPacksTo(final String description, final boolean[] isNull) { + final int expectedNulls = countNulls(isNull); + final byte[] expected = expectedBitmap(isNull); + + final BaseChunkWriter.ValidityBuffer lazy = fill(new BaseChunkWriter.ValidityBuffer(isNull.length), isNull); + assertThat(lazy.nullCount()).as("%s: lazy nullCount", description).isEqualTo(expectedNulls); + if (expectedNulls == 0) { + assertThatThrownBy(lazy::bytes) + .as("%s: lazy bytes() with no null appended", description) + .isInstanceOf(IllegalStateException.class); + } else { + assertThat(lazy.bytes()).as("%s: lazy bitmap", description).containsExactly(expected); + } + + final BaseChunkWriter.ValidityBuffer packed = + fill(BaseChunkWriter.ValidityBuffer.packed(isNull.length), isNull); + assertThat(packed.nullCount()).as("%s: packed nullCount", description).isEqualTo(expectedNulls); + assertThat(packed.bytes()).as("%s: packed bitmap", description).containsExactly(expected); + } + + /** Sizes that bracket the word boundaries: empty, sub-word, exact multiples, and one either side. */ + private static final int[] SIZES = {0, 1, 7, 8, 62, 63, 64, 65, 66, 127, 128, 129, 133, 192, 193}; + + /** Positions where a run of nulls or valid elements starting there straddles a word boundary. */ + private static final int[] BOUNDARY_POSITIONS = {0, 1, 61, 62, 63, 64, 65, 66, 126, 127, 128, 129, 130}; + + @Test + public void allElementsNull() { + for (final int size : SIZES) { + final boolean[] isNull = new boolean[size]; + Arrays.fill(isNull, true); + assertPacksTo("all null, size " + size, isNull); + } + } + + @Test + public void allElementsValid() { + // No null ever arrives: the lazy buffer must refuse to hand out bytes, while the packed buffer -- the shape + // BooleanChunkWriter uses for a payload of all-TRUE values -- must produce the full all-ones bitmap. + for (final int size : SIZES) { + assertPacksTo("all valid, size " + size, new boolean[size]); + } + } + + @Test + public void firstNullAtEveryPosition() { + final int size = 200; + for (int nullAt = 0; nullAt < size; ++nullAt) { + final boolean[] isNull = new boolean[size]; + isNull[nullAt] = true; + assertPacksTo("single null at " + nullAt, isNull); + } + } + + @Test + public void firstValidAtEveryPosition() { + final int size = 200; + for (int validAt = 0; validAt < size; ++validAt) { + final boolean[] isNull = new boolean[size]; + Arrays.fill(isNull, true); + isNull[validAt] = false; + assertPacksTo("single valid at " + validAt, isNull); + } + } + + @Test + public void validRunThenNullsFromBoundary() { + // Valid up to the pivot, null from there on: the first null lands at the pivot, so allocate() must back-fill + // the words already passed with all-ones. + final int size = 200; + for (final int pivot : BOUNDARY_POSITIONS) { + final boolean[] isNull = new boolean[size]; + for (int ii = pivot; ii < size; ++ii) { + isNull[ii] = true; + } + assertPacksTo("valid then null from " + pivot, isNull); + } + } + + @Test + public void nullRunThenValidFromBoundary() { + // Null up to the pivot, valid from there on: the first valid element lands at the pivot. + final int size = 200; + for (final int pivot : BOUNDARY_POSITIONS) { + final boolean[] isNull = new boolean[size]; + for (int ii = 0; ii < pivot && ii < size; ++ii) { + isNull[ii] = true; + } + assertPacksTo("null then valid from " + pivot, isNull); + } + } + + @Test + public void singleNullOrValidRunSpanningEachBoundary() { + // A short run placed so that it straddles a word boundary, in both polarities. + final int size = 200; + for (final int start : BOUNDARY_POSITIONS) { + for (final int length : new int[] {1, 2, 3, 64, 65}) { + if (start + length > size) { + continue; + } + final boolean[] nullRun = new boolean[size]; + for (int ii = start; ii < start + length; ++ii) { + nullRun[ii] = true; + } + assertPacksTo("null run [" + start + "," + (start + length) + ")", nullRun); + + final boolean[] validRun = new boolean[size]; + Arrays.fill(validRun, true); + for (int ii = start; ii < start + length; ++ii) { + validRun[ii] = false; + } + assertPacksTo("valid run [" + start + "," + (start + length) + ")", validRun); + } + } + } + + @Test + public void trailingBitsPastLastElementAreClear() { + // The wire format leaves the tail of the final word undefined, but we must keep emitting zeros there: the + // bytes are compared against other Barrage implementations byte for byte. + for (final int size : SIZES) { + if ((size & 63) == 0) { + continue; + } + for (final boolean fillNull : new boolean[] {false, true}) { + final boolean[] isNull = new boolean[size]; + Arrays.fill(isNull, fillNull); + final byte[] bytes = fill(BaseChunkWriter.ValidityBuffer.packed(size), isNull).bytes(); + for (int bit = size; bit < bytes.length * 8; ++bit) { + assertThat((bytes[bit / 8] >> (bit & 7)) & 1) + .as("size %d fillNull %b: bit %d past the last element", size, fillNull, bit) + .isEqualTo(0); + } + } + } + } + + @Test + public void setNextAreNullMatchesIndividualCalls() { + // NullChunkWriter reports a whole column of nulls at once; that shortcut has to land on the same bytes as the + // element-at-a-time path. A suffix follows the bulk run so that any drift in the bulk position bookkeeping + // shows up as misplaced bits: within the run itself no bit is ever set, which hides such drift. + final int suffixLength = 70; + // A prefix of only valid elements leaves the buffer unmaterialized when the bulk run starts; a prefix that + // already contains a null means the bulk run must append to an existing buffer rather than rebuild it. + for (final boolean prefixHasNull : new boolean[] {false, true}) { + for (final int prefixLength : new int[] {0, 1, 62, 63, 64, 65}) { + for (final int nullRun : new int[] {0, 1, 2, 63, 64, 65, 128}) { + final int size = prefixLength + nullRun + suffixLength; + final boolean[] isNull = new boolean[size]; + if (prefixHasNull && prefixLength > 0) { + // one null early in the prefix, so the buffer is already materialized + isNull[0] = true; + } + for (int ii = prefixLength; ii < prefixLength + nullRun; ++ii) { + isNull[ii] = true; + } + for (int ii = 0; ii < suffixLength; ++ii) { + isNull[prefixLength + nullRun + ii] = ii % 3 == 0; + } + + final BaseChunkWriter.ValidityBuffer bulk = new BaseChunkWriter.ValidityBuffer(size); + for (int ii = 0; ii < prefixLength; ++ii) { + bulk.setNextIsNull(isNull[ii]); + } + bulk.setNextAreNull(nullRun); + for (int ii = prefixLength + nullRun; ii < size; ++ii) { + bulk.setNextIsNull(isNull[ii]); + } + + final String description = prefixLength + " prefix (hasNull " + prefixHasNull + + "), setNextAreNull(" + nullRun + "), then " + suffixLength + " mixed"; + assertThat(bulk.nullCount()).as("%s: nullCount", description).isEqualTo(countNulls(isNull)); + assertThat(bulk.bytes()).as("%s: bitmap", description).containsExactly(expectedBitmap(isNull)); + } + } + } + } + + @Test + public void bytesIsRepeatableAndIndependentOfNullCountOrder() { + final boolean[] isNull = new boolean[133]; + for (int ii = 0; ii < isNull.length; ++ii) { + isNull[ii] = ii == 63 || ii == 64 || ii >= 130; + } + final byte[] expected = expectedBitmap(isNull); + + // bytes() first, then nullCount() + final BaseChunkWriter.ValidityBuffer bytesFirst = new BaseChunkWriter.ValidityBuffer(isNull.length); + for (final boolean elementIsNull : isNull) { + bytesFirst.setNextIsNull(elementIsNull); + } + assertThat(bytesFirst.bytes()).as("bytes() before nullCount()").containsExactly(expected); + assertThat(bytesFirst.nullCount()).as("nullCount() after bytes()").isEqualTo(countNulls(isNull)); + assertThat(bytesFirst.bytes()).as("bytes() called twice").containsExactly(expected); + + // and the production order, nullCount() first + assertPacksTo("nullCount() before bytes()", isNull); + } + + @Test + public void randomPatterns() { + final long seed = new Random().nextLong(); + System.out.println("ValidityBufferTest.randomPatterns seed: " + seed); + final Random random = new Random(seed); + for (final int size : SIZES) { + for (int trial = 0; trial < 20; ++trial) { + final boolean[] isNull = new boolean[size]; + for (int ii = 0; ii < size; ++ii) { + isNull[ii] = random.nextInt(4) == 0; + } + assertPacksTo("seed " + seed + ", random size " + size + " trial " + trial, isNull); + } + } + } +} diff --git a/replication/static/src/main/java/io/deephaven/replicators/ReplicateBarrageUtils.java b/replication/static/src/main/java/io/deephaven/replicators/ReplicateBarrageUtils.java index 1b983a101bc..7aa4c9dedbe 100644 --- a/replication/static/src/main/java/io/deephaven/replicators/ReplicateBarrageUtils.java +++ b/replication/static/src/main/java/io/deephaven/replicators/ReplicateBarrageUtils.java @@ -24,9 +24,7 @@ public static void main(final String[] args) throws IOException { ReplicatePrimitiveCode.charToAllButBoolean("replicateBarrageUtils", CHUNK_PACKAGE + "/CharChunkReader.java"); - // ReplicatePrimitiveCode.floatToAllFloatingPoints("replicateBarrageUtils", - // CHUNK_PACKAGE + "/FloatChunkReader.java", "Float16"); - fixupDoubleChunkReader(CHUNK_PACKAGE + "/DoubleChunkReader.java"); + fixupByteChunkReader(CHUNK_PACKAGE + "/ByteChunkReader.java"); ReplicatePrimitiveCode.charToAllButBoolean("replicateBarrageUtils", CHUNK_PACKAGE + "/array/CharArrayExpansionKernel.java"); @@ -63,27 +61,29 @@ private static void fixupDictionaryWriterValueMap( FileUtils.writeLines(file, lines); } - private static void fixupDoubleChunkReader(final @NotNull String path) throws IOException { + /** + * A single byte needs no byte-order decoding, so the byte reader transfers the payload straight into the chunk's + * backing array instead of staging it in an intermediate buffer and decoding element by element. + */ + private static void fixupByteChunkReader(final @NotNull String path) throws IOException { final File file = new File(path); List lines = FileUtils.readLines(file, Charset.defaultCharset()); - lines = globalReplacements(lines, - "Float16.toDouble", "Float16.toFloat", - "doubleing point precision", "floating point precision", - "half-precision doubles", "half-precision floats"); - lines = replaceRegion(lines, "PrecisionSingleDhNulls", List.of( - " final float v = is.readFloat();", - " chunk.set(offset + ii, doubleCast(v));")); - lines = replaceRegion(lines, "PrecisionDoubleDhNulls", List.of( - " chunk.set(offset + ii, is.readDouble());")); - lines = replaceRegion(lines, "PrecisionSingleValidityBuffer", List.of( - " elementSize = Float.BYTES;", - " supplier = () -> doubleCast(is.readFloat());")); - lines = replaceRegion(lines, "PrecisionDoubleValidityBuffer", List.of( - " supplier = is::readDouble;")); - lines = replaceRegion(lines, "FPCastHelper", List.of( - " private static double doubleCast(float a) {", - " return a == QueryConstants.NULL_FLOAT ? QueryConstants.NULL_DOUBLE : (double) a;", - " }")); + lines = replaceRegion(lines, "PayloadDhNulls", List.of( + " // Bytes have no endianness, so transfer the payload straight into the chunk's backing array in", + " // bounded windows rather than decoding element by element through a staging buffer.", + " for (int ei = 0; ei < numElements;) {", + " final int length = Math.min(BULK_READ_ELEMENTS, numElements - ei);", + " is.readFully(chunk.array(), chunk.arrayOffset() + offset + ei, length);", + " ei += length;", + " }")); + lines = replaceRegion(lines, "PayloadValidityBuffer", List.of( + " // The payload carries a value slot for every element, including nulls; transfer it straight into", + " // the chunk's backing array in bounded windows, then overwrite the invalid positions with null.", + " for (int ei = 0; ei < numElements;) {", + " final int length = Math.min(BULK_READ_ELEMENTS, numElements - ei);", + " is.readFully(chunk.array(), chunk.arrayOffset() + offset + ei, length);", + " ei += length;", + " }")); FileUtils.writeLines(file, lines); } } diff --git a/web/client-api/src/main/java/io/deephaven/web/super/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java b/web/client-api/src/main/resources/io/deephaven/web/super/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java similarity index 78% rename from web/client-api/src/main/java/io/deephaven/web/super/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java rename to web/client-api/src/main/resources/io/deephaven/web/super/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java index 8b1f34ecd87..0163d942552 100644 --- a/web/client-api/src/main/java/io/deephaven/web/super/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java +++ b/web/client-api/src/main/resources/io/deephaven/web/super/io/deephaven/extensions/barrage/chunk/LittleEndianCodec.java @@ -33,6 +33,15 @@ static short getShort(final byte[] b, final int o) { return (short) ((b[o] & 0xFF) | (b[o + 1] & 0xFF) << 8); } + static char getChar(final byte[] b, final int o) { + return (char) ((b[o] & 0xFF) | (b[o + 1] & 0xFF) << 8); + } + + /** A single byte has no byte order; present so the replicated readers/writers can share one shape. */ + static byte getByte(final byte[] b, final int o) { + return b[o]; + } + static double getDouble(final byte[] b, final int o) { return Double.longBitsToDouble(getLong(b, o)); } @@ -64,6 +73,16 @@ static void putShort(final byte[] b, final int o, final short v) { b[o + 1] = (byte) (v >> 8); } + static void putChar(final byte[] b, final int o, final char v) { + b[o] = (byte) v; + b[o + 1] = (byte) (v >> 8); + } + + /** A single byte has no byte order; present so the replicated readers/writers can share one shape. */ + static void putByte(final byte[] b, final int o, final byte v) { + b[o] = v; + } + static void putDouble(final byte[] b, final int o, final double v) { putLong(b, o, Double.doubleToLongBits(v)); }