natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch cassandra-3.11 into trunk

This commit is contained in:
Benjamin Lerer 2017-01-27 10:57:46 +01:00
parent 113b92bcb2 84d836137d
commit 80728df563
5 changed files with 182 additions and 212 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -30,6 +30,7 @@
* Tables in system_distributed should not use gcgs of 0 (CASSANDRA-12954)
3.10
* Add duration type to the protocol V5 (CASSANDRA-12850)
* Fix duration type validation (CASSANDRA-13143)
* Fix flaky GcCompactionTest (CASSANDRA-12664)
* Fix TestHintedHandoff.hintedhandoff_decom_test (CASSANDRA-13058)

149
doc/native_protocol_v5.spec
View File

@ -202,51 +202,69 @@ Table of Contents
To describe the layout of the frame body for the messages in Section 4, we
define the following:
[int] A 4 bytes integer
[long] A 8 bytes integer
[byte] A 1 byte unsigned integer
[short] A 2 bytes unsigned integer
[string] A [short] n, followed by n bytes representing an UTF-8
string.
[long string] An [int] n, followed by n bytes representing an UTF-8 string.
[uuid] A 16 bytes long uuid.
[string list] A [short] n, followed by n [string].
[bytes] A [int] n, followed by n bytes if n >= 0. If n < 0,
no byte should follow and the value represented is `null`.
[value] A [int] n, followed by n bytes if n >= 0.
If n == -1 no byte should follow and the value represented is `null`.
If n == -2 no byte should follow and the value represented is
`not set` not resulting in any change to the existing value.
n < -2 is an invalid value and results in an error.
[short bytes] A [short] n, followed by n bytes if n >= 0.
[int] A 4 bytes integer
[long] A 8 bytes integer
[byte] A 1 byte unsigned integer
[short] A 2 bytes unsigned integer
[string] A [short] n, followed by n bytes representing an UTF-8
string.
[long string] An [int] n, followed by n bytes representing an UTF-8 string.
[uuid] A 16 bytes long uuid.
[string list] A [short] n, followed by n [string].
[bytes] A [int] n, followed by n bytes if n >= 0. If n < 0,
no byte should follow and the value represented is `null`.
[value] A [int] n, followed by n bytes if n >= 0.
If n == -1 no byte should follow and the value represented is `null`.
If n == -2 no byte should follow and the value represented is
`not set` not resulting in any change to the existing value.
n < -2 is an invalid value and results in an error.
[short bytes] A [short] n, followed by n bytes if n >= 0.
[option] A pair of <id><value> where <id> is a [short] representing
the option id and <value> depends on that option (and can be
of size 0). The supported id (and the corresponding <value>)
will be described when this is used.
[option list] A [short] n, followed by n [option].
[inet] An address (ip and port) to a node. It consists of one
[byte] n, that represents the address size, followed by n
[byte] representing the IP address (in practice n can only be
either 4 (IPv4) or 16 (IPv6)), following by one [int]
representing the port.
[inetaddr] An IP address (without a port) to a node. It consists of one
[byte] n, that represents the address size, followed by n
[byte] representing the IP address.
[consistency] A consistency level specification. This is a [short]
representing a consistency level with the following
correspondance:
0x0000 ANY
0x0001 ONE
0x0002 TWO
0x0003 THREE
0x0004 QUORUM
0x0005 ALL
0x0006 LOCAL_QUORUM
0x0007 EACH_QUORUM
0x0008 SERIAL
0x0009 LOCAL_SERIAL
0x000A LOCAL_ONE
[unsigned vint] An unsigned variable length integer. A vint is encoded with the most significant byte (MSB) first.
The most significant byte will contains the information about how many extra bytes need to be read
as well as the most significant bits of the integer.
The number of extra bytes to read is encoded as 1 bits on the left side.
For example, if we need to read 2 more bytes the first byte will start with 110
(e.g. 256 000 will be encoded on 3 bytes as [110]00011 11101000 00000000)
If the encoded integer is 8 bytes long the vint will be encoded on 9 bytes and the first
byte will be: 11111111
[vint] A signed variable length integer. This is encoded using zig-zag encoding and then sent
like an [unsigned vint]. Zig-zag encoding converts numbers as follows:
0 = 0, -1 = 1, 1 = 2, -2 = 3, 2 = 4, -3 = 5, 3 = 6 and so forth.
The purpose is to send small negative values as small unsigned values, so that we save bytes on the wire.
To encode a value n use "(n >> 31) ^ (n << 1)" for 32 bit values, and "(n >> 63) ^ (n << 1)"
for 64 bit values where "^" is the xor operation, "<<" is the left shift operation and ">>" is
the arithemtic right shift operation (highest-order bit is replicated).
Decode with "(n >> 1) ^ -(n & 1)".
[option] A pair of <id><value> where <id> is a [short] representing
the option id and <value> depends on that option (and can be
of size 0). The supported id (and the corresponding <value>)
will be described when this is used.
[option list] A [short] n, followed by n [option].
[inet] An address (ip and port) to a node. It consists of one
[byte] n, that represents the address size, followed by n
[byte] representing the IP address (in practice n can only be
either 4 (IPv4) or 16 (IPv6)), following by one [int]
representing the port.
[inetaddr] An IP address (without a port) to a node. It consists of one
[byte] n, that represents the address size, followed by n
[byte] representing the IP address.
[consistency] A consistency level specification. This is a [short]
representing a consistency level with the following
correspondance:
0x0000 ANY
0x0001 ONE
0x0002 TWO
0x0003 THREE
0x0004 QUORUM
0x0005 ALL
0x0006 LOCAL_QUORUM
0x0007 EACH_QUORUM
0x0008 SERIAL
0x0009 LOCAL_SERIAL
0x000A LOCAL_ONE
[string map] A [short] n, followed by n pair <k><v> where <k> and <v>
are [string].
@ -613,6 +631,7 @@ Table of Contents
0x0012 Time
0x0013 Smallint
0x0014 Tinyint
0x0015 Duration
0x0020 List: the value is an [option], representing the type
of the elements of the list.
0x0021 Map: the value is two [option], representing the types of the
@ -881,77 +900,86 @@ Table of Contents
An 8 byte floating point number in the IEEE 754 binary64 format.
6.8 float
6.8 duration
A duration is composed of 3 signed variable length integers ([vint]s).
The first [vint] represents a number of months, the second [vint] represents
a number of days, and the last [vint] represents a number of nanoseconds.
A duration can either be positive or negative. If a duration is positive
all the integers must be positive or zero. If a duration is
negative all the numbers must be negative or zero.
6.9 float
A 4 byte floating point number in the IEEE 754 binary32 format.
6.9 inet
6.10 inet
A 4 byte or 16 byte sequence denoting an IPv4 or IPv6 address, respectively.
6.10 int
6.11 int
A 4 byte two's complement integer.
6.11 list
6.12 list
A [int] n indicating the number of elements in the list, followed by n
elements. Each element is [bytes] representing the serialized value.
6.12 map
6.13 map
A [int] n indicating the number of key/value pairs in the map, followed by
n entries. Each entry is composed of two [bytes] representing the key
and value.
6.13 set
6.14 set
A [int] n indicating the number of elements in the set, followed by n
elements. Each element is [bytes] representing the serialized value.
6.14 smallint
6.15 smallint
A 2 byte two's complement integer.
6.15 text
6.16 text
A sequence of bytes conforming to the UTF-8 specifications.
6.16 time
6.17 time
An 8 byte two's complement long representing nanoseconds since midnight.
Valid values are in the range 0 to 86399999999999
6.17 timestamp
6.18 timestamp
An 8 byte two's complement integer representing a millisecond-precision
offset from the unix epoch (00:00:00, January 1st, 1970). Negative values
represent a negative offset from the epoch.
6.18 timeuuid
6.19 timeuuid
A 16 byte sequence representing a version 1 UUID as defined by RFC 4122.
6.19 tinyint
6.20 tinyint
A 1 byte two's complement integer.
6.20 tuple
6.21 tuple
A sequence of [bytes] values representing the items in a tuple. The encoding
of each element depends on the data type for that position in the tuple.
Null values may be represented by using length -1 for the [bytes]
representation of an element.
6.21 uuid
6.22 uuid
A 16 byte sequence representing any valid UUID as defined by RFC 4122.
6.22 varchar
6.23 varchar
An alias of the "text" type.
6.23 varint
6.24 varint
A variable-length two's complement encoding of a signed integer.
@ -1182,3 +1210,4 @@ Table of Contents
a failure reason code which indicates why the request failed on that node.
* Enlarged flag's bitmaps for QUERY, EXECUTE and BATCH messages from [byte] to [int]
(Sections 4.1.4, 4.1.6 and 4.1.7).
* Add the duration data type

