diff --git a/modules/accord b/modules/accord index fc14a154fd..134df57677 160000 --- a/modules/accord +++ b/modules/accord @@ -1 +1 @@ -Subproject commit fc14a154fd514d4ab40b37508fb9497f786835e0 +Subproject commit 134df57677bbd5092994923a4dc2f15cd1d033d1 diff --git a/src/java/org/apache/cassandra/concurrent/ExecutionFailure.java b/src/java/org/apache/cassandra/concurrent/ExecutionFailure.java index 27ab885e23..dc1262b3c1 100644 --- a/src/java/org/apache/cassandra/concurrent/ExecutionFailure.java +++ b/src/java/org/apache/cassandra/concurrent/ExecutionFailure.java @@ -19,6 +19,7 @@ package org.apache.cassandra.concurrent; import java.util.concurrent.Callable; +import java.util.concurrent.CancellationException; import java.util.concurrent.Future; import org.apache.cassandra.concurrent.DebuggableTask.RunnableDebuggableTask; @@ -26,6 +27,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.apache.cassandra.db.compaction.CompactionInterruptedException; +import org.apache.cassandra.exceptions.RequestTimeoutException; import org.apache.cassandra.utils.Closeable; import org.apache.cassandra.utils.JVMStabilityInspector; import org.apache.cassandra.utils.WithResources; @@ -49,6 +51,9 @@ public class ExecutionFailure { try { + if (t instanceof RequestTimeoutException || t instanceof CancellationException) + return; + if (t instanceof CompactionInterruptedException) { // TODO: should we check to see there aren't nested CompactionInterruptedException? diff --git a/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java b/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java index edc28b58e2..381bac4971 100644 --- a/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java +++ b/src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java @@ -318,13 +318,13 @@ public class ByteBufferAccessor implements ValueAccessor @Override public int putLeastSignificantBytes(ByteBuffer dst, int offset, long register, int bytes) { - if (dst.remaining() < Long.BYTES) + int pos = dst.position() + offset; + if (dst.limit() - pos < Long.BYTES) { return ValueAccessor.putLeastSignificantBytes(this, dst, offset, register, bytes); } else { - int pos = dst.position() + offset; dst.putLong(pos, register << (64 - (bytes * 8))); } return bytes; @@ -333,13 +333,13 @@ public class ByteBufferAccessor implements ValueAccessor @Override public long getLeastSignificantBytes(ByteBuffer dst, int offset, int bytes) { - if (dst.remaining() < Long.BYTES) + int pos = dst.position() + offset; + if (dst.limit() - pos < Long.BYTES) { return ValueAccessor.getLeastSignificantBytes(this, dst, offset, bytes); } else { - int pos = dst.position() + offset; return dst.getLong(pos) >>> (64 - (bytes * 8)); } } diff --git a/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java b/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java index 362063d683..4916c73262 100644 --- a/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java +++ b/src/java/org/apache/cassandra/db/marshal/ValueAccessor.java @@ -525,23 +525,23 @@ public interface ValueAccessor break; case 3: accessor.putShort(dst, offset, (short)(register >>> 8)); - accessor.putByte(dst, offset, (byte)register); + accessor.putByte(dst, offset + 2, (byte)register); break; case 4: accessor.putInt(dst, offset, (int)register); break; case 5: accessor.putInt(dst, offset, (int)(register >>> 8)); - accessor.putByte(dst, offset, (byte)register); + accessor.putByte(dst, offset + 4, (byte)register); break; case 6: accessor.putInt(dst, offset, (int)(register >>> 16)); - accessor.putShort(dst, offset, (short)register); + accessor.putShort(dst, offset + 4, (short)register); break; case 7: accessor.putInt(dst, offset, (int)(register >>> 24)); - accessor.putShort(dst, offset, (short)(register >> 8)); - accessor.putByte(dst, offset, (byte)register); + accessor.putShort(dst, offset + 4, (short)(register >> 8)); + accessor.putByte(dst, offset + 6, (byte)register); break; case 8: accessor.putLong(dst, offset, register); @@ -557,23 +557,23 @@ public interface ValueAccessor switch (bytes) { case 0: return 0; - case 1: return accessor.getByte(dst, offset); - case 2: return accessor.getShort(dst, offset); + case 1: return accessor.getByte(dst, offset) & 0xffL; + case 2: return accessor.getShort(dst, offset) & 0xffffL; case 3: - return ((long)accessor.getShort(dst, offset) << 8) - | (long)accessor.getByte(dst, offset + 2); + return ((accessor.getShort(dst, offset) & 0xffffL) << 8) + | (accessor.getByte(dst, offset + 2) & 0xffL); case 4: - return accessor.getInt(dst, offset); + return accessor.getInt(dst, offset) & 0xffffffffL; case 5: - return ((long)accessor.getInt(dst, offset) << 8) - | (long)accessor.getByte(dst, offset + 4); + return ((accessor.getInt(dst, offset) & 0xffffffffL) << 8) + | (accessor.getByte(dst, offset + 4) & 0xffL); case 6: - return ((long)accessor.getInt(dst, offset) << 16) - | (long)accessor.getShort(dst, offset + 4); + return ((accessor.getInt(dst, offset) & 0xffffffffL) << 16) + | (accessor.getShort(dst, offset + 4) & 0xffffL); case 7: - return ((long)accessor.getInt(dst, offset) << 24) - | ((long)accessor.getShort(dst, offset + 4) << 8) - | (long)accessor.getByte(dst, offset + 6); + return ((accessor.getInt(dst, offset) & 0xffffffffL) << 24) + | ((accessor.getShort(dst, offset + 4) & 0xffffL) << 8) + | (accessor.getByte(dst, offset + 6) & 0xffL); case 8: return accessor.getLong(dst, offset); default: throw new IllegalArgumentException(); } diff --git a/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java b/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java index b92bf69e85..56be6b9eed 100644 --- a/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java +++ b/src/java/org/apache/cassandra/index/accord/RouteIndexFormat.java @@ -249,7 +249,7 @@ public class RouteIndexFormat } } - static List readSegements(Map index) throws IOException + static List readSegments(Map index) throws IOException { List segments = new ArrayList<>(); diff --git a/src/java/org/apache/cassandra/index/accord/SSTableIndex.java b/src/java/org/apache/cassandra/index/accord/SSTableIndex.java index c6085b8827..03c669f4bd 100644 --- a/src/java/org/apache/cassandra/index/accord/SSTableIndex.java +++ b/src/java/org/apache/cassandra/index/accord/SSTableIndex.java @@ -74,7 +74,7 @@ public class SSTableIndex extends SharedCloseableImpl Map files = new EnumMap<>(IndexComponent.class); for (IndexComponent c : id.getLiveComponents()) files.put(c, new FileHandle.Builder(id.fileFor(c)).mmapped(true).complete()); - List segments = RouteIndexFormat.readSegements(files); + List segments = RouteIndexFormat.readSegments(files); files.remove(IndexComponent.SEGMENT).close(); files.remove(IndexComponent.METADATA).close(); Cleanup cleanup = new Cleanup(files); diff --git a/src/java/org/apache/cassandra/io/AsymmetricUnversionedSerializer.java b/src/java/org/apache/cassandra/io/AsymmetricUnversionedSerializer.java index eae92e087e..570741903f 100644 --- a/src/java/org/apache/cassandra/io/AsymmetricUnversionedSerializer.java +++ b/src/java/org/apache/cassandra/io/AsymmetricUnversionedSerializer.java @@ -42,6 +42,11 @@ public interface AsymmetricUnversionedSerializer } } + default void skip(DataInputPlus in) throws IOException + { + deserialize(in); + } + default ByteBuffer serializeUnchecked(In t) { try diff --git a/src/java/org/apache/cassandra/net/Verb.java b/src/java/org/apache/cassandra/net/Verb.java index 6d3e6f95db..e9258b5f9a 100644 --- a/src/java/org/apache/cassandra/net/Verb.java +++ b/src/java/org/apache/cassandra/net/Verb.java @@ -338,7 +338,7 @@ public enum Verb ACCORD_AWAIT_ASYNC_RSP_REQ (139, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.asyncReply), AccordService::requestHandlerOrNoop ), ACCORD_WAIT_UNTIL_APPLIED_REQ (140, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(ReadDataSerializers.waitUntilApplied), AccordService::requestHandlerOrNoop, ACCORD_READ_RSP ), ACCORD_RECOVER_AWAIT_RSP (141, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.recoverReply), AccordService::responseHandlerOrNoop ), - ACCORD_RECOVER_AWAIT_REQ (142, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.recoverRequest), AccordService::requestHandlerOrNoop, ACCORD_RECOVER_AWAIT_RSP), + ACCORD_RECOVER_AWAIT_REQ (142, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AwaitSerializers.recoverRequest), AccordService::requestHandlerOrNoop, ACCORD_RECOVER_AWAIT_RSP), ACCORD_INFORM_DURABLE_REQ (143, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(InformDurableSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ), ACCORD_CHECK_STATUS_RSP (144, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CheckStatusSerializers.reply), AccordService::responseHandlerOrNoop ), ACCORD_CHECK_STATUS_REQ (145, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(CheckStatusSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_CHECK_STATUS_RSP ), @@ -350,15 +350,14 @@ public enum Verb ACCORD_GET_LATEST_DEPS_REQ (151, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(LatestDepsSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_LATEST_DEPS_RSP), ACCORD_GET_MAX_CONFLICT_RSP (152, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.reply), AccordService::responseHandlerOrNoop ), ACCORD_GET_MAX_CONFLICT_REQ (153, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetMaxConflictSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_MAX_CONFLICT_RSP), - ACCORD_GET_DURABLE_BEFORE_RSP (154, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.reply), AccordService::responseHandlerOrNoop ), - ACCORD_GET_DURABLE_BEFORE_REQ (155, P2, readTimeout, IMMEDIATE, () -> accordEmbedded(GetDurableBeforeSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_DURABLE_BEFORE_RSP ), + ACCORD_GET_DURABLE_BEFORE_RSP (154, P2, readTimeout, MISC, () -> accordEmbedded(GetDurableBeforeSerializers.reply), AccordService::responseHandlerOrNoop ), + ACCORD_GET_DURABLE_BEFORE_REQ (155, P2, readTimeout, MISC, () -> accordEmbedded(GetDurableBeforeSerializers.request), AccordService::requestHandlerOrNoop, ACCORD_GET_DURABLE_BEFORE_RSP ), ACCORD_SET_SHARD_DURABLE_REQ (156, P2, rpcTimeout, MISC, () -> accordEmbedded(SetDurableSerializers.shardDurable), AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ), ACCORD_SET_GLOBALLY_DURABLE_REQ (157, P2, rpcTimeout, MISC, () -> accordEmbedded(SetDurableSerializers.globallyDurable),AccordService::requestHandlerOrNoop, ACCORD_SIMPLE_RSP ), ACCORD_SYNC_NOTIFY_RSP (158, P2, writeTimeout, MISC, () -> accordEmbedded(EnumSerializer.simpleReply), RESPONSE_HANDLER), ACCORD_SYNC_NOTIFY_REQ (159, P2, writeTimeout, MISC, () -> accordEmbedded(Notification.serializer), () -> AccordSyncPropagator.verbHandler, ACCORD_SYNC_NOTIFY_RSP ), - CONSENSUS_KEY_MIGRATION (160, P1, writeTimeout, MISC, () -> accordEmbedded(ConsensusKeyMigrationFinished.serializer),() -> ConsensusKeyMigrationState.consensusKeyMigrationFinishedHandler), ACCORD_INTEROP_READ_RSP (161, P2, writeTimeout, IMMEDIATE, () -> accordEmbedded(AccordInteropRead.replySerializer), AccordService::responseHandlerOrNoop), diff --git a/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java b/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java index 668736c842..18f3de0633 100644 --- a/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java +++ b/src/java/org/apache/cassandra/service/accord/AbstractAccordSegmentCompactor.java @@ -106,7 +106,7 @@ public abstract class AbstractAccordSegmentCompactor implements SegmentCompac } // nothing to compact (all segments empty, should never happen, but it is theoretically possible?) - exit early - // TODO: investigate how this comes to be, check if there is a cleanup issue + // TODO (required): investigate how this comes to be, check if there is a cleanup issue if (readers.isEmpty()) return Collections.emptyList(); diff --git a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java index efc504ecb9..4cec60f50c 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java +++ b/src/java/org/apache/cassandra/service/accord/AccordCommandStores.java @@ -263,6 +263,6 @@ public class AccordCommandStores extends CommandStores implements CacheSize throw new RuntimeException(e); } } - //TODO shutdown isn't useful by itself, we need a way to "wait" as well. Should be AutoCloseable or offer awaitTermination as well (think Shutdownable interface) + //TODO (expected): shutdown isn't useful by itself, we need a way to "wait" as well. Should be AutoCloseable or offer awaitTermination as well (think Shutdownable interface) } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java index 07eade798c..2ab5c98641 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordDataStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordDataStore.java @@ -71,10 +71,10 @@ public class AccordDataStore implements DataStore } @Override - public AsyncResult snapshot(Ranges ranges, TxnId before) // TODO: does this have to go to journal, too? + public AsyncResult snapshot(Ranges ranges, TxnId before) { AsyncResults.SettableResult result = new AsyncResults.SettableResult<>(); - // TODO: maintain a list of Accord tables, perhaps in ClusterMetadata? + // TODO (desired): maintain a list of Accord tables, perhaps in ClusterMetadata? ClusterMetadata metadata = ClusterMetadata.current(); Object2ObjectHashMap tables = new Object2ObjectHashMap<>(); for (Range range : ranges) diff --git a/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java b/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java index d26e6e5abe..c5442c644d 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java +++ b/src/java/org/apache/cassandra/service/accord/AccordExecutorLoops.java @@ -24,8 +24,7 @@ import java.util.function.Function; import java.util.function.IntFunction; import accord.utils.Invariants; - -import io.netty.util.collection.LongObjectHashMap; +import org.agrona.collections.Long2ObjectHashMap; import org.apache.cassandra.service.accord.AccordExecutor.Mode; import org.apache.cassandra.utils.concurrent.Condition; @@ -37,7 +36,7 @@ import static org.apache.cassandra.utils.Clock.Global.nanoTime; class AccordExecutorLoops { - private final LongObjectHashMap loops; + private final Long2ObjectHashMap loops; private final AtomicInteger running = new AtomicInteger(); private final Condition terminated = Condition.newOneTimeCondition(); @@ -46,7 +45,7 @@ class AccordExecutorLoops { Invariants.require(mode == RUN_WITH_LOCK ? threads == 1 : threads >= 1); running.addAndGet(threads); - loops = new LongObjectHashMap<>(threads); + loops = new Long2ObjectHashMap<>(threads, 0.65f); for (int i = 0; i < threads; ++i) { Thread thread = executorFactory().startThread(name.apply(i), wrap(loopFactory.apply(mode)), NON_DAEMON, INFINITE_LOOP); diff --git a/src/java/org/apache/cassandra/service/accord/AccordJournal.java b/src/java/org/apache/cassandra/service/accord/AccordJournal.java index 7a29afb061..66bb8bdffa 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordJournal.java +++ b/src/java/org/apache/cassandra/service/accord/AccordJournal.java @@ -384,7 +384,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier private BUILDER readAll(JournalKey key) { BUILDER builder = (BUILDER) key.type.serializer.mergerFor(); - // TODO: this can be further improved to avoid allocating lambdas + // TODO (expected): this can be further improved to avoid allocating lambdas AccordJournalValueSerializers.FlyweightSerializer serializer = (AccordJournalValueSerializers.FlyweightSerializer) key.type.serializer; // TODO (expected): for those where we store an image, read only the first entry we find in DESC order journalTable.readAll(key, (in, userVersion) -> serializer.deserialize(key, builder, in, userVersion)); @@ -572,7 +572,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier CommandSerializers.ballot.serialize(command.promised(), out); break; case PARTICIPANTS: - CommandSerializers.participants.serialize(command.participants(), out, userVersion); + CommandSerializers.participants.serialize(command.participants(), out); break; case PARTIAL_TXN: CommandSerializers.partialTxn.serialize(command.partialTxn(), out, userVersion); @@ -705,7 +705,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier break; case PARTICIPANTS: Invariants.require(participants != null); - CommandSerializers.participants.serialize(participants, out, userVersion); + CommandSerializers.participants.serialize(participants, out); break; case PARTIAL_TXN: Invariants.require(partialTxn != null); @@ -783,7 +783,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier promised = CommandSerializers.ballot.deserialize(in); break; case PARTICIPANTS: - participants = CommandSerializers.participants.deserialize(in, userVersion); + participants = CommandSerializers.participants.deserialize(in); break; case PARTIAL_TXN: partialTxn = CommandSerializers.partialTxn.deserialize(in, userVersion); @@ -832,7 +832,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier CommandSerializers.ballot.skip(in); break; case PARTICIPANTS: - CommandSerializers.participants.deserialize(in, userVersion); + CommandSerializers.participants.deserialize(in); break; case PARTIAL_TXN: CommandSerializers.partialTxn.deserialize(in, userVersion); diff --git a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java index efe9f39b25..e87250fa52 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java +++ b/src/java/org/apache/cassandra/service/accord/AccordKeyspace.java @@ -63,9 +63,7 @@ import org.apache.cassandra.db.filter.DataLimits; import org.apache.cassandra.db.filter.RowFilter; import org.apache.cassandra.db.lifecycle.View; import org.apache.cassandra.db.marshal.ByteBufferAccessor; -import org.apache.cassandra.db.marshal.ByteType; import org.apache.cassandra.db.marshal.BytesType; -import org.apache.cassandra.db.marshal.CompositeType; import org.apache.cassandra.db.marshal.Int32Type; import org.apache.cassandra.db.marshal.LongType; import org.apache.cassandra.db.marshal.TupleType; @@ -113,6 +111,7 @@ import org.apache.cassandra.utils.CloseableIterator; import org.apache.cassandra.utils.MergeIterator; import org.apache.cassandra.utils.btree.BTreeSet; import org.apache.cassandra.utils.concurrent.OpOrder; +import org.apache.cassandra.utils.vint.VIntCoding; import static java.lang.String.format; import static java.util.Collections.emptyMap; @@ -140,14 +139,12 @@ public class AccordKeyspace TableMetadata.Builder builder = parse(tableName, "accord journal", "CREATE TABLE %s (" - + "store_id int," - + "type tinyint," - + "id blob," + + "key blob," + "descriptor bigint," + "offset int," + "user_version int," + "record blob," - + "PRIMARY KEY((store_id, type, id), descriptor, offset)" + + "PRIMARY KEY((key), descriptor, offset)" + ") WITH CLUSTERING ORDER BY (descriptor DESC, offset DESC)" + " WITH compression = {'class':'NoopCompressor'};") .compaction(CompactionParams.lcs(emptyMap())) @@ -417,9 +414,7 @@ public class AccordKeyspace */ private static CloseableIterator keyIterator(Memtable memtable, AbstractBounds range) { - // TODO (required): why are we replacing the right bound with max bound? - AbstractBounds memtableRange = range.withNewRight(memtable.metadata().partitioner.getMinimumToken().maxKeyBound()); - DataRange dataRange = new DataRange(memtableRange, new ClusteringIndexSliceFilter(Slices.ALL, false)); + DataRange dataRange = new DataRange(range, new ClusteringIndexSliceFilter(Slices.ALL, false)); UnfilteredPartitionIterator iter = memtable.partitionIterator(ColumnFilter.NONE, dataRange, SSTableReadsListener.NOOP_LISTENER); int rangeStartCmpMin = range.isStartInclusive() ? 0 : 1; @@ -537,7 +532,7 @@ public class AccordKeyspace return cell.accessor().toBuffer(cell.value()); } - // TODO: convert to byte array + // TODO (desired): convert to byte array private static ByteBuffer cellValue(Row row, ColumnMetadata column) { Cell cell = row.getCell(column); @@ -546,60 +541,38 @@ public class AccordKeyspace public static class JournalColumns { - static final ClusteringComparator keyComparator = Journal.partitionKeyAsClusteringComparator(); - static final CompositeType partitionKeyType = (CompositeType) Journal.partitionKeyType; - public static final ColumnMetadata store_id = getColumn(Journal, "store_id"); - public static final ColumnMetadata type = getColumn(Journal, "type"); - public static final ColumnMetadata id = getColumn(Journal, "id"); + public static final ColumnMetadata key = getColumn(Journal, "key"); public static final ColumnMetadata record = getColumn(Journal, "record"); public static final ColumnMetadata user_version = getColumn(Journal, "user_version"); public static final RegularAndStaticColumns regular = new RegularAndStaticColumns(Columns.NONE, Columns.from(Arrays.asList(record, user_version))); public static DecoratedKey decorate(JournalKey key) { - ByteBuffer id = ByteBuffer.allocate(CommandSerializers.txnId.serializedSize()); - CommandSerializers.txnId.serialize(key.id, id); - id.flip(); - ByteBuffer pk = keyComparator.make(key.commandStoreId, (byte)key.type.id, id).serializeAsPartitionKey(); - Invariants.require(getTxnId(splitPartitionKey(pk)).equals(key.id)); + int commandStoreIdBytes = VIntCoding.computeUnsignedVIntSize(key.commandStoreId); + int length = commandStoreIdBytes + 1; + if (key.type == JournalKey.Type.COMMAND_DIFF) + length += CommandSerializers.txnId.serializedSize(key.id); + ByteBuffer pk = ByteBuffer.allocate(length); + ByteBufferAccessor.instance.putUnsignedVInt32(pk, 0, key.commandStoreId); + pk.put(commandStoreIdBytes, (byte)key.type.id); + if (key.type == JournalKey.Type.COMMAND_DIFF) + CommandSerializers.txnId.serializeComparable(key.id, pk, ByteBufferAccessor.instance, commandStoreIdBytes + 1); return Journal.partitioner.decorateKey(pk); } - public static ByteBuffer[] splitPartitionKey(DecoratedKey key) - { - return JournalColumns.partitionKeyType.split(key.getKey()); - } - - public static ByteBuffer[] splitPartitionKey(ByteBuffer key) - { - return JournalColumns.partitionKeyType.split(key); - } - public static int getStoreId(DecoratedKey pk) { - return getStoreId(splitPartitionKey(pk)); - } - - public static int getStoreId(ByteBuffer[] partitionKeyComponents) - { - return Int32Type.instance.compose(partitionKeyComponents[store_id.position()]); - } - - public static JournalKey.Type getType(ByteBuffer[] partitionKeyComponents) - { - return JournalKey.Type.fromId(ByteType.instance.compose(partitionKeyComponents[type.position()])); - } - - public static TxnId getTxnId(ByteBuffer[] partitionKeyComponents) - { - ByteBuffer buffer = partitionKeyComponents[id.position()]; - return CommandSerializers.txnId.deserialize(buffer, buffer.position()); + return VIntCoding.readUnsignedVInt32(pk.getKey(), 0); } public static JournalKey getJournalKey(DecoratedKey key) { - ByteBuffer[] parts = splitPartitionKey(key); - return new JournalKey(getTxnId(parts), getType(parts), getStoreId(parts)); + ByteBuffer bb = key.getKey(); + int storeId = ByteBufferAccessor.instance.getUnsignedVInt32(bb, 0); + int offset = VIntCoding.readLengthOfVInt(bb, 0); + JournalKey.Type type = JournalKey.Type.fromId(bb.get(offset)); + TxnId txnId = type != JournalKey.Type.COMMAND_DIFF ? TxnId.NONE : CommandSerializers.txnId.deserializeComparable(bb, ByteBufferAccessor.instance, offset + 1); + return new JournalKey(txnId, type, storeId); } } diff --git a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java index 54846f5e8f..5a5e09d4f6 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java +++ b/src/java/org/apache/cassandra/service/accord/AccordObjectSizes.java @@ -58,6 +58,7 @@ import accord.primitives.TxnId; import accord.primitives.Unseekables; import accord.primitives.Writes; import accord.utils.ImmutableBitSet; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.dht.Murmur3Partitioner; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.service.accord.api.TokenKey; @@ -116,9 +117,12 @@ public class AccordObjectSizes public static long ranges(Ranges ranges) { long size = EMPTY_RANGES_SIZE; - size += ObjectSizes.sizeOfReferenceArray(ranges.size()); - // TODO: many ranges are fixed size, can compute by multiplication - for (int i = 0, mi = ranges.size() ; i < mi ; i++) + int numberOfRanges = ranges.size(); + size += ObjectSizes.sizeOfReferenceArray(numberOfRanges); + if (numberOfRanges > 1 && DatabaseDescriptor.getPartitioner().isFixedLength()) + return size + numberOfRanges * range(ranges.get(0)); + + for (int i = 0 ; i < numberOfRanges ; i++) size += range(ranges.get(i)); return size; } @@ -145,9 +149,12 @@ public class AccordObjectSizes private static long routingKeysOnly(AbstractKeys keys) { - // TODO: many routing keys are fixed size, can compute by multiplication - long size = ObjectSizes.sizeOfReferenceArray(keys.size()); - for (int i=0, mi=keys.size(); i 1 && DatabaseDescriptor.getPartitioner().isFixedLength()) + return size + numberOfKeys * key(keys.get(0)); + + for (int i=0 ; i < numberOfKeys; i++) size += key(keys.get(i)); return size; } @@ -163,7 +170,7 @@ public class AccordObjectSizes { return EMPTY_FULL_KEY_ROUTE_SIZE + routingKeysOnly(route) - + key(route.homeKey()); // TODO: we will probably dedup homeKey, serializer dependent, but perhaps this is an acceptable error + + key(route.homeKey()); // TODO (desired): we will probably dedup homeKey, serializer dependent, but perhaps this is an acceptable error } private static final long EMPTY_PARTIAL_KEY_ROUTE_KEYS_SIZE = measure(new PartialKeyRoute(new TokenKey(null, null), new RoutingKey[0])); @@ -187,7 +194,7 @@ public class AccordObjectSizes { return EMPTY_FULL_RANGE_ROUTE_SIZE + ranges(route) - + key(route.homeKey()); // TODO: we will probably dedup homeKey, serializer dependent, but perhaps this is an acceptable error + + key(route.homeKey()); // TODO (desired): we will probably dedup homeKey, serializer dependent, but perhaps this is an acceptable error } private static final long EMPTY_PARTIAL_RANGE_ROUTE_KEYS_SIZE = measure(new PartialRangeRoute(new TokenKey(null, null), new Range[0])); diff --git a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java index 89f3214dc7..5ace8976c8 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSafeCommandStore.java @@ -196,7 +196,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore> chains = new ArrayList<>(); for (TxnId blockedBy : cmdTxnState.blockedBy) { diff --git a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java index 5e857d618e..66ae868231 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java +++ b/src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java @@ -313,7 +313,6 @@ public class AccordSyncPropagator { Invariants.require(msg.payload == SimpleReply.Ok, "Unexpected message: %s", msg); Set completedEpochs = new HashSet<>(); - // TODO review is it a good idea to call the listener while not holding the `AccordSyncPropagator` lock? synchronized (AccordSyncPropagator.this) { pending.ack(to, notification); diff --git a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java index 12fd39a135..9562aea86b 100644 --- a/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java +++ b/src/java/org/apache/cassandra/service/accord/AccordVerbHandler.java @@ -57,8 +57,7 @@ public class AccordVerbHandler implements IVerbHandler T request = message.payload; /* - * TODO (desired): messages without side-effects don't go through the journal, - * and as such are retained on heap until the node catches up to waitForEpoch, + * TODO (desired): messages are retained on heap until the node catches up to waitForEpoch, * which can be problematic in absense of proper Accord<->Messaging backpressure */ Node.Id fromNodeId = endpointMapper.mappedId(message.from()); diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java index 7c2cdf972c..3bcb271381 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordAgent.java @@ -18,6 +18,7 @@ package org.apache.cassandra.service.accord.api; +import java.util.concurrent.CancellationException; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; @@ -59,6 +60,7 @@ import accord.utils.async.AsyncChains; import accord.utils.async.AsyncResult; import accord.utils.async.AsyncResults; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.exceptions.RequestTimeoutException; import org.apache.cassandra.metrics.AccordMetrics; import org.apache.cassandra.net.ResponseContext; import org.apache.cassandra.service.accord.AccordService; @@ -114,13 +116,12 @@ public class AccordAgent implements Agent @Override public void onRecover(Node node, Result success, Throwable fail) { - // TODO: this } @Override public void onInconsistentTimestamp(Command command, Timestamp prev, Timestamp next) { - // TODO: this + // TODO (expected): better reporting AssertionError error = new AssertionError("Inconsistent execution timestamp detected for txnId " + command.txnId() + ": " + prev + " != " + next); onUncaughtException(error); throw error; @@ -142,6 +143,8 @@ public class AccordAgent implements Agent @Override public void onUncaughtException(Throwable t) { + if (t instanceof RequestTimeoutException || t instanceof CancellationException) + return; logger.error("Uncaught accord exception", t); JVMStabilityInspector.uncaughtException(Thread.currentThread(), t); } diff --git a/src/java/org/apache/cassandra/service/accord/api/AccordRoutableKey.java b/src/java/org/apache/cassandra/service/accord/api/AccordRoutableKey.java index 8a2d5c1ce7..25d409bfde 100644 --- a/src/java/org/apache/cassandra/service/accord/api/AccordRoutableKey.java +++ b/src/java/org/apache/cassandra/service/accord/api/AccordRoutableKey.java @@ -56,7 +56,7 @@ public abstract class AccordRoutableKey implements RoutableKey static final int PREFIX_MASK = 0xF0; static final int SUFFIX_MASK = 0x0F; - final TableId table; // TODO (desired): use an id (TrM) + final TableId table; // TODO (desired): use a long id (TrM) protected AccordRoutableKey(TableId table) { diff --git a/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java b/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java index 2e47d0678f..2c0619fd9e 100644 --- a/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java +++ b/src/java/org/apache/cassandra/service/accord/api/PartitionKey.java @@ -21,14 +21,10 @@ package org.apache.cassandra.service.accord.api; import java.io.IOException; import java.nio.ByteBuffer; -import com.google.common.base.Preconditions; - import accord.api.Key; -import accord.primitives.Routable; import accord.utils.Invariants; import org.apache.cassandra.db.DecoratedKey; import org.apache.cassandra.db.SinglePartitionReadCommand; -import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.db.marshal.ByteBufferAccessor; import org.apache.cassandra.db.marshal.ValueAccessor; import org.apache.cassandra.db.partitions.Partition; @@ -39,6 +35,7 @@ import org.apache.cassandra.io.util.DataOutputPlus; import org.apache.cassandra.schema.TableId; import org.apache.cassandra.utils.ByteBufferUtil; import org.apache.cassandra.utils.ObjectSizes; +import org.apache.cassandra.utils.vint.VIntCoding; import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner; @@ -115,23 +112,16 @@ public final class PartitionKey extends AccordRoutableKey implements Key return partitionKey().toString(); } - // TODO: callers to this method are not correctly handling ranges - public static PartitionKey toPartitionKey(Routable routable) - { - return (PartitionKey) routable; - } - public static final Serializer serializer = new Serializer(); public static class Serializer implements AccordKeySerializer { - // TODO: add vint to value accessor and use vints private Serializer() {} @Override public void serialize(PartitionKey key, DataOutputPlus out) throws IOException { key.table().serializeCompact(out); - ByteBufferUtil.writeWithShortLength(key.partitionKey().getKey(), out); + ByteBufferUtil.writeWithVIntLength(key.partitionKey().getKey(), out); } public int serialize(PartitionKey key, V dst, ValueAccessor accessor, int offset) @@ -140,9 +130,8 @@ public final class PartitionKey extends AccordRoutableKey implements Key position += key.table().serializeCompact(dst, accessor, position); ByteBuffer bytes = key.partitionKey().getKey(); Invariants.require(key.partitionKey().getPartitioner() == getPartitioner()); - int numBytes = ByteBufferAccessor.instance.size(bytes); - Preconditions.checkState(numBytes <= Short.MAX_VALUE); - position += accessor.putShort(dst, position, (short) numBytes); + int numBytes = bytes.remaining(); + position += accessor.putUnsignedVInt32(dst, position, numBytes); position += accessor.copyByteBufferTo(bytes, 0, dst, position, numBytes); return position - offset; @@ -152,14 +141,14 @@ public final class PartitionKey extends AccordRoutableKey implements Key public void skip(DataInputPlus in) throws IOException { TableId.skipCompact(in); - ByteBufferUtil.skipShortLength(in); + ByteBufferUtil.skipWithVIntLength(in); } @Override public PartitionKey deserialize(DataInputPlus in) throws IOException { TableId tableId = TableId.deserializeCompact(in).intern(); - DecoratedKey key = getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(in)); + DecoratedKey key = getPartitioner().decorateKey(ByteBufferUtil.readWithVIntLength(in)); return new PartitionKey(tableId, key); } @@ -167,8 +156,8 @@ public final class PartitionKey extends AccordRoutableKey implements Key { TableId tableId = TableId.deserializeCompact(src, accessor, offset).intern(); offset += tableId.serializedCompactSize(); - int numBytes = accessor.getShort(src, offset); - offset += TypeSizes.SHORT_SIZE; + int numBytes = accessor.getUnsignedVInt32(src, offset); + offset += VIntCoding.readLengthOfVInt(src, accessor, offset); ByteBuffer bytes = ByteBuffer.allocate(numBytes); accessor.copyTo(src, offset, bytes, ByteBufferAccessor.instance, 0, numBytes); DecoratedKey key = getPartitioner().decorateKey(bytes); @@ -178,7 +167,7 @@ public final class PartitionKey extends AccordRoutableKey implements Key @Override public long serializedSize(PartitionKey key) { - return key.table().serializedCompactSize() + ByteBufferUtil.serializedSizeWithShortLength(key.partitionKey().getKey()); + return key.table().serializedCompactSize() + ByteBufferUtil.serializedSizeWithVIntLength(key.partitionKey().getKey()); } } } diff --git a/src/java/org/apache/cassandra/service/accord/api/TokenKey.java b/src/java/org/apache/cassandra/service/accord/api/TokenKey.java index 24aa2ac9bc..7511cb70f1 100644 --- a/src/java/org/apache/cassandra/service/accord/api/TokenKey.java +++ b/src/java/org/apache/cassandra/service/accord/api/TokenKey.java @@ -182,15 +182,11 @@ public final class TokenKey extends AccordRoutableKey implements RoutingKey, Ran public boolean isMin() { - //TODO (review): some code paths don't care if before/after are used, but some are not fully correct (range.isFullRange) -// return sentinel == MIN_TABLE_SENTINEL; return (sentinel & PREFIX_MASK) == (MIN_TABLE_SENTINEL & PREFIX_MASK); } public boolean isMax() { - //TODO (review): some code paths don't care if before/after are used, but some are not fully correct (range.isFullRange) -// return sentinel == MAX_TABLE_SENTINEL; return (sentinel & PREFIX_MASK) == (MAX_TABLE_SENTINEL & PREFIX_MASK); } 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 1bdc29d012..3fa2194087 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropExecution.java @@ -228,9 +228,9 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen public void sendReadCommand(Message message, InetAddressAndPort to, RequestCallback callback) { Node.Id id = endpointMapper.mappedId(to); - // TODO (nicetohave): It would be better to use the re-use the command from the transaction but it's fragile - // to try and figure out exactly what changed for things like read repair and short read protection - // Also this read scope doesn't reflect the contents of this particular read and is larger than it needs to be + // TODO (desired): It would be better to use the re-use the command from the transaction but it's fragile + // to try and figure out exactly what changed for things like read repair and short read protection + // Also this read scope doesn't reflect the contents of this particular read and is larger than it needs to be // TODO (required): understand interop and whether StableFastPath is appropriate AccordInteropStableThenRead commit = new AccordInteropStableThenRead(id, allTopologies, txnId, Kind.StableFastPath, executeAt, txn, deps, route, message.payload); node.send(id, commit, executor, new AccordInteropRead.ReadCallback(id, to, message, callback, this)); diff --git a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java index f74b6625c2..995d93b6b0 100644 --- a/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java +++ b/src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java @@ -129,7 +129,6 @@ public class AccordInteropReadRepair extends ReadData public AccordInteropReadRepair(TxnId txnId, Participants scope, long executeAtEpoch, Mutation mutation) { - // TODO (review): remove followup read - Is there anything left to be done for this or can I remove it? super(txnId, scope, executeAtEpoch); this.mutation = mutation; } diff --git a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java index ff4a6e6ecb..a01580363f 100644 --- a/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java +++ b/src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java @@ -111,7 +111,7 @@ public interface AccordTopologyUpdate out.writeUnsignedVInt32(e.getKey()); RangesForEpochSerializer.instance.serialize(e.getValue(), out); } - //TODO (performance): local to what? Rather than serializing local we can serialize the node its relative too? that why when we deserialize we do globa.forNode(node) + //TODO (desired): local to what? Rather than serializing local we can serialize the node its relative too? that why when we deserialize we do globa.forNode(node) // this also decreases the size as we don't have redundent shards TopologySerializers.topology.serialize(from.local, out); TopologySerializers.topology.serialize(from.global, out); diff --git a/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java index 6ec96618a1..25a40a6a41 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java @@ -27,7 +27,6 @@ import accord.primitives.Ballot; import accord.primitives.Participants; import accord.primitives.Route; import accord.primitives.SaveStatus; -import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.UnversionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; import org.apache.cassandra.io.util.DataOutputPlus; @@ -63,6 +62,11 @@ public class BeginInvalidationSerializers public static final UnversionedSerializer reply = new UnversionedSerializer<>() { + private static final int ACCEPTED_FAST_PATH = 0x1; + private static final int HAS_TRUNCATED = 0x2; + private static final int HAS_ROUTE = 0x4; + private static final int HAS_HOME_KEY = 0x8; + @Override public void serialize(InvalidateReply reply, DataOutputPlus out) throws IOException { @@ -70,24 +74,28 @@ public class BeginInvalidationSerializers CommandSerializers.ballot.serialize(reply.accepted, out); CommandSerializers.saveStatus.serialize(reply.maxStatus, out); CommandSerializers.saveStatus.serialize(reply.maxKnowledgeStatus, out); - out.writeBoolean(reply.acceptedFastPath); - KeySerializers.nullableParticipants.serialize(reply.truncated, out); - KeySerializers.nullableRoute.serialize(reply.route, out); - KeySerializers.nullableRoutingKey.serialize(reply.homeKey, out); + int flags = (reply.acceptedFastPath ? ACCEPTED_FAST_PATH : 0) + | (reply.truncated != null ? HAS_TRUNCATED : 0) + | (reply.route != null ? HAS_ROUTE : 0) + | (reply.homeKey != null && reply.route == null ? HAS_HOME_KEY : 0); + out.writeByte(flags); + if (reply.truncated != null) KeySerializers.participants.serialize(reply.truncated, out); + if (reply.route != null) KeySerializers.route.serialize(reply.route, out); + else if (reply.homeKey != null) KeySerializers.routingKey.serialize(reply.homeKey, out); } @Override public InvalidateReply deserialize(DataInputPlus in) throws IOException { - // TODO (expected): use headers instead of nullable+bool serializers Ballot supersededBy = CommandSerializers.ballot.deserialize(in); Ballot accepted = CommandSerializers.ballot.deserialize(in); SaveStatus maxStatus = CommandSerializers.saveStatus.deserialize(in); SaveStatus maxKnowledgeStatus = CommandSerializers.saveStatus.deserialize(in); - boolean acceptedFastPath = in.readBoolean(); - Participants truncated = KeySerializers.nullableParticipants.deserialize(in); - Route route = KeySerializers.nullableRoute.deserialize(in); - RoutingKey homeKey = KeySerializers.nullableRoutingKey.deserialize(in); + byte flags = in.readByte(); + boolean acceptedFastPath = (flags & ACCEPTED_FAST_PATH) != 0; + Participants truncated = (flags & HAS_TRUNCATED) != 0 ? KeySerializers.participants.deserialize(in) : null; + Route route = (flags & HAS_ROUTE) != 0 ? KeySerializers.route.deserialize(in) : null; + RoutingKey homeKey = (flags & HAS_HOME_KEY) != 0 ? KeySerializers.routingKey.deserialize(in) : route != null ? route.homeKey() : null; return new InvalidateReply(supersededBy, accepted, maxStatus, maxKnowledgeStatus, acceptedFastPath, truncated, route, homeKey); } @@ -98,10 +106,10 @@ public class BeginInvalidationSerializers + CommandSerializers.ballot.serializedSize(reply.accepted) + CommandSerializers.saveStatus.serializedSize(reply.maxStatus) + CommandSerializers.saveStatus.serializedSize(reply.maxKnowledgeStatus) - + TypeSizes.BOOL_SIZE - + KeySerializers.nullableParticipants.serializedSize(reply.truncated) - + KeySerializers.nullableRoute.serializedSize(reply.route) - + KeySerializers.nullableRoutingKey.serializedSize(reply.homeKey); + + 1 + + (reply.truncated != null ? KeySerializers.participants.serializedSize(reply.truncated) : 0) + + (reply.route != null ? KeySerializers.route.serializedSize(reply.route) : 0) + + (reply.homeKey != null && reply.route == null ? KeySerializers.routingKey.serializedSize(reply.homeKey) : 0); } }; } 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 ec3fa6e109..a231b21251 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/CommandSerializers.java @@ -22,7 +22,6 @@ import java.io.IOException; import java.nio.ByteBuffer; import com.google.common.annotations.VisibleForTesting; -import com.google.common.base.Preconditions; import accord.api.Query; import accord.api.Read; @@ -36,6 +35,7 @@ import accord.primitives.Known; import accord.primitives.Known.KnownDeps; import accord.primitives.PartialTxn; import accord.primitives.Participants; +import accord.primitives.Routable; import accord.primitives.Route; import accord.primitives.SaveStatus; import accord.primitives.Seekables; @@ -45,9 +45,11 @@ import accord.primitives.Timestamp; import accord.primitives.TimestampWithUniqueHlc; import accord.primitives.Txn; import accord.primitives.TxnId; +import accord.primitives.Unseekables; import accord.primitives.Writes; import accord.utils.Invariants; import org.apache.cassandra.db.TypeSizes; +import org.apache.cassandra.db.marshal.ByteBufferAccessor; import org.apache.cassandra.db.marshal.ValueAccessor; import org.apache.cassandra.io.UnversionedSerializer; import org.apache.cassandra.io.VersionedSerializer; @@ -67,9 +69,8 @@ public class CommandSerializers { } - public static final TimestampSerializer txnId = new TimestampSerializer<>(TxnId::fromBits); - public static final TimestampSerializer timestamp = new TimestampSerializer<>(Timestamp::fromBits); - public static final UnversionedSerializer nullableTimestamp = NullableSerializer.wrap(timestamp); + public static final VariableWidthTimestampSerializer txnId = new VariableWidthTimestampSerializer<>(TxnId::fromValues); + public static final VariableWidthTimestampSerializer timestamp = new VariableWidthTimestampSerializer<>(Timestamp::fromValues); 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(); @@ -314,95 +315,202 @@ public class CommandSerializers } } - // TODO (expected): optimise using subset serializers, or perhaps simply with some deduping key serializer - public static class StoreParticipantsSerializer implements IVersionedSerializer + public static class StoreParticipantsSerializer implements UnversionedSerializer { static final int HAS_ROUTE = 0x1; - static final int HAS_TOUCHED_EQUALS_ROUTE = 0x2; - static final int TOUCHES_EQUALS_HAS_TOUCHED = 0x4; - static final int OWNS_EQUALS_TOUCHES = 0x8; - static final int EXECUTES_IS_NULL = 0x10; - static final int EXECUTES_IS_OWNS = 0x20; - static final int WAITSON_IS_OWNS = 0x40; + static final int ROUTE_EQUALS_SUPERSET = 0x2; + static final int HAS_TOUCHED_EQUALS_SUPERSET = 0x4; + static final int TOUCHES_EQUALS_HAS_TOUCHED = 0x8; + static final int OWNS_EQUALS_TOUCHES = 0x10; + static final int EXECUTES_IS_NULL = 0x20; + static final int EXECUTES_IS_OWNS = 0x40; + static final int WAITSON_IS_OWNS = 0x80; @Override - public void serialize(StoreParticipants t, DataOutputPlus out, Version version) throws IOException + public void serialize(StoreParticipants t, DataOutputPlus out) throws IOException { - boolean hasRoute = t.route() != null; - boolean hasTouchedEqualsRoute = t.route() == t.hasTouched(); - boolean touchesEqualsHasTouched = t.touches() == t.hasTouched(); - boolean ownsEqualsTouches = t.owns() == t.touches(); - boolean executesIsNull = t.executes() == null; - boolean executesIsOwns = !executesIsNull && t.executes() == t.owns(); - boolean waitsOnIsOwns = !executesIsNull && t.waitsOn() == t.owns(); + Participants hasTouched = t.hasTouched(); + Route route = t.route(); + Participants owns = t.owns(); + Participants executes = t.executes(); + Participants touches = t.touches(); + boolean hasRoute = route != null; + boolean touchesEqualsHasTouched = touches == hasTouched; + boolean ownsEqualsTouches = owns == touches; + boolean executesIsNull = executes == null; + boolean executesIsOwns = !executesIsNull && executes == owns; + boolean waitsOnIsOwns = !executesIsNull && t.waitsOn() == owns; + boolean encodeSubsets = hasTouched.domain() == Routable.Domain.Key; + Participants superset = !hasRoute ? hasTouched : encodeSubsets ? route.with((Participants)hasTouched) : route; + boolean routeEqualsSuperset = route == superset; + boolean hasTouchedEqualsSuperset = hasTouched == superset; out.writeByte((hasRoute ? HAS_ROUTE : 0) - | (hasTouchedEqualsRoute ? HAS_TOUCHED_EQUALS_ROUTE : 0) + | (routeEqualsSuperset ? ROUTE_EQUALS_SUPERSET : 0) + | (hasTouchedEqualsSuperset ? HAS_TOUCHED_EQUALS_SUPERSET : 0) | (touchesEqualsHasTouched ? TOUCHES_EQUALS_HAS_TOUCHED : 0) | (ownsEqualsTouches ? OWNS_EQUALS_TOUCHES : 0) | (executesIsNull ? EXECUTES_IS_NULL : 0) | (executesIsOwns ? EXECUTES_IS_OWNS : 0) | (waitsOnIsOwns ? WAITSON_IS_OWNS : 0) ); - if (hasRoute) KeySerializers.route.serialize(t.route(), out); - if (!hasTouchedEqualsRoute) KeySerializers.participants.serialize(t.hasTouched(), out); - if (!touchesEqualsHasTouched) KeySerializers.participants.serialize(t.touches(), out); - if (!ownsEqualsTouches) KeySerializers.participants.serialize(t.owns(), out); - if (!executesIsNull && !executesIsOwns) KeySerializers.participants.serialize(t.executes(), out); - if (!executesIsNull && !waitsOnIsOwns) KeySerializers.participants.serialize(t.waitsOn(), out); + + KeySerializers.participants.serialize(superset, out); + if (encodeSubsets) + { + if (hasRoute && !routeEqualsSuperset) KeySerializers.route.serializeSubset(route, superset, out); + if (!hasTouchedEqualsSuperset) KeySerializers.participants.serializeSubset(hasTouched, superset, out); + if (!touchesEqualsHasTouched) KeySerializers.participants.serializeSubset(touches, superset, out); + if (!ownsEqualsTouches) KeySerializers.participants.serializeSubset(owns, superset, out); + if (!executesIsNull && !executesIsOwns) KeySerializers.participants.serializeSubset(executes, superset, out); + if (!executesIsNull && !waitsOnIsOwns) KeySerializers.participants.serializeSubset(t.waitsOn(), superset, out); + } + else + { + if (hasRoute && !routeEqualsSuperset) KeySerializers.route.serialize(route, out); + if (!hasTouchedEqualsSuperset) KeySerializers.participants.serialize(hasTouched, out); + if (!touchesEqualsHasTouched) KeySerializers.participants.serialize(touches, out); + if (!ownsEqualsTouches) KeySerializers.participants.serialize(owns, out); + if (!executesIsNull && !executesIsOwns) KeySerializers.participants.serialize(executes, out); + if (!executesIsNull && !waitsOnIsOwns) KeySerializers.participants.serialize(t.waitsOn(), out); + } } - public void skip(DataInputPlus in, Version version) throws IOException + public void skip(DataInputPlus in) throws IOException { int flags = in.readByte(); - if (0 != (flags & HAS_ROUTE)) KeySerializers.route.skip(in); - if (0 == (flags & HAS_TOUCHED_EQUALS_ROUTE)) KeySerializers.participants.skip(in); - if (0 == (flags & TOUCHES_EQUALS_HAS_TOUCHED)) KeySerializers.participants.skip(in); - if (0 == (flags & OWNS_EQUALS_TOUCHES)) KeySerializers.participants.skip(in); - if (0 == (flags & (EXECUTES_IS_OWNS | EXECUTES_IS_NULL))) KeySerializers.participants.skip(in); - if (0 == (flags & (WAITSON_IS_OWNS | EXECUTES_IS_NULL))) KeySerializers.participants.skip(in); + Unseekables.UnseekablesKind kind = KeySerializers.participants.readKind(in); + int supersetCount = KeySerializers.participants.countAndSkip(kind, in); + boolean skipSubset = kind.domain() == Routable.Domain.Key; + if (skipSubset) + { + if (0 != (flags & HAS_ROUTE) && 0 == (flags & ROUTE_EQUALS_SUPERSET)) KeySerializers.route.skipSubset(supersetCount, in); + if (0 == (flags & HAS_TOUCHED_EQUALS_SUPERSET)) KeySerializers.participants.skipSubset(supersetCount, in); + if (0 == (flags & TOUCHES_EQUALS_HAS_TOUCHED)) KeySerializers.participants.skipSubset(supersetCount, in); + if (0 == (flags & OWNS_EQUALS_TOUCHES)) KeySerializers.participants.skipSubset(supersetCount, in); + if (0 == (flags & (EXECUTES_IS_OWNS | EXECUTES_IS_NULL))) KeySerializers.participants.skipSubset(supersetCount, in); + if (0 == (flags & (WAITSON_IS_OWNS | EXECUTES_IS_NULL))) KeySerializers.participants.skipSubset(supersetCount, in); + } + else + { + if (0 != (flags & HAS_ROUTE) && 0 == (flags & ROUTE_EQUALS_SUPERSET)) KeySerializers.route.skip(in); + if (0 == (flags & HAS_TOUCHED_EQUALS_SUPERSET)) KeySerializers.participants.skip(in); + if (0 == (flags & TOUCHES_EQUALS_HAS_TOUCHED)) KeySerializers.participants.skip(in); + if (0 == (flags & OWNS_EQUALS_TOUCHES)) KeySerializers.participants.skip(in); + if (0 == (flags & (EXECUTES_IS_OWNS | EXECUTES_IS_NULL))) KeySerializers.participants.skip(in); + if (0 == (flags & (WAITSON_IS_OWNS | EXECUTES_IS_NULL))) KeySerializers.participants.skip(in); + } } @Override - public StoreParticipants deserialize(DataInputPlus in, Version version) throws IOException + public StoreParticipants deserialize(DataInputPlus in) throws IOException { int flags = in.readByte(); - Route route = 0 == (flags & HAS_ROUTE) ? null : KeySerializers.route.deserialize(in); - Participants hasTouched = 0 != (flags & HAS_TOUCHED_EQUALS_ROUTE) ? route : KeySerializers.participants.deserialize(in); - Participants touches = 0 != (flags & TOUCHES_EQUALS_HAS_TOUCHED) ? hasTouched : KeySerializers.participants.deserialize(in); - Participants owns = 0 != (flags & OWNS_EQUALS_TOUCHES) ? touches : KeySerializers.participants.deserialize(in); - Participants executes = 0 != (flags & EXECUTES_IS_NULL) ? null : 0 != (flags & EXECUTES_IS_OWNS) ? owns : KeySerializers.participants.deserialize(in); - Participants waitsOn = 0 != (flags & EXECUTES_IS_NULL) ? null : 0 != (flags & WAITSON_IS_OWNS) ? owns : KeySerializers.participants.deserialize(in); - return StoreParticipants.create(route, owns, executes, waitsOn, touches, hasTouched); + Participants superset = KeySerializers.participants.deserialize(in); + boolean decodeSubset = superset.domain() == Routable.Domain.Key; + if (decodeSubset) + { + Route route = 0 == (flags & HAS_ROUTE) ? null : 0 != (flags & ROUTE_EQUALS_SUPERSET) ? (Route)superset : KeySerializers.route.deserializeSubset(superset, in); + Participants hasTouched = 0 != (flags & HAS_TOUCHED_EQUALS_SUPERSET) ? superset : KeySerializers.participants.deserializeSubset(superset, in); + Participants touches = 0 != (flags & TOUCHES_EQUALS_HAS_TOUCHED) ? hasTouched : KeySerializers.participants.deserializeSubset(superset, in); + Participants owns = 0 != (flags & OWNS_EQUALS_TOUCHES) ? touches : KeySerializers.participants.deserializeSubset(superset, in); + Participants executes = 0 != (flags & EXECUTES_IS_NULL) ? null : 0 != (flags & EXECUTES_IS_OWNS) ? owns : KeySerializers.participants.deserializeSubset(superset, in); + Participants waitsOn = 0 != (flags & EXECUTES_IS_NULL) ? null : 0 != (flags & WAITSON_IS_OWNS) ? owns : KeySerializers.participants.deserializeSubset(superset, in); + return StoreParticipants.create(route, owns, executes, waitsOn, touches, hasTouched); + } + else + { + Route route = 0 == (flags & HAS_ROUTE) ? null : 0 != (flags & ROUTE_EQUALS_SUPERSET) ? (Route)superset : KeySerializers.route.deserialize(in); + Participants hasTouched = 0 != (flags & HAS_TOUCHED_EQUALS_SUPERSET) ? superset : KeySerializers.participants.deserialize(in); + Participants touches = 0 != (flags & TOUCHES_EQUALS_HAS_TOUCHED) ? hasTouched : KeySerializers.participants.deserialize(in); + Participants owns = 0 != (flags & OWNS_EQUALS_TOUCHES) ? touches : KeySerializers.participants.deserialize(in); + Participants executes = 0 != (flags & EXECUTES_IS_NULL) ? null : 0 != (flags & EXECUTES_IS_OWNS) ? owns : KeySerializers.participants.deserialize(in); + Participants waitsOn = 0 != (flags & EXECUTES_IS_NULL) ? null : 0 != (flags & WAITSON_IS_OWNS) ? owns : KeySerializers.participants.deserialize(in); + return StoreParticipants.create(route, owns, executes, waitsOn, touches, hasTouched); + } } @Override - public long serializedSize(StoreParticipants t, Version version) + public long serializedSize(StoreParticipants t) { - boolean hasRoute = t.route() != null; - boolean hasTouchedEqualsRoute = t.route() == t.hasTouched(); - boolean touchesEqualsHasTouched = t.touches() == t.hasTouched(); - boolean ownsEqualsTouches = t.owns() == t.touches(); - boolean executesIsNotNullAndNotOwns = t.executes() != null && t.owns() != t.executes(); - long size = 1; - if (hasRoute) size += KeySerializers.route.serializedSize(t.route()); - if (!hasTouchedEqualsRoute) size += KeySerializers.participants.serializedSize(t.hasTouched()); - if (!touchesEqualsHasTouched) size += KeySerializers.participants.serializedSize(t.touches()); - if (!ownsEqualsTouches) size += KeySerializers.participants.serializedSize(t.owns()); - if (executesIsNotNullAndNotOwns) size += KeySerializers.participants.serializedSize(t.executes()); + Participants hasTouched = t.hasTouched(); + Route route = t.route(); + Participants owns = t.owns(); + Participants executes = t.executes(); + Participants touches = t.touches(); + boolean hasRoute = route != null; + boolean touchesEqualsHasTouched = touches == hasTouched; + boolean ownsEqualsTouches = owns == touches; + boolean executesIsNull = executes == null; + boolean executesIsOwns = !executesIsNull && executes == owns; + boolean waitsOnIsOwns = !executesIsNull && t.waitsOn() == owns; + boolean encodeSubsets = hasTouched.domain() == Routable.Domain.Key; + Participants superset = !hasRoute ? hasTouched : encodeSubsets ? route.with((Participants)hasTouched) : route; + boolean routeEqualsSuperset = route == superset; + boolean hasTouchedEqualsSuperset = hasTouched == superset; + long size = 1 + KeySerializers.participants.serializedSize(superset); + if (encodeSubsets) + { + if (hasRoute && !routeEqualsSuperset) size += KeySerializers.route.serializedSubsetSize(route, superset); + if (!hasTouchedEqualsSuperset) size += KeySerializers.participants.serializedSubsetSize(hasTouched, superset); + if (!touchesEqualsHasTouched) size += KeySerializers.participants.serializedSubsetSize(touches, superset); + if (!ownsEqualsTouches) size += KeySerializers.participants.serializedSubsetSize(owns, superset); + if (!executesIsNull && !executesIsOwns) size += KeySerializers.participants.serializedSubsetSize(executes, superset); + if (!executesIsNull && !waitsOnIsOwns) size += KeySerializers.participants.serializedSubsetSize(t.waitsOn(), superset); + } + else + { + if (hasRoute && !routeEqualsSuperset) size += KeySerializers.route.serializedSize(route); + if (!hasTouchedEqualsSuperset) size += KeySerializers.participants.serializedSize(hasTouched); + if (!touchesEqualsHasTouched) size += KeySerializers.participants.serializedSize(touches); + if (!ownsEqualsTouches) size += KeySerializers.participants.serializedSize(owns); + if (!executesIsNull && !executesIsOwns) size += KeySerializers.participants.serializedSize(executes); + if (!executesIsNull && !waitsOnIsOwns) size += KeySerializers.participants.serializedSize(t.waitsOn()); + } return size; } } - public static class TimestampSerializer implements UnversionedSerializer + public static class VariableWidthTimestampSerializer implements UnversionedSerializer { - interface Factory + private static final int NODE_SHIFT = 0; + private static final int NODE_MASK = 0x3; + private static final int NODE_MIN_LENGTH = 1; + private static final int FLAGS_SHIFT = NODE_SHIFT + Integer.bitCount(NODE_MASK); + private static final int FLAGS_MASK = 0x1; + private static final int FLAGS_MIN_LENGTH = 1; + private static final int HLC_SHIFT = FLAGS_SHIFT + Integer.bitCount(FLAGS_MASK); + private static final int HLC_MASK = 0x3; + private static final int HLC_MIN_LENGTH = 5; + private static final int EPOCH_SHIFT = HLC_SHIFT + Integer.bitCount(HLC_MASK); + private static final int EPOCH_MASK = 0x3; + private static final int EPOCH_MIN_LENGTH = 3; + static final byte NULL_BYTE = (byte) 0x80; + static { - T create(long msb, long lsb, Node.Id node); + Invariants.require(EPOCH_MASK << EPOCH_SHIFT >= 0); } - private final TimestampSerializer.Factory factory; + interface Factory + { + T create(long epoch, long hlc, int flags, Node.Id node); + } - private TimestampSerializer(TimestampSerializer.Factory factory) + private final VariableWidthTimestampSerializer.Factory factory; + + T decodeSpecial(int encodingFlags) + { + Invariants.require(encodingFlags == NULL_BYTE); + return null; + } + + byte encodeSpecial(T value) + { + if (value != null) + return 0; + return NULL_BYTE; + } + + private VariableWidthTimestampSerializer(VariableWidthTimestampSerializer.Factory factory) { this.factory = factory; } @@ -410,121 +518,271 @@ public class CommandSerializers @Override public void serialize(T ts, DataOutputPlus out) throws IOException { - out.writeLong(ts.msb); - out.writeLong(ts.lsb); - TopologySerializers.nodeId.serialize(ts.node, out); + { + byte specialByte = encodeSpecial(ts); + if (specialByte != 0) + { + Invariants.require(specialByte < 0); + out.writeByte(specialByte); + return; + } + } + 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); + int encodingFlags = 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); + out.writeByte(encodingFlags); + out.writeLeastSignificantBytes(epoch, epochLength); + out.writeLeastSignificantBytes(hlc, hlcLength); + out.writeLeastSignificantBytes(flags, flagsLength); + out.writeLeastSignificantBytes(ts.node.id, nodeLength); + } + + // 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) + { + int position = offset; + Invariants.require(encodeSpecial(ts) == 0); + 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); + + long pack = packLength(epochLength, epochLength * 8, EPOCH_MIN_LENGTH, EPOCH_MASK); + pack |= epoch; + pack <<= 5; + pack |= packLength(hlcLength, 3, HLC_MIN_LENGTH, HLC_MASK); + pack |= hlc >>> ((hlcLength*8)-3); + accessor.putLeastSignificantBytes(dst, position, pack, epochLength + 1); + position += epochLength + 1; + + hlc <<= 3; + hlc |= packLength(flagsLength, 2, FLAGS_MIN_LENGTH, FLAGS_MASK); + hlc |= flags >>> ((flagsLength * 8) - 2); + accessor.putLeastSignificantBytes(dst, position, hlc, hlcLength); + position += hlcLength; + + pack = (long)flags << (2 + nodeLength * 8); + pack |= packLength(nodeLength, nodeLength * 8, NODE_MIN_LENGTH, NODE_MASK); + pack |= ts.node.id & 0xffffffffL; + accessor.putLeastSignificantBytes(dst, position, pack, flagsLength + nodeLength); + position += flagsLength + nodeLength; + return position - offset; } public int serialize(T ts, V dst, ValueAccessor accessor, int offset) { + { + byte specialByte = encodeSpecial(ts); + if (specialByte != 0) + { + Invariants.require(specialByte < 0); + accessor.putByte(dst, offset, specialByte); + return 1; + } + } + + 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); + int encodingFlags = 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); + int position = offset; - position += accessor.putLong(dst, position, ts.msb); - position += accessor.putLong(dst, position, ts.lsb); - position += TopologySerializers.nodeId.serialize(ts.node, dst, accessor, position); - int size = position - offset; - Preconditions.checkState(size == serializedSize()); - return size; + position += accessor.putByte(dst, position, (byte)encodingFlags); + position += accessor.putLeastSignificantBytes(dst, position, epoch, epochLength); + position += accessor.putLeastSignificantBytes(dst, position, hlc, hlcLength); + position += accessor.putLeastSignificantBytes(dst, position, flags, flagsLength); + position += accessor.putLeastSignificantBytes(dst, position, ts.node.id, nodeLength); + return position - offset; + } + + public ByteBuffer serialize(T ts) + { + int size = Math.toIntExact(serializedSize(ts)); + ByteBuffer result = ByteBuffer.allocate(size); + serialize(ts, result, ByteBufferAccessor.instance, 0); + return result; } public void serialize(T ts, ByteBuffer out) { - out.putLong(ts.msb); - out.putLong(ts.lsb); - TopologySerializers.nodeId.serialize(ts.node, out); + int position = out.position(); + position += serialize(ts, out, ByteBufferAccessor.instance, 0); + out.position(position); } public void skip(DataInputPlus in) throws IOException { - in.skipBytesFully(serializedSize()); + int encodingFlags = in.readByte(); + if (encodingFlags < 0) + return; + 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); } @Override public T deserialize(DataInputPlus in) throws IOException { - return factory.create(in.readLong(), - in.readLong(), - TopologySerializers.nodeId.deserialize(in)); + int encodingFlags = in.readByte(); + if (encodingFlags < 0) + return decodeSpecial(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); + long epoch = in.readLeastSignificantBytes(epochLength); + long hlc = in.readLeastSignificantBytes(hlcLength); + int flags = Math.toIntExact(in.readLeastSignificantBytes(flagsLength)); + int nodeId = (int)in.readLeastSignificantBytes(nodeLength); + return factory.create(epoch, hlc, flags, new Node.Id(nodeId)); } public T deserialize(V src, ValueAccessor accessor, int offset) { - long msb = accessor.getLong(src, offset); - offset += TypeSizes.LONG_SIZE; - long lsb = accessor.getLong(src, offset); - offset += TypeSizes.LONG_SIZE; - Node.Id node = TopologySerializers.nodeId.deserialize(src, accessor, offset); - return factory.create(msb, lsb, node); + int encodingFlags = accessor.getByte(src, offset); + if (encodingFlags < 0) + return decodeSpecial(encodingFlags); + ++offset; + 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); + long epoch = accessor.getLeastSignificantBytes(src, offset, epochLength); + offset += epochLength; + long hlc = accessor.getLeastSignificantBytes(src, offset, hlcLength); + offset += hlcLength; + int flags = Math.toIntExact(accessor.getLeastSignificantBytes(src, offset, flagsLength)); + offset += flagsLength; + int nodeId = (int)accessor.getLeastSignificantBytes(src, offset, nodeLength); + return factory.create(epoch, hlc, flags, new Node.Id(nodeId)); } public T deserialize(ByteBuffer buffer, int position) { - long msb = buffer.getLong(position); - position += TypeSizes.LONG_SIZE; - long lsb = buffer.getLong(position); - position += TypeSizes.LONG_SIZE; - Node.Id node = TopologySerializers.nodeId.deserialize(buffer, position); - return factory.create(msb, lsb, node); + return deserialize(buffer, ByteBufferAccessor.instance, position); + } + + public T deserialize(ByteBuffer buffer) + { + return deserialize(buffer, ByteBufferAccessor.instance, 0); + } + + // exactly the same fundamental format as deserialize(), only we interleave the length bits with the values, maintaining ordering + public T deserializeComparable(V src, ValueAccessor accessor, int offset) + { + int b = accessor.getByte(src, offset++); + int epochLength = decodeLength(b, 5, EPOCH_MIN_LENGTH, EPOCH_MASK); + long bits64 = accessor.getLeastSignificantBytes(src, offset, epochLength); + offset += epochLength; + long epoch = (b&0x1fL) << (epochLength*8 - 5); + epoch |= bits64 >>> 5; + + int hlcLength = decodeLength((int)bits64, 3, HLC_MIN_LENGTH, HLC_MASK); + long hlc = (bits64 & 0x7L) << (hlcLength*8 - 3); + bits64 = accessor.getLeastSignificantBytes(src, offset, hlcLength); + offset += hlcLength; + hlc |= bits64 >>> 3; + + int flagsLength = decodeLength((int)bits64, 2, FLAGS_MIN_LENGTH, FLAGS_MASK); + int flags = ((int)bits64 & 0x3) << (flagsLength*8-2); + int bits32 = (int) accessor.getLeastSignificantBytes(src, offset, flagsLength); + offset += flagsLength; + flags |= bits32 >>> 2; + + int nodeLength = decodeLength(bits32, 0, NODE_MIN_LENGTH, NODE_MASK); + int node = (int) accessor.getLeastSignificantBytes(src, offset, nodeLength); + return factory.create(epoch, hlc, flags, new Node.Id(node)); } @Override public long serializedSize(T ts) { - return serializedSize(); + if (encodeSpecial(ts) != 0) + return 1; + int epochLength = length(ts.epoch(), EPOCH_MIN_LENGTH); + int hlcLength = length(ts.hlc(), HLC_MIN_LENGTH); + int flagsLength = length(ts.flags(), FLAGS_MIN_LENGTH); + int nodeLength = length(ts.node.id, NODE_MIN_LENGTH); + return 1 + epochLength + hlcLength + flagsLength + nodeLength; } - public int serializedSize() + private static int length(long value, int minLength) { - return Math.toIntExact(TypeSizes.LONG_SIZE + // ts.msb - TypeSizes.LONG_SIZE + // ts.lsb - TopologySerializers.nodeId.serializedSize(null)); // ts.node + int length = ((64 + 7) - Long.numberOfLeadingZeros(value))/8; + return Math.max(length, minLength); + } + + private static int length(int value, int minLength) + { + int length = ((32 + 7) - Integer.numberOfLeadingZeros(value))/8; + return Math.max(length, minLength); + } + + private static int encodeLength(int length, int shift, int minLength, int mask) + { + int encoded = length - minLength; + Invariants.require(encoded <= mask); + return encoded << shift; + } + + private static long packLength(int length, int shift, int minLength, int mask) + { + int encoded = length - minLength; + Invariants.require(encoded <= mask); + return (long)encoded << shift; + } + + private static int decodeLength(int encodingFlags, int shift, int minLength, int mask) + { + return minLength + ((encodingFlags >>> shift) & mask); } } - public static class BallotSerializer implements UnversionedSerializer + public static class BallotSerializer extends VariableWidthTimestampSerializer { - final TimestampSerializer wrapped = new TimestampSerializer<>(Ballot::fromBits); - - @Override - public void serialize(Ballot t, DataOutputPlus out) throws IOException + private static final byte ZERO_BYTE = (byte) 0x81; + private static final byte MAX_BYTE = (byte) 0x82; + private BallotSerializer() { - if (t == null || t.equals(Ballot.ZERO) || t.equals(Ballot.MAX)) - { - out.writeByte(t == null ? 1 : t.equals(Ballot.ZERO) ? 2 : 3); - } - else - { - out.writeByte(0); - wrapped.serialize(t, out); - } + super(Ballot::fromValues); } @Override - public Ballot deserialize(DataInputPlus in) throws IOException + byte encodeSpecial(Ballot value) { - int flags = in.readByte(); - switch (flags) - { - default: throw new IOException("Corrupted input: expected [0..3], received: " + flags); - case 0: return wrapped.deserialize(in); - case 1: return null; - case 2: return Ballot.ZERO; - case 3: return Ballot.MAX; - } - } - - public void skip(DataInputPlus in) throws IOException - { - int flags = in.readByte(); - if (flags == 0) - wrapped.skip(in); + if (value == null) return NULL_BYTE; + if (value == Ballot.ZERO) return ZERO_BYTE; + if (value == Ballot.MAX) return MAX_BYTE; + return 0; } @Override - public long serializedSize(Ballot t) + Ballot decodeSpecial(int specialByte) { - if (t == null || t.equals(Ballot.ZERO) || t.equals(Ballot.MAX)) - return 1; - return 1 + wrapped.serializedSize(); + if (specialByte == NULL_BYTE) return null; + if (specialByte == ZERO_BYTE) return Ballot.ZERO; + if (specialByte == MAX_BYTE) return Ballot.MAX; + throw new IllegalArgumentException("Unexpected specialByte: " + specialByte); } } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java index 35b1896735..7d45a6006d 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java @@ -92,7 +92,7 @@ public class FetchSerializers FetchResponse response = (FetchResponse) reply; serializeNullable(response.unavailable, out, KeySerializers.ranges); serializeNullable(response.data, out, streamDataSerializer); - CommandSerializers.nullableTimestamp.serialize(response.safeToReadAfter, out); + CommandSerializers.timestamp.serialize(response.safeToReadAfter, out); } @Override @@ -104,7 +104,7 @@ public class FetchSerializers return new FetchResponse(deserializeNullable(in, KeySerializers.ranges), deserializeNullable(in, streamDataSerializer), - CommandSerializers.nullableTimestamp.deserialize(in)); + CommandSerializers.timestamp.deserialize(in)); } @Override @@ -117,7 +117,7 @@ public class FetchSerializers return TypeSizes.BYTE_SIZE + serializedNullableSize(response.unavailable, KeySerializers.ranges) + serializedNullableSize(response.data, streamDataSerializer) - + CommandSerializers.nullableTimestamp.serializedSize(response.safeToReadAfter); + + CommandSerializers.timestamp.serializedSize(response.safeToReadAfter); } }; diff --git a/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java b/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java index 61f7a02dc9..eb2cded874 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java @@ -19,6 +19,8 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; +import java.util.function.BiFunction; +import java.util.function.IntFunction; import accord.api.RoutingKey; import accord.primitives.AbstractKeys; @@ -26,10 +28,11 @@ import accord.primitives.AbstractRanges; import accord.primitives.AbstractUnseekableKeys; import accord.primitives.Range; import accord.primitives.Ranges; +import accord.primitives.Routable; import accord.primitives.RoutableKey; import accord.primitives.Routables; import accord.primitives.RoutingKeys; -import accord.primitives.Unseekables; +import accord.utils.UnhandledEnum; import net.nicoulaj.compilecommand.annotations.DontInline; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.util.DataInputPlus; @@ -69,62 +72,13 @@ public interface IVersionedWithKeysSerializer, T> extends */ long serializedSize(K keys, T t, Version version); - final class NullableWithKeysSerializer, T> implements IVersionedWithKeysSerializer - { - final IVersionedWithKeysSerializer wrapped; - public NullableWithKeysSerializer(IVersionedWithKeysSerializer wrapped) - { - this.wrapped = wrapped; - } - - @Override - public void serialize(T t, DataOutputPlus out, Version version) throws IOException - { - out.writeByte(t == null ? 0 : 1); - if (t != null) wrapped.serialize(t, out, version); - } - - @Override - public T deserialize(DataInputPlus in, Version version) throws IOException - { - if (in.readByte() == 0) return null; - return wrapped.deserialize(in, version); - } - - @Override - public long serializedSize(T t, Version version) - { - return t == null ? 1 : 1 + wrapped.serializedSize(t, version); - } - - @Override - public void serialize(K keys, T t, DataOutputPlus out, Version version) throws IOException - { - out.writeByte(t == null ? 0 : 1); - if (t != null) wrapped.serialize(keys, t, out, version); - } - - @Override - public T deserialize(K keys, DataInputPlus in, Version version) throws IOException - { - if (in.readByte() == 0) return null; - return wrapped.deserialize(keys, in, version); - } - - @Override - public long serializedSize(K keys, T t, Version version) - { - return t == null ? 1 : 1 + wrapped.serializedSize(keys, t, version); - } - } - abstract class AbstractWithKeysSerializer { /** * If both ends have a pre-shared superset of the columns we are serializing, we can send them much * more efficiently. Both ends must provide the identically same set of columns. */ - protected void serializeSubset(Routables serialize, Routables superset, DataOutputPlus out) throws IOException + protected void serializeSubsetInternal(Routables serialize, Routables superset, DataOutputPlus out) throws IOException { /** * We weight this towards small sets, and sets where the majority of items are present, since @@ -148,7 +102,7 @@ public interface IVersionedWithKeysSerializer, T> extends { switch (serialize.domain()) { - default: throw new AssertionError("Unhandled domain: " + serialize.domain()); + default: throw UnhandledEnum.unknown(serialize.domain()); case Key: out.writeUnsignedVInt(encodeBitmap((AbstractUnseekableKeys)serialize, (AbstractUnseekableKeys)superset, supersetCount)); break; @@ -161,7 +115,7 @@ public interface IVersionedWithKeysSerializer, T> extends { switch (serialize.domain()) { - default: throw new AssertionError("Unhandled domain: " + serialize.domain()); + default: throw UnhandledEnum.unknown(serialize.domain()); case Key: serializeLargeSubset((AbstractUnseekableKeys)serialize, serializeCount, (AbstractUnseekableKeys)superset, supersetCount, out); break; @@ -172,7 +126,7 @@ public interface IVersionedWithKeysSerializer, T> extends } } - public long serializedSubsetSize(Routables serialize, Routables superset) + public long serializedSubsetSizeInternal(Routables serialize, Routables superset) { int columnCount = serialize.size(); int supersetCount = superset.size(); @@ -184,7 +138,7 @@ public interface IVersionedWithKeysSerializer, T> extends { switch (serialize.domain()) { - default: throw new AssertionError("Unhandled domain: " + serialize.domain()); + default: throw UnhandledEnum.unknown(serialize.domain()); case Key: return TypeSizes.sizeofUnsignedVInt(encodeBitmap((AbstractUnseekableKeys)serialize, (AbstractUnseekableKeys)superset, supersetCount)); case Range: @@ -195,7 +149,7 @@ public interface IVersionedWithKeysSerializer, T> extends { switch (serialize.domain()) { - default: throw new AssertionError("Unhandled domain: " + serialize.domain()); + default: throw UnhandledEnum.unknown(serialize.domain()); case Key: return serializeLargeSubsetSize((AbstractUnseekableKeys)serialize, columnCount, (AbstractUnseekableKeys)superset, supersetCount); case Range: @@ -204,55 +158,6 @@ public interface IVersionedWithKeysSerializer, T> extends } } - public Unseekables deserializeSubset(Unseekables superset, DataInputPlus in) throws IOException - { - long encoded = in.readUnsignedVInt(); - int supersetCount = superset.size(); - if (encoded == 0L) - { - return superset; - } - else if (supersetCount >= 64) - { - return deserializeLargeSubset(in, superset, supersetCount, (int) encoded); - } - else - { - encoded ^= -1L >>> (64 - supersetCount); - int deserializeCount = Long.bitCount(encoded); - switch (superset.domain()) - { - default: throw new AssertionError("Unhandled domain: " + superset.domain()); - case Key: - { - AbstractUnseekableKeys keys = (AbstractUnseekableKeys) superset; - RoutingKey[] out = new RoutingKey[deserializeCount]; - int count = 0; - while (encoded != 0) - { - long lowestBit = Long.lowestOneBit(encoded); - out[count++] = keys.get(Long.numberOfTrailingZeros(lowestBit)); - encoded ^= lowestBit; - } - return RoutingKeys.ofSortedUnique(out); - } - case Range: - { - AbstractRanges ranges = (AbstractRanges)superset; - Range[] out = new Range[deserializeCount]; - int count = 0; - while (encoded != 0) - { - long lowestBit = Long.lowestOneBit(encoded); - out[count++] = ranges.get(Long.numberOfTrailingZeros(lowestBit)); - encoded ^= lowestBit; - } - return Ranges.ofSortedAndDeoverlapped(out); - } - } - } - } - // encodes a 1 bit for every *missing* column, on the assumption presence is more common, // and because this is consistent with encoding 0 to represent all present private static long encodeBitmap(AbstractKeys serialize, AbstractKeys superset, int supersetCount) @@ -323,44 +228,112 @@ public interface IVersionedWithKeysSerializer, T> extends } } - @DontInline - private Unseekables deserializeLargeSubset(DataInputPlus in, Unseekables superset, int supersetCount, int delta) throws IOException + public Routables deserializeSubsetInternal(Routables superset, DataInputPlus in) throws IOException { - int deserializeCount = supersetCount - delta; switch (superset.domain()) { - default: throw new AssertionError("Unhandled domain: " + superset.domain()); - case Key: - { - AbstractUnseekableKeys keys = (AbstractUnseekableKeys) superset; - RoutingKey[] out = new RoutingKey[deserializeCount]; - int supersetIndex = 0; - int count = 0; - while (count < deserializeCount) - { - int takeCount = in.readUnsignedVInt32(); - while (takeCount-- > 0) out[count++] = keys.get(supersetIndex++); - supersetIndex += in.readUnsignedVInt32(); - } - return RoutingKeys.ofSortedUnique(out); - } - case Range: - { - AbstractRanges ranges = (AbstractRanges)superset; - Range[] out = new Range[deserializeCount]; - int supersetIndex = 0; - int count = 0; - while (count < deserializeCount) - { - int takeCount = in.readUnsignedVInt32(); - while (takeCount-- > 0) out[count++] = ranges.get(supersetIndex++); - supersetIndex += in.readUnsignedVInt32(); - } - return Ranges.ofSortedAndDeoverlapped(out); - } + default: throw UnhandledEnum.unknown(superset.domain()); + case Key: return deserializeRoutingKeySubset((AbstractUnseekableKeys) superset, in, (ks, s) -> ks == null ? s : RoutingKeys.of(ks)); + case Range: return deserializeRangeSubset((AbstractRanges) superset, in, (rs, s) -> rs == null ? s : Ranges.of(rs)); } } + public void skipSubsetInternal(int supersetCount, DataInputPlus in) throws IOException + { + long encoded = in.readUnsignedVInt(); + if (supersetCount <= 64) + return; + + int deserializeCount = supersetCount - (int)encoded; + int count = 0; + while (count < deserializeCount) + { + count += in.readUnsignedVInt32(); + in.readUnsignedVInt32(); + } + } + + public T deserializeRoutingKeySubset(S superset, DataInputPlus in, BiFunction result) throws IOException + { + long encoded = in.readUnsignedVInt(); + int supersetCount = superset.size(); + if (encoded == 0L) + return result.apply(null, superset); + else if (supersetCount >= 64) + return result.apply(deserializeLargeRoutingKeySubset(in, superset, supersetCount, (int) encoded), superset); + else + return result.apply(deserializeSmallRoutingKeySubset(encoded, superset, supersetCount), superset); + } + + public T deserializeRangeSubset(S superset, DataInputPlus in, BiFunction result) throws IOException + { + long encoded = in.readUnsignedVInt(); + int supersetCount = superset.size(); + if (encoded == 0L) + return result.apply(null, superset); + else if (supersetCount >= 64) + return result.apply(deserializeLargeRangeSubset(in, superset, supersetCount, (int) encoded), superset); + else + return result.apply(deserializeSmallRangeSubsetArray(encoded, superset, supersetCount), superset); + } + + private RoutingKey[] deserializeSmallRoutingKeySubset(long encoded, AbstractUnseekableKeys superset, int supersetCount) + { + return deserializeSmallSubsetArray(encoded, superset, supersetCount, RoutingKey[]::new); + } + + private Range[] deserializeSmallRangeSubsetArray(long encoded, AbstractRanges superset, int supersetCount) + { + return deserializeSmallSubsetArray(encoded, superset, supersetCount, Range[]::new); + } + + private R[] deserializeSmallSubsetArray(long encoded, Routables superset, int supersetCount, IntFunction allocator) + { + encoded ^= -1L >>> (64 - supersetCount); + int deserializeCount = Long.bitCount(encoded); + R[] out = allocator.apply(deserializeCount); + int count = 0; + while (encoded != 0) + { + long lowestBit = Long.lowestOneBit(encoded); + out[count++] = superset.get(Long.numberOfTrailingZeros(lowestBit)); + encoded ^= lowestBit; + } + return out; + } + + @DontInline + private RoutingKey[] deserializeLargeRoutingKeySubset(DataInputPlus in, AbstractUnseekableKeys superset, int supersetCount, int delta) throws IOException + { + int deserializeCount = supersetCount - delta; + RoutingKey[] out = new RoutingKey[deserializeCount]; + int supersetIndex = 0; + int count = 0; + while (count < deserializeCount) + { + int takeCount = in.readUnsignedVInt32(); + while (takeCount-- > 0) out[count++] = superset.get(supersetIndex++); + supersetIndex += in.readUnsignedVInt32(); + } + return out; + } + + @DontInline + private Range[] deserializeLargeRangeSubset(DataInputPlus in, AbstractRanges superset, int supersetCount, int delta) throws IOException + { + int deserializeCount = supersetCount - delta; + Range[] out = new Range[deserializeCount]; + int supersetIndex = 0; + int count = 0; + while (count < deserializeCount) + { + int takeCount = in.readUnsignedVInt32(); + while (takeCount-- > 0) out[count++] = superset.get(supersetIndex++); + supersetIndex += in.readUnsignedVInt32(); + } + return out; + } + @DontInline private long serializeLargeSubsetSize(AbstractKeys serialize, int serializeCount, AbstractKeys superset, int supersetCount) { diff --git a/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java index c800974e88..a7d181e705 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java @@ -38,7 +38,7 @@ public class InformDurableSerializers { out.writeVInt(msg.minEpoch - msg.waitForEpoch); out.writeVInt(msg.maxEpoch - msg.waitForEpoch); - CommandSerializers.nullableTimestamp.serialize(msg.executeAt, out); + CommandSerializers.timestamp.serialize(msg.executeAt, out); CommandSerializers.durability.serialize(msg.durability, out); } @@ -47,7 +47,7 @@ public class InformDurableSerializers { long minEpoch = waitForEpoch + in.readVInt(); long maxEpoch = waitForEpoch + in.readVInt(); - Timestamp executeAt = CommandSerializers.nullableTimestamp.deserialize(in); + Timestamp executeAt = CommandSerializers.timestamp.deserialize(in); Status.Durability durability = CommandSerializers.durability.deserialize(in); return InformDurable.SerializationSupport.create(txnId, scope, executeAt, minEpoch, waitForEpoch, maxEpoch, durability); } @@ -57,7 +57,7 @@ public class InformDurableSerializers { return TypeSizes.sizeofVInt(msg.minEpoch - msg.waitForEpoch) + TypeSizes.sizeofVInt(msg.maxEpoch - msg.waitForEpoch) - + CommandSerializers.nullableTimestamp.serializedSize(msg.executeAt) + + CommandSerializers.timestamp.serializedSize(msg.executeAt) + CommandSerializers.durability.serializedSize(msg.durability); } }; 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 b8c3ceefd2..0c96f24207 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/KeySerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/KeySerializers.java @@ -19,11 +19,7 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; -import java.nio.ByteBuffer; -import java.util.EnumSet; -import java.util.Map; import java.util.Objects; -import java.util.TreeMap; import java.util.function.Function; import java.util.function.IntFunction; @@ -33,15 +29,18 @@ import accord.api.Key; import accord.api.RoutingKey; import accord.primitives.AbstractKeys; import accord.primitives.AbstractRanges; +import accord.primitives.AbstractUnseekableKeys; import accord.primitives.FullKeyRoute; import accord.primitives.FullRangeRoute; import accord.primitives.FullRoute; +import accord.primitives.KeyRoute; import accord.primitives.Keys; import accord.primitives.PartialKeyRoute; import accord.primitives.PartialRangeRoute; import accord.primitives.PartialRoute; import accord.primitives.Participants; import accord.primitives.Range; +import accord.primitives.RangeRoute; import accord.primitives.Ranges; import accord.primitives.Routable; import accord.primitives.RoutableKey; @@ -53,6 +52,8 @@ import accord.primitives.Seekables; import accord.primitives.Unseekables; import accord.primitives.Unseekables.UnseekablesKind; import accord.utils.Invariants; +import accord.utils.TinyEnumSet; +import accord.utils.UnhandledEnum; import org.apache.cassandra.db.TypeSizes; import org.apache.cassandra.io.UnversionedSerializer; import org.apache.cassandra.io.util.DataInputPlus; @@ -69,14 +70,14 @@ import static accord.utils.ArrayBuffers.cachedInts; public class KeySerializers { public static final AccordKeySerializer key; - public static final UnversionedSerializer routingKey; + public static final AccordSearchableKeySerializer routingKey; public static final UnversionedSerializer nullableRoutingKey; - public static final AbstractSearchableKeysSerializer routingKeys; + public static final AbstractSearchableRoutingKeysSerializer routingKeys; public static final UnversionedSerializer keys; - public static final AbstractSearchableKeysSerializer partialKeyRoute; - public static final AbstractSearchableKeysSerializer fullKeyRoute; + public static final AbstractSearchableRoutingKeysSerializer partialKeyRoute; + public static final AbstractSearchableRoutingKeysSerializer fullKeyRoute; public static final UnversionedSerializer range; public static final AbstractRangesSerializer ranges; @@ -130,11 +131,11 @@ public class KeySerializers final AccordSearchableKeySerializer routingKey; final UnversionedSerializer nullableRoutingKey; - final AbstractSearchableKeysSerializer routingKeys; + final AbstractSearchableRoutingKeysSerializer routingKeys; final UnversionedSerializer keys; - final AbstractSearchableKeysSerializer partialKeyRoute; - final AbstractSearchableKeysSerializer fullKeyRoute; + final AbstractSearchableRoutingKeysSerializer partialKeyRoute; + final AbstractSearchableRoutingKeysSerializer fullKeyRoute; final UnversionedSerializer range; final AbstractRangesSerializer ranges; @@ -168,7 +169,7 @@ public class KeySerializers this.range = range; this.nullableRoutingKey = NullableSerializer.wrap(routingKey); - this.routingKeys = new AbstractSearchableKeysSerializer<>(routingKey, RoutingKey[]::new) + this.routingKeys = new AbstractSearchableRoutingKeysSerializer<>(routingKey) { @Override RoutingKeys deserialize(DataInputPlus in, RoutingKey[] keys) { @@ -184,53 +185,25 @@ public class KeySerializers } }; - this.partialKeyRoute = new AbstractSearchableKeysSerializer<>(routingKey, RoutingKey[]::new) + this.partialKeyRoute = new AbstractKeyRouteSerializer<>(routingKey) { - @Override PartialKeyRoute deserialize(DataInputPlus in, RoutingKey[] keys) throws IOException + @Override + PartialKeyRoute construct(RoutingKey homeKey, RoutingKey[] keys) { - RoutingKey homeKey = routingKey.deserialize(in); return PartialKeyRoute.SerializationSupport.create(homeKey, keys); } - - @Override - public void serialize(PartialKeyRoute route, DataOutputPlus out) throws IOException - { - super.serialize(route, out); - routingKey.serialize(route.homeKey, out); - } - - @Override - public long serializedSize(PartialKeyRoute routables) - { - return super.serializedSize(routables) - + routingKey.serializedSize(routables.homeKey); - } }; - this.fullKeyRoute = new AbstractSearchableKeysSerializer<>(routingKey, RoutingKey[]::new) + this.fullKeyRoute = new AbstractKeyRouteSerializer<>(routingKey) { - @Override FullKeyRoute deserialize(DataInputPlus in, RoutingKey[] keys) throws IOException + @Override + FullKeyRoute construct(RoutingKey homeKey, RoutingKey[] keys) { - RoutingKey homeKey = routingKey.deserialize(in); return FullKeyRoute.SerializationSupport.create(homeKey, keys); } - - @Override - public void serialize(FullKeyRoute route, DataOutputPlus out) throws IOException - { - super.serialize(route, out); - routingKey.serialize(route.homeKey, out); - } - - @Override - public long serializedSize(FullKeyRoute routables) - { - return super.serializedSize(routables) - + routingKey.serializedSize(routables.homeKey); - } }; - this.ranges = new AbstractRangesSerializer<>(routingKey) + this.ranges = new AbstractRangesSerializer<>() { @Override public Ranges deserialize(DataInputPlus in, Range[] ranges) @@ -239,64 +212,35 @@ public class KeySerializers } }; - - this.partialRangeRoute = new AbstractRangesSerializer<>(routingKey) + this.partialRangeRoute = new AbstractRangeRouteSerializer<>() { - @Override PartialRangeRoute deserialize(DataInputPlus in, Range[] rs) throws IOException + @Override + PartialRangeRoute construct(RoutingKey homeKey, Range[] rs) { - RoutingKey homeKey = routingKey.deserialize(in); return PartialRangeRoute.SerializationSupport.create(homeKey, rs); } - - @Override - public void serialize(PartialRangeRoute route, DataOutputPlus out) throws IOException - { - super.serialize(route, out); - routingKey.serialize(route.homeKey, out); - } - - @Override - public long serializedSize(PartialRangeRoute rs) - { - return super.serializedSize(rs) - + routingKey.serializedSize(rs.homeKey); - } }; - this.fullRangeRoute = new AbstractRangesSerializer<>(routingKey) + this.fullRangeRoute = new AbstractRangeRouteSerializer<>() { - @Override FullRangeRoute deserialize(DataInputPlus in, Range[] Ranges) throws IOException + @Override + FullRangeRoute construct(RoutingKey homeKey, Range[] Ranges) { - RoutingKey homeKey = routingKey.deserialize(in); return FullRangeRoute.SerializationSupport.create(homeKey, Ranges); } - - @Override - public void serialize(FullRangeRoute route, DataOutputPlus out) throws IOException - { - super.serialize(route, out); - routingKey.serialize(route.homeKey, out); - } - - @Override - public long serializedSize(FullRangeRoute ranges) - { - return super.serializedSize(ranges) - + routingKey.serializedSize(ranges.homeKey()); - } }; - Function, AbstractRoutablesSerializer> factory = (a) -> new AbstractRoutablesSerializer<>(a, routingKeys, partialKeyRoute, fullKeyRoute, ranges, partialRangeRoute, fullRangeRoute); + Function, AbstractRoutablesSerializer> factory = (a) -> new AbstractRoutablesSerializer<>(a, routingKeys, partialKeyRoute, fullKeyRoute, ranges, partialRangeRoute, fullRangeRoute); - this.route = (AbstractRoutablesSerializer>) factory.apply(EnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.FullKeyRoute, UnseekablesKind.PartialRangeRoute, UnseekablesKind.FullRangeRoute)); + this.route = (AbstractRoutablesSerializer>) factory.apply(TinyEnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.FullKeyRoute, UnseekablesKind.PartialRangeRoute, UnseekablesKind.FullRangeRoute)); this.nullableRoute = NullableSerializer.wrap(route); - this.partialRoute = (AbstractRoutablesSerializer>) factory.apply(EnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.PartialRangeRoute)); - this.fullRoute = (AbstractRoutablesSerializer>) factory.apply(EnumSet.of(UnseekablesKind.FullKeyRoute, UnseekablesKind.FullRangeRoute)); + this.partialRoute = (AbstractRoutablesSerializer>) factory.apply(TinyEnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.PartialRangeRoute)); + this.fullRoute = (AbstractRoutablesSerializer>) factory.apply(TinyEnumSet.of(UnseekablesKind.FullKeyRoute, UnseekablesKind.FullRangeRoute)); this.nullableFullRoute = NullableSerializer.wrap(fullRoute); - this.unseekables = (AbstractRoutablesSerializer>) factory.apply(EnumSet.allOf(UnseekablesKind.class)); - this.participants = (AbstractRoutablesSerializer>) factory.apply(EnumSet.allOf(UnseekablesKind.class)); + this.unseekables = (AbstractRoutablesSerializer>) factory.apply(TinyEnumSet.allOf(UnseekablesKind.class)); + this.participants = (AbstractRoutablesSerializer>) factory.apply(TinyEnumSet.allOf(UnseekablesKind.class)); this.nullableParticipants = NullableSerializer.wrap(participants); this.seekables = new AbstractSeekablesSerializer(keys, ranges); @@ -305,18 +249,18 @@ public class KeySerializers public static class AbstractRoutablesSerializer> implements UnversionedSerializer { - final EnumSet permitted; - final AbstractSearchableKeysSerializer routingKeys; - final AbstractSearchableKeysSerializer partialKeyRoute; - final AbstractSearchableKeysSerializer fullKeyRoute; + final TinyEnumSet permitted; + final AbstractSearchableRoutingKeysSerializer routingKeys; + final AbstractSearchableRoutingKeysSerializer partialKeyRoute; + final AbstractSearchableRoutingKeysSerializer fullKeyRoute; final AbstractRangesSerializer ranges; final AbstractRangesSerializer partialRangeRoute; final AbstractRangesSerializer fullRangeRoute; - protected AbstractRoutablesSerializer(EnumSet permitted, - AbstractSearchableKeysSerializer routingKeys, - AbstractSearchableKeysSerializer partialKeyRoute, - AbstractSearchableKeysSerializer fullKeyRoute, + protected AbstractRoutablesSerializer(TinyEnumSet permitted, + AbstractSearchableRoutingKeysSerializer routingKeys, + AbstractSearchableRoutingKeysSerializer partialKeyRoute, + AbstractSearchableRoutingKeysSerializer fullKeyRoute, AbstractRangesSerializer ranges, AbstractRangesSerializer partialRangeRoute, AbstractRangesSerializer fullRangeRoute) @@ -334,8 +278,7 @@ public class KeySerializers public void serialize(RS t, DataOutputPlus out) throws IOException { UnseekablesKind kind = t.kind(); - if (!permitted.contains(kind)) - throw new IllegalArgumentException(); + Invariants.requireArgument(permitted.contains(kind)); switch (kind) { @@ -367,6 +310,41 @@ public class KeySerializers } } + public void serializeSubset(RS t, Unseekables superset, DataOutputPlus out) throws IOException + { + UnseekablesKind kind = t.kind(); + Invariants.requireArgument(permitted.contains(kind)); + + switch (kind) + { + default: throw new AssertionError(); + case RoutingKeys: + out.writeByte(1); + routingKeys.serializeSubset((RoutingKeys)t, superset, out); + break; + case PartialKeyRoute: + out.writeByte(2); + partialKeyRoute.serializeSubset((PartialKeyRoute)t, superset, out); + break; + case FullKeyRoute: + out.writeByte(3); + fullKeyRoute.serializeSubset((FullKeyRoute)t, superset, out); + break; + case RoutingRanges: + out.writeByte(4); + ranges.serializeSubset((Ranges)t, superset, out); + break; + case PartialRangeRoute: + out.writeByte(5); + partialRangeRoute.serializeSubset((PartialRangeRoute)t, superset, out); + break; + case FullRangeRoute: + out.writeByte(6); + fullRangeRoute.serializeSubset((FullRangeRoute)t, superset, out); + break; + } + } + @Override public RS deserialize(DataInputPlus in) throws IOException { @@ -375,7 +353,7 @@ public class KeySerializers RS result; switch (b) { - default: throw new IOException("Corrupted input: expected byte 1, 2, 3, 4 or 5; received " + b); + default: throw new IOException("Corrupted input: expected byte 1, 2, 3, 4, 5 or 6; received " + b); case 1: kind = UnseekablesKind.RoutingKeys; result = (RS)routingKeys.deserialize(in); break; case 2: kind = UnseekablesKind.PartialKeyRoute; result = (RS)partialKeyRoute.deserialize(in); break; case 3: kind = UnseekablesKind.FullKeyRoute; result = (RS)fullKeyRoute.deserialize(in); break; @@ -387,18 +365,92 @@ public class KeySerializers return result; } + public RS deserializeSubset(Unseekables superset, DataInputPlus in) throws IOException + { + byte b = in.readByte(); + UnseekablesKind kind; + RS result; + switch (b) + { + default: throw new IOException("Corrupted input: expected byte 1, 2, 3, 4 or 5; received " + b); + case 1: kind = UnseekablesKind.RoutingKeys; result = (RS)routingKeys.deserializeSubset((AbstractUnseekableKeys) superset, in); break; + case 2: kind = UnseekablesKind.PartialKeyRoute; result = (RS)partialKeyRoute.deserializeSubset((AbstractUnseekableKeys) superset, in); break; + case 3: kind = UnseekablesKind.FullKeyRoute; result = (RS)fullKeyRoute.deserializeSubset((AbstractUnseekableKeys) superset, in); break; + case 4: kind = UnseekablesKind.RoutingRanges; result = (RS)ranges.deserializeSubset((AbstractRanges) superset, in); break; + case 5: kind = UnseekablesKind.PartialRangeRoute; result = (RS)partialRangeRoute.deserializeSubset((AbstractRanges) superset, in); break; + case 6: kind = UnseekablesKind.FullRangeRoute; result = (RS)fullRangeRoute.deserializeSubset((AbstractRanges) superset, in); break; + } + Invariants.require(permitted.contains(kind)); + return result; + } + public void skip(DataInputPlus in) throws IOException + { + countAndSkip(in); + } + + public void skip(UnseekablesKind kind, DataInputPlus in) throws IOException + { + countAndSkip(kind, in); + } + + // return number of elements skipped + public int countAndSkip(DataInputPlus in) throws IOException { byte b = in.readByte(); switch (b) { default: throw new IOException("Corrupted input: expected byte 1, 2, 3, 4 or 5; received " + b); - case 1: routingKeys.skip(in); break; - case 2: partialKeyRoute.skip(in); break; - case 3: fullKeyRoute.skip(in); break; - case 4: ranges.skip(in); break; - case 5: partialRangeRoute.skip(in); break; - case 6: fullRangeRoute.skip(in); break; + case 1: return routingKeys.countAndSkip(in); + case 2: return partialKeyRoute.countAndSkip(in); + case 3: return fullKeyRoute.countAndSkip(in); + case 4: return ranges.countAndSkip(in); + case 5: return partialRangeRoute.countAndSkip(in); + case 6: return fullRangeRoute.countAndSkip(in); + } + } + + public int countAndSkip(UnseekablesKind kind, DataInputPlus in) throws IOException + { + switch (kind) + { + default: throw UnhandledEnum.unknown(kind); + case RoutingKeys: return routingKeys.countAndSkip(in); + case PartialKeyRoute: return partialKeyRoute.countAndSkip(in); + case FullKeyRoute: return fullKeyRoute.countAndSkip(in); + case RoutingRanges: return ranges.countAndSkip(in); + case PartialRangeRoute: return partialRangeRoute.countAndSkip(in); + case FullRangeRoute: return fullRangeRoute.countAndSkip(in); + } + } + + public Unseekables.UnseekablesKind readKind(DataInputPlus in) throws IOException + { + byte b = in.readByte(); + switch (b) + { + default: throw new IOException("Corrupted input: expected byte 1, 2, 3, 4 or 5; received " + b); + case 1: return UnseekablesKind.RoutingKeys; + case 2: return UnseekablesKind.PartialKeyRoute; + case 3: return UnseekablesKind.FullKeyRoute; + case 4: return UnseekablesKind.RoutingRanges; + case 5: return UnseekablesKind.PartialRangeRoute; + case 6: return UnseekablesKind.FullRangeRoute; + } + } + + public void skipSubset(int supersetCount, DataInputPlus in) throws IOException + { + byte b = in.readByte(); + switch (b) + { + default: throw new IOException("Corrupted input: expected byte 1, 2, 3, 4 or 5; received " + b); + case 1: routingKeys.skipSubset(supersetCount, in); break; + case 2: partialKeyRoute.skipSubset(supersetCount, in); break; + case 3: fullKeyRoute.skipSubset(supersetCount, in); break; + case 4: ranges.skipSubset(supersetCount, in); break; + case 5: partialRangeRoute.skipSubset(supersetCount, in); break; + case 6: fullRangeRoute.skipSubset(supersetCount, in); break; } } @@ -422,6 +474,26 @@ public class KeySerializers return 1 + fullRangeRoute.serializedSize((FullRangeRoute)t); } } + + public long serializedSubsetSize(RS t, Unseekables superset) + { + switch (t.kind()) + { + default: throw new AssertionError(); + case RoutingKeys: + return 1 + routingKeys.serializedSubsetSize((RoutingKeys)t, superset); + case PartialKeyRoute: + return 1 + partialKeyRoute.serializedSubsetSize((PartialKeyRoute)t, superset); + case FullKeyRoute: + return 1 + fullKeyRoute.serializedSubsetSize((FullKeyRoute)t, superset); + case RoutingRanges: + return 1 + ranges.serializedSubsetSize((Ranges)t, superset); + case PartialRangeRoute: + return 1 + partialRangeRoute.serializedSubsetSize((PartialRangeRoute)t, superset); + case FullRangeRoute: + return 1 + fullRangeRoute.serializedSubsetSize((FullRangeRoute)t, superset); + } + } } public static final UnversionedSerializer seekable = new UnversionedSerializer<>() @@ -574,30 +646,28 @@ 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> implements UnversionedSerializer + public abstract static class AbstractSearchableSerializer> extends IVersionedWithKeysSerializer.AbstractWithKeysSerializer implements UnversionedSerializer { - final AccordSearchableKeySerializer keySerializer; final IntFunction allocate; - public AbstractSearchableSerializer(AccordSearchableKeySerializer keySerializer, IntFunction allocate) + public AbstractSearchableSerializer(IntFunction allocate) { - this.keySerializer = keySerializer; this.allocate = allocate; } private int serializedSizeOfPrefix(Object prefix) { - return keySerializer.serializedSizeOfPrefix(prefix); + return routingKey.serializedSizeOfPrefix(prefix); } private void serializePrefix(Object prefix, DataOutputPlus out) throws IOException { - keySerializer.serializePrefix(prefix, out); + routingKey.serializePrefix(prefix, out); } private Object deserializePrefix(DataInputPlus in) throws IOException { - return keySerializer.deserializePrefix(in); + return routingKey.deserializePrefix(in); } // if we store Ranges, we have twice as many indexes @@ -651,6 +721,11 @@ 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 { @@ -696,12 +771,24 @@ public class KeySerializers serializeWithoutPrefixOrLength(keys.get(i), out); } + public void serializeSubset(RS keys, Routables superset, DataOutputPlus out) throws IOException + { + serializeSubsetInternal(keys, superset, out); + } + public void skip(DataInputPlus in) throws IOException + { + countAndSkip(in); + } + + // return number of elements skipped + public int countAndSkip(DataInputPlus in) throws IOException { int remaining = in.readUnsignedVInt32(); if (remaining == 0) - return; + return 0; + int total = 0; while (remaining > 0) { int count = remaining - in.readUnsignedVInt32(); @@ -718,7 +805,14 @@ public class KeySerializers int end = in.readInt(); in.skipBytesFully(end); } + total += count; } + return total; + } + + public void skipSubset(int supersetCount, DataInputPlus in) throws IOException + { + skipSubsetInternal(supersetCount, in); } @Override @@ -769,17 +863,17 @@ 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 AbstractSearchableKeysSerializer> extends AbstractSearchableSerializer implements UnversionedSerializer + public abstract static class AbstractSearchableRoutingKeysSerializer extends AbstractSearchableSerializer implements UnversionedSerializer { - public AbstractSearchableKeysSerializer(AccordSearchableKeySerializer keySerializer, IntFunction allocate) + public AbstractSearchableRoutingKeysSerializer(AccordSearchableKeySerializer serializer) { - super(keySerializer, allocate); + super(RoutingKey[]::new); } @Override final int fixedKeyLengthForPrefix(Object prefix) { - return keySerializer.fixedKeyLengthForPrefix(prefix); + return routingKey.fixedKeyLengthForPrefix(prefix); } @Override @@ -789,15 +883,15 @@ public class KeySerializers } @Override - final int serializedSizeWithoutPrefix(K routable) + final int serializedSizeWithoutPrefix(RoutingKey routable) { - return keySerializer.serializedSizeWithoutPrefix(routable); + return routingKey.serializedSizeWithoutPrefix(routable); } @Override - final void serializeWithoutPrefixOrLength(K routable, DataOutputPlus out) throws IOException + final void serializeWithoutPrefixOrLength(RoutingKey routable, DataOutputPlus out) throws IOException { - keySerializer.serializeWithoutPrefixOrLength(routable, out); + routingKey.serializeWithoutPrefixOrLength(routable, out); } @Override @@ -812,29 +906,117 @@ public class KeySerializers } @Override - final K deserializeWithPrefix(Object prefix, int length, DataInputPlus in) throws IOException + final RoutingKey deserializeWithPrefix(Object prefix, int length, DataInputPlus in) throws IOException { - return keySerializer.deserializeWithPrefix(prefix, length, in); + return routingKey.deserializeWithPrefix(prefix, length, in); } @Override - final K deserializeWithPrefix(Object prefix, int lengthIndex, int[] lengths, DataInputPlus in) throws IOException + final RoutingKey deserializeWithPrefix(Object prefix, int lengthIndex, int[] lengths, DataInputPlus in) throws IOException { - return keySerializer.deserializeWithPrefix(prefix, lengths[lengthIndex], in); + return routingKey.deserializeWithPrefix(prefix, lengths[lengthIndex], in); + } + + public KS deserializeSubset(AbstractUnseekableKeys superset, DataInputPlus in) throws IOException + { + RoutingKey[] keys = deserializeRoutingKeySubset(superset, in, (ks, s) -> ks == null ? s.unsafeKeys() : ks); + return deserialize(in, keys); + } + } + + public abstract static class AbstractKeyRouteSerializer extends AbstractSearchableRoutingKeysSerializer + { + public AbstractKeyRouteSerializer(AccordSearchableKeySerializer serializer) + { + super(serializer); + } + + abstract KS construct(RoutingKey homeKey, RoutingKey[] keys); + + @Override + KS deserialize(DataInputPlus in, RoutingKey[] keys) throws IOException + { + int i = in.readUnsignedVInt32(); + RoutingKey homeKey = i == 0 ? routingKey.deserialize(in) : keys[i - 1]; + return construct(homeKey, keys); + } + + @Override + public int countAndSkip(DataInputPlus in) throws IOException + { + int count = super.countAndSkip(in); + completeSkip(in); + return count; + } + + @Override + public void skipSubset(int supersetCount, DataInputPlus in) throws IOException + { + skipSubsetInternal(supersetCount, in); + completeSkip(in); + } + + @Override + public void serialize(KS route, DataOutputPlus out) throws IOException + { + super.serialize(route, out); + completeSerialize(route, out); + } + + @Override + public void serializeSubset(KS route, Routables superset, DataOutputPlus out) throws IOException + { + super.serializeSubset(route, superset, out); + completeSerialize(route, out); + } + + @Override + public long serializedSize(KS route) + { + return super.serializedSize(route) + + completeSerializedSize(route); + } + + @Override + public long serializedSubsetSize(KS route, Routables superset) + { + return super.serializedSubsetSize(route, superset) + + completeSerializedSize(route); + } + + private void completeSerialize(KS route, DataOutputPlus out) throws IOException + { + int i = route.indexOf(route.homeKey()); + out.writeUnsignedVInt32(Math.max(0, 1 + i)); + if (i < 0) routingKey.serialize(route.homeKey, out); + } + + private void completeSkip(DataInputPlus in) throws IOException + { + int i = in.readUnsignedVInt32(); + if (i == 0) routingKey.skip(in); + } + + private long completeSerializedSize(KS route) + { + int i = route.indexOf(route.homeKey()); + long size = TypeSizes.sizeofUnsignedVInt(Math.max(0, 1 + i)); + if (i < 0) size += routingKey.serializedSize(route.homeKey); + return size; } } public abstract static class AbstractRangesSerializer extends AbstractSearchableSerializer implements UnversionedSerializer { - public AbstractRangesSerializer(AccordSearchableKeySerializer keySerializer) + public AbstractRangesSerializer() { - super(keySerializer, Range[]::new); + super(Range[]::new); } @Override int fixedKeyLengthForPrefix(Object prefix) { - return keySerializer.fixedKeyLengthForPrefix(prefix) * 2; + return routingKey.fixedKeyLengthForPrefix(prefix) * 2; } @Override @@ -846,15 +1028,15 @@ public class KeySerializers @Override final int serializedSizeWithoutPrefix(Range range) { - return keySerializer.serializedSizeWithoutPrefix(range.start()) - + keySerializer.serializedSizeWithoutPrefix(range.end()); + return routingKey.serializedSizeWithoutPrefix(range.start()) + + routingKey.serializedSizeWithoutPrefix(range.end()); } @Override final void serializeWithoutPrefixOrLength(Range key, DataOutputPlus out) throws IOException { - keySerializer.serializeWithoutPrefixOrLength(key.start(), out); - keySerializer.serializeWithoutPrefixOrLength(key.end(), out); + routingKey.serializeWithoutPrefixOrLength(key.start(), out); + routingKey.serializeWithoutPrefixOrLength(key.end(), out); } @Override @@ -864,9 +1046,9 @@ public class KeySerializers for (int i = startIndex; i < endIndex; ++i) { Range r = ranges.get(i); - endOffset += keySerializer.serializedSizeWithoutPrefix(r.start()); + endOffset += routingKey.serializedSizeWithoutPrefix(r.start()); out.writeInt(endOffset); - endOffset += keySerializer.serializedSizeWithoutPrefix(r.end()); + endOffset += routingKey.serializedSizeWithoutPrefix(r.end()); out.writeInt(endOffset); } } @@ -874,40 +1056,84 @@ public class KeySerializers @Override final Range deserializeWithPrefix(Object prefix, int length, DataInputPlus in) throws IOException { - RoutingKey start = keySerializer.deserializeWithPrefix(prefix, length/2, in); - RoutingKey end = keySerializer.deserializeWithPrefix(prefix, length/2, in); + RoutingKey start = routingKey.deserializeWithPrefix(prefix, length/2, in); + RoutingKey end = routingKey.deserializeWithPrefix(prefix, length/2, in); return start.rangeFactory().newRange(start, end); } @Override final Range deserializeWithPrefix(Object prefix, int lengthIndex, int[] lengths, DataInputPlus in) throws IOException { - RoutingKey start = keySerializer.deserializeWithPrefix(prefix, lengths[lengthIndex * 2], in); - RoutingKey end = keySerializer.deserializeWithPrefix(prefix, lengths[lengthIndex * 2 + 1], in); + RoutingKey start = routingKey.deserializeWithPrefix(prefix, lengths[lengthIndex * 2], in); + RoutingKey end = routingKey.deserializeWithPrefix(prefix, lengths[lengthIndex * 2 + 1], in); return start.rangeFactory().newRange(start, end); } + + public RS deserializeSubset(AbstractRanges superset, DataInputPlus in) throws IOException + { + Range[] ranges = deserializeRangeSubset(superset, in, (rs, s) -> rs == null ? s.unsafeRanges() : rs); + return deserialize(in, ranges); + } } - public static Map rangesToBlobMap(Ranges ranges) + public abstract static class AbstractRangeRouteSerializer extends AbstractRangesSerializer { - TreeMap result = new TreeMap<>(); - for (Range range : ranges) + public AbstractRangeRouteSerializer() { - result.put(TokenKey.serializer.serialize((TokenKey) range.start()), - TokenKey.serializer.serialize((TokenKey) range.end())); + super(); + } + + abstract RS construct(RoutingKey homeKey, Range[] ranges); + + @Override + RS deserialize(DataInputPlus in, Range[] ranges) throws IOException + { + RoutingKey homeKey = routingKey.deserialize(in); + return construct(homeKey, ranges); + } + + @Override + public int countAndSkip(DataInputPlus in) throws IOException + { + int count = super.countAndSkip(in); + routingKey.skip(in); + return count; + } + + @Override + public void skipSubset(int supersetCount, DataInputPlus in) throws IOException + { + super.skipSubset(supersetCount, in); + routingKey.skip(in); + } + + @Override + public void serialize(RS route, DataOutputPlus out) throws IOException + { + super.serialize(route, out); + routingKey.serialize(route.homeKey, out); + } + + @Override + public void serializeSubset(RS route, Routables superset, DataOutputPlus out) throws IOException + { + super.serializeSubset(route, superset, out); + routingKey.serialize(route.homeKey, out); + } + + @Override + public long serializedSize(RS route) + { + return super.serializedSize(route) + + routingKey.serializedSize(route.homeKey); + } + + @Override + public long serializedSubsetSize(RS route, Routables superset) + { + return super.serializedSubsetSize(route, superset) + routingKey.serializedSize(route.homeKey); } - return result; } - public static Ranges blobMapToRanges(Map blobMap) - { - int i = 0; - Range[] ranges = new Range[blobMap.size()]; - for (Map.Entry e : blobMap.entrySet()) - { - ranges[i++] = TokenRange.create(TokenKey.serializer.deserialize(e.getKey()), - TokenKey.serializer.deserialize(e.getValue())); - } - return Ranges.of(ranges); - } + } diff --git a/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java b/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java index 0101efa04a..0e3413905f 100644 --- a/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java +++ b/src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java @@ -26,7 +26,7 @@ import org.apache.cassandra.io.util.DataOutputPlus; public class ResultSerializers { - // TODO (expected): this is meant to encode e.g. whether the transaction's condition met or not + // TODO (expected): this is meant to encode e.g. whether the transaction's condition met or not for clients to later query public static final Result APPLIED = new Result() { @Override diff --git a/src/java/org/apache/cassandra/utils/vint/VIntCoding.java b/src/java/org/apache/cassandra/utils/vint/VIntCoding.java index 9a0e2c5cb7..a444f4147b 100644 --- a/src/java/org/apache/cassandra/utils/vint/VIntCoding.java +++ b/src/java/org/apache/cassandra/utils/vint/VIntCoding.java @@ -308,6 +308,17 @@ public class VIntCoding return retval; } + public static int readLengthOfVInt(V input, ValueAccessor accessor, int position) + { + byte firstByte = accessor.getByte(input, position); + if (firstByte >= 0) + return 1; + + int extraBytes = accord.utils.VIntCoding.numberOfExtraBytesToRead(firstByte); + return 1 + extraBytes; + } + + /** * Computes size of an unsigned vint that starts at readerIndex of the provided ByteBuf. * @@ -380,6 +391,16 @@ public class VIntCoding return checkedCast(readUnsignedVInt(input, position)); } + public static int readLengthOfVInt(ByteBuffer in, int position) + { + byte firstByte = in.get(position); + if (firstByte >= 0) + return 1; + + int extraBytes = numberOfExtraBytesToRead(firstByte); + return 1 + extraBytes; + } + // & this with the first byte to give the value part for a given extraBytesToRead encoded in the byte public static int firstByteValueMask(int extraBytesToRead) { diff --git a/test/unit/org/apache/cassandra/cql3/validation/entities/VectorsTest.java b/test/unit/org/apache/cassandra/cql3/validation/entities/VectorsTest.java index 7154d0295e..2d77428dae 100644 --- a/test/unit/org/apache/cassandra/cql3/validation/entities/VectorsTest.java +++ b/test/unit/org/apache/cassandra/cql3/validation/entities/VectorsTest.java @@ -25,8 +25,6 @@ import org.junit.Test; import org.apache.cassandra.cql3.CQLTester; -import static org.apache.cassandra.ServerTestUtils.daemonInitialization; - public class VectorsTest extends CQLTester { @BeforeClass diff --git a/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java b/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java index c9fe2e9060..70506bb722 100644 --- a/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java +++ b/test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java @@ -396,7 +396,7 @@ public class CheckpointIntervalArrayIndexTest Map files = new EnumMap<>(IndexComponent.class); for (IndexComponent c : descriptor.getLiveComponents()) files.put(c, new FileHandle.Builder(descriptor.fileFor(c)).mmapped(true).complete()); - List segments = RouteIndexFormat.readSegements(files); + List segments = RouteIndexFormat.readSegments(files); files.remove(IndexComponent.SEGMENT).close(); files.remove(IndexComponent.METADATA).close(); diff --git a/test/unit/org/apache/cassandra/io/Serializers.java b/test/unit/org/apache/cassandra/io/Serializers.java index d911e07aed..16b56b9c39 100644 --- a/test/unit/org/apache/cassandra/io/Serializers.java +++ b/test/unit/org/apache/cassandra/io/Serializers.java @@ -19,6 +19,7 @@ package org.apache.cassandra.io; import java.io.IOException; +import java.nio.ByteBuffer; import accord.utils.LazyToString; import accord.utils.ReflectionUtils; @@ -37,9 +38,14 @@ public class Serializers long expectedSize = serializer.serializedSize(input); serializer.serialize(input, output); Assertions.assertThat(output.getLength()).describedAs("The serialized size and bytes written do not match").isEqualTo(expectedSize); - DataInputBuffer in = new DataInputBuffer(output.unsafeGetBufferAndFlip(), false); + ByteBuffer buffer = output.unsafeGetBufferAndFlip(); + DataInputBuffer in = new DataInputBuffer(buffer, false); T read = serializer.deserialize(in); Assertions.assertThat(read).describedAs("The deserialized output does not match the serialized input; difference %s", new LazyToString(() -> ReflectionUtils.recursiveEquals(read, input).toString())).isEqualTo(input); + Assertions.assertThat(buffer.remaining()).describedAs("deserialize did not consume all the serialized input").isEqualTo(0); + buffer.flip(); + serializer.skip(in); + Assertions.assertThat(buffer.remaining()).describedAs("skip did not consume all the serialized input").isEqualTo(0); } public static void testSerde(AsymmetricUnversionedSerializer serializer, T input) throws IOException diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/CommandSerializersTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/CommandSerializersTest.java index 18fee68f5f..c4635cdde6 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/CommandSerializersTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/CommandSerializersTest.java @@ -19,21 +19,28 @@ package org.apache.cassandra.service.accord.serializers; import java.io.IOException; +import java.nio.ByteBuffer; import org.junit.BeforeClass; import org.junit.Test; +import accord.local.Node; import accord.primitives.PartialTxn; import accord.primitives.Ranges; +import accord.primitives.Timestamp; import accord.primitives.Txn; +import accord.primitives.TxnId; import accord.utils.AccordGens; import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.db.marshal.ByteBufferAccessor; import org.apache.cassandra.io.Serializers; import org.apache.cassandra.io.util.DataOutputBuffer; import org.apache.cassandra.schema.KeyspaceParams; import org.apache.cassandra.service.accord.AccordTestUtils; import org.apache.cassandra.service.accord.TokenRange; import org.apache.cassandra.service.accord.api.PartitionKey; +import org.apache.cassandra.utils.FastByteOperations; +import org.assertj.core.api.Assertions; import static accord.utils.Property.qt; import static org.apache.cassandra.config.DatabaseDescriptor.getPartitioner; @@ -69,6 +76,58 @@ public class CommandSerializersTest public void txnIdSerde() { DataOutputBuffer output = new DataOutputBuffer(); - qt().forAll(AccordGens.txnIds()).check(txnId -> Serializers.testSerde(output, CommandSerializers.txnId, txnId)); + qt().forAll(AccordGens.txnIds()).check(txnId -> { + Serializers.testSerde(output, CommandSerializers.txnId, txnId); + ByteBuffer tmp = output.buffer(); + tmp.clear(); + CommandSerializers.txnId.serialize(txnId, tmp); + tmp.flip(); + TxnId rt = CommandSerializers.txnId.deserialize(tmp); + Assertions.assertThat(rt).isEqualTo(txnId); + }); + } + + @Test + public void txnIdComparable() + { + qt().forAll(AccordGens.txnIds(), AccordGens.txnIds()).check(CommandSerializersTest::testComparable); + qt().forAll(AccordGens.txnIds()).check((a) -> { + ByteBuffer abb = ByteBuffer.allocate((int) CommandSerializers.txnId.serializedSize(a)); + CommandSerializers.txnId.serializeComparable(a, abb, ByteBufferAccessor.instance, 0); + if (a.epoch() < Timestamp.MAX_EPOCH) + testComparable(a, TxnId.fromValues(a.epoch() + 1, a.hlc(), a.flags(), a.node)); + if (a.epoch() > 0) + testComparable(a, TxnId.fromValues(a.epoch() - 1, a.hlc(), a.flags(), a.node)); + if (a.hlc() < Timestamp.MAX.hlc()) + testComparable(a, TxnId.fromValues(a.epoch(), a.hlc() + 1, a.flags(), a.node)); + if (a.hlc() > 0) + testComparable(a, TxnId.fromValues(a.epoch(), a.hlc() - 1, a.flags(), a.node)); + if (a.flags() < Timestamp.MAX.flags()) + testComparable(a, TxnId.fromValues(a.epoch(), a.hlc(), a.flags() + 1, a.node)); + if (a.flags() != 0) + testComparable(a, TxnId.fromValues(a.epoch(), a.hlc(), a.flags() - 1, a.node)); + if (a.node.id > 0) + testComparable(a, TxnId.fromValues(a.epoch(), a.hlc(), a.flags(), new Node.Id(a.node.id - 1))); + if (a.node.id < Integer.MAX_VALUE) + testComparable(a, TxnId.fromValues(a.epoch(), a.hlc(), a.flags(), new Node.Id(a.node.id + 1))); + }); + } + + private static void testComparable(TxnId a, TxnId b) + { + ByteBuffer abb = ByteBuffer.allocate((int) CommandSerializers.txnId.serializedSize(a)); + CommandSerializers.txnId.serializeComparable(a, abb, ByteBufferAccessor.instance, 0); + TxnId art = CommandSerializers.txnId.deserializeComparable(abb, ByteBufferAccessor.instance, 0); + Assertions.assertThat(art).isEqualTo(a); + testComparable(abb, a, b); + } + + private static void testComparable(ByteBuffer abb, TxnId a, TxnId b) + { + ByteBuffer bbb = ByteBuffer.allocate((int) CommandSerializers.txnId.serializedSize(b)); + CommandSerializers.txnId.serializeComparable(b, bbb, ByteBufferAccessor.instance, 0); + Assertions.assertThat(FastByteOperations.compareUnsigned(abb, bbb)).isEqualTo(a.compareTo(b)); + TxnId brt = CommandSerializers.txnId.deserializeComparable(bbb, ByteBufferAccessor.instance, 0); + Assertions.assertThat(brt).isEqualTo(b); } } diff --git a/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java b/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java index 5b1d5ca42e..0c964e5d12 100644 --- a/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java +++ b/test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java @@ -18,18 +18,47 @@ package org.apache.cassandra.service.accord.serializers; +import java.io.IOException; +import java.util.Arrays; + import org.junit.Test; +import accord.api.RoutingKey; +import accord.local.StoreParticipants; +import accord.primitives.AbstractRanges; +import accord.primitives.AbstractUnseekableKeys; +import accord.primitives.FullKeyRoute; +import accord.primitives.FullRangeRoute; +import accord.primitives.KeyRoute; +import accord.primitives.PartialKeyRoute; +import accord.primitives.PartialRangeRoute; +import accord.primitives.Participants; +import accord.primitives.Range; +import accord.primitives.RangeRoute; import accord.primitives.Ranges; +import accord.primitives.Route; +import accord.primitives.RoutingKeys; +import accord.primitives.Unseekable; import accord.utils.Gen; +import accord.utils.RandomSource; +import accord.utils.RandomTestRunner; +import accord.utils.UnhandledEnum; import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.dht.IPartitioner; import org.apache.cassandra.dht.Murmur3Partitioner; +import org.apache.cassandra.dht.Token; import org.apache.cassandra.io.Serializers; import org.apache.cassandra.io.util.DataOutputBuffer; +import org.apache.cassandra.schema.TableId; +import org.apache.cassandra.service.accord.TokenRange; +import org.apache.cassandra.service.accord.api.TokenKey; import org.apache.cassandra.utils.AccordGenerators; +import org.apache.cassandra.utils.CassandraGenerators; import static accord.utils.Property.qt; +import static org.apache.cassandra.utils.AccordGenerators.fromQT; import static org.apache.cassandra.utils.AccordGenerators.maybeUpdatePartitioner; +import static org.apache.cassandra.utils.AccordGenerators.partitioner; public class KeySerializersTest { @@ -50,9 +79,149 @@ public class KeySerializersTest }); } + @Test + public void storeParticipants() + { + DataOutputBuffer output = new DataOutputBuffer(); + for (int i = 0 ; i < 10000 ; ++i) + { + RandomTestRunner.test().check(rs -> testTwo(rs, output)); + } + } + + private void testTwo(RandomSource rs, DataOutputBuffer output) + { + IPartitioner partitioner = partitioner().next(rs); + DatabaseDescriptor.setPartitionerUnsafe(partitioner); + testOne(rs, output, keyRoute(partitioner, rs)); + testOne(rs, output, rangeRoute(partitioner, rs)); + } + + private void testOne(RandomSource rs, DataOutputBuffer output, Participants superset) + { + Route route = null; + if (rs.nextBoolean()) superset = ((Route)superset).participantsOnly(); + else route = (Route)subset(rs, superset, false); + Participants hasTouched = subset(rs, superset, true); + Participants touches = subset(rs, hasTouched, true); + Participants owns = subset(rs, touches, true); + Participants executes = rs.nextBoolean() ? subset(rs, owns, true) : null; + Participants waitsOn = executes != null ? subset(rs, executes, true) : null; + StoreParticipants participants = StoreParticipants.create(route, owns, executes, waitsOn, touches, hasTouched); + try + { + Serializers.testSerde(output, CommandSerializers.participants, participants); + } + catch (IOException e) + { + throw new RuntimeException(e); + } + } + + private static KeyRoute keyRoute(IPartitioner partitioner, RandomSource rs) + { + TableId tableId = fromQT(CassandraGenerators.TABLE_ID_GEN).next(rs); + Gen tokenGen = fromQT(CassandraGenerators.token(partitioner)); + Gen keyGen = AccordGenerators.routingKeyGen(ignore -> tableId, tokenGen, partitioner); + RoutingKey[] ks = new RoutingKey[rs.nextInt(1, 10)]; + for (int i = 0 ; i < ks.length ; ++i) + ks[i] = keyGen.next(rs); + Arrays.sort(ks); + int count = 1; + for (int i = 1 ; i < ks.length ; ++i) + { + if (!ks[count - 1].equals(ks[i])) + ks[count++] = ks[i]; + } + if (count != ks.length) + ks = Arrays.copyOf(ks, count); + + float f = rs.nextFloat(); + if (f < 0.66f) + { + int homeKey = rs.nextInt(ks.length); + return f < 0.33f ? new FullKeyRoute(ks[homeKey], ks) : new PartialKeyRoute(ks[homeKey], ks); + } + return new PartialKeyRoute(keyGen.next(rs), ks); + } + + private static RangeRoute rangeRoute(IPartitioner partitioner, RandomSource rs) + { + TableId tableId = fromQT(CassandraGenerators.TABLE_ID_GEN).next(rs); + Gen tokenGen = fromQT(CassandraGenerators.token(partitioner)); + Gen keyGen = AccordGenerators.routingKeyGen(ignore -> tableId, tokenGen, partitioner); + TokenKey[] ks = new TokenKey[rs.nextInt(1, 10) * 2]; + for (int i = 0 ; i < ks.length ; ++i) + ks[i] = keyGen.next(rs); + Arrays.sort(ks); + int count = 1; + for (int i = 1 ; i < ks.length ; ++i) + { + if (!ks[count - 1].equals(ks[i])) + ks[count++] = ks[i]; + } + Range[] ranges = new Range[count / 2]; + for (int i = 0 ; i < ranges.length ; ++i) + ranges[i] = TokenRange.create(ks[i*2], ks[i*2+1]); + + float f = rs.nextFloat(); + if (ranges.length > 0 && f < 0.66f) + { + RoutingKey homeKey = rs.nextBoolean() ? ks[rs.nextInt(ranges.length * 2)] : ranges[rs.nextInt(ranges.length)].someIntersectingRoutingKey(null); + return f < 0.33f ? new FullRangeRoute(homeKey, ranges) : new PartialRangeRoute(homeKey, ranges); + } + return new PartialRangeRoute(keyGen.next(rs), ranges); + } + + private static Participants subset(RandomSource rs, Participants superset, boolean changeType) + { + if (rs.nextBoolean()) + return changeType && superset instanceof Route && rs.nextBoolean() ? ((Route)superset).participantsOnly() : superset; + + int count = superset.isEmpty() ? 0 : rs.nextInt(superset.size()); + Participants subset = selectSubset(rs, count, superset); + if (superset instanceof Route && (!changeType || rs.nextBoolean())) + return superset.intersecting(subset); + return subset; + } + + private static Participants selectSubset(RandomSource rs, int count, Participants superset) + { + switch (superset.domain()) + { + default: throw UnhandledEnum.unknown(superset.domain()); + case Key: + { + AbstractUnseekableKeys in = (AbstractUnseekableKeys) superset; + RoutingKey[] out = new RoutingKey[count]; + int j = 0; + for (int i = 0 ; i < out.length ; ++i) + { + j += count == (in.size() - j) ? 0 : rs.nextInt(0, in.size() - j); + out[i] = in.get(j); + } + return (Participants) RoutingKeys.of(out); + } + + case Range: + { + AbstractRanges in = (AbstractRanges) superset; + Range[] out = new Range[count]; + int j = 0; + for (int i = 0 ; i < out.length ; ++i) + { + j += count == (in.size() - j) ? 0 : rs.nextInt(0, in.size() - j); + out[i] = in.get(j); + } + return (Participants) Ranges.of(out); + } + + } + + } + private static Gen rangesGen() { - return AccordGenerators.partitioner() - .flatMap(AccordGenerators::rangesSplitOrArbitrary); + return partitioner().flatMap(AccordGenerators::rangesSplitOrArbitrary); } } \ No newline at end of file