Revert "Revert "Materialized Views""

This reverts commit 24d185d.

CHANGES.txt

@@ -21,6 +21,7 @@
  * Populate TokenMetadata early during startup (CASSANDRA-9317)
  * Undeprecate cache recentHitRate (CASSANDRA-6591)
  * Add support for selectively varint encoding fields (CASSANDRA-9499, 9865)
+ * Materialized Views (CASSANDRA-6477)
 
 
 2.2.1

NEWS.txt

@@ -16,8 +16,18 @@ using the provided 'sstableupgrade' tool.
 3.0
 ===
 
+New features
+------------
+   - Materialized Views, which allow for server-side denormalization, is now
+     available. Materialized views provide an alternative to secondary indexes
+     for non-primary key queries, and perform much better for indexing high
+     cardinality columns. 
+     See http://www.datastax.com/dev/blog/new-in-cassandra-3-0-materialized-views 
+
 Upgrading
 ---------
+   - New write stages have been added for batchlog and materialized view mutations
+     you can set their size in cassandra.yaml
    - User defined functions are now executed in a sandbox.
      To use UDFs and UDAs, you have to enable them in cassandra.yaml.
    - New SSTable version 'la' with improved bloom-filter false-positive handling

conf/cassandra.yaml

@@ -342,6 +342,11 @@ seed_provider:
 concurrent_reads: 32
 concurrent_writes: 32
 concurrent_counter_writes: 32
+concurrent_batchlog_writes: 32
+
+# For materialized view writes, as there is a read involved, so this should
+# be limited by the less of concurrent reads or concurrent writes.
+concurrent_materialized_view_writes: 32
 
 # Maximum memory to use for pooling sstable buffers. Defaults to the smaller
 # of 1/4 of heap or 512MB. This pool is allocated off-heap, so is in addition

doc/cql3/CQL.textile

@@ -479,6 +479,61 @@ The @DROP INDEX@ statement is used to drop an existing secondary index. The argu
 
 If the index does not exists, the statement will return an error, unless @IF EXISTS@ is used in which case the operation is a no-op.
 
