# Copyright DataStax, Inc.

#

# Licensed 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.

from binascii import unhexlify

from bisect import bisect_right

from collections import defaultdict, Mapping

from functools import total_ordering

from hashlib import md5

from itertools import islice, cycle

import json

import logging

import re

import six

from six.moves import zip

import sys

from threading import RLock

import struct

import random

murmur3 = None

try:

from cassandra.murmur3 import murmur3

except ImportError as e:

pass

from cassandra import SignatureDescriptor, ConsistencyLevel, InvalidRequest, Unauthorized

import cassandra.cqltypes as types

from cassandra.encoder import Encoder

from cassandra.marshal import varint_unpack

from cassandra.protocol import QueryMessage

from cassandra.query import dict_factory, bind_params, Statement

from cassandra.util import OrderedDict

from cassandra.hosts import HostDistance

log = logging.getLogger(__name__)

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',

'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',

'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',

'tinyint', 'to', 'token', 'trigger', 'truncate', 'ttl', 'tuple', 'type', 'unlogged', 'update', 'use', 'user',

'users', 'using', 'uuid', 'values', 'varchar', 'varint', 'view', 'where', 'with', 'writetime'

))

"""

Set of keywords in CQL.

Derived from .../cassandra/src/java/org/apache/cassandra/cql3/Cql.g

"""

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',

'frozen', 'function', 'functions', 'inet', 'initcond', 'input', 'int', 'json', 'key', 'keys', 'keyspaces',

'language', 'list', 'login', 'map', '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'

))

"""

Set of unreserved keywords in CQL.

Derived from .../cassandra/src/java/org/apache/cassandra/cql3/Cql.g

"""

cql_keywords_reserved = cql_keywords - cql_keywords_unreserved

"""

Set of reserved keywords in CQL.

"""

_encoder = Encoder()

class Metadata(object):

"""

Holds a representation of the cluster schema and topology.

"""

cluster_name = None

""" The string name of the cluster. """

keyspaces = None

"""

A map from keyspace names to matching :class:`~.KeyspaceMetadata` instances.

"""

partitioner = None

"""

The string name of the partitioner for the cluster.

"""

token_map = None

""" A :class:`~.TokenMap` instance describing the ring topology. """

def __init__(self):

self.keyspaces = {}

self._hosts = {}

self._hosts_lock = RLock()

def export_schema_as_string(self):

"""

Returns a string that can be executed as a query in order to recreate

the entire schema. The string is formatted to be human readable.

"""

return "\n\n".join(ks.export_as_string() for ks in self.keyspaces.values())

def refresh(self, connection, timeout, target_type=None, change_type=None, **kwargs):

server_version = self.get_host(connection.host).release_version

parser = get_schema_parser(connection, server_version, timeout)

if not target_type:

self._rebuild_all(parser)

return

tt_lower = target_type.lower()

try:

parse_method = getattr(parser, 'get_' + tt_lower)

meta = parse_method(self.keyspaces, **kwargs)

if meta:

update_method = getattr(self, '_update_' + tt_lower)

update_method(meta)

else:

drop_method = getattr(self, '_drop_' + tt_lower)

drop_method(**kwargs)

except AttributeError:

raise ValueError("Unknown schema target_type: '%s'" % target_type)

def _rebuild_all(self, parser):

current_keyspaces = set()

for keyspace_meta in parser.get_all_keyspaces():

current_keyspaces.add(keyspace_meta.name)

old_keyspace_meta = self.keyspaces.get(keyspace_meta.name, None)

self.keyspaces[keyspace_meta.name] = keyspace_meta

if old_keyspace_meta:

self._keyspace_updated(keyspace_meta.name)

else:

self._keyspace_added(keyspace_meta.name)

# remove not-just-added keyspaces

removed_keyspaces = [name for name in self.keyspaces.keys()

if name not in current_keyspaces]

self.keyspaces = dict((name, meta) for name, meta in self.keyspaces.items()

if name in current_keyspaces)

for ksname in removed_keyspaces:

self._keyspace_removed(ksname)

def _update_keyspace(self, keyspace_meta, new_user_types=None):

ks_name = keyspace_meta.name

old_keyspace_meta = self.keyspaces.get(ks_name, None)

self.keyspaces[ks_name] = keyspace_meta

if old_keyspace_meta:

