Merged metadata.py

Author: aboudreault

cassandra/metadata.py

@@ -17,7 +17,6 @@
 from collections import defaultdict
 from functools import total_ordering
 from hashlib import md5
-from itertools import islice, cycle
 import json
 import logging
 import re
@@ -50,10 +49,10 @@
 cql_keywords = set((
     'add', 'aggregate', 'all', 'allow', 'alter', 'and', 'apply', 'as', 'asc', 'ascii', 'authorize', 'batch', 'begin',
     'bigint', 'blob', 'boolean', 'by', 'called', 'clustering', 'columnfamily', 'compact', 'contains', 'count',
-    'counter', 'create', 'custom', 'date', 'decimal', 'delete', 'desc', 'describe', 'distinct', 'double', 'drop',
+    'counter', 'create', 'custom', 'date', 'decimal', 'delete', 'desc', 'describe', 'deterministic', 'distinct', 'double', 'drop',
     'entries', 'execute', 'exists', 'filtering', 'finalfunc', 'float', 'from', 'frozen', 'full', 'function',
     'functions', 'grant', 'if', 'in', 'index', 'inet', 'infinity', 'initcond', 'input', 'insert', 'int', 'into', 'is', 'json',
-    'key', 'keys', 'keyspace', 'keyspaces', 'language', 'limit', 'list', 'login', 'map', 'materialized', 'modify', 'nan', 'nologin',
+    'key', 'keys', 'keyspace', 'keyspaces', 'language', 'limit', 'list', 'login', 'map', 'materialized', 'modify', 'monotonic', 'nan', 'nologin',
     'norecursive', 'nosuperuser', 'not', 'null', 'of', 'on', 'options', 'or', 'order', 'password', 'permission',
     'permissions', 'primary', 'rename', 'replace', 'returns', 'revoke', 'role', 'roles', 'schema', 'select', 'set',
     'sfunc', 'smallint', 'static', 'storage', 'stype', 'superuser', 'table', 'text', 'time', 'timestamp', 'timeuuid',
@@ -68,9 +67,9 @@
 
 cql_keywords_unreserved = set((
     'aggregate', 'all', 'as', 'ascii', 'bigint', 'blob', 'boolean', 'called', 'clustering', 'compact', 'contains',
-    'count', 'counter', 'custom', 'date', 'decimal', 'distinct', 'double', 'exists', 'filtering', 'finalfunc', 'float',
+    'count', 'counter', 'custom', 'date', 'decimal', 'deterministic', 'distinct', 'double', 'exists', 'filtering', 'finalfunc', 'float',
     'frozen', 'function', 'functions', 'inet', 'initcond', 'input', 'int', 'json', 'key', 'keys', 'keyspaces',
-    'language', 'list', 'login', 'map', 'nologin', 'nosuperuser', 'options', 'password', 'permission', 'permissions',
+    'language', 'list', 'login', 'map', 'monotonic', 'nologin', 'nosuperuser', 'options', 'password', 'permission', 'permissions',
     'returns', 'role', 'roles', 'sfunc', 'smallint', 'static', 'storage', 'stype', 'superuser', 'text', 'time',
     'timestamp', 'timeuuid', 'tinyint', 'trigger', 'ttl', 'tuple', 'type', 'user', 'users', 'uuid', 'values', 'varchar',
     'varint', 'writetime'
@@ -125,7 +124,8 @@ def export_schema_as_string(self):
     def refresh(self, connection, timeout, target_type=None, change_type=None, **kwargs):
 
         server_version = self.get_host(connection.endpoint).release_version
-        parser = get_schema_parser(connection, server_version, timeout)
+        dse_version = self.get_host(connection.endpoint).dse_version
+        parser = get_schema_parser(connection, server_version, dse_version, timeout)
 
         if not target_type:
             self._rebuild_all(parser)
@@ -661,7 +661,7 @@ class KeyspaceMetadata(object):
     virtual = False
     """
     A boolean indicating if this is a virtual keyspace or not. Always ``False``
-    for clusters running pre-4.0 versions of Cassandra.
+    for clusters running Cassandra pre-4.0 and DSE pre-6.7 versions.
 
     .. versionadded:: 3.15
     """
@@ -893,8 +893,15 @@ class Aggregate(object):
     Type of the aggregate state
     """
 
+    deterministic = None
+    """
+    Flag indicating if this function is guaranteed to produce the same result
+    for a particular input and state. This is available only with DSE >=6.0.
+    """
+
     def __init__(self, keyspace, name, argument_types, state_func,
-                 state_type, final_func, initial_condition, return_type):
+                 state_type, final_func, initial_condition, return_type,
+                 deterministic):
         self.keyspace = keyspace
         self.name = name
         self.argument_types = argument_types
@@ -903,6 +910,7 @@ def __init__(self, keyspace, name, argument_types, state_func,
         self.final_func = final_func
         self.initial_condition = initial_condition
         self.return_type = return_type
+        self.deterministic = deterministic
 
     def as_cql_query(self, formatted=False):
         """
@@ -923,6 +931,7 @@ def as_cql_query(self, formatted=False):
 
         ret += ''.join((sep, 'FINALFUNC ', protect_name(self.final_func))) if self.final_func else ''
         ret += ''.join((sep, 'INITCOND ', self.initial_condition)) if self.initial_condition is not None else ''
+        ret += '{}DETERMINISTIC'.format(sep) if self.deterministic else ''
 
         return ret
 
@@ -984,8 +993,27 @@ class Function(object):
     (convenience function to avoid handling nulls explicitly if the result will just be null)
     """
 
+    deterministic = None
+    """
+    Flag indicating if this function is guaranteed to produce the same result
+    for a particular input. This is available only for DSE >=6.0.
+    """
+
+    monotonic = None
+    """
+    Flag indicating if this function is guaranteed to increase or decrease
+    monotonically on any of its arguments. This is available only for DSE >=6.0.
+    """
+
+    monotonic_on = None
+    """
+    A list containing the argument or arguments over which this function is
+    monotonic. This is available only for DSE >=6.0.
+    """
+
     def __init__(self, keyspace, name, argument_types, argument_names,
-                 return_type, language, body, called_on_null_input):
+                 return_type, language, body, called_on_null_input,
+                 deterministic, monotonic, monotonic_on):
         self.keyspace = keyspace
         self.name = name
         self.argument_types = argument_types
@@ -996,6 +1024,9 @@ def __init__(self, keyspace, name, argument_types, argument_names,
         self.language = language
         self.body = body
         self.called_on_null_input = called_on_null_input
+        self.deterministic = deterministic
+        self.monotonic = monotonic
+        self.monotonic_on = monotonic_on
 
     def as_cql_query(self, formatted=False):
         """
@@ -1012,10 +1043,25 @@ def as_cql_query(self, formatted=False):
         lang = self.language
         body = self.body
         on_null = "CALLED" if self.called_on_null_input else "RETURNS NULL"
+        deterministic_token = ('DETERMINISTIC{}'.format(sep)
+                               if self.deterministic else
+                               '')
+        monotonic_tokens = ''  # default for nonmonotonic function
+        if self.monotonic:
+            # monotonic on all arguments; ignore self.monotonic_on
+            monotonic_tokens = 'MONOTONIC{}'.format(sep)
+        elif self.monotonic_on:
+            # if monotonic == False and monotonic_on is nonempty, we know that
+            # monotonicity was specified with MONOTONIC ON <arg>, so there's
+            # exactly 1 value there
+            monotonic_tokens = 'MONOTONIC ON {}{}'.format(self.monotonic_on[0],
+                                                          sep)
 
         return "CREATE FUNCTION %(keyspace)s.%(name)s(%(arg_list)s)%(sep)s" \
                "%(on_null)s ON NULL INPUT%(sep)s" \
                "RETURNS %(typ)s%(sep)s" \
+               "%(deterministic_token)s" \
+               "%(monotonic_tokens)s" \
                "LANGUAGE %(lang)s%(sep)s" \
                "AS $$%(body)s$$" % locals()
 
@@ -1102,7 +1148,7 @@ def primary_key(self):
     virtual = False
     """
     A boolean indicating if this is a virtual table or not. Always ``False``
-    for clusters running pre-4.0 versions of Cassandra.
+    for clusters running Cassandra pre-4.0 and DSE pre-6.7 versions.
 
     .. versionadded:: 3.15
     """
@@ -1733,6 +1779,9 @@ def _query_build_rows(self, query_string, build_func):
 
 
 class SchemaParserV22(_SchemaParser):
+    """
+    For C* 2.2+
+    """
     _SELECT_KEYSPACES = "SELECT * FROM system.schema_keyspaces"
     _SELECT_COLUMN_FAMILIES = "SELECT * FROM system.schema_columnfamilies"
     _SELECT_COLUMNS = "SELECT * FROM system.schema_columns"
@@ -1884,10 +1933,14 @@ def _build_user_type(cls, usertype_row):
     @classmethod
     def _build_function(cls, function_row):
         return_type = cls._schema_type_to_cql(function_row['return_type'])
+        deterministic = function_row.get('deterministic', False)
+        monotonic = function_row.get('monotonic', False)
+        monotonic_on = function_row.get('monotonic_on', ())
         return Function(function_row['keyspace_name'], function_row['function_name'],
                         function_row[cls._function_agg_arument_type_col], function_row['argument_names'],
                         return_type, function_row['language'], function_row['body'],
-                        function_row['called_on_null_input'])
+                        function_row['called_on_null_input'],
+                        deterministic, monotonic, monotonic_on)
 
     @classmethod
     def _build_aggregate(cls, aggregate_row):
@@ -1899,7 +1952,8 @@ def _build_aggregate(cls, aggregate_row):
         return_type = cls._schema_type_to_cql(aggregate_row['return_type'])
         return Aggregate(aggregate_row['keyspace_name'], aggregate_row['aggregate_name'],
                          aggregate_row['signature'], aggregate_row['state_func'], state_type,
-                         aggregate_row['final_func'], initial_condition, return_type)
+                         aggregate_row['final_func'], initial_condition, return_type,
+                         aggregate_row.get('deterministic', False))
 
     def _build_table_metadata(self, row, col_rows=None, trigger_rows=None):
         keyspace_name = row["keyspace_name"]
@@ -2230,6 +2284,9 @@ def _schema_type_to_cql(type_string):
 
 
 class SchemaParserV3(SchemaParserV22):
+    """
+    For C* 3.0+
+    """
     _SELECT_KEYSPACES = "SELECT * FROM system_schema.keyspaces"
     _SELECT_TABLES = "SELECT * FROM system_schema.tables"
     _SELECT_COLUMNS = "SELECT * FROM system_schema.columns"
@@ -2316,7 +2373,8 @@ def _build_keyspace_metadata_internal(row):
     def _build_aggregate(aggregate_row):
         return Aggregate(aggregate_row['keyspace_name'], aggregate_row['aggregate_name'],
                          aggregate_row['argument_types'], aggregate_row['state_func'], aggregate_row['state_type'],
-                         aggregate_row['final_func'], aggregate_row['initcond'], aggregate_row['return_type'])
+                         aggregate_row['final_func'], aggregate_row['initcond'], aggregate_row['return_type'],
+                         aggregate_row.get('deterministic', False))
 
     def _build_table_metadata(self, row, col_rows=None, trigger_rows=None, index_rows=None, virtual=False):
         keyspace_name = row["keyspace_name"]
@@ -2498,6 +2556,14 @@ def _schema_type_to_cql(type_string):
         return type_string
 
 
+class SchemaParserDSE60(SchemaParserV3):
+    """
+    For DSE 6.0+
+    """
+    recognized_table_options = (SchemaParserV3.recognized_table_options +
+                                ("nodesync",))
+
+
 class SchemaParserV4(SchemaParserV3):
 
     recognized_table_options = tuple(
@@ -2629,10 +2695,25 @@ def _build_keyspace_metadata_internal(row):
         return super(SchemaParserV4, SchemaParserV4)._build_keyspace_metadata_internal(row)
 
 
+class SchemaParserDSE67(SchemaParserV4):
+    """
+    For DSE 6.7+
+    """
+    recognized_table_options = (SchemaParserV4.recognized_table_options +
+                                ("nodesync",))
+
+
 class TableMetadataV3(TableMetadata):
+    """
+    For C* 3.0+. `option_maps` take a superset of map names, so if  nothing
+    changes structurally, new option maps can just be appended to the list.
+    """
     compaction_options = {}
 
-    option_maps = ['compaction', 'compression', 'caching']
+    option_maps = [
+        'compaction', 'compression', 'caching',
+        'nodesync'  # added DSE 6.0
+    ]
 
     @property
     def is_cql_compatible(self):
@@ -2768,11 +2849,18 @@ def export_as_string(self):
         return self.as_cql_query(formatted=True) + ";"
 
 
-def get_schema_parser(connection, server_version, timeout):
+def get_schema_parser(connection, server_version, dse_version, timeout):
     version = Version(server_version)
+    if dse_version:
+        v = Version(dse_version)
+        if v >= Version('6.7.0'):
+            return SchemaParserDSE67(connection, timeout)
+        elif v >= Version('6.0.0'):
+            return SchemaParserDSE60(connection, timeout)
+
     if version >= Version('4.0.0'):
         return SchemaParserV4(connection, timeout)
-    if version >= Version('3.0.0'):
+    elif version >= Version('3.0.0'):
         return SchemaParserV3(connection, timeout)
     else:
         # we could further specialize by version. Right now just refactoring the
@@ -2791,6 +2879,14 @@ def _cql_from_cass_type(cass_type):
         return cass_type.cql_parameterized_type()
 
 
+class RLACTableExtension(RegisteredTableExtension):
+    name = "DSE_RLACA"
+
+    @classmethod
+    def after_table_cql(cls, table_meta, ext_key, ext_blob):
+        return "RESTRICT ROWS ON %s.%s USING %s;" % (protect_name(table_meta.keyspace_name),
+                                                     protect_name(table_meta.name),
+                                                     protect_name(ext_blob.decode('utf-8')))
 NO_VALID_REPLICA = object()