add CQL TTL support

patch by pyaskevich; reviewed by jbellis for CASSANDRA-2476

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

CHANGES.txt

@@ -1,12 +1,15 @@
 1.0-dev
+
+
+0.8.1
  * 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.1
  * add support for comparator parameters and a generic ReverseType
    (CASSANDRA-2355)
  * add CompositeType and DynamicCompositeType (CASSANDRA-2231)
+ * add CQL TTL support (CASSANDRA-2476)
+
 
 0.8.0-?
  * faster flushes and compaction from fixing excessively pessimistic 

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> [AND TIMESTAMP <timestamp>]];
+INSERT INTO <COLUMN FAMILY> (KEY, <col>, <col>, ...) VALUES (<key>, <val>, <val>, ...) [USING CONSISTENCY <LEVEL> [AND TIMESTAMP <timestamp>] [AND TTL <timeToLive>]];
 
 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> [AND TIMESTAMP <timestamp>]]
+UPDATE <COLUMN FAMILY> [USING <CONSISTENCY> [AND TIMESTAMP <timestamp>] [AND TTL <timeToLive>]]
         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.
@@ -111,6 +111,13 @@ UPDATE ... [USING TIMESTAMP <timestamp>] ...
 
 @UPDATE@ supports setting client-supplied optional timestamp for modification.
 
+h3. TTL
+
+bc.
+UPDATE ... [USING TTL <timeToLive>] ...
+
+@UPDATE@ supports setting time to live (TTL) for each of the columns in @UPDATE@ statement.
+
 h3. Specifying Columns and Row
 
 bc. 

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

@@ -33,17 +33,19 @@ public abstract class AbstractModification
     protected final String columnFamily;
     protected final ConsistencyLevel cLevel;
     protected final Long timestamp;
+    protected final int timeToLive;
 
-    public AbstractModification(String columnFamily, ConsistencyLevel cLevel)
+    public AbstractModification(String columnFamily, Attributes attrs)
     {
-        this(columnFamily, cLevel, null);
+        this(columnFamily, attrs.getConsistencyLevel(), attrs.getTimestamp(), attrs.getTimeToLive());
     }
 
-    public AbstractModification(String columnFamily, ConsistencyLevel cLevel, Long timestamp)
+    public AbstractModification(String columnFamily, ConsistencyLevel cLevel, Long timestamp, int timeToLive)
     {
         this.columnFamily = columnFamily;
         this.cLevel = cLevel;
         this.timestamp = timestamp;
+        this.timeToLive = timeToLive;
     }
 
     public String getColumnFamily()
@@ -76,6 +78,11 @@ public abstract class AbstractModification
         return timestamp != null;
     }
 
+    public int getTimeToLive()
+    {
+        return timeToLive;
+    }
+
     /**
      * Convert statement into a list of mutations to apply on the server
      *

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

@@ -0,0 +1,79 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.cassandra.cql;
+
+import org.apache.cassandra.thrift.ConsistencyLevel;
+
+/**
+ * Class to contain attributes for statements
+ */
+public class Attributes
+{
+    private ConsistencyLevel cLevel;
+    private Long timestamp;
+    private int timeToLive;
+
+    public Attributes()
+    {}
+
+    public Attributes(ConsistencyLevel cLevel, Long timestamp, int timeToLive)
+    {
+        this.cLevel = cLevel;
+        this.timestamp = timestamp;
+        this.timeToLive = timeToLive;
+    }
+
+    public ConsistencyLevel getConsistencyLevel()
+    {
+        return cLevel;
+    }
+
+    public void setConsistencyLevel(ConsistencyLevel cLevel)
+    {
+        this.cLevel = cLevel;
+    }
+
+    public Long getTimestamp()
+    {
+        return timestamp;
+    }
+
+    public void setTimestamp(Long timestamp)
+    {
+        this.timestamp = timestamp;
+    }
+
+    public int getTimeToLive()
+    {
+        return timeToLive;
+    }
+
+    public void setTimeToLive(int timeToLive)
+    {
+        this.timeToLive = timeToLive;
+    }
+
+    public String toString()
+    {
+        return String.format("Attributes(consistency=%s, timestamp=%s, timeToLive=%s)", cLevel, timestamp, timeToLive);
+    }
+
+}

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

