cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From slebre...@apache.org
Subject git commit: Allow preparing timestamp, ttl and limit in queries
Date Tue, 28 May 2013 09:09:24 GMT
Updated Branches:
  refs/heads/trunk bc3597d35 -> 524261f88


Allow preparing timestamp, ttl and limit in queries

patch by slebresne; reviewed by iamaleksey for CASSANDRA-4450


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

Branch: refs/heads/trunk
Commit: 524261f88cd2adcd623de3604e735b282dd5caac
Parents: bc3597d
Author: Sylvain Lebresne <sylvain@datastax.com>
Authored: Mon Apr 29 09:27:36 2013 +0200
Committer: Sylvain Lebresne <sylvain@datastax.com>
Committed: Tue May 28 11:09:16 2013 +0200

----------------------------------------------------------------------
 CHANGES.txt                                        |    1 +
 src/java/org/apache/cassandra/cql3/Attributes.java |  110 ++++++++++++++-
 src/java/org/apache/cassandra/cql3/Cql.g           |   35 +++--
 .../cassandra/cql3/statements/BatchStatement.java  |   22 ++--
 .../cassandra/cql3/statements/DeleteStatement.java |    4 +-
 .../cql3/statements/ModificationStatement.java     |   28 ++--
 .../cassandra/cql3/statements/SelectStatement.java |  103 +++++++++-----
 .../cassandra/cql3/statements/UpdateStatement.java |    8 +-
 .../cassandra/transport/messages/BatchMessage.java |    2 +-
 9 files changed, 225 insertions(+), 88 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index e233ba0..e630d23 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -53,6 +53,7 @@
  * Track max/min column names in sstables to be able to optimize slice
    queries (CASSANDRA-5514)
  * Binary protocol: allow batching already prepared statements (CASSANDRA-4693)
+ * Allow preparing timestamp, ttl and limit in CQL3 queries (CASSANDRA-4450)
 
 1.2.6
  * (Hadoop) Fix InputKeyRange in CFIF (CASSANDRA-5536)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/src/java/org/apache/cassandra/cql3/Attributes.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Attributes.java b/src/java/org/apache/cassandra/cql3/Attributes.java
index 62f98b2..511f34e 100644
--- a/src/java/org/apache/cassandra/cql3/Attributes.java
+++ b/src/java/org/apache/cassandra/cql3/Attributes.java
@@ -17,7 +17,13 @@
  */
 package org.apache.cassandra.cql3;
 
+import java.nio.ByteBuffer;
+import java.util.List;
+
 import org.apache.cassandra.db.ExpiringColumn;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.LongType;
