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

Key transaction recovery should witness range transactions 

patch by Benedict; reviewed by David for CASSANDRA-20105
This commit is contained in:
Benedict Elliott Smith 2024-11-11 17:07:03 +00:00 committed by David Capwell
parent 03cd31fc57
commit f2ea674140
19 changed files with 980 additions and 758 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit ad6d9c748984d64518377510434e59923b7c3183
Subproject commit a271897790aa3816c3dea2125b1e374b091bc090

280
src/java/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndex.java
View File

@ -95,6 +95,26 @@ public class CheckpointIntervalArrayIndex
return (a, b) -> ByteArrayUtil.compareUnsigned(a, 0, b, 0, a.length);
}
@Override
public int compareEndTo(Interval interval, byte[] key)
{
return ByteArrayUtil.compareUnsigned(interval.end, key);
}
@Override
public int compareStartTo(Interval interval, byte[] key)
{
int c = ByteArrayUtil.compareUnsigned(interval.start, key);
if (c == 0) c = -1;
return c;
}
@Override
public boolean endInclusive(Interval[] checksumedRandomAccessReader)
{
return true;
}
@Override
public int binarySearch(Interval[] intervals, int from, int to, byte[] find, AsymmetricComparator<byte[], Interval> comparator, SortedArrays.Search op)
{
@ -520,109 +540,24 @@ public class CheckpointIntervalArrayIndex
this.checkpoints = new CheckpointReader(checkpointFile, checkpointPosition);
}
public Stats intersects(byte[] start, byte[] end, Consumer<Interval> callback) throws IOException
// contains
public Stats contains(byte[] key, Consumer<Interval> callback) throws IOException
{
byte[] keyBuffer = new byte[reader.bytesPerKey];
byte[] recordBuffer = new byte[reader.recordSize - Integer.BYTES];
Stats stats = new Stats();
long startNanos = Clock.Global.nanoTime();
try (ChecksumedRandomAccessReader indexInput = new ChecksumedRandomAccessReader(reader.fh.createReader(), CHECKSUM_SUPPLIER))
{
Interval buffer = new Interval();
Accessor<ChecksumedRandomAccessReader, byte[], byte[]> accessor = new Accessor<>()
{
@Override
public int size(ChecksumedRandomAccessReader indexInput)
{
return reader.count;
}
@Override
public byte[] get(ChecksumedRandomAccessReader indexInput, int index)
{
try
{
return reader.getRecord(indexInput, stats, SortedListReader.SeekReason.GET, recordBuffer, index);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
@Override
public byte[] start(ChecksumedRandomAccessReader indexInput, int index)
{
try
{
return reader.readStart(indexInput, stats, recordBuffer, keyBuffer, index);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
@Override
public byte[] start(byte[] bytes)
{
return reader.copyStart(bytes, keyBuffer);
}
@Override
public byte[] end(ChecksumedRandomAccessReader indexInput, int index)
{
try
{
return reader.readEnd(indexInput, stats, recordBuffer, keyBuffer, index);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
@Override
public byte[] end(byte[] bytes)
{
return reader.copyEnd(bytes, keyBuffer);
}
@Override
public Comparator<byte[]> keyComparator()
{
return (a, b) -> ByteArrayUtil.compareUnsigned(a, 0, b, 0, a.length);
}
@Override
public int binarySearch(ChecksumedRandomAccessReader indexInput, int from, int to, byte[] find, AsymmetricComparator<byte[], byte[]> comparator, SortedArrays.Search op)
{
try
{
return reader.binarySearch(indexInput, stats, recordBuffer, from, to, find, comparator, op);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
};
CheckpointIntervalArray<ChecksumedRandomAccessReader, byte[], byte[]> searcher = new CheckpointIntervalArray<>(accessor, indexInput, checkpoints.bounds, checkpoints.headers, checkpoints.lists, checkpoints.maxScanAndCheckpointMatches);
searcher.forEachRange(start, end, (i1, i2, i3, i4, index) -> {
stats.matches++;
callback.accept(reader.copyTo(accessor.get(indexInput, index), buffer));
return run(ctx -> {
ctx.searcher.forEachKey(key, (i1, i2, i3, i4, index) -> {
ctx.stats.matches++;
callback.accept(reader.copyTo(ctx.accessor.get(ctx.indexInput, index), ctx.buffer));
}, (i1, i2, i3, i4, startIdx, endIdx) -> {
try
{
if (startIdx == endIdx)
return;
reader.maybeSeek(indexInput, stats, SortedListReader.SeekReason.SCAN, reader.fileOffsetStart(startIdx));
reader.maybeSeek(ctx.indexInput, ctx.stats, SortedListReader.SeekReason.SCAN, reader.fileOffsetStart(startIdx));
for (int i = startIdx; i < endIdx; i++)
{
stats.matches++;
reader.getCurrentRecord(indexInput, stats, recordBuffer);
callback.accept(reader.copyTo(recordBuffer, buffer));
ctx.stats.matches++;
reader.getCurrentRecord(ctx.indexInput, ctx.stats, ctx.recordBuffer);
callback.accept(reader.copyTo(ctx.recordBuffer, ctx.buffer));
}
}
catch (IOException e)
@ -630,12 +565,49 @@ public class CheckpointIntervalArrayIndex
throw new UncheckedIOException(e);
}
}, 0, 0, 0, 0, 0);
});
}
public Stats intersects(byte[] start, byte[] end, Consumer<Interval> callback) throws IOException
{
return run(ctx -> {
ctx.searcher.forEachRange(start, end, (i1, i2, i3, i4, index) -> {
ctx.stats.matches++;
callback.accept(reader.copyTo(ctx.accessor.get(ctx.indexInput, index), ctx.buffer));
}, (i1, i2, i3, i4, startIdx, endIdx) -> {
try
{
if (startIdx == endIdx)
return;
reader.maybeSeek(ctx.indexInput, ctx.stats, SortedListReader.SeekReason.SCAN, reader.fileOffsetStart(startIdx));
for (int i = startIdx; i < endIdx; i++)
{
ctx.stats.matches++;
reader.getCurrentRecord(ctx.indexInput, ctx.stats, ctx.recordBuffer);
callback.accept(reader.copyTo(ctx.recordBuffer, ctx.buffer));
}
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}, 0, 0, 0, 0, 0);
});
}
private Stats run(Consumer<Context> fn) throws IOException
{
long startNanos = Clock.Global.nanoTime();
Context ctx = new Context();
try (ctx)
{
fn.accept(ctx);
}
finally
{
stats.durationNs = Clock.Global.nanoTime() - startNanos;
ctx.stats.durationNs = Clock.Global.nanoTime() - startNanos;
}
return stats;
return ctx.stats;
}
@Override
@ -644,6 +616,120 @@ public class CheckpointIntervalArrayIndex
FileUtils.closeQuietly(checkpoints);
FileUtils.closeQuietly(reader);
}
private class Context implements Closeable
{
final byte[] keyBuffer = new byte[reader.bytesPerKey];
final byte[] recordBuffer = new byte[reader.recordSize - Integer.BYTES];
final Stats stats = new Stats();
final ChecksumedRandomAccessReader indexInput = new ChecksumedRandomAccessReader(reader.fh.createReader(), CHECKSUM_SUPPLIER);
final Interval buffer = new Interval();
final Accessor<ChecksumedRandomAccessReader, byte[], byte[]> accessor = new Accessor<>()
{
@Override
public int size(ChecksumedRandomAccessReader indexInput)
{
return reader.count;
}
@Override
public byte[] get(ChecksumedRandomAccessReader indexInput, int index)
{
try
{
return reader.getRecord(indexInput, stats, SortedListReader.SeekReason.GET, recordBuffer, index);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
@Override
public byte[] start(ChecksumedRandomAccessReader indexInput, int index)
{
try
{
return reader.readStart(indexInput, stats, recordBuffer, keyBuffer, index);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
@Override
public byte[] start(byte[] bytes)
{
return reader.copyStart(bytes, keyBuffer);
}
@Override
public byte[] end(ChecksumedRandomAccessReader indexInput, int index)
{
try
{
return reader.readEnd(indexInput, stats, recordBuffer, keyBuffer, index);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
@Override
public byte[] end(byte[] bytes)
{
return reader.copyEnd(bytes, keyBuffer);
}
@Override
public Comparator<byte[]> keyComparator()
{
return (a, b) -> ByteArrayUtil.compareUnsigned(a, 0, b, 0, a.length);
}
@Override
public int compareEndTo(byte[] range, byte[] key)
{
return ByteArrayUtil.compareUnsigned(end(range), key);
}
@Override
public int compareStartTo(byte[] range, byte[] key)
{
int c = ByteArrayUtil.compareUnsigned(start(range), key);
if (c == 0) c = 1;
return c;
}
@Override
public boolean endInclusive(ChecksumedRandomAccessReader checksumedRandomAccessReader)
{
return true;
}
@Override
public int binarySearch(ChecksumedRandomAccessReader indexInput, int from, int to, byte[] find, AsymmetricComparator<byte[], byte[]> comparator, SortedArrays.Search op)
{
try
{
return reader.binarySearch(indexInput, stats, recordBuffer, from, to, find, comparator, op);
}
catch (IOException e)
{
throw new UncheckedIOException(e);
}
}
};
final CheckpointIntervalArray<ChecksumedRandomAccessReader, byte[], byte[]> searcher = new CheckpointIntervalArray<>(accessor, indexInput, checkpoints.bounds, checkpoints.headers, checkpoints.lists, checkpoints.maxScanAndCheckpointMatches);
@Override
public void close() throws IOException
{
indexInput.close();
}
}
}
public static class Interval implements Comparable<Interval>

5
src/java/org/apache/cassandra/index/accord/MemtableIndex.java
View File

@ -67,4 +67,9 @@ public class MemtableIndex
{
return memoryIndex.search(storeId, tableId, start, startInclusive, end, endInclusive);
}
public Collection<ByteBuffer> search(int storeId, TableId tableId, byte[] key)
{
return memoryIndex.search(storeId, tableId, key);
}
}

1
src/java/org/apache/cassandra/index/accord/MemtableIndexManager.java
View File

@ -38,4 +38,5 @@ public interface MemtableIndexManager
void renewMemtable(Memtable renewed);
NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] start, boolean startInclusive, byte[] end, boolean endInclusive);
NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] key);
}

26
src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java
View File

@ -45,9 +45,9 @@ import org.apache.cassandra.db.DecoratedKey;
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.utils.ByteArrayUtil;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FastByteOperations;
import org.apache.cassandra.utils.ObjectSizes;
import org.apache.cassandra.utils.RTree;
import org.apache.cassandra.utils.RangeTree;
@ -83,7 +83,9 @@ public class RangeMemoryIndex
@Override
public boolean contains(byte[] start, byte[] end, byte[] bytes)
{
throw new UnsupportedOperationException();
// bytes are ordered, start is exclusive, end is inclusive
return FastByteOperations.compareUnsigned(start, bytes) < 0
&& FastByteOperations.compareUnsigned(end, bytes) >= 0;
}
@Override
@ -135,8 +137,8 @@ public class RangeMemoryIndex
int storeId = AccordKeyspace.CommandRows.getStoreId(key);
TableId tableId = ts.table();
Group group = new Group(storeId, tableId);
byte[] start = OrderedRouteSerializer.serializeRoutingKeyNoTable((AccordRoutingKey) ts.start());
byte[] end = OrderedRouteSerializer.serializeRoutingKeyNoTable((AccordRoutingKey) ts.end());
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());
@ -159,7 +161,7 @@ public class RangeMemoryIndex
return pks;
}
private TreeMap<Range, Set<DecoratedKey>> search(RangeTree<byte[], Range, DecoratedKey> tokensToPks, byte[] start, byte[] end)
public TreeMap<Range, Set<DecoratedKey>> search(RangeTree<byte[], Range, DecoratedKey> tokensToPks, byte[] start, byte[] end)
{
TreeMap<Range, Set<DecoratedKey>> matches = new TreeMap<>();
@ -167,6 +169,20 @@ public class RangeMemoryIndex
return matches;
}
public synchronized NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] key)
{
RangeTree<byte[], Range, DecoratedKey> rangesToPks = map.get(new Group(storeId, tableId));
if (rangesToPks == null || rangesToPks.isEmpty())
return Collections.emptyNavigableSet();
TreeMap<Range, Set<DecoratedKey>> matches = new TreeMap<>();
rangesToPks.searchToken(key, e -> matches.computeIfAbsent(e.getKey(), ignore -> new HashSet<>()).add(e.getValue()));
TreeSet<ByteBuffer> pks = new TreeSet<>();
matches.values().forEach(s -> s.forEach(d -> pks.add(d.getKey())));
return pks;
}
public synchronized boolean isEmpty()
{
return map.isEmpty();

54
src/java/org/apache/cassandra/index/accord/RouteIndex.java
View File

@ -414,11 +414,13 @@ public class RouteIndex implements Index, INotificationConsumer
}
if (start == null || end == null || storeId == null)
return null;
int finalStoreId = storeId;
ByteBuffer finalStart = start;
boolean finalStartInclusive = startInclusive;
ByteBuffer finalEnd = end;
boolean finalEndInclusive = endInclusive;
if (start.equals(end))
return keySearcher(command, storeId, start);
return rangeSearcher(command, storeId, start, startInclusive, end, endInclusive);
}
private Searcher keySearcher(ReadCommand command, Integer storeId, ByteBuffer key)
{
return new Searcher()
{
@Override
@ -431,14 +433,50 @@ public class RouteIndex implements Index, INotificationConsumer
public UnfilteredPartitionIterator search(ReadExecutionController executionController)
{
// find all partitions from memtable / sstable
NavigableSet<ByteBuffer> partitions = search(finalStoreId, finalStart, finalStartInclusive, finalEnd, finalEndInclusive);
NavigableSet<ByteBuffer> partitions = search(storeId, key);
// do SinglePartitionReadCommand per partition
return new SearchIterator(executionController, command, partitions);
}
NavigableSet<ByteBuffer> search(int storeId, ByteBuffer key)
{
TableId tableId;
byte[] start;
{
AccordRoutingKey route = OrderedRouteSerializer.deserializeRoutingKey(key);
tableId = route.table();
start = OrderedRouteSerializer.serializeRoutingKeyNoTable(route);
}
NavigableSet<ByteBuffer> matches = sstableManager.search(storeId, tableId, start);
matches.addAll(memtableIndexManager.search(storeId, tableId, start));
return matches;
}
};
}
private Searcher rangeSearcher(ReadCommand command, int storeId, ByteBuffer start, boolean startInclusive, ByteBuffer end, boolean endInclusive)
{
return new Searcher()
{
@Override
public ReadCommand command()
{
return command;
}
@Override
public UnfilteredPartitionIterator search(ReadExecutionController executionController)
{
// 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);
}
NavigableSet<ByteBuffer> search(int storeId,
ByteBuffer startTableWithToken, boolean startInclusive,
ByteBuffer endTableWithToken, boolean endInclusive)
ByteBuffer startTableWithToken, boolean startInclusive,
ByteBuffer endTableWithToken, boolean endInclusive)
{
TableId tableId;
byte[] start;

8
src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java
View File

@ -108,4 +108,12 @@ public class RouteMemtableIndexManager implements MemtableIndexManager
liveMemtableIndexMap.values().forEach(m -> matches.addAll(m.search(storeId, tableId, start, startInclusive, end, endInclusive)));
return matches;
}
@Override
public NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] key)
{
TreeSet<ByteBuffer> matches = new TreeSet<>();
liveMemtableIndexMap.values().forEach(m -> matches.addAll(m.search(storeId, tableId, key)));
return matches;
}
}

10
src/java/org/apache/cassandra/index/accord/RouteSSTableManager.java
View File

@ -88,4 +88,14 @@ public class RouteSSTableManager implements SSTableManager
matches.addAll(index.search(group, start, startInclusive, end, endInclusive));
return matches;
}
@Override
public synchronized NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] key)
{
Group group = new Group(storeId, tableId);
TreeSet<ByteBuffer> matches = new TreeSet<>();
for (SSTableIndex index : sstables.values())
matches.addAll(index.search(group, key));
return matches;
}
}

