From fb05d224e7ceb28dfb1b8dd327c951cff532c57a Mon Sep 17 00:00:00 2001 From: Benedict Elliott Smith Date: Thu, 13 Mar 2025 11:55:29 +0000 Subject: [PATCH] Fix: - Accord Journal purging was disabled - remove unique_id from schema keyspace - avoid String.format in Compactor hot path - avoid string concatenation on hot path; improve segment compactor partition build efficiency - Partial compaction should update records in place to ensure truncation of discontiguous compactions do not lead to an incorrect field version being used - StoreParticipants.touches behaviour for RX was erroneously modified; should touch all non-redundant ranges including those no longer owned - SetShardDurable should correctly set DurableBefore Majority/Universal based on the Durability parameter - fix erroneous prunedBefore invariant - Journal compaction should not rewrite fields shadowed by a newer record - Don't save updates to ERASED commands - Simplify CommandChange.getFlags - fix handling of Durability for Invalidated - Don't use ApplyAt for GC_BEFORE with partial input, as might be a saveStatus >= ApplyAtKnown but with executeAt < ApplyAtKnown patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20441 --- modules/accord | 2 +- .../apache/cassandra/config/AccordSpec.java | 6 +- .../db/compaction/CompactionIterator.java | 381 ++++++++++++------ .../cassandra/schema/SchemaKeyspace.java | 10 +- .../AbstractAccordSegmentCompactor.java | 56 ++- .../service/accord/AccordCommandStore.java | 6 +- .../service/accord/AccordJournal.java | 37 +- .../accord/AccordJournalValueSerializers.java | 41 +- .../service/accord/AccordKeyspace.java | 2 + .../service/accord/AccordService.java | 2 +- .../accord/journal/AccordTopologyUpdate.java | 11 +- .../utils/NativeSSTableLoaderClient.java | 6 +- .../service/accord/AccordJournalBurnTest.java | 99 +++-- .../accord/AccordJournalCompactionTest.java | 8 +- .../CompactionAccordIteratorsTest.java | 6 +- .../AccordConfigurationServiceTest.java | 3 +- .../accord/AccordJournalOrderTest.java | 4 +- .../accord/AccordJournalTestParams.java | 77 ---- .../service/accord/AccordTestUtils.java | 2 +- .../service/accord/CommandChangeTest.java | 10 +- .../accord/SimulatedAccordCommandStore.java | 7 +- 21 files changed, 454 insertions(+), 322 deletions(-) delete mode 100644 test/unit/org/apache/cassandra/service/accord/AccordJournalTestParams.java diff --git a/modules/accord b/modules/accord index c7379e12bd..0a5446a365 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit c7379e12bd8f8732004cb77264801fe157af1dbe +Subproject commit 0a5446a365fe7eb1a15b6a2d0d72f9475c51bc47 diff --git a/src/java/org/apache/cassandra/config/AccordSpec.java b/src/java/org/apache/cassandra/config/AccordSpec.java index b81792df0c..a85437624f 100644 --- a/src/java/org/apache/cassandra/config/AccordSpec.java +++ b/src/java/org/apache/cassandra/config/AccordSpec.java @@ -158,10 +158,10 @@ public class AccordSpec public volatile DurationSpec.IntSecondsBound fast_path_update_delay = null; public volatile DurationSpec.IntSecondsBound gc_delay = new DurationSpec.IntSecondsBound("5m"); - public volatile int shard_durability_target_splits = 128; + public volatile int shard_durability_target_splits = 16; public volatile DurationSpec.IntSecondsBound durability_txnid_lag = new DurationSpec.IntSecondsBound(5); - public volatile DurationSpec.IntSecondsBound shard_durability_cycle = new DurationSpec.IntSecondsBound(15, TimeUnit.MINUTES); - public volatile DurationSpec.IntSecondsBound global_durability_cycle = new DurationSpec.IntSecondsBound(10, TimeUnit.MINUTES); + public volatile DurationSpec.IntSecondsBound shard_durability_cycle = new DurationSpec.IntSecondsBound(5, TimeUnit.MINUTES); + public volatile DurationSpec.IntSecondsBound global_durability_cycle = new DurationSpec.IntSecondsBound(5, TimeUnit.MINUTES); public enum TransactionalRangeMigration { diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 1250c69fb0..53b4d57f2e 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -19,6 +19,7 @@ package org.apache.cassandra.db.compaction; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; @@ -27,8 +28,8 @@ import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.LongPredicate; import java.util.function.Supplier; -import javax.annotation.Nonnull; +import com.google.common.annotations.VisibleForTesting; import com.google.common.collect.ImmutableSet; import com.google.common.collect.Ordering; import org.slf4j.Logger; @@ -39,6 +40,9 @@ import accord.local.DurableBefore; import accord.local.RedundantBefore; import accord.utils.Invariants; import accord.utils.UnhandledEnum; +import accord.utils.btree.BTree; +import accord.utils.btree.BulkIterator; +import accord.utils.btree.UpdateFunction; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.cql3.ColumnIdentifier; import org.apache.cassandra.db.AbstractCompactionController; @@ -57,6 +61,9 @@ import org.apache.cassandra.db.partitions.PartitionUpdate; import org.apache.cassandra.db.partitions.PurgeFunction; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.db.rows.BTreeRow; +import org.apache.cassandra.db.rows.BufferCell; +import org.apache.cassandra.db.rows.ColumnData; import org.apache.cassandra.db.rows.RangeTombstoneBoundMarker; import org.apache.cassandra.db.rows.RangeTombstoneMarker; import org.apache.cassandra.db.rows.Row; @@ -82,8 +89,9 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; +import org.apache.cassandra.service.ClientState; import org.apache.cassandra.service.accord.AccordJournal; -import org.apache.cassandra.service.accord.AccordJournalValueSerializers; +import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightImage; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightSerializer; import org.apache.cassandra.service.accord.AccordKeyspace; import org.apache.cassandra.service.accord.AccordKeyspace.CommandsForKeyAccessor; @@ -92,7 +100,6 @@ import org.apache.cassandra.service.accord.IAccordService; import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfo; import org.apache.cassandra.service.accord.IAccordService.AccordCompactionInfos; import org.apache.cassandra.service.accord.JournalKey; -import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.service.accord.api.TokenKey; import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate; import org.apache.cassandra.service.accord.serializers.Version; @@ -158,7 +165,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte public CompactionIterator(OperationType type, List scanners, AbstractCompactionController controller, long nowInSec, TimeUUID compactionId) { - this(type, scanners, controller, nowInSec, compactionId, ActiveCompactionsTracker.NOOP, null, AccordService::instance); + this(type, scanners, controller, nowInSec, compactionId, ActiveCompactionsTracker.NOOP, null); } public CompactionIterator(OperationType type, @@ -170,7 +177,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte TopPartitionTracker.Collector topPartitionCollector) { this(type, scanners, controller, nowInSec, compactionId, activeCompactions, topPartitionCollector, - AccordService::instance); + AccordService.isSetup() ? AccordService.instance() : null); } public CompactionIterator(OperationType type, @@ -180,7 +187,23 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte TimeUUID compactionId, ActiveCompactionsTracker activeCompactions, TopPartitionTracker.Collector topPartitionCollector, - @Nonnull Supplier accordService) + IAccordService accord) + { + this(type, scanners, controller, nowInSec, compactionId, activeCompactions, topPartitionCollector, + () -> accord.getCompactionInfo(), + () -> Version.fromVersion(accord.journalConfiguration().userVersion())); + } + + @VisibleForTesting + public CompactionIterator(OperationType type, + List scanners, + AbstractCompactionController controller, + long nowInSec, + TimeUUID compactionId, + ActiveCompactionsTracker activeCompactions, + TopPartitionTracker.Collector topPartitionCollector, + Supplier compactionInfos, + Supplier accordVersion) { this.controller = controller; this.type = type; @@ -206,13 +229,13 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte if (topPartitionCollector != null) // need to count tombstones before they are purged merged = Transformation.apply(merged, new TopPartitionTracker.TombstoneCounter(topPartitionCollector, nowInSec)); merged = Transformation.apply(merged, new GarbageSkipper(controller)); - Transformation purger = purger(controller.cfs, accordService); + Transformation purger = purger(controller.cfs, compactionInfos, accordVersion); merged = Transformation.apply(merged, purger); merged = DuplicateRowChecker.duringCompaction(merged, type); compacted = Transformation.apply(merged, new AbortableUnfilteredPartitionTransformation(this)); } - private Transformation purger(ColumnFamilyStore cfs, Supplier accordService) + private Transformation purger(ColumnFamilyStore cfs, Supplier compactionInfos, Supplier version) { if (isPaxos(cfs) && paxosStatePurging() != legacy) return new PaxosPurger(); @@ -222,9 +245,9 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte return new Purger(controller, nowInSec); if (isAccordJournal(cfs)) - return new AccordJournalPurger(accordService, cfs); + return new AccordJournalPurger(compactionInfos.get(), version.get(), cfs); if (isAccordCommandsForKey(cfs)) - return new AccordCommandsForKeyPurger(AccordKeyspace.CFKAccessor, accordService); + return new AccordCommandsForKeyPurger(AccordKeyspace.CFKAccessor, compactionInfos); throw new IllegalArgumentException("Unhandled accord table: " + cfs.keyspace.getName() + '.' + cfs.name); } @@ -792,10 +815,10 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte int storeId; TokenKey tokenKey; - AccordCommandsForKeyPurger(CommandsForKeyAccessor accessor, Supplier accordService) + AccordCommandsForKeyPurger(CommandsForKeyAccessor accessor, Supplier compactionInfos) { this.accessor = accessor; - this.compactionInfos = accordService.get().getCompactionInfo(); + this.compactionInfos = compactionInfos.get(); } protected void beginPartition(UnfilteredRowIterator partition) @@ -836,31 +859,21 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte final AccordCompactionInfos infos; final ColumnMetadata recordColumn; final ColumnMetadata versionColumn; - final AccordService service; - final AccordAgent agent; - AccordCompactionInfo info; - JournalKey key = null; - Object builder = null; - FlyweightSerializer serializer = null; + JournalKey key; + AccordRowCompactor compactor; // Initialize topology serializer during compaction to avoid deserializing redundant epochs - FlyweightSerializer topologySerializer; - Object[] highestClustering = null; + FlyweightSerializer topologySerializer; final Version userVersion; - long lastDescriptor = -1; - int lastOffset = -1; - public AccordJournalPurger(Supplier serviceSupplier, ColumnFamilyStore cfs) + public AccordJournalPurger(AccordCompactionInfos compactionInfos, Version version, ColumnFamilyStore cfs) { - service = (AccordService) serviceSupplier.get(); - // TODO: test serialization version logic - userVersion = Version.fromVersion(service.journalConfiguration().userVersion()); + this.userVersion = version; - this.agent = service.agent(); - this.infos = service.getCompactionInfo(); + this.infos = compactionInfos; this.recordColumn = cfs.metadata().getColumn(ColumnIdentifier.getInterned("record", false)); this.versionColumn = cfs.metadata().getColumn(ColumnIdentifier.getInterned("user_version", false)); - this.topologySerializer = (FlyweightSerializer) (FlyweightSerializer) new AccordTopologyUpdate.AccumulatingSerializer(() -> infos.minEpoch); + this.topologySerializer = (FlyweightSerializer) (FlyweightSerializer) new AccordTopologyUpdate.AccumulatingSerializer(() -> infos.minEpoch); } @SuppressWarnings("unchecked") @@ -868,88 +881,36 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte protected void beginPartition(UnfilteredRowIterator partition) { key = AccordKeyspace.JournalColumns.getJournalKey(partition.partitionKey()); - serializer = (AccordJournalValueSerializers.FlyweightSerializer) key.type.serializer; - builder = serializer.mergerFor(key); - lastDescriptor = -1; - lastOffset = -1; - highestClustering = null; + if (compactor == null || compactor.serializer != key.type.serializer) + { + switch (key.type) + { + case COMMAND_DIFF: + compactor = new AccordCommandRowCompactor(infos, userVersion, nowInSec); + break; + case TOPOLOGY_UPDATE: + compactor = new AccordMergingCompactor(topologySerializer, userVersion); + break; + default: + compactor = new AccordMergingCompactor(key.type.serializer, userVersion); + } + } + compactor.reset(key); } @Override protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition) { - beginPartition(partition); - - if (partition.isEmpty()) - return null; + if (!partition.hasNext()) + return partition; try { - List rows = new ArrayList<>(); + beginPartition(partition); while (partition.hasNext()) - { - Row row = (Row) partition.next(); - rows.add(row); - collect(row); - } + collect((Row)partition.next()); - if (key.type != JournalKey.Type.COMMAND_DIFF) - { - PartitionUpdate.SimpleBuilder newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partition.partitionKey()); - try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get()) - { - serializer.reserialize(key, builder, out, userVersion); - newVersion.row(highestClustering) - .add("record", out.asNewBuffer()) - .add("user_version", userVersion.version); - } - catch (IOException e) - { - throw new RuntimeException(e); - } - - return newVersion.build().unfilteredIterator(); - } - - AccordJournal.Builder commandBuilder = (AccordJournal.Builder) builder; - if (commandBuilder.isEmpty()) - { - Invariants.require(rows.isEmpty()); - return partition; - } - - if (info != null && info.commandStoreId != key.commandStoreId) info = null; - if (info == null) info = infos.get(key.commandStoreId); - // TODO (required): should return null only if commandStore has been removed - if (info == null) - return partition; - DurableBefore durableBefore = infos.durableBefore; - Cleanup cleanup = commandBuilder.maybeCleanup(PARTIAL, agent, info.redundantBefore, durableBefore); - if (cleanup != NO) - { - switch (cleanup) - { - default: throw new UnhandledEnum(cleanup); - case EXPUNGE: - return null; - case ERASE: - return PartitionUpdate.fullPartitionDelete(metadata(), partition.partitionKey(), Long.MAX_VALUE, nowInSec).unfilteredIterator(); - case TRUNCATE: - case INVALIDATE: - case TRUNCATE_WITH_OUTCOME: - case VESTIGIAL: - PartitionUpdate.SimpleBuilder newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partition.partitionKey()); - - Row.SimpleBuilder rowBuilder = newVersion.row(rows.get(rows.size() - 1).clustering().getBufferArray()); - rowBuilder.add("record", commandBuilder.asByteBuffer(userVersion)) - .add("user_version", userVersion.version); - - return newVersion.build().unfilteredIterator(); - } - } - - return PartitionUpdate.multiRowUpdate(AccordKeyspace.Journal, partition.partitionKey(), rows) - .unfilteredIterator(); + return compactor.result(key, partition.partitionKey()); } catch (UnknownTableException e) { @@ -962,47 +923,218 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte } } - @Override - protected Row applyToRow(Row row) - { - return row; - } - protected void collect(Row row) throws IOException { updateProgress(); - ByteBuffer record = row.getCell(recordColumn).buffer(); + ByteBuffer bytes = row.getCell(recordColumn).buffer(); + Version userVersion = Version.fromVersion(Int32Type.instance.compose(row.getCell(versionColumn).buffer())); + compactor.collect(key, row, bytes, userVersion); + } + } + + static abstract class AccordRowCompactor + { + final FlyweightSerializer serializer; + + AccordRowCompactor(FlyweightSerializer serializer) + { + this.serializer = serializer; + } + + abstract void reset(JournalKey key); + abstract void collect(JournalKey key, Row row, ByteBuffer bytes, Version userVersion) throws IOException; + abstract UnfilteredRowIterator result(JournalKey journalKey, DecoratedKey partitionKey) throws IOException; + } + + static class AccordMergingCompactor extends AccordRowCompactor + { + final T builder; + final Version userVersion; + Object[] highestClustering; + long lastDescriptor; + int lastOffset; + + AccordMergingCompactor(FlyweightSerializer serializer, Version userVersion) + { + super(serializer); + this.builder = serializer.mergerFor(); + this.userVersion = userVersion; + } + + @Override + void reset(JournalKey key) + { + builder.reset(key); + lastDescriptor = -1; + lastOffset = -1; + highestClustering = null; + } + + @Override + protected void collect(JournalKey key, Row row, ByteBuffer bytes, Version userVersion) throws IOException + { + if (highestClustering == null) + highestClustering = row.clustering().getBufferArray(); + long descriptor = LongType.instance.compose(row.clustering().bufferAt(0)); int offset = Int32Type.instance.compose(row.clustering().bufferAt(1)); if (lastOffset != -1) { Invariants.require(descriptor <= lastDescriptor, - "Descriptors were accessed out of order: %d was accessed after %d", descriptor, lastDescriptor); + "Descriptors were accessed out of order: %d was accessed after %d", descriptor, lastDescriptor); Invariants.require(descriptor != lastDescriptor || - offset < lastOffset, - "Offsets within %s were accessed out of order: %d was accessed after %s", offset, lastOffset); + offset < lastOffset, + "Offsets within %d were accessed out of order: %d was accessed after %s", descriptor, offset, lastOffset); } lastDescriptor = descriptor; lastOffset = offset; - try (DataInputBuffer in = new DataInputBuffer(record, false)) + try (DataInputBuffer in = new DataInputBuffer(bytes, false)) { - Version userVersion = Version.fromVersion(Int32Type.instance.compose(row.getCell(versionColumn).buffer())); - if (key.type == JournalKey.Type.TOPOLOGY_UPDATE) - topologySerializer.deserialize(key, builder, in, userVersion); - else - serializer.deserialize(key, builder, in, userVersion); - if (highestClustering == null) // we iterate highest to lowest - highestClustering = row.clustering().getBufferArray(); + serializer.deserialize(key, builder, in, userVersion); } } @Override - protected Row applyToStatic(Row row) + UnfilteredRowIterator result(JournalKey journalKey, DecoratedKey partitionKey) throws IOException { - checkState(row.isStatic() && row.isEmpty()); - return row; + PartitionUpdate.SimpleBuilder newVersion = PartitionUpdate.simpleBuilder(AccordKeyspace.Journal, partitionKey); + try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get()) + { + serializer.reserialize(journalKey, builder, out, userVersion); + newVersion.row(highestClustering) + .add("record", out.asNewBuffer()) + .add("user_version", userVersion.version); + } + + return newVersion.build().unfilteredIterator(); + } + } + + static class AccordCommandRowEntry + { + final AccordJournal.Builder builder = new AccordJournal.Builder(); + Row row; + boolean modified; + + void init(JournalKey key, Row row, ByteBuffer bytes, Version userVersion) throws IOException + { + this.row = row; + this.builder.reset(key); + try (DataInputBuffer in = new DataInputBuffer(bytes, false)) + { + builder.deserializeNext(in, userVersion); + } + } + + void clear() + { + row = null; + modified = false; + builder.clear(); + } + } + + static class AccordCommandRowCompactor extends AccordRowCompactor + { + static final Object[] rowTemplate = BTree.build(BulkIterator.of(new Object[2]), 2, UpdateFunction.noOp); + final long timestamp = ClientState.getTimestamp(); + final AccordCompactionInfos infos; + final Version userVersion; + final ColumnData userVersionCell; + final long nowInSec; + + final AccordJournal.Builder mainBuilder = new AccordJournal.Builder(); + final List entries = new ArrayList<>(); + final ArrayDeque reuseEntries = new ArrayDeque<>(); + AccordCompactionInfo info; + + AccordCommandRowCompactor(AccordCompactionInfos infos, Version userVersion, long nowInSec) + { + super((FlyweightSerializer) JournalKey.Type.COMMAND_DIFF.serializer); + this.infos = infos; + this.userVersion = userVersion; + this.userVersionCell = BufferCell.live(AccordKeyspace.JournalColumns.user_version, timestamp, Int32Type.instance.decompose(userVersion.version)); + this.nowInSec = nowInSec; + } + + @Override + void reset(JournalKey key) + { + mainBuilder.reset(key); + reuseEntries.addAll(entries); + for (int i = 0; i < entries.size() ; ++i) + entries.get(i).clear(); + entries.clear(); + } + + @Override + void collect(JournalKey key, Row row, ByteBuffer bytes, Version userVersion) throws IOException + { + AccordCommandRowEntry e = reuseEntries.pollLast(); + if (e == null) + e = new AccordCommandRowEntry(); + entries.add(e); + e.init(key, row, bytes, userVersion); + e.modified |= e.builder.clearSuperseded(false, mainBuilder); + mainBuilder.fillInMissingOrCleanup(false, e.builder); + } + + @Override + UnfilteredRowIterator result(JournalKey journalKey, DecoratedKey partitionKey) throws IOException + { + if (mainBuilder.isEmpty()) + return null; + + if (info != null && info.commandStoreId != journalKey.commandStoreId) info = null; + if (info == null) info = infos.get(journalKey.commandStoreId); + // TODO (required): should return null only if commandStore has been removed + if (info == null) + return null; + + DurableBefore durableBefore = infos.durableBefore; + Cleanup cleanup = mainBuilder.maybeCleanup(false, PARTIAL, info.redundantBefore, durableBefore); + if (cleanup != NO) + { + switch (cleanup) + { + default: throw new UnhandledEnum(cleanup); + case EXPUNGE: + return null; + case ERASE: + return PartitionUpdate.fullPartitionDelete(AccordKeyspace.Journal, partitionKey, Long.MAX_VALUE, nowInSec).unfilteredIterator(); + + case TRUNCATE: + case TRUNCATE_WITH_OUTCOME: + case INVALIDATE: + case VESTIGIAL: + for (int i = 0, size = entries.size(); i < size ; i++) + { + AccordCommandRowEntry entry = entries.get(i); + if (i == 0) entry.modified |= entry.builder.addCleanup(false, cleanup); + else entry.modified |= entry.builder.cleanup(false, cleanup); + } + } + } + + PartitionUpdate.Builder newVersion = new PartitionUpdate.Builder(AccordKeyspace.Journal, partitionKey, AccordKeyspace.JournalColumns.regular, entries.size()); + for (int i = 0, size = entries.size() ; i < size ; ++i) + { + AccordCommandRowEntry entry = entries.get(i); + if (!entry.modified) + { + newVersion.add(entry.row); + } + else if (entry.builder.flags() != 0) + { + Object[] newRow = rowTemplate.clone(); + newRow[0] = BufferCell.live(AccordKeyspace.JournalColumns.record, timestamp, entry.builder.asByteBuffer(userVersion)); + newRow[1] = userVersionCell; + newVersion.add(BTreeRow.create(entry.row.clustering(), entry.row.primaryKeyLivenessInfo(), entry.row.deletion(), newRow)); + } + } + return newVersion.build().unfilteredIterator(); } } @@ -1049,9 +1181,8 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte private static boolean requiresAccordSpecificPurger(ColumnFamilyStore cfs) { return cfs.getKeyspaceName().equals(SchemaConstants.ACCORD_KEYSPACE_NAME) && - ImmutableSet.of(AccordKeyspace.JOURNAL, - AccordKeyspace.COMMANDS_FOR_KEY) - .contains(cfs.getTableName()); + (cfs.getTableName().contains(AccordKeyspace.JOURNAL) || + AccordKeyspace.COMMANDS_FOR_KEY.equals(cfs.getTableName())); } private static boolean isAccordTable(ColumnFamilyStore cfs, String name) diff --git a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java index 65dc76a042..393e0ba75e 100644 --- a/src/java/org/apache/cassandra/schema/SchemaKeyspace.java +++ b/src/java/org/apache/cassandra/schema/SchemaKeyspace.java @@ -62,6 +62,7 @@ import static org.apache.cassandra.config.CassandraRelevantProperties.IGNORE_COR import static org.apache.cassandra.config.CassandraRelevantProperties.TEST_FLUSH_LOCAL_SCHEMA_CHANGES; import static org.apache.cassandra.cql3.QueryProcessor.executeInternal; import static org.apache.cassandra.cql3.QueryProcessor.executeOnceInternal; +import static org.apache.cassandra.schema.ColumnMetadata.NO_UNIQUE_ID; import static org.apache.cassandra.schema.SchemaKeyspaceTables.*; import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized; import static org.apache.cassandra.utils.LocalizeString.toUpperCaseLocalized; @@ -143,7 +144,6 @@ public final class SchemaKeyspace + "clustering_order text," + "column_name_bytes blob," + "kind text," - + "unique_id int," + "position int," + "type text," + "PRIMARY KEY ((keyspace_name), table_name, column_name))"); @@ -172,7 +172,6 @@ public final class SchemaKeyspace + "dropped_time timestamp," + "kind text," + "type text," - + "unique_id int," + "PRIMARY KEY ((keyspace_name), table_name, column_name))"); private static final TableMetadata Triggers = @@ -735,7 +734,6 @@ public final class SchemaKeyspace .add("column_name_bytes", column.name.bytes) .add("kind", toLowerCaseLocalized(column.kind.toString())) .add("position", column.position()) - .add("unique_id", column.uniqueId) .add("clustering_order", toLowerCaseLocalized(column.clusteringOrder().toString())) .add("type", type.asCQL3Type().toString()); @@ -1120,7 +1118,6 @@ public final class SchemaKeyspace ColumnMetadata.Kind kind = ColumnMetadata.Kind.valueOf(toUpperCaseLocalized(row.getString("kind"))); - int uniqueId = row.getInt("unique_id", ColumnMetadata.NO_UNIQUE_ID); int position = row.getInt("position"); ClusteringOrder order = ClusteringOrder.valueOf(toUpperCaseLocalized(row.getString("clustering_order"))); @@ -1176,7 +1173,7 @@ public final class SchemaKeyspace mask = new ColumnMask((ScalarFunction) function, values); } - return new ColumnMetadata(keyspace, table, name, type, uniqueId, position, kind, mask); + return new ColumnMetadata(keyspace, table, name, type, NO_UNIQUE_ID, position, kind, mask); } private static Map fetchDroppedColumns(String keyspace, String table) @@ -1196,7 +1193,6 @@ public final class SchemaKeyspace String keyspace = row.getString("keyspace_name"); String table = row.getString("table_name"); String name = row.getString("column_name"); - int uniqueId = row.getInt("unique_id", ColumnMetadata.NO_UNIQUE_ID); /* * we never store actual UDT names in dropped column types (so that we can safely drop types if nothing refers to * them anymore), so before storing dropped columns in schema we expand UDTs to tuples. See expandUserTypes method. @@ -1209,7 +1205,7 @@ public final class SchemaKeyspace assert kind == ColumnMetadata.Kind.REGULAR || kind == ColumnMetadata.Kind.STATIC : "Unexpected dropped column kind: " + kind; - ColumnMetadata column = new ColumnMetadata(keyspace, table, ColumnIdentifier.getInterned(name, true), type, uniqueId, ColumnMetadata.NO_POSITION, kind, null); + ColumnMetadata column = new ColumnMetadata(keyspace, table, ColumnIdentifier.getInterned(name, true), type, NO_UNIQUE_ID, ColumnMetadata.NO_POSITION, kind, null); long droppedTime = TimeUnit.MILLISECONDS.toMicros(row.getLong("dropped_time")); return new DroppedColumn(column, droppedTime); } diff --git a/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java b/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java index aa00a3d43d..668736c842 100644 --- a/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java +++ b/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java @@ -26,10 +26,20 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import accord.utils.Invariants; +import accord.utils.btree.BTree; +import accord.utils.btree.BulkIterator; +import accord.utils.btree.UpdateFunction; +import org.apache.cassandra.db.BufferClustering; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.DecoratedKey; +import org.apache.cassandra.db.LivenessInfo; +import org.apache.cassandra.db.marshal.Int32Type; +import org.apache.cassandra.db.marshal.LongType; import org.apache.cassandra.db.partitions.PartitionUpdate; -import org.apache.cassandra.db.partitions.PartitionUpdate.SimpleBuilder; +import org.apache.cassandra.db.rows.BTreeRow; +import org.apache.cassandra.db.rows.BufferCell; +import org.apache.cassandra.db.rows.ColumnData; +import org.apache.cassandra.db.rows.Row; import org.apache.cassandra.exceptions.UnknownTableException; import org.apache.cassandra.io.sstable.SSTableTxnWriter; import org.apache.cassandra.io.util.DataInputBuffer; @@ -37,6 +47,8 @@ import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.journal.SegmentCompactor; import org.apache.cassandra.journal.StaticSegment; import org.apache.cassandra.journal.StaticSegment.KeyOrderReader; +import org.apache.cassandra.service.ClientState; +import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightImage; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightSerializer; import org.apache.cassandra.utils.NoSpamLogger; @@ -51,12 +63,17 @@ public abstract class AbstractAccordSegmentCompactor implements SegmentCompac protected static final Logger logger = LoggerFactory.getLogger(AbstractAccordSegmentCompactor.class); private static final NoSpamLogger.NoSpamLogStatement unknownTable = NoSpamLogger.getStatement(logger, "Unknown (probably dropped) TableId {} reading {}; skipping record", 1L, MINUTES); + static final Object[] rowTemplate = BTree.build(BulkIterator.of(new Object[2]), 2, UpdateFunction.noOp); + protected final Version userVersion; + protected final ColumnData userVersionCell; protected final ColumnFamilyStore cfs; + protected final long timestamp = ClientState.getTimestamp(); public AbstractAccordSegmentCompactor(Version userVersion, ColumnFamilyStore cfs) { this.userVersion = userVersion; + this.userVersionCell = BufferCell.live(AccordKeyspace.JournalColumns.user_version, timestamp, Int32Type.instance.decompose(userVersion.version)); this.cfs = cfs; } @@ -96,8 +113,8 @@ public abstract class AbstractAccordSegmentCompactor implements SegmentCompac initializeWriter(); JournalKey key = null; - Object builder = null; - FlyweightSerializer serializer = null; + FlyweightImage builder = null; + FlyweightSerializer serializer = null; long firstDescriptor = -1, lastDescriptor = -1; int firstOffset = -1, lastOffset = -1; try @@ -110,8 +127,9 @@ public abstract class AbstractAccordSegmentCompactor implements SegmentCompac maybeWritePartition(key, builder, serializer, firstDescriptor, firstOffset); switchPartitions(); key = reader.key(); - serializer = (FlyweightSerializer) key.type.serializer; - builder = serializer.mergerFor(key); + serializer = (FlyweightSerializer) key.type.serializer; + builder = serializer.mergerFor(); + builder.reset(key); firstDescriptor = lastDescriptor = -1; firstOffset = lastOffset = -1; } @@ -122,7 +140,10 @@ public abstract class AbstractAccordSegmentCompactor implements SegmentCompac do { if (builder == null) - builder = serializer.mergerFor(key); + { + builder = serializer.mergerFor(); + builder.reset(key); + } try (DataInputBuffer in = new DataInputBuffer(reader.record(), false)) { @@ -178,30 +199,27 @@ public abstract class AbstractAccordSegmentCompactor implements SegmentCompac private JournalKey prevKey; private DecoratedKey prevDecoratedKey; - private void maybeWritePartition(JournalKey key, Object builder, FlyweightSerializer serializer, long descriptor, int offset) throws IOException + private void maybeWritePartition(JournalKey key, FlyweightImage builder, FlyweightSerializer serializer, long descriptor, int offset) throws IOException { if (builder != null) { DecoratedKey decoratedKey = AccordKeyspace.JournalColumns.decorate(key); - - if (prevKey != null) - { - Invariants.requireArgument((decoratedKey.compareTo(prevDecoratedKey) >= 0 ? 1 : -1) == (JournalKey.SUPPORT.compare(key, prevKey) >= 0 ? 1 : -1), - String.format("Partition key and JournalKey didn't have matching order, which may imply a serialization issue.\n%s (%s)\n%s (%s)", - key, decoratedKey, prevKey, prevDecoratedKey)); - } + Invariants.requireArgument(prevKey == null || ((decoratedKey.compareTo(prevDecoratedKey) >= 0 ? 1 : -1) == (JournalKey.SUPPORT.compare(key, prevKey) >= 0 ? 1 : -1)), + "Partition key and JournalKey didn't have matching order, which may imply a serialization issue.\n%s (%s)\n%s (%s)", + key, decoratedKey, prevKey, prevDecoratedKey); prevKey = key; prevDecoratedKey = decoratedKey; - SimpleBuilder partitionBuilder = PartitionUpdate.simpleBuilder(cfs.metadata(), decoratedKey); + Object[] rowData = rowTemplate.clone(); try (DataOutputBuffer out = DataOutputBuffer.scratchBuffer.get()) { serializer.reserialize(key, builder, out, userVersion); - partitionBuilder.row(descriptor, offset) - .add("record", out.asNewBuffer()) - .add("user_version", userVersion.version); + rowData[0] = BufferCell.live(AccordKeyspace.JournalColumns.record, timestamp, out.asNewBuffer()); } - writer().append(partitionBuilder.build().unfilteredIterator()); + rowData[1] = userVersionCell; + Row row = BTreeRow.create(BufferClustering.make(LongType.instance.decompose(descriptor), Int32Type.instance.decompose(offset)), LivenessInfo.EMPTY, Row.Deletion.LIVE, rowData); + PartitionUpdate update = PartitionUpdate.singleRowUpdate(AccordKeyspace.Journal, decoratedKey, row); + writer().append(update.unfilteredIterator()); } } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index dad05aa7a5..4e8629696a 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -220,9 +220,11 @@ public class AccordCommandStore extends CommandStore @Override public void markShardDurable(SafeCommandStore safeStore, TxnId globalSyncId, Ranges ranges, Status.Durability durability) { - store.snapshot(ranges, globalSyncId); + if (durability.compareTo(Status.Durability.UniversalOrInvalidated) >= 0) + store.snapshot(ranges, globalSyncId); super.markShardDurable(safeStore, globalSyncId, ranges, durability); - commandsForRanges.gcBefore(globalSyncId, ranges); + if (durability.compareTo(Status.Durability.UniversalOrInvalidated) >= 0) + commandsForRanges.gcBefore(globalSyncId, ranges); } @Override diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 0ac5c87807..f4159f5731 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -67,9 +67,9 @@ import org.apache.cassandra.journal.SegmentCompactor; import org.apache.cassandra.journal.StaticSegment; import org.apache.cassandra.journal.ValueSerializer; import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightImage; import org.apache.cassandra.service.accord.AccordJournalValueSerializers.IdentityAccumulator; import org.apache.cassandra.service.accord.JournalKey.JournalKeySupport; -import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate; import org.apache.cassandra.service.accord.serializers.CommandSerializers; import org.apache.cassandra.service.accord.serializers.CommandSerializers.ExecuteAtSerializer; @@ -109,21 +109,19 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier @VisibleForTesting protected final AccordJournalTable journalTable; private final Params params; - private final AccordAgent agent; Node node; enum Status { INITIALIZED, STARTING, REPLAY, STARTED, TERMINATING, TERMINATED } private volatile Status status = Status.INITIALIZED; - public AccordJournal(Params params, AccordAgent agent) + public AccordJournal(Params params) { - this(params, agent, new File(DatabaseDescriptor.getAccordJournalDirectory()), Keyspace.open(AccordKeyspace.metadata().name).getColumnFamilyStore(AccordKeyspace.JOURNAL)); + this(params, new File(DatabaseDescriptor.getAccordJournalDirectory()), Keyspace.open(AccordKeyspace.metadata().name).getColumnFamilyStore(AccordKeyspace.JOURNAL)); } @VisibleForTesting - public AccordJournal(Params params, AccordAgent agent, File directory, ColumnFamilyStore cfs) + public AccordJournal(Params params, File directory, ColumnFamilyStore cfs) { - this.agent = agent; Version userVersion = Version.fromVersion(params.userVersion()); this.journal = new Journal<>("AccordJournal", directory, params, JournalKey.SUPPORT, // In Accord, we are using streaming serialization, i.e. Reader/Writer interfaces instead of materializing objects @@ -232,7 +230,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier public Command loadCommand(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore) { Builder builder = load(commandStoreId, txnId); - builder.maybeCleanup(FULL, agent, redundantBefore, durableBefore); + builder.maybeCleanup(true, FULL, redundantBefore, durableBefore); return builder.construct(redundantBefore); } @@ -243,7 +241,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier if (builder.isEmpty()) return null; - Cleanup cleanup = builder.shouldCleanup(FULL, node.agent(), redundantBefore, durableBefore); + Cleanup cleanup = builder.shouldCleanup(FULL, redundantBefore, durableBefore); switch (cleanup) { case VESTIGIAL: @@ -384,9 +382,9 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier return loadDiffs(commandStoreId, txnId, Load.ALL); } - private BUILDER readAll(JournalKey key) + private BUILDER readAll(JournalKey key) { - BUILDER builder = (BUILDER) key.type.serializer.mergerFor(key); + BUILDER builder = (BUILDER) key.type.serializer.mergerFor(); // TODO: this can be further improved to avoid allocating lambdas AccordJournalValueSerializers.FlyweightSerializer serializer = (AccordJournalValueSerializers.FlyweightSerializer) key.type.serializer; // TODO (expected): for those where we store an image, read only the first entry we find in DESC order @@ -618,11 +616,16 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier } } - public static class Builder extends CommandChange.Builder + public static class Builder extends CommandChange.Builder implements FlyweightImage { public Builder() { - super(null, Load.ALL); + this(Load.ALL); + } + + public Builder(Load load) + { + super(null, load); } public Builder(TxnId txnId) @@ -634,6 +637,12 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier { super(txnId, load); } + + public void reset(JournalKey key) + { + reset(key.id); + } + public ByteBuffer asByteBuffer(Version userVersion) throws IOException { try (DataOutputBuffer out = new DataOutputBuffer()) @@ -728,11 +737,11 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier Invariants.require(txnId != null); int readFlags = in.readInt(); Invariants.require(readFlags != 0); - nextCalled = true; + hasUpdate = true; count++; // batch-apply any new nulls - setNulls(readFlags); + setNulls(false, readFlags); // iterator sets low 16 bits; low readFlag bits are nulls, so masking with ~readFlags restricts to non-null changed fields int iterable = toIterableSetFields(readFlags) & ~readFlags; for (Field field = nextSetField(iterable) ; field != null; field = nextSetField(iterable = unsetIterable(field, iterable))) diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java index 25e37e0b0b..e72beaab2c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalValueSerializers.java @@ -34,7 +34,6 @@ import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate; import org.apache.cassandra.service.accord.serializers.CommandStoreSerializers; import org.apache.cassandra.service.accord.serializers.Version; -import static accord.api.Journal.Load.ALL; import static accord.local.CommandStores.RangesForEpoch; // TODO (required): test with large collection values, and perhaps split out some fields if they have a tendency to grow larger @@ -42,9 +41,14 @@ import static accord.local.CommandStores.RangesForEpoch; // TODO (required): versioning public class AccordJournalValueSerializers { - public interface FlyweightSerializer + public interface FlyweightImage { - IMAGE mergerFor(JournalKey key); + void reset(JournalKey key); + } + + public interface FlyweightSerializer + { + IMAGE mergerFor(); void serialize(JournalKey key, ENTRY from, DataOutputPlus out, Version userVersion) throws IOException; @@ -57,9 +61,9 @@ public class AccordJournalValueSerializers implements FlyweightSerializer { @Override - public AccordJournal.Builder mergerFor(JournalKey journalKey) + public AccordJournal.Builder mergerFor() { - return new AccordJournal.Builder(journalKey.id, ALL); + return new AccordJournal.Builder(); } @Override @@ -91,7 +95,7 @@ public class AccordJournalValueSerializers } } - public abstract static class Accumulator + public abstract static class Accumulator implements FlyweightImage { protected A accumulated; @@ -115,10 +119,19 @@ public class AccordJournalValueSerializers public static class IdentityAccumulator extends Accumulator { + final T initial; boolean hasRead; public IdentityAccumulator(T initial) { super(initial); + this.initial = initial; + } + + @Override + public void reset(JournalKey key) + { + hasRead = false; + accumulated = initial; } @Override @@ -135,7 +148,7 @@ public class AccordJournalValueSerializers implements FlyweightSerializer> { @Override - public IdentityAccumulator mergerFor(JournalKey journalKey) + public IdentityAccumulator mergerFor() { return new IdentityAccumulator<>(RedundantBefore.EMPTY); } @@ -184,6 +197,12 @@ public class AccordJournalValueSerializers super(DurableBefore.EMPTY); } + @Override + public void reset(JournalKey key) + { + accumulated = DurableBefore.EMPTY; + } + @Override protected DurableBefore accumulate(DurableBefore oldValue, DurableBefore newValue) { @@ -194,7 +213,7 @@ public class AccordJournalValueSerializers public static class DurableBeforeSerializer implements FlyweightSerializer { - public DurableBeforeAccumulator mergerFor(JournalKey journalKey) + public DurableBeforeAccumulator mergerFor() { return new DurableBeforeAccumulator(); } @@ -231,7 +250,7 @@ public class AccordJournalValueSerializers implements FlyweightSerializer, IdentityAccumulator>> { @Override - public IdentityAccumulator> mergerFor(JournalKey key) + public IdentityAccumulator> mergerFor() { return new IdentityAccumulator<>(ImmutableSortedMap.of(TxnId.NONE, Ranges.EMPTY)); } @@ -259,7 +278,7 @@ public class AccordJournalValueSerializers implements FlyweightSerializer, IdentityAccumulator>> { @Override - public IdentityAccumulator> mergerFor(JournalKey key) + public IdentityAccumulator> mergerFor() { return new IdentityAccumulator<>(ImmutableSortedMap.of(Timestamp.NONE, Ranges.EMPTY)); } @@ -287,7 +306,7 @@ public class AccordJournalValueSerializers implements FlyweightSerializer> { public static final RangesForEpochSerializer instance = new RangesForEpochSerializer(); - public IdentityAccumulator mergerFor(JournalKey key) + public IdentityAccumulator mergerFor() { return new IdentityAccumulator<>(null); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index fed2683e32..efe9f39b25 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -21,6 +21,7 @@ package org.apache.cassandra.service.accord; import java.io.IOException; import java.nio.ByteBuffer; import java.util.ArrayList; +import java.util.Arrays; import java.util.Iterator; import java.util.List; import java.util.Set; @@ -552,6 +553,7 @@ public class AccordKeyspace public static final ColumnMetadata id = getColumn(Journal, "id"); public static final ColumnMetadata record = getColumn(Journal, "record"); public static final ColumnMetadata user_version = getColumn(Journal, "user_version"); + public static final RegularAndStaticColumns regular = new RegularAndStaticColumns(Columns.NONE, Columns.from(Arrays.asList(record, user_version))); public static DecoratedKey decorate(JournalKey key) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java index 90b906bfde..75f5031ea4 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordService.java +++ b/src/java/org/apache/cassandra/service/accord/AccordService.java @@ -296,7 +296,7 @@ public class AccordService implements IAccordService, Shutdownable final RequestCallbacks callbacks = new RequestCallbacks(time); this.scheduler = new AccordScheduler(); this.dataStore = new AccordDataStore(); - this.journal = new AccordJournal(DatabaseDescriptor.getAccord().journal, agent); + this.journal = new AccordJournal(DatabaseDescriptor.getAccord().journal); this.configService = new AccordConfigurationService(localId); this.fastPathCoordinator = AccordFastPathCoordinator.create(localId, configService); this.messageSink = new AccordMessageSink(agent, configService, callbacks); diff --git a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java index 86eff7369e..ff4a6e6ecb 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java @@ -45,8 +45,6 @@ import org.apache.cassandra.service.accord.serializers.KeySerializers; import org.apache.cassandra.service.accord.serializers.TopologySerializers; import org.apache.cassandra.service.accord.serializers.Version; -import static org.apache.cassandra.service.accord.JournalKey.Type.TOPOLOGY_UPDATE; - public interface AccordTopologyUpdate { Kind kind(); @@ -367,6 +365,12 @@ public interface AccordTopologyUpdate super(new TreeMap<>()); } + @Override + public void reset(JournalKey key) + { + accumulated = new TreeMap<>(); + } + @Override public void update(AccordTopologyUpdate newValue) { @@ -414,9 +418,8 @@ public interface AccordTopologyUpdate } @Override - public Accumulator mergerFor(JournalKey key) + public Accumulator mergerFor() { - Invariants.require(key.type == TOPOLOGY_UPDATE); return new Accumulator(); } diff --git a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java index a0580390de..03292d9d8c 100644 --- a/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java +++ b/src/java/org/apache/cassandra/utils/NativeSSTableLoaderClient.java @@ -213,9 +213,8 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client ColumnIdentifier name = new ColumnIdentifier(row.getBytes("column_name_bytes"), row.getString("column_name")); int position = row.getInt("position"); - int uniqueId = row.isNull("unique_id") ? ColumnMetadata.NO_UNIQUE_ID : row.getInt("unique_id"); org.apache.cassandra.schema.ColumnMetadata.Kind kind = ColumnMetadata.Kind.valueOf(toUpperCaseLocalized(row.getString("kind"))); - return new ColumnMetadata(keyspace, table, name, type, uniqueId, position, kind, null); + return new ColumnMetadata(keyspace, table, name, type, ColumnMetadata.NO_UNIQUE_ID, position, kind, null); } private static DroppedColumn createDroppedColumnFromRow(Row row, String keyspace, String table) @@ -223,8 +222,7 @@ public class NativeSSTableLoaderClient extends SSTableLoader.Client String name = row.getString("column_name"); AbstractType type = CQLTypeParser.parse(keyspace, row.getString("type"), Types.none()); ColumnMetadata.Kind kind = ColumnMetadata.Kind.valueOf(toUpperCaseLocalized(row.getString("kind"))); - int uniqueId = row.isNull("unique_id") ? ColumnMetadata.NO_UNIQUE_ID : row.getInt("unique_id"); - ColumnMetadata column = new ColumnMetadata(keyspace, table, ColumnIdentifier.getInterned(name, true), type, uniqueId, ColumnMetadata.NO_POSITION, kind, null); + ColumnMetadata column = new ColumnMetadata(keyspace, table, ColumnIdentifier.getInterned(name, true), type, ColumnMetadata.NO_UNIQUE_ID, ColumnMetadata.NO_POSITION, kind, null); long droppedTime = row.getTimestamp("dropped_time").getTime(); return new DroppedColumn(column, droppedTime); } diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java index f60026b892..fbf2a9c874 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java @@ -43,14 +43,15 @@ import accord.local.CommandStores; import accord.local.Node; import accord.primitives.EpochSupplier; import accord.utils.DefaultRandom; +import accord.utils.Invariants; import accord.utils.RandomSource; import org.apache.cassandra.ServerTestUtils; import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Directories; import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.compaction.ActiveCompactionsTracker; import org.apache.cassandra.db.compaction.CompactionController; import org.apache.cassandra.db.compaction.CompactionIterator; -import org.apache.cassandra.db.compaction.CompactionManager; import org.apache.cassandra.db.compaction.OperationType; import org.apache.cassandra.db.compaction.writers.CompactionAwareWriter; import org.apache.cassandra.db.compaction.writers.DefaultCompactionWriter; @@ -61,11 +62,12 @@ import org.apache.cassandra.io.sstable.format.SSTableReader; import org.apache.cassandra.io.util.File; import org.apache.cassandra.journal.SegmentCompactor; import org.apache.cassandra.journal.StaticSegment; +import org.apache.cassandra.journal.TestParams; import org.apache.cassandra.schema.Schema; +import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.Tables; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.service.accord.serializers.CommandSerializers; import org.apache.cassandra.service.accord.serializers.DepsSerializers; import org.apache.cassandra.service.accord.serializers.KeySerializers; @@ -163,22 +165,30 @@ public class AccordJournalBurnTest extends BurnTestBase operations, 10 + random.nextInt(30), new RandomDelayQueue.Factory(random).get(), - (node, agent, randomSource) -> { + (nodeId, randomSource) -> { try { File directory = new File(Files.createTempDirectory(Integer.toString(counter.incrementAndGet()))); directory.deleteRecursiveOnExit(); - ColumnFamilyStore cfs = ks.getColumnFamilyStore("journal_" + node); + ColumnFamilyStore cfs = ks.getColumnFamilyStore("journal_" + nodeId); cfs.disableAutoCompaction(); - AccordJournal journal = new AccordJournal(new AccordJournalTestParams() + AccordJournal journal = new AccordJournal(new TestParams() { @Override public int segmentSize() { - return 32 * 1024 * 1024; + return 1 * 1024 * 1024; } - }, new AccordAgent(), directory, cfs) + }, directory, cfs) { + @Override + public AccordJournal start(Node node) + { + super.start(node); + unsafeSetStarted(); + return this; + } + @Override public void saveCommand(int store, CommandUpdate update, @Nullable Runnable onFlush) { @@ -236,45 +246,48 @@ public class AccordJournalBurnTest extends BurnTestBase this.journal.closeCurrentSegmentForTestingIfNonEmpty(); this.journal.runCompactorForTesting(); - List all = new ArrayList<>(cfs.getLiveSSTables()); + Set orig = cfs.getLiveSSTables(); + List all = new ArrayList<>(orig); if (all.size() <= 1) return; - Set sstables = new HashSet<>(); - - int min, max; - while (true) + Set selected = new HashSet<>(); + int count = all.size(); + int removeCount = random.nextInt(1, count); + while (removeCount-- > 0) { - int tmp1 = randomSource.nextInt(0, all.size()); - int tmp2 = randomSource.nextInt(0, all.size()); - if (tmp1 != tmp2 && Math.abs(tmp1 - tmp2) >= 1) - { - min = Math.min(tmp1, tmp2); - max = Math.max(tmp1, tmp2); - break; - } + int removeIndex = random.nextInt(count); + SSTableReader reader = all.get(removeIndex); + if (reader == null) + continue; + all.set(removeIndex, null); + selected.add(reader); + --count; } - // Random subset - for (int i = min; i < max; i++) - sstables.add(all.get(i)); - List scanners = sstables.stream().map(SSTableReader::getScanner).collect(Collectors.toList()); + if (selected.isEmpty()) + return; + List scanners = selected.stream().map(SSTableReader::getScanner).collect(Collectors.toList()); Collection newSStables; - try (LifecycleTransaction txn = cfs.getTracker().tryModify(sstables, OperationType.COMPACTION); - CompactionController controller = new CompactionController(cfs, sstables, 0); - CompactionIterator ci = new CompactionIterator(OperationType.COMPACTION, scanners, controller, 0, nextTimeUUID())) + try (LifecycleTransaction txn = cfs.getTracker().tryModify(selected, OperationType.COMPACTION); + CompactionController controller = new CompactionController(cfs, selected, 0); + CompactionIterator ci = new CompactionIterator(OperationType.COMPACTION, + scanners, + controller, + 0, + nextTimeUUID(), + ActiveCompactionsTracker.NOOP, null, + () -> getCompactionInfo(node, cfs.getTableId()), + () -> Version.V1)) { - CompactionManager.instance.active.beginCompaction(ci); - try (CompactionAwareWriter writer = getCompactionAwareWriter(cfs, cfs.getDirectories(), txn, sstables)) + try (CompactionAwareWriter writer = getCompactionAwareWriter(cfs, cfs.getDirectories(), txn, selected)) { while (ci.hasNext()) - { writer.append(ci.next()); - ci.setTargetDirectory(writer.getSStableDirectory().path()); - } + ci.setTargetDirectory(writer.getSStableDirectory().path()); // point of no return newSStables = writer.finish(); } @@ -282,11 +295,9 @@ public class AccordJournalBurnTest extends BurnTestBase { throw new RuntimeException(e); } - finally - { - CompactionManager.instance.active.finishCompaction(ci); - } } + + Invariants.require(!orig.equals(cfs.getLiveSSTables())); } @@ -299,8 +310,6 @@ public class AccordJournalBurnTest extends BurnTestBase } }; - journal.start(null); - journal.unsafeSetStarted(); return journal; } catch (Throwable t) @@ -316,4 +325,18 @@ public class AccordJournalBurnTest extends BurnTestBase throw SimulationException.wrap(seed, t); } } + + public static IAccordService.AccordCompactionInfos getCompactionInfo(Node node, TableId tableId) + { + IAccordService.AccordCompactionInfos compactionInfos = new IAccordService.AccordCompactionInfos(node.durableBefore(), node.topology().minEpoch()); + node.commandStores().forEachCommandStore(commandStore -> { + compactionInfos.put(commandStore.id(), new IAccordService.AccordCompactionInfo(commandStore.id(), + commandStore.unsafeGetRedundantBefore(), + commandStore.unsafeGetRangesForEpoch(), + tableId)); + }); + return compactionInfos; + } + + } diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java index 6844f58827..8471fdd829 100644 --- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java +++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalCompactionTest.java @@ -45,9 +45,9 @@ import org.apache.cassandra.db.ColumnFamilyStore; import org.apache.cassandra.db.Keyspace; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.io.util.File; +import org.apache.cassandra.journal.TestParams; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.utils.AccordGenerators; import static accord.api.Journal.FieldUpdates; @@ -94,7 +94,7 @@ public class AccordJournalCompactionTest Gen> safeToReadGen = AccordGenerators.safeToReadGen(DatabaseDescriptor.getPartitioner()); Gen rangesForEpochGen = AccordGenerators.rangesForEpoch(DatabaseDescriptor.getPartitioner()); - AccordJournal journal = new AccordJournal(new AccordJournalTestParams() + AccordJournal journal = new AccordJournal(new TestParams() { @Override public int segmentSize() @@ -107,7 +107,7 @@ public class AccordJournalCompactionTest { return false; } - }, new AccordAgent()); + }); try { journal.start(null); @@ -164,4 +164,4 @@ public class AccordJournalCompactionTest Gen keyDepsGen = AccordGenerators.keyDepsGen(DatabaseDescriptor.getPartitioner()); return AccordGens.deps(keyDepsGen::next, (rs) -> Deps.NONE.rangeDeps); } -} \ No newline at end of file +} diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java index 9691756cc3..0d6d11216f 100644 --- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java +++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java @@ -36,6 +36,7 @@ import org.junit.Test; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import accord.api.Agent; import accord.api.Key; import accord.api.Result; import accord.local.CheckedCommands; @@ -75,6 +76,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterator; import org.apache.cassandra.distributed.shared.WithProperties; import org.apache.cassandra.io.sstable.ISSTableScanner; import org.apache.cassandra.io.sstable.format.SSTableReader; +import org.apache.cassandra.journal.TestParams; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.SchemaConstants; import org.apache.cassandra.schema.TableMetadata; @@ -265,7 +267,9 @@ public class CompactionAccordIteratorsTest IAccordService.AccordCompactionInfo compactionInfo = new IAccordService.AccordCompactionInfo(commandStore.id(), redundantBefore, commandStore.unsafeGetRangesForEpoch(), ((AccordCommandStore)commandStore).tableId()); IAccordService.AccordCompactionInfos compactionInfos = new IAccordService.AccordCompactionInfos(durableBefore, 0); compactionInfos.put(commandStore.id(), compactionInfo); + when(mockAccordService.agent()).thenReturn(mock(Agent.class)); when(mockAccordService.getCompactionInfo()).thenReturn(compactionInfos); + when(mockAccordService.journalConfiguration()).thenReturn(new TestParams()); return mockAccordService; } @@ -374,7 +378,7 @@ public class CompactionAccordIteratorsTest nextInputScanners.add(scanners.remove(random.nextInt(scanners.size()))); } try (CompactionController controller = new CompactionController(ColumnFamilyStore.getIfExists(ACCORD_KEYSPACE_NAME, cfs.name), Collections.emptySet(), 0); - CompactionIterator compactionIterator = new CompactionIterator(OperationType.COMPACTION, nextInputScanners, controller, FBUtilities.nowInSeconds(), null, ActiveCompactionsTracker.NOOP, null, () -> mockAccordService)) + CompactionIterator compactionIterator = new CompactionIterator(OperationType.COMPACTION, nextInputScanners, controller, FBUtilities.nowInSeconds(), null, ActiveCompactionsTracker.NOOP, null, mockAccordService)) { while (compactionIterator.hasNext()) { diff --git a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java index b2d9c07ad2..77d6e687d2 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordConfigurationServiceTest.java @@ -62,7 +62,6 @@ import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.schema.Tables; -import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.service.accord.journal.AccordTopologyUpdate; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ValidatingClusterMetadataService; @@ -229,7 +228,7 @@ public class AccordConfigurationServiceTest directory.deleteRecursiveOnExit(); Keyspace ks = Schema.instance.getKeyspaceInstance("system_accord"); ColumnFamilyStore cfs = ks.getColumnFamilyStore(AccordKeyspace.JOURNAL); - AccordJournal journal = new AccordJournal(new TestParams(), new AccordAgent(), directory, cfs); + AccordJournal journal = new AccordJournal(new TestParams(), directory, cfs); journal.start(null); journal.unsafeSetStarted(); return journal; diff --git a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java index 0445e3d75e..03666a777b 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordJournalOrderTest.java @@ -43,12 +43,12 @@ import org.apache.cassandra.config.CassandraRelevantProperties; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.ByteOrderedPartitioner; import org.apache.cassandra.io.util.File; +import org.apache.cassandra.journal.TestParams; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.schema.Schema; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.schema.TableMetadata; import org.apache.cassandra.service.StorageService; -import org.apache.cassandra.service.accord.api.AccordAgent; import org.apache.cassandra.service.accord.api.TokenKey; import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.utils.StorageCompatibilityMode; @@ -74,7 +74,7 @@ public class AccordJournalOrderTest { if (new File(DatabaseDescriptor.getAccordJournalDirectory()).exists()) ServerTestUtils.cleanupDirectory(DatabaseDescriptor.getAccordJournalDirectory()); - AccordJournal accordJournal = new AccordJournal(AccordJournalTestParams.INSTANCE, new AccordAgent()); + AccordJournal accordJournal = new AccordJournal(TestParams.INSTANCE); accordJournal.start(null); RandomSource randomSource = RandomSource.wrap(new Random(0)); TxnId id1 = AccordGens.txnIds().next(randomSource); diff --git a/test/unit/org/apache/cassandra/service/accord/AccordJournalTestParams.java b/test/unit/org/apache/cassandra/service/accord/AccordJournalTestParams.java deleted file mode 100644 index c9a14efd7f..0000000000 --- a/test/unit/org/apache/cassandra/service/accord/AccordJournalTestParams.java +++ /dev/null @@ -1,77 +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.concurrent.TimeUnit; - -import org.apache.cassandra.journal.Params; -import org.apache.cassandra.service.accord.serializers.Version; - -public class AccordJournalTestParams implements Params -{ - public static final AccordJournalTestParams INSTANCE = new AccordJournalTestParams(); - - @Override - public int segmentSize() - { - return 32 << 20; - } - - @Override - public FailurePolicy failurePolicy() - { - return FailurePolicy.STOP; - } - - @Override - public FlushMode flushMode() - { - return FlushMode.GROUP; - } - - @Override - public boolean enableCompaction() - { - return false; - } - - @Override - public long compactionPeriod(TimeUnit units) - { - return units.convert(60, TimeUnit.SECONDS); - } - - @Override - public long flushPeriod(TimeUnit units) - { - return units.convert(1, TimeUnit.SECONDS); - } - - @Override - public long periodicBlockPeriod(TimeUnit units) - { - return units.convert(2, TimeUnit.SECONDS); - } - - @Override - public int userVersion() - { - return Version.LATEST.version; - } -} diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java index 0cb57ad9c1..1bb8614ca4 100644 --- a/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java +++ b/test/unit/org/apache/cassandra/service/accord/AccordTestUtils.java @@ -353,7 +353,7 @@ public class AccordTestUtils ServerTestUtils.cleanupDirectory(DatabaseDescriptor.getAccordJournalDirectory()); AccordSpec.JournalSpec spec = new AccordSpec.JournalSpec(); spec.flushPeriod = new DurationSpec.IntSecondsBound(1); - AccordJournal journal = new AccordJournal(spec, agent); + AccordJournal journal = new AccordJournal(spec); journal.start(null); CommandStore.EpochUpdateHolder holder = new CommandStore.EpochUpdateHolder(); diff --git a/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java b/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java index d87ca5292f..6991290021 100644 --- a/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java +++ b/test/unit/org/apache/cassandra/service/accord/CommandChangeTest.java @@ -69,8 +69,14 @@ public class CommandChangeTest @Test public void allNull() { - int flags = getFlags(null, null); - assertMissing(flags, ALL); + int flags = getFlags(null, Command.NotDefined.uninitialised(TxnId.NONE)); + EnumSet missing = EnumSet.allOf(Field.class); + missing.remove(Field.SAVE_STATUS); + missing.remove(Field.PARTICIPANTS); + missing.remove(Field.PROMISED); + missing.remove(Field.ACCEPTED); + missing.remove(Field.DURABILITY); + assertMissing(flags, missing); } @Test diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index 9a68d127ce..ef7e92c6ea 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -31,7 +31,6 @@ import java.util.function.Predicate; import java.util.function.ToLongFunction; import javax.annotation.Nullable; -import accord.api.Agent; import accord.api.Journal; import accord.api.LocalListeners; import accord.api.ProgressLog; @@ -234,7 +233,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable } }; - this.journal = new DefaultJournal(nodeId, agent, rs.fork()); + this.journal = new DefaultJournal(nodeId, rs.fork()); this.commandStore = new AccordCommandStore(0, storeService, agent, @@ -465,9 +464,9 @@ public class SimulatedAccordCommandStore implements AutoCloseable private static class DefaultJournal extends InMemoryJournal implements RangeSearcher.Supplier { private final RouteInMemoryIndex index = new RouteInMemoryIndex<>(); - private DefaultJournal(Node.Id id, Agent agent, RandomSource rs) + private DefaultJournal(Node.Id id, RandomSource rs) { - super(id, agent, rs); + super(id, rs); } @Override