diff --git a/modules/accord b/modules/accord index a8916a18a0..5cbe8d62f1 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit a8916a18a012063f4da74aaccd39f0b828c99da0 +Subproject commit 5cbe8d62f15cc7d66af2c7686f14b8fa52b1d35b diff --git a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java index 4b09ada872..1480069953 100644 --- a/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java +++ b/src/java/org/apache/cassandra/db/compaction/CompactionIterator.java @@ -1031,7 +1031,7 @@ public class CompactionIterator extends CompactionInfo.Holder implements Unfilte { row = null; modified = false; - builder.clear(); + builder.reset(); } } diff --git a/src/java/org/apache/cassandra/index/accord/MemtableIndex.java b/src/java/org/apache/cassandra/index/accord/MemtableIndex.java index b66820c91f..c225bbe5f0 100644 --- a/src/java/org/apache/cassandra/index/accord/MemtableIndex.java +++ b/src/java/org/apache/cassandra/index/accord/MemtableIndex.java @@ -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 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 onMatch) { memoryIndex.search(storeId, tableId, key, - minTxnId, maxTxnId, minDecidedId, + minTxnId, maxTxnId, decidedRX, onMatch); } } diff --git a/src/java/org/apache/cassandra/index/accord/MemtableIndexManager.java b/src/java/org/apache/cassandra/index/accord/MemtableIndexManager.java index 061d4165fe..359a214bb1 100644 --- a/src/java/org/apache/cassandra/index/accord/MemtableIndexManager.java +++ b/src/java/org/apache/cassandra/index/accord/MemtableIndexManager.java @@ -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 onMatch); void search(int storeId, TableId tableId, byte[] key, - TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, + TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX, Consumer onMatch); } diff --git a/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java b/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java index 01653f1349..2ced2a8ffc 100644 --- a/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java +++ b/src/java/org/apache/cassandra/index/accord/RangeMemoryIndex.java @@ -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> 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> 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 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 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() diff --git a/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java b/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java index de08bbc967..a05e2e8354 100644 --- a/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java +++ b/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java @@ -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 diff --git a/src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java b/src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java index 4b089d067c..849ecd9d70 100644 --- a/src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java +++ b/src/java/org/apache/cassandra/index/accord/RouteJournalIndex.java @@ -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 partitions = search(storeId, key, - minTxnId, maxTxnId, minDecidedId); + minTxnId, maxTxnId, decidedRX); // do SinglePartitionReadCommand per partition return new SearchIterator(command, partitions); } NavigableSet 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 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 partitions = search(storeId, start, end, - minTxnId, maxTxnId, minDecidedId); + minTxnId, maxTxnId, decidedRX); // do SinglePartitionReadCommand per partition return new SearchIterator(command, partitions); } NavigableSet 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 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); } }; diff --git a/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java b/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java index 00cac068f3..67174d7ae0 100644 --- a/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java +++ b/src/java/org/apache/cassandra/index/accord/RouteMemtableIndexManager.java @@ -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 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 onMatch) { liveMemtableIndexMap.values().forEach(m -> m.search(storeId, tableId, key, - minTxnId, maxTxnId, minDecidedId, + minTxnId, maxTxnId, decidedRX, onMatch)); } } diff --git a/src/java/org/apache/cassandra/index/accord/RouteSSTableManager.java b/src/java/org/apache/cassandra/index/accord/RouteSSTableManager.java index e1bb01ff51..af3074be28 100644 --- a/src/java/org/apache/cassandra/index/accord/RouteSSTableManager.java +++ b/src/java/org/apache/cassandra/index/accord/RouteSSTableManager.java @@ -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 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 onMatch) + public synchronized void search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, Consumer 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); } } diff --git a/src/java/org/apache/cassandra/index/accord/SSTableIndex.java b/src/java/org/apache/cassandra/index/accord/SSTableIndex.java index 7041d244e9..8163c05783 100644 --- a/src/java/org/apache/cassandra/index/accord/SSTableIndex.java +++ b/src/java/org/apache/cassandra/index/accord/SSTableIndex.java @@ -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 onMatch) + public void search(Key group, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX, Consumer onMatch) { List 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 onMatch) + public void search(Key group, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX, Consumer onMatch) { List 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; diff --git a/src/java/org/apache/cassandra/index/accord/SSTableManager.java b/src/java/org/apache/cassandra/index/accord/SSTableManager.java index ea3c4de578..f1d80f8983 100644 --- a/src/java/org/apache/cassandra/index/accord/SSTableManager.java +++ b/src/java/org/apache/cassandra/index/accord/SSTableManager.java @@ -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 removed, Iterable added); boolean isIndexComplete(SSTableReader reader); - void search(int storeId, TableId tableId, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, Consumer onMatch); - void search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, Consumer onMatch); + void search(int storeId, TableId tableId, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, Consumer onMatch); + void search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, Consumer onMatch); } diff --git a/src/java/org/apache/cassandra/io/AsymmetricParameterisedVersionedSerializer.java b/src/java/org/apache/cassandra/io/AsymmetricParameterisedVersionedSerializer.java index a1a295f83d..3b08d4e039 100644 --- a/src/java/org/apache/cassandra/io/AsymmetricParameterisedVersionedSerializer.java +++ b/src/java/org/apache/cassandra/io/AsymmetricParameterisedVersionedSerializer.java @@ -57,6 +57,11 @@ public interface AsymmetricParameterisedVersionedSerializer 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)) diff --git a/src/java/org/apache/cassandra/io/AsymmetricVersionedSerializer.java b/src/java/org/apache/cassandra/io/AsymmetricVersionedSerializer.java index 2dac3a5777..f0f4cdbf24 100644 --- a/src/java/org/apache/cassandra/io/AsymmetricVersionedSerializer.java +++ b/src/java/org/apache/cassandra/io/AsymmetricVersionedSerializer.java @@ -54,6 +54,12 @@ public interface AsymmetricVersionedSerializer } } 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)) diff --git a/src/java/org/apache/cassandra/io/util/DataInputBuffer.java b/src/java/org/apache/cassandra/io/util/DataInputBuffer.java index 9df9861cca..2059e38dfe 100644 --- a/src/java/org/apache/cassandra/io/util/DataInputBuffer.java +++ b/src/java/org/apache/cassandra/io/util/DataInputBuffer.java @@ -67,6 +67,11 @@ public class DataInputBuffer extends RebufferingInputStream return buffer.remaining(); } + public ByteBuffer buffer() + { + return buffer; + } + @Override public void close() {} } diff --git a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java index e322a51ee8..2da70603bd 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCacheEntry.java @@ -383,6 +383,13 @@ public class AccordCacheEntry 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()); diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java index 5806ae391f..e283d9f5e0 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStore.java @@ -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() diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 8adeb6ca87..20bc467c15 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -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; } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java index 3c97b02dcd..d461f6a8dc 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournalTable.java @@ -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 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 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 inMemory = index.search(commandStoreId, range, minTxnId, maxTxnId, minDecidedId).results(); - CloseableIterator table = tableSearch(commandStoreId, range.start(), range.end(), minTxnId, maxTxnId, minDecidedId); - return new DefaultResult(minTxnId, maxTxnId, minDecidedId, MergeIterator.get(Arrays.asList(inMemory, table))); + CloseableIterator inMemory = index.search(commandStoreId, range, minTxnId, maxTxnId, decidedRX).results(); + CloseableIterator 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 inMemory = index.search(commandStoreId, key, minTxnId, maxTxnId, minDecidedId).results(); - CloseableIterator table = tableSearch(commandStoreId, key, minTxnId, maxTxnId, minDecidedId); - return new DefaultResult(minTxnId, maxTxnId, minDecidedId, MergeIterator.get(Arrays.asList(inMemory, table))); + CloseableIterator inMemory = index.search(commandStoreId, key, minTxnId, maxTxnId, decidedRX).results(); + CloseableIterator table = tableSearch(commandStoreId, key, minTxnId, maxTxnId); + return new DefaultResult(minTxnId, maxTxnId, decidedRX, MergeIterator.get(Arrays.asList(inMemory, table))); } - private CloseableIterator tableSearch(int store, TokenKey start, TokenKey end, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId) + private CloseableIterator 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 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 tableSearch(int store, TokenKey key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId) + private CloseableIterator 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 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); } diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index ac112186f9..31a2bf942f 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -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; } diff --git a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java index b905ec519a..bbbb7679d3 100644 --- a/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java +++ b/src/java/org/apache/cassandra/service/accord/CommandsForRanges.java @@ -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 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 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 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(); + } } } } diff --git a/src/java/org/apache/cassandra/service/accord/RangeSearcher.java b/src/java/org/apache/cassandra/service/accord/RangeSearcher.java index 5c373ab481..fdaab601aa 100644 --- a/src/java/org/apache/cassandra/service/accord/RangeSearcher.java +++ b/src/java/org/apache/cassandra/service/accord/RangeSearcher.java @@ -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 results; private boolean consumed = false; - public DefaultResult(TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, CloseableIterator results) + public DefaultResult(TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, CloseableIterator 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; } diff --git a/src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java b/src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java index 7d513a9cb9..65e0bb0921 100644 --- a/src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java +++ b/src/java/org/apache/cassandra/service/accord/RouteInMemoryIndex.java @@ -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 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 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 search(int storeId, TableId tableId, byte[] start, byte[] end, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId) + private synchronized NavigableSet 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 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 result = search(commandStoreId, key.table(), OrderedRouteSerializer.serializeTokenOnly(key), minTxnId, maxTxnId, minDecidedId); - return new DefaultResult(minTxnId, maxTxnId, minDecidedId, CloseableIterator.wrap(result.iterator())); + NavigableSet result = search(commandStoreId, key.table(), OrderedRouteSerializer.serializeTokenOnly(key), minTxnId, maxTxnId, decidedRX); + return new DefaultResult(minTxnId, maxTxnId, decidedRX, CloseableIterator.wrap(result.iterator())); } - private synchronized NavigableSet search(int storeId, TableId tableId, byte[] key, TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId) + private synchronized NavigableSet 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 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 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> 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> 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 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> 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> 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 implements RangeSearcher } private void search(byte[] start, byte[] end, - TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, + TxnId minTxnId, Timestamp maxTxnId, @Nullable DecidedRX decidedRX, Consumer> 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 implements RangeSearcher } private void search(byte[] key, - TxnId minTxnId, Timestamp maxTxnId, @Nullable TxnId minDecidedId, + TxnId minTxnId, Timestamp maxTxnId, @Nullable MaxDecidedRX.DecidedRX decidedRX, Consumer> 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; diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java index af1782db2a..864752d9c6 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -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) diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java index 6981761612..d1f81512b9 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CheckStatusSerializers.java @@ -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; } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java index 9b07e707d2..3348fc1f2c 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java @@ -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 = new VariableWidthTimestampSerializer<>(TxnId::fromValues); - public static final VariableWidthTimestampSerializer timestamp = new VariableWidthTimestampSerializer<>(Timestamp::fromValues); + public static final VariableWidthTimestampSerializer txnId = new VariableWidthTimestampSerializer<>(TxnId::fromValues, TxnId::fromBits, TxnId[]::new); + public static final VariableWidthTimestampSerializer timestamp = new VariableWidthTimestampSerializer<>(Timestamp::fromValues, Timestamp::fromBits, Timestamp[]::new); public static final BallotSerializer ballot = new BallotSerializer(); // permits null public static final UnversionedSerializer 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 create(long epoch, long hlc, int flags, Node.Id node); - } - - private final VariableWidthTimestampSerializer.Factory factory; + private final Timestamp.ValueFactory factory; + private final Timestamp.RawFactory rawFactory; + private final IntFunction allocator; T decodeSpecial(int encodingFlags) { @@ -510,9 +513,11 @@ public class CommandSerializers return NULL_BYTE; } - private VariableWidthTimestampSerializer(VariableWidthTimestampSerializer.Factory factory) + private VariableWidthTimestampSerializer(Timestamp.ValueFactory factory, Timestamp.RawFactory rawFactory, IntFunction 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 int serializeComparable(T ts, V dst, ValueAccessor 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 @@ -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) { diff --git a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java index daf6f923d2..f400d66d5e 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/DepsSerializers.java @@ -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 tokenRange; - public static final DepsSerializer deps; + public static final AbstractDepsSerializer deps; public static final UnversionedSerializer nullableDeps; - public static final DepsSerializer partialDeps; + public static final AbstractDepsSerializer partialDeps; + public static final AbstractDepsSerializer partialDepsById; public static final UnversionedSerializer 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 implements UnversionedSerializer + public static abstract class AbstractDepsSerializer implements UnversionedSerializer { - protected UnversionedSerializer tokenRange; - public DepsSerializer(UnversionedSerializer 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 tokenRange; + public AbstractDepsSerializer(boolean forceByTxnId, UnversionedSerializer 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 + { + public PartialDepsSerializer(boolean preferByTxnId, UnversionedSerializer 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 + { + public DepsSerializer(boolean preferByTxnId, UnversionedSerializer 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 tokenRange; - final DepsSerializer deps; + final DepsSerializer deps; final UnversionedSerializer nullableDeps; - final DepsSerializer partialDeps; + final PartialDepsSerializer partialDeps; + final PartialDepsSerializer partialDepsById; final UnversionedSerializer nullablePartialDeps; public Impl(UnversionedSerializer 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); } } + } \ No newline at end of file diff --git a/src/java/org/apache/cassandra/service/accord/serializers/KeySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/KeySerializers.java index 3ab41a74df..d8778fc67b 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/KeySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/KeySerializers.java @@ -73,16 +73,17 @@ public class KeySerializers public static final AccordSearchableKeySerializer routingKey; public static final UnversionedSerializer nullableRoutingKey; - public static final AbstractSearchableRoutingKeysSerializer routingKeys; + public static final AbstractKeyRoutablesSerializer routingKeys; public static final UnversionedSerializer keys; - public static final AbstractSearchableRoutingKeysSerializer partialKeyRoute; - public static final AbstractSearchableRoutingKeysSerializer fullKeyRoute; + public static final AbstractKeyRoutablesSerializer partialKeyRoute; + public static final AbstractKeyRoutablesSerializer fullKeyRoute; public static final UnversionedSerializer range; - public static final AbstractRangesSerializer ranges; - public static final AbstractRangesSerializer partialRangeRoute; - public static final AbstractRangesSerializer fullRangeRoute; + public static final AbstractRangesSerializer rangeArray; + public static final AbstractRangeRoutablesSerializer ranges; + public static final AbstractRangeRoutablesSerializer partialRangeRoute; + public static final AbstractRangeRoutablesSerializer fullRangeRoute; public static final AbstractRoutablesSerializer> route; public static final UnversionedSerializer> 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; final UnversionedSerializer nullableRoutingKey; - final AbstractSearchableRoutingKeysSerializer routingKeys; + final AbstractKeyRoutablesSerializer routingKeys; final UnversionedSerializer keys; - final AbstractSearchableRoutingKeysSerializer partialKeyRoute; - final AbstractSearchableRoutingKeysSerializer fullKeyRoute; + final AbstractKeyRoutablesSerializer partialKeyRoute; + final AbstractKeyRoutablesSerializer fullKeyRoute; final UnversionedSerializer range; - final AbstractRangesSerializer ranges; - final AbstractRangesSerializer partialRangeRoute; - final AbstractRangesSerializer fullRangeRoute; + final AbstractRangesSerializer rangeArray; + final AbstractRangeRoutablesSerializer ranges; + final AbstractRangeRoutablesSerializer partialRangeRoute; + final AbstractRangeRoutablesSerializer fullRangeRoute; final AbstractRoutablesSerializer> route; final UnversionedSerializer> 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> implements UnversionedSerializer { final TinyEnumSet permitted; - final AbstractSearchableRoutingKeysSerializer routingKeys; - final AbstractSearchableRoutingKeysSerializer partialKeyRoute; - final AbstractSearchableRoutingKeysSerializer fullKeyRoute; - final AbstractRangesSerializer ranges; - final AbstractRangesSerializer partialRangeRoute; - final AbstractRangesSerializer fullRangeRoute; + final AbstractKeyRoutablesSerializer routingKeys; + final AbstractKeyRoutablesSerializer partialKeyRoute; + final AbstractKeyRoutablesSerializer fullKeyRoute; + final AbstractRangeRoutablesSerializer ranges; + final AbstractRangeRoutablesSerializer partialRangeRoute; + final AbstractRangeRoutablesSerializer fullRangeRoute; protected AbstractRoutablesSerializer(TinyEnumSet permitted, - AbstractSearchableRoutingKeysSerializer routingKeys, - AbstractSearchableRoutingKeysSerializer partialKeyRoute, - AbstractSearchableRoutingKeysSerializer fullKeyRoute, - AbstractRangesSerializer ranges, - AbstractRangesSerializer partialRangeRoute, - AbstractRangesSerializer fullRangeRoute) + AbstractKeyRoutablesSerializer routingKeys, + AbstractKeyRoutablesSerializer partialKeyRoute, + AbstractKeyRoutablesSerializer fullKeyRoute, + AbstractRangeRoutablesSerializer ranges, + AbstractRangeRoutablesSerializer partialRangeRoute, + AbstractRangeRoutablesSerializer 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> { final UnversionedSerializer keys; - final AbstractRangesSerializer ranges; + final AbstractRangeRoutablesSerializer ranges; - public AbstractSeekablesSerializer(UnversionedSerializer keys, AbstractRangesSerializer ranges) + public AbstractSeekablesSerializer(UnversionedSerializer keys, AbstractRangeRoutablesSerializer 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> extends IVersionedWithKeysSerializer.AbstractWithKeysSerializer implements UnversionedSerializer + public abstract static class AbstractSearchableSerializer extends IVersionedWithKeysSerializer.AbstractWithKeysSerializer implements UnversionedSerializer { final IntFunction 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 extends AbstractSearchableSerializer implements UnversionedSerializer + public abstract static class AbstractKeyRoutablesSerializer extends AbstractSearchableSerializer implements UnversionedSerializer { - public AbstractSearchableRoutingKeysSerializer(AccordSearchableKeySerializer 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 extends AbstractSearchableRoutingKeysSerializer + public abstract static class AbstractKeyRouteSerializer extends AbstractKeyRoutablesSerializer { - public AbstractKeyRouteSerializer(AccordSearchableKeySerializer 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 extends AbstractSearchableSerializer implements UnversionedSerializer + public abstract static class AbstractRangesSerializer extends AbstractSearchableSerializer implements UnversionedSerializer { 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 extends AbstractRangesSerializer implements UnversionedSerializer + { + @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 extends AbstractRangesSerializer + public abstract static class AbstractRangeRouteSerializer extends AbstractRangeRoutablesSerializer { public AbstractRangeRouteSerializer() { diff --git a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java index fcb680326d..3eb2fa4984 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/RecoverySerializers.java @@ -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 request = new WithUnsyncedSerializer() { @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())); } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java b/src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java new file mode 100644 index 0000000000..8321e147fd --- /dev/null +++ b/src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java @@ -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 + { + long get(In in, int i); + } + + @Inline + public static void serializePacked(SerializeAdapter 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 + { + void accept(Out out, int i, long v); + } + + @Inline + public static void deserializePacked(DeserializeAdapter 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; + } +} diff --git a/src/java/org/apache/cassandra/service/accord/serializers/TableMetadatasAndKeys.java b/src/java/org/apache/cassandra/service/accord/serializers/TableMetadatasAndKeys.java index 293264bfa5..47b218944a 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/TableMetadatasAndKeys.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/TableMetadatasAndKeys.java @@ -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(); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java index af7e5d66bb..5770125a0d 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/WaitingOnSerializer.java @@ -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 diff --git a/src/java/org/apache/cassandra/service/accord/txn/AccordUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/AccordUpdate.java index 453ae24cc9..89536d2058 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/AccordUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/AccordUpdate.java @@ -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) { diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java index 17f69aefae..61602324af 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnNamedRead.java @@ -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 @@ -431,6 +432,14 @@ public class TxnNamedRead extends AbstractParameterisedVersionedSerialized 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 { diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java index aef4342c67..f88b1eb3b8 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnUpdate.java @@ -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) { diff --git a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java index 8008df78ab..31efe97026 100644 --- a/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java +++ b/src/java/org/apache/cassandra/service/accord/txn/TxnWrite.java @@ -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 implements Write { @@ -215,6 +216,14 @@ public class TxnWrite extends AbstractKeySorted 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 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) { diff --git a/src/java/org/apache/cassandra/utils/ArraySerializers.java b/src/java/org/apache/cassandra/utils/ArraySerializers.java index cad27d68fe..7ca6afa8a7 100644 --- a/src/java/org/apache/cassandra/utils/ArraySerializers.java +++ b/src/java/org/apache/cassandra/utils/ArraySerializers.java @@ -69,6 +69,13 @@ public class ArraySerializers return items; } + public static void skipArray(DataInputPlus in, UnversionedSerializer serializer) throws IOException + { + int size = in.readUnsignedVInt32(); + for (int i = 0; i < size; i++) + serializer.skip(in); + } + public static T[] deserializeArray(DataInputPlus in, int version, IVersionedSerializer serializer, IntFunction arrayFactory) throws IOException { int size = in.readUnsignedVInt32(); @@ -87,6 +94,14 @@ public class ArraySerializers return items; } + public static void skipArray(DataInputPlus in, Version version, AsymmetricVersionedSerializer serializer) throws IOException + { + int size = in.readUnsignedVInt32(); + for (int i = 0; i < size; i++) + serializer.skip(in, version); + } + + public static T[] deserializeArray(P p, DataInputPlus in, Version version, ParameterisedVersionedSerializer serializer, IntFunction arrayFactory) throws IOException { int size = in.readUnsignedVInt32(); @@ -96,6 +111,13 @@ public class ArraySerializers return items; } + public static void skipArray(P p, DataInputPlus in, Version version, ParameterisedVersionedSerializer serializer) throws IOException + { + int size = in.readUnsignedVInt32(); + for (int i = 0; i < size; i++) + serializer.skip(p, in, version); + } + public static long serializedArraySize(T[] array, UnversionedSerializer serializer) { long size = sizeofUnsignedVInt(array.length); diff --git a/test/unit/org/apache/cassandra/index/accord/AccordIndexUtil.java b/test/unit/org/apache/cassandra/index/accord/AccordIndexUtil.java index d791c5906e..093001021c 100644 --- a/test/unit/org/apache/cassandra/index/accord/AccordIndexUtil.java +++ b/test/unit/org/apache/cassandra/index/accord/AccordIndexUtil.java @@ -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())); + } } diff --git a/test/unit/org/apache/cassandra/index/accord/RangeMemoryIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RangeMemoryIndexTest.java index 9bef0aecbc..8e4fed96f4 100644 --- a/test/unit/org/apache/cassandra/index/accord/RangeMemoryIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/RangeMemoryIndexTest.java @@ -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 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 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 search(TokenRange range, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId) + public NavigableSet 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 search(TokenKey key, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId) + public NavigableSet 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 search(Predicate test, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId) + public NavigableSet search(Predicate test, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX) { NavigableSet 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); } diff --git a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java index 7b32878e71..d0c2d87f1e 100644 --- a/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/RouteIndexTest.java @@ -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 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 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 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 run(Sut sut) throws Throwable { Set 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 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 run(Sut sut) throws Throwable { Set 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) diff --git a/test/unit/org/apache/cassandra/service/accord/RouteInMemoryIndexTest.java b/test/unit/org/apache/cassandra/service/accord/RouteInMemoryIndexTest.java index b51c778161..1d95a6d304 100644 --- a/test/unit/org/apache/cassandra/service/accord/RouteInMemoryIndexTest.java +++ b/test/unit/org/apache/cassandra/service/accord/RouteInMemoryIndexTest.java @@ -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 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 actual = new ArrayList<>(); - index.search(0, range, minTxnId, maxTxnId, minDecidedId).consume(actual::add); + index.search(0, range, minTxnId, maxTxnId, decidedRX).consume(actual::add); List 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 actual = new ArrayList<>(); - index.search(0, key, minTxnId, maxTxnId, minDecidedId).consume(actual::add); + index.search(0, key, minTxnId, maxTxnId, decidedRX).consume(actual::add); List 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 test, TxnId minTxnId, TxnId maxTxnId, @Nullable TxnId minDecidedId) + public RangeSearcher.Result search(Predicate test, TxnId minTxnId, TxnId maxTxnId, @Nullable DecidedRX decidedRX) { TreeSet 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) diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java index f47f56a9b0..868134292c 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStore.java @@ -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); diff --git a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java index 8a5570d736..f7778c628c 100644 --- a/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java +++ b/test/unit/org/apache/cassandra/service/accord/SimulatedAccordCommandStoreTestBase.java @@ -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); diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java index aa97ee534b..85d2618f6e 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/WaitingOnSerializerTest.java @@ -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); }