83
src/java/org/apache/cassandra/index/accord/RoutesSearcher.java
View File

@ -19,6 +19,7 @@
package org.apache.cassandra.index.accord;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.function.Consumer;
import accord.primitives.Timestamp;
@ -54,7 +55,7 @@ public class RoutesSearcher
private final DataLimits limits = DataLimits.NONE;
private final DataRange dataRange = DataRange.allData(cfs.getPartitioner());
private CloseableIterator<Entry> searchKeysAccord(int store, AccordRoutingKey start, AccordRoutingKey end)
private CloseableIterator<Entry> searchRange(int store, AccordRoutingKey start, AccordRoutingKey end)
{
RowFilter rowFilter = RowFilter.create(false);
rowFilter.add(participants, Operator.GT, OrderedRouteSerializer.serializeRoutingKey(start));
@ -99,22 +100,80 @@ public class RoutesSearcher
}
}
public void intersects(int store, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
private CloseableIterator<Entry> searchKey(int store, AccordRoutingKey key)
{
intersects(store, range.start(), range.end(), minTxnId, maxTxnId, forEach);
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));
PartitionRangeReadCommand 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;
}
};
}
}
void intersects(int store, AccordRoutingKey start, AccordRoutingKey end, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
public void intersects(int storeId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, Consumer<TxnId> forEach)
{
try (CloseableIterator<Entry> it = searchKeysAccord(store, start, end))
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))
{
while (it.hasNext())
{
Entry next = it.next();
if (next.store_id != store) 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);
}
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);
}
}

