Use byte[] directly in QueryOptions instead of ByteBuffer and convert them to ArrayCell instead of BufferCell.

Additionally replace List with array for bind values (we know the size in advance during a decoding), so in total: List<List> is replaced with byte[][] QueryOptions classes support both ways to get values now: using an old API with ByteBuffer and a new API with byte[].

Patch by Dmitry Konstantinov; reviewed by Michael Semb Wever for CASSANDRA-20166

Author: netudima

CHANGES.txt

@@ -1,4 +1,5 @@
 5.1
+ * Use byte[] directly in QueryOptions instead of ByteBuffer and convert them to ArrayCell instead of BufferCell to reduce allocations (CASSANDRA-20166)
  * Log queries scanning too many SSTables per read (CASSANDRA-21048)
  * Extend nodetool verify to (optionally) validate SAI files (CASSANDRA-20949)
  * Fix CompressionDictionary being closed while still in use (CASSANDRA-21047)

src/java/org/apache/cassandra/audit/AuditLogManager.java

@@ -22,7 +22,6 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
-import java.nio.ByteBuffer;
 import java.security.AccessControlContext;
 import java.security.AccessController;
 import java.security.Principal;
@@ -317,7 +316,7 @@ else if (statement != null)
     public void batchSuccess(BatchStatement.Type batchType,
                              List<? extends CQLStatement> statements,
                              List<String> queries,
-                             List<List<ByteBuffer>> values,
+                             List<byte[][]> values,
                              QueryOptions options,
                              QueryState state,
                              long queryStartTimeMillis,
@@ -358,7 +357,7 @@ public void batchSuccess(BatchStatement.Type batchType,
         }
     }
 