keyspace_meta.tables = old_keyspace_meta.tables

keyspace_meta.user_types = new_user_types if new_user_types is not None else old_keyspace_meta.user_types

keyspace_meta.indexes = old_keyspace_meta.indexes

keyspace_meta.functions = old_keyspace_meta.functions

keyspace_meta.aggregates = old_keyspace_meta.aggregates

keyspace_meta.views = old_keyspace_meta.views

if (keyspace_meta.replication_strategy != old_keyspace_meta.replication_strategy):

self._keyspace_updated(ks_name)

else:

self._keyspace_added(ks_name)

def _drop_keyspace(self, keyspace):

if self.keyspaces.pop(keyspace, None):

self._keyspace_removed(keyspace)

def _update_table(self, meta):

try:

keyspace_meta = self.keyspaces[meta.keyspace_name]

# this is unfortunate, but protocol v4 does not differentiate

# between events for tables and views. <parser>.get_table will

# return one or the other based on the query results.

# Here we deal with that.

if isinstance(meta, TableMetadata):

keyspace_meta._add_table_metadata(meta)

else:

keyspace_meta._add_view_metadata(meta)

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _drop_table(self, keyspace, table):

try:

keyspace_meta = self.keyspaces[keyspace]

keyspace_meta._drop_table_metadata(table) # handles either table or view

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _update_type(self, type_meta):

try:

self.keyspaces[type_meta.keyspace].user_types[type_meta.name] = type_meta

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _drop_type(self, keyspace, type):

try:

self.keyspaces[keyspace].user_types.pop(type, None)

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _update_function(self, function_meta):

try:

self.keyspaces[function_meta.keyspace].functions[function_meta.signature] = function_meta

except KeyError:

# can happen if keyspace disappears while processing async event

pass

def _drop_function(self, keyspace, function):

try:

self.keyspaces[keyspace].functions.pop(function.signature, None)

except KeyError:

pass

def _update_aggregate(self, aggregate_meta):

try:

self.keyspaces[aggregate_meta.keyspace].aggregates[aggregate_meta.signature] = aggregate_meta

except KeyError:

pass

def _drop_aggregate(self, keyspace, aggregate):

try:

self.keyspaces[keyspace].aggregates.pop(aggregate.signature, None)

except KeyError:

pass

def _keyspace_added(self, ksname):

if self.token_map:

self.token_map.rebuild_keyspace(ksname, build_if_absent=False)

def _keyspace_updated(self, ksname):

if self.token_map:

self.token_map.rebuild_keyspace(ksname, build_if_absent=False)

def _keyspace_removed(self, ksname):

if self.token_map:

self.token_map.remove_keyspace(ksname)

def rebuild_token_map(self, partitioner, token_map):

"""

Rebuild our view of the topology from fresh rows from the

system topology tables.

For internal use only.

"""

self.partitioner = partitioner

if partitioner.endswith('RandomPartitioner'):

token_class = MD5Token

elif partitioner.endswith('Murmur3Partitioner'):

token_class = Murmur3Token

elif partitioner.endswith('ByteOrderedPartitioner'):

token_class = BytesToken

else:

self.token_map = None

return

token_to_host_owner = {}

ring = []

for host, token_strings in six.iteritems(token_map):

for token_string in token_strings:

token = token_class.from_string(token_string)

ring.append(token)

token_to_host_owner[token] = host

all_tokens = sorted(ring)

self.token_map = TokenMap(

token_class, token_to_host_owner, all_tokens, self)

def get_replicas(self, keyspace, key):

"""

Returns a list of :class:`.Host` instances that are replicas for a given

partition key.

"""

t = self.token_map

if not t:

return []

try:

return t.get_replicas(keyspace, t.token_class.from_key(key))

except NoMurmur3:

return []

def can_support_partitioner(self):

if self.partitioner.endswith('Murmur3Partitioner') and murmur3 is None:

return False

else:

return True

def add_or_return_host(self, host):

"""

Returns a tuple (host, new), where ``host`` is a Host

instance, and ``new`` is a bool indicating whether

the host was newly added.

"""

with self._hosts_lock:

try:

return self._hosts[host.address], False

except KeyError:

self._hosts[host.address] = host

return host, True

def remove_host(self, host):

with self._hosts_lock:

return bool(self._hosts.pop(host.address, False))

