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

Improve Journal table key serialization 

Also improve:
 - TxnId serialization
 - StoreParticipants serialization
 - compareUnsigned Node.Id for consistency with serialized TxnId

patch by Benedict; reviewed by Alex Petrov for CASSANDRA-20546
This commit is contained in:
Benedict Elliott Smith 2025-04-07 12:46:32 +01:00 committed by David Capwell
parent 68aea4b15f
commit 19048ef570
39 changed files with 1272 additions and 582 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
modules/accord

@ -1 +1 @@
Subproject commit fc14a154fd514d4ab40b37508fb9497f786835e0
Subproject commit 134df57677bbd5092994923a4dc2f15cd1d033d1

5
src/java/org/apache/cassandra/concurrent/ExecutionFailure.java
View File

@ -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?

8
src/java/org/apache/cassandra/db/marshal/ByteBufferAccessor.java
View File

@ -318,13 +318,13 @@ public class ByteBufferAccessor implements ValueAccessor<ByteBuffer>
@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<ByteBuffer>
@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));
}
}

34
src/java/org/apache/cassandra/db/marshal/ValueAccessor.java
View File

@ -525,23 +525,23 @@ public interface ValueAccessor<V>
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<V>
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();
}

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

@ -249,7 +249,7 @@ public class RouteIndexFormat
}
}
static List<Segment> readSegements(Map<IndexComponent, FileHandle> index) throws IOException
static List<Segment> readSegments(Map<IndexComponent, FileHandle> index) throws IOException
{
List<Segment> segments = new ArrayList<>();

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

@ -74,7 +74,7 @@ public class SSTableIndex extends SharedCloseableImpl
Map<IndexComponent, FileHandle> files = new EnumMap<>(IndexComponent.class);
for (IndexComponent c : id.getLiveComponents())
files.put(c, new FileHandle.Builder(id.fileFor(c)).mmapped(true).complete());
List<Segment> segments = RouteIndexFormat.readSegements(files);
List<Segment> segments = RouteIndexFormat.readSegments(files);
files.remove(IndexComponent.SEGMENT).close();
files.remove(IndexComponent.METADATA).close();
Cleanup cleanup = new Cleanup(files);

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

@ -42,6 +42,11 @@ public interface AsymmetricUnversionedSerializer<In, Out>
}
}
default void skip(DataInputPlus in) throws IOException
{
deserialize(in);
}
default ByteBuffer serializeUnchecked(In t)
{
try

7
src/java/org/apache/cassandra/net/Verb.java
View File

@ -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),

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

@ -106,7 +106,7 @@ public abstract class AbstractAccordSegmentCompactor<V> 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();

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

@ -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)
}
}

4
src/java/org/apache/cassandra/service/accord/AccordDataStore.java
View File

@ -71,10 +71,10 @@ public class AccordDataStore implements DataStore
}
@Override
public AsyncResult<Void> snapshot(Ranges ranges, TxnId before) // TODO: does this have to go to journal, too?
public AsyncResult<Void> snapshot(Ranges ranges, TxnId before)
{
AsyncResults.SettableResult<Void> 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<TableId, SnapshotBounds> tables = new Object2ObjectHashMap<>();
for (Range range : ranges)

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

@ -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<Thread> loops;
private final Long2ObjectHashMap<Thread> 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);

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

@ -384,7 +384,7 @@ public class AccordJournal implements accord.api.Journal, RangeSearcher.Supplier
private <BUILDER extends FlyweightImage> 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<?, BUILDER> serializer = (AccordJournalValueSerializers.FlyweightSerializer<?, BUILDER>) 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);

71
src/java/org/apache/cassandra/service/accord/AccordKeyspace.java
View File

@ -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<DecoratedKey> keyIterator(Memtable memtable, AbstractBounds<PartitionPosition> range)
{
// TODO (required): why are we replacing the right bound with max bound?
AbstractBounds<PartitionPosition> 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);
}
}

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

@ -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<RoutingKey> 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<mi; i++)
int numberOfKeys = keys.size();
long size = ObjectSizes.sizeOfReferenceArray(numberOfKeys);
if (numberOfKeys > 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]));

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

