From 7278bf6a75bd8581fa0be7a14df4f1e898d4c40c Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Fri, 14 Aug 2026 11:21:16 +0800 Subject: [PATCH 1/2] fix: IndexKey timestamp by separating routing keys from Retina IndexKey binding --- .../pixels/sink/event/RowChangeEvent.java | 103 +++++++++++++----- .../sink/event/RowChangeEventFactory.java | 2 +- .../sink/writer/AbstractBucketedWriter.java | 8 +- .../pixels/sink/writer/NoneWriter.java | 7 +- .../writer/retina/RetinaPayloadBuilder.java | 3 + .../writer/retina/SinkContextManager.java | 1 + .../retina/TableSingleRecordWriter.java | 10 +- .../pixels/sink/cdc/CdcOrderContractTest.java | 8 +- .../sink/cdc/RecordingRetinaWriter.java | 1 - .../retina/RetinaPayloadBuilderTest.java | 32 +++++- .../retina/RetinaWriterIntegrationTest.java | 6 + 11 files changed, 133 insertions(+), 48 deletions(-) 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 cd053bc..235fcf4 100644 --- a/src/main/java/io/pixelsdb/pixels/sink/event/RowChangeEvent.java +++ b/src/main/java/io/pixelsdb/pixels/sink/event/RowChangeEvent.java @@ -54,12 +54,21 @@ public class RowChangeEvent private TableMetadata tableMetadata = null; private Map beforeValueMap; private Map afterValueMap; + /** + * Primary-key bytes used for bucket routing / PK-change detection. + * Independent of TransService timestamp. + */ + @Getter + private ByteString beforeRoutingKey; + @Getter + private ByteString afterRoutingKey; @Getter private IndexProto.IndexKey beforeKey; @Getter private IndexProto.IndexKey afterKey; - private boolean indexKeyInited = false; + private boolean routingKeyInited = false; + private boolean indexKeyBound = false; @Getter private long tableId; @@ -76,7 +85,7 @@ public RowChangeEvent(SinkProto.RowRecord rowRecord) throws SinkException this.tableMetadata = tableMetadataRegistry.getMetadata( rowRecord.getSource().getDb(), rowRecord.getSource().getTable()); init(); - initIndexKey(); + initRoutingKey(); } public RowChangeEvent(SinkProto.RowRecord rowRecord, TypeDescription schema) throws SinkException @@ -135,9 +144,12 @@ private void initColumnValueMap(SinkProto.RowValue rowValue, Map map.put(schema.getFieldNames().get(i), rowValue.getValuesList().get(i))); } - public void initIndexKey() throws SinkException + /** + * Builds primary-key bytes for routing. Safe before TransService timestamp is known. + */ + public void initRoutingKey() throws SinkException { - if (indexKeyInited) + if (routingKeyInited) { return; } @@ -149,49 +161,89 @@ public void initIndexKey() throws SinkException if (!this.tableMetadata.hasPrimaryIndex()) { + routingKeyInited = true; return; } if (hasBeforeData()) { - this.beforeKey = generateIndexKey(tableMetadata, beforeValueMap); + this.beforeRoutingKey = generateRoutingKey(tableMetadata, beforeValueMap); } if (hasAfterData()) { - this.afterKey = generateIndexKey(tableMetadata, afterValueMap); + this.afterRoutingKey = generateRoutingKey(tableMetadata, afterValueMap); } - indexKeyInited = true; + routingKeyInited = true; } - public void updateIndexKey() throws SinkException + /** + * Builds Retina {@link IndexProto.IndexKey} from routing bytes and {@link #timeStamp}. + * Call after the TransService timestamp is assigned. + */ + public void bindIndexKey() throws SinkException { + initRoutingKey(); + + if (this.tableMetadata == null || !this.tableMetadata.hasPrimaryIndex()) + { + return; + } + + SinglePointIndex index = tableMetadata.getIndex(); + long tableIdForKey = tableMetadata.getTable().getId(); if (hasBeforeData()) { - this.beforeKey = generateIndexKey(tableMetadata, beforeValueMap); + this.beforeKey = toIndexKey(beforeRoutingKey, index.getId(), tableIdForKey); } if (hasAfterData()) { - this.afterKey = generateIndexKey(tableMetadata, afterValueMap); + this.afterKey = toIndexKey(afterRoutingKey, index.getId(), tableIdForKey); } + + indexKeyBound = true; + } + + /** + * @deprecated Use {@link #initRoutingKey()} for bucketing and {@link #bindIndexKey()} after + * timestamp assignment. Kept for callers that only need routing. + */ + @Deprecated + public void initIndexKey() throws SinkException + { + initRoutingKey(); + } + + /** + * @deprecated Use {@link #bindIndexKey()}. + */ + @Deprecated + public void updateIndexKey() throws SinkException + { + bindIndexKey(); + } + + public boolean isIndexKeyBound() + { + return indexKeyBound; } public int getBeforeBucketFromIndex() { - assert indexKeyInited; + assert routingKeyInited; if (hasBeforeData()) { - return getBucketFromIndexKey(beforeKey); + return getBucketIdFromByteBuffer(beforeRoutingKey); } throw new IllegalCallerException("Event dosen't have before data"); } public boolean isPkChanged() throws SinkException { - if (!indexKeyInited) + if (!routingKeyInited) { - initIndexKey(); + initRoutingKey(); } if (getOp() != SinkProto.OperationType.UPDATE) @@ -199,26 +251,23 @@ public boolean isPkChanged() throws SinkException return false; } - ByteString beforeKey = getBeforeKey().getKey(); - ByteString afterKey = getAfterKey().getKey(); - - return !beforeKey.equals(afterKey); + return !beforeRoutingKey.equals(afterRoutingKey); } public int getAfterBucketFromIndex() { - assert indexKeyInited; + assert routingKeyInited; if (hasAfterData()) { - return getBucketFromIndexKey(afterKey); + return getBucketIdFromByteBuffer(afterRoutingKey); } throw new IllegalCallerException("Event dosen't have after data"); } - private IndexProto.IndexKey generateIndexKey(TableMetadata tableMetadata, Map rowValue) + private ByteString generateRoutingKey( + TableMetadata tableMetadata, Map rowValue) { List keyColumnNames = tableMetadata.getKeyColumnNames(); - SinglePointIndex index = tableMetadata.getIndex(); int len = keyColumnNames.size(); List keyColumnValues = new ArrayList<>(len); int keySize = 0; @@ -234,12 +283,16 @@ private IndexProto.IndexKey generateIndexKey(TableMetadata tableMetadata, Map batch) try { List tableUpdateData = new LinkedList<>(); - for (RowChangeEvent event : batch) - { - event.setTimeStamp(pixelsTransContext.getTimestamp()); - event.updateIndexKey(); - } - + // Timestamp + IndexKey binding happen inside RetinaPayloadBuilder. RetinaProto.TableUpdateData update = buildTableUpdateDataFromBatch(pixelsTransContext, batch); if (update != null) @@ -97,9 +92,6 @@ public void flush(List batch) } } ); - } catch (SinkException e) - { - throw new RuntimeException(e); } finally { writeLock.unlock(); diff --git a/src/test/java/io/pixelsdb/pixels/sink/cdc/CdcOrderContractTest.java b/src/test/java/io/pixelsdb/pixels/sink/cdc/CdcOrderContractTest.java index f9e28bd..4e0e85c 100644 --- a/src/test/java/io/pixelsdb/pixels/sink/cdc/CdcOrderContractTest.java +++ b/src/test/java/io/pixelsdb/pixels/sink/cdc/CdcOrderContractTest.java @@ -157,7 +157,7 @@ private RowChangeEvent event( metadataFixture.metadata().getTypeDescription(), metadataFixture.metadata()); event.setTimeStamp(7001L); - event.initIndexKey(); + event.initRoutingKey(); return event; } @@ -202,9 +202,9 @@ List primaryKeys() { return emitted.stream() .map(event -> event.hasAfterData() - ? event.getAfterKey() - : event.getBeforeKey()) - .map(key -> readLong(key.getKey())) + ? event.getAfterRoutingKey() + : event.getBeforeRoutingKey()) + .map(CdcOrderContractTest::readLong) .toList(); } } diff --git a/src/test/java/io/pixelsdb/pixels/sink/cdc/RecordingRetinaWriter.java b/src/test/java/io/pixelsdb/pixels/sink/cdc/RecordingRetinaWriter.java index 68ab554..325f080 100644 --- a/src/test/java/io/pixelsdb/pixels/sink/cdc/RecordingRetinaWriter.java +++ b/src/test/java/io/pixelsdb/pixels/sink/cdc/RecordingRetinaWriter.java @@ -82,7 +82,6 @@ public boolean writeRow(RowChangeEvent row) int ordinal = recordedRequests.size() + 1; long timestamp = TIMESTAMP_BASE + ordinal; row.setTimeStamp(timestamp); - row.updateIndexKey(); RetinaProto.TableUpdateData tableUpdate = RetinaPayloadBuilder.buildTableUpdateData( 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 3802ff5..d13a25b 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 @@ -141,6 +141,15 @@ private static RowChangeEvent event( SinkProto.RowValue before, SinkProto.RowValue after, TableMetadata metadata) throws Exception + { + return eventWithRoutingOnly(operation, before, after, metadata); + } + + private static RowChangeEvent eventWithRoutingOnly( + SinkProto.OperationType operation, + SinkProto.RowValue before, + SinkProto.RowValue after, + TableMetadata metadata) throws Exception { SinkProto.RowRecord.Builder record = SinkProto.RowRecord.newBuilder() .setOp(operation) @@ -158,8 +167,7 @@ private static RowChangeEvent event( RowChangeEvent event = new RowChangeEvent( record.build(), metadata.getTypeDescription(), metadata); - event.setTimeStamp(TIMESTAMP); - event.initIndexKey(); + event.initRoutingKey(); return event; } @@ -202,4 +210,24 @@ private static ByteString bytes(String value) { return ByteString.copyFromUtf8(value); } + + @Test + void shouldBindIndexKeyWithPayloadTimestamp() throws Exception + { + TableMetadata metadata = tableMetadata(); + RowChangeEvent insert = eventWithRoutingOnly( + SinkProto.OperationType.INSERT, null, row("1", "insert"), metadata); + + assertEquals(bytes("1"), insert.getAfterRoutingKey()); + assertEquals(null, insert.getAfterKey()); + + RetinaProto.TableUpdateData tableUpdate = + RetinaPayloadBuilder.buildTableUpdateData( + TABLE_NAME, TIMESTAMP, List.of(insert)); + + assertEquals(TIMESTAMP, tableUpdate.getTimestamp()); + assertEquals(TIMESTAMP, tableUpdate.getInsertData(0).getIndexKeys(0).getTimestamp()); + assertEquals(bytes("1"), tableUpdate.getInsertData(0).getIndexKeys(0).getKey()); + assertEquals(TIMESTAMP, insert.getAfterKey().getTimestamp()); + } } diff --git a/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaWriterIntegrationTest.java b/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaWriterIntegrationTest.java index 32e4bb9..ab5c3ae 100644 --- a/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaWriterIntegrationTest.java +++ b/src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaWriterIntegrationTest.java @@ -118,6 +118,7 @@ public void insertSingleRecord() throws RetinaException, SinkException, TransExc ); RowChangeEvent rowChangeEvent = new RowChangeEvent(builder.build()); rowChangeEvent.setTimeStamp(timeStamp); + rowChangeEvent.bindIndexKey(); IndexProto.IndexKey indexKey = rowChangeEvent.getAfterKey(); RetinaProto.InsertData.Builder insertDataBuilder = RetinaProto.InsertData.newBuilder(); insertDataBuilder. @@ -184,6 +185,7 @@ public void updateSingleRecord() throws RetinaException, SinkException, TransExc RowChangeEvent rowChangeEvent = new RowChangeEvent(builder.build()); rowChangeEvent.setTimeStamp(timeStamp); + rowChangeEvent.bindIndexKey(); RetinaProto.DeleteData.Builder deleteDataBuilder = RetinaProto.DeleteData.newBuilder(); deleteDataBuilder .addIndexKeys(rowChangeEvent.getBeforeKey()); @@ -274,6 +276,7 @@ public void testCheckingAccountInsertPerformance() throws RowChangeEvent rowChangeEvent = new RowChangeEvent(rowBuilder.build()); rowChangeEvent.setTimeStamp(timeStamp); + rowChangeEvent.bindIndexKey(); // InsertData RetinaProto.InsertData.Builder insertDataBuilder = RetinaProto.InsertData.newBuilder() @@ -403,6 +406,7 @@ public void testCheckingAccountUpdatePerformance() throws RowChangeEvent rowChangeEvent = new RowChangeEvent(rowBuilder.build()); rowChangeEvent.setTimeStamp(timeStamp); + rowChangeEvent.bindIndexKey(); // deleteData RetinaProto.DeleteData.Builder deleteDataBuilder = RetinaProto.DeleteData.newBuilder() @@ -536,8 +540,10 @@ public void testInsertTwoTablePerformance() throws ); RowChangeEvent rowChangeEvent = new RowChangeEvent(rowBuilder.build()); rowChangeEvent.setTimeStamp(timeStamp); + rowChangeEvent.bindIndexKey(); RowChangeEvent rowChangeEvent2 = new RowChangeEvent(rowBuilder.build()); rowChangeEvent2.setTimeStamp(timeStamp); + rowChangeEvent2.bindIndexKey(); // InsertData RetinaProto.InsertData.Builder insertDataBuilder = RetinaProto.InsertData.newBuilder() .addColValues(ByteString.copyFrom(cols[0])) From 0b680489c891f3d98654fa9d5d9122e64d33f55b Mon Sep 17 00:00:00 2001 From: Dongyang Geng Date: Mon, 17 Aug 2026 13:14:04 +0800 Subject: [PATCH 2/2] fix: vectorlayout --- src/test/java/io/pixelsdb/pixels/sink/event/RowBatchTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/test/java/io/pixelsdb/pixels/sink/event/RowBatchTest.java b/src/test/java/io/pixelsdb/pixels/sink/event/RowBatchTest.java index d86cc55..3deb665 100644 --- a/src/test/java/io/pixelsdb/pixels/sink/event/RowBatchTest.java +++ b/src/test/java/io/pixelsdb/pixels/sink/event/RowBatchTest.java @@ -31,7 +31,7 @@ void integerRowBatchTest() TypeDescription schema = TypeDescription.createSchemaFromStrings( java.util.List.of("id"), java.util.List.of("int")); VectorizedRowBatch rowBatch = schema.createRowBatch( - 3, TypeDescription.Mode.CREATE_INT_VECTOR_FOR_INT); + 3, TypeDescription.VectorLayout.NONE); IntColumnVector vector = (IntColumnVector) rowBatch.cols[0]; vector.add(10); vector.add(20); @@ -52,7 +52,7 @@ void varcharRowBatchTest() { TypeDescription schema = TypeDescription.createSchemaFromStrings( java.util.List.of("name"), java.util.List.of("varchar(100)")); - VectorizedRowBatch rowBatch = schema.createRowBatch(3, TypeDescription.Mode.CREATE_INT_VECTOR_FOR_INT); + VectorizedRowBatch rowBatch = schema.createRowBatch(3, TypeDescription.VectorLayout.NONE); BinaryColumnVector v = (BinaryColumnVector) rowBatch.cols[0]; v.add("rr"); v.add("zz");