@@ -43,18 +43,22 @@ public class BatchStatement
     // global timestamp to apply for each mutation
     protected final Long timestamp;
 
+    // global time to live
+    protected final int timeToLive;
+
     /**
      * Creates a new BatchStatement from a list of statements and a
      * Thrift consistency level.
      *
      * @param statements a list of UpdateStatements
-     * @param level Thrift consistency level enum
+     * @param attrs additional attributes for statement (CL, timestamp, timeToLive)
      */
-    public BatchStatement(List<AbstractModification> statements, ConsistencyLevel level, Long timestamp)
+    public BatchStatement(List<AbstractModification> statements, Attributes attrs)
     {
         this.statements = statements;
-        consistency = level;
+        this.consistency = attrs.getConsistencyLevel();
-        this.timestamp = timestamp;
+        this.timestamp = attrs.getTimestamp();
+        this.timeToLive = attrs.getTimeToLive();
     }
 
     public List<AbstractModification> getStatements()
@@ -67,6 +71,11 @@ public class BatchStatement
         return consistency;
     }
 
+    public int getTimeToLive()
+    {
+        return timeToLive;
+    }
+
     public List<RowMutation> getMutations(String keyspace, ClientState clientState) throws InvalidRequestException
     {
         List<RowMutation> batch = new LinkedList<RowMutation>();

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

@@ -201,9 +201,7 @@ whereClause returns [WhereClause clause]
  */
 insertStatement returns [UpdateStatement expr]
     : {
-          Long timestamp = null;
+          Attributes attrs = new Attributes();
-          ConsistencyLevel cLevel = null;
-
           Map<Term, Term> columns = new HashMap<Term, Term>();
 
           List<Term> columnNames  = new ArrayList<Term>();
@@ -213,19 +211,20 @@ 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); })+ ')'
-        ( usingClause[cLevel, timestamp] )?
+        ( usingClause[attrs] )?
       {
-          return new UpdateStatement($columnFamily.text, cLevel, columnNames, columnValues, Collections.singletonList(key), timestamp);
+          return new UpdateStatement($columnFamily.text, columnNames, columnValues, Collections.singletonList(key), attrs);
       }
     ;
 
-usingClause[ConsistencyLevel cLevel, Long timestamp]
+usingClause[Attributes attrs]
-    : K_USING usingClauseObjective[cLevel, timestamp] ( K_AND? usingClauseObjective[cLevel, timestamp] )?
+    : K_USING usingClauseObjective[attrs] ( K_AND? usingClauseObjective[attrs] )*
     ;
 
-usingClauseObjective[ConsistencyLevel cLevel, Long timestamp]
+usingClauseObjective[Attributes attrs]
-    : K_CONSISTENCY K_LEVEL  { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); }
+    : K_CONSISTENCY K_LEVEL  { attrs.setConsistencyLevel(ConsistencyLevel.valueOf($K_LEVEL.text)); }
-    | K_TIMESTAMP ts=INTEGER { timestamp = Long.valueOf($ts.text); }
+    | K_TIMESTAMP ts=INTEGER { attrs.setTimestamp(Long.valueOf($ts.text)); }
+    | K_TTL t=INTEGER        { attrs.setTimeToLive(Integer.parseInt($t.text)); }
     ;
 
 /**
@@ -254,15 +253,16 @@ usingClauseObjective[ConsistencyLevel cLevel, Long timestamp]
  */
 batchStatement returns [BatchStatement expr]
     : {
-          Long timestamp = null;
+          Attributes attrs = new Attributes();
-          ConsistencyLevel cLevel = ConsistencyLevel.ONE;
+          attrs.setConsistencyLevel(ConsistencyLevel.ONE);
+
           List<AbstractModification> statements = new ArrayList<AbstractModification>();
       }
-      K_BEGIN K_BATCH ( usingClause[cLevel, timestamp] )?
+      K_BEGIN K_BATCH ( usingClause[attrs] )?
           s1=batchStatementObjective ';'? { statements.add(s1); } ( sN=batchStatementObjective ';'? { statements.add(sN); } )*
       K_APPLY K_BATCH endStmnt
       {
-          return new BatchStatement(statements, cLevel, timestamp);
+          return new BatchStatement(statements, attrs);
       }
     ;
 
