cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From slebre...@apache.org
Subject [1/2] git commit: Move resultset type information into prepare, not execute
Date Tue, 02 Jul 2013 18:51:15 GMT
Updated Branches:
  refs/heads/trunk ec3b8f817 -> b068a9c4f


Move resultset type information into prepare, not execute

patch by slebresne; reviewed by iamaleksey for CASSANDRA-5649


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

Branch: refs/heads/trunk
Commit: dad2f11aa660de36896859ca27b454cce26b4590
Parents: ec3b8f8
Author: Sylvain Lebresne <sylvain@datastax.com>
Authored: Mon Jul 1 14:50:50 2013 +0200
Committer: Sylvain Lebresne <sylvain@datastax.com>
Committed: Tue Jul 2 20:48:01 2013 +0200

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 doc/native_protocol_v2.spec                     | 57 ++++++++++---
 .../apache/cassandra/cql3/QueryProcessor.java   |  2 +-
 .../org/apache/cassandra/cql3/ResultSet.java    | 84 ++++++++++++++------
 .../cql3/statements/SelectStatement.java        |  5 ++
 .../cassandra/cql3/statements/Selection.java    |  5 ++
 .../org/apache/cassandra/transport/Client.java  |  2 +-
 .../cassandra/transport/SimpleClient.java       |  2 +-
 .../transport/messages/ExecuteMessage.java      | 69 ++++++++++++++--
 .../transport/messages/QueryMessage.java        | 84 +++++++++++++++++---
 .../transport/messages/ResultMessage.java       | 32 ++++++--
 11 files changed, 278 insertions(+), 65 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 2520b23..a50038d 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -69,6 +69,7 @@
  * Rename Table to Keyspace (CASSANDRA-5613)
  * Ensure changing column_index_size_in_kb on different nodes don't corrupt the
    sstable (CASSANDRA-5454)
+ * Move resultset type information into prepare, not execute (CASSANDRA-5649)
 
 
 1.2.7

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/doc/native_protocol_v2.spec
----------------------------------------------------------------------
diff --git a/doc/native_protocol_v2.spec b/doc/native_protocol_v2.spec
index e0ac541..196b4f5 100644
--- a/doc/native_protocol_v2.spec
+++ b/doc/native_protocol_v2.spec
@@ -279,15 +279,23 @@ Table of Contents
 4.1.4. QUERY
 
   Performs a CQL query. The body of the message must be:
-    <query><consistency><result_page_size>[<n><value_1>...<value_n>]
+    <query><consistency><flags>[<result_page_size>][<n><value_1>...<value_n>]
   where:
+    - <flags> is a [byte] whose bits define the options for this query and
+      in particular influence what the remainder of the message contains.
+      A flag is set if the bit corresponding to its `mask` is set. Supported
+      flags are, given there mask:
+        0x01: Page_size. In that case, <result_page_size> is an [int]
+              controlling the desired page size of the result (in CQL3 rows).
+              See the section on paging (Section 7) for more details.
+        0x02: Values. In that case, a [short] <n> followed by <n> [bytes]
+              values are provided. Those value are used for bound variables in
+              the query.
+        0x04: Skip_metadata. If present, the Result Set returned as a response
+              to that query (if any) will have the NO_METADATA flag (see
+              Section 4.2.5.2).
     - <query> the query, [long string].
     - <consistency> is the [consistency] level for the operation.
-    - <result_page_size> is an [int] controlling the desired page size of the
-      result (in CQL3 rows). A negative value disable paging of the result. See the
-      section on paging (Section 7) for more details.
-    - optional: <n> [short], the number of following values.
-    - optional: <value_1>...<value_n> are [bytes] to use for bound variables
in the query.
 
   Note that the consistency is ignored by some queries (USE, CREATE, ALTER,
   TRUNCATE, ...).
@@ -308,7 +316,7 @@ Table of Contents
 4.1.6. EXECUTE
 
   Executes a prepared query. The body of the message must be:
