feat: Barrage Refactor Read/Write Chunk Factories

cpp-client/build.gradle

@@ -114,6 +114,7 @@ def testCppClient = Docker.registerDockerTask(project, 'testCppClient') {
         environmentVariable 'DH_HOST', deephavenDocker.containerName.get()
         environmentVariable 'DH_PORT', '10000'
     }
+    waitTimeMinutes = 1
     containerDependencies.dependsOn = [deephavenDocker.healthyTask]
     containerDependencies.finalizedBy = deephavenDocker.endTask
     network = deephavenDocker.networkName.get()

engine/api/src/main/java/io/deephaven/engine/table/ColumnDefinition.java

@@ -400,6 +400,15 @@ public <Other> ColumnDefinition<Other> withDataType(@NotNull final Class<Other>
                 : fromGenericType(name, newDataType, componentType, columnType);
     }
 
+    public <Other> ColumnDefinition<Other> withDataType(
+            @NotNull final Class<Other> newDataType,
+            @Nullable final Class<?> newComponentType) {
+        // noinspection unchecked
+        return dataType == newDataType && componentType == newComponentType
+                ? (ColumnDefinition<Other>) this
+                : fromGenericType(name, newDataType, newComponentType, columnType);
+    }
+
     public ColumnDefinition<?> withName(@NotNull final String newName) {
         return newName.equals(name) ? this : new ColumnDefinition<>(newName, dataType, componentType, columnType);
     }

engine/api/src/main/java/io/deephaven/engine/table/Table.java

@@ -217,6 +217,10 @@ public interface Table extends
      * Set this attribute to enable collection of barrage performance stats.
      */
     String BARRAGE_PERFORMANCE_KEY_ATTRIBUTE = "BarragePerformanceTableKey";
+    /**
+     * Set this to control the schema used for barrage serialization.
+     */
+    String BARRAGE_SCHEMA_ATTRIBUTE = "BarrageSchema";
 
     // -----------------------------------------------------------------------------------------------------------------
     // ColumnSources for fetching data by row key

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -74,6 +75,12 @@ public final boolean get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return false;
+    }
+    // endregion isNull
+
     @Override
     public BooleanChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -78,6 +79,12 @@ public final byte get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == QueryConstants.NULL_BYTE;
+    }
+    // endregion isNull
+
     @Override
     public ByteChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -3,6 +3,7 @@
 //
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -73,6 +74,12 @@ public final char get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == QueryConstants.NULL_CHAR;
+    }
+    // endregion isNull
+
     @Override
     public CharChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -77,6 +78,12 @@ public final double get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == QueryConstants.NULL_DOUBLE;
+    }
+    // endregion isNull
+
     @Override
     public DoubleChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -77,6 +78,12 @@ public final float get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == QueryConstants.NULL_FLOAT;
+    }
+    // endregion isNull
+
     @Override
     public FloatChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -77,6 +78,12 @@ public final int get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == QueryConstants.NULL_INT;
+    }
+    // endregion isNull
+
     @Override
     public IntChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -77,6 +78,12 @@ public final long get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == QueryConstants.NULL_LONG;
+    }
+    // endregion isNull
+
     @Override
     public LongChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -77,6 +78,12 @@ public final T get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == null;
+    }
+    // endregion isNull
+
     @Override
     public ObjectChunk<T, ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -10,7 +10,8 @@
  * {@link Chunk} that may have its backing storage reset to a slice of that belonging to another {@link Chunk} or a
  * native array.
  */
