[Issue #1372] read TIME as picosecond LongColumnVector for zero-copy Trino

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_LONG_TIME))
+                {
+                    return new LongTimeColumnVector(maxSize, precision);
+                }
                 return new TimeColumnVector(maxSize, precision);
             case TIMESTAMP:
                 return new TimestampColumnVector(maxSize, precision);
@@ -1414,6 +1418,11 @@ 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.LongTimeColumnVector} for TIME type,
+         * storing picoseconds of day for Trino-native zero-copy.
+         */
+        public static final int TIME_AS_LONG_TIME = 0x04;
 
         public static boolean match(int layout1, int layout2)
         {

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:

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
+ * <https://www.gnu.org/licenses/>.
+ */
+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.
+ * <p>
+ * 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;
+        }
+    }
+}

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 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
@@ -178,6 +179,17 @@ public boolean isReadShortColumnAsLongVector()
         return readShortColumnAsLongVector;
     }
 
+    public PixelsReaderOption readTimeColumnAsLongTimeVector(boolean readTimeColumnAsLongTimeVector)
+    {
+        this.readTimeColumnAsLongTimeVector = readTimeColumnAsLongTimeVector;
+        return this;
+    }
+
+    public boolean isReadTimeColumnAsLongTimeVector()
+    {
+        return readTimeColumnAsLongTimeVector;
+    }
+
     public PixelsReaderOption exposeHiddenColumn(boolean exposeHiddenColumn)
     {
         this.exposeHiddenColumn = exposeHiddenColumn;

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.isReadTimeColumnAsLongTimeVector() ? TypeDescription.VectorLayout.TIME_AS_LONG_TIME : 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);