-    <id><n><value_1>....<value_n><consistency><result_page_size>
+    <id><n><value_1>....<value_n><consistency><flags>[<result_page_size>]
   where:
     - <id> is the prepared query ID. It's the [short bytes] returned as a
       response to a PREPARE message.
@@ -316,9 +324,16 @@ Table of Contents
     - <value_1>...<value_n> are the [bytes] to use for bound variables in the
       prepared query.
     - <consistency> is the [consistency] level for the operation.
-    - <result_page_size> is an [int] controlling the desired page size of the
-      result (in CQL3 rows). A negative value disable paging of the result. See the
-      section on paging (Section 7) for more details.
+    - <flags> is a [byte] whose bits define the options for this query and
+      in particular influence what the remainder of the message contains.
+      A flag is set if the bit corresponding to its `mask` is set. Supported
+      flags are, given there mask:
+        0x01: Page size. In that case, <result_page_size> is an [int]
+              controlling the desired page size of the result (in CQL3 rows).
+              See the section on paging (Section 7) for more details.
+        0x02: Skip metadata. If present, the Result Set returned as a response
+              to that query (if any) will have the NO_METADATA flag (see
+              Section 4.2.5.2).
 
   Note that the consistency is ignored by some (prepared) queries (USE, CREATE,
   ALTER, TRUNCATE, ...).
@@ -461,7 +476,7 @@ Table of Contents
     <metadata><rows_count><rows_content>
   where:
     - <metadata> is composed of:
-        <flags><columns_count><global_table_spec>?<col_spec_1>...<col_spec_n>
+        <flags><columns_count>[<global_table_spec>?<col_spec_1>...<col_spec_n>]
       where:
         - <flags> is an [int]. The bits of <flags> provides information on the
           formatting of the remaining informations. A flag is set if the bit
@@ -476,6 +491,11 @@ Table of Contents
                       and NEXT cannot and should not be used. If no result
                       paging has been requested in the QUERY/EXECUTE/BATCH
                       message, this will never be set.
+            0x0003    No_metadata: if set, the <metadata> is only composed of
+                      these <flags> and the <column_count> but no other
+                      information (so no <global_table_spec> nor <col_spec_i>).
+                      This will only ever be the case if this was requested
+                      during the query (see QUERY and RESULT messages).
         - <columns_count> is an [int] representing the number of columns selected
           by the query this result is of. It defines the number of <col_spec_i>
           elements in and the number of element for each row in <rows_content>.
@@ -532,10 +552,21 @@ Table of Contents
 4.2.5.4. Prepared
 
   The result to a PREPARE message. The rest of the body of a Prepared result is:
-    <id><metadata>
+    <id><metadata><result_metadata>
   where:
     - <id> is [short bytes] representing the prepared query ID.
-    - <metadata> is defined exactly as for a Rows RESULT (See section 4.2.5.2).
+    - <metadata> is defined exactly as for a Rows RESULT (See section 4.2.5.2) and
+      is the specification for the variable bound in this prepare statement.
+    - <result_metadata> is defined exactly as <metadata> but correspond to the
+      metadata for the resultSet that execute this query will yield. Note that
+      <result_metadata> may be empty (have the No_metadata flag and 0 columns, See
+      section 4.2.5.2) and will be for any query that is not a Select. There is
+      in fact never a guarantee that this will non-empty so client should protect
+      themselves accordingly. The presence of this information is an
+      optimization that allows to later execute the statement that has been
+      prepared without requesting the metadata (Skip_metadata flag in EXECUTE).
+      Clients can safely discard this metadata if they do not want to take
+      advantage of that optimization.
 
   Note that prepared query ID return is global to the node on which the query
   has been prepared. It can be used on any connection to that node and this

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/cql3/QueryProcessor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/QueryProcessor.java b/src/java/org/apache/cassandra/cql3/QueryProcessor.java
index b2da29d..476ca88 100644
--- a/src/java/org/apache/cassandra/cql3/QueryProcessor.java
+++ b/src/java/org/apache/cassandra/cql3/QueryProcessor.java
@@ -224,7 +224,7 @@ public class QueryProcessor
                                        statementId,
                                        prepared.statement.getBoundsTerms()));
             preparedStatements.put(statementId, prepared.statement);