def get_host(self, address):

return self._hosts.get(address)

def all_hosts(self):

"""

Returns a list of all known :class:`.Host` instances in the cluster.

"""

with self._hosts_lock:

return list(self._hosts.values())

REPLICATION_STRATEGY_CLASS_PREFIX = "org.apache.cassandra.locator."

def trim_if_startswith(s, prefix):

if s.startswith(prefix):

return s[len(prefix):]

return s

_replication_strategies = {}

class ReplicationStrategyTypeType(type):

def __new__(metacls, name, bases, dct):

dct.setdefault('name', name)

cls = type.__new__(metacls, name, bases, dct)

if not name.startswith('_'):

_replication_strategies[name] = cls

return cls

@six.add_metaclass(ReplicationStrategyTypeType)

class _ReplicationStrategy(object):

options_map = None

@classmethod

def create(cls, strategy_class, options_map):

if not strategy_class:

return None

strategy_name = trim_if_startswith(strategy_class, REPLICATION_STRATEGY_CLASS_PREFIX)

rs_class = _replication_strategies.get(strategy_name, None)

if rs_class is None:

rs_class = _UnknownStrategyBuilder(strategy_name)

_replication_strategies[strategy_name] = rs_class

try:

rs_instance = rs_class(options_map)

except Exception as exc:

log.warning("Failed creating %s with options %s: %s", strategy_name, options_map, exc)

return None

return rs_instance

def make_token_replica_map(self, token_to_host_owner, ring):

raise NotImplementedError()

def export_for_schema(self):

raise NotImplementedError()

ReplicationStrategy = _ReplicationStrategy

class _UnknownStrategyBuilder(object):

def __init__(self, name):

self.name = name

def __call__(self, options_map):

strategy_instance = _UnknownStrategy(self.name, options_map)

return strategy_instance

class _UnknownStrategy(ReplicationStrategy):

def __init__(self, name, options_map):

self.name = name

self.options_map = options_map.copy() if options_map is not None else dict()

self.options_map['class'] = self.name

def __eq__(self, other):

return (isinstance(other, _UnknownStrategy)

and self.name == other.name

and self.options_map == other.options_map)

def export_for_schema(self):

"""

Returns a string version of these replication options which are

suitable for use in a CREATE KEYSPACE statement.

"""

if self.options_map:

return dict((str(key), str(value)) for key, value in self.options_map.items())

return "{'class': '%s'}" % (self.name, )

def make_token_replica_map(self, token_to_host_owner, ring):

return {}

class SimpleStrategy(ReplicationStrategy):

replication_factor = None

"""

The replication factor for this keyspace.

"""

def __init__(self, options_map):

try:

self.replication_factor = int(options_map['replication_factor'])

except Exception:

raise ValueError("SimpleStrategy requires an integer 'replication_factor' option")

def make_token_replica_map(self, token_to_host_owner, ring):

replica_map = {}

for i in range(len(ring)):

j, hosts = 0, list()

while len(hosts) < self.replication_factor and j < len(ring):

token = ring[(i + j) % len(ring)]

host = token_to_host_owner[token]

if host not in hosts:

hosts.append(host)

j += 1

replica_map[ring[i]] = hosts

return replica_map

def export_for_schema(self):

"""

Returns a string version of these replication options which are

suitable for use in a CREATE KEYSPACE statement.

"""

return "{'class': 'SimpleStrategy', 'replication_factor': '%d'}" \

% (self.replication_factor,)

def __eq__(self, other):

if not isinstance(other, SimpleStrategy):

return False

return self.replication_factor == other.replication_factor

class NetworkTopologyStrategy(ReplicationStrategy):

dc_replication_factors = None

"""

A map of datacenter names to the replication factor for that DC.

"""

def __init__(self, dc_replication_factors):

self.dc_replication_factors = dict(

(str(k), int(v)) for k, v in dc_replication_factors.items())

def make_token_replica_map(self, token_to_host_owner, ring):

dc_rf_map = dict((dc, int(rf))

for dc, rf in self.dc_replication_factors.items() if rf > 0)

# build a map of DCs to lists of indexes into `ring` for tokens that

# belong to that DC

dc_to_token_offset = defaultdict(list)

dc_racks = defaultdict(set)

hosts_per_dc = defaultdict(set)

for i, token in enumerate(ring):

