cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From slebre...@apache.org
Subject [1/2] DC-local CAS
Date Fri, 26 Jul 2013 15:29:35 GMT
Updated Branches:
  refs/heads/trunk afe4d555b -> bfd73beaf


http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/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 2f9627b..8b9f7df 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -138,8 +138,10 @@ public class SelectStatement implements CQLStatement
         // Nothing to do, all validation has been done by RawStatement.prepare()
     }
 
-    public ResultMessage.Rows execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer>
variables, int pageSize, PagingState pagingState) throws RequestExecutionException, RequestValidationException
+    public ResultMessage.Rows execute(QueryState state, QueryOptions options) throws RequestExecutionException,
RequestValidationException
     {
+        ConsistencyLevel cl = options.getConsistency();
+        List<ByteBuffer> variables = options.getValues();
         if (cl == null)
             throw new InvalidRequestException("Invalid empty consistency level");
 
@@ -158,18 +160,19 @@ public class SelectStatement implements CQLStatement
             command = commands == null ? null : new Pageable.ReadCommands(commands);
         }
 
+        int pageSize = options.getPageSize();
         // A count query will never be paged for the user, but we always page it internally
to avoid OOM.
         // If we user provided a pageSize we'll use that to page internally (because why
not), otherwise we use our default
-        if (parameters.isCount && pageSize < 0)
+        if (parameters.isCount && pageSize <= 0)
             pageSize = DEFAULT_COUNT_PAGE_SIZE;
 
-        if (pageSize < 0 || command == null || !QueryPagers.mayNeedPaging(command, pageSize))
+        if (pageSize <= 0 || command == null || !QueryPagers.mayNeedPaging(command, pageSize))
         {
             return execute(command, cl, variables, limit, now);
         }
         else
         {
-            QueryPager pager = QueryPagers.pager(command, cl, pagingState);
+            QueryPager pager = QueryPagers.pager(command, cl, options.getPagingState());
             if (parameters.isCount)
                 return pageCountQuery(pager, variables, pageSize, now);
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java b/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java
index c66608b..318995e 100644
--- a/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/TruncateStatement.java
@@ -55,7 +55,7 @@ public class TruncateStatement extends CFStatement implements CQLStatement
         ThriftValidation.validateColumnFamily(keyspace(), columnFamily());
     }
 