@@ -287,19 +287,18 @@ batchStatementObjective returns [AbstractModification statement]
  */
 updateStatement returns [UpdateStatement expr]
     : {
-          Long timestamp = null;
+          Attributes attrs = new Attributes();
-          ConsistencyLevel cLevel = null;
           Map<Term, Term> columns = new HashMap<Term, Term>();
           List<Term> keyList = null;
       }
       K_UPDATE columnFamily=( IDENT | STRING_LITERAL | INTEGER )
-          ( usingClause[cLevel, timestamp] )?
+          ( usingClause[attrs] )?
           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, timestamp);
+          return new UpdateStatement($columnFamily.text, columns, keyList, attrs);
       }
     ;
 
@@ -321,7 +320,8 @@ deleteStatement returns [DeleteStatement expr]
       }
       K_DELETE
           ( cols=termList { columnsList = $cols.items; })?
-          K_FROM columnFamily=( IDENT | STRING_LITERAL | INTEGER ) ( K_USING K_CONSISTENCY K_LEVEL )?
+          K_FROM columnFamily=( IDENT | STRING_LITERAL | INTEGER )
+          ( K_USING K_CONSISTENCY K_LEVEL { cLevel = ConsistencyLevel.valueOf($K_LEVEL.text); } )?
           K_WHERE ( K_KEY '=' key=term           { keyList = Collections.singletonList(key); }
                   | K_KEY K_IN '(' keys=termList { keyList = $keys.items; } ')'
                   )?
@@ -466,6 +466,7 @@ 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;
+K_TTL:         T T L;
 
 // 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, null);
+        super(columnFamily, cLevel, null, 0);
 
         this.columns = columns;
         this.keys = keys;

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

@@ -559,6 +559,9 @@ public class QueryProcessor
             case BATCH:
                 BatchStatement batch = (BatchStatement) statement.statement;
 
+                if (batch.getTimeToLive() != 0)
+                    throw new InvalidRequestException("Global TTL on the BATCH statement is not supported.");
+
                 for (AbstractModification up : batch.getStatements())
                 {
                     if (up.isSetConsistencyLevel())

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

@@ -53,32 +53,20 @@ public class UpdateStatement extends AbstractModification
      * level, and key term.
      * 
      * @param columnFamily column family name
-     * @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()
+     * @param attrs additional attributes for statement (CL, timestamp, timeToLive)
      */
-    public UpdateStatement(String columnFamily, ConsistencyLevel cLevel, Map<Term, Term> columns, List<Term> keys, Long timestamp)
+    public UpdateStatement(String columnFamily,
+                           Map<Term, Term> columns,
+                           List<Term> keys,
+                           Attributes attrs)
     {
-        super(columnFamily, cLevel, timestamp);
+        super(columnFamily, attrs);
 
         this.columns = columns;
         this.keys = keys;
     }
-
-    /**
-     * Creates a new UpdateStatement from a column family name, columns map,
-     * and key term.
-     * 
-     * @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, Long timestamp)
-    {
-        this(columnFamily, null, columns, keys, timestamp);
-    }
     
     /**
      * Creates a new UpdateStatement from a column family name, a consistency level,
@@ -86,20 +74,18 @@ public class UpdateStatement extends AbstractModification
      * alternate update format, <code>INSERT</code>.
      * 
      * @param columnFamily column family name
-     * @param cLevel the thrift consistency level
      * @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()
+     * @param attrs additional attributes for statement (CL, timestamp, timeToLive)
      */
     public UpdateStatement(String columnFamily,
-                           ConsistencyLevel cLevel,
                            List<Term> columnNames,
                            List<Term> columnValues,
                            List<Term> keys,
-                           Long timestamp)
+                           Attributes attrs)
     {
-        super(columnFamily, cLevel, timestamp);
+        super(columnFamily, attrs);
 
         this.columnNames = columnNames;
         this.columnValues = columnValues;
@@ -184,7 +170,8 @@ public class UpdateStatement extends AbstractModification
             validateColumn(metadata, colName, colValue);
             rm.add(new QueryPath(columnFamily, null, colName),
                    colValue,
-                   (timestamp == null) ? getTimestamp() : timestamp);
+                   (timestamp == null) ? getTimestamp() : timestamp,
+                   getTimeToLive());
         }
 
         return rm;