-public interface ResettableReadOnlyChunk<ATTR_BASE extends Any> extends ResettableChunk<ATTR_BASE>, PoolableChunk {
+public interface ResettableReadOnlyChunk<ATTR_BASE extends Any>
+        extends ResettableChunk<ATTR_BASE>, PoolableChunk<ATTR_BASE> {
 
     /**
      * Reset the data and bounds of this chunk to a range or sub-range of the specified {@link Chunk}.

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

@@ -11,7 +11,7 @@
  * {@link WritableChunk} or a native array.
  */
 public interface ResettableWritableChunk<ATTR_BASE extends Any>
-        extends ResettableChunk<ATTR_BASE>, WritableChunk<ATTR_BASE>, PoolableChunk {
+        extends ResettableChunk<ATTR_BASE>, WritableChunk<ATTR_BASE>, PoolableChunk<ATTR_BASE> {
 
     @Override
     <ATTR extends ATTR_BASE> WritableChunk<ATTR> resetFromChunk(WritableChunk<ATTR> other, int offset, int capacity);

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

@@ -7,6 +7,7 @@
 // @formatter:off
 package io.deephaven.chunk;
 
+import io.deephaven.util.QueryConstants;
 import io.deephaven.util.type.ArrayTypeUtils;
 import io.deephaven.chunk.attributes.Any;
 
@@ -77,6 +78,12 @@ public final short get(int index) {
         return data[offset + index];
     }
 
+    // region isNull
+    public final boolean isNull(int index) {
+        return data[offset + index] == QueryConstants.NULL_SHORT;
+    }
+    // endregion isNull
+
     @Override
     public ShortChunk<ATTR> slice(int offset, int capacity) {
         ChunkHelpers.checkSliceArgs(size, offset, capacity);

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

@@ -14,7 +14,7 @@
  *
  * @param <ATTR> Descriptive attribute that applies to the elements stored within this WritableChunk
  */
-public interface WritableChunk<ATTR extends Any> extends Chunk<ATTR>, PoolableChunk {
+public interface WritableChunk<ATTR extends Any> extends Chunk<ATTR>, PoolableChunk<ATTR> {
     @Override
     WritableChunk<ATTR> slice(int offset, int capacity);
 

engine/chunk/src/main/java/io/deephaven/chunk/util/pools/PoolableChunk.java

@@ -4,11 +4,12 @@
 package io.deephaven.chunk.util.pools;
 
 import io.deephaven.chunk.Chunk;
+import io.deephaven.chunk.attributes.Any;
 import io.deephaven.util.SafeCloseable;
 
 /**
  * Marker interface for {@link Chunk} subclasses that can be kept with in a {@link ChunkPool}, and whose
  * {@link #close()} method will return them to the appropriate pool.
  */
-public interface PoolableChunk extends SafeCloseable {
+public interface PoolableChunk<ATTR extends Any> extends Chunk<ATTR>, SafeCloseable {
 }

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

@@ -360,6 +360,9 @@ public enum CopyAttributeOperation {
                 CopyAttributeOperation.Flatten, // add flatten for now because web flattens all views
                 CopyAttributeOperation.Preview));
 
+        tempMap.put(BARRAGE_SCHEMA_ATTRIBUTE, EnumSet.of(
+                CopyAttributeOperation.Filter));
+
         attributeToCopySet = Collections.unmodifiableMap(tempMap);
     }
 

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

@@ -1411,7 +1411,12 @@ private static boolean snapshotColumnsParallel(
             final ExecutionContext executionContext,
             @NotNull final BarrageMessage snapshot) {
         final JobScheduler jobScheduler = new OperationInitializerJobScheduler();
-        final CompletableFuture<Void> waitForParallelSnapshot = new CompletableFuture<>();
+        final CompletableFuture<Void> waitForParallelSnapshot = new CompletableFuture<>() {
+            @Override
+            public boolean completeExceptionally(Throwable ex) {
+                return super.completeExceptionally(ex);
+            }
+        };
         jobScheduler.iterateParallel(
                 executionContext,
                 logOutput -> logOutput.append("snapshotColumnsParallel"),

engine/table/src/main/java/io/deephaven/engine/table/impl/sources/ReinterpretUtils.java

@@ -4,6 +4,7 @@
 package io.deephaven.engine.table.impl.sources;
 
 import io.deephaven.chunk.ChunkType;
+import io.deephaven.engine.table.ColumnDefinition;
 import io.deephaven.engine.table.ColumnSource;
 import io.deephaven.engine.table.WritableColumnSource;
 import org.jetbrains.annotations.NotNull;
@@ -212,6 +213,27 @@ public static ColumnSource<?>[] maybeConvertToPrimitive(@NotNull final ColumnSou
         return result;
     }
 
+    /**
+     * If {@code columnDefinition.getDataType()} or {@code columnDefinition.getComponentType} are something that we
+     * prefer to handle as a primitive, do the appropriate conversion.
+     *
+     * @param columnDefinition The column definition to convert
+     * @return if possible, {@code columnDefinition} converted to a primitive, otherewise {@code columnDefinition}
+     */
+    @NotNull
+    public static ColumnDefinition<?> maybeConvertToPrimitive(@NotNull final ColumnDefinition<?> columnDefinition) {
+        final Class<?> dataType = ReinterpretUtils.maybeConvertToPrimitiveDataType(columnDefinition.getDataType());
+        Class<?> componentType = columnDefinition.getComponentType();
+        if (componentType != null) {
+            componentType = ReinterpretUtils.maybeConvertToPrimitiveDataType(componentType);
+        }
+        if (columnDefinition.getDataType() == dataType
+                && columnDefinition.getComponentType() == componentType) {
+            return columnDefinition;
+        }
+        return columnDefinition.withDataType(dataType, componentType);
+    }
+
     /**
      * If {@code source} is something that we prefer to handle as a primitive, do the appropriate conversion.
      *
@@ -265,6 +287,7 @@ public static ChunkType maybeConvertToWritablePrimitiveChunkType(@NotNull final
         }
         if (dataType == Instant.class) {
             // Note that storing ZonedDateTime as a primitive is lossy on the time zone.
+            // TODO (https://github.com/deephaven/deephaven-core/issues/5241): Inconsistent handling of ZonedDateTime
             return ChunkType.Long;
         }
         return ChunkType.fromElementType(dataType);
@@ -283,6 +306,8 @@ public static Class<?> maybeConvertToPrimitiveDataType(@NotNull final Class<?> d
             return byte.class;
         }
         if (dataType == Instant.class || dataType == ZonedDateTime.class) {
+            // Note: not all ZonedDateTime sources are convertible to long, so this doesn't match column source behavior
+            // TODO (https://github.com/deephaven/deephaven-core/issues/5241): Inconsistent handling of ZonedDateTime
             return long.class;
         }
         return dataType;

engine/table/src/main/java/io/deephaven/engine/updategraph/impl/PeriodicUpdateGraph.java

@@ -100,6 +100,9 @@ public static Builder newBuilder(final String name) {
 
     public static final String DEFAULT_TARGET_CYCLE_DURATION_MILLIS_PROP =
             "PeriodicUpdateGraph.targetCycleDurationMillis";
+    public static final int DEFAULT_TARGET_CYCLE_DURATION_MILLIS =
+            Configuration.getInstance().getIntegerWithDefault(DEFAULT_TARGET_CYCLE_DURATION_MILLIS_PROP, 1000);
+
     private final long defaultTargetCycleDurationMillis;
     private volatile long targetCycleDurationMillis;
     private final ThreadInitializationFactory threadInitializationFactory;
@@ -1166,8 +1169,7 @@ public static PeriodicUpdateGraph getInstance(final String name) {
     public static final class Builder {
         private final boolean allowUnitTestMode =
                 Configuration.getInstance().getBooleanWithDefault(ALLOW_UNIT_TEST_MODE_PROP, false);
-        private long targetCycleDurationMillis =
-                Configuration.getInstance().getIntegerWithDefault(DEFAULT_TARGET_CYCLE_DURATION_MILLIS_PROP, 1000);
+        private long targetCycleDurationMillis = DEFAULT_TARGET_CYCLE_DURATION_MILLIS;
         private long minimumCycleDurationToLogNanos = DEFAULT_MINIMUM_CYCLE_DURATION_TO_LOG_NANOSECONDS;
 
         private String name;