62
src/java/org/apache/cassandra/transport/DataType.java
View File

@ -33,7 +33,7 @@ import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.utils.Pair;
public enum DataType implements OptionCodec.Codecable<DataType>
public enum DataType
{
CUSTOM (0, null, ProtocolVersion.V1),
ASCII (1, AsciiType.instance, ProtocolVersion.V1),
@ -56,13 +56,14 @@ public enum DataType implements OptionCodec.Codecable<DataType>
TIME (18, TimeType.instance, ProtocolVersion.V4),
SMALLINT (19, ShortType.instance, ProtocolVersion.V4),
BYTE (20, ByteType.instance, ProtocolVersion.V4),
DURATION (21, DurationType.instance, ProtocolVersion.V5),
LIST (32, null, ProtocolVersion.V1),
MAP (33, null, ProtocolVersion.V1),
SET (34, null, ProtocolVersion.V1),
UDT (48, null, ProtocolVersion.V3),
TUPLE (49, null, ProtocolVersion.V3);
public static final OptionCodec<DataType> codec = new OptionCodec<DataType>(DataType.class);
public static final Codec codec = new Codec();
private final int id;
private final ProtocolVersion protocolVersion;
@ -302,4 +303,61 @@ public enum DataType implements OptionCodec.Codecable<DataType>
{
return protocolVersion;
}
public static final class Codec
{
private final DataType[] ids;
public Codec()
{
DataType[] values = DataType.values();
ids = new DataType[getMaxId(values) + 1];
for (DataType opt : values)
{
int id = opt.getId(opt.getProtocolVersion());
DataType existingType = ids[id];
if (existingType != null)
throw new IllegalStateException(String.format("Duplicate option id %d", id));
ids[id] = opt;
}
}
private int getMaxId(DataType[] values)
{
int maxId = -1;
for (DataType opt : values)
maxId = Math.max(maxId, opt.getId(ProtocolVersion.CURRENT));
return maxId;
}
private DataType fromId(int id)
{
DataType opt = ids[id];
if (opt == null)
throw new ProtocolException(String.format("Unknown option id %d", id));
return opt;
}
public Pair<DataType, Object> decodeOne(ByteBuf body, ProtocolVersion version)
{
DataType opt = fromId(body.readUnsignedShort());
Object value = opt.readValue(body, version);
return Pair.create(opt, value);
}
public void writeOne(Pair<DataType, Object> option, ByteBuf dest, ProtocolVersion version)
{
DataType opt = option.left;
Object obj = option.right;
dest.writeShort(opt.getId(version));
opt.writeValue(obj, dest, version);
}
public int oneSerializedSize(Pair<DataType, Object> option, ProtocolVersion version)
{
DataType opt = option.left;
Object obj = option.right;
return 2 + opt.serializedValueSize(obj, version);
}
}
}

