Add support in the binary protocol to allow transactions to have multiple conditions

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

Author: dcapwell

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Add support in the binary protocol to allow transactions to have multiple conditions (CASSANDRA-20883)
  * Enable CQLSSTableWriter to create SSTables compressed with a dictionary (CASSANDRA-20938)
  * Support ZSTD dictionary compression (CASSANDRA-17021)
  * Fix ExceptionsTable when stacktrace has zero elements (CASSANDRA-20992)

src/java/org/apache/cassandra/io/AsymmetricParameterisedUnversionedSerializer.java

@@ -74,5 +74,10 @@ default Out deserializeUnchecked(P p, ByteBuffer buffer)
         }
     }
 
+    default void skip(P p, DataInputPlus in) throws IOException
+    {
+        deserialize(p, in);
+    }
+
     long serializedSize(In t, P p);
 }

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

@@ -54,6 +54,11 @@ public static TokenRange create(TokenKey start, TokenKey end)
         return new TokenRange(start, end);
     }
 
+    public static TokenRange create(TableId tableId, Token start, Token end)
+    {
+        return new TokenRange(new TokenKey(tableId, start), new TokenKey(tableId, end));
+    }
+
     public static TokenRange createUnsafe(TokenKey start, TokenKey end)
     {
         return new TokenRange(start, end);

src/java/org/apache/cassandra/service/accord/serializers/SerializePacked.java

@@ -23,27 +23,99 @@
 import accord.utils.BitUtils;
 import accord.utils.Invariants;
 import net.nicoulaj.compilecommand.annotations.Inline;
+import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.DataOutputPlus;
 
 /**
  * A set of simple utilities to quickly serialize/deserialize arrays/lists of values that each require <= 64 bits to represent.
  * These are packed into an "array" of fixed bit width, so that the total size consumed is ceil((bits*elements)/8).
  * This can (in future) be read directly without deserialization, by indexing into the byte stream directly.
+ * <p/>
+ * The serialized value is optimized for values in the range 0 to 256 (negative will be rejected), and should produce
+ * output smaller or equal to vint serialization; when values are larger than 256, then the packing can produce 1 extra
+ * serialized byte.  Serialization is safe in these cases, and faster to skip.
  */
 public class SerializePacked
 {
-    public static void serializePackedInts(int[] vs, int from, int to, int max, DataOutputPlus out) throws IOException
+    public static void serializePackedSortedIntsAndLength(int[] vs, DataOutputPlus out) throws IOException
+    {
+        out.writeUnsignedVInt32(vs.length);
+        serializePackedSortedInts(vs, out);
+    }
+
+    public static void serializePackedSortedInts(int[] vs, DataOutputPlus out) throws IOException
+    {
+        if (vs.length == 0)
+            return;
+
+        int last = vs[vs.length - 1];
+        Invariants.require(last >= 0,
+                () -> String.format("Found a negative value at offset %d; value %d", (Object) (vs.length - 1), (Object) last));
+        out.writeUnsignedVInt32(last);
+        serializePackedInts(vs, 0, vs.length - 1, last, out);
+    }
+
+    public static int[] deserializePackedSortedIntsAndLength(DataInputPlus in) throws IOException
+    {
+        return deserializePackedSortedInts(in.readUnsignedVInt32(), in);
+    }
+
+    public static int[] deserializePackedSortedInts(int length, DataInputPlus in) throws IOException
+    {
+        if (length == 0)
+            return new int[0];
+
+        int last = in.readUnsignedVInt32();
+        int[] vs = new int[length];
+        deserializePackedInts(vs, 0, length - 1, last, in);
+        vs[length - 1] = last;
+        return vs;
+    }
+
+    public static void skipPackedSortedIntsAndLength(DataInputPlus in) throws IOException
+    {
+        skipPackedSortedInts(in.readUnsignedVInt32(), in);
+    }
+
+    public static void skipPackedSortedInts(int length, DataInputPlus in) throws IOException
+    {
+        if (length > 0)
+        {
+            int last = in.readUnsignedVInt32();
+            skipPackedInts(0, length - 1, last, in);
+        }
+    }
+
+    public static long serializedSizeOfPackedSortedIntsAndLength(int[] vs)
+    {
+        return TypeSizes.sizeofUnsignedVInt(vs.length) + serializedSizeOfPackedSortedInts(vs);
+    }
+
+    public static long serializedSizeOfPackedSortedInts(int[] vs)
+    {
+        if (vs.length == 0)
+            return 0;
+        int last = vs[vs.length - 1];
+        return TypeSizes.sizeofUnsignedVInt(last) + serializedPackedSize(vs.length - 1, last);
+    }
+
+    public static void serializePackedInts(int[] vs, int from, int to, long max, DataOutputPlus out) throws IOException
     {
         serializePacked((in, i) -> in[i], vs, from, to, max, out);
     }
 
-    public static void deserializePackedInts(int[] vs, int from, int to, int max, DataInputPlus in) throws IOException
+    public static void deserializePackedInts(int[] vs, int from, int to, long max, DataInputPlus in) throws IOException
     {
         deserializePacked((out, i, v) -> out[i] = (int)v, vs, from, to, max, in);
     }
 
-    public static long serializedPackedIntsSize(int[] vs, int from, int to, int max)
+    public static void skipPackedInts(int from, int to, long max, DataInputPlus in) throws IOException
+    {
+        in.skipBytesFully(serializedPackedSize(to - from, max));
+    }
+
+    public static long serializedPackedIntsSize(int[] vs, int from, int to, long max)
     {
         return serializedPackedSize(to - from, max);
     }
@@ -60,12 +132,15 @@ public static <In> void serializePacked(SerializeAdapter<In> adapter, In in, int
         if (bitsPerEntry == 0)
             return;
 
+        long outOfRange = -1L << bitsPerEntry;
         long buffer = 0L;
         int bufferCount = 0;
         for (int i = from; i < to; i++)
         {
             long v = adapter.get(in, i);
-            Invariants.require(v <= max);
+            int finalI = i;
+            Invariants.require(v >= 0 && (v & outOfRange) == 0,
+                    () -> String.format(v < 0 ? "Found a negative value at offset %d; value %d" : "Value out of range at offset %d; value %d", (Object) finalI, (Object) v));
             buffer |= v << bufferCount;
             bufferCount = bufferCount + bitsPerEntry;
             if (bufferCount >= 64)

src/java/org/apache/cassandra/service/accord/serializers/TableMetadatas.java

@@ -22,6 +22,9 @@
 import java.util.AbstractList;
 import java.util.Arrays;
 import java.util.Comparator;
+import java.util.List;
+
+import com.google.common.annotations.VisibleForTesting;
 
 import accord.utils.Invariants;
 import accord.utils.SortedArrays;
@@ -99,6 +102,14 @@ public static Complete of(TableMetadata metadata)
         return new One(metadata);
     }
 
+    @VisibleForTesting
+    public static Complete of(List<TableMetadata> values)
+    {
+        Collector collector = new Collector();
+        collector.addAll(values);
+        return collector.build();
+    }
+
     public static Complete ofSortedUnique(TableMetadata ... metadatas)
     {
         if (metadatas.length == 0)

src/java/org/apache/cassandra/service/accord/txn/AccordUpdate.java

@@ -72,7 +72,7 @@ public void failCondition()
 
     }
 
-    public boolean checkCondition(Data data)
+    public boolean checkAnyConditionMatch(Data data)
     {
         throw new UnsupportedOperationException();
     }

src/java/org/apache/cassandra/service/accord/txn/TxnQuery.java

@@ -119,7 +119,7 @@ public Result doCompute(TxnId txnId, Timestamp executeAt, Seekables<?, ?> keys,
 
             AccordUpdate accordUpdate = (AccordUpdate)update;
             TxnData txnData = (TxnData)data;
-            boolean conditionCheck = accordUpdate.checkCondition(data);
+            boolean conditionCheck = accordUpdate.checkAnyConditionMatch(data);
             // If the condition applied an empty result indicates success
             if (conditionCheck)
                 return new TxnData();