host = token_to_host_owner[token]

dc_to_token_offset[host.datacenter].append(i)

if host.datacenter and host.rack:

dc_racks[host.datacenter].add(host.rack)

hosts_per_dc[host.datacenter].add(host)

# A map of DCs to an index into the dc_to_token_offset value for that dc.

# This is how we keep track of advancing around the ring for each DC.

dc_to_current_index = defaultdict(int)

replica_map = defaultdict(list)

for i in range(len(ring)):

replicas = replica_map[ring[i]]

# go through each DC and find the replicas in that DC

for dc in dc_to_token_offset.keys():

if dc not in dc_rf_map:

continue

# advance our per-DC index until we're up to at least the

# current token in the ring

token_offsets = dc_to_token_offset[dc]

index = dc_to_current_index[dc]

num_tokens = len(token_offsets)

while index < num_tokens and token_offsets[index] < i:

index += 1

dc_to_current_index[dc] = index

replicas_remaining = dc_rf_map[dc]

replicas_this_dc = 0

skipped_hosts = []

racks_placed = set()

racks_this_dc = dc_racks[dc]

hosts_this_dc = len(hosts_per_dc[dc])

for token_offset in islice(cycle(token_offsets), index, index + num_tokens):

host = token_to_host_owner[ring[token_offset]]

if replicas_remaining == 0 or replicas_this_dc == hosts_this_dc:

break

if host in replicas:

continue

if host.rack in racks_placed and len(racks_placed) < len(racks_this_dc):

skipped_hosts.append(host)

continue

replicas.append(host)

replicas_this_dc += 1

replicas_remaining -= 1

racks_placed.add(host.rack)

if len(racks_placed) == len(racks_this_dc):

for host in skipped_hosts:

if replicas_remaining == 0:

break

replicas.append(host)

replicas_remaining -= 1

del skipped_hosts[:]

return replica_map

def export_for_schema(self):

"""

Returns a string version of these replication options which are

suitable for use in a CREATE KEYSPACE statement.

"""

ret = "{'class': 'NetworkTopologyStrategy'"

for dc, repl_factor in sorted(self.dc_replication_factors.items()):

ret += ", '%s': '%d'" % (dc, repl_factor)

return ret + "}"

def __eq__(self, other):

if not isinstance(other, NetworkTopologyStrategy):

return False

return self.dc_replication_factors == other.dc_replication_factors

class LocalStrategy(ReplicationStrategy):

def __init__(self, options_map):

pass

def make_token_replica_map(self, token_to_host_owner, ring):

return {}

def export_for_schema(self):

"""

Returns a string version of these replication options which are

suitable for use in a CREATE KEYSPACE statement.

"""

return "{'class': 'LocalStrategy'}"

def __eq__(self, other):

return isinstance(other, LocalStrategy)

class KeyspaceMetadata(object):

"""

A representation of the schema for a single keyspace.

"""

name = None

""" The string name of the keyspace. """

durable_writes = True

"""

A boolean indicating whether durable writes are enabled for this keyspace

or not.

"""

replication_strategy = None

"""

A :class:`.ReplicationStrategy` subclass object.

"""

tables = None

"""

A map from table names to instances of :class:`~.TableMetadata`.

"""

indexes = None

"""

A dict mapping index names to :class:`.IndexMetadata` instances.

"""

user_types = None

"""

A map from user-defined type names to instances of :class:`~cassandra.metadata.UserType`.

.. versionadded:: 2.1.0

"""

functions = None

"""

A map from user-defined function signatures to instances of :class:`~cassandra.metadata.Function`.

.. versionadded:: 2.6.0

"""

aggregates = None

"""

A map from user-defined aggregate signatures to instances of :class:`~cassandra.metadata.Aggregate`.

.. versionadded:: 2.6.0

"""

views = None

"""

A dict mapping view names to :class:`.MaterializedViewMetadata` instances.

"""

_exc_info = None

""" set if metadata parsing failed """

def __init__(self, name, durable_writes, strategy_class, strategy_options):

self.name = name

self.durable_writes = durable_writes

self.replication_strategy = ReplicationStrategy.create(strategy_class, strategy_options)

self.tables = {}

self.indexes = {}

self.user_types = {}

self.functions = {}

self.aggregates = {}

self.views = {}

def export_as_string(self):