+
+h3(#createMVStmt). CREATE MATERIALIZED VIEW
+
+__Syntax:__
+
+bc(syntax)..
+<create-table-stmt> ::= CREATE MATERIALIZED VIEW ( IF NOT EXISTS )? <viewname> AS
+                          SELECT ( '(' <identifier> ( ',' <identifier> ) * ')' | '*' )
+                          FROM <tablename>
+                          WHERE ( <identifier> IS NOT NULL ( AND <identifier> IS NOT NULL )* )?
+                          PRIMARY KEY '(' <partition-key> ( ',' <identifier> )* ')'
+                          ( WITH <option> ( AND <option>)* )?
+__Sample:__
+
+bc(sample)..
+CREATE MATERIALIZED VIEW monkeySpecies_by_population AS
+    SELECT *
+    FROM monkeySpecies
+    WHERE population IS NOT NULL AND species IS NOT NULL
+    PRIMARY KEY (population, species)
+    WITH comment='Allow query by population instead of species';
+p.
+The @CREATE MATERIALIZED VIEW@ statement creates a new materialized view. Each such view is a set of _rows_ which corresponds to rows which are present in the underlying, or base, table specified in the @SELECT@ statement. A materialized view cannot be directly updated, but updates to the base table will cause corresponding updates in the view.
+
+Attempting to create an already existing materialized view will return an error unless the @IF NOT EXISTS@ option is used. If it is used, the statement will be a no-op if the materialized view already exists.
+
+h4(#createMVWhere). @WHERE <identifier> IS NOT NULL@
+
+The where clause is required to explicitly exclude all primary key columns' null values. Any row which contains null values in the primary key will not be present in the materialized view.
+
+h3(#alterMVStmt). ALTER MATERIALIZED VIEW
+
+__Syntax:__
+
+bc(syntax). <alter-materialized-view-stmt> ::= ALTER MATERIALIZED VIEW <viewname>
+                                                 WITH <option> ( AND <option> )*
+
+p.
+The @ALTER MATERIALIZED VIEW@ statement allows options to be update; these options are the same as <a href="#createTableOptions">@CREATE TABLE@'s options</a>.
+
+
+h3(#dropMVStmt). DROP MATERIALIZED VIEW
+
+__Syntax:__
+
+bc(syntax). <drop-materialized-stmt> ::= DROP MATERIALIZED VIEW ( IF EXISTS )? <tablename>
+
+__Sample:__
+
+bc(sample). DROP MATERIALIZED VIEW monkeySpecies_by_population;
+
+The @DROP MATERIALIZED VIEW@ statement is used to drop an existing materialized view.
+
+If the materialized view does not exists, the statement will return an error, unless @IF EXISTS@ is used in which case the operation is a no-op.
+
 h3(#createTypeStmt). CREATE TYPE
 
 __Syntax:__

pylib/cqlshlib/cql3handling.py

@@ -44,7 +44,8 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet):
         'function', 'aggregate', 'keyspace', 'schema', 'columnfamily', 'table', 'index', 'on', 'drop',
         'primary', 'into', 'values', 'date', 'time', 'timestamp', 'ttl', 'alter', 'add', 'type',
         'compact', 'storage', 'order', 'by', 'asc', 'desc', 'clustering',
-        'token', 'writetime', 'map', 'list', 'to', 'custom', 'if', 'not'
+        'token', 'writetime', 'map', 'list', 'to', 'custom', 'if', 'not',
+        'materialized', 'view'
     ))
 
     unreserved_keywords = set((
@@ -244,13 +245,15 @@ def dequote_value(cqlword):
 <schemaChangeStatement> ::= <createKeyspaceStatement>
                           | <createColumnFamilyStatement>
                           | <createIndexStatement>
+                          | <createMaterializedViewStatement>
                           | <createUserTypeStatement>
                           | <createFunctionStatement>
                           | <createAggregateStatement>
                           | <createTriggerStatement>
                           | <dropKeyspaceStatement>
                           | <dropColumnFamilyStatement>
                           | <dropIndexStatement>
+                          | <dropMaterializedViewStatement>
                           | <dropUserTypeStatement>
                           | <dropFunctionStatement>
                           | <dropAggregateStatement>
@@ -1090,6 +1093,11 @@ def create_cf_composite_primary_key_comma_completer(ctxt, cass):
                                ( "USING" <stringLiteral> ( "WITH" "OPTIONS" "=" <mapLiteral> )? )?
                          ;
 
+<createMaterializedViewStatement> ::= "CREATE" "MATERIALIZED" "VIEW" ("IF" "NOT" "EXISTS")? <columnFamilyName>?
+                                      "AS" <selectStatement>
+                                      "PRIMARY" "KEY" <pkDef>
+                                    ;
+
 <createUserTypeStatement> ::= "CREATE" "TYPE" ( ks=<nonSystemKeyspaceName> dot="." )? typename=<cfOrKsName> "(" newcol=<cident> <storageType>
                                 ( "," [newcolname]=<cident> <storageType> )*
                             ")"
@@ -1146,6 +1154,9 @@ def create_index_col_completer(ctxt, cass):
 <dropIndexStatement> ::= "DROP" "INDEX" ("IF" "EXISTS")? idx=<indexName>
                        ;
 
+<dropMaterializedViewStatement> ::= "DROP" "MATERIALIZED" "VIEW" ("IF" "EXISTS")? mv=<columnFamilyName>
+                                  ;
+
 <dropUserTypeStatement> ::= "DROP" "TYPE" ut=<userTypeName>
                           ;
 

src/java/org/apache/cassandra/concurrent/Stage.java

@@ -27,6 +27,8 @@ public enum Stage
     READ,
     MUTATION,
     COUNTER_MUTATION,
+    BATCHLOG_MUTATION,
+    MATERIALIZED_VIEW_MUTATION,
     GOSSIP,
     REQUEST_RESPONSE,
     ANTI_ENTROPY,
@@ -60,6 +62,8 @@ public String getJmxType()
                 return "internal";
             case MUTATION:
             case COUNTER_MUTATION:
+            case BATCHLOG_MUTATION:
+            case MATERIALIZED_VIEW_MUTATION:
             case READ:
             case REQUEST_RESPONSE:
             case READ_REPAIR:

src/java/org/apache/cassandra/concurrent/StageManager.java

@@ -47,6 +47,8 @@ public class StageManager
     {
         stages.put(Stage.MUTATION, multiThreadedLowSignalStage(Stage.MUTATION, getConcurrentWriters()));
         stages.put(Stage.COUNTER_MUTATION, multiThreadedLowSignalStage(Stage.COUNTER_MUTATION, getConcurrentCounterWriters()));
+        stages.put(Stage.BATCHLOG_MUTATION, multiThreadedLowSignalStage(Stage.BATCHLOG_MUTATION, getConcurrentBatchlogWriters()));
+        stages.put(Stage.MATERIALIZED_VIEW_MUTATION, multiThreadedLowSignalStage(Stage.MATERIALIZED_VIEW_MUTATION, getConcurrentMaterializedViewWriters()));
         stages.put(Stage.READ, multiThreadedLowSignalStage(Stage.READ, getConcurrentReaders()));
         stages.put(Stage.REQUEST_RESPONSE, multiThreadedLowSignalStage(Stage.REQUEST_RESPONSE, FBUtilities.getAvailableProcessors()));
         stages.put(Stage.INTERNAL_RESPONSE, multiThreadedStage(Stage.INTERNAL_RESPONSE, FBUtilities.getAvailableProcessors()));