+import org.apache.cassandra.db.marshal.MarshalException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 
 /**
@@ -26,15 +32,107 @@ import org.apache.cassandra.exceptions.InvalidRequestException;
  */
 public class Attributes
 {
-    public Long timestamp;
-    public int timeToLive;
+    private final Term timestamp;
+    private final Term timeToLive;
+
+    public static Attributes none()
+    {
+        return new Attributes(null, null);
+    }
+
+    private Attributes(Term timestamp, Term timeToLive)
+    {
+        this.timestamp = timestamp;
+        this.timeToLive = timeToLive;
+    }
+
+    public boolean isTimestampSet()
+    {
+        return timestamp != null;
+    }
+
+    public boolean isTimeToLiveSet()
+    {
+        return timeToLive != null;
+    }
+
+    public long getTimestamp(long now, List<ByteBuffer> variables) throws InvalidRequestException
+    {
+        if (timestamp == null)
+            return now;
+
+        ByteBuffer tval = timestamp.bindAndGet(variables);
+        if (tval == null)
+            throw new InvalidRequestException("Invalid null value of timestamp");
 
-    public void validate() throws InvalidRequestException
+        try
+        {
+            LongType.instance.validate(tval);
+        }
+        catch (MarshalException e)
+        {
+            throw new InvalidRequestException("Invalid timestamp value");
+        }
+
+        return LongType.instance.compose(tval);
+    }
+
+    public int getTimeToLive(List<ByteBuffer> variables) throws InvalidRequestException
     {
-        if (timeToLive < 0)
+        if (timeToLive == null)
+            return 0;
+
+        ByteBuffer tval = timeToLive.bindAndGet(variables);
+        if (tval == null)
+            throw new InvalidRequestException("Invalid null value of TTL");
+
+        try
+        {
+            Int32Type.instance.validate(tval);
+        }
+        catch (MarshalException e)
+        {
+            throw new InvalidRequestException("Invalid timestamp value");
+        }
+
+        int ttl = Int32Type.instance.compose(tval);
+        if (ttl < 0)
             throw new InvalidRequestException("A TTL must be greater or equal to 0");
 
-        if (timeToLive > ExpiringColumn.MAX_TTL)
-            throw new InvalidRequestException(String.format("ttl is too large. requested
(%d) maximum (%d)", timeToLive, ExpiringColumn.MAX_TTL));
+        if (ttl > ExpiringColumn.MAX_TTL)
+            throw new InvalidRequestException(String.format("ttl is too large. requested
(%d) maximum (%d)", ttl, ExpiringColumn.MAX_TTL));
+
+        return ttl;
+    }
+
+    public void collectMarkerSpecification(ColumnSpecification[] boundNames)
+    {
+        if (timestamp != null)
+            timestamp.collectMarkerSpecification(boundNames);
+        if (timeToLive != null)
+            timeToLive.collectMarkerSpecification(boundNames);
+    }
+
+    public static class Raw
+    {
+        public Term.Raw timestamp;
+        public Term.Raw timeToLive;
+
+        public Attributes prepare(String ksName, String cfName) throws InvalidRequestException
+        {
+            Term ts = timestamp == null ? null : timestamp.prepare(timestampReceiver(ksName,
cfName));
+            Term ttl = timeToLive == null ? null : timeToLive.prepare(timeToLiveReceiver(ksName,
cfName));
+            return new Attributes(ts, ttl);
+        }
+
+        private ColumnSpecification timestampReceiver(String ksName, String cfName)
+        {
+            return new ColumnSpecification(ksName, cfName, new ColumnIdentifier("[timestamp]",
true), LongType.instance);
+        }
+
+        private ColumnSpecification timeToLiveReceiver(String ksName, String cfName)
+        {
+            return new ColumnSpecification(ksName, cfName, new ColumnIdentifier("[ttl]",
true), Int32Type.instance);
+        }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/src/java/org/apache/cassandra/cql3/Cql.g
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Cql.g b/src/java/org/apache/cassandra/cql3/Cql.g
index 8d31de9..913f6ea 100644
--- a/src/java/org/apache/cassandra/cql3/Cql.g
+++ b/src/java/org/apache/cassandra/cql3/Cql.g
@@ -212,7 +212,7 @@ selectStatement returns [SelectStatement.RawStatement expr]
     @init {
         boolean isCount = false;
         ColumnIdentifier countAlias = null;
-        int limit = Integer.MAX_VALUE;
+        Term.Raw limit = null;
         Map<ColumnIdentifier, Boolean> orderings = new LinkedHashMap<ColumnIdentifier,
Boolean>();
         boolean allowFiltering = false;
     }
@@ -221,15 +221,14 @@ selectStatement returns [SelectStatement.RawStatement expr]
       K_FROM cf=columnFamilyName
       ( K_WHERE wclause=whereClause )?
       ( K_ORDER K_BY orderByClause[orderings] ( ',' orderByClause[orderings] )* )?
-      ( K_LIMIT rows=INTEGER { limit = Integer.parseInt($rows.text); } )?
+      ( K_LIMIT rows=intValue { limit = rows; } )?
       ( K_ALLOW K_FILTERING  { allowFiltering = true; } )?
       {
-          SelectStatement.Parameters params = new SelectStatement.Parameters(limit,
-                                                                             orderings,
+          SelectStatement.Parameters params = new SelectStatement.Parameters(orderings,
                                                                              isCount,
                                                                              countAlias,
                                                                              allowFiltering);
-          $expr = new SelectStatement.RawStatement(cf, params, sclause, wclause);
+          $expr = new SelectStatement.RawStatement(cf, params, sclause, wclause, limit);
       }
     ;
 
@@ -283,7 +282,7 @@ orderByClause[Map<ColumnIdentifier, Boolean> orderings]
  */
 insertStatement returns [UpdateStatement.ParsedInsert expr]
     @init {
-        Attributes attrs = new Attributes();
+        Attributes.Raw attrs = new Attributes.Raw();
         List<ColumnIdentifier> columnNames  = new ArrayList<ColumnIdentifier>();
         List<Term.Raw> values = new ArrayList<Term.Raw>();
     }
@@ -297,21 +296,21 @@ insertStatement returns [UpdateStatement.ParsedInsert expr]
       }
     ;
 
-usingClause[Attributes attrs]
+usingClause[Attributes.Raw attrs]
     : K_USING usingClauseObjective[attrs] ( K_AND? usingClauseObjective[attrs] )*
     ;
 
-usingClauseDelete[Attributes attrs]
+usingClauseDelete[Attributes.Raw attrs]
     : K_USING usingClauseDeleteObjective[attrs] ( K_AND? usingClauseDeleteObjective[attrs]
)*
     ;
 
-usingClauseDeleteObjective[Attributes attrs]
-    : K_TIMESTAMP ts=INTEGER { attrs.timestamp = Long.valueOf($ts.text); }
+usingClauseDeleteObjective[Attributes.Raw attrs]
+    : K_TIMESTAMP ts=intValue { attrs.timestamp = ts; }
     ;
 
-usingClauseObjective[Attributes attrs]
+usingClauseObjective[Attributes.Raw attrs]
     : usingClauseDeleteObjective[attrs]
-    | K_TTL t=INTEGER { attrs.timeToLive = Integer.valueOf($t.text); }
+    | K_TTL t=intValue { attrs.timeToLive = t; }
     ;
 
 /**
@@ -322,7 +321,7 @@ usingClauseObjective[Attributes attrs]
  */
 updateStatement returns [UpdateStatement.ParsedUpdate expr]
     @init {
-        Attributes attrs = new Attributes();
+        Attributes.Raw attrs = new Attributes.Raw();
         List<Pair<ColumnIdentifier, Operation.RawUpdate>> operations = new ArrayList<Pair<ColumnIdentifier,
Operation.RawUpdate>>();
         boolean ifNotExists = false;
     }
@@ -354,7 +353,7 @@ updateCondition returns [List<Pair<ColumnIdentifier, Operation.RawUpdate>>
condi
  */
 deleteStatement returns [DeleteStatement.Parsed expr]
     @init {
-        Attributes attrs = new Attributes();
+        Attributes.Raw attrs = new Attributes.Raw();
         List<Operation.RawDeletion> columnDeletions = Collections.emptyList();
     }
     : K_DELETE ( dels=deleteSelection { columnDeletions = dels; } )?
@@ -410,7 +409,7 @@ batchStatement returns [BatchStatement.Parsed expr]
     @init {
         BatchStatement.Type type = BatchStatement.Type.LOGGED;
         List<ModificationStatement.Parsed> statements = new ArrayList<ModificationStatement.Parsed>();
-        Attributes attrs = new Attributes();
+        Attributes.Raw attrs = new Attributes.Raw();
     }
     : K_BEGIN
       ( K_UNLOGGED { type = BatchStatement.Type.UNLOGGED; } | K_COUNTER { type = BatchStatement.Type.COUNTER;
} )?
@@ -738,6 +737,12 @@ value returns [Term.Raw value]
     | QMARK                { $value = new AbstractMarker.Raw(++currentBindMarkerIdx); }
     ;
 
+intValue returns [Term.Raw value]
+    :
+    | t=INTEGER { $value = Constants.Literal.integer($t.text); }
+    | QMARK     { $value = new AbstractMarker.Raw(++currentBindMarkerIdx); }
+    ;
+
 functionName returns [String s]
     : f=IDENT                       { $s = $f.text; }
     | u=unreserved_function_keyword { $s = u; }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
index d6d0e16..777c80f 100644
--- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
@@ -76,23 +76,16 @@ public class BatchStatement implements CQLStatement
 
     public void validate(ClientState state) throws InvalidRequestException
     {
-        if (attrs.timeToLive != 0)
+        if (attrs.isTimeToLiveSet())
             throw new InvalidRequestException("Global TTL on the BATCH statement is not supported.");
 
         for (ModificationStatement statement : statements)
         {
-            statement.validate(state);
-
-            if (attrs.timestamp != null && statement.isSetTimestamp())
+            if (attrs.isTimestampSet() && statement.isTimestampSet())
                 throw new InvalidRequestException("Timestamp must be set either on BATCH
or individual statements");
         }
     }
 
-    public long getTimestamp(long now)
-    {
-        return attrs.timestamp == null ? now : attrs.timestamp;
-    }
-
     private Collection<? extends IMutation> getMutations(List<ByteBuffer> variables,
boolean local, ConsistencyLevel cl, long now)
     throws RequestExecutionException, RequestValidationException
     {
@@ -125,7 +118,7 @@ public class BatchStatement implements CQLStatement
     throws RequestExecutionException, RequestValidationException
     {
         // Group mutation together, otherwise they won't get applied atomically
-        for (IMutation m : statement.getMutations(variables, local, cl, getTimestamp(now),
true))
+        for (IMutation m : statement.getMutations(variables, local, cl, attrs.getTimestamp(now,
variables), true))
         {
             Pair<String, ByteBuffer> key = Pair.create(m.getTable(), m.key());
             IMutation existing = mutations.get(key);
@@ -179,10 +172,10 @@ public class BatchStatement implements CQLStatement
     public static class Parsed extends CFStatement
     {
         private final Type type;
-        private final Attributes attrs;
+        private final Attributes.Raw attrs;
         private final List<ModificationStatement.Parsed> parsedStatements;
 
-        public Parsed(Type type, Attributes attrs, List<ModificationStatement.Parsed>
parsedStatements)
+        public Parsed(Type type, Attributes.Raw attrs, List<ModificationStatement.Parsed>
parsedStatements)
         {
             super(null);
             this.type = type;
@@ -217,7 +210,10 @@ public class BatchStatement implements CQLStatement
                 statements.add(stmt);
             }
 
-            return new ParsedStatement.Prepared(new BatchStatement(getBoundsTerms(), type,
statements, attrs), Arrays.<ColumnSpecification>asList(boundNames));
+            Attributes prepAttrs = attrs.prepare("[batch]", "[batch]");
+            prepAttrs.collectMarkerSpecification(boundNames);
+
+            return new ParsedStatement.Prepared(new BatchStatement(getBoundsTerms(), type,
statements, prepAttrs), Arrays.<ColumnSpecification>asList(boundNames));
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java b/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
index db3c41c..54a1034 100644
--- a/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
@@ -93,7 +93,7 @@ public class DeleteStatement extends ModificationStatement
         private final List<Relation> whereClause;
 
         public Parsed(CFName name,
-                      Attributes attrs,
+                      Attributes.Raw attrs,
                       List<Operation.RawDeletion> deletions,
                       List<Relation> whereClause,
                       List<Pair<ColumnIdentifier, Operation.RawUpdate>> conditions)
@@ -103,7 +103,7 @@ public class DeleteStatement extends ModificationStatement
             this.whereClause = whereClause;
         }
 
-        protected ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames) throws InvalidRequestException
+        protected ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames, Attributes attrs) throws InvalidRequestException
         {
             DeleteStatement stmt = new DeleteStatement(getBoundsTerms(), cfDef.cfm, attrs);
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index 7766f94..f6b7140 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -85,19 +85,19 @@ public abstract class ModificationStatement implements CQLStatement
         return cfm.getDefaultValidator().isCommutative();
     }
 
-    public int getTimeToLive()
+    public long getTimestamp(long now, List<ByteBuffer> variables) throws InvalidRequestException
     {
-        return attrs.timeToLive;
+        return attrs.getTimestamp(now, variables);
     }
 
-    public long getTimestamp(long now)
+    public boolean isTimestampSet()
     {
-        return attrs.timestamp == null ? now : attrs.timestamp;
+        return attrs.isTimestampSet();
     }
 
-    public boolean isSetTimestamp()
+    public int getTimeToLive(List<ByteBuffer> variables) throws InvalidRequestException
     {
-        return attrs.timestamp != null;
+        return attrs.getTimeToLive(variables);
     }
 
     public void checkAccess(ClientState state) throws InvalidRequestException, UnauthorizedException
@@ -107,7 +107,6 @@ public abstract class ModificationStatement implements CQLStatement
 
     public void validate(ClientState state) throws InvalidRequestException
     {
-        attrs.validate();
     }
 
     public void addOperation(Operation op)
@@ -363,7 +362,7 @@ public abstract class ModificationStatement implements CQLStatement
             throw new InvalidRequestException("IN on the partition key is not supported with
conditional updates");
 
         ColumnNameBuilder clusteringPrefix = createClusteringPrefixBuilder(variables);
-        UpdateParameters params = new UpdateParameters(cfm, variables, getTimestamp(queryState.getTimestamp()),
getTimeToLive(), null);
+        UpdateParameters params = new UpdateParameters(cfm, variables, getTimestamp(queryState.getTimestamp(),
variables), getTimeToLive(variables), null);
 
         ByteBuffer key = keys.get(0);
         ThriftValidation.validateKey(cfm, key);
@@ -407,7 +406,7 @@ public abstract class ModificationStatement implements CQLStatement
 
         // Some lists operation requires reading
         Map<ByteBuffer, ColumnGroupMap> rows = readRequiredRows(keys, clusteringPrefix,
local, cl);
-        UpdateParameters params = new UpdateParameters(cfm, variables, getTimestamp(now),
getTimeToLive(), rows);
+        UpdateParameters params = new UpdateParameters(cfm, variables, getTimestamp(now,
variables), getTimeToLive(variables), rows);
 
         Collection<IMutation> mutations = new ArrayList<IMutation>();
         for (ByteBuffer key: keys)
@@ -449,11 +448,11 @@ public abstract class ModificationStatement implements CQLStatement
 
     public static abstract class Parsed extends CFStatement
     {
-        protected final Attributes attrs;
+        protected final Attributes.Raw attrs;
         private final List<Pair<ColumnIdentifier, Operation.RawUpdate>> conditions;
         private final boolean ifNotExists;
 
-        protected Parsed(CFName name, Attributes attrs, List<Pair<ColumnIdentifier,
Operation.RawUpdate>> conditions, boolean ifNotExists)
+        protected Parsed(CFName name, Attributes.Raw attrs, List<Pair<ColumnIdentifier,
Operation.RawUpdate>> conditions, boolean ifNotExists)
         {
             super(name);
             this.attrs = attrs;
@@ -473,7 +472,10 @@ public abstract class ModificationStatement implements CQLStatement
             CFMetaData metadata = ThriftValidation.validateColumnFamily(keyspace(), columnFamily());
             CFDefinition cfDef = metadata.getCfDef();
 
-            ModificationStatement stmt = prepareInternal(cfDef, boundNames);
+            Attributes preparedAttributes = attrs.prepare(keyspace(), columnFamily());
+            preparedAttributes.collectMarkerSpecification(boundNames);
+
+            ModificationStatement stmt = prepareInternal(cfDef, boundNames, preparedAttributes);
 
             if (ifNotExists || (conditions != null && !conditions.isEmpty()))
             {
@@ -528,6 +530,6 @@ public abstract class ModificationStatement implements CQLStatement
             return stmt;
         }
 
-        protected abstract ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames) throws InvalidRequestException;
+        protected abstract ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames, Attributes attrs) throws InvalidRequestException;
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/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 d45b730..9630771 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -58,6 +58,7 @@ public class SelectStatement implements CQLStatement
     public final CFDefinition cfDef;
     public final Parameters parameters;
     private final Selection selection;
+    private final Term limit;
 
     private final Restriction[] keyRestrictions;
     private final Restriction[] columnRestrictions;
@@ -92,7 +93,7 @@ public class SelectStatement implements CQLStatement
         }
     }
 
-    public SelectStatement(CFDefinition cfDef, int boundTerms, Parameters parameters, Selection
selection)
+    public SelectStatement(CFDefinition cfDef, int boundTerms, Parameters parameters, Selection
selection, Term limit)
     {
         this.cfDef = cfDef;
         this.boundTerms = boundTerms;
@@ -100,6 +101,7 @@ public class SelectStatement implements CQLStatement
         this.keyRestrictions = new Restriction[cfDef.keys.size()];
         this.columnRestrictions = new Restriction[cfDef.columns.size()];
         this.parameters = parameters;
+        this.limit = limit;
     }
 
     public int getBoundsTerms()
@@ -124,17 +126,18 @@ public class SelectStatement implements CQLStatement
 
         cl.validateForRead(keyspace());
 
+        int limit = getLimit(variables);
         List<Row> rows = isKeyRange || usesSecondaryIndexing
-                       ? StorageProxy.getRangeSlice(getRangeCommand(variables), cl)
-                       : StorageProxy.read(getSliceCommands(variables), cl);
+                       ? StorageProxy.getRangeSlice(getRangeCommand(variables, limit), cl)
+                       : StorageProxy.read(getSliceCommands(variables, limit), cl);
 
-        return processResults(rows, variables);
+        return processResults(rows, variables, limit);
     }
 
-    private ResultMessage.Rows processResults(List<Row> rows, List<ByteBuffer>
variables) throws RequestValidationException
+    private ResultMessage.Rows processResults(List<Row> rows, List<ByteBuffer>
variables, int limit) throws RequestValidationException
     {
         // Even for count, we need to process the result as it'll group some column together
in sparse column families
-        ResultSet rset = process(rows, variables);
+        ResultSet rset = process(rows, variables, limit);
         rset = parameters.isCount ? rset.makeCountResult(parameters.countAlias) : rset;
         return new ResultMessage.Rows(rset);
     }
@@ -150,17 +153,19 @@ public class SelectStatement implements CQLStatement
 
     public ResultMessage.Rows executeInternal(QueryState state) throws RequestExecutionException,
RequestValidationException
     {
+        List<ByteBuffer> variables = Collections.<ByteBuffer>emptyList();
+        int limit = getLimit(variables);
         List<Row> rows = isKeyRange || usesSecondaryIndexing
-                       ? RangeSliceVerbHandler.executeLocally(getRangeCommand(Collections.<ByteBuffer>emptyList()))
-                       : readLocally(keyspace(), getSliceCommands(Collections.<ByteBuffer>emptyList()));
+                       ? RangeSliceVerbHandler.executeLocally(getRangeCommand(variables,
limit))
+                       : readLocally(keyspace(), getSliceCommands(variables, limit));
 
-        return processResults(rows, Collections.<ByteBuffer>emptyList());
+        return processResults(rows, variables, limit);
     }
 
     public ResultSet process(List<Row> rows) throws InvalidRequestException
     {
         assert !parameters.isCount; // not yet needed
-        return process(rows, Collections.<ByteBuffer>emptyList());
+        return process(rows, Collections.<ByteBuffer>emptyList(), getLimit(Collections.<ByteBuffer>emptyList()));
     }
 
     public String keyspace()
@@ -173,12 +178,12 @@ public class SelectStatement implements CQLStatement
         return cfDef.cfm.cfName;
     }
 
-    private List<ReadCommand> getSliceCommands(List<ByteBuffer> variables) throws
RequestValidationException
+    private List<ReadCommand> getSliceCommands(List<ByteBuffer> variables, int
limit) throws RequestValidationException
     {
         Collection<ByteBuffer> keys = getKeys(variables);
         List<ReadCommand> commands = new ArrayList<ReadCommand>(keys.size());
 
-        IDiskAtomFilter filter = makeFilter(variables);
+        IDiskAtomFilter filter = makeFilter(variables, limit);
         // Note that we use the total limit for every key, which is potentially inefficient.
         // However, IN + LIMIT is not a very sensible choice.
         for (ByteBuffer key : keys)
@@ -192,9 +197,9 @@ public class SelectStatement implements CQLStatement
         return commands;
     }
 
-    private RangeSliceCommand getRangeCommand(List<ByteBuffer> variables) throws RequestValidationException
+    private RangeSliceCommand getRangeCommand(List<ByteBuffer> variables, int limit)
throws RequestValidationException
     {
-        IDiskAtomFilter filter = makeFilter(variables);
+        IDiskAtomFilter filter = makeFilter(variables, limit);
         List<IndexExpression> expressions = getIndexExpressions(variables);
         // The LIMIT provided by the user is the number of CQL row he wants returned.
         // We want to have getRangeSlice to count the number of columns, not the number of
keys.
@@ -203,7 +208,7 @@ public class SelectStatement implements CQLStatement
                                      filter,
                                      getKeyBounds(variables),
                                      expressions,
-                                     getLimit(),
+                                     limit,
                                      true,
                                      false);
     }
@@ -252,7 +257,7 @@ public class SelectStatement implements CQLStatement
         return bounds;
     }
 
-    private IDiskAtomFilter makeFilter(List<ByteBuffer> variables)
+    private IDiskAtomFilter makeFilter(List<ByteBuffer> variables, int limit)
     throws InvalidRequestException
     {
         if (isColumnRange())
@@ -266,7 +271,7 @@ public class SelectStatement implements CQLStatement
                                                 getRequestedBound(Bound.END, variables));
             SliceQueryFilter filter = new SliceQueryFilter(new ColumnSlice[]{slice},
                                                            isReversed,
-                                                           getLimit(),
+                                                           limit,
                                                            toGroup);
             QueryProcessor.validateSliceFilter(cfDef.cfm, filter);
             return filter;
@@ -279,13 +284,35 @@ public class SelectStatement implements CQLStatement
         }
     }
 
-    private int getLimit()
+    private int getLimit(List<ByteBuffer> variables) throws InvalidRequestException
     {
+        int l = Integer.MAX_VALUE;
+        if (limit != null)
+        {
+            ByteBuffer b = limit.bindAndGet(variables);
+            if (b == null)
+                throw new InvalidRequestException("Invalid null value of limit");
+
+            try
+            {
+                Int32Type.instance.validate(b);
+                l = Int32Type.instance.compose(b);
+            }
+            catch (MarshalException e)
+            {
+                throw new InvalidRequestException("Invalid limit value");
+            }
+        }
+
+        if (l <= 0)
+            throw new InvalidRequestException("LIMIT must be strictly positive");
+
         // Internally, we don't support exclusive bounds for slices. Instead,
         // we query one more element if necessary and exclude
-        return sliceRestriction != null && !sliceRestriction.isInclusive(Bound.START)
&& parameters.limit != Integer.MAX_VALUE
-             ? parameters.limit + 1
-             : parameters.limit;
+        if (sliceRestriction != null && !sliceRestriction.isInclusive(Bound.START)
&& l != Integer.MAX_VALUE)
+            l += 1;
+
+        return l;
     }
 
     private Collection<ByteBuffer> getKeys(final List<ByteBuffer> variables)
throws InvalidRequestException
@@ -634,7 +661,7 @@ public class SelectStatement implements CQLStatement
         };
     }
 
