ifesdjeen commented on code in PR #3818:
URL: https://github.com/apache/cassandra/pull/3818#discussion_r1925644497
##########
test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java:
##########
@@ -101,7 +101,7 @@ public void beforeTest() throws Throwable
@Test
public void testOne()
{
- long seed = System.nanoTime();
+ long seed = 882712153357083l;
Review Comment:
nit: debug leftover
##########
src/java/org/apache/cassandra/service/accord/AccordJournal.java:
##########
@@ -243,10 +243,9 @@ public Command.Minimal loadMinimal(int commandStoreId,
TxnId txnId, Load load, R
if (builder.isEmpty())
return null;
- Cleanup cleanup = builder.shouldCleanup(node.agent(), redundantBefore,
durableBefore, false);
+ Cleanup cleanup = builder.shouldCleanup(FULL, node.agent(),
redundantBefore, durableBefore);
switch (cleanup)
{
- case EXPUNGE_PARTIAL:
Review Comment:
on Accord side, we have the following logic:
```
Cleanup cleanup = builder.shouldCleanup(FULL, agent,
redundantBefore, durableBefore);
switch (cleanup)
{
case VESTIGIAL:
case ERASE:
case EXPUNGE:
return null;
}
```
should it be the same here? (I did try to make this logic common, but
decided to put the patch aside until we stabilize tests)
##########
src/java/org/apache/cassandra/service/accord/AccordJournal.java:
##########
@@ -225,10 +226,9 @@ public boolean awaitTermination(long timeout, TimeUnit
units) throws Interrupted
public Command loadCommand(int commandStoreId, TxnId txnId,
RedundantBefore redundantBefore, DurableBefore durableBefore)
{
Builder builder = load(commandStoreId, txnId);
- Cleanup cleanup = builder.shouldCleanup(agent, redundantBefore,
durableBefore, false);
+ Cleanup cleanup = builder.shouldCleanup(FULL, agent, redundantBefore,
durableBefore);
Review Comment:
on Accord side, we have the following logic:
```
switch (cleanup)
{
case VESTIGIAL:
case EXPUNGE:
return ErasedSafeCommand.erased(txnId, ErasedOrVestigial);
case ERASE:
return ErasedSafeCommand.erased(txnId, Erased);
}
```
looks like here we need an equivalent, or?
##########
src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java:
##########
@@ -73,22 +68,17 @@ public static ByteBuffer serialize(TxnId txnId, WaitingOn
waitingOn) throws IOEx
if (waitingOn.appliedOrInvalidated != null)
appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
- Timestamp executeAtLeast = waitingOn.executeAtLeast();
- ByteBuffer out = ByteBuffer.allocate(1 + (executeAtLeast == null ? 0 :
CommandSerializers.timestamp.serializedSize())
- +
TypeSizes.sizeofUnsignedVInt(keyCount) +
TypeSizes.sizeofUnsignedVInt(txnIdCount)
+ ByteBuffer out =
ByteBuffer.allocate(TypeSizes.sizeofUnsignedVInt(waitingOnLength)
+ + (waitingOn.appliedOrInvalidated
== null ? 0 : TypeSizes.sizeof(waitingOnLength - appliedOrInvalidatedLength))
Review Comment:
I think we need to have `TypeSizes.sizeofUnsignedVInt` here instead of just
`sizeOf`
##########
src/java/org/apache/cassandra/service/accord/AccordJournal.java:
##########
@@ -727,16 +742,30 @@ private void deserialize(Field field, DataInputPlus in,
int userVersion) throws
switch (field)
{
case EXECUTE_AT:
- executeAt = CommandSerializers.timestamp.deserialize(in,
userVersion);
+ {
Review Comment:
Would you mind to extract this and deser logic into a class; maybe under
CommandSerializers, if not - under this class?
##########
src/java/org/apache/cassandra/service/accord/AccordJournal.java:
##########
@@ -759,16 +788,12 @@ private void deserialize(Field field, DataInputPlus in,
int userVersion) throws
byte[] waitingOnBytes = new byte[size];
in.readFully(waitingOnBytes);
ByteBuffer buffer = ByteBuffer.wrap(waitingOnBytes);
- waitingOn = (localTxnId, deps) -> {
- try
- {
- Invariants.nonNull(deps);
- return WaitingOnSerializer.deserialize(localTxnId,
deps.keyDeps.keys(), deps.rangeDeps, deps.directKeyDeps, buffer);
- }
- catch (IOException e)
- {
- throw Throwables.unchecked(e);
- }
+ waitingOn = (localTxnId, deps, executeAtLeast, uniqueHlc)
-> {
+ Invariants.nonNull(deps);
+ Command.WaitingOn basic =
WaitingOnSerializer.deserializeBasic(localTxnId, deps.keyDeps.keys(),
deps.rangeDeps, deps.directKeyDeps, buffer);
Review Comment:
We never serialize min unique HLC in this code.
##########
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java:
##########
@@ -912,9 +913,9 @@ protected UnfilteredRowIterator
applyToPartition(UnfilteredRowIterator partition
RedundantBefore redundantBefore =
redundantBefores.get(key.commandStoreId);
DurableBefore durableBefore =
durableBefores.get(key.commandStoreId);
- Cleanup cleanup = commandBuilder.shouldCleanup(agent,
redundantBefore, durableBefore, true);
+ Cleanup cleanup = commandBuilder.shouldCleanup(PARTIAL, agent,
redundantBefore, durableBefore);
if (cleanup == ERASE)
- return PartitionUpdate.fullPartitionDelete(metadata(),
partition.partitionKey(), maxSeenTimestamp, nowInSec).unfilteredIterator();
+ return PartitionUpdate.fullPartitionDelete(metadata(),
partition.partitionKey(), Long.MAX_VALUE, nowInSec).unfilteredIterator();
Review Comment:
I think this change makes maxSeenTimestamp write-only, so probably we can
drop it entirely
##########
src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java:
##########
@@ -73,22 +68,17 @@ public static ByteBuffer serialize(TxnId txnId, WaitingOn
waitingOn) throws IOEx
if (waitingOn.appliedOrInvalidated != null)
appliedOrInvalidatedLength = (txnIdCount + 63) / 64;
- Timestamp executeAtLeast = waitingOn.executeAtLeast();
- ByteBuffer out = ByteBuffer.allocate(1 + (executeAtLeast == null ? 0 :
CommandSerializers.timestamp.serializedSize())
- +
TypeSizes.sizeofUnsignedVInt(keyCount) +
TypeSizes.sizeofUnsignedVInt(txnIdCount)
+ ByteBuffer out =
ByteBuffer.allocate(TypeSizes.sizeofUnsignedVInt(waitingOnLength)
+ + (waitingOn.appliedOrInvalidated
== null ? 0 : TypeSizes.sizeof(waitingOnLength - appliedOrInvalidatedLength))
+ TypeSizes.LONG_SIZE *
(waitingOnLength + appliedOrInvalidatedLength));
- if (executeAtLeast == null) out.put((byte)0);
- else
- {
- out.put((byte) 1);
- CommandSerializers.timestamp.serialize(executeAtLeast, out);
- }
- VIntCoding.writeUnsignedVInt32(keyCount, out);
- VIntCoding.writeUnsignedVInt32(txnIdCount, out);
+ VIntCoding.writeUnsignedVInt32(waitingOnLength, out);
serialize(waitingOnLength, waitingOn.waitingOn, out);
if (appliedOrInvalidatedLength > 0)
Review Comment:
I think we need to use `waitingOn.appliedOrInvalidated != null` here, as
there seem to be cases where we can have empty appliedOrInvalidated length, in
which case we will compute size differently. An alternative is of course, when
computing size, to use `> 0` as a check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