@ -196,7 +196,6 @@ public class AccordSafeCommandStore extends AbstractSafeCommandStore<AccordSafeC
@Override
public NodeCommandStoreService node()
{
// TODO: safe command store should not have arbitrary time
return commandStore.node();
}

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

@ -781,7 +781,7 @@ public class AccordService implements IAccordService, Shutdownable
CommandStoreTxnBlockedGraph.TxnState cmdTxnState = populate(state, safeCommand.current());
if (cmdTxnState.notBlocked())
return null;
//TODO (safety): check depth
//TODO (expected): check depth
List<AsyncChain<Void>> chains = new ArrayList<>();
for (TxnId blockedBy : cmdTxnState.blockedBy)
{

1
src/java/org/apache/cassandra/service/accord/AccordSyncPropagator.java
View File

@ -313,7 +313,6 @@ public class AccordSyncPropagator
{
Invariants.require(msg.payload == SimpleReply.Ok, "Unexpected message: %s", msg);
Set<Long> 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);

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

@ -57,8 +57,7 @@ public class AccordVerbHandler<T extends Request> implements IVerbHandler<T>
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());

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

@ -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);
}

2
src/java/org/apache/cassandra/service/accord/api/AccordRoutableKey.java
View File

@ -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)
{

29
src/java/org/apache/cassandra/service/accord/api/PartitionKey.java
View File

@ -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<PartitionKey>
{
// 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 <V> int serialize(PartitionKey key, V dst, ValueAccessor<V> 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());
}
}
}

4
src/java/org/apache/cassandra/service/accord/api/TokenKey.java
View File

@ -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);
}

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

@ -228,9 +228,9 @@ public class AccordInteropExecution implements ReadCoordinator, MaximalCommitSen
public void sendReadCommand(Message<ReadCommand> message, InetAddressAndPort to, RequestCallback<ReadResponse> 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));

1
src/java/org/apache/cassandra/service/accord/interop/AccordInteropReadRepair.java
View File

@ -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;
}

2
src/java/org/apache/cassandra/service/accord/journal/AccordTopologyUpdate.java
View File

@ -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);

36
src/java/org/apache/cassandra/service/accord/serializers/BeginInvalidationSerializers.java
View File

@ -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<InvalidateReply> 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);
}
};
}

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

