cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bened...@apache.org
Subject [2/2] cassandra git commit: Serialize ClusteringPrefix in microbatches, using vint encoding
Date Thu, 23 Jul 2015 22:26:48 GMT
Serialize ClusteringPrefix in microbatches, using vint encoding

patch by benedict; reviewed by sylvain for CASSANDRA-9708


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/5786b320
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/5786b320
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/5786b320

Branch: refs/heads/trunk
Commit: 5786b3204d6da352124338c0130451e27dd056b0
Parents: c4c9eae
Author: Benedict Elliott Smith <benedict@apache.org>
Authored: Wed Jun 17 09:58:41 2015 +0100
Committer: Benedict Elliott Smith <benedict@apache.org>
Committed: Thu Jul 23 23:26:24 2015 +0100

----------------------------------------------------------------------
 .../apache/cassandra/db/ClusteringPrefix.java   | 129 +++++++++----------
 src/java/org/apache/cassandra/db/TypeSizes.java |   5 +
 .../cassandra/db/rows/UnfilteredSerializer.java |   2 +-
 .../cassandra/cql3/SerializationMirrorTest.java |  63 +++++++++
 4 files changed, 133 insertions(+), 66 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/5786b320/src/java/org/apache/cassandra/db/ClusteringPrefix.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ClusteringPrefix.java b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
index 7b9d582..713ad1b 100644
--- a/src/java/org/apache/cassandra/db/ClusteringPrefix.java
+++ b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
@@ -286,109 +286,103 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
 
         void serializeValuesWithoutSize(ClusteringPrefix clustering, DataOutputPlus out,
int version, List<AbstractType<?>> types) throws IOException
         {
-            if (clustering.size() == 0)
-                return;
-
-            writeHeader(clustering, out);
-            for (int i = 0; i < clustering.size(); i++)
+            int offset = 0;
+            int clusteringSize = clustering.size();
+            // serialize in batches of 32, to avoid garbage when deserializing headers
+            while (offset < clusteringSize)
             {
-                ByteBuffer v = clustering.get(i);
-                if (v == null || !v.hasRemaining())
-                    continue; // handled in the header
-
-                types.get(i).writeValue(v, out);
+                // we micro-batch the headers, so that we can incur fewer method calls,
+                // and generate no garbage on deserialization;
+                // we piggyback on vint encoding so that, typically, only 1 byte is used
per 32 clustering values,
+                // i.e. more than we ever expect to see
+                int limit = Math.min(clusteringSize, offset + 32);
+                out.writeUnsignedVInt(makeHeader(clustering, offset, limit));
+                while (offset < limit)
+                {
+                    ByteBuffer v = clustering.get(offset);
+                    if (v != null && v.hasRemaining())
+                        types.get(offset).writeValue(v, out);
+                    offset++;
+                }
             }
         }
 
         long valuesWithoutSizeSerializedSize(ClusteringPrefix clustering, int version, List<AbstractType<?>>
types)
         {
-            if (clustering.size() == 0)
-                return 0;
-
-            long size = headerBytesCount(clustering.size());
-            for (int i = 0; i < clustering.size(); i++)
+            long result = 0;
+            int offset = 0;
+            int clusteringSize = clustering.size();
+            while (offset < clusteringSize)
+            {
+                int limit = Math.min(clusteringSize, offset + 32);
+                result += TypeSizes.sizeofUnsignedVInt(makeHeader(clustering, offset, limit));
+                offset = limit;
+            }
+            for (int i = 0; i < clusteringSize; i++)
             {
                 ByteBuffer v = clustering.get(i);
                 if (v == null || !v.hasRemaining())
                     continue; // handled in the header
 
-                size += types.get(i).writtenLength(v);
+                result += types.get(i).writtenLength(v);
             }
-            return size;
+            return result;
         }
 
         ByteBuffer[] deserializeValuesWithoutSize(DataInputPlus in, int size, int version,
List<AbstractType<?>> types) throws IOException
         {
             // Callers of this method should handle the case where size = 0 (in all case
we want to return a special value anyway).
             assert size > 0;
-
             ByteBuffer[] values = new ByteBuffer[size];