34
src/java/org/apache/cassandra/index/accord/SSTableIndex.java
View File

@ -81,6 +81,40 @@ public class SSTableIndex extends SharedCloseableImpl
return new SSTableIndex(id, files, segments, cleanup);
}
public Collection<? extends ByteBuffer> search(Group group, byte[] key)
{
List<Segment> matches = segments.stream().filter(s -> {
Segment.Metadata metadata = s.groups.get(group);
if (metadata == null) return false;
return ByteArrayUtil.compareUnsigned(metadata.minTerm, key) < 0
&& ByteArrayUtil.compareUnsigned(metadata.maxTerm, key) >= 0;
})
.collect(Collectors.toList());
if (matches.isEmpty()) return Collections.emptyList();
if (matches.size() == 1) return search(matches.get(0), group, key);
Set<ByteBuffer> found = new HashSet<>();
for (Segment s : matches)
found.addAll(search(s, group, key));
return found;
}
private Collection<? extends ByteBuffer> search(Segment segment, Group group, byte[] key)
{
Set<ByteBuffer> matches = new HashSet<>();
Segment.Metadata metadata = Objects.requireNonNull(segment.groups.get(group), () -> "Unknown group: " + group);
try
{
SegmentSearcher searcher = new SegmentSearcher(fileFor(IndexComponent.CINTIA_SORTED_LIST), metadata.metas.get(IndexComponent.CINTIA_SORTED_LIST).offset,
fileFor(IndexComponent.CINTIA_CHECKPOINTS), metadata.metas.get(IndexComponent.CINTIA_CHECKPOINTS).offset);
searcher.contains(key, interval -> matches.add(ByteBuffer.wrap(interval.value)));
}
catch (IOException e)
{
throw new FSReadError(e, id.fileFor(IndexComponent.CINTIA_SORTED_LIST));
}
return matches;
}
public Collection<? extends ByteBuffer> search(Group group, byte[] start, boolean startInclusive, byte[] end, boolean endInclusive)
{
List<Segment> matches = segments.stream().filter(s -> {

1
src/java/org/apache/cassandra/index/accord/SSTableManager.java
View File

@ -31,4 +31,5 @@ public interface SSTableManager
boolean isIndexComplete(SSTableReader reader);
NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] start, boolean startInclusive, byte[] end, boolean endInclusive);
NavigableSet<ByteBuffer> search(int storeId, TableId tableId, byte[] key);
}

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

@ -67,6 +67,7 @@ import org.apache.cassandra.utils.NoSpamLogger.NoSpamLogStatement;
import org.apache.cassandra.utils.ObjectSizes;
import static accord.utils.Invariants.checkState;
import static accord.utils.Invariants.illegalState;
import static org.apache.cassandra.net.MessagingService.current_version;
import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.EVICTED;
import static org.apache.cassandra.service.accord.AccordCacheEntry.Status.LOADED;
@ -659,7 +660,7 @@ public class AccordCache implements CacheSize
public void unregister(Listener<K, V> l)
{
if (!tryUnregister(l))
throw new AssertionError("Listener was not registered");
throw illegalState("Listener was not registered");
}
public boolean tryUnregister(Listener<K, V> l)

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

@ -142,7 +142,7 @@ public class AccordCommandStore extends CommandStore
private final Executor taskExecutor;
private final ExclusiveCaches caches;
private long lastSystemTimestampMicros = Long.MIN_VALUE;
private final CommandsForRangesLoader commandsForRangesLoader;
private final CommandsForRanges.Manager commandsForRanges;
private AccordSafeCommandStore current;
private Thread currentThread;
@ -174,7 +174,7 @@ public class AccordCommandStore extends CommandStore
}
this.taskExecutor = executor.executor(this);
this.commandsForRangesLoader = new CommandsForRangesLoader(this);
this.commandsForRanges = new CommandsForRanges.Manager(this);
loadRedundantBefore(journal.loadRedundantBefore(id()));
loadBootstrapBeganAt(journal.loadBootstrapBeganAt(id()));
loadSafeToRead(journal.loadSafeToRead(id()));
@ -187,16 +187,16 @@ public class AccordCommandStore extends CommandStore
new AccordCommandStore(id, node, agent, dataStore, progressLogFactory, listenerFactory, rangesForEpoch, journal, executorFactory.apply(id));
}
public CommandsForRangesLoader diskCommandsForRanges()
public CommandsForRanges.Manager diskCommandsForRanges()
{
return commandsForRangesLoader;
return commandsForRanges;
}
public void markShardDurable(SafeCommandStore safeStore, TxnId globalSyncId, Ranges ranges)
{
store.snapshot(ranges, globalSyncId);
super.markShardDurable(safeStore, globalSyncId, ranges);
commandsForRangesLoader.gcBefore(globalSyncId, ranges);
commandsForRanges.gcBefore(globalSyncId, ranges);
}
@Override

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

