STAR-813: New MessagingService version for Stargazer (#218)

mike-tr-adamson · djatnieks · commit a05d1e29941d · 2023-09-11T20:10:41.000-07:00
(cherry picked from commit ac0be81) (cherry picked from commit 39fb781) (cherry picked from commit 1937874) (cherry picked from commit af134b4) (cherry picked from commit a37767a) (cherry picked from commit c8a0639)
diff --git a/src/java/org/apache/cassandra/db/CounterMutation.java b/src/java/org/apache/cassandra/db/CounterMutation.java
@@ -47,6 +47,7 @@
 import org.apache.cassandra.utils.btree.BTreeSet;
 
 import static java.util.concurrent.TimeUnit.*;
+import static org.apache.cassandra.net.MessagingService.VERSION_SG_10;
 import static org.apache.cassandra.net.MessagingService.VERSION_30;
 import static org.apache.cassandra.net.MessagingService.VERSION_3014;
 import static org.apache.cassandra.net.MessagingService.VERSION_40;
@@ -326,6 +327,7 @@ public long getTimeout(TimeUnit unit)
     private int serializedSize30;
     private int serializedSize3014;
     private int serializedSize40;
+    private int serializedSizeSG10;
 
     public int serializedSize(int version)
     {
@@ -343,6 +345,10 @@ public int serializedSize(int version)
                 if (serializedSize40 == 0)
                     serializedSize40 = (int) serializer.serializedSize(this, VERSION_40);
                 return serializedSize40;
+            case VERSION_SG_10:
+                if (serializedSizeSG10 == 0)
+                    serializedSizeSG10 = (int) serializer.serializedSize(this, VERSION_SG_10);
+                return serializedSizeSG10;
             default:
                 throw new IllegalStateException("Unknown serialization version: " + version);
         }
diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java
@@ -39,6 +39,7 @@
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
+import static org.apache.cassandra.net.MessagingService.VERSION_SG_10;
 import static org.apache.cassandra.net.MessagingService.VERSION_30;
 import static org.apache.cassandra.net.MessagingService.VERSION_3014;
 import static org.apache.cassandra.net.MessagingService.VERSION_40;
@@ -282,6 +283,7 @@ public String toString(boolean shallow)
     private int serializedSize30;
     private int serializedSize3014;
     private int serializedSize40;
+    private int serializedSizeSG10;
 
     public int serializedSize(int version)
     {
@@ -299,6 +301,10 @@ public int serializedSize(int version)
                 if (serializedSize40 == 0)
                     serializedSize40 = (int) serializer.serializedSize(this, VERSION_40);
                 return serializedSize40;
+            case VERSION_SG_10:
+                if (serializedSizeSG10 == 0)
+                    serializedSizeSG10 = (int) serializer.serializedSize(this, VERSION_SG_10);
+                return serializedSizeSG10;
             default:
                 throw new IllegalStateException("Unknown serialization version: " + version);
         }
diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogDescriptor.java
@@ -62,13 +62,15 @@ public class CommitLogDescriptor
     public static final int VERSION_40 = 7;
     // For compatibility with CNDB
     public static final int VERSION_DSE_68 = 680;
+    // Stargazer 1.0 messaging
+    public static final int VERSION_SG_10 = 100;
 
     /**
      * Increment this number if there is a changes in the commit log disc layout or MessagingVersion changes.
      * Note: make sure to handle {@link #getMessagingVersion()}
      */
     @VisibleForTesting
-    public static final int current_version = VERSION_40;
+    public static final int current_version = VERSION_SG_10;
 
     final int version;
     public final long id;
@@ -212,6 +214,8 @@ public int getMessagingVersion()
             case VERSION_40:
             case VERSION_DSE_68:
                 return MessagingService.VERSION_40;
+            case VERSION_SG_10:
+                return MessagingService.VERSION_SG_10;
             default:
                 throw new IllegalStateException("Unknown commitlog version " + version);
         }
diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java
@@ -46,6 +46,7 @@
 import org.apache.cassandra.index.IndexRegistry;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
+import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.IndexMetadata;
 import org.apache.cassandra.schema.TableMetadata;