-            int[] header = readHeader(size, in);
-            for (int i = 0; i < size; i++)
+            int offset = 0;
+            while (offset < size)
             {
-                values[i] = isNull(header, i)
-                          ? null
-                          : (isEmpty(header, i) ? ByteBufferUtil.EMPTY_BYTE_BUFFER : types.get(i).readValue(in));
+                long header = in.readUnsignedVInt();
+                int limit = Math.min(size, offset + 32);
+                while (offset < limit)
+                {
+                    values[offset] = isNull(header, offset)
+                                ? null
+                                : (isEmpty(header, offset) ? ByteBufferUtil.EMPTY_BYTE_BUFFER
: types.get(offset).readValue(in));
+                    offset++;
+                }
             }
             return values;
         }
 
-        private int headerBytesCount(int size)
-        {
-            // For each component, we store 2 bit to know if the component is empty or null
(or neither).
-            // We thus handle 4 component per byte
-            return size / 4 + (size % 4 == 0 ? 0 : 1);
-        }
-
         /**
          * Whatever the type of a given clustering column is, its value can always be either
empty or null. So we at least need to distinguish those
          * 2 values, and because we want to be able to store fixed width values without appending
their (fixed) size first, we need a way to encode
          * empty values too. So for that, every clustering prefix includes a "header" that
contains 2 bits per element in the prefix. For each element,
          * those 2 bits encode whether the element is null, empty, or none of those.
          */
-        private void writeHeader(ClusteringPrefix clustering, DataOutputPlus out) throws
IOException
+        private static long makeHeader(ClusteringPrefix clustering, int offset, int limit)
         {
-            int nbBytes = headerBytesCount(clustering.size());
-            for (int i = 0; i < nbBytes; i++)
+            long header = 0;
+            for (int i = offset ; i < limit ; i++)
             {
-                int b = 0;
-                for (int j = 0; j < 4; j++)
-                {
-                    int c = i * 4 + j;
-                    if (c >= clustering.size())
-                        break;
-
-                    ByteBuffer v = clustering.get(c);
-                    if (v == null)
-                        b |= (1 << (j * 2) + 1);
-                    else if (!v.hasRemaining())
-                        b |= (1 << (j * 2));
-                }
-                out.writeByte((byte)b);
+                ByteBuffer v = clustering.get(i);
+                // no need to do modulo arithmetic for i, since the left-shift execute on
the modulus of RH operand by definition
+                if (v == null)
+                    header |= (1L << (i * 2) + 1);
+                else if (!v.hasRemaining())
+                    header |= (1L << (i * 2));
             }
-        }
-
-        private int[] readHeader(int size, DataInputPlus in) throws IOException
-        {
-            int nbBytes = headerBytesCount(size);
-            int[] header = new int[nbBytes];
-            for (int i = 0; i < nbBytes; i++)
-                header[i] = in.readUnsignedByte();
             return header;
         }
 
-        private static boolean isNull(int[] header, int i)
+        // no need to do modulo arithmetic for i, since the left-shift execute on the modulus
of RH operand by definition
+        private static boolean isNull(long header, int i)
         {
-            int b = header[i / 4];
-            int mask = 1 << ((i % 4) * 2) + 1;
-            return (b & mask) != 0;
+            long mask = 1L << (i * 2) + 1;
+            return (header & mask) != 0;
         }
 
-        private static boolean isEmpty(int[] header, int i)
+        // no need to do modulo arithmetic for i, since the left-shift execute on the modulus
of RH operand by definition
+        private static boolean isEmpty(long header, int i)
         {
-            int b = header[i / 4];
-            int mask = 1 << ((i % 4) * 2);
-            return (b & mask) != 0;
+            long mask = 1L << (i * 2);
+            return (header & mask) != 0;
         }
     }
 
@@ -408,7 +402,7 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
         private final SerializationHeader serializationHeader;
 
         private boolean nextIsRow;
-        private int[] nextHeader;
+        private long nextHeader;
 
         private int nextSize;
         private ClusteringPrefix.Kind nextKind;
@@ -428,7 +422,6 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
             this.nextIsRow = UnfilteredSerializer.kind(flags) == Unfiltered.Kind.ROW;
             this.nextKind = nextIsRow ? Kind.CLUSTERING : ClusteringPrefix.Kind.values()[in.readByte()];
             this.nextSize = nextIsRow ? comparator.size() : in.readUnsignedShort();