@ -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> txnId = new TimestampSerializer<>(TxnId::fromBits);
public static final TimestampSerializer<Timestamp> timestamp = new TimestampSerializer<>(Timestamp::fromBits);
public static final UnversionedSerializer<Timestamp> nullableTimestamp = NullableSerializer.wrap(timestamp);
public static final VariableWidthTimestampSerializer<TxnId> txnId = new VariableWidthTimestampSerializer<>(TxnId::fromValues);
public static final VariableWidthTimestampSerializer<Timestamp> timestamp = new VariableWidthTimestampSerializer<>(Timestamp::fromValues);
public static final BallotSerializer ballot = new BallotSerializer(); // permits null
public static final UnversionedSerializer<Txn.Kind> kind = EncodeAsVInt32.of(Txn.Kind.class);
public static final StoreParticipantsSerializer participants = new StoreParticipantsSerializer();
@ -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<StoreParticipants>
public static class StoreParticipantsSerializer implements UnversionedSerializer<StoreParticipants>
{
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<T extends Timestamp> implements UnversionedSerializer<T>
public static class VariableWidthTimestampSerializer<T extends Timestamp> implements UnversionedSerializer<T>
{
interface Factory<T extends Timestamp>
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<T> factory;
interface Factory<T extends Timestamp>
{
T create(long epoch, long hlc, int flags, Node.Id node);
}
private TimestampSerializer(TimestampSerializer.Factory<T> factory)
private final VariableWidthTimestampSerializer.Factory<T> 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<T> 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 <V> int serializeComparable(T ts, V dst, ValueAccessor<V> 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 <V> int serialize(T ts, V dst, ValueAccessor<V> 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 <V> T deserialize(V src, ValueAccessor<V> 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 <V> T deserializeComparable(V src, ValueAccessor<V> 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<Ballot>
public static class BallotSerializer extends VariableWidthTimestampSerializer<Ballot>
{
final TimestampSerializer<Ballot> 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);
}
}

6
src/java/org/apache/cassandra/service/accord/serializers/FetchSerializers.java
View File

@ -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);
}
};

247
src/java/org/apache/cassandra/service/accord/serializers/IVersionedWithKeysSerializer.java
View File

@ -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<K extends Routables<?>, T> extends
*/
long serializedSize(K keys, T t, Version version);
final class NullableWithKeysSerializer<K extends Routables<?>, T> implements IVersionedWithKeysSerializer<K, T>
{
final IVersionedWithKeysSerializer<K, T> wrapped;
public NullableWithKeysSerializer(IVersionedWithKeysSerializer<K, T> 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<K extends Routables<?>, 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<K extends Routables<?>, 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<K extends Routables<?>, 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<K extends Routables<?>, 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<K extends Routables<?>, 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<K extends Routables<?>, 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 <K extends RoutableKey> long encodeBitmap(AbstractKeys<K> serialize, AbstractKeys<K> superset, int supersetCount)
@ -323,44 +228,112 @@ public interface IVersionedWithKeysSerializer<K extends Routables<?>, 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, S extends AbstractUnseekableKeys> T deserializeRoutingKeySubset(S superset, DataInputPlus in, BiFunction<RoutingKey[], S, T> 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, S extends AbstractRanges> T deserializeRangeSubset(S superset, DataInputPlus in, BiFunction<Range[], S, T> 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 extends Routable> R[] deserializeSmallSubsetArray(long encoded, Routables<R> superset, int supersetCount, IntFunction<R[]> 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 <K extends RoutableKey> long serializeLargeSubsetSize(AbstractKeys<K> serialize, int serializeCount, AbstractKeys<K> superset, int supersetCount)
{

6
src/java/org/apache/cassandra/service/accord/serializers/InformDurableSerializers.java
View File

@ -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);
}
};

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

@ -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> key;
public static final UnversionedSerializer<RoutingKey> routingKey;
public static final AccordSearchableKeySerializer<RoutingKey> routingKey;
public static final UnversionedSerializer<RoutingKey> nullableRoutingKey;
public static final AbstractSearchableKeysSerializer<RoutingKey, RoutingKeys> routingKeys;
public static final AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys;
public static final UnversionedSerializer<Keys> keys;
public static final AbstractSearchableKeysSerializer<?, PartialKeyRoute> partialKeyRoute;
public static final AbstractSearchableKeysSerializer<?, FullKeyRoute> fullKeyRoute;
public static final AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute;
public static final AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute;
public static final UnversionedSerializer<Range> range;
public static final AbstractRangesSerializer<Ranges> ranges;
@ -130,11 +131,11 @@ public class KeySerializers
final AccordSearchableKeySerializer<RoutingKey> routingKey;
final UnversionedSerializer<RoutingKey> nullableRoutingKey;
final AbstractSearchableKeysSerializer<RoutingKey, RoutingKeys> routingKeys;
final AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys;
final UnversionedSerializer<Keys> keys;
final AbstractSearchableKeysSerializer<?, PartialKeyRoute> partialKeyRoute;
final AbstractSearchableKeysSerializer<?, FullKeyRoute> fullKeyRoute;
final AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute;
final AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute;
final UnversionedSerializer<Range> range;
final AbstractRangesSerializer<Ranges> 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<EnumSet<UnseekablesKind>, AbstractRoutablesSerializer<?>> factory = (a) -> new AbstractRoutablesSerializer<>(a, routingKeys, partialKeyRoute, fullKeyRoute, ranges, partialRangeRoute, fullRangeRoute);
Function<TinyEnumSet<UnseekablesKind>, AbstractRoutablesSerializer<?>> factory = (a) -> new AbstractRoutablesSerializer<>(a, routingKeys, partialKeyRoute, fullKeyRoute, ranges, partialRangeRoute, fullRangeRoute);
this.route = (AbstractRoutablesSerializer<Route<?>>) factory.apply(EnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.FullKeyRoute, UnseekablesKind.PartialRangeRoute, UnseekablesKind.FullRangeRoute));
this.route = (AbstractRoutablesSerializer<Route<?>>) factory.apply(TinyEnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.FullKeyRoute, UnseekablesKind.PartialRangeRoute, UnseekablesKind.FullRangeRoute));
this.nullableRoute = NullableSerializer.wrap(route);
this.partialRoute = (AbstractRoutablesSerializer<PartialRoute<?>>) factory.apply(EnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.PartialRangeRoute));
this.fullRoute = (AbstractRoutablesSerializer<FullRoute<?>>) factory.apply(EnumSet.of(UnseekablesKind.FullKeyRoute, UnseekablesKind.FullRangeRoute));
this.partialRoute = (AbstractRoutablesSerializer<PartialRoute<?>>) factory.apply(TinyEnumSet.of(UnseekablesKind.PartialKeyRoute, UnseekablesKind.PartialRangeRoute));
this.fullRoute = (AbstractRoutablesSerializer<FullRoute<?>>) factory.apply(TinyEnumSet.of(UnseekablesKind.FullKeyRoute, UnseekablesKind.FullRangeRoute));
this.nullableFullRoute = NullableSerializer.wrap(fullRoute);
this.unseekables = (AbstractRoutablesSerializer<Unseekables<?>>) factory.apply(EnumSet.allOf(UnseekablesKind.class));
this.participants = (AbstractRoutablesSerializer<Participants<?>>) factory.apply(EnumSet.allOf(UnseekablesKind.class));
this.unseekables = (AbstractRoutablesSerializer<Unseekables<?>>) factory.apply(TinyEnumSet.allOf(UnseekablesKind.class));
this.participants = (AbstractRoutablesSerializer<Participants<?>>) 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<RS extends Unseekables<?>> implements UnversionedSerializer<RS>
{
final EnumSet<UnseekablesKind> permitted;
final AbstractSearchableKeysSerializer<RoutingKey, RoutingKeys> routingKeys;
final AbstractSearchableKeysSerializer<?, PartialKeyRoute> partialKeyRoute;
final AbstractSearchableKeysSerializer<?, FullKeyRoute> fullKeyRoute;
final TinyEnumSet<UnseekablesKind> permitted;
final AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys;
final AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute;
final AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute;
final AbstractRangesSerializer<Ranges> ranges;
final AbstractRangesSerializer<PartialRangeRoute> partialRangeRoute;
final AbstractRangesSerializer<FullRangeRoute> fullRangeRoute;
protected AbstractRoutablesSerializer(EnumSet<UnseekablesKind> permitted,
AbstractSearchableKeysSerializer<RoutingKey, RoutingKeys> routingKeys,
AbstractSearchableKeysSerializer<?, PartialKeyRoute> partialKeyRoute,
AbstractSearchableKeysSerializer<?, FullKeyRoute> fullKeyRoute,
protected AbstractRoutablesSerializer(TinyEnumSet<UnseekablesKind> permitted,
AbstractSearchableRoutingKeysSerializer<RoutingKeys> routingKeys,
AbstractSearchableRoutingKeysSerializer<PartialKeyRoute> partialKeyRoute,
AbstractSearchableRoutingKeysSerializer<FullKeyRoute> fullKeyRoute,
AbstractRangesSerializer<Ranges> ranges,
AbstractRangesSerializer<PartialRangeRoute> partialRangeRoute,
AbstractRangesSerializer<FullRangeRoute> 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> 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<K extends RoutableKey, R extends Routable, RS extends Routables<R>> implements UnversionedSerializer<RS>
public abstract static class AbstractSearchableSerializer<K extends RoutableKey, R extends Routable, RS extends Routables<R>> extends IVersionedWithKeysSerializer.AbstractWithKeysSerializer implements UnversionedSerializer<RS>
{
final AccordSearchableKeySerializer<K> keySerializer;
final IntFunction<R[]> allocate;
public AbstractSearchableSerializer(AccordSearchableKeySerializer<K> keySerializer, IntFunction<R[]> allocate)
public AbstractSearchableSerializer(IntFunction<R[]> 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<K extends RoutableKey, KS extends AbstractKeys<K>> extends AbstractSearchableSerializer<K, K, KS> implements UnversionedSerializer<KS>
public abstract static class AbstractSearchableRoutingKeysSerializer<KS extends AbstractUnseekableKeys> extends AbstractSearchableSerializer<RoutingKey, RoutingKey, KS> implements UnversionedSerializer<KS>
{
public AbstractSearchableKeysSerializer(AccordSearchableKeySerializer<K> keySerializer, IntFunction<K[]> allocate)
public AbstractSearchableRoutingKeysSerializer(AccordSearchableKeySerializer<RoutingKey> 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<KS extends KeyRoute> extends AbstractSearchableRoutingKeysSerializer<KS>
{
public AbstractKeyRouteSerializer(AccordSearchableKeySerializer<RoutingKey> 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<RS extends AbstractRanges> extends AbstractSearchableSerializer<RoutingKey, Range, RS> implements UnversionedSerializer<RS>
{
public AbstractRangesSerializer(AccordSearchableKeySerializer<RoutingKey> 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<ByteBuffer, ByteBuffer> rangesToBlobMap(Ranges ranges)
public abstract static class AbstractRangeRouteSerializer<RS extends RangeRoute> extends AbstractRangesSerializer<RS>
{
TreeMap<ByteBuffer, ByteBuffer> 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<ByteBuffer, ByteBuffer> blobMap)
{
int i = 0;
Range[] ranges = new Range[blobMap.size()];
for (Map.Entry<ByteBuffer, ByteBuffer> e : blobMap.entrySet())
{
ranges[i++] = TokenRange.create(TokenKey.serializer.deserialize(e.getKey()),
TokenKey.serializer.deserialize(e.getValue()));
}
return Ranges.of(ranges);
}
}

2
src/java/org/apache/cassandra/service/accord/serializers/ResultSerializers.java
View File

@ -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

21
src/java/org/apache/cassandra/utils/vint/VIntCoding.java
View File

@ -308,6 +308,17 @@ public class VIntCoding
return retval;
}
public static <V> int readLengthOfVInt(V input, ValueAccessor<V> 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)
{

2
test/unit/org/apache/cassandra/cql3/validation/entities/VectorsTest.java
View File

@ -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

2
test/unit/org/apache/cassandra/index/accord/CheckpointIntervalArrayIndexTest.java
View File

@ -396,7 +396,7 @@ public class CheckpointIntervalArrayIndexTest
Map<IndexComponent, FileHandle> files = new EnumMap<>(IndexComponent.class);
for (IndexComponent c : descriptor.getLiveComponents())
files.put(c, new FileHandle.Builder(descriptor.fileFor(c)).mmapped(true).complete());
List<Segment> segments = RouteIndexFormat.readSegements(files);
List<Segment> segments = RouteIndexFormat.readSegments(files);
files.remove(IndexComponent.SEGMENT).close();
files.remove(IndexComponent.METADATA).close();

8
test/unit/org/apache/cassandra/io/Serializers.java
View File

@ -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 <T> void testSerde(AsymmetricUnversionedSerializer<T, T> serializer, T input) throws IOException

61
test/unit/org/apache/cassandra/service/accord/serializers/CommandSerializersTest.java
View File

@ -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);
}
}

173
test/unit/org/apache/cassandra/service/accord/serializers/KeySerializersTest.java
View File

@ -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<Token> tokenGen = fromQT(CassandraGenerators.token(partitioner));
Gen<TokenKey> 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<Token> tokenGen = fromQT(CassandraGenerators.token(partitioner));
Gen<TokenKey> 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 <R extends Unseekable> Participants<R> selectSubset(RandomSource rs, int count, Participants<R> 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<R>) 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<R>) Ranges.of(out);
}
}
}
private static Gen<Ranges> rangesGen()
{
return AccordGenerators.partitioner()
.flatMap(AccordGenerators::rangesSplitOrArbitrary);
return partitioner().flatMap(AccordGenerators::rangesSplitOrArbitrary);
}
}