Accord: Topology serializer has a lot of repeated data, can dedup to shrink the cost

patch by David Capwell; reviewed by Benedict Elliott Smith for CASSANDRA-20715

Author: dcapwell

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Accord: Topology serializer has a lot of repeated data, can dedup to shrink the cost (CASSANDRA-20715)
  * Stream individual files in their own transactions and hand over ownership to a parent transaction on completion (CASSANDRA-20728)
  * Limit the number of held heap dumps to not consume disk space excessively (CASSANDRA-20457)
  * Accord: BEGIN TRANSACTIONs IF condition logic does not properly support meaningless emptiness and null values (CASSANDRA-20667)

modules/accord

@@ -25,6 +25,8 @@
 import java.util.function.LongUnaryOperator;
 
 import javax.annotation.Nullable;
+
+import org.apache.cassandra.io.UnversionedSerializer;
 import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.commons.lang3.ArrayUtils;
 
@@ -52,6 +54,10 @@
 public final class TableId implements Comparable<TableId>
 {
     public static final long MAGIC = 1956074401491665062L;
+    /**
+     * Represents a placeholder for cases where a table id is defined, but has no meaning.
+     */
+    public static final TableId UNDEFINED = TableId.fromRaw(Long.MIN_VALUE, Long.MIN_VALUE);
     public static final long EMPTY_SIZE = ObjectSizes.measureDeep(new UUID(0, 0));
     private static final int MAGIC_BYTE = (int) ((flipSign(MAGIC) >>> 56) & 0xf0);
 
@@ -402,6 +408,27 @@ public long serializedSize(TableId t, int version)
         }
     };
 
+    public static final UnversionedSerializer<TableId> compactComparableSerializer = new UnversionedSerializer<TableId>()
+    {
+        @Override
+        public void serialize(TableId t, DataOutputPlus out) throws IOException
+        {
+            t.serializeCompactComparable(out);
+        }
+
+        @Override
+        public TableId deserialize(DataInputPlus in) throws IOException
+        {
+            return TableId.deserializeCompactComparable(in);
+        }
+
+        @Override
+        public long serializedSize(TableId t)
+        {
+            return t.serializedCompactComparableSize();
+        }
+    };
+
     public static final MetadataSerializer<TableId> metadataSerializer = new MetadataSerializer<TableId>()
     {
         @Override

src/java/org/apache/cassandra/schema/TableId.java

@@ -22,6 +22,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
+import java.util.Objects;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -90,6 +91,20 @@ public FetchTopologies(long minEpoch, long maxEpoch)
         this.maxEpoch = maxEpoch;
     }
 
+    @Override
+    public boolean equals(Object o)
+    {
+        if (o == null || getClass() != o.getClass()) return false;
+        FetchTopologies that = (FetchTopologies) o;
+        return minEpoch == that.minEpoch && maxEpoch == that.maxEpoch;
+    }
+
+    @Override
+    public int hashCode()
+    {
+        return Objects.hash(minEpoch, maxEpoch);
+    }
+
     public static final UnversionedSerializer<TopologyRange> responseSerializer = new UnversionedSerializer<>()
     {
             @Override
@@ -101,7 +116,7 @@ public void serialize(TopologyRange t, DataOutputPlus out) throws IOException
                 out.writeUnsignedVInt32(t.topologies.size());
 
                 for (Topology topology : t.topologies)
-                    TopologySerializers.topology.serialize(topology, out);
+                    TopologySerializers.compactTopology.serialize(topology, out);
             }
 
             @Override
@@ -113,7 +128,7 @@ public TopologyRange deserialize(DataInputPlus in) throws IOException
                 int count = in.readUnsignedVInt32();
                 List<Topology> topologies = new ArrayList<>(count);
                 for (int i = 0; i < count; ++i)
-                    topologies.add(TopologySerializers.topology.deserialize(in));
+                    topologies.add(TopologySerializers.compactTopology.deserialize(in));
                 return new TopologyRange(min, current, firstNonEmpty, topologies);
             }
 
@@ -125,7 +140,7 @@ public long serializedSize(TopologyRange t)
                 size += TypeSizes.sizeofUnsignedVInt(t.firstNonEmpty);
                 size += TypeSizes.sizeofUnsignedVInt(t.topologies.size());
                 for (Topology topology : t.topologies)
-                    size += TopologySerializers.topology.serializedSize(topology);
+                    size += TopologySerializers.compactTopology.serializedSize(topology);
                 return size;
             }
         };

src/java/org/apache/cassandra/service/accord/FetchTopologies.java

@@ -87,7 +87,7 @@ public boolean isFullRange()
     }
 
     @VisibleForTesting
-    public Range withTable(TableId table)
+    public TokenRange withTable(TableId table)
     {
         return new TokenRange(start().withTable(table), end().withTable(table));
     }
@@ -148,5 +148,29 @@ public long serializedSize(TokenRange range)
             return TokenKey.serializer.serializedSize(range.start())
                    + TokenKey.serializer.serializedSize(range.end());
         }