"""

Returns a CQL query string that can be used to recreate the entire keyspace,

including user-defined types and tables.

"""

cql = "\n\n".join([self.as_cql_query() + ';']

+ self.user_type_strings()

+ [f.export_as_string() for f in self.functions.values()]

+ [a.export_as_string() for a in self.aggregates.values()]

+ [t.export_as_string() for t in self.tables.values()])

if self._exc_info:

import traceback

ret = "/*\nWarning: Keyspace %s is incomplete because of an error processing metadata.\n" % \

(self.name)

for line in traceback.format_exception(*self._exc_info):

ret += line

ret += "\nApproximate structure, for reference:\n(this should not be used to reproduce this schema)\n\n%s\n*/" % cql

return ret

return cql

def as_cql_query(self):

"""

Returns a CQL query string that can be used to recreate just this keyspace,

not including user-defined types and tables.

"""

ret = "CREATE KEYSPACE %s WITH replication = %s " % (

protect_name(self.name),

self.replication_strategy.export_for_schema())

return ret + (' AND durable_writes = %s' % ("true" if self.durable_writes else "false"))

def user_type_strings(self):

user_type_strings = []

user_types = self.user_types.copy()

keys = sorted(user_types.keys())

for k in keys:

if k in user_types:

self.resolve_user_types(k, user_types, user_type_strings)

return user_type_strings

def resolve_user_types(self, key, user_types, user_type_strings):

user_type = user_types.pop(key)

for type_name in user_type.field_types:

for sub_type in types.cql_types_from_string(type_name):

if sub_type in user_types:

self.resolve_user_types(sub_type, user_types, user_type_strings)

user_type_strings.append(user_type.export_as_string())

def _add_table_metadata(self, table_metadata):

old_indexes = {}

old_meta = self.tables.get(table_metadata.name, None)

if old_meta:

# views are not queried with table, so they must be transferred to new

table_metadata.views = old_meta.views

# indexes will be updated with what is on the new metadata

old_indexes = old_meta.indexes

# note the intentional order of add before remove

# this makes sure the maps are never absent something that existed before this update

for index_name, index_metadata in six.iteritems(table_metadata.indexes):

self.indexes[index_name] = index_metadata

for index_name in (n for n in old_indexes if n not in table_metadata.indexes):

self.indexes.pop(index_name, None)

self.tables[table_metadata.name] = table_metadata

def _drop_table_metadata(self, table_name):

table_meta = self.tables.pop(table_name, None)

if table_meta:

for index_name in table_meta.indexes:

self.indexes.pop(index_name, None)

for view_name in table_meta.views:

self.views.pop(view_name, None)

return

# we can't tell table drops from views, so drop both

# (name is unique among them, within a keyspace)

view_meta = self.views.pop(table_name, None)

if view_meta:

try:

self.tables[view_meta.base_table_name].views.pop(table_name, None)

except KeyError:

pass

def _add_view_metadata(self, view_metadata):

try:

self.tables[view_metadata.base_table_name].views[view_metadata.name] = view_metadata

self.views[view_metadata.name] = view_metadata

except KeyError:

pass

class UserType(object):

"""

A user defined type, as created by ``CREATE TYPE`` statements.

User-defined types were introduced in Cassandra 2.1.

.. versionadded:: 2.1.0

"""

keyspace = None

"""

The string name of the keyspace in which this type is defined.

"""

name = None

"""

The name of this type.

"""

field_names = None

"""

An ordered list of the names for each field in this user-defined type.

"""

field_types = None

"""

An ordered list of the types for each field in this user-defined type.

"""

def __init__(self, keyspace, name, field_names, field_types):

self.keyspace = keyspace

self.name = name

# non-frozen collections can return None

self.field_names = field_names or []

self.field_types = field_types or []

def as_cql_query(self, formatted=False):

"""

Returns a CQL query that can be used to recreate this type.

If `formatted` is set to :const:`True`, extra whitespace will

be added to make the query more readable.

"""

ret = "CREATE TYPE %s.%s (%s" % (

protect_name(self.keyspace),

protect_name(self.name),

"\n" if formatted else "")

if formatted:

field_join = ",\n"

padding = " "

else:

field_join = ", "

padding = ""

fields = []

for field_name, field_type in zip(self.field_names, self.field_types):

fields.append("%s %s" % (protect_name(field_name), field_type))

