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 a2ab248..4e2d79b 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
@@ -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;
/**
@@ -30,6 +31,41 @@ private RetinaPayloadBuilder()
{
}
+ /**
+ * Builds table updates while preserving the source operation order.
+ *
+ * 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}.
+ */
+ public static List buildTableUpdateDataList(
+ String tableName,
+ long timestamp,
+ List events) throws SinkException
+ {
+ if (events == null || events.isEmpty())
+ {
+ throw new IllegalArgumentException("events is empty");
+ }
+
+ List 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,
@@ -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())
diff --git a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableCrossTxWriter.java b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableCrossTxWriter.java
index ce1a535..1c90b06 100644
--- a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableCrossTxWriter.java
+++ b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableCrossTxWriter.java
@@ -72,17 +72,9 @@ public void flush(List 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<>();
}
@@ -91,13 +83,8 @@ public void flush(List 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());
@@ -149,6 +136,26 @@ public void flush(List batch)
}
}
+ private void addTransactionBatch(
+ String txId,
+ List transactionBatch,
+ List tableUpdateData,
+ List txIds,
+ List fullTableNames,
+ List tableUpdateCount)
+ {
+ List 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 txIds)
{
for (String writeTxId : txIds)
@@ -187,13 +194,13 @@ private void updateCtxCounters(List txIds, List fullTableName, L
writeLock.unlock();
}
- protected RetinaProto.TableUpdateData buildTableUpdateDataFromBatch(
+ protected List buildTableUpdateDataFromBatch(
String txId, List smallBatch)
{
SinkContext sinkContext = SinkContextManager.getInstance().getSinkContext(txId);
if (sinkContext == null)
{
- return null;
+ return List.of();
}
try
{
@@ -212,7 +219,7 @@ protected RetinaProto.TableUpdateData buildTableUpdateDataFromBatch(
}
try
{
- return RetinaPayloadBuilder.buildTableUpdateData(
+ return RetinaPayloadBuilder.buildTableUpdateDataList(
tableName, sinkContext.getTimestamp(), smallBatch);
} catch (SinkException e)
{
diff --git a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleRecordWriter.java b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleRecordWriter.java
index 39391fb..3d2c6a2 100644
--- a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleRecordWriter.java
+++ b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleRecordWriter.java
@@ -52,12 +52,9 @@ public void flush(List batch)
{
List tableUpdateData = new LinkedList<>();
// Timestamp + IndexKey binding happen inside RetinaPayloadBuilder.
- RetinaProto.TableUpdateData update =
+ List updates =
buildTableUpdateDataFromBatch(pixelsTransContext, batch);
- if (update != null)
- {
- tableUpdateData.add(update);
- }
+ tableUpdateData.addAll(updates);
// flushRateLimiter.acquire(batch.size());
long txStartTime = System.currentTimeMillis();
@@ -98,12 +95,12 @@ public void flush(List batch)
}
}
- protected RetinaProto.TableUpdateData buildTableUpdateDataFromBatch(
+ protected List buildTableUpdateDataFromBatch(
TransContext transContext, List smallBatch)
{
try
{
- return RetinaPayloadBuilder.buildTableUpdateData(
+ return RetinaPayloadBuilder.buildTableUpdateDataList(
tableName, transContext.getTimestamp(), smallBatch);
} catch (SinkException e)
{
diff --git a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleTxWriter.java b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleTxWriter.java
index e8cef3e..0aa47ab 100644
--- a/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleTxWriter.java
+++ b/src/main/java/io/pixelsdb/pixels/sink/writer/retina/TableSingleTxWriter.java
@@ -82,8 +82,8 @@ public void flush(List batchToFlush)
RowChangeEvent event1 = batch.get(0);
try
{
- List tableUpdateData = List.of(
- RetinaPayloadBuilder.buildTableUpdateData(tableName, 0L, batch));
+ List tableUpdateData =
+ RetinaPayloadBuilder.buildTableUpdateDataList(tableName, 0L, batch);
delegate.writeTrans(event1.getSchemaName(), tableUpdateData);
sinkContext.updateCounter(fullTableName, batch.size());
// ---- Outside lock: build proto and write ----
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 d13a25b..28d0d07 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
@@ -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
{
@@ -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 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 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 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
{
@@ -111,16 +156,16 @@ void shouldPropagateIsNullAlongsideColValues() throws Exception
.build(),
metadata);
- RetinaProto.TableUpdateData tableUpdate =
- RetinaPayloadBuilder.buildTableUpdateData(
+ List 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(