Support collections natively in the binary protocol patch by slebresne; reviewed by Paul Cannon for CASSANDRA-4453
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/ba231f4e Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/ba231f4e Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/ba231f4e Branch: refs/heads/trunk Commit: ba231f4eb53f01ceb297f85d52592600f14a9bbb Parents: 5e5fbc6 Author: Jonathan Ellis <[email protected]> Authored: Tue Aug 14 16:50:39 2012 -0500 Committer: Jonathan Ellis <[email protected]> Committed: Tue Aug 14 16:51:08 2012 -0500 ---------------------------------------------------------------------- doc/native_protocol.spec | 8 ++ src/java/org/apache/cassandra/cql3/ResultSet.java | 5 +- .../org/apache/cassandra/transport/DataType.java | 78 ++++++++++++++- .../apache/cassandra/transport/OptionCodec.java | 10 ++- 4 files changed, 91 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/ba231f4e/doc/native_protocol.spec ---------------------------------------------------------------------- diff --git a/doc/native_protocol.spec b/doc/native_protocol.spec index ae3c1fd..0092f5d 100644 --- a/doc/native_protocol.spec +++ b/doc/native_protocol.spec @@ -49,6 +49,8 @@ Table of Contents . . +---------------------------------------- + The protocol is big-endian (network byte order). + Each frame contains a fixed size header (8 bytes) followed by a variable size body. The header is described in Section 2. The content of the body depends on the header opcode value (the body can in particular be empty for some @@ -355,6 +357,12 @@ Table of Contents 0x000D Varchar 0x000E Varint 0x000F Timeuuid + 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 + keys and values of the map + 0x0022 Set: the value is an [option], representing the type + of the elements of the set - <rows_count> is an [int] representing the number of rows present in this result. Those rows are serialized in the <rows_content> part. - <rows_content> is composed of <row_1>...<row_m> where m is <rows_count>. http://git-wip-us.apache.org/repos/asf/cassandra/blob/ba231f4e/src/java/org/apache/cassandra/cql3/ResultSet.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/cql3/ResultSet.java b/src/java/org/apache/cassandra/cql3/ResultSet.java index cb5e89f..152edb9 100644 --- a/src/java/org/apache/cassandra/cql3/ResultSet.java +++ b/src/java/org/apache/cassandra/cql3/ResultSet.java @@ -204,7 +204,6 @@ public class ResultSet public static class Metadata { - private static OptionCodec<DataType> dataTypeCodec = new OptionCodec<DataType>(DataType.class); public static final CBCodec<Metadata> codec = new Codec(); public final EnumSet<Flag> flags; @@ -277,7 +276,7 @@ public class ResultSet String ksName = globalTablesSpec ? globalKsName : CBUtil.readString(body); String cfName = globalTablesSpec ? globalCfName : CBUtil.readString(body); ColumnIdentifier colName = new ColumnIdentifier(CBUtil.readString(body), true); - AbstractType type = DataType.toType(dataTypeCodec.decodeOne(body)); + AbstractType type = DataType.toType(DataType.codec.decodeOne(body)); names.add(new ColumnSpecification(ksName, cfName, colName, type)); } return new Metadata(flags, names); @@ -309,7 +308,7 @@ public class ResultSet builder.addString(name.cfName); } builder.addString(name.toString()); - builder.add(dataTypeCodec.encodeOne(DataType.fromType(name.type))); + builder.add(DataType.codec.encodeOne(DataType.fromType(name.type))); } return builder.build(); } http://git-wip-us.apache.org/repos/asf/cassandra/blob/ba231f4e/src/java/org/apache/cassandra/transport/DataType.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/transport/DataType.java b/src/java/org/apache/cassandra/transport/DataType.java index 5254945..9a8c2f0 100644 --- a/src/java/org/apache/cassandra/transport/DataType.java +++ b/src/java/org/apache/cassandra/transport/DataType.java @@ -17,8 +17,11 @@ */ package org.apache.cassandra.transport; +import java.util.ArrayList; +import java.util.Arrays; import java.util.HashMap; import java.util.Map; +import java.util.List; import com.google.common.base.Charsets; import org.jboss.netty.buffer.ChannelBuffer; @@ -44,7 +47,12 @@ public enum DataType implements OptionCodec.Codecable<DataType> UUID (12, UUIDType.instance), VARCHAR (13, UTF8Type.instance), VARINT (14, IntegerType.instance), - TIMEUUID (15, TimeUUIDType.instance); + TIMEUUID (15, TimeUUIDType.instance), + LIST (32, null), + MAP (33, null), + SET (34, null); + + public static final OptionCodec<DataType> codec = new OptionCodec<DataType>(DataType.class); private final int id; private final AbstractType type; @@ -75,6 +83,15 @@ public enum DataType implements OptionCodec.Codecable<DataType> { case CUSTOM: return CBUtil.readString(cb); + case LIST: + return DataType.toType(codec.decodeOne(cb)); + case SET: + return DataType.toType(codec.decodeOne(cb)); + case MAP: + List<AbstractType> l = new ArrayList<AbstractType>(2); + l.add(DataType.toType(codec.decodeOne(cb))); + l.add(DataType.toType(codec.decodeOne(cb))); + return l; default: return null; } @@ -88,6 +105,17 @@ public enum DataType implements OptionCodec.Codecable<DataType> assert value instanceof String; cb.writeBytes(CBUtil.stringToCB((String)value)); break; + case LIST: + cb.writeBytes(codec.encodeOne(DataType.fromType((AbstractType)value))); + break; + case SET: + cb.writeBytes(codec.encodeOne(DataType.fromType((AbstractType)value))); + break; + case MAP: + List<AbstractType> l = (List<AbstractType>)value; + cb.writeBytes(codec.encodeOne(DataType.fromType(l.get(0)))); + cb.writeBytes(codec.encodeOne(DataType.fromType(l.get(1)))); + break; } } @@ -97,6 +125,15 @@ public enum DataType implements OptionCodec.Codecable<DataType> { case CUSTOM: return 2 + ((String)value).getBytes(Charsets.UTF_8).length; + case LIST: + case SET: + return codec.oneSerializedSize(DataType.fromType((AbstractType)value)); + case MAP: + List<AbstractType> l = (List<AbstractType>)value; + int s = 0; + s += codec.oneSerializedSize(DataType.fromType(l.get(0))); + s += codec.oneSerializedSize(DataType.fromType(l.get(1))); + return s; default: return 0; } @@ -106,19 +143,50 @@ public enum DataType implements OptionCodec.Codecable<DataType> { DataType dt = dataTypeMap.get(type); if (dt == null) + { + if (type.isCollection()) + { + if (type instanceof ListType) + { + return Pair.<DataType, Object>create(LIST, ((ListType)type).elements); + } + else if (type instanceof MapType) + { + MapType mt = (MapType)type; + return Pair.<DataType, Object>create(MAP, Arrays.asList(mt.keys, mt.values)); + } + else + { + assert type instanceof SetType; + return Pair.<DataType, Object>create(LIST, ((SetType)type).elements); + } + } return Pair.<DataType, Object>create(CUSTOM, type.toString()); + } else + { return Pair.create(dt, null); + } } public static AbstractType toType(Pair<DataType, Object> entry) { try { - if (entry.left == CUSTOM) - return TypeParser.parse((String)entry.right); - else - return entry.left.type; + switch (entry.left) + { + case CUSTOM: + return TypeParser.parse((String)entry.right); + case LIST: + return ListType.getInstance((AbstractType)entry.right); + case SET: + return SetType.getInstance((AbstractType)entry.right); + case MAP: + List<AbstractType> l = (List<AbstractType>)entry.right; + return MapType.getInstance(l.get(0), l.get(1)); + default: + return entry.left.type; + } } catch (ConfigurationException e) { http://git-wip-us.apache.org/repos/asf/cassandra/blob/ba231f4e/src/java/org/apache/cassandra/transport/OptionCodec.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/transport/OptionCodec.java b/src/java/org/apache/cassandra/transport/OptionCodec.java index f5e702f..7652c24 100644 --- a/src/java/org/apache/cassandra/transport/OptionCodec.java +++ b/src/java/org/apache/cassandra/transport/OptionCodec.java @@ -109,11 +109,17 @@ public class OptionCodec<T extends Enum<T> & OptionCodec.Codecable<T>> T opt = option.left; Object obj = option.right; - int l = 2 + opt.serializedValueSize(obj); - ChannelBuffer cb = ChannelBuffers.buffer(l); + ChannelBuffer cb = ChannelBuffers.buffer(oneSerializedSize(option)); cb.writeShort(opt.getId()); opt.writeValue(obj, cb); return cb; } + + public int oneSerializedSize(Pair<T, Object> option) + { + T opt = option.left; + Object obj = option.right; + return 2 + opt.serializedValueSize(obj); + } }