ret += field_join.join("%s%s" % (padding, field) for field in fields)

ret += "\n)" if formatted else ")"

return ret

def export_as_string(self):

return self.as_cql_query(formatted=True) + ';'

class Aggregate(object):

"""

A user defined aggregate function, as created by ``CREATE AGGREGATE`` statements.

Aggregate functions were introduced in Cassandra 2.2

.. versionadded:: 2.6.0

"""

keyspace = None

"""

The string name of the keyspace in which this aggregate is defined

"""

name = None

"""

The name of this aggregate

"""

argument_types = None

"""

An ordered list of the types for each argument to the aggregate

"""

final_func = None

"""

Name of a final function

"""

initial_condition = None

"""

Initial condition of the aggregate

"""

return_type = None

"""

Return type of the aggregate

"""

state_func = None

"""

Name of a state function

"""

state_type = None

"""

Type of the aggregate state

"""

def __init__(self, keyspace, name, argument_types, state_func,

state_type, final_func, initial_condition, return_type):

self.keyspace = keyspace

self.name = name

self.argument_types = argument_types

self.state_func = state_func

self.state_type = state_type

self.final_func = final_func

self.initial_condition = initial_condition

self.return_type = return_type

def as_cql_query(self, formatted=False):

"""

Returns a CQL query that can be used to recreate this aggregate.

If `formatted` is set to :const:`True`, extra whitespace will

be added to make the query more readable.

"""

sep = '\n ' if formatted else ' '

keyspace = protect_name(self.keyspace)

name = protect_name(self.name)

type_list = ', '.join(self.argument_types)

state_func = protect_name(self.state_func)

state_type = self.state_type

ret = "CREATE AGGREGATE %(keyspace)s.%(name)s(%(type_list)s)%(sep)s" \

"SFUNC %(state_func)s%(sep)s" \

"STYPE %(state_type)s" % locals()

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 ''

return ret

def export_as_string(self):

return self.as_cql_query(formatted=True) + ';'

@property

def signature(self):

return SignatureDescriptor.format_signature(self.name, self.argument_types)

class Function(object):

"""

A user defined function, as created by ``CREATE FUNCTION`` statements.

User-defined functions were introduced in Cassandra 2.2

.. versionadded:: 2.6.0

"""

keyspace = None

"""

The string name of the keyspace in which this function is defined

"""

name = None

"""

The name of this function

"""

argument_types = None

"""

An ordered list of the types for each argument to the function

"""

argument_names = None

"""

An ordered list of the names of each argument to the function

"""

return_type = None

"""

Return type of the function

"""

language = None

"""

Language of the function body

"""

body = None

"""

Function body string

"""

called_on_null_input = None

"""

Flag indicating whether this function should be called for rows with null values

(convenience function to avoid handling nulls explicitly if the result will just be null)

"""

def __init__(self, keyspace, name, argument_types, argument_names,

return_type, language, body, called_on_null_input):

self.keyspace = keyspace

self.name = name

self.argument_types = argument_types

# argument_types (frozen<list<>>) will always be a list

# argument_name is not frozen in C* < 3.0 and may return None

self.argument_names = argument_names or []

self.return_type = return_type

self.language = language

self.body = body

self.called_on_null_input = called_on_null_input

def as_cql_query(self, formatted=False):

"""

Returns a CQL query that can be used to recreate this function.

If `formatted` is set to :const:`True`, extra whitespace will

be added to make the query more readable.

"""

sep = '\n ' if formatted else ' '

keyspace = protect_name(self.keyspace)

name = protect_name(self.name)

arg_list = ', '.join(["%s %s" % (protect_name(n), t)

for n, t in zip(self.argument_names, self.argument_types)])

typ = self.return_type

lang = self.language

body = self.body

on_null = "CALLED" if self.called_on_null_input else "RETURNS NULL"

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" \

"LANGUAGE %(lang)s%(sep)s" \

"AS $$%(body)s$$" % locals()

def export_as_string(self):

return self.as_cql_query(formatted=True) + ';'

@property

def signature(self):

return SignatureDescriptor.format_signature(self.name, self.argument_types)

class TableMetadata(object):

"""

A representation of the schema for a single table.

"""

keyspace_name = None

""" String name of this Table's keyspace """

name = None

""" The string name of the table. """