-    public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer>
variables, int pageSize, PagingState pagingState) throws InvalidRequestException, TruncateException
+    public ResultMessage execute(QueryState state, QueryOptions options) throws InvalidRequestException,
TruncateException
     {
         try
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/src/java/org/apache/cassandra/cql3/statements/UseStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/UseStatement.java b/src/java/org/apache/cassandra/cql3/statements/UseStatement.java
index c2e3c34..ab42fed 100644
--- a/src/java/org/apache/cassandra/cql3/statements/UseStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UseStatement.java
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
 import java.util.List;
 
 import org.apache.cassandra.cql3.CQLStatement;
+import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.exceptions.UnauthorizedException;
@@ -52,7 +53,7 @@ public class UseStatement extends ParsedStatement implements CQLStatement
     {
     }
 
-    public ResultMessage execute(ConsistencyLevel cl, QueryState state, List<ByteBuffer>
variables, int pageSize, PagingState pagingState) throws InvalidRequestException
+    public ResultMessage execute(QueryState state, QueryOptions options) throws InvalidRequestException
     {
         state.getClientState().setKeyspace(keyspace);
         return new ResultMessage.SetKeyspace(keyspace);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/src/java/org/apache/cassandra/db/ConsistencyLevel.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
index d642d08..956ab58 100644
--- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java
+++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
@@ -48,7 +48,8 @@ public enum ConsistencyLevel
     ALL         (5),
     LOCAL_QUORUM(6),
     EACH_QUORUM (7),
-    SERIAL      (8);
+    SERIAL      (8),
+    LOCAL_SERIAL(9);
 
     private static final Logger logger = LoggerFactory.getLogger(ConsistencyLevel.class);
 
@@ -277,11 +278,13 @@ public enum ConsistencyLevel
                 requireNetworkTopologyStrategy(keyspaceName);
                 break;
             case SERIAL:
+            case LOCAL_SERIAL:
                 throw new InvalidRequestException("You must use conditional updates for serializable
writes");
         }
     }
 
-    public void validateForCas(String keyspaceName) throws InvalidRequestException
+    // This is the same than validateForWrite really, but we include a slightly different
error message for SERIAL/LOCAL_SERIAL
+    public void validateForCasCommit(String keyspaceName) throws InvalidRequestException
     {
         switch (this)
         {
@@ -289,11 +292,23 @@ public enum ConsistencyLevel
             case EACH_QUORUM:
                 requireNetworkTopologyStrategy(keyspaceName);
                 break;
-            case ANY:
-                throw new InvalidRequestException("ANY is not supported with CAS. Use SERIAL
if you mean, make sure it is accepted but I don't care how many replicas commit it for non-SERIAL
reads");
+            case SERIAL:
+            case LOCAL_SERIAL:
+                throw new InvalidRequestException(this + " is not supported as conditional
update commit consistency. Use ANY if you mean \"make sure it is accepted but I don't care
how many replicas commit it for non-SERIAL reads\"");
         }
     }
 
+    public void validateForCas() throws InvalidRequestException
+    {
+        if (!isSerialConsistency())
+            throw new InvalidRequestException("Invalid consistency for conditional update.
Must be one of SERIAL or LOCAL_SERIAL");
+    }
+
+    public boolean isSerialConsistency()
+    {
+        return this == SERIAL || this == LOCAL_SERIAL;
+    }
+
     public void validateCounterForWrite(CFMetaData metadata) throws InvalidRequestException
     {
         if (this == ConsistencyLevel.ANY)
@@ -304,7 +319,7 @@ public enum ConsistencyLevel
         {
             throw new InvalidRequestException("cannot achieve CL > CL.ONE without replicate_on_write
on columnfamily " + metadata.cfName);
         }
-        else if (this == ConsistencyLevel.SERIAL)
+        else if (isSerialConsistency())
         {
             throw new InvalidRequestException("Counter operations are inherently non-serializable");
         }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 443d6b8..9b7da6d 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -29,6 +29,7 @@ import javax.management.MBeanServer;
 import javax.management.ObjectName;
 
 import com.google.common.base.Function;
+import com.google.common.base.Predicate;
 import com.google.common.collect.*;
 import com.google.common.util.concurrent.Uninterruptibles;
 import org.apache.commons.lang.StringUtils;
@@ -195,16 +196,25 @@ public class StorageProxy implements StorageProxyMBean
      * the first live column of the row).
      * @param expected the expected column values. This can be null to check for existence
(see {@code prefix}).
      * @param updates the value to insert if {@code expected matches the current values}.
-     * @param consistencyLevel the consistency for the operation.
+     * @param consistencyForPaxos the consistency for the paxos prepare and propose round.
This can only be either SERIAL or LOCAL_SERIAL.
+     * @param consistencyForCommit the consistency for write done during the commit phase.
This can be anything, except SERIAL or LOCAL_SERIAL.
      *
      * @return null if the operation succeeds in updating the row, or the current values
for the columns contained in
      * expected (since, if the CAS doesn't succeed, it means the current value do not match
the one in expected). If
      * expected == null and the CAS is unsuccessfull, the first live column of the CF is
returned.
      */
-    public static ColumnFamily cas(String keyspaceName, String cfName, ByteBuffer key, ColumnNameBuilder
prefix, ColumnFamily expected, ColumnFamily updates, ConsistencyLevel consistencyLevel)
+    public static ColumnFamily cas(String keyspaceName,
+                                   String cfName,
+                                   ByteBuffer key,
+                                   ColumnNameBuilder prefix,
+                                   ColumnFamily expected,
+                                   ColumnFamily updates,
+                                   ConsistencyLevel consistencyForPaxos,
+                                   ConsistencyLevel consistencyForCommit)
     throws UnavailableException, IsBootstrappingException, ReadTimeoutException, WriteTimeoutException,
InvalidRequestException
     {
-        consistencyLevel.validateForCas(keyspaceName);
+        consistencyForPaxos.validateForCas();
+        consistencyForCommit.validateForCasCommit(keyspaceName);
 
         CFMetaData metadata = Schema.instance.getCFMetaData(keyspaceName, cfName);
 
@@ -213,11 +223,11 @@ public class StorageProxy implements StorageProxyMBean
         while (System.nanoTime() - start < timeout)
         {
             // for simplicity, we'll do a single liveness check at the start of each attempt
-            Pair<List<InetAddress>, Integer> p = getPaxosParticipants(keyspaceName,
key);
+            Pair<List<InetAddress>, Integer> p = getPaxosParticipants(keyspaceName,
key, consistencyForPaxos);
             List<InetAddress> liveEndpoints = p.left;
             int requiredParticipants = p.right;
 
-            UUID ballot = beginAndRepairPaxos(start, key, metadata, liveEndpoints, requiredParticipants);
+            UUID ballot = beginAndRepairPaxos(start, key, metadata, liveEndpoints, requiredParticipants,
consistencyForPaxos);
 
             // read the current value and compare with expected
             Tracing.trace("Reading existing values for CAS precondition");
@@ -250,10 +260,10 @@ public class StorageProxy implements StorageProxyMBean
             Tracing.trace("CAS precondition is met; proposing client-requested updates for
{}", ballot);
             if (proposePaxos(proposal, liveEndpoints, requiredParticipants))
             {
-                if (consistencyLevel == ConsistencyLevel.SERIAL)
+                if (consistencyForCommit == ConsistencyLevel.ANY)
                     sendCommit(proposal, liveEndpoints);
                 else
-                    commitPaxos(proposal, consistencyLevel);
+                    commitPaxos(proposal, consistencyForCommit);
                 Tracing.trace("CAS successful");
                 return null;
             }
@@ -263,7 +273,7 @@ public class StorageProxy implements StorageProxyMBean
             // continue to retry
         }
 
-        throw new WriteTimeoutException(WriteType.CAS, ConsistencyLevel.SERIAL, -1, -1);
+        throw new WriteTimeoutException(WriteType.CAS, consistencyForPaxos, -1, -1);
     }
 
     private static boolean hasLiveColumns(ColumnFamily cf, long now)
@@ -301,15 +311,35 @@ public class StorageProxy implements StorageProxyMBean
         return true;
     }
 
-    private static Pair<List<InetAddress>, Integer> getPaxosParticipants(String
keyspaceName, ByteBuffer key) throws UnavailableException
+    private static Predicate<InetAddress> sameDCPredicateFor(final String dc)
+    {
+        final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
+        return new Predicate<InetAddress>()
+        {
+            public boolean apply(InetAddress host)
+            {
+                return dc.equals(snitch.getDatacenter(host));
+            }
+        };
+    }
+
+    private static Pair<List<InetAddress>, Integer> getPaxosParticipants(String
keyspaceName, ByteBuffer key, ConsistencyLevel consistencyForPaxos) throws UnavailableException
     {
         Token tk = StorageService.getPartitioner().getToken(key);
         List<InetAddress> naturalEndpoints = StorageService.instance.getNaturalEndpoints(keyspaceName,
tk);
         Collection<InetAddress> pendingEndpoints = StorageService.instance.getTokenMetadata().pendingEndpointsFor(tk,
keyspaceName);
+        if (consistencyForPaxos == ConsistencyLevel.LOCAL_SERIAL)
+        {
+            // Restrict naturalEndpoints and pendingEndpoints to node in the local DC only
+            String localDc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(FBUtilities.getBroadcastAddress());
+            Predicate<InetAddress> isLocalDc = sameDCPredicateFor(localDc);
+            naturalEndpoints = ImmutableList.copyOf(Iterables.filter(naturalEndpoints, isLocalDc));
+            pendingEndpoints = ImmutableList.copyOf(Iterables.filter(pendingEndpoints, isLocalDc));
+        }
         int requiredParticipants = pendingEndpoints.size() + 1 + naturalEndpoints.size()
/ 2; // See CASSANDRA-833
         List<InetAddress> liveEndpoints = ImmutableList.copyOf(Iterables.filter(Iterables.concat(naturalEndpoints,
pendingEndpoints), IAsyncCallback.isAlive));
         if (liveEndpoints.size() < requiredParticipants)
-            throw new UnavailableException(ConsistencyLevel.SERIAL, requiredParticipants,
liveEndpoints.size());
+            throw new UnavailableException(consistencyForPaxos, requiredParticipants, liveEndpoints.size());
         return Pair.create(liveEndpoints, requiredParticipants);
     }
 
@@ -319,7 +349,7 @@ public class StorageProxy implements StorageProxyMBean
      * @return the Paxos ballot promised by the replicas if no in-progress requests were
seen and a quorum of
      * nodes have seen the mostRecentCommit.  Otherwise, return null.
      */
-    private static UUID beginAndRepairPaxos(long start, ByteBuffer key, CFMetaData metadata,
List<InetAddress> liveEndpoints, int requiredParticipants)
+    private static UUID beginAndRepairPaxos(long start, ByteBuffer key, CFMetaData metadata,
List<InetAddress> liveEndpoints, int requiredParticipants, ConsistencyLevel consistencyForPaxos)
     throws WriteTimeoutException
     {
         long timeout = TimeUnit.MILLISECONDS.toNanos(DatabaseDescriptor.getCasContentionTimeout());
@@ -376,7 +406,7 @@ public class StorageProxy implements StorageProxyMBean
             return ballot;
         }
 
-        throw new WriteTimeoutException(WriteType.CAS, ConsistencyLevel.SERIAL, -1, -1);
+        throw new WriteTimeoutException(WriteType.CAS, consistencyForPaxos, -1, -1);
     }
 
     /**
@@ -1079,26 +1109,26 @@ public class StorageProxy implements StorageProxyMBean
         List<Row> rows = null;
         try
         {
-            if (consistency_level == ConsistencyLevel.SERIAL)
+            if (consistency_level.isSerialConsistency())
             {
                 // make sure any in-progress paxos writes are done (i.e., committed to a
majority of replicas), before performing a quorum read
                 if (commands.size() > 1)
-                    throw new InvalidRequestException("SERIAL consistency may only be requested
for one row at a time");
+                    throw new InvalidRequestException("SERIAL/LOCAL_SERIAL consistency may
only be requested for one row at a time");
                 ReadCommand command = commands.get(0);
 
                 CFMetaData metadata = Schema.instance.getCFMetaData(command.ksName, command.cfName);
-                Pair<List<InetAddress>, Integer> p = getPaxosParticipants(command.ksName,
command.key);
+                Pair<List<InetAddress>, Integer> p = getPaxosParticipants(command.ksName,
command.key, consistency_level);
                 List<InetAddress> liveEndpoints = p.left;
                 int requiredParticipants = p.right;
 
                 // does the work of applying in-progress writes; throws UAE or timeout if
it can't
                 try
                 {
-                    beginAndRepairPaxos(start, command.key, metadata, liveEndpoints, requiredParticipants);
+                    beginAndRepairPaxos(start, command.key, metadata, liveEndpoints, requiredParticipants,
consistency_level);
                 }
                 catch (WriteTimeoutException e)
                 {
-                    throw new ReadTimeoutException(ConsistencyLevel.SERIAL, -1, -1, false);
+                    throw new ReadTimeoutException(consistency_level, -1, -1, false);
                 }
 
                 rows = fetchRows(commands, ConsistencyLevel.QUORUM);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/src/java/org/apache/cassandra/thrift/CassandraServer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java
index 15ab9d0..e9473d9 100644
--- a/src/java/org/apache/cassandra/thrift/CassandraServer.java
+++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java
@@ -43,6 +43,7 @@ import org.apache.cassandra.config.KSMetaData;
 import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.cql.CQLStatement;
 import org.apache.cassandra.cql.QueryProcessor;
+import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.context.CounterContext;
 import org.apache.cassandra.db.filter.IDiskAtomFilter;
@@ -691,7 +692,12 @@ public class CassandraServer implements Cassandra.Iface
         }
     }
 
-    public CASResult cas(ByteBuffer key, String column_family, List<Column> expected,
List<Column> updates, ConsistencyLevel consistency_level)
+    public CASResult cas(ByteBuffer key,
+                         String column_family,
+                         List<Column> expected,
+                         List<Column> updates,
+                         ConsistencyLevel serial_consistency_level,
+                         ConsistencyLevel commit_consistency_level)
     throws InvalidRequestException, UnavailableException, TimedOutException
     {
         if (startSessionIfRequested())
@@ -747,7 +753,14 @@ public class CassandraServer implements Cassandra.Iface
             }
 
             schedule(DatabaseDescriptor.getWriteRpcTimeout());
-            ColumnFamily result = StorageProxy.cas(cState.getKeyspace(), column_family, key,
null, cfExpected, cfUpdates, ThriftConversion.fromThrift(consistency_level));
+            ColumnFamily result = StorageProxy.cas(cState.getKeyspace(),
+                                                   column_family,
+                                                   key,
+                                                   null,
+                                                   cfExpected,
+                                                   cfUpdates,
+                                                   ThriftConversion.fromThrift(serial_consistency_level),
+                                                   ThriftConversion.fromThrift(commit_consistency_level));
             return result == null
                  ? new CASResult(true)
                  : new CASResult(false).setCurrent_values(thriftifyColumnsAsColumns(result.getSortedColumns(),
System.currentTimeMillis()));
@@ -2024,11 +2037,8 @@ public class CassandraServer implements Cassandra.Iface
             logger.trace("Retrieved prepared statement #{} with {} bind markers", itemId,
statement.getBoundsTerms());
 
             return org.apache.cassandra.cql3.QueryProcessor.processPrepared(statement,
-                                                                            ThriftConversion.fromThrift(cLevel),
                                                                             cState.getQueryState(),
-                                                                            bindVariables,
-                                                                            -1,
-                                                                            null).toThriftResult();
+                                                                            new QueryOptions(ThriftConversion.fromThrift(cLevel),
bindVariables)).toThriftResult();
         }
         catch (RequestExecutionException e)
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/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 12c46dd..f0b700c 100644
--- a/src/java/org/apache/cassandra/transport/Client.java
+++ b/src/java/org/apache/cassandra/transport/Client.java
@@ -32,11 +32,13 @@ import java.util.Map;
 import com.google.common.base.Splitter;
 
 import org.apache.cassandra.auth.IAuthenticator;
+import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.transport.messages.*;
 import org.apache.cassandra.utils.Hex;
+import org.apache.cassandra.utils.MD5Digest;
 
 import static org.apache.cassandra.config.EncryptionOptions.ClientEncryptionOptions;
 
@@ -126,7 +128,7 @@ public class Client extends SimpleClient
                     return null;
                 }
             }
-            return new QueryMessage(query, ConsistencyLevel.ONE, Collections.<ByteBuffer>emptyList(),
pageSize);
+            return new QueryMessage(query, new QueryOptions(ConsistencyLevel.ONE, Collections.<ByteBuffer>emptyList(),
false, pageSize, null, null));
         }
         else if (msgType.equals("PREPARE"))
         {
@@ -154,7 +156,7 @@ public class Client extends SimpleClient
                     }
                     values.add(bb);
                 }
-                return new ExecuteMessage(id, values, ConsistencyLevel.ONE, -1);
+                return new ExecuteMessage(MD5Digest.wrap(id), new QueryOptions(ConsistencyLevel.ONE,
values));
             }
             catch (Exception e)
             {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/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 df4f811..cfe1bab 100644
--- a/src/java/org/apache/cassandra/transport/SimpleClient.java
+++ b/src/java/org/apache/cassandra/transport/SimpleClient.java
@@ -32,6 +32,7 @@ import javax.net.ssl.SSLEngine;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.security.SSLFactory;
 import org.apache.cassandra.transport.messages.CredentialsMessage;
@@ -41,6 +42,7 @@ import org.apache.cassandra.transport.messages.PrepareMessage;
 import org.apache.cassandra.transport.messages.QueryMessage;
 import org.apache.cassandra.transport.messages.ResultMessage;
 import org.apache.cassandra.transport.messages.StartupMessage;
+import org.apache.cassandra.utils.MD5Digest;
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelFuture;
@@ -158,7 +160,7 @@ public class SimpleClient
 
     public ResultMessage execute(String query, List<ByteBuffer> values, ConsistencyLevel
consistencyLevel)
     {
-        Message.Response msg = execute(new QueryMessage(query, consistencyLevel, values,
-1));
+        Message.Response msg = execute(new QueryMessage(query, new QueryOptions(consistencyLevel,
values)));
         assert msg instanceof ResultMessage;
         return (ResultMessage)msg;
     }
@@ -172,7 +174,7 @@ public class SimpleClient
 
     public ResultMessage executePrepared(byte[] statementId, List<ByteBuffer> values,
ConsistencyLevel consistency)
     {
-        Message.Response msg = execute(new ExecuteMessage(statementId, values, consistency,
-1));
+        Message.Response msg = execute(new ExecuteMessage(MD5Digest.wrap(statementId), new
QueryOptions(consistency, values)));
         assert msg instanceof ResultMessage;
         return (ResultMessage)msg;
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/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 f83df9d..c297426 100644
--- a/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/ExecuteMessage.java
@@ -26,9 +26,11 @@ import java.util.UUID;
 
 import com.google.common.collect.ImmutableMap;
 import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
 
 import org.apache.cassandra.cql3.CQLStatement;
 import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.exceptions.PreparedQueryNotFoundException;
 import org.apache.cassandra.service.QueryState;
@@ -40,131 +42,65 @@ 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,
-        PAGING_STATE;
-
-        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)
         {
             byte[] id = CBUtil.readBytes(body);
 
-            int count = body.readUnsignedShort();
-            List<ByteBuffer> values = new ArrayList<ByteBuffer>(count);
-            for (int i = 0; i < count; i++)
-                values.add(CBUtil.readValue(body));
-
-            ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
+            if (version == 1)
+            {
+                int count = body.readUnsignedShort();
+                List<ByteBuffer> values = new ArrayList<ByteBuffer>(count);
+                for (int i = 0; i < count; i++)
+                    values.add(CBUtil.readValue(body));
 
-            int resultPageSize = -1;
-            boolean skipMetadata = false;
-            PagingState pagingState = null;
-            if (version >= 2)
+                ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
+                return new ExecuteMessage(id, values, consistency);
+            }
+            else
             {
-                EnumSet<Flag> flags = Flag.deserialize((int)body.readByte());
-                if (flags.contains(Flag.PAGE_SIZE))
-                    resultPageSize = body.readInt();
-                skipMetadata = flags.contains(Flag.SKIP_METADATA);
-                if (flags.contains(Flag.PAGING_STATE))
-                    pagingState = PagingState.deserialize(CBUtil.readValue(body));
+                return new ExecuteMessage(MD5Digest.wrap(id), QueryOptions.codec.decode(body,
version));
             }
-            return new ExecuteMessage(MD5Digest.wrap(id), values, consistency, resultPageSize,
skipMetadata, pagingState);
         }
 
         public ChannelBuffer encode(ExecuteMessage msg, int version)
         {
-            // We have:
-            //   - statementId
-            //   - Number of values
-            //   - The values
-            //   - options
-            int vs = msg.values.size();
-
-            EnumSet<Flag> flags = EnumSet.noneOf(Flag.class);
-            if (msg.resultPageSize >= 0)
-                flags.add(Flag.PAGE_SIZE);
-            if (msg.skipMetadata)
-                flags.add(Flag.SKIP_METADATA);
-            if (msg.pagingState != null)
-                flags.add(Flag.PAGING_STATE);
-
-            assert flags.isEmpty() || version >= 2;
-
-            int nbBuff = 3;
-            if (version >= 2)
+            ChannelBuffer idBuffer = CBUtil.bytesToCB(msg.statementId.bytes);
+            ChannelBuffer optBuffer;
+            if (version == 1)
             {
-                nbBuff++; // the flags themselves
-                if (flags.contains(Flag.PAGE_SIZE))
-                    nbBuff++;
-            }
-            CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(nbBuff, 0, vs + (flags.contains(Flag.PAGING_STATE)
? 1 : 0));
-            builder.add(CBUtil.bytesToCB(msg.statementId.bytes));
-            builder.add(CBUtil.shortToCB(vs));
+                CBUtil.BufferBuilder builder = new CBUtil.BufferBuilder(2, 0, msg.options.getValues().size());
+                builder.add(CBUtil.shortToCB(msg.options.getValues().size()));
 
-            // Values
-            for (ByteBuffer value : msg.values)
-                builder.addValue(value);
+                // Values
+                for (ByteBuffer value : msg.options.getValues())
+                    builder.addValue(value);
 
-            builder.add(CBUtil.consistencyLevelToCB(msg.consistency));
-
-            if (version >= 2)
+                builder.add(CBUtil.consistencyLevelToCB(msg.options.getConsistency()));
+                optBuffer = builder.build();
+            }
+            else
             {
-                builder.add(CBUtil.byteToCB((byte)Flag.serialize(flags)));
-                if (flags.contains(Flag.PAGE_SIZE))
-                    builder.add(CBUtil.intToCB(msg.resultPageSize));
-                if (flags.contains(Flag.PAGING_STATE))
-                    builder.addValue(msg.pagingState == null ? null : msg.pagingState.serialize());
+                optBuffer = QueryOptions.codec.encode(msg.options, version);
             }
-            return builder.build();
+            return ChannelBuffers.wrappedBuffer(idBuffer, optBuffer);
         }
     };
 
     public final MD5Digest statementId;
-    public final List<ByteBuffer> values;
-    public final ConsistencyLevel consistency;
-    public final int resultPageSize;
-    public final boolean skipMetadata;
-    public final PagingState pagingState;
+    public final QueryOptions options;
 
-    public ExecuteMessage(byte[] statementId, List<ByteBuffer> values, ConsistencyLevel
consistency, int resultPageSize)
+    public ExecuteMessage(byte[] statementId, List<ByteBuffer> values, ConsistencyLevel
consistency)
     {
-        this(MD5Digest.wrap(statementId), values, consistency, resultPageSize, false, null);
+        this(MD5Digest.wrap(statementId), new QueryOptions(consistency, values));
     }
 
-    public ExecuteMessage(MD5Digest statementId, List<ByteBuffer> values, ConsistencyLevel
consistency, int resultPageSize, boolean skipMetadata, PagingState pagingState)
+    public ExecuteMessage(MD5Digest statementId, QueryOptions options)
     {
         super(Message.Type.EXECUTE);
         this.statementId = statementId;
-        this.values = values;
-        this.consistency = consistency;
-        this.resultPageSize = resultPageSize;
-        this.skipMetadata = skipMetadata;
-        this.pagingState = pagingState;
+        this.options = options;
     }
 
     public ChannelBuffer encode(int version)
@@ -181,7 +117,7 @@ public class ExecuteMessage extends Message.Request
             if (statement == null)
                 throw new PreparedQueryNotFoundException(statementId);
 
-            if (resultPageSize == 0)
+            if (options.getPageSize() == 0)
                 throw new ProtocolException("The page size cannot be 0");
 
             UUID tracingId = null;
@@ -196,15 +132,15 @@ public class ExecuteMessage extends Message.Request
                 state.createTracingSession();
 
                 ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
-                if (resultPageSize > 0)
-                    builder.put("page_size", Integer.toString(resultPageSize));
+                if (options.getPageSize() > 0)
+                    builder.put("page_size", Integer.toString(options.getPageSize()));
 
                 // TODO we don't have [typed] access to CQL bind variables here.  CASSANDRA-4560
is open to add support.
                 Tracing.instance.begin("Execute CQL3 prepared query", builder.build());
             }
 
-            Message.Response response = QueryProcessor.processPrepared(statement, consistency,
state, values, resultPageSize, pagingState);
-            if (skipMetadata && response instanceof ResultMessage.Rows)
+            Message.Response response = QueryProcessor.processPrepared(statement, state,
options);
+            if (options.skipMetadata() && response instanceof ResultMessage.Rows)
                 ((ResultMessage.Rows)response).result.metadata.setSkipMetadata();
 
             if (tracingId != null)
@@ -225,6 +161,6 @@ public class ExecuteMessage extends Message.Request
     @Override
     public String toString()
     {
-        return "EXECUTE " + statementId + " with " + values.size() + " values at consistency
" + consistency;
+        return "EXECUTE " + statementId + " with " + options.getValues().size() + " values
at consistency " + options.getConsistency();
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/bfd73bea/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 9e8050c..2b2583e 100644
--- a/src/java/org/apache/cassandra/transport/messages/QueryMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/QueryMessage.java
@@ -26,8 +26,10 @@ import java.util.UUID;
 
 import com.google.common.collect.ImmutableMap;
 import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
 
 import org.apache.cassandra.cql3.QueryProcessor;
+import org.apache.cassandra.cql3.QueryOptions;
 import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.exceptions.*;
 import org.apache.cassandra.service.QueryState;
@@ -41,147 +43,43 @@ 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,
-        PAGING_STATE;
-
-        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)
         {
             String query = CBUtil.readLongString(body);
-            ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
-
-            int resultPageSize = -1;
-            List<ByteBuffer> values = Collections.emptyList();
-            boolean skipMetadata = false;
-            PagingState pagingState = null;
-            if (version >= 2)
+            if (version == 1)
             {
-                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);
-
-                if (flags.contains(Flag.PAGING_STATE))
-                    pagingState = PagingState.deserialize(CBUtil.readValue(body));
+                ConsistencyLevel consistency = CBUtil.readConsistencyLevel(body);
+                return new QueryMessage(query, consistency);
+            }
+            else
+            {
+                return new QueryMessage(query, QueryOptions.codec.decode(body, version));
             }
-            return new QueryMessage(query, consistency, values, resultPageSize, skipMetadata,
pagingState);
         }
 
         public ChannelBuffer encode(QueryMessage msg, int version)
         {
-            // We have:
-            //   - query
-            //   - options
-            //     * optional:
-            //   - Number of values
-            //   - The values
-            int vs = msg.values.size();
-
-            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);
-            if (msg.pagingState != null)
-                flags.add(Flag.PAGING_STATE);
-
-            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 + (flags.contains(Flag.PAGING_STATE)
? 1 : 0));
-            builder.add(CBUtil.longStringToCB(msg.query));
-            builder.add(CBUtil.consistencyLevelToCB(msg.consistency));
-            if (version >= 2)
-            {
-                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)
-                        builder.addValue(value);
-                }
-                if (flags.contains(Flag.PAGING_STATE))
-                    builder.addValue(msg.pagingState == null ? null : msg.pagingState.serialize());
-            }
-            return builder.build();
+            return ChannelBuffers.wrappedBuffer(CBUtil.longStringToCB(msg.query),
+                                                (version == 1 ? CBUtil.consistencyLevelToCB(msg.options.getConsistency())
+                                                              : QueryOptions.codec.encode(msg.options,
version)));
         }
     };
 
     public final String query;
-    public final ConsistencyLevel consistency;
-    public final int resultPageSize;
-    public final List<ByteBuffer> values;
-    public final boolean skipMetadata;
-    public final PagingState pagingState;
+    public final QueryOptions options;
 
     public QueryMessage(String query, ConsistencyLevel consistency)
     {
-        this(query, consistency, Collections.<ByteBuffer>emptyList(), -1);
-    }
-
-    public QueryMessage(String query, ConsistencyLevel consistency, List<ByteBuffer>
values, int resultPageSize)
-    {
-        this(query, consistency, values, resultPageSize, false, null);
+        this(query, new QueryOptions(consistency, Collections.<ByteBuffer>emptyList()));
     }
 
-    public QueryMessage(String query, ConsistencyLevel consistency, List<ByteBuffer>
values, int resultPageSize, boolean skipMetadata, PagingState pagingState)
+    public QueryMessage(String query, QueryOptions options)
     {
         super(Type.QUERY);
         this.query = query;
-        this.consistency = consistency;
-        this.resultPageSize = resultPageSize;
-        this.values = values;
-        this.skipMetadata = skipMetadata;
-        this.pagingState = pagingState;
+        this.options = options;
     }
 
     public ChannelBuffer encode(int version)
@@ -193,7 +91,7 @@ public class QueryMessage extends Message.Request
     {
         try
         {
-            if (resultPageSize == 0)
+            if (options.getPageSize() == 0)
                 throw new ProtocolException("The page size cannot be 0");
 
             UUID tracingId = null;
@@ -209,14 +107,14 @@ public class QueryMessage extends Message.Request
 
                 ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
                 builder.put("query", query);
-                if (resultPageSize > 0)
-                    builder.put("page_size", Integer.toString(resultPageSize));
+                if (options.getPageSize() > 0)
+                    builder.put("page_size", Integer.toString(options.getPageSize()));
 
                 Tracing.instance.begin("Execute CQL3 query", builder.build());
             }
 
-            Message.Response response = QueryProcessor.process(query, values, consistency,
state, resultPageSize, pagingState);
-            if (skipMetadata && response instanceof ResultMessage.Rows)
+            Message.Response response = QueryProcessor.process(query, state, options);
+            if (options.skipMetadata() && response instanceof ResultMessage.Rows)
                 ((ResultMessage.Rows)response).result.metadata.setSkipMetadata();
 
             if (tracingId != null)


Mime
View raw message