You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by be...@apache.org on 2024/10/11 15:32:02 UTC
(cassandra) branch cep-15-accord updated: do not deserialize deps in CommandsForRangesLoader unless required
This is an automated email from the ASF dual-hosted git repository.
benedict pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-15-accord by this push:
new 4abe3f84b1 do not deserialize deps in CommandsForRangesLoader unless required
4abe3f84b1 is described below
commit 4abe3f84b15cbca35d839a6d21e7942a2a6c5608
Author: Benedict Elliott Smith <be...@apache.org>
AuthorDate: Fri Oct 11 16:27:33 2024 +0100
do not deserialize deps in CommandsForRangesLoader unless required
---
modules/accord | 2 +-
.../service/accord/AccordCommandStore.java | 7 ++
.../cassandra/service/accord/AccordJournal.java | 24 +++++-
.../accord/AccordJournalValueSerializers.java | 3 +-
.../service/accord/CommandsForRangesLoader.java | 69 ++++++++++++++---
.../apache/cassandra/service/accord/IJournal.java | 1 +
.../cassandra/service/accord/SavedCommand.java | 88 +++++++++++++++++++---
.../cassandra/service/accord/MockJournal.java | 13 ++++
.../cassandra/service/accord/SavedCommandTest.java | 4 +-
9 files changed, 183 insertions(+), 28 deletions(-)
diff --git a/modules/accord b/modules/accord
index dd04c81214..9cc9b3a12e 160000
--- a/modules/accord
+++ b/modules/accord
@@ -1 +1 @@
-Subproject commit dd04c81214be9213cf503025bf6ed4cbf118be00
+Subproject commit 9cc9b3a12e8271f9be2d047719a04318de275fba
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index 196d5d4913..f07901d502 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -71,6 +71,7 @@ import org.apache.cassandra.concurrent.SequentialExecutorPlus;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.service.accord.SavedCommand.MinimalCommand;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.service.accord.async.AsyncOperation;
import org.apache.cassandra.service.accord.events.CacheEvents;
@@ -83,6 +84,7 @@ import static accord.primitives.Status.Committed;
import static accord.primitives.Status.Invalidated;
import static accord.primitives.Status.Truncated;
import static accord.utils.Invariants.checkState;
+import static org.apache.cassandra.service.accord.SavedCommand.Load.MINIMAL;
public class AccordCommandStore extends CommandStore
{
@@ -541,6 +543,11 @@ public class AccordCommandStore extends CommandStore
return journal.loadCommand(id, txnId, unsafeGetRedundantBefore(), durableBefore());
}
+ public MinimalCommand loadMinimal(TxnId txnId)
+ {
+ return journal.loadMinimal(id, txnId, MINIMAL, unsafeGetRedundantBefore(), durableBefore());
+ }
+
public interface Loader
{
Promise<?> load(Command next);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index 26b868e5d1..309905648a 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -193,6 +193,21 @@ public class AccordJournal implements IJournal, Shutdownable
return builder.construct();
}
+ @Override
+ public SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId, SavedCommand.Load load, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ {
+ SavedCommand.Builder builder = loadDiffs(commandStoreId, txnId, load);
+ Cleanup cleanup = builder.shouldCleanup(redundantBefore, durableBefore);
+ switch (cleanup)
+ {
+ case EXPUNGE_PARTIAL:
+ case EXPUNGE:
+ case ERASE:
+ return null;
+ }
+ return builder.asMinimal();
+ }
+
@VisibleForTesting
public RedundantBefore loadRedundantBefore(int store)
{
@@ -299,14 +314,19 @@ public class AccordJournal implements IJournal, Shutdownable
}
@VisibleForTesting
- public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId)
+ public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId, SavedCommand.Load load)
{
JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, commandStoreId);
- SavedCommand.Builder builder = new SavedCommand.Builder(txnId);
+ SavedCommand.Builder builder = new SavedCommand.Builder(txnId, load);
journalTable.readAll(key, builder::deserializeNext);
return builder;
}
+ public SavedCommand.Builder loadDiffs(int commandStoreId, TxnId txnId)
+ {
+ return loadDiffs(commandStoreId, txnId, SavedCommand.Load.ALL);
+ }
+
private <BUILDER> BUILDER readAll(JournalKey key)
{
BUILDER builder = (BUILDER) key.type.serializer.mergerFor(key);
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
index 60a1ef4f58..3f68a6eb1c 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java
@@ -38,6 +38,7 @@ import org.apache.cassandra.service.accord.serializers.CommandStoreSerializers;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
import static accord.local.CommandStores.RangesForEpoch;
+import static org.apache.cassandra.service.accord.SavedCommand.Load.ALL;
import static org.apache.cassandra.service.accord.serializers.DepsSerializer.deps;
// TODO (required): test with large collection values, and perhaps split out some fields if they have a tendency to grow larger
@@ -63,7 +64,7 @@ public class AccordJournalValueSerializers
@Override
public SavedCommand.Builder mergerFor(JournalKey journalKey)
{
- return new SavedCommand.Builder(journalKey.id);
+ return new SavedCommand.Builder(journalKey.id, ALL);
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
index fa3ee4a846..6f4c4afaa0 100644
--- a/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
+++ b/src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
@@ -227,13 +227,27 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener<TxnId,
{
if (cacheHits.containsKey(txnId))
continue;
- var cmd = store.loadCommand(txnId);
- if (cmd == null)
- continue; // unknown command
- var summary = create(cmd, ranges, findAsDep, redundantBefore);
- if (summary == null)
- continue;
- map.put(txnId, summary);
+ if (findAsDep == null)
+ {
+ var cmd = store.loadMinimal(txnId);
+ if (cmd == null)
+ continue; // unknown command
+ var summary = create(cmd, ranges, redundantBefore);
+ if (summary == null)
+ continue;
+ map.put(txnId, summary);
+
+ }
+ else
+ {
+ var cmd = store.loadCommand(txnId);
+ if (cmd == null)
+ continue; // unknown command
+ var summary = create(cmd, ranges, findAsDep, redundantBefore);
+ if (summary == null)
+ continue;
+ map.put(txnId, summary);
+ }
}
return map;
}
@@ -260,13 +274,11 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener<TxnId,
if (redundantBefore != null)
{
- Ranges durableAlready = Ranges.of(redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> {
+ Ranges newRanges = redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> {
if (e.gcBefore.compareTo(cmd.txnId()) < 0)
return accum;
- accum.add(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end));
- return accum;
- }, new ArrayList<Range>(), ignore -> false).toArray(Range[]::new));
- Ranges newRanges = ranges.without(durableAlready);
+ return accum.without(Ranges.of(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end)));
+ }, ranges, ignore -> false);
if (newRanges.isEmpty())
return null;
@@ -277,6 +289,39 @@ public class CommandsForRangesLoader implements AccordStateCache.Listener<TxnId,
return new Summary(cmd.txnId(), cmd.executeAt(), saveStatus, ranges, findAsDep, hasAsDep);
}
+ private static Summary create(SavedCommand.MinimalCommand cmd, Ranges cacheRanges, @Nullable RedundantBefore redundantBefore)
+ {
+ //TODO (required, correctness): C* did Invalidated, accord-core did Erased... what is correct?
+ SaveStatus saveStatus = cmd.saveStatus;
+ if (saveStatus == SaveStatus.Invalidated
+ || saveStatus == SaveStatus.Erased
+ || !saveStatus.hasBeen(Status.PreAccepted))
+ return null;
+
+ var keysOrRanges = cmd.participants.touches().toRanges();
+ if (keysOrRanges.domain() != Domain.Range)
+ throw new AssertionError(String.format("Txn keys are not range for %s", cmd.participants));
+ Ranges ranges = (Ranges) keysOrRanges;
+
+ ranges = ranges.slice(cacheRanges, Routables.Slice.Minimal);
+ if (ranges.isEmpty())
+ return null;
+
+ if (redundantBefore != null)
+ {
+ Ranges newRanges = redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> {
+ if (e.gcBefore.compareTo(cmd.txnId) < 0)
+ return accum;
+ return accum.without(Ranges.of(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end)));
+ }, ranges, ignore -> false);
+
+ if (newRanges.isEmpty())
+ return null;
+ }
+
+ return new Summary(cmd.txnId, cmd.executeAt, saveStatus, ranges, null, false);
+ }
+
public void mergeHistoricalTransaction(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
{
historicalTransaction.merge(txnId, ranges, remappingFunction);
diff --git a/src/java/org/apache/cassandra/service/accord/IJournal.java b/src/java/org/apache/cassandra/service/accord/IJournal.java
index 7de9d68602..f5bde09566 100644
--- a/src/java/org/apache/cassandra/service/accord/IJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/IJournal.java
@@ -34,6 +34,7 @@ import accord.utils.PersistentField.Persister;
public interface IJournal
{
Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore);
+ SavedCommand.MinimalCommand loadMinimal(int commandStoreId, TxnId txnId, SavedCommand.Load load, RedundantBefore redundantBefore, DurableBefore durableBefore);
RedundantBefore loadRedundantBefore(int commandStoreId);
NavigableMap<TxnId, Ranges> loadBootstrapBeganAt(int commandStoreId);
diff --git a/src/java/org/apache/cassandra/service/accord/SavedCommand.java b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
index be2f9c6289..71582deb9b 100644
--- a/src/java/org/apache/cassandra/service/accord/SavedCommand.java
+++ b/src/java/org/apache/cassandra/service/accord/SavedCommand.java
@@ -59,7 +59,12 @@ import static accord.primitives.Known.KnownDeps.NoDeps;
import static accord.primitives.SaveStatus.TruncatedApplyWithOutcome;
import static accord.primitives.Status.Durability.NotDurable;
import static accord.utils.Invariants.illegalState;
+import static org.apache.cassandra.service.accord.SavedCommand.Fields.DURABILITY;
+import static org.apache.cassandra.service.accord.SavedCommand.Fields.EXECUTE_AT;
import static org.apache.cassandra.service.accord.SavedCommand.Fields.PARTICIPANTS;
+import static org.apache.cassandra.service.accord.SavedCommand.Fields.SAVE_STATUS;
+import static org.apache.cassandra.service.accord.SavedCommand.Fields.WRITES;
+import static org.apache.cassandra.service.accord.SavedCommand.Load.ALL;
public class SavedCommand
{
@@ -231,8 +236,8 @@ public class SavedCommand
flags = collectFlags(before, after, Command::executeAt, true, Fields.EXECUTE_AT, flags);
flags = collectFlags(before, after, Command::executesAtLeast, true, Fields.EXECUTES_AT_LEAST, flags);
- flags = collectFlags(before, after, Command::saveStatus, false, Fields.SAVE_STATUS, flags);
- flags = collectFlags(before, after, Command::durability, false, Fields.DURABILITY, flags);
+ flags = collectFlags(before, after, Command::saveStatus, false, SAVE_STATUS, flags);
+ flags = collectFlags(before, after, Command::durability, false, DURABILITY, flags);
flags = collectFlags(before, after, Command::acceptedOrCommitted, false, Fields.ACCEPTED, flags);
flags = collectFlags(before, after, Command::promised, false, Fields.PROMISED, flags);
@@ -243,7 +248,7 @@ public class SavedCommand
flags = collectFlags(before, after, SavedCommand::getWaitingOn, false, Fields.WAITING_ON, flags);
- flags = collectFlags(before, after, Command::writes, false, Fields.WRITES, flags);
+ flags = collectFlags(before, after, Command::writes, false, WRITES, flags);
return flags;
}
@@ -323,8 +328,51 @@ public class SavedCommand
return oldFlags & ~(1 << field.ordinal());
}
+ public enum Load
+ {
+ ALL(0),
+ PURGEABLE(SAVE_STATUS, PARTICIPANTS, DURABILITY, EXECUTE_AT, WRITES),
+ MINIMAL(SAVE_STATUS, PARTICIPANTS, EXECUTE_AT);
+
+ final int mask;
+
+ Load(int mask)
+ {
+ this.mask = mask;
+ }
+
+ Load(Fields ... fields)
+ {
+ int mask = -1;
+ for (Fields field : fields)
+ mask &= ~(1<< field.ordinal());
+ this.mask = mask;
+ }
+ }
+
+ public static class MinimalCommand
+ {
+ public final TxnId txnId;
+ public final SaveStatus saveStatus;
+ public final StoreParticipants participants;
+ public final Status.Durability durability;
+ public final Timestamp executeAt;
+ public final Writes writes;
+
+ public MinimalCommand(TxnId txnId, SaveStatus saveStatus, StoreParticipants participants, Status.Durability durability, Timestamp executeAt, Writes writes)
+ {
+ this.txnId = txnId;
+ this.saveStatus = saveStatus;
+ this.participants = participants;
+ this.durability = durability;
+ this.executeAt = executeAt;
+ this.writes = writes;
+ }
+ }
+
public static class Builder
{
+ final int mask;
int flags;
TxnId txnId;
@@ -350,13 +398,25 @@ public class SavedCommand
boolean nextCalled;
int count;
- public Builder(TxnId txnId)
+ public Builder(TxnId txnId, Load load)
{
+ this.mask = load.mask;
init(txnId);
}
+ public Builder(TxnId txnId)
+ {
+ this(txnId, ALL);
+ }
+
+ public Builder(Load load)
+ {
+ this.mask = load.mask;
+ }
+
public Builder()
{
+ this(ALL);
}
public TxnId txnId()
@@ -522,13 +582,13 @@ public class SavedCommand
public Builder expungePartial(Cleanup cleanup, SaveStatus saveStatus, boolean includeOutcome)
{
Invariants.checkState(txnId != null);
- Builder builder = new Builder(txnId);
+ Builder builder = new Builder(txnId, ALL);
builder.count++;
builder.nextCalled = true;
Invariants.checkState(saveStatus != null);
- builder.flags = setFieldChanged(Fields.SAVE_STATUS, builder.flags);
+ builder.flags = setFieldChanged(SAVE_STATUS, builder.flags);
builder.saveStatus = saveStatus;
builder.flags = setFieldChanged(Fields.CLEANUP, builder.flags);
builder.cleanup = cleanup;
@@ -539,7 +599,7 @@ public class SavedCommand
}
if (durability != null)
{
- builder.flags = setFieldChanged(Fields.DURABILITY, builder.flags);
+ builder.flags = setFieldChanged(DURABILITY, builder.flags);
builder.durability = durability;
}
if (participants != null)
@@ -549,7 +609,7 @@ public class SavedCommand
}
if (includeOutcome && builder.writes != null)
{
- builder.flags = setFieldChanged(Fields.WRITES, builder.flags);
+ builder.flags = setFieldChanged(WRITES, builder.flags);
builder.writes = writes;
}
@@ -559,7 +619,7 @@ public class SavedCommand
public Builder saveStatusOnly()
{
Invariants.checkState(txnId != null);
- Builder builder = new Builder(txnId);
+ Builder builder = new Builder(txnId, ALL);
builder.count++;
builder.nextCalled = true;
@@ -567,7 +627,7 @@ public class SavedCommand
// TODO: these accesses can be abstracted away
if (saveStatus != null)
{
- builder.flags = setFieldChanged(Fields.SAVE_STATUS, builder.flags);
+ builder.flags = setFieldChanged(SAVE_STATUS, builder.flags);
builder.saveStatus = saveStatus;
}
@@ -583,6 +643,11 @@ public class SavedCommand
}
}
+ public MinimalCommand asMinimal()
+ {
+ return new MinimalCommand(txnId, saveStatus, participants, durability, executeAt, writes);
+ }
+
public static Route<?> deserializeRouteOrNull(DataInputPlus in, int userVersion) throws IOException
{
int flags = in.readInt();
@@ -595,6 +660,7 @@ public class SavedCommand
public void serialize(DataOutputPlus out, int userVersion) throws IOException
{
+ Invariants.checkState(mask == 0);
out.writeInt(validateFlags(flags));
int iterable = toIterableSetFields(flags);
@@ -665,7 +731,7 @@ public class SavedCommand
while (iterable != 0)
{
Fields field = nextSetField(iterable);
- if (getFieldChanged(field, this.flags))
+ if (getFieldChanged(field, this.flags) || getFieldIsNull(field, mask))
{
if (!getFieldIsNull(field, flags))
skip(field, in, userVersion);
diff --git a/test/unit/org/apache/cassandra/service/accord/MockJournal.java b/test/unit/org/apache/cassandra/service/accord/MockJournal.java
index 64caa97216..4c1c038676 100644
--- a/test/unit/org/apache/cassandra/service/accord/MockJournal.java
+++ b/test/unit/org/apache/cassandra/service/accord/MockJournal.java
@@ -54,6 +54,8 @@ import org.apache.cassandra.service.accord.AccordJournalValueSerializers.Durable
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.HistoricalTransactionsAccumulator;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.IdentityAccumulator;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.RedundantBeforeAccumulator;
+import org.apache.cassandra.service.accord.SavedCommand.Load;
+import org.apache.cassandra.service.accord.SavedCommand.MinimalCommand;
import org.apache.cassandra.service.accord.serializers.CommandSerializers;
public class MockJournal implements IJournal
@@ -81,6 +83,17 @@ public class MockJournal implements IJournal
return reconstructFromDiff(new ArrayList<>(saved));
}
+ @Override
+ public MinimalCommand loadMinimal(int store, TxnId txnId, Load load, RedundantBefore redundantBefore, DurableBefore durableBefore)
+ {
+ JournalKey key = new JournalKey(txnId, JournalKey.Type.COMMAND_DIFF, store);
+ List<LoadedDiff> saved = commands.get(key);
+ if (saved == null)
+ return null;
+ Command command = reconstructFromDiff(new ArrayList<>(saved));
+ return new MinimalCommand(command.txnId(), command.saveStatus(), command.participants(), command.durability(), command.executeAt(), command.writes());
+ }
+
@Override
public RedundantBefore loadRedundantBefore(int store)
{
diff --git a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java
index 99963c3af0..3cfc1c91c5 100644
--- a/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/SavedCommandTest.java
@@ -40,12 +40,14 @@ import org.apache.cassandra.schema.Schema;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.SavedCommand.Fields;
+import org.apache.cassandra.service.accord.SavedCommand.Load;
import org.apache.cassandra.service.consensus.TransactionalMode;
import org.apache.cassandra.utils.AccordGenerators;
import org.assertj.core.api.SoftAssertions;
import static accord.utils.Property.qt;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
+import static org.apache.cassandra.service.accord.SavedCommand.Load.ALL;
import static org.apache.cassandra.service.accord.SavedCommand.getFlags;
public class SavedCommandTest
@@ -96,7 +98,7 @@ public class SavedCommandTest
out.clear();
Command orig = cmdBuilder.build(saveStatus);
SavedCommand.serialize(null, orig, out, userVersion);
- SavedCommand.Builder builder = new SavedCommand.Builder(orig.txnId());
+ SavedCommand.Builder builder = new SavedCommand.Builder(orig.txnId(), Load.ALL);
builder.deserializeNext(new DataInputBuffer(out.unsafeGetBufferAndFlip(), false), userVersion);
// We are not persisting the result, so force it for strict equality
builder.forceResult(orig.result());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org