Ryan's feedback and bug fixes

nbauernfeind · Sep 17, 2024 · 8876208 · 8876208
1 parent 28b25d1
commit 8876208
Show file tree

Hide file tree

Showing 57 changed files with 913 additions and 734 deletions.
diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/BooleanChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/BooleanChunk.java
@@ -75,9 +75,11 @@ public final boolean get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int index) {
-        return data[offset + index] == QueryConstants.NULL_BOOLEAN;
+    // region isNull
+    public final boolean isNull(int index) {
+        return false;
     }
+    // endregion isNull
 
     @Override
     public BooleanChunk<ATTR> slice(int offset, int capacity) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/ByteChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/ByteChunk.java
@@ -79,9 +79,11 @@ public final byte get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/CharChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/CharChunk.java
@@ -74,9 +74,11 @@ public final char get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/Chunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/Chunk.java
@@ -109,12 +109,6 @@ default void copyToBuffer(int srcOffset, @NotNull Buffer destBuffer, int destOff
      */
     int size();
 
-    /**
-     * @return whether The value offset is null
-     * @param index The index to check
-     */
-    boolean isNullAt(int index);
-
     /**
      * @return The underlying chunk type
      */

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/DoubleChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/DoubleChunk.java
@@ -78,9 +78,11 @@ public final double get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/FloatChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/FloatChunk.java
@@ -78,9 +78,11 @@ public final float get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/IntChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/IntChunk.java
@@ -78,9 +78,11 @@ public final int get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/LongChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/LongChunk.java
@@ -78,9 +78,11 @@ public final long get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/ObjectChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/ObjectChunk.java
@@ -78,9 +78,11 @@ public final T get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/chunk/src/main/java/io/deephaven/chunk/ShortChunk.java b/engine/chunk/src/main/java/io/deephaven/chunk/ShortChunk.java
@@ -78,9 +78,11 @@ public final short get(int index) {
         return data[offset + index];
     }
 
-    public final boolean isNullAt(int 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) {

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/QueryTable.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/QueryTable.java
@@ -2613,7 +2613,6 @@ private Table snapshotIncrementalInternal(final Table base, final boolean doInit
                     new ListenerRecorder("snapshotIncremental (triggerTable)", this, resultTable);
             addUpdateListener(triggerListenerRecorder);
 
-            dropColumns(getColumnSourceMap().keySet());
             final SnapshotIncrementalListener listener =
                     new SnapshotIncrementalListener(this, resultTable, resultColumns,
                             baseListenerRecorder, triggerListenerRecorder, baseTable, triggerColumns);

diff --git a/engine/table/src/main/java/io/deephaven/engine/table/impl/sources/ReinterpretUtils.java b/engine/table/src/main/java/io/deephaven/engine/table/impl/sources/ReinterpretUtils.java
@@ -265,6 +265,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);
@@ -284,6 +285,7 @@ public static Class<?> maybeConvertToPrimitiveDataType(@NotNull final Class<?> d
         }
         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;

diff --git a/...sions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageMessageWriterImpl.java b/...sions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageMessageWriterImpl.java
@@ -25,7 +25,6 @@
 import io.deephaven.engine.rowset.*;
 import io.deephaven.engine.rowset.impl.ExternalizableRowSetUtils;
 import io.deephaven.engine.table.impl.util.BarrageMessage;
-import io.deephaven.extensions.barrage.chunk.ChunkReader;
 import io.deephaven.extensions.barrage.chunk.ChunkWriter;
 import io.deephaven.extensions.barrage.chunk.SingleElementListHeaderWriter;
 import io.deephaven.extensions.barrage.util.ExposedByteArrayOutputStream;
@@ -75,7 +74,7 @@ public class BarrageMessageWriterImpl implements BarrageMessageWriter {
     public interface RecordBatchMessageView extends MessageView {
         boolean isViewport();
 
-        ChunkReader.Options options();
+        BarrageOptions options();
 
         RowSet addRowOffsets();
 
@@ -354,7 +353,7 @@ public boolean isViewport() {
         }
 
         @Override
-        public ChunkReader.Options options() {
+        public BarrageOptions options() {
             return options;
         }
 
@@ -533,7 +532,7 @@ public boolean isViewport() {
         }
 
         @Override
-        public ChunkReader.Options options() {
+        public BarrageOptions options() {
             return options;
         }
 

diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageOptions.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageOptions.java
@@ -0,0 +1,39 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.extensions.barrage;
+
+import io.deephaven.util.QueryConstants;
+
+public interface BarrageOptions {
+    /**
+     * @return whether we encode the validity buffer to express null values or {@link QueryConstants QueryConstants'}
+     *         NULL values.
+     */
+    boolean useDeephavenNulls();
+
+    /**
+     * @return the conversion mode to use for object columns
+     */
+    ColumnConversionMode columnConversionMode();
+
+    /**
+     * @return the ideal number of records to send per record batch
+     */
+    int batchSize();
+
+    /**
+     * @return the maximum number of bytes that should be sent in a single message.
+     */
+    int maxMessageSize();
+
+    /**
+     * Some Flight clients cannot handle modifications that have irregular column counts. These clients request that the
+     * server wrap all columns in a list to enable each column having a variable length.
+     *
+     * @return true if the columns should be wrapped in a list
+     */
+    default boolean columnsAsList() {
+        return false;
+    }
+}
diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageSnapshotOptions.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageSnapshotOptions.java
@@ -6,13 +6,12 @@
 import com.google.flatbuffers.FlatBufferBuilder;
 import io.deephaven.annotations.BuildableStyle;
 import io.deephaven.barrage.flatbuf.BarrageSnapshotRequest;
-import io.deephaven.extensions.barrage.chunk.ChunkReader;
 import org.immutables.value.Value.Default;
 import org.immutables.value.Value.Immutable;
 
 @Immutable
 @BuildableStyle
-public abstract class BarrageSnapshotOptions implements ChunkReader.Options {
+public abstract class BarrageSnapshotOptions implements BarrageOptions {
     public static Builder builder() {
         return ImmutableBarrageSnapshotOptions.builder();
     }

diff --git a/...ons/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageSubscriptionOptions.java b/...ons/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageSubscriptionOptions.java
@@ -6,13 +6,12 @@
 import com.google.flatbuffers.FlatBufferBuilder;
 import io.deephaven.annotations.BuildableStyle;
 import io.deephaven.barrage.flatbuf.BarrageSubscriptionRequest;
-import io.deephaven.extensions.barrage.chunk.ChunkReader;
 import org.immutables.value.Value.Default;
 import org.immutables.value.Value.Immutable;
 
 @Immutable
 @BuildableStyle
-public abstract class BarrageSubscriptionOptions implements ChunkReader.Options {
+public abstract class BarrageSubscriptionOptions implements BarrageOptions {
 
     public static Builder builder() {
         return ImmutableBarrageSubscriptionOptions.builder();

diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageTypeInfo.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/BarrageTypeInfo.java
@@ -0,0 +1,55 @@
+//
+// Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
+//
+package io.deephaven.extensions.barrage;
+
+import org.apache.arrow.flatbuf.Field;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Describes type info used by factory implementations when creating a ChunkReader.
+ */
+public class BarrageTypeInfo {
+    /**
+     * Factory method to create a TypeInfo instance.
+     *
+     * @param type the Java type to be read into the chunk
+     * @param componentType the Java type of nested components
+     * @param arrowField the Arrow type to be read into the chunk
+     * @return a TypeInfo instance
+     */
+    public static BarrageTypeInfo make(
+            @NotNull final Class<?> type,
+            @Nullable final Class<?> componentType,
+            @NotNull final Field arrowField) {
+        return new BarrageTypeInfo(type, componentType, arrowField);
+    }
+
+    private final Class<?> type;
+    @Nullable
+    private final Class<?> componentType;
+    private final Field arrowField;
+
+    public BarrageTypeInfo(
+            @NotNull final Class<?> type,
+            @Nullable final Class<?> componentType,
+            @NotNull final Field arrowField) {
+        this.type = type;
+        this.componentType = componentType;
+        this.arrowField = arrowField;
+    }
+
+    public Class<?> type() {
+        return type;
+    }
+
+    @Nullable
+    public Class<?> componentType() {
+        return componentType;
+    }
+
+    public Field arrowField() {
+        return arrowField;
+    }
+}
diff --git a/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/ChunkListWriter.java b/extensions/barrage/src/main/java/io/deephaven/extensions/barrage/ChunkListWriter.java
@@ -5,7 +5,6 @@
 
 import io.deephaven.chunk.Chunk;
 import io.deephaven.chunk.attributes.Values;
-import io.deephaven.extensions.barrage.chunk.ChunkReader;
 import io.deephaven.extensions.barrage.chunk.ChunkWriter;
 import io.deephaven.util.SafeCloseable;
 import org.jetbrains.annotations.NotNull;
@@ -41,7 +40,7 @@ public ChunkWriter.Context<SourceChunkType>[] chunks() {
         return contexts;
     }
 
-    public ChunkWriter.DrainableColumn empty(@NotNull final ChunkReader.Options options) throws IOException {
+    public ChunkWriter.DrainableColumn empty(@NotNull final BarrageOptions options) throws IOException {
         return writer.getEmptyInputStream(options);
     }
 

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
@@ -10,8 +10,8 @@
 import java.util.function.Function;
 import java.util.function.IntFunction;
 
-public abstract class BaseChunkReader<ReadChunkType extends WritableChunk<Values>>
-        implements ChunkReader<ReadChunkType> {
+public abstract class BaseChunkReader<READ_CHUNK_TYPE extends WritableChunk<Values>>
+        implements ChunkReader<READ_CHUNK_TYPE> {
 
     protected static <T extends WritableChunk<Values>> T castOrCreateChunk(
             final WritableChunk<Values> outChunk,