natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

- 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
This commit is contained in:
Caleb Rackliffe 2023-09-20 11:25:51 -05:00 committed by David Capwell
parent cfc63edcb4
commit 08428a2750
13 changed files with 144 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit 1d6028ca20553d1c1a6fe2809b204254955da3b3
Subproject commit df492dfd2ffe993c33761d0531ac5b979b80f080

62
src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
View File

@ -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<blob>, "
+ "PRIMARY KEY((store_id, domain, txn_id))"
@ -267,7 +265,6 @@ public class AccordKeyspace
static final LocalVersionedSerializer<PartialTxn> partialTxn = localSerializer(CommandSerializers.partialTxn);
static final LocalVersionedSerializer<PartialDeps> partialDeps = localSerializer(DepsSerializer.partialDeps);
static final LocalVersionedSerializer<Writes> writes = localSerializer(CommandSerializers.writes);
static final LocalVersionedSerializer<TxnData> result = localSerializer(TxnData.serializer);
static final LocalVersionedSerializer<Command.DurableAndIdempotentListener> listeners = localSerializer(ListenerSerializers.listener);
static final LocalVersionedSerializer<Topology> topology = localSerializer(TopologySerializers.topology);
static final LocalVersionedSerializer<ReducingRangeMap<Timestamp>> 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:

10
src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java
View File

@ -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;

8
src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java
View File

@ -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);
}
};

12
src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java
View File

@ -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;
}
};

24
src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java
View File

@ -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));

21
src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
View File

@ -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<TxnWrite.Fragment> fragments = deserialize(this.fragments, TxnWrite.Fragment.serializer);
List<TxnWrite.Update> 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<TxnWrite.Update> completeUpdatesForKey(RoutableKey key)
{
List<TxnWrite.Fragment> fragments = deserialize(this.fragments, TxnWrite.Fragment.serializer);
List<TxnWrite.Update> 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<TxnUpdate> serializer = new IVersionedSerializer<TxnUpdate>()

54
src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java
View File

@ -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<TxnWrite.Update> 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<PartitionUpdate>
{
@ -218,10 +216,20 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> 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<TxnWrite.Update> implements Writ
};
}
private TxnWrite(Update[] items)
private final boolean isConditionMet;
private TxnWrite(Update[] items, boolean isConditionMet)
{
super(items);
this.isConditionMet = isConditionMet;
}
public TxnWrite(List<Update> items)
public TxnWrite(List<Update> items, boolean isConditionMet)
{
super(items);
this.isConditionMet = isConditionMet;
}
@Override
@ -343,7 +355,7 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
}
@Override
public AsyncChain<Void> apply(Seekable key, SafeCommandStore safeStore, Timestamp executeAt, DataStore store)
public AsyncChain<Void> 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<TxnWrite.Update> implements Writ
int nowInSeconds = cfk.nowInSecondsFor(executeAt, true);
List<AsyncChain<Void>> 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<Update> 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<TxnWrite.Update> 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);
}
};
}

26
test/distributed/org/apache/cassandra/distributed/test/accord/AccordCQLTest.java
View File

@ -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
{

1
test/distributed/org/apache/cassandra/distributed/test/accord/AccordIntegrationTest.java
View File

@ -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);

4
test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
View File

@ -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));

2
test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
View File

@ -121,7 +121,7 @@ public class AccordCommandStoreTest
attrs.addListener(new Command.ProxyListener(oldTxnId1));
Pair<Writes, Result> 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);

6
test/unit/org/apache/cassandra/service/accord/AccordSyncPropagatorTest.java
View File

@ -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
}
}
}
}
}