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

Remove TimestampsForKey 

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20174
This commit is contained in:
Benedict Elliott Smith 2024-11-28 16:39:56 +00:00 committed by David Capwell
parent 44cd181438
commit 52242f23a8
22 changed files with 133 additions and 747 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit f543851dffbf2907580f91608be3c1aadc4ccb95
Subproject commit 520cc1072d44a5f7617566b6667e915532b89033

89
src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
View File

@ -45,7 +45,6 @@ import accord.local.StoreParticipants;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Status.Durability;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.agrona.collections.Int2ObjectHashMap;
@ -99,7 +98,6 @@ import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordKeyspace.CommandRows;
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns;
import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor;
import org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyRows;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.JournalKey;
@ -121,14 +119,9 @@ import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.exp
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.saveStatusOnly;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.truncatedApply;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeysAccessor;
import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyColumns.last_executed_micros;
import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyColumns.last_executed_timestamp;
import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyColumns.last_write_timestamp;
import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyRows.truncateTimestampsForKeyRow;
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeDurabilityOrNull;
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeParticipantsOrNull;
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeSaveStatusOrNull;
import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeTimestampOrNull;
/**
* Merge multiple iterators over the content of sstable into a "compacted" iterator.
@ -241,8 +234,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
if (isAccordCommands(cfs))
return new AccordCommandsPurger(accordService);
if (isAccordTimestampsForKey(cfs))
return new AccordTimestampsForKeyPurger(accordService);
if (isAccordJournal(cfs))
return new AccordJournalPurger(accordService, cfs);
if (isAccordCommandsForKey(cfs))
@ -897,80 +888,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
}
class AccordTimestampsForKeyPurger extends AbstractPurger
{
final Int2ObjectHashMap<RedundantBefore> redundantBefores;
int storeId;
TokenKey partitionKey;
AccordTimestampsForKeyPurger(Supplier<IAccordService> accordService)
{
this.redundantBefores = accordService.get().getCompactionInfo().redundantBefores;
}
protected void beginPartition(UnfilteredRowIterator partition)
{
ByteBuffer[] partitionKeyComponents = TimestampsForKeyRows.splitPartitionKey(partition.partitionKey());
storeId = TimestampsForKeyRows.getStoreId(partitionKeyComponents);
partitionKey = TimestampsForKeyRows.getKey(partitionKeyComponents);
}
@Override
protected Row applyToRow(Row row)
{
updateProgress();
RedundantBefore redundantBefore = redundantBefores.get(storeId);
// TODO (expected): if the store has been retired, this should return null
if (redundantBefore == null)
return row;
RedundantBefore.Entry redundantBeforeEntry = redundantBefore.get(partitionKey.toUnseekable());
if (redundantBeforeEntry == null)
return row;
TxnId redundantBeforeTxnId = redundantBeforeEntry.shardRedundantBefore();
Cell lastExecuteMicrosCell = row.getCell(last_executed_micros);
Long last_execute_micros = null;
if (lastExecuteMicrosCell != null && !lastExecuteMicrosCell.accessor().isEmpty(lastExecuteMicrosCell.value()))
last_execute_micros = lastExecuteMicrosCell.accessor().getLong(lastExecuteMicrosCell.value(), 0);
if (last_execute_micros != null && last_execute_micros < redundantBeforeTxnId.hlc())
{
lastExecuteMicrosCell = null;
}
Cell lastExecuteCell = row.getCell(last_executed_timestamp);
Timestamp last_execute = deserializeTimestampOrNull(lastExecuteCell);
if (last_execute != null && last_execute.compareTo(redundantBeforeTxnId) < 0)
{
lastExecuteCell = null;
}
Cell lastWriteCell = row.getCell(last_write_timestamp);
Timestamp last_write = deserializeTimestampOrNull(lastWriteCell);
if (last_write != null && last_write.compareTo(redundantBeforeTxnId) < 0)
{
lastWriteCell = null;
}
// No need to emit a tombstone as earlier versions of the row will also be nulled out
// when compacted later or loaded into a commands for key
if (lastExecuteMicrosCell == null &&
lastExecuteCell == null &&
lastWriteCell == null)
return null;
return truncateTimestampsForKeyRow(nowInSec, row, lastExecuteMicrosCell, lastExecuteCell, lastWriteCell);
}
@Override
protected Row applyToStatic(Row row)
{
checkState(row.isStatic() && row.isEmpty());
return row;
}
}
class AccordCommandsForKeyPurger extends AbstractPurger
{
final CommandsForKeyAccessor accessor;
@ -1226,7 +1143,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
{
return cfs.getKeyspaceName().equals(SchemaConstants.ACCORD_KEYSPACE_NAME) &&
ImmutableSet.of(AccordKeyspace.COMMANDS,
AccordKeyspace.TIMESTAMPS_FOR_KEY,
AccordKeyspace.JOURNAL,
AccordKeyspace.COMMANDS_FOR_KEY)
.contains(cfs.getTableName());
@ -1247,11 +1163,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
return isAccordTable(cfs, AccordKeyspace.COMMANDS);
}
private static boolean isAccordTimestampsForKey(ColumnFamilyStore cfs)
{
return isAccordTable(cfs, AccordKeyspace.TIMESTAMPS_FOR_KEY);
}
private static boolean isAccordCommandsForKey(ColumnFamilyStore cfs)
{
return isAccordTable(cfs, AccordKeyspace.COMMANDS_FOR_KEY);

1
src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
View File

@ -216,7 +216,6 @@ public class AccordDebugKeyspace extends VirtualKeyspace
{
addRow(ds, executor.executorId(), AccordKeyspace.COMMANDS, cache.commands.statsSnapshot());
addRow(ds, executor.executorId(), AccordKeyspace.COMMANDS_FOR_KEY, cache.commandsForKey.statsSnapshot());
addRow(ds, executor.executorId(), AccordKeyspace.TIMESTAMPS_FOR_KEY, cache.timestampsForKey.statsSnapshot());
}
}
return ds;

59
src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
View File

@ -40,7 +40,6 @@ import accord.api.ProgressLog;
import accord.api.RoutingKey;
import accord.impl.AbstractLoader;
import accord.impl.AbstractSafeCommandStore.CommandStoreCaches;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.CommandStores;
@ -86,14 +85,12 @@ public class AccordCommandStore extends CommandStore
{
private final AccordCache global;
private final AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands;
private final AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey>.Instance timestampsForKeys;
private final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeys;
Caches(AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commandCache, AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey>.Instance timestampsForKeyCache, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeyCache)
Caches(AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commandCache, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeyCache)
{
this.global = global;
this.commands = commandCache;
this.timestampsForKeys = timestampsForKeyCache;
this.commandsForKeys = commandsForKeyCache;
}
@ -107,24 +104,19 @@ public class AccordCommandStore extends CommandStore
return commands;
}
public final AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey>.Instance timestampsForKeys()
{
return timestampsForKeys;
}
public final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeys()
{
return commandsForKeys;
}
}
public static final class ExclusiveCaches extends Caches implements CommandStoreCaches<AccordSafeCommand, AccordSafeTimestampsForKey, AccordSafeCommandsForKey>
public static final class ExclusiveCaches extends Caches implements CommandStoreCaches<AccordSafeCommand, AccordSafeCommandsForKey>
{
private final Lock lock;
public ExclusiveCaches(Lock lock, AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands, AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey>.Instance timestampsForKeys, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeys)
public ExclusiveCaches(Lock lock, AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeys)
{
super(global, commands, timestampsForKeys, commandsForKeys);
super(global, commands, commandsForKeys);
this.lock = lock;
}
@ -141,12 +133,6 @@ public class AccordCommandStore extends CommandStore
return commandsForKeys().acquireIfLoaded(key);
}
@Override
public AccordSafeTimestampsForKey acquireTfkIfLoaded(RoutingKey key)
{
return timestampsForKeys().acquireIfLoaded(key);
}
@Override
public void close()
{
@ -183,14 +169,12 @@ public class AccordCommandStore extends CommandStore
this.executor = executor;
final AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands;
final AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey>.Instance timestampsForKey;
final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKey;
try (AccordExecutor.ExclusiveGlobalCaches exclusive = executor.lockCaches())
{
commands = exclusive.commands.newInstance(this);
timestampsForKey = exclusive.timestampsForKey.newInstance(this);
commandsForKey = exclusive.commandsForKey.newInstance(this);
this.caches = new ExclusiveCaches(executor.lock, exclusive.global, commands, timestampsForKey, commandsForKey);
this.caches = new ExclusiveCaches(executor.lock, exclusive.global, commands, commandsForKey);
}
this.taskExecutor = executor.executor(this);
@ -321,20 +305,6 @@ public class AccordCommandStore extends CommandStore
((AccordJournal) journal).sanityCheck(id, command);
}
boolean validateTimestampsForKey(RoutableKey key, TimestampsForKey evicting)
{
if (!Invariants.isParanoid())
return true;
TimestampsForKey reloaded = AccordKeyspace.unsafeLoadTimestampsForKey(id, (TokenKey) key);
return Objects.equals(evicting, reloaded);
}
TimestampsForKey loadTimestampsForKey(RoutableKey key)
{
return AccordKeyspace.loadTimestampsForKey(id, (TokenKey) key);
}
CommandsForKey loadCommandsForKey(RoutableKey key)
{
return AccordKeyspace.loadCommandsForKey(id, (TokenKey) key);
@ -349,12 +319,6 @@ public class AccordCommandStore extends CommandStore
return Objects.equals(evicting, reloaded);
}
@Nullable
Runnable saveTimestampsForKey(RoutingKey key, TimestampsForKey after, Object serialized)
{
return AccordKeyspace.getTimestampsForKeyUpdater(this, after, nextSystemTimestampMicros());
}
@Nullable
Runnable saveCommandsForKey(RoutingKey key, CommandsForKey after, Object serialized)
{
@ -573,13 +537,12 @@ public class AccordCommandStore extends CommandStore
@Override
public void apply(Command command, OnDone onDone)
{
PreLoadContext context = context(command, KeyHistory.TIMESTAMPS);
store.execute(context,
safeStore -> {
applyWrites(command.txnId(), safeStore, (safeCommand, cmd) -> {
Commands.applyWrites(safeStore, context, cmd).begin(store.agent);
});
})
PreLoadContext context = context(command, SYNC);
store.execute(context, safeStore -> {
applyWrites(command.txnId(), safeStore, (safeCommand, cmd) -> {
Commands.applyWrites(safeStore, context, cmd).begin(store.agent);
});
})
.begin((unused, throwable) -> {
if (throwable != null)
onDone.failure(throwable);

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

@ -28,7 +28,6 @@ import java.util.function.IntFunction;
import accord.api.Agent;
import accord.api.RoutingKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.cfk.CommandsForKey;
import accord.primitives.TxnId;
@ -77,9 +76,9 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo
{
final Lock lock;
public ExclusiveGlobalCaches(Lock lock, AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand> commands, AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKey, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKey)
public ExclusiveGlobalCaches(Lock lock, AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand> commands, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKey)
{
super(global, commands, timestampsForKey, commandsForKey);
super(global, commands, commandsForKey);
this.lock = lock;
}
@ -94,14 +93,12 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo
{
public final AccordCache global;
public final AccordCache.Type<TxnId, Command, AccordSafeCommand> commands;
public final AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKey;
public final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKey;
public GlobalCaches(AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand> commands, AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKey, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKey)
public GlobalCaches(AccordCache global, AccordCache.Type<TxnId, Command, AccordSafeCommand> commands, AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKey)
{
this.global = global;
this.commands = commands;
this.timestampsForKey = timestampsForKey;
this.commandsForKey = commandsForKey;
}
}
@ -153,22 +150,14 @@ public abstract class AccordExecutor implements CacheSize, AccordCacheEntry.OnLo
this.agent = agent;
final AccordCache.Type<TxnId, Command, AccordSafeCommand> commands;
final AccordCache.Type<RoutingKey, TimestampsForKey, AccordSafeTimestampsForKey> timestampsForKey;
final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey> commandsForKey;
commands = cache.newType(TxnId.class, COMMAND_ADAPTER);
registerJfrListener(executorId, commands, "Command");
timestampsForKey = cache.newType(RoutingKey.class,
AccordCommandStore::loadTimestampsForKey,
AccordCommandStore::saveTimestampsForKey,
Function.identity(),
AccordCommandStore::validateTimestampsForKey,
AccordObjectSizes::timestampsForKey,
AccordSafeTimestampsForKey::new);
registerJfrListener(executorId, timestampsForKey, "TimestampsForKey");
commandsForKey = cache.newType(RoutingKey.class, CFK_ADAPTER);
registerJfrListener(executorId, commandsForKey, "CommandsForKey");
this.caches = new ExclusiveGlobalCaches(lock, cache, commands, timestampsForKey, commandsForKey);
this.caches = new ExclusiveGlobalCaches(lock, cache, commands, commandsForKey);
ScheduledExecutors.scheduledFastTasks.scheduleAtFixedRate(() -> {
executeDirectlyWithLock(cache::processNoEvictQueue);
}, 1L, 1L, TimeUnit.SECONDS);

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

@ -41,8 +41,6 @@ import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.RoutingKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.CommandStore;
import accord.local.Node;
@ -168,12 +166,11 @@ public class AccordKeyspace
public static final String JOURNAL = "journal";
public static final String COMMANDS = "commands";
public static final String TIMESTAMPS_FOR_KEY = "timestamps_for_key";
public static final String COMMANDS_FOR_KEY = "commands_for_key";
public static final String TOPOLOGIES = "topologies";
public static final String EPOCH_METADATA = "epoch_metadata";
public static final Set<String> TABLE_NAMES = ImmutableSet.of(COMMANDS, TIMESTAMPS_FOR_KEY, COMMANDS_FOR_KEY,
public static final Set<String> TABLE_NAMES = ImmutableSet.of(COMMANDS, COMMANDS_FOR_KEY,
TOPOLOGIES, EPOCH_METADATA,
JOURNAL);
@ -473,131 +470,6 @@ public class AccordKeyspace
}
}
public static final TableMetadata TimestampsForKeys =
parse(TIMESTAMPS_FOR_KEY,
"accord timestamps per key",
"CREATE TABLE %s ("
+ "store_id int, "
+ "routing_key blob, " // can't use "token" as this is restricted word in CQL
+ format("last_executed_timestamp %s, ", TIMESTAMP_TUPLE)
+ "last_executed_micros bigint, "
+ format("last_write_id %s, ", TIMESTAMP_TUPLE)
+ format("last_write_timestamp %s, ", TIMESTAMP_TUPLE)
+ "PRIMARY KEY((store_id, routing_key))"
+ ')')
.partitioner(new LocalCompositePrefixPartitioner(Int32Type.instance, BytesType.instance))
.build();
public static class TimestampsForKeyColumns
{
static final ClusteringComparator keyComparator = TimestampsForKeys.partitionKeyAsClusteringComparator();
static final CompositeType partitionKeyType = (CompositeType) TimestampsForKeys.partitionKeyType;
static final ColumnMetadata store_id = getColumn(TimestampsForKeys, "store_id");
static final ColumnMetadata routing_key = getColumn(TimestampsForKeys, "routing_key");
public static final ColumnMetadata last_executed_timestamp = getColumn(TimestampsForKeys, "last_executed_timestamp");
public static final ColumnMetadata last_executed_micros = getColumn(TimestampsForKeys, "last_executed_micros");
public static final ColumnMetadata last_write_timestamp = getColumn(TimestampsForKeys, "last_write_timestamp");
public static final ColumnMetadata last_write_id = getColumn(TimestampsForKeys, "last_write_id");
static final Columns columns = Columns.from(Lists.newArrayList(last_executed_timestamp, last_executed_micros, last_write_timestamp));
static final ColumnFilter allColumns = ColumnFilter.all(TimestampsForKeys);
static DecoratedKey decorateKey(int storeId, RoutingKey key)
{
return TimestampsForKeys.partitioner.decorateKey(makeKey(storeId, key));
}
static ByteBuffer makeKey(int storeId, RoutingKey key)
{
TokenKey pk = (TokenKey) key;
return keyComparator.make(storeId, serializeRoutingKey(pk)).serializeAsPartitionKey();
}
static ByteBuffer makeKey(int storeId, TimestampsForKey timestamps)
{
return makeKey(storeId, timestamps.key());
}
}
public static class TimestampsForKeyRows extends TimestampsForKeyColumns
{
public static ByteBuffer[] splitPartitionKey(DecoratedKey key)
{
return partitionKeyType.split(key.getKey());
}
public static int getStoreId(ByteBuffer[] partitionKeyComponents)
{
return Int32Type.instance.compose(partitionKeyComponents[store_id.position()]);
}
public static TokenKey getKey(DecoratedKey key)
{
return getKey(splitPartitionKey(key));
}
public static TokenKey getKey(ByteBuffer[] partitionKeyComponents)
{
return (TokenKey) deserializeRoutingKey(partitionKeyComponents[routing_key.position()]);
}
@Nullable
public static Timestamp getLastExecutedTimestamp(Row row)
{
Cell cell = row.getCell(last_executed_timestamp);
if (cell == null)
return null;
return deserializeTimestampOrNull(cell.value(), cell.accessor(), Timestamp::fromBits);
}
public static long getLastExecutedMicros(Row row)
{
Cell cell = row.getCell(last_executed_micros);
if (cell == null || cell.accessor().isEmpty(cell.value()))
return Long.MIN_VALUE;
return cell.accessor().getLong(cell.value(), 0);
}
@Nullable
public static Timestamp getLastWriteTimestamp(Row row)
{
Cell cell = row.getCell(last_write_timestamp);
if (cell == null)
return null;
return deserializeTimestampOrNull(cell.value(), cell.accessor(), Timestamp::fromBits);
}
public static Row truncateTimestampsForKeyRow(long nowInSec, Row row, Cell lastExecuteMicrosCell, Cell lastExecuteCell, Cell lastWriteCell)
{
checkArgument(lastExecuteMicrosCell == null || lastExecuteMicrosCell.column() == last_executed_micros);
checkArgument(lastExecuteCell == null || lastExecuteCell.column() == last_executed_timestamp);
checkArgument(lastWriteCell == null || lastWriteCell.column() == last_write_timestamp);
long timestamp = row.primaryKeyLivenessInfo().timestamp();
int colCount = 0;
if (lastExecuteMicrosCell != null)
colCount++;
if (lastExecuteCell != null)
colCount++;
if (lastWriteCell != null)
colCount++;
checkState(columns.size() >= colCount, "CommandsForKeyColumns.static_columns_metadata should include all the columns");
Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(colCount);
int colIndex = 0;
if (lastExecuteMicrosCell != null)
newLeaf[colIndex++] = lastExecuteMicrosCell;
if (lastExecuteCell != null)
newLeaf[colIndex++] = lastExecuteCell;
if (lastWriteCell != null)
newLeaf[colIndex] = lastWriteCell;
return BTreeRow.create(row.clustering(), LivenessInfo.create(timestamp, nowInSec),
Deletion.LIVE, newLeaf);
}
}
private static final LocalCompositePrefixPartitioner CFKPartitioner = new LocalCompositePrefixPartitioner(Int32Type.instance, UUIDType.instance, BytesType.instance);
public static final TableMetadata CommandsForKeys = commandsForKeysTable(COMMANDS_FOR_KEY);
@ -765,16 +637,16 @@ public class AccordKeyspace
return KeyspaceMetadata.create(ACCORD_KEYSPACE_NAME, KeyspaceParams.local(), tables(), Views.none(), Types.none(), UserFunctions.none());
}
public static Tables tables = Tables.of(Commands, TimestampsForKeys, CommandsForKeys, Topologies, EpochMetadata, Journal);
public static Tables TABLES = Tables.of(Commands, CommandsForKeys, Topologies, EpochMetadata, Journal);
public static Tables tables()
{
return tables;
return TABLES;
}
public static void truncateAllCaches()
{
Keyspace ks = Keyspace.open(ACCORD_KEYSPACE_NAME);
for (String table : new String[]{ TimestampsForKeys.name, CommandsForKeys.name })
for (String table : new String[]{ CommandsForKeys.name })
ks.getColumnFamilyStore(table).truncateBlocking();
}
@ -1100,105 +972,6 @@ public class AccordKeyspace
return (TokenKey) AccordRoutingKeyByteSource.Serializer.fromComparableBytes(ByteBufferAccessor.instance, tokenBytes, tableId, currentVersion, null);
}
public static PartitionUpdate getTimestampsForKeyUpdate(int storeId, TimestampsForKey current, long timestampMicros)
{
try
{
// TODO: convert to byte arrays
ValueAccessor<ByteBuffer> accessor = ByteBufferAccessor.instance;
Row.Builder builder = BTreeRow.unsortedBuilder();
builder.newRow(Clustering.EMPTY);
int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(timestampMicros);
LivenessInfo livenessInfo = LivenessInfo.create(timestampMicros, nowInSeconds);
builder.addPrimaryKeyLivenessInfo(livenessInfo);
addCell(TimestampsForKeyColumns.last_executed_timestamp, TimestampsForKey::lastExecutedTimestamp, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, current);
addCell(TimestampsForKeyColumns.last_executed_micros, TimestampsForKey::rawLastExecutedHlc, accessor::valueOf, builder, timestampMicros, nowInSeconds, current);
addCell(TimestampsForKeyColumns.last_write_timestamp, TimestampsForKey::lastWriteTimestamp, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, current);
Row row = builder.build();
if (row.columnCount() == 0)
return null;
ByteBuffer key = TimestampsForKeyColumns.makeKey(storeId, current.key());
return singleRowUpdate(TimestampsForKeys, key, row);
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
public static Runnable getTimestampsForKeyUpdater(AccordCommandStore commandStore, TimestampsForKey liveTimestamps, long timestampMicros)
{
PartitionUpdate upd = getTimestampsForKeyUpdate(commandStore.id(), liveTimestamps, timestampMicros);
return () -> {
ColumnFamilyStore cfs = AccordColumnFamilyStores.timestampsForKey;
try (OpOrder.Group group = Keyspace.writeOrder.start())
{
cfs.getCurrentMemtable().put(upd, UpdateTransaction.NO_OP, group, true);
}
};
}
public static UntypedResultSet loadTimestampsForKeyRow(int commandStoreId, TokenKey key)
{
String cql = "SELECT * FROM " + ACCORD_KEYSPACE_NAME + '.' + TIMESTAMPS_FOR_KEY + ' ' +
"WHERE store_id = ? " +
"AND routing_key = ?";
return executeInternal(cql, commandStoreId, serializeRoutingKey(key));
}
private static SinglePartitionReadCommand getTimestampsForKeyRead(int storeId, TokenKey key, long nowInSeconds)
{
return SinglePartitionReadCommand.create(TimestampsForKeys, nowInSeconds,
TimestampsForKeyColumns.allColumns,
RowFilter.none(),
DataLimits.NONE,
TimestampsForKeyColumns.decorateKey(storeId, key),
FULL_PARTITION);
}
public static TimestampsForKey loadTimestampsForKey(int commandStoreId, TokenKey key)
{
return unsafeLoadTimestampsForKey(commandStoreId, key);
}
public static TimestampsForKey unsafeLoadTimestampsForKey(int commandStoreId, TokenKey key)
{
long timestampMicros = TimeUnit.MILLISECONDS.toMicros(Global.currentTimeMillis());
int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(timestampMicros);
SinglePartitionReadCommand command = getTimestampsForKeyRead(commandStoreId, key, nowInSeconds);
try (ReadExecutionController controller = command.executionController();
FilteredPartitions partitions = FilteredPartitions.filter(command.executeLocally(controller), nowInSeconds))
{
if (!partitions.hasNext())
return null;
try (RowIterator partition = partitions.next())
{
Invariants.checkState(partition.hasNext());
Row row = partition.next();
TokenKey checkKey = TimestampsForKeyRows.getKey(partition.partitionKey());
checkState(checkKey.equals(key));
Timestamp lastExecutedTimestamp = deserializeTimestampOrDefault(cellValue(row, TimestampsForKeyColumns.last_executed_timestamp), ByteBufferAccessor.instance, Timestamp::fromBits, Timestamp.NONE);
ByteBuffer lastExecutedMicrosBB = cellValue(row, TimestampsForKeyColumns.last_executed_micros);
long lastExecutedMicros = lastExecutedMicrosBB == null || !lastExecutedMicrosBB.hasRemaining() ? 0 : lastExecutedMicrosBB.getLong(lastExecutedMicrosBB.position());
TxnId lastWriteId = deserializeTimestampOrDefault(cellValue(row, TimestampsForKeyColumns.last_write_id), ByteBufferAccessor.instance, TxnId::fromBits, TxnId.NONE);
Timestamp lastWriteTimestamp = deserializeTimestampOrDefault(cellValue(row, TimestampsForKeyColumns.last_write_timestamp), ByteBufferAccessor.instance, Timestamp::fromBits, Timestamp.NONE);
return TimestampsForKey.SerializerSupport.create(key, lastExecutedTimestamp, lastExecutedMicros, lastWriteId, lastWriteTimestamp);
}
}
catch (Throwable t)
{
logger.error("Exception loading AccordTimestampsForKey " + key, t);
throw t;
}
}
private static DecoratedKey makeKeySeparateTable(CommandsForKeyAccessor accessor, int commandStoreId, TokenKey key)
{
ByteBuffer pk = accessor.keyComparator.make(commandStoreId,
@ -1670,6 +1443,5 @@ public class AccordKeyspace
{
public static final ColumnFamilyStore journal = Schema.instance.getColumnFamilyStoreInstance(Journal.id);
public static final ColumnFamilyStore commandsForKey = Schema.instance.getColumnFamilyStoreInstance(CommandsForKeys.id);
public static final ColumnFamilyStore timestampsForKey = Schema.instance.getColumnFamilyStoreInstance(TimestampsForKeys.id);
}
}

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

@ -24,7 +24,6 @@ import java.util.function.ToLongFunction;
import accord.api.Key;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.impl.TimestampsForKey;
import accord.local.Command;
import accord.local.Command.WaitingOn;
import accord.local.CommonAttributes;
@ -358,19 +357,9 @@ public class AccordObjectSizes
return size;
}
private static long EMPTY_TFK_SIZE = measure(TimestampsForKey.SerializerSupport.create(null, null, 0, null, null));
public static long timestampsForKey(TimestampsForKey timestamps)
{
long size = EMPTY_TFK_SIZE;
size += timestamp(timestamps.lastExecutedTimestamp());
size += timestamp(timestamps.lastWriteTimestamp());
return size;
}
private static long EMPTY_CFK_SIZE = measure(new CommandsForKey(null));
private static long EMPTY_INFO_SIZE = measure(CommandsForKey.NO_INFO);
private static long EMPTY_INFO_EXTRA_ADDITIONAL_SIZE = measure(TxnInfo.create(TxnId.NONE, ACCEPTED, false, false, TxnId.NONE, NO_TXNIDS, Ballot.MAX)) - EMPTY_INFO_SIZE;
private static long EMPTY_INFO_EXTRA_ADDITIONAL_SIZE = measure(TxnInfo.create(TxnId.NONE, ACCEPTED, false, TxnId.NONE, NO_TXNIDS, Ballot.MAX)) - EMPTY_INFO_SIZE;
public static long commandsForKey(CommandsForKey cfk)
{
long size = EMPTY_CFK_SIZE;

26
src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java
View File

@ -43,7 +43,7 @@ import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
import static accord.utils.Invariants.illegalState;
public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeCommand, AccordSafeTimestampsForKey, AccordSafeCommandsForKey, AccordCommandStore.ExclusiveCaches>
public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeCommand, AccordSafeCommandsForKey, AccordCommandStore.ExclusiveCaches>
{
final AccordTask<?> task;
private final @Nullable CommandsForRanges commandsForRanges;
@ -130,21 +130,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
}
}
protected AccordSafeTimestampsForKey add(AccordSafeTimestampsForKey safeTfk, ExclusiveCaches caches)
{
Object check = task.ensureTimestampsForKey().putIfAbsent(safeTfk.key(), safeTfk);
if (check == null)
{
safeTfk.preExecute();
return safeTfk;
}
else
{
caches.timestampsForKeys().release(safeTfk, task);
throw illegalState("Attempted to take a duplicate reference to CFK for %s", safeTfk.key());
}
}
@Override
protected AccordSafeCommandsForKey getInternal(RoutingKey key)
{
@ -154,15 +139,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
return commandsForKey.get(key);
}
protected AccordSafeTimestampsForKey timestampsForKeyInternal(RoutingKey key)
{
Map<RoutingKey, AccordSafeTimestampsForKey> timestampsForKey = task.timestampsForKey();
if (timestampsForKey == null)
return null;
return timestampsForKey.get(key);
}
@Override
public AccordCommandStore commandStore()
{

140
src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java
View File

@ -1,140 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
import accord.api.RoutingKey;
import accord.impl.SafeTimestampsForKey;
import accord.impl.TimestampsForKey;
import accord.primitives.Timestamp;
public class AccordSafeTimestampsForKey extends SafeTimestampsForKey implements AccordSafeState<RoutingKey, TimestampsForKey>
{
private boolean invalidated;
private final AccordCacheEntry<RoutingKey, TimestampsForKey> global;
private TimestampsForKey original;
private TimestampsForKey current;
public AccordSafeTimestampsForKey(AccordCacheEntry<RoutingKey, TimestampsForKey> global)
{
super(global.key());
this.global = global;
this.original = null;
this.current = null;
}
@Override
public boolean equals(Object o)
{
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
AccordSafeTimestampsForKey that = (AccordSafeTimestampsForKey) o;
return Objects.equals(original, that.original) && Objects.equals(current, that.current);
}
@Override
public int hashCode()
{
throw new UnsupportedOperationException();
}
@Override
public String toString()
{
return "AccordSafeTimestampsForKey{" +
"invalidated=" + invalidated +
", global=" + global +
", original=" + original +
", current=" + current +
'}';
}
@Override
public AccordCacheEntry<RoutingKey, TimestampsForKey> global()
{
checkNotInvalidated();
return global;
}
@Override
public TimestampsForKey current()
{
checkNotInvalidated();
return current;
}
@Override
@VisibleForTesting
public void set(TimestampsForKey cfk)
{
checkNotInvalidated();
this.current = cfk;
}
public TimestampsForKey original()
{
checkNotInvalidated();
return original;
}
@Override
public void preExecute()
{
checkNotInvalidated();
original = global.getExclusive();
current = original;
if (isUnset())
initialize();
}
@Override
public void invalidate()
{
invalidated = true;
}
@Override
public boolean invalidated()
{
return invalidated;
}
public long lastExecutedMicros()
{
return current().lastExecutedHlc();
}
public static long timestampMicrosFor(TimestampsForKey timestamps, Timestamp executeAt, boolean isForWriteTxn)
{
return timestamps.hlcFor(executeAt, isForWriteTxn);
}
public static int nowInSecondsFor(TimestampsForKey timestamps, Timestamp executeAt, boolean isForWriteTxn)
{
timestamps.validateExecuteAtTime(executeAt, isForWriteTxn);
// we use the executeAt time instead of the monotonic database timestamp to prevent uneven
// ttl expiration in extreme cases, ie 1M+ writes/second to a key causing timestamps to overflow
// into the next second on some keys and not others.
return Math.toIntExact(TimeUnit.MICROSECONDS.toSeconds(timestamps.lastExecutedTimestamp().hlc()));
}
}

3
src/java/org/apache/cassandra/service/accord/AccordService.java
View File

@ -72,7 +72,6 @@ import accord.impl.DefaultRemoteListeners;
import accord.impl.DurabilityScheduling;
import accord.impl.RequestCallbacks;
import accord.impl.SizeOfIntersectionSorter;
import accord.impl.TimestampsForKey;
import accord.impl.progresslog.DefaultProgressLogs;
import accord.local.Command;
import accord.local.CommandStore;
@ -271,7 +270,6 @@ public class AccordService implements IAccordService, Shutdownable
private static void replayJournal(AccordService as)
{
logger.info("Starting journal replay.");
TimestampsForKey.unsafeSetReplay(true);
CommandsForKey.disableLinearizabilityViolationsReporting();
AccordKeyspace.truncateAllCaches();
@ -280,7 +278,6 @@ public class AccordService implements IAccordService, Shutdownable
logger.info("Waiting for command stores to quiesce.");
((AccordCommandStores)as.node.commandStores()).waitForQuiescense();
CommandsForKey.enableLinearizabilityViolationsReporting();
TimestampsForKey.unsafeSetReplay(false);
as.journal.unsafeSetStarted();
logger.info("Finished journal replay.");

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

@ -63,7 +63,6 @@ import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.utils.NoSpamLogger;
import org.apache.cassandra.utils.concurrent.Condition;
import static accord.local.KeyHistory.TIMESTAMPS;
import static accord.primitives.Routable.Domain.Key;
import static accord.primitives.Txn.Kind.EphemeralRead;
import static accord.utils.Invariants.illegalState;
@ -192,7 +191,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
// TODO (expected): merge all of these maps into one
@Nullable Object2ObjectHashMap<TxnId, AccordSafeCommand> commands;
@Nullable Object2ObjectHashMap<RoutingKey, AccordSafeTimestampsForKey> timestampsForKey;
@Nullable Object2ObjectHashMap<RoutingKey, AccordSafeCommandsForKey> commandsForKey;
@Nullable Object2ObjectHashMap<Object, AccordSafeState<?, ?>> loading;
// TODO (expected): collection supporting faster deletes but still fast poll (e.g. some ordered collection)
@ -246,7 +244,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
+ ", waitingToLoad: " + summarise(waitingToLoad)
+ ", loading:" + summarise(loading, AccordSafeState::global)
+ ", cfks:" + summarise(commandsForKey, AccordSafeState::global)
+ ", tfks:" + summarise(timestampsForKey, AccordSafeState::global)
+ ", txns:" + summarise(commands, AccordSafeState::global);
}
@ -340,7 +337,7 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
presetupExclusive(txnId, AccordTask::ensureCommands, parent.commands, commandStore.cachesUnsafe().commands());
}
if ((preLoadContext.keyHistory() == TIMESTAMPS ? parent.timestampsForKey : parent.commandsForKey) == null) return;
if (parent.commandsForKey == null) return;
if (preLoadContext.keys().domain() != Key) return;
switch (preLoadContext.keyHistory())
{
@ -348,11 +345,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
case NONE:
break;
case TIMESTAMPS:
for (RoutingKey key : (AbstractUnseekableKeys)preLoadContext.keys())
presetupExclusive(key, AccordTask::ensureTimestampsForKey, parent.timestampsForKey, commandStore.cachesUnsafe().timestampsForKeys());
break;
case ASYNC:
case RECOVER:
case INCR:
@ -401,16 +393,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
case NONE:
break;
case TIMESTAMPS:
{
boolean hasPreSetup = timestampsForKey != null;
for (RoutingKey key : keys)
{
if (hasPreSetup && completePresetupExclusive(key, timestampsForKey, caches.timestampsForKeys())) continue;
setupExclusive(key, AccordTask::ensureTimestampsForKey, caches.timestampsForKeys());
}
break;
}
case RECOVER:
if (!isToCompleteRangeScan)
{
@ -445,9 +427,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
case INCR:
throw new AssertionError("Incremental mode should only be used with an explicit list of keys");
case TIMESTAMPS:
throw new AssertionError("TimestampsForKey unsupported for range transactions");
case RECOVER:
case SYNC:
hasRanges = true;
@ -521,18 +500,9 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
AccordSafeState<?, ?> safeRef = loading == null ? null : loading.remove(state.key());
Invariants.checkState(safeRef != null && safeRef.global() == state, "Expected to find %s loading; found %s", state, this, AccordTask::toDescription);
if (safeRef.getClass() == AccordSafeCommand.class)
{
ensureCommands().put((TxnId)state.key(), (AccordSafeCommand) safeRef);
}
else if (safeRef.getClass() == AccordSafeCommandsForKey.class)
{
ensureCommandsForKey().put((RoutingKey) state.key(), (AccordSafeCommandsForKey) safeRef);
}
else
{
Invariants.checkState (safeRef.getClass() == AccordSafeTimestampsForKey.class);
ensureTimestampsForKey().put((RoutingKey) state.key(), (AccordSafeTimestampsForKey) safeRef);
}
ensureCommandsForKey().put((RoutingKey) state.key(), (AccordSafeCommandsForKey) safeRef);
if (!loading.isEmpty())
return false;
@ -584,18 +554,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
return commands;
}
public Map<RoutingKey, AccordSafeTimestampsForKey> timestampsForKey()
{
return timestampsForKey;
}
public Map<RoutingKey, AccordSafeTimestampsForKey> ensureTimestampsForKey()
{
if (timestampsForKey == null)
timestampsForKey = new Object2ObjectHashMap<>();
return timestampsForKey;
}
public Map<RoutingKey, AccordSafeCommandsForKey> commandsForKey()
{
return commandsForKey;
@ -683,8 +641,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
commands.forEach((k, v) -> v.preExecute());
if (commandsForKey != null)
commandsForKey.forEach((k, v) -> v.preExecute());
if (timestampsForKey != null)
timestampsForKey.forEach((k, v) -> v.preExecute());
}
@Override
@ -852,12 +808,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
commands.clear();
commands = null;
}
if (timestampsForKey != null)
{
timestampsForKey.forEach((k, v) -> caches.timestampsForKeys().release(v, this));
timestampsForKey.clear();
timestampsForKey = null;
}
if (commandsForKey != null)
{
commandsForKey.forEach((k, v) -> caches.commandsForKeys().release(v, this));
@ -892,12 +842,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
commands.clear();
commands = null;
}
if (timestampsForKey != null)
{
safeRelease(timestampsForKey, caches.timestampsForKeys(), suppressedBy);
timestampsForKey.clear();
timestampsForKey = null;
}
if (commandsForKey != null)
{
safeRelease(commandsForKey, caches.commandsForKeys(), suppressedBy);
@ -945,8 +889,6 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
{
if (commands != null)
commands.forEach((k, v) -> v.revert());
if (timestampsForKey != null)
timestampsForKey.forEach((k, v) -> v.revert());
if (commandsForKey != null)
commandsForKey.forEach((k, v) -> v.revert());
}

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

@ -90,7 +90,7 @@ public class ApplySerializers
@Override
public long serializedBodySize(A apply, int version)
{
return TypeSizes.sizeofVInt(apply.minEpoch - apply.waitForEpoch)
return TypeSizes.sizeofVInt(apply.minEpoch - apply.waitForEpoch)
+ kind.serializedSize(apply.kind, version)
+ CommandSerializers.timestamp.serializedSize(apply.executeAt, version)
+ DepsSerializers.partialDeps.serializedSize(apply.deps, version)

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

@ -31,6 +31,7 @@ import accord.messages.CheckStatus.CheckStatusReply;
import accord.primitives.Ballot;
import accord.primitives.Known;
import accord.primitives.KnownMap;
import accord.primitives.KnownMap.MinMax;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Participants;
@ -82,7 +83,7 @@ public class CheckStatusSerializers
RoutingKey[] starts = new RoutingKey[size + 1];
for (int i = 0 ; i <= size ; ++i)
starts[i] = KeySerializers.routingKey.deserialize(in, version);
KnownMap.MinMax[] values = new KnownMap.MinMax[size];
MinMax[] values = new MinMax[size];
for (int i = 0 ; i < size ; ++i)
{
int kind = in.readByte();

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

@ -23,6 +23,7 @@ import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import accord.api.Result;
import accord.api.RoutingKey;
import accord.messages.BeginRecovery;
import accord.messages.BeginRecovery.RecoverNack;
import accord.messages.BeginRecovery.RecoverOk;
@ -30,6 +31,7 @@ import accord.messages.BeginRecovery.RecoverReply;
import accord.primitives.Ballot;
import accord.primitives.Deps;
import accord.primitives.FullRoute;
import accord.primitives.Known.KnownDeps;
import accord.primitives.LatestDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Route;
@ -44,7 +46,6 @@ import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.accord.serializers.TxnRequestSerializer.WithUnsyncedSerializer;
import static accord.messages.BeginRecovery.RecoverReply.Kind.Ok;
import static org.apache.cassandra.service.accord.serializers.LatestDepsSerializers.latestDeps;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
@ -178,4 +179,80 @@ public class RecoverySerializers
+ (reply.kind() == Ok ? serializedOkSize((RecoverOk) reply, version) : serializedNackSize((RecoverNack) reply, version));
}
};
public static final IVersionedSerializer<LatestDeps> latestDeps = new IVersionedSerializer<>()
{
@Override
public void serialize(LatestDeps t, DataOutputPlus out, int version) throws IOException
{
out.writeUnsignedVInt32(t.size());
for (int i = 0 ; i < t.size() ; ++i)
{
RoutingKey start = t.startAt(i);
KeySerializers.routingKey.serialize(start, out, version);
LatestDeps.LatestEntry e = t.valueAt(i);
if (e == null)
{
CommandSerializers.nullableKnownDeps.serialize(null, out, version);
}
else
{
CommandSerializers.nullableKnownDeps.serialize(e.known, out, version);
CommandSerializers.ballot.serialize(e.ballot, out, version);
DepsSerializers.nullableDeps.serialize(e.coordinatedDeps, out, version);
DepsSerializers.nullableDeps.serialize(e.localDeps, out, version);
}
}
KeySerializers.routingKey.serialize(t.startAt(t.size()), out, version);
}
@Override
public LatestDeps deserialize(DataInputPlus in, int version) throws IOException
{
int size = in.readUnsignedVInt32();
RoutingKey[] starts = new RoutingKey[size + 1];
LatestDeps.LatestEntry[] values = new LatestDeps.LatestEntry[size];
for (int i = 0 ; i < size ; ++i)
{
starts[i] = KeySerializers.routingKey.deserialize(in, version);
KnownDeps knownDeps = CommandSerializers.nullableKnownDeps.deserialize(in, version);
if (knownDeps == null)
continue;
Ballot ballot = CommandSerializers.ballot.deserialize(in, version);
Deps coordinatedDeps = DepsSerializers.nullableDeps.deserialize(in, version);
Deps localDeps = DepsSerializers.nullableDeps.deserialize(in, version);
values[i] = new LatestDeps.LatestEntry(knownDeps, ballot, coordinatedDeps, localDeps);
}
starts[size] = KeySerializers.routingKey.deserialize(in, version);
return LatestDeps.SerializerSupport.create(true, starts, values);
}
@Override
public long serializedSize(LatestDeps t, int version)
{
long size = 0;
size += TypeSizes.sizeofUnsignedVInt(t.size());
for (int i = 0 ; i < t.size() ; ++i)
{
RoutingKey start = t.startAt(i);
size += KeySerializers.routingKey.serializedSize(start, version);
LatestDeps.LatestEntry e = t.valueAt(i);
if (e == null)
{
size += CommandSerializers.nullableKnownDeps.serializedSize(null, version);
}
else
{
size += CommandSerializers.nullableKnownDeps.serializedSize(e.known, version);
size += CommandSerializers.ballot.serializedSize(e.ballot, version);
size += DepsSerializers.nullableDeps.serializedSize(e.coordinatedDeps, version);
size += DepsSerializers.nullableDeps.serializedSize(e.localDeps, version);
}
}
size += KeySerializers.routingKey.serializedSize(t.startAt(t.size()), version);
return size;
}
};
}

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

@ -26,6 +26,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import javax.annotation.Nonnull;
import com.google.common.base.Function;
@ -35,11 +36,10 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.api.DataStore;
import accord.api.Key;
import accord.api.RoutingKey;
import accord.api.Write;
import accord.impl.TimestampsForKey;
import accord.impl.TimestampsForKeys;
import accord.local.SafeCommandStore;
import accord.local.cfk.SafeCommandsForKey;
import accord.primitives.PartialTxn;
import accord.primitives.RoutableKey;
import accord.primitives.Seekable;
@ -64,7 +64,6 @@ import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.service.accord.AccordSafeTimestampsForKey;
import org.apache.cassandra.service.accord.api.PartitionKey;
import org.apache.cassandra.utils.BooleanSerializer;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -389,10 +388,11 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
// 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;
// any that aren't can just use executeAt.hlc
TimestampsForKey cfk = TimestampsForKeys.updateLastExecutionTimestamps(safeStore, ((Key) key).toUnseekable(), txnId, executeAt, true);
long timestamp = AccordSafeTimestampsForKey.timestampMicrosFor(cfk, executeAt, true);
SafeCommandsForKey safeCfk = safeStore.get((RoutingKey) key.toUnseekable());
long timestamp = safeCfk.current().uniqueHlc(safeStore, txnId, executeAt);
// TODO (low priority - do we need to compute nowInSeconds, or can we just use executeAt?)
int nowInSeconds = AccordSafeTimestampsForKey.nowInSecondsFor(cfk, executeAt, true);
int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(executeAt.hlc());
List<AsyncChain<Void>> results = new ArrayList<>();

3
test/conf/logback-burntest.xml
View File

@ -63,4 +63,7 @@
<appender-ref ref="ASYNCFILE" />
<appender-ref ref="STDOUT" />
</root>
<logger name="accord.impl.DurabilityScheduling" level="ERROR"/>
</configuration>

13
test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
View File

@ -115,16 +115,15 @@ public class AccordJournalBurnTest extends BurnTestBase
{
ServerTestUtils.daemonInitialization();
TableMetadata[] metadatas = new TableMetadata[5 + nodes.size()];
TableMetadata[] metadatas = new TableMetadata[4 + nodes.size()];
metadatas[0] = AccordKeyspace.Commands;
metadatas[1] = AccordKeyspace.TimestampsForKeys;
metadatas[2] = AccordKeyspace.CommandsForKeys;
metadatas[3] = AccordKeyspace.Topologies;
metadatas[4] = AccordKeyspace.EpochMetadata;
metadatas[1] = AccordKeyspace.CommandsForKeys;
metadatas[2] = AccordKeyspace.Topologies;
metadatas[3] = AccordKeyspace.EpochMetadata;
for (int i = 0; i < nodes.size(); i++)
metadatas[5 + i] = AccordKeyspace.journalMetadata("journal_" + nodes.get(i));
metadatas[4 + i] = AccordKeyspace.journalMetadata("journal_" + nodes.get(i));
AccordKeyspace.tables = Tables.of(metadatas);
AccordKeyspace.TABLES = Tables.of(metadatas);
setUp();
}
Keyspace ks = Schema.instance.getKeyspaceInstance("system_accord");

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

@ -70,7 +70,6 @@ import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ColumnFamilyStore.FlushReason;
import org.apache.cassandra.db.Keyspace;
@ -98,7 +97,6 @@ import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
import org.assertj.core.api.Assertions;
import static accord.impl.TimestampsForKey.NO_LAST_EXECUTED_HLC;
import static accord.local.KeyHistory.SYNC;
import static accord.local.PreLoadContext.contextFor;
import static accord.primitives.Routable.Domain.Range;
@ -113,8 +111,6 @@ import static org.apache.cassandra.service.accord.AccordKeyspace.COMMANDS_FOR_KE
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsColumns;
import static org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeysAccessor;
import static org.apache.cassandra.service.accord.AccordKeyspace.TIMESTAMPS_FOR_KEY;
import static org.apache.cassandra.service.accord.AccordKeyspace.TimestampsForKeyRows;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
@ -141,7 +137,6 @@ public class CompactionAccordIteratorsTest
private static final TxnId GT_SECOND_TXN_ID = AccordTestUtils.txnId(EPOCH, SECOND_TXN_ID.hlc() + 1, NODE);
static ColumnFamilyStore commands;
static ColumnFamilyStore timestampsForKey;
static ColumnFamilyStore commandsForKey;
static TableMetadata table;
static FullRoute<?> route;
@ -165,9 +160,6 @@ public class CompactionAccordIteratorsTest
commands = ColumnFamilyStore.getIfExists(SchemaConstants.ACCORD_KEYSPACE_NAME, AccordKeyspace.COMMANDS);
commands.disableAutoCompaction();
timestampsForKey = ColumnFamilyStore.getIfExists(SchemaConstants.ACCORD_KEYSPACE_NAME, TIMESTAMPS_FOR_KEY);
timestampsForKey.disableAutoCompaction();
commandsForKey = ColumnFamilyStore.getIfExists(SchemaConstants.ACCORD_KEYSPACE_NAME, COMMANDS_FOR_KEY);
commandsForKey.disableAutoCompaction();
@ -245,35 +237,14 @@ public class CompactionAccordIteratorsTest
private void testAccordCommandsForKeyPurger(boolean singleCompaction) throws Throwable
{
this.singleCompaction = singleCompaction;
testAccordTimestampsForKeyPurger(null, expectedAccordTimestampsForKeyNoChange());
testAccordCommandsForKeyPurger(null, expectedAccordCommandsForKeyNoChange());
testAccordTimestampsForKeyPurger(redundantBefore(LT_TXN_ID), expectedAccordTimestampsForKeyNoChange());
testAccordCommandsForKeyPurger(redundantBefore(LT_TXN_ID), expectedAccordCommandsForKeyNoChange());
// will erase one more than expected as converted to ExclusiveSyncPoint id which is > base id
testAccordTimestampsForKeyPurger(redundantBefore(TXN_ID), expectedAccordTimestampsForKeyEraseOne());
testAccordCommandsForKeyPurger(redundantBefore(TXN_ID), expectedAccordCommandsForKeyEraseOne());
testAccordTimestampsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordTimestampsForKeyEraseAll());
testAccordCommandsForKeyPurger(redundantBefore(GT_TXN_ID), expectedAccordCommandsForKeyEraseAll());
testAccordTimestampsForKeyPurger(redundantBefore(GT_SECOND_TXN_ID), expectedAccordTimestampsForKeyEraseAll());
testAccordCommandsForKeyPurger(redundantBefore(GT_SECOND_TXN_ID), expectedAccordCommandsForKeyEraseAll());
}
private static Consumer<List<Partition>> expectedAccordTimestampsForKeyNoChange()
{
return partitions -> {
assertEquals(1, partitions.size());
Partition partition = partitions.get(0);
Row row = partition.getRow(Clustering.EMPTY);
assertEquals(TXN_ID, TimestampsForKeyRows.getLastExecutedTimestamp(row));
assertEquals(TXN_ID, TimestampsForKeyRows.getLastWriteTimestamp(row));
// last_executed_micros is only persisted if it doesn't match txnId.hlc, which only happens in the
// case of an hlc collision. Each txnId in this test has a unique hlc
assertEquals(NO_LAST_EXECUTED_HLC, TimestampsForKeyRows.getLastExecutedMicros(row));
};
}
private static Consumer<List<Partition>> expectedAccordCommandsForKeyNoChange()
{
return partitions -> {
@ -303,26 +274,11 @@ public class CompactionAccordIteratorsTest
};
}
private static Consumer<List<Partition>> expectedAccordTimestampsForKeyEraseAll()
{
return partitions -> assertEquals(0, partitions.size());
}
private static Consumer<List<Partition>> expectedAccordCommandsForKeyEraseAll()
{
return partitions -> assertEquals(0, partitions.size());
}
private void testAccordTimestampsForKeyPurger(RedundantBefore redundantBefore, Consumer<List<Partition>> expectedResult) throws Throwable
{
testWithCommandStore((commandStore) -> {
IAccordService mockAccordService = mockAccordService(commandStore, redundantBefore, DurableBefore.EMPTY);
ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(ACCORD_KEYSPACE_NAME, TIMESTAMPS_FOR_KEY);
List<Partition> result = compactCFS(mockAccordService, cfs);
expectedResult.accept(result);
}, true);
}
private void testAccordCommandsForKeyPurger(RedundantBefore redundantBefore, Consumer<List<Partition>> expectedResult) throws Throwable
{
testWithCommandStore((commandStore) -> {
@ -463,7 +419,6 @@ public class CompactionAccordIteratorsTest
}
});
commands.forceBlockingFlush(FlushReason.UNIT_TESTS);
timestampsForKey.forceBlockingFlush(FlushReason.UNIT_TESTS);
commandsForKey.forceBlockingFlush(FlushReason.UNIT_TESTS);
while (commandStore.executor().hasTasks())
LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));

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

@ -31,8 +31,6 @@ import org.slf4j.LoggerFactory;
import accord.api.Key;
import accord.api.Result;
import accord.impl.TimestampsForKey;
import accord.impl.TimestampsForKeys;
import accord.local.Command;
import accord.local.CommonAttributes;
import accord.local.StoreParticipants;
@ -154,46 +152,6 @@ public class AccordCommandStoreTest
Assert.assertEquals(expected, actual);
}
@Test
public void timestampsForKeyLoadSave()
{
AtomicLong clock = new AtomicLong(0);
AccordCommandStore commandStore = createAccordCommandStore(clock::incrementAndGet, "ks", "tbl");
PartialTxn txn = createPartialTxn(1);
TokenKey key = ((PartitionKey) getOnlyElement(txn.keys())).toUnseekable();
TxnId txnId1 = txnId(1, clock.incrementAndGet(), 1);
TxnId txnId2 = txnId(1, clock.incrementAndGet(), 1);
Command command1 = preaccepted(txnId1, txn, timestamp(1, clock.incrementAndGet(), 1));
Command command2 = preaccepted(txnId2, txn, timestamp(1, clock.incrementAndGet(), 1));
AccordSafeTimestampsForKey tfk = new AccordSafeTimestampsForKey(loaded(key, null));
tfk.initialize();
TimestampsForKeys.updateLastExecutionTimestamps(null, tfk, txnId1, txnId1, true);
Assert.assertEquals(txnId1.hlc(), AccordSafeTimestampsForKey.timestampMicrosFor(tfk.current(), txnId1, true));
TimestampsForKeys.updateLastExecutionTimestamps(null, tfk, txnId2, txnId2, true);
Assert.assertEquals(txnId2.hlc(), AccordSafeTimestampsForKey.timestampMicrosFor(tfk.current(), txnId2, true));
Assert.assertEquals(txnId2, tfk.current().lastExecutedTimestamp());
Assert.assertEquals(txnId2.hlc(), tfk.lastExecutedMicros());
AccordSafeCommandsForKey cfk = new AccordSafeCommandsForKey(loaded(key, null));
cfk.initialize();
cfk.set(cfk.current().update(command1).cfk());
cfk.set(cfk.current().update(command2).cfk());
AccordKeyspace.getTimestampsForKeyUpdater(commandStore, tfk.current(), commandStore.nextSystemTimestampMicros()).run();
logger.info("E: {}", tfk);
TimestampsForKey actual = AccordKeyspace.loadTimestampsForKey(commandStore.id(), key);
logger.info("A: {}", actual);
Assert.assertEquals(tfk.current(), actual);
}
@Test
public void commandsForKeyLoadSave()
{
@ -208,9 +166,6 @@ public class AccordCommandStoreTest
Command command1 = preaccepted(txnId1, txn, timestamp(1, clock.incrementAndGet(), 1));
Command command2 = preaccepted(txnId2, txn, timestamp(1, clock.incrementAndGet(), 1));
AccordSafeTimestampsForKey tfk = new AccordSafeTimestampsForKey(loaded(key, null));
tfk.initialize();
AccordSafeCommandsForKey cfk = new AccordSafeCommandsForKey(loaded(key, null));
cfk.initialize();

1
test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
View File

@ -115,7 +115,6 @@ public class AccordTaskTest
public void before()
{
QueryProcessor.executeInternal(String.format("TRUNCATE %s.%s", SchemaConstants.ACCORD_KEYSPACE_NAME, AccordKeyspace.COMMANDS));
QueryProcessor.executeInternal(String.format("TRUNCATE %s.%s", SchemaConstants.ACCORD_KEYSPACE_NAME, AccordKeyspace.TIMESTAMPS_FOR_KEY));
QueryProcessor.executeInternal(String.format("TRUNCATE %s.%s", SchemaConstants.ACCORD_KEYSPACE_NAME, AccordKeyspace.COMMANDS_FOR_KEY));
}

3
test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java
View File

@ -144,9 +144,6 @@ public class SimulatedAccordTaskTest extends SimulatedAccordCommandStoreTestBase
instance.commandStore.cachesUnsafe().commandsForKeys().forEach(e -> {
Assertions.assertThat(e.references()).isEqualTo(0);
});
instance.commandStore.cachesUnsafe().timestampsForKeys().forEach(e -> {
Assertions.assertThat(e.references()).isEqualTo(0);
});
}
}

20
test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java
View File

@ -310,14 +310,14 @@ public class CommandsForKeySerializerTest
for (int j = 0 ; j < i ; ++j)
{
InternalStatus status = InternalStatus.from(cmds[j].saveStatus);
if (status == null || !status.hasExecuteAtAndDeps) continue;
if (status == null || !status.hasExecuteAtOrDeps()) continue;
if (cmds[j].txnId.kind().witnesses(cmds[i].txnId) && status.depsKnownBefore(cmds[j].txnId, cmds[j].executeAt).compareTo(cmds[i].txnId) > 0 && Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
continue outer;
}
for (int j = i + 1 ; j < cmds.length ; ++j)
{
InternalStatus status = InternalStatus.from(cmds[j].saveStatus);
if (status == null || !status.hasExecuteAtAndDeps) continue;
if (status == null || !status.hasExecuteAtOrDeps()) continue;
if (cmds[j].txnId.kind().witnesses(cmds[i].txnId) && Collections.binarySearch(cmds[j].missing, cmds[i].txnId) < 0)
continue outer;
}
@ -485,8 +485,10 @@ public class CommandsForKeySerializerTest
}
TxnInfo info = cfk.get(i);
InternalStatus expectStatus = InternalStatus.from(cmd.saveStatus);
if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVE;
if (expectStatus.hasExecuteAtAndDeps)
if (expectStatus == InternalStatus.APPLIED_NOT_DURABLE && cmd.isDurable)
expectStatus = InternalStatus.APPLIED_DURABLE;
if (expectStatus == null) expectStatus = InternalStatus.TRANSITIVE_VISIBLE;
if (expectStatus.hasExecuteAt())
Assert.assertEquals(cmd.executeAt, info.executeAt);
Assert.assertEquals(expectStatus, info.status());
Assert.assertArrayEquals(cmd.missing.toArray(TxnId[]::new), info.missing());
@ -526,7 +528,7 @@ public class CommandsForKeySerializerTest
for (int i = 0; i < info.length; i++)
{
InternalStatus status = rs.pick(statuses);
info[i] = TxnInfo.create(ids[i], status, false, true, ids[i], TxnId.NO_TXNIDS, Ballot.ZERO);
info[i] = TxnInfo.create(ids[i], status, true, ids[i], TxnId.NO_TXNIDS, Ballot.ZERO);
}
Gen<Unmanaged.Pending> pendingGen = Gens.enums().allMixedDistribution(Unmanaged.Pending.class).next(rs);
@ -545,7 +547,7 @@ public class CommandsForKeySerializerTest
{
int idx = Arrays.binarySearch(ids, u.txnId);
if (idx < 0)
missing.add(TxnInfo.create(u.txnId, InternalStatus.TRANSITIVE, false, true, u.txnId, Ballot.ZERO));
missing.add(TxnInfo.create(u.txnId, InternalStatus.TRANSITIVE, true, u.txnId, Ballot.ZERO));
}
if (!missing.isEmpty())
{
@ -555,7 +557,7 @@ public class CommandsForKeySerializerTest
}
else unmanaged = CommandsForKey.NO_PENDING_UNMANAGED;
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, unmanaged, TxnId.NONE, NO_BOUNDS_INFO);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk, info, 0, unmanaged, TxnId.NONE, NO_BOUNDS_INFO);
ByteBuffer buffer = Serialize.toBytesWithoutKey(expected);
CommandsForKey roundTrip = Serialize.fromBytes(pk, buffer);
@ -571,8 +573,8 @@ public class CommandsForKeySerializerTest
TokenKey pk = new TokenKey(TableId.fromString("1b255f4d-ef25-40a6-0000-000000000009"), token);
TxnId txnId = TxnId.fromValues(11,34052499,2,1);
CommandsForKey expected = CommandsForKey.SerializerSupport.create(pk,
new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, false, true, txnId, TxnId.NO_TXNIDS, Ballot.ZERO) },
CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE, NO_BOUNDS_INFO);
new TxnInfo[] { TxnInfo.create(txnId, InternalStatus.PREACCEPTED_OR_ACCEPTED_INVALIDATE, true, txnId, TxnId.NO_TXNIDS, Ballot.ZERO) },
0, CommandsForKey.NO_PENDING_UNMANAGED, TxnId.NONE, NO_BOUNDS_INFO);
ByteBuffer buffer = Serialize.toBytesWithoutKey(expected);
CommandsForKey roundTrip = Serialize.fromBytes(pk, buffer);