forked from deephaven/deephaven-core
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Commit #1 reading chunks, checkpoint to talk to nate, next will try r…
…eading schema at beginning of stream
- Loading branch information
Showing
9 changed files
with
385 additions
and
190 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
116 changes: 116 additions & 0 deletions
116
...ions/barrage/src/main/java/io/deephaven/extensions/barrage/chunk/ChunkReadingFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| // | ||
| // Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending | ||
| // | ||
| package io.deephaven.extensions.barrage.chunk; | ||
|
|
||
| import io.deephaven.chunk.ChunkType; | ||
| import io.deephaven.chunk.WritableChunk; | ||
| import io.deephaven.chunk.attributes.Values; | ||
| import io.deephaven.extensions.barrage.util.StreamReaderOptions; | ||
| import org.apache.arrow.flatbuf.Field; | ||
| import org.apache.arrow.flatbuf.Type; | ||
|
|
||
| import java.io.DataInput; | ||
| import java.io.IOException; | ||
| import java.util.Iterator; | ||
| import java.util.PrimitiveIterator; | ||
|
|
||
| /** | ||
| * | ||
| */ | ||
| public interface ChunkReadingFactory { | ||
| /** | ||
| * | ||
| */ | ||
| class ChunkTypeInfo { | ||
| private final ChunkType chunkType; | ||
| private final Class<?> type; | ||
| private final Class<?> componentType; | ||
| private final Field arrowField; | ||
|
|
||
| public ChunkTypeInfo(ChunkType chunkType, Class<?> type, Class<?> componentType, Field arrowField) { | ||
| this.chunkType = chunkType; | ||
| this.type = type; | ||
| this.componentType = componentType; | ||
| this.arrowField = arrowField; | ||
| } | ||
|
|
||
| public ChunkType chunkType() { | ||
| return chunkType; | ||
| } | ||
|
|
||
| public Class<?> type() { | ||
| return type; | ||
| } | ||
|
|
||
| public Class<?> componentType() { | ||
| return componentType; | ||
| } | ||
|
|
||
| public Field arrowField() { | ||
| return arrowField; | ||
| } | ||
|
|
||
| public Field componentArrowField() { | ||
| if (arrowField.typeType() != Type.List) { | ||
| throw new IllegalStateException("Not a flight List"); | ||
| } | ||
| if (arrowField.childrenLength() != 1) { | ||
| throw new IllegalStateException("Incorrect number of child Fields"); | ||
| } | ||
| return arrowField.children(0); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * | ||
| * @param options | ||
| * @param factor | ||
| * @param typeInfo | ||
| * @param fieldNodeIter | ||
| * @param bufferInfoIter | ||
| * @param is | ||
| * @param outChunk | ||
| * @param outOffset | ||
| * @param totalRows | ||
| * @return | ||
| * @throws IOException | ||
| */ | ||
| WritableChunk<Values> extractChunkFromInputStream( | ||
| final StreamReaderOptions options, | ||
| final int factor, | ||
| final ChunkTypeInfo typeInfo, | ||
| final Iterator<ChunkInputStreamGenerator.FieldNodeInfo> fieldNodeIter, | ||
| final PrimitiveIterator.OfLong bufferInfoIter, | ||
| final DataInput is, | ||
| final WritableChunk<Values> outChunk, | ||
| final int outOffset, | ||
| final int totalRows) throws IOException; | ||
|
|
||
| /** | ||
| * | ||
| * @param options | ||
| * @param typeInfo | ||
| * @param fieldNodeIter | ||
| * @param bufferInfoIter | ||
| * @param is | ||
| * @param outChunk | ||
| * @param offset | ||
| * @param totalRows | ||
| * @return | ||
| * @throws IOException | ||
| */ | ||
| default WritableChunk<Values> extractChunkFromInputStream( | ||
| final StreamReaderOptions options, | ||
| final ChunkTypeInfo typeInfo, | ||
| final Iterator<ChunkInputStreamGenerator.FieldNodeInfo> fieldNodeIter, | ||
| final PrimitiveIterator.OfLong bufferInfoIter, | ||
| final DataInput is, | ||
| final WritableChunk<Values> outChunk, | ||
| final int offset, | ||
| final int totalRows) throws IOException { | ||
| return extractChunkFromInputStream(options, 1, typeInfo, fieldNodeIter, bufferInfoIter, is, outChunk, offset, | ||
| totalRows); | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.