-            this.nextHeader = serializer.readHeader(nextSize, in);
             this.deserializedSize = 0;
 
             // The point of the deserializer is that some of the clustering prefix won't
actually be used (because they are not
@@ -478,6 +471,9 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
             if (deserializedSize == nextSize)
                 return false;
 
+            if ((deserializedSize % 32) == 0)
+                nextHeader = in.readUnsignedVInt();
+
             int i = deserializedSize++;
             nextValues[i] = Serializer.isNull(nextHeader, i)
                           ? null
@@ -513,9 +509,12 @@ public interface ClusteringPrefix extends IMeasurableMemory, Clusterable
         {
             for (int i = deserializedSize; i < nextSize; i++)
             {
+                if ((i % 32) == 0)
+                    nextHeader = in.readUnsignedVInt();
                 if (!Serializer.isNull(nextHeader, i) && !Serializer.isEmpty(nextHeader,
i))
                     serializationHeader.clusteringTypes().get(i).skipValue(in);
             }
+            deserializedSize = nextSize;
             return nextKind;
         }
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5786b320/src/java/org/apache/cassandra/db/TypeSizes.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/TypeSizes.java b/src/java/org/apache/cassandra/db/TypeSizes.java
index 73766c8..7e5bd87 100644
--- a/src/java/org/apache/cassandra/db/TypeSizes.java
+++ b/src/java/org/apache/cassandra/db/TypeSizes.java
@@ -102,4 +102,9 @@ public final class TypeSizes
     {
         return VIntCoding.computeVIntSize(value);
     }
+
+    public static int sizeofUnsignedVInt(long value)
+    {
+        return VIntCoding.computeUnsignedVIntSize(value);
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5786b320/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
index 4072f8d..11fa800 100644
--- a/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
+++ b/src/java/org/apache/cassandra/db/rows/UnfilteredSerializer.java
@@ -21,8 +21,8 @@ import java.io.IOException;
 
 import org.apache.cassandra.config.ColumnDefinition;
 import org.apache.cassandra.db.*;
-import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.io.util.DataInputPlus;
+import org.apache.cassandra.io.util.DataOutputPlus;
 import org.apache.cassandra.utils.SearchIterator;
 
 /**

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5786b320/test/unit/org/apache/cassandra/cql3/SerializationMirrorTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/cql3/SerializationMirrorTest.java b/test/unit/org/apache/cassandra/cql3/SerializationMirrorTest.java
new file mode 100644
index 0000000..49f77a7
--- /dev/null
+++ b/test/unit/org/apache/cassandra/cql3/SerializationMirrorTest.java
@@ -0,0 +1,63 @@
+/*
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*    http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing,
+* software distributed under the License is distributed on an
+* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+* KIND, either express or implied.  See the License for the
+* specific language governing permissions and limitations
+* under the License.
+*/
+package org.apache.cassandra.cql3;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
+
+import org.junit.Test;
+
+import junit.framework.Assert;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class SerializationMirrorTest extends CQLTester
+{
+
+    @Test
+    public void testManyClusterings() throws Throwable
+    {
+        StringBuilder table = new StringBuilder("CREATE TABLE %s (a TEXT");
+        StringBuilder cols = new StringBuilder();
+        StringBuilder args = new StringBuilder("?");
+        List<Object> vals = new ArrayList<>();
+        vals.add("a");
+        for (int i = 0 ; i < 40 ; i++)
+        {
+            table.append(", c").append(i).append(" text");
+            cols.append(", c").append(i);
+            if (ThreadLocalRandom.current().nextBoolean())
+                vals.add(Integer.toString(i));
+            else
+                vals.add("");
+            args.append(",?");
+        }
+        args.append(",?");
+        vals.add("value");
+        table.append(", v text, PRIMARY KEY ((a)").append(cols).append("))");
+        createTable(table.toString());
+
+        execute("INSERT INTO %s (a" + cols + ", v) VALUES (" + args+ ")", vals.toArray());
+        flush();
+        UntypedResultSet.Row row = execute("SELECT * FROM %s").one();
+        for (int i = 0 ; i < row.getColumns().size() ; i++)
+            Assert.assertEquals(vals.get(i), row.getString(i == 0 ? "a" : i < 41 ? "c"
+ (i - 1) : "v"));
+    }
+
+}


Mime
View raw message