[PIXELS-1368] encode SHORT as 2-byte values to match Pixels ShortColumnVector

src/main/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverter.java

@@ -64,6 +64,7 @@ private static void buildInt32(int value, int capacity, SinkProto.ColumnValue.Bu
         switch (capacity)
         {
             case Byte.BYTES -> buffer.put((byte) value);
+            case Short.BYTES -> buffer.putShort((short) value);
             case Integer.BYTES -> buffer.putInt(value);
             default -> throw new IllegalArgumentException("Unsupported integer width: " + capacity);
         }
@@ -347,8 +348,7 @@ private SinkProto.ColumnValue.Builder encodeSignedInteger(
                 min = BigInteger.valueOf(Short.MIN_VALUE);
                 max = BigInteger.valueOf(Short.MAX_VALUE);
                 valueBits = Short.SIZE;
-                // Pixels encodes SHORT with the same width as INT.
-                wireWidth = Integer.BYTES;
+                wireWidth = Short.BYTES;
                 break;
             case INT:
                 min = BigInteger.valueOf(Integer.MIN_VALUE);
@@ -646,7 +646,7 @@ private SinkProto.ColumnValue.Builder parseValue(JsonNode valueNode, String fiel
             }
             case SHORT:
             {
-                buildInt32(valueNode.asInt(), Integer.BYTES, columnValueBuilder);
+                buildInt32(valueNode.asInt(), Short.BYTES, columnValueBuilder);
                 break;
             }
             case INT:
@@ -757,7 +757,7 @@ private SinkProto.ColumnValue.Builder parseCanonicalValue(
                 buildInt32(((Number) raw).intValue(), Byte.BYTES, builder);
                 break;
             case SHORT:
-                buildInt32(((Number) raw).intValue(), Integer.BYTES, builder);
+                buildInt32(((Number) raw).intValue(), Short.BYTES, builder);
                 break;
             case INT:
                 buildInt32(((Number) raw).intValue(), Integer.BYTES, builder);

src/test/java/io/pixelsdb/pixels/sink/cdc/AbstractCdcValidationTest.java

@@ -557,8 +557,8 @@ protected static ByteString byteValue(int value)
 
     protected static ByteString shortValue(int value)
     {
-        // Pixels encodes SHORT with the same width as INT.
-        return intValue((short) value);
+        return ByteString.copyFrom(
+                ByteBuffer.allocate(Short.BYTES).putShort((short) value).array());
     }
 
     protected static ByteString intValue(int value)

src/test/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverterTest.java

@@ -127,7 +127,7 @@ void shouldEncodeIntegerWidthsExpectedByPixelsColumnVectors() throws Exception
         new DebeziumRowValueConverter(typeDescription).parse(row, builder);
 
         assertEquals(1, builder.getValues(0).getValue().size());
-        assertEquals(4, builder.getValues(1).getValue().size());
+        assertEquals(2, builder.getValues(1).getValue().size());
         assertEquals(4, builder.getValues(2).getValue().size());
     }
 
@@ -184,7 +184,7 @@ void shouldEncodeConnectPrimitiveValuesAsExactBytes() throws Exception
         assertArrayEquals(new byte[]{1}, bytes(value, 0));
         assertArrayEquals(new byte[]{0}, bytes(value, 1));
         assertArrayEquals(new byte[]{(byte) 0x80}, bytes(value, 2));
-        assertArrayEquals(intBytes((short) 0x8123), bytes(value, 3));
+        assertArrayEquals(shortBytes((short) 0x8123), bytes(value, 3));
         assertArrayEquals(intBytes(0x81234567), bytes(value, 4));
         assertArrayEquals(longBytes(0x8123456789ABCDEFL), bytes(value, 5));
         assertArrayEquals(intBytes(Float.floatToIntBits(-12.5f)), bytes(value, 6));
@@ -424,6 +424,11 @@ private static Schema decimalSchema(int scale, int precision)
                 .build();
     }
 
+    private static byte[] shortBytes(short value)
+    {
+        return ByteBuffer.allocate(Short.BYTES).putShort(value).array();
+    }
+
     private static byte[] intBytes(int value)
     {
         return ByteBuffer.allocate(Integer.BYTES).putInt(value).array();