-    private ResultSet process(List<Row> rows, List<ByteBuffer> variables) throws
InvalidRequestException
+    private ResultSet process(List<Row> rows, List<ByteBuffer> variables, int
limit) throws InvalidRequestException
     {
         Selection.ResultSetBuilder result = selection.resultSetBuilder();
         for (org.apache.cassandra.db.Row row : rows)
@@ -740,7 +767,7 @@ public class SelectStatement implements CQLStatement
             cqlRows.reverse();
 
         // Trim result if needed to respect the limit
-        cqlRows.trim(parameters.limit);
+        cqlRows.trim(limit);
         return cqlRows;
     }
 
@@ -839,22 +866,21 @@ public class SelectStatement implements CQLStatement
         private final Parameters parameters;
         private final List<RawSelector> selectClause;
         private final List<Relation> whereClause;
+        private final Term.Raw limit;
 
-        public RawStatement(CFName cfName, Parameters parameters, List<RawSelector>
selectClause, List<Relation> whereClause)
+        public RawStatement(CFName cfName, Parameters parameters, List<RawSelector>
selectClause, List<Relation> whereClause, Term.Raw limit)
         {
             super(cfName);
             this.parameters = parameters;
             this.selectClause = selectClause;
             this.whereClause = whereClause == null ? Collections.<Relation>emptyList()
: whereClause;
+            this.limit = limit;
         }
 
         public ParsedStatement.Prepared prepare() throws InvalidRequestException
         {
             CFMetaData cfm = ThriftValidation.validateColumnFamily(keyspace(), columnFamily());
 
-            if (parameters.limit <= 0)
-                throw new InvalidRequestException("LIMIT must be strictly positive");
-
             CFDefinition cfDef = cfm.getCfDef();
 
             ColumnSpecification[] names = new ColumnSpecification[getBoundsTerms()];
@@ -868,7 +894,14 @@ public class SelectStatement implements CQLStatement
                                 ? Selection.wildcard(cfDef)
                                 : Selection.fromSelectors(cfDef, selectClause);
 
-            SelectStatement stmt = new SelectStatement(cfDef, getBoundsTerms(), parameters,
selection);
+            Term prepLimit = null;
+            if (limit != null)
+            {
+                prepLimit = limit.prepare(limitReceiver());
+                prepLimit.collectMarkerSpecification(names);
+            }
+
+            SelectStatement stmt = new SelectStatement(cfDef, getBoundsTerms(), parameters,
selection, prepLimit);
 
             /*
              * WHERE clause. For a given entity, rules are:
@@ -1209,6 +1242,11 @@ public class SelectStatement implements CQLStatement
                                                });
         }
 
+        private ColumnSpecification limitReceiver()
+        {
+            return new ColumnSpecification(keyspace(), columnFamily(), new ColumnIdentifier("[limit]",
true), Int32Type.instance);
+        }
+
         Restriction updateRestriction(CFDefinition.Name name, Restriction restriction, Relation
newRel, ColumnSpecification[] boundNames) throws InvalidRequestException
         {
             ColumnSpecification receiver = name;
@@ -1265,12 +1303,11 @@ public class SelectStatement implements CQLStatement
         @Override
         public String toString()
         {
-            return String.format("SelectRawStatement[name=%s, selectClause=%s, whereClause=%s,
isCount=%s, limit=%s]",
+            return String.format("SelectRawStatement[name=%s, selectClause=%s, whereClause=%s,
isCount=%s]",
                     cfName,
                     selectClause,
                     whereClause,
-                    parameters.isCount,
-                    parameters.limit);
+                    parameters.isCount);
         }
     }
 
@@ -1409,15 +1446,13 @@ public class SelectStatement implements CQLStatement
 
     public static class Parameters
     {
-        private final int limit;
         private final Map<ColumnIdentifier, Boolean> orderings;
         private final boolean isCount;
         private final ColumnIdentifier countAlias;
         private final boolean allowFiltering;
 
-        public Parameters(int limit, Map<ColumnIdentifier, Boolean> orderings, boolean
isCount, ColumnIdentifier countAlias, boolean allowFiltering)
+        public Parameters(Map<ColumnIdentifier, Boolean> orderings, boolean isCount,
ColumnIdentifier countAlias, boolean allowFiltering)
         {
-            this.limit = limit;
             this.orderings = orderings;
             this.isCount = isCount;
             this.countAlias = countAlias;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index bc05dd6..cff4105 100644
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@ -115,14 +115,14 @@ public class UpdateStatement extends ModificationStatement
          * @param columnValues list of column values (corresponds to names)
          * @param attrs additional attributes for statement (CL, timestamp, timeToLive)
          */
-        public ParsedInsert(CFName name, Attributes attrs, List<ColumnIdentifier> columnNames,
List<Term.Raw> columnValues)
+        public ParsedInsert(CFName name, Attributes.Raw attrs, List<ColumnIdentifier>
columnNames, List<Term.Raw> columnValues)
         {
             super(name, attrs, Collections.<Pair<ColumnIdentifier, Operation.RawUpdate>>emptyList(),
false);
             this.columnNames = columnNames;
             this.columnValues = columnValues;
         }
 
-        protected ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames) throws InvalidRequestException
+        protected ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames, Attributes attrs) throws InvalidRequestException
         {
             UpdateStatement stmt = new UpdateStatement(getBoundsTerms(), cfDef.cfm, attrs);
 
@@ -182,7 +182,7 @@ public class UpdateStatement extends ModificationStatement
          * @param whereClause the where clause
          */
         public ParsedUpdate(CFName name,
-                            Attributes attrs,
+                            Attributes.Raw attrs,
                             List<Pair<ColumnIdentifier, Operation.RawUpdate>>
updates,
                             List<Relation> whereClause,
                             List<Pair<ColumnIdentifier, Operation.RawUpdate>>
conditions,
@@ -193,7 +193,7 @@ public class UpdateStatement extends ModificationStatement
             this.whereClause = whereClause;
         }
 
-        protected ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames) throws InvalidRequestException
+        protected ModificationStatement prepareInternal(CFDefinition cfDef, ColumnSpecification[]
boundNames, Attributes attrs) throws InvalidRequestException
         {
             UpdateStatement stmt = new UpdateStatement(getBoundsTerms(), cfDef.cfm, attrs);
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/524261f8/src/java/org/apache/cassandra/transport/messages/BatchMessage.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/transport/messages/BatchMessage.java b/src/java/org/apache/cassandra/transport/messages/BatchMessage.java
index 3bec918..ed8aaaf 100644
--- a/src/java/org/apache/cassandra/transport/messages/BatchMessage.java
+++ b/src/java/org/apache/cassandra/transport/messages/BatchMessage.java
@@ -221,7 +221,7 @@ public class BatchMessage extends Message.Request
 
             // Note: It's ok at this point to pass a bogus value for the number of bound
terms in the BatchState ctor
             // (and no value would be really correct, so we prefer passing a clearly wrong
one).
-            BatchStatement batch = new BatchStatement(-1, type, statements, new Attributes());
+            BatchStatement batch = new BatchStatement(-1, type, statements, Attributes.none());
             Message.Response response = QueryProcessor.processBatch(batch, consistency, state,
values);
 
             if (tracingId != null)


Mime
View raw message