From 08428a27508fd5c2150eef7aad364882c75a2b5c Mon Sep 17 00:00:00 2001 From: Caleb Rackliffe Date: Wed, 20 Sep 2023 11:25:51 -0500 Subject: [PATCH] - Avoid persisting fragments that do not require completion as Updates in TxnWrite, as they can simply be pulled from PartialTxn when needed in Write#apply() - Avoid serializing full TxnData instances to Accord state tables patch by Caleb Rackliffe; reviewed by David Capwell, Benedict Elliot Smith, and Ariel Weisberg for CASSANDRA-18355 --- modules/accord | 2 +- .../service/accord/AccordKeyspace.java | 62 +++++++------------ .../service/accord/AccordObjectSizes.java | 10 +-- .../accord/serializers/ApplySerializers.java | 8 +-- .../serializers/CheckStatusSerializers.java | 12 ++-- .../serializers/RecoverySerializers.java | 24 ++++--- .../service/accord/txn/TxnUpdate.java | 21 ++++++- .../service/accord/txn/TxnWrite.java | 54 ++++++++++++---- .../test/accord/AccordCQLTest.java | 26 ++++++++ .../test/accord/AccordIntegrationTest.java | 1 - .../CompactionAccordIteratorsTest.java | 4 +- .../accord/AccordCommandStoreTest.java | 2 +- .../accord/AccordSyncPropagatorTest.java | 6 +- 13 files changed, 144 insertions(+), 88 deletions(-) diff --git a/modules/accord b/modules/accord index 1d6028ca20..df492dfd2f 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit 1d6028ca20553d1c1a6fe2809b204254955da3b3 +Subproject commit df492dfd2ffe993c33761d0531ac5b979b80f080 diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index bfaf764592..837ad19c0a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -154,7 +154,6 @@ import org.apache.cassandra.service.accord.serializers.KeySerializers; import org.apache.cassandra.service.accord.serializers.ListenerSerializers; import org.apache.cassandra.service.accord.serializers.TopologySerializers; import org.apache.cassandra.service.accord.serializers.WaitingOnSerializer; -import org.apache.cassandra.service.accord.txn.TxnData; import org.apache.cassandra.transport.Dispatcher; import org.apache.cassandra.utils.Clock; import org.apache.cassandra.utils.btree.BTree; @@ -251,7 +250,6 @@ public class AccordKeyspace + format("accepted_ballot %s,", TIMESTAMP_TUPLE) + "dependencies blob," + "writes blob," - + "result blob," + "waiting_on blob," + "listeners set, " + "PRIMARY KEY((store_id, domain, txn_id))" @@ -267,7 +265,6 @@ public class AccordKeyspace static final LocalVersionedSerializer partialTxn = localSerializer(CommandSerializers.partialTxn); static final LocalVersionedSerializer partialDeps = localSerializer(DepsSerializer.partialDeps); static final LocalVersionedSerializer writes = localSerializer(CommandSerializers.writes); - static final LocalVersionedSerializer result = localSerializer(TxnData.serializer); static final LocalVersionedSerializer listeners = localSerializer(ListenerSerializers.listener); static final LocalVersionedSerializer topology = localSerializer(TopologySerializers.topology); static final LocalVersionedSerializer> rejectBefore = localSerializer(CommandStoreSerializers.rejectBefore); @@ -305,13 +302,12 @@ public class AccordKeyspace static final ColumnMetadata accepted_ballot = getColumn(Commands, "accepted_ballot"); static final ColumnMetadata dependencies = getColumn(Commands, "dependencies"); static final ColumnMetadata writes = getColumn(Commands, "writes"); - static final ColumnMetadata result = getColumn(Commands, "result"); static final ColumnMetadata waiting_on = getColumn(Commands, "waiting_on"); static final ColumnMetadata listeners = getColumn(Commands, "listeners"); public static ColumnMetadata[][] TRUNCATE_FIELDS = new ColumnMetadata[][] { new ColumnMetadata[] { durability, execute_at, route, status }, - new ColumnMetadata[] { durability, execute_at, result, route, status, writes }, + new ColumnMetadata[] { durability, execute_at, route, status, writes }, }; static @@ -386,61 +382,56 @@ public class AccordKeyspace } } - private static Object[] truncatedApplyLeaf(long newTimestamp, SaveStatus newSaveStatus, Cell durabilityCell, Cell executeAtCell, @Nullable Cell resultCell, Cell routeCell, @Nullable Cell writesCell, boolean updateTimestamps) + private static Object[] truncatedApplyLeaf(long newTimestamp, SaveStatus newSaveStatus, Cell durabilityCell, Cell executeAtCell, Cell routeCell, @Nullable Cell writesCell, boolean updateTimestamps) { checkArgument(durabilityCell.column() == CommandsColumns.durability); checkArgument(executeAtCell.column() == CommandsColumns.execute_at); - checkArgument(resultCell == null || resultCell.column() == CommandsColumns.result); checkArgument(routeCell.column() == CommandsColumns.route); checkArgument(writesCell == null || writesCell.column() == CommandsColumns.writes); - boolean includeOutcome = resultCell != null; + boolean includeOutcome = writesCell != null; Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(TRUNCATE_FIELDS[includeOutcome ? 1 : 0].length); int colIndex = 0; newLeaf[colIndex++] = updateTimestamps ? durabilityCell.withUpdatedTimestamp(newTimestamp) : durabilityCell; newLeaf[colIndex++] = updateTimestamps ? executeAtCell.withUpdatedTimestamp(newTimestamp) : executeAtCell; - if (includeOutcome) - newLeaf[colIndex++] = updateTimestamps ? resultCell.withUpdatedTimestamp(newTimestamp) : resultCell; newLeaf[colIndex++] = updateTimestamps ? routeCell.withUpdatedTimestamp(newTimestamp) : routeCell; // Status always needs to use the new timestamp since we are replacing the existing value // All the other columns are being retained unmodified with at most updated timestamps to accomdate deletion newLeaf[colIndex++] = BufferCell.live(status, newTimestamp, ByteBufferAccessor.instance.valueOf(newSaveStatus.ordinal())); if (includeOutcome) + //noinspection UnusedAssignment newLeaf[colIndex++] = updateTimestamps ? writesCell.withUpdatedTimestamp(newTimestamp) : writesCell; return newLeaf; } - public static Row truncatedApply(SaveStatus newSaveStatus, Row row, long nowInSec, Durability durability, Cell durabilityCell, Cell executeAtCell, Cell routeCell, boolean withOutcome) + public static Row truncatedApply(SaveStatus newSaveStatus, Row row, long nowInSec, Durability durability, Cell durabilityCell, Cell executeAtCell, Cell routeCell, boolean withOutcome) { checkArgument(durabilityCell.column() == CommandsColumns.durability); checkArgument(executeAtCell.column() == CommandsColumns.execute_at); checkArgument(routeCell.column() == CommandsColumns.route); long oldTimestamp = row.primaryKeyLivenessInfo().timestamp(); long newTimestamp = oldTimestamp + 1; - Cell resultCell = withOutcome ? row.getCell(CommandsColumns.result) : null; - Cell writesCell = withOutcome ? row.getCell(CommandsColumns.writes) : null; - checkState((resultCell != null) == (writesCell != null), "result and writes should always be set together"); - boolean doDeletion = true; + Cell writesCell = withOutcome ? row.getCell(CommandsColumns.writes) : null; + // If durability is not universal we don't want to delete older versions of the row that might have recorded // a higher durability value. maybeDropTruncatedCommandColumns will take care of dropping things even if we don't drop via tombstones. // durability should be the only column that could have an older value that is insufficient for propagating forward - if (durability != Durability.Universal) - doDeletion = false; + boolean doDeletion = durability == Durability.Universal; + // We may not have what we need to generate a deletion and include the outcome in the truncated row // so need to wait until we can have the outcome to issue the deletion otherwise it would be shadowed and lost - if (withOutcome && resultCell == null) + if (withOutcome && writesCell == null) doDeletion = false; - Object[] newLeaf = truncatedApplyLeaf(newTimestamp, newSaveStatus, durabilityCell, executeAtCell, resultCell, routeCell, writesCell, doDeletion); + Object[] newLeaf = truncatedApplyLeaf(newTimestamp, newSaveStatus, durabilityCell, executeAtCell, routeCell, writesCell, doDeletion); // Including a deletion allows future compactions to drop data before it gets to the purger // but it is pretty optional because maybeDropTruncatedCommandColumns will drop the extra columns // regardless Row.Deletion deletion = doDeletion ? new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false) : Deletion.LIVE; - return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec), - deletion, newLeaf); + return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec), deletion, newLeaf); } - public static Row maybeDropTruncatedCommandColumns(Row row, boolean withOutcome, Cell durabilityCell, Cell executeAtCell, Cell routeCell, Cell statusCell) + public static Row maybeDropTruncatedCommandColumns(Row row, boolean withOutcome, Cell durabilityCell, Cell executeAtCell, Cell routeCell, Cell statusCell) { checkArgument(durabilityCell.column() == CommandsColumns.durability); checkArgument(executeAtCell.column() == CommandsColumns.execute_at); @@ -449,38 +440,29 @@ public class AccordKeyspace int colCount = row.columnCount(); // If it's the exact length of the post truncate column count without outcome fields // then it is exactly the columns needed for getting this far and withOutcome doesn't matter since - // nothing additional is available to include anyways + // nothing additional is available to include anyway if (colCount == TRUNCATE_FIELDS[0].length) return row; - Cell resultCell = row.getCell(CommandsColumns.result); - Cell writesCell = row.getCell(CommandsColumns.writes); - checkState((resultCell != null) == (writesCell != null), "result and writes should always be set together"); - boolean includeOutcome = withOutcome && resultCell != null; + Cell writesCell = row.getCell(CommandsColumns.writes); // This has just the columns needed for truncation with outcome so return it unmodified - if (colCount == TRUNCATE_FIELDS[1].length && includeOutcome) + if (colCount == TRUNCATE_FIELDS[1].length && withOutcome) return row; // Construct a replacement with just the available columns that are still needed - Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(TRUNCATE_FIELDS[includeOutcome ? 1 : 0].length); + Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(TRUNCATE_FIELDS[withOutcome ? 1 : 0].length); int colIndex = 0; newLeaf[colIndex++] = durabilityCell; newLeaf[colIndex++] = executeAtCell; - if (includeOutcome) - newLeaf[colIndex++] = resultCell; newLeaf[colIndex++] = routeCell; newLeaf[colIndex++] = statusCell; - if (includeOutcome) + if (withOutcome && writesCell != null) + //noinspection UnusedAssignment newLeaf[colIndex++] = writesCell; return BTreeRow.create(row.clustering(), row.primaryKeyLivenessInfo(), row.deletion(), newLeaf); } - public static Result getResult(Row row) throws IOException - { - return deserializeWithVersionOr(row, result, LocalVersionedSerializers.result, () -> null); - } - public static Writes getWrites(Row row) throws IOException { return deserializeWithVersionOr(row, writes, LocalVersionedSerializers.writes, () -> null); @@ -918,7 +900,6 @@ public class AccordKeyspace addSetChanges(CommandsColumns.listeners, Command::durableListeners, v -> serialize(v, LocalVersionedSerializers.listeners), builder, timestampMicros, nowInSeconds, original, command); addCellIfModified(CommandsColumns.writes, Command::writes, v -> serialize(v, LocalVersionedSerializers.writes), builder, timestampMicros, nowInSeconds, original, command); - addCellIfModified(CommandsColumns.result, Command::result, v -> serialize((TxnData) v, LocalVersionedSerializers.result), builder, timestampMicros, nowInSeconds, original, command); // TODO review this is just to work around Truncated not being committed but having a status after committed // so status claims it is committed. @@ -1314,7 +1295,6 @@ public class AccordKeyspace Ballot promised = deserializeTimestampOrNull(row, "promised_ballot", Ballot::fromBits); Ballot accepted = deserializeTimestampOrNull(row, "accepted_ballot", Ballot::fromBits); Writes writes = deserializeWithVersionOr(row, "writes", LocalVersionedSerializers.writes, () -> null); - Result result = deserializeWithVersionOr(row, "result", LocalVersionedSerializers.result, () -> null); switch (status.status) { @@ -1331,9 +1311,9 @@ public class AccordKeyspace return Command.SerializerSupport.committed(attributes, status, executeAt, promised, accepted, waitingOn); case PreApplied: case Applied: - return Command.SerializerSupport.executed(attributes, status, executeAt, promised, accepted, waitingOn, writes, result); + return Command.SerializerSupport.executed(attributes, status, executeAt, promised, accepted, waitingOn, writes, Result.APPLIED); case Truncated: - return Command.SerializerSupport.truncatedApply(attributes, status, executeAt, writes, result); + return Command.SerializerSupport.truncatedApply(attributes, status, executeAt, writes, Result.APPLIED); case Invalidated: return Command.SerializerSupport.invalidated(txnId, attributes.durableListeners()); default: diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index a0a5202c56..bff25a9c05 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -330,7 +330,9 @@ public class AccordObjectSizes size += sizeNullable(command.partialDeps(), AccordObjectSizes::dependencies); size += sizeNullable(command.accepted(), AccordObjectSizes::timestamp); size += sizeNullable(command.writes(), AccordObjectSizes::writes); - size += sizeNullable(command.result(), AccordObjectSizes::results); + + if (command.result() instanceof TxnData) + size += sizeNullable(command.result(), AccordObjectSizes::results); if (!(command instanceof Command.Committed)) return size; @@ -352,9 +354,9 @@ public class AccordObjectSizes return size; } - private static long EMPTY_CFK_SIZE = measure(CommandsForKey.SerializerSupport.create(null, null, null, 0, null, null, - ImmutableSortedMap.of(), - ImmutableSortedMap.of())); + private static final long EMPTY_CFK_SIZE = measure(CommandsForKey.SerializerSupport.create(null, null, null, 0, null, null, + ImmutableSortedMap.of(), + ImmutableSortedMap.of())); public static long commandsForKey(CommandsForKey cfk) { long size = EMPTY_CFK_SIZE; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java index 99c8410403..9ad1fcd953 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java @@ -20,13 +20,13 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; +import accord.api.Result; import accord.messages.Apply; import accord.primitives.PartialRoute; import accord.primitives.TxnId; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.service.accord.txn.TxnData; import org.apache.cassandra.utils.NullableSerializer; public class ApplySerializers @@ -41,7 +41,6 @@ public class ApplySerializers DepsSerializer.partialDeps.serialize(apply.deps, out, version); NullableSerializer.serializeNullable(apply.txn, out, version, CommandSerializers.partialTxn); CommandSerializers.writes.serialize(apply.writes, out, version); - TxnData.serializer.serialize((TxnData) apply.result, out, version); } @Override @@ -53,7 +52,7 @@ public class ApplySerializers DepsSerializer.partialDeps.deserialize(in, version), NullableSerializer.deserializeNullable(in, version, CommandSerializers.partialTxn), CommandSerializers.writes.deserialize(in, version), - TxnData.serializer.deserialize(in, version)); + Result.APPLIED); } @Override @@ -63,8 +62,7 @@ public class ApplySerializers + CommandSerializers.timestamp.serializedSize(apply.executeAt, version) + DepsSerializer.partialDeps.serializedSize(apply.deps, version) + NullableSerializer.serializedNullableSize(apply.txn, version, CommandSerializers.partialTxn) - + CommandSerializers.writes.serializedSize(apply.writes, version) - + TxnData.serializer.serializedSize((TxnData) apply.result, version); + + CommandSerializers.writes.serializedSize(apply.writes, version); } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java index 37d42395e7..4ce9eb7e92 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java @@ -43,7 +43,6 @@ import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.service.accord.txn.TxnData; import static accord.messages.CheckStatus.SerializationSupport.createOk; import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable; @@ -121,7 +120,6 @@ public class CheckStatusSerializers serializeNullable(okFull.partialTxn, out, version, CommandSerializers.partialTxn); serializeNullable(okFull.committedDeps, out, version, DepsSerializer.partialDeps); serializeNullable(okFull.writes, out, version, CommandSerializers.writes); - serializeNullable((TxnData) okFull.result, out, version, TxnData.serializer); } @Override @@ -154,7 +152,14 @@ public class CheckStatusSerializers PartialTxn partialTxn = deserializeNullable(in, version, CommandSerializers.partialTxn); PartialDeps committedDeps = deserializeNullable(in, version, DepsSerializer.partialDeps); Writes writes = deserializeNullable(in, version, CommandSerializers.writes); - Result result = deserializeNullable(in, version, TxnData.serializer); + + Result result = null; + if (status == SaveStatus.PreApplied || status == SaveStatus.Applied + || status == SaveStatus.TruncatedApply || status == SaveStatus.TruncatedApplyWithOutcome || status == SaveStatus.TruncatedApplyWithDeps) + result = Result.APPLIED; + else if (status == SaveStatus.Invalidated) + result = Result.INVALIDATED; + return createOk(truncated, invalidIfNotAtLeast, status, maxStatus, promised, accepted, executeAt, isCoordinating, durability, route, homeKey, partialTxn, committedDeps, writes, result); } @@ -187,7 +192,6 @@ public class CheckStatusSerializers size += serializedNullableSize(okFull.partialTxn, version, CommandSerializers.partialTxn); size += serializedNullableSize(okFull.committedDeps, version, DepsSerializer.partialDeps); size += serializedNullableSize(okFull.writes, version, CommandSerializers.writes); - size += serializedNullableSize((TxnData) okFull.result, version, TxnData.serializer); return size; } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java index 5d3a5a93da..19e13cbe01 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java @@ -41,7 +41,6 @@ import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.IVersionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; -import org.apache.cassandra.service.accord.txn.TxnData; import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable; import static org.apache.cassandra.utils.NullableSerializer.serializeNullable; @@ -96,11 +95,10 @@ public class RecoverySerializers DepsSerializer.deps.serialize(recoverOk.earlierAcceptedNoWitness, out, version); out.writeBoolean(recoverOk.rejectsFastPath); serializeNullable(recoverOk.writes, out, version, CommandSerializers.writes); - serializeNullable((TxnData) recoverOk.result, out, version, TxnData.serializer); } @Override - public final void serialize(RecoverReply reply, DataOutputPlus out, int version) throws IOException + public void serialize(RecoverReply reply, DataOutputPlus out, int version) throws IOException { out.writeBoolean(reply.isOk()); if (!reply.isOk()) @@ -120,14 +118,23 @@ public class RecoverySerializers } @Override - public final RecoverReply deserialize(DataInputPlus in, int version) throws IOException + public RecoverReply deserialize(DataInputPlus in, int version) throws IOException { boolean isOk = in.readBoolean(); if (!isOk) return deserializeNack(CommandSerializers.nullableBallot.deserialize(in, version), in, version); - return deserializeOk(CommandSerializers.txnId.deserialize(in, version), - CommandSerializers.status.deserialize(in, version), + TxnId id = CommandSerializers.txnId.deserialize(in, version); + Status status = CommandSerializers.status.deserialize(in, version); + + Result result = null; + if (status == Status.PreApplied || status == Status.Applied || status == Status.Truncated) + result = Result.APPLIED; + else if (status == Status.Invalidated) + result = Result.INVALIDATED; + + return deserializeOk(id, + status, CommandSerializers.ballot.deserialize(in, version), deserializeNullable(in, version, CommandSerializers.timestamp), DepsSerializer.partialDeps.deserialize(in, version), @@ -136,7 +143,7 @@ public class RecoverySerializers DepsSerializer.deps.deserialize(in, version), in.readBoolean(), deserializeNullable(in, version, CommandSerializers.writes), - deserializeNullable(in, version, TxnData.serializer), + result, in, version); } @@ -158,12 +165,11 @@ public class RecoverySerializers size += DepsSerializer.deps.serializedSize(recoverOk.earlierAcceptedNoWitness, version); size += TypeSizes.sizeof(recoverOk.rejectsFastPath); size += serializedNullableSize(recoverOk.writes, version, CommandSerializers.writes); - size += serializedNullableSize((TxnData) recoverOk.result, version, TxnData.serializer); return size; } @Override - public final long serializedSize(RecoverReply reply, int version) + public long serializedSize(RecoverReply reply, int version) { return TypeSizes.sizeof(reply.isOk()) + (reply.isOk() ? serializedOkSize((RecoverOk) reply, version) : serializedNackSize((RecoverNack) reply, version)); diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index a35ae36020..af0b6b771e 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -33,6 +33,7 @@ import accord.api.Update; import accord.api.Write; import accord.primitives.Keys; import accord.primitives.Ranges; +import accord.primitives.RoutableKey; import accord.primitives.Timestamp; import org.apache.cassandra.cql3.QueryOptions; import org.apache.cassandra.db.TypeSizes; @@ -174,7 +175,7 @@ public class TxnUpdate implements Update public Write apply(Timestamp executeAt, Data data) { if (!checkCondition(data)) - return TxnWrite.EMPTY; + return TxnWrite.EMPTY_CONDITION_FAILED; List fragments = deserialize(this.fragments, TxnWrite.Fragment.serializer); List updates = new ArrayList<>(fragments.size()); @@ -182,9 +183,23 @@ public class TxnUpdate implements Update AccordUpdateParameters parameters = new AccordUpdateParameters((TxnData) data, options); for (TxnWrite.Fragment fragment : fragments) - updates.add(fragment.complete(parameters)); + // Filter out fragments that already constitute complete updates to avoid persisting them via TxnWrite: + if (!fragment.isComplete()) + updates.add(fragment.complete(parameters)); - return new TxnWrite(updates); + return new TxnWrite(updates, true); + } + + public List completeUpdatesForKey(RoutableKey key) + { + List fragments = deserialize(this.fragments, TxnWrite.Fragment.serializer); + List updates = new ArrayList<>(fragments.size()); + + for (TxnWrite.Fragment fragment : fragments) + if (fragment.isComplete() && fragment.key.equals(key)) + updates.add(fragment.toUpdate()); + + return updates; } public static final IVersionedSerializer serializer = new IVersionedSerializer() diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java index 7c5794a66d..89599bc90e 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java @@ -27,16 +27,13 @@ import java.util.List; import java.util.Objects; import java.util.Set; +import accord.primitives.*; import com.google.common.base.Preconditions; import com.google.common.collect.Iterables; import accord.api.DataStore; import accord.api.Write; import accord.local.SafeCommandStore; -import accord.primitives.RoutableKey; -import accord.primitives.Seekable; -import accord.primitives.Timestamp; -import accord.primitives.Writes; import accord.utils.async.AsyncChain; import accord.utils.async.AsyncChains; import org.apache.cassandra.concurrent.Stage; @@ -56,6 +53,7 @@ import org.apache.cassandra.schema.ColumnMetadata; import org.apache.cassandra.service.accord.AccordSafeCommandsForKey; import org.apache.cassandra.service.accord.AccordSafeCommandStore; import org.apache.cassandra.service.accord.api.PartitionKey; +import org.apache.cassandra.utils.BooleanSerializer; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.ObjectSizes; @@ -66,9 +64,9 @@ import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize; public class TxnWrite extends AbstractKeySorted implements Write { - public static final TxnWrite EMPTY = new TxnWrite(Collections.emptyList()); + public static final TxnWrite EMPTY_CONDITION_FAILED = new TxnWrite(Collections.emptyList(), false); - private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY); + private static final long EMPTY_SIZE = ObjectSizes.measure(EMPTY_CONDITION_FAILED); public static class Update extends AbstractSerialized { @@ -218,10 +216,20 @@ public class TxnWrite extends AbstractKeySorted implements Writ return "Fragment{key=" + key + ", index=" + index + ", baseUpdate=" + baseUpdate + ", referenceOps=" + referenceOps + '}'; } + public boolean isComplete() + { + return referenceOps.isEmpty(); + } + + public Update toUpdate() + { + return new Update(key, index, baseUpdate); + } + public Update complete(AccordUpdateParameters parameters) { - if (referenceOps.isEmpty()) - return new Update(key, index, baseUpdate); + if (isComplete()) + return toUpdate(); DecoratedKey key = baseUpdate.partitionKey(); PartitionUpdate.Builder updateBuilder = new PartitionUpdate.Builder(baseUpdate.metadata(), @@ -314,14 +322,18 @@ public class TxnWrite extends AbstractKeySorted implements Writ }; } - private TxnWrite(Update[] items) + private final boolean isConditionMet; + + private TxnWrite(Update[] items, boolean isConditionMet) { super(items); + this.isConditionMet = isConditionMet; } - public TxnWrite(List items) + public TxnWrite(List items, boolean isConditionMet) { super(items); + this.isConditionMet = isConditionMet; } @Override @@ -343,7 +355,7 @@ public class TxnWrite extends AbstractKeySorted implements Writ } @Override - public AsyncChain apply(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store) + public AsyncChain apply(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store, PartialTxn txn) { // TODO (expected, efficiency): 99.9999% of the time we can just use executeAt.hlc(), so can avoid bringing // cfk into memory by retaining at all times in memory key ranges that are dirty and must use this logic; @@ -355,8 +367,22 @@ public class TxnWrite extends AbstractKeySorted implements Writ int nowInSeconds = cfk.nowInSecondsFor(executeAt, true); List> results = new ArrayList<>(); + + // Apply updates not specified fully by the client but built from fragments completed by data from reads. + // This occurs, for example, when an UPDATE statement uses a value assigned by a LET statement. forEachWithKey((PartitionKey) key, write -> results.add(write.write(timestamp, nowInSeconds))); + if (isConditionMet) + { + // Apply updates that are fully specified by the client and not reliant on data from reads. + // ex. INSERT INTO tbl (a, b, c) VALUES (1, 2, 3) + // These updates are persisted only in TxnUpdate and not in TxnWrite to avoid duplication. + TxnUpdate txnUpdate = (TxnUpdate) txn.update(); + assert txnUpdate != null : "PartialTxn should contain an update if we're applying a write!"; + List updates = txnUpdate.completeUpdatesForKey((RoutableKey) key); + updates.forEach(update -> results.add(update.write(timestamp, nowInSeconds))); + } + if (results.isEmpty()) return Writes.SUCCESS; @@ -379,19 +405,21 @@ public class TxnWrite extends AbstractKeySorted implements Writ @Override public void serialize(TxnWrite write, DataOutputPlus out, int version) throws IOException { + BooleanSerializer.serializer.serialize(write.isConditionMet, out, version); serializeArray(write.items, out, version, Update.serializer); } @Override public TxnWrite deserialize(DataInputPlus in, int version) throws IOException { - return new TxnWrite(deserializeArray(in, version, Update.serializer, Update[]::new)); + boolean isConditionMet = BooleanSerializer.serializer.deserialize(in, version); + return new TxnWrite(deserializeArray(in, version, Update.serializer, Update[]::new), isConditionMet); } @Override public long serializedSize(TxnWrite write, int version) { - return serializedArraySize(write.items, version, Update.serializer); + return BooleanSerializer.serializer.serializedSize(write.isConditionMet, version) + serializedArraySize(write.items, version, Update.serializer); } }; } diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java index 8dada5ba9e..8a515bb614 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java @@ -508,6 +508,32 @@ public class AccordCQLTest extends AccordTestBase }); } + @Test + public void testFailedConditionWithCompleteInsert() throws Throwable + { + test(cluster -> + { + cluster.coordinator(1).execute("INSERT INTO " + currentTable + " (k, c, v) VALUES (1, 0, 3);", ConsistencyLevel.ALL); + + String query = "BEGIN TRANSACTION\n" + + " LET row0 = (SELECT * FROM " + currentTable + " WHERE k = ? AND c = ?);\n" + + " LET row1 = (SELECT * FROM " + currentTable + " WHERE k = ? AND c = ?);\n" + + " SELECT row1.v;\n" + + " IF row0 IS NULL AND row1.v = ? THEN\n" + + " INSERT INTO " + currentTable + " (k, c, v) VALUES (?, ?, ?);\n" + + " END IF\n" + + "COMMIT TRANSACTION"; + SimpleQueryResult result = cluster.coordinator(1).executeWithResult(query, ConsistencyLevel.ANY, 0, 0, 1, 0, 2, 0, 0, 1); + assertEquals(ImmutableList.of("row1.v"), result.names()); + assertThat(result).hasSize(1).contains(3); + + String check = "BEGIN TRANSACTION\n" + + " SELECT * FROM " + currentTable + " WHERE k=0 AND c=0;\n" + + "COMMIT TRANSACTION"; + assertEmptyWithPreemptedRetry(cluster, check); + }); + } + @Test public void testReversedClusteringReference() throws Exception { diff --git a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java index d128ead489..ba9e1b801c 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java @@ -28,7 +28,6 @@ import org.apache.cassandra.distributed.impl.Instance; import org.apache.cassandra.net.Message; import org.apache.cassandra.net.Verb; -@SuppressWarnings("Convert2MethodRef") public class AccordIntegrationTest extends AccordTestBase { private static final Logger logger = LoggerFactory.getLogger(AccordIntegrationTest.class); diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index c51e161d4d..2f9328b7fc 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -85,7 +85,6 @@ import org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns; import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyRows; import org.apache.cassandra.service.accord.AccordTestUtils; import org.apache.cassandra.service.accord.IAccordService; -import org.apache.cassandra.service.accord.txn.TxnData; import org.apache.cassandra.utils.FBUtilities; import org.apache.cassandra.utils.Pair; @@ -129,7 +128,7 @@ public class CompactionAccordIteratorsTest static ColumnFamilyStore commands; static ColumnFamilyStore commandsForKey; static TableMetadata table; - static FullRoute route; + static FullRoute route; Random random; /* @@ -304,7 +303,6 @@ public class CompactionAccordIteratorsTest for (ColumnMetadata cm : CommandsColumns.TRUNCATE_FIELDS[1]) assertNotNull(row.getColumnData(cm)); assertEquals(TXN_ID, CommandRows.getTxnId(partitionKeyComponents)); - assertEquals(1, ((TxnData)CommandRows.getResult(row)).entrySet().size()); assertNotNull(CommandRows.getWrites(row)); assertEquals(Durability.Local, CommandRows.getDurability(row)); assertEquals(TXN_ID, CommandRows.getExecuteAt(row)); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java index f5148fa486..021cbd4897 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java @@ -121,7 +121,7 @@ public class AccordCommandStoreTest attrs.addListener(new Command.ProxyListener(oldTxnId1)); Pair result = AccordTestUtils.processTxnResult(commandStore, txnId, txn, executeAt); Command command = Command.SerializerSupport.executed(attrs, SaveStatus.Applied, executeAt, promised, accepted, - waitingOn, result.left, result.right); + waitingOn, result.left, Result.APPLIED); AccordSafeCommand safeCommand = new AccordSafeCommand(loaded(txnId, null)); safeCommand.set(command); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java index 5115dfdc47..73830bda44 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java @@ -100,7 +100,7 @@ public class AccordSyncPropagatorTest RandomDelayQueue delayQueue = new RandomDelayQueue.Factory(rs).get(); PendingQueue queue = new PropagatingPendingQueue(failures, delayQueue); Agent agent = new TestAgent.RethrowAgent(); - SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, agent, rs.fork()); + SimulatedDelayedExecutorService globalExecutor = new SimulatedDelayedExecutorService(queue, agent); ScheduledExecutorPlus scheduler = new AdaptingScheduledExecutorPlus(globalExecutor); Cluster cluster = new Cluster(nodes, rs, scheduler); @@ -361,7 +361,7 @@ public class AccordSyncPropagatorTest { if (self.equals(ep)) return true; - return !nodeRuns.computeIfAbsent(ep, ignore -> Gens.bools().runs(.01)).next(rs); + return !nodeRuns.computeIfAbsent(ep, ignore -> Gens.bools().biasedRepeatingRuns(.01)).next(rs); } @Override @@ -505,4 +505,4 @@ public class AccordSyncPropagatorTest } } } -} \ No newline at end of file +}