121
src/java/org/apache/cassandra/transport/OptionCodec.java
View File

@ -1,121 +0,0 @@
/*
* 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.transport;
import java.lang.reflect.Array;
import java.util.EnumMap;
import java.util.Map;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.cassandra.utils.Pair;
public class OptionCodec<T extends Enum<T> & OptionCodec.Codecable<T>>
{
public interface Codecable<T extends Enum<T>>
{
public int getId(ProtocolVersion version);
public Object readValue(ByteBuf cb, ProtocolVersion version);
public void writeValue(Object value, ByteBuf cb, ProtocolVersion version);
public int serializedValueSize(Object obj, ProtocolVersion version);
}
private final Class<T> klass;
private final T[] ids;
@SuppressWarnings({"unchecked"})
public OptionCodec(Class<T> klass)
{
this.klass = klass;
T[] values = klass.getEnumConstants();
int maxId = -1;
for (T opt : values)
maxId = Math.max(maxId, opt.getId(ProtocolVersion.CURRENT));
ids = (T[])Array.newInstance(klass, maxId + 1);
for (T opt : values)
{
if (ids[opt.getId(ProtocolVersion.CURRENT)] != null)
throw new IllegalStateException(String.format("Duplicate option id %d", opt.getId(ProtocolVersion.CURRENT)));
ids[opt.getId(ProtocolVersion.CURRENT)] = opt;
}
}
private T fromId(int id)
{
T opt = ids[id];
if (opt == null)
throw new ProtocolException(String.format("Unknown option id %d", id));
return opt;
}
public Map<T, Object> decode(ByteBuf body, ProtocolVersion version)
{
EnumMap<T, Object> options = new EnumMap<T, Object>(klass);
int n = body.readUnsignedShort();
for (int i = 0; i < n; i++)
{
T opt = fromId(body.readUnsignedShort());
Object value = opt.readValue(body, version);
if (options.containsKey(opt))
throw new ProtocolException(String.format("Duplicate option %s in message", opt.name()));
options.put(opt, value);
}
return options;
}
public ByteBuf encode(Map<T, Object> options, ProtocolVersion version)
{
int optLength = 2;
for (Map.Entry<T, Object> entry : options.entrySet())
optLength += 2 + entry.getKey().serializedValueSize(entry.getValue(), version);
ByteBuf cb = Unpooled.buffer(optLength);
cb.writeShort(options.size());
for (Map.Entry<T, Object> entry : options.entrySet())
{
T opt = entry.getKey();
cb.writeShort(opt.getId(version));
opt.writeValue(entry.getValue(), cb, version);
}
return cb;
}
public Pair<T, Object> decodeOne(ByteBuf body, ProtocolVersion version)
{
T opt = fromId(body.readUnsignedShort());
Object value = opt.readValue(body, version);
return Pair.create(opt, value);
}
public void writeOne(Pair<T, Object> option, ByteBuf dest, ProtocolVersion version)
{
T opt = option.left;
Object obj = option.right;
dest.writeShort(opt.getId(version));
opt.writeValue(obj, dest, version);
}
public int oneSerializedSize(Pair<T, Object> option, ProtocolVersion version)
{
T opt = option.left;
Object obj = option.right;
return 2 + opt.serializedValueSize(obj, version);
}
}

61
test/unit/org/apache/cassandra/transport/DataTypeTest.java
View File

@ -26,10 +26,13 @@ import java.util.Map;
import org.junit.Test;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.AsciiType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.utils.Pair;
import static org.junit.Assert.assertEquals;
@ -43,7 +46,7 @@ public class DataTypeTest
if (isComplexType(type))
continue;
Map<DataType, Object> options = Collections.singletonMap(type, (Object)type.toString());
Pair<DataType, Object> options = Pair.create(type, (Object)type.toString());
for (ProtocolVersion version : ProtocolVersion.SUPPORTED)
testEncodeDecode(type, options, version);
}
@ -53,7 +56,7 @@ public class DataTypeTest
public void TestListDataTypeSerialization()
{
DataType type = DataType.LIST;
Map<DataType, Object> options = Collections.singletonMap(type, (Object)LongType.instance);
Pair<DataType, Object> options = Pair.create(type, (Object)LongType.instance);
for (ProtocolVersion version : ProtocolVersion.SUPPORTED)
testEncodeDecode(type, options, version);
}
@ -65,41 +68,41 @@ public class DataTypeTest
List<AbstractType> value = new ArrayList<>();
value.add(LongType.instance);
value.add(AsciiType.instance);
Map<DataType, Object> options = Collections.singletonMap(type, (Object)value);
Pair<DataType, Object> options = Pair.create(type, (Object)value);
for (ProtocolVersion version : ProtocolVersion.SUPPORTED)
testEncodeDecode(type, options, version);
}
private void testEncodeDecode(DataType type, Map<DataType, Object> options, ProtocolVersion version)
private void testEncodeDecode(DataType type, Pair<DataType, Object> options, ProtocolVersion version)
{
ByteBuf dest = type.codec.encode(options, version);
Map<DataType, Object> results = type.codec.decode(dest, version);
int optLength = DataType.codec.oneSerializedSize(options, version);
ByteBuf dest = Unpooled.buffer(optLength);
DataType.codec.writeOne(options, dest, version);
Pair<DataType, Object> result = DataType.codec.decodeOne(dest, version);
for (DataType key : results.keySet())
System.out.println(result + "version " + version);
int ssize = type.serializedValueSize(result.right, version);
int esize = version.isSmallerThan(type.getProtocolVersion()) ? 2 + TypeSizes.encodedUTF8Length(result.right.toString()) : 0;
switch (type)
{
int ssize = type.serializedValueSize(results.get(key), version);
int esize = version.isSmallerThan(type.getProtocolVersion()) ? 2 + TypeSizes.encodedUTF8Length(results.get(key).toString()) : 0;
switch (type)
{
case LIST:
case SET:
esize += 2;
break;
case MAP:
esize += 4;
break;
case CUSTOM:
esize = 8;
break;
}
assertEquals(esize, ssize);
DataType expected = version.isSmallerThan(type.getProtocolVersion())
? DataType.CUSTOM
: type;
assertEquals(expected, key);
case LIST:
case SET:
esize += 2;
break;
case MAP:
esize += 4;
break;
case CUSTOM:
esize = 8;
break;
}
}
assertEquals(esize, ssize);
DataType expected = version.isSmallerThan(type.getProtocolVersion())
? DataType.CUSTOM
: type;
assertEquals(expected, result.left);
}
private boolean isComplexType(DataType type)
{