diff --git a/CHANGES.txt b/CHANGES.txt
index 0318318de7..bd3514967b 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -10,6 +10,8 @@
  * Avoid creating empty and non cleaned writer during compaction (CASSANDRA-3616)
  * stop thrift service in shutdown hook so we can quiesce MessagingService
    (CASSANDRA-3335)
+ * (CQL) compaction_strategy_options and compression_parameters for
+   CREATE COLUMNFAMILY statement (CASSANDRA-3374)
 Merged from 0.8:
  * avoid logging (harmless) exception when GC takes < 1ms (CASSANDRA-3656)
  * prevent new nodes from thinking down nodes are up forever (CASSANDRA-3626)
diff --git a/doc/cql/CQL.textile b/doc/cql/CQL.textile
index 435c83a924..79c1e83c4b 100644
--- a/doc/cql/CQL.textile
+++ b/doc/cql/CQL.textile
@@ -488,9 +488,14 @@ bc(syntax).
 <createColumnFamilyStatement> ::= "CREATE" "COLUMNFAMILY" <name>
                                     "(" <term> <storageType> "PRIMARY" "KEY"
                                         ( "," <term> <storageType> )* ")"
-                                   ( "WITH" <identifier> "=" <cfOptionVal>
-                                     ( "AND" <identifier> "=" <cfOptionVal> )* )?
+                                   ( "WITH" <optionName> "=" <cfOptionVal>
+                                     ( "AND" <optionName> "=" <cfOptionVal> )* )?
                                 ;
+<optionName> ::= <identifier>
+               | <optionName> ":" <identifier>
+               | <optionName> ":" <integer>
+               ;
+
 <cfOptionVal> ::= <storageType>
                 | <identifier>
                 | <stringLiteral>
@@ -549,6 +554,8 @@ A number of optional keyword arguments can be supplied to control the configurat
 |row_cache_save_period_in_seconds|0|Number of seconds between saving row caches.|
 |key_cache_save_period_in_seconds|14400|Number of seconds between saving key caches.|
 |replicate_on_write|false| |
+|compaction_strategy_options|none|CompactionStrategy specific options such as "sstable_size_in_mb" for LeveledCompactionStrategy and "min_sstable_size" for SizeTieredCompactionStrategy|
+|compression_parameters|none|Compression parameters such as "sstable_compressor" and "chunk_length_kb"|
 
 h2. CREATE INDEX
 
diff --git a/src/java/org/apache/cassandra/cql/Cql.g b/src/java/org/apache/cassandra/cql/Cql.g
index 6971381fae..d016b0c3fe 100644
--- a/src/java/org/apache/cassandra/cql/Cql.g
+++ b/src/java/org/apache/cassandra/cql/Cql.g
@@ -377,8 +377,8 @@ createKeyspaceStatement returns [CreateKeyspaceStatement expr]
 createColumnFamilyStatement returns [CreateColumnFamilyStatement expr]
     : K_CREATE K_COLUMNFAMILY name=( IDENT | STRING_LITERAL | INTEGER ) { $expr = new CreateColumnFamilyStatement($name.text); }
       ( '(' createCfamColumns[expr] ( ',' createCfamColumns[expr] )* ')' )?
