From 943def671853c69e016b41d22cdf6d18e1d33469 Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Thu, 13 Aug 2026 10:58:06 +0800 Subject: [PATCH] fix: encode SHORT as 2-byte values to match Pixels ShortColumnVector --- .../debezium/support/DebeziumRowValueConverter.java | 8 ++++---- .../pixels/sink/cdc/AbstractCdcValidationTest.java | 4 ++-- .../debezium/support/DebeziumRowValueConverterTest.java | 9 +++++++-- 3 files changed, 13 insertions(+), 8 deletions(-) diff --git a/src/main/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverter.java b/src/main/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverter.java index ce5f31d..578e247 100644 --- a/src/main/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverter.java +++ b/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); diff --git a/src/test/java/io/pixelsdb/pixels/sink/cdc/AbstractCdcValidationTest.java b/src/test/java/io/pixelsdb/pixels/sink/cdc/AbstractCdcValidationTest.java index db0f057..571f3d4 100644 --- a/src/test/java/io/pixelsdb/pixels/sink/cdc/AbstractCdcValidationTest.java +++ b/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) diff --git a/src/test/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverterTest.java b/src/test/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverterTest.java index 2d4dcc8..51291a4 100644 --- a/src/test/java/io/pixelsdb/pixels/sink/conversion/debezium/support/DebeziumRowValueConverterTest.java +++ b/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();