From ff89d74ca362cd5806a11415a36345dfca5380ea Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Fri, 14 Aug 2026 16:16:11 +0800 Subject: [PATCH 1/3] feat: read TIME column as picosecond long vector for zero-copy Trino integration --- .../pixelsdb/pixels/core/TypeDescription.java | 8 + .../core/reader/PixelsReaderOption.java | 12 ++ .../reader/PixelsRecordReaderBufferImpl.java | 7 +- .../core/reader/PixelsRecordReaderImpl.java | 3 +- .../reader/PixelsRecordReaderStreamImpl.java | 3 +- .../pixels/core/reader/TimeColumnReader.java | 138 ++++++++++++++---- .../pixels/core/utils/DatetimeUtils.java | 1 + .../pixels/core/vector/LongColumnVector.java | 20 +++ .../core/vector/VectorizedRowBatch.java | 33 ++++- .../core/reader/TestTimeColumnReader.java | 95 ++++++++++++ .../core/vector/TestLongColumnVector.java | 68 +++++++++ .../executor/predicate/ColumnFilter.java | 63 +++++++- .../executor/predicate/TestPredicate.java | 30 ++++ 13 files changed, 440 insertions(+), 41 deletions(-) create mode 100644 pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongColumnVector.java diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java index bff682aebf..862a646946 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java @@ -1249,6 +1249,10 @@ private ColumnVector createColumn(int maxSize, int vectorLayout, boolean... useE case DATE: return new DateColumnVector(maxSize); case TIME: + if (VectorLayout.match(vectorLayout, VectorLayout.TIME_AS_PICO_LONG)) + { + return new LongColumnVector(maxSize); + } return new TimeColumnVector(maxSize, precision); case TIMESTAMP: return new TimestampColumnVector(maxSize, precision); @@ -1414,6 +1418,10 @@ public static final class VectorLayout * instead of {@link io.pixelsdb.pixels.core.vector.IntColumnVector}. */ public static final int INT_AS_LONG = 0x02; + /** + * Create {@link io.pixelsdb.pixels.core.vector.LongColumnVector} containing picoseconds for TIME type. + */ + public static final int TIME_AS_PICO_LONG = 0x04; public static boolean match(int layout1, int layout2) { diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java index 0444438dd6..463241d2b3 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java @@ -35,6 +35,7 @@ public class PixelsReaderOption private boolean enableEncodedColumnVector = false; // whether read encoded column vectors directly when possible private boolean readIntColumnAsLongVector = false; // whether read int32 columns as long column vectors, for backward compatibility of old query engines private boolean readShortColumnAsLongVector = false; // whether read int16 columns as long column vectors, for backward compatibility of old query engines + private boolean readTimeColumnAsLongVector = false; // whether read time columns as picoseconds in long vectors private boolean exposeHiddenColumn = false; // whether expose the hidden commit timestamp column in the result batch private long transId = -1L; private long transTimestamp = -1L; // -1 means no need to consider the timestamp when reading data @@ -178,6 +179,17 @@ public boolean isReadShortColumnAsLongVector() return readShortColumnAsLongVector; } + public PixelsReaderOption readTimeColumnAsLongVector(boolean readTimeColumnAsLongVector) + { + this.readTimeColumnAsLongVector = readTimeColumnAsLongVector; + return this; + } + + public boolean isReadTimeColumnAsLongVector() + { + return readTimeColumnAsLongVector; + } + public PixelsReaderOption exposeHiddenColumn(boolean exposeHiddenColumn) { this.exposeHiddenColumn = exposeHiddenColumn; diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java index 77f96fec9c..a9f01895ad 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java @@ -118,7 +118,8 @@ public PixelsRecordReaderBufferImpl(PixelsReaderOption option, this.option = option; this.vectorLayout = (option.isReadIntColumnAsLongVector() ? TypeDescription.VectorLayout.INT_AS_LONG : 0) | - (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0); + (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0) | + (option.isReadTimeColumnAsLongVector() ? TypeDescription.VectorLayout.TIME_AS_PICO_LONG : 0); this.activeMemtableData = activeMemtableData; this.fileIds = fileIds; this.storage = storage; @@ -171,7 +172,7 @@ private void startPrefetching() { memoryUsage.addAndGet(activeMemtableData.length); ByteBuffer buffer = ByteBuffer.wrap(activeMemtableData); - VectorizedRowBatch batch = VectorizedRowBatch.deserialize(buffer); + VectorizedRowBatch batch = VectorizedRowBatch.deserialize(buffer, vectorLayout); memoryUsage.addAndGet(batch.getMemoryUsage()); prefetchQueue.put(batch); } catch (Exception e) @@ -219,7 +220,7 @@ private void startPrefetching() buffer = getMemtableDataFromStorage(path); // CPU Intensive Operation - VectorizedRowBatch batch = VectorizedRowBatch.deserialize(buffer); + VectorizedRowBatch batch = VectorizedRowBatch.deserialize(buffer, vectorLayout); memoryUsage.addAndGet(batch.getMemoryUsage()); // Put result into the queue (blocks if queue is full) prefetchQueue.put(batch); diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java index f613ec00e6..3f91b42054 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java @@ -173,7 +173,8 @@ public PixelsRecordReaderImpl(PhysicalReader physicalReader, this.RGLen = option.getRGLen(); this.enableEncodedVector = option.isEnableEncodedColumnVector(); this.vectorLayout = (option.isReadIntColumnAsLongVector() ? TypeDescription.VectorLayout.INT_AS_LONG : 0) | - (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0); + (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0) | + (option.isReadTimeColumnAsLongVector() ? TypeDescription.VectorLayout.TIME_AS_PICO_LONG : 0); this.enableMetrics = enableMetrics; this.metricsDir = metricsDir; this.readPerfMetrics = new ReadPerfMetrics(); diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java index 4626c67a2a..c8cf76d2b4 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java @@ -124,7 +124,8 @@ public PixelsRecordReaderStreamImpl(PhysicalReader physicalReader, this.streamHeader = streamHeader; this.option = option; this.vectorLayout = (option.isReadIntColumnAsLongVector() ? TypeDescription.VectorLayout.INT_AS_LONG : 0) | - (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0); + (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0) | + (option.isReadTimeColumnAsLongVector() ? TypeDescription.VectorLayout.TIME_AS_PICO_LONG : 0); this.includedColumnTypes = new ArrayList<>(); checkBeforeRead(); } diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java index fa28847161..3a79a6deab 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java @@ -26,6 +26,7 @@ import io.pixelsdb.pixels.core.utils.Bitmap; import io.pixelsdb.pixels.core.utils.ByteBufferInputStream; import io.pixelsdb.pixels.core.vector.ColumnVector; +import io.pixelsdb.pixels.core.vector.LongColumnVector; import io.pixelsdb.pixels.core.vector.TimeColumnVector; import java.io.IOException; @@ -34,6 +35,8 @@ import java.nio.ByteOrder; import java.util.Arrays; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; + /** * Pixels time column reader. * All time values are translated to the specified time zone after read from file. @@ -84,7 +87,8 @@ public void close() throws IOException * @param size number of values to read * @param pixelStride the stride (number of rows) in a pixels. * @param vectorIndex the index from where we start reading values into the vector - * @param vector vector to read values into + * @param vector vector to read values into, it is a {@link LongColumnVector} of picoseconds of day + * if the read option requires the time column to be read as a long vector * @param chunkIndex the metadata of the column chunk to read. * @throws IOException */ @@ -93,7 +97,9 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, int offset, int size, int pixelStride, final int vectorIndex, ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex) throws IOException { - TimeColumnVector columnVector = (TimeColumnVector) vector; + // longColumnVector is not null if the time values are to be stored as picoseconds of day. + LongColumnVector longColumnVector = vector instanceof LongColumnVector ? (LongColumnVector) vector : null; + TimeColumnVector timeColumnVector = longColumnVector == null ? (TimeColumnVector) vector : null; boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); @@ -135,24 +141,37 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, hasNull = chunkIndex.getPixelStatistics(pixelId).getStatistic().getHasNull(); if (hasNull) { - BitUtils.bitWiseDeCompact(columnVector.isNull, i, numToRead, + BitUtils.bitWiseDeCompact(vector.isNull, i, numToRead, inputBuffer, isNullOffset, isNullSkipBits, littleEndian); isNullOffset += bytesToDeCompact; isNullSkipBits = endOfPixels ? 0 : (numToRead + isNullSkipBits) % 8; - columnVector.noNulls = false; + vector.noNulls = false; } else { - Arrays.fill(columnVector.isNull, i, i + numToRead, false); + Arrays.fill(vector.isNull, i, i + numToRead, false); } // read content if (decoding) { for (int j = i; j < i + numToRead; ++j) { - if (!(hasNull && columnVector.isNull[j])) + if (!(hasNull && vector.isNull[j])) { - columnVector.set(j, (int) decoder.next()); + int millis = (int) decoder.next(); + if (longColumnVector != null) + { + longColumnVector.vector[j] = millis * PICOSECONDS_PER_MILLISECOND; + longColumnVector.isNull[j] = false; + if (j >= longColumnVector.getWriteIndex()) + { + longColumnVector.setWriteIndex(j + 1); + } + } + else + { + timeColumnVector.set(j, millis); + } } } } @@ -163,17 +182,38 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, for (int j = i; j < i + numToRead; ++j) { // Issue #791: do not call the set() method, as it may clear the isNull flag of null values. - columnVector.times[j] = inputBuffer.getInt(); + int millis = inputBuffer.getInt(); + if (longColumnVector != null) + { + longColumnVector.vector[j] = millis * PICOSECONDS_PER_MILLISECOND; + } + else + { + timeColumnVector.times[j] = millis; + } } } else { for (int j = i; j < i + numToRead; ++j) { - if (!(hasNull && columnVector.isNull[j])) + if (!(hasNull && vector.isNull[j])) { // If time column is not encoded, it is written as integers instead of longs. - columnVector.set(j, inputBuffer.getInt()); + int millis = inputBuffer.getInt(); + if (longColumnVector != null) + { + longColumnVector.vector[j] = millis * PICOSECONDS_PER_MILLISECOND; + longColumnVector.isNull[j] = false; + if (j >= longColumnVector.getWriteIndex()) + { + longColumnVector.setWriteIndex(j + 1); + } + } + else + { + timeColumnVector.set(j, millis); + } } } } @@ -194,7 +234,8 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, * @param size number of values to read * @param pixelStride the stride (number of rows) in a pixels. * @param vectorIndex the index from where we start reading values into the vector - * @param vector vector to read values into + * @param vector vector to read values into, it is a {@link LongColumnVector} of picoseconds of day + * if the read option requires the time column to be read as a long vector * @param chunkIndex the metadata of the column chunk to read. * @param selected whether the value is selected, use the vectorIndex as the 0 offset of the selected * @throws IOException @@ -204,7 +245,9 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, int offset, int size, int pixelStride, final int vectorIndex, ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex, Bitmap selected) throws IOException { - TimeColumnVector columnVector = (TimeColumnVector) vector; + // longColumnVector is not null if the time values are to be stored as picoseconds of day. + LongColumnVector longColumnVector = vector instanceof LongColumnVector ? (LongColumnVector) vector : null; + TimeColumnVector timeColumnVector = longColumnVector == null ? (TimeColumnVector) vector : null; boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); @@ -227,7 +270,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, // read without copying the de-compacted content and isNull int numLeft = size, numToRead, bytesToDeCompact, vectorWriteIndex = vectorIndex; boolean[] isNull = null; - boolean endOfPixel; + boolean endOfPixels; if (decoding || !nullsPadding) { isNull = new boolean[size]; @@ -238,14 +281,14 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, { // read to the end of the current pixel numToRead = pixelStride - elementIndex % pixelStride; - endOfPixel = true; + endOfPixels = true; } else { numToRead = numLeft; - endOfPixel = false; + endOfPixels = false; } - bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixel ? 7 : 0)) / 8; + bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixels ? 7 : 0)) / 8; // read isNull int pixelId = elementIndex / pixelStride; @@ -255,7 +298,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, if (!decoding && nullsPadding) { // read isNull directly into the vector of the column chunk - BitUtils.bitWiseDeCompact(columnVector.isNull, vectorWriteIndex, numToRead, inputBuffer, + BitUtils.bitWiseDeCompact(vector.isNull, vectorWriteIndex, numToRead, inputBuffer, isNullOffset, isNullSkipBits, littleEndian, selected, i - vectorIndex); } else @@ -263,19 +306,19 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, // need to keep isNull for later use BitUtils.bitWiseDeCompact(isNull, i - vectorIndex, numToRead, inputBuffer, isNullOffset, isNullSkipBits, littleEndian); - // update columnVector.isNull + // update vector.isNull int k = vectorWriteIndex; for (int j = i; j < i + numToRead; ++j) { if (selected.get(j - vectorIndex)) { - columnVector.isNull[k++] = isNull[j - vectorIndex]; + vector.isNull[k++] = isNull[j - vectorIndex]; } } } isNullOffset += bytesToDeCompact; - isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; - columnVector.noNulls = false; + isNullSkipBits = endOfPixels ? 0 : (numToRead + isNullSkipBits) % 8; + vector.noNulls = false; } else { @@ -283,7 +326,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, { Arrays.fill(isNull, i - vectorIndex, i - vectorIndex + numToRead, false); } - // update columnVector.isNull later to avoid bitmap unnecessary traversal + // update vector.isNull later to avoid bitmap unnecessary traversal } // read content @@ -294,10 +337,23 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, { if (!(hasNull && isNull[j - vectorIndex])) { - int value = (int) decoder.next(); + int millis = (int) decoder.next(); if (selected.get(j - vectorIndex)) { - columnVector.set(vectorWriteIndex++, value); + if (longColumnVector != null) + { + longColumnVector.vector[vectorWriteIndex] = millis * PICOSECONDS_PER_MILLISECOND; + longColumnVector.isNull[vectorWriteIndex] = false; + if (vectorWriteIndex >= longColumnVector.getWriteIndex()) + { + longColumnVector.setWriteIndex(vectorWriteIndex + 1); + } + vectorWriteIndex++; + } + else + { + timeColumnVector.set(vectorWriteIndex++, millis); + } } } else if (selected.get(j - vectorIndex)) @@ -312,11 +368,18 @@ else if (selected.get(j - vectorIndex)) { for (int j = i; j < i + numToRead; ++j) { - int value = inputBuffer.getInt(); + int millis = inputBuffer.getInt(); if (selected.get(j - vectorIndex)) { // Issue #791: do not call the set() method, as it may clear the isNull flag of null values. - columnVector.times[vectorWriteIndex++] = value; + if (longColumnVector != null) + { + longColumnVector.vector[vectorWriteIndex++] = millis * PICOSECONDS_PER_MILLISECOND; + } + else + { + timeColumnVector.times[vectorWriteIndex++] = millis; + } } } } @@ -326,11 +389,24 @@ else if (selected.get(j - vectorIndex)) { if (!(hasNull && isNull[j - vectorIndex])) { - int value = inputBuffer.getInt(); + // If time column is not encoded, it is written as integers instead of longs. + int millis = inputBuffer.getInt(); if (selected.get(j - vectorIndex)) { - // If time column is not encoded, it is written as integers instead of longs. - columnVector.set(vectorWriteIndex++, value); + if (longColumnVector != null) + { + longColumnVector.vector[vectorWriteIndex] = millis * PICOSECONDS_PER_MILLISECOND; + longColumnVector.isNull[vectorWriteIndex] = false; + if (vectorWriteIndex >= longColumnVector.getWriteIndex()) + { + longColumnVector.setWriteIndex(vectorWriteIndex + 1); + } + vectorWriteIndex++; + } + else + { + timeColumnVector.set(vectorWriteIndex++, millis); + } } } else if (selected.get(j - vectorIndex)) @@ -341,10 +417,10 @@ else if (selected.get(j - vectorIndex)) } } - // update columnVector.isNull if has no nulls + // update vector.isNull if has no nulls if (!hasNull) { - Arrays.fill(columnVector.isNull, originalVectorWriteIndex, vectorWriteIndex, false); + Arrays.fill(vector.isNull, originalVectorWriteIndex, vectorWriteIndex, false); } // update variables diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java index e37c21e482..9b2569bf60 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java @@ -42,6 +42,7 @@ public class DatetimeUtils private static final long NANOS_PER_MILLIS = 1000_000L; private static final long MICROS_PER_SEC = 1000_000L; private static final long NANOS_PER_MICROS = 1000L; + public static final long PICOSECONDS_PER_MILLISECOND = 1_000_000_000L; public static long microsToMillis(long micros) { diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java index 0771255dbd..f3bdf1fadd 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java @@ -22,12 +22,14 @@ import com.google.flatbuffers.FlatBufferBuilder; import io.pixelsdb.pixels.core.flat.ColumnVectorFlat; import io.pixelsdb.pixels.core.flat.LongColumnVectorFlat; +import io.pixelsdb.pixels.core.flat.TimeColumnVectorFlat; import io.pixelsdb.pixels.core.utils.Bitmap; import java.nio.ByteBuffer; import java.util.Arrays; import static com.google.common.base.Preconditions.checkArgument; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; import static java.util.Objects.requireNonNull; /** @@ -365,4 +367,22 @@ public static LongColumnVector deserialize(LongColumnVectorFlat flat) vector.deserializeBase(flat.base()); return vector; } + + /** + * Deserialize a time vector directly into Trino's physical representation. + * Pixels stores TIME values as milliseconds of day, while Trino stores them as + * picoseconds of day. + */ + public static LongColumnVector deserializeTime(TimeColumnVectorFlat flat) + { + int length = flat.base().length(); + LongColumnVector vector = new LongColumnVector(length); + for (int i = 0; i < flat.timesLength(); ++i) + { + vector.vector[i] = (long) flat.times(i) * PICOSECONDS_PER_MILLISECOND; + } + vector.deserializeBase(flat.base()); + vector.memoryUsage += (long) (Long.BYTES - Integer.BYTES) * length; + return vector; + } } diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java index 6ded9a50b7..fa8caa4f46 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java @@ -20,6 +20,7 @@ package io.pixelsdb.pixels.core.vector; import com.google.flatbuffers.FlatBufferBuilder; +import io.pixelsdb.pixels.core.TypeDescription; import io.pixelsdb.pixels.core.flat.*; import io.pixelsdb.pixels.core.utils.Bitmap; @@ -400,8 +401,12 @@ public byte[] serialize() public static VectorizedRowBatch deserialize(byte[] data) { - ByteBuffer buffer = ByteBuffer.wrap(data); - return deserialize(buffer); + return deserialize(ByteBuffer.wrap(data)); + } + + public static VectorizedRowBatch deserialize(byte[] data, int vectorLayout) + { + return deserialize(ByteBuffer.wrap(data), vectorLayout); } /** Issue-1078: @@ -413,6 +418,18 @@ public static VectorizedRowBatch deserialize(byte[] data) * @return the deserialized VectorizedRowBatch */ public static VectorizedRowBatch deserialize(ByteBuffer buffer) + { + return deserialize(buffer, TypeDescription.VectorLayout.NONE); + } + + /** + * Deserialize a row batch using the requested physical vector layout. + * + * @param buffer the ByteBuffer containing serialized batch data + * @param vectorLayout requested physical vector layout + * @return the deserialized row batch + */ + public static VectorizedRowBatch deserialize(ByteBuffer buffer, int vectorLayout) { VectorizedRowBatchFlat batchFlat = VectorizedRowBatchFlat.getRootAsVectorizedRowBatchFlat(buffer); @@ -461,7 +478,17 @@ public static VectorizedRowBatch deserialize(ByteBuffer buffer) batch.cols[i] = LongDecimalColumnVector.deserialize((LongDecimalColumnVectorFlat) batchFlat.cols(new LongDecimalColumnVectorFlat(), i)); break; case ColumnVectorFlat.TimeColumnVectorFlat: - batch.cols[i] = TimeColumnVector.deserialize((TimeColumnVectorFlat) batchFlat.cols(new TimeColumnVectorFlat(), i)); + TimeColumnVectorFlat timeFlat = + (TimeColumnVectorFlat) batchFlat.cols(new TimeColumnVectorFlat(), i); + if (TypeDescription.VectorLayout.match( + vectorLayout, TypeDescription.VectorLayout.TIME_AS_PICO_LONG)) + { + batch.cols[i] = LongColumnVector.deserializeTime(timeFlat); + } + else + { + batch.cols[i] = TimeColumnVector.deserialize(timeFlat); + } break; case ColumnVectorFlat.TimestampColumnVectorFlat: batch.cols[i] = TimestampColumnVector.deserialize((TimestampColumnVectorFlat) batchFlat.cols(new TimestampColumnVectorFlat(), i)); diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java index f22460f52b..a054c8e798 100644 --- a/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java +++ b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java @@ -23,6 +23,7 @@ import io.pixelsdb.pixels.core.TypeDescription; import io.pixelsdb.pixels.core.encoding.EncodingLevel; import io.pixelsdb.pixels.core.utils.Bitmap; +import io.pixelsdb.pixels.core.vector.LongColumnVector; import io.pixelsdb.pixels.core.vector.TimeColumnVector; import io.pixelsdb.pixels.core.writer.PixelsWriterOption; import io.pixelsdb.pixels.core.writer.TimeColumnWriter; @@ -32,6 +33,11 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + /** * @author hank * @create 2023-08-20 Zermatt @@ -217,6 +223,95 @@ public void testSelected() throws IOException } } + @Test + public void testLongVectorPicoseconds() throws IOException + { + int pixelStride = 4; + int[] millis = {0, 1, 3_723_004, 86_399_999}; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL2).nullsPadding(false); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + TimeColumnVector source = new TimeColumnVector(millis.length + 1, 3); + for (int value : millis) + { + source.add(value); + } + source.addNull(); + columnWriter.write(source, millis.length + 1); + columnWriter.flush(); + columnWriter.close(); + + TypeDescription timeType = TypeDescription.createTime(3); + assertTrue(timeType.createRowBatch( + millis.length + 1, TypeDescription.VectorLayout.TIME_AS_PICO_LONG) + .cols[0] instanceof LongColumnVector); + + LongColumnVector target = new LongColumnVector(millis.length + 1); + TimeColumnReader reader = new TimeColumnReader(timeType); + reader.read( + ByteBuffer.wrap(columnWriter.getColumnChunkContent()), + columnWriter.getColumnChunkEncoding().build(), + 0, + millis.length + 1, + pixelStride, + 0, + target, + columnWriter.getColumnChunkIndex().build()); + reader.close(); + + for (int i = 0; i < millis.length; ++i) + { + assertFalse(target.isNull[i]); + assertEquals((long) millis[i] * PICOSECONDS_PER_MILLISECOND, target.vector[i]); + } + assertTrue(target.isNull[millis.length]); + } + + @Test + public void testSelectedLongVectorWithNullPadding() throws IOException + { + int pixelStride = 4; + int[] millis = {10, 20, 30, 40, 50}; + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(true); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + TimeColumnVector source = new TimeColumnVector(millis.length, 3); + source.add(millis[0]); + source.addNull(); + for (int i = 2; i < millis.length; ++i) + { + source.add(millis[i]); + } + columnWriter.write(source, millis.length); + columnWriter.flush(); + columnWriter.close(); + + Bitmap selected = new Bitmap(millis.length, true); + selected.clear(0); + selected.clear(3); + LongColumnVector target = new LongColumnVector(3); + TimeColumnReader reader = new TimeColumnReader(TypeDescription.createTime(3)); + reader.readSelected( + ByteBuffer.wrap(columnWriter.getColumnChunkContent()), + columnWriter.getColumnChunkEncoding().build(), + 0, + millis.length, + pixelStride, + 0, + target, + columnWriter.getColumnChunkIndex().build(), + selected); + reader.close(); + + assertTrue(target.isNull[0]); + assertEquals((long) millis[2] * PICOSECONDS_PER_MILLISECOND, target.vector[1]); + assertEquals((long) millis[4] * PICOSECONDS_PER_MILLISECOND, target.vector[2]); + } + @Test public void testLarge() throws IOException { diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongColumnVector.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongColumnVector.java new file mode 100644 index 0000000000..655d60cef6 --- /dev/null +++ b/pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongColumnVector.java @@ -0,0 +1,68 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.core.vector; + +import io.pixelsdb.pixels.core.TypeDescription; +import org.junit.Test; + +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * @author hank + * @create 2026-08-09 + */ +public class TestLongColumnVector +{ + @Test + public void testDeserializeTimeWithLongLayout() + { + int[] millis = {0, 1, 3_723_004, 86_399_999}; + VectorizedRowBatch source = TypeDescription.createTime(3).createRowBatch(millis.length + 1); + TimeColumnVector timeVector = (TimeColumnVector) source.cols[0]; + for (int value : millis) + { + timeVector.add(value); + } + timeVector.addNull(); + source.size = millis.length + 1; + + byte[] serialized = source.serialize(); + VectorizedRowBatch defaultBatch = VectorizedRowBatch.deserialize(serialized); + assertTrue(defaultBatch.cols[0] instanceof TimeColumnVector); + assertEquals(millis[2], ((TimeColumnVector) defaultBatch.cols[0]).times[2]); + + VectorizedRowBatch longBatch = VectorizedRowBatch.deserialize( + serialized, TypeDescription.VectorLayout.TIME_AS_PICO_LONG); + assertTrue(longBatch.cols[0] instanceof LongColumnVector); + LongColumnVector longVector = (LongColumnVector) longBatch.cols[0]; + assertEquals(timeVector.getLength(), longVector.getLength()); + assertEquals(timeVector.getWriteIndex(), longVector.getWriteIndex()); + assertFalse(longVector.noNulls); + for (int i = 0; i < millis.length; ++i) + { + assertFalse(longVector.isNull[i]); + assertEquals(millis[i] * PICOSECONDS_PER_MILLISECOND, longVector.vector[i]); + } + assertTrue(longVector.isNull[millis.length]); + } +} diff --git a/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java b/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java index 5c2324490a..1be2b910fb 100644 --- a/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java +++ b/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java @@ -38,6 +38,7 @@ import java.util.Set; import static com.google.common.base.Preconditions.checkArgument; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; import static java.util.Objects.requireNonNull; /** @@ -378,8 +379,16 @@ public void doFilter(ColumnVector columnVector, int start, int length, Bitmap re doFilter(dacv.dates, dacv.noNulls ? null : dacv.isNull, start, length, result); return; case TIME: - TimeColumnVector tcv = (TimeColumnVector) columnVector; - doFilter(tcv.times, tcv.noNulls ? null : tcv.isNull, start, length, result); + if (columnVector instanceof LongColumnVector) + { + LongColumnVector ltcv = (LongColumnVector) columnVector; + doFilterTime(ltcv.vector, ltcv.noNulls ? null : ltcv.isNull, start, length, result); + } + else + { + TimeColumnVector tcv = (TimeColumnVector) columnVector; + doFilter(tcv.times, tcv.noNulls ? null : tcv.isNull, start, length, result); + } return; case TIMESTAMP: TimestampColumnVector tscv = (TimestampColumnVector) columnVector; @@ -667,6 +676,56 @@ private void doFilter(long[] vector, boolean[] isNull, int start, int length, Bi } } + private void doFilterTime(long[] vector, boolean[] isNull, int start, int length, Bitmap result) + { + boolean noNulls = isNull == null; + if (!this.filter.ranges.isEmpty()) + { + for (Range range : this.filter.ranges) + { + long lowerBound = range.lowerBound.type != Bound.Type.UNBOUNDED ? + ((Integer) range.lowerBound.value + + (range.lowerBound.type == Bound.Type.EXCLUDED ? 1L : 0L)) * + PICOSECONDS_PER_MILLISECOND : Long.MIN_VALUE; + long upperBound = range.upperBound.type != Bound.Type.UNBOUNDED ? + ((Integer) range.upperBound.value - + (range.upperBound.type == Bound.Type.EXCLUDED ? 1L : 0L)) * + PICOSECONDS_PER_MILLISECOND : Long.MAX_VALUE; + for (int i = start; i < start + length; ++i) + { + if (this.filter.allowNull && !noNulls && isNull[i] || + (noNulls || !isNull[i]) && vector[i] >= lowerBound && vector[i] <= upperBound) + { + result.set(i); + } + } + } + } + else + { + Set picoIncludes = new HashSet<>(includes.size()); + for (T value : includes) + { + picoIncludes.add((Integer) value * PICOSECONDS_PER_MILLISECOND); + } + Set picoExcludes = new HashSet<>(excludes.size()); + for (T value : excludes) + { + picoExcludes.add((Integer) value * PICOSECONDS_PER_MILLISECOND); + } + for (int i = start; i < start + length; ++i) + { + if (this.filter.allowNull && !noNulls && isNull[i] || + (noNulls || !isNull[i]) && + ((!picoIncludes.isEmpty() && picoIncludes.contains(vector[i])) || + (!picoExcludes.isEmpty() && !picoExcludes.contains(vector[i])))) + { + result.set(i); + } + } + } + } + /** * For Decimal, the values in the filter are Long, therefore we create Decimals using the * same precision and scale in the column vector. However, this method is currently not diff --git a/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java b/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java index 5872ccf8e9..1b62a4a376 100644 --- a/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java +++ b/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java @@ -28,6 +28,7 @@ import io.pixelsdb.pixels.core.reader.PixelsRecordReader; import io.pixelsdb.pixels.core.utils.Bitmap; import io.pixelsdb.pixels.core.utils.Decimal; +import io.pixelsdb.pixels.core.vector.LongColumnVector; import io.pixelsdb.pixels.core.vector.VectorizedRowBatch; import org.junit.Test; @@ -36,6 +37,9 @@ import java.util.SortedMap; import java.util.TreeMap; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; +import static org.junit.Assert.assertEquals; + /** * Created at: 07/04/2022 * Author: hank @@ -78,6 +82,32 @@ public void testLongFilterSerDe() System.out.println(columnFilter1.getFilter().getDiscreteValueCount()); } + @Test + public void testLongTimeVectorFilterUsesMillisBounds() + { + Filter timeFilter = new Filter<>(Integer.TYPE, false, false, false, false); + timeFilter.addRange( + new Bound<>(Bound.Type.INCLUDED, 100), + new Bound<>(Bound.Type.INCLUDED, 200)); + ColumnFilter columnFilter = + new ColumnFilter<>("time", TypeDescription.Category.TIME, timeFilter); + LongColumnVector vector = new LongColumnVector(5); + int[] millis = {99, 100, 150, 200, 201}; + for (int i = 0; i < millis.length; ++i) + { + vector.vector[i] = (long) millis[i] * PICOSECONDS_PER_MILLISECOND; + } + + Bitmap result = new Bitmap(millis.length, false); + columnFilter.doFilter(vector, 0, millis.length, result); + + assertEquals(false, result.get(0)); + assertEquals(true, result.get(1)); + assertEquals(true, result.get(2)); + assertEquals(true, result.get(3)); + assertEquals(false, result.get(4)); + } + @Test public void testStringFilterSerDe() { From 61d837146bac2280609ab152104ea10c25b7d22b Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Mon, 17 Aug 2026 16:47:42 +0800 Subject: [PATCH 2/3] fix: endOfPixel --- .../pixels/core/reader/TimeColumnReader.java | 20 +++++++++---------- 1 file changed, 10 insertions(+), 10 deletions(-) diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java index 3a79a6deab..5f244a9b77 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java @@ -121,21 +121,21 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, // read without copying the de-compacted content and isNull int numLeft = size, numToRead, bytesToDeCompact; - boolean endOfPixels; + boolean endOfPixel; for (int i = vectorIndex; numLeft > 0;) { if (elementIndex / pixelStride < (elementIndex + numLeft) / pixelStride) { // read to the end of the current pixel numToRead = pixelStride - elementIndex % pixelStride; - endOfPixels = true; + endOfPixel = true; } else { numToRead = numLeft; - endOfPixels = false; + endOfPixel = false; } - bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixels ? 7 : 0)) / 8; + bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixel ? 7 : 0)) / 8; // read isNull int pixelId = elementIndex / pixelStride; hasNull = chunkIndex.getPixelStatistics(pixelId).getStatistic().getHasNull(); @@ -144,7 +144,7 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, BitUtils.bitWiseDeCompact(vector.isNull, i, numToRead, inputBuffer, isNullOffset, isNullSkipBits, littleEndian); isNullOffset += bytesToDeCompact; - isNullSkipBits = endOfPixels ? 0 : (numToRead + isNullSkipBits) % 8; + isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; vector.noNulls = false; } else @@ -270,7 +270,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, // read without copying the de-compacted content and isNull int numLeft = size, numToRead, bytesToDeCompact, vectorWriteIndex = vectorIndex; boolean[] isNull = null; - boolean endOfPixels; + boolean endOfPixel; if (decoding || !nullsPadding) { isNull = new boolean[size]; @@ -281,14 +281,14 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, { // read to the end of the current pixel numToRead = pixelStride - elementIndex % pixelStride; - endOfPixels = true; + endOfPixel = true; } else { numToRead = numLeft; - endOfPixels = false; + endOfPixel = false; } - bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixels ? 7 : 0)) / 8; + bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixel ? 7 : 0)) / 8; // read isNull int pixelId = elementIndex / pixelStride; @@ -317,7 +317,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, } } isNullOffset += bytesToDeCompact; - isNullSkipBits = endOfPixels ? 0 : (numToRead + isNullSkipBits) % 8; + isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; vector.noNulls = false; } else From 04ead63be2d3a0a9c3d16a08cc4ff74e1a5955b3 Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Mon, 17 Aug 2026 22:29:27 +0800 Subject: [PATCH 3/3] refactor: use dedicated long-time vectors for TIME reads --- .../pixelsdb/pixels/core/TypeDescription.java | 9 +- .../pixels/core/reader/ColumnReader.java | 9 +- .../core/reader/LongTimeColumnReader.java | 306 ++++++++++++++++ .../core/reader/PixelsReaderOption.java | 10 +- .../reader/PixelsRecordReaderBufferImpl.java | 2 +- .../core/reader/PixelsRecordReaderImpl.java | 2 +- .../reader/PixelsRecordReaderStreamImpl.java | 2 +- .../pixels/core/reader/TimeColumnReader.java | 130 ++----- .../pixels/core/utils/DatetimeUtils.java | 2 +- .../pixels/core/vector/LongColumnVector.java | 20 -- .../core/vector/LongTimeColumnVector.java | 98 +++++ .../core/vector/VectorizedRowBatch.java | 12 +- .../core/reader/TestLongTimeColumnReader.java | 335 ++++++++++++++++++ .../core/reader/TestTimeColumnReader.java | 99 +----- ...tor.java => TestLongTimeColumnVector.java} | 41 ++- .../executor/predicate/ColumnFilter.java | 14 +- .../executor/predicate/TestPredicate.java | 8 +- 17 files changed, 836 insertions(+), 263 deletions(-) create mode 100644 pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/LongTimeColumnReader.java create mode 100644 pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongTimeColumnVector.java create mode 100644 pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestLongTimeColumnReader.java rename pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/{TestLongColumnVector.java => TestLongTimeColumnVector.java} (57%) diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java index 862a646946..a7abe46cae 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java @@ -1249,9 +1249,9 @@ private ColumnVector createColumn(int maxSize, int vectorLayout, boolean... useE case DATE: return new DateColumnVector(maxSize); case TIME: - if (VectorLayout.match(vectorLayout, VectorLayout.TIME_AS_PICO_LONG)) + if (VectorLayout.match(vectorLayout, VectorLayout.TIME_AS_LONG_TIME)) { - return new LongColumnVector(maxSize); + return new LongTimeColumnVector(maxSize, precision); } return new TimeColumnVector(maxSize, precision); case TIMESTAMP: @@ -1419,9 +1419,10 @@ public static final class VectorLayout */ public static final int INT_AS_LONG = 0x02; /** - * Create {@link io.pixelsdb.pixels.core.vector.LongColumnVector} containing picoseconds for TIME type. + * Create {@link io.pixelsdb.pixels.core.vector.LongTimeColumnVector} for TIME type, + * storing picoseconds of day for Trino-native zero-copy. */ - public static final int TIME_AS_PICO_LONG = 0x04; + public static final int TIME_AS_LONG_TIME = 0x04; public static boolean match(int layout1, int layout2) { diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ColumnReader.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ColumnReader.java index 294783f3e2..bd6188772d 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ColumnReader.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ColumnReader.java @@ -101,7 +101,14 @@ public static ColumnReader newColumnReader(TypeDescription type, PixelsReaderOpt case DATE: return new DateColumnReader(type); case TIME: - return new TimeColumnReader(type); + if (option.isReadTimeColumnAsLongTimeVector()) + { + return new LongTimeColumnReader(type); + } + else + { + return new TimeColumnReader(type); + } case TIMESTAMP: return new TimestampColumnReader(type); case BINARY: diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/LongTimeColumnReader.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/LongTimeColumnReader.java new file mode 100644 index 0000000000..6024c579fb --- /dev/null +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/LongTimeColumnReader.java @@ -0,0 +1,306 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.core.reader; + +import io.pixelsdb.pixels.core.PixelsProto; +import io.pixelsdb.pixels.core.TypeDescription; +import io.pixelsdb.pixels.core.encoding.RunLenIntDecoder; +import io.pixelsdb.pixels.core.utils.BitUtils; +import io.pixelsdb.pixels.core.utils.Bitmap; +import io.pixelsdb.pixels.core.utils.ByteBufferInputStream; +import io.pixelsdb.pixels.core.vector.ColumnVector; +import io.pixelsdb.pixels.core.vector.LongTimeColumnVector; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.util.Arrays; + +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOS_PER_MILLIS; + +/** + * Reads TIME column chunks (on-disk millis of day as int) into a + * {@link LongTimeColumnVector} in picoseconds of day. + *

+ * Selected when {@link PixelsReaderOption#isReadTimeColumnAsLongTimeVector()} is true, + * analogous to {@link LongColumnReader} for SHORT/INT → long output vector layout. + * + * @author gengdy + * @create 2026-08-17 + */ +public class LongTimeColumnReader extends ColumnReader +{ + private ByteBuffer inputBuffer = null; + private InputStream inputStream = null; + private RunLenIntDecoder decoder = null; + + LongTimeColumnReader(TypeDescription type) + { + super(type); + } + + @Override + public void close() throws IOException + { + if (inputStream != null) + { + inputStream.close(); + } + } + + @Override + public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, + int offset, int size, int pixelStride, final int vectorIndex, + ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex) throws IOException + { + LongTimeColumnVector columnVector = (LongTimeColumnVector) vector; + boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); + boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); + boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); + if (offset == 0) + { + if (inputStream != null) + { + inputStream.close(); + } + this.inputBuffer = input; + this.inputBuffer.order(littleEndian ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN); + inputStream = new ByteBufferInputStream(inputBuffer, inputBuffer.position(), inputBuffer.limit()); + decoder = new RunLenIntDecoder(inputStream, true); + isNullOffset = inputBuffer.position() + chunkIndex.getIsNullOffset(); + isNullSkipBits = 0; + hasNull = true; + elementIndex = 0; + } + + int numLeft = size, numToRead, bytesToDeCompact; + boolean endOfPixel; + for (int i = vectorIndex; numLeft > 0;) + { + if (elementIndex / pixelStride < (elementIndex + numLeft) / pixelStride) + { + numToRead = pixelStride - elementIndex % pixelStride; + endOfPixel = true; + } + else + { + numToRead = numLeft; + endOfPixel = false; + } + bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixel ? 7 : 0)) / 8; + int pixelId = elementIndex / pixelStride; + hasNull = chunkIndex.getPixelStatistics(pixelId).getStatistic().getHasNull(); + if (hasNull) + { + BitUtils.bitWiseDeCompact(columnVector.isNull, i, numToRead, + inputBuffer, isNullOffset, isNullSkipBits, littleEndian); + isNullOffset += bytesToDeCompact; + isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; + columnVector.noNulls = false; + } + else + { + Arrays.fill(columnVector.isNull, i, i + numToRead, false); + } + if (decoding) + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && columnVector.isNull[j])) + { + int millis = (int) decoder.next(); + columnVector.set(j, millis * PICOS_PER_MILLIS); + } + } + } + else + { + if (nullsPadding) + { + for (int j = i; j < i + numToRead; ++j) + { + // Issue #791: do not call set(), as it may clear the isNull flag of null values. + int millis = inputBuffer.getInt(); + columnVector.vector[j] = millis * PICOS_PER_MILLIS; + } + } + else + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && columnVector.isNull[j])) + { + int millis = inputBuffer.getInt(); + columnVector.set(j, millis * PICOS_PER_MILLIS); + } + } + } + } + numLeft -= numToRead; + elementIndex += numToRead; + i += numToRead; + } + } + + @Override + public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, + int offset, int size, int pixelStride, final int vectorIndex, + ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex, Bitmap selected) throws IOException + { + LongTimeColumnVector columnVector = (LongTimeColumnVector) vector; + boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); + boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); + boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); + if (offset == 0) + { + if (inputStream != null) + { + inputStream.close(); + } + this.inputBuffer = input; + this.inputBuffer.order(littleEndian ? ByteOrder.LITTLE_ENDIAN : ByteOrder.BIG_ENDIAN); + inputStream = new ByteBufferInputStream(inputBuffer, inputBuffer.position(), inputBuffer.limit()); + decoder = new RunLenIntDecoder(inputStream, true); + isNullOffset = inputBuffer.position() + chunkIndex.getIsNullOffset(); + isNullSkipBits = 0; + hasNull = true; + elementIndex = 0; + } + + int numLeft = size, numToRead, bytesToDeCompact, vectorWriteIndex = vectorIndex; + boolean[] isNull = null; + boolean endOfPixel; + if (decoding || !nullsPadding) + { + isNull = new boolean[size]; + } + for (int i = vectorIndex; numLeft > 0;) + { + if (elementIndex / pixelStride < (elementIndex + numLeft) / pixelStride) + { + numToRead = pixelStride - elementIndex % pixelStride; + endOfPixel = true; + } + else + { + numToRead = numLeft; + endOfPixel = false; + } + bytesToDeCompact = (numToRead + isNullSkipBits + (endOfPixel ? 7 : 0)) / 8; + + int pixelId = elementIndex / pixelStride; + hasNull = chunkIndex.getPixelStatistics(pixelId).getStatistic().getHasNull(); + if (hasNull) + { + if (!decoding && nullsPadding) + { + BitUtils.bitWiseDeCompact(columnVector.isNull, vectorWriteIndex, numToRead, inputBuffer, + isNullOffset, isNullSkipBits, littleEndian, selected, i - vectorIndex); + } + else + { + BitUtils.bitWiseDeCompact(isNull, i - vectorIndex, numToRead, inputBuffer, + isNullOffset, isNullSkipBits, littleEndian); + int k = vectorWriteIndex; + for (int j = i; j < i + numToRead; ++j) + { + if (selected.get(j - vectorIndex)) + { + columnVector.isNull[k++] = isNull[j - vectorIndex]; + } + } + } + isNullOffset += bytesToDeCompact; + isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; + columnVector.noNulls = false; + } + else + { + if (decoding || !nullsPadding) + { + Arrays.fill(isNull, i - vectorIndex, i - vectorIndex + numToRead, false); + } + } + + int originalVectorWriteIndex = vectorWriteIndex; + if (decoding) + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && isNull[j - vectorIndex])) + { + int millis = (int) decoder.next(); + if (selected.get(j - vectorIndex)) + { + columnVector.set(vectorWriteIndex++, millis * PICOS_PER_MILLIS); + } + } + else if (selected.get(j - vectorIndex)) + { + vectorWriteIndex++; + } + } + } + else + { + if (nullsPadding) + { + for (int j = i; j < i + numToRead; ++j) + { + int millis = inputBuffer.getInt(); + if (selected.get(j - vectorIndex)) + { + // Issue #791: do not call set(), as it may clear the isNull flag of null values. + columnVector.vector[vectorWriteIndex++] = millis * PICOS_PER_MILLIS; + } + } + } + else + { + for (int j = i; j < i + numToRead; ++j) + { + if (!(hasNull && isNull[j - vectorIndex])) + { + int millis = inputBuffer.getInt(); + if (selected.get(j - vectorIndex)) + { + columnVector.set(vectorWriteIndex++, millis * PICOS_PER_MILLIS); + } + } + else if (selected.get(j - vectorIndex)) + { + vectorWriteIndex++; + } + } + } + } + + if (!hasNull) + { + Arrays.fill(columnVector.isNull, originalVectorWriteIndex, vectorWriteIndex, false); + } + + numLeft -= numToRead; + elementIndex += numToRead; + i += numToRead; + } + } +} diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java index 463241d2b3..d24f0f6efc 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsReaderOption.java @@ -35,7 +35,7 @@ public class PixelsReaderOption private boolean enableEncodedColumnVector = false; // whether read encoded column vectors directly when possible private boolean readIntColumnAsLongVector = false; // whether read int32 columns as long column vectors, for backward compatibility of old query engines private boolean readShortColumnAsLongVector = false; // whether read int16 columns as long column vectors, for backward compatibility of old query engines - private boolean readTimeColumnAsLongVector = false; // whether read time columns as picoseconds in long vectors + private boolean readTimeColumnAsLongTimeVector = false; // whether read TIME as LongTimeColumnVector (picoseconds), for Trino-native layout private boolean exposeHiddenColumn = false; // whether expose the hidden commit timestamp column in the result batch private long transId = -1L; private long transTimestamp = -1L; // -1 means no need to consider the timestamp when reading data @@ -179,15 +179,15 @@ public boolean isReadShortColumnAsLongVector() return readShortColumnAsLongVector; } - public PixelsReaderOption readTimeColumnAsLongVector(boolean readTimeColumnAsLongVector) + public PixelsReaderOption readTimeColumnAsLongTimeVector(boolean readTimeColumnAsLongTimeVector) { - this.readTimeColumnAsLongVector = readTimeColumnAsLongVector; + this.readTimeColumnAsLongTimeVector = readTimeColumnAsLongTimeVector; return this; } - public boolean isReadTimeColumnAsLongVector() + public boolean isReadTimeColumnAsLongTimeVector() { - return readTimeColumnAsLongVector; + return readTimeColumnAsLongTimeVector; } public PixelsReaderOption exposeHiddenColumn(boolean exposeHiddenColumn) diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java index a9f01895ad..54dd92cbce 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderBufferImpl.java @@ -119,7 +119,7 @@ public PixelsRecordReaderBufferImpl(PixelsReaderOption option, this.option = option; this.vectorLayout = (option.isReadIntColumnAsLongVector() ? TypeDescription.VectorLayout.INT_AS_LONG : 0) | (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0) | - (option.isReadTimeColumnAsLongVector() ? TypeDescription.VectorLayout.TIME_AS_PICO_LONG : 0); + (option.isReadTimeColumnAsLongTimeVector() ? TypeDescription.VectorLayout.TIME_AS_LONG_TIME : 0); this.activeMemtableData = activeMemtableData; this.fileIds = fileIds; this.storage = storage; diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java index 3f91b42054..f3d2145ae0 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderImpl.java @@ -174,7 +174,7 @@ public PixelsRecordReaderImpl(PhysicalReader physicalReader, this.enableEncodedVector = option.isEnableEncodedColumnVector(); this.vectorLayout = (option.isReadIntColumnAsLongVector() ? TypeDescription.VectorLayout.INT_AS_LONG : 0) | (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0) | - (option.isReadTimeColumnAsLongVector() ? TypeDescription.VectorLayout.TIME_AS_PICO_LONG : 0); + (option.isReadTimeColumnAsLongTimeVector() ? TypeDescription.VectorLayout.TIME_AS_LONG_TIME : 0); this.enableMetrics = enableMetrics; this.metricsDir = metricsDir; this.readPerfMetrics = new ReadPerfMetrics(); diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java index c8cf76d2b4..d90ad51bf8 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/PixelsRecordReaderStreamImpl.java @@ -125,7 +125,7 @@ public PixelsRecordReaderStreamImpl(PhysicalReader physicalReader, this.option = option; this.vectorLayout = (option.isReadIntColumnAsLongVector() ? TypeDescription.VectorLayout.INT_AS_LONG : 0) | (option.isReadShortColumnAsLongVector() ? TypeDescription.VectorLayout.SHORT_AS_LONG : 0) | - (option.isReadTimeColumnAsLongVector() ? TypeDescription.VectorLayout.TIME_AS_PICO_LONG : 0); + (option.isReadTimeColumnAsLongTimeVector() ? TypeDescription.VectorLayout.TIME_AS_LONG_TIME : 0); this.includedColumnTypes = new ArrayList<>(); checkBeforeRead(); } diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java index 5f244a9b77..972392281b 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/TimeColumnReader.java @@ -26,7 +26,6 @@ import io.pixelsdb.pixels.core.utils.Bitmap; import io.pixelsdb.pixels.core.utils.ByteBufferInputStream; import io.pixelsdb.pixels.core.vector.ColumnVector; -import io.pixelsdb.pixels.core.vector.LongColumnVector; import io.pixelsdb.pixels.core.vector.TimeColumnVector; import java.io.IOException; @@ -35,11 +34,13 @@ import java.nio.ByteOrder; import java.util.Arrays; -import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; - /** * Pixels time column reader. * All time values are translated to the specified time zone after read from file. + *

+ * Reads into {@link TimeColumnVector} (millis of day). For Trino-native picoseconds + * output vector layout, use {@link LongTimeColumnReader} instead (selected by + * {@link PixelsReaderOption#readTimeColumnAsLongTimeVector(boolean)}). * * @author hank * @create 2021-04-28 @@ -87,8 +88,7 @@ public void close() throws IOException * @param size number of values to read * @param pixelStride the stride (number of rows) in a pixels. * @param vectorIndex the index from where we start reading values into the vector - * @param vector vector to read values into, it is a {@link LongColumnVector} of picoseconds of day - * if the read option requires the time column to be read as a long vector + * @param vector vector to read values into * @param chunkIndex the metadata of the column chunk to read. * @throws IOException */ @@ -97,9 +97,7 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, int offset, int size, int pixelStride, final int vectorIndex, ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex) throws IOException { - // longColumnVector is not null if the time values are to be stored as picoseconds of day. - LongColumnVector longColumnVector = vector instanceof LongColumnVector ? (LongColumnVector) vector : null; - TimeColumnVector timeColumnVector = longColumnVector == null ? (TimeColumnVector) vector : null; + TimeColumnVector columnVector = (TimeColumnVector) vector; boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); @@ -141,37 +139,24 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, hasNull = chunkIndex.getPixelStatistics(pixelId).getStatistic().getHasNull(); if (hasNull) { - BitUtils.bitWiseDeCompact(vector.isNull, i, numToRead, + BitUtils.bitWiseDeCompact(columnVector.isNull, i, numToRead, inputBuffer, isNullOffset, isNullSkipBits, littleEndian); isNullOffset += bytesToDeCompact; isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; - vector.noNulls = false; + columnVector.noNulls = false; } else { - Arrays.fill(vector.isNull, i, i + numToRead, false); + Arrays.fill(columnVector.isNull, i, i + numToRead, false); } // read content if (decoding) { for (int j = i; j < i + numToRead; ++j) { - if (!(hasNull && vector.isNull[j])) + if (!(hasNull && columnVector.isNull[j])) { - int millis = (int) decoder.next(); - if (longColumnVector != null) - { - longColumnVector.vector[j] = millis * PICOSECONDS_PER_MILLISECOND; - longColumnVector.isNull[j] = false; - if (j >= longColumnVector.getWriteIndex()) - { - longColumnVector.setWriteIndex(j + 1); - } - } - else - { - timeColumnVector.set(j, millis); - } + columnVector.set(j, (int) decoder.next()); } } } @@ -182,38 +167,17 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, for (int j = i; j < i + numToRead; ++j) { // Issue #791: do not call the set() method, as it may clear the isNull flag of null values. - int millis = inputBuffer.getInt(); - if (longColumnVector != null) - { - longColumnVector.vector[j] = millis * PICOSECONDS_PER_MILLISECOND; - } - else - { - timeColumnVector.times[j] = millis; - } + columnVector.times[j] = inputBuffer.getInt(); } } else { for (int j = i; j < i + numToRead; ++j) { - if (!(hasNull && vector.isNull[j])) + if (!(hasNull && columnVector.isNull[j])) { // If time column is not encoded, it is written as integers instead of longs. - int millis = inputBuffer.getInt(); - if (longColumnVector != null) - { - longColumnVector.vector[j] = millis * PICOSECONDS_PER_MILLISECOND; - longColumnVector.isNull[j] = false; - if (j >= longColumnVector.getWriteIndex()) - { - longColumnVector.setWriteIndex(j + 1); - } - } - else - { - timeColumnVector.set(j, millis); - } + columnVector.set(j, inputBuffer.getInt()); } } } @@ -234,8 +198,7 @@ public void read(ByteBuffer input, PixelsProto.ColumnEncoding encoding, * @param size number of values to read * @param pixelStride the stride (number of rows) in a pixels. * @param vectorIndex the index from where we start reading values into the vector - * @param vector vector to read values into, it is a {@link LongColumnVector} of picoseconds of day - * if the read option requires the time column to be read as a long vector + * @param vector vector to read values into * @param chunkIndex the metadata of the column chunk to read. * @param selected whether the value is selected, use the vectorIndex as the 0 offset of the selected * @throws IOException @@ -245,9 +208,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, int offset, int size, int pixelStride, final int vectorIndex, ColumnVector vector, PixelsProto.ColumnChunkIndex chunkIndex, Bitmap selected) throws IOException { - // longColumnVector is not null if the time values are to be stored as picoseconds of day. - LongColumnVector longColumnVector = vector instanceof LongColumnVector ? (LongColumnVector) vector : null; - TimeColumnVector timeColumnVector = longColumnVector == null ? (TimeColumnVector) vector : null; + TimeColumnVector columnVector = (TimeColumnVector) vector; boolean nullsPadding = chunkIndex.hasNullsPadding() && chunkIndex.getNullsPadding(); boolean decoding = encoding.getKind().equals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH); boolean littleEndian = chunkIndex.hasLittleEndian() && chunkIndex.getLittleEndian(); @@ -298,7 +259,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, if (!decoding && nullsPadding) { // read isNull directly into the vector of the column chunk - BitUtils.bitWiseDeCompact(vector.isNull, vectorWriteIndex, numToRead, inputBuffer, + BitUtils.bitWiseDeCompact(columnVector.isNull, vectorWriteIndex, numToRead, inputBuffer, isNullOffset, isNullSkipBits, littleEndian, selected, i - vectorIndex); } else @@ -306,19 +267,19 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, // need to keep isNull for later use BitUtils.bitWiseDeCompact(isNull, i - vectorIndex, numToRead, inputBuffer, isNullOffset, isNullSkipBits, littleEndian); - // update vector.isNull + // update columnVector.isNull int k = vectorWriteIndex; for (int j = i; j < i + numToRead; ++j) { if (selected.get(j - vectorIndex)) { - vector.isNull[k++] = isNull[j - vectorIndex]; + columnVector.isNull[k++] = isNull[j - vectorIndex]; } } } isNullOffset += bytesToDeCompact; isNullSkipBits = endOfPixel ? 0 : (numToRead + isNullSkipBits) % 8; - vector.noNulls = false; + columnVector.noNulls = false; } else { @@ -326,7 +287,7 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, { Arrays.fill(isNull, i - vectorIndex, i - vectorIndex + numToRead, false); } - // update vector.isNull later to avoid bitmap unnecessary traversal + // update columnVector.isNull later to avoid bitmap unnecessary traversal } // read content @@ -337,23 +298,10 @@ public void readSelected(ByteBuffer input, PixelsProto.ColumnEncoding encoding, { if (!(hasNull && isNull[j - vectorIndex])) { - int millis = (int) decoder.next(); + int value = (int) decoder.next(); if (selected.get(j - vectorIndex)) { - if (longColumnVector != null) - { - longColumnVector.vector[vectorWriteIndex] = millis * PICOSECONDS_PER_MILLISECOND; - longColumnVector.isNull[vectorWriteIndex] = false; - if (vectorWriteIndex >= longColumnVector.getWriteIndex()) - { - longColumnVector.setWriteIndex(vectorWriteIndex + 1); - } - vectorWriteIndex++; - } - else - { - timeColumnVector.set(vectorWriteIndex++, millis); - } + columnVector.set(vectorWriteIndex++, value); } } else if (selected.get(j - vectorIndex)) @@ -368,18 +316,11 @@ else if (selected.get(j - vectorIndex)) { for (int j = i; j < i + numToRead; ++j) { - int millis = inputBuffer.getInt(); + int value = inputBuffer.getInt(); if (selected.get(j - vectorIndex)) { // Issue #791: do not call the set() method, as it may clear the isNull flag of null values. - if (longColumnVector != null) - { - longColumnVector.vector[vectorWriteIndex++] = millis * PICOSECONDS_PER_MILLISECOND; - } - else - { - timeColumnVector.times[vectorWriteIndex++] = millis; - } + columnVector.times[vectorWriteIndex++] = value; } } } @@ -390,23 +331,10 @@ else if (selected.get(j - vectorIndex)) if (!(hasNull && isNull[j - vectorIndex])) { // If time column is not encoded, it is written as integers instead of longs. - int millis = inputBuffer.getInt(); + int value = inputBuffer.getInt(); if (selected.get(j - vectorIndex)) { - if (longColumnVector != null) - { - longColumnVector.vector[vectorWriteIndex] = millis * PICOSECONDS_PER_MILLISECOND; - longColumnVector.isNull[vectorWriteIndex] = false; - if (vectorWriteIndex >= longColumnVector.getWriteIndex()) - { - longColumnVector.setWriteIndex(vectorWriteIndex + 1); - } - vectorWriteIndex++; - } - else - { - timeColumnVector.set(vectorWriteIndex++, millis); - } + columnVector.set(vectorWriteIndex++, value); } } else if (selected.get(j - vectorIndex)) @@ -417,10 +345,10 @@ else if (selected.get(j - vectorIndex)) } } - // update vector.isNull if has no nulls + // update columnVector.isNull if has no nulls if (!hasNull) { - Arrays.fill(vector.isNull, originalVectorWriteIndex, vectorWriteIndex, false); + Arrays.fill(columnVector.isNull, originalVectorWriteIndex, vectorWriteIndex, false); } // update variables diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java index 9b2569bf60..2cc82c936e 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/utils/DatetimeUtils.java @@ -42,7 +42,7 @@ public class DatetimeUtils private static final long NANOS_PER_MILLIS = 1000_000L; private static final long MICROS_PER_SEC = 1000_000L; private static final long NANOS_PER_MICROS = 1000L; - public static final long PICOSECONDS_PER_MILLISECOND = 1_000_000_000L; + public static final long PICOS_PER_MILLIS = 1_000_000_000L; public static long microsToMillis(long micros) { diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java index f3bdf1fadd..0771255dbd 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongColumnVector.java @@ -22,14 +22,12 @@ import com.google.flatbuffers.FlatBufferBuilder; import io.pixelsdb.pixels.core.flat.ColumnVectorFlat; import io.pixelsdb.pixels.core.flat.LongColumnVectorFlat; -import io.pixelsdb.pixels.core.flat.TimeColumnVectorFlat; import io.pixelsdb.pixels.core.utils.Bitmap; import java.nio.ByteBuffer; import java.util.Arrays; import static com.google.common.base.Preconditions.checkArgument; -import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; import static java.util.Objects.requireNonNull; /** @@ -367,22 +365,4 @@ public static LongColumnVector deserialize(LongColumnVectorFlat flat) vector.deserializeBase(flat.base()); return vector; } - - /** - * Deserialize a time vector directly into Trino's physical representation. - * Pixels stores TIME values as milliseconds of day, while Trino stores them as - * picoseconds of day. - */ - public static LongColumnVector deserializeTime(TimeColumnVectorFlat flat) - { - int length = flat.base().length(); - LongColumnVector vector = new LongColumnVector(length); - for (int i = 0; i < flat.timesLength(); ++i) - { - vector.vector[i] = (long) flat.times(i) * PICOSECONDS_PER_MILLISECOND; - } - vector.deserializeBase(flat.base()); - vector.memoryUsage += (long) (Long.BYTES - Integer.BYTES) * length; - return vector; - } } diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongTimeColumnVector.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongTimeColumnVector.java new file mode 100644 index 0000000000..8d87b2488a --- /dev/null +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/LongTimeColumnVector.java @@ -0,0 +1,98 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.core.vector; + +import com.google.flatbuffers.FlatBufferBuilder; +import io.pixelsdb.pixels.core.flat.TimeColumnVectorFlat; + +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOS_PER_MILLIS; + +/** + * TIME column vector in Trino-native picoseconds-of-day layout. + *

+ * Stores values in a {@code long[]} so they can be zero-copied into Trino + * {@code LongArrayBlock}. Pixels on-disk and serialized TIME is milliseconds + * of day; conversion to picoseconds happens while reading. + * + * @author gengdy + * @create 2026-08-17 + */ +public class LongTimeColumnVector extends LongColumnVector +{ + private final int precision; + + public LongTimeColumnVector(int precision) + { + this(VectorizedRowBatch.DEFAULT_SIZE, precision); + } + + public LongTimeColumnVector(int len, int precision) + { + super(len); + if (precision != 3) + { + // TODO: support more precisions. + throw new UnsupportedOperationException("Time type currently only supports precision 3"); + } + this.precision = precision; + } + + public int getPrecision() + { + return precision; + } + + public void set(int elementNum, long picoOfDay) + { + if (elementNum >= writeIndex) + { + writeIndex = elementNum + 1; + } + this.isNull[elementNum] = false; + this.vector[elementNum] = picoOfDay; + } + + @Override + public byte getFlatBufferType() + { + throw new UnsupportedOperationException("LongTimeColumnVector is a read-only vector layout"); + } + + @Override + public int serialize(FlatBufferBuilder builder) + { + throw new UnsupportedOperationException("LongTimeColumnVector is a read-only vector layout"); + } + + /** + * Deserialize native millis-of-day TIME directly into picoseconds-of-day layout. + */ + public static LongTimeColumnVector deserialize(TimeColumnVectorFlat flat) + { + LongTimeColumnVector vector = new LongTimeColumnVector(flat.base().length(), flat.precision()); + for (int i = 0; i < flat.timesLength(); ++i) + { + vector.vector[i] = (long) flat.times(i) * PICOS_PER_MILLIS; + } + vector.deserializeBase(flat.base()); + vector.memoryUsage += (long) (Long.BYTES - Integer.BYTES) * vector.length; + return vector; + } +} diff --git a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java index fa8caa4f46..4d88c5e274 100644 --- a/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java +++ b/pixels-core/src/main/java/io/pixelsdb/pixels/core/vector/VectorizedRowBatch.java @@ -423,10 +423,10 @@ public static VectorizedRowBatch deserialize(ByteBuffer buffer) } /** - * Deserialize a row batch using the requested physical vector layout. + * Deserialize a row batch using the requested read vector layout. * * @param buffer the ByteBuffer containing serialized batch data - * @param vectorLayout requested physical vector layout + * @param vectorLayout requested read vector layout * @return the deserialized row batch */ public static VectorizedRowBatch deserialize(ByteBuffer buffer, int vectorLayout) @@ -478,12 +478,10 @@ public static VectorizedRowBatch deserialize(ByteBuffer buffer, int vectorLayout batch.cols[i] = LongDecimalColumnVector.deserialize((LongDecimalColumnVectorFlat) batchFlat.cols(new LongDecimalColumnVectorFlat(), i)); break; case ColumnVectorFlat.TimeColumnVectorFlat: - TimeColumnVectorFlat timeFlat = - (TimeColumnVectorFlat) batchFlat.cols(new TimeColumnVectorFlat(), i); - if (TypeDescription.VectorLayout.match( - vectorLayout, TypeDescription.VectorLayout.TIME_AS_PICO_LONG)) + TimeColumnVectorFlat timeFlat = (TimeColumnVectorFlat) batchFlat.cols(new TimeColumnVectorFlat(), i); + if (TypeDescription.VectorLayout.match(vectorLayout, TypeDescription.VectorLayout.TIME_AS_LONG_TIME)) { - batch.cols[i] = LongColumnVector.deserializeTime(timeFlat); + batch.cols[i] = LongTimeColumnVector.deserialize(timeFlat); } else { diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestLongTimeColumnReader.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestLongTimeColumnReader.java new file mode 100644 index 0000000000..db1ddff4b3 --- /dev/null +++ b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestLongTimeColumnReader.java @@ -0,0 +1,335 @@ +/* + * Copyright 2026 PixelsDB. + * + * This file is part of Pixels. + * + * Pixels is free software: you can redistribute it and/or modify + * it under the terms of the Affero GNU General Public License as + * published by the Free Software Foundation, either version 3 of + * the License, or (at your option) any later version. + * + * Pixels is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * Affero GNU General Public License for more details. + * + * You should have received a copy of the Affero GNU General Public + * License along with Pixels. If not, see + * . + */ +package io.pixelsdb.pixels.core.reader; + +import io.pixelsdb.pixels.core.PixelsProto; +import io.pixelsdb.pixels.core.TypeDescription; +import io.pixelsdb.pixels.core.encoding.EncodingLevel; +import io.pixelsdb.pixels.core.utils.Bitmap; +import io.pixelsdb.pixels.core.vector.LongTimeColumnVector; +import io.pixelsdb.pixels.core.vector.TimeColumnVector; +import io.pixelsdb.pixels.core.writer.PixelsWriterOption; +import io.pixelsdb.pixels.core.writer.TimeColumnWriter; +import org.junit.Test; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; + +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOS_PER_MILLIS; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Tests for {@link LongTimeColumnReader}: on-disk TIME (millis int) projected to + * {@link LongTimeColumnVector} (picoseconds), selected by + * {@link PixelsReaderOption#readTimeColumnAsLongTimeVector(boolean)}. + * + * @author gengdy + * @create 2026-08-17 + */ +public class TestLongTimeColumnReader +{ + private static TimeColumnVector createSampleMillisVector(int numRows) + { + TimeColumnVector vector = new TimeColumnVector(numRows, 3); + vector.add(100); + vector.add(103); + vector.add(106); + vector.add(34); + vector.addNull(); + vector.add(54); + vector.add(55); + vector.add(67); + vector.addNull(); + vector.add(34); + vector.add(555); + vector.add(565); + vector.add(234); + vector.add(675); + vector.add(235); + vector.add(32434); + vector.addNull(); + vector.add(6); + vector.add(7); + vector.add(65656565); + vector.add(3434); + vector.add(54578); + return vector; + } + + private static void assertMillisProjectedToPicos(TimeColumnVector source, + LongTimeColumnVector target, int numRows) + { + assertEquals(source.noNulls, target.noNulls); + for (int i = 0; i < numRows; ++i) + { + assertEquals("isNull mismatch at row " + i, source.isNull[i], target.isNull[i]); + if (source.noNulls || !source.isNull[i]) + { + assertEquals("pico mismatch at row " + i, + (long) source.times[i] * PICOS_PER_MILLIS, target.vector[i]); + } + } + } + + @Test + public void testFactorySelectsLongTimeColumnReader() + { + TypeDescription timeType = TypeDescription.createTime(3); + assertTrue(ColumnReader.newColumnReader(timeType, new PixelsReaderOption()) + instanceof TimeColumnReader); + assertTrue(ColumnReader.newColumnReader(timeType, + new PixelsReaderOption().readTimeColumnAsLongTimeVector(true)) + instanceof LongTimeColumnReader); + } + + @Test + public void testNullsPadding() throws IOException + { + int pixelsStride = 10; + int numRows = 22; + TimeColumnVector source = createSampleMillisVector(numRows); + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(true); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + columnWriter.write(source, numRows); + columnWriter.flush(); + columnWriter.close(); + + LongTimeColumnVector target = new LongTimeColumnVector(numRows, 3); + LongTimeColumnReader reader = new LongTimeColumnReader(TypeDescription.createTime(3)); + reader.read(ByteBuffer.wrap(columnWriter.getColumnChunkContent()), + columnWriter.getColumnChunkEncoding().build(), 0, numRows, + pixelsStride, 0, target, columnWriter.getColumnChunkIndex().build()); + reader.close(); + + assertMillisProjectedToPicos(source, target, numRows); + } + + @Test + public void testWithoutNullsPadding() throws IOException + { + int pixelsStride = 10; + int numRows = 22; + TimeColumnVector source = createSampleMillisVector(numRows); + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(false); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + columnWriter.write(source, numRows); + columnWriter.flush(); + columnWriter.close(); + + LongTimeColumnVector target = new LongTimeColumnVector(numRows, 3); + LongTimeColumnReader reader = new LongTimeColumnReader(TypeDescription.createTime(3)); + reader.read(ByteBuffer.wrap(columnWriter.getColumnChunkContent()), + columnWriter.getColumnChunkEncoding().build(), 0, numRows, + pixelsStride, 0, target, columnWriter.getColumnChunkIndex().build()); + reader.close(); + + assertMillisProjectedToPicos(source, target, numRows); + } + + @Test + public void testRunLengthEncoded() throws IOException + { + int pixelStride = 4; + int[] millis = {0, 1, 3_723_004, 86_399_999}; + TimeColumnVector source = new TimeColumnVector(millis.length + 1, 3); + for (int value : millis) + { + source.add(value); + } + source.addNull(); + + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL2).nullsPadding(false); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + columnWriter.write(source, millis.length + 1); + columnWriter.flush(); + columnWriter.close(); + + assertEquals(PixelsProto.ColumnEncoding.Kind.RUNLENGTH, + columnWriter.getColumnChunkEncoding().build().getKind()); + + LongTimeColumnVector target = new LongTimeColumnVector(millis.length + 1, 3); + LongTimeColumnReader reader = new LongTimeColumnReader(TypeDescription.createTime(3)); + reader.read(ByteBuffer.wrap(columnWriter.getColumnChunkContent()), + columnWriter.getColumnChunkEncoding().build(), 0, millis.length + 1, + pixelStride, 0, target, columnWriter.getColumnChunkIndex().build()); + reader.close(); + + for (int i = 0; i < millis.length; ++i) + { + assertFalse(target.isNull[i]); + assertEquals((long) millis[i] * PICOS_PER_MILLIS, target.vector[i]); + } + assertTrue(target.isNull[millis.length]); + } + + @Test + public void testSelectedWithNullPadding() throws IOException + { + int pixelStride = 4; + int[] millis = {10, 20, 30, 40, 50}; + TimeColumnVector source = new TimeColumnVector(millis.length, 3); + source.add(millis[0]); + source.addNull(); + for (int i = 2; i < millis.length; ++i) + { + source.add(millis[i]); + } + + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(true); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + columnWriter.write(source, millis.length); + columnWriter.flush(); + columnWriter.close(); + + Bitmap selected = new Bitmap(millis.length, true); + selected.clear(0); + selected.clear(3); + LongTimeColumnVector target = new LongTimeColumnVector(3, 3); + LongTimeColumnReader reader = new LongTimeColumnReader(TypeDescription.createTime(3)); + reader.readSelected( + ByteBuffer.wrap(columnWriter.getColumnChunkContent()), + columnWriter.getColumnChunkEncoding().build(), + 0, millis.length, pixelStride, 0, target, + columnWriter.getColumnChunkIndex().build(), selected); + reader.close(); + + assertTrue(target.isNull[0]); + assertEquals((long) millis[2] * PICOS_PER_MILLIS, target.vector[1]); + assertEquals((long) millis[4] * PICOS_PER_MILLIS, target.vector[2]); + } + + @Test + public void testSelectedWithoutNullPadding() throws IOException + { + int pixelsStride = 10; + int numRows = 22; + TimeColumnVector source = createSampleMillisVector(numRows); + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(pixelsStride).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(false); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + columnWriter.write(source, numRows); + columnWriter.flush(); + columnWriter.close(); + + Bitmap selected = new Bitmap(numRows, true); + selected.clear(0); + selected.clear(10); + selected.clear(20); + + LongTimeColumnVector target = new LongTimeColumnVector(numRows, 3); + LongTimeColumnReader reader = new LongTimeColumnReader(TypeDescription.createTime(3)); + reader.readSelected(ByteBuffer.wrap(columnWriter.getColumnChunkContent()), + columnWriter.getColumnChunkEncoding().build(), 0, numRows, + pixelsStride, 0, target, columnWriter.getColumnChunkIndex().build(), selected); + reader.close(); + + for (int i = 0, j = 0; i < numRows; ++i) + { + if (i % 10 != 0) + { + assertEquals(source.isNull[i], target.isNull[j]); + if (source.noNulls || !source.isNull[i]) + { + assertEquals((long) source.times[i] * PICOS_PER_MILLIS, target.vector[j]); + } + j++; + } + } + } + + @Test + public void testVectorLayoutCreatesLongTimeColumnVector() + { + TypeDescription timeType = TypeDescription.createTime(3); + assertTrue(timeType.createRowBatch(4, TypeDescription.VectorLayout.TIME_AS_LONG_TIME) + .cols[0] instanceof LongTimeColumnVector); + assertTrue(timeType.createRowBatch(4).cols[0] instanceof TimeColumnVector); + } + + @Test + public void testLargeFragmented() throws IOException + { + int numBatches = 15; + int numRows = 1024; + TimeColumnVector origin = new TimeColumnVector(numRows, 3); + for (int j = 0; j < numRows; j++) + { + if (j % 100 == 0) + { + origin.addNull(); + } + else + { + origin.add(1000); + } + } + + PixelsWriterOption writerOption = new PixelsWriterOption() + .pixelStride(10000).byteOrder(ByteOrder.LITTLE_ENDIAN) + .encodingLevel(EncodingLevel.EL0).nullsPadding(true); + TimeColumnWriter columnWriter = new TimeColumnWriter( + TypeDescription.createTime(3), writerOption); + for (int i = 0; i < numBatches; i++) + { + columnWriter.write(origin, numRows); + } + columnWriter.flush(); + columnWriter.close(); + + byte[] content = columnWriter.getColumnChunkContent(); + PixelsProto.ColumnChunkIndex chunkIndex = columnWriter.getColumnChunkIndex().build(); + PixelsProto.ColumnEncoding encoding = columnWriter.getColumnChunkEncoding().build(); + LongTimeColumnReader reader = new LongTimeColumnReader(TypeDescription.createTime(3)); + LongTimeColumnVector target = new LongTimeColumnVector(numBatches * numRows, 3); + ByteBuffer buffer = ByteBuffer.wrap(content); + reader.read(buffer, encoding, 0, 123, 10000, 0, target, chunkIndex); + reader.read(buffer, encoding, 123, 456, 10000, 123, target, chunkIndex); + reader.read(buffer, encoding, 123 + 456, numBatches * numRows - 123 - 456, + 10000, 123 + 456, target, chunkIndex); + reader.close(); + + for (int i = 0; i < numBatches * numRows; i++) + { + assertEquals(origin.isNull[i % numRows], target.isNull[i]); + if (target.noNulls || !target.isNull[i]) + { + assertEquals((long) origin.times[i % numRows] * PICOS_PER_MILLIS, + target.vector[i]); + } + } + } +} diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java index a054c8e798..c36c7c91be 100644 --- a/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java +++ b/pixels-core/src/test/java/io/pixelsdb/pixels/core/reader/TestTimeColumnReader.java @@ -23,7 +23,6 @@ import io.pixelsdb.pixels.core.TypeDescription; import io.pixelsdb.pixels.core.encoding.EncodingLevel; import io.pixelsdb.pixels.core.utils.Bitmap; -import io.pixelsdb.pixels.core.vector.LongColumnVector; import io.pixelsdb.pixels.core.vector.TimeColumnVector; import io.pixelsdb.pixels.core.writer.PixelsWriterOption; import io.pixelsdb.pixels.core.writer.TimeColumnWriter; @@ -33,12 +32,11 @@ import java.nio.ByteBuffer; import java.nio.ByteOrder; -import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertTrue; - /** + * Tests for {@link TimeColumnReader} reading into {@link TimeColumnVector} + * (millis of day). For picoseconds / {@code LongTimeColumnVector} output layout, + * see {@link TestLongTimeColumnReader}. + * * @author hank * @create 2023-08-20 Zermatt */ @@ -223,95 +221,6 @@ public void testSelected() throws IOException } } - @Test - public void testLongVectorPicoseconds() throws IOException - { - int pixelStride = 4; - int[] millis = {0, 1, 3_723_004, 86_399_999}; - PixelsWriterOption writerOption = new PixelsWriterOption() - .pixelStride(pixelStride).byteOrder(ByteOrder.LITTLE_ENDIAN) - .encodingLevel(EncodingLevel.EL2).nullsPadding(false); - TimeColumnWriter columnWriter = new TimeColumnWriter( - TypeDescription.createTime(3), writerOption); - TimeColumnVector source = new TimeColumnVector(millis.length + 1, 3); - for (int value : millis) - { - source.add(value); - } - source.addNull(); - columnWriter.write(source, millis.length + 1); - columnWriter.flush(); - columnWriter.close(); - - TypeDescription timeType = TypeDescription.createTime(3); - assertTrue(timeType.createRowBatch( - millis.length + 1, TypeDescription.VectorLayout.TIME_AS_PICO_LONG) - .cols[0] instanceof LongColumnVector); - - LongColumnVector target = new LongColumnVector(millis.length + 1); - TimeColumnReader reader = new TimeColumnReader(timeType); - reader.read( - ByteBuffer.wrap(columnWriter.getColumnChunkContent()), - columnWriter.getColumnChunkEncoding().build(), - 0, - millis.length + 1, - pixelStride, - 0, - target, - columnWriter.getColumnChunkIndex().build()); - reader.close(); - - for (int i = 0; i < millis.length; ++i) - { - assertFalse(target.isNull[i]); - assertEquals((long) millis[i] * PICOSECONDS_PER_MILLISECOND, target.vector[i]); - } - assertTrue(target.isNull[millis.length]); - } - - @Test - public void testSelectedLongVectorWithNullPadding() throws IOException - { - int pixelStride = 4; - int[] millis = {10, 20, 30, 40, 50}; - PixelsWriterOption writerOption = new PixelsWriterOption() - .pixelStride(pixelStride).byteOrder(ByteOrder.LITTLE_ENDIAN) - .encodingLevel(EncodingLevel.EL0).nullsPadding(true); - TimeColumnWriter columnWriter = new TimeColumnWriter( - TypeDescription.createTime(3), writerOption); - TimeColumnVector source = new TimeColumnVector(millis.length, 3); - source.add(millis[0]); - source.addNull(); - for (int i = 2; i < millis.length; ++i) - { - source.add(millis[i]); - } - columnWriter.write(source, millis.length); - columnWriter.flush(); - columnWriter.close(); - - Bitmap selected = new Bitmap(millis.length, true); - selected.clear(0); - selected.clear(3); - LongColumnVector target = new LongColumnVector(3); - TimeColumnReader reader = new TimeColumnReader(TypeDescription.createTime(3)); - reader.readSelected( - ByteBuffer.wrap(columnWriter.getColumnChunkContent()), - columnWriter.getColumnChunkEncoding().build(), - 0, - millis.length, - pixelStride, - 0, - target, - columnWriter.getColumnChunkIndex().build(), - selected); - reader.close(); - - assertTrue(target.isNull[0]); - assertEquals((long) millis[2] * PICOSECONDS_PER_MILLISECOND, target.vector[1]); - assertEquals((long) millis[4] * PICOSECONDS_PER_MILLISECOND, target.vector[2]); - } - @Test public void testLarge() throws IOException { diff --git a/pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongColumnVector.java b/pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongTimeColumnVector.java similarity index 57% rename from pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongColumnVector.java rename to pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongTimeColumnVector.java index 655d60cef6..6e8ccbad2d 100644 --- a/pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongColumnVector.java +++ b/pixels-core/src/test/java/io/pixelsdb/pixels/core/vector/TestLongTimeColumnVector.java @@ -22,19 +22,21 @@ import io.pixelsdb.pixels.core.TypeDescription; import org.junit.Test; -import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOS_PER_MILLIS; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; /** - * @author hank - * @create 2026-08-09 + * Tests for {@link LongTimeColumnVector} read layout and native TIME deserialization. + * + * @author gengdy + * @create 2026-08-17 */ -public class TestLongColumnVector +public class TestLongTimeColumnVector { @Test - public void testDeserializeTimeWithLongLayout() + public void testDeserializeTimeWithRequestedLayout() { int[] millis = {0, 1, 3_723_004, 86_399_999}; VectorizedRowBatch source = TypeDescription.createTime(3).createRowBatch(millis.length + 1); @@ -51,18 +53,27 @@ public void testDeserializeTimeWithLongLayout() assertTrue(defaultBatch.cols[0] instanceof TimeColumnVector); assertEquals(millis[2], ((TimeColumnVector) defaultBatch.cols[0]).times[2]); - VectorizedRowBatch longBatch = VectorizedRowBatch.deserialize( - serialized, TypeDescription.VectorLayout.TIME_AS_PICO_LONG); - assertTrue(longBatch.cols[0] instanceof LongColumnVector); - LongColumnVector longVector = (LongColumnVector) longBatch.cols[0]; - assertEquals(timeVector.getLength(), longVector.getLength()); - assertEquals(timeVector.getWriteIndex(), longVector.getWriteIndex()); - assertFalse(longVector.noNulls); + VectorizedRowBatch longTimeBatch = VectorizedRowBatch.deserialize( + serialized, TypeDescription.VectorLayout.TIME_AS_LONG_TIME); + assertTrue(longTimeBatch.cols[0] instanceof LongTimeColumnVector); + LongTimeColumnVector longTimeVector = (LongTimeColumnVector) longTimeBatch.cols[0]; + assertEquals(timeVector.getLength(), longTimeVector.getLength()); + assertEquals(timeVector.getWriteIndex(), longTimeVector.getWriteIndex()); + assertEquals(timeVector.noNulls, longTimeVector.noNulls); + assertEquals(timeVector.getPrecision(), longTimeVector.getPrecision()); for (int i = 0; i < millis.length; ++i) { - assertFalse(longVector.isNull[i]); - assertEquals(millis[i] * PICOSECONDS_PER_MILLISECOND, longVector.vector[i]); + assertFalse(longTimeVector.isNull[i]); + assertEquals((long) millis[i] * PICOS_PER_MILLIS, longTimeVector.vector[i]); } - assertTrue(longVector.isNull[millis.length]); + assertTrue(longTimeVector.isNull[millis.length]); + } + + @Test(expected = UnsupportedOperationException.class) + public void testLongTimeColumnVectorCannotBeSerialized() + { + TypeDescription.createTime(3) + .createRowBatch(1, TypeDescription.VectorLayout.TIME_AS_LONG_TIME) + .serialize(); } } diff --git a/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java b/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java index 1be2b910fb..4c6b615b43 100644 --- a/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java +++ b/pixels-executor/src/main/java/io/pixelsdb/pixels/executor/predicate/ColumnFilter.java @@ -38,7 +38,7 @@ import java.util.Set; import static com.google.common.base.Preconditions.checkArgument; -import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOS_PER_MILLIS; import static java.util.Objects.requireNonNull; /** @@ -379,9 +379,9 @@ public void doFilter(ColumnVector columnVector, int start, int length, Bitmap re doFilter(dacv.dates, dacv.noNulls ? null : dacv.isNull, start, length, result); return; case TIME: - if (columnVector instanceof LongColumnVector) + if (columnVector instanceof LongTimeColumnVector) { - LongColumnVector ltcv = (LongColumnVector) columnVector; + LongTimeColumnVector ltcv = (LongTimeColumnVector) columnVector; doFilterTime(ltcv.vector, ltcv.noNulls ? null : ltcv.isNull, start, length, result); } else @@ -686,11 +686,11 @@ private void doFilterTime(long[] vector, boolean[] isNull, int start, int length long lowerBound = range.lowerBound.type != Bound.Type.UNBOUNDED ? ((Integer) range.lowerBound.value + (range.lowerBound.type == Bound.Type.EXCLUDED ? 1L : 0L)) * - PICOSECONDS_PER_MILLISECOND : Long.MIN_VALUE; + PICOS_PER_MILLIS : Long.MIN_VALUE; long upperBound = range.upperBound.type != Bound.Type.UNBOUNDED ? ((Integer) range.upperBound.value - (range.upperBound.type == Bound.Type.EXCLUDED ? 1L : 0L)) * - PICOSECONDS_PER_MILLISECOND : Long.MAX_VALUE; + PICOS_PER_MILLIS : Long.MAX_VALUE; for (int i = start; i < start + length; ++i) { if (this.filter.allowNull && !noNulls && isNull[i] || @@ -706,12 +706,12 @@ private void doFilterTime(long[] vector, boolean[] isNull, int start, int length Set picoIncludes = new HashSet<>(includes.size()); for (T value : includes) { - picoIncludes.add((Integer) value * PICOSECONDS_PER_MILLISECOND); + picoIncludes.add((Integer) value * PICOS_PER_MILLIS); } Set picoExcludes = new HashSet<>(excludes.size()); for (T value : excludes) { - picoExcludes.add((Integer) value * PICOSECONDS_PER_MILLISECOND); + picoExcludes.add((Integer) value * PICOS_PER_MILLIS); } for (int i = start; i < start + length; ++i) { diff --git a/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java b/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java index 1b62a4a376..626f41fde9 100644 --- a/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java +++ b/pixels-executor/src/test/java/io/pixelsdb/pixels/executor/predicate/TestPredicate.java @@ -28,7 +28,7 @@ import io.pixelsdb.pixels.core.reader.PixelsRecordReader; import io.pixelsdb.pixels.core.utils.Bitmap; import io.pixelsdb.pixels.core.utils.Decimal; -import io.pixelsdb.pixels.core.vector.LongColumnVector; +import io.pixelsdb.pixels.core.vector.LongTimeColumnVector; import io.pixelsdb.pixels.core.vector.VectorizedRowBatch; import org.junit.Test; @@ -37,7 +37,7 @@ import java.util.SortedMap; import java.util.TreeMap; -import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOSECONDS_PER_MILLISECOND; +import static io.pixelsdb.pixels.core.utils.DatetimeUtils.PICOS_PER_MILLIS; import static org.junit.Assert.assertEquals; /** @@ -91,11 +91,11 @@ public void testLongTimeVectorFilterUsesMillisBounds() new Bound<>(Bound.Type.INCLUDED, 200)); ColumnFilter columnFilter = new ColumnFilter<>("time", TypeDescription.Category.TIME, timeFilter); - LongColumnVector vector = new LongColumnVector(5); + LongTimeColumnVector vector = new LongTimeColumnVector(5, 3); int[] millis = {99, 100, 150, 200, 201}; for (int i = 0; i < millis.length; ++i) { - vector.vector[i] = (long) millis[i] * PICOSECONDS_PER_MILLISECOND; + vector.vector[i] = (long) millis[i] * PICOS_PER_MILLIS; } Bitmap result = new Bitmap(millis.length, false);