+    }
+
+    public static final UnversionedSerializer<TokenRange> noTableSerializer = new UnversionedSerializer<TokenRange>()
+    {
+        @Override
+        public void serialize(TokenRange t, DataOutputPlus out) throws IOException
+        {
+            TokenKey.noTableSerializer.serialize(t.start(), out);
+            TokenKey.noTableSerializer.serialize(t.end(), out);
+        }
+
+        @Override
+        public TokenRange deserialize(DataInputPlus in) throws IOException
+        {
+            return TokenRange.create(TokenKey.noTableSerializer.deserialize(TableId.UNDEFINED, in),
+                                     TokenKey.noTableSerializer.deserialize(TableId.UNDEFINED, in));
+        }
+
+        @Override
+        public long serializedSize(TokenRange t)
+        {
+            return TokenKey.noTableSerializer.serializedSize(t.start())
+                   + TokenKey.noTableSerializer.serializedSize(t.end());
+        }
     };
 }

src/java/org/apache/cassandra/service/accord/TokenRange.java

@@ -40,6 +40,7 @@
 import org.apache.cassandra.db.marshal.ValueAccessor;
 import org.apache.cassandra.dht.IPartitioner;
 import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.io.ParameterisedUnversionedSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.schema.TableId;
@@ -230,6 +231,55 @@ public static TokenKey before(TableId table, Token token)
         return new TokenKey(table, BEFORE_TOKEN_SENTINEL, token);
     }
 
+    public static final NoTableSerializer noTableSerializer = new NoTableSerializer();
+
+
+    public static class NoTableSerializer implements ParameterisedUnversionedSerializer<TokenKey, TableId>
+    {
+        @Override
+        public void serialize(TokenKey key, TableId tableId, DataOutputPlus out) throws IOException
+        {
+            IPartitioner partitioner = key.token.getPartitioner();
+            int fixedLength = partitioner.accordFixedLength();
+            if (fixedLength < 0)
+            {
+                int len = partitioner.accordSerializedSize(key.token);
+                out.writeUnsignedVInt32(len);
+            }
+            serializer.serializeWithoutPrefixOrLength(key, out);
+        }
+
+        public void serialize(TokenKey key, DataOutputPlus out) throws IOException
+        {
+            serialize(key, key.table, out);
+        }
+
+        @Override
+        public long serializedSize(TokenKey key, TableId tableId)
+        {
+            IPartitioner partitioner = key.token.getPartitioner();
+            int tokenSize = partitioner.accordFixedLength();
+            if (tokenSize >= 0)
+                return 2 + tokenSize;
+            tokenSize = partitioner.accordSerializedSize(key.token);
+            return 2 + tokenSize + VIntCoding.sizeOfUnsignedVInt(tokenSize);
+        }
+
+        public long serializedSize(TokenKey key)
+        {
+            return serializedSize(key, key.table);
+        }
+
+        @Override
+        public TokenKey deserialize(TableId tableId, DataInputPlus in) throws IOException
+        {
+            IPartitioner partitioner = getPartitioner();
+            int len = partitioner.accordFixedLength();
+            if (len < 0) len = in.readUnsignedVInt32();
+            return serializer.deserializeWithPrefix(tableId, len + 2, in, partitioner);
+        }
+    }
+
     public static final class Serializer implements AccordSearchableKeySerializer<TokenKey>
     {
         private Serializer() {}

src/java/org/apache/cassandra/service/accord/api/TokenKey.java

@@ -114,8 +114,8 @@ public void serialize(Journal.TopologyUpdate from, DataOutputPlus out) throws IO
             }
             //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);
+            TopologySerializers.compactTopology.serialize(from.local, out);
+            TopologySerializers.compactTopology.serialize(from.global, out);
         }
 
         @Override
@@ -129,8 +129,8 @@ public Journal.TopologyUpdate deserialize(DataInputPlus in) throws IOException
                 CommandStores.RangesForEpoch rangesForEpoch = RangesForEpochSerializer.instance.deserialize(in);
                 commandStores.put(commandStoreId, rangesForEpoch);
             }
-            Topology local = TopologySerializers.topology.deserialize(in);
-            Topology global = TopologySerializers.topology.deserialize(in);
+            Topology local = TopologySerializers.compactTopology.deserialize(in);
+            Topology global = TopologySerializers.compactTopology.deserialize(in);
             return new Journal.TopologyUpdate(commandStores, local, global);
         }
 
@@ -144,8 +144,8 @@ public long serializedSize(Journal.TopologyUpdate from)
                 size += RangesForEpochSerializer.instance.serializedSize(e.getValue());
             }
 
-            size += TopologySerializers.topology.serializedSize(from.local);
-            size += TopologySerializers.topology.serializedSize(from.global);
+            size += TopologySerializers.compactTopology.serializedSize(from.local);
+            size += TopologySerializers.compactTopology.serializedSize(from.global);
             return size;
         }
     }