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

Accord fixes: 

- Fix erroneous call to registerWithDelay with absolute deadline
 - ExecuteSyncPoint should discount votes from unbootstrapped replicas
 - Bootstrap no longer needs to first ensure durability (now ExecuteSyncPoint discounts votes from unbootstrapped replicas)
 - Progress stall with home shard stopping before Stable is propagated to all shards
 - Only upgrade Durability.HasPhase if we have queried all shards
 - Incorrectly inferring Durability in CheckStatus.finish
 - Incorrectly inferring/propagating stable from fast unstable reply; clarify to prevent further mistakes
 - MaxDecidedRX should track separate hlc bounds for filtering non-RX dependencies
 - minGcBefore.hlc() can move backwards across epochs, even if each shard moves forwards; must track minHlc directly
 - slowTimeout init in ExecuteTxn.LocalExecute
 - TxnId.parse for RV
Also improve:
 - Don't query recovery state if fast path durably decided
 - Support deferred partial dep deserialisation
 - Support both orientations of KeyDeps/RangeDeps
 - Support partialDeps as ByteBuffer
Also improve in Cassandra:
 - Deps serialization and Journal skipping
 - Don't inflate when reading from cache for CommandsForRanges

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20847
This commit is contained in:
Benedict Elliott Smith 2025-08-17 21:11:39 +01:00
parent 9d5cef7f8c
commit fdff319739
44 changed files with 1127 additions and 429 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit a8916a18a012063f4da74aaccd39f0b828c99da0
Subproject commit 5cbe8d62f15cc7d66af2c7686f14b8fa52b1d35b

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

@ -1031,7 +1031,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte
{
row = null;
modified = false;
builder.clear();
builder.reset();
}
}

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

@ -25,6 +25,7 @@ import java.util.function.Consumer;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.db.Clustering;
@ -68,21 +69,21 @@ public class MemtableIndex
}
public void search(int storeId, TableId tableId, byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch)
{
memoryIndex.search(storeId, tableId,
start, end,
minTxnId, maxTxnId, minDecidedId,
minTxnId, maxTxnId, decidedRX,
onMatch);
}
public void search(int storeId, TableId tableId, byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch)
{
memoryIndex.search(storeId, tableId, key,
minTxnId, maxTxnId, minDecidedId,
minTxnId, maxTxnId, decidedRX,
onMatch);
}
}

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

@ -23,6 +23,7 @@ import java.util.function.Consumer;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.db.DecoratedKey;
@ -42,10 +43,10 @@ public interface MemtableIndexManager
void renewMemtable(Memtable renewed);
void search(int storeId, TableId tableId, byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch);
void search(int storeId, TableId tableId, byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch);
}

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

@ -34,6 +34,7 @@ import javax.annotation.concurrent.GuardedBy;
import com.google.common.annotations.VisibleForTesting;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.primitives.Participants;
import accord.primitives.Routable;
import accord.primitives.Timestamp;
@ -86,12 +87,12 @@ public class RangeMemoryIndex
}
void search(byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<Map.Entry<RangeMemoryIndex.Range, DecoratedKey>> fn)
{
if (this.minTxnId.compareTo(maxTxnId) > 0 || this.maxTxnId.compareTo(minTxnId) < 0)
return;
if (maxRXId != null && !RouteIndexFormat.includeByMinDecidedId(minDecidedId, maxRXId))
if (maxRXId != null && !RouteIndexFormat.includeByDecidedRX(decidedRX, maxRXId))
return;
tree.search(new Range(start, end), e -> {
TxnId id = AccordKeyspace.JournalColumns.getJournalKey(e.getValue()).id;
@ -101,12 +102,12 @@ public class RangeMemoryIndex
}
void searchToken(byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<Map.Entry<RangeMemoryIndex.Range, DecoratedKey>> fn)
{
if (this.minTxnId.compareTo(maxTxnId) > 0 || this.maxTxnId.compareTo(minTxnId) < 0)
return;
if (maxRXId != null && !RouteIndexFormat.includeByMinDecidedId(minDecidedId, maxRXId))
if (maxRXId != null && !RouteIndexFormat.includeByDecidedRX(decidedRX, maxRXId))
return;
tree.searchToken(key, e -> {
TxnId id = AccordKeyspace.JournalColumns.getJournalKey(e.getValue()).id;
@ -198,25 +199,25 @@ public class RangeMemoryIndex
public synchronized void search(int storeId, TableId tableId,
byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch)
{
Group group = map.get(new Key(storeId, tableId));
if (group == null) return;
if (group.tree.isEmpty()) return;
group.search(start, end, minTxnId, maxTxnId, minDecidedId, e -> onMatch.accept(e.getValue().getKey()));
group.search(start, end, minTxnId, maxTxnId, decidedRX, e -> onMatch.accept(e.getValue().getKey()));
}
public synchronized void search(int storeId, TableId tableId, byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch)
{
Group group = map.get(new Key(storeId, tableId));
if (group == null) return;
if (group.tree.isEmpty()) return;
group.searchToken(key, minTxnId, maxTxnId, minDecidedId, e -> onMatch.accept(e.getValue().getKey()));
group.searchToken(key, minTxnId, maxTxnId, decidedRX, e -> onMatch.accept(e.getValue().getKey()));
}
public synchronized boolean isEmpty()

7
src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java
View File

@ -35,6 +35,7 @@ import javax.annotation.Nullable;
import com.google.common.collect.Maps;
import accord.local.MaxDecidedRX;
import accord.local.StoreParticipants;
import accord.primitives.Participants;
import accord.primitives.Txn;
@ -92,10 +93,10 @@ public class RouteIndexFormat
return touches.deserialize(bytes);
}
public static boolean includeByMinDecidedId(@Nullable TxnId minDecidedId, TxnId txnId)
public static boolean includeByDecidedRX(@Nullable MaxDecidedRX.DecidedRX decidedRX, TxnId txnId)
{
if (minDecidedId == null || txnId.equals(TxnId.NONE) || !txnId.is(Txn.Kind.ExclusiveSyncPoint)) return true;
return txnId.compareTo(minDecidedId) >= 0;
if (decidedRX == null || txnId.equals(TxnId.NONE) || !txnId.is(Txn.Kind.ExclusiveSyncPoint)) return true;
return decidedRX.includeDecided(txnId);
}
public interface Writer extends SSTableFlushObserver

30
src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java
View File

@ -38,6 +38,7 @@ import com.google.common.base.Splitter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.config.DatabaseDescriptor;
@ -390,7 +391,6 @@ public class RouteJournalIndex implements Index, INotificationConsumer
Integer storeId = null;
TxnId minTxnId = TxnId.NONE;
Timestamp maxTxnId = TxnId.MAX;
@Nullable TxnId minDecidedId = null;
for (RowFilter.Expression e : expressions)
{
if (e.column() == AccordJournalTable.SyntheticColumn.participants.metadata)
@ -429,10 +429,6 @@ public class RouteJournalIndex implements Index, INotificationConsumer
return null;
}
}
else if (e.column() == AccordJournalTable.SyntheticColumn.min_decided_id.metadata)
{
minDecidedId = CommandSerializers.txnId.deserialize(e.getIndexValue());
}
else
{
String cqlString;
@ -450,12 +446,12 @@ public class RouteJournalIndex implements Index, INotificationConsumer
if (start == null || end == null || storeId == null)
return null;
if (start.equals(end))
return keySearcher(command, storeId, start, minTxnId, maxTxnId, minDecidedId);
return rangeSearcher(command, storeId, start, end, minTxnId, maxTxnId, minDecidedId);
return keySearcher(command, storeId, start, minTxnId, maxTxnId, null);
return rangeSearcher(command, storeId, start, end, minTxnId, maxTxnId, null);
}
private Searcher keySearcher(ReadCommand command, Integer storeId, ByteBuffer key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX)
{
return new Searcher()
{
@ -470,13 +466,13 @@ public class RouteJournalIndex implements Index, INotificationConsumer
{
// find all partitions from memtable / sstable
NavigableSet<ByteBuffer> partitions = search(storeId, key,
minTxnId, maxTxnId, minDecidedId);
minTxnId, maxTxnId, decidedRX);
// do SinglePartitionReadCommand per partition
return new SearchIterator(command, partitions);
}
NavigableSet<ByteBuffer> search(int storeId, ByteBuffer key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX)
{
TableId tableId;
byte[] start;
@ -487,9 +483,9 @@ public class RouteJournalIndex implements Index, INotificationConsumer
}
// store matches in a hash set so add is O(1), and the sorting is done after collecting all matches
Set<ByteBuffer> matches = new HashSet<>();
sstableManager.search(storeId, tableId, start, minTxnId, maxTxnId, minDecidedId, matches::add);
sstableManager.search(storeId, tableId, start, minTxnId, maxTxnId, decidedRX, matches::add);
memtableIndexManager.search(storeId, tableId, start,
minTxnId, maxTxnId, minDecidedId,
minTxnId, maxTxnId, decidedRX,
matches::add);
return new TreeSet<>(matches);
}
@ -498,7 +494,7 @@ public class RouteJournalIndex implements Index, INotificationConsumer
private Searcher rangeSearcher(ReadCommand command, int storeId,
ByteBuffer start, ByteBuffer end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX)
{
return new Searcher()
{
@ -514,14 +510,14 @@ public class RouteJournalIndex implements Index, INotificationConsumer
// find all partitions from memtable / sstable
NavigableSet<ByteBuffer> partitions = search(storeId,
start, end,
minTxnId, maxTxnId, minDecidedId);
minTxnId, maxTxnId, decidedRX);
// do SinglePartitionReadCommand per partition
return new SearchIterator(command, partitions);
}
NavigableSet<ByteBuffer> search(int storeId,
ByteBuffer startTableWithToken, ByteBuffer endTableWithToken,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX)
{
TableId tableId;
byte[] start;
@ -534,8 +530,8 @@ public class RouteJournalIndex implements Index, INotificationConsumer
byte[] end = OrderedRouteSerializer.serializeTokenOnly(OrderedRouteSerializer.deserialize(endTableWithToken));
// store matches in a hash set so add is O(1), and the sorting is done after collecting all matches
Set<ByteBuffer> matches = new HashSet<>();
sstableManager.search(storeId, tableId, start, end, minTxnId, maxTxnId, minDecidedId, matches::add);
memtableIndexManager.search(storeId, tableId, start, end, minTxnId, maxTxnId, minDecidedId, matches::add);
sstableManager.search(storeId, tableId, start, end, minTxnId, maxTxnId, decidedRX, matches::add);
memtableIndexManager.search(storeId, tableId, start, end, minTxnId, maxTxnId, decidedRX, matches::add);
return new TreeSet<>(matches);
}
};

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

@ -26,6 +26,7 @@ import java.util.function.Consumer;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.db.DecoratedKey;
@ -106,22 +107,22 @@ public class RouteMemtableIndexManager implements MemtableIndexManager
@Override
public void search(int storeId, TableId tableId, byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch)
{
liveMemtableIndexMap.values().forEach(m -> m.search(storeId, tableId,
start, end,
minTxnId, maxTxnId, minDecidedId,
minTxnId, maxTxnId, decidedRX,
onMatch));
}
@Override
public void search(int storeId, TableId tableId, byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch)
{
liveMemtableIndexMap.values().forEach(m -> m.search(storeId, tableId, key,
minTxnId, maxTxnId, minDecidedId,
minTxnId, maxTxnId, decidedRX,
onMatch));
}
}

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

@ -29,6 +29,7 @@ import java.util.function.Consumer;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.io.FSReadError;
@ -72,6 +73,7 @@ public class RouteSSTableManager implements SSTableManager
}
catch (IOException e)
{
if (notComplete == null) notComplete = new ArrayList<>();
notComplete.add(sstable);
}
}
@ -88,7 +90,7 @@ public class RouteSSTableManager implements SSTableManager
@Override
public synchronized void search(int storeId, TableId tableId,
byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX,
Consumer<ByteBuffer> onMatch)
{
Key group = new Key(storeId, tableId);
@ -96,7 +98,7 @@ public class RouteSSTableManager implements SSTableManager
{
try
{
index.search(group, start, end, minTxnId, maxTxnId, minDecidedId, onMatch);
index.search(group, start, end, minTxnId, maxTxnId, decidedRX, onMatch);
}
catch (Throwable t)
{
@ -107,10 +109,10 @@ public class RouteSSTableManager implements SSTableManager
}
@Override
public synchronized void search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, Consumer<ByteBuffer> onMatch)
public synchronized void search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, Consumer<ByteBuffer> onMatch)
{
Key group = new Key(storeId, tableId);
for (SSTableIndex index : sstables.values())
index.search(group, key, minTxnId, maxTxnId, minDecidedId, onMatch);
index.search(group, key, minTxnId, maxTxnId, decidedRX, onMatch);
}
}

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

@ -29,6 +29,7 @@ import java.util.stream.Collectors;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.index.accord.CheckpointIntervalArrayIndex.SegmentSearcher;
@ -82,14 +83,14 @@ public class SSTableIndex extends SharedCloseableImpl
return new SSTableIndex(id, files, segments, cleanup);
}
public void search(Key group, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, Consumer<ByteBuffer> onMatch)
public void search(Key group, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX, Consumer<ByteBuffer> onMatch)
{
List<Segment> matches = segments.stream().filter(s -> {
Segment.Metadata metadata = s.groups.get(group);
if (metadata == null) return false;
if (metadata.maxTxnId.compareTo(minTxnId) < 0 || metadata.minTxnId.compareTo(maxTxnId) > 0)
return false;
if (!RouteIndexFormat.includeByMinDecidedId(minDecidedId, metadata.maxRxId))
if (!RouteIndexFormat.includeByDecidedRX(decidedRX, metadata.maxRxId))
return false;
return ByteArrayUtil.compareUnsigned(metadata.minTerm, key) < 0
&& ByteArrayUtil.compareUnsigned(metadata.maxTerm, key) >= 0;
@ -115,14 +116,14 @@ public class SSTableIndex extends SharedCloseableImpl
}
}
public void search(Key group, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, Consumer<ByteBuffer> onMatch)
public void search(Key group, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX, Consumer<ByteBuffer> onMatch)
{
List<Segment> matches = segments.stream().filter(s -> {
Segment.Metadata metadata = s.groups.get(group);
if (metadata == null) return false;
if (metadata.maxTxnId.compareTo(minTxnId) < 0 || metadata.minTxnId.compareTo(maxTxnId) > 0)
return false;
if (!RouteIndexFormat.includeByMinDecidedId(minDecidedId, metadata.maxRxId))
if (!RouteIndexFormat.includeByDecidedRX(decidedRX, metadata.maxRxId))
return false;
if (ByteArrayUtil.compareUnsigned(metadata.minTerm, end) >= 0)
return false;

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

@ -24,6 +24,7 @@ import java.util.function.Consumer;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.io.sstable.format.SSTableReader;
@ -34,6 +35,6 @@ public interface SSTableManager
void onSSTableChanged(Collection<SSTableReader> removed, Iterable<SSTableReader> added);
boolean isIndexComplete(SSTableReader reader);
void search(int storeId, TableId tableId, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, Consumer<ByteBuffer> onMatch);
void search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, Consumer<ByteBuffer> onMatch);
void search(int storeId, TableId tableId, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, Consumer<ByteBuffer> onMatch);
void search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, Consumer<ByteBuffer> onMatch);
}

5
src/java/org/apache/cassandra/io/AsymmetricParameterisedVersionedSerializer.java
View File

@ -57,6 +57,11 @@ public interface AsymmetricParameterisedVersionedSerializer<In, P, Out, Version>
Out deserialize(P p, DataInputPlus in, Version version) throws IOException;
default void skip(P p, DataInputPlus in, Version version) throws IOException
{
deserialize(p, in, version);
}
default Out deserialize(P p, ByteBuffer buffer, Version version) throws IOException
{
try (DataInputBuffer in = new DataInputBuffer(buffer, true))

6
src/java/org/apache/cassandra/io/AsymmetricVersionedSerializer.java
View File

@ -54,6 +54,12 @@ public interface AsymmetricVersionedSerializer<In, Out, Version>
}
}
Out deserialize(DataInputPlus in, Version version) throws IOException;
default void skip(DataInputPlus in, Version version) throws IOException
{
deserialize(in, version);
}
default Out deserialize(ByteBuffer buffer, Version version) throws IOException
{
try (DataInputBuffer in = new DataInputBuffer(buffer, true))

5
src/java/org/apache/cassandra/io/util/DataInputBuffer.java
View File

@ -67,6 +67,11 @@ public class DataInputBuffer extends RebufferingInputStream
return buffer.remaining();
}
public ByteBuffer buffer()
{
return buffer;
}
@Override
public void close() {}
}

7
src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java
View File

