Merge branch 'cassandra-2.1' into trunk

Conflicts:
	CHANGES.txt

CHANGES.txt

@@ -141,7 +141,10 @@
  * Force config client mode in CQLSSTableWriter (CASSANDRA-8281)
  * Fix sstableupgrade throws exception (CASSANDRA-8688)
 Merged from 2.0:
- * Fix SSTableSimpleUnsortedWriter ConcurrentModificationException (CASSANDRA-8619)
+ * Prevent non-zero default_time_to_live on tables with counters
+   (CASSANDRA-8678)
+ * Fix SSTableSimpleUnsortedWriter ConcurrentModificationException
+   (CASSANDRA-8619)
  * Round up time deltas lower than 1ms in BulkLoader (CASSANDRA-8645)
  * Add batch remove iterator to ABSC (CASSANDRA-8414, 8666)
  * Round up time deltas lower than 1ms in BulkLoader (CASSANDRA-8645)

src/java/org/apache/cassandra/cql3/statements/AlterTableStatement.java

@@ -260,6 +260,10 @@ public class AlterTableStatement extends SchemaAlteringStatement
                     throw new InvalidRequestException(String.format("ALTER TABLE WITH invoked, but no parameters found"));
 
                 cfProps.validate();
+
+                if (meta.isCounter() && cfProps.getDefaultTimeToLive() > 0)
+                    throw new InvalidRequestException("Cannot set default_time_to_live on a table with counters");
+
                 cfProps.applyToCFMetadata(cfm);
                 break;
             case RENAME:

src/java/org/apache/cassandra/cql3/statements/CFPropDefs.java

@@ -166,6 +166,11 @@ public class CFPropDefs extends PropertyDefinitions
         return options;
     }
 
+    public Integer getDefaultTimeToLive() throws SyntaxException
+    {
+        return getInt(KW_DEFAULT_TIME_TO_LIVE, 0);
+    }
+
     public void applyToCFMetadata(CFMetaData cfm) throws ConfigurationException, SyntaxException
     {
         if (hasProperty(KW_COMMENT))

src/java/org/apache/cassandra/cql3/statements/CreateTableStatement.java

@@ -210,6 +210,7 @@ public class CreateTableStatement extends SchemaAlteringStatement
 
             CreateTableStatement stmt = new CreateTableStatement(cfName, properties, ifNotExists, staticColumns);
 
+            boolean hasCounters = false;
             Map<ByteBuffer, CollectionType> definedMultiCellCollections = null;
             for (Map.Entry<ColumnIdentifier, CQL3Type.Raw> entry : definitions.entrySet())
             {
@@ -221,6 +222,9 @@ public class CreateTableStatement extends SchemaAlteringStatement
                         definedMultiCellCollections = new HashMap<>();
                     definedMultiCellCollections.put(id.bytes, (CollectionType) pt.getType());
                 }
+                else if (entry.getValue().isCounter())
+                    hasCounters = true;
+
                 stmt.columns.put(id, pt.getType()); // we'll remove what is not a column below
             }
 
@@ -228,6 +232,8 @@ public class CreateTableStatement extends SchemaAlteringStatement
                 throw new InvalidRequestException("No PRIMARY KEY specifed (exactly one required)");
             else if (keyAliases.size() > 1)
                 throw new InvalidRequestException("Multiple PRIMARY KEYs specifed (exactly one required)");
+            else if (hasCounters && properties.getDefaultTimeToLive() > 0)
+                throw new InvalidRequestException("Cannot set default_time_to_live on a table with counters");
 
             List<ColumnIdentifier> kAliases = keyAliases.get(0);