Skip to content

[PIXELS-1398] preserve Retina operation order #20

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
bianhq merged 1 commit into from
Sep 18, 2026
Merged

[PIXELS-1398] preserve Retina operation order #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions src/main/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import io.pixelsdb.pixels.sink.event.RowChangeEvent;
import io.pixelsdb.pixels.sink.exception.SinkException;

import java.util.ArrayList;
import java.util.List;

/**
Expand All @@ -30,6 +31,41 @@ private RetinaPayloadBuilder()
{
}

/**
* Builds table updates while preserving the source operation order.
*
* <p>Retina processes the operation fields inside one {@code TableUpdateData}
* in its own fixed order. Therefore, only adjacent events with the same
* operation may share a payload. The order of the returned payloads matches
* the order of {@code events}.</p>
*/
public static List<RetinaProto.TableUpdateData> buildTableUpdateDataList(
String tableName,
long timestamp,
List<RowChangeEvent> events) throws SinkException
{
if (events == null || events.isEmpty())
{
throw new IllegalArgumentException("events is empty");
}

List<RetinaProto.TableUpdateData> updates = new ArrayList<>();
int groupStart = 0;
for (int i = 1; i <= events.size(); i++)
{
if (i == events.size() || events.get(i).getOp() != events.get(groupStart).getOp())
{
updates.add(buildTableUpdateData(
tableName, timestamp, events.subList(groupStart, i)));
groupStart = i;
}
}
return updates;
}

/**
* Builds one table update from a non-empty, homogeneous operation group.
*/
public static RetinaProto.TableUpdateData buildTableUpdateData(
String tableName,
long timestamp,
Expand All @@ -41,6 +77,14 @@ public static RetinaProto.TableUpdateData buildTableUpdateData(
}

RowChangeEvent firstEvent = events.get(0);
for (RowChangeEvent event : events)
{
if (event.getOp() != firstEvent.getOp())
{
throw new IllegalArgumentException(
"TableUpdateData cannot contain different operation types");
}
}
RetinaProto.TableUpdateData.Builder builder = RetinaProto.TableUpdateData.newBuilder()
.setTimestamp(timestamp)
.setPrimaryIndexId(firstEvent.getTableMetadata().getPrimaryIndexKeyId())
Expand Down
47 changes: 27 additions & 20 deletions src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableCrossTxWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,17 +72,9 @@ public void flush(List<RowChangeEvent> batch)
{
if (smallBatch != null && !smallBatch.isEmpty())
{
RetinaProto.TableUpdateData update =
buildTableUpdateDataFromBatch(txId, smallBatch);
if (update == null)
{
continue;
}
tableUpdateData.add(update);
tableUpdateCount.add(smallBatch.size());
addTransactionBatch(txId, smallBatch, tableUpdateData,
txIds, fullTableName, tableUpdateCount);
}
txIds.add(currTxId);
fullTableName.add(event.getFullTableName());
txId = currTxId;
smallBatch = new LinkedList<>();
}
Expand All @@ -91,13 +83,8 @@ public void flush(List<RowChangeEvent> batch)

if (smallBatch != null)
{
RetinaProto.TableUpdateData update =
buildTableUpdateDataFromBatch(txId, smallBatch);
if (update != null)
{
tableUpdateData.add(update);
tableUpdateCount.add(smallBatch.size());
}
addTransactionBatch(txId, smallBatch, tableUpdateData,
txIds, fullTableName, tableUpdateCount);
}

// flushRateLimiter.acquire(batch.size());
Expand Down Expand Up @@ -149,6 +136,26 @@ public void flush(List<RowChangeEvent> batch)
}
}

private void addTransactionBatch(
String txId,
List<RowChangeEvent> transactionBatch,
List<RetinaProto.TableUpdateData> tableUpdateData,
List<String> txIds,
List<String> fullTableNames,
List<Integer> tableUpdateCount)
{
List<RetinaProto.TableUpdateData> updates =
buildTableUpdateDataFromBatch(txId, transactionBatch);
if (updates.isEmpty())
{
return;
}
tableUpdateData.addAll(updates);
txIds.add(txId);
fullTableNames.add(transactionBatch.get(0).getFullTableName());
tableUpdateCount.add(transactionBatch.size());
}