@ -383,6 +383,13 @@ public class AccordCacheEntry<K, V> extends IntrusiveLinkedListNode
return (V)unwrap();
}
public Object getOrShrunkExclusive()
{
Invariants.require(owner == null || owner.commandStore == null || owner.commandStore.executor().isOwningThread());
Invariants.require(isLoaded(), "%s", this);
return unwrap();
}
public V tryGetExclusive()
{
Invariants.require(owner == null || owner.commandStore == null || owner.commandStore.executor().isOwningThread());

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

@ -79,7 +79,6 @@ import org.apache.cassandra.utils.Clock;
import static accord.api.Journal.CommandUpdate;
import static accord.api.Journal.FieldUpdates;
import static accord.api.Journal.Load.MINIMAL;
import static accord.utils.Invariants.require;
import static org.apache.cassandra.journal.Params.ReplayMode.ONLY_NON_DURABLE;
@ -435,7 +434,12 @@ public class AccordCommandStore extends CommandStore
public Command.Minimal loadMinimal(TxnId txnId)
{
return journal.loadMinimal(id, txnId, MINIMAL, unsafeGetRedundantBefore(), durableBefore());
return journal.loadMinimal(id, txnId, unsafeGetRedundantBefore(), durableBefore());
}
public Command.MinimalWithDeps loadMinimalWithDeps(TxnId txnId)
{
return journal.loadMinimalWithDeps(id, txnId, unsafeGetRedundantBefore(), durableBefore());
}
public AccordCompactionInfo getCompactionInfo()

90
src/java/org/apache/cassandra/service/accord/AccordJournal.java
View File

@ -50,6 +50,8 @@ import accord.local.DurableBefore;
import accord.local.Node;
import accord.local.RedundantBefore;
import accord.primitives.EpochSupplier;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Ranges;
import accord.primitives.Route;
import accord.primitives.SaveStatus;
@ -67,6 +69,7 @@ import org.apache.cassandra.concurrent.Shutdownable;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -94,6 +97,9 @@ import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.ExecutorUtils;
import org.apache.cassandra.utils.concurrent.Semaphore;
import static accord.api.Journal.Load.ALL;
import static accord.api.Journal.Load.MINIMAL;
import static accord.api.Journal.Load.MINIMAL_WITH_DEPS;
import static accord.impl.CommandChange.Field.CLEANUP;
import static accord.impl.CommandChange.anyFieldChanged;
import static accord.impl.CommandChange.describeFlags;
@ -278,10 +284,9 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
return result;
}
@Override
public Command.Minimal loadMinimal(int commandStoreId, TxnId txnId, Load load, RedundantBefore redundantBefore, DurableBefore durableBefore)
// applies cleanup and returns null if no command should be returned
public static Builder cleanupAndFilter(Builder builder, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
Builder builder = loadDiffs(commandStoreId, txnId, load);
if (builder.isEmpty())
return null;
@ -294,7 +299,21 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
return null;
}
Invariants.require(builder.saveStatus() != null, "No saveSatus loaded, but next was called and cleanup was not: %s", builder);
return builder.asMinimal();
return builder;
}
@Override
public Command.Minimal loadMinimal(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
Builder builder = cleanupAndFilter(loadDiffs(commandStoreId, txnId, MINIMAL), redundantBefore, durableBefore);
return builder == null ? null : builder.asMinimal();
}
@Override
public Command.MinimalWithDeps loadMinimalWithDeps(int commandStoreId, TxnId txnId, RedundantBefore redundantBefore, DurableBefore durableBefore)
{
Builder builder = cleanupAndFilter(loadDiffs(commandStoreId, txnId, MINIMAL_WITH_DEPS), redundantBefore, durableBefore);
return builder == null ? null : builder.asMinimalWithDeps();
}
@Override
@ -831,7 +850,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
CommandSerializers.partialTxn.serialize(command.partialTxn(), out, userVersion);
break;
case PARTIAL_DEPS:
DepsSerializers.partialDeps.serialize(command.partialDeps(), out);
DepsSerializers.partialDepsById.serialize(command.partialDeps(), out);
break;
case WAITING_ON:
Command.WaitingOn waitingOn = command.waitingOn();
@ -878,6 +897,8 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
public static class Builder extends CommandChange.Builder implements FlyweightImage
{
private final boolean deserializeDeps;
public Builder()
{
this(Load.ALL);
@ -885,17 +906,35 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
public Builder(Load load)
{
super(null, load);
this(null, load);
}
public Builder(TxnId txnId)
{
super(txnId, Load.ALL);
this(txnId, Load.ALL);
}
public Builder(TxnId txnId, Load load)
{
super(txnId, load);
deserializeDeps = load == ALL;
}
@Override
public PartialDeps partialDeps()
{
if (partialDeps instanceof ByteBuffer)
{
try
{
partialDeps = DepsSerializers.partialDepsById.deserialize((ByteBuffer) partialDeps);
}
catch (IOException e)
{
throw new IllegalStateException("Failed to materialise partially deserialised deps", e);
}
}
return (PartialDeps) partialDeps;
}
public void reset(JournalKey key)
@ -970,15 +1009,17 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
break;
case PARTIAL_TXN:
Invariants.require(partialTxn != null, "%s", this);
CommandSerializers.partialTxn.serialize(partialTxn, out, userVersion);
if (partialTxn instanceof ByteBuffer) out.write(((ByteBuffer) partialTxn).duplicate());
else CommandSerializers.partialTxn.serialize((PartialTxn) partialTxn, out, userVersion);
break;
case PARTIAL_DEPS:
Invariants.require(partialDeps != null, "%s", this);
DepsSerializers.partialDeps.serialize(partialDeps, out);
if (partialDeps instanceof ByteBuffer) out.write(((ByteBuffer) partialDeps).duplicate());
else DepsSerializers.partialDepsById.serialize((PartialDeps) partialDeps, out);
break;
case WAITING_ON:
Invariants.require(waitingOn != null, "%s", this);
((WaitingOnSerializer.Provider)waitingOn).reserialize(out);
((WaitingOnSerializer.WaitingOnBitSetsAndLength)waitingOn).reserialize(out);
break;
case WRITES:
Invariants.require(writes != null, "%s", this);
@ -1050,10 +1091,24 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
partialTxn = CommandSerializers.partialTxn.deserialize(in, userVersion);
break;
case PARTIAL_DEPS:
partialDeps = DepsSerializers.partialDeps.deserialize(in);
// TODO (required): this optimisation will be easily disabled;
// should either operate natively on ByteBuffer
// or else use some explicit API for copying bytes while skipping
if (deserializeDeps || !(in instanceof DataInputBuffer))
{
partialDeps = DepsSerializers.partialDepsById.deserialize(in);
}
else
{
ByteBuffer buf = ((DataInputBuffer)in).buffer();
int start = buf.position();
DepsSerializers.partialDepsById.skip(in);
int end = buf.position();
partialDeps = buf.duplicate().position(start).limit(end);
}
break;
case WAITING_ON:
waitingOn = WaitingOnSerializer.deserializeProvider(txnId, in);
waitingOn = WaitingOnSerializer.deserializeBitSets(txnId, in);
break;
case WRITES:
writes = CommandSerializers.writes.deserialize(in, userVersion);
@ -1093,25 +1148,24 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
CommandSerializers.ballot.skip(in);
break;
case PARTICIPANTS:
CommandSerializers.participants.deserialize(in);
CommandSerializers.participants.skip(in);
break;
case PARTIAL_TXN:
CommandSerializers.partialTxn.deserialize(in, userVersion);
CommandSerializers.partialTxn.skip(in, userVersion);
break;
case PARTIAL_DEPS:
// TODO (expected): skip
DepsSerializers.partialDeps.deserialize(in);
DepsSerializers.partialDepsById.skip(in);
break;
case WAITING_ON:
WaitingOnSerializer.skip(txnId, in);
break;
case WRITES:
// TODO (expected): skip
CommandSerializers.writes.deserialize(in, userVersion);
CommandSerializers.writes.skip(in, userVersion);
break;
case RESULT:
// TODO (expected): skip
ResultSerializers.result.deserialize(in);
ResultSerializers.result.skip(in);
break;
}
}

30
src/java/org/apache/cassandra/service/accord/AccordJournalTable.java
View File

@ -32,6 +32,7 @@ import com.google.common.collect.AbstractIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.MaxDecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import accord.utils.Invariants;
@ -230,8 +231,7 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
{
participants("participants", BytesType.instance),
store_id("store_id", Int32Type.instance),
txn_id("txn_id", BytesType.instance),
min_decided_id("min_decided_id", BytesType.instance);
txn_id("txn_id", BytesType.instance);
public final ColumnMetadata metadata;
@ -253,22 +253,22 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
}
@Override
public Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
public Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX)
{
CloseableIterator<TxnId> inMemory = index.search(commandStoreId, range, minTxnId, maxTxnId, minDecidedId).results();
CloseableIterator<TxnId> table = tableSearch(commandStoreId, range.start(), range.end(), minTxnId, maxTxnId, minDecidedId);
return new DefaultResult(minTxnId, maxTxnId, minDecidedId, MergeIterator.get(Arrays.asList(inMemory, table)));
CloseableIterator<TxnId> inMemory = index.search(commandStoreId, range, minTxnId, maxTxnId, decidedRX).results();
CloseableIterator<TxnId> table = tableSearch(commandStoreId, range.start(), range.end(), minTxnId, maxTxnId, decidedRX);
return new DefaultResult(minTxnId, maxTxnId, decidedRX, MergeIterator.get(Arrays.asList(inMemory, table)));
}
@Override
public Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
public Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX)
{
CloseableIterator<TxnId> inMemory = index.search(commandStoreId, key, minTxnId, maxTxnId, minDecidedId).results();
CloseableIterator<TxnId> table = tableSearch(commandStoreId, key, minTxnId, maxTxnId, minDecidedId);
return new DefaultResult(minTxnId, maxTxnId, minDecidedId, MergeIterator.get(Arrays.asList(inMemory, table)));
CloseableIterator<TxnId> inMemory = index.search(commandStoreId, key, minTxnId, maxTxnId, decidedRX).results();
CloseableIterator<TxnId> table = tableSearch(commandStoreId, key, minTxnId, maxTxnId);
return new DefaultResult(minTxnId, maxTxnId, decidedRX, MergeIterator.get(Arrays.asList(inMemory, table)));
}
private CloseableIterator<TxnId> tableSearch(int store, TokenKey start, TokenKey end, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
private CloseableIterator<TxnId> tableSearch(int store, TokenKey start, TokenKey end, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX)
{
RowFilter rowFilter = RowFilter.create(false);
rowFilter.add(AccordJournalTable.SyntheticColumn.participants.metadata, Operator.GT, OrderedRouteSerializer.serialize(start));
@ -276,13 +276,10 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
rowFilter.add(AccordJournalTable.SyntheticColumn.store_id.metadata, Operator.EQ, Int32Type.instance.decompose(store));
rowFilter.add(AccordJournalTable.SyntheticColumn.txn_id.metadata, Operator.GTE, CommandSerializers.txnId.serialize(minTxnId));
rowFilter.add(AccordJournalTable.SyntheticColumn.txn_id.metadata, Operator.LTE, CommandSerializers.timestamp.serialize(maxTxnId));
if (minDecidedId != null)
rowFilter.add(AccordJournalTable.SyntheticColumn.min_decided_id.metadata, Operator.GTE, CommandSerializers.txnId.serialize(minDecidedId));
return process(store, rowFilter);
}
private CloseableIterator<TxnId> tableSearch(int store, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
private CloseableIterator<TxnId> tableSearch(int store, TokenKey key, TxnId minTxnId, Timestamp maxTxnId)
{
RowFilter rowFilter = RowFilter.create(false);
rowFilter.add(AccordJournalTable.SyntheticColumn.participants.metadata, Operator.GTE, OrderedRouteSerializer.serialize(key));
@ -290,9 +287,6 @@ public class AccordJournalTable<K extends JournalKey, V> implements RangeSearche
rowFilter.add(AccordJournalTable.SyntheticColumn.store_id.metadata, Operator.EQ, Int32Type.instance.decompose(store));
rowFilter.add(AccordJournalTable.SyntheticColumn.txn_id.metadata, Operator.GTE, CommandSerializers.txnId.serialize(minTxnId));
rowFilter.add(AccordJournalTable.SyntheticColumn.txn_id.metadata, Operator.LTE, CommandSerializers.timestamp.serialize(maxTxnId));
if (minDecidedId != null)
rowFilter.add(AccordJournalTable.SyntheticColumn.min_decided_id.metadata, Operator.GTE, CommandSerializers.txnId.serialize(minDecidedId));
return process(store, rowFilter);
}

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

@ -270,13 +270,10 @@ public class AccordObjectSizes
size += range(dependencies.rangeDeps.range(i));
size += ObjectSizes.sizeOfReferenceArray(dependencies.rangeDeps.rangeCount());
for (int i = 0 ; i < dependencies.keyDeps.txnIdCount() ; ++i)
size += timestamp(dependencies.keyDeps.txnId(i));
for (int i = 0 ; i < dependencies.rangeDeps.txnIdCount() ; ++i)
size += timestamp(dependencies.rangeDeps.txnId(i));
size += KeyDeps.SerializerSupport.keysToTxnIdsCount(dependencies.keyDeps) * 4L;
size += RangeDeps.SerializerSupport.rangesToTxnIdsCount(dependencies.rangeDeps) * 4L;
size += dependencies.keyDeps.txnIdCount() * TIMESTAMP_SIZE;
size += dependencies.rangeDeps.txnIdCount() * TIMESTAMP_SIZE;
size += KeyDeps.SerializerSupport.keysToTxnIds(dependencies.keyDeps).length * 4L;
size += RangeDeps.SerializerSupport.rangesToTxnIds(dependencies.rangeDeps).length * 4L;
return size;
}

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

@ -18,6 +18,8 @@
package org.apache.cassandra.service.accord;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;
@ -31,6 +33,7 @@ import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import javax.annotation.Nullable;
import accord.api.Journal;
import accord.api.RoutingKey;
import accord.local.Command;
import accord.local.CommandSummaries;
@ -54,12 +57,17 @@ import accord.utils.Invariants;
import accord.utils.SymmetricComparator;
import accord.utils.UnhandledEnum;
import org.agrona.collections.Object2ObjectHashMap;
import org.apache.cassandra.exceptions.UnknownTableException;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.service.accord.api.TokenKey;
import org.apache.cassandra.service.accord.serializers.Version;
import org.apache.cassandra.utils.btree.BTree;
import org.apache.cassandra.utils.btree.BTreeSet;
import org.apache.cassandra.utils.btree.IntervalBTree;
import org.apache.cassandra.utils.concurrent.IntrusiveStack;
import static accord.api.Journal.Load.MINIMAL;
import static accord.api.Journal.Load.MINIMAL_WITH_DEPS;
import static org.apache.cassandra.utils.btree.IntervalBTree.InclusiveEndHelper.endWithStart;
import static org.apache.cassandra.utils.btree.IntervalBTree.InclusiveEndHelper.keyEndWithStart;
import static org.apache.cassandra.utils.btree.IntervalBTree.InclusiveEndHelper.keyStartWithEnd;
@ -375,11 +383,11 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
{
case Range:
for (Unseekable range : searchKeysOrRanges)
manager.searcher.search(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, minDecidedId).consume(forEach);
manager.searcher.search(manager.commandStore.id(), (TokenRange) range, minTxnId, maxTxnId, decidedRx).consume(forEach);
break;
case Key:
for (Unseekable key : searchKeysOrRanges)
manager.searcher.search(manager.commandStore.id(), (TokenKey) key, minTxnId, maxTxnId, minDecidedId).consume(forEach);
manager.searcher.search(manager.commandStore.id(), (TokenKey) key, minTxnId, maxTxnId, decidedRx).consume(forEach);
}
}
@ -448,7 +456,7 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
}
else
{
Command cmd = manager.commandStore.loadCommand(txnId);
Command.MinimalWithDeps cmd = manager.commandStore.loadMinimalWithDeps(txnId);
if (cmd != null)
return ifRelevant(cmd);
}
@ -477,19 +485,38 @@ public class CommandsForRanges extends TreeMap<Timestamp, Summary> implements Co
}
TxnId txnId = state.key();
if (!txnId.isVisible() || txnId.compareTo(minTxnId) < 0 || txnId.compareTo(maxTxnId) >= 0)
if (!isMaybeRelevant(txnId))
return null;
Command command = state.getExclusive();
Object command = state.getOrShrunkExclusive();
if (command == null)
return null;
return ifRelevant(command);
}
public Summary ifRelevant(Command.Minimal cmd)
{
Invariants.require(findAsDep == null);
return ifRelevant(cmd.txnId, cmd.executeAt, cmd.saveStatus, cmd.durability, cmd.participants, null);
if (command instanceof Command)
return ifRelevant((Command) command);
Invariants.require(command instanceof ByteBuffer);
AccordJournal.Builder builder = new AccordJournal.Builder(txnId, findAsDep == null ? MINIMAL : MINIMAL_WITH_DEPS);
ByteBuffer buffer = (ByteBuffer) command;
buffer.mark();
try (DataInputBuffer buf = new DataInputBuffer(buffer, false))
{
builder.deserializeNext(buf, Version.LATEST);
if (findAsDep == null) return ifRelevant(builder.asMinimal());
else return ifRelevant(builder.asMinimalWithDeps());
}
catch (UnknownTableException e)
{
return null;
}
catch (IOException e)
{
throw new RuntimeException(e);
}
finally
{
buffer.reset();
}
}
}
}