@@ -224,11 +211,13 @@ public class UpdateStatement extends AbstractModification
     
     public String toString()
     {
-        return String.format("UpdateStatement(columnFamily=%s, keys=%s, columns=%s, consistency=%s)",
+        return String.format("UpdateStatement(columnFamily=%s, keys=%s, columns=%s, consistency=%s, timestamp=%s, timeToLive=%s)",
                              columnFamily,
                              keys,
                              columns,
-                             cLevel);
+                             getConsistencyLevel(),
+                             timestamp,
+                             timeToLive);
     }
     
     public AbstractType<?> getKeyType(String keyspace)

test/system/test_cql.py

@@ -807,6 +807,15 @@ class TestCql(ThriftTester):
                         APPLY BATCH
                       """)
 
+        # BATCH should not allow setting global TTL
+        assert_raises(cql.ProgrammingError,
+                      cursor.execute,
+                      """
+                        BEGIN BATCH USING TTL 130374
+                          UPDATE StandardString1 SET name = 'name here' WHERE KEY = 'TimestampedUser4'
+                        APPLY BATCH
+                      """)
+
         assert_raises(cql.ProgrammingError,
                       cursor.execute,
                       """
@@ -842,7 +851,7 @@ class TestCql(ThriftTester):
             assert r[2] == "p4ssw0rd", \
                    "unrecognized value '%s'" % r[1]
 
-    def test_insert_with_timestamp(self):
+    def test_insert_with_timestamp_and_ttl(self):
         "insert statement should support setting timestamp"
         cursor = init()
         cursor.compression = 'NONE'
@@ -877,7 +886,50 @@ class TestCql(ThriftTester):
         assert r[1] == "name here", \
                "unrecognized value '%s'" % r[1]
 
-    def test_update_with_timestamp(self):
+        # and INSERT with TTL
+        cursor.execute("INSERT INTO StandardString1 (KEY, name) VALUES ('TimestampedUser2', 'name here') USING TTL 5678")
+
+        # 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 INSERT with CONSISTENCY, TIMESTAMP and TTL together
+        cursor.execute("INSERT INTO StandardString1 (KEY, name) VALUES ('TimestampedUser3', 'name here') USING TTL 4587 AND TIMESTAMP 1303743619771318 AND CONSISTENCY ONE")
+
+        # 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]
+
+        # and INSERT with TTL
+        cursor.execute("INSERT INTO StandardString1 (KEY, name) VALUES ('TimestampedUser14', 'name here') USING TTL 1 AND CONSISTENCY ONE")
+
+        # wait for column to expire
+        time.sleep(5)
+
+        # try to read it
+        cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser14'")
+        assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount
+
+        r = cursor.fetchone()
+        assert len(r) == 1, "expected 0 results, got %d" % len(r)
+
+    def test_update_with_timestamp_and_ttl(self):
         "update statement should support setting timestamp"
         cursor = init()
         cursor.compression = 'NONE'
@@ -911,3 +963,46 @@ class TestCql(ThriftTester):
         r = cursor.fetchone()
         assert r[1] == "name here", \
                "unrecognized value '%s'" % r[1]
+
+        # UPDATE with TTL
+        cursor.execute("UPDATE StandardString1 USING TTL 13030 SET name = 'name here' WHERE KEY = 'TimestampedUser4'")
+
+        # try to read it
+        cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser4'")
+        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]
+
+        # UPDATE with CONSISTENCY, TIMESTAMP and TTL together
+        cursor.execute("UPDATE StandardString1 USING CONSISTENCY ONE AND TIMESTAMP 1303743619771318 AND TTL 13037 SET name = 'name here' WHERE KEY = 'TimestampedUser5'")
+
+        # try to read it
+        cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser5'")
+        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]
+
+        # UPDATE with TTL
+        cursor.execute("UPDATE StandardString1 USING CONSISTENCY ONE TTL 1 SET name = 'name here' WHERE KEY = 'TimestampedUser6'")
+
+        # wait for column to expire
+        time.sleep(5)
+
+        # try to read it
+        cursor.execute("SELECT * FROM StandardString1 WHERE KEY = 'TimestampedUser6'")
+        assert cursor.rowcount == 1, "expected 1 results, got %d" % cursor.rowcount
+
+        r = cursor.fetchone()
+        assert len(r) == 1, "expected 0 results, got %d" % len(r)