add support for bind variables to non-prepared statements.

patch by marcuse, reviewed by pcmanus for CASSANDRA-5349

doc/native_protocol_v2.spec

@@ -263,8 +263,13 @@ Table of Contents
 
 4.1.4. QUERY
 
-  Performs a CQL query. The body of the message consists of a CQL query as a [long
-  string] followed by the [consistency] for the operation.
+  Performs a CQL query. The body of the message must be:
+    <query><consistency>[<n><value_1>...<value_n>]
+  where:
+    - <query> the query, [long string].
+    - <consistency> is the [consistency] level for the operation.
+    - 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, ...).
@@ -638,3 +643,10 @@ Table of Contents
               executed if the provide prepared statement ID is not known by
               this host. The rest of the ERROR message body will be [short
               bytes] representing the unknown ID.
+
+8. Changes from v1
+  * Protocol is versioned to allow old client connects to a newer server, if a newer
+    client connects to an older server, it needs to check if it gets a
+    ProtocolException on connection and try connecting with a lower version.
+  * A query can now have bind variables even though the statement is not
+    prepared. (see Section 4.1.4)

src/java/org/apache/cassandra/cql3/QueryProcessor.java

@@ -133,12 +133,18 @@ private static ResultMessage processStatement(CQLStatement statement, Consistenc
 
     public static ResultMessage process(String queryString, ConsistencyLevel cl, QueryState queryState)
     throws RequestExecutionException, RequestValidationException
+    {
+        return process(queryString, Collections.<ByteBuffer>emptyList(), cl, queryState);
+    }
+
+    public static ResultMessage process(String queryString, List<ByteBuffer> variables, ConsistencyLevel cl, QueryState queryState)
+    throws RequestExecutionException, RequestValidationException
     {
         logger.trace("CQL QUERY: {}", queryString);
         CQLStatement prepared = getStatement(queryString, queryState.getClientState()).statement;
-        if (prepared.getBoundsTerms() > 0)
-            throw new InvalidRequestException("Cannot execute query with bind variables");
-        return processStatement(prepared, cl, queryState, Collections.<ByteBuffer>emptyList());
+        if (prepared.getBoundsTerms() != variables.size())
+            throw new InvalidRequestException("Invalid amount of bind variables");
+        return processStatement(prepared, cl, queryState, variables);
     }
 
     public static UntypedResultSet process(String query, ConsistencyLevel cl) throws RequestExecutionException

src/java/org/apache/cassandra/transport/SimpleClient.java

@@ -154,6 +154,13 @@ public ResultMessage execute(String query, ConsistencyLevel consistency)
         return (ResultMessage)msg;
     }
 
+    public ResultMessage execute(String query, List<ByteBuffer> values, ConsistencyLevel consistencyLevel)
+    {
+        Message.Response msg = execute(new QueryMessage(query, values, consistencyLevel));
+        assert msg instanceof ResultMessage;
+        return (ResultMessage)msg;
+    }
+
     public ResultMessage.Prepared prepare(String query)
     {
         Message.Response msg = execute(new PrepareMessage(query));

src/java/org/apache/cassandra/transport/messages/QueryMessage.java

@@ -17,8 +17,11 @@
  */
 package org.apache.cassandra.transport.messages;
 
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.UUID;
-
 import com.google.common.collect.ImmutableMap;
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
@@ -42,24 +45,53 @@ public QueryMessage decode(ChannelBuffer body, int version)
         {
             String query = CBUtil.readLongString(body);
             ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
-            return new QueryMessage(query, consistency);
+            List<ByteBuffer> values = new ArrayList<ByteBuffer>();
+            if (body.readable())
+            {
+                int paramCount = body.readUnsignedShort();
+                for (int i = 0; i < paramCount; i++)
+                     values.add(CBUtil.readValue(body));
+            }
+            return new QueryMessage(query, values, consistency);
         }
 
         public ChannelBuffer encode(QueryMessage msg)
         {
-
-            return ChannelBuffers.wrappedBuffer(CBUtil.longStringToCB(msg.query), CBUtil.consistencyLevelToCB(msg.consistency));
+            // We have:
+            //   - query
+            //   - options
+            //     * optional:
+            //   - Number of values
+            //   - The values
+            int vs = msg.values.size();
+            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(3, 0, vs);
+            builder.add(CBUtil.longStringToCB(msg.query));
+            builder.add(CBUtil.consistencyLevelToCB(msg.consistency));
+            if (vs > 0 && msg.getVersion() > 1)
+            {
+                builder.add(CBUtil.shortToCB(vs));
+                for (ByteBuffer value : msg.values)
+                    builder.addValue(value);
+            }
+            return builder.build();
         }
     };
 
     public final String query;
     public final ConsistencyLevel consistency;
+    public final List<ByteBuffer> values;
 
     public QueryMessage(String query, ConsistencyLevel consistency)
     {
-        super(Message.Type.QUERY);
+        this(query, Collections.<ByteBuffer>emptyList(), consistency);
+    }
+
+    public QueryMessage(String query, List<ByteBuffer> values, ConsistencyLevel consistency)
+    {
+        super(Type.QUERY);
         this.query = query;
         this.consistency = consistency;
+        this.values = values;
     }
 
     public ChannelBuffer encode()
@@ -84,7 +116,7 @@ public Message.Response execute(QueryState state)
                 Tracing.instance().begin("Execute CQL3 query", ImmutableMap.of("query", query));
             }
 
-            Message.Response response = QueryProcessor.process(query, consistency, state);
+            Message.Response response = QueryProcessor.process(query, values, consistency, state);
 
             if (tracingId != null)
                 response.setTracingId(tracingId);