add timestamp support to cqlINSERT,UPDATE,and BATCH

patch by Pavel Yaskevich; reviewed by jbellis for CASSANDRA-2555

git-svn-id: https://svn.apache.org/repos/asf/cassandra/trunk@1097511 13f79535-47bb-0310-9956-ffa450edef68

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-?

doc/cql/CQL.textile

@@ -74,7 +74,7 @@ h2. INSERT
 _Synopsis:_
 
 bc.
-INSERT INTO <COLUMN FAMILY> (KEY, <col>, <col>, ...) VALUES (<key>, <val>, <val>, ...) [USING CONSISTENCY <LEVEL];
+INSERT INTO <COLUMN FAMILY> (KEY, <col>, <col>, ...) VALUES (<key>, <val>, <val>, ...) [USING CONSISTENCY <LEVEL> [AND TIMESTAMP <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 <COLUMN FAMILY> [USING <CONSISTENCY>]
+UPDATE <COLUMN FAMILY> [USING <CONSISTENCY> [AND TIMESTAMP <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 <CONSISTENCY>] ...
 
 Following the column family identifier is an optional "consistency level specification":#consistency.
 
+h3. Timestamp
+
+bc.
+UPDATE ... [USING TIMESTAMP <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 <LEVEL>]
+BATCH BEGIN BATCH [USING CONSISTENCY <LEVEL> [AND TIMESTAMP <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._

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<RowMutation> 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<RowMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp)
+            throws org.apache.cassandra.thrift.InvalidRequestException;
 }

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<AbstractModification> statements, ConsistencyLevel level)
+    public BatchStatement(List<AbstractModification> statements, ConsistencyLevel level, Long timestamp)
     {
         this.statements = statements;
         consistency = level;
+        this.timestamp = timestamp;
     }
 
     public List<AbstractModification> 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()
     {

src/java/org/apache/cassandra/cql/Cql.g

@@ -193,13 +193,17 @@ whereClause returns [WhereClause clause]
  * VALUES
  *    (<key>, <value>, <value>, ...)
  * (USING
- *    CONSISTENCY <level>)?;
+ *    CONSISTENCY <level>
+ *   (AND TIMESTAMP <long>)?
+ * )?;
  *
  * Consistency level is set to ONE by default
  */
 insertStatement returns [UpdateStatement expr]
     : {
+          Long timestamp = null;
           ConsistencyLevel cLevel = null;
+
           Map<Term, Term> columns = new HashMap<Term, Term>();
 
           List<Term> columnNames  = new ArrayList<Term>();
@@ -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 <LVL>]
  *   UPDATE <CF> 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<AbstractModification> statements = new ArrayList<AbstractModification>();
       }
-      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
  *     <CF>
- * USING
+ * (USING
  *     CONSISTENCY.ONE
+ *    (AND TIMESTAMP <long>)?
+ * )?
  * SET
  *     name1 = value1,
  *     name2 = value2
@@ -271,18 +287,19 @@ batchStatementObjective returns [AbstractModification statement]
  */
 updateStatement returns [UpdateStatement expr]
     : {
+          Long timestamp = null;
           ConsistencyLevel cLevel = null;
           Map<Term, Term> columns = new HashMap<Term, Term>();
           List<Term> 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');

src/java/org/apache/cassandra/cql/DeleteStatement.java

@@ -48,7 +48,7 @@ public class DeleteStatement extends AbstractModification
     
     public DeleteStatement(List<Term> columns, String columnFamily, ConsistencyLevel cLevel, List<Term> 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<RowMutation> prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException
+    {
+        return prepareRowMutations(keyspace, clientState, null);
+    }
+
+    /** {@inheritDoc} */
+    public List<RowMutation> 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);
                 }
             }
 

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());

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<Term, Term> columns, List<Term> keys)
+    public UpdateStatement(String columnFamily, ConsistencyLevel cLevel, Map<Term, Term> columns, List<Term> 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<Term, Term> columns, List<Term> keys)
+    public UpdateStatement(String columnFamily, Map<Term, Term> columns, List<Term> 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<Term> columnNames, List<Term> columnValues, List<Term> keys)
+    public UpdateStatement(String columnFamily,
+                           ConsistencyLevel cLevel,
+                           List<Term> columnNames,
+                           List<Term> columnValues,
+                           List<Term> 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<RowMutation> prepareRowMutations(String keyspace, ClientState clientState) throws InvalidRequestException
+    {
+        return prepareRowMutations(keyspace, clientState, null);
+    }
+
+    /** {@inheritDoc} */
+    public List<RowMutation> prepareRowMutations(String keyspace, ClientState clientState, Long timestamp) throws InvalidRequestException
     {
         List<String> cfamsSeen = new ArrayList<String>();
 
@@ -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;

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]
+
+