@@ -451,22 +452,33 @@ public static class Serializer
         {
             public void serialize(FilterElement operation, DataOutputPlus out, int version) throws IOException
             {
-                out.writeBoolean(operation.isDisjunction);
+                assert (!operation.isDisjunction && operation.children().isEmpty()) || version == MessagingService.VERSION_SG_10 :
+                "Attempting to serialize a disjunct row filter to a node that doesn't support disjunction";
+
                 out.writeUnsignedVInt(operation.expressions.size());
                 for (Expression expr : operation.expressions)
                     Expression.serializer.serialize(expr, out, version);
+
+                if (version < MessagingService.VERSION_SG_10)
+                    return;
+
+                out.writeBoolean(operation.isDisjunction);
                 out.writeUnsignedVInt(operation.children.size());
                 for (FilterElement child : operation.children)
                     serialize(child, out, version);
             }
 
             public FilterElement deserialize(DataInputPlus in, int version, TableMetadata metadata) throws IOException
             {
-                boolean isDisjunction = in.readBoolean();
                 int size = (int)in.readUnsignedVInt();
                 List<Expression> expressions = new ArrayList<>(size);
                 for (int i = 0; i < size; i++)
                     expressions.add(Expression.serializer.deserialize(in, version, metadata));
+
+                if (version < MessagingService.VERSION_SG_10)
+                    return new FilterElement(false, expressions, Collections.emptyList());
+
+                boolean isDisjunction = in.readBoolean();
                 size = (int)in.readUnsignedVInt();
                 List<FilterElement> children = new ArrayList<>(size);
                 for (int i  = 0; i < size; i++)
@@ -476,9 +488,14 @@ public FilterElement deserialize(DataInputPlus in, int version, TableMetadata me
 
             public long serializedSize(FilterElement operation, int version)
             {
-                long size = 1 + TypeSizes.sizeofUnsignedVInt(operation.expressions.size());
+                long size = TypeSizes.sizeofUnsignedVInt(operation.expressions.size());
                 for (Expression expr : operation.expressions)
                     size += Expression.serializer.serializedSize(expr, version);
+
+                if (version < MessagingService.VERSION_SG_10)
+                    return size;
+
+                size++; // isDisjunction boolean
                 size += TypeSizes.sizeofUnsignedVInt(operation.children.size());
                 for (FilterElement child : operation.children)
                     size += serializedSize(child, version);
diff --git a/src/java/org/apache/cassandra/hints/HintsDescriptor.java b/src/java/org/apache/cassandra/hints/HintsDescriptor.java
@@ -64,7 +64,8 @@ final class HintsDescriptor
 
     static final int VERSION_30 = 1;
     static final int VERSION_40 = 2;
-    static final int CURRENT_VERSION = VERSION_40;
+    static final int VERSION_SG_10 = 100;
+    static final int CURRENT_VERSION = VERSION_SG_10;
 
     static final String COMPRESSION = "compression";
     static final String ENCRYPTION = "encryption";
@@ -220,6 +221,8 @@ static int messagingVersion(int hintsVersion)
                 return MessagingService.VERSION_30;
             case VERSION_40:
                 return MessagingService.VERSION_40;
+            case VERSION_SG_10:
+                return MessagingService.VERSION_SG_10;
             default:
                 throw new AssertionError();
         }
diff --git a/src/java/org/apache/cassandra/io/sstable/format/trieindex/TrieIndexFormat.java b/src/java/org/apache/cassandra/io/sstable/format/trieindex/TrieIndexFormat.java
@@ -323,7 +323,7 @@ static class TrieIndexVersion extends Version
             hasAccurateLegacyMinMax = version.compareTo("ac") >= 0;
             hasOriginatingHostId = version.matches("(a[d-z])|(b[b-z])") || version.compareTo("ca") >= 0;
             hasMaxColumnValueLengths = version.matches("b[a-z]"); // DSE only field
-            correspondingMessagingVersion = version.compareTo("ca") >= 0 ? MessagingService.VERSION_40 : MessagingService.VERSION_3014;
+            correspondingMessagingVersion = version.compareTo("ca") >= 0 ? MessagingService.VERSION_SG_10 : MessagingService.VERSION_3014;
         }
 
         // this is for the ab version which was used in the LABS, and then has been renamed to ba
diff --git a/src/java/org/apache/cassandra/net/Message.java b/src/java/org/apache/cassandra/net/Message.java
@@ -56,6 +56,7 @@
 import static org.apache.cassandra.net.MessagingService.VERSION_30;
 import static org.apache.cassandra.net.MessagingService.VERSION_3014;
 import static org.apache.cassandra.net.MessagingService.VERSION_40;
+import static org.apache.cassandra.net.MessagingService.VERSION_SG_10;
 import static org.apache.cassandra.net.MessagingService.instance;
 import static org.apache.cassandra.utils.MonotonicClock.approxTime;
 import static org.apache.cassandra.utils.vint.VIntCoding.computeUnsignedVIntSize;
@@ -1367,6 +1368,7 @@ private static <In,Out> IVersionedAsymmetricSerializer<In, Out> getPayloadSerial
     private int serializedSize30;
     private int serializedSize3014;
     private int serializedSize40;
+    private int serializedSizeSG10;
 
     /**
      * Serialized size of the entire message, for the provided messaging version. Caches the calculated value.
@@ -1387,6 +1389,10 @@ public int serializedSize(int version)
                 if (serializedSize40 == 0)
                     serializedSize40 = serializer.serializedSize(this, VERSION_40);
                 return serializedSize40;
+            case VERSION_SG_10:
+                if (serializedSizeSG10 == 0)
+                    serializedSizeSG10 = (int) serializer.serializedSize(this, VERSION_SG_10);
+                return serializedSizeSG10;
             default:
                 throw new IllegalStateException();
         }
@@ -1395,6 +1401,7 @@ public int serializedSize(int version)
     private int payloadSize30   = -1;
     private int payloadSize3014 = -1;
     private int payloadSize40   = -1;
+    private int payloadSizeSG10 = -1;
 
     private int payloadSize(int version)
     {
@@ -1412,6 +1419,10 @@ private int payloadSize(int version)
                 if (payloadSize40 < 0)
                     payloadSize40 = serializer.payloadSize(this, VERSION_40);
                 return payloadSize40;
+            case VERSION_SG_10:
+                if (payloadSizeSG10 < 0)
+                    payloadSizeSG10 = serializer.payloadSize(this, VERSION_SG_10);
+                return payloadSizeSG10;
             default:
                 throw new IllegalStateException();
         }
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -203,16 +203,20 @@ public class MessagingService extends MessagingServiceMBeanImpl
     public static final int VERSION_30 = 10;
     public static final int VERSION_3014 = 11;
     public static final int VERSION_40 = 12;
+    // Current Stargazer version while we have serialization differences
+    // If differences get merged upstream then we can revert to OS versioning
+    public static final int VERSION_SG_10 = 100;
     public static final int minimum_version = VERSION_30;
-    public static final int current_version = VERSION_40;
+    public static final int current_version = VERSION_SG_10;
     static AcceptVersions accept_messaging = new AcceptVersions(minimum_version, current_version);
     static AcceptVersions accept_streaming = new AcceptVersions(current_version, current_version);
 
     public enum Version
     {
         VERSION_30(10),
         VERSION_3014(11),
-        VERSION_40(12);
+        VERSION_40(12),
+        STARGAZER_10(100);
 
         public final int value;
 
diff --git a/test/unit/org/apache/cassandra/net/FramingTest.java b/test/unit/org/apache/cassandra/net/FramingTest.java
@@ -52,6 +52,8 @@
 import static java.lang.Math.*;
 import static org.apache.cassandra.net.MessagingService.VERSION_30;
 import static org.apache.cassandra.net.MessagingService.VERSION_3014;
+import static org.apache.cassandra.net.MessagingService.VERSION_40;
+import static org.apache.cassandra.net.MessagingService.VERSION_SG_10;
 import static org.apache.cassandra.net.MessagingService.current_version;
 import static org.apache.cassandra.net.MessagingService.minimum_version;
 import static org.apache.cassandra.net.OutboundConnections.LARGE_MESSAGE_THRESHOLD;
@@ -251,13 +253,14 @@ public void testSerializeSizeMatchesEdgeCases() // See CASSANDRA-16103
 
     private void burnRandomLegacy(int count)
     {
+        int[] versions = new int[] { VERSION_30, VERSION_3014, VERSION_40, VERSION_SG_10 };
         SecureRandom seed = new SecureRandom();
         Random random = new Random();
         for (int i = 0 ; i < count ; ++i)
         {
             long innerSeed = seed.nextLong();
             float ratio = seed.nextFloat();
-            int version = minimum_version + random.nextInt(1 + current_version - minimum_version);
+            int version = versions[random.nextInt(4)];
             logger.debug("seed: {}, ratio: {}, version: {}", innerSeed, ratio, version);
             random.setSeed(innerSeed);
             testRandomSequenceOfMessages(random, ratio, version, new FrameDecoderLegacy(GlobalBufferPoolAllocator.instance, version));

Original file line number	Diff line number	Diff line change
`@@ -323,7 +323,7 @@ static class TrieIndexVersion extends Version`
`323`	`323`	`hasAccurateLegacyMinMax = version.compareTo("ac") >= 0;`
`324`	`324`	`hasOriginatingHostId = version.matches("(a[d-z])\|(b[b-z])") \|\| version.compareTo("ca") >= 0;`
`325`	`325`	`hasMaxColumnValueLengths = version.matches("b[a-z]"); // DSE only field`
`326`		`- correspondingMessagingVersion = version.compareTo("ca") >= 0 ? MessagingService.VERSION_40 : MessagingService.VERSION_3014;`
	`326`	`+ correspondingMessagingVersion = version.compareTo("ca") >= 0 ? MessagingService.VERSION_SG_10 : MessagingService.VERSION_3014;`
`327`	`327`	`}`
`328`	`328`
`329`	`329`	`// this is for the ab version which was used in the LABS, and then has been renamed to ba`