-    public void batchFailure(BatchStatement.Type batchType, List<? extends CQLStatement> statements, List<String> queries, List<List<ByteBuffer>> values, QueryOptions options, QueryState state, Exception cause)
+    public void batchFailure(BatchStatement.Type batchType, List<? extends CQLStatement> statements, List<String> queries, List<byte[][]> values, QueryOptions options, QueryState state, Exception cause)
     {
         String auditMessage = String.format("BATCH of %d statements at consistency %s", statements.size(), options.getConsistency());
         AuditLogEntry entry = new AuditLogEntry.Builder(state).setOperation(auditMessage)

src/java/org/apache/cassandra/cql3/BatchQueryOptions.java

@@ -31,6 +31,8 @@
 import org.apache.commons.lang3.builder.ToStringBuilder;
 import org.apache.commons.lang3.builder.ToStringStyle;
 
+import static org.apache.cassandra.utils.ByteArrayUtil.convertToByteBufferValue;
+
 public abstract class BatchQueryOptions
 {
     public static BatchQueryOptions DEFAULT = withoutPerStatementVariables(QueryOptions.DEFAULT);
@@ -49,7 +51,7 @@ public static BatchQueryOptions withoutPerStatementVariables(QueryOptions option
         return new WithoutPerStatementVariables(options, Collections.<Object>emptyList());
     }
 
-    public static BatchQueryOptions withPerStatementVariables(QueryOptions options, List<List<ByteBuffer>> variables, List<Object> queryOrIdList)
+    public static BatchQueryOptions withPerStatementVariables(QueryOptions options, List<byte[][]> variables, List<Object> queryOrIdList)
     {
         return new WithPerStatementVariables(options, variables, queryOrIdList);
     }
@@ -91,6 +93,50 @@ public long getNowInSeconds(QueryState state)
         return wrapped.getNowInSeconds(state);
     }
 
+    private static class BatchQueryOptionsWrapper extends QueryOptions.QueryOptionsWrapper {
+        private final byte[][] valuesAsByteArray;
+        private List<ByteBuffer> values; // initialized on demand
+
+        BatchQueryOptionsWrapper(QueryOptions wrapped, byte[][] vars)
+        {
+            super(wrapped);
+            this.valuesAsByteArray = vars;
+        }
+        public List<ByteBuffer> getValues()
+        {
+            if (values == null)
+            {
+                values = new ArrayList<>(valuesAsByteArray.length);
+                for (byte[] byteArrayValue : valuesAsByteArray)
+                    values.add(convertToByteBufferValue(byteArrayValue));
+            }
+            return values;
+        }
+
+        public int getValuesSize()
+        {
+            return valuesAsByteArray.length;
+        }
+
+        public ByteBuffer getValue(int index)
+        {
+            if (values == null) // we convert values to ByteBuffer in a lazy way, on demand
+                return convertToByteBufferValue(valuesAsByteArray[index]);
+            else
+                return values.get(index);
+        }
+
+        public boolean isByteArrayValuesGetSupported()
+        {
+            return true;
+        }
+
+        public byte[][] getByteArrayValues()
+        {
+            return valuesAsByteArray;
+        }
+    }
+
     private static class WithoutPerStatementVariables extends BatchQueryOptions
     {
         private WithoutPerStatementVariables(QueryOptions wrapped, List<Object> queryOrIdList)
@@ -108,20 +154,12 @@ private static class WithPerStatementVariables extends BatchQueryOptions
     {
         private final List<QueryOptions> perStatementOptions;
 
-        private WithPerStatementVariables(QueryOptions wrapped, List<List<ByteBuffer>> variables, List<Object> queryOrIdList)
+        private WithPerStatementVariables(QueryOptions wrapped, List<byte[][]> variables, List<Object> queryOrIdList)
         {
             super(wrapped, queryOrIdList);
             this.perStatementOptions = new ArrayList<>(variables.size());
-            for (final List<ByteBuffer> vars : variables)
-            {
-                perStatementOptions.add(new QueryOptions.QueryOptionsWrapper(wrapped)
-                {
-                    public List<ByteBuffer> getValues()
-                    {
-                        return vars;
-                    }
-                });
-            }
+            for (final byte[][] vars : variables)
+                perStatementOptions.add(new BatchQueryOptionsWrapper(wrapped, vars));
         }
 
         public QueryOptions forStatement(int i)

src/java/org/apache/cassandra/cql3/QueryEvents.java

@@ -18,7 +18,6 @@
 
 package org.apache.cassandra.cql3;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
@@ -143,10 +142,14 @@ public void notifyExecuteFailure(QueryHandler.Prepared prepared,
         }
     }
 
+    /**
+     * Note: we use {@code byte[][]} for values of batch statement elements instead of {@code List<ByteBuffer>} here
+     * to reduce memory allocation when we decode and process a batch statement
+     */
     public void notifyBatchSuccess(BatchStatement.Type batchType,
                                    List<? extends CQLStatement> statements,
                                    List<String> queries,
-                                   List<List<ByteBuffer>> values,
+                                   List<byte[][]> values,
                                    QueryOptions options,
                                    QueryState state,
                                    long queryTime,
@@ -164,10 +167,14 @@ public void notifyBatchSuccess(BatchStatement.Type batchType,
         }
     }
 
+    /**
+     * Note: we use {@code byte[][]} for values of batch statement elements instead of {@code List<ByteBuffer>} here
+     * to reduce memory allocation when we decode and process a batch statement
+     */
     public void notifyBatchFailure(List<QueryHandler.Prepared> prepared,
                                    BatchStatement.Type batchType,
                                    List<Object> queryOrIdList,
-                                   List<List<ByteBuffer>> values,
+                                   List<byte[][]> values,
                                    QueryOptions options,
                                    QueryState state,
                                    Exception cause)
@@ -288,15 +295,15 @@ default void executeFailure(@Nullable CQLStatement statement,
         default void batchSuccess(BatchStatement.Type batchType,
                                   List<? extends CQLStatement> statements,
                                   List<String> queries,
-                                  List<List<ByteBuffer>> values,
+                                  List<byte[][]> values,
                                   QueryOptions options,
                                   QueryState state,
                                   long queryTime,
                                   Message.Response response) {}
         default void batchFailure(BatchStatement.Type batchType,
                                   List<? extends CQLStatement> statements,
                                   List<String> queries,
-                                  List<List<ByteBuffer>> values,
+                                  List<byte[][]> values,
                                   QueryOptions options,
                                   QueryState state,
                                   Exception cause) {}