feat: Barrage Refactor Read/Write Chunk Factories

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

@@ -168,7 +168,7 @@ public int available() throws IOException {
          * @formatter:on
          */
         protected boolean sendValidityBuffer() {
-            return fieldNullable && nullCount() != 0;
+            return nullCount() != 0;
         }
 
         @Override

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

@@ -91,7 +91,7 @@ protected void writePayload(
                 }
 
                 final BigInteger truncatedValue;
-                boolean isNegative = value.compareTo(BigDecimal.ZERO) < 0;
+                boolean isNegative = value.signum() < 0;
                 if (isNegative) {
                     // negative values are sign extended to match truncationMask's byte length; operate on abs-value
                     truncatedValue = value.unscaledValue().negate().and(truncationMask).negate();

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

@@ -32,7 +32,8 @@ public interface ChunkWriter<SOURCE_CHUNK_TYPE extends Chunk<Values>> {
     /**
      * Creator of {@link ChunkWriter} instances.
      * <p>
-     * This API may not be stable, while the JS API's usages of it are implemented.
+     * TODO: https://github.com/deephaven/deephaven-core/issues/5927 This API may not be stable, while the JS API's
+     * usages of it are implemented.
      */
     interface Factory {
         /**

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

@@ -93,7 +93,7 @@ ChunkReader<? extends WritableChunk<Values>> make(
     }
 
     // allow subclasses to modify this as they wish
-    protected final Map<ArrowType.ArrowTypeID, Map<Class<?>, ChunkReaderFactory>> registeredFactories =
+    private final Map<ArrowType.ArrowTypeID, Map<Class<?>, ChunkReaderFactory>> registeredFactories =
             new HashMap<>();
 
     protected DefaultChunkReaderFactory() {
@@ -282,6 +282,8 @@ public <T extends WritableChunk<Values>> ChunkReader<T> newReaderPojo(
         }
 
         if (typeId == ArrowType.ArrowTypeID.Map) {
+            // TODO: https://github.com/deephaven/deephaven-core/issues/6637
+            //  should we allow the user to supply the collector?
             final Field structField = field.getChildren().get(0);
             final BarrageTypeInfo<Field> keyTypeInfo = BarrageUtil.getDefaultType(structField.getChildren().get(0));
             final BarrageTypeInfo<Field> valueTypeInfo = BarrageUtil.getDefaultType(structField.getChildren().get(1));
@@ -293,9 +295,10 @@ public <T extends WritableChunk<Values>> ChunkReader<T> newReaderPojo(
             return (ChunkReader<T>) new MapChunkReader<>(keyReader, valueReader);
         }
 
-        // TODO: if (typeId == ArrowType.ArrowTypeID.Struct) {
-        // expose transformer API of Map<String, Chunk<Values>> -> T
-        // maybe defaults to Map<String, Object>
+        // TODO: struct support - https://github.com/deephaven/deephaven-core/issues/6636
+        // expose @FunctionInterface of Map<String, Chunk<Values>> -> T
+        // maybe defaults to Map<String, Object>?
+        // if (typeId == ArrowType.ArrowTypeID.Struct) {
 
         if (typeId == ArrowType.ArrowTypeID.Union) {
             final ArrowType.Union unionType = (ArrowType.Union) field.getType();
@@ -716,7 +719,7 @@ private static ChunkReader<WritableObjectChunk<BigInteger, Values>> fpToBigInteg
                         final short value = src.get(ii);
                         dst.set(dstOffset + ii, value == QueryConstants.NULL_SHORT
                                 ? null
-                                : BigInteger.valueOf((long) Float16.toFloat(value)));
+                                : BigDecimal.valueOf(Float16.toFloat(value)).toBigInteger());
                     }
                 });
 
@@ -726,7 +729,7 @@ private static ChunkReader<WritableObjectChunk<BigInteger, Values>> fpToBigInteg
                         final float value = src.get(ii);
                         dst.set(dstOffset + ii, value == QueryConstants.NULL_FLOAT
                                 ? null
-                                : BigInteger.valueOf((long) value));
+                                : BigDecimal.valueOf(value).toBigInteger());
                     }
                 });
             case DOUBLE:
@@ -735,7 +738,7 @@ private static ChunkReader<WritableObjectChunk<BigInteger, Values>> fpToBigInteg
                         final double value = src.get(ii);
                         dst.set(dstOffset + ii, value == QueryConstants.NULL_DOUBLE
                                 ? null
-                                : BigInteger.valueOf((long) value));
+                                : BigDecimal.valueOf(value).toBigInteger());
                     }
                 });
 

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

@@ -261,10 +261,6 @@ public <T extends Chunk<Values>> ChunkWriter<T> newWriterPojo(
             if (useVectorKernels) {
                 Class<?> componentType =
                         VectorExpansionKernel.getComponentType(typeInfo.type(), typeInfo.componentType());
-                if (componentType == Boolean.class) {
-                    // Boolean's are stored in ObjectVector and should use ObjectVectorExpansionKernel
-                    componentType = Object.class;
-                }
                 componentTypeInfo = new BarrageTypeInfo<>(
                         componentType,
                         componentType.getComponentType(),
@@ -298,7 +294,6 @@ public <T extends Chunk<Values>> ChunkWriter<T> newWriterPojo(
         }
 
         if (typeId == ArrowType.ArrowTypeID.Map) {
-            // TODO: should we allow the user to supply the collector?
             final Field structField = field.getChildren().get(0);
             final BarrageTypeInfo<Field> keyTypeInfo = BarrageUtil.getDefaultType(structField.getChildren().get(0));
             final BarrageTypeInfo<Field> valueTypeInfo = BarrageUtil.getDefaultType(structField.getChildren().get(1));
@@ -311,8 +306,9 @@ public <T extends Chunk<Values>> ChunkWriter<T> newWriterPojo(
                     keyWriter, valueWriter, keyTypeInfo.chunkType(), valueTypeInfo.chunkType(), field.isNullable());
         }
 
-        // TODO: if (typeId == ArrowType.ArrowTypeID.Struct) {
-        // expose transformer API of Map<String, Chunk<Values>> -> T
+        // TODO: struct support - https://github.com/deephaven/deephaven-core/issues/6636
+        // expose @FunctionalInterface of Map<String, Chunk<Values>> -> T?
+        // if (typeId == ArrowType.ArrowTypeID.Struct) {
 
         if (typeId == ArrowType.ArrowTypeID.Union) {
             final ArrowType.Union unionType = (ArrowType.Union) field.getType();
@@ -422,8 +418,9 @@ private static ChunkWriter<Chunk<Values>> timestampFromZonedDateTime(
                     chunk.set(ii, longChunk.isNull(ii) ? value : value / factor);
                 }
             } else {
+                final ObjectChunk<ZonedDateTime, Values> zdtChunk = source.asObjectChunk();
                 for (int ii = 0; ii < source.size(); ++ii) {
-                    final ZonedDateTime value = source.<ZonedDateTime>asObjectChunk().get(ii);
+                    final ZonedDateTime value = zdtChunk.get(ii);
                     chunk.set(ii, value == null
                             ? QueryConstants.NULL_LONG
                             : DateTimeUtils.epochNanos(value) / factor);

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

@@ -67,7 +67,8 @@ <A extends Any> WritableChunk<A> expand(
      * <li>If {@code offsets} is {@code null}, each row is assumed to have a fixed size defined by
      * {@code sizePerElement}.</li>
      * <li>If {@code lengths} is {@code null}, each row's size is determined by differences between adjacent elements in
-     * {@code offsets}.</li>
+     * {@code offsets}. An element's length is determined by subtracting its offset from the next offset, therefore
+     * offsets must contain one more element than the number of elements.</li>
      * <li>If both {@code offsets} and {@code lengths} are provided, {@code lengths} determines the row sizes.</li>
      * </ul>
      *

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

@@ -232,7 +232,7 @@ public void visitBuffers(final BufferListener listener) {
             }
             listener.noteLogicalBuffer(padBufferSize(numOffsetBytes));
 
-            // a validity buffer for the inner struct ??
+            // a validity buffer for the inner struct; (the buffer is odd since each entry needs to be valid)
             listener.noteLogicalBuffer(0);
 
             // payload

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/array/BooleanArrayExpansionKernel.java

@@ -40,15 +40,14 @@ public <A extends Any> WritableChunk<A> expand(
         final ObjectChunk<boolean[], A> typedSource = source.asObjectChunk();
 
         long totalSize = 0;
-        for (int ii = 0; ii < typedSource.size(); ++ii) {
-            int rowLen;
-            if (fixedSizeLength != 0) {
-                rowLen = Math.abs(fixedSizeLength);
-            } else {
+        if (fixedSizeLength != 0) {
+            totalSize = typedSource.size() * (long) fixedSizeLength;
+        } else {
+            for (int ii = 0; ii < typedSource.size(); ++ii) {
                 final boolean[] row = typedSource.get(ii);
-                rowLen = row == null ? 0 : row.length;
+                int rowLen = row == null ? 0 : row.length;
+                totalSize += rowLen;
             }
-            totalSize += rowLen;
         }
         final WritableByteChunk<A> result = WritableByteChunk.makeWritableChunk(
                 LongSizedDataStructure.intSize(DEBUG_NAME, totalSize));
@@ -64,18 +63,12 @@ public <A extends Any> WritableChunk<A> expand(
             }
             int written = 0;
             if (row != null) {
-                int offset = 0;
                 if (fixedSizeLength != 0) {
                     // limit length to fixedSizeLength
-                    written = Math.min(row.length, Math.abs(fixedSizeLength));
-                    if (fixedSizeLength < 0 && written < row.length) {
-                        // read from the end of the array when fixedSizeLength is negative
-                        offset = row.length - written;
-                    }
+                    written = Math.min(row.length, fixedSizeLength);
                 } else {
                     written = row.length;
                 }
-
                 // copy the row into the result
                 for (int j = 0; j < written; ++j) {
                     final byte value = row[j] ? BooleanUtils.TRUE_BOOLEAN_AS_BYTE : BooleanUtils.FALSE_BOOLEAN_AS_BYTE;
@@ -84,7 +77,7 @@ public <A extends Any> WritableChunk<A> expand(
             }
             if (fixedSizeLength != 0) {
                 final int toNull = LongSizedDataStructure.intSize(
-                        DEBUG_NAME, Math.max(0, Math.abs(fixedSizeLength) - written));
+                        DEBUG_NAME, Math.max(0, fixedSizeLength - written));
                 if (toNull > 0) {
                     // fill the rest of the row with nulls
                     result.fillWithNullValue(lenWritten + written, toNull);
@@ -135,9 +128,6 @@ public <A extends Any> WritableObjectChunk<boolean[], A> contract(
             final int rowLen = computeSize(ii, sizePerElement, offsets, lengths);
             if (rowLen == 0) {
                 result.set(outOffset + ii, ZERO_LEN_ARRAY);
-            } else if (rowLen < 0) {
-                // note that this may occur when data sent from a native arrow client is null
-                result.set(outOffset + ii, null);
             } else {
                 final boolean[] row = new boolean[rowLen];
                 for (int j = 0; j < rowLen; ++j) {

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/array/BoxedBooleanArrayExpansionKernel.java

@@ -40,15 +40,14 @@ public <A extends Any> WritableChunk<A> expand(
         final ObjectChunk<Boolean[], A> typedSource = source.asObjectChunk();
 
         long totalSize = 0;
-        for (int ii = 0; ii < typedSource.size(); ++ii) {
-            int rowLen;
-            if (fixedSizeLength > 0) {
-                rowLen = Math.abs(fixedSizeLength);
-            } else {
+        if (fixedSizeLength != 0) {
+            totalSize = source.size() * (long) fixedSizeLength;
+        } else {
+            for (int ii = 0; ii < source.size(); ++ii) {
                 final Boolean[] row = typedSource.get(ii);
-                rowLen = row == null ? 0 : row.length;
+                final int rowLen = row == null ? 0 : row.length;
+                totalSize += rowLen;
             }
-            totalSize += rowLen;
         }
         final WritableByteChunk<A> result = WritableByteChunk.makeWritableChunk(
                 LongSizedDataStructure.intSize(DEBUG_NAME, totalSize));
@@ -64,27 +63,22 @@ public <A extends Any> WritableChunk<A> expand(
             }
             int written = 0;
             if (row != null) {
-                int offset = 0;
                 if (fixedSizeLength != 0) {
                     // limit length to fixedSizeLength
-                    written = Math.min(row.length, Math.abs(fixedSizeLength));
-                    if (fixedSizeLength < 0 && written < row.length) {
-                        // read from the end of the array when fixedSizeLength is negative
-                        offset = row.length - written;
-                    }
+                    written = Math.min(row.length, fixedSizeLength);
                 } else {
                     written = row.length;
                 }
 
                 // copy the row into the result
                 for (int j = 0; j < written; ++j) {
-                    final byte value = BooleanUtils.booleanAsByte(row[offset + j]);
+                    final byte value = BooleanUtils.booleanAsByte(row[j]);
                     result.set(lenWritten + j, value);
                 }
             }
             if (fixedSizeLength != 0) {
                 final int toNull = LongSizedDataStructure.intSize(
-                        DEBUG_NAME, Math.max(0, Math.abs(fixedSizeLength) - written));
+                        DEBUG_NAME, Math.max(0, fixedSizeLength - written));
                 if (toNull > 0) {
                     // fill the rest of the row with nulls
                     result.fillWithNullValue(lenWritten + written, toNull);
@@ -135,9 +129,6 @@ public <A extends Any> WritableObjectChunk<Boolean[], A> contract(
             final int rowLen = computeSize(ii, sizePerElement, offsets, lengths);
             if (rowLen == 0) {
                 result.set(outOffset + ii, ZERO_LEN_ARRAY);
-            } else if (rowLen < 0) {
-                // note that this may occur when data sent from a native arrow client is null
-                result.set(outOffset + ii, null);
             } else {
                 final Boolean[] row = new Boolean[rowLen];
                 int numSent = Math.min(rowLen, typedSource.size() - offset);

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/array/ByteArrayExpansionKernel.java

@@ -41,15 +41,14 @@ public <A extends Any> WritableChunk<A> expand(
         }
 
         long totalSize = 0;
-        for (int ii = 0; ii < source.size(); ++ii) {
-            final int rowLen;
-            if (fixedSizeLength != 0) {
-                rowLen = Math.abs(fixedSizeLength);
-            } else {
+        if (fixedSizeLength != 0) {
+            totalSize = source.size() * (long) fixedSizeLength;
+        } else {
+            for (int ii = 0; ii < source.size(); ++ii) {
                 final byte[] row = source.get(ii);
-                rowLen = row == null ? 0 : row.length;
+                final int rowLen = row == null ? 0 : row.length;
+                totalSize += rowLen;
             }
-            totalSize += rowLen;
         }
         final WritableByteChunk<A> result = WritableByteChunk.makeWritableChunk(
                 LongSizedDataStructure.intSize(DEBUG_NAME, totalSize));
@@ -65,23 +64,18 @@ public <A extends Any> WritableChunk<A> expand(
             }
             int written = 0;
             if (row != null) {
-                int offset = 0;
                 if (fixedSizeLength != 0) {
                     // limit length to fixedSizeLength
-                    written = Math.min(row.length, Math.abs(fixedSizeLength));
-                    if (fixedSizeLength < 0 && written < row.length) {
-                        // read from the end of the array when fixedSizeLength is negative
-                        offset = row.length - written;
-                    }
+                    written = Math.min(row.length, fixedSizeLength);
                 } else {
                     written = row.length;
                 }
                 // copy the row into the result
-                result.copyFromArray(row, offset, lenWritten, written);
+                result.copyFromArray(row, 0, lenWritten, written);
             }
             if (fixedSizeLength != 0) {
                 final int toNull = LongSizedDataStructure.intSize(
-                        DEBUG_NAME, Math.max(0, Math.abs(fixedSizeLength) - written));
+                        DEBUG_NAME, Math.max(0, fixedSizeLength - written));
                 if (toNull > 0) {
                     // fill the rest of the row with nulls
                     result.fillWithNullValue(lenWritten + written, toNull);
@@ -135,9 +129,6 @@ public <A extends Any> WritableObjectChunk<byte[], A> contract(
             final int rowLen = computeSize(ii, sizePerElement, offsets, lengths);
             if (rowLen == 0) {
                 result.set(outOffset + ii, ZERO_LEN_ARRAY);
-            } else if (rowLen < 0) {
-                // note that this may occur when data sent from a native arrow client is null
-                result.set(outOffset + ii, null);
             } else {
                 final byte[] row = new byte[rowLen];
                 typedSource.copyToArray(offset, row, 0, rowLen);

extensions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/array/CharArrayExpansionKernel.java

@@ -37,15 +37,14 @@ public <A extends Any> WritableChunk<A> expand(
         }
 
         long totalSize = 0;
-        for (int ii = 0; ii < source.size(); ++ii) {
-            final int rowLen;
-            if (fixedSizeLength != 0) {
-                rowLen = Math.abs(fixedSizeLength);
-            } else {
+        if (fixedSizeLength != 0) {
+            totalSize = source.size() * (long) fixedSizeLength;
+        } else {
+            for (int ii = 0; ii < source.size(); ++ii) {
                 final char[] row = source.get(ii);
-                rowLen = row == null ? 0 : row.length;
+                final int rowLen = row == null ? 0 : row.length;
+                totalSize += rowLen;
             }
-            totalSize += rowLen;
         }
         final WritableCharChunk<A> result = WritableCharChunk.makeWritableChunk(
                 LongSizedDataStructure.intSize(DEBUG_NAME, totalSize));
@@ -61,23 +60,18 @@ public <A extends Any> WritableChunk<A> expand(
             }
             int written = 0;
             if (row != null) {
-                int offset = 0;
                 if (fixedSizeLength != 0) {
                     // limit length to fixedSizeLength
-                    written = Math.min(row.length, Math.abs(fixedSizeLength));
-                    if (fixedSizeLength < 0 && written < row.length) {
-                        // read from the end of the array when fixedSizeLength is negative
-                        offset = row.length - written;
-                    }
+                    written = Math.min(row.length, fixedSizeLength);
                 } else {
                     written = row.length;
                 }
                 // copy the row into the result
-                result.copyFromArray(row, offset, lenWritten, written);
+                result.copyFromArray(row, 0, lenWritten, written);
             }
             if (fixedSizeLength != 0) {
                 final int toNull = LongSizedDataStructure.intSize(
-                        DEBUG_NAME, Math.max(0, Math.abs(fixedSizeLength) - written));
+                        DEBUG_NAME, Math.max(0, fixedSizeLength - written));
                 if (toNull > 0) {
                     // fill the rest of the row with nulls
                     result.fillWithNullValue(lenWritten + written, toNull);
@@ -131,9 +125,6 @@ public <A extends Any> WritableObjectChunk<char[], A> contract(
             final int rowLen = computeSize(ii, sizePerElement, offsets, lengths);
             if (rowLen == 0) {
                 result.set(outOffset + ii, ZERO_LEN_ARRAY);
-            } else if (rowLen < 0) {
-                // note that this may occur when data sent from a native arrow client is null
-                result.set(outOffset + ii, null);
             } else {
                 final char[] row = new char[rowLen];
                 typedSource.copyToArray(offset, row, 0, rowLen);