-      ( K_WITH prop1=IDENT '=' arg1=createCfamKeywordArgument { $expr.addProperty($prop1.text, $arg1.arg); }
-          ( K_AND propN=IDENT '=' argN=createCfamKeywordArgument { $expr.addProperty($propN.text, $argN.arg); } )*
+      ( K_WITH prop1=(COMPIDENT | IDENT) '=' arg1=createCfamKeywordArgument { $expr.addProperty($prop1.text, $arg1.arg); }
+          ( K_AND propN=(COMPIDENT | IDENT) '=' argN=createCfamKeywordArgument { $expr.addProperty($propN.text, $argN.arg); } )*
       )?
       endStmnt
     ;
diff --git a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
index f23c2ee993..1095a72143 100644
--- a/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
+++ b/src/java/org/apache/cassandra/cql/CreateColumnFamilyStatement.java
@@ -42,6 +42,7 @@ import org.apache.cassandra.db.marshal.TypeParser;
 import org.apache.cassandra.thrift.InvalidRequestException;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
+import org.apache.cassandra.io.compress.CompressionParameters;
 
 /** A <code>CREATE COLUMNFAMILY</code> parsed from a CQL query statement. */
 public class CreateColumnFamilyStatement
@@ -61,12 +62,16 @@ public class CreateColumnFamilyStatement
     private static final String KW_KEYCACHESAVEPERIODSECS = "key_cache_save_period_in_seconds";
     private static final String KW_REPLICATEONWRITE = "replicate_on_write";
     private static final String KW_ROW_CACHE_PROVIDER = "row_cache_provider";
+    private static final String KW_COMPACTION_STRATEGY_CLASS = "compaction_strategy_class";
     
     // Maps CQL short names to the respective Cassandra comparator/validator class names
     public  static final Map<String, String> comparators = new HashMap<String, String>();
     private static final Set<String> keywords = new HashSet<String>();
     private static final Set<String> obsoleteKeywords = new HashSet<String>();
-    
+
+    private static final String COMPACTION_OPTIONS_PREFIX = "compaction_strategy_options";
+    private static final String COMPRESSION_PARAMETERS_PREFIX = "compression_parameters";
+
     static
     {
         comparators.put("ascii", "AsciiType");
@@ -97,6 +102,7 @@ public class CreateColumnFamilyStatement
         keywords.add(KW_KEYCACHESAVEPERIODSECS);
         keywords.add(KW_REPLICATEONWRITE);
         keywords.add(KW_ROW_CACHE_PROVIDER);
+        keywords.add(KW_COMPACTION_STRATEGY_CLASS);
 
         obsoleteKeywords.add("memtable_throughput_in_mb");
         obsoleteKeywords.add("memtable_operations_in_millions");
@@ -108,6 +114,8 @@ public class CreateColumnFamilyStatement
     private final Map<String, String> properties = new HashMap<String, String>();
     private List<String> keyValidator = new ArrayList<String>();
     private ByteBuffer keyAlias = null;
+    private final Map<String, String> compactionStrategyOptions = new HashMap<String, String>();
+    private final Map<String, String> compressionParameters = new HashMap<String, String>();
 
     public CreateColumnFamilyStatement(String name)
     {
@@ -117,6 +125,34 @@ public class CreateColumnFamilyStatement
     /** Perform validation of parsed params */
     private void validate() throws InvalidRequestException
     {
+        // we need to remove parent:key = value pairs from the main properties
+        Set<String> propsToRemove = new HashSet<String>();
+
+        // check if we have compaction/compression options
+        for (String property : properties.keySet())
+        {
+            if (!property.contains(":"))
+                continue;
+
+            String key = property.split(":")[1];
+            String val = properties.get(property);
+
+            if (property.startsWith(COMPACTION_OPTIONS_PREFIX))
+            {
+                compactionStrategyOptions.put(key, val);
+                propsToRemove.add(property);
+            }
+
+            if (property.startsWith(COMPRESSION_PARAMETERS_PREFIX))
+            {
+                compressionParameters.put(key, val);
+                propsToRemove.add(property);
+            }
+        }
+
+        for (String property : propsToRemove)
+            properties.remove(property);
+
         // Column family name
         if (!name.matches("\\w+"))
             throw new InvalidRequestException(String.format("\"%s\" is not a valid column family name", name));
@@ -302,6 +338,8 @@ public class CreateColumnFamilyStatement
                    .keyValidator(TypeParser.parse(comparators.get(getKeyType())))
                    .rowCacheProvider(FBUtilities.newCacheProvider(getPropertyString(KW_ROW_CACHE_PROVIDER, CFMetaData.DEFAULT_ROW_CACHE_PROVIDER.getClass().getName())))
                    .keyAlias(keyAlias)
+                   .compactionStrategyOptions(compactionStrategyOptions)
+                   .compressionParameters(CompressionParameters.create(compressionParameters))
                    .validate();
         }
         catch (ConfigurationException e)