You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sl...@apache.org on 2013/05/28 11:09:24 UTC
git commit: Allow preparing timestamp, ttl and limit in queries
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 <sy...@datastax.com>
Authored: Mon Apr 29 09:27:36 2013 +0200
Committer: Sylvain Lebresne <sy...@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)