17
src/java/org/apache/cassandra/service/accord/RangeSearcher.java
View File

@ -22,6 +22,7 @@ import java.util.function.Consumer;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX;
import accord.primitives.Timestamp;
import accord.primitives.TxnId;
import org.apache.cassandra.index.accord.RouteIndexFormat;
@ -30,8 +31,8 @@ import org.apache.cassandra.utils.CloseableIterator;
public interface RangeSearcher
{
Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId);
Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId);
Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX);
Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX);
static RangeSearcher extractRangeSearcher(Object o)
{
@ -55,15 +56,15 @@ public interface RangeSearcher
{
private final TxnId minTxnId;
private final Timestamp maxTxnId;
private final @Nullable TxnId minDecidedId;
private final @Nullable MaxDecidedRX.DecidedRX decidedRX;
private final CloseableIterator<TxnId> results;
private boolean consumed = false;
public DefaultResult(TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, CloseableIterator<TxnId> results)
public DefaultResult(TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, CloseableIterator<TxnId> results)
{
this.minTxnId = minTxnId;
this.maxTxnId = maxTxnId;
this.minDecidedId = minDecidedId;
this.decidedRX = decidedRX;
this.results = results;
}
@ -84,7 +85,7 @@ public interface RangeSearcher
{
TxnId next = results.next();
if (next.compareTo(minTxnId) >= 0 && next.compareTo(maxTxnId) < 0 && RouteIndexFormat.includeByMinDecidedId(minDecidedId, next))
if (next.compareTo(minTxnId) >= 0 && next.compareTo(maxTxnId) < 0 && RouteIndexFormat.includeByDecidedRX(decidedRX, next))
forEach.accept(next);
}
}
@ -120,13 +121,13 @@ public interface RangeSearcher
instance;
@Override
public Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
public Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX)
{
return NoopResult.instance;
}
@Override
public Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
public Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX)
{
return NoopResult.instance;
}

46
src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java
View File