@ -45,7 +45,6 @@ import accord.local.RedundantBefore;
import accord.local.SafeCommandStore;
import accord.local.cfk.CommandsForKey;
import accord.primitives.AbstractKeys;
import accord.primitives.AbstractRanges;
import accord.primitives.AbstractUnseekableKeys;
import accord.primitives.Participants;
import accord.primitives.Ranges;
@ -93,18 +92,21 @@ public class AccordSafeCommandStore extends SafeCommandStore
}
@Override
public PreLoadContext canExecute(PreLoadContext context)
public PreLoadContext canExecute(PreLoadContext with)
{
if (context.isEmpty()) return context;
if (context.keys().domain() == Routable.Domain.Range)
return context.isSubsetOf(this.context) ? context : null;
if (with.isEmpty()) return with;
if (with.keys().domain() == Routable.Domain.Range)
return with.isSubsetOf(this.context) ? with : null;
if (!context().keyHistory().satisfies(with.keyHistory()))
return null;
try (ExclusiveCaches caches = commandStore.tryLockCaches())
{
if (caches == null)
return context.isSubsetOf(this.context) ? context : null;
return with.isSubsetOf(this.context) ? with : null;
for (TxnId txnId : context.txnIds())
for (TxnId txnId : with.txnIds())
{
if (null != getInternal(txnId))
continue;
@ -116,14 +118,14 @@ public class AccordSafeCommandStore extends SafeCommandStore
add(safeCommand, caches);
}
KeyHistory keyHistory = context.keyHistory();
KeyHistory keyHistory = with.keyHistory();
if (keyHistory == KeyHistory.NONE)
return context;
return with;
List<RoutingKey> unavailable = null;
AbstractUnseekableKeys keys = (AbstractUnseekableKeys) context.keys();
if (keys.size() == 0)
return context;
return with;
for (int i = 0 ; i < keys.size() ; ++i)
{
@ -158,7 +160,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
}
if (unavailable == null)
return context;
return with;
if (unavailable.size() == keys.size())
return null;
@ -344,6 +346,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
private <O> O mapReduce(Routables<?> keysOrRanges, BiFunction<CommandsSummary, O, O> map, O accumulate)
{
Invariants.checkState(context.keys().containsAll(keysOrRanges), "Attempted to access keysOrRanges outside of what was asked for; asked for %s, accessed %s", context.keys(), keysOrRanges);
accumulate = mapReduceForRange(keysOrRanges, map, accumulate);
return mapReduceForKey(keysOrRanges, map, accumulate);
}
@ -353,25 +356,6 @@ public class AccordSafeCommandStore extends SafeCommandStore
if (commandsForRanges == null)
return accumulate;
switch (keysOrRanges.domain())
{
case Key:
{
AbstractKeys<Key> keys = (AbstractKeys<Key>) keysOrRanges;
if (!commandsForRanges.ranges.intersects(keys))
return accumulate;
}
break;
case Range:
{
AbstractRanges ranges = (AbstractRanges) keysOrRanges;
if (!commandsForRanges.ranges.intersects(ranges))
return accumulate;
}
break;
default:
throw new AssertionError("Unknown domain: " + keysOrRanges.domain());
}
return map.apply(commandsForRanges, accumulate);
}
@ -419,7 +403,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
public <P1, T> T mapReduceActive(Unseekables<?> keysOrRanges, @Nullable Timestamp withLowerTxnId, Txn.Kind.Kinds testKind, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
{
return mapReduce(keysOrRanges, (summary, in) -> {
return summary.mapReduceActive(withLowerTxnId, testKind, map, p1, in);
return summary.mapReduceActive(keysOrRanges, withLowerTxnId, testKind, map, p1, in);
}, accumulate);
}
@ -427,7 +411,7 @@ public class AccordSafeCommandStore extends SafeCommandStore
public <P1, T> T mapReduceFull(Unseekables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
{
return mapReduce(keysOrRanges, (summary, in) -> {
return summary.mapReduceFull(testTxnId, testKind, testStartedAt, testDep, testStatus, map, p1, in);
return summary.mapReduceFull(keysOrRanges, testTxnId, testKind, testStartedAt, testDep, testStatus, map, p1, in);
}, accumulate);
}

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

@ -23,7 +23,6 @@ import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
@ -190,13 +189,15 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
private final PreLoadContext preLoadContext;
private final String loggingId;
// TODO (expected): merge all of these maps into one
@Nullable Object2ObjectHashMap<TxnId, AccordSafeCommand> commands;
@Nullable Object2ObjectHashMap<RoutingKey, AccordSafeTimestampsForKey> timestampsForKey;
@Nullable Object2ObjectHashMap<RoutingKey, AccordSafeCommandsForKey> commandsForKey;
@Nullable Object2ObjectHashMap<Object, AccordSafeState<?, ?>> loading;
// TODO (expected): collection supporting faster deletes but still fast poll (e.g. some ordered collection)
@Nullable ArrayDeque<AccordCacheEntry<?, ?>> waitingToLoad;
@Nullable RangeScanner rangeScanner;
@Nullable RangeTxnScanner rangeScanner;
boolean hasRanges;
@Nullable CommandsForRanges commandsForRanges;
@Nullable private TaskQueue queued;
@ -292,6 +293,7 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
this.state = state;
if (state == WAITING_TO_RUN)
{
Invariants.checkState(rangeScanner == null || rangeScanner.scanned);
Invariants.checkState(loading == null && waitingToLoad == null, "WAITING_TO_RUN => no loading or waiting; found %s", this, AccordTask::toDescription);
loadedAt = nanoTime();
}
@ -385,12 +387,12 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
switch (preLoadContext.keys().domain())
{
case Key: setupKeyLoadsExclusive(caches, (AbstractUnseekableKeys)preLoadContext.keys()); break;
case Key: setupKeyLoadsExclusive(caches, (AbstractUnseekableKeys)preLoadContext.keys(), false); break;
case Range: setupRangeLoadsExclusive(caches);
}
}
private void setupKeyLoadsExclusive(Caches caches, Iterable<? extends RoutingKey> keys)
private void setupKeyLoadsExclusive(Caches caches, Iterable<? extends RoutingKey> keys, boolean isToCompleteRangeScan)
{
switch (preLoadContext.keyHistory())
{
@ -408,8 +410,14 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
}
break;
}
case ASYNC:
case RECOVER:
if (!isToCompleteRangeScan)
{
Invariants.checkState(rangeScanner == null);
rangeScanner = new RangeTxnScanner();
}
case ASYNC:
case INCR:
case SYNC:
{
@ -441,7 +449,8 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
case RECOVER:
case SYNC:
rangeScanner = new RangeScanner(caches.commandsForKeys());
hasRanges = true;
rangeScanner = new RangeTxnAndKeyScanner(caches.commandsForKeys());
}
}
@ -789,14 +798,15 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
state(FAILED);
}
public RangeScanner rangeScanner()
@Nullable
public RangeTxnScanner rangeScanner()
{
return rangeScanner;
}
public boolean hasRanges()
{
return rangeScanner != null;
return hasRanges;
}
@Override
@ -832,8 +842,7 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
// TODO (expected): we should destructively iterate to avoid invoking second time in fail; or else read and set to null
if (rangeScanner != null)
{
caches.commands().tryUnregister(rangeScanner.commandWatcher);
caches.commandsForKeys().tryUnregister(rangeScanner.keyWatcher);
rangeScanner.cleanup(caches);
rangeScanner = null;
}
if (commands != null)
@ -941,7 +950,36 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
commandsForKey.forEach((k, v) -> v.revert());
}
public class RangeScanner implements Runnable
protected void addToQueue(TaskQueue queue)
{
Invariants.checkState(queue.kind == state || (queue.kind == State.WAITING_TO_LOAD && state == WAITING_TO_SCAN_RANGES), "Invalid queue type: %s vs %s", queue.kind, this, AccordTask::toDescription);
Invariants.checkState(this.queued == null, "Already queued with state: %s", this, AccordTask::toDescription);
queued = queue;
queue.append(this);
}
@Nullable
TaskQueue<?> queued()
{
return queued;
}
TaskQueue<?> unqueue()
{
TaskQueue<?> wasQueued = queued;
queued.remove(this);
queued = null;
return wasQueued;
}
TaskQueue<?> unqueueIfQueued()
{
if (queued == null)
return null;
return unqueue();
}
public class RangeTxnAndKeyScanner extends RangeTxnScanner
{
class KeyWatcher implements AccordCache.Listener<RoutingKey, CommandsForKey>
{
@ -953,63 +991,28 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
}
}
class CommandWatcher implements AccordCache.Listener<TxnId, Command>
{
@Override
public void onUpdate(AccordCacheEntry<TxnId, Command> state)
{
CommandsForRangesLoader.Summary summary = summaryLoader.from(state);
if (summary != null)
summaries.put(summary.txnId, summary);
}
}
final ConcurrentHashMap<TxnId, CommandsForRangesLoader.Summary> summaries = new ConcurrentHashMap<>();
// TODO (expected): produce key summaries to avoid locking all in memory
final Set<AccordRoutingKey.TokenKey> intersectingKeys = new ObjectHashSet<>();
final KeyWatcher keyWatcher = new KeyWatcher();
final CommandWatcher commandWatcher = new CommandWatcher();
final Ranges ranges = ((AbstractRanges) preLoadContext.keys()).toRanges();
final AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeyCache;
public RangeScanner(AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeyCache)
public RangeTxnAndKeyScanner(AccordCache.Type<RoutingKey, CommandsForKey, AccordSafeCommandsForKey>.Instance commandsForKeyCache)
{
this.commandsForKeyCache = commandsForKeyCache;
}
CommandsForRangesLoader.Loader summaryLoader;
boolean scanned;
@Override
public void run()
void runInternal()
{
try
for (Range range : ranges)
{
for (Range range : ranges)
{
AccordKeyspace.findAllKeysBetween(commandStore.id(),
(AccordRoutingKey) range.start(), range.startInclusive(),
(AccordRoutingKey) range.end(), range.endInclusive(),
intersectingKeys::add);
}
Collection<TxnId> txnIds = summaryLoader.intersects();
for (TxnId txnId : txnIds)
{
if (summaries.containsKey(txnId))
continue;
CommandsForRangesLoader.Summary summary = summaryLoader.load(txnId);
if (summary != null)
summaries.putIfAbsent(txnId, summary);
}
AccordKeyspace.findAllKeysBetween(commandStore.id(),
(AccordRoutingKey) range.start(), range.startInclusive(),
(AccordRoutingKey) range.end(), range.endInclusive(),
intersectingKeys::add);
}
catch (Throwable t)
{
commandStore.executor().onScannedRanges(AccordTask.this, t);
throw t;
}
commandStore.executor().onScannedRanges(AccordTask.this, null);
super.runInternal();
}
private void reference(AccordCacheEntry<RoutingKey, CommandsForKey> entry)
@ -1041,74 +1044,127 @@ public abstract class AccordTask<R> extends Task implements Runnable, Function<S
}
}
public void start(BiFunction<Task, Runnable, Cancellable> executor)
void startInternal(Caches caches)
{
Caches caches = commandStore.cachesExclusive();
state(SCANNING_RANGES);
for (RoutingKey key : caches.commandsForKeys().keySet())
{
if (ranges.contains(key))
intersectingKeys.add((AccordRoutingKey.TokenKey) key);
}
summaryLoader = commandStore.diskCommandsForRanges().loader(preLoadContext.primaryTxnId(), preLoadContext.keyHistory(), ranges);
summaryLoader.forEachInCache(summary -> summaries.put(summary.txnId, summary), caches);
caches.commandsForKeys().register(keyWatcher);
caches.commands().register(commandWatcher);
// TODO (expected): support cancellation here
super.startInternal(caches);
}
void scannedInternal()
{
if (commandsForKey != null)
intersectingKeys.removeAll(commandsForKey.keySet());
if (loading != null)
intersectingKeys.removeAll(loading.keySet());
setupKeyLoadsExclusive(commandStore.cachesExclusive(), intersectingKeys, true);
super.scannedInternal();
}
void cleanup(Caches caches)
{
caches.commandsForKeys().tryUnregister(keyWatcher);
super.cleanup(caches);
}
CommandsForRanges finish(Caches caches)
{
caches.commandsForKeys().unregister(keyWatcher);
return super.finish(caches);
}
}
public class RangeTxnScanner implements Runnable
{
class CommandWatcher implements AccordCache.Listener<TxnId, Command>
{
@Override
public void onUpdate(AccordCacheEntry<TxnId, Command> state)
{
CommandsForRanges.Summary summary = summaryLoader.from(state);
if (summary != null)
summaries.put(summary.txnId, summary);
}
}
final ConcurrentHashMap<TxnId, CommandsForRanges.Summary> summaries = new ConcurrentHashMap<>();
// TODO (expected): produce key summaries to avoid locking all in memory
final CommandWatcher commandWatcher = new CommandWatcher();
final Unseekables<?> keysOrRanges = preLoadContext.keys();
CommandsForRanges.Loader summaryLoader;
boolean scanned;
@Override
public void run()
{
try
{
runInternal();
}
catch (Throwable t)
{
commandStore.executor().onScannedRanges(AccordTask.this, t);
throw t;
}
commandStore.executor().onScannedRanges(AccordTask.this, null);
}
void runInternal()
{
summaryLoader.intersects(txnId -> {
if (summaries.containsKey(txnId))
return;
CommandsForRanges.Summary summary = summaryLoader.load(txnId);
if (summary != null)
summaries.putIfAbsent(txnId, summary);
});
}
public void start(BiFunction<Task, Runnable, Cancellable> executor)
{
Caches caches = commandStore.cachesExclusive();
state(SCANNING_RANGES);
startInternal(caches);
executor.apply(AccordTask.this, this);
}
void startInternal(Caches caches)
{
summaryLoader = commandStore.diskCommandsForRanges().loader(preLoadContext.primaryTxnId(), preLoadContext.keyHistory(), keysOrRanges);
summaryLoader.forEachInCache(summary -> summaries.put(summary.txnId, summary), caches);
caches.commands().register(commandWatcher);
}
public void scannedExclusive()
{
Invariants.checkState(state == SCANNING_RANGES, "Expected SCANNING_RANGES; found %s", AccordTask.this, AccordTask::toDescription);
scanned = true;
if (commandsForKey != null)
intersectingKeys.removeAll(commandsForKey.keySet());
if (loading != null)
intersectingKeys.removeAll(loading.keySet());
setupKeyLoadsExclusive(commandStore.cachesExclusive(), intersectingKeys);
scannedInternal();
if (loading == null) state(WAITING_TO_RUN);
else if (waitingToLoad == null) state(LOADING);
else state(State.WAITING_TO_LOAD);
}
void scannedInternal()
{
}
void cleanup(Caches caches)
{
caches.commands().tryUnregister(commandWatcher);
}
CommandsForRanges finish(Caches caches)
{
caches.commandsForKeys().unregister(keyWatcher);
caches.commands().unregister(commandWatcher);
return CommandsForRanges.create(ranges, new TreeMap<>(summaries));
return new CommandsForRanges(summaries);
}
}
protected void addToQueue(TaskQueue queue)
{
Invariants.checkState(queue.kind == state || (queue.kind == State.WAITING_TO_LOAD && state == WAITING_TO_SCAN_RANGES), "Invalid queue type: %s vs %s", queue.kind, this, AccordTask::toDescription);
Invariants.checkState(this.queued == null, "Already queued with state: %s", this, AccordTask::toDescription);
queued = queue;
queue.append(this);
}
TaskQueue<?> queued()
{
return queued;
}
TaskQueue<?> unqueue()
{
TaskQueue<?> wasQueued = queued;
queued.remove(this);
queued = null;
return wasQueued;
}
TaskQueue<?> unqueueIfQueued()
{
if (queued == null)
return null;
return unqueue();
}
}

