diff --git a/CHANGES.txt b/CHANGES.txt index ea5d6cdacc..1c447ccf31 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,6 +1,7 @@ 1.0-dev * add support for insert, delete in cql BATCH (CASSANDRA-2537) * add support for IN to cql SELECT, UPDATE (CASSANDRA-2553) + * add timestamp support to cql INSERT, UPDATE, and BATCH (CASSANDRA-2555) 0.8.0-? diff --git a/doc/cql/CQL.textile b/doc/cql/CQL.textile index 8d280d2ba5..1a0cbfd0b2 100644 --- a/doc/cql/CQL.textile +++ b/doc/cql/CQL.textile @@ -74,7 +74,7 @@ h2. INSERT _Synopsis:_ bc. -INSERT INTO (KEY, , , ...) VALUES (, , , ...) [USING CONSISTENCY (KEY, , , ...) VALUES (, , , ...) [USING CONSISTENCY [AND TIMESTAMP ]]; An @INSERT@ is used to write one or more columns to a record in a Cassandra column family. No results are returned. @@ -85,7 +85,7 @@ h2. UPDATE _Synopsis:_ bc. -UPDATE [USING ] +UPDATE [USING [AND TIMESTAMP ]] SET name1 = value1, name2 = value2 WHERE KEY = keyname; An @UPDATE@ is used to write one or more columns to a record in a Cassandra column family. No results are returned. @@ -104,6 +104,13 @@ UPDATE ... [USING ] ... Following the column family identifier is an optional "consistency level specification":#consistency. +h3. Timestamp + +bc. +UPDATE ... [USING TIMESTAMP ] ... + +@UPDATE@ supports setting client-supplied optional timestamp for modification. + h3. Specifying Columns and Row bc. @@ -156,10 +163,12 @@ h2. BATCH _Synopsis:_ bc. -BATCH BEGIN BATCH [USING CONSISTENCY ] +BATCH BEGIN BATCH [USING CONSISTENCY [AND TIMESTAMP ]] INSERT or UPDATE or DELETE statements separated by semicolon or "end of line" APPLY BATCH +@BATCH@ supports setting client-supplied optional global timestamp which will be used for each of the operations included in batch. + A single consistency level is used for the entire batch, it appears after the @BEGIN BATCH@ statement, and uses the standard "consistency level specification":#consistency. Batch default to @CONSISTENCY.ONE@ when left unspecified. _NOTE: While there are no isolation guarantees, @UPDATE@ queries are atomic within a give record._ diff --git a/src/java/org/apache/cassandra/cql/AbstractModification.java b/src/java/org/apache/cassandra/cql/AbstractModification.java index 74e8e6d64f..e4d154c7d6 100644 --- a/src/java/org/apache/cassandra/cql/AbstractModification.java +++ b/src/java/org/apache/cassandra/cql/AbstractModification.java @@ -32,11 +32,18 @@ public abstract class AbstractModification protected final String columnFamily; protected final ConsistencyLevel cLevel; + protected final Long timestamp; public AbstractModification(String columnFamily, ConsistencyLevel cLevel) + { + this(columnFamily, cLevel, null); + } + + public AbstractModification(String columnFamily, ConsistencyLevel cLevel, Long timestamp) { this.columnFamily = columnFamily; this.cLevel = cLevel; + this.timestamp = timestamp; } public String getColumnFamily() @@ -59,6 +66,16 @@ public abstract class AbstractModification return cLevel != null; } + public long getTimestamp() + { + return timestamp == null ? System.currentTimeMillis() : timestamp; + } + + public boolean isSetTimestamp() + { + return timestamp != null; + } + /** * Convert statement into a list of mutations to apply on the server * @@ -71,4 +88,18 @@ public abstract class AbstractModification */ public abstract List prepareRowMutations(String keyspace, ClientState clientState) throws org.apache.cassandra.thrift.InvalidRequestException; + + /** + * Convert statement into a list of mutations to apply on the server + * + * @param keyspace The working keyspace + * @param clientState current client status + * @param timestamp global timestamp to use for all mutations + * + * @return list of the mutations + * + * @throws org.apache.cassandra.thrift.InvalidRequestException on the wrong request + */ + public abstract List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) + throws org.apache.cassandra.thrift.InvalidRequestException; } diff --git a/src/java/org/apache/cassandra/cql/BatchStatement.java b/src/java/org/apache/cassandra/cql/BatchStatement.java index e609374fe5..e025a00a20 100644 --- a/src/java/org/apache/cassandra/cql/BatchStatement.java +++ b/src/java/org/apache/cassandra/cql/BatchStatement.java @@ -40,6 +40,8 @@ public class BatchStatement // global consistency level protected final ConsistencyLevel consistency; + // global timestamp to apply for each mutation + protected final Long timestamp; /** * Creates a new BatchStatement from a list of statements and a @@ -48,10 +50,11 @@ public class BatchStatement * @param statements a list of UpdateStatements * @param level Thrift consistency level enum */ - public BatchStatement(List statements, ConsistencyLevel level) + public BatchStatement(List statements, ConsistencyLevel level, Long timestamp) { this.statements = statements; consistency = level; + this.timestamp = timestamp; } public List getStatements() @@ -70,12 +73,16 @@ public class BatchStatement for (AbstractModification statement : statements) { - batch.addAll(statement.prepareRowMutations(keyspace, clientState)); + batch.addAll(statement.prepareRowMutations(keyspace, clientState, timestamp)); } return batch; } + public boolean isSetTimestamp() + { + return timestamp != null; + } public String toString() { diff --git a/src/java/org/apache/cassandra/cql/Cql.g b/src/java/org/apache/cassandra/cql/Cql.g index d055cbfa3b..e9bd5dbae4 100644 --- a/src/java/org/apache/cassandra/cql/Cql.g +++ b/src/java/org/apache/cassandra/cql/Cql.g @@ -193,13 +193,17 @@ whereClause returns [WhereClause clause] * VALUES * (, , , ...) * (USING - * CONSISTENCY )?; + * CONSISTENCY + * (AND TIMESTAMP )? + * )?; * * Consistency level is set to ONE by default */ insertStatement returns [UpdateStatement expr] : { + Long timestamp = null; ConsistencyLevel cLevel = null; + Map columns = new HashMap(); List columnNames = new ArrayList(); @@ -209,12 +213,21 @@ insertStatement returns [UpdateStatement expr] '(' K_KEY ( ',' column_name=term { columnNames.add($column_name.item); } )+ ')' K_VALUES '(' key=term ( ',' column_value=term { columnValues.add($column_value.item); })+ ')' - ( K_USING K_CONSISTENCY K_LEVEL { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); } )? + ( usingClause[cLevel, timestamp] )? { - return new UpdateStatement($columnFamily.text, cLevel, columnNames, columnValues, Collections.singletonList(key)); + return new UpdateStatement($columnFamily.text, cLevel, columnNames, columnValues, Collections.singletonList(key), timestamp); } ; +usingClause[ConsistencyLevel cLevel, Long timestamp] + : K_USING usingClauseObjective[cLevel, timestamp] ( K_AND? usingClauseObjective[cLevel, timestamp] )? + ; + +usingClauseObjective[ConsistencyLevel cLevel, Long timestamp] + : K_CONSISTENCY K_LEVEL { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); } + | K_TIMESTAMP ts=INTEGER { timestamp = Long.valueOf($ts.text); } + ; + /** * BEGIN BATCH [USING CONSISTENCY ] * UPDATE SET name1 = value1 WHERE KEY = keyname1; @@ -241,14 +254,15 @@ insertStatement returns [UpdateStatement expr] */ batchStatement returns [BatchStatement expr] : { + Long timestamp = null; ConsistencyLevel cLevel = ConsistencyLevel.ONE; List statements = new ArrayList(); } - K_BEGIN K_BATCH ( K_USING K_CONSISTENCY K_LEVEL { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); } )? + K_BEGIN K_BATCH ( usingClause[cLevel, timestamp] )? s1=batchStatementObjective ';'? { statements.add(s1); } ( sN=batchStatementObjective ';'? { statements.add(sN); } )* K_APPLY K_BATCH endStmnt { - return new BatchStatement(statements, cLevel); + return new BatchStatement(statements, cLevel, timestamp); } ; @@ -261,8 +275,10 @@ batchStatementObjective returns [AbstractModification statement] /** * UPDATE * - * USING + * (USING * CONSISTENCY.ONE + * (AND TIMESTAMP )? + * )? * SET * name1 = value1, * name2 = value2 @@ -271,18 +287,19 @@ batchStatementObjective returns [AbstractModification statement] */ updateStatement returns [UpdateStatement expr] : { + Long timestamp = null; ConsistencyLevel cLevel = null; Map columns = new HashMap(); List keyList = null; } K_UPDATE columnFamily=( IDENT | STRING_LITERAL | INTEGER ) - (K_USING K_CONSISTENCY K_LEVEL { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); })? + ( usingClause[cLevel, timestamp] )? K_SET termPair[columns] (',' termPair[columns])* K_WHERE ( K_KEY '=' key=term { keyList = Collections.singletonList(key); } | K_KEY K_IN '(' keys=termList { keyList = $keys.items; } ')' ) { - return new UpdateStatement($columnFamily.text, cLevel, columns, keyList); + return new UpdateStatement($columnFamily.text, cLevel, columns, keyList, timestamp); } ; @@ -448,6 +465,7 @@ K_DROP: D R O P; K_PRIMARY: P R I M A R Y; K_INTO: I N T O; K_VALUES: V A L U E S; +K_TIMESTAMP: T I M E S T A M P; // Case-insensitive alpha characters fragment A: ('a'|'A'); diff --git a/src/java/org/apache/cassandra/cql/DeleteStatement.java b/src/java/org/apache/cassandra/cql/DeleteStatement.java index cc6437fe72..87461bee98 100644 --- a/src/java/org/apache/cassandra/cql/DeleteStatement.java +++ b/src/java/org/apache/cassandra/cql/DeleteStatement.java @@ -48,7 +48,7 @@ public class DeleteStatement extends AbstractModification public DeleteStatement(List columns, String columnFamily, ConsistencyLevel cLevel, List keys) { - super(columnFamily, cLevel); + super(columnFamily, cLevel, null); this.columns = columns; this.keys = keys; @@ -66,6 +66,12 @@ public class DeleteStatement extends AbstractModification /** {@inheritDoc} */ public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException + { + return prepareRowMutations(keyspace, clientState, null); + } + + /** {@inheritDoc} */ + public List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException { clientState.hasColumnFamilyAccess(columnFamily, Permission.WRITE); CFMetaData metadata = validateColumnFamily(keyspace, columnFamily, false); @@ -87,7 +93,7 @@ public class DeleteStatement extends AbstractModification { ByteBuffer columnName = column.getByteBuffer(comparator); validateColumnName(columnName); - rm.delete(new QueryPath(columnFamily, null, columnName), System.currentTimeMillis()); + rm.delete(new QueryPath(columnFamily, null, columnName), (timestamp == null) ? getTimestamp() : timestamp); } } diff --git a/src/java/org/apache/cassandra/cql/QueryProcessor.java b/src/java/org/apache/cassandra/cql/QueryProcessor.java index 9ae4a36024..40ea5c0798 100644 --- a/src/java/org/apache/cassandra/cql/QueryProcessor.java +++ b/src/java/org/apache/cassandra/cql/QueryProcessor.java @@ -530,10 +530,16 @@ public class QueryProcessor BatchStatement batch = (BatchStatement) statement.statement; for (AbstractModification up : batch.getStatements()) + { if (up.isSetConsistencyLevel()) throw new InvalidRequestException( "Consistency level must be set on the BATCH, not individual statements"); + if (batch.isSetTimestamp() && up.isSetTimestamp()) + throw new InvalidRequestException( + "Timestamp must be set either on BATCH or individual statements"); + } + try { StorageProxy.mutate(batch.getMutations(keyspace, clientState), batch.getConsistencyLevel()); diff --git a/src/java/org/apache/cassandra/cql/UpdateStatement.java b/src/java/org/apache/cassandra/cql/UpdateStatement.java index 0605d65d82..ef8e15c77b 100644 --- a/src/java/org/apache/cassandra/cql/UpdateStatement.java +++ b/src/java/org/apache/cassandra/cql/UpdateStatement.java @@ -56,10 +56,11 @@ public class UpdateStatement extends AbstractModification * @param cLevel the thrift consistency level * @param columns a map of column name/values pairs * @param keys the keys to update + * @param timestamp timestamp to use for mutation, if set to null then System.currentTimeMillis() */ - public UpdateStatement(String columnFamily, ConsistencyLevel cLevel, Map columns, List keys) + public UpdateStatement(String columnFamily, ConsistencyLevel cLevel, Map columns, List keys, Long timestamp) { - super(columnFamily, cLevel); + super(columnFamily, cLevel, timestamp); this.columns = columns; this.keys = keys; @@ -72,10 +73,11 @@ public class UpdateStatement extends AbstractModification * @param columnFamily column family name * @param columns a map of column name/values pairs * @param keys the keys to update + * @param timestamp timestamp to use for mutation, if set to null then System.currentTimeMillis() */ - public UpdateStatement(String columnFamily, Map columns, List keys) + public UpdateStatement(String columnFamily, Map columns, List keys, Long timestamp) { - this(columnFamily, null, columns, keys); + this(columnFamily, null, columns, keys, timestamp); } /** @@ -88,10 +90,16 @@ public class UpdateStatement extends AbstractModification * @param columnNames list of column names * @param columnValues list of column values (corresponds to names) * @param keys the keys to update + * @param timestamp timestamp to use for mutation, if set to null then System.currentTimeMillis() */ - public UpdateStatement(String columnFamily, ConsistencyLevel cLevel, List columnNames, List columnValues, List keys) + public UpdateStatement(String columnFamily, + ConsistencyLevel cLevel, + List columnNames, + List columnValues, + List keys, + Long timestamp) { - super(columnFamily, cLevel); + super(columnFamily, cLevel, timestamp); this.columnNames = columnNames; this.columnValues = columnValues; @@ -121,6 +129,12 @@ public class UpdateStatement extends AbstractModification /** {@inheritDoc} */ public List prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException + { + return prepareRowMutations(keyspace, clientState, null); + } + + /** {@inheritDoc} */ + public List prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException { List cfamsSeen = new ArrayList(); @@ -137,7 +151,7 @@ public class UpdateStatement extends AbstractModification for (Term key: keys) { - rowMutations.add(mutationForKey(keyspace, key.getByteBuffer(getKeyType(keyspace)), metadata)); + rowMutations.add(mutationForKey(keyspace, key.getByteBuffer(getKeyType(keyspace)), metadata, timestamp)); } return rowMutations; @@ -149,12 +163,13 @@ public class UpdateStatement extends AbstractModification * @param keyspace working keyspace * @param key key to change * @param metadata information about CF + * @param timestamp global timestamp to use for every key mutation * * @return row mutation * * @throws InvalidRequestException on the wrong request */ - private RowMutation mutationForKey(String keyspace, ByteBuffer key, CFMetaData metadata) throws InvalidRequestException + private RowMutation mutationForKey(String keyspace, ByteBuffer key, CFMetaData metadata, Long timestamp) throws InvalidRequestException { validateKey(key); @@ -167,7 +182,9 @@ public class UpdateStatement extends AbstractModification ByteBuffer colValue = column.getValue().getByteBuffer(getValueValidator(keyspace, colName)); validateColumn(metadata, colName, colValue); - rm.add(new QueryPath(columnFamily, null, colName), colValue, System.currentTimeMillis()); + rm.add(new QueryPath(columnFamily, null, colName), + colValue, + (timestamp == null) ? getTimestamp() : timestamp); } return rm; diff --git a/test/system/test_cql.py b/test/system/test_cql.py index e3abf425f2..2e4aab914f 100644 --- a/test/system/test_cql.py +++ b/test/system/test_cql.py @@ -746,6 +746,24 @@ class TestCql(ThriftTester): assert r[1] == "bVal", \ "unrecognized value '%s'" % r[1] + # using all 3 types of statements allowed in batch to test timestamp + cursor.execute(""" + BEGIN BATCH USING CONSISTENCY ONE AND TIMESTAMP 1303743619771318 + INSERT INTO StandardString1 (KEY, name) VALUES ('TimestampedUser4', 'sname') + UPDATE StandardString1 SET name = 'name here' WHERE KEY = 'TimestampedUser4' + DELETE name FROM StandardString1 WHERE KEY = 'TimestampedUser4' + APPLY BATCH + """) + + # BATCH should not allow setting individual timestamp when global timestamp is set + assert_raises(cql.ProgrammingError, + cursor.execute, + """ + BEGIN BATCH USING TIMESTAMP 1303743619771456 + UPDATE USING TIMESTAMP 1303743619771318 StandardString1 SET name = 'name here' WHERE KEY = 'TimestampedUser4' + APPLY BATCH + """) + assert_raises(cql.ProgrammingError, cursor.execute, """ @@ -780,3 +798,75 @@ class TestCql(ThriftTester): "unrecognized value '%s'" % r[1] assert r[2] == "p4ssw0rd", \ "unrecognized value '%s'" % r[1] + + def test_insert_with_timestamp(self): + "insert statement should support setting timestamp" + cursor = init() + cursor.compression = 'NONE' + + # insert to the StandardString1 + cursor.execute("INSERT INTO StandardString1 (KEY, name) VALUES ('TimestampedUser', 'name here') USING TIMESTAMP 1303743619771318") + + # try to read it + cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser'") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "name", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == "name here", \ + "unrecognized value '%s'" % r[1] + + # and INSERT with CONSISTENCY and TIMESTAMP together + cursor.execute("INSERT INTO StandardString1 (KEY, name) VALUES ('TimestampedUser1', 'name here') USING TIMESTAMP 1303743619771318 AND CONSISTENCY ONE") + + # try to read it + cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser1'") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "name", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == "name here", \ + "unrecognized value '%s'" % r[1] + + def test_update_with_timestamp(self): + "update statement should support setting timestamp" + cursor = init() + cursor.compression = 'NONE' + + # insert to the StandardString1 + cursor.execute("UPDATE StandardString1 USING TIMESTAMP 1303743619771318 SET name = 'name here' WHERE KEY = 'TimestampedUser2'") + + # try to read it + cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser2'") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "name", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == "name here", \ + "unrecognized value '%s'" % r[1] + + # and UPDATE with CONSISTENCY and TIMESTAMP together + cursor.execute("UPDATE StandardString1 USING CONSISTENCY ONE AND TIMESTAMP 1303743619771318 SET name = 'name here' WHERE KEY = 'TimestampedUser3'") + + # try to read it + cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser3'") + assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount + colnames = [col_d[0] for col_d in cursor.description] + + assert colnames[1] == "name", \ + "unrecognized name '%s'" % colnames[1] + + r = cursor.fetchone() + assert r[1] == "name here", \ + "unrecognized value '%s'" % r[1] + +