diff --git a/modules/accord b/modules/accord index f543851dff..520cc1072d 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit f543851dffbf2907580f91608be3c1aadc4ccb95 +Subproject commit 520cc1072d44a5f7617566b6667e915532b89033 diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index a2d7595c9d..09cea4dada 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -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 redundantBefores; - int storeId; - TokenKey partitionKey; - - AccordTimestampsForKeyPurger(Supplier 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); diff --git a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java index 4d3999247f..b01e736065 100644 --- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java +++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java @@ -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; diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index b0185ad4ae..35e8dca12a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -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.Instance commands; - private final AccordCache.Type.Instance timestampsForKeys; private final AccordCache.Type.Instance commandsForKeys; - Caches(AccordCache global, AccordCache.Type.Instance commandCache, AccordCache.Type.Instance timestampsForKeyCache, AccordCache.Type.Instance commandsForKeyCache) + Caches(AccordCache global, AccordCache.Type.Instance commandCache, AccordCache.Type.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.Instance timestampsForKeys() - { - return timestampsForKeys; - } - public final AccordCache.Type.Instance commandsForKeys() { return commandsForKeys; } } - public static final class ExclusiveCaches extends Caches implements CommandStoreCaches + public static final class ExclusiveCaches extends Caches implements CommandStoreCaches { private final Lock lock; - public ExclusiveCaches(Lock lock, AccordCache global, AccordCache.Type.Instance commands, AccordCache.Type.Instance timestampsForKeys, AccordCache.Type.Instance commandsForKeys) + public ExclusiveCaches(Lock lock, AccordCache global, AccordCache.Type.Instance commands, AccordCache.Type.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.Instance commands; - final AccordCache.Type.Instance timestampsForKey; final AccordCache.Type.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); diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java index 73962c3a9e..3ccbe4cc2e 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutor.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutor.java @@ -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 commands, AccordCache.Type timestampsForKey, AccordCache.Type commandsForKey) + public ExclusiveGlobalCaches(Lock lock, AccordCache global, AccordCache.Type commands, AccordCache.Type 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 commands; - public final AccordCache.Type timestampsForKey; public final AccordCache.Type commandsForKey; - public GlobalCaches(AccordCache global, AccordCache.Type commands, AccordCache.Type timestampsForKey, AccordCache.Type commandsForKey) + public GlobalCaches(AccordCache global, AccordCache.Type commands, AccordCache.Type 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 commands; - final AccordCache.Type timestampsForKey; final AccordCache.Type 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); diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index d7587c462f..bcad001c4c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -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 TABLE_NAMES = ImmutableSet.of(COMMANDS, TIMESTAMPS_FOR_KEY, COMMANDS_FOR_KEY, + public static final Set 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 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); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index 280ba08036..1679f3da58 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -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; diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index ddaacc4241..8e4948a49a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -43,7 +43,7 @@ import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches; import static accord.utils.Invariants.illegalState; -public class AccordSafeCommandStore extends AbstractSafeCommandStore +public class AccordSafeCommandStore extends AbstractSafeCommandStore { final AccordTask task; private final @Nullable CommandsForRanges commandsForRanges; @@ -130,21 +130,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore timestampsForKey = task.timestampsForKey(); - if (timestampsForKey == null) - return null; - - return timestampsForKey.get(key); - } - @Override public AccordCommandStore commandStore() { diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java b/src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java deleted file mode 100644 index 6406f53704..0000000000 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeTimestampsForKey.java +++ /dev/null @@ -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 -{ - private boolean invalidated; - private final AccordCacheEntry global; - private TimestampsForKey original; - private TimestampsForKey current; - - public AccordSafeTimestampsForKey(AccordCacheEntry 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 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())); - } -} diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 63fe397cc4..e1aec28521 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -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."); diff --git a/src/java/org/apache/cassandra/service/accord/AccordTask.java b/src/java/org/apache/cassandra/service/accord/AccordTask.java index d9632efd2d..ea4277391a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordTask.java +++ b/src/java/org/apache/cassandra/service/accord/AccordTask.java @@ -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 extends Task implements Runnable, Function commands; - @Nullable Object2ObjectHashMap timestampsForKey; @Nullable Object2ObjectHashMap commandsForKey; @Nullable Object2ObjectHashMap> loading; // TODO (expected): collection supporting faster deletes but still fast poll (e.g. some ordered collection) @@ -246,7 +244,6 @@ public abstract class AccordTask extends Task implements Runnable, Function extends Task implements Runnable, Function extends Task implements Runnable, Function extends Task implements Runnable, Function extends Task implements Runnable, Function extends Task implements Runnable, Function 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 extends Task implements Runnable, Function timestampsForKey() - { - return timestampsForKey; - } - - public Map ensureTimestampsForKey() - { - if (timestampsForKey == null) - timestampsForKey = new Object2ObjectHashMap<>(); - return timestampsForKey; - } - public Map commandsForKey() { return commandsForKey; @@ -683,8 +641,6 @@ public abstract class AccordTask extends Task implements Runnable, Function 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 extends Task implements Runnable, Function 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 extends Task implements Runnable, Function extends Task implements Runnable, Function v.revert()); - if (timestampsForKey != null) - timestampsForKey.forEach((k, v) -> v.revert()); if (commandsForKey != null) commandsForKey.forEach((k, v) -> v.revert()); } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java index 839d2e4d72..9422738e3d 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ApplySerializers.java @@ -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) diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java index dd56dd28f9..7c44b60835 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java @@ -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(); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java index 3b4a78d404..683131c1da 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java @@ -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 = 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; + } + }; } diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java index 48260635e4..3deca7c171 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java @@ -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 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> results = new ArrayList<>(); diff --git a/test/conf/logback-burntest.xml b/test/conf/logback-burntest.xml index e1e48a9d3f..3aada72e8f 100644 --- a/test/conf/logback-burntest.xml +++ b/test/conf/logback-burntest.xml @@ -63,4 +63,7 @@ + + + diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java index 255c2e0758..25f9c61d38 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java @@ -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"); diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index 5e45b39251..c6e6809457 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -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> 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> expectedAccordCommandsForKeyNoChange() { return partitions -> { @@ -303,26 +274,11 @@ public class CompactionAccordIteratorsTest }; } - private static Consumer> expectedAccordTimestampsForKeyEraseAll() - { - return partitions -> assertEquals(0, partitions.size()); - } - private static Consumer> expectedAccordCommandsForKeyEraseAll() { return partitions -> assertEquals(0, partitions.size()); } - private void testAccordTimestampsForKeyPurger(RedundantBefore redundantBefore, Consumer> expectedResult) throws Throwable - { - testWithCommandStore((commandStore) -> { - IAccordService mockAccordService = mockAccordService(commandStore, redundantBefore, DurableBefore.EMPTY); - ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(ACCORD_KEYSPACE_NAME, TIMESTAMPS_FOR_KEY); - List result = compactCFS(mockAccordService, cfs); - expectedResult.accept(result); - }, true); - } - private void testAccordCommandsForKeyPurger(RedundantBefore redundantBefore, Consumer> 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)); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java index ea5f7c2b60..a24f2c876d 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java @@ -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(); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java index c53319101f..8aa15daaed 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java @@ -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)); } diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java index 1fd77bed77..c8d3f82052 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordTaskTest.java @@ -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); - }); } } diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java index 190f1b1104..4a3bc8cfd9 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandsForKeySerializerTest.java @@ -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 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);