You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by dc...@apache.org on 2025/01/18 00:05:39 UTC
(cassandra) branch cep-15-accord updated: Migrate route index from commands table to journal, and drop the commands table
This is an automated email from the ASF dual-hosted git repository.
dcapwell pushed a commit to branch cep-15-accord
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-15-accord by this push:
new d834c5d634 Migrate route index from commands table to journal, and drop the commands table
d834c5d634 is described below
commit d834c5d6340b27521d938f18adafe9f6ebb89d87
Author: David Capwell <dc...@apache.org>
AuthorDate: Mon Dec 9 16:56:44 2024 -0800
Migrate route index from commands table to journal, and drop the commands table
patch by David Capwell; reviewed by Alex Petrov for CASSANDRA-20144
---
.../db/compaction/CompactionIterator.java | 121 +---
.../cassandra/db/virtual/AccordDebugKeyspace.java | 2 +-
.../index/accord/CheckpointIntervalArrayIndex.java | 6 +-
.../cassandra/index/accord/IndexMetrics.java | 2 +-
.../cassandra/index/accord/RangeMemoryIndex.java | 27 +-
.../cassandra/index/accord/RouteIndexFormat.java | 72 ++-
.../{RouteIndex.java => RouteJournalIndex.java} | 176 +++---
.../index/accord/RouteMemtableIndexManager.java | 16 +-
.../index/accord/RouteSecondaryIndexBuilder.java | 4 +-
.../cassandra/index/accord/RoutesSearcher.java | 185 ------
.../apache/cassandra/journal/ActiveSegment.java | 4 +-
.../org/apache/cassandra/journal/Compactor.java | 3 -
.../apache/cassandra/journal/InMemoryIndex.java | 7 +-
src/java/org/apache/cassandra/journal/Journal.java | 21 +-
src/java/org/apache/cassandra/journal/Segment.java | 5 +
.../org/apache/cassandra/journal/Segments.java | 2 +-
.../cassandra/service/accord/AccordCache.java | 2 +-
.../service/accord/AccordCommandStore.java | 25 +-
.../service/accord/AccordConfigurationService.java | 2 +-
.../cassandra/service/accord/AccordJournal.java | 60 +-
.../service/accord/AccordJournalTable.java | 180 +++++-
.../cassandra/service/accord/AccordKeyspace.java | 490 ++--------------
.../cassandra/service/accord/AccordService.java | 29 +-
.../service/accord/CommandsForRanges.java | 10 +-
.../cassandra/service/accord/IndexRange.java | 60 ++
.../cassandra/service/accord/RangeSearcher.java | 128 ++++
.../service/accord/RouteInMemoryIndex.java | 234 ++++++++
.../serializers/AccordRoutingKeyByteSource.java | 10 +-
.../org/apache/cassandra/utils/MergeIterator.java | 29 +
src/java/org/apache/cassandra/utils/RTree.java | 2 +-
.../fuzz/topology/AccordTopologyMixupTest.java | 1 +
.../topology/HarryOnAccordTopologyMixupTest.java | 3 +-
.../service/accord/AccordJournalBurnTest.java | 11 +-
.../concurrent/ForwardingExecutorFactory.java | 136 +++++
test/unit/org/apache/cassandra/cql3/CQLTester.java | 13 +-
.../compaction/CompactionAccordIteratorsTest.java | 152 +----
.../org/apache/cassandra/dht/IPartitionerTest.java | 5 +-
.../index/accord/AccordIndexStressTest.java | 527 -----------------
.../cassandra/index/accord/RouteIndexTest.java | 641 +++++++++++++--------
.../apache/cassandra/io/FileSystemListeners.java | 101 ++++
.../io/filesystem/ListenableFileSystem.java | 60 +-
.../service/accord/AccordCommandStoreTest.java | 1 -
.../service/accord/AccordKeyspaceTest.java | 4 -
.../cassandra/service/accord/AccordTaskTest.java | 9 -
.../accord/SimulatedAccordCommandStore.java | 38 +-
.../cassandra/utils/CassandraGeneratorsTest.java | 2 +-
46 files changed, 1737 insertions(+), 1881 deletions(-)
diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
index c740e69db9..663d33f1ba 100644
--- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
+++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java
@@ -35,17 +35,10 @@ import com.google.common.collect.Ordering;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import accord.api.Agent;
import accord.local.Cleanup;
import accord.local.CommandStores;
-import accord.local.CommandStores.RangesForEpoch;
import accord.local.DurableBefore;
import accord.local.RedundantBefore;
-import accord.local.StoreParticipants;
-import accord.primitives.SaveStatus;
-import accord.primitives.Status;
-import accord.primitives.Status.Durability;
-import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.agrona.collections.Int2ObjectHashMap;
import org.apache.cassandra.config.DatabaseDescriptor;
@@ -66,8 +59,6 @@ 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.Cell;
import org.apache.cassandra.db.rows.RangeTombstoneBoundMarker;
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
import org.apache.cassandra.db.rows.Row;
@@ -96,8 +87,6 @@ import org.apache.cassandra.service.accord.AccordJournal;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers;
import org.apache.cassandra.service.accord.AccordJournalValueSerializers.FlyweightSerializer;
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.AccordService;
import org.apache.cassandra.service.accord.IAccordService;
@@ -109,19 +98,11 @@ import org.apache.cassandra.service.paxos.uncommitted.PaxosRows;
import org.apache.cassandra.utils.TimeUUID;
import static accord.local.Cleanup.ERASE;
-import static accord.local.Cleanup.TRUNCATE_WITH_OUTCOME;
-import static accord.local.Cleanup.shouldCleanupPartial;
import static com.google.common.base.Preconditions.checkState;
import static java.util.concurrent.TimeUnit.MICROSECONDS;
import static org.apache.cassandra.config.Config.PaxosStatePurging.legacy;
import static org.apache.cassandra.config.DatabaseDescriptor.paxosStatePurging;
-import static org.apache.cassandra.service.accord.AccordKeyspace.CommandRows.expungePartial;
-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.deserializeDurabilityOrNull;
-import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeParticipantsOrNull;
-import static org.apache.cassandra.service.accord.AccordKeyspace.deserializeSaveStatusOrNull;
/**
* Merge multiple iterators over the content of sstable into a "compacted" iterator.
@@ -232,8 +213,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
if (!requiresAccordSpecificPurger(cfs))
return new Purger(controller, nowInSec);
- if (isAccordCommands(cfs))
- return new AccordCommandsPurger(accordService);
if (isAccordJournal(cfs))
return new AccordJournalPurger(accordService, cfs);
if (isAccordCommandsForKey(cfs))
@@ -796,98 +775,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
}
}
- class AccordCommandsPurger extends AbstractPurger
- {
- final Agent agent;
- final Int2ObjectHashMap<RedundantBefore> redundantBefores;
- final Int2ObjectHashMap<DurableBefore> durableBefores;
- final Int2ObjectHashMap<RangesForEpoch> ranges;
-
- int storeId;
- TxnId txnId;
-
- AccordCommandsPurger(Supplier<IAccordService> accordService)
- {
- IAccordService service = accordService.get();
- IAccordService.CompactionInfo compactionInfo = service.getCompactionInfo();
- this.agent = service.agent();
- this.redundantBefores = compactionInfo.redundantBefores;
- this.ranges = compactionInfo.ranges;
- this.durableBefores = compactionInfo.durableBefores;
- }
-
- protected void beginPartition(UnfilteredRowIterator partition)
- {
- ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
- storeId = CommandRows.getStoreId(partitionKeyComponents);
- txnId = CommandRows.getTxnId(partitionKeyComponents);
- }
-
- @Override
- protected Row applyToRow(Row row)
- {
- updateProgress();
-
- RedundantBefore redundantBefore = redundantBefores.get(storeId);
- DurableBefore durableBefore = durableBefores.get(storeId);
- // TODO (expected): if the store has been retired, this should return null
- if (redundantBefore == null)
- return row;
-
- Cell durabilityCell = row.getCell(CommandsColumns.durability);
- Durability durability = deserializeDurabilityOrNull(durabilityCell);
- Cell executeAtCell = row.getCell(CommandsColumns.execute_at);
- Cell participantsCell = row.getCell(CommandsColumns.participants);
- StoreParticipants participants = deserializeParticipantsOrNull(participantsCell);
- Cell statusCell = row.getCell(CommandsColumns.status);
- SaveStatus saveStatus = deserializeSaveStatusOrNull(statusCell);
-
- if (saveStatus == null)
- return row;
-
- if (saveStatus.is(Status.Invalidated))
- return saveStatusOnly(saveStatus, row, nowInSec);
-
- Cleanup cleanup = shouldCleanupPartial(agent, txnId, saveStatus, durability, participants,
- redundantBefore, durableBefore);
- switch (cleanup)
- {
- default: throw new AssertionError(String.format("Unexpected cleanup task: %s", cleanup));
- case EXPUNGE:
- return null;
-
- case EXPUNGE_PARTIAL:
- return expungePartial(row, durabilityCell, executeAtCell, participantsCell);
-
- case ERASE:
- // Emit a tombstone so if this is slicing the command and making it not possible to determine if it
- // can be truncated later it can still be dropped via the tombstone.
- // Eventually the tombstone can be dropped by `durableBefore.min(txnId) == Universal`
- // We can still encounter sliced command state just because compaction inputs are random
- return BTreeRow.emptyDeletedRow(row.clustering(), new Row.Deletion(DeletionTime.build(row.primaryKeyLivenessInfo().timestamp(), nowInSec), false));
-
- case VESTIGIAL:
- case INVALIDATE:
- return saveStatusOnly(cleanup.appliesIfNot, row, nowInSec);
-
- case TRUNCATE_WITH_OUTCOME:
- case TRUNCATE:
- return truncatedApply(cleanup.appliesIfNot, row, nowInSec, durability, durabilityCell, executeAtCell, participantsCell, cleanup == TRUNCATE_WITH_OUTCOME);
-
- case NO:
- // TODO (required): when we port this to journal, make sure to expunge extra fields beyond those we need to retain
- return row;
- }
- }
-
- @Override
- protected Row applyToStatic(Row row)
- {
- checkState(row.isStatic() && row.isEmpty());
- return row;
- }
- }
-
class AccordCommandsForKeyPurger extends AbstractPurger
{
final CommandsForKeyAccessor accessor;
@@ -1142,8 +1029,7 @@ 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.COMMANDS,
- AccordKeyspace.JOURNAL,
+ ImmutableSet.of(AccordKeyspace.JOURNAL,
AccordKeyspace.COMMANDS_FOR_KEY)
.contains(cfs.getTableName());
}
@@ -1158,11 +1044,6 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
return cfs.getKeyspaceName().equals(SchemaConstants.ACCORD_KEYSPACE_NAME) && cfs.name.startsWith(AccordKeyspace.JOURNAL);
}
- private static boolean isAccordCommands(ColumnFamilyStore cfs)
- {
- return isAccordTable(cfs, AccordKeyspace.COMMANDS);
- }
-
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 b01e736065..d8b81b866f 100644
--- a/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
+++ b/src/java/org/apache/cassandra/db/virtual/AccordDebugKeyspace.java
@@ -214,7 +214,7 @@ public class AccordDebugKeyspace extends VirtualKeyspace
{
try (AccordExecutor.ExclusiveGlobalCaches cache = executor.lockCaches())
{
- addRow(ds, executor.executorId(), AccordKeyspace.COMMANDS, cache.commands.statsSnapshot());
+ addRow(ds, executor.executorId(), "commands", cache.commands.statsSnapshot());
addRow(ds, executor.executorId(), AccordKeyspace.COMMANDS_FOR_KEY, cache.commandsForKey.statsSnapshot());
}
}
diff --git a/src/java/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndex.java b/src/java/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndex.java
index 2cee3ed260..9fbb3634b9 100644
--- a/src/java/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndex.java
+++ b/src/java/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndex.java
@@ -157,7 +157,7 @@ public class CheckpointIntervalArrayIndex
// write values
callback.preWalk(sortedIntervals);
int count = 0;
- for (var it : sortedIntervals)
+ for (Interval it : sortedIntervals)
{
validate(count, it);
@@ -207,7 +207,7 @@ public class CheckpointIntervalArrayIndex
this.fh = fh;
try (RandomAccessReader reader = fh.createReader();
- var in = new ChecksumedRandomAccessReader(reader, CHECKSUM_SUPPLIER))
+ ChecksumedRandomAccessReader in = new ChecksumedRandomAccessReader(reader, CHECKSUM_SUPPLIER))
{
if (pos != -1)
in.seek(pos);
@@ -401,7 +401,7 @@ public class CheckpointIntervalArrayIndex
this.maxScanAndCheckpointMatches = maxScanAndCheckpointMatches;
}
}
- var c = new CheckpointIntervalArrayBuilder<>(LIST_INTERVAL_ACCESSOR, sortedIntervals, ACCURATE, LINKS).build((ignore, bounds, headers, lists, max) -> new Checkpoints(bounds, headers, lists, max));
+ Checkpoints c = new CheckpointIntervalArrayBuilder<>(LIST_INTERVAL_ACCESSOR, sortedIntervals, ACCURATE, LINKS).build((ignore, bounds, headers, lists, max) -> new Checkpoints(bounds, headers, lists, max));
out.resetChecksum(); // reset checksum so it only covers this metadata
out.writeUnsignedVInt32(c.maxScanAndCheckpointMatches);
write(c.bounds);
diff --git a/src/java/org/apache/cassandra/index/accord/IndexMetrics.java b/src/java/org/apache/cassandra/index/accord/IndexMetrics.java
index d992c8a16b..cb12afe147 100644
--- a/src/java/org/apache/cassandra/index/accord/IndexMetrics.java
+++ b/src/java/org/apache/cassandra/index/accord/IndexMetrics.java
@@ -40,7 +40,7 @@ public class IndexMetrics
private final String indexName;
public final Timer memtableIndexWriteLatency;
- public IndexMetrics(RouteIndex index)
+ public IndexMetrics(RouteJournalIndex index)
{
this.ks = index.baseCfs().getKeyspaceName();
this.table = index.baseCfs().name;
diff --git a/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java b/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java
index 984fee05a2..637cf65b39 100644
--- a/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java
+++ b/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java
@@ -24,6 +24,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
+import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -122,7 +123,7 @@ public class RangeMemoryIndex
if (route == null || route.domain() != Routable.Domain.Range)
return 0;
long sum = 0;
- for (var keyOrRange : route)
+ for (Unseekable keyOrRange : route)
sum += add(key, keyOrRange);
return sum;
}
@@ -133,12 +134,12 @@ public class RangeMemoryIndex
throw new IllegalArgumentException("Unexpected domain: " + keyOrRange.domain());
TokenRange ts = (TokenRange) keyOrRange;
- var storeId = AccordKeyspace.CommandRows.getStoreId(key);
- var tableId = ts.table();
- var group = new Group(storeId, tableId);
- var start = OrderedRouteSerializer.serializeRoutingKeyNoTable(ts.start());
- var end = OrderedRouteSerializer.serializeRoutingKeyNoTable(ts.end());
- var range = new Range(start, end);
+ int storeId = AccordKeyspace.JournalColumns.getStoreId(key);
+ TableId tableId = ts.table();
+ Group group = new Group(storeId, tableId);
+ byte[] start = OrderedRouteSerializer.serializeRoutingKeyNoTable(ts.start());
+ byte[] end = OrderedRouteSerializer.serializeRoutingKeyNoTable(ts.end());
+ Range range = new Range(start, end);
map.computeIfAbsent(group, ignore -> createRangeTree()).add(range, key);
Metadata metadata = groupMetadata.computeIfAbsent(group, ignore -> new Metadata());
@@ -149,10 +150,10 @@ public class RangeMemoryIndex
public synchronized NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] start, boolean startInclusive, byte[] end, boolean endInclusive)
{
- var rangesToPks = map.get(new Group(storeId, tableId));
+ RangeTree<byte[], Range, DecoratedKey> rangesToPks = map.get(new Group(storeId, tableId));
if (rangesToPks == null || rangesToPks.isEmpty())
return Collections.emptyNavigableSet();
- var matches = search(rangesToPks, start, end);
+ TreeMap<Range, Set<DecoratedKey>> matches = search(rangesToPks, start, end);
if (matches.isEmpty())
return Collections.emptyNavigableSet();
TreeSet<ByteBuffer> pks = new TreeSet<>();
@@ -196,9 +197,9 @@ public class RangeMemoryIndex
List<Group> groups = new ArrayList<>(map.keySet());
groups.sort(Comparator.naturalOrder());
- for (var group : groups)
+ for (Group group : groups)
{
- var submap = map.get(group);
+ RangeTree<byte[], Range, DecoratedKey> submap = map.get(group);
if (submap.isEmpty()) // is this possible? put here for safty so list is never empty
continue;
Metadata metadata = groupMetadata.get(group);
@@ -210,8 +211,8 @@ public class RangeMemoryIndex
.sorted(Comparator.naturalOrder())
.collect(Collectors.toList());
- var writer = new CheckpointIntervalArrayIndex.SegmentWriter(id, list.get(0).start.length, list.get(0).value.length);
- var meta = writer.write(list.toArray(CheckpointIntervalArrayIndex.Interval[]::new));
+ CheckpointIntervalArrayIndex.SegmentWriter writer = new CheckpointIntervalArrayIndex.SegmentWriter(id, list.get(0).start.length, list.get(0).value.length);
+ EnumMap<IndexDescriptor.IndexComponent, Segment.ComponentMetadata> meta = writer.write(list.toArray(CheckpointIntervalArrayIndex.Interval[]::new));
if (meta.isEmpty()) // don't include empty segments
continue;
output.put(group, new Segment.Metadata(meta, metadata.minTerm, metadata.maxTerm));
diff --git a/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java b/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java
index bd722ca9f0..cd8d9a8c41 100644
--- a/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java
+++ b/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java
@@ -19,6 +19,7 @@
package org.apache.cassandra.index.accord;
import java.io.IOException;
+import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Comparator;
@@ -32,7 +33,10 @@ import java.util.zip.Checksum;
import com.google.common.collect.Maps;
+import accord.local.StoreParticipants;
+import accord.primitives.TxnId;
import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Unfiltered;
@@ -43,6 +47,10 @@ import org.apache.cassandra.io.util.ChecksumedSequentialWriter;
import org.apache.cassandra.io.util.FileHandle;
import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.serializers.UUIDSerializer;
+import org.apache.cassandra.service.accord.AccordJournal;
+import org.apache.cassandra.service.accord.AccordJournalTable;
+import org.apache.cassandra.service.accord.AccordKeyspace;
+import org.apache.cassandra.service.accord.JournalKey;
import org.apache.cassandra.utils.ByteArrayUtil;
import org.apache.cassandra.utils.Throwables;
@@ -61,10 +69,11 @@ public class RouteIndexFormat
public static class SSTableIndexWriter extends MemtableRouteIndexWriter
{
- private final RouteIndex index;
+ private final RouteJournalIndex index;
private DecoratedKey current;
+ private JournalKey journalKey;
- public SSTableIndexWriter(RouteIndex index, IndexDescriptor id)
+ public SSTableIndexWriter(RouteJournalIndex index, IndexDescriptor id)
{
super(id, new MemtableIndex());
this.index = index;
@@ -74,6 +83,7 @@ public class RouteIndexFormat
public void startPartition(DecoratedKey key, long keyPosition, long keyPositionForSASI)
{
this.current = key;
+ this.journalKey = AccordKeyspace.JournalColumns.getJournalKey(key);
}
@Override
@@ -81,16 +91,58 @@ public class RouteIndexFormat
{
// there is some duplication from org.apache.cassandra.index.accord.RouteMemtableIndexManager.index
// should this be cleaned up?
- if (!unfiltered.isRow())
+ if (!unfiltered.isRow() || !RouteJournalIndex.allowed(journalKey))
return;
Row row = (Row) unfiltered;
- // simplified version of org.apache.cassandra.index.sai.utils.IndexTermType.valueOf
- Cell<?> cell = row.getCell(index.column());
- ByteBuffer value = cell == null || !cell.isLive(nowInSeconds()) ? null : cell.buffer();
+ ByteBuffer value = extractParticipants(index, journalKey.id, row);
indexer.index(current, row.clustering(), value);
}
}
+ public static ByteBuffer extractParticipants(RouteJournalIndex index, TxnId txnId, Row row)
+ {
+ boolean recordNull = row.getCell(AccordKeyspace.JournalColumns.record) == null;
+ boolean userVersionNull = row.getCell(AccordKeyspace.JournalColumns.user_version) == null;
+ if (recordNull != userVersionNull)
+ throw new IllegalStateException(String.format("Record is %s, but user_version is %s",
+ (recordNull ? "null" : "defined"),
+ (userVersionNull ? "null" : "defined")));
+ if (recordNull)
+ return null;
+ Cell<?> recordCell = row.getCell(AccordKeyspace.JournalColumns.record);
+ Cell<?> user_versionCell = row.getCell(AccordKeyspace.JournalColumns.user_version);
+ long nowInSec = nowInSeconds();
+ boolean recordLive = recordCell.isLive(nowInSec);
+ boolean user_versionLive = user_versionCell.isLive(nowInSec);
+ if (recordLive != user_versionLive)
+ throw new IllegalStateException(String.format("Record is %s, but user_version is %s",
+ (recordLive ? "live" : "dead"),
+ (user_versionLive ? "live" : "dead")));
+ if (!recordLive)
+ return null;
+ ByteBuffer record = recordCell.buffer();
+ int user_version = Int32Type.instance.compose(user_versionCell.buffer());
+ AccordJournal.Builder builder = extract(txnId, record, user_version);
+ StoreParticipants participants = builder.participants();
+ if (participants == null)
+ return null;
+ try
+ {
+ return AccordKeyspace.LocalVersionedSerializers.serialize(participants);
+ }
+ catch (IOException e)
+ {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ public static AccordJournal.Builder extract(TxnId txnId, ByteBuffer record, int user_version)
+ {
+ AccordJournal.Builder builder = new AccordJournal.Builder(txnId, AccordJournal.Load.ALL);
+ AccordJournalTable.readBuffer(record, builder::deserializeNext, user_version);
+ return builder;
+ }
+
public static class MemtableRouteIndexWriter implements Writer
{
private final IndexDescriptor id;
@@ -170,8 +222,8 @@ public class RouteIndexFormat
{
List<Segment> segments = new ArrayList<>();
- try (var metaReader = new ChecksumedRandomAccessReader(index.get(IndexComponent.METADATA).createReader(), CHECKSUM_SUPPLIER);
- var segmentReader = new ChecksumedRandomAccessReader(index.get(IndexComponent.SEGMENT).createReader(), CHECKSUM_SUPPLIER))
+ try (ChecksumedRandomAccessReader metaReader = new ChecksumedRandomAccessReader(index.get(IndexComponent.METADATA).createReader(), CHECKSUM_SUPPLIER);
+ ChecksumedRandomAccessReader segmentReader = new ChecksumedRandomAccessReader(index.get(IndexComponent.SEGMENT).createReader(), CHECKSUM_SUPPLIER))
{
while (metaReader.getFilePointer() < metaReader.length())
{
@@ -218,8 +270,8 @@ public class RouteIndexFormat
List<Group> groups = new ArrayList<>(segment.groups.keySet());
groups.sort(Comparator.naturalOrder());
- try (var segmentWriter = ChecksumedSequentialWriter.open(id.fileFor(IndexComponent.SEGMENT), true, CHECKSUM_SUPPLIER);
- var metadataWriter = ChecksumedSequentialWriter.open(id.fileFor(IndexComponent.METADATA), true, CHECKSUM_SUPPLIER))
+ try (ChecksumedSequentialWriter segmentWriter = ChecksumedSequentialWriter.open(id.fileFor(IndexComponent.SEGMENT), true, CHECKSUM_SUPPLIER);
+ ChecksumedSequentialWriter metadataWriter = ChecksumedSequentialWriter.open(id.fileFor(IndexComponent.METADATA), true, CHECKSUM_SUPPLIER))
{
long startPointer = segmentWriter.getFilePointer();
for (Group group : groups)
diff --git a/src/java/org/apache/cassandra/index/accord/RouteIndex.java b/src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java
similarity index 90%
rename from src/java/org/apache/cassandra/index/accord/RouteIndex.java
rename to src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java
index 1ebeb54198..c03f840533 100644
--- a/src/java/org/apache/cassandra/index/accord/RouteIndex.java
+++ b/src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java
@@ -29,12 +29,14 @@ import java.util.Set;
import java.util.concurrent.Callable;
import java.util.stream.Collectors;
+import com.google.common.base.Splitter;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import accord.primitives.TxnId;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.Operator;
-import org.apache.cassandra.cql3.statements.schema.IndexTarget;
import org.apache.cassandra.db.CassandraWriteContext;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.ColumnFamilyStore;
@@ -62,7 +64,6 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.IndexRegistry;
-import org.apache.cassandra.index.TargetParser;
import org.apache.cassandra.index.sai.StorageAttachedIndex;
import org.apache.cassandra.index.transactions.IndexTransaction;
import org.apache.cassandra.io.sstable.Component;
@@ -81,25 +82,24 @@ 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.StorageService;
+import org.apache.cassandra.service.accord.AccordJournalTable;
import org.apache.cassandra.service.accord.AccordKeyspace;
+import org.apache.cassandra.service.accord.JournalKey;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.utils.AbstractIterator;
-import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.Future;
import org.apache.cassandra.utils.concurrent.FutureCombiner;
-public class RouteIndex implements Index, INotificationConsumer
+public class RouteJournalIndex implements Index, INotificationConsumer
{
public enum RegisterStatus
{PENDING, REGISTERED, UNREGISTERED}
- private static final Logger logger = LoggerFactory.getLogger(RouteIndex.class);
+ private static final Logger logger = LoggerFactory.getLogger(RouteJournalIndex.class);
private static final Component.Type type = Component.Type.createSingleton("AccordRoute", "AccordRoute.*.db", true, null);
private final ColumnFamilyStore baseCfs;
- private final ColumnMetadata column;
private final IndexMetadata indexMetadata;
private final IndexMetrics indexMetrics;
private final MemtableIndexManager memtableIndexManager;
@@ -108,35 +108,43 @@ public class RouteIndex implements Index, INotificationConsumer
private volatile boolean initBuildStarted = false;
private volatile RegisterStatus registerStatus = RegisterStatus.PENDING;
- public RouteIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ public RouteJournalIndex(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
{
- if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
- if (!AccordKeyspace.COMMANDS.equals(baseCfs.name))
- throw new IllegalArgumentException("Route index is only allowed for accord commands table; given " + baseCfs().metadata());
-
- TableMetadata tableMetadata = baseCfs.metadata();
- Pair<ColumnMetadata, IndexTarget.Type> target = TargetParser.parse(tableMetadata, indexMetadata);
- if (!AccordKeyspace.CommandsColumns.participants.name.equals(target.left.name))
- throw new IllegalArgumentException("Attempted to index the wrong column; needed " + AccordKeyspace.CommandsColumns.participants.name + " but given " + target.left.name);
-
- if (target.right != IndexTarget.Type.VALUES)
- throw new IllegalArgumentException("Attempted to index " + AccordKeyspace.CommandsColumns.participants.name + " with index type " + target.right + "; only " + IndexTarget.Type.VALUES + " is supported");
+ validateTargets(baseCfs, indexMetadata);
this.baseCfs = baseCfs;
+ // type is only IndexTarget.Type.VALUES
this.indexMetadata = indexMetadata;
+
this.memtableIndexManager = new RouteMemtableIndexManager(this);
this.sstableManager = new RouteSSTableManager();
this.indexMetrics = new IndexMetrics(this);
- this.column = target.left;
Tracker tracker = baseCfs.getTracker();
tracker.subscribe(this);
}
- public ColumnMetadata column()
+ public static boolean allowed(JournalKey id)
+ {
+ return id.type == JournalKey.Type.COMMAND_DIFF && allowed(id.id);
+ }
+
+ public static boolean allowed(TxnId id)
{
- return column;
+ return id.domain().isRange();
+ }
+
+ private static void validateTargets(ColumnFamilyStore baseCfs, IndexMetadata indexMetadata)
+ {
+ // this contains 2 columns....
+ if (!SchemaConstants.ACCORD_KEYSPACE_NAME.equals(baseCfs.getKeyspaceName()))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table; given " + baseCfs.metadata());
+ if (!AccordKeyspace.JOURNAL.equals(baseCfs.name))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table; given " + baseCfs.metadata());
+ Set<String> columns = Splitter.on(',').trimResults().omitEmptyStrings().splitToStream(indexMetadata.options.get("target")).collect(Collectors.toSet());
+ Set<String> expected = Set.of("record", "user_version");
+ if (!expected.equals(columns))
+ throw new IllegalArgumentException("Route index is only allowed for accord journal table, and on the record/user_value columns; given " + baseCfs.metadata() + " and columns " + columns);
}
public IndexMetrics indexMetrics()
@@ -144,6 +152,11 @@ public class RouteIndex implements Index, INotificationConsumer
return indexMetrics;
}
+ public RegisterStatus registerStatus()
+ {
+ return registerStatus;
+ }
+
public ColumnFamilyStore baseCfs()
{
return baseCfs;
@@ -182,8 +195,6 @@ public class RouteIndex implements Index, INotificationConsumer
@Override
public Callable<?> getInitializationTask()
{
- //TODO (now): in SAI startup doesn't validate... what are the downstream issues this can face? Corrupt indexes not being detected?
- boolean starting = StorageService.instance.isStarting();
return () -> {
if (baseCfs.indexManager.isIndexQueryable(this))
{
@@ -250,11 +261,6 @@ public class RouteIndex implements Index, INotificationConsumer
return status == Status.BUILD_SUCCEEDED || status == Status.UNKNOWN;
}
- public RegisterStatus registerStatus()
- {
- return registerStatus;
- }
-
@Override
public synchronized void register(IndexRegistry registry)
{
@@ -308,10 +314,6 @@ public class RouteIndex implements Index, INotificationConsumer
// only internal can write... so it must be valid no?
}
-
- //TODO (now): flesh this stuff out...
-
-
@Override
public SSTableFlushObserver getFlushObserver(Descriptor descriptor,
LifecycleNewTracker tracker)
@@ -383,7 +385,7 @@ public class RouteIndex implements Index, INotificationConsumer
Integer storeId = null;
for (RowFilter.Expression e : expressions)
{
- if (e.column() == AccordKeyspace.CommandsColumns.participants)
+ if (e.column() == AccordJournalTable.SyntheticColumn.participants.metadata)
{
switch (e.operator())
{
@@ -407,10 +409,14 @@ public class RouteIndex implements Index, INotificationConsumer
return null;
}
}
- else if (e.column() == AccordKeyspace.CommandsColumns.store_id && e.operator() == Operator.EQ)
+ else if (e.column() == AccordJournalTable.SyntheticColumn.store_id.metadata && e.operator() == Operator.EQ)
{
storeId = Int32Type.instance.compose(e.getIndexValue());
}
+ else
+ {
+ throw new IllegalArgumentException("Unexpected expression: " + e.toCQLString());
+ }
}
if (start == null || end == null || storeId == null)
return null;
@@ -435,7 +441,7 @@ public class RouteIndex implements Index, INotificationConsumer
// find all partitions from memtable / sstable
NavigableSet<ByteBuffer> partitions = search(storeId, key);
// do SinglePartitionReadCommand per partition
- return new SearchIterator(executionController, command, partitions);
+ return new SearchIterator(command, partitions);
}
NavigableSet<ByteBuffer> search(int storeId, ByteBuffer key)
@@ -471,7 +477,7 @@ public class RouteIndex implements Index, INotificationConsumer
// find all partitions from memtable / sstable
NavigableSet<ByteBuffer> partitions = search(storeId, start, startInclusive, end, endInclusive);
// do SinglePartitionReadCommand per partition
- return new SearchIterator(executionController, command, partitions);
+ return new SearchIterator(command, partitions);
}
NavigableSet<ByteBuffer> search(int storeId,
@@ -494,17 +500,57 @@ public class RouteIndex implements Index, INotificationConsumer
};
}
+ @Override
+ public void handleNotification(INotification notification, Object sender)
+ {
+ // unfortunately, we can only check the type of notification via instanceof :(
+ if (notification instanceof SSTableAddedNotification)
+ {
+ SSTableAddedNotification notice = (SSTableAddedNotification) notification;
+ sstableManager.onSSTableChanged(Collections.emptySet(), notice.added);
+ }
+ else if (notification instanceof SSTableListChangedNotification)
+ {
+ SSTableListChangedNotification notice = (SSTableListChangedNotification) notification;
+ sstableManager.onSSTableChanged(notice.removed, notice.added);
+ }
+ else if (notification instanceof MemtableRenewedNotification)
+ {
+ memtableIndexManager.renewMemtable(((MemtableRenewedNotification) notification).renewed);
+ }
+ else if (notification instanceof MemtableDiscardedNotification)
+ {
+ memtableIndexManager.discardMemtable(((MemtableDiscardedNotification) notification).memtable);
+ }
+ }
+
+ //TODO (coverage): everything below here never triggered...
+
+ @Override
+ public boolean dependsOn(ColumnMetadata column)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public AbstractType<?> customExpressionValueType()
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long getEstimatedResultRows()
+ {
+ throw new UnsupportedOperationException();
+ }
+
private class SearchIterator extends AbstractIterator<UnfilteredRowIterator> implements UnfilteredPartitionIterator
{
- private final ReadExecutionController executionController;
- private final ReadCommand command;
private final TableMetadata metadata;
private final Iterator<ByteBuffer> partitions;
- private SearchIterator(ReadExecutionController executionController, ReadCommand command, NavigableSet<ByteBuffer> partitions)
+ private SearchIterator(ReadCommand command, NavigableSet<ByteBuffer> partitions)
{
- this.executionController = executionController;
- this.command = command;
this.metadata = command.metadata();
this.partitions = partitions.iterator();
}
@@ -595,48 +641,4 @@ public class RouteIndex implements Index, INotificationConsumer
}
}
-
- @Override
- public void handleNotification(INotification notification, Object sender)
- {
- // unfortunately, we can only check the type of notification via instanceof :(
- if (notification instanceof SSTableAddedNotification)
- {
- SSTableAddedNotification notice = (SSTableAddedNotification) notification;
- sstableManager.onSSTableChanged(Collections.emptySet(), notice.added);
- }
- else if (notification instanceof SSTableListChangedNotification)
- {
- SSTableListChangedNotification notice = (SSTableListChangedNotification) notification;
- sstableManager.onSSTableChanged(notice.removed, notice.added);
- }
- else if (notification instanceof MemtableRenewedNotification)
- {
- memtableIndexManager.renewMemtable(((MemtableRenewedNotification) notification).renewed);
- }
- else if (notification instanceof MemtableDiscardedNotification)
- {
- memtableIndexManager.discardMemtable(((MemtableDiscardedNotification) notification).memtable);
- }
- }
-
- //TODO (coverage): everything below here never triggered...
-
- @Override
- public boolean dependsOn(ColumnMetadata column)
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public AbstractType<?> customExpressionValueType()
- {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public long getEstimatedResultRows()
- {
- throw new UnsupportedOperationException();
- }
}
diff --git a/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java b/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
index a03eda1182..f979ea5cfe 100644
--- a/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
+++ b/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
@@ -28,19 +28,19 @@ import java.util.concurrent.TimeUnit;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
import org.apache.cassandra.db.memtable.Memtable;
-import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordKeyspace;
+import org.apache.cassandra.service.accord.JournalKey;
import static org.apache.cassandra.utils.Clock.Global.nanoTime;
-import static org.apache.cassandra.utils.Clock.Global.nowInSeconds;
public class RouteMemtableIndexManager implements MemtableIndexManager
{
private final ConcurrentMap<Memtable, MemtableIndex> liveMemtableIndexMap = new ConcurrentHashMap<>();
- private final RouteIndex index;
+ private final RouteJournalIndex index;
- public RouteMemtableIndexManager(RouteIndex index)
+ public RouteMemtableIndexManager(RouteJournalIndex index)
{
this.index = index;
}
@@ -50,6 +50,9 @@ public class RouteMemtableIndexManager implements MemtableIndexManager
{
if (row.isStatic())
return 0;
+ JournalKey journalKey = AccordKeyspace.JournalColumns.getJournalKey(key);
+ if (!RouteJournalIndex.allowed(journalKey))
+ return 0;
//TODO (performance): we dropped jdk8 and this was fixed in jdk8... so do we need to do this still?
MemtableIndex current = liveMemtableIndexMap.get(mt);
@@ -63,10 +66,7 @@ public class RouteMemtableIndexManager implements MemtableIndexManager
long bytes = 0;
- // simplified version of org.apache.cassandra.index.sai.utils.IndexTermType.valueOf
- Cell<?> cell = row.getCell(index.column());
- ByteBuffer value = cell == null || !cell.isLive(nowInSeconds()) ? null : cell.buffer();
-
+ ByteBuffer value = RouteIndexFormat.extractParticipants(index, journalKey.id, row);
bytes += target.index(key, row.clustering(), value);
index.indexMetrics().memtableIndexWriteLatency.update(nanoTime() - start, TimeUnit.NANOSECONDS);
return bytes;
diff --git a/src/java/org/apache/cassandra/index/accord/RouteSecondaryIndexBuilder.java b/src/java/org/apache/cassandra/index/accord/RouteSecondaryIndexBuilder.java
index 8caf981353..428ed774c0 100644
--- a/src/java/org/apache/cassandra/index/accord/RouteSecondaryIndexBuilder.java
+++ b/src/java/org/apache/cassandra/index/accord/RouteSecondaryIndexBuilder.java
@@ -49,7 +49,7 @@ public class RouteSecondaryIndexBuilder extends SecondaryIndexBuilder
private static final Logger logger = LoggerFactory.getLogger(RouteSecondaryIndexBuilder.class);
private final TimeUUID compactionId = nextTimeUUID();
- private final RouteIndex index;
+ private final RouteJournalIndex index;
private final TableMetadata metadata;
private final Tracker tracker;
private final SSTableManager sstableManager;
@@ -59,7 +59,7 @@ public class RouteSecondaryIndexBuilder extends SecondaryIndexBuilder
private final long totalSizeInBytes;
private long bytesProcessed = 0;
- public RouteSecondaryIndexBuilder(RouteIndex index,
+ public RouteSecondaryIndexBuilder(RouteJournalIndex index,
SSTableManager sstableManager,
List<SSTableReader> sstables,
boolean isFullRebuild,
diff --git a/src/java/org/apache/cassandra/index/accord/RoutesSearcher.java b/src/java/org/apache/cassandra/index/accord/RoutesSearcher.java
deleted file mode 100644
index 5955b7ff0e..0000000000
--- a/src/java/org/apache/cassandra/index/accord/RoutesSearcher.java
+++ /dev/null
@@ -1,185 +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.index.accord;
-
-import java.nio.ByteBuffer;
-import java.util.Iterator;
-import java.util.function.Consumer;
-
-import accord.primitives.Timestamp;
-import accord.primitives.TxnId;
-import org.apache.cassandra.cql3.Operator;
-import org.apache.cassandra.db.ColumnFamilyStore;
-import org.apache.cassandra.db.DataRange;
-import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.db.PartitionRangeReadCommand;
-import org.apache.cassandra.db.ReadExecutionController;
-import org.apache.cassandra.db.filter.ColumnFilter;
-import org.apache.cassandra.db.filter.DataLimits;
-import org.apache.cassandra.db.filter.RowFilter;
-import org.apache.cassandra.db.marshal.Int32Type;
-import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
-import org.apache.cassandra.db.rows.UnfilteredRowIterator;
-import org.apache.cassandra.index.Index;
-import org.apache.cassandra.schema.ColumnMetadata;
-import org.apache.cassandra.service.accord.AccordKeyspace;
-import org.apache.cassandra.service.accord.TokenRange;
-import org.apache.cassandra.service.accord.api.AccordRoutingKey;
-import org.apache.cassandra.utils.CloseableIterator;
-import org.apache.cassandra.utils.FBUtilities;
-
-public class RoutesSearcher
-{
- private final ColumnFamilyStore cfs = Keyspace.open("system_accord").getColumnFamilyStore("commands");
- private final Index index = cfs.indexManager.getIndexByName("route");
- private final ColumnMetadata participants = AccordKeyspace.CommandsColumns.participants;
- private final ColumnMetadata store_id = AccordKeyspace.CommandsColumns.store_id;
- private final ColumnMetadata txn_id = AccordKeyspace.CommandsColumns.txn_id;
- private final ColumnFilter columnFilter = ColumnFilter.selectionBuilder().add(store_id).add(txn_id).build();
- private final DataLimits limits = DataLimits.NONE;
- private final DataRange dataRange = DataRange.allData(cfs.getPartitioner());
-
- private CloseableIterator<Entry> searchRange(int store, AccordRoutingKey start, AccordRoutingKey end)
- {
- RowFilter rowFilter = RowFilter.create(false);
- rowFilter.add(participants, Operator.GT, OrderedRouteSerializer.serializeRoutingKey(start));
- rowFilter.add(participants, Operator.LTE, OrderedRouteSerializer.serializeRoutingKey(end));
- rowFilter.add(store_id, Operator.EQ, Int32Type.instance.decompose(store));
-
- var cmd = PartitionRangeReadCommand.create(cfs.metadata(),
- FBUtilities.nowInSeconds(),
- columnFilter,
- rowFilter,
- limits,
- dataRange);
- Index.Searcher s = index.searcherFor(cmd);
- try (var controller = cmd.executionController())
- {
- UnfilteredPartitionIterator partitionIterator = s.search(controller);
- return new CloseableIterator<>()
- {
- private final Entry entry = new Entry();
- @Override
- public void close()
- {
- partitionIterator.close();
- }
-
- @Override
- public boolean hasNext()
- {
- return partitionIterator.hasNext();
- }
-
- @Override
- public Entry next()
- {
- UnfilteredRowIterator next = partitionIterator.next();
- var partitionKeyComponents = AccordKeyspace.CommandRows.splitPartitionKey(next.partitionKey());
- entry.store_id = AccordKeyspace.CommandRows.getStoreId(partitionKeyComponents);
- entry.txnId = AccordKeyspace.CommandRows.getTxnId(partitionKeyComponents);
- return entry;
- }
- };
- }
- }
-
- private CloseableIterator<Entry> searchKey(int store, AccordRoutingKey key)
- {
- RowFilter rowFilter = RowFilter.create(false);
- rowFilter.add(participants, Operator.GTE, OrderedRouteSerializer.serializeRoutingKey(key));
- rowFilter.add(participants, Operator.LTE, OrderedRouteSerializer.serializeRoutingKey(key));
- rowFilter.add(store_id, Operator.EQ, Int32Type.instance.decompose(store));
-
- var cmd = PartitionRangeReadCommand.create(cfs.metadata(),
- FBUtilities.nowInSeconds(),
- columnFilter,
- rowFilter,
- limits,
- dataRange);
- Index.Searcher s = index.searcherFor(cmd);
- try (ReadExecutionController controller = cmd.executionController())
- {
- UnfilteredPartitionIterator partitionIterator = s.search(controller);
- return new CloseableIterator<>()
- {
- private final Entry entry = new Entry();
- @Override
- public void close()
- {
- partitionIterator.close();
- }
-
- @Override
- public boolean hasNext()
- {
- return partitionIterator.hasNext();
- }
-
- @Override
- public Entry next()
- {
- UnfilteredRowIterator next = partitionIterator.next();
- ByteBuffer[] partitionKeyComponents = AccordKeyspace.CommandRows.splitPartitionKey(next.partitionKey());
- entry.store_id = AccordKeyspace.CommandRows.getStoreId(partitionKeyComponents);
- entry.txnId = AccordKeyspace.CommandRows.getTxnId(partitionKeyComponents);
- return entry;
- }
- };
- }
- }
-
- public void intersects(int storeId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
- {
- intersects(storeId, range.start(), range.end(), minTxnId, maxTxnId, forEach);
- }
-
- void intersects(int storeId, AccordRoutingKey start, AccordRoutingKey end, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
- {
- try (CloseableIterator<RoutesSearcher.Entry> it = searchRange(storeId, start, end))
- {
- consume(it, storeId, minTxnId, maxTxnId, forEach);
- }
- }
-
- public void intersects(int storeId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
- {
- try (CloseableIterator<RoutesSearcher.Entry> it = searchKey(storeId, key))
- {
- consume(it, storeId, minTxnId, maxTxnId, forEach);
- }
- }
-
- private void consume(Iterator<Entry> it, int storeId, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
- {
- while (it.hasNext())
- {
- Entry next = it.next();
- if (next.store_id != storeId) continue; // the index should filter out, but just in case...
- if (next.txnId.compareTo(minTxnId) >= 0 && next.txnId.compareTo(maxTxnId) < 0)
- forEach.accept(next.txnId);
- }
- }
-
- private static final class Entry
- {
- public int store_id;
- public TxnId txnId;
- }
-}
diff --git a/src/java/org/apache/cassandra/journal/ActiveSegment.java b/src/java/org/apache/cassandra/journal/ActiveSegment.java
index f9ff3fd937..9e7d87b6e7 100644
--- a/src/java/org/apache/cassandra/journal/ActiveSegment.java
+++ b/src/java/org/apache/cassandra/journal/ActiveSegment.java
@@ -37,7 +37,7 @@ import org.apache.cassandra.utils.concurrent.WaitQueue;
import static org.apache.cassandra.utils.Simulate.With.MONITORS;
@Simulate(with=MONITORS)
-final class ActiveSegment<K, V> extends Segment<K, V>
+public final class ActiveSegment<K, V> extends Segment<K, V>
{
final FileChannel channel;
@@ -98,7 +98,7 @@ final class ActiveSegment<K, V> extends Segment<K, V>
}
@Override
- InMemoryIndex<K> index()
+ public InMemoryIndex<K> index()
{
return index;
}
diff --git a/src/java/org/apache/cassandra/journal/Compactor.java b/src/java/org/apache/cassandra/journal/Compactor.java
index 0062b7471d..6525df5d05 100644
--- a/src/java/org/apache/cassandra/journal/Compactor.java
+++ b/src/java/org/apache/cassandra/journal/Compactor.java
@@ -76,9 +76,6 @@ public final class Compactor<K, V> implements Runnable, Shutdownable
try
{
Collection<StaticSegment<K, V>> newSegments = segmentCompactor.compact(toCompact);
- // No-op compaction
- if (newSegments == null)
- return;
for (StaticSegment<K, V> segment : newSegments)
toCompact.remove(segment);
diff --git a/src/java/org/apache/cassandra/journal/InMemoryIndex.java b/src/java/org/apache/cassandra/journal/InMemoryIndex.java
index baa56853cc..49fe4d1367 100644
--- a/src/java/org/apache/cassandra/journal/InMemoryIndex.java
+++ b/src/java/org/apache/cassandra/journal/InMemoryIndex.java
@@ -32,7 +32,7 @@ import org.apache.cassandra.journal.StaticSegment.SequentialReader;
/**
* An index for a segment that's still being updated by journal writers concurrently.
*/
-final class InMemoryIndex<K> extends Index<K>
+public final class InMemoryIndex<K> extends Index<K>
{
private static final long[] EMPTY = new long[0];
@@ -55,6 +55,11 @@ final class InMemoryIndex<K> extends Index<K>
this.lastId = new AtomicReference<>();
}
+ public int size()
+ {
+ return index.size();
+ }
+
public void update(K id, int offset, int size)
{
long currentOffsetAndSize = composeOffsetAndSize(offset, size);
diff --git a/src/java/org/apache/cassandra/journal/Journal.java b/src/java/org/apache/cassandra/journal/Journal.java
index 14bafa375d..19d1c1d78f 100644
--- a/src/java/org/apache/cassandra/journal/Journal.java
+++ b/src/java/org/apache/cassandra/journal/Journal.java
@@ -754,7 +754,7 @@ public class Journal<K, V> implements Shutdownable
return oldest;
}
- ActiveSegment<K, V> currentActiveSegment()
+ public ActiveSegment<K, V> currentActiveSegment()
{
return currentSegment;
}
@@ -906,8 +906,19 @@ public class Journal<K, V> implements Shutdownable
@VisibleForTesting
public void truncateForTesting()
{
- advanceSegment(null);
- segments.set(Segments.none());
+ ActiveSegment<?, ?> discarding = currentSegment;
+ if (!discarding.isEmpty()) // if there is no data in the segement then ignore it
+ {
+ closeCurrentSegmentForTestingIfNonEmpty();
+ //TODO (desired): wait for the ActiveSegment to get released, else can see weird race conditions;
+ // this thread will see the static segmenet and will release it (which will delete the file),
+ // and the sync thread will then try to release and will fail as the file no longer exists...
+ while (discarding.selfRef().globalCount() > 0) {}
+ }
+
+ Segments<K, V> statics = swapSegments(s -> s.select(Segment::isActive)).select(Segment::isStatic);
+ for (Segment<K, V> segment : statics.all())
+ ((StaticSegment) segment).discard(this);
}
public interface Writer
@@ -939,6 +950,8 @@ public class Journal<K, V> implements Shutdownable
StaticSegment.KeyOrderReader<K> reader = staticSegment.keyOrderReader();
if (reader.advance())
this.readers.add(reader);
+ else
+ reader.close();
}
}
@@ -962,6 +975,8 @@ public class Journal<K, V> implements Shutdownable
reader.accept(next.descriptor.timestamp, next.offset, next.key(), next.record(), next.hosts(), next.descriptor.userVersion);
if (next.advance())
readers.add(next);
+ else
+ next.close();
}
}
diff --git a/src/java/org/apache/cassandra/journal/Segment.java b/src/java/org/apache/cassandra/journal/Segment.java
index 6f8cde0358..2e441a1398 100644
--- a/src/java/org/apache/cassandra/journal/Segment.java
+++ b/src/java/org/apache/cassandra/journal/Segment.java
@@ -71,6 +71,11 @@ public abstract class Segment<K, V> implements SelfRefCounted<Segment<K, V>>, Co
abstract ActiveSegment<K, V> asActive();
abstract StaticSegment<K, V> asStatic();
+ public long id()
+ {
+ return descriptor.timestamp;
+ }
+
/*
* Reading entries (by id, by offset, iterate)
*/
diff --git a/src/java/org/apache/cassandra/journal/Segments.java b/src/java/org/apache/cassandra/journal/Segments.java
index 6601d75549..7d63b86481 100644
--- a/src/java/org/apache/cassandra/journal/Segments.java
+++ b/src/java/org/apache/cassandra/journal/Segments.java
@@ -75,7 +75,7 @@ class Segments<K, V>
Invariants.checkArgument(activeSegment.descriptor.equals(staticSegment.descriptor));
Long2ObjectHashMap<Segment<K, V>> newSegments = new Long2ObjectHashMap<>(segments);
Segment<K, V> oldValue = newSegments.put(staticSegment.descriptor.timestamp, staticSegment);
- Invariants.checkState(oldValue == activeSegment);
+ Invariants.checkState(oldValue == activeSegment, () -> String.format("old value %s != new %s", oldValue, activeSegment));
return new Segments<>(newSegments);
}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCache.java b/src/java/org/apache/cassandra/service/accord/AccordCache.java
index cbdba5224f..78fb502c59 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCache.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCache.java
@@ -1191,7 +1191,7 @@ public class AccordCache implements CacheSize
return null;
}
- return commandStore.appendToKeyspace(txnId, value);
+ return null;
}
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
index 667199c6e1..25f09d2347 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java
@@ -55,7 +55,6 @@ import accord.primitives.PartialTxn;
import accord.primitives.Participants;
import accord.primitives.RangeDeps;
import accord.primitives.Ranges;
-import accord.primitives.Routable;
import accord.primitives.RoutableKey;
import accord.primitives.Status;
import accord.primitives.Timestamp;
@@ -63,7 +62,6 @@ import accord.primitives.TxnId;
import accord.utils.Invariants;
import accord.utils.async.AsyncChain;
import accord.utils.async.AsyncChains;
-import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.utils.Clock;
@@ -142,6 +140,7 @@ public class AccordCommandStore extends CommandStore
public final String loggingId;
private final Journal journal;
+ private final RangeSearcher rangeSearcher;
private final AccordExecutor executor;
private final Executor taskExecutor;
private final ExclusiveCaches caches;
@@ -166,6 +165,7 @@ public class AccordCommandStore extends CommandStore
super(id, node, agent, dataStore, progressLogFactory, listenerFactory, epochUpdateHolder);
loggingId = String.format("[%s]", id);
this.journal = journal;
+ this.rangeSearcher = RangeSearcher.extractRangeSearcher(journal);
this.executor = executor;
final AccordCache.Type<TxnId, Command, AccordSafeCommand>.Instance commands;
@@ -255,22 +255,6 @@ public class AccordCommandStore extends CommandStore
return caches;
}
- @Nullable
- @VisibleForTesting
- public Runnable appendToKeyspace(TxnId txnId, Command after)
- {
- if (txnId.is(Routable.Domain.Key))
- return null;
-
- Mutation mutation = AccordKeyspace.getCommandMutation(this.id, after, nextSystemTimestampMicros());
-
- // TODO (required): make sure we test recovering when this has failed to be persisted
- if (null != mutation)
- return mutation::applyUnsafe;
-
- return null;
- }
-
public void persistFieldUpdates(FieldUpdates fieldUpdates, Runnable onFlush)
{
journal.saveStoreState(id, fieldUpdates, onFlush);
@@ -485,6 +469,11 @@ public class AccordCommandStore extends CommandStore
return journal.loadMinimal(id, txnId, MINIMAL, unsafeGetRedundantBefore(), durableBefore());
}
+ public RangeSearcher rangeSearcher()
+ {
+ return rangeSearcher;
+ }
+
public Loader loader()
{
return loader;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
index bd1605c6bc..65e19e79b3 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordConfigurationService.java
@@ -204,7 +204,7 @@ public class AccordConfigurationService extends AbstractConfigurationService<Acc
}
}
- final ChangeListener listener = new MetadataChangeListener();
+ public final ChangeListener listener = new MetadataChangeListener();
private class MetadataChangeListener implements ChangeListener
{
@Override
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
index d171a4c021..5a0a6e7b5b 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java
@@ -19,6 +19,7 @@ package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.NavigableMap;
@@ -29,8 +30,6 @@ import java.util.function.BiConsumer;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import accord.impl.CommandChange;
import accord.impl.CommandChange.Field;
@@ -64,6 +63,7 @@ import org.apache.cassandra.journal.Compactor;
import org.apache.cassandra.journal.Journal;
import org.apache.cassandra.journal.Params;
import org.apache.cassandra.journal.RecordPointer;
+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.IdentityAccumulator;
@@ -92,7 +92,7 @@ import static accord.primitives.SaveStatus.ErasedOrVestigial;
import static accord.primitives.Status.Truncated;
import static org.apache.cassandra.service.accord.AccordJournalValueSerializers.DurableBeforeAccumulator;
-public class AccordJournal implements accord.api.Journal, Shutdownable
+public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier, Shutdownable
{
static
{
@@ -100,8 +100,6 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
Invariants.checkState(MessagingService.current_version == MessagingService.VERSION_51, "Expected current version to be %d but given %d", MessagingService.VERSION_51, MessagingService.current_version);
}
- private static final Logger logger = LoggerFactory.getLogger(AccordJournal.class);
-
private static final Set<Integer> SENTINEL_HOSTS = Collections.singleton(0);
static final ThreadLocal<byte[]> keyCRCBytes = ThreadLocal.withInitial(() -> new byte[JournalKeySupport.TOTAL_SIZE]);
@@ -124,6 +122,18 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
public AccordJournal(Params params, AccordAgent agent, File directory, ColumnFamilyStore cfs)
{
this.agent = agent;
+ AccordSegmentCompactor<Object> compactor = new AccordSegmentCompactor<>(params.userVersion(), cfs) {
+ @Nullable
+ @Override
+ public Collection<StaticSegment<JournalKey, Object>> compact(Collection<StaticSegment<JournalKey, Object>> staticSegments)
+ {
+ if (journalTable == null)
+ throw new IllegalStateException("Unsafe access to AccordJournal during <init>; journalTable was touched before it was published");
+ Collection<StaticSegment<JournalKey, Object>> result = super.compact(staticSegments);
+ journalTable.safeNotify(index -> index.remove(staticSegments));
+ return result;
+ }
+ };
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
new ValueSerializer<>()
@@ -140,11 +150,17 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
throw new UnsupportedOperationException();
}
},
- new AccordSegmentCompactor<>(params.userVersion(), cfs));
+ compactor);
this.journalTable = new AccordJournalTable<>(journal, JournalKey.SUPPORT, cfs, params.userVersion());
this.params = params;
}
+ @VisibleForTesting
+ public int inMemorySize()
+ {
+ return journal.currentActiveSegment().index().size();
+ }
+
public AccordJournal start(Node node)
{
Invariants.checkState(status == Status.INITIALIZED);
@@ -268,7 +284,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
}
@Override
- public void saveCommand(int store, CommandUpdate update, Runnable onFlush)
+ public void saveCommand(int store, CommandUpdate update, @Nullable Runnable onFlush)
{
Writer diff = Writer.make(update.before, update.after);
if (diff == null)
@@ -280,6 +296,12 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
JournalKey key = new JournalKey(update.txnId, JournalKey.Type.COMMAND_DIFF, store);
RecordPointer pointer = journal.asyncWrite(key, diff, SENTINEL_HOSTS);
+ if (journalTable.shouldIndex(key)
+ && diff.hasParticipants()
+ && diff.after.route() != null)
+ journal.onDurable(pointer, () ->
+ journalTable.safeNotify(index ->
+ index.update(pointer.segment, key.commandStoreId, key.id, diff.after.route())));
if (onFlush != null)
journal.onDurable(pointer, onFlush);
}
@@ -400,6 +422,7 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
public void truncateForTesting()
{
journal.truncateForTesting();
+ journalTable.safeNotify(RouteInMemoryIndex::truncateForTesting);
}
@VisibleForTesting
@@ -441,6 +464,11 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
try (DataInputBuffer in = new DataInputBuffer(buffer, false))
{
builder.deserializeNext(in, userVersion);
+ if (journalTable.shouldIndex(finalKey)
+ && builder.participants() != null
+ && builder.participants().route() != null)
+ journalTable.safeNotify(index ->
+ index.update(segment, finalKey.commandStoreId, finalKey.id, builder.participants().route()));
}
catch (IOException e)
{
@@ -505,6 +533,12 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
status = Status.STARTED;
}
+ @Override
+ public RangeSearcher rangeSearcher()
+ {
+ return journalTable.rangeSearcher();
+ }
+
public static class Writer implements Journal.Writer
{
private final Command after;
@@ -601,6 +635,16 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
iterable = unsetIterableFields(field, iterable);
}
}
+
+ private boolean hasField(Field fields)
+ {
+ return !getFieldIsNull(fields, flags);
+ }
+
+ public boolean hasParticipants()
+ {
+ return hasField(Field.PARTICIPANTS);
+ }
}
public static class Builder extends CommandChange.Builder
@@ -788,4 +832,4 @@ public class AccordJournal implements accord.api.Journal, Shutdownable
}
}
}
-}
\ No newline at end of file
+}
diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
index 891abcf1df..9981f5a957 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
@@ -21,21 +21,38 @@ import java.io.Closeable;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.List;
+import java.util.function.Consumer;
+import javax.annotation.Nullable;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.agrona.collections.IntHashSet;
import org.agrona.collections.LongHashSet;
import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ColumnFamilyStore.RefViewFragment;
+import org.apache.cassandra.db.DataRange;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.EmptyIterators;
+import org.apache.cassandra.db.PartitionRangeReadCommand;
+import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.Slices;
import org.apache.cassandra.db.StorageHook;
import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.filter.DataLimits;
+import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
@@ -44,6 +61,9 @@ import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.db.rows.UnfilteredRowIterators;
+import org.apache.cassandra.index.Index;
+import org.apache.cassandra.index.accord.OrderedRouteSerializer;
+import org.apache.cassandra.index.accord.RouteJournalIndex;
import org.apache.cassandra.io.sstable.ISSTableScanner;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.util.DataInputBuffer;
@@ -53,11 +73,18 @@ import org.apache.cassandra.journal.Journal;
import org.apache.cassandra.journal.KeySupport;
import org.apache.cassandra.journal.RecordConsumer;
import org.apache.cassandra.schema.ColumnMetadata;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.CloseableIterator;
+import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.JVMStabilityInspector;
+import org.apache.cassandra.utils.MergeIterator;
import static org.apache.cassandra.io.sstable.SSTableReadsListener.NOOP_LISTENER;
-public class AccordJournalTable<K extends JournalKey, V>
+public class AccordJournalTable<K extends JournalKey, V> implements RangeSearcher.Supplier
{
+ private static final Logger logger = LoggerFactory.getLogger(AccordJournalTable.class);
+
private static final IntHashSet SENTINEL_HOSTS = new IntHashSet();
private final Journal<K, V> journal;
@@ -67,6 +94,13 @@ public class AccordJournalTable<K extends JournalKey, V>
private final ColumnMetadata versionColumn;
private final KeySupport<K> keySupport;
+ /**
+ * Access to this field should only ever be handled by {@link #safeNotify(Consumer)}. There is an assumption that
+ * an error in the index should not cause the journal to crash, so {@link #safeNotify(Consumer)} exists to make sure
+ * this property holds true.
+ */
+ @Nullable
+ private final RouteInMemoryIndex<Object> index;
private final int accordJournalVersion;
public AccordJournalTable(Journal<K, V> journal, KeySupport<K> keySupport, ColumnFamilyStore cfs, int accordJournalVersion)
@@ -77,6 +111,31 @@ public class AccordJournalTable<K extends JournalKey, V>
this.versionColumn = cfs.metadata().getColumn(ColumnIdentifier.getInterned("user_version", false));
this.keySupport = keySupport;
this.accordJournalVersion = accordJournalVersion;
+
+ this.index = cfs.indexManager.getIndexByName(AccordKeyspace.JOURNAL_INDEX_NAME) != null
+ ? new RouteInMemoryIndex<>()
+ : null;
+ }
+
+ boolean shouldIndex(JournalKey key)
+ {
+ if (index == null) return false;
+ return RouteJournalIndex.allowed(key);
+ }
+
+ void safeNotify(Consumer<RouteInMemoryIndex<Object>> fn)
+ {
+ if (index == null)
+ return;
+ try
+ {
+ fn.accept(index);
+ }
+ catch (Throwable t)
+ {
+ JVMStabilityInspector.inspectThrowable(t);
+ logger.warn("Failure updating index", t);
+ }
}
public void forceCompaction()
@@ -84,6 +143,14 @@ public class AccordJournalTable<K extends JournalKey, V>
cfs.forceMajorCompaction();
}
+ @Override
+ public RangeSearcher rangeSearcher()
+ {
+ if (index == null)
+ return RangeSearcher.NoopRangeSearcher.instance;
+ return new TableRangeSearcher();
+ }
+
public interface Reader
{
void read(DataInputPlus input, int userVersion) throws IOException;
@@ -159,6 +226,117 @@ public class AccordJournalTable<K extends JournalKey, V>
}
}
+ /**
+ * When using {@link PartitionRangeReadCommand} we need to work with {@link RowFilter} which works with columns.
+ * But the index doesn't care about table based queries and needs to be queried using the fields in the index, to
+ * support that this enum exists. This enum represents the fields present in the index and can be used to apply
+ * filters to the index.
+ */
+ public enum SyntheticColumn
+ {
+ participants("participants", BytesType.instance),
+ store_id("store_id", Int32Type.instance),
+ txn_id("txn_id", AccordKeyspace.TIMESTAMP_TYPE);
+
+ public final ColumnMetadata metadata;
+
+ SyntheticColumn(String name, AbstractType<?> type)
+ {
+ this.metadata = new ColumnMetadata("journal", "routes", new ColumnIdentifier(name, false), type, ColumnMetadata.NO_POSITION, ColumnMetadata.Kind.REGULAR, null);
+ }
+ }
+
+ private class TableRangeSearcher implements RangeSearcher
+ {
+ private final Index tableIndex;
+
+ private TableRangeSearcher()
+ {
+ this.tableIndex = cfs.indexManager.getIndexByName("record");
+ if (!cfs.indexManager.isIndexQueryable(tableIndex))
+ throw new AssertionError("Journal record index is not queryable");
+ }
+
+ @Override
+ public Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId)
+ {
+ CloseableIterator<TxnId> inMemory = index.search(commandStoreId, range, minTxnId, maxTxnId).results();
+ CloseableIterator<TxnId> table = tableSearch(commandStoreId, range.start(), range.end());
+ return new DefaultResult(minTxnId, maxTxnId, MergeIterator.get(Arrays.asList(inMemory, table)));
+ }
+
+ @Override
+ public Result search(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId)
+ {
+ CloseableIterator<TxnId> inMemory = index.search(commandStoreId, key, minTxnId, maxTxnId).results();
+ CloseableIterator<TxnId> table = tableSearch(commandStoreId, key);
+ return new DefaultResult(minTxnId, maxTxnId, MergeIterator.get(Arrays.asList(inMemory, table)));
+ }
+
+ private CloseableIterator<TxnId> tableSearch(int store, AccordRoutingKey start, AccordRoutingKey end)
+ {
+ RowFilter rowFilter = RowFilter.create(false);
+ rowFilter.add(AccordJournalTable.SyntheticColumn.participants.metadata, Operator.GT, OrderedRouteSerializer.serializeRoutingKey(start));
+ rowFilter.add(AccordJournalTable.SyntheticColumn.participants.metadata, Operator.LTE, OrderedRouteSerializer.serializeRoutingKey(end));
+ rowFilter.add(AccordJournalTable.SyntheticColumn.store_id.metadata, Operator.EQ, Int32Type.instance.decompose(store));
+
+ return process(store, rowFilter);
+ }
+
+ private CloseableIterator<TxnId> tableSearch(int store, AccordRoutingKey key)
+ {
+ RowFilter rowFilter = RowFilter.create(false);
+ rowFilter.add(AccordJournalTable.SyntheticColumn.participants.metadata, Operator.GTE, OrderedRouteSerializer.serializeRoutingKey(key));
+ rowFilter.add(AccordJournalTable.SyntheticColumn.participants.metadata, Operator.LTE, OrderedRouteSerializer.serializeRoutingKey(key));
+ rowFilter.add(AccordJournalTable.SyntheticColumn.store_id.metadata, Operator.EQ, Int32Type.instance.decompose(store));
+
+ return process(store, rowFilter);
+ }
+
+ private CloseableIterator<TxnId> process(int storeId, RowFilter rowFilter)
+ {
+ PartitionRangeReadCommand cmd = PartitionRangeReadCommand.create(cfs.metadata(),
+ FBUtilities.nowInSeconds(),
+ ColumnFilter.selectionBuilder()
+ .add(AccordJournalTable.SyntheticColumn.store_id.metadata)
+ .add(AccordJournalTable.SyntheticColumn.txn_id.metadata)
+ .build(),
+ rowFilter,
+ DataLimits.NONE,
+ DataRange.allData(cfs.getPartitioner()));
+ Index.Searcher s = tableIndex.searcherFor(cmd);
+ try (ReadExecutionController controller = cmd.executionController())
+ {
+ UnfilteredPartitionIterator partitionIterator = s.search(controller);
+ return new CloseableIterator<>()
+ {
+
+ @Override
+ public void close()
+ {
+ partitionIterator.close();
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ return partitionIterator.hasNext();
+ }
+
+ @Override
+ public TxnId next()
+ {
+ UnfilteredRowIterator next = partitionIterator.next();
+ JournalKey partitionKeyComponents = AccordKeyspace.JournalColumns.getJournalKey(next.partitionKey());
+ Invariants.checkState(partitionKeyComponents.commandStoreId == storeId,
+ () -> String.format("table index returned a command store other than the exepcted one; expected %d != %d", storeId, partitionKeyComponents.commandStoreId));
+ return partitionKeyComponents.id;
+ }
+ };
+ }
+ }
+ }
+
/**
* Perform a read from Journal table, followed by the reads from all journal segments.
* <p>
diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
index 5e3085cf18..f0662b6662 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
@@ -21,7 +21,6 @@ package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.ByteBuffer;
-import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Objects;
@@ -30,7 +29,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
-import java.util.function.Function;
import java.util.stream.Collectors;
import javax.annotation.Nullable;
@@ -41,8 +39,6 @@ import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import accord.local.Command;
-import accord.local.CommandStore;
import accord.local.Node;
import accord.local.RedundantBefore;
import accord.local.StoreParticipants;
@@ -65,10 +61,7 @@ import org.apache.cassandra.db.ClusteringComparator;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Columns;
import org.apache.cassandra.db.DecoratedKey;
-import org.apache.cassandra.db.DeletionTime;
import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.db.LivenessInfo;
-import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.PartitionPosition;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.RegularAndStaticColumns;
@@ -79,7 +72,6 @@ import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.filter.DataLimits;
import org.apache.cassandra.db.filter.RowFilter;
-import org.apache.cassandra.db.marshal.ByteArrayAccessor;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.ByteType;
import org.apache.cassandra.db.marshal.BytesType;
@@ -94,21 +86,18 @@ import org.apache.cassandra.db.rows.BTreeRow;
import org.apache.cassandra.db.rows.BufferCell;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.Row;
-import org.apache.cassandra.db.rows.Row.Deletion;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.transform.FilteredPartitions;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.dht.Bounds;
-import org.apache.cassandra.dht.ByteOrderedPartitioner;
import org.apache.cassandra.dht.ExcludingBounds;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.IncludingExcludingBounds;
import org.apache.cassandra.dht.LocalCompositePrefixPartitioner;
import org.apache.cassandra.dht.LocalPartitioner;
-import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
-import org.apache.cassandra.index.accord.RouteIndex;
+import org.apache.cassandra.index.accord.RouteJournalIndex;
import org.apache.cassandra.index.transactions.UpdateTransaction;
import org.apache.cassandra.io.IVersionedSerializer;
import org.apache.cassandra.io.LocalVersionedSerializer;
@@ -139,24 +128,18 @@ import org.apache.cassandra.service.accord.serializers.CommandSerializers;
import org.apache.cassandra.service.accord.serializers.KeySerializers;
import org.apache.cassandra.utils.Clock.Global;
import org.apache.cassandra.utils.CloseableIterator;
-import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.BTreeSet;
-import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.concurrent.OpOrder;
-import static accord.utils.Invariants.checkArgument;
import static accord.utils.Invariants.checkState;
import static java.lang.String.format;
import static java.util.Collections.emptyMap;
import static org.apache.cassandra.cql3.QueryProcessor.executeInternal;
import static org.apache.cassandra.db.partitions.PartitionUpdate.singleRowUpdate;
import static org.apache.cassandra.db.rows.BTreeRow.singleCellRow;
-import static org.apache.cassandra.db.rows.BufferCell.live;
-import static org.apache.cassandra.db.rows.BufferCell.tombstone;
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
import static org.apache.cassandra.service.accord.serializers.AccordRoutingKeyByteSource.currentVersion;
import static org.apache.cassandra.service.accord.serializers.KeySerializers.blobMapToRanges;
-import static org.apache.cassandra.utils.ByteBufferUtil.EMPTY_BYTE_BUFFER;
import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
public class AccordKeyspace
@@ -164,20 +147,17 @@ public class AccordKeyspace
private static final Logger logger = LoggerFactory.getLogger(AccordKeyspace.class);
public static final String JOURNAL = "journal";
- public static final String COMMANDS = "commands";
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 String JOURNAL_INDEX_NAME = "record";
- public static final Set<String> TABLE_NAMES = ImmutableSet.of(COMMANDS, COMMANDS_FOR_KEY,
+ public static final Set<String> TABLE_NAMES = ImmutableSet.of(COMMANDS_FOR_KEY,
TOPOLOGIES, EPOCH_METADATA,
JOURNAL);
// TODO (desired): implement a custom type so we can get correct sort order
- private static final TupleType TIMESTAMP_TYPE = new TupleType(Lists.newArrayList(LongType.instance, LongType.instance, Int32Type.instance));
- private static final String TIMESTAMP_TUPLE = TIMESTAMP_TYPE.asCQL3Type().toString();
- private static final TupleType KEY_TYPE = new TupleType(Arrays.asList(UUIDType.instance, BytesType.instance));
- private static final String KEY_TUPLE = KEY_TYPE.asCQL3Type().toString();
+ public static final TupleType TIMESTAMP_TYPE = new TupleType(Lists.newArrayList(LongType.instance, LongType.instance, Int32Type.instance));
private static final ClusteringIndexFilter FULL_PARTITION = new ClusteringIndexNamesFilter(BTreeSet.of(new ClusteringComparator(), Clustering.EMPTY), false);
@@ -201,71 +181,32 @@ public class AccordKeyspace
private static SchemaProvider schema = Objects.requireNonNull(Schema.instance);
}
- private enum TokenType
- {
- Murmur3((byte) 1),
- ByteOrdered((byte) 2),
- ;
-
- private final byte value;
-
- TokenType(byte b)
- {
- this.value = b;
- }
-
- static TokenType valueOf(Token token)
- {
- if (token instanceof Murmur3Partitioner.LongToken)
- return Murmur3;
- if (token instanceof ByteOrderedPartitioner.BytesToken)
- return ByteOrdered;
- throw new IllegalArgumentException("Unexpected token type: " + token.getClass());
- }
- }
-
- public static TableMetadata journalMetadata(String tableName)
- {
- return parse(tableName,
- "accord journal",
- "CREATE TABLE %s ("
- + "store_id int,"
- + "type tinyint,"
- + "id blob,"
- + "descriptor bigint,"
- + "offset int,"
- + "user_version int,"
- + "record blob,"
- + "PRIMARY KEY((store_id, type, id), descriptor, offset)"
- + ") WITH CLUSTERING ORDER BY (descriptor DESC, offset DESC)" +
- " WITH compression = {'class':'NoopCompressor'};")
- .compaction(CompactionParams.lcs(emptyMap()))
- .bloomFilterFpChance(0.01)
- .partitioner(new LocalPartitioner(BytesType.instance))
- .build();
- }
-
- public static final TableMetadata Journal = journalMetadata(JOURNAL);
-
- // TODO: store timestamps as blobs (confirm there are no negative numbers, or offset)
- public static final TableMetadata Commands =
- parse(COMMANDS,
- "accord commands",
- "CREATE TABLE %s ("
- + "store_id int,"
- + "domain int," // this is stored as part of txn_id, used currently for cheaper scans of the table
- + format("txn_id %s,", TIMESTAMP_TUPLE)
- + "status int,"
- + "participants blob,"
- + "durability int,"
- + format("execute_at %s,", TIMESTAMP_TUPLE)
- + "PRIMARY KEY((store_id, domain, txn_id))"
- + ')')
- .partitioner(new LocalPartitioner(CompositeType.getInstance(Int32Type.instance, Int32Type.instance, TIMESTAMP_TYPE)))
- .indexes(Indexes.builder()
- .add(IndexMetadata.fromSchemaMetadata("route", IndexMetadata.Kind.CUSTOM, ImmutableMap.of("class_name", RouteIndex.class.getCanonicalName(), "target", "participants")))
- .build())
- .build();
+ public static TableMetadata journalMetadata(String tableName, boolean index)
+ {
+ TableMetadata.Builder builder = parse(tableName,
+ "accord journal",
+ "CREATE TABLE %s ("
+ + "store_id int,"
+ + "type tinyint,"
+ + "id blob,"
+ + "descriptor bigint,"
+ + "offset int,"
+ + "user_version int,"
+ + "record blob,"
+ + "PRIMARY KEY((store_id, type, id), descriptor, offset)"
+ + ") WITH CLUSTERING ORDER BY (descriptor DESC, offset DESC)" +
+ " WITH compression = {'class':'NoopCompressor'};")
+ .compaction(CompactionParams.lcs(emptyMap()))
+ .bloomFilterFpChance(0.01)
+ .partitioner(new LocalPartitioner(BytesType.instance));
+ if (index)
+ builder.indexes(Indexes.builder()
+ .add(IndexMetadata.fromSchemaMetadata(JOURNAL_INDEX_NAME, IndexMetadata.Kind.CUSTOM, ImmutableMap.of("class_name", RouteJournalIndex.class.getCanonicalName(), "target", "record,user_version")))
+ .build());
+ return builder.build();
+ }
+
+ public static final TableMetadata Journal = journalMetadata(JOURNAL, true);
// TODO: naming is not very clearly distinct from the base serializers
public static class LocalVersionedSerializers
@@ -276,6 +217,16 @@ public class AccordKeyspace
{
return new LocalVersionedSerializer<>(AccordSerializerVersion.CURRENT, AccordSerializerVersion.serializer, serializer);
}
+
+ public static ByteBuffer serialize(StoreParticipants value) throws IOException
+ {
+ int size = Math.toIntExact(participants.serializedSize(value));
+ try (DataOutputBuffer buffer = new DataOutputBuffer(size))
+ {
+ participants.serialize(value, buffer);
+ return buffer.buffer(true);
+ }
+ }
}
private static ColumnMetadata getColumn(TableMetadata metadata, String name)
@@ -286,189 +237,6 @@ public class AccordKeyspace
return column;
}
- public static class CommandsColumns
- {
- static final ClusteringComparator keyComparator = Commands.partitionKeyAsClusteringComparator();
- static final CompositeType partitionKeyType = (CompositeType) Commands.partitionKeyType;
- public static final ColumnMetadata txn_id = getColumn(Commands, "txn_id");
- public static final ColumnMetadata store_id = getColumn(Commands, "store_id");
- public static final ColumnMetadata status = getColumn(Commands, "status");
- public static final ColumnMetadata participants = getColumn(Commands, "participants");
- public static final ColumnMetadata durability = getColumn(Commands, "durability");
- public static final ColumnMetadata execute_at = getColumn(Commands, "execute_at");
-
- public static final ColumnMetadata[] TRUNCATE_FIELDS = new ColumnMetadata[] { durability, execute_at, participants, status };
- public static final ColumnMetadata[] SAVE_STATUS_ONLY_FIELDS = new ColumnMetadata[] { status };
-
- static
- {
- for (int i = 1 ; i < TRUNCATE_FIELDS.length ; ++i)
- Invariants.checkState(TRUNCATE_FIELDS[i - 1].compareTo(TRUNCATE_FIELDS[i]) < 0);
- }
- }
-
- public static class CommandRows extends CommandsColumns
- {
- 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 int getStoreId(DecoratedKey pk)
- {
- var array = splitPartitionKey(pk);
- return getStoreId(array);
- }
-
- public static TxnId getTxnId(ByteBuffer[] partitionKeyComponents)
- {
- return deserializeTimestampOrNull(partitionKeyComponents[txn_id.position()], ByteBufferAccessor.instance, TxnId::fromBits);
- }
-
- @Nullable
- public static Timestamp getExecuteAt(Row row)
- {
- Cell cell = row.getCell(execute_at);
- if (cell == null)
- return null;
- return deserializeTimestampOrNull(cell.value(), cell.accessor(), Timestamp::fromBits);
- }
-
- @Nullable
- public static SaveStatus getStatus(Row row)
- {
- Cell cell = row.getCell(status);
- if (cell == null)
- return null;
- int ordinal = cell.accessor().getInt(cell.value(), 0);
- return CommandSerializers.saveStatus.forOrdinal(ordinal);
- }
-
- @Nullable
- public static Status.Durability getDurability(Row row)
- {
- Cell cell = row.getCell(durability);
- if (cell == null)
- return null;
- int ordinal = cell.accessor().getInt(cell.value(), 0);
- return CommandSerializers.durability.forOrdinal(ordinal);
- }
-
- @Nullable
- public static Route<?> getRoute(Row row)
- {
- Cell<?> cell = row.getCell(participants);
- StoreParticipants participants = deserializeParticipantsOrNull(cell);
- return participants == null ? null : participants.route();
- }
-
- private static Object[] truncatedApplyLeaf(long newTimestamp, SaveStatus newSaveStatus, @Nullable Cell<?> durabilityCell, @Nullable Cell<?> executeAtCell, @Nullable Cell<?> participantsCell, boolean updateTimestamps)
- {
- int count = 1 + (durabilityCell != null ? 1 : 0) + (executeAtCell != null ? 1 : 0) + (participantsCell != null ? 1 : 0);
- Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(count);
- int colIndex = 0;
- if (durabilityCell != null)
- {
- checkArgument(durabilityCell.column() == CommandsColumns.durability);
- newLeaf[colIndex++] = updateTimestamps ? durabilityCell.withUpdatedTimestamp(newTimestamp) : durabilityCell;
- }
- if (executeAtCell != null)
- {
- checkArgument(executeAtCell.column() == CommandsColumns.execute_at);
- newLeaf[colIndex++] = updateTimestamps ? executeAtCell.withUpdatedTimestamp(newTimestamp) : executeAtCell;
- }
- if (participantsCell != null)
- {
- checkArgument(participantsCell.column() == CommandsColumns.participants);
- newLeaf[colIndex++] = updateTimestamps ? participantsCell.withUpdatedTimestamp(newTimestamp) : participantsCell;
- }
- newLeaf[colIndex] = BufferCell.live(status, newTimestamp, ByteBufferAccessor.instance.valueOf(newSaveStatus.ordinal()));
- return newLeaf;
- }
-
- private static Object[] expungePartialLeaf(Cell<?> durabilityCell, Cell<?> executeAtCell, Cell<?> participantsCell)
- {
- int count = (durabilityCell != null ? 1 : 0) + (executeAtCell != null ? 1 : 0) + (participantsCell != null ? 1 : 0);
- if (count == 0)
- return null;
-
- Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(count);
- int colIndex = 0;
- if (durabilityCell != null)
- {
- checkArgument(durabilityCell.column() == CommandsColumns.durability);
- newLeaf[colIndex++] = durabilityCell;
- }
- if (executeAtCell != null)
- {
- checkArgument(executeAtCell.column() == CommandsColumns.execute_at);
- newLeaf[colIndex++] = executeAtCell;
- }
- if (participantsCell != null)
- {
- checkArgument(participantsCell.column() == CommandsColumns.participants);
- newLeaf[colIndex] = participantsCell;
- }
- return newLeaf;
- }
-
- public static Row saveStatusOnly(SaveStatus newSaveStatus, Row row, long nowInSec)
- {
- long oldTimestamp = row.primaryKeyLivenessInfo().timestamp();
- long newTimestamp = oldTimestamp + 1;
-
- Object[] newLeaf = saveStatusOnlyLeaf(newTimestamp, newSaveStatus);
-
- // Including a deletion allows future compactions to drop data before it gets to the purger
- // but it is pretty optional because maybeDropTruncatedCommandColumns will drop the extra columns
- // regardless
- Row.Deletion deletion = new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false);
- return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec), deletion, newLeaf);
- }
-
- private static Object[] saveStatusOnlyLeaf(long newTimestamp, SaveStatus newSaveStatus)
- {
- Object[] newLeaf = BTree.unsafeAllocateNonEmptyLeaf(SAVE_STATUS_ONLY_FIELDS.length);
- int colIndex = 0;
- // Status always needs to use the new timestamp since we are replacing the existing value
- // All the other columns are being retained unmodified with at most updated timestamps to accomdate deletion
- //noinspection UnusedAssignment
- newLeaf[colIndex++] = BufferCell.live(status, newTimestamp, ByteBufferAccessor.instance.valueOf(newSaveStatus.ordinal()));
- return newLeaf;
- }
-
- public static Row truncatedApply(SaveStatus newSaveStatus, Row row, long nowInSec, Durability durability, @Nullable Cell<?> durabilityCell, @Nullable Cell<?> executeAtCell, @Nullable Cell<?> participantsCell, boolean withOutcome)
- {
- long oldTimestamp = row.primaryKeyLivenessInfo().timestamp();
- long newTimestamp = oldTimestamp + 1;
- boolean doDeletion = durability == Durability.Universal;
-
- // We may not have what we need to generate a deletion and include the outcome in the truncated row
- // so need to wait until we can have the outcome to issue the deletion otherwise it would be shadowed and lost
- if (withOutcome)
- doDeletion = false;
-
- Object[] newLeaf = truncatedApplyLeaf(newTimestamp, newSaveStatus, durabilityCell, executeAtCell, participantsCell, doDeletion);
-
- // Including a deletion allows future compactions to drop data before it gets to the purger
- // but it is pretty optional because maybeDropTruncatedCommandColumns will drop the extra columns
- // regardless
- Row.Deletion deletion = doDeletion ? new Row.Deletion(DeletionTime.build(oldTimestamp, nowInSec), false) : Deletion.LIVE;
- return BTreeRow.create(row.clustering(), LivenessInfo.create(newTimestamp, nowInSec), deletion, newLeaf);
- }
-
- public static Row expungePartial(Row row, @Nullable Cell<?> durabilityCell, @Nullable Cell<?> executeAtCell, @Nullable Cell<?> participantsCell)
- {
- Object[] newLeaf = expungePartialLeaf(durabilityCell, executeAtCell, participantsCell);
- return BTreeRow.create(row.clustering(), row.primaryKeyLivenessInfo(), 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);
@@ -636,7 +404,7 @@ 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, CommandsForKeys, Topologies, EpochMetadata, Journal);
+ public static Tables TABLES = Tables.of(CommandsForKeys, Topologies, EpochMetadata, Journal);
public static Tables tables()
{
return TABLES;
@@ -661,11 +429,6 @@ public class AccordKeyspace
}
}
- private static <T> ByteBuffer serializeOrNull(T obj, LocalVersionedSerializer<T> serializer) throws IOException
- {
- return obj != null ? serialize(obj, serializer) : EMPTY_BYTE_BUFFER;
- }
-
private static <T> T deserialize(ByteBuffer bytes, LocalVersionedSerializer<T> serializer) throws IOException
{
try (DataInputBuffer in = new DataInputBuffer(bytes, true))
@@ -674,120 +437,6 @@ public class AccordKeyspace
}
}
- private interface SerializeFunction<V>
- {
- ByteBuffer apply(V v) throws IOException;
- }
-
- private static <C, V> boolean valueModified(Function<C, V> get, C original, C current)
- {
- V prev = original != null ? get.apply(original) : null;
- V value = get.apply(current);
-
- return prev != value;
- }
-
- private static <C, V> void addCellIfModified(ColumnMetadata column, Function<C, V> get, SerializeFunction<V> serialize, Row.Builder builder, long timestampMicros, int nowInSeconds, C original, C current) throws IOException
- {
- if (valueModified(get, original, current))
- {
- V newValue = get.apply(current);
- if (newValue == null) builder.addCell(tombstone(column, timestampMicros, nowInSeconds));
- else builder.addCell(live(column, timestampMicros, serialize.apply(newValue)));
- }
- }
-
- private static <C, V> void addCell(ColumnMetadata column, Function<C, V> get, SerializeFunction<V> serialize, Row.Builder builder, long timestampMicros, int nowInSeconds, C current) throws IOException
- {
- V newValue = get.apply(current);
- if (newValue == null) builder.addCell(tombstone(column, timestampMicros, nowInSeconds));
- else builder.addCell(live(column, timestampMicros, serialize.apply(newValue)));
- }
-
- private static <C extends Command, V> void addCell(ColumnMetadata column, Function<C, V> get, LocalVersionedSerializer<V> serializer, Row.Builder builder, long timestampMicros, int nowInSeconds, C command) throws IOException
- {
- addCell(column, get, v -> serializeOrNull(v, serializer), builder, timestampMicros, nowInSeconds, command);
- }
-
- private static <C extends Command, V> void addCellIfModified(ColumnMetadata column, Function<C, V> get, LocalVersionedSerializer<V> serializer, Row.Builder builder, long timestampMicros, int nowInSeconds, C original, C command) throws IOException
- {
- addCellIfModified(column, get, v -> serializeOrNull(v, serializer), builder, timestampMicros, nowInSeconds, original, command);
- }
-
- private static <C extends Command, V extends Enum<V>> void addEnumCellIfModified(ColumnMetadata column, Function<C, V> get, Row.Builder builder, long timestampMicros, int nowInSeconds, C original, C command) throws IOException
- {
- // TODO: convert to byte arrays
- ValueAccessor<ByteBuffer> accessor = ByteBufferAccessor.instance;
- addCellIfModified(column, get, v -> accessor.valueOf(v.ordinal()), builder, timestampMicros, nowInSeconds, original, command);
- }
-
- private static <C extends Command, V extends Enum<V>> void addEnumCell(ColumnMetadata column, Function<C, V> get, Row.Builder builder, long timestampMicros, int nowInSeconds, C command) throws IOException
- {
- // TODO: convert to byte arrays
- ValueAccessor<ByteBuffer> accessor = ByteBufferAccessor.instance;
- addCell(column, get, v -> accessor.valueOf(v.ordinal()), builder, timestampMicros, nowInSeconds, command);
- }
-
- public static Mutation getCommandMutation(AccordCommandStore commandStore, AccordSafeCommand liveCommand, long timestampMicros)
- {
- return getCommandMutation(commandStore.id(), liveCommand.current(), timestampMicros);
- }
-
- public static Mutation getCommandMutation(int storeId, Command command, long timestampMicros)
- {
- if (command.saveStatus() == SaveStatus.Uninitialised)
- return null;
-
- try
- {
- Row.Builder builder = BTreeRow.unsortedBuilder();
- builder.newRow(Clustering.EMPTY);
- int nowInSeconds = (int) TimeUnit.MICROSECONDS.toSeconds(timestampMicros);
- builder.addPrimaryKeyLivenessInfo(LivenessInfo.create(timestampMicros, nowInSeconds));
-
- addEnumCell(CommandsColumns.durability, Command::durability, builder, timestampMicros, nowInSeconds, command);
- addCell(CommandsColumns.participants, Command::participants, LocalVersionedSerializers.participants, builder, timestampMicros, nowInSeconds, command);
- addEnumCell(CommandsColumns.status, Command::saveStatus, builder, timestampMicros, nowInSeconds, command);
- addCell(CommandsColumns.execute_at, Command::executeAt, AccordKeyspace::serializeTimestamp, builder, timestampMicros, nowInSeconds, command);
-
- Row row = builder.build();
- if (row.columnCount() == 0)
- return null;
-
- ByteBuffer key = CommandsColumns.keyComparator.make(storeId,
- command.txnId().domain().ordinal(),
- serializeTimestamp(command.txnId())).serializeAsPartitionKey();
- PartitionUpdate update = singleRowUpdate(Commands, key, row);
- return new Mutation(update);
- }
- catch (IOException e)
- {
- throw new RuntimeException(e);
- }
- }
-
- public static ByteBuffer serializeToken(Token token)
- {
- return serializeToken(token, ByteBufferAccessor.instance);
- }
-
- public static ByteBuffer serializeTableId(TableId tableId)
- {
- ByteBuffer buffer = ByteBuffer.allocate(tableId.serializedSize());
- tableId.serialize(buffer, ByteBufferAccessor.instance, 0);
- return buffer;
- }
-
- private static <V> V serializeToken(Token token, ValueAccessor<V> accessor)
- {
- TokenType type = TokenType.valueOf(token);
- byte[] ordered = token.getPartitioner().getTokenFactory().toOrderedByteArray(token, ByteComparable.Version.OSS50);
- V value = accessor.allocate(ordered.length + 1);
- accessor.putByte(value, 0, type.value);
- ByteArrayAccessor.instance.copyTo(ordered, 0, value, accessor, 1, ordered.length);
- return value;
- }
-
public static ByteBuffer serializeTimestamp(Timestamp timestamp)
{
return TIMESTAMP_TYPE.pack(bytes(timestamp.msb), bytes(timestamp.lsb), bytes(timestamp.node.id));
@@ -819,14 +468,6 @@ public class AccordKeyspace
return Timestamp.fromBits(accessor.getLong(split.get(0), 0), accessor.getLong(split.get(1), 0), new Node.Id(accessor.getInt(split.get(2), 0)));
}
- public static <V, T extends Timestamp> T deserializeTimestampOrDefault(V value, ValueAccessor<V> accessor, TimestampFactory<T> factory, T defaultVal)
- {
- if (value == null || accessor.isEmpty(value))
- return defaultVal;
- List<V> split = TIMESTAMP_TYPE.unpack(value, accessor);
- return factory.create(accessor.getLong(split.get(0), 0), accessor.getLong(split.get(1), 0), new Node.Id(accessor.getInt(split.get(2), 0)));
- }
-
public static <V, T extends Timestamp> T deserializeTimestampOrNull(V value, ValueAccessor<V> accessor, TimestampFactory<T> factory)
{
if (value == null || accessor.isEmpty(value))
@@ -840,11 +481,6 @@ public class AccordKeyspace
return deserializeTimestampOrNull(row.getBlob(name), factory);
}
- private static <T extends Timestamp> T deserializeTimestampOrDefault(UntypedResultSet.Row row, String name, TimestampFactory<T> factory, T defaultVal)
- {
- return deserializeTimestampOrDefault(row.getBlob(name), ByteBufferAccessor.instance, factory, defaultVal);
- }
-
public static Durability deserializeDurabilityOrNull(Cell cell)
{
return cell == null ? null : CommandSerializers.durability.forOrdinal(cell.accessor().getInt(cell.value(), 0));
@@ -855,20 +491,6 @@ public class AccordKeyspace
return cell == null ? null : CommandSerializers.saveStatus.forOrdinal(cell.accessor().getInt(cell.value(), 0));
}
- @VisibleForTesting
- public static UntypedResultSet loadCommandRow(CommandStore commandStore, TxnId txnId)
- {
- String cql = "SELECT * FROM " + ACCORD_KEYSPACE_NAME + '.' + COMMANDS + ' ' +
- "WHERE store_id = ? " +
- "AND domain = ? " +
- "AND txn_id=(?, ?, ?)";
-
- return executeInternal(cql,
- commandStore.id(),
- txnId.domain().ordinal(),
- txnId.msb, txnId.lsb, txnId.node.id);
- }
-
/**
* Calculates token bounds based on key prefixes.
*/
@@ -979,28 +601,11 @@ public class AccordKeyspace
return accessor.table.partitioner.decorateKey(pk);
}
- @VisibleForTesting
- public static ByteBuffer serializeRoutingKey(AccordRoutingKey routingKey)
- {
- byte[] bytes = getRoutingKeySerializer(routingKey).serialize(routingKey);
- return ByteBuffer.wrap(bytes);
- }
-
public static ByteBuffer serializeRoutingKeyNoTable(AccordRoutingKey key)
{
return CommandsForKeysAccessor.serializeKeyNoTable(key);
}
- public static AccordRoutingKey deserializeRoutingKey(ByteBuffer buffer)
- {
- return AccordRoutingKeyByteSource.Serializer.fromComparableBytes(ByteBufferAccessor.instance, buffer, currentVersion, null);
- }
-
- private static AccordRoutingKeyByteSource.Serializer serializer(AccordRoutingKey routingKey)
- {
- return serializer(routingKey.table());
- }
-
public static AccordRoutingKeyByteSource.Serializer serializer(TableId tableId)
{
return TABLE_SERIALIZERS.computeIfAbsent(tableId, id -> AccordRoutingKeyByteSource.variableLength(partitioner(tableId)));
@@ -1052,11 +657,6 @@ public class AccordKeyspace
return (cell != null && !cell.isTombstone()) ? cellValue(cell) : null;
}
- private static <T> ByteBuffer clusteringValue(Clustering<T> clustering, int idx)
- {
- return clustering.accessor().toBuffer(clustering.get(idx));
- }
-
private static SinglePartitionReadCommand getCommandsForKeyRead(CommandsForKeyAccessor accessor, int storeId, TokenKey key, long nowInSeconds)
{
return SinglePartitionReadCommand.create(accessor.table, nowInSeconds,
@@ -1100,11 +700,6 @@ public class AccordKeyspace
}
}
- public static CommandsForKey unsafeLoadCommandsForKey(int commandStoreId, TokenKey key)
- {
- return unsafeLoadCommandsForKey(CommandsForKeysAccessor, commandStoreId, key);
- }
-
public static CommandsForKey loadCommandsForKey(int commandStoreId, TokenKey key)
{
return unsafeLoadCommandsForKey(CommandsForKeysAccessor, commandStoreId, key);
@@ -1188,6 +783,7 @@ public class AccordKeyspace
public static final ColumnMetadata type = getColumn(Journal, "type");
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 DecoratedKey decorate(JournalKey key)
{
@@ -1428,7 +1024,7 @@ public class AccordKeyspace
@VisibleForTesting
public static void unsafeClear()
{
- for (var store : Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStores())
+ for (ColumnFamilyStore store : Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME).getColumnFamilyStores())
store.truncateBlockingWithoutSnapshot();
TABLE_SERIALIZERS.clear();
SchemaHolder.schema = Schema.instance;
diff --git a/src/java/org/apache/cassandra/service/accord/AccordService.java b/src/java/org/apache/cassandra/service/accord/AccordService.java
index f101d67e47..5b14572614 100644
--- a/src/java/org/apache/cassandra/service/accord/AccordService.java
+++ b/src/java/org/apache/cassandra/service/accord/AccordService.java
@@ -245,6 +245,8 @@ public class AccordService implements IAccordService, Shutdownable
public synchronized static void startup(NodeId tcmId)
{
+ if (instance != null)
+ throw new IllegalStateException("Attempt to call startup multiple times");
if (!DatabaseDescriptor.getAccordTransactionsEnabled())
{
instance = NOOP_SERVICE;
@@ -269,7 +271,8 @@ public class AccordService implements IAccordService, Shutdownable
replayJournal(as);
}
- private static void replayJournal(AccordService as)
+ @VisibleForTesting
+ public static void replayJournal(AccordService as)
{
logger.info("Starting journal replay.");
CommandsForKey.disableLinearizabilityViolationsReporting();
@@ -320,7 +323,8 @@ public class AccordService implements IAccordService, Shutdownable
return instance != null;
}
- private AccordService(Id localId)
+ @VisibleForTesting
+ public AccordService(Id localId)
{
Invariants.checkState(localId != null, "static localId must be set before instantiating AccordService");
logger.info("Starting accord with nodeId {}", localId);
@@ -428,7 +432,7 @@ public class AccordService implements IAccordService, Shutdownable
{
for (InetAddressAndPort peer : group.endpoints())
{
- if (!alive.contains(peer)) continue;
+ if (peer.equals(self) || !alive.contains(peer)) continue;
for (TableMetadata table : tables)
peers.computeIfAbsent(peer, i -> new HashSet<>()).add(AccordTopology.fullRange(table.id));
}
@@ -1224,14 +1228,17 @@ public class AccordService implements IAccordService, Shutdownable
Int2ObjectHashMap<RedundantBefore> redundantBefores = new Int2ObjectHashMap<>();
Int2ObjectHashMap<DurableBefore> durableBefores = new Int2ObjectHashMap<>();
Int2ObjectHashMap<RangesForEpoch> ranges = new Int2ObjectHashMap<>();
- AsyncChains.getBlockingAndRethrow(node.commandStores().forEach(safeStore -> {
- synchronized (redundantBefores)
- {
- redundantBefores.put(safeStore.commandStore().id(), safeStore.redundantBefore());
- ranges.put(safeStore.commandStore().id(), safeStore.ranges());
- durableBefores.put(safeStore.commandStore().id(), safeStore.durableBefore());
- }
- }));
+ if (node.commandStores().all().length > 0)
+ {
+ AsyncChains.getBlockingAndRethrow(node.commandStores().forEach(safeStore -> {
+ synchronized (redundantBefores)
+ {
+ redundantBefores.put(safeStore.commandStore().id(), safeStore.redundantBefore());
+ ranges.put(safeStore.commandStore().id(), safeStore.ranges());
+ durableBefores.put(safeStore.commandStore().id(), safeStore.durableBefore());
+ }
+ }));
+ }
return new CompactionInfo(redundantBefores, ranges, durableBefores);
}
diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
index c803852522..9a42e8cc93 100644
--- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
+++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
@@ -40,7 +40,6 @@ import accord.primitives.Unseekable;
import accord.primitives.Unseekables;
import accord.utils.Invariants;
import org.agrona.collections.ObjectHashSet;
-import org.apache.cassandra.index.accord.RoutesSearcher;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import static accord.local.CommandSummaries.SummaryStatus.NOT_DIRECTLY_WITNESSED;
@@ -62,7 +61,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
public static class Manager implements AccordCache.Listener<TxnId, Command>
{
private final AccordCommandStore commandStore;
- private final RoutesSearcher searcher = new RoutesSearcher();
+ private final RangeSearcher searcher;
private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
@@ -73,6 +72,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
{
caches.commands().register(this);
}
+ this.searcher = commandStore.rangeSearcher();
}
@Override
@@ -137,16 +137,16 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
{
case Range:
for (Unseekable range : searchKeysOrRanges)
- manager.searcher.intersects(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, forEach);
+ manager.searcher.search(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId).consume(forEach);
break;
case Key:
for (Unseekable key : searchKeysOrRanges)
- manager.searcher.intersects(manager.commandStore.id(), (AccordRoutingKey) key, minTxnId, maxTxnId, forEach);
+ manager.searcher.search(manager.commandStore.id(), (AccordRoutingKey) key, minTxnId, maxTxnId).consume(forEach);
}
if (!manager.transitive.isEmpty())
{
- for (var e : manager.transitive.tailMap(minTxnId, true).entrySet())
+ for (Map.Entry<TxnId, Ranges> e : manager.transitive.tailMap(minTxnId, true).entrySet())
{
if (e.getValue().intersects(searchKeysOrRanges))
forEach.accept(e.getKey());
diff --git a/src/java/org/apache/cassandra/service/accord/IndexRange.java b/src/java/org/apache/cassandra/service/accord/IndexRange.java
new file mode 100644
index 0000000000..426dde5a5b
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/accord/IndexRange.java
@@ -0,0 +1,60 @@
+/*
+ * 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 org.apache.cassandra.cache.IMeasurableMemory;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.ObjectSizes;
+
+public class IndexRange implements Comparable<IndexRange>, IMeasurableMemory
+{
+ private static final long EMPTY_SIZE = ObjectSizes.measure(new IndexRange(null, null));
+
+ public final byte[] start, end;
+
+ public IndexRange(byte[] start, byte[] end)
+ {
+ this.start = start;
+ this.end = end;
+ }
+
+ @Override
+ public int compareTo(IndexRange other)
+ {
+ int rc = ByteArrayUtil.compareUnsigned(start, 0, other.start, 0, start.length);
+ if (rc == 0)
+ rc = ByteArrayUtil.compareUnsigned(end, 0, other.end, 0, end.length);
+ return rc;
+ }
+
+ @Override
+ public long unsharedHeapSize()
+ {
+ return EMPTY_SIZE + ObjectSizes.sizeOfArray(start) * 2;
+ }
+
+ public boolean intersects(byte[] start, byte[] end)
+ {
+ if (ByteArrayUtil.compareUnsigned(this.start, 0, end, 0, end.length) >= 0)
+ return false;
+ if (ByteArrayUtil.compareUnsigned(this.end, 0, start, 0, start.length) <= 0)
+ return false;
+ return true;
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/accord/RangeSearcher.java b/src/java/org/apache/cassandra/service/accord/RangeSearcher.java
new file mode 100644
index 0000000000..3bfab38e38
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/accord/RangeSearcher.java
@@ -0,0 +1,128 @@
+/*
+ * 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.function.Consumer;
+
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.CloseableIterator;
+
+public interface RangeSearcher
+{
+ Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId);
+ Result search(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId);
+
+ static RangeSearcher extractRangeSearcher(Object o)
+ {
+ if (o instanceof RangeSearcher.Supplier)
+ return ((RangeSearcher.Supplier) o).rangeSearcher();
+ return NoopRangeSearcher.instance;
+ }
+
+ interface Supplier
+ {
+ RangeSearcher rangeSearcher();
+ }
+
+ interface Result
+ {
+ void consume(Consumer<TxnId> forEach);
+ CloseableIterator<TxnId> results();
+ }
+
+ class DefaultResult implements Result
+ {
+ private final TxnId minTxnId;
+ private final Timestamp maxTxnId;
+ private final CloseableIterator<TxnId> results;
+ private boolean consumed = false;
+
+ public DefaultResult(TxnId minTxnId, Timestamp maxTxnId, CloseableIterator<TxnId> results)
+ {
+ this.minTxnId = minTxnId;
+ this.maxTxnId = maxTxnId;
+ this.results = results;
+ }
+
+ @Override
+ public CloseableIterator<TxnId> results()
+ {
+ consume();
+ return results;
+ }
+
+ @Override
+ public void consume(Consumer<TxnId> forEach)
+ {
+ consume();
+ try (results)
+ {
+ while (results.hasNext())
+ {
+ TxnId next = results.next();
+ if (next.compareTo(minTxnId) >= 0 && next.compareTo(maxTxnId) < 0)
+ forEach.accept(next);
+ }
+ }
+ }
+
+ private void consume()
+ {
+ if (consumed)
+ throw new IllegalStateException("Attempted to consume an already consumed result");
+ consumed = true;
+ }
+ }
+
+ enum NoopResult implements Result
+ {
+ instance;
+
+ @Override
+ public void consume(Consumer<TxnId> forEach)
+ {
+
+ }
+
+ @Override
+ public CloseableIterator<TxnId> results()
+ {
+ return CloseableIterator.empty();
+ }
+ }
+
+ enum NoopRangeSearcher implements RangeSearcher
+ {
+ instance;
+
+ @Override
+ public Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId)
+ {
+ return NoopResult.instance;
+ }
+
+ @Override
+ public Result search(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId)
+ {
+ return NoopResult.instance;
+ }
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java b/src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java
new file mode 100644
index 0000000000..722d7d6f0c
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java
@@ -0,0 +1,234 @@
+/*
+ * 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.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NavigableSet;
+import java.util.TreeSet;
+import java.util.function.Consumer;
+
+import accord.primitives.Routable;
+import accord.primitives.Route;
+import accord.primitives.Timestamp;
+import accord.primitives.TxnId;
+import accord.primitives.Unseekable;
+import accord.utils.Invariants;
+import org.agrona.collections.Int2ObjectHashMap;
+import org.agrona.collections.Long2ObjectHashMap;
+import org.apache.cassandra.index.accord.OrderedRouteSerializer;
+import org.apache.cassandra.index.accord.RouteJournalIndex;
+import org.apache.cassandra.journal.StaticSegment;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.api.AccordRoutingKey;
+import org.apache.cassandra.utils.ByteArrayUtil;
+import org.apache.cassandra.utils.CloseableIterator;
+import org.apache.cassandra.utils.FastByteOperations;
+import org.apache.cassandra.utils.RTree;
+import org.apache.cassandra.utils.RangeTree;
+
+public class RouteInMemoryIndex<V> implements RangeSearcher
+{
+ private final Long2ObjectHashMap<SegmentIndex> segmentIndexes = new Long2ObjectHashMap<>();
+
+ public synchronized void update(long segment, int commandStoreId, TxnId id, Route<?> route)
+ {
+ if (!RouteJournalIndex.allowed(id))
+ return;
+ Invariants.nonNull(route, "route");
+ segmentIndexes.computeIfAbsent(segment, SegmentIndex::new).add(commandStoreId, id, route);
+ }
+
+ public synchronized void remove(Collection<StaticSegment<JournalKey, V>> oldSegments)
+ {
+ // As of this writing compact in accord journal takes StaticSegments, writes them to a SSTable, and pushes to a table;
+ // it then stops managing those segments... for this reason compactedSegments is normally empty and none of the
+ // oldSegments are expected to be tracked anymore, so this index should remove the reference (there is normal table 2i to pick up the job)
+ oldSegments.forEach(s -> segmentIndexes.remove(s.id()));
+ }
+
+ @Override
+ public RangeSearcher.Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId)
+ {
+ NavigableSet<TxnId> result = search(commandStoreId, range.table(),
+ OrderedRouteSerializer.serializeRoutingKeyNoTable(range.start()),
+ OrderedRouteSerializer.serializeRoutingKeyNoTable(range.end()));
+ return new DefaultResult(minTxnId, maxTxnId, CloseableIterator.wrap(result.iterator()));
+ }
+
+ private synchronized NavigableSet<TxnId> search(int storeId, TableId tableId, byte[] start, byte[] end)
+ {
+ TreeSet<TxnId> matches = new TreeSet<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, start, end, e -> matches.add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableSet() : matches;
+ }
+
+ @Override
+ public RangeSearcher.Result search(int commandStoreId, AccordRoutingKey key, TxnId minTxnId, Timestamp maxTxnId)
+ {
+ NavigableSet<TxnId> result = search(commandStoreId, key.table(), OrderedRouteSerializer.serializeRoutingKeyNoTable(key));
+ return new DefaultResult(minTxnId, maxTxnId, CloseableIterator.wrap(result.iterator()));
+ }
+
+ private synchronized NavigableSet<TxnId> search(int storeId, TableId tableId, byte[] key)
+ {
+ TreeSet<TxnId> matches = new TreeSet<>();
+ segmentIndexes.values().forEach(s -> s.search(storeId, tableId, key, e -> matches.add(e.getValue())));
+ return matches.isEmpty() ? Collections.emptyNavigableSet() : matches;
+ }
+
+ public synchronized void truncateForTesting()
+ {
+ segmentIndexes.clear();
+ }
+
+ private static class SegmentIndex
+ {
+ private final Int2ObjectHashMap<StoreIndex> storeIndexes = new Int2ObjectHashMap<>();
+
+ private SegmentIndex(long segment)
+ {
+ }
+
+ public void add(int commandStoreId, TxnId id, Route<?> route)
+ {
+ storeIndexes.computeIfAbsent(commandStoreId, i -> new StoreIndex()).add(id, route);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ StoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, start, end, fn);
+ }
+
+ public void search(int storeId, TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ StoreIndex idx = storeIndexes.get(storeId);
+ if (idx == null) return;
+ idx.search(tableId, key, fn);
+ }
+ }
+
+ private static class StoreIndex
+ {
+ private final Map<TableId, TableIndex> tableIndex = new HashMap<>();
+
+ private StoreIndex()
+ {
+ }
+
+ public void add(TxnId id, Route<?> route)
+ {
+ for (Unseekable keyOrRange : route)
+ add(id, keyOrRange);
+ }
+
+ private void add(TxnId id, Unseekable keyOrRange)
+ {
+ if (keyOrRange.domain() != Routable.Domain.Range)
+ throw new IllegalArgumentException("Unexpected domain: " + keyOrRange.domain());
+ TokenRange ts = (TokenRange) keyOrRange;
+ TableId tableId = ts.table();
+ tableIndex.computeIfAbsent(tableId, i -> new TableIndex()).add(id, ts);
+ }
+
+ public void search(TableId tableId, byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ TableIndex index = tableIndex.get(tableId);
+ if (index == null) return;
+ index.search(start, end, fn);
+ }
+
+ public void search(TableId tableId, byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ TableIndex index = tableIndex.get(tableId);
+ if (index == null) return;
+ index.search(key, fn);
+ }
+ }
+
+ private static class TableIndex
+ {
+ private final RangeTree<byte[], IndexRange, TxnId> index = createRangeTree();
+
+ private TableIndex()
+ {
+ }
+
+ public void add(TxnId id, TokenRange ts)
+ {
+ byte[] start = OrderedRouteSerializer.serializeRoutingKeyNoTable(ts.start());
+ byte[] end = OrderedRouteSerializer.serializeRoutingKeyNoTable(ts.end());
+ IndexRange range = new IndexRange(start, end);
+
+ index.add(range, id);
+ }
+
+ public void search(byte[] start, byte[] end, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ index.search(new IndexRange(start, end), fn);
+ }
+
+ public void search(byte[] key, Consumer<Map.Entry<IndexRange, TxnId>> fn)
+ {
+ index.searchToken(key, fn);
+ }
+ }
+
+ private static RangeTree<byte[], IndexRange, TxnId> createRangeTree()
+ {
+ return new RTree<>((a, b) -> ByteArrayUtil.compareUnsigned(a, 0, b, 0, a.length), new RangeTree.Accessor<>()
+ {
+ @Override
+ public byte[] start(IndexRange range)
+ {
+ return range.start;
+ }
+
+ @Override
+ public byte[] end(IndexRange range)
+ {
+ return range.end;
+ }
+
+ @Override
+ public boolean contains(byte[] start, byte[] end, byte[] bytes)
+ {
+ // bytes are ordered, start is exclusive, end is inclusive
+ return FastByteOperations.compareUnsigned(start, bytes) < 0
+ && FastByteOperations.compareUnsigned(end, bytes) >= 0;
+ }
+
+ @Override
+ public boolean intersects(IndexRange range, byte[] start, byte[] end)
+ {
+ return range.intersects(start, end);
+ }
+
+ @Override
+ public boolean intersects(IndexRange left, IndexRange right)
+ {
+ return left.intersects(right.start, right.end);
+ }
+ });
+ }
+}
diff --git a/src/java/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSource.java b/src/java/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSource.java
index f7d2b2acd6..101475fbfd 100644
--- a/src/java/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSource.java
+++ b/src/java/org/apache/cassandra/service/accord/serializers/AccordRoutingKeyByteSource.java
@@ -160,7 +160,7 @@ public class AccordRoutingKeyByteSource
throw new IOException("Unable to read prefix");
ByteSource.Peekable component = progress(bs);
- var prefix = ByteSourceInverse.getOptionalSignedFixedLength(ByteArrayAccessor.instance, component, 1);
+ byte[] prefix = ByteSourceInverse.getOptionalSignedFixedLength(ByteArrayAccessor.instance, component, 1);
if (prefix == null)
throw new IOException("Unable to read prefix; prefix was null");
@@ -192,7 +192,7 @@ public class AccordRoutingKeyByteSource
public ByteSource asComparableBytes(AccordRoutingKey key)
{
- var uuid = key.table().asUUID();
+ UUID uuid = key.table().asUUID();
ByteSource[] srcs = { LongType.instance.asComparableBytes(LongType.instance.decompose(uuid.getMostSignificantBits()), ByteComparable.Version.OSS50),
LongType.instance.asComparableBytes(LongType.instance.decompose(uuid.getLeastSignificantBits()), ByteComparable.Version.OSS50),
asComparableBytesNoTable(key) };
@@ -235,7 +235,7 @@ public class AccordRoutingKeyByteSource
public static <V> AccordRoutingKey fromComparableBytes(ValueAccessor<V> accessor, V data, ByteComparable.Version version, @Nullable IPartitioner partitioner)
{
- var bs = ByteSource.peekable(ByteSource.fixedLength(accessor, data));
+ ByteSource.Peekable bs = ByteSource.peekable(ByteSource.fixedLength(accessor, data));
long[] uuidValues = new long[2];
for (int i = 0; i < 2; i++)
{
@@ -251,7 +251,7 @@ public class AccordRoutingKeyByteSource
public static <V> AccordRoutingKey fromComparableBytes(ValueAccessor<V> accessor, V data, TableId tableId, ByteComparable.Version version, @Nullable IPartitioner partitioner)
{
- var bs = ByteSource.peekable(ByteSource.fixedLength(accessor, data));
+ ByteSource.Peekable bs = ByteSource.peekable(ByteSource.fixedLength(accessor, data));
return fromComparableBytes(bs, tableId, version, partitioner);
}
@@ -277,7 +277,7 @@ public class AccordRoutingKeyByteSource
throw new IllegalStateException("Unable to read prefix");
ByteSource.Peekable component = progress(bs);
- var prefix = ByteSourceInverse.getOptionalSignedFixedLength(ByteArrayAccessor.instance, component, 1);
+ byte[] prefix = ByteSourceInverse.getOptionalSignedFixedLength(ByteArrayAccessor.instance, component, 1);
if (prefix == null)
throw new IllegalStateException("Unable to read prefix; prefix was null");
switch (prefix[0])
diff --git a/src/java/org/apache/cassandra/utils/MergeIterator.java b/src/java/org/apache/cassandra/utils/MergeIterator.java
index 7ffb55a2db..9818e93083 100644
--- a/src/java/org/apache/cassandra/utils/MergeIterator.java
+++ b/src/java/org/apache/cassandra/utils/MergeIterator.java
@@ -19,6 +19,8 @@ package org.apache.cassandra.utils;
import java.util.*;
+import accord.utils.Invariants;
+
/** Merges sorted input iterators which individually contain unique items. */
public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implements IMergeIterator<In, Out>
{
@@ -44,6 +46,33 @@ public abstract class MergeIterator<In,Out> extends AbstractIterator<Out> implem
return new ManyToOne<>(sources, comparator, reducer);
}
+ public static <E extends Comparable<? super E>> MergeIterator<E, E> get(List<? extends Iterator<E>> sources)
+ {
+ return get(sources, Comparator.naturalOrder(), new Reducer<>()
+ {
+ private E first = null;
+ @Override
+ protected void onKeyChange()
+ {
+ first = null;
+ }
+
+ @Override
+ public void reduce(int idx, E current)
+ {
+ if (first == null)
+ first = current;
+ }
+
+ @Override
+ protected E getReduced()
+ {
+ Invariants.checkState(first != null);
+ return first;
+ }
+ });
+ }
+
public Iterable<? extends Iterator<In>> iterators()
{
return iterators;
diff --git a/src/java/org/apache/cassandra/utils/RTree.java b/src/java/org/apache/cassandra/utils/RTree.java
index f539512d26..5b7affe373 100644
--- a/src/java/org/apache/cassandra/utils/RTree.java
+++ b/src/java/org/apache/cassandra/utils/RTree.java
@@ -156,7 +156,7 @@ public class RTree<Token, Range, Value> implements RangeTree<Token, Range, Value
public int remove(Range key, Value value)
{
- var match = Map.entry(key, value);
+ Map.Entry<Range, Value> match = Map.entry(key, value);
return node.removeIf(match::equals);
}
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java
index 8c56f706da..3bc6046a3f 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/AccordTopologyMixupTest.java
@@ -253,6 +253,7 @@ public class AccordTopologyMixupTest extends TopologyMixupTestBase<AccordTopolog
{
c.set("accord.command_store_shard_count", 1)
.set("accord.queue_shard_count", 1)
+ .set("concurrent_accord_operations", 1)
.set("paxos_variant", Config.PaxosVariant.v2.name());
}
diff --git a/test/distributed/org/apache/cassandra/fuzz/topology/HarryOnAccordTopologyMixupTest.java b/test/distributed/org/apache/cassandra/fuzz/topology/HarryOnAccordTopologyMixupTest.java
index b9bd6573e9..21a2ea2ff5 100644
--- a/test/distributed/org/apache/cassandra/fuzz/topology/HarryOnAccordTopologyMixupTest.java
+++ b/test/distributed/org/apache/cassandra/fuzz/topology/HarryOnAccordTopologyMixupTest.java
@@ -73,7 +73,8 @@ public class HarryOnAccordTopologyMixupTest extends HarryTopologyMixupTest
{
super.onConfigure(config);
config.set("accord.command_store_shard_count", 1)
- .set("accord.queue_shard_count", 1);
+ .set("accord.queue_shard_count", 1)
+ .set("concurrent_accord_operations", 1);
}
@Override
diff --git a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
index caec9ec444..75fdd8fea8 100644
--- a/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
+++ b/test/distributed/org/apache/cassandra/service/accord/AccordJournalBurnTest.java
@@ -123,13 +123,12 @@ public class AccordJournalBurnTest extends BurnTestBase
{
ServerTestUtils.daemonInitialization();
- TableMetadata[] metadatas = new TableMetadata[4 + nodes.size()];
- metadatas[0] = AccordKeyspace.Commands;
- metadatas[1] = AccordKeyspace.CommandsForKeys;
- metadatas[2] = AccordKeyspace.Topologies;
- metadatas[3] = AccordKeyspace.EpochMetadata;
+ TableMetadata[] metadatas = new TableMetadata[3 + nodes.size()];
+ metadatas[0] = AccordKeyspace.CommandsForKeys;
+ metadatas[1] = AccordKeyspace.Topologies;
+ metadatas[2] = AccordKeyspace.EpochMetadata;
for (int i = 0; i < nodes.size(); i++)
- metadatas[4 + i] = AccordKeyspace.journalMetadata("journal_" + nodes.get(i));
+ metadatas[3 + i] = AccordKeyspace.journalMetadata("journal_" + nodes.get(i), false);
AccordKeyspace.TABLES = Tables.of(metadatas);
setUp();
diff --git a/test/unit/org/apache/cassandra/concurrent/ForwardingExecutorFactory.java b/test/unit/org/apache/cassandra/concurrent/ForwardingExecutorFactory.java
new file mode 100644
index 0000000000..b6bae2f1a4
--- /dev/null
+++ b/test/unit/org/apache/cassandra/concurrent/ForwardingExecutorFactory.java
@@ -0,0 +1,136 @@
+/*
+ * 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.concurrent;
+
+public class ForwardingExecutorFactory implements ExecutorFactory
+{
+ private final ExecutorFactory delegate;
+
+ public ForwardingExecutorFactory(ExecutorFactory delegate)
+ {
+ this.delegate = delegate;
+ }
+
+ protected ExecutorFactory delegate()
+ {
+ return delegate;
+ }
+
+ @Override
+ public ExecutorPlus pooled(String name, int threads)
+ {
+ return delegate().pooled(name, threads);
+ }
+
+ @Override
+ public ExecutorBuilderFactory<ExecutorPlus, SequentialExecutorPlus> withJmxInternal()
+ {
+ return delegate().withJmxInternal();
+ }
+
+ @Override
+ public ScheduledExecutorPlus scheduled(String name)
+ {
+ return delegate().scheduled(name);
+ }
+
+ @Override
+ public ScheduledExecutorPlus scheduled(String name, SimulatorSemantics simulatorSemantics)
+ {
+ return delegate().scheduled(name, simulatorSemantics);
+ }
+
+ @Override
+ public ScheduledExecutorPlus scheduled(boolean executeOnShutdown, String name)
+ {
+ return delegate().scheduled(executeOnShutdown, name);
+ }
+
+ @Override
+ public ScheduledExecutorPlus scheduled(boolean executeOnShutdown, String name, int priority)
+ {
+ return delegate().scheduled(executeOnShutdown, name, priority);
+ }
+
+ @Override
+ public Thread startThread(String name, Runnable runnable)
+ {
+ return delegate().startThread(name, runnable);
+ }
+
+ @Override
+ public Interruptible infiniteLoop(String name, Interruptible.SimpleTask task, InfiniteLoopExecutor.SimulatorSafe simulatorSafe)
+ {
+ return delegate().infiniteLoop(name, task, simulatorSafe);
+ }
+
+ @Override
+ public ExecutorBuilder<? extends SequentialExecutorPlus> configureSequential(String name)
+ {
+ return delegate().configureSequential(name);
+ }
+
+ @Override
+ public SequentialExecutorPlus sequential(String name)
+ {
+ return delegate().sequential(name);
+ }
+
+ @Override
+ public ExecutorBuilder<? extends ExecutorPlus> configurePooled(String name, int threads)
+ {
+ return delegate().configurePooled(name, threads);
+ }
+
+ @Override
+ public ExecutorBuilderFactory<ExecutorPlus, SequentialExecutorPlus> withJmx(String jmxPath)
+ {
+ return delegate().withJmx(jmxPath);
+ }
+
+ @Override
+ public LocalAwareSubFactory localAware()
+ {
+ return delegate().localAware();
+ }
+
+ @Override
+ public ScheduledExecutorPlus scheduled(boolean executeOnShutdown, String name, int priority, SimulatorSemantics simulatorSemantics)
+ {
+ return delegate().scheduled(executeOnShutdown, name, priority, simulatorSemantics);
+ }
+
+ @Override
+ public Thread startThread(String name, Runnable runnable, InfiniteLoopExecutor.Daemon daemon)
+ {
+ return delegate().startThread(name, runnable, daemon);
+ }
+
+ @Override
+ public Interruptible infiniteLoop(String name, Interruptible.Task task, InfiniteLoopExecutor.SimulatorSafe simulatorSafe, InfiniteLoopExecutor.Daemon daemon, InfiniteLoopExecutor.Interrupts interrupts)
+ {
+ return delegate().infiniteLoop(name, task, simulatorSafe, daemon, interrupts);
+ }
+
+ @Override
+ public ThreadGroup newThreadGroup(String name)
+ {
+ return delegate().newThreadGroup(name);
+ }
+}
diff --git a/test/unit/org/apache/cassandra/cql3/CQLTester.java b/test/unit/org/apache/cassandra/cql3/CQLTester.java
index d764133226..6568b566d9 100644
--- a/test/unit/org/apache/cassandra/cql3/CQLTester.java
+++ b/test/unit/org/apache/cassandra/cql3/CQLTester.java
@@ -3441,17 +3441,26 @@ public abstract class CQLTester
* Used by {@link #cleanupFileSystemListeners()} to know if file system listeners should be removed at the start
* of a test; can disable for cases where listeners are needed cross mutliple tests.
*/
- protected boolean cleanupFileSystemListeners = true;
+ protected static boolean cleanupFileSystemListeners = true;
@BeforeClass
public static void setUpClass()
+ {
+ prePrepareServer();
+
+ // Once per-JVM is enough
+ prepareServer();
+ }
+
+ protected static void prePrepareServer()
{
fs = FileSystems.newGlobalInMemoryFileSystem();
CassandraRelevantProperties.IGNORE_MISSING_NATIVE_FILE_HINTS.setBoolean(true);
FileSystems.maybeCreateTmp();
- CQLTester.setUpClass();
+ CQLTester.prePrepareServer();
}
+
@Before
public void cleanupFileSystemListeners()
{
diff --git a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
index c6e6809457..2746306534 100644
--- a/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
+++ b/test/unit/org/apache/cassandra/db/compaction/CompactionAccordIteratorsTest.java
@@ -18,16 +18,13 @@
package org.apache.cassandra.db.compaction;
-import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.LockSupport;
-import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.stream.Collectors;
@@ -60,7 +57,6 @@ import accord.primitives.Route;
import accord.primitives.SaveStatus;
import accord.primitives.Seekable;
import accord.primitives.Status;
-import accord.primitives.Status.Durability;
import accord.primitives.Txn;
import accord.primitives.Txn.Kind;
import accord.primitives.TxnId;
@@ -81,21 +77,18 @@ 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.schema.ColumnMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.service.accord.AccordCommandStore;
import org.apache.cassandra.service.accord.AccordExecutor;
-import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordService;
import org.apache.cassandra.service.accord.AccordTestUtils;
import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.Pair;
-import org.assertj.core.api.Assertions;
import static accord.local.KeyHistory.SYNC;
import static accord.local.PreLoadContext.contextFor;
@@ -103,17 +96,10 @@ import static accord.primitives.Routable.Domain.Range;
import static accord.utils.async.AsyncChains.getUninterruptibly;
import static org.apache.cassandra.Util.spinAssertEquals;
import static org.apache.cassandra.cql3.statements.schema.CreateTableStatement.parse;
-import static org.apache.cassandra.db.compaction.CompactionAccordIteratorsTest.DurableBeforeType.MAJORITY;
-import static org.apache.cassandra.db.compaction.CompactionAccordIteratorsTest.DurableBeforeType.NOT_DURABLE;
-import static org.apache.cassandra.db.compaction.CompactionAccordIteratorsTest.DurableBeforeType.UNIVERSAL;
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
import static org.apache.cassandra.service.accord.AccordKeyspace.COMMANDS_FOR_KEY;
-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.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -136,7 +122,6 @@ public class CompactionAccordIteratorsTest
private static final TxnId[] TXN_IDS = new TxnId[]{ TXN_ID, SECOND_TXN_ID };
private static final TxnId GT_SECOND_TXN_ID = AccordTestUtils.txnId(EPOCH, SECOND_TXN_ID.hlc() + 1, NODE);
- static ColumnFamilyStore commands;
static ColumnFamilyStore commandsForKey;
static TableMetadata table;
static FullRoute<?> route;
@@ -157,9 +142,6 @@ public class CompactionAccordIteratorsTest
parse("CREATE TABLE tbl (k int, c int, v int, PRIMARY KEY (k, c)) WITH transactional_mode = 'full'", "ks"));
StorageService.instance.initServer();
- commands = ColumnFamilyStore.getIfExists(SchemaConstants.ACCORD_KEYSPACE_NAME, AccordKeyspace.COMMANDS);
- commands.disableAutoCompaction();
-
commandsForKey = ColumnFamilyStore.getIfExists(SchemaConstants.ACCORD_KEYSPACE_NAME, COMMANDS_FOR_KEY);
commandsForKey.disableAutoCompaction();
@@ -176,52 +158,6 @@ public class CompactionAccordIteratorsTest
random = new Random(seed);
}
- // This isn't attempting to be an exhaustive test of Commands.shouldCleanup just that the return values
- // are handled correctly and that the interaction between the CompactionIterator and shoudCleanup seems reasonable
- @Test
- public void testAccordCommandsPurgerSingleCompaction() throws Throwable
- {
- testAccordCommandsPurger(true);
- }
-
- @Test
- public void testAccordCommandsPurgerMultipleCompactions() throws Throwable
- {
- testAccordCommandsPurger(false);
- }
-
- private void testAccordCommandsPurger(boolean singleCompaction) throws Throwable
- {
- this.singleCompaction = singleCompaction;
- // Null redudnant before should make no change since we have no information on this CommandStore
- testAccordCommandsPurger(null, DurableBefore.EMPTY, expectAccordCommandsNoChange());
- // Universally and locally durable (and global to boot) should be erased since literally everyone knows about it
- testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(UNIVERSAL), expectAccordCommandsErase());
- // Universally durable but not locally; we're stale, but shouldn't erase
- testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(UNIVERSAL), expectAccordCommandsNoChange());
- // redundantBefore will be > txnId as must be exclusive sync point, so will truncate
- testAccordCommandsPurger(redundantBefore(TXN_ID), durableBefore(MAJORITY), expectAccordCommandsTruncated());
- testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(MAJORITY), expectAccordCommandsNoChange());
- // Durable at a majority can be truncated with minimal data preserved, it must be redundant for this to occur
- testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(MAJORITY), expectAccordCommandsTruncated());
- // Not durable can be truncated, but needs the outcome preserved, it must be redundant for this to occur
- testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(NOT_DURABLE), expectAccordCommandsTruncatedWithOutcome());
- // Since it is redudnant but not known durable (outside of local)
- testAccordCommandsPurger(redundantBefore(GT_TXN_ID), durableBefore(DurableBeforeType.EMPTY), expectAccordCommandsTruncatedWithOutcome());
- // Never makes it past redundantBefore being LT_TXN_ID
- testAccordCommandsPurger(redundantBefore(LT_TXN_ID), durableBefore(DurableBeforeType.EMPTY), expectAccordCommandsNoChange());
- }
-
- private void testAccordCommandsPurger(RedundantBefore redundantBefore, DurableBefore durableBefore, Consumer<List<Partition>> expectedResult) throws Throwable
- {
- testWithCommandStore((commandStore) -> {
- IAccordService mockAccordService = mockAccordService(commandStore, redundantBefore, durableBefore);
- ColumnFamilyStore cfs = ColumnFamilyStore.getIfExists(ACCORD_KEYSPACE_NAME, AccordKeyspace.COMMANDS);
- List<Partition> result = compactCFS(mockAccordService, cfs);
- expectedResult.accept(result);
- }, false);
- }
-
@Test
public void testAccordCommandsForKeyPurgerSingleCompaction() throws Throwable
{
@@ -289,70 +225,6 @@ public class CompactionAccordIteratorsTest
}, true);
}
- Consumer<List<Partition>> expectAccordCommandsErase()
- {
- return partitions -> assertTrue(partitions.isEmpty());
- }
-
- Consumer<List<Partition>> expectAccordCommandsTruncatedWithOutcome()
- {
- return partitions -> {
- assertEquals(1, partitions.size());
- Partition partition = partitions.get(0);
- assertEquals(1, Iterators.size(partition.unfilteredIterator()));
- ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
- Row row = (Row) partition.unfilteredIterator().next();
- assertEquals(CommandsColumns.TRUNCATE_FIELDS.length, row.columnCount());
- for (ColumnMetadata cm : CommandsColumns.TRUNCATE_FIELDS)
- assertNotNull(row.getColumnData(cm));
- assertEquals(TXN_ID, CommandRows.getTxnId(partitionKeyComponents));
- assertEquals(Durability.Local, CommandRows.getDurability(row));
- assertEquals(TXN_ID, CommandRows.getExecuteAt(row));
- assertEquals(route, CommandRows.getRoute(row));
- assertEquals(SaveStatus.TruncatedApplyWithOutcome, AccordKeyspace.CommandRows.getStatus(row));
- };
- }
-
- Consumer<List<Partition>> expectAccordCommandsTruncated()
- {
- return partitions -> {
- assertEquals(1, partitions.size());
- Partition partition = partitions.get(0);
- assertEquals(1, Iterators.size(partition.unfilteredIterator()));
- ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
- Row row = (Row) partition.unfilteredIterator().next();
- assertEquals(CommandsColumns.TRUNCATE_FIELDS.length, row.columnCount());
- for (ColumnMetadata cm : CommandsColumns.TRUNCATE_FIELDS)
- assertNotNull(row.getColumnData(cm));
- assertEquals(TXN_ID, CommandRows.getTxnId(partitionKeyComponents));
- assertEquals(Durability.Local, CommandRows.getDurability(row));
- assertEquals(TXN_ID, CommandRows.getExecuteAt(row));
- assertEquals(route, CommandRows.getRoute(row));
- assertEquals(SaveStatus.TruncatedApply, AccordKeyspace.CommandRows.getStatus(row));
- };
- }
-
- Consumer<List<Partition>> expectAccordCommandsNoChange()
- {
- return partitions -> {
- assertEquals(1, partitions.size());
- Partition partition = partitions.get(0);
- assertEquals(1, Iterators.size(partition.unfilteredIterator()));
- ByteBuffer[] partitionKeyComponents = CommandRows.splitPartitionKey(partition.partitionKey());
- Row row = (Row) partition.unfilteredIterator().next();
-
- // execute_atleast is null, so when we read from the scanner the column won't be present in the partition
- Assertions.assertThat(new ArrayList<>(row.columns())).isEqualTo(commands.metadata().regularColumns().stream().filter(c -> !c.name.toString().equals("execute_atleast")).collect(Collectors.toList()));
- for (ColumnMetadata cm : commands.metadata().regularColumns())
- {
- if (cm.name.toString().equals("execute_atleast")) continue;
- assertNotNull(row.getColumnData(cm));
- }
- assertEquals(TXN_ID, CommandRows.getTxnId(partitionKeyComponents));
- assertEquals(SaveStatus.Applied, AccordKeyspace.CommandRows.getStatus(row));
- };
- }
-
private static RedundantBefore redundantBefore(TxnId txnId)
{
Ranges ranges = AccordTestUtils.fullRange(AccordTestUtils.keys(table, 42));
@@ -418,7 +290,6 @@ public class CompactionAccordIteratorsTest
cache.global.setCapacity(cacheSize);
}
});
- commands.forceBlockingFlush(FlushReason.UNIT_TESTS);
commandsForKey.forceBlockingFlush(FlushReason.UNIT_TESTS);
while (commandStore.executor().hasTasks())
LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
@@ -449,20 +320,20 @@ public class CompactionAccordIteratorsTest
PartialTxn partialTxn = txn.slice(commandStore.unsafeGetRangesForEpoch().currentRanges(), true);
Route<?> partialRoute = route.slice(commandStore.unsafeGetRangesForEpoch().currentRanges());
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
- CheckedCommands.preaccept(safe, txnId, partialTxn, route, appendDiffToKeyspace(commandStore));
+ CheckedCommands.preaccept(safe, txnId, partialTxn, route, (a, b) -> {});
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
- CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, txnId, partialDeps, appendDiffToKeyspace(commandStore));
+ CheckedCommands.accept(safe, txnId, Ballot.ZERO, partialRoute, txnId, partialDeps, (a, b) -> {});
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
- CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, txnId, partialDeps, appendDiffToKeyspace(commandStore));
+ CheckedCommands.commit(safe, SaveStatus.Stable, Ballot.ZERO, txnId, route, partialTxn, txnId, partialDeps, (a, b) -> {});
}).beginAsResult());
flush(commandStore);
getUninterruptibly(commandStore.execute(contextFor(txnId, route, SYNC), safe -> {
Pair<Writes, Result> result = AccordTestUtils.processTxnResultDirect(safe, txnId, partialTxn, txnId);
- CheckedCommands.apply(safe, txnId, route, txnId, partialDeps, partialTxn, result.left, result.right, appendDiffToKeyspace(commandStore));
+ CheckedCommands.apply(safe, txnId, route, txnId, partialDeps, partialTxn, result.left, result.right, (a, b) -> {});
}).beginAsResult());
flush(commandStore);
// The apply chain is asychronous, so it is easiest to just spin until it is applied
@@ -471,18 +342,11 @@ public class CompactionAccordIteratorsTest
return getUninterruptibly(commandStore.submit(contextFor(txnId, route, SYNC), safe -> {
StoreParticipants participants = StoreParticipants.all(route);
Command command = safe.get(txnId, participants).current();
- appendDiffToKeyspace(commandStore).accept(null, command);
return command.hasBeen(Status.Applied);
}).beginAsResult());
});
flush(commandStore);
}
- UntypedResultSet commandsTable = QueryProcessor.executeInternal("SELECT * FROM " + ACCORD_KEYSPACE_NAME + "." + AccordKeyspace.COMMANDS + ";");
- logger.info(commandsTable.toStringUnsafe());
- assertEquals(txnIds.length, commandsTable.size());
- Iterator<UntypedResultSet.Row> commandsTableIterator = commandsTable.iterator();
- for (TxnId txnId : txnIds)
- assertEquals(txnId, AccordKeyspace.deserializeTimestampOrNull(commandsTableIterator.next().getBytes("txn_id"), TxnId::fromBits));
UntypedResultSet commandsForKeyTable = QueryProcessor.executeInternal("SELECT * FROM " + ACCORD_KEYSPACE_NAME + "." + COMMANDS_FOR_KEY + ";");
logger.info(commandsForKeyTable.toStringUnsafe());
assertEquals(1, commandsForKeyTable.size());
@@ -493,14 +357,6 @@ public class CompactionAccordIteratorsTest
test.test(commandStore);
}
- // This little bit of magic is required because we do not expose range commands explicitly, but still need to compact them
- private static BiConsumer<Command, Command> appendDiffToKeyspace(AccordCommandStore commandStore)
- {
- return (before, after) -> {
- AccordKeyspace.getCommandMutation(commandStore.id(), after, commandStore.nextSystemTimestampMicros()).applyUnsafe();
- };
- }
-
private List<Partition> compactCFS(IAccordService mockAccordService, ColumnFamilyStore cfs)
{
List<ISSTableScanner> scanners = cfs.getLiveSSTables().stream().map(SSTableReader::getScanner).collect(Collectors.toList());
diff --git a/test/unit/org/apache/cassandra/dht/IPartitionerTest.java b/test/unit/org/apache/cassandra/dht/IPartitionerTest.java
index e531e25da5..d2196bdf5a 100644
--- a/test/unit/org/apache/cassandra/dht/IPartitionerTest.java
+++ b/test/unit/org/apache/cassandra/dht/IPartitionerTest.java
@@ -34,6 +34,7 @@ import org.apache.cassandra.utils.CassandraGenerators;
import org.apache.cassandra.utils.bytecomparable.ByteComparable;
import org.apache.cassandra.utils.bytecomparable.ByteSource;
import org.assertj.core.api.Assertions;
+import org.quicktheories.core.Gen;
import org.reflections.Reflections;
import org.reflections.scanners.Scanners;
import org.reflections.util.ConfigurationBuilder;
@@ -93,7 +94,9 @@ public class IPartitionerTest
@Test
public void byteCompareSerde()
{
- qt().forAll(AccordGenerators.fromQT(CassandraGenerators.token())).check(token -> {
+ // make sure to use simplify as local partitioner can have a type that could generate data too large causing this test to be flakey
+ Gen<Token> qt = CassandraGenerators.partitioners().flatMap(p -> CassandraGenerators.token(CassandraGenerators.simplify(p)));
+ qt().forAll(AccordGenerators.fromQT(qt)).check(token -> {
var p = token.getPartitioner();
var comparable = Objects.requireNonNull(ByteSource.peekable(p.getTokenFactory().asComparableBytes(token, ByteComparable.Version.OSS50)));
Token read = p.getTokenFactory().fromComparableBytes(comparable, ByteComparable.Version.OSS50);
diff --git a/test/unit/org/apache/cassandra/index/accord/AccordIndexStressTest.java b/test/unit/org/apache/cassandra/index/accord/AccordIndexStressTest.java
deleted file mode 100644
index 79cc0b66b0..0000000000
--- a/test/unit/org/apache/cassandra/index/accord/AccordIndexStressTest.java
+++ /dev/null
@@ -1,527 +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.index.accord;
-
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.TreeSet;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-import java.util.stream.LongStream;
-
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import accord.api.RoutingKey;
-import accord.local.Node;
-import accord.local.StoreParticipants;
-import accord.primitives.SaveStatus;
-import accord.primitives.Status;
-import accord.primitives.FullKeyRoute;
-import accord.primitives.Range;
-import accord.primitives.Ranges;
-import accord.primitives.Routable;
-import accord.primitives.Route;
-import accord.primitives.Timestamp;
-import accord.primitives.Txn;
-import accord.primitives.TxnId;
-import accord.utils.RandomSource;
-import org.agrona.collections.Int2ObjectHashMap;
-import org.agrona.collections.Long2ObjectHashMap;
-import org.agrona.collections.ObjectHashSet;
-import org.apache.cassandra.config.CassandraRelevantProperties;
-import org.apache.cassandra.cql3.CQLTester;
-import org.apache.cassandra.cql3.UntypedResultSet;
-import org.apache.cassandra.db.ColumnFamilyStore;
-import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.dht.Murmur3Partitioner;
-import org.apache.cassandra.exceptions.ReadSizeAbortException;
-import org.apache.cassandra.schema.TableId;
-import org.apache.cassandra.service.accord.AccordKeyspace;
-import org.apache.cassandra.service.accord.TokenRange;
-import org.apache.cassandra.service.accord.api.AccordRoutingKey;
-import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
-import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.Interval;
-import org.apache.cassandra.utils.IntervalTree;
-import org.apache.cassandra.utils.ObjectSizes;
-import org.assertj.core.api.Assertions;
-
-import static accord.utils.Property.qt;
-import static org.apache.cassandra.utils.Clock.Global.nanoTime;
-
-/**
- * This test validates the system_accord.commands.route index to make sure it returns the right values... the way the
- * test is strucutred allows pluggability but runs with fixed configs while running in CI...
- *
- * If you are interested in testing different cases, the following tunables exist:
- *
- * <table>
- * <li>size: how many read/writes should be done</li>
- * <li>pattern: how ranges are layed out. By default we test NO_OVERLAP so there are no conflicts</li>
- * </table>
- */
-public class AccordIndexStressTest extends CQLTester
-{
- private static final Logger logger = LoggerFactory.getLogger(AccordIndexStressTest.class);
-
- static
- {
- // The plan is to migrate away from SAI, so rather than hacking around timeout issues; just disable for now
- CassandraRelevantProperties.SAI_TEST_DISABLE_TIMEOUT.setBoolean(true);
- }
- private static final boolean VALIDATE = true;
- private static final boolean INCLUDE_FLUSH = true;
- private static final long SLOW_NS = TimeUnit.MILLISECONDS.toNanos(25);
- private static final Node.Id NODE = new Node.Id(42);
- private final Routable.Domain domain = Routable.Domain.Range;
- private final Int2ObjectHashMap<Map<TableId, Long2ObjectHashMap<List<TxnId>>>> storeToTableToRoutingKeysToTxns = new Int2ObjectHashMap<>();
- private final Int2ObjectHashMap<Map<TableId, Map<TokenRange, List<TxnId>>>> storeToTableToRangesToTxns = new Int2ObjectHashMap<>();
- private final RoutesSearcher searcher = new RoutesSearcher();
-
- enum Size
- {tiny, small, medium, large, benchmark}
-
- private final Size size = Size.small;
-
- private enum Read
- {INDEX, CQL}
-
- private Read read = Read.INDEX;
-
- private enum Pattern { RANDOM, NO_OVERLAP }
- private final Pattern pattern = Pattern.NO_OVERLAP;
-
- @Test
- public void test()
- {
- var tables = IntStream.range(0, 10)
- .mapToObj(i -> TableId.fromUUID(new UUID(0, i)))
- .collect(Collectors.toList());
- int numWrites, numReads;
- switch (size)
- {
- case tiny:
- numWrites = 100;
- numReads = 10;
- break;
- case small:
- numWrites = 100_000;
- numReads = 1_000;
- break;
- case medium:
- numWrites = 1_000_000;
- numReads = 1_000;
- break;
- case large:
- numWrites = 10_000_000;
- numReads = 1_000;
- break;
- case benchmark:
- numWrites = 1_000_000;
- numReads = numWrites * 10;
- break;
- default:
- throw new AssertionError("Unknown size: " + size);
- }
- var minToken = 0;
- var maxToken = (1 << 8) * numWrites;
- int numStores = 10;
- qt().withSeed(-1464527987857660885L).withExamples(1).check(rs -> {
- timed("write(" + numWrites + ")", () -> writeRecords(rs, numStores, tables, minToken, maxToken, numWrites));
- if (INCLUDE_FLUSH)
- timed("flush(writes=" + numWrites + ")", () -> FBUtilities.waitOnFutures(Keyspace.open("system_accord").flush(ColumnFamilyStore.FlushReason.UNIT_TESTS)));
- var warmupReads = Math.max(1, (int) (numReads * .2));
- timed("warmup read(" + warmupReads + ")", () -> readRecords(rs, warmupReads));
- timed("read(" + numReads + ")", () -> readRecords(rs, numReads));
- });
- }
-
- private static void timed(String name, Runnable fn)
- {
- logger.warn("Task {} starting...", name);
- long startNs = nanoTime();
- try
- {
- fn.run();
- }
- catch (Throwable t)
- {
- logger.warn("Task {} failed after {}ms", name, TimeUnit.NANOSECONDS.toMillis(nanoTime() - startNs));
- throw t;
- }
- logger.warn("Task {} completed after {}ms", name, TimeUnit.NANOSECONDS.toMillis(nanoTime() - startNs));
- }
-
- private static class RangeWrapper
- {
- final TokenRange[] ranges;
- final IntervalTree<RoutingKey, TxnId, Interval<RoutingKey, TxnId>> tree;
-
- RangeWrapper(TokenRange[] ranges, IntervalTree<RoutingKey, TxnId, Interval<RoutingKey, TxnId>> tree)
- {
- this.ranges = ranges;
- this.tree = tree;
- }
- }
-
- private Int2ObjectHashMap<Map<TableId, long[]>> store2Table2Tokens;
- private Int2ObjectHashMap<Map<TableId, RangeWrapper>> store2Table2Ranges;
-
- private void readRecords(RandomSource rs, int numRecords)
- {
- logger.warn("The bookkeeping is {} bytes", ObjectSizes.measureDeep(storeToTableToRoutingKeysToTxns));
- // sort the tokens
- if (domain == Routable.Domain.Key && store2Table2Tokens == null)
- {
- store2Table2Tokens = new Int2ObjectHashMap<>();
- timed("Model building: key", () -> storeToTableToRoutingKeysToTxns.forEachInt((storeId, actual) -> {
- Map<TableId, long[]> map = new HashMap<>();
- for (var e : actual.entrySet())
- {
- var keys = e.getValue().keySet();
- long[] tokens = new long[keys.size()];
- var it = keys.iterator();
- for (int i = 0; it.hasNext(); i++)
- tokens[i] = it.nextLong();
- Arrays.sort(tokens);
- map.put(e.getKey(), tokens);
- }
- store2Table2Tokens.put(storeId, map);
- }));
- store2Table2Ranges = null;
- }
- else if (domain == Routable.Domain.Range && store2Table2Ranges == null)
- {
- store2Table2Ranges = new Int2ObjectHashMap<>();
- timed("Model building: range", () -> storeToTableToRangesToTxns.forEachInt((storeId, actual) -> {
- Map<TableId, RangeWrapper> map = new HashMap<>();
- for (var e : actual.entrySet())
- {
- TableId tableId = e.getKey();
- Map<TokenRange, List<TxnId>> range2Txns = e.getValue();
- var keys = range2Txns.keySet();
- TokenRange[] ranges = new TokenRange[keys.size()];
- var it = keys.iterator();
- var builder = new IntervalTree.Builder<RoutingKey, TxnId, Interval<RoutingKey, TxnId>>();
- for (int i = 0; it.hasNext(); i++)
- {
- TokenRange r = ranges[i] = it.next();
- List<TxnId> txns = range2Txns.get(r);
- txns.forEach(txnId -> builder.add(new Interval<>(r.start(), r.end(), txnId)));
- }
- Arrays.sort(ranges, Range::compare);
- map.put(tableId, new RangeWrapper(ranges, builder.build()));
- }
- store2Table2Ranges.put(storeId, map);
- }));
- store2Table2Tokens = null;
- }
- long[] samples = new long[numRecords];
- int[] counts = new int[numRecords];
- int size = 0;
- int numReadSizeAborts = 0;
- try
- {
- for (int i = 0; i < numRecords; i++)
- {
- int store;
- TableId table;
- Set<TxnId> expected = new HashSet<>();
- TokenKey start, end;
- switch (domain)
- {
- case Key:
- {
- store = rs.pickUnorderedSet(storeToTableToRoutingKeysToTxns.keySet());
- var actual = this.storeToTableToRoutingKeysToTxns.get(store);
- var tableToTokens = store2Table2Tokens.get(store);
-
- table = rs.pickUnorderedSet(actual.keySet());
- var tokens = tableToTokens.get(table);
-
- var offset = rs.nextInt(0, tokens.length);
- var endOffset = offset == tokens.length - 1 ? tokens.length - 1 : offset + rs.nextInt(1, Math.min(3, tokens.length - offset));
- IntStream.range(offset + 1, endOffset + 1).mapToLong(o -> tokens[o]).forEach(token -> expected.addAll(actual.get(table).get(token)));
-
- start = new TokenKey(table, new Murmur3Partitioner.LongToken(tokens[offset]));
- end = new TokenKey(table, new Murmur3Partitioner.LongToken(tokens[endOffset]));
- }
- break;
- case Range:
- {
- store = rs.pickUnorderedSet(storeToTableToRangesToTxns.keySet());
- var tableToRangesToTxns = storeToTableToRangesToTxns.get(store);
- var tableToRanges = store2Table2Ranges.get(store);
-
- table = rs.pickUnorderedSet(tableToRangesToTxns.keySet());
- var wrapper = tableToRanges.get(table);
- var ranges = wrapper.ranges;
- var tree = wrapper.tree;
- var range = rs.pick(ranges);
- var a = tokenValue(range.start());
- var b = tokenValue(range.end());
- start = new TokenKey(table, new Murmur3Partitioner.LongToken(a + 1));
- end = new TokenKey(table, new Murmur3Partitioner.LongToken(b - 1));
- expected.addAll(tree.search(start, end));
- assert !expected.isEmpty();
- }
- break;
- default:
- throw new IllegalArgumentException("Unknown domain: " + domain);
- }
-
- var startNs = nanoTime();
- // TODO (desired): randomise lower bound for reading
- Set<TxnId> actual = read(store, start, end, TxnId.NONE, Timestamp.MAX);
- var durationNs = nanoTime() - startNs;
- samples[size] = durationNs;
- counts[size++] = actual.size();
- if (slow(durationNs))
- logger.warn("Slow search: i={}, store={}, [{}, {}), results={}", i, store, start, end, actual.size());
- if (VALIDATE)
- {
- try
- {
- Assertions.assertThat(actual).describedAs("[%s, %s)", start, end).isEqualTo(expected);
- }
- catch (Throwable t)
- {
- logSamples(samples, counts, size);
- throw t;
- }
- }
- }
- }
- finally
- {
- logger.info("Number of aborts due to size: {}", numReadSizeAborts);
- logSamples(samples, counts, size);
- }
- }
-
- private static long tokenValue(RoutingKey start)
- {
- return ((AccordRoutingKey.TokenKey) start).token().getLongValue();
- }
-
- private static boolean slow(long durationNs)
- {
- return durationNs >= SLOW_NS;
- }
-
- private Set<TxnId> read(int store, AccordRoutingKey start, AccordRoutingKey end, TxnId minTxnId, Timestamp maxTxnId)
- {
- switch (read)
- {
- case INDEX:
- return readIndex(store, start, end, minTxnId, maxTxnId);
- case CQL:
- return readCQL(store, start, end, minTxnId, maxTxnId);
- default:
- throw new AssertionError("Unknown read type: " + read);
- }
- }
-
- private Set<TxnId> readIndex(int store, AccordRoutingKey start, AccordRoutingKey end, TxnId minTxnId, Timestamp maxTxnId)
- {
- Set<TxnId> out = new ObjectHashSet<>();
- searcher.intersects(store, start, end, minTxnId, maxTxnId, out::add);
- return out;
- }
-
- private Set<TxnId> readCQL(int store, AccordRoutingKey start, AccordRoutingKey end, TxnId minTxnId, Timestamp maxTxnId)
- {
- Set<TxnId> actual = new ObjectHashSet<>();
- try
- {
- UntypedResultSet results = execute("SELECT txn_id FROM system_accord.commands WHERE store_id = ? AND route > ? AND route <= ?", store, OrderedRouteSerializer.serializeRoutingKey(start), OrderedRouteSerializer.serializeRoutingKey(end));
- for (var row : results)
- {
- TxnId txnId = AccordKeyspace.deserializeTxnId(row);
- if (txnId.compareTo(minTxnId) >= 0 && txnId.compareTo(maxTxnId) < 0)
- actual.add(txnId);
- }
- }
- catch (ReadSizeAbortException e)
- {
- // don't count it...
- logger.warn("Abort query to [{}, {}) do to size", start, end);
- return null;
- }
- return actual;
- }
-
- private void logSamples(long[] samples, int[] counts, int size)
- {
- if (size == 0)
- {
- logger.warn("No logs sampled");
- return;
- }
- Arrays.sort(samples, 0, size);
- Arrays.sort(counts, 0, size);
- StringBuilder sb = new StringBuilder();
- sb.append("Samples: ").append(size);
- sb.append("\nLatenciy:");
- sb.append("\n Min (micro): ").append(TimeUnit.NANOSECONDS.toMicros(samples[0]));
- sb.append("\n Max (micro): ").append(TimeUnit.NANOSECONDS.toMicros(samples[size - 1]));
- sb.append("\n Median (micro): ").append(TimeUnit.NANOSECONDS.toMicros(samples[size / 2]));
- sb.append("\n Avg (micro): ").append(TimeUnit.NANOSECONDS.toMicros((long) LongStream.of(samples).limit(size).average().getAsDouble()));
- sb.append("\nCounts:");
- sb.append("\n Min: ").append(counts[0]);
- sb.append("\n Max: ").append(counts[size - 1]);
- sb.append("\n Median: ").append(counts[size / 2]);
- sb.append("\n Avg: ").append((int) IntStream.of(counts).limit(size).average().getAsDouble());
- logger.info(sb.toString());
- }
-
- private void writeRecords(RandomSource rs,
- int numStores,
- List<TableId> tables,
- int minToken, int maxToken,
- int numRecords)
- {
- var cql = "INSERT INTO system_accord.commands (store_id, domain, txn_id, status, participants, durability) VALUES (?, ?, ?, ?, ?, ?)";
- for (int i = 0; i < numRecords; i++)
- {
- int store = rs.nextInt(0, numStores);
- TxnId txnId = new TxnId(0, 1000 + i, Txn.Kind.Write, domain, NODE);
- int domain = txnId.domain().ordinal();
- int status = SaveStatus.PreCommitted.ordinal();
- ByteBuffer routeBB;
- try
- {
- StoreParticipants participants = StoreParticipants.all(createRoute(rs, numRecords, i, rs.nextInt(1, 20), tables, minToken, maxToken));
- for (var u : participants.route())
- {
- switch (u.domain())
- {
- case Key:
- {
- AccordRoutingKey key = (AccordRoutingKey) u;
- var table = key.table();
- var token = key.token().getLongValue();
- storeToTableToRoutingKeysToTxns.computeIfAbsent(store, ignore -> new HashMap<>())
- .computeIfAbsent(table, ignore -> new Long2ObjectHashMap<>())
- .computeIfAbsent(token, ignore -> new ArrayList<>())
- .add(txnId);
- }
- break;
- case Range:
- {
- TokenRange range = (TokenRange) u;
- var table = range.table();
- storeToTableToRangesToTxns.computeIfAbsent(store, ignore -> new HashMap<>())
- .computeIfAbsent(table, ignore -> new HashMap<>())
- .computeIfAbsent(range, ignore -> new ArrayList<>())
- .add(txnId);
- }
- break;
- default:
- throw new AssertionError("Unexpected domain: " + u.domain());
- }
- }
- routeBB = AccordKeyspace.serializeParticipants(participants);
- }
- catch (IOException e)
- {
- throw new UncheckedIOException(e);
- }
- int durability = Status.Durability.NotDurable.ordinal();
-
- execute(cql, store, domain, AccordKeyspace.serializeTimestamp(txnId), status, routeBB, durability);
- }
- }
-
- private Route<?> createRoute(RandomSource rs, int numRecords, int index, int numKeys, List<TableId> tables, int minToken, int maxToken)
- {
- switch (domain)
- {
- case Key:
- {
- TreeSet<AccordRoutingKey> keys = new TreeSet<>();
- while (keys.size() < numKeys)
- {
- var table = rs.pick(tables);
- var token = new Murmur3Partitioner.LongToken(rs.nextInt(minToken, maxToken));
- keys.add(new TokenKey(table, token));
- }
- return new FullKeyRoute(keys.first(), keys.toArray(RoutingKey[]::new));
- }
- case Range:
- {
- TreeSet<TokenRange> ranges = new TreeSet<>(Range::compareTo);
- RoutingKey routingKey = null;
- var domain = maxToken - minToken + 1;
- var delta = domain / numRecords;
- var sub_delta = delta / numKeys;
- while (ranges.size() < numKeys)
- {
- var table = rs.pick(tables);
- int a, b;
- switch (pattern)
- {
- case RANDOM:
- {
- a = rs.nextInt(minToken, maxToken);
- b = rs.nextInt(minToken, maxToken);
- while (a == b)
- b = rs.nextInt(minToken, maxToken);
- if (a > b)
- {
- var tmp = a;
- a = b;
- b = tmp;
- }
- }
- break;
- case NO_OVERLAP:
- {
- a = delta * index + (sub_delta * ranges.size());
- b = a + sub_delta;
- }
- break;
- default:
- throw new IllegalArgumentException("Unknown pattern: " + pattern);
- }
- ranges.add(TokenRange.create(new TokenKey(table, new Murmur3Partitioner.LongToken(a)), new TokenKey(table, new Murmur3Partitioner.LongToken(b))));
- if (routingKey == null)
- {
- routingKey = new TokenKey(table, new Murmur3Partitioner.LongToken(b));
- }
- }
- return Ranges.ofSorted(ranges.toArray(Range[]::new)).toRoute(routingKey);
- }
- default:
- throw new IllegalArgumentException("Unknown domain: " + domain);
- }
- }
-}
diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
index 4832a32306..a6a5ebd545 100644
--- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
@@ -24,26 +24,36 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.TreeSet;
-import java.util.UUID;
+import java.util.function.Supplier;
import java.util.stream.Collectors;
-import java.util.stream.IntStream;
import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
+import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import accord.api.Journal;
import accord.api.RoutingKey;
+import accord.local.CommandStores;
+import accord.local.DurableBefore;
import accord.local.Node;
+import accord.local.RedundantBefore;
+import accord.local.RedundantStatus;
import accord.local.StoreParticipants;
-import accord.primitives.SaveStatus;
-import accord.primitives.Status.Durability;
+import accord.primitives.Ballot;
+import accord.primitives.Deps;
import accord.primitives.FullKeyRoute;
+import accord.primitives.PartialDeps;
+import accord.primitives.Participants;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Routable.Domain;
import accord.primitives.Route;
+import accord.primitives.SaveStatus;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
@@ -55,8 +65,14 @@ import accord.utils.RandomSource;
import org.agrona.collections.Int2ObjectHashMap;
import org.agrona.collections.Long2ObjectHashMap;
import org.agrona.collections.ObjectHashSet;
+import org.apache.cassandra.concurrent.ExecutorFactory;
+import org.apache.cassandra.concurrent.ForwardingExecutorFactory;
+import org.apache.cassandra.concurrent.ForwardingExecutorPlus;
+import org.apache.cassandra.concurrent.ImmediateExecutor;
+import org.apache.cassandra.concurrent.SequentialExecutorPlus;
import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.config.OptionaldPositiveInt;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
@@ -64,54 +80,92 @@ import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.dht.Murmur3Partitioner.LongToken;
import org.apache.cassandra.schema.SchemaConstants;
import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.accord.AccordJournal;
import org.apache.cassandra.service.accord.AccordKeyspace;
import org.apache.cassandra.service.accord.AccordService;
+import org.apache.cassandra.service.accord.AccordTestUtils;
+import org.apache.cassandra.service.accord.AccordTopology;
+import org.apache.cassandra.service.accord.IAccordService;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import org.apache.cassandra.service.accord.api.AccordRoutingKey.TokenKey;
+import org.apache.cassandra.service.consensus.TransactionalMode;
+import org.apache.cassandra.tcm.ClusterMetadata;
+import org.apache.cassandra.tcm.membership.NodeId;
+import org.apache.cassandra.utils.AccordGenerators;
import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.utils.LazyToString;
import org.apache.cassandra.utils.RTree;
import org.apache.cassandra.utils.RangeTree;
import org.assertj.core.api.Assertions;
+import org.mockito.Mockito;
import static accord.utils.Property.commands;
import static accord.utils.Property.stateful;
+import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
public class RouteIndexTest extends CQLTester.InMemory
{
- static
- {
- // since this test does frequent truncates, the info table gets updated and forced flushed... which is 90% of the cost of this test...
- // this flag disables that flush
- CassandraRelevantProperties.UNSAFE_SYSTEM.setBoolean(true);
- }
-
private static final Node.Id NODE = new Node.Id(42);
private static final int MIN_TOKEN = 0;
private static final int MAX_TOKEN = 1 << 18;
private static final int TOKEN_RANGE_SIZE = MAX_TOKEN - MIN_TOKEN + 1;
- private static final Gen.IntGen NUM_STORES_GEN = Gens.ints().between(1, 10);
- private static final Gen.IntGen NUM_TABLES_GEN = Gens.ints().between(1, 10);
+ private static final int MAX_STORES = 10;
+ private static final Gen.IntGen NUM_STORES_GEN = Gens.ints().between(1, MAX_STORES);
private static final Gen<Gen.IntGen> TOKEN_DISTRIBUTION = Gens.mixedDistribution(MIN_TOKEN, MAX_TOKEN + 1);
private static final Gen<Gen.IntGen> RANGE_SIZE_DISTRIBUTION = Gens.mixedDistribution(10, (int) (TOKEN_RANGE_SIZE * .01));
private static final Gen<Gen<Domain>> DOMAIN_DISTRIBUTION = Gens.mixedDistribution(Domain.values());
- private static RoutesSearcher ROUTES_SEARCHER = null;
@BeforeClass
public static void setUpClass()
{
- CQLTester.InMemory.setUpClass();
+ // since this test does frequent truncates, the info table gets updated and forced flushed... which is 90% of the cost of this test...
+ // this flag disables that flush
+ CassandraRelevantProperties.UNSAFE_SYSTEM.setBoolean(true);
+
+ DatabaseDescriptor.daemonInitialization();
+ DatabaseDescriptor.setAccordTransactionsEnabled(true);
+ // disable journal compaction so the test can control when it happens
+ DatabaseDescriptor.getAccord().enable_journal_compaction = false;
DatabaseDescriptor.setIncrementalBackupsEnabled(false);
- ROUTES_SEARCHER = new RoutesSearcher();
+ DatabaseDescriptor.setAutoSnapshot(false);
+
+ CQLTester.InMemory.prePrepareServer();
+
+ // Journal will async release segment references and close files... this adds possible race condition issues with truncate
+ // so make these steps happen inline.
+ // Simulatored executors for the journal would be ideal, but given the blocking nature only a few executors can be simulated
+ // without The Simulator, so full simulation is out of scope of this test.
+ ExecutorFactory.Global.unsafeSet(new ForwardingExecutorFactory(ExecutorFactory.Global.executorFactory()) {
+ @Override
+ public SequentialExecutorPlus sequential(String name)
+ {
+ if (name.endsWith("-releaser") || name.endsWith("-closer"))
+ return new ForwardingExecutorPlus(ImmediateExecutor.INSTANCE);
+ return super.sequential(name);
+ }
+ });
+
+ CQLTester.InMemory.prepareServer();
+ }
+
+ private static TableId tableId = null;
+
+ @Before
+ public void setupTable()
+ {
+ if (tableId != null) return;
+ String name = createTable("CREATE TABLE %s (pk int primary key) WITH " + TransactionalMode.full.asCqlParam());
+ tableId = Keyspace.open(KEYSPACE).getColumnFamilyStore(name).metadata().id;
}
- private Command<State, ColumnFamilyStore, ?> insert(RandomSource rs, State state)
+ private Command<State, Sut, ?> insert(RandomSource rs, State state)
{
- int storeId = rs.nextInt(0, state.numStores);
Domain domain = state.domainGen.next(rs);
TxnId txnId = state.nextTxnId(domain);
Route<?> route = createRoute(state, rs, domain, rs.nextInt(1, 20));
- return new InsertTxn(storeId, txnId, SaveStatus.PreAccepted, Durability.NotDurable, route);
+ int storeId = state.accordService.node().commandStores().select(route.homeKey()).id();
+ return new InsertTxn(storeId, txnId, SaveStatus.PreAccepted, route);
}
private static KeySearch keySearchExisting(RandomSource rs, State state)
@@ -120,38 +174,7 @@ public class RouteIndexTest extends CQLTester.InMemory
var tables = state.storeToTableToRangesToTxns.get(storeId);
TableId tableId = rs.pickUnorderedSet(tables.keySet());
var ranges = tables.get(tableId);
- TreeSet<TokenRange> distinctRanges = ranges.stream().map(Map.Entry::getKey).collect(Collectors.toCollection(() -> new TreeSet<>(TokenRange::compareTo)));
- TokenRange range;
- if (distinctRanges.size() == 1)
- {
- range = Iterables.getFirst(distinctRanges, null);
- }
- else
- {
- switch (rs.nextInt(0, 2))
- {
- case 0: // perfect match
- range = rs.pickOrderedSet(distinctRanges);
- break;
- case 1: // mutli-match
- {
- TokenRange a = rs.pickOrderedSet(distinctRanges);
- TokenRange b = rs.pickOrderedSet(distinctRanges);
- while (a.equals(b))
- b = rs.pickOrderedSet(distinctRanges);
- if (b.compareTo(a) < 0)
- {
- TokenRange tmp = a;
- a = b;
- b = tmp;
- }
- range = TokenRange.create((AccordRoutingKey) a.start(), (AccordRoutingKey) b.end());
- }
- break;
- default:
- throw new AssertionError();
- }
- }
+ TokenRange range = selectExistingRange(rs, ranges);
// have a key, so find a key within the range
long start = range.start().kindOfRoutingKey() == AccordRoutingKey.RoutingKeyKind.SENTINEL ? Long.MIN_VALUE : ((LongToken) range.start().token()).token;
@@ -166,47 +189,15 @@ public class RouteIndexTest extends CQLTester.InMemory
var tables = state.storeToTableToRangesToTxns.get(storeId);
TableId tableId = rs.pickUnorderedSet(tables.keySet());
var ranges = tables.get(tableId);
- TreeSet<TokenRange> distinctRanges = ranges.stream().map(Map.Entry::getKey).collect(Collectors.toCollection(() -> new TreeSet<>(TokenRange::compareTo)));
- TokenRange range;
- if (distinctRanges.size() == 1)
- {
- range = Iterables.getFirst(distinctRanges, null);
- }
- else
- {
- switch (rs.nextInt(0, 2))
- {
- case 0: // perfect match
- range = rs.pickOrderedSet(distinctRanges);
- break;
- case 1: // mutli-match
- {
- TokenRange a = rs.pickOrderedSet(distinctRanges);
- TokenRange b = rs.pickOrderedSet(distinctRanges);
- while (a.equals(b))
- b = rs.pickOrderedSet(distinctRanges);
- if (b.compareTo(a) < 0)
- {
- TokenRange tmp = a;
- a = b;
- b = tmp;
- }
- range = TokenRange.create((AccordRoutingKey) a.start(), (AccordRoutingKey) b.end());
- }
- break;
- default:
- throw new AssertionError();
- }
- }
- return new RangeSearch(storeId, range);
+ return new RangeSearch(storeId, selectExistingRange(rs, ranges));
}
- private static Command<State, ColumnFamilyStore, ?> rangeSearch(RandomSource rs, State state)
+ private static Command<State, Sut, ?> rangeSearch(RandomSource rs, State state)
{
return new RangeSearch(rs.nextInt(0, state.numStores), state.rangeGen.next(rs));
}
- private static Command<State, ColumnFamilyStore, ?> keySearch(RandomSource rs, State state)
+ private static Command<State, Sut, ?> keySearch(RandomSource rs, State state)
{
return new KeySearch(rs.nextInt(0, state.numStores), new TokenKey(rs.pick(state.tables), new LongToken(state.tokenGen.nextInt(rs))));
}
@@ -216,11 +207,14 @@ public class RouteIndexTest extends CQLTester.InMemory
{
cfs().disableAutoCompaction(); // let the test control compaction
//TODO (coverage): include with the ability to mark ranges as durable for compaction cleanup
- AccordService.unsafeSetNoop(); // disable accord service since compaction touches it. It would be nice to include this for cleanup support....
- stateful().withExamples(50).check(commands(() -> State::new, i -> cfs())
- .destroySut(sut -> sut.truncateBlocking())
- .add(FLUSH)
- .add(COMPACT)
+ stateful().withExamples(10).withSteps(500).check(commands(() -> State::new, Sut::new)
+ .destroyState(State::close)
+ .destroySut(Sut::close)
+ .addIf(State::mayFlush, CLOSE)
+ .add(COMPACTOR)
+ .addIf(State::mayCompact, COMPACT)
+ .add(PURGE)
+ .add(RESTART)
.add(this::insert)
.add(RouteIndexTest::rangeSearch)
.add(RouteIndexTest::keySearch)
@@ -229,144 +223,31 @@ public class RouteIndexTest extends CQLTester.InMemory
.build());
}
- private static ColumnFamilyStore cfs()
- {
- return Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME)
- .getColumnFamilyStore(AccordKeyspace.COMMANDS);
- }
-
- private static Gen<TokenRange> rangeGen(RandomSource rand, List<TableId> tables)
+ private static class InsertTxn implements UnitCommand<State, Sut>
{
- Gen.IntGen tokenGen = TOKEN_DISTRIBUTION.next(rand);
- Gen<TableId> tableIdGen = Gens.mixedDistribution(tables).next(rand);
- switch (rand.nextInt(0, 3))
- {
- case 0: // pure random
- return rs -> {
- int a = tokenGen.nextInt(rs);
- int b = tokenGen.nextInt(rs);
- while (a == b)
- b = tokenGen.nextInt(rs);
- if (a > b)
- {
- int tmp = a;
- a = b;
- b = tmp;
- }
- TableId tableId = tableIdGen.next(rs);
- return TokenRange.create(new TokenKey(tableId, new LongToken(a)),
- new TokenKey(tableId, new LongToken(b)));
- };
- case 1: // small range
- Gen.IntGen rangeSizeGen = RANGE_SIZE_DISTRIBUTION.next(rand);
- return rs -> {
- int a = tokenGen.nextInt(rs);
- int rangeSize = rangeSizeGen.nextInt(rs);
- int b = a + rangeSize;
- if (b > MAX_TOKEN)
- {
- b = a;
- a = b - rangeSize;
- }
- TableId tableId = tableIdGen.next(rs);
- return TokenRange.create(new TokenKey(tableId, new LongToken(a)),
- new TokenKey(tableId, new LongToken(b)));
- };
- case 2: // single element
- return rs -> {
- int a = tokenGen.nextInt(rs);
- int b = a + 1;
- TableId tableId = tableIdGen.next(rs);
- return TokenRange.create(new TokenKey(tableId, new LongToken(a)),
- new TokenKey(tableId, new LongToken(b)));
- };
- default:
- throw new AssertionError();
- }
- }
-
- private static Route<?> createRoute(State state, RandomSource rs, Domain domain, int numKeys)
- {
- switch (domain)
- {
- case Key:
- {
- TreeSet<AccordRoutingKey> keys = new TreeSet<>();
- while (keys.size() < numKeys)
- {
- var table = rs.pick(state.tables);
- var token = new LongToken(state.tokenGen.nextInt(rs));
- keys.add(new TokenKey(table, token));
- }
- return new FullKeyRoute(keys.first(), keys.toArray(RoutingKey[]::new));
- }
- case Range:
- {
- TreeSet<TokenRange> set = new TreeSet<>(Range::compareTo);
- while (set.size() < numKeys)
- set.add(state.rangeGen.next(rs));
- return Ranges.ofSorted(set.toArray(Range[]::new)).toRoute(set.first().end());
- }
- default:
- throw new IllegalArgumentException("Unknown domain: " + domain);
- }
- }
-
- private class InsertTxn implements UnitCommand<State, ColumnFamilyStore>
- {
- private static final String cql = "INSERT INTO system_accord.commands (store_id, domain, txn_id, status, participants, durability) VALUES (?, ?, ?, ?, ?, ?)";
private final int storeId;
private final TxnId txnId;
private final SaveStatus saveStatus;
- private final Durability durability;
private final StoreParticipants participants;
- private InsertTxn(int storeId, TxnId txnId, SaveStatus saveStatus, Durability durability, Route<?> route)
+ private InsertTxn(int storeId, TxnId txnId, SaveStatus saveStatus, Route<?> route)
{
this.storeId = storeId;
this.txnId = txnId;
this.saveStatus = saveStatus;
- this.durability = durability;
this.participants = StoreParticipants.all(route);
}
@Override
public void applyUnit(State state)
{
- for (var u : participants.route())
- {
- switch (u.domain())
- {
- case Key:
- {
- AccordRoutingKey key = (AccordRoutingKey) u;
- var table = key.table();
- var token = key.token().getLongValue();
- state.storeToTableToRoutingKeysToTxns.computeIfAbsent(storeId, ignore -> new HashMap<>())
- .computeIfAbsent(table, ignore -> new Long2ObjectHashMap<>())
- .computeIfAbsent(token, ignore -> new ArrayList<>())
- .add(txnId);
- }
- break;
- case Range:
- {
- TokenRange range = (TokenRange) u;
- var table = range.table();
- state.storeToTableToRangesToTxns.computeIfAbsent(storeId, ignore -> new HashMap<>())
- .computeIfAbsent(table, ignore -> rangeTree())
- .add(range, txnId);
- }
- break;
- default:
- throw new AssertionError("Unexpected domain: " + u.domain());
- }
- }
+ state.insertTxn(storeId, txnId, participants.route());
}
@Override
- public void runUnit(ColumnFamilyStore sut) throws Throwable
+ public void runUnit(Sut sut)
{
- execute(cql, storeId, txnId.domain().ordinal(), AccordKeyspace.serializeTimestamp(txnId), saveStatus.ordinal(), AccordKeyspace.serializeParticipants(participants), durability.ordinal());
+ sut.insertTxn(storeId, txnId, saveStatus, participants);
}
@Override
@@ -376,13 +257,12 @@ public class RouteIndexTest extends CQLTester.InMemory
"storeId=" + storeId +
", txnId=" + txnId +
", saveStatus=" + saveStatus +
- ", durability=" + durability +
", participants=" + participants +
'}';
}
}
- private static class KeySearch implements Command<State, ColumnFamilyStore, Set<TxnId>>
+ private static class KeySearch implements Command<State, Sut, Set<TxnId>>
{
private final int storeId;
private final AccordRoutingKey key;
@@ -406,16 +286,16 @@ public class RouteIndexTest extends CQLTester.InMemory
}
@Override
- public Set<TxnId> run(ColumnFamilyStore sut) throws Throwable
+ public Set<TxnId> run(Sut sut) throws Throwable
{
Set<TxnId> result = new ObjectHashSet<>();
- ROUTES_SEARCHER.intersects(storeId, key, TxnId.NONE, Timestamp.MAX, result::add);
+ sut.journal.get().rangeSearcher().search(storeId, key, TxnId.NONE, Timestamp.MAX).consume(result::add);
return result;
}
@Override
public void checkPostconditions(State state, Set<TxnId> expected,
- ColumnFamilyStore sut, Set<TxnId> actual)
+ Sut sut, Set<TxnId> actual)
{
Assertions.assertThat(actual).describedAs("Unexpected txns for key %s", key).isEqualTo(expected);
}
@@ -430,7 +310,7 @@ public class RouteIndexTest extends CQLTester.InMemory
}
}
- private static class RangeSearch implements Command<State, ColumnFamilyStore, Set<TxnId>>
+ private static class RangeSearch implements Command<State, Sut, Set<TxnId>>
{
private final int storeId;
private final TokenRange range;
@@ -454,18 +334,18 @@ public class RouteIndexTest extends CQLTester.InMemory
}
@Override
- public Set<TxnId> run(ColumnFamilyStore sut) throws Throwable
+ public Set<TxnId> run(Sut sut) throws Throwable
{
Set<TxnId> result = new ObjectHashSet<>();
- ROUTES_SEARCHER.intersects(storeId, range, TxnId.NONE, Timestamp.MAX, result::add);
+ sut.journal.get().rangeSearcher().search(storeId, range, TxnId.NONE, Timestamp.MAX).consume(result::add);
return result;
}
@Override
public void checkPostconditions(State state, Set<TxnId> expected,
- ColumnFamilyStore sut, Set<TxnId> actual)
+ Sut sut, Set<TxnId> actual)
{
- Assertions.assertThat(actual).describedAs("Unexpected txns for range %s", range).isEqualTo(expected);
+ Assertions.assertThat(actual).describedAs("Unexpected txns for range %s; missing %s, added %s", range, LazyToString.lazy(() -> Sets.difference(expected, actual).toString()), LazyToString.lazy(() -> Sets.difference(actual, expected).toString())).isEqualTo(expected);
}
@Override
@@ -478,7 +358,7 @@ public class RouteIndexTest extends CQLTester.InMemory
}
}
- private static abstract class CassandraCommand implements UnitCommand<State, ColumnFamilyStore>
+ private static abstract class CassandraCommand implements UnitCommand<State, Sut>
{
private final String name;
@@ -500,33 +380,72 @@ public class RouteIndexTest extends CQLTester.InMemory
}
}
- private static final CassandraCommand FLUSH = new CassandraCommand("Flush")
+ private static final CassandraCommand CLOSE = new CassandraCommand("Close Current Segment")
+ {
+ @Override
+ public void runUnit(Sut sut)
+ {
+ sut.journal.get().closeCurrentSegmentForTestingIfNonEmpty();
+ }
+ };
+
+ private static final CassandraCommand COMPACTOR = new CassandraCommand("Compactor")
{
@Override
- public void runUnit(ColumnFamilyStore sut)
+ public void runUnit(Sut sut)
{
- sut.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
+ sut.journal.get().runCompactorForTesting();
}
};
private static final CassandraCommand COMPACT = new CassandraCommand("Compact")
{
@Override
- public void runUnit(ColumnFamilyStore sut)
+ public void runUnit(Sut sut)
{
try
{
- sut.enableAutoCompaction();
- FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(sut));
+ sut.cfs.enableAutoCompaction();
+ FBUtilities.waitOnFutures(CompactionManager.instance.submitBackground(sut.cfs));
}
finally
{
- sut.disableAutoCompaction();
+ sut.cfs.disableAutoCompaction();
}
}
};
- private static class State
+ private static final CassandraCommand PURGE = new CassandraCommand("Purge")
+ {
+ @Override
+ public void runUnit(Sut sut)
+ {
+ sut.journal.get().purge(sut.stores.get());
+ }
+ };
+
+ private static final UnitCommand<State, Sut> RESTART = new UnitCommand<State, Sut>()
+ {
+ @Override
+ public void applyUnit(State state) throws Throwable
+ {
+ state.restartAccord();
+ }
+
+ @Override
+ public void runUnit(Sut sut)
+ {
+ // no-op
+ }
+
+ @Override
+ public String detailed(State state)
+ {
+ return "Restart Accord";
+ }
+ };
+
+ private static class State implements AutoCloseable
{
private final Int2ObjectHashMap<Map<TableId, Long2ObjectHashMap<List<TxnId>>>> storeToTableToRoutingKeysToTxns = new Int2ObjectHashMap<>();
private final Int2ObjectHashMap<Map<TableId, RangeTree<AccordRoutingKey, TokenRange, TxnId>>> storeToTableToRangesToTxns = new Int2ObjectHashMap<>();
@@ -536,17 +455,40 @@ public class RouteIndexTest extends CQLTester.InMemory
private final Gen.IntGen tokenGen;
private final Gen<TokenRange> rangeGen;
private final Gen<Domain> domainGen;
+ private final ColumnFamilyStore journalTable;
+ // private final AccordJournal journal;
+ private AccordService accordService;
private int hlc = 1000;
public State(RandomSource rs)
{
numStores = NUM_STORES_GEN.nextInt(rs);
- tables = IntStream.range(0, NUM_TABLES_GEN.nextInt(rs))
- .mapToObj(i -> TableId.fromUUID(new UUID(0, i)))
- .collect(Collectors.toList());
+ DatabaseDescriptor.getAccord().command_store_shard_count = new OptionaldPositiveInt(numStores);
+ tables = Collections.singletonList(tableId);
tokenGen = TOKEN_DISTRIBUTION.next(rs);
rangeGen = rangeGen(rs, tables);
domainGen = DOMAIN_DISTRIBUTION.next(rs);
+
+ this.journalTable = Keyspace.open(ACCORD_KEYSPACE_NAME).getColumnFamilyStore(AccordKeyspace.JOURNAL);
+
+ accordService = startAccord();
+ accordService.configurationService().listener.notifyPostCommit(null, ClusterMetadata.current(), false);
+ accordService.epochReady(ClusterMetadata.current().epoch).awaitUninterruptibly();
+ }
+
+ AccordService startAccord()
+ {
+ NodeId tcmNodeId = ClusterMetadata.current().myNodeId();
+ AccordService as = new AccordService(AccordTopology.tcmIdToAccord(tcmNodeId));
+ as.startup();
+
+ // the reason for the mocking is to speed up compaction. Collecting the info from the stores has been slow and its always empty in this test... so stub it out to speed up the test
+ AccordService mock = Mockito.spy(as);
+ Mockito.doReturn(emptyCompactionInfo()).when(mock).getCompactionInfo();
+ AccordService.unsafeSetNewAccordService(mock);
+
+ AccordService.replayJournal(as);
+ return as;
}
TxnId nextTxnId(Domain domain)
@@ -554,6 +496,48 @@ public class RouteIndexTest extends CQLTester.InMemory
return new TxnId(1, hlc++, Txn.Kind.Write, domain, NODE);
}
+ void insertTxn(int storeId, TxnId txnId, Route<?> route)
+ {
+ for (var u : Objects.requireNonNull(route))
+ {
+ switch (u.domain())
+ {
+ case Key:
+ {
+ AccordRoutingKey key = (AccordRoutingKey) u;
+ var table = key.table();
+ var token = key.token().getLongValue();
+ storeToTableToRoutingKeysToTxns.computeIfAbsent(storeId, ignore -> new HashMap<>())
+ .computeIfAbsent(table, ignore -> new Long2ObjectHashMap<>())
+ .computeIfAbsent(token, ignore -> new ArrayList<>())
+ .add(txnId);
+ }
+ break;
+ case Range:
+ {
+ TokenRange range = (TokenRange) u;
+ var table = range.table();
+ storeToTableToRangesToTxns.computeIfAbsent(storeId, ignore -> new HashMap<>())
+ .computeIfAbsent(table, ignore -> rangeTree())
+ .add(range, txnId);
+ }
+ break;
+ default:
+ throw new AssertionError("Unexpected domain: " + u.domain());
+ }
+ }
+ }
+
+ public boolean mayFlush()
+ {
+ return accordService.journal().inMemorySize() > 0;
+ }
+
+ public boolean mayCompact()
+ {
+ return journalTable.getLiveSSTables().size() > 1;
+ }
+
@Override
public String toString()
{
@@ -562,6 +546,54 @@ public class RouteIndexTest extends CQLTester.InMemory
", tables=" + tables +
'}';
}
+
+ @Override
+ public void close()
+ {
+ accordService.shutdown();
+ }
+
+ private void restartAccord()
+ {
+ accordService.shutdown();
+ accordService = startAccord();
+ }
+ }
+
+ public static class Sut implements AutoCloseable
+ {
+ private final ColumnFamilyStore cfs;
+ private final Supplier<CommandStores> stores;
+ private final Supplier<AccordJournal> journal;
+
+ public Sut(State state)
+ {
+ cfs = cfs();
+ this.stores = () -> state.accordService.node().commandStores();
+ this.journal = () -> state.accordService.journal();
+ }
+
+ void insertTxn(int storeId, TxnId txnId, SaveStatus saveStatus, StoreParticipants participants)
+ {
+ Txn txn = toTxn(txnId, participants);
+ AccordGenerators.CommandBuilder builder = new AccordGenerators.CommandBuilder(txnId, txn, txnId, txn.slice(participants.owns().toRanges(), true), PartialDeps.NONE, Ballot.ZERO, Ballot.ZERO, accord.local.Command.WaitingOn.none(txnId.domain(), Deps.NONE));
+ var cmd = builder.build(saveStatus);
+ journal.get().saveCommand(storeId, new Journal.CommandUpdate(null, cmd), () -> {});
+ }
+
+ private static Txn toTxn(TxnId txnId, StoreParticipants participants)
+ {
+ Ranges ranges = Objects.requireNonNull(participants.route()).toRanges();
+ return AccordTestUtils.createTxn(txnId.kind(), ranges);
+ }
+
+ @Override
+ public void close() throws Exception
+ {
+ journal.get().truncateForTesting();
+ cfs.forceBlockingFlush(ColumnFamilyStore.FlushReason.UNIT_TESTS);
+ cfs.truncateBlocking();
+ }
}
private static RangeTree<AccordRoutingKey, TokenRange, TxnId> rangeTree()
@@ -569,30 +601,167 @@ public class RouteIndexTest extends CQLTester.InMemory
return RTree.create(ACCESSOR);
}
- private static final RangeTree.Accessor<AccordRoutingKey, TokenRange> ACCESSOR = new RangeTree.Accessor<AccordRoutingKey, TokenRange>()
+ private static final RangeTree.Accessor<AccordRoutingKey, TokenRange> ACCESSOR = new RangeTree.Accessor<>()
{
@Override
public AccordRoutingKey start(TokenRange tokenRange)
{
- return (AccordRoutingKey) tokenRange.start();
+ return tokenRange.start();
}
@Override
public AccordRoutingKey end(TokenRange tokenRange)
{
- return (AccordRoutingKey) tokenRange.end();
+ return tokenRange.end();
}
@Override
public boolean contains(AccordRoutingKey start, AccordRoutingKey end, AccordRoutingKey accordRoutingKey)
{
- return TokenRange.create(start, end).contains(accordRoutingKey);
+ return new TokenRange(start, end).contains(accordRoutingKey);
}
@Override
public boolean intersects(TokenRange tokenRange, AccordRoutingKey start, AccordRoutingKey end)
{
- return tokenRange.compareIntersecting(TokenRange.create(start, end)) == 0;
+ return tokenRange.compareIntersecting(new TokenRange(start, end)) == 0;
}
};
+
+ private static IAccordService.CompactionInfo emptyCompactionInfo()
+ {
+ Int2ObjectHashMap<RedundantBefore> redundantBefores = new Int2ObjectHashMap<>();
+ Int2ObjectHashMap<DurableBefore> durableBefores = new Int2ObjectHashMap<>();
+ Int2ObjectHashMap<CommandStores.RangesForEpoch> ranges = new Int2ObjectHashMap<>();
+ RedundantBefore redundantBefore = Mockito.spy(RedundantBefore.EMPTY);
+ Mockito.doReturn(RedundantStatus.LIVE).when(redundantBefore).status(Mockito.any(), (Participants<?>) Mockito.any());
+ Mockito.doReturn(RedundantStatus.LIVE).when(redundantBefore).status(Mockito.any(), (RoutingKey) Mockito.any());
+ for (int i = 0; i < MAX_STORES; i++)
+ {
+ redundantBefores.put(i, redundantBefore);
+ durableBefores.put(i, DurableBefore.EMPTY);
+ ranges.put(i, new CommandStores.RangesForEpoch(1, Ranges.EMPTY));
+ }
+ return new IAccordService.CompactionInfo(redundantBefores, ranges, durableBefores);
+ }
+
+ private static ColumnFamilyStore cfs()
+ {
+ return Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME)
+ .getColumnFamilyStore(AccordKeyspace.JOURNAL);
+ }
+
+ private static Gen<TokenRange> rangeGen(RandomSource rand, List<TableId> tables)
+ {
+ Gen.IntGen tokenGen = TOKEN_DISTRIBUTION.next(rand);
+ Gen<TableId> tableIdGen = Gens.mixedDistribution(tables).next(rand);
+ switch (rand.nextInt(0, 3))
+ {
+ case 0: // pure random
+ return rs -> {
+ int a = tokenGen.nextInt(rs);
+ int b = tokenGen.nextInt(rs);
+ while (a == b)
+ b = tokenGen.nextInt(rs);
+ if (a > b)
+ {
+ int tmp = a;
+ a = b;
+ b = tmp;
+ }
+ TableId tableId = tableIdGen.next(rs);
+ return new TokenRange(new TokenKey(tableId, new LongToken(a)),
+ new TokenKey(tableId, new LongToken(b)));
+ };
+ case 1: // small range
+ Gen.IntGen rangeSizeGen = RANGE_SIZE_DISTRIBUTION.next(rand);
+ return rs -> {
+ int a = tokenGen.nextInt(rs);
+ int rangeSize = rangeSizeGen.nextInt(rs);
+ int b = a + rangeSize;
+ if (b > MAX_TOKEN)
+ {
+ b = a;
+ a = b - rangeSize;
+ }
+ TableId tableId = tableIdGen.next(rs);
+ return new TokenRange(new TokenKey(tableId, new LongToken(a)),
+ new TokenKey(tableId, new LongToken(b)));
+ };
+ case 2: // single element
+ return rs -> {
+ int a = tokenGen.nextInt(rs);
+ int b = a + 1;
+ TableId tableId = tableIdGen.next(rs);
+ return new TokenRange(new TokenKey(tableId, new LongToken(a)),
+ new TokenKey(tableId, new LongToken(b)));
+ };
+ default:
+ throw new AssertionError();
+ }
+ }
+
+ private static Route<?> createRoute(State state, RandomSource rs, Domain domain, int numKeys)
+ {
+ switch (domain)
+ {
+ case Key:
+ {
+ TreeSet<AccordRoutingKey> keys = new TreeSet<>();
+ while (keys.size() < numKeys)
+ {
+ var table = rs.pick(state.tables);
+ var token = new LongToken(state.tokenGen.nextInt(rs));
+ keys.add(new TokenKey(table, token));
+ }
+ return new FullKeyRoute(keys.first(), keys.toArray(RoutingKey[]::new));
+ }
+ case Range:
+ {
+ TreeSet<TokenRange> set = new TreeSet<>(Range::compareTo);
+ while (set.size() < numKeys)
+ set.add(state.rangeGen.next(rs));
+ return Ranges.ofSorted(set.toArray(Range[]::new)).toRoute(set.first().end());
+ }
+ default:
+ throw new IllegalArgumentException("Unknown domain: " + domain);
+ }
+ }
+
+ private static TokenRange selectExistingRange(RandomSource rs, RangeTree<AccordRoutingKey, TokenRange, TxnId> ranges)
+ {
+ TreeSet<TokenRange> distinctRanges = ranges.stream().map(Map.Entry::getKey).collect(Collectors.toCollection(() -> new TreeSet<>(TokenRange::compareTo)));
+ TokenRange range;
+ if (distinctRanges.size() == 1)
+ {
+ range = Iterables.getFirst(distinctRanges, null);
+ }
+ else
+ {
+ switch (rs.nextInt(0, 2))
+ {
+ case 0: // perfect match
+ range = rs.pickOrderedSet(distinctRanges);
+ break;
+ case 1: // mutli-match
+ {
+ TokenRange a = rs.pickOrderedSet(distinctRanges);
+ TokenRange b = rs.pickOrderedSet(distinctRanges);
+ while (a.equals(b))
+ b = rs.pickOrderedSet(distinctRanges);
+ if (b.compareTo(a) < 0)
+ {
+ TokenRange tmp = a;
+ a = b;
+ b = tmp;
+ }
+ range = new TokenRange(a.start(), b.end());
+ }
+ break;
+ default:
+ throw new AssertionError();
+ }
+ }
+ return range;
+ }
}
\ No newline at end of file
diff --git a/test/unit/org/apache/cassandra/io/FileSystemListeners.java b/test/unit/org/apache/cassandra/io/FileSystemListeners.java
new file mode 100644
index 0000000000..2e3f784717
--- /dev/null
+++ b/test/unit/org/apache/cassandra/io/FileSystemListeners.java
@@ -0,0 +1,101 @@
+/*
+ * 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.io;
+
+import java.nio.channels.FileChannel;
+import java.nio.file.OpenOption;
+import java.nio.file.Path;
+import java.nio.file.attribute.FileAttribute;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
+import org.apache.cassandra.io.filesystem.ListenableFileSystem;
+
+/**
+ * Listeners to aid in debugging and monitoring of disk issues. The class is normally unused as its target is to help
+ * debug issues, for this reason its likely dead code most of the time... it's here when you need it!
+ */
+@SuppressWarnings("unused")
+public class FileSystemListeners
+{
+ /**
+ * Adds a listener to {@link ListenableFileSystem} that will track open/close/delete operations. If a path delete
+ * is detected before the file is closed, this will throw an exception as this could be due to a possible filesystem leak.
+ * A filesystem leak can be very dangerous as the allocated disk space keeps growing and growing, but when you look
+ * at the file system you won't see all the space as taken; the files were deleted but since the file handles are still
+ * open the kernel will keep the data around.
+ */
+ public static class FileLeak implements ListenableFileSystem.OnPostOpen, ListenableFileSystem.OnPreClose, ListenableFileSystem.OnPreDelete
+ {
+ @Nullable
+ private final IdentityHashMap<FileChannel, Throwable> whereDidChannelsComeFrom;
+ private final Map<Path, Set<FileChannel>> pathToChannels = new HashMap<>();
+
+ public FileLeak(boolean trackOpenSource)
+ {
+ whereDidChannelsComeFrom = trackOpenSource ? new IdentityHashMap<>() : null;
+ }
+
+ public FileLeak()
+ {
+ this(false);
+ }
+
+ @Override
+ public synchronized void postOpen(Path path,
+ Set<? extends OpenOption> options,
+ FileAttribute<?>[] attrs,
+ FileChannel channel)
+ {
+ pathToChannels.computeIfAbsent(path, i -> Collections.newSetFromMap(new IdentityHashMap<>())).add(channel);
+ if (whereDidChannelsComeFrom != null)
+ whereDidChannelsComeFrom.put(channel, new Throwable("here"));
+ }
+
+ @Override
+ public synchronized void preClose(Path path, FileChannel channel)
+ {
+ Set<FileChannel> channels = pathToChannels.get(path);
+ if (channels == null || !channels.remove(channel))
+ return; // listener was added after the open?
+ if (channels.isEmpty())
+ pathToChannels.remove(path);
+ }
+
+ @Override
+ public synchronized void preDelete(Path path)
+ {
+ var cs = pathToChannels.get(path);
+ if (cs == null) return;
+ AssertionError e = new AssertionError("File leak (delete before close) detected on path " + path + "; " + cs.size() + " open handels detected");
+ if (whereDidChannelsComeFrom != null)
+ {
+ var sources = cs.stream().map(whereDidChannelsComeFrom::get).collect(Collectors.toList());
+ sources.forEach(e::addSuppressed);
+
+ }
+ throw e;
+ }
+ }
+}
diff --git a/test/unit/org/apache/cassandra/io/filesystem/ListenableFileSystem.java b/test/unit/org/apache/cassandra/io/filesystem/ListenableFileSystem.java
index 659d34dfe2..83c2f4d225 100644
--- a/test/unit/org/apache/cassandra/io/filesystem/ListenableFileSystem.java
+++ b/test/unit/org/apache/cassandra/io/filesystem/ListenableFileSystem.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.io.filesystem;
import java.io.EOFException;
import java.io.File;
+import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -35,6 +36,7 @@ import java.nio.channels.SeekableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.nio.file.FileSystem;
import java.nio.file.FileSystems;
+import java.nio.file.NoSuchFileException;
import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
@@ -75,6 +77,16 @@ public class ListenableFileSystem extends ForwardingFileSystem
void postOpen(Path path, Set<? extends OpenOption> options, FileAttribute<?>[] attrs, FileChannel channel) throws IOException;
}
+ public interface OnPreClose extends Listener
+ {
+ void preClose(Path path, FileChannel channel) throws IOException;
+ }
+
+ public interface OnPreDelete extends Listener
+ {
+ void preDelete(Path path) throws IOException;
+ }
+
public interface OnPreRead extends Listener
{
void preRead(Path path, FileChannel channel, long position, ByteBuffer dst) throws IOException;
@@ -153,6 +165,8 @@ public class ListenableFileSystem extends ForwardingFileSystem
private final List<OnPreOpen> onPreOpen = new CopyOnWriteArrayList<>();
private final List<OnPostOpen> onPostOpen = new CopyOnWriteArrayList<>();
+ private final List<OnPreClose> onPreClose = new CopyOnWriteArrayList<>();
+ private final List<OnPreDelete> onPreDelete = new CopyOnWriteArrayList<>();
private final List<OnPreTransferTo> onPreTransferTo = new CopyOnWriteArrayList<>();
private final List<OnPostTransferTo> onPostTransferTo = new CopyOnWriteArrayList<>();
private final List<OnPreRead> onPreRead = new CopyOnWriteArrayList<>();
@@ -197,6 +211,16 @@ public class ListenableFileSystem extends ForwardingFileSystem
onPostOpen.add((OnPostOpen) listener);
matches.add(onPostOpen);
}
+ if (listener instanceof OnPreClose)
+ {
+ onPreClose.add((OnPreClose) listener);
+ matches.add(onPreClose);
+ }
+ if (listener instanceof OnPreDelete)
+ {
+ onPreDelete.add((OnPreDelete) listener);
+ matches.add(onPreDelete);
+ }
if (listener instanceof OnPreRead)
{
onPreRead.add((OnPreRead) listener);
@@ -298,6 +322,11 @@ public class ListenableFileSystem extends ForwardingFileSystem
});
}
+ public Unsubscribable onPreClose(OnPreClose callback)
+ {
+ return listen(callback);
+ }
+
public Unsubscribable onPreRead(OnPreRead callback)
{
return listen(callback);
@@ -544,6 +573,13 @@ public class ListenableFileSystem extends ForwardingFileSystem
return ListenableFileSystem.this.unwrap(p);
}
+ @Override
+ public void delete(Path path) throws IOException
+ {
+ notifyListeners(onPreDelete, l -> l.preDelete(path));
+ super.delete(path);
+ }
+
@Override
public OutputStream newOutputStream(Path path, OpenOption... options) throws IOException
{
@@ -796,18 +832,30 @@ public class ListenableFileSystem extends ForwardingFileSystem
long pos = position;
try
{
- while (local.hasRemaining())
+ // the channel could be closed... so always create a new channel to avoid this problem
+ try (FileChannel channel = provider().newFileChannel(path, Set.of(StandardOpenOption.WRITE)))
+ {
+ while (local.hasRemaining())
+ {
+ int wrote = channel.write(local, pos);
+ if (wrote == -1)
+ throw new EOFException();
+ pos += wrote;
+ }
+ }
+ catch (NoSuchFileException | FileNotFoundException e)
{
- int wrote = write(local, pos);
- if (wrote == -1)
- throw new EOFException();
- pos += wrote;
+ // nothing to see here
}
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
+ synchronized (this)
+ {
+ mutable.set(null);
+ }
};
MemoryUtil.setAttachment(bb, forcer);
if (!mutable.compareAndSet(null, mapped))
@@ -823,8 +871,8 @@ public class ListenableFileSystem extends ForwardingFileSystem
@Override
protected void implCloseChannel() throws IOException
{
+ notifyListeners(onPreClose, l -> l.preClose(path, this));
super.implCloseChannel();
- mutable.set(null);
}
}
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
index 505b8e8a0d..547ee77568 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordCommandStoreTest.java
@@ -137,7 +137,6 @@ public class AccordCommandStoreTest
safeCommand.set(expected);
AccordTestUtils.appendCommandsBlocking(commandStore, null, expected);
- AccordKeyspace.getCommandMutation(commandStore, safeCommand, commandStore.nextSystemTimestampMicros()).apply();
logger.info("E: {}", expected);
Command actual = commandStore.loadCommand(txnId);
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
index 87388b877a..12dacb7762 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordKeyspaceTest.java
@@ -53,7 +53,6 @@ import org.apache.cassandra.config.CassandraRelevantProperties;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.db.Keyspace;
-import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.LocalPartitioner;
@@ -120,9 +119,6 @@ public class AccordKeyspaceTest extends CQLTester.InMemory
AccordTestUtils.appendCommandsBlocking(store, null, committed);
- Mutation mutation = AccordKeyspace.getCommandMutation(store, safeCommand, 42);
- mutation.apply();
-
Command loaded = store.loadCommand(id);
Assertions.assertThat(loaded).isEqualTo(committed);
}
diff --git a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
index 8aa15daaed..6317008c86 100644
--- a/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
+++ b/test/unit/org/apache/cassandra/service/accord/AccordTaskTest.java
@@ -67,7 +67,6 @@ import accord.utils.Gens;
import accord.utils.RandomSource;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.cql3.QueryProcessor;
-import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.transform.FilteredPartitions;
@@ -114,7 +113,6 @@ public class AccordTaskTest
@Before
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.COMMANDS_FOR_KEY));
}
@@ -134,9 +132,6 @@ public class AccordTaskTest
Assert.assertNull(instance.ifInitialised(txnId));
Assert.assertNull(instance.ifLoadedAndInitialised(txnId));
}));
-
- UntypedResultSet result = AccordKeyspace.loadCommandRow(commandStore, txnId);
- Assert.assertTrue(result.isEmpty());
}
@Test
@@ -150,9 +145,6 @@ public class AccordTaskTest
SafeCommand command = safe.get(txnId, participants);
Assert.assertNotNull(command);
}));
-
- UntypedResultSet result = AccordKeyspace.loadCommandRow(commandStore, txnId);
- Assert.assertTrue(result.isEmpty());
}
@Test
@@ -182,7 +174,6 @@ public class AccordTaskTest
AccordSafeCommand safeCommand = new AccordSafeCommand(loaded(txnId, null));
safeCommand.set(command);
- AccordKeyspace.getCommandMutation(commandStore, safeCommand, commandStore.nextSystemTimestampMicros()).apply();
appendDiffToLog(commandStore).accept(null, command);
return command;
}
diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
index 17117222ae..43cd4402e7 100644
--- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
+++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
@@ -30,6 +30,7 @@ import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.ToLongFunction;
+import accord.api.Agent;
import accord.api.Journal;
import accord.api.LocalListeners;
import accord.api.ProgressLog;
@@ -62,6 +63,7 @@ import accord.primitives.FullRoute;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.RoutableKey;
+import accord.primitives.Route;
import accord.primitives.RoutingKeys;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@@ -209,7 +211,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
}
};
- this.journal = new InMemoryJournal(nodeId, agent);
+ this.journal = new DefaultJournal(nodeId, agent);
this.commandStore = new AccordCommandStore(0,
storeService,
agent,
@@ -440,4 +442,38 @@ public class SimulatedAccordCommandStore implements AutoCloseable
{
commandStore.shutdown();
}
+
+ private static class DefaultJournal extends InMemoryJournal implements RangeSearcher.Supplier
+ {
+ private final RouteInMemoryIndex<?> index = new RouteInMemoryIndex<>();
+ private DefaultJournal(Node.Id id, Agent agent)
+ {
+ super(id, agent);
+ }
+
+ @Override
+ public void saveCommand(int store, CommandUpdate update, Runnable onFlush)
+ {
+ super.saveCommand(store, update, onFlush);
+ if (!update.after.txnId().domain().isRange())
+ return;
+ Command after = update.after;
+ Route<?> route = after.participants().route();
+ if (route != null)
+ index.update(0, store, after.txnId(), route);
+ }
+
+ @Override
+ public void purge(CommandStores commandStores)
+ {
+ super.purge(commandStores);
+ index.truncateForTesting();
+ }
+
+ @Override
+ public RangeSearcher rangeSearcher()
+ {
+ return index;
+ }
+ }
}
diff --git a/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java b/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java
index 856e9a0780..e443c10d95 100644
--- a/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java
+++ b/test/unit/org/apache/cassandra/utils/CassandraGeneratorsTest.java
@@ -54,7 +54,7 @@ public class CassandraGeneratorsTest
@Test
public void partitionerKeys()
{
- qt().forAll(Gens.random(), toGen(CassandraGenerators.partitioners()))
+ qt().forAll(Gens.random(), toGen(CassandraGenerators.partitioners().map(CassandraGenerators::simplify)))
.check((rs, p) -> Assertions.assertThat(toGen(CassandraGenerators.decoratedKeys(i -> p)).next(rs)).isNotNull());
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org