334
src/java/org/apache/cassandra/service/accord/CommandsForRanges.java
View File

@ -19,27 +19,42 @@
package org.apache.cassandra.service.accord;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.impl.CommandsSummary;
import accord.local.Command;
import accord.local.KeyHistory;
import accord.local.RedundantBefore;
import accord.local.SafeCommandStore.CommandFunction;
import accord.local.SafeCommandStore.TestDep;
import accord.local.SafeCommandStore.TestStartedAt;
import accord.local.SafeCommandStore.TestStatus;
import accord.local.StoreParticipants;
import accord.primitives.PartialDeps;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Routable;
import accord.primitives.Routables;
import accord.primitives.SaveStatus;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
import accord.primitives.TxnId;
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.SafeCommandStore.TestDep.ANY_DEPS;
import static accord.local.SafeCommandStore.TestDep.WITH;
@ -48,56 +63,41 @@ import static accord.local.SafeCommandStore.TestStatus.ANY_STATUS;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Status.Stable;
import static accord.primitives.Status.Truncated;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
public class CommandsForRanges implements CommandsSummary
public class CommandsForRanges extends TreeMap<Timestamp, CommandsForRanges.Summary> implements CommandsSummary
{
public final Ranges ranges;
private final NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map;
private CommandsForRanges(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
public CommandsForRanges(Map<? extends Timestamp, ? extends Summary> m)
{
this.ranges = ranges;
this.map = map;
}
public static CommandsForRanges create(Ranges ranges, NavigableMap<Timestamp, CommandsForRangesLoader.Summary> map)
{
return new CommandsForRanges(ranges, map);
}
@VisibleForTesting
public int size()
{
return map.size();
super(m);
}
@Override
public <P1, T> T mapReduceFull(TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
public <P1, T> T mapReduceFull(Routables<?> keysOrRanges, TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
{
return mapReduce(testTxnId, testTxnId, testKind, testStartedAt, testDep, testStatus, map, p1, accumulate);
return mapReduce(keysOrRanges, testTxnId, testTxnId, testKind, testStartedAt, testDep, testStatus, map, p1, accumulate);
}
@Override
public <P1, T> T mapReduceActive(Timestamp startedBefore, Txn.Kind.Kinds testKind, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
public <P1, T> T mapReduceActive(Routables<?> keysOrRanges, Timestamp startedBefore, Txn.Kind.Kinds testKind, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
{
return mapReduce(startedBefore, null, testKind, STARTED_BEFORE, ANY_DEPS, ANY_STATUS, map, p1, accumulate);
return mapReduce(keysOrRanges, startedBefore, null, testKind, STARTED_BEFORE, ANY_DEPS, ANY_STATUS, map, p1, accumulate);
}
private <P1, T> T mapReduce(@Nonnull Timestamp testTimestamp, @Nullable TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
private <P1, T> T mapReduce(Routables<?> keysOrRanges, @Nonnull Timestamp testTimestamp, @Nullable TxnId testTxnId, Txn.Kind.Kinds testKind, TestStartedAt testStartedAt, TestDep testDep, TestStatus testStatus, CommandFunction<P1, T, T> map, P1 p1, T accumulate)
{
// TODO (required): reconsider how we build this, to avoid having to provide range keys in order (or ensure our range search does this for us)
Map<Range, List<CommandsForRangesLoader.Summary>> collect = new TreeMap<>(Range::compare);
NavigableMap<Timestamp, CommandsForRangesLoader.Summary> submap;
Map<Range, List<Summary>> collect = new TreeMap<>(Range::compare);
NavigableMap<Timestamp, Summary> submap;
switch (testStartedAt)
{
case STARTED_AFTER:
submap = this.map.tailMap(testTimestamp, false);
submap = tailMap(testTimestamp, false);
break;
case STARTED_BEFORE:
submap = this.map.headMap(testTimestamp, false);
submap = headMap(testTimestamp, false);
break;
case ANY:
submap = this.map;
submap = this;
break;
default:
throw new AssertionError("Unknown started at: " + testStartedAt);
@ -111,11 +111,11 @@ public class CommandsForRanges implements CommandsSummary
if (summary.saveStatus != null && summary.saveStatus.compareTo(SaveStatus.Erased) >= 0)
return;
// TODO (expected): share this logic with InMemoryCommandsStore for improved BurnTest coverage
switch (testStatus)
{
default: throw new AssertionError("Unhandled TestStatus: " + testStatus);
case ANY_STATUS:
//TODO (now, symitry): how do we map to TRANSITIVELY_KNOWN?
break;
case IS_PROPOSED:
switch (summary.saveStatus.status)
@ -124,7 +124,6 @@ public class CommandsForRanges implements CommandsSummary
case PreCommitted:
case Committed:
case Accepted:
case AcceptedInvalidate:
}
break;
case IS_STABLE:
@ -143,7 +142,6 @@ public class CommandsForRanges implements CommandsSummary
if (executeAt.compareTo(testTxnId) <= 0)
return;
// TODO (required): we must ensure these txnId are limited to those we intersect in this command store
// We are looking for transactions A that have (or have not) B as a dependency.
// If B covers ranges [1..3] and A covers [2..3], but the command store only covers ranges [1..2],
// we could have A adopt B as a dependency on [3..3] only, and have that A intersects B on this
@ -160,33 +158,279 @@ public class CommandsForRanges implements CommandsSummary
Invariants.checkState(testTxnId.equals(summary.findAsDep));
}
// TODO (required): ensure we are excluding any ranges that are now shard-redundant (not sure if this is enforced yet)
for (Range range : summary.ranges)
{
if (!this.ranges.intersects(range))
continue;
collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
if (keysOrRanges.intersects(range))
collect.computeIfAbsent(range, ignore -> new ArrayList<>()).add(summary);
}
}));
for (Map.Entry<Range, List<CommandsForRangesLoader.Summary>> e : collect.entrySet())
for (Map.Entry<Range, List<Summary>> e : collect.entrySet())
{
for (CommandsForRangesLoader.Summary command : e.getValue())
for (Summary command : e.getValue())
accumulate = map.apply(p1, e.getKey(), command.txnId, command.executeAt, accumulate);
}
return accumulate;
}
public CommandsForRanges slice(Ranges slice)
public static class Summary
{
Ranges ranges = this.ranges.slice(slice, Minimal);
NavigableMap<Timestamp, CommandsForRangesLoader.Summary> copy = new TreeMap<>();
for (Map.Entry<Timestamp, CommandsForRangesLoader.Summary> e : map.entrySet())
public final @Nonnull TxnId txnId;
public final @Nonnull Timestamp executeAt;
public final @Nonnull SaveStatus saveStatus;
public final @Nonnull Ranges ranges;
public final TxnId findAsDep;
public final boolean hasAsDep;
@VisibleForTesting
Summary(@Nonnull TxnId txnId, @Nonnull Timestamp executeAt, @Nonnull SaveStatus saveStatus, @Nonnull Ranges ranges, TxnId findAsDep, boolean hasAsDep)
{
if (!e.getValue().ranges.intersects(slice)) continue;
copy.put(e.getKey(), e.getValue().slice(slice));
this.txnId = txnId;
this.executeAt = executeAt;
this.saveStatus = saveStatus;
this.ranges = ranges;
this.findAsDep = findAsDep;
this.hasAsDep = hasAsDep;
}
public Summary slice(Ranges slice)
{
return new Summary(txnId, executeAt, saveStatus, ranges.slice(slice, Minimal), findAsDep, hasAsDep);
}
@Override
public String toString()
{
return "Summary{" +
"txnId=" + txnId +
", executeAt=" + executeAt +
", saveStatus=" + saveStatus +
", ranges=" + ranges +
", findAsDep=" + findAsDep +
", hasAsDep=" + hasAsDep +
'}';
}
}
public static class Manager implements AccordCache.Listener<TxnId, Command>
{
private final AccordCommandStore commandStore;
private final RoutesSearcher searcher = new RoutesSearcher();
private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
public Manager(AccordCommandStore commandStore)
{
this.commandStore = commandStore;
try (AccordCommandStore.ExclusiveCaches caches = commandStore.lockCaches())
{
caches.commands().register(this);
}
}
@Override
public void onAdd(AccordCacheEntry<TxnId, Command> state)
{
TxnId txnId = state.key();
if (txnId.is(Routable.Domain.Range))
cachedRangeTxns.add(txnId);
}
@Override
public void onEvict(AccordCacheEntry<TxnId, Command> state)
{
TxnId txnId = state.key();
if (txnId.is(Routable.Domain.Range))
cachedRangeTxns.remove(txnId);
}
public CommandsForRanges.Loader loader(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Unseekables<?> keysOrRanges)
{
RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
TxnId minTxnId = redundantBefore.min(keysOrRanges, e -> e.gcBefore);
Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVER || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVER ? primaryTxnId : null;
return new CommandsForRanges.Loader(this, keysOrRanges, redundantBefore, minTxnId, maxTxnId, findAsDep);
}
public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
{
transitive.merge(txnId, ranges, remappingFunction);
}
public void gcBefore(TxnId gcBefore, Ranges ranges)
{
Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
while (iterator.hasNext())
{
Map.Entry<TxnId, Ranges> e = iterator.next();
Ranges newRanges = e.getValue().without(ranges);
if (newRanges.isEmpty())
iterator.remove();
e.setValue(newRanges);
}
}
}
public static class Loader
{
private final Manager manager;
final Unseekables<?> searchKeysOrRanges;
final RedundantBefore redundantBefore;
final TxnId minTxnId;
final Timestamp maxTxnId;
@Nullable final TxnId findAsDep;
public Loader(Manager manager, Unseekables<?> searchKeysOrRanges, RedundantBefore redundantBefore, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
{
this.manager = manager;
this.searchKeysOrRanges = searchKeysOrRanges;
this.redundantBefore = redundantBefore;
this.minTxnId = minTxnId;
this.maxTxnId = maxTxnId;
this.findAsDep = findAsDep;
}
public void intersects(Consumer<TxnId> forEach)
{
switch (searchKeysOrRanges.domain())
{
case Range:
for (Unseekable range : searchKeysOrRanges)
manager.searcher.intersects(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, forEach);
break;
case Key:
for (Unseekable key : searchKeysOrRanges)
manager.searcher.intersects(manager.commandStore.id(), (AccordRoutingKey) key, minTxnId, maxTxnId, forEach);
}
if (!manager.transitive.isEmpty())
{
for (Map.Entry<TxnId, Ranges> e : manager.transitive.tailMap(minTxnId, true).entrySet())
{
if (e.getValue().intersects(searchKeysOrRanges))
forEach.accept(e.getKey());
}
}
}
public void forEachInCache(Consumer<Summary> forEach, AccordCommandStore.Caches caches)
{
for (TxnId txnId : manager.cachedRangeTxns)
{
AccordCacheEntry<TxnId, Command> state = caches.commands().getUnsafe(txnId);
Summary summary = from(state);
if (summary != null)
forEach.accept(summary);
}
}
public Summary load(TxnId txnId)
{
if (findAsDep == null)
{
SavedCommand.MinimalCommand cmd = manager.commandStore.loadMinimal(txnId);
if (cmd != null)
return from(cmd);
}
else
{
Command cmd = manager.commandStore.loadCommand(txnId);
if (cmd != null)
return from(cmd);
}
Ranges ranges = manager.transitive.get(txnId);
if (ranges == null)
return null;
ranges = ranges.intersecting(searchKeysOrRanges);
if (ranges.isEmpty())
return null;
return new Summary(txnId, txnId, SaveStatus.NotDefined, ranges, null, false);
}
public Summary from(AccordCacheEntry<TxnId, Command> state)
{
if (state.key().domain() != Routable.Domain.Range)
return null;
switch (state.status())
{
default: throw new AssertionError("Unhandled status: " + state.status());
case LOADING:
case WAITING_TO_LOAD:
case UNINITIALIZED:
return null;
case LOADED:
case MODIFIED:
case SAVING:
case FAILED_TO_SAVE:
}
TxnId txnId = state.key();
if (!txnId.isVisible() || txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) >= 0)
return null;
Command command = state.getExclusive();
if (command == null)
return null;
return from(command);
}
public Summary from(Command cmd)
{
return from(cmd.txnId(), cmd.executeAt(), cmd.saveStatus(), cmd.participants(), cmd.partialDeps());
}
public Summary from(SavedCommand.MinimalCommand cmd)
{
Invariants.checkState(findAsDep == null);
return from(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.participants, null);
}
private Summary from(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, StoreParticipants participants, @Nullable PartialDeps partialDeps)
{
if (participants == null)
return null;
Ranges keysOrRanges = participants.touches().toRanges();
if (keysOrRanges.domain() != Routable.Domain.Range)
throw new AssertionError(String.format("Txn keys are not range for %s", participants));
Ranges ranges = keysOrRanges;
ranges = ranges.intersecting(searchKeysOrRanges, Minimal);
if (ranges.isEmpty())
return null;
if (redundantBefore != null)
{
Ranges newRanges = redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> {
if (e.shardAppliedOrInvalidatedBefore.compareTo(txnId) < 0)
return accum;
return accum.without(Ranges.of(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end)));
}, ranges, ignore -> false);
if (newRanges.isEmpty())
return null;
ranges = newRanges;
}
Invariants.checkState(partialDeps != null || findAsDep == null || !saveStatus.known.deps.hasProposedOrDecidedDeps());
boolean hasAsDep = false;
if (partialDeps != null)
{
Ranges depRanges = partialDeps.rangeDeps.ranges(txnId);
if (depRanges != null && depRanges.containsAll(ranges))
hasAsDep = true;
}
return new Summary(txnId, executeAt, saveStatus, ranges, findAsDep, hasAsDep);
}
return new CommandsForRanges(ranges, copy);
}
}

299
src/java/org/apache/cassandra/service/accord/CommandsForRangesLoader.java
View File

@ -1,299 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.util.Collection;
import java.util.Iterator;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.local.Command;
import accord.local.KeyHistory;
import accord.local.RedundantBefore;
import accord.local.StoreParticipants;
import accord.primitives.PartialDeps;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Routable.Domain;
import accord.primitives.SaveStatus;
import accord.primitives.Status;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import org.agrona.collections.ObjectHashSet;
import org.apache.cassandra.index.accord.RoutesSearcher;
import org.apache.cassandra.service.accord.AccordCommandStore.Caches;
import org.apache.cassandra.service.accord.AccordCommandStore.ExclusiveCaches;
import org.apache.cassandra.service.accord.api.AccordRoutingKey;
import static accord.primitives.Routables.Slice.Minimal;
import static accord.primitives.Txn.Kind.ExclusiveSyncPoint;
public class CommandsForRangesLoader implements AccordCache.Listener<TxnId, Command>
{
private final AccordCommandStore commandStore;
private final RoutesSearcher searcher = new RoutesSearcher();
private final NavigableMap<TxnId, Ranges> transitive = new TreeMap<>();
private final ObjectHashSet<TxnId> cachedRangeTxns = new ObjectHashSet<>();
public CommandsForRangesLoader(AccordCommandStore commandStore)
{
this.commandStore = commandStore;
try (ExclusiveCaches caches = commandStore.lockCaches())
{
caches.commands().register(this);
}
}
@Override
public void onAdd(AccordCacheEntry<TxnId, Command> state)
{
TxnId txnId = state.key();
if (txnId.is(Domain.Range))
cachedRangeTxns.add(txnId);
}
@Override
public void onEvict(AccordCacheEntry<TxnId, Command> state)
{
TxnId txnId = state.key();
if (txnId.is(Domain.Range))
cachedRangeTxns.remove(txnId);
}
public Loader loader(@Nullable TxnId primaryTxnId, KeyHistory keyHistory, Ranges ranges)
{
RedundantBefore redundantBefore = commandStore.unsafeGetRedundantBefore();
TxnId minTxnId = redundantBefore.min(ranges, e -> e.gcBefore);
Timestamp maxTxnId = primaryTxnId == null || keyHistory == KeyHistory.RECOVER || !primaryTxnId.is(ExclusiveSyncPoint) ? Timestamp.MAX : primaryTxnId;
TxnId findAsDep = primaryTxnId != null && keyHistory == KeyHistory.RECOVER ? primaryTxnId : null;
return new Loader(ranges, redundantBefore, minTxnId, maxTxnId, findAsDep);
}
public void mergeTransitive(TxnId txnId, Ranges ranges, BiFunction<? super Ranges, ? super Ranges, ? extends Ranges> remappingFunction)
{
transitive.merge(txnId, ranges, remappingFunction);
}
public void gcBefore(TxnId gcBefore, Ranges ranges)
{
Iterator<Map.Entry<TxnId, Ranges>> iterator = transitive.headMap(gcBefore).entrySet().iterator();
while (iterator.hasNext())
{
Map.Entry<TxnId, Ranges> e = iterator.next();
Ranges newRanges = e.getValue().without(ranges);
if (newRanges.isEmpty())
iterator.remove();
e.setValue(newRanges);
}
}
public static class Summary
{
public final TxnId txnId;
@Nullable public final Timestamp executeAt;
@Nullable public final SaveStatus saveStatus;
@Nullable public final Ranges ranges;
// TODO (required): this logic is still broken (was already): needs to consider exact range matches
public final TxnId findAsDep;
public final boolean hasAsDep;
@VisibleForTesting
Summary(TxnId txnId, @Nullable Timestamp executeAt, SaveStatus saveStatus, Ranges ranges, TxnId findAsDep, boolean hasAsDep)
{
this.txnId = txnId;
this.executeAt = executeAt;
this.saveStatus = saveStatus;
this.ranges = ranges;
this.findAsDep = findAsDep;
this.hasAsDep = hasAsDep;
}
public Summary slice(Ranges slice)
{
return new Summary(txnId, executeAt, saveStatus, ranges == null ? null : ranges.slice(slice, Minimal), findAsDep, hasAsDep);
}
@Override
public String toString()
{
return "Summary{" +
"txnId=" + txnId +
", executeAt=" + executeAt +
", saveStatus=" + saveStatus +
", ranges=" + ranges +
", findAsDep=" + findAsDep +
", hasAsDep=" + hasAsDep +
'}';
}
}
public class Loader
{
final Ranges searchRanges;
final RedundantBefore redundantBefore;
final TxnId minTxnId;
final Timestamp maxTxnId;
@Nullable final TxnId findAsDep;
public Loader(Ranges searchRanges, RedundantBefore redundantBefore, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId findAsDep)
{
this.searchRanges = searchRanges;
this.redundantBefore = redundantBefore;
this.minTxnId = minTxnId;
this.maxTxnId = maxTxnId;
this.findAsDep = findAsDep;
}
public Collection<TxnId> intersects()
{
ObjectHashSet<TxnId> txnIds = new ObjectHashSet<>();
for (Range range : searchRanges)
{
searcher.intersects(commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, txnIds::add);
}
if (!transitive.isEmpty())
{
for (Map.Entry<TxnId, Ranges> e : transitive.tailMap(minTxnId, true).entrySet())
{
if (e.getValue().intersects(searchRanges))
txnIds.add(e.getKey());
}
}
return txnIds;
}
public void forEachInCache(Consumer<Summary> forEach, Caches caches)
{
for (TxnId txnId : cachedRangeTxns)
{
AccordCacheEntry<TxnId, Command> state = caches.commands().getUnsafe(txnId);
Summary summary = from(state);
if (summary != null)
forEach.accept(summary);
}
}
public Summary load(TxnId txnId)
{
if (findAsDep == null)
{
SavedCommand.MinimalCommand cmd = commandStore.loadMinimal(txnId);
return cmd == null ? null : from(cmd);
}
else
{
Command cmd = commandStore.loadCommand(txnId);
return cmd == null ? null : from(cmd);
}
}
public Summary from(AccordCacheEntry<TxnId, Command> state)
{
if (state.key().domain() != Domain.Range)
return null;
switch (state.status())
{
default: throw new AssertionError("Unhandled status: " + state.status());
case LOADING:
case WAITING_TO_LOAD:
case UNINITIALIZED:
return null;
case LOADED:
case MODIFIED:
case SAVING:
case FAILED_TO_SAVE:
}
TxnId txnId = state.key();
if (!txnId.isVisible() || txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) >= 0)
return null;
Command command = state.getExclusive();
if (command == null)
return null;
return from(command);
}
public Summary from(Command cmd)
{
return from(cmd.txnId(), cmd.executeAt(), cmd.saveStatus(), cmd.participants(), cmd.partialDeps());
}
public Summary from(SavedCommand.MinimalCommand cmd)
{
Invariants.checkState(findAsDep == null);
return from(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.participants, null);
}
private Summary from(TxnId txnId, Timestamp executeAt, SaveStatus saveStatus, StoreParticipants participants, @Nullable PartialDeps partialDeps)
{
if (saveStatus == SaveStatus.Invalidated
|| saveStatus == SaveStatus.Erased
|| !saveStatus.hasBeen(Status.PreAccepted))
return null;
if (participants == null)
return null;
Ranges keysOrRanges = participants.touches().toRanges();
if (keysOrRanges.domain() != Domain.Range)
throw new AssertionError(String.format("Txn keys are not range for %s", participants));
Ranges ranges = keysOrRanges;
ranges = ranges.slice(searchRanges, Minimal);
if (ranges.isEmpty())
return null;
if (redundantBefore != null)
{
Ranges newRanges = redundantBefore.foldlWithBounds(ranges, (e, accum, start, end) -> {
if (e.gcBefore.compareTo(txnId) < 0)
return accum;
return accum.without(Ranges.of(new TokenRange((AccordRoutingKey) start, (AccordRoutingKey) end)));
}, ranges, ignore -> false);
if (newRanges.isEmpty())
return null;
ranges = newRanges;
}
Invariants.checkState(partialDeps != null || findAsDep == null || !saveStatus.known.deps.hasProposedOrDecidedDeps());
boolean hasAsDep = false;
if (partialDeps != null)
{
Ranges depRanges = partialDeps.rangeDeps.ranges(txnId);
if (depRanges != null && depRanges.containsAll(ranges))
hasAsDep = true;
}
return new Summary(txnId, executeAt, saveStatus, ranges, findAsDep, hasAsDep);
}
}
}