@ -32,6 +32,8 @@ import javax.annotation.Nullable;
import com.google.common.annotations.VisibleForTesting;
import accord.local.MaxDecidedRX;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.primitives.Route;
import accord.primitives.Timestamp;
import accord.primitives.Txn;
@ -81,35 +83,35 @@ public class RouteInMemoryIndex<V> implements RangeSearcher
}
@Override
public RangeSearcher.Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
public RangeSearcher.Result search(int commandStoreId, TokenRange range, TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX)
{
NavigableSet<TxnId> result = search(commandStoreId, range.table(),
OrderedRouteSerializer.serializeTokenOnly(range.start()),
OrderedRouteSerializer.serializeTokenOnly(range.end()),
minTxnId, maxTxnId, minDecidedId);
return new DefaultResult(minTxnId, maxTxnId, minDecidedId, CloseableIterator.wrap(result.iterator()));
minTxnId, maxTxnId, decidedRX);
return new DefaultResult(minTxnId, maxTxnId, decidedRX, CloseableIterator.wrap(result.iterator()));
}
private synchronized NavigableSet<TxnId> search(int storeId, TableId tableId, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
private synchronized NavigableSet<TxnId> search(int storeId, TableId tableId, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX)
{
// store matches in a hash set so add is O(1), and the sorting is done after collecting all matches
Set<TxnId> matches = new HashSet<>();
segmentIndexes.values().forEach(s -> s.search(storeId, tableId, start, end, minTxnId, maxTxnId, minDecidedId, e -> matches.add(e.getValue())));
segmentIndexes.values().forEach(s -> s.search(storeId, tableId, start, end, minTxnId, maxTxnId, decidedRX, e -> matches.add(e.getValue())));
return matches.isEmpty() ? Collections.emptyNavigableSet() : new TreeSet<>(matches);
}
@Override
public RangeSearcher.Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
public RangeSearcher.Result search(int commandStoreId, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX)
{
NavigableSet<TxnId> result = search(commandStoreId, key.table(), OrderedRouteSerializer.serializeTokenOnly(key), minTxnId, maxTxnId, minDecidedId);
return new DefaultResult(minTxnId, maxTxnId, minDecidedId, CloseableIterator.wrap(result.iterator()));
NavigableSet<TxnId> result = search(commandStoreId, key.table(), OrderedRouteSerializer.serializeTokenOnly(key), minTxnId, maxTxnId, decidedRX);
return new DefaultResult(minTxnId, maxTxnId, decidedRX, CloseableIterator.wrap(result.iterator()));
}
private synchronized NavigableSet<TxnId> search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId)
private synchronized NavigableSet<TxnId> search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX)
{
// store matches in a hash set so add is O(1), and the sorting is done after collecting all matches
Set<TxnId> matches = new HashSet<>();
segmentIndexes.values().forEach(s -> s.search(storeId, tableId, key, minTxnId, maxTxnId, minDecidedId, e -> matches.add(e.getValue())));
segmentIndexes.values().forEach(s -> s.search(storeId, tableId, key, minTxnId, maxTxnId, decidedRX, e -> matches.add(e.getValue())));
return matches.isEmpty() ? Collections.emptyNavigableSet() : new TreeSet<>(matches);
}
@ -133,22 +135,22 @@ public class RouteInMemoryIndex<V> implements RangeSearcher
private void search(int storeId, TableId tableId,
byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<Map.Entry<IndexRange, TxnId>> fn)
{
StoreIndex idx = storeIndexes.get(storeId);
if (idx == null) return;
idx.search(tableId, start, end, minTxnId, maxTxnId, minDecidedId, fn);
idx.search(tableId, start, end, minTxnId, maxTxnId, decidedRX, fn);
}
private void search(int storeId, TableId tableId,
byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<Map.Entry<IndexRange, TxnId>> fn)
{
StoreIndex idx = storeIndexes.get(storeId);
if (idx == null) return;
idx.search(tableId, key, minTxnId, maxTxnId, minDecidedId, fn);
idx.search(tableId, key, minTxnId, maxTxnId, decidedRX, fn);
}
}
@ -176,22 +178,22 @@ public class RouteInMemoryIndex<V> implements RangeSearcher
public void search(TableId tableId,
byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<Map.Entry<IndexRange, TxnId>> fn)
{
TableIndex index = tableIndex.get(tableId);
if (index == null) return;
index.search(start, end, minTxnId, maxTxnId, minDecidedId, fn);
index.search(start, end, minTxnId, maxTxnId, decidedRX, fn);
}
public void search(TableId tableId,
byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX,
Consumer<Map.Entry<IndexRange, TxnId>> fn)
{
TableIndex index = tableIndex.get(tableId);
if (index == null) return;
index.search(key, minTxnId, maxTxnId, minDecidedId, fn);
index.search(key, minTxnId, maxTxnId, decidedRX, fn);
}
}
@ -225,12 +227,12 @@ public class RouteInMemoryIndex<V> implements RangeSearcher
}
private void search(byte[] start, byte[] end,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX,
Consumer<Map.Entry<IndexRange, TxnId>> fn)
{
if (minTxnId.compareTo(max) > 0) return;
if (maxTxnId.compareTo(min) < 0) return;
if (maxRX != null && !RouteIndexFormat.includeByMinDecidedId(min, maxRX)) return;
if (maxRX != null && !RouteIndexFormat.includeByDecidedRX(decidedRX, maxRX)) return;
index.search(new IndexRange(start, end), e -> {
if (minTxnId.compareTo(e.getValue()) > 0) return;
if (maxTxnId.compareTo(e.getValue()) < 0) return;
@ -239,12 +241,12 @@ public class RouteInMemoryIndex<V> implements RangeSearcher
}
private void search(byte[] key,
TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId,
TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX,
Consumer<Map.Entry<IndexRange, TxnId>> fn)
{
if (minTxnId.compareTo(max) > 0) return;
if (maxTxnId.compareTo(min) < 0) return;
if (maxRX != null && !RouteIndexFormat.includeByMinDecidedId(min, maxRX)) return;
if (maxRX != null && !RouteIndexFormat.includeByDecidedRX(decidedRX, maxRX)) return;
index.searchToken(key, e -> {
if (minTxnId.compareTo(e.getValue()) > 0) return;
if (maxTxnId.compareTo(e.getValue()) < 0) return;

2
src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java
View File

@ -453,7 +453,7 @@ public class AccordInteropExecution implements ReadCoordinator
// Provide request callbacks with a way to send maximal commits on Insufficient responses
public void sendMaximalCommit(Id to)
{
Commit.stableMaximal(node, to, txn, txnId, executeAt, route, deps);
node.send(to, new Commit(Kind.StableWithTxnAndDeps, to, allTopologies, txnId, txn, route, ballot, executeAt, deps));
}
public void maybeUpdateUniqueHlc(long uniqueHlc)

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

@ -19,6 +19,7 @@
package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import java.util.Objects;
import accord.api.Result;
import accord.api.RoutingKey;
@ -31,7 +32,7 @@ import accord.messages.CheckStatus.CheckStatusReply;
import accord.primitives.Ballot;
import accord.primitives.Known;
import accord.primitives.KnownMap;
import accord.primitives.KnownMap.MinMax;
import accord.primitives.KnownMap.MinAndMaxKnown;
import accord.primitives.PartialDeps;
import accord.primitives.PartialTxn;
import accord.primitives.Participants;
@ -63,17 +64,17 @@ public class CheckStatusSerializers
KeySerializers.routingKey.serialize(knownMap.startAt(i), out);
for (int i = 0 ; i < size ; ++i)
{
KnownMap.MinMax minMax = knownMap.valueAt(i);
if (minMax == null)
MinAndMaxKnown minAndMax = knownMap.valueAt(i);
if (minAndMax == null)
{
out.writeByte(0);
continue;
}
boolean equal = minMax.min.equals(minMax);
boolean equal = Objects.equals(minAndMax.minOwned, minAndMax.max);
out.writeByte(equal ? 1 : 2);
known.serialize(minMax.min, out);
known.serialize(minAndMax.minOwned, out);
if (!equal)
known.serialize(minMax, out);
known.serialize(minAndMax.max, out);
}
}
@ -84,7 +85,7 @@ public class CheckStatusSerializers
RoutingKey[] starts = new RoutingKey[size + 1];
for (int i = 0 ; i <= size ; ++i)
starts[i] = KeySerializers.routingKey.deserialize(in);
MinMax[] values = new MinMax[size];
MinAndMaxKnown[] values = new MinAndMaxKnown[size];
for (int i = 0 ; i < size ; ++i)
{
int kind = in.readByte();
@ -92,7 +93,7 @@ public class CheckStatusSerializers
continue;
Known min = known.deserialize(in);
Known max = kind == 1 ? min : known.deserialize(in);
values[i] = new KnownMap.MinMax(min, max);
values[i] = new MinAndMaxKnown(min, max);
}
return KnownMap.SerializerSupport.create(true, starts, values);
}
@ -106,14 +107,14 @@ public class CheckStatusSerializers
result += KeySerializers.routingKey.serializedSize(knownMap.startAt(i));
for (int i = 0 ; i < size ; ++i)
{
KnownMap.MinMax minMax = knownMap.valueAt(i);
MinAndMaxKnown minMax = knownMap.valueAt(i);
result += TypeSizes.BYTE_SIZE;
if (minMax == null)
continue;
boolean equal = minMax.min.equals(minMax);
result += known.serializedSize(minMax.min);
boolean equal = Objects.equals(minMax.minOwned, minMax.max);
result += known.serializedSize(minMax.minOwned);
if (!equal)
result += known.serializedSize(minMax);
result += known.serializedSize(minMax.max);
}
return result;
}

220
src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.function.IntFunction;
import com.google.common.annotations.VisibleForTesting;
@ -47,7 +48,10 @@ import accord.primitives.Txn;
import accord.primitives.TxnId;
import accord.primitives.Unseekables;
import accord.primitives.Writes;
import accord.utils.ArrayBuffers;
import accord.utils.BitUtils;
import accord.utils.Invariants;
import accord.utils.VIntCoding;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.ByteBufferAccessor;
import org.apache.cassandra.db.marshal.ValueAccessor;
@ -62,14 +66,16 @@ import org.apache.cassandra.service.accord.txn.TxnRead;
import org.apache.cassandra.service.accord.txn.TxnWrite;
import org.apache.cassandra.utils.NullableSerializer;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializedPackedBitsSize;
public class CommandSerializers
{
private CommandSerializers()
{
}
public static final VariableWidthTimestampSerializer<TxnId> txnId = new VariableWidthTimestampSerializer<>(TxnId::fromValues);
public static final VariableWidthTimestampSerializer<Timestamp> timestamp = new VariableWidthTimestampSerializer<>(Timestamp::fromValues);
public static final VariableWidthTimestampSerializer<TxnId> txnId = new VariableWidthTimestampSerializer<>(TxnId::fromValues, TxnId::fromBits, TxnId[]::new);
public static final VariableWidthTimestampSerializer<Timestamp> timestamp = new VariableWidthTimestampSerializer<>(Timestamp::fromValues, Timestamp::fromBits, Timestamp[]::new);
public static final BallotSerializer ballot = new BallotSerializer(); // permits null
public static final UnversionedSerializer<Txn.Kind> kind = EncodeAsVInt32.of(Txn.Kind.class);
public static final StoreParticipantsSerializer participants = new StoreParticipantsSerializer();
@ -490,12 +496,9 @@ public class CommandSerializers
Invariants.require(EPOCH_SHIFT + Integer.bitCount(EPOCH_MASK) < 8);
}
interface Factory<T extends Timestamp>
{
T create(long epoch, long hlc, int flags, Node.Id node);
}
private final VariableWidthTimestampSerializer.Factory<T> factory;
private final Timestamp.ValueFactory<T> factory;
private final Timestamp.RawFactory<T> rawFactory;
private final IntFunction<T[]> allocator;
T decodeSpecial(int encodingFlags)
{
@ -510,9 +513,11 @@ public class CommandSerializers
return NULL_BYTE;
}
private VariableWidthTimestampSerializer(VariableWidthTimestampSerializer.Factory<T> factory)
private VariableWidthTimestampSerializer(Timestamp.ValueFactory<T> factory, Timestamp.RawFactory<T> rawFactory, IntFunction<T[]> allocator)
{
this.factory = factory;
this.rawFactory = rawFactory;
this.allocator = allocator;
}
@Override
@ -546,6 +551,73 @@ public class CommandSerializers
out.writeLeastSignificantBytes(ts.node.id, nodeLength);
}
public void serializeArray(T[] ts, DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt32(ts.length);
if (ts.length == 0)
return;
long minEpoch = Long.MAX_VALUE, maxEpoch = 0;
long minHlc = Long.MAX_VALUE, maxHlc = 0;
int minFlags = 0xFFFF, maxFlags = 0;
int minNodeId = Integer.MAX_VALUE, maxNodeId = 0;
for (int i = 0; i < ts.length; i++)
{
T t = ts[i];
long epoch = t.epoch();
minEpoch = Math.min(epoch, minEpoch);
maxEpoch = Math.max(epoch, maxEpoch);
long hlc = t.hlc();
minHlc = Math.min(hlc, minHlc);
maxHlc = Math.max(hlc, maxHlc);
int flags = t.flags();
minFlags = Math.min(flags, minFlags);
maxFlags = Math.max(flags, maxFlags);
int nodeId = t.node.id;
minNodeId = Math.min(nodeId, minNodeId);
maxNodeId = Math.max(nodeId, maxNodeId);
}
int epochBits = BitUtils.numberOfBitsToRepresent(maxEpoch - minEpoch);
int hlcBits = BitUtils.numberOfBitsToRepresent(maxHlc - minHlc);
int flagBits = BitUtils.numberOfBitsToRepresent(maxFlags - minFlags);
int nodeBits = BitUtils.numberOfBitsToRepresent(maxNodeId - minNodeId);
// we could pack these a bit more tightly if we wanted to
out.writeUnsignedVInt(minEpoch);
out.writeUnsignedVInt(minHlc);
out.writeUnsignedVInt32(minFlags);
out.writeUnsignedVInt32(minNodeId);
out.writeByte(epochBits);
out.writeByte(hlcBits);
out.writeByte(flagBits);
out.writeByte(nodeBits);
long finalMinEpoch = minEpoch;
SerializePacked.serializePacked((in, i) -> in[i].epoch() - finalMinEpoch, ts, 0, ts.length, maxEpoch - minEpoch, out);
long finalMinHlc = minHlc;
SerializePacked.serializePacked((in, i) -> in[i].hlc() - finalMinHlc, ts, 0, ts.length, maxHlc - minHlc, out);
long finalMinFlags = minFlags;
SerializePacked.serializePacked((in, i) -> in[i].flags() - finalMinFlags, ts, 0, ts.length, maxFlags - minFlags, out);
long finalMinNodeId = minNodeId;
SerializePacked.serializePacked((in, i) -> in[i].node.id - finalMinNodeId, ts, 0, ts.length, maxNodeId - minNodeId, out);
}
public int flags(T ts)
{
long epoch = ts.epoch();
long hlc = ts.hlc();
int flags = ts.flags();
int epochLength = length(epoch, EPOCH_MIN_LENGTH);
int hlcLength = length(hlc, HLC_MIN_LENGTH);
int flagsLength = length(flags, FLAGS_MIN_LENGTH);
int nodeLength = length(ts.node.id, NODE_MIN_LENGTH);
return encodeLength(epochLength, EPOCH_SHIFT, EPOCH_MIN_LENGTH, EPOCH_MASK)
| encodeLength(hlcLength, HLC_SHIFT, HLC_MIN_LENGTH, HLC_MASK)
| encodeLength(flagsLength, FLAGS_SHIFT, FLAGS_MIN_LENGTH, FLAGS_MASK)
| encodeLength(nodeLength, NODE_SHIFT, NODE_MIN_LENGTH, NODE_MASK);
}
// exactly the same fundamental format as serialize(), only we interleave the length bits with the values, maintaining ordering
public <V> int serializeComparable(T ts, V dst, ValueAccessor<V> accessor, int offset)
{
@ -635,11 +707,16 @@ public class CommandSerializers
int encodingFlags = in.readByte();
if (encodingFlags < 0)
return;
in.skipBytesFully(lengthWithFlags(encodingFlags));
}
public int lengthWithFlags(int encodingFlags)
{
int epochLength = decodeLength(encodingFlags, EPOCH_SHIFT, EPOCH_MIN_LENGTH, EPOCH_MASK);
int hlcLength = decodeLength(encodingFlags, HLC_SHIFT, HLC_MIN_LENGTH, HLC_MASK);
int flagsLength = decodeLength(encodingFlags, FLAGS_SHIFT, FLAGS_MIN_LENGTH, FLAGS_MASK);
int nodeLength = decodeLength(encodingFlags, NODE_SHIFT, NODE_MIN_LENGTH, NODE_MASK);
in.skipBytesFully(epochLength + hlcLength + flagsLength + nodeLength);
return epochLength + hlcLength + flagsLength + nodeLength;
}
@Override
@ -648,10 +725,59 @@ public class CommandSerializers
int encodingFlags = in.readByte();
if (encodingFlags < 0)
return decodeSpecial(encodingFlags);
return deserializeFixed(encodingFlags, in);
}
public T[] deserializeArray(DataInputPlus in) throws IOException
{
int length = in.readUnsignedVInt32();
if (length == 0)
return allocator.apply(0);
// we could pack these a bit more tightly if we wanted to
long minEpoch = in.readUnsignedVInt();
long minHlc = in.readUnsignedVInt();
int minFlags = in.readUnsignedVInt32();
int minNodeId = in.readUnsignedVInt32();
int epochBits = in.readByte();
int hlcBits = in.readByte();
int flagBits = in.readByte();
int nodeBits = in.readByte();
long[] bits = ArrayBuffers.cachedLongs().getLongs(length * 2);
SerializePacked.deserializePacked((out, i, v) -> out[i*2] = Timestamp.epochMsb(minEpoch + v), bits, 0, length, mask(epochBits), in);
SerializePacked.deserializePacked((out, i, v) -> {
long hlc = minHlc + v;
out[i*2] |= Timestamp.hlcMsb(hlc);
out[i*2+1] = Timestamp.hlcLsb(hlc);
}, bits, 0, length, mask(hlcBits), in);
SerializePacked.deserializePacked((out, i, v) -> out[i*2 + 1] |= minFlags + v, bits, 0, length, mask(flagBits), in);
T[] ts = allocator.apply(length);
SerializePacked.deserializePacked((out, i, v) -> {
Node.Id id = new Node.Id(minNodeId + (int)v);
ts[i] = rawFactory.create(bits[i*2], bits[i*2 + 1], id);
}, bits, 0, length, mask(nodeBits), in);
ArrayBuffers.cachedLongs().forceDiscard(bits);
return ts;
}
private static long mask(int bits)
{
return bits == 0 ? 0 : -1L >>> (64 - bits);
}
public T deserializeFixed(int encodingFlags, DataInputPlus in) throws IOException
{
Invariants.require(((byte)encodingFlags) >= 0);
int epochLength = decodeLength(encodingFlags, EPOCH_SHIFT, EPOCH_MIN_LENGTH, EPOCH_MASK);
int hlcLength = decodeLength(encodingFlags, HLC_SHIFT, HLC_MIN_LENGTH, HLC_MASK);
int flagsLength = decodeLength(encodingFlags, FLAGS_SHIFT, FLAGS_MIN_LENGTH, FLAGS_MASK);
int nodeLength = decodeLength(encodingFlags, NODE_SHIFT, NODE_MIN_LENGTH, NODE_MASK);
return deserialize(epochLength, hlcLength, flagsLength, nodeLength, in);
}
private T deserialize(int epochLength, int hlcLength, int flagsLength, int nodeLength, DataInputPlus in) throws IOException
{
long epoch = in.readLeastSignificantBytes(epochLength);
long hlc = in.readLeastSignificantBytes(hlcLength);
int flags = Math.toIntExact(in.readLeastSignificantBytes(flagsLength));
@ -728,6 +854,49 @@ public class CommandSerializers
return 1 + epochLength + hlcLength + flagsLength + nodeLength;
}
public long serializedArraySize(T[] ts)
{
if (ts.length == 0)
return 1;
long minEpoch = Long.MAX_VALUE, maxEpoch = 0;
long minHlc = Long.MAX_VALUE, maxHlc = 0;
int minFlags = 0xFFFF, maxFlags = 0;
int minNodeId = Integer.MAX_VALUE, maxNodeId = 0;
for (int i = 0; i < ts.length; i++)
{
T t = ts[i];
long epoch = t.epoch();
minEpoch = Math.min(epoch, minEpoch);
maxEpoch = Math.max(epoch, maxEpoch);
long hlc = t.hlc();
minHlc = Math.min(hlc, minHlc);
maxHlc = Math.max(hlc, maxHlc);
int flags = t.flags();
minFlags = Math.min(flags, minFlags);
maxFlags = Math.max(flags, maxFlags);
int nodeId = t.node.id;
minNodeId = Math.min(nodeId, minNodeId);
maxNodeId = Math.max(nodeId, maxNodeId);
}
int epochBits = BitUtils.numberOfBitsToRepresent(maxEpoch - minEpoch);
int hlcBits = BitUtils.numberOfBitsToRepresent(maxHlc - minHlc);
int flagBits = BitUtils.numberOfBitsToRepresent(maxFlags - minFlags);
int nodeBits = BitUtils.numberOfBitsToRepresent(maxNodeId - minNodeId);
return VIntCoding.sizeOfUnsignedVInt(ts.length)
+ VIntCoding.sizeOfUnsignedVInt(minEpoch)
+ VIntCoding.sizeOfUnsignedVInt(minHlc)
+ VIntCoding.sizeOfUnsignedVInt(minFlags)
+ VIntCoding.sizeOfUnsignedVInt(minNodeId)
+ 4
+ serializedPackedBitsSize(ts.length, epochBits)
+ serializedPackedBitsSize(ts.length, hlcBits)
+ serializedPackedBitsSize(ts.length, flagBits)
+ serializedPackedBitsSize(ts.length, nodeBits);
}
private static int length(long value, int minLength)
{
int length = ((64 + 7) - Long.numberOfLeadingZeros(value))/8;
@ -747,6 +916,12 @@ public class CommandSerializers
return encoded << shift;
}
private static int reencodePartDecodedLength(int length, int shift, int mask)
{
Invariants.require(length <= mask);
return length << shift;
}
private static long packLength(int length, int shift, int minLength, int mask)
{
int encoded = length - minLength;
@ -758,6 +933,11 @@ public class CommandSerializers
{
return minLength + ((encodingFlags >>> shift) & mask);
}
private static int maxPartDecoded(int flagsa, int flagsb, int shift, int mask)
{
return Math.max(((flagsa >>> shift) & mask), (flagsb >>> shift) & mask);
}
}
public static class BallotSerializer extends VariableWidthTimestampSerializer<Ballot>
@ -766,7 +946,7 @@ public class CommandSerializers
private static final byte MAX_BYTE = (byte) 0x82;
private BallotSerializer()
{
super(Ballot::fromValues);
super(Ballot::fromValues, Ballot::fromBits, Ballot[]::new);
}
@Override
@ -923,9 +1103,9 @@ public class CommandSerializers
{
txnId.serialize(writes.txnId, out);
ExecuteAtSerializer.serialize(writes.txnId, writes.executeAt, out);
KeySerializers.seekables.serialize(writes.keys, out);
boolean hasWrite = writes.write != null;
out.writeBoolean(hasWrite);
KeySerializers.seekables.serialize(writes.keys, out);
if (hasWrite)
CommandSerializers.write.serialize(writes.write, writes.keys, out, version);
}
@ -935,14 +1115,28 @@ public class CommandSerializers
{
TxnId id = txnId.deserialize(in);
Timestamp executeAt = ExecuteAtSerializer.deserialize(id, in);
Seekables seekables = KeySerializers.seekables.deserialize(in);
boolean hasWrite = in.readBoolean();
Seekables seekables = KeySerializers.seekables.deserialize(in);
Write write = null;
if (hasWrite)
write = CommandSerializers.write.deserialize(seekables, in, version);
return new Writes(id, executeAt, seekables, write);
}
@Override
public void skip(DataInputPlus in, Version version) throws IOException
{
txnId.skip(in);
ExecuteAtSerializer.skip(null, in);
boolean hasWrite = in.readBoolean();
if (hasWrite)
{
Seekables seekables = KeySerializers.seekables.deserialize(in);
CommandSerializers.write.skip(seekables, in, version);
}
else KeySerializers.seekables.skip(in);
}
@Override
public long serializedSize(Writes writes, Version version)
{

307
src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java
View File

@ -20,7 +20,6 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
import accord.primitives.Deps;
import accord.primitives.KeyDeps;
@ -30,6 +29,8 @@ import accord.primitives.Range;
import accord.primitives.RangeDeps;
import accord.primitives.RoutingKeys;
import accord.primitives.TxnId;
import accord.utils.Invariants;
import accord.utils.VIntCoding;
import org.apache.cassandra.io.UnversionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
@ -37,17 +38,21 @@ import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.utils.NullableSerializer;
import static accord.primitives.KeyDeps.SerializerSupport.keysToTxnIds;
import static accord.primitives.KeyDeps.SerializerSupport.keysToTxnIdsCount;
import static accord.primitives.KeyDeps.SerializerSupport.txnIdsToKeys;
import static accord.primitives.RangeDeps.SerializerSupport.ranges;
import static accord.primitives.RangeDeps.SerializerSupport.rangesToTxnIds;
import static accord.primitives.RangeDeps.SerializerSupport.rangesToTxnIdsCount;
import static org.apache.cassandra.db.TypeSizes.sizeofUnsignedVInt;
import static accord.primitives.RangeDeps.SerializerSupport.txnIdsToRanges;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.deserializePackedInts;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializePackedInts;
import static org.apache.cassandra.service.accord.serializers.SerializePacked.serializedPackedIntsSize;
public class DepsSerializers
{
public static final UnversionedSerializer<Range> tokenRange;
public static final DepsSerializer<Deps> deps;
public static final AbstractDepsSerializer<Deps> deps;
public static final UnversionedSerializer<Deps> nullableDeps;
public static final DepsSerializer<PartialDeps> partialDeps;
public static final AbstractDepsSerializer<PartialDeps> partialDeps;
public static final AbstractDepsSerializer<PartialDeps> partialDepsById;
public static final UnversionedSerializer<PartialDeps> nullablePartialDeps;
static
@ -58,14 +63,21 @@ public class DepsSerializers
deps = serializers.deps;
nullableDeps = serializers.nullableDeps;
partialDeps = serializers.partialDeps;
partialDepsById = serializers.partialDepsById;
nullablePartialDeps = serializers.nullablePartialDeps;
}
public static abstract class DepsSerializer<D extends Deps> implements UnversionedSerializer<D>
public static abstract class AbstractDepsSerializer<D extends Deps> implements UnversionedSerializer<D>
{
protected UnversionedSerializer<Range> tokenRange;
public DepsSerializer(UnversionedSerializer<Range> tokenRange)
static final int KEYS_BY_TXNID = 0x1;
static final int RANGES_BY_TXNID = 0x2;
static final int FLAGS_SIZE = VIntCoding.sizeOfUnsignedVInt(KEYS_BY_TXNID | RANGES_BY_TXNID);
final boolean forceByTxnId;
protected final UnversionedSerializer<Range> tokenRange;
public AbstractDepsSerializer(boolean forceByTxnId, UnversionedSerializer<Range> tokenRange)
{
this.forceByTxnId = forceByTxnId;
this.tokenRange = tokenRange;
}
@ -74,164 +86,225 @@ public class DepsSerializers
@Override
public void serialize(D deps, DataOutputPlus out) throws IOException
{
boolean keysByTxnId = forceByTxnId || deps.keyDeps.hasByTxnId();
boolean rangesByTxnId = forceByTxnId || deps.rangeDeps.hasByTxnId();
out.writeUnsignedVInt32((keysByTxnId ? KEYS_BY_TXNID : 0) | (rangesByTxnId ? RANGES_BY_TXNID : 0));
{
KeyDeps keyDeps = deps.keyDeps;
KeySerializers.routingKeys.serialize(keyDeps.keys(), out);
int txnIdCount = keyDeps.txnIdCount();
out.writeUnsignedVInt32(txnIdCount);
for (int i = 0; i < txnIdCount; i++)
CommandSerializers.txnId.serialize(keyDeps.txnId(i), out);
int keysToTxnIdsCount = keysToTxnIdsCount(keyDeps);
out.writeUnsignedVInt32(keysToTxnIdsCount);
for (int i = 0; i < keysToTxnIdsCount; i++)
out.writeUnsignedVInt32(keysToTxnIds(keyDeps, i));
CommandSerializers.txnId.serializeArray(KeyDeps.SerializerSupport.txnIds(keyDeps), out);
if (keysByTxnId) serializePackedXtoY(txnIdsToKeys(keyDeps), keyDeps.txnIdCount(), keyDeps.keys().size(), out);
else serializePackedXtoY(keysToTxnIds(keyDeps), keyDeps.keys().size(), keyDeps.txnIdCount(), out);
}
{
RangeDeps rangeDeps = deps.rangeDeps;
int rangeCount = rangeDeps.rangeCount();
out.writeUnsignedVInt32(rangeCount);
for (int i = 0; i < rangeCount; i++)
tokenRange.serialize(rangeDeps.range(i), out);
KeySerializers.rangeArray.serialize(ranges(rangeDeps), out);
CommandSerializers.txnId.serializeArray(RangeDeps.SerializerSupport.txnIds(rangeDeps), out);
if (rangesByTxnId) serializePackedXtoY(txnIdsToRanges(rangeDeps), rangeDeps.txnIdCount(), rangeDeps.rangeCount(), out);
else serializePackedXtoY(rangesToTxnIds(rangeDeps), rangeDeps.rangeCount(), rangeDeps.txnIdCount(), out);
}
}
int txnIdCount = rangeDeps.txnIdCount();
out.writeUnsignedVInt32(txnIdCount);
for (int i = 0; i < txnIdCount; i++)
CommandSerializers.txnId.serialize(rangeDeps.txnId(i), out);
private static void serializePackedXtoY(int[] xtoy, int xCount, int yCount, DataOutputPlus out) throws IOException
{
out.writeUnsignedVInt32(xtoy.length);
int rangesToTxnIdsCount = rangesToTxnIdsCount(rangeDeps);
out.writeUnsignedVInt32(rangesToTxnIdsCount);
for (int i = 0; i < rangesToTxnIdsCount; i++)
out.writeUnsignedVInt32(rangesToTxnIds(rangeDeps, i));
if ((xCount <= 1 || yCount <= 1) && (xtoy.length == xCount + yCount || xCount == 0 || yCount == 0))
{
// no point serializing as can be directly inferred
if (Invariants.isParanoid())
{
if (xCount == 1)
{
Invariants.require(xtoy[0] == xtoy.length, "%d != %d", xtoy[0], xtoy.length);
for (int i = 0 ; i < yCount ; ++i) Invariants.require(xtoy[1 + i] == i, "%d != %d", xtoy[1 + i], i);
}
else if (yCount == 1)
{
for (int i = 0 ; i < xCount ; ++i) Invariants.require(xtoy[i] == xCount + i + 1, "%d != %d", xtoy[i], xCount + i + 1);
for (int i = xCount ; i < xtoy.length ; ++i) Invariants.require(xtoy[i] == 0, "%d != %d", xtoy[i], 0);
}
else if (yCount == 0)
{
for (int i = 0 ; i < xCount ; ++i) Invariants.require(xtoy[i] == xCount, "%d != %d", xtoy[i], xCount);
}
else
{
Invariants.require(xtoy.length == 0);
}
}
}
else
{
serializePackedInts(xtoy, 0, xCount, xtoy.length, out);
serializePackedInts(xtoy, xCount, xtoy.length, yCount - 1, out);
}
}
@Override
public D deserialize(DataInputPlus in) throws IOException
{
int flags = in.readUnsignedVInt32();
KeyDeps keyDeps;
{
RoutingKeys keys = KeySerializers.routingKeys.deserialize(in);
int txnIdCount = in.readUnsignedVInt32();
TxnId[] txnIds = new TxnId[txnIdCount];
for (int i = 0; i < txnIdCount; i++)
txnIds[i] = CommandSerializers.txnId.deserialize(in);
int keysToTxnIdsCount = in.readUnsignedVInt32();
int[] keysToTxnIds = new int[keysToTxnIdsCount];
for (int i = 0; i < keysToTxnIdsCount; i++)
keysToTxnIds[i] = in.readUnsignedVInt32();
keyDeps = KeyDeps.SerializerSupport.create(keys, txnIds, keysToTxnIds);
TxnId[] txnIds = CommandSerializers.txnId.deserializeArray(in);
int[] txnIdsToKeys = null, keysToTxnIds = null;
if (0 != (flags & KEYS_BY_TXNID)) txnIdsToKeys = deserializePackedXtoY(txnIds.length, keys.size(), in);
else keysToTxnIds = deserializePackedXtoY(keys.size(), txnIds.length, in);
keyDeps = KeyDeps.SerializerSupport.create(keys, txnIds, keysToTxnIds, txnIdsToKeys);
}
RangeDeps rangeDeps;
{
int rangeCount = Ints.checkedCast(in.readUnsignedVInt32());
Range[] ranges = new Range[rangeCount];
for (int i = 0; i < rangeCount; i++)
ranges[i] = tokenRange.deserialize(in);
int txnIdCount = in.readUnsignedVInt32();
TxnId[] txnIds = new TxnId[txnIdCount];
for (int i = 0; i < txnIdCount; i++)
txnIds[i] = CommandSerializers.txnId.deserialize(in);
int rangesToTxnIdsCount = in.readUnsignedVInt32();
int[] rangesToTxnIds = new int[rangesToTxnIdsCount];
for (int i = 0; i < rangesToTxnIdsCount; i++)
rangesToTxnIds[i] = in.readUnsignedVInt32();
rangeDeps = RangeDeps.SerializerSupport.create(ranges, txnIds, rangesToTxnIds);
Range[] ranges = KeySerializers.rangeArray.deserialize(in);
TxnId[] txnIds = CommandSerializers.txnId.deserializeArray(in);
int[] txnIdsToRanges = null, rangesToTxnIds = null;
if (0 != (flags & RANGES_BY_TXNID)) txnIdsToRanges = deserializePackedXtoY(txnIds.length, ranges.length, in);
else rangesToTxnIds = deserializePackedXtoY(ranges.length, txnIds.length, in);
rangeDeps = RangeDeps.SerializerSupport.create(ranges, txnIds, rangesToTxnIds, txnIdsToRanges);
}
return deserialize(keyDeps, rangeDeps, in);
}
private static int[] deserializePackedXtoY(int xCount, int yCount, DataInputPlus in) throws IOException
{
int length = in.readUnsignedVInt32();
int[] xtoy = new int[length];
if ((xCount <= 1 || yCount <= 1) && (xtoy.length == xCount + yCount || xCount == 0 || yCount == 0))
{
// no point serializing as can be directly inferred
if (xCount == 1)
{
xtoy[0] = xtoy.length;
for (int i = 0 ; i < yCount ; ++i)
xtoy[1 + i] = i;
}
else if (yCount == 1)
{
for (int i = 0 ; i < xCount ; ++i)
xtoy[i] = xCount + i + 1;
}
else if (yCount == 0)
{
for (int i = 0 ; i < xCount ; ++i)
xtoy[i] = xCount;
}
else
{
Invariants.require(length == 0);
}
}
else
{
deserializePackedInts(xtoy, 0, xCount, xtoy.length, in);
deserializePackedInts(xtoy, xCount, xtoy.length, yCount - 1, in);
}
return xtoy;
}
@Override
public long serializedSize(D deps)
{
long size;
boolean keysByTxnId = forceByTxnId || deps.keyDeps.hasByTxnId();
boolean rangesByTxnId = forceByTxnId || deps.rangeDeps.hasByTxnId();
long size = FLAGS_SIZE;
{
KeyDeps keyDeps = deps.keyDeps;
size = KeySerializers.routingKeys.serializedSize(deps.keyDeps.keys());
int txnIdCount = keyDeps.txnIdCount();
size += sizeofUnsignedVInt(txnIdCount);
for (int i = 0; i < txnIdCount; i++)
size += CommandSerializers.txnId.serializedSize(keyDeps.txnId(i));
int keysToTxnIdsCount = keysToTxnIdsCount(keyDeps);
size += sizeofUnsignedVInt(keysToTxnIdsCount);
for (int i = 0; i < keysToTxnIdsCount; i++)
size += sizeofUnsignedVInt(keysToTxnIds(keyDeps, i));
size += KeySerializers.routingKeys.serializedSize(keyDeps.keys());
size += CommandSerializers.txnId.serializedArraySize(KeyDeps.SerializerSupport.txnIds(keyDeps));
size += keysByTxnId ? serializedPackedXtoYSize(txnIdsToKeys(keyDeps), keyDeps.txnIdCount(), keyDeps.keys().size())
: serializedPackedXtoYSize(keysToTxnIds(keyDeps), keyDeps.keys().size(), keyDeps.txnIdCount());
}
{
RangeDeps rangeDeps = deps.rangeDeps;
int rangeCount = rangeDeps.rangeCount();
size += sizeofUnsignedVInt(rangeCount);
for (int i = 0; i < rangeCount; ++i)
size += tokenRange.serializedSize(rangeDeps.range(i));
int txnIdCount = rangeDeps.txnIdCount();
size += sizeofUnsignedVInt(txnIdCount);
for (int i = 0; i < txnIdCount; i++)
size += CommandSerializers.txnId.serializedSize(rangeDeps.txnId(i));
int rangesToTxnIdsCount = rangesToTxnIdsCount(rangeDeps);
size += sizeofUnsignedVInt(rangesToTxnIdsCount);
for (int i = 0; i < rangesToTxnIdsCount; i++)
size += sizeofUnsignedVInt(rangesToTxnIds(rangeDeps, i));
size += KeySerializers.rangeArray.serializedSize(ranges(rangeDeps));
size += CommandSerializers.txnId.serializedArraySize(RangeDeps.SerializerSupport.txnIds(rangeDeps));
size += rangesByTxnId ? serializedPackedXtoYSize(txnIdsToRanges(rangeDeps), rangeDeps.txnIdCount(), rangeDeps.rangeCount())
: serializedPackedXtoYSize(rangesToTxnIds(rangeDeps), rangeDeps.rangeCount(), rangeDeps.txnIdCount());
}
return size;
}
private static long serializedPackedXtoYSize(int[] xtoy, int xCount, int yCount)
{
long size = VIntCoding.sizeOfUnsignedVInt(xtoy.length);
if ((xCount <= 1 || yCount <= 1) && (xtoy.length == xCount + yCount || xCount == 0 || yCount == 0))
{
// no point serializing as can be directly inferred
}
else
{
size += serializedPackedIntsSize(xtoy, 0, xCount, xtoy.length);
size += serializedPackedIntsSize(xtoy, xCount, xtoy.length, yCount - 1);
}
return size;
}
}
static class PartialDepsSerializer extends AbstractDepsSerializer<PartialDeps>
{
public PartialDepsSerializer(boolean preferByTxnId, UnversionedSerializer<Range> tokenRange)
{
super(preferByTxnId, tokenRange);
}
@Override
PartialDeps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, DataInputPlus in) throws IOException
{
Participants<?> covering = KeySerializers.participants.deserialize(in);
return new PartialDeps(covering, keyDeps, rangeDeps);
}
@Override
public void serialize(PartialDeps partialDeps, DataOutputPlus out) throws IOException
{
super.serialize(partialDeps, out);
KeySerializers.participants.serialize(partialDeps.covering, out);
}
@Override
public long serializedSize(PartialDeps partialDeps)
{
return super.serializedSize(partialDeps)
+ KeySerializers.participants.serializedSize(partialDeps.covering);
}
}
static class DepsSerializer extends AbstractDepsSerializer<Deps>
{
public DepsSerializer(boolean preferByTxnId, UnversionedSerializer<Range> tokenRange)
{
super(preferByTxnId, tokenRange);
}
@Override
Deps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, DataInputPlus in) throws IOException
{
return new Deps(keyDeps, rangeDeps);
}
}
@VisibleForTesting
public static class Impl
{
final UnversionedSerializer<Range> tokenRange;
final DepsSerializer<Deps> deps;
final DepsSerializer deps;
final UnversionedSerializer<Deps> nullableDeps;
final DepsSerializer<PartialDeps> partialDeps;
final PartialDepsSerializer partialDeps;
final PartialDepsSerializer partialDepsById;
final UnversionedSerializer<PartialDeps> nullablePartialDeps;
public Impl(UnversionedSerializer<Range> tokenRange)
{
this.tokenRange = tokenRange;
this.deps = new DepsSerializer<>(tokenRange)
{
@Override
Deps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, DataInputPlus in)
{
return new Deps(keyDeps, rangeDeps);
}
};
this.deps = new DepsSerializer(false, tokenRange);
this.nullableDeps = NullableSerializer.wrap(deps);
this.partialDeps = new DepsSerializer<>(tokenRange)
{
@Override
PartialDeps deserialize(KeyDeps keyDeps, RangeDeps rangeDeps, DataInputPlus in) throws IOException
{
Participants<?> covering = KeySerializers.participants.deserialize(in);
return new PartialDeps(covering, keyDeps, rangeDeps);
}
@Override
public void serialize(PartialDeps partialDeps, DataOutputPlus out) throws IOException
{
super.serialize(partialDeps, out);
KeySerializers.participants.serialize(partialDeps.covering, out);
}
@Override
public long serializedSize(PartialDeps partialDeps)
{
return super.serializedSize(partialDeps)
+ KeySerializers.participants.serializedSize(partialDeps.covering);
}
};
this.partialDeps = new PartialDepsSerializer(false, tokenRange);
this.partialDepsById = new PartialDepsSerializer(true, tokenRange);
this.nullablePartialDeps = NullableSerializer.wrap(partialDeps);
}
}
}

