From c131d309381c1e90c9cf90679f8317f2ebb96180 Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Tue, 11 Aug 2026 14:52:42 +0800 Subject: [PATCH] fix: Propagate SQL NULL via Retina isNull instead of empty bytes --- .../support/DebeziumRowValueConverter.java | 11 ++++-- .../pixels/sink/event/RowChangeEvent.java | 14 +++++++ .../writer/retina/RetinaPayloadBuilder.java | 6 ++- .../DebeziumRowValueConverterTest.java | 4 ++ .../retina/RetinaPayloadBuilderTest.java | 38 +++++++++++++++++++ 5 files changed, 67 insertions(+), 6 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..6b382b2 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 @@ -128,7 +128,9 @@ private SinkProto.ColumnValue.Builder parseConnectValue( } if (raw == null) { - return SinkProto.ColumnValue.newBuilder().setValue(ByteString.EMPTY); + return SinkProto.ColumnValue.newBuilder() + .setValue(ByteString.EMPTY) + .setIsNull(true); } if (schemaName != null) @@ -631,8 +633,8 @@ private SinkProto.ColumnValue.Builder parseValue(JsonNode valueNode, String fiel if (valueNode == null || valueNode.isNull()) { return SinkProto.ColumnValue.newBuilder() - // .setName(fieldName) - .setValue(ByteString.EMPTY); + .setValue(ByteString.EMPTY) + .setIsNull(true); } SinkProto.ColumnValue.Builder columnValueBuilder = SinkProto.ColumnValue.newBuilder(); @@ -743,7 +745,8 @@ private SinkProto.ColumnValue.Builder parseCanonicalValue( if (raw == null) { return SinkProto.ColumnValue.newBuilder() - .setValue(ByteString.EMPTY); + .setValue(ByteString.EMPTY) + .setIsNull(true); } if (raw instanceof JsonNode jsonNode) { diff --git a/src/main/java/io/pixelsdb/pixels/sink/event/RowChangeEvent.java b/src/main/java/io/pixelsdb/pixels/sink/event/RowChangeEvent.java index 19336e7..cd053bc 100644 --- a/src/main/java/io/pixelsdb/pixels/sink/event/RowChangeEvent.java +++ b/src/main/java/io/pixelsdb/pixels/sink/event/RowChangeEvent.java @@ -336,6 +336,20 @@ public List getAfterData() return colValueList; } + /** + * Parallel to {@link #getAfterData()}: {@code true} means SQL NULL at that column. + */ + public List getAfterIsNull() + { + List colValues = rowRecord.getAfter().getValuesList(); + List isNullList = new ArrayList<>(colValues.size()); + for (SinkProto.ColumnValue col : colValues) + { + isNullList.add(col.getIsNull()); + } + return isNullList; + } + @Override public String toString() { diff --git a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilder.java b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilder.java index bda2633..0ce2212 100644 --- a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilder.java +++ b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilder.java @@ -78,14 +78,16 @@ private static void addRowChange( { RetinaProto.InsertData.Builder insertData = RetinaProto.InsertData.newBuilder() .addIndexKeys(event.getAfterKey()) - .addAllColValues(event.getAfterData()); + .addAllColValues(event.getAfterData()) + .addAllIsNull(event.getAfterIsNull()); builder.addInsertData(insertData); } case UPDATE -> { RetinaProto.UpdateData.Builder updateData = RetinaProto.UpdateData.newBuilder() .addIndexKeys(event.getAfterKey()) - .addAllColValues(event.getAfterData()); + .addAllColValues(event.getAfterData()) + .addAllIsNull(event.getAfterIsNull()); builder.addUpdateData(updateData); } case DELETE -> 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..4399adf 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 @@ -34,6 +34,7 @@ import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -69,7 +70,9 @@ void shouldEncodeCanonicalStructJsonAndAvroValues() throws Exception assertEquals("TDSQL value ", value.getValues(1).getValue().toStringUtf8()); assertArrayEquals(longBytes(2471035L), bytes(value, 2)); assertEquals(0, value.getValues(3).getValue().size()); + assertTrue(value.getValues(3).getIsNull()); assertEquals("", value.getValues(4).getValue().toStringUtf8()); + assertFalse(value.getValues(4).getIsNull()); ObjectNode jsonRow = new ObjectMapper().createObjectNode(); jsonRow.put("id", 9223372036854775806L); @@ -195,6 +198,7 @@ void shouldEncodeConnectPrimitiveValuesAsExactBytes() throws Exception assertArrayEquals( new byte[]{0x10, 0x20, (byte) 0xfe, 0x00}, bytes(value, 11)); assertArrayEquals(new byte[0], bytes(value, 12)); + assertTrue(value.getValues(12).getIsNull()); } @Test diff --git a/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilderTest.java b/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilderTest.java index aa2bf58..3802ff5 100644 --- a/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilderTest.java +++ b/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilderTest.java @@ -69,12 +69,15 @@ void shouldBuildCompleteUpdateRecordRequest() throws Exception assertEquals(2, tableUpdate.getInsertDataCount()); assertRow(tableUpdate.getInsertData(0).getIndexKeys(0), tableUpdate.getInsertData(0).getColValuesList(), "1", "insert"); + assertEquals(List.of(false, false), tableUpdate.getInsertData(0).getIsNullList()); assertRow(tableUpdate.getInsertData(1).getIndexKeys(0), tableUpdate.getInsertData(1).getColValuesList(), "2", "snapshot"); + assertEquals(List.of(false, false), tableUpdate.getInsertData(1).getIsNullList()); assertEquals(1, tableUpdate.getUpdateDataCount()); assertRow(tableUpdate.getUpdateData(0).getIndexKeys(0), tableUpdate.getUpdateData(0).getColValuesList(), "3", "after"); + assertEquals(List.of(false, false), tableUpdate.getUpdateData(0).getIsNullList()); assertEquals(1, tableUpdate.getDeleteDataCount()); assertEquals(bytes("4"), tableUpdate.getDeleteData(0).getIndexKeys(0).getKey()); @@ -85,6 +88,41 @@ void shouldBuildCompleteUpdateRecordRequest() throws Exception assertEquals(request, RetinaProto.UpdateRecordRequest.parseFrom(request.toByteArray())); } + @Test + void shouldPropagateIsNullAlongsideColValues() throws Exception + { + TableMetadata metadata = tableMetadata(); + RowChangeEvent withNull = event( + SinkProto.OperationType.INSERT, + null, + SinkProto.RowValue.newBuilder() + .addValues(SinkProto.ColumnValue.newBuilder().setValue(bytes("5"))) + .addValues(SinkProto.ColumnValue.newBuilder() + .setValue(ByteString.EMPTY) + .setIsNull(true)) + .build(), + metadata); + RowChangeEvent emptyString = event( + SinkProto.OperationType.UPDATE, + row("6", "before"), + SinkProto.RowValue.newBuilder() + .addValues(SinkProto.ColumnValue.newBuilder().setValue(bytes("6"))) + .addValues(SinkProto.ColumnValue.newBuilder().setValue(ByteString.EMPTY)) + .build(), + metadata); + + RetinaProto.TableUpdateData tableUpdate = + RetinaPayloadBuilder.buildTableUpdateData( + TABLE_NAME, TIMESTAMP, List.of(withNull, emptyString)); + + assertEquals(List.of(false, true), tableUpdate.getInsertData(0).getIsNullList()); + assertEquals(bytes("5"), tableUpdate.getInsertData(0).getColValues(0)); + assertEquals(ByteString.EMPTY, tableUpdate.getInsertData(0).getColValues(1)); + + assertEquals(List.of(false, false), tableUpdate.getUpdateData(0).getIsNullList()); + assertEquals(ByteString.EMPTY, tableUpdate.getUpdateData(0).getColValues(1)); + } + private static void assertRow( io.pixelsdb.pixels.index.IndexProto.IndexKey indexKey, List columnValues,