private void failCtxs(List<String> txIds)
{
for (String writeTxId : txIds)
Expand Down Expand Up @@ -187,13 +194,13 @@ private void updateCtxCounters(List<String> txIds, List<String> fullTableName, L
writeLock.unlock();
}

protected RetinaProto.TableUpdateData buildTableUpdateDataFromBatch(
protected List<RetinaProto.TableUpdateData> buildTableUpdateDataFromBatch(
String txId, List<RowChangeEvent> smallBatch)
{
SinkContext sinkContext = SinkContextManager.getInstance().getSinkContext(txId);
if (sinkContext == null)
{
return null;
return List.of();
}
try
{
Expand All @@ -212,7 +219,7 @@ protected RetinaProto.TableUpdateData buildTableUpdateDataFromBatch(
}
try
{
return RetinaPayloadBuilder.buildTableUpdateData(
return RetinaPayloadBuilder.buildTableUpdateDataList(
tableName, sinkContext.getTimestamp(), smallBatch);
} catch (SinkException e)
{
Expand Down
11 changes: 4 additions & 7 deletions src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleRecordWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,12 +52,9 @@ public void flush(List<RowChangeEvent> batch)
{
List<RetinaProto.TableUpdateData> tableUpdateData = new LinkedList<>();
// Timestamp + IndexKey binding happen inside RetinaPayloadBuilder.
RetinaProto.TableUpdateData update =
List<RetinaProto.TableUpdateData> updates =
buildTableUpdateDataFromBatch(pixelsTransContext, batch);
if (update != null)
{
tableUpdateData.add(update);
}
tableUpdateData.addAll(updates);

// flushRateLimiter.acquire(batch.size());
long txStartTime = System.currentTimeMillis();
Expand Down Expand Up @@ -98,12 +95,12 @@ public void flush(List<RowChangeEvent> batch)
}
}

protected RetinaProto.TableUpdateData buildTableUpdateDataFromBatch(
protected List<RetinaProto.TableUpdateData> buildTableUpdateDataFromBatch(
TransContext transContext, List<RowChangeEvent> smallBatch)
{
try
{
return RetinaPayloadBuilder.buildTableUpdateData(
return RetinaPayloadBuilder.buildTableUpdateDataList(
tableName, transContext.getTimestamp(), smallBatch);
} catch (SinkException e)
{
Expand Down
4 changes: 2 additions & 2 deletions src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleTxWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -82,8 +82,8 @@ public void flush(List<RowChangeEvent> batchToFlush)
RowChangeEvent event1 = batch.get(0);
try
{
List<RetinaProto.TableUpdateData> tableUpdateData = List.of(
RetinaPayloadBuilder.buildTableUpdateData(tableName, 0L, batch));
List<RetinaProto.TableUpdateData> tableUpdateData =
RetinaPayloadBuilder.buildTableUpdateDataList(tableName, 0L, batch);
delegate.writeTrans(event1.getSchemaName(), tableUpdateData);
sinkContext.updateCounter(fullTableName, batch.size());
// ---- Outside lock: build proto and write ----
Expand Down
109 changes: 77 additions & 32 deletions src/test/java/io/pixelsdb/pixels/sink/writer/retina/RetinaPayloadBuilderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.List;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

class RetinaPayloadBuilderTest
{
Expand All @@ -51,43 +52,87 @@ void shouldBuildCompleteUpdateRecordRequest() throws Exception
RowChangeEvent delete = event(
SinkProto.OperationType.DELETE, row("4", "delete"), null, metadata);

RetinaProto.TableUpdateData tableUpdate =
RetinaPayloadBuilder.buildTableUpdateData(
List<RetinaProto.TableUpdateData> tableUpdates =
RetinaPayloadBuilder.buildTableUpdateDataList(
TABLE_NAME, TIMESTAMP, List.of(insert, snapshot, update, delete));
RetinaProto.UpdateRecordRequest request =
RetinaPayloadBuilder.buildUpdateRecordRequest(
"fixed-token", SCHEMA_NAME, 23, List.of(tableUpdate));
"fixed-token", SCHEMA_NAME, 23, tableUpdates);

assertEquals("fixed-token", request.getHeader().getToken());
assertEquals(SCHEMA_NAME, request.getSchemaName());
assertEquals(23, request.getVirtualNodeId());
assertEquals(List.of(tableUpdate), request.getTableUpdateDataList());
assertEquals(TABLE_NAME, tableUpdate.getTableName());
assertEquals(INDEX_ID, tableUpdate.getPrimaryIndexId());
assertEquals(TIMESTAMP, tableUpdate.getTimestamp());
assertEquals(tableUpdates, request.getTableUpdateDataList());
assertEquals(4, tableUpdates.size());
for (RetinaProto.TableUpdateData tableUpdate : tableUpdates)
{
assertEquals(TABLE_NAME, tableUpdate.getTableName());
assertEquals(INDEX_ID, tableUpdate.getPrimaryIndexId());
assertEquals(TIMESTAMP, tableUpdate.getTimestamp());
}

assertEquals(1, tableUpdates.get(0).getInsertDataCount());
assertRow(tableUpdates.get(0).getInsertData(0).getIndexKeys(0),
tableUpdates.get(0).getInsertData(0).getColValuesList(), "1", "insert");
assertEquals(List.of(false, false), tableUpdates.get(0).getInsertData(0).getIsNullList());
assertEquals(1, tableUpdates.get(1).getInsertDataCount());
assertRow(tableUpdates.get(1).getInsertData(0).getIndexKeys(0),
tableUpdates.get(1).getInsertData(0).getColValuesList(), "2", "snapshot");
assertEquals(List.of(false, false), tableUpdates.get(1).getInsertData(0).getIsNullList());

assertEquals(1, tableUpdates.get(2).getUpdateDataCount());
assertRow(tableUpdates.get(2).getUpdateData(0).getIndexKeys(0),
tableUpdates.get(2).getUpdateData(0).getColValuesList(), "3", "after");
assertEquals(List.of(false, false), tableUpdates.get(2).getUpdateData(0).getIsNullList());

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());
assertEquals(INDEX_ID, tableUpdate.getDeleteData(0).getIndexKeys(0).getIndexId());
assertEquals(TABLE_ID, tableUpdate.getDeleteData(0).getIndexKeys(0).getTableId());
assertEquals(TIMESTAMP, tableUpdate.getDeleteData(0).getIndexKeys(0).getTimestamp());
assertEquals(1, tableUpdates.get(3).getDeleteDataCount());
assertEquals(bytes("4"), tableUpdates.get(3).getDeleteData(0).getIndexKeys(0).getKey());
assertEquals(INDEX_ID, tableUpdates.get(3).getDeleteData(0).getIndexKeys(0).getIndexId());
assertEquals(TABLE_ID, tableUpdates.get(3).getDeleteData(0).getIndexKeys(0).getTableId());
assertEquals(TIMESTAMP, tableUpdates.get(3).getDeleteData(0).getIndexKeys(0).getTimestamp());

assertEquals(request, RetinaProto.UpdateRecordRequest.parseFrom(request.toByteArray()));
}

@Test
void shouldGroupOnlyAdjacentEventsWithTheSameOperation() throws Exception
{
TableMetadata metadata = tableMetadata();
List<RowChangeEvent> events = List.of(
event(SinkProto.OperationType.INSERT, null, row("1", "first"), metadata),
event(SinkProto.OperationType.INSERT, null, row("2", "second"), metadata),
event(SinkProto.OperationType.DELETE, row("3", "third"), null, metadata),
event(SinkProto.OperationType.DELETE, row("4", "fourth"), null, metadata),
event(SinkProto.OperationType.INSERT, null, row("5", "fifth"), metadata));

List<RetinaProto.TableUpdateData> updates =
RetinaPayloadBuilder.buildTableUpdateDataList(TABLE_NAME, TIMESTAMP, events);

assertEquals(3, updates.size());
assertEquals(2, updates.get(0).getInsertDataCount());
assertEquals(2, updates.get(1).getDeleteDataCount());
assertEquals(1, updates.get(2).getInsertDataCount());
assertEquals(bytes("1"), updates.get(0).getInsertData(0).getIndexKeys(0).getKey());
assertEquals(bytes("2"), updates.get(0).getInsertData(1).getIndexKeys(0).getKey());
assertEquals(bytes("3"), updates.get(1).getDeleteData(0).getIndexKeys(0).getKey());
assertEquals(bytes("4"), updates.get(1).getDeleteData(1).getIndexKeys(0).getKey());
assertEquals(bytes("5"), updates.get(2).getInsertData(0).getIndexKeys(0).getKey());
}

@Test
void shouldRejectMixedOperationsInOneTableUpdateData() throws Exception
{
TableMetadata metadata = tableMetadata();
RowChangeEvent insert = event(
SinkProto.OperationType.INSERT, null, row("1", "insert"), metadata);
RowChangeEvent delete = event(
SinkProto.OperationType.DELETE, row("1", "delete"), null, metadata);

assertThrows(IllegalArgumentException.class,
() -> RetinaPayloadBuilder.buildTableUpdateData(
TABLE_NAME, TIMESTAMP, List.of(insert, delete)));
}

@Test
void shouldPropagateIsNullAlongsideColValues() throws Exception
{
Expand All @@ -111,16 +156,16 @@ void shouldPropagateIsNullAlongsideColValues() throws Exception
.build(),
metadata);

RetinaProto.TableUpdateData tableUpdate =
RetinaPayloadBuilder.buildTableUpdateData(
List<RetinaProto.TableUpdateData> tableUpdates =
RetinaPayloadBuilder.buildTableUpdateDataList(
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, true), tableUpdates.get(0).getInsertData(0).getIsNullList());
assertEquals(bytes("5"), tableUpdates.get(0).getInsertData(0).getColValues(0));
assertEquals(ByteString.EMPTY, tableUpdates.get(0).getInsertData(0).getColValues(1));

assertEquals(List.of(false, false), tableUpdate.getUpdateData(0).getIsNullList());
assertEquals(ByteString.EMPTY, tableUpdate.getUpdateData(0).getColValues(1));
assertEquals(List.of(false, false), tableUpdates.get(1).getUpdateData(0).getIsNullList());
assertEquals(ByteString.EMPTY, tableUpdates.get(1).getUpdateData(0).getColValues(1));
}

private static void assertRow(
Expand Down
Loading