189
src/java/org/apache/cassandra/service/accord/serializers/KeySerializers.java
View File

@ -73,16 +73,17 @@ public class KeySerializers
public static final AccordSearchableKeySerializer<RoutingKey> routingKey;
public static final UnversionedSerializer<RoutingKey> nullableRoutingKey;
public static final AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys;
public static final AbstractKeyRoutablesSerializer<RoutingKeys> routingKeys;
public static final UnversionedSerializer<Keys> keys;
public static final AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute;
public static final AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute;
public static final AbstractKeyRoutablesSerializer<PartialKeyRoute> partialKeyRoute;
public static final AbstractKeyRoutablesSerializer<FullKeyRoute> fullKeyRoute;
public static final UnversionedSerializer<Range> range;
public static final AbstractRangesSerializer<Ranges> ranges;
public static final AbstractRangesSerializer<PartialRangeRoute> partialRangeRoute;
public static final AbstractRangesSerializer<FullRangeRoute> fullRangeRoute;
public static final AbstractRangesSerializer<Range[]> rangeArray;
public static final AbstractRangeRoutablesSerializer<Ranges> ranges;
public static final AbstractRangeRoutablesSerializer<PartialRangeRoute> partialRangeRoute;
public static final AbstractRangeRoutablesSerializer<FullRangeRoute> fullRangeRoute;
public static final AbstractRoutablesSerializer<Route<?>> route;
public static final UnversionedSerializer<Route<?>> nullableRoute;
@ -109,6 +110,7 @@ public class KeySerializers
fullKeyRoute = impl.fullKeyRoute;
range = impl.range;
rangeArray = impl.rangeArray;
ranges = impl.ranges;
partialRangeRoute = impl.partialRangeRoute;
fullRangeRoute = impl.fullRangeRoute;
@ -131,16 +133,17 @@ public class KeySerializers
final AccordSearchableKeySerializer<RoutingKey> routingKey;
final UnversionedSerializer<RoutingKey> nullableRoutingKey;
final AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys;
final AbstractKeyRoutablesSerializer<RoutingKeys> routingKeys;
final UnversionedSerializer<Keys> keys;
final AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute;
final AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute;
final AbstractKeyRoutablesSerializer<PartialKeyRoute> partialKeyRoute;
final AbstractKeyRoutablesSerializer<FullKeyRoute> fullKeyRoute;
final UnversionedSerializer<Range> range;
final AbstractRangesSerializer<Ranges> ranges;
final AbstractRangesSerializer<PartialRangeRoute> partialRangeRoute;
final AbstractRangesSerializer<FullRangeRoute> fullRangeRoute;
final AbstractRangesSerializer<Range[]> rangeArray;
final AbstractRangeRoutablesSerializer<Ranges> ranges;
final AbstractRangeRoutablesSerializer<PartialRangeRoute> partialRangeRoute;
final AbstractRangeRoutablesSerializer<FullRangeRoute> fullRangeRoute;
final AbstractRoutablesSerializer<Route<?>> route;
final UnversionedSerializer<Route<?>> nullableRoute;
@ -169,7 +172,7 @@ public class KeySerializers
this.range = range;
this.nullableRoutingKey = NullableSerializer.wrap(routingKey);
this.routingKeys = new AbstractSearchableRoutingKeysSerializer<>(routingKey)
this.routingKeys = new AbstractKeyRoutablesSerializer<>()
{
@Override RoutingKeys deserialize(DataInputPlus in, RoutingKey[] keys)
{
@ -185,7 +188,7 @@ public class KeySerializers
}
};
this.partialKeyRoute = new AbstractKeyRouteSerializer<>(routingKey)
this.partialKeyRoute = new AbstractKeyRouteSerializer<>()
{
@Override
PartialKeyRoute construct(RoutingKey homeKey, RoutingKey[] keys)
@ -194,7 +197,7 @@ public class KeySerializers
}
};
this.fullKeyRoute = new AbstractKeyRouteSerializer<>(routingKey)
this.fullKeyRoute = new AbstractKeyRouteSerializer<>()
{
@Override
FullKeyRoute construct(RoutingKey homeKey, RoutingKey[] keys)
@ -203,7 +206,7 @@ public class KeySerializers
}
};
this.ranges = new AbstractRangesSerializer<>()
this.ranges = new AbstractRangeRoutablesSerializer<>()
{
@Override
public Ranges deserialize(DataInputPlus in, Range[] ranges)
@ -212,6 +215,12 @@ public class KeySerializers
}
};
this.rangeArray = new AbstractRangesSerializer<>()
{
@Override Range[] getArray(Range[] ranges) { return ranges; }
@Override public Range[] deserialize(DataInputPlus in, Range[] ranges) { return ranges; }
};
this.partialRangeRoute = new AbstractRangeRouteSerializer<>()
{
@Override
@ -250,20 +259,20 @@ public class KeySerializers
public static class AbstractRoutablesSerializer<RS extends Unseekables<?>> implements UnversionedSerializer<RS>
{
final TinyEnumSet<UnseekablesKind> permitted;
final AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys;
final AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute;
final AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute;
final AbstractRangesSerializer<Ranges> ranges;
final AbstractRangesSerializer<PartialRangeRoute> partialRangeRoute;
final AbstractRangesSerializer<FullRangeRoute> fullRangeRoute;
final AbstractKeyRoutablesSerializer<RoutingKeys> routingKeys;
final AbstractKeyRoutablesSerializer<PartialKeyRoute> partialKeyRoute;
final AbstractKeyRoutablesSerializer<FullKeyRoute> fullKeyRoute;
final AbstractRangeRoutablesSerializer<Ranges> ranges;
final AbstractRangeRoutablesSerializer<PartialRangeRoute> partialRangeRoute;
final AbstractRangeRoutablesSerializer<FullRangeRoute> fullRangeRoute;
protected AbstractRoutablesSerializer(TinyEnumSet<UnseekablesKind> permitted,
AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys,
AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute,
AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute,
AbstractRangesSerializer<Ranges> ranges,
AbstractRangesSerializer<PartialRangeRoute> partialRangeRoute,
AbstractRangesSerializer<FullRangeRoute> fullRangeRoute)
AbstractKeyRoutablesSerializer<RoutingKeys> routingKeys,
AbstractKeyRoutablesSerializer<PartialKeyRoute> partialKeyRoute,
AbstractKeyRoutablesSerializer<FullKeyRoute> fullKeyRoute,
AbstractRangeRoutablesSerializer<Ranges> ranges,
AbstractRangeRoutablesSerializer<PartialRangeRoute> partialRangeRoute,
AbstractRangeRoutablesSerializer<FullRangeRoute> fullRangeRoute)
{
this.permitted = permitted;
this.routingKeys = routingKeys;
@ -527,6 +536,18 @@ public class KeySerializers
}
}
@Override
public void skip(DataInputPlus in) throws IOException
{
byte b = in.readByte();
switch (b)
{
default: throw new IOException("Corrupted input: expected byte 1 or 2, received " + b);
case 0: PartitionKey.serializer.skip(in); break;
case 1: TokenRange.serializer.skip(in); break;
}
}
@Override
public long serializedSize(Seekable seekable)
{
@ -544,9 +565,9 @@ public class KeySerializers
public static class AbstractSeekablesSerializer implements UnversionedSerializer<Seekables<?, ?>>
{
final UnversionedSerializer<Keys> keys;
final AbstractRangesSerializer<Ranges> ranges;
final AbstractRangeRoutablesSerializer<Ranges> ranges;
public AbstractSeekablesSerializer(UnversionedSerializer<Keys> keys, AbstractRangesSerializer<Ranges> ranges)
public AbstractSeekablesSerializer(UnversionedSerializer<Keys> keys, AbstractRangeRoutablesSerializer<Ranges> ranges)
{
this.keys = keys;
this.ranges = ranges;
@ -646,7 +667,7 @@ public class KeySerializers
// this serializer is designed to permits using the collection in its serialized form with minimal in-memory state.
// it also saves some memory by avoiding duplicating prefixes (which happens to also assist faster lookups)
public abstract static class AbstractSearchableSerializer<K extends RoutableKey, R extends Routable, RS extends Routables<R>> extends IVersionedWithKeysSerializer.AbstractWithKeysSerializer implements UnversionedSerializer<RS>
public abstract static class AbstractSearchableSerializer<R extends Routable, RS> extends IVersionedWithKeysSerializer.AbstractWithKeysSerializer implements UnversionedSerializer<RS>
{
final IntFunction<R[]> allocate;
@ -675,29 +696,35 @@ public class KeySerializers
abstract int fixedKeyLengthForPrefix(Object prefix);
abstract int serializedSizeWithoutPrefix(R routable);
abstract void serializeWithoutPrefixOrLength(R routable, DataOutputPlus out) throws IOException;
abstract void serializeOffsets(RS unseekables, int start, int end, DataOutputPlus out) throws IOException;
abstract void serializeOffsets(R[] keys, int start, int end, DataOutputPlus out) throws IOException;
abstract R deserializeWithPrefix(Object prefix, int length, DataInputPlus in) throws IOException;
abstract R deserializeWithPrefix(Object prefix, int lengthIndex, int[] lengths, DataInputPlus in) throws IOException;
abstract R[] getArray(RS routables);
abstract RS deserialize(DataInputPlus in, R[] keys) throws IOException;
@Override
public long serializedSize(RS routables)
{
int count = routables.size();
return serializedArraySize(getArray(routables));
}
protected long serializedArraySize(R[] rs)
{
int count = rs.length;
long size = TypeSizes.sizeofUnsignedVInt(count);
if (count == 0)
return size;
Object prefix = routables.get(0).prefix();
Object prefix = rs[0].prefix();
int prefixStart = 0;
for (int i = 1 ; i <= count ; ++i)
{
Object nextPrefix = null;
if (i < count)
{
nextPrefix = routables.get(i).prefix();
nextPrefix = rs[i].prefix();
if (Objects.equals(prefix, nextPrefix))
continue;
}
@ -708,7 +735,7 @@ public class KeySerializers
if (fixedLength < 0)
{
size += 4L * recordCountToLengthCount(i - prefixStart);
size += serializedSizeOfKeysWithoutPrefix(routables, prefixStart, i);
size += serializedSizeOfKeysWithoutPrefix(rs, prefixStart, i);
}
else
{
@ -721,27 +748,27 @@ public class KeySerializers
return size;
}
public long serializedSubsetSize(RS keys, Routables<?> superset)
{
return serializedSubsetSizeInternal(keys, superset);
}
@Override
public void serialize(RS keys, DataOutputPlus out) throws IOException
{
int size = keys.size();
serializeArray(getArray(keys), out);
}
public void serializeArray(R[] rs, DataOutputPlus out) throws IOException
{
int size = rs.length;
out.writeUnsignedVInt32(size);
if (size == 0)
return;
Object prefix = keys.get(0).prefix();
Object prefix = rs[0].prefix();
int prefixStart = 0;
for (int i = 1 ; i <= size ; ++i)
{
Object nextPrefix = null;
if (i < size)
{
nextPrefix = keys.get(i).prefix();
nextPrefix = rs[i].prefix();
if (Objects.equals(prefix, nextPrefix))
continue;
}
@ -750,30 +777,25 @@ public class KeySerializers
serializePrefix(prefix, out);
int fixedLength = fixedKeyLengthForPrefix(prefix);
if (fixedLength < 0)
serializeOffsets(keys, prefixStart, i, out);
serializeKeysWithoutPrefix(keys, prefixStart, i, out);
serializeOffsets(rs, prefixStart, i, out);
serializeKeysWithoutPrefix(rs, prefixStart, i, out);
prefixStart = i;
prefix = nextPrefix;
}
}
private long serializedSizeOfKeysWithoutPrefix(RS keys, int start, int end)
private long serializedSizeOfKeysWithoutPrefix(R[] keys, int start, int end)
{
long size = 0;
for (int i = start; i < end; ++i)
size += serializedSizeWithoutPrefix(keys.get(i));
size += serializedSizeWithoutPrefix(keys[i]);
return size;
}
private void serializeKeysWithoutPrefix(RS keys, int start, int end, DataOutputPlus out) throws IOException
private void serializeKeysWithoutPrefix(R[] rs, int start, int end, DataOutputPlus out) throws IOException
{
for (int i = start; i < end; ++i)
serializeWithoutPrefixOrLength(keys.get(i), out);
}
public void serializeSubset(RS keys, Routables<?> superset, DataOutputPlus out) throws IOException
{
serializeSubsetInternal(keys, superset, out);
serializeWithoutPrefixOrLength(rs[i], out);
}
public void skip(DataInputPlus in) throws IOException
@ -863,13 +885,19 @@ public class KeySerializers
// this serializer is designed to permits using the collection in its serialized form with minimal in-memory state.
// it also saves some memory by avoiding duplicating prefixes (which happens to also assist faster lookups)
public abstract static class AbstractSearchableRoutingKeysSerializer<KS extends AbstractUnseekableKeys> extends AbstractSearchableSerializer<RoutingKey, RoutingKey, KS> implements UnversionedSerializer<KS>
public abstract static class AbstractKeyRoutablesSerializer<KS extends AbstractUnseekableKeys> extends AbstractSearchableSerializer<RoutingKey, KS> implements UnversionedSerializer<KS>
{
public AbstractSearchableRoutingKeysSerializer(AccordSearchableKeySerializer<RoutingKey> serializer)
public AbstractKeyRoutablesSerializer()
{
super(RoutingKey[]::new);
}
@Override
RoutingKey[] getArray(KS keys)
{
return keys.unsafeKeys();
}
@Override
final int fixedKeyLengthForPrefix(Object prefix)
{
@ -895,12 +923,12 @@ public class KeySerializers
}
@Override
final void serializeOffsets(KS keys, int startIndex, int endIndex, DataOutputPlus out) throws IOException
final void serializeOffsets(RoutingKey[] keys, int startIndex, int endIndex, DataOutputPlus out) throws IOException
{
int endOffset = 0;
for (int i = startIndex; i < endIndex; ++i)
{
endOffset += serializedSizeWithoutPrefix(keys.get(i));
endOffset += serializedSizeWithoutPrefix(keys[i]);
out.writeInt(endOffset);
}
}
@ -922,13 +950,23 @@ public class KeySerializers
RoutingKey[] keys = deserializeSubset(superset, in, (ks, s) -> ks == null ? s.unsafeKeys() : ks, RoutingKey[]::new);
return deserialize(in, keys);
}
public long serializedSubsetSize(KS keys, Routables<?> superset)
{
return serializedSubsetSizeInternal(keys, superset);
}
public void serializeSubset(KS keys, Routables<?> superset, DataOutputPlus out) throws IOException
{
serializeSubsetInternal(keys, superset, out);
}
}
public abstract static class AbstractKeyRouteSerializer<KS extends KeyRoute> extends AbstractSearchableRoutingKeysSerializer<KS>
public abstract static class AbstractKeyRouteSerializer<KS extends KeyRoute> extends AbstractKeyRoutablesSerializer<KS>
{
public AbstractKeyRouteSerializer(AccordSearchableKeySerializer<RoutingKey> serializer)
public AbstractKeyRouteSerializer()
{
super(serializer);
super();
}
abstract KS construct(RoutingKey homeKey, RoutingKey[] keys);
@ -1006,7 +1044,7 @@ public class KeySerializers
}
}
public abstract static class AbstractRangesSerializer<RS extends AbstractRanges> extends AbstractSearchableSerializer<RoutingKey, Range, RS> implements UnversionedSerializer<RS>
public abstract static class AbstractRangesSerializer<RS> extends AbstractSearchableSerializer<Range, RS> implements UnversionedSerializer<RS>
{
public AbstractRangesSerializer()
{
@ -1040,12 +1078,12 @@ public class KeySerializers
}
@Override
final void serializeOffsets(RS ranges, int startIndex, int endIndex, DataOutputPlus out) throws IOException
final void serializeOffsets(Range[] ranges, int startIndex, int endIndex, DataOutputPlus out) throws IOException
{
int endOffset = 0;
for (int i = startIndex; i < endIndex; ++i)
{
Range r = ranges.get(i);
Range r = ranges[i];
endOffset += routingKey.serializedSizeWithoutPrefix(r.start());
out.writeInt(endOffset);
endOffset += routingKey.serializedSizeWithoutPrefix(r.end());
@ -1068,6 +1106,25 @@ public class KeySerializers
RoutingKey end = routingKey.deserializeWithPrefix(prefix, lengths[lengthIndex * 2 + 1], in);
return start.rangeFactory().newRange(start, end);
}
}
public abstract static class AbstractRangeRoutablesSerializer<RS extends AbstractRanges> extends AbstractRangesSerializer<RS> implements UnversionedSerializer<RS>
{
@Override
Range[] getArray(RS ranges)
{
return ranges.unsafeRanges();
}
public long serializedSubsetSize(RS ranges, Routables<?> superset)
{
return serializedSubsetSizeInternal(ranges, superset);
}
public void serializeSubset(RS ranges, Routables<?> superset, DataOutputPlus out) throws IOException
{
serializeSubsetInternal(ranges, superset, out);
}
public RS deserializeSubset(AbstractRanges superset, DataInputPlus in) throws IOException
{
@ -1076,7 +1133,7 @@ public class KeySerializers
}
}
public abstract static class AbstractRangeRouteSerializer<RS extends RangeRoute> extends AbstractRangesSerializer<RS>
public abstract static class AbstractRangeRouteSerializer<RS extends RangeRoute> extends AbstractRangeRoutablesSerializer<RS>
{
public AbstractRangeRouteSerializer()
{

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

@ -46,23 +46,31 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.service.accord.serializers.CommandSerializers.ExecuteAtSerializer;
import org.apache.cassandra.service.accord.serializers.TxnRequestSerializer.WithUnsyncedSerializer;
import org.apache.cassandra.utils.vint.VIntCoding;
import static accord.messages.BeginRecovery.RecoverReply.Kind.Ok;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
public class RecoverySerializers
{
static final int HAS_ROUTE = 0x1;
static final int HAS_EXECUTE_AT_EPOCH = 0x2;
static final int IS_FAST_PATH_DECIDED = 0x4;
static final int SIZE_OF_FLAGS = VIntCoding.computeUnsignedVIntSize(HAS_ROUTE | HAS_EXECUTE_AT_EPOCH | IS_FAST_PATH_DECIDED);
public static final IVersionedSerializer<BeginRecovery> request = new WithUnsyncedSerializer<BeginRecovery>()
{
@Override
public void serializeBody(BeginRecovery recover, DataOutputPlus out, Version version) throws IOException
{
CommandSerializers.partialTxn.serialize(recover.partialTxn, out, version);
int flags = (recover.route != null ? HAS_ROUTE : 0)
| (recover.executeAtOrTxnIdEpoch != recover.txnId.epoch() ? HAS_EXECUTE_AT_EPOCH : 0)
| (recover.isFastPathDecided ? IS_FAST_PATH_DECIDED : 0);
CommandSerializers.ballot.serialize(recover.ballot, out);
serializeNullable(recover.route, out, KeySerializers.fullRoute);
out.writeUnsignedVInt(recover.executeAtOrTxnIdEpoch - recover.txnId.epoch());
out.writeUnsignedVInt32(flags);
if (recover.route != null)
KeySerializers.fullRoute.serialize(recover.route, out);
if (0 != (flags & HAS_EXECUTE_AT_EPOCH))
out.writeUnsignedVInt(recover.executeAtOrTxnIdEpoch - recover.txnId.epoch());
}
@Override
@ -70,9 +78,15 @@ public class RecoverySerializers
{
PartialTxn partialTxn = CommandSerializers.partialTxn.deserialize(in, version);
Ballot ballot = CommandSerializers.ballot.deserialize(in);
@Nullable FullRoute<?> route = deserializeNullable(in, KeySerializers.fullRoute);
long executeAtOrTxnIdEpoch = in.readUnsignedVInt32() + txnId.epoch();
return BeginRecovery.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, partialTxn, ballot, route, executeAtOrTxnIdEpoch);
int flags = in.readUnsignedVInt32();
FullRoute<?> route = null;
if (0 != (flags & HAS_ROUTE))
route = KeySerializers.fullRoute.deserialize(in);
long executeAtOrTxnIdEpoch = txnId.epoch();
if (0 != (flags & HAS_EXECUTE_AT_EPOCH))
executeAtOrTxnIdEpoch += in.readUnsignedVInt32();
boolean isFastPathDecided = 0 != (flags & IS_FAST_PATH_DECIDED);
return BeginRecovery.SerializationSupport.create(txnId, scope, waitForEpoch, minEpoch, partialTxn, ballot, route, executeAtOrTxnIdEpoch, isFastPathDecided);
}
@Override
@ -80,8 +94,9 @@ public class RecoverySerializers
{
return CommandSerializers.partialTxn.serializedSize(recover.partialTxn, version)
+ CommandSerializers.ballot.serializedSize(recover.ballot)
+ serializedNullableSize(recover.route, KeySerializers.fullRoute)
+ TypeSizes.sizeofUnsignedVInt(recover.executeAtOrTxnIdEpoch - recover.txnId.epoch());
+ SIZE_OF_FLAGS
+ (recover.route == null ? 0 : KeySerializers.fullRoute.serializedSize(recover.route))
+ (recover.executeAtOrTxnIdEpoch == recover.txnId.epoch() ? 0 : TypeSizes.sizeofUnsignedVInt(recover.executeAtOrTxnIdEpoch - recover.txnId.epoch()));
}
};

145
src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java Normal file
View File

@ -0,0 +1,145 @@
/*
* 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.serializers;
import java.io.IOException;
import accord.utils.BitUtils;
import accord.utils.Invariants;
import net.nicoulaj.compilecommand.annotations.Inline;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
/**
* A set of simple utilities to quickly serialize/deserialize arrays/lists of values that each require <= 64 bits to represent.
* These are packed into an "array" of fixed bit width, so that the total size consumed is ceil((bits*elements)/8).
* This can (in future) be read directly without deserialization, by indexing into the byte stream directly.
*/
public class SerializePacked
{
public static void serializePackedInts(int[] vs, int from, int to, int max, DataOutputPlus out) throws IOException
{
serializePacked((in, i) -> in[i], vs, from, to, max, out);
}
public static void deserializePackedInts(int[] vs, int from, int to, int max, DataInputPlus in) throws IOException
{
deserializePacked((out, i, v) -> out[i] = (int)v, vs, from, to, max, in);
}
public static long serializedPackedIntsSize(int[] vs, int from, int to, int max)
{
return serializedPackedSize(to - from, max);
}
public interface SerializeAdapter<In>
{
long get(In in, int i);
}
@Inline
public static <In> void serializePacked(SerializeAdapter<In> adapter, In in, int from, int to, long max, DataOutputPlus out) throws IOException
{
int bitsPerEntry = BitUtils.numberOfBitsToRepresent(max);
if (bitsPerEntry == 0)
return;
long buffer = 0L;
int bufferCount = 0;
for (int i = from; i < to; i++)
{
long v = adapter.get(in, i);
Invariants.require(v <= max);
buffer |= v << bufferCount;
bufferCount = bufferCount + bitsPerEntry;
if (bufferCount >= 64)
{
out.writeLong(buffer);
bufferCount -= 64;
buffer = v >>> (bitsPerEntry - bufferCount);
}
}
if (bufferCount > 0)
out.writeLeastSignificantBytes(buffer, (bufferCount + 7) / 8);
}
public interface DeserializeAdapter<Out>
{
void accept(Out out, int i, long v);
}
@Inline
public static <Out> void deserializePacked(DeserializeAdapter<Out> consumer, Out out, int from, int to, long max, DataInputPlus in) throws IOException
{
int bitsPerEntry = BitUtils.numberOfBitsToRepresent(max);
if (bitsPerEntry == 0)
{
for (int i = from; i < to ; ++i)
consumer.accept(out, i, 0);
return;
}
long mask = -1L >>> (64 - bitsPerEntry);
int remainingBytes = (bitsPerEntry * (to - from) + 7) / 8;
long buffer = 0L;
int bufferCount = 0;
for (int i = from; i < to; i++)
{
long v = buffer & mask;
if (bufferCount >= bitsPerEntry)
{
bufferCount -= bitsPerEntry;
buffer >>>= bitsPerEntry;
}
else
{
int newBufferCount;
if (remainingBytes >= 8)
{
buffer = in.readLong();
newBufferCount = 64;
remainingBytes -= 8;
}
else
{
Invariants.require(remainingBytes > 0);
newBufferCount = remainingBytes * 8;
buffer = in.readLeastSignificantBytes(remainingBytes);
remainingBytes = 0;
}
int readExtra = bitsPerEntry - bufferCount;
long extraBits = buffer & (mask >>> bufferCount);
v |= extraBits << bufferCount;
bufferCount = newBufferCount - readExtra;
buffer >>>= readExtra;
}
Invariants.require(v <= max);
consumer.accept(out, i, v);
}
}
public static long serializedPackedSize(int count, long max)
{
return serializedPackedBitsSize(count, BitUtils.numberOfBitsToRepresent(max));
}
public static long serializedPackedBitsSize(int count, int bitsPerEntry)
{
return ((long)bitsPerEntry * count + 7)/8;
}
}

11
src/java/org/apache/cassandra/service/accord/serializers/TableMetadatasAndKeys.java
View File

@ -134,6 +134,11 @@ public class TableMetadatasAndKeys extends IVersionedWithKeysSerializer.Abstract
return (Keys)deserializeSubsetInternal(this.keys, in);
}
public void skipKeys(DataInputPlus in) throws IOException
{
skipSubsetInternal(this.keys.size(), in);
}
public void serializeSeekable(Seekable seekable, DataOutputPlus out) throws IOException
{
int index = keys.indexOf(seekable);
@ -162,6 +167,12 @@ public class TableMetadatasAndKeys extends IVersionedWithKeysSerializer.Abstract
return key;
}
public void skipSeekable(DataInputPlus in) throws IOException
{
int offset = in.readUnsignedVInt32();
if (offset <= 0) KeySerializers.seekable.skip(in);
}
public PartitionKey deserializeKey(DataInputPlus in) throws IOException
{
int offset = in.readUnsignedVInt32();

17
src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java
View File

@ -20,7 +20,7 @@ package org.apache.cassandra.service.accord.serializers;
import java.io.IOException;
import accord.impl.CommandChange.WaitingOnProvider;
import accord.impl.CommandChange.WaitingOnBitSets;
import accord.local.Command;
import accord.local.Command.WaitingOn;
import accord.primitives.PartialDeps;
@ -57,21 +57,18 @@ public class WaitingOnSerializer
}
}
public static final class Provider implements WaitingOnProvider
public static final class WaitingOnBitSetsAndLength extends WaitingOnBitSets
{
final ImmutableBitSet waitingOn, appliedOrInvalidated;
final int waitingOnLength, appliedOrInvalidatedLength;
public Provider(ImmutableBitSet waitingOn, ImmutableBitSet appliedOrInvalidated, int waitingOnLength, int appliedOrInvalidatedLength)
public WaitingOnBitSetsAndLength(ImmutableBitSet waitingOn, ImmutableBitSet appliedOrInvalidated, int waitingOnLength, int appliedOrInvalidatedLength)
{
this.waitingOn = waitingOn;
this.appliedOrInvalidated = appliedOrInvalidated;
super(waitingOn, appliedOrInvalidated);
this.waitingOnLength = waitingOnLength;
this.appliedOrInvalidatedLength = appliedOrInvalidatedLength;
}
@Override
public WaitingOn provide(TxnId txnId, PartialDeps deps, Timestamp executeAtLeast, long uniqueHlc)
public WaitingOn construct(PartialDeps deps, Timestamp executeAtLeast, long uniqueHlc)
{
Invariants.nonNull(deps);
RoutingKeys keys = deps.keyDeps.keys();
@ -98,7 +95,7 @@ public class WaitingOnSerializer
}
}
public static WaitingOnProvider deserializeProvider(TxnId txnId, DataInputPlus in) throws IOException
public static WaitingOnBitSets deserializeBitSets(TxnId txnId, DataInputPlus in) throws IOException
{
ImmutableBitSet waitingOn, appliedOrInvalidated = null;
int waitingOnLength, appliedOrInvalidatedLength = 0;
@ -110,7 +107,7 @@ public class WaitingOnSerializer
appliedOrInvalidated = deserialize(appliedOrInvalidatedLength, in);
}
return new Provider(waitingOn, appliedOrInvalidated, waitingOnLength, appliedOrInvalidatedLength);
return new WaitingOnBitSetsAndLength(waitingOn, appliedOrInvalidated, waitingOnLength, appliedOrInvalidatedLength);
}
public static void skip(TxnId txnId, DataInputPlus in) throws IOException

7
src/java/org/apache/cassandra/service/accord/txn/AccordUpdate.java
View File

@ -121,6 +121,13 @@ public abstract class AccordUpdate implements Update
return (AccordUpdate) serializerFor(kind).deserialize(tablesAndKeys, in, version);
}
@Override
public void skip(TableMetadatasAndKeys tablesAndKeys, DataInputPlus in, Version version) throws IOException
{
Kind kind = Kind.valueOf(in.readByte());
serializerFor(kind).skip(tablesAndKeys, in, version);
}
@Override
public long serializedSize(AccordUpdate update, TableMetadatasAndKeys tablesAndKeys, Version version)
{

9
src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java
View File

@ -73,6 +73,7 @@ import static com.google.common.base.Preconditions.checkState;
import static org.apache.cassandra.io.util.DataOutputBuffer.scratchBuffer;
import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.skipWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
public class TxnNamedRead extends AbstractParameterisedVersionedSerialized<ReadCommand, TableMetadatas>
@ -431,6 +432,14 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized<ReadC
return new TxnNamedRead(name, key, bytes);
}
@Override
public void skip(TableMetadatasAndKeys tablesAndKeys, DataInputPlus in, Version version) throws IOException
{
in.readInt();
tablesAndKeys.skipSeekable(in);
if (in.readByte() != 1) skipWithVIntLength(in);
}
@Override
public long serializedSize(TxnNamedRead read, TableMetadatasAndKeys tablesAndKeys, Version version)
{

18
src/java/org/apache/cassandra/service/accord/txn/TxnRead.java
View File

@ -71,6 +71,7 @@ import static org.apache.cassandra.service.accord.txn.TxnData.txnDataName;
import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
import static org.apache.cassandra.utils.ArraySerializers.skipArray;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializedNullableSize;
@ -420,6 +421,23 @@ public class TxnRead extends AbstractKeySorted<TxnNamedRead> implements Read
}
}
public void skip(TableMetadatasAndKeys tablesAndKeys, DataInputPlus in, Version version) throws IOException
{
byte type = in.readByte();
switch (type)
{
default:
throw new IllegalStateException("Unhandled type " + type);
case TYPE_EMPTY_KEY:
case TYPE_EMPTY_RANGE:
return;
case TYPE_NOT_EMPTY:
skipArray(tablesAndKeys, in, version, TxnNamedRead.serializer);
deserializeNullable(in, consistencyLevelSerializer);
}
}
@Override
public TxnRead deserialize(TableMetadatasAndKeys tablesAndKeys, DataInputPlus in, Version version) throws IOException
{

12
src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java
View File

@ -66,8 +66,10 @@ import static org.apache.cassandra.service.accord.AccordSerializers.consistencyL
import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
import static org.apache.cassandra.utils.ArraySerializers.skipArray;
import static org.apache.cassandra.utils.ByteBufferUtil.readWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.serializedSizeWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.skipWithVIntLength;
import static org.apache.cassandra.utils.ByteBufferUtil.writeWithVIntLength;
import static org.apache.cassandra.utils.NullableSerializer.deserializeNullable;
import static org.apache.cassandra.utils.NullableSerializer.serializeNullable;
@ -292,6 +294,16 @@ public class TxnUpdate extends AccordUpdate
return new TxnUpdate(tablesAndKeys.tables, keys, fragments, new SerializedTxnCondition(condition), consistencyLevel, preserveTimestamps ? PreserveTimestamp.yes : PreserveTimestamp.no);
}
@Override
public void skip(TableMetadatasAndKeys tablesAndKeys, DataInputPlus in, Version version) throws IOException
{
in.readByte();
tablesAndKeys.skipKeys(in);
skipWithVIntLength(in);
skipArray(in, ByteBufferUtil.byteBufferSerializer);
deserializeNullable(in, consistencyLevelSerializer);
}
@Override
public long serializedSize(TxnUpdate update, TableMetadatasAndKeys tablesAndKeys, Version version)
{

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

@ -77,6 +77,7 @@ import static org.apache.cassandra.db.rows.DeserializationHelper.Flag.FROM_REMOT
import static org.apache.cassandra.utils.ArraySerializers.deserializeArray;
import static org.apache.cassandra.utils.ArraySerializers.serializeArray;
import static org.apache.cassandra.utils.ArraySerializers.serializedArraySize;
import static org.apache.cassandra.utils.ArraySerializers.skipArray;
public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Write
{
@ -215,6 +216,14 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
return new Update(key, index, bytes);
}
@Override
public void skip(TableMetadatasAndKeys tablesAndKeys, DataInputPlus in, Version version) throws IOException
{
PartitionKey key = tablesAndKeys.deserializeKey(in);
int index = in.readInt();
ByteBufferUtil.skipWithVIntLength(in);
}
@Override
public long serializedSize(Update write, TableMetadatasAndKeys tablesAndKeys, Version version)
{
@ -493,6 +502,14 @@ public class TxnWrite extends AbstractKeySorted<TxnWrite.Update> implements Writ
return new TxnWrite(tables, deserializeArray(new TableMetadatasAndKeys(tables, keys), in, version, Update.serializer, Update[]::new), isConditionMet);
}
@Override
public void skip(Seekables keys, DataInputPlus in, Version version) throws IOException
{
TableMetadatas tables = TableMetadatas.deserializeSelf(in);
BooleanSerializer.serializer.deserialize(in);
skipArray(new TableMetadatasAndKeys(tables, keys), in, version, Update.serializer);
}
@Override
public long serializedSize(TxnWrite write, Seekables keys, Version version)
{

22
src/java/org/apache/cassandra/utils/ArraySerializers.java
View File

@ -69,6 +69,13 @@ public class ArraySerializers
return items;
}
public static <T, P, Version> void skipArray(DataInputPlus in, UnversionedSerializer<T> serializer) throws IOException
{
int size = in.readUnsignedVInt32();
for (int i = 0; i < size; i++)
serializer.skip(in);
}
public static <T> T[] deserializeArray(DataInputPlus in, int version, IVersionedSerializer<T> serializer, IntFunction<T[]> arrayFactory) throws IOException
{
int size = in.readUnsignedVInt32();
@ -87,6 +94,14 @@ public class ArraySerializers
return items;
}
public static <T, Version> void skipArray(DataInputPlus in, Version version, AsymmetricVersionedSerializer<T, ?, Version> serializer) throws IOException
{
int size = in.readUnsignedVInt32();
for (int i = 0; i < size; i++)
serializer.skip(in, version);
}
public static <T, P, Version> T[] deserializeArray(P p, DataInputPlus in, Version version, ParameterisedVersionedSerializer<T, P, Version> serializer, IntFunction<T[]> arrayFactory) throws IOException
{
int size = in.readUnsignedVInt32();
@ -96,6 +111,13 @@ public class ArraySerializers
return items;
}
public static <T, P, Version> void skipArray(P p, DataInputPlus in, Version version, ParameterisedVersionedSerializer<T, P, Version> serializer) throws IOException
{
int size = in.readUnsignedVInt32();
for (int i = 0; i < size; i++)
serializer.skip(p, in, version);
}
public static <T> long serializedArraySize(T[] array, UnversionedSerializer<T> serializer)
{
long size = sizeofUnsignedVInt(array.length);

6
test/unit/org/apache/cassandra/index/accord/AccordIndexUtil.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.index.accord;
import javax.annotation.Nullable;
import accord.local.MaxDecidedRX;
import accord.primitives.TxnId;
import org.apache.cassandra.service.accord.TokenRange;
import org.apache.cassandra.service.accord.api.TokenKey;
@ -40,4 +41,9 @@ public class AccordIndexUtil
{
return "T:" + (txnId == null ? "null" : Long.toString(txnId.hlc()));
}
public static String normalize(@Nullable MaxDecidedRX.DecidedRX decidedRX)
{
return "T:" + (decidedRX == null ? "null" : Long.toString(decidedRX.any.hlc()));
}
}

36
test/unit/org/apache/cassandra/index/accord/RangeMemoryIndexTest.java
View File

@ -29,6 +29,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.local.Node;
import accord.primitives.Routable;
import accord.primitives.Txn;
@ -113,13 +114,14 @@ public class RangeMemoryIndexTest
return TxnRange.next(rs, minKnown, maxKnown, RangeMemoryIndexTest::idFor);
}
private static @Nullable TxnId nextMinDecidedId(RandomSource rs, State state)
private static @Nullable DecidedRX nextDecidedRX(RandomSource rs, State state)
{
if (rs.decide(state.minDecidedIdNull)) return null;
long maxKnown = state.operations;
long minKnown = state.model.isEmpty() ? maxKnown : state.model.minTime();
if (minKnown == maxKnown) return idFor(maxKnown);
return idFor(rs.nextLong(minKnown, maxKnown));
TxnId txnId = minKnown == maxKnown ? idFor(maxKnown)
: idFor(rs.nextLong(minKnown, maxKnown));
return new DecidedRX(txnId, txnId);
}
private static DecoratedKey pk(TxnId txnId)
@ -159,11 +161,11 @@ public class RangeMemoryIndexTest
var txnRange = nextTxnRange(rs, state);
byte[] start = OrderedRouteSerializer.serializeTokenOnly(range.start());
byte[] end = OrderedRouteSerializer.serializeTokenOnly(range.end());
@Nullable TxnId minDecidedId = nextMinDecidedId(rs, state);
return new Property.SimpleCommand<>("search(" + normalize(range) + ", " + txnRange + ", " + normalize(minDecidedId) + ')', s2 -> {
@Nullable DecidedRX decidedRX = nextDecidedRX(rs, state);
return new Property.SimpleCommand<>("search(" + normalize(range) + ", " + txnRange + ", " + decidedRX + ')', s2 -> {
TreeSet<TxnId> actual = new TreeSet<>();
state.index.search(STORE, TABLE_ID, start, end, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId, bb -> actual.add(AccordKeyspace.JournalColumns.getJournalKey(bb).id));
var expected = state.model.search(range, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId);
state.index.search(STORE, TABLE_ID, start, end, txnRange.minTxnId, txnRange.maxTxnId, decidedRX, bb -> actual.add(AccordKeyspace.JournalColumns.getJournalKey(bb).id));
var expected = state.model.search(range, txnRange.minTxnId, txnRange.maxTxnId, decidedRX);
Assertions.assertThat(actual).isEqualTo(expected);
});
}
@ -173,11 +175,11 @@ public class RangeMemoryIndexTest
TokenKey key = tokenKey(rs.nextLong(MIN_TOKEN, MAX_TOKEN + 1));
var txnRange = nextTxnRange(rs, state);
var start = OrderedRouteSerializer.serializeTokenOnly(key);
@Nullable TxnId minDecidedId = nextMinDecidedId(rs, state);
return new Property.SimpleCommand<>("search(" + normalize(key) + ", " + txnRange + ", " + normalize(minDecidedId) + ')', s2 -> {
@Nullable DecidedRX decidedRX = nextDecidedRX(rs, state);
return new Property.SimpleCommand<>("search(" + normalize(key) + ", " + txnRange + ", " + decidedRX + ')', s2 -> {
TreeSet<TxnId> actual = new TreeSet<>();
state.index.search(STORE, TABLE_ID, start, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId, bb -> actual.add(AccordKeyspace.JournalColumns.getJournalKey(bb).id));
var expected = state.model.search(key, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId);
state.index.search(STORE, TABLE_ID, start, txnRange.minTxnId, txnRange.maxTxnId, decidedRX, bb -> actual.add(AccordKeyspace.JournalColumns.getJournalKey(bb).id));
var expected = state.model.search(key, txnRange.minTxnId, txnRange.maxTxnId, decidedRX);
Assertions.assertThat(actual).isEqualTo(expected);
});
}
@ -219,23 +221,23 @@ public class RangeMemoryIndexTest
return values.isEmpty();
}
public NavigableSet<TxnId> search(TokenRange range, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
public NavigableSet<TxnId> search(TokenRange range, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
return search(r -> r.compareIntersecting(range) == 0, minTxnId, maxTxnId, minDecidedId);
return search(r -> r.compareIntersecting(range) == 0, minTxnId, maxTxnId, decidedRX);
}
public NavigableSet<TxnId> search(TokenKey key, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
public NavigableSet<TxnId> search(TokenKey key, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
return search(r -> r.contains(key), minTxnId, maxTxnId, minDecidedId);
return search(r -> r.contains(key), minTxnId, maxTxnId, decidedRX);
}
public NavigableSet<TxnId> search(Predicate<TokenRange> test, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
public NavigableSet<TxnId> search(Predicate<TokenRange> test, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
NavigableSet<TxnId> result = new TreeSet<>();
for (var value : values)
{
if (value.txnId.compareTo(minTxnId) < 0 || value.txnId.compareTo(maxTxnId) > 0) continue;
if (minDecidedId != null && minDecidedId.compareTo(maxRXId) > 0) continue;
if (decidedRX != null && decidedRX.excludeDecided(maxRXId)) continue;
if (test.test(value.range))
result.add(value.txnId);
}

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

@ -35,6 +35,7 @@ import accord.api.RoutingKey;
import accord.local.CommandStores;
import accord.local.CommandStores.RangesForEpoch;
import accord.local.DurableBefore;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.local.Node;
import accord.local.RedundantBefore;
import accord.local.StoreParticipants;
@ -101,7 +102,6 @@ import static accord.local.RedundantStatus.SomeStatus.NONE;
import static accord.utils.Property.commands;
import static accord.utils.Property.stateful;
import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner;
import static org.apache.cassandra.index.accord.AccordIndexUtil.normalize;
import static org.apache.cassandra.schema.SchemaConstants.ACCORD_KEYSPACE_NAME;
public class RouteIndexTest extends CQLTester
@ -179,8 +179,8 @@ public class RouteIndexTest extends CQLTester
long start = range.start().isMin() ? Long.MIN_VALUE : ((LongToken) range.start().token()).token;
long end = range.end().isMax() ? Long.MAX_VALUE : ((LongToken) range.end().token()).token;
long token = 1 + rs.nextLong(start, end);
@Nullable TxnId minDecidedId = state.nextMinDecidedId(rs);
return new KeySearch(storeId, new TokenKey(tableId, new LongToken(token)), state.nextTxnRange(rs), minDecidedId);
@Nullable DecidedRX decidedRX = state.nextDecidedRX(rs);
return new KeySearch(storeId, new TokenKey(tableId, new LongToken(token)), state.nextTxnRange(rs), decidedRX);
}
private static RangeSearch rangeSearchExisting(RandomSource rs, State state)
@ -189,20 +189,20 @@ public class RouteIndexTest extends CQLTester
var tables = state.storeToTableToRangesToTxns.get(storeId);
TableId tableId = rs.pickUnorderedSet(tables.keySet());
var ranges = tables.get(tableId);
@Nullable TxnId minDecidedId = state.nextMinDecidedId(rs);
return new RangeSearch(storeId, selectExistingRange(rs, ranges), state.nextTxnRange(rs), minDecidedId);
@Nullable DecidedRX decidedRX = state.nextDecidedRX(rs);
return new RangeSearch(storeId, selectExistingRange(rs, ranges), state.nextTxnRange(rs), decidedRX);
}
private static Command<State, Sut, ?> rangeSearch(RandomSource rs, State state)
{
@Nullable TxnId minDecidedId = state.nextMinDecidedId(rs);
return new RangeSearch(rs.nextInt(0, state.numStores), state.rangeGen.next(rs), state.nextTxnRange(rs), minDecidedId);
@Nullable DecidedRX decidedRX = state.nextDecidedRX(rs);
return new RangeSearch(rs.nextInt(0, state.numStores), state.rangeGen.next(rs), state.nextTxnRange(rs), decidedRX);
}
private static Command<State, Sut, ?> keySearch(RandomSource rs, State state)
{
@Nullable TxnId minDecidedId = state.nextMinDecidedId(rs);
return new KeySearch(rs.nextInt(0, state.numStores), new TokenKey(rs.pick(state.tables), new LongToken(state.tokenGen.nextInt(rs))), state.nextTxnRange(rs), minDecidedId);
@Nullable DecidedRX decidedRX = state.nextDecidedRX(rs);
return new KeySearch(rs.nextInt(0, state.numStores), new TokenKey(rs.pick(state.tables), new LongToken(state.tokenGen.nextInt(rs))), state.nextTxnRange(rs), decidedRX);
}
@Test
@ -270,14 +270,14 @@ public class RouteIndexTest extends CQLTester
private final int storeId;
private final TokenKey key;
private final TxnRange txnRange;
private final @Nullable TxnId minDecidedId;
private final @Nullable DecidedRX decidedRX;
private KeySearch(int storeId, TokenKey key, TxnRange txnRange, @Nullable TxnId minDecidedId)
private KeySearch(int storeId, TokenKey key, TxnRange txnRange, @Nullable DecidedRX decidedRX)
{
this.storeId = storeId;
this.key = key;
this.txnRange = txnRange;
this.minDecidedId = minDecidedId;
this.decidedRX = decidedRX;
}
@Override
@ -290,7 +290,7 @@ public class RouteIndexTest extends CQLTester
Set<TxnId> matches = new HashSet<>();
ranges.searchToken(key, e -> {
TxnId txnId = e.getValue();
if (minDecidedId != null && txnId.is(Txn.Kind.ExclusiveSyncPoint) && minDecidedId.compareTo(txnId) > 0)
if (decidedRX != null && txnId.is(Txn.Kind.ExclusiveSyncPoint) && decidedRX.excludeDecided(txnId))
return;
if (txnRange.includes(txnId))
matches.add(txnId);
@ -302,7 +302,7 @@ public class RouteIndexTest extends CQLTester
public Set<TxnId> run(Sut sut) throws Throwable
{
Set<TxnId> result = new ObjectHashSet<>();
sut.journal.get().rangeSearcher().search(storeId, key, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId).consume(result::add);
sut.journal.get().rangeSearcher().search(storeId, key, txnRange.minTxnId, txnRange.maxTxnId, decidedRX).consume(result::add);
return result;
}
@ -319,7 +319,7 @@ public class RouteIndexTest extends CQLTester
return "KeySearch{" +
"storeId=" + storeId +
", key=" + key +
", minDecidedId=" + normalize(minDecidedId) +
", decidedRX=" + decidedRX +
'}';
}
}
@ -329,14 +329,14 @@ public class RouteIndexTest extends CQLTester
private final int storeId;
private final TokenRange range;
private final TxnRange txnRange;
private final TxnId minDecidedId;
private final DecidedRX decidedRX;
private RangeSearch(int storeId, TokenRange range, TxnRange txnRange, @Nullable TxnId minDecidedId)
private RangeSearch(int storeId, TokenRange range, TxnRange txnRange, DecidedRX decidedRX)
{
this.storeId = storeId;
this.range = range;
this.txnRange = txnRange;
this.minDecidedId = minDecidedId;
this.decidedRX = decidedRX;
}
@Override
@ -349,7 +349,7 @@ public class RouteIndexTest extends CQLTester
Set<TxnId> matches = new HashSet<>();
ranges.search(range, e -> {
TxnId txnId = e.getValue();
if (minDecidedId != null && txnId.is(Txn.Kind.ExclusiveSyncPoint) && minDecidedId.compareTo(txnId) > 0) return;
if (decidedRX != null && txnId.is(Txn.Kind.ExclusiveSyncPoint) && decidedRX.excludeDecided(txnId)) return;
if (txnRange.includes(txnId))
matches.add(txnId);
});
@ -360,7 +360,7 @@ public class RouteIndexTest extends CQLTester
public Set<TxnId> run(Sut sut) throws Throwable
{
Set<TxnId> result = new ObjectHashSet<>();
sut.journal.get().rangeSearcher().search(storeId, range, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId).consume(result::add);
sut.journal.get().rangeSearcher().search(storeId, range, txnRange.minTxnId, txnRange.maxTxnId, decidedRX).consume(result::add);
return result;
}
@ -377,7 +377,7 @@ public class RouteIndexTest extends CQLTester
return "RangeSearch{" +
"storeId=" + storeId +
", range=" + range +
", minDecidedId=" + normalize(minDecidedId) +
", minDecidedId=" + decidedRX +
'}';
}
}
@ -540,13 +540,14 @@ public class RouteIndexTest extends CQLTester
return TxnRange.next(rs, minKnown, maxKnown, hlc -> idFor(Domain.Key, hlc));
}
private @Nullable TxnId nextMinDecidedId(RandomSource rs)
private @Nullable DecidedRX nextDecidedRX(RandomSource rs)
{
if (rs.decide(minDecidedIdNull)) return null;
long maxKnown = hlc;
long minKnown = MIN_TIMESTAMP;
if (minKnown == maxKnown) return idFor(Domain.Range, maxKnown);
return idFor(Domain.Range, rs.nextLong(minKnown, maxKnown));
TxnId txnId = minKnown == maxKnown ? idFor(Domain.Range, maxKnown)
: idFor(Domain.Range, rs.nextLong(minKnown, maxKnown));
return new DecidedRX(txnId, txnId);
}
void insertTxn(int storeId, TxnId txnId, Route<?> route)

42
test/unit/org/apache/cassandra/service/accord/RouteInMemoryIndexTest.java
View File

@ -29,6 +29,7 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import accord.local.MaxDecidedRX.DecidedRX;
import accord.local.Node;
import accord.primitives.FullRangeRoute;
import accord.primitives.Range;
@ -119,13 +120,14 @@ public class RouteInMemoryIndexTest
return TxnRange.next(rs, minKnown, maxKnown, RouteInMemoryIndexTest::idFor);
}
private static @Nullable TxnId nextMinDecidedId(RandomSource rs, State state)
private static @Nullable DecidedRX nextDecidedRX(RandomSource rs, State state)
{
if (rs.decide(state.minDecidedIdNull)) return null;
long maxKnown = state.operations;
long minKnown = state.model.isEmpty() ? maxKnown : state.model.minTime();
if (minKnown == maxKnown) return idFor(maxKnown);
return idFor(rs.nextLong(minKnown, maxKnown));
TxnId txnId = minKnown == maxKnown ? idFor(maxKnown)
: idFor(rs.nextLong(minKnown, maxKnown));
return new DecidedRX(txnId, txnId);
}
private static class State
@ -197,33 +199,33 @@ public class RouteInMemoryIndexTest
{
var range = nextRange(rs);
var txnRange = nextTxnRange(rs, state);
@Nullable TxnId minDecidedId = nextMinDecidedId(rs, state);
return new Property.SimpleCommand<>("Search " + normalize(range) + ", txn_id range " + txnRange + ", minDecidedId " + normalize(minDecidedId), s2 -> s2.assertSearchMatch(range, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId));
@Nullable DecidedRX decidedRX = nextDecidedRX(rs, state);
return new Property.SimpleCommand<>("Search " + normalize(range) + ", txn_id range " + txnRange + ", minDecidedId " + normalize(decidedRX), s2 -> s2.assertSearchMatch(range, txnRange.minTxnId, txnRange.maxTxnId, decidedRX));
}
public static Property.Command<State, Void, ?> keySearch(RandomSource rs, State state)
{
TokenKey key = tokenKey(rs.nextLong(MIN_TOKEN, MAX_TOKEN + 1));
var txnRange = nextTxnRange(rs, state);
@Nullable TxnId minDecidedId = nextMinDecidedId(rs, state);
return new Property.SimpleCommand<>("Search " + normalize(key) + ", txn_id range " + txnRange + ", minDecidedId " + normalize(minDecidedId), s2 -> s2.assertSearchMatch(key, txnRange.minTxnId, txnRange.maxTxnId, minDecidedId));
@Nullable DecidedRX decidedRX = nextDecidedRX(rs, state);
return new Property.SimpleCommand<>("Search " + normalize(key) + ", txn_id range " + txnRange + ", decidedRX " + normalize(decidedRX), s2 -> s2.assertSearchMatch(key, txnRange.minTxnId, txnRange.maxTxnId, decidedRX));
}
private void assertSearchMatch(TokenRange range, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
private void assertSearchMatch(TokenRange range, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
List<TxnId> actual = new ArrayList<>();
index.search(0, range, minTxnId, maxTxnId, minDecidedId).consume(actual::add);
index.search(0, range, minTxnId, maxTxnId, decidedRX).consume(actual::add);
List<TxnId> expected = new ArrayList<>();
model.search(range, minTxnId, maxTxnId, minDecidedId).consume(expected::add);
model.search(range, minTxnId, maxTxnId, decidedRX).consume(expected::add);
Assertions.assertThat(actual).isEqualTo(expected);
}
private void assertSearchMatch(TokenKey key, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
private void assertSearchMatch(TokenKey key, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
List<TxnId> actual = new ArrayList<>();
index.search(0, key, minTxnId, maxTxnId, minDecidedId).consume(actual::add);
index.search(0, key, minTxnId, maxTxnId, decidedRX).consume(actual::add);
List<TxnId> expected = new ArrayList<>();
model.search(key, minTxnId, maxTxnId, minDecidedId).consume(expected::add);
model.search(key, minTxnId, maxTxnId, decidedRX).consume(expected::add);
Assertions.assertThat(actual).isEqualTo(expected);
}
}
@ -276,22 +278,22 @@ public class RouteInMemoryIndexTest
segments.computeIfAbsent(segment, i -> new Segment()).add(range, txnId);
}
public RangeSearcher.Result search(TokenRange range, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
public RangeSearcher.Result search(TokenRange range, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
return search(vrange -> range.compareIntersecting(vrange) == 0, minTxnId, maxTxnId, minDecidedId);
return search(vrange -> range.compareIntersecting(vrange) == 0, minTxnId, maxTxnId, decidedRX);
}
public RangeSearcher.Result search(TokenKey key, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
public RangeSearcher.Result search(TokenKey key, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
return search(range -> range.contains(key), minTxnId, maxTxnId, minDecidedId);
return search(range -> range.contains(key), minTxnId, maxTxnId, decidedRX);
}
public RangeSearcher.Result search(Predicate<TokenRange> test, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId)
public RangeSearcher.Result search(Predicate<TokenRange> test, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX)
{
TreeSet<TxnId> result = new TreeSet<>();
for (var segment: segments.values())
{
if (!RouteIndexFormat.includeByMinDecidedId(minDecidedId, segment.maxRXId)) continue;
if (!RouteIndexFormat.includeByDecidedRX(decidedRX, segment.maxRXId)) continue;
for (var value : segment.values)
{
if (value.txnId.compareTo(minTxnId) < 0 || value.txnId.compareTo(maxTxnId) > 0) continue;
@ -299,7 +301,7 @@ public class RouteInMemoryIndexTest
result.add(value.txnId);
}
}
return new RangeSearcher.DefaultResult(minTxnId, maxTxnId, minDecidedId, CloseableIterator.wrap(result.iterator()));
return new RangeSearcher.DefaultResult(minTxnId, maxTxnId, decidedRX, CloseableIterator.wrap(result.iterator()));
}
void remove(long segment)

2
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java
View File

@ -459,7 +459,7 @@ public class SimulatedAccordCommandStore implements AutoCloseable
{
TxnId txnId = nextTxnId(txn.kind(), txn.keys().domain());
Ballot ballot = Ballot.fromValues(storeService.epoch(), storeService.now(), nodeId);
BeginRecovery br = new BeginRecovery(nodeId, topologies, txnId, null, txn, route, ballot);
BeginRecovery br = new BeginRecovery(nodeId, topologies, txnId, null, false, txn, route, ballot);
return Pair.create(txnId, processAsync(br, safe -> {
var reply = br.apply(safe);

2
test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java
View File

@ -287,7 +287,7 @@ public abstract class SimulatedAccordCommandStoreTestBase extends CQLTester
});
var delay = preAcceptAsync.flatMap(ignore -> AsyncChains.ofCallable(instance.unorderedScheduled, () -> {
Ballot ballot = Ballot.fromValues(instance.storeService.epoch(), instance.storeService.now(), nodeId);
return new BeginRecovery(nodeId, new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, instance.topology), txnId, null, txn, route, ballot);
return new BeginRecovery(nodeId, new Topologies.Single(SizeOfIntersectionSorter.SUPPLIER, instance.topology), txnId, null, false, txn, route, ballot);
}));
var recoverAsync = delay.flatMap(br -> instance.processAsync(br, safe -> {
var reply = br.apply(safe);

4
test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java
View File

@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
import org.junit.BeforeClass;
import org.junit.Test;
import accord.impl.CommandChange;
import accord.local.Command;
import accord.primitives.Deps;
import accord.primitives.KeyDeps;
@ -68,7 +69,8 @@ public class WaitingOnSerializerTest
try (DataInputBuffer buf = new DataInputBuffer(bb, true))
{
PartialDeps deps = new PartialDeps(RoutingKeys.EMPTY, KeyDeps.none(waitingOn.keys), waitingOn.directRangeDeps);
Command.WaitingOn read = WaitingOnSerializer.deserializeProvider(txnId, buf).provide(txnId, deps, null, 0);
CommandChange.WaitingOnBitSets bitSets = WaitingOnSerializer.deserializeBitSets(txnId, buf);
Command.WaitingOn read = new Command.WaitingOn(deps.keyDeps.keys(), deps.rangeDeps, bitSets.waitingOn, bitSets.appliedOrInvalidated);
Assertions.assertThat(read).isEqualTo(waitingOn);
Assertions.assertThat(buf.available()).isEqualTo(0);
}