[Issue #1368] split unified integer column into dedicated SHORT/INT/LONG types

flatbuffers/rowBatch.fbs

@@ -93,6 +93,11 @@ table IntColumnVectorFlat {
     vector : [int];
 }
 
+table ShortColumnVectorFlat {
+    base : ColumnVectorBaseFlat;
+    vector : [short];
+}
+
 table LongColumnVectorFlat {
     base : ColumnVectorBaseFlat;
     vector : [long];
@@ -138,7 +143,8 @@ union ColumnVectorFlat {
     LongDecimalColumnVectorFlat,
     TimeColumnVectorFlat,
     TimestampColumnVectorFlat,
-    VectorColumnVectorFlat
+    VectorColumnVectorFlat,
+    ShortColumnVectorFlat
 }
 
 table VectorizedRowBatchFlat {

pixels-amphi/src/main/java/io/pixelsdb/pixels/amphi/downloader/PeerDownloader.java

@@ -361,7 +361,13 @@ private static GenericRecord castToGenericRecord(Schema schema, List<Column> col
                     record.put(columns.get(i).getName(), bcv.vector[rowIdx]);
                     break;
                 case SHORT:
+                    ShortColumnVector scv = (ShortColumnVector) columnVectors.get(i);
+                    record.put(columns.get(i).getName(), (int) scv.vector[rowIdx]);
+                    break;
                 case INT:
+                    IntColumnVector icv = (IntColumnVector) columnVectors.get(i);
+                    record.put(columns.get(i).getName(), icv.vector[rowIdx]);
+                    break;
                 case LONG:
                     LongColumnVector lcv = (LongColumnVector) columnVectors.get(i);
                     record.put(columns.get(i).getName(), lcv.vector[rowIdx]);

pixels-cli/src/main/java/io/pixelsdb/pixels/cli/load/IndexedPixelsConsumer.java

@@ -308,7 +308,7 @@ public PerVirtualNodeWriter(PixelsWriter writer, File file, Path path, NodeProto
              this.prevRgId = this.rgId;
              this.rgRowOffset = 0;
              this.rowCounter = 0;
-             this.rowBatch = schema.createRowBatchWithHiddenColumn(pixelStride, TypeDescription.Mode.NONE);
+             this.rowBatch = schema.createRowBatchWithHiddenColumn(pixelStride);
              this.vNodeId = vNodeId;
              this.indexService = indexServices.computeIfAbsent(node.getAddress(), nodeInfo ->
                      RPCIndexService.CreateInstance(nodeInfo, indexServerPort));

pixels-cli/src/main/java/io/pixelsdb/pixels/cli/load/SimplePixelsConsumer.java

@@ -57,7 +57,7 @@ public SimplePixelsConsumer(BlockingQueue<String> queue, Parameters parameters,
                                  ConcurrentLinkedQueue<LoadedInfo> loadedInfos)
      {
          super(queue, parameters, loadedInfos);
-         this.rowBatch = schema.createRowBatchWithHiddenColumn(pixelStride, TypeDescription.Mode.NONE);
+         this.rowBatch = schema.createRowBatchWithHiddenColumn(pixelStride);
      }
 
      @Override

pixels-core/src/main/java/io/pixelsdb/pixels/core/TypeDescription.java

@@ -199,8 +199,8 @@ public enum Category
          */
         BOOLEAN(true, boolean.class, byte.class, "boolean"),
         BYTE(true, byte.class, byte.class, "tinyint", "byte"),
-        SHORT(true, short.class, long.class, "smallint", "short"),
-        INT(true, int.class, long.class, "integer", "int"),
+        SHORT(true, short.class, short.class, "smallint", "short"),
+        INT(true, int.class, int.class, "integer", "int"),
         LONG(true, long.class, long.class, "bigint", "long"),
         FLOAT(true, float.class, long.class, "float", "real"),
         DOUBLE(true, double.class, long.class, "double"),
@@ -1222,7 +1222,7 @@ public int getMaximumId()
         return maxId;
     }
 
-    private ColumnVector createColumn(int maxSize, int mode, boolean... useEncodedVector)
+    private ColumnVector createColumn(int maxSize, int vectorLayout, boolean... useEncodedVector)
     {
         requireNonNull(useEncodedVector, "columnsEncoded should not be null");
         // the length of useEncodedVector is already checked, not need to check again.
@@ -1232,15 +1232,17 @@ private ColumnVector createColumn(int maxSize, int mode, boolean... useEncodedVe
             case BYTE:
                 return new ByteColumnVector(maxSize);
             case SHORT:
-            case INT:
-                if (Mode.match(mode, Mode.CREATE_INT_VECTOR_FOR_INT))
+                if (VectorLayout.match(vectorLayout, VectorLayout.SHORT_AS_LONG))
                 {
-                    return new IntColumnVector(maxSize);
+                    return new LongColumnVector(maxSize);
                 }
-                else
+                return new ShortColumnVector(maxSize);
+            case INT:
+                if (VectorLayout.match(vectorLayout, VectorLayout.INT_AS_LONG))
                 {
                     return new LongColumnVector(maxSize);
                 }
+                return new IntColumnVector(maxSize);
             case LONG:
                 return new LongColumnVector(maxSize);
             case DATE:
@@ -1276,7 +1278,7 @@ private ColumnVector createColumn(int maxSize, int mode, boolean... useEncodedVe
                 ColumnVector[] fieldVector = new ColumnVector[children.size()];
                 for (int i = 0; i < fieldVector.length; ++i)
                 {
-                    fieldVector[i] = children.get(i).createColumn(maxSize, mode, useEncodedVector[i]);
+                    fieldVector[i] = children.get(i).createColumn(maxSize, vectorLayout, useEncodedVector[i]);
                 }
                 return new StructColumnVector(maxSize, fieldVector);
             }
@@ -1287,7 +1289,7 @@ private ColumnVector createColumn(int maxSize, int mode, boolean... useEncodedVe
         }
     }
 
-    public VectorizedRowBatch createRowBatch(int maxSize, int mode, boolean... useEncodedVector)
+    public VectorizedRowBatch createRowBatch(int maxSize, int vectorLayout, boolean... useEncodedVector)
     {
         VectorizedRowBatch result;
         if (category == Category.STRUCT)
@@ -1299,7 +1301,7 @@ public VectorizedRowBatch createRowBatch(int maxSize, int mode, boolean... useEn
             for (int i = 0; i < result.cols.length; ++i)
             {
                 String fieldName = fieldNames.get(i);
-                ColumnVector cv = children.get(i).createColumn(maxSize, mode,
+                ColumnVector cv = children.get(i).createColumn(maxSize, vectorLayout,
                         useEncodedVector.length != 0 && useEncodedVector[i]);
                 int originId = columnNames.indexOf(fieldName);
                 if (originId >= 0)
@@ -1319,24 +1321,24 @@ public VectorizedRowBatch createRowBatch(int maxSize, int mode, boolean... useEn
             checkArgument(useEncodedVector.length == 0 || useEncodedVector.length == 1,
                     "for null structure type, there can be only 0 or 1 elements in useEncodedVector");
             result = new VectorizedRowBatch(1, maxSize);
-            result.cols[0] = createColumn(maxSize, mode,
+            result.cols[0] = createColumn(maxSize, vectorLayout,
                     useEncodedVector.length == 1 && useEncodedVector[0]);
         }
         result.reset();
         return result;
     }
 
-    public VectorizedRowBatch createRowBatch()
+    public VectorizedRowBatch createRowBatch(int maxSize, boolean... useEncodedVector)
     {
-        return createRowBatch(VectorizedRowBatch.DEFAULT_SIZE, Mode.NONE);
+        return createRowBatch(maxSize, VectorLayout.NONE, useEncodedVector);
     }
 
-    public VectorizedRowBatch createRowBatch(int size)
+    public VectorizedRowBatch createRowBatch()
     {
-        return createRowBatch(size, Mode.NONE);
+        return createRowBatch(VectorizedRowBatch.DEFAULT_SIZE, VectorLayout.NONE);
     }
 
-    public VectorizedRowBatch createRowBatchWithHiddenColumn(int maxSize, int mode, boolean... useEncodedVector)
+    public VectorizedRowBatch createRowBatchWithHiddenColumn(int maxSize, int vectorLayout, boolean... useEncodedVector)
     {
         VectorizedRowBatch result;
         if (category == Category.STRUCT)
@@ -1349,7 +1351,7 @@ public VectorizedRowBatch createRowBatchWithHiddenColumn(int maxSize, int mode,
             for (int i = 0; i < result.cols.length - 1; ++i)
             {
                 String fieldName = fieldNames.get(i);
-                ColumnVector cv = children.get(i).createColumn(maxSize, mode,
+                ColumnVector cv = children.get(i).createColumn(maxSize, vectorLayout,
                         useEncodedVector.length != 0 && useEncodedVector[i]);
                 int originId = columnNames.indexOf(fieldName);
                 if (originId >= 0)
@@ -1371,17 +1373,51 @@ public VectorizedRowBatch createRowBatchWithHiddenColumn(int maxSize, int mode,
             checkArgument(useEncodedVector.length == 0 || useEncodedVector.length == 1,
                     "for null structure type, there can be only 0 or 1 elements in useEncodedVector");
             result = new VectorizedRowBatch(2, maxSize);
-            result.cols[0] = createColumn(maxSize, mode,
+            result.cols[0] = createColumn(maxSize, vectorLayout,
                     useEncodedVector.length == 1 && useEncodedVector[0]);
             result.cols[1] = new LongColumnVector(maxSize);
         }
         result.reset();
         return result;
     }
 
+    public VectorizedRowBatch createRowBatchWithHiddenColumn(int maxSize, boolean... useEncodedVector)
+    {
+        return createRowBatchWithHiddenColumn(maxSize, VectorLayout.NONE, useEncodedVector);
+    }
+
     public VectorizedRowBatch createRowBatchWithHiddenColumn()
     {
-        return createRowBatchWithHiddenColumn(VectorizedRowBatch.DEFAULT_SIZE, Mode.NONE);
+        return createRowBatchWithHiddenColumn(VectorizedRowBatch.DEFAULT_SIZE, VectorLayout.NONE);
+    }
+
+    /**
+     * The column vector layouts used when creating column vectors and row batches.
+     * These flags control whether short/int columns are widened into
+     * {@link io.pixelsdb.pixels.core.vector.LongColumnVector} for backward
+     * compatibility with old query engines (e.g. Trino 405, Presto 0.279),
+     * or created as their native dedicated vectors
+     * (e.g. {@link io.pixelsdb.pixels.core.vector.ShortColumnVector},
+     * {@link io.pixelsdb.pixels.core.vector.IntColumnVector}).
+     */
+    public static final class VectorLayout
+    {
+        public static final int NONE = 0;
+        /**
+         * Read SHORT columns as {@link io.pixelsdb.pixels.core.vector.LongColumnVector}
+         * instead of {@link io.pixelsdb.pixels.core.vector.ShortColumnVector}.
+         */
+        public static final int SHORT_AS_LONG = 0x01;
+        /**
+         * Read INT columns as {@link io.pixelsdb.pixels.core.vector.LongColumnVector}
+         * instead of {@link io.pixelsdb.pixels.core.vector.IntColumnVector}.
+         */
+        public static final int INT_AS_LONG = 0x02;
+
+        public static boolean match(int layout1, int layout2)
+        {
+            return (layout1 & layout2) != 0;
+        }
     }
 
     /**
@@ -1674,23 +1710,6 @@ public TypeDescription findSubtype(int goal)
         }
     }
 
-    /**
-     * The type related modes used when creating column vectors and row batches.
-     */
-    public static final class Mode
-    {
-        public static final int NONE = 0;
-        /**
-         * Create {@link IntColumnVector} for INT type.
-         */
-        public static final int CREATE_INT_VECTOR_FOR_INT = 0x01;
-
-        public static boolean match(int mode1, int mode2)
-        {
-            return (mode1 & mode2) != 0;
-        }
-    }
-
     /**
      * Serializes one cell from a {@link ColumnVector} at the given row index into
      * the canonical byte format, using the same encoding as {@link #convertSqlStringToByte}.
@@ -1708,19 +1727,26 @@ public byte[] convertColumnVectorToByte(ColumnVector col, int row)
             case BYTE:
                 return new byte[]{((ByteColumnVector) col).vector[row]};
             case SHORT:
+            {
+                short shortValue = col instanceof ShortColumnVector ?
+                        ((ShortColumnVector) col).vector[row] :
+                        (short) ((LongColumnVector) col).vector[row];
+                return ByteBuffer.allocate(Short.BYTES).putShort(shortValue).array();
+            }
             case INT:
             {
-                int value = col instanceof IntColumnVector ?
+                int intValue = col instanceof IntColumnVector ?
                         ((IntColumnVector) col).vector[row] :
                         (int) ((LongColumnVector) col).vector[row];
-                return ByteBuffer.allocate(Integer.BYTES).putInt(value).array();
+                return ByteBuffer.allocate(Integer.BYTES).putInt(intValue).array();
             }
             case LONG:
                 return ByteBuffer.allocate(Long.BYTES).putLong(((LongColumnVector) col).vector[row]).array();
             case DATE:
                 return ByteBuffer.allocate(Integer.BYTES).putInt(((DateColumnVector) col).dates[row]).array();
             case TIME:
-                return ByteBuffer.allocate(Integer.BYTES).putInt(((TimeColumnVector) col).times[row]).array();
+                return ByteBuffer.allocate(Integer.BYTES)
+                        .putInt(((TimeColumnVector) col).times[row]).array();
             case TIMESTAMP:
                 return ByteBuffer.allocate(Long.BYTES).putLong(((TimestampColumnVector) col).times[row]).array();
             case FLOAT:
@@ -1781,6 +1807,11 @@ public byte[] convertSqlStringToByte(String value)
                 return new byte[]{parsedByte};
             }
             case SHORT:
+            {
+                short shortValue = Short.parseShort(value);
+                bytes = ByteBuffer.allocate(Short.BYTES).putShort(shortValue).array();
+                break;
+            }
             case INT:
             {
                 int intValue = Integer.parseInt(value);

pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/Encoder.java

@@ -79,7 +79,12 @@ public byte[] encode(int[] values, int offset, int length) throws IOException
         throw new PixelsEncodingException("Encoding int values is not supported");
     }
 
-    public byte[] encode(short[] values)
+    public byte[] encode(short[] values) throws IOException
+    {
+        throw new PixelsEncodingException("Encoding short values is not supported");
+    }
+
+    public byte[] encode(short[] values, int offset, int length) throws IOException
     {
         throw new PixelsEncodingException("Encoding short values is not supported");
     }

pixels-core/src/main/java/io/pixelsdb/pixels/core/encoding/RunLenIntEncoder.java

@@ -113,6 +113,19 @@ public byte[] encode(int[] values, int offset, int length) throws IOException
         return result;
     }
 
+    @Override
+    public byte[] encode(short[] values, int offset, int length) throws IOException
+    {
+        for (int i = 0; i < length; i++)
+        {
+            this.write(values[i+offset]);
+        }
+        flush();
+        byte[] result = outputStream.toByteArray();
+        outputStream.reset();
+        return result;
+    }
+
     @Override
     public byte[] encode(long[] values) throws IOException
     {
@@ -125,6 +138,12 @@ public byte[] encode(int[] values) throws IOException
         return encode(values, 0, values.length);
     }
 
+    @Override
+    public byte[] encode(short[] values) throws IOException
+    {
+        return encode(values, 0, values.length);
+    }
+
     @Override
     public void close() throws IOException
     {

pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/ColumnReader.java

@@ -64,15 +64,23 @@ public static ColumnReader newColumnReader(TypeDescription type, PixelsReaderOpt
             case BYTE:
                 return new ByteColumnReader(type);
             case SHORT:
-            case INT:
-                if (option.isReadIntColumnAsIntVector())
+                if (option.isReadShortColumnAsLongVector())
                 {
-                    return new IntColumnReader(type);
+                    return new LongColumnReader(type);
                 }
                 else
+                {
+                    return new ShortColumnReader(type);
+                }
+            case INT:
+                if (option.isReadIntColumnAsLongVector())
                 {
                     return new LongColumnReader(type);
                 }
+                else
+                {
+                    return new IntColumnReader(type);
+                }
             case LONG:
                 return new LongColumnReader(type);
             case DOUBLE:

pixels-core/src/main/java/io/pixelsdb/pixels/core/reader/IntColumnReader.java

@@ -37,14 +37,19 @@
 
 /**
  * This is the column reader for integer (int32) columns.
- * In some query engines (e.g., Trino 405 and Presto 0.279), the integer column should be read into long[]
- * (i.e., {@link LongColumnVector}) in memory. In this case, the integer column should be read by
- * {@link LongColumnReader} instead of {@link IntColumnReader}.
+ * <p>
+ * In some query engines (e.g., Trino 405 and Presto 0.279), the integer column
+ * should be read into {@code long[]} (i.e., {@link LongColumnVector}) in memory. 
+ * In that case, set {@link PixelsReaderOption#readIntColumnAsLongVector(boolean)}
+ * to {@code true} so that {@link LongColumnReader} is used instead of this reader.
+ * <p>
+ * However, in some other query engines (e.g., Trino 466), the integer column 
+ * should be read into {@code int[]} (i.e., {@link IntColumnVector})
+ * by this reader, which is the default behavior.
  *
- * However, in some other query engines (e.g., Trino 466), the integer column should be read into int[]
- * (i.e., {@link IntColumnVector}) by {@link IntColumnReader}.
- * @author hank
+ * @author hank, gengdy
  * @create 2024-12-02
+ * @update 2026-08-07
  */
 public class IntColumnReader extends ColumnReader
 {
@@ -333,7 +338,7 @@ else if (selected.get(j - vectorIndex))
                 {
                     for (int j = i; j < i + numToRead; ++j)
                     {
-                        if (!(hasNull && isNull[j]))
+                        if (!(hasNull && isNull[j - vectorIndex]))
                         {
                             int value = inputBuffer.getInt();
                             if (selected.get(j - vectorIndex))