-            return new ResultMessage.Prepared(statementId, prepared.boundNames);
+            return new ResultMessage.Prepared(statementId, prepared);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/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 df892b7..8d6995f 100644
--- a/src/java/org/apache/cassandra/cql3/ResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/ResultSet.java
@@ -60,14 +60,14 @@ public class ResultSet
 
     public void addRow(List<ByteBuffer> row)
     {
-        assert row.size() == metadata.names.size();
+        assert row.size() == metadata.columnCount;
         rows.add(row);
     }
 
     public void addColumnValue(ByteBuffer value)
     {
-        if (rows.isEmpty() || lastRow().size() == metadata.names.size())
-            rows.add(new ArrayList<ByteBuffer>(metadata.names.size()));
+        if (rows.isEmpty() || lastRow().size() == metadata.columnCount)
+            rows.add(new ArrayList<ByteBuffer>(metadata.columnCount));
 
         lastRow().add(value);
     }
@@ -94,6 +94,7 @@ public class ResultSet
 
     public ResultSet makeCountResult(ColumnIdentifier alias)
     {
+        assert metadata.names != null;
         String ksName = metadata.names.get(0).ksName;
         String cfName = metadata.names.get(0).cfName;
         long count = rows.size();
@@ -111,6 +112,8 @@ public class ResultSet
 
     public CqlResult toThriftResult()
     {
+        assert metadata.names != null;
+
         String UTF8 = "UTF8Type";
         CqlMetadata schema = new CqlMetadata(new HashMap<ByteBuffer, String>(),
                 new HashMap<ByteBuffer, String>(),
@@ -187,7 +190,7 @@ public class ResultSet
             ResultSet rs = new ResultSet(m, new ArrayList<List<ByteBuffer>>(rowCount));
 
             // rows
-            int totalValues = rowCount * m.names.size();
+            int totalValues = rowCount * m.columnCount;
             for (int i = 0; i < totalValues; i++)
                 rs.addColumnValue(CBUtil.readValue(body));
 
@@ -196,7 +199,7 @@ public class ResultSet
 
         public ChannelBuffer encode(ResultSet rs, int version)
         {
-            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(2, 0, rs.metadata.names.size()
* rs.rows.size());
+            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(2, 0, rs.metadata.columnCount
* rs.rows.size());
             builder.add(Metadata.codec.encode(rs.metadata, version));
             builder.add(CBUtil.intToCB(rs.rows.size()));
 
@@ -214,8 +217,11 @@ public class ResultSet
     {
         public static final CBCodec<Metadata> codec = new Codec();
 
+        public static final Metadata empty = new Metadata(EnumSet.of(Flag.NO_METADATA), 0);
+
         public final EnumSet<Flag> flags;
         public final List<ColumnSpecification> names;
+        public final int columnCount;
 
         public Metadata(List<ColumnSpecification> names)
         {
@@ -228,10 +234,21 @@ public class ResultSet
         {
             this.flags = flags;
             this.names = names;
+            this.columnCount = names.size();
+        }
+
+        private Metadata(EnumSet<Flag> flags, int columnCount)
+        {
+            this.flags = flags;
+            this.names = null;
+            this.columnCount = columnCount;
         }
 
         private boolean allInSameCF()
         {
+            if (names == null)
+                return false;
+
             assert !names.isEmpty();
 
             Iterator<ColumnSpecification> iter = names.iterator();
@@ -254,11 +271,19 @@ public class ResultSet
         public String toString()
         {
             StringBuilder sb = new StringBuilder();
-            for (ColumnSpecification name : names)
+
+            if (names == null)
             {
-                sb.append("[").append(name.toString());
-                sb.append("(").append(name.ksName).append(", ").append(name.cfName).append(")");
-                sb.append(", ").append(name.type).append("]");
+                sb.append("[").append(columnCount).append("columns]");
+            }
+            else
+            {
+                for (ColumnSpecification name : names)
+                {
+                    sb.append("[").append(name.toString());
+                    sb.append("(").append(name.ksName).append(", ").append(name.cfName).append(")");
+                    sb.append(", ").append(name.type).append("]");
+                }
             }
             if (flags.contains(Flag.HAS_MORE_PAGES))
                 sb.append(" (to be continued)");
@@ -274,6 +299,9 @@ public class ResultSet
                 int columnCount = body.readInt();
 
                 EnumSet<Flag> flags = Flag.deserialize(iflags);
+                if (flags.contains(Flag.NO_METADATA))
+                    return new Metadata(flags, columnCount);
+
                 boolean globalTablesSpec = flags.contains(Flag.GLOBAL_TABLES_SPEC);
 
                 String globalKsName = null;
@@ -299,34 +327,39 @@ public class ResultSet
 
             public ChannelBuffer encode(Metadata m, int version)
             {
+                boolean noMetadata = m.flags.contains(Flag.NO_METADATA);
                 boolean globalTablesSpec = m.flags.contains(Flag.GLOBAL_TABLES_SPEC);
 
-                int stringCount = globalTablesSpec ? 2 + m.names.size() : 3* m.names.size();
-                CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(1 + m.names.size(),
stringCount, 0);
+                int stringCount = noMetadata ? 0 : (globalTablesSpec ? 2 + columnCount :
3 * columnCount);
+                CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(1 + (noMetadata :
0 ? columnCount), stringCount, 0);
 
                 ChannelBuffer header = ChannelBuffers.buffer(8);
 
-                assert version > 1 || !m.flags.contains(Flag.HAS_MORE_PAGES);
+                assert version > 1 || (!m.flags.contains(Flag.HAS_MORE_PAGES) &&
!noMetadata);
 
                 header.writeInt(Flag.serialize(m.flags));
-                header.writeInt(m.names.size());
+                header.writeInt(columnCount);
+
                 builder.add(header);
 
-                if (globalTablesSpec)
+                if (!noMetadata)
                 {
-                    builder.addString(m.names.get(0).ksName);
-                    builder.addString(m.names.get(0).cfName);
-                }
+                    if (globalTablesSpec)
+                    {
+                        builder.addString(m.names.get(0).ksName);
+                        builder.addString(m.names.get(0).cfName);
+                    }
 
-                for (ColumnSpecification name : m.names)
-                {
-                    if (!globalTablesSpec)
+                    for (ColumnSpecification name : m.names)
                     {
-                        builder.addString(name.ksName);
-                        builder.addString(name.cfName);
+                        if (!globalTablesSpec)
+                        {
+                            builder.addString(name.ksName);
+                            builder.addString(name.cfName);
+                        }
+                        builder.addString(name.toString());
+                        builder.add(DataType.codec.encodeOne(DataType.fromType(name.type)));
                     }
-                    builder.addString(name.toString());
-                    builder.add(DataType.codec.encodeOne(DataType.fromType(name.type)));
                 }
                 return builder.build();
             }
@@ -337,7 +370,8 @@ public class ResultSet
     {
         // The order of that enum matters!!
         GLOBAL_TABLES_SPEC,
-        HAS_MORE_PAGES;
+        HAS_MORE_PAGES,
+        NO_METADATA;
 
         public static EnumSet<Flag> deserialize(int flags)
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index b323a33..994c331 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -120,6 +120,11 @@ public class SelectStatement implements CQLStatement
         return new SelectStatement(cfDef, 0, defaultParameters, selection, null);
     }
 
+    public ResultSet.Metadata getResultMetadata()
+    {
+        return selection.getResultMetadata();
+    }
+
     public int getBoundsTerms()
     {
         return boundTerms;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/cql3/statements/Selection.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/Selection.java b/src/java/org/apache/cassandra/cql3/statements/Selection.java
index af0804a..600ee1b 100644
--- a/src/java/org/apache/cassandra/cql3/statements/Selection.java
+++ b/src/java/org/apache/cassandra/cql3/statements/Selection.java
@@ -49,6 +49,11 @@ public abstract class Selection
         this.collectTTLs = collectTTLs;
     }
 
+    public ResultSet.Metadata getResultMetadata()
+    {
+        return new ResultSet.Metadata(metadata);
+    }
+
     public static Selection wildcard(CFDefinition cfDef)
     {
         List<CFDefinition.Name> all = new ArrayList<CFDefinition.Name>();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/transport/Client.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/Client.java b/src/java/org/apache/cassandra/transport/Client.java
index 2f3f3bd..013de9f 100644
--- a/src/java/org/apache/cassandra/transport/Client.java
+++ b/src/java/org/apache/cassandra/transport/Client.java
@@ -126,7 +126,7 @@ public class Client extends SimpleClient
                     return null;
                 }
             }
-            return new QueryMessage(query, Collections.<ByteBuffer>emptyList(), ConsistencyLevel.ONE,
pageSize);
+            return new QueryMessage(query, ConsistencyLevel.ONE, Collections.<ByteBuffer>emptyList(),
pageSize);
         }
         else if (msgType.equals("NEXT"))
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/transport/SimpleClient.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/SimpleClient.java b/src/java/org/apache/cassandra/transport/SimpleClient.java
index 3a9c286..b9dd3f5 100644
--- a/src/java/org/apache/cassandra/transport/SimpleClient.java
+++ b/src/java/org/apache/cassandra/transport/SimpleClient.java
@@ -156,7 +156,7 @@ public class SimpleClient
 
     public ResultMessage execute(String query, List<ByteBuffer> values, ConsistencyLevel
consistencyLevel)
     {
-        Message.Response msg = execute(new QueryMessage(query, values, consistencyLevel,
-1));
+        Message.Response msg = execute(new QueryMessage(query, consistencyLevel, values,
-1));
         assert msg instanceof ResultMessage;
         return (ResultMessage)msg;
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
index 2d6b67b..8082e23 100644
--- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.transport.messages;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.UUID;
 
@@ -38,6 +39,33 @@ import org.apache.cassandra.utils.UUIDGen;
 
 public class ExecuteMessage extends Message.Request
 {
+    public static enum Flag
+    {
+        // The order of that enum matters!!
+        PAGE_SIZE,
+        SKIP_METADATA;
+
+        public static EnumSet<Flag> deserialize(int flags)
+        {
+            EnumSet<Flag> set = EnumSet.noneOf(Flag.class);
+            Flag[] values = Flag.values();
+            for (int n = 0; n < values.length; n++)
+            {
+                if ((flags & (1 << n)) != 0)
+                    set.add(values[n]);
+            }
+            return set;
+        }
+
+        public static int serialize(EnumSet<Flag> flags)
+        {
+            int i = 0;
+            for (Flag flag : flags)
+                i |= 1 << flag.ordinal();
+            return i;
+        }
+    }
+
     public static final Message.Codec<ExecuteMessage> codec = new Message.Codec<ExecuteMessage>()
     {
         public ExecuteMessage decode(ChannelBuffer body, int version)
@@ -52,9 +80,15 @@ public class ExecuteMessage extends Message.Request
             ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
 
             int resultPageSize = -1;
+            boolean skipMetadata = false;
             if (version >= 2)
-                resultPageSize = body.readInt();
-            return new ExecuteMessage(id, values, consistency, resultPageSize);
+            {
+                EnumSet<Flag> flags = Flag.deserialize((int)body.readByte());
+                if (flags.contains(Flag.PAGE_SIZE))
+                    resultPageSize = body.readInt();
+                skipMetadata = flags.contains(Flag.SKIP_METADATA);
+            }
+            return new ExecuteMessage(MD5Digest.wrap(id), values, consistency, resultPageSize,
skipMetadata);
         }
 
         public ChannelBuffer encode(ExecuteMessage msg, int version)
@@ -65,7 +99,23 @@ public class ExecuteMessage extends Message.Request
             //   - The values
             //   - options
             int vs = msg.values.size();
-            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(4, 0, vs);
+
+            EnumSet<Flag> flags = EnumSet.noneOf(Flag.class);
+            if (msg.resultPageSize >= 0)
+                flags.add(Flag.PAGE_SIZE);
+            if (msg.skipMetadata)
+                flags.add(Flag.SKIP_METADATA);
+
+            assert flags.isEmpty() || version >= 2;
+
+            int nbBuff = 3;
+            if (version >= 2)
+            {
+                nbBuff++; // the flags themselves
+                if (flags.contains(Flag.PAGE_SIZE))
+                    nbBuff++;
+            }
+            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(nbBuff, 0, vs);
             builder.add(CBUtil.bytesToCB(msg.statementId.bytes));
             builder.add(CBUtil.shortToCB(vs));
 
@@ -75,9 +125,12 @@ public class ExecuteMessage extends Message.Request
 
             builder.add(CBUtil.consistencyLevelToCB(msg.consistency));
 
-            assert msg.resultPageSize == -1 || version >= 2;
             if (version >= 2)
-                builder.add(CBUtil.intToCB(msg.resultPageSize));
+            {
+                builder.add(CBUtil.byteToCB((byte)Flag.serialize(flags)));
+                if (flags.contains(Flag.PAGE_SIZE))
+                    builder.add(CBUtil.intToCB(msg.resultPageSize));
+            }
             return builder.build();
         }
     };
@@ -86,19 +139,21 @@ public class ExecuteMessage extends Message.Request
     public final List<ByteBuffer> values;
     public final ConsistencyLevel consistency;
     public final int resultPageSize;
+    public final boolean skipMetadata;
 
     public ExecuteMessage(byte[] statementId, List<ByteBuffer> values, ConsistencyLevel
consistency, int resultPageSize)
     {
-        this(MD5Digest.wrap(statementId), values, consistency, resultPageSize);
+        this(MD5Digest.wrap(statementId), values, consistency, resultPageSize, false);
     }
 
-    public ExecuteMessage(MD5Digest statementId, List<ByteBuffer> values, ConsistencyLevel
consistency, int resultPageSize)
+    public ExecuteMessage(MD5Digest statementId, List<ByteBuffer> values, ConsistencyLevel
consistency, int resultPageSize, boolean skipMetadata)
     {
         super(Message.Type.EXECUTE);
         this.statementId = statementId;
         this.values = values;
         this.consistency = consistency;
         this.resultPageSize = resultPageSize;
+        this.skipMetadata = skipMetadata;
     }
 
     public ChannelBuffer encode()

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java
index 1ec556c..c3f9d4c 100644
--- a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java
@@ -20,6 +20,7 @@ package org.apache.cassandra.transport.messages;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.EnumSet;
 import java.util.List;
 import java.util.UUID;
 
@@ -39,6 +40,34 @@ import org.apache.cassandra.utils.UUIDGen;
  */
 public class QueryMessage extends Message.Request
 {
+    public static enum Flag
+    {
+        // The order of that enum matters!!
+        PAGE_SIZE,
+        VALUES,
+        SKIP_METADATA;
+
+        public static EnumSet<Flag> deserialize(int flags)
+        {
+            EnumSet<Flag> set = EnumSet.noneOf(Flag.class);
+            Flag[] values = Flag.values();
+            for (int n = 0; n < values.length; n++)
+            {
+                if ((flags & (1 << n)) != 0)
+                    set.add(values[n]);
+            }
+            return set;
+        }
+
+        public static int serialize(EnumSet<Flag> flags)
+        {
+            int i = 0;
+            for (Flag flag : flags)
+                i |= 1 << flag.ordinal();
+            return i;
+        }
+    }
+
     public static final Message.Codec<QueryMessage> codec = new Message.Codec<QueryMessage>()
     {
         public QueryMessage decode(ChannelBuffer body, int version)
@@ -48,19 +77,25 @@ public class QueryMessage extends Message.Request
 
             int resultPageSize = -1;
             List<ByteBuffer> values = Collections.emptyList();
-
+            boolean skipMetadata = false;
             if (version >= 2)
             {
-                resultPageSize = body.readInt();
-                if (body.readable())
+                EnumSet<Flag> flags = Flag.deserialize((int)body.readByte());
+
+                if (flags.contains(Flag.PAGE_SIZE))
+                    resultPageSize = body.readInt();
+
+                if (flags.contains(Flag.VALUES))
                 {
                     int paramCount = body.readUnsignedShort();
                     values = paramCount == 0 ? Collections.<ByteBuffer>emptyList()
: new ArrayList<ByteBuffer>(paramCount);
                     for (int i = 0; i < paramCount; i++)
                         values.add(CBUtil.readValue(body));
                 }
+
+                skipMetadata = flags.contains(Flag.SKIP_METADATA);
             }
-            return new QueryMessage(query, values, consistency, resultPageSize);
+            return new QueryMessage(query, consistency, values, resultPageSize, skipMetadata);
         }
 
         public ChannelBuffer encode(QueryMessage msg, int version)
@@ -72,15 +107,35 @@ public class QueryMessage extends Message.Request
             //   - Number of values
             //   - The values
             int vs = msg.values.size();
-            assert (msg.resultPageSize == -1 && vs == 0) || version >= 2 : "Version
1 of the protocol support neither a page size nor values";
 
-            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(2 + (version == 1 ? 0
: 1 + (vs > 0 ? 1 : 0)), 0, vs);
+            EnumSet<Flag> flags = EnumSet.noneOf(Flag.class);
+            if (msg.resultPageSize >= 0)
+                flags.add(Flag.PAGE_SIZE);
+            if (vs > 0)
+                flags.add(Flag.VALUES);
+            if (msg.skipMetadata)
+                flags.add(Flag.SKIP_METADATA);
+
+            assert flags.isEmpty() || version >= 2 : "Version 1 of the protocol supports
no option after the consistency level";
+
+            int nbBuff = 2;
+            if (version >= 2)
+            {
+                nbBuff++; // the flags themselves
+                if (flags.contains(Flag.PAGE_SIZE))
+                    nbBuff++;
+                if (flags.contains(Flag.VALUES))
+                    nbBuff++;
+            }
+            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(nbBuff, 0, vs);
             builder.add(CBUtil.longStringToCB(msg.query));
             builder.add(CBUtil.consistencyLevelToCB(msg.consistency));
             if (version >= 2)
             {
-                builder.add(CBUtil.intToCB(msg.resultPageSize));
-                if (vs > 0)
+                builder.add(CBUtil.byteToCB((byte)Flag.serialize(flags)));
+                if (flags.contains(Flag.PAGE_SIZE))
+                    builder.add(CBUtil.intToCB(msg.resultPageSize));
+                if (flags.contains(Flag.VALUES))
                 {
                     builder.add(CBUtil.shortToCB(vs));
                     for (ByteBuffer value : msg.values)
@@ -95,19 +150,26 @@ public class QueryMessage extends Message.Request
     public final ConsistencyLevel consistency;
     public final int resultPageSize;
     public final List<ByteBuffer> values;
+    public final boolean skipMetadata;
 
     public QueryMessage(String query, ConsistencyLevel consistency)
     {
-        this(query, Collections.<ByteBuffer>emptyList(), consistency, -1);
+        this(query, consistency, Collections.<ByteBuffer>emptyList(), -1);
+    }
+
+    public QueryMessage(String query, ConsistencyLevel consistency, List<ByteBuffer>
values, int resultPageSize)
+    {
+        this(query, consistency, values, resultPageSize, false);
     }
 
-    public QueryMessage(String query, List<ByteBuffer> values, ConsistencyLevel consistency,
int resultPageSize)
+    public QueryMessage(String query, ConsistencyLevel consistency, List<ByteBuffer>
values, int resultPageSize, boolean skipMetadata)
     {
         super(Type.QUERY);
         this.query = query;
-        this.resultPageSize = resultPageSize;
         this.consistency = consistency;
+        this.resultPageSize = resultPageSize;
         this.values = values;
+        this.skipMetadata = skipMetadata;
     }
 
     public ChannelBuffer encode()

http://git-wip-us.apache.org/repos/asf/cassandra/blob/dad2f11a/src/java/org/apache/cassandra/transport/messages/ResultMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/messages/ResultMessage.java b/src/java/org/apache/cassandra/transport/messages/ResultMessage.java
index 33975a4..9300f42 100644
--- a/src/java/org/apache/cassandra/transport/messages/ResultMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/ResultMessage.java
@@ -23,7 +23,10 @@ import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
 
 import org.apache.cassandra.cql3.ColumnSpecification;
+import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.ResultSet;
+import org.apache.cassandra.cql3.statements.SelectStatement;
+import org.apache.cassandra.cql3.statements.ParsedStatement;
 import org.apache.cassandra.transport.*;
 import org.apache.cassandra.thrift.CqlPreparedResult;
 import org.apache.cassandra.thrift.CqlResult;
@@ -240,7 +243,10 @@ public abstract class ResultMessage extends Message.Response
             public ResultMessage decode(ChannelBuffer body, int version)
             {
                 MD5Digest id = MD5Digest.wrap(CBUtil.readBytes(body));
-                return new Prepared(id, -1, ResultSet.Metadata.codec.decode(body, version));
+                ResultSet.Metadata metadata = ResultSet.Metadata.codec.decode(body, version);
+                ResultSet.Metadata resultMetadata = ResultSet.Metadata.codec.decode(body,
version);
+
+                return new Prepared(id, -1, metadata, resultMetadata);
             }
 
             public ChannelBuffer encode(ResultMessage msg, int version)
@@ -248,32 +254,46 @@ public abstract class ResultMessage extends Message.Response
                 assert msg instanceof Prepared;
                 Prepared prepared = (Prepared)msg;
                 assert prepared.statementId != null;
-                return ChannelBuffers.wrappedBuffer(CBUtil.bytesToCB(prepared.statementId.bytes),
ResultSet.Metadata.codec.encode(prepared.metadata, version));
+
+
+                return ChannelBuffers.wrappedBuffer(CBUtil.bytesToCB(prepared.statementId.bytes),
+                                                    ResultSet.Metadata.codec.encode(prepared.metadata,
version),
+                                                    ResultSet.Metadata.codec.encode(prepared.resultMetadata,
version));
             }
         };
 
         public final MD5Digest statementId;
         public final ResultSet.Metadata metadata;
+        public final ResultSet.Metadata resultMetadata;
 
         // statement id for CQL-over-thrift compatibility. The binary protocol ignore that.
         private final int thriftStatementId;
 
-        public Prepared(MD5Digest statementId, List<ColumnSpecification> names)
+        public Prepared(MD5Digest statementId, ParsedStatement.Prepared prepared)
         {
-            this(statementId, -1, new ResultSet.Metadata(names));
+            this(statementId, -1, new ResultSet.Metadata(prepared.boundNames), extractResultMetadata(prepared.statement));
         }
 
         public static Prepared forThrift(int statementId, List<ColumnSpecification>
names)
         {
-            return new Prepared(null, statementId, new ResultSet.Metadata(names));
+            return new Prepared(null, statementId, new ResultSet.Metadata(names), ResultSet.Metadata.empty);
         }
 
-        private Prepared(MD5Digest statementId, int thriftStatementId, ResultSet.Metadata
metadata)
+        private Prepared(MD5Digest statementId, int thriftStatementId, ResultSet.Metadata
metadata, ResultSet.Metadata resultMetadata)
         {
             super(Kind.PREPARED);
             this.statementId = statementId;
             this.thriftStatementId = thriftStatementId;
             this.metadata = metadata;
+            this.resultMetadata = resultMetadata;
+        }
+
+        private static ResultSet.Metadata extractResultMetadata(CQLStatement statement)
+        {
+            if (!(statement instanceof SelectStatement))
+                return ResultSet.Metadata.empty;
+
+            return ((SelectStatement)statement).getResultMetadata();
         }
 
         protected ChannelBuffer encodeBody()


Mime
View raw message