155
test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java
View File

@ -105,29 +105,62 @@ public class RouteIndexTest extends CQLTester.InMemory
ROUTES_SEARCHER = new RoutesSearcher();
}
@Test
public void test()
private Command<State, ColumnFamilyStore, ?> insert(RandomSource rs, State state)
{
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)
.add((rs, 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);
})
.add((rs, state) -> new RangeSearch(rs.nextInt(0, state.numStores), state.rangeGen.next(rs)))
.addIf(state -> !state.storeToTableToRangesToTxns.isEmpty(), RouteIndexTest::rangeSearch)
.build());
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);
}
private static RangeSearch rangeSearch(RandomSource rs, State state)
private static KeySearch keySearchExisting(RandomSource rs, State state)
{
int storeId = rs.pickUnorderedSet(state.storeToTableToRangesToTxns.keySet());
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 = new TokenRange((AccordRoutingKey) a.start(), (AccordRoutingKey) b.end());
}
break;
default:
throw new AssertionError();
}
}
// 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;
long end = range.end().kindOfRoutingKey() == AccordRoutingKey.RoutingKeyKind.SENTINEL ? Long.MAX_VALUE : ((LongToken) range.end().token()).token;
long token = 1 + rs.nextLong(start, end);
return new KeySearch(storeId, new TokenKey(tableId, new LongToken(token)));
}
private static RangeSearch rangeSearchExisting(RandomSource rs, State state)
{
int storeId = rs.pickUnorderedSet(state.storeToTableToRangesToTxns.keySet());
var tables = state.storeToTableToRangesToTxns.get(storeId);
@ -168,6 +201,34 @@ public class RouteIndexTest extends CQLTester.InMemory
return new RangeSearch(storeId, range);
}
private static Command<State, ColumnFamilyStore, ?> 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)
{
return new KeySearch(rs.nextInt(0, state.numStores), new TokenKey(rs.pick(state.tables), new LongToken(state.tokenGen.nextInt(rs))));
}
@Test
public void test()
{
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)
.add(this::insert)
.add(RouteIndexTest::rangeSearch)
.add(RouteIndexTest::keySearch)
.addIf(state -> !state.storeToTableToRangesToTxns.isEmpty(), RouteIndexTest::rangeSearchExisting)
.addIf(state -> !state.storeToTableToRangesToTxns.isEmpty(), RouteIndexTest::keySearchExisting)
.build());
}
private static ColumnFamilyStore cfs()
{
return Keyspace.open(SchemaConstants.ACCORD_KEYSPACE_NAME)
@ -321,6 +382,54 @@ public class RouteIndexTest extends CQLTester.InMemory
}
}
private static class KeySearch implements Command<State, ColumnFamilyStore, Set<TxnId>>
{
private final int storeId;
private final AccordRoutingKey key;
private KeySearch(int storeId, AccordRoutingKey key)
{
this.storeId = storeId;
this.key = key;
}
@Override
public Set<TxnId> apply(State state) throws Throwable
{
var tables = state.storeToTableToRangesToTxns.get(storeId);
if (tables == null) return Collections.emptySet();
var ranges = tables.get(key.table());
if (ranges == null) return Collections.emptySet();
Set<TxnId> matches = new HashSet<>();
ranges.searchToken(key, e -> matches.add(e.getValue()));
return matches;
}
@Override
public Set<TxnId> run(ColumnFamilyStore sut) throws Throwable
{
Set<TxnId> result = new ObjectHashSet<>();
ROUTES_SEARCHER.intersects(storeId, key, TxnId.NONE, Timestamp.MAX, result::add);
return result;
}
@Override
public void checkPostconditions(State state, Set<TxnId> expected,
ColumnFamilyStore sut, Set<TxnId> actual)
{
Assertions.assertThat(actual).describedAs("Unexpected txns for key %s", key).isEqualTo(expected);
}
@Override
public String toString()
{
return "KeySearch{" +
"storeId=" + storeId +
", key=" + key +
'}';
}
}
private static class RangeSearch implements Command<State, ColumnFamilyStore, Set<TxnId>>
{
private final int storeId;
@ -347,9 +456,9 @@ public class RouteIndexTest extends CQLTester.InMemory
@Override
public Set<TxnId> run(ColumnFamilyStore sut) throws Throwable
{
Set<TxnId> out = new ObjectHashSet<>();
ROUTES_SEARCHER.intersects(storeId, range, TxnId.NONE, Timestamp.MAX, out::add);
return out;
Set<TxnId> result = new ObjectHashSet<>();
ROUTES_SEARCHER.intersects(storeId, range, TxnId.NONE, Timestamp.MAX, result::add);
return result;
}
@Override

131
test/unit/org/apache/cassandra/service/accord/CommandsForRangesTest.java
View File

@ -1,131 +0,0 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.service.accord;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.TreeMap;
import org.junit.Test;
import accord.api.RoutingKey;
import accord.impl.IntKey;
import accord.primitives.SaveStatus;
import accord.primitives.Range;
import accord.primitives.Ranges;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.AccordGens;
import accord.utils.Gen;
import accord.utils.Gens;
import org.apache.cassandra.service.accord.CommandsForRangesLoader.Summary;
import static accord.utils.Property.qt;
import static org.assertj.core.api.Assertions.assertThat;
public class CommandsForRangesTest
{
private static final AccordGens.RangeFactory<IntKey.Routing> ROUTING_RANGE_FACTORY = (i, a, b) -> IntKey.range(a, b);
private static final Gen.IntGen SMALL_INTS = Gens.ints().between(1, 10);
private static final Gen<Ranges> RANGES_GEN = AccordGens.ranges(SMALL_INTS, AccordGens.intRoutingKey(), ROUTING_RANGE_FACTORY);
private static final Gen<TxnId> TXN_ID_GEN = AccordGens.txnIds();
private static final Gen<CommandsForRanges> CFK_GEN = rs -> {
Ranges ranges = RANGES_GEN.next(rs);
int numTxn = 10;
TreeMap<TxnId, Summary> map = new TreeMap<>();
for (int i = 0; i < numTxn; i++)
{
TxnId id = TXN_ID_GEN.next(rs);
map.put(id, new Summary(id, id, SaveStatus.ReadyToExecute, ranges, null, false));
}
return CommandsForRanges.create(ranges, new TreeMap<Timestamp, Summary>(map));
};
private static final IntKey.Routing MIN = IntKey.routing(Integer.MIN_VALUE);
private static final IntKey.Routing MAX = IntKey.routing(Integer.MAX_VALUE);
@Test
public void sliceEmptyWhenOutside()
{
qt().check(rs -> {
CommandsForRanges cfr = CFK_GEN.next(rs);
for (Range range : allOutside(cfr.ranges))
{
Ranges slice = Ranges.single(range);
CommandsForRanges subset = cfr.slice(slice);
assertThat(subset.ranges).isEmpty();
assertThat(subset.size()).isEqualTo(0);
}
});
}
@Test
public void sliceSameNoop()
{
qt().check(rs -> {
CommandsForRanges cfr = CFK_GEN.next(rs);
CommandsForRanges subset = cfr.slice(cfr.ranges);
assertThat(subset.ranges).isEqualTo(cfr.ranges);
assertThat(subset.size()).isEqualTo(cfr.size());
});
}
private static List<Range> allOutside(Ranges ranges)
{
if (ranges.isEmpty()) return Collections.emptyList();
List<Range> matches = new ArrayList<>();
{
Range first = ranges.get(0);
if (!first.start().equals(MIN))
{
int start = Integer.MIN_VALUE;
int end = key(first.start());
matches.add(IntKey.range(start, end));
}
}
if (ranges.size() > 1)
{
{
Range last = ranges.get(ranges.size() - 1);
if (!last.end().equals(MAX))
{
int start = key(last.end());
int end = Integer.MAX_VALUE;
matches.add(IntKey.range(start, end));
}
}
for (int i = 1; i < ranges.size(); i++)
{
Range previous = ranges.get(i - 1);
Range next = ranges.get(i - 1);
int start = key(previous.end());
int end = key(next.start());
if (start < end)
matches.add(IntKey.range(start, end));
}
}
return matches;
}
private static int key(RoutingKey key)
{
return ((IntKey.Routing) key).key;
}
}