Support collections natively in the binary protocol

patch by slebresne; reviewed by Paul Cannon for CASSANDRA-4453

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>.

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();
             }

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)
+            switch (entry.left)
-                return TypeParser.parse((String)entry.right);
+            {
-            else
+                case CUSTOM:
-                return entry.left.type;
+                    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)
         {

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(oneSerializedSize(option));
-        ChannelBuffer cb = ChannelBuffers.buffer(l);
 
         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);
+    }
 }