text: # Programmer: Navraj Chohan <nlake44@gmail.com>
  
  """
   Cassandra Interface for AppScale
  """
  import cassandra
  import datetime
  import logging
  import struct
  import sys
  import time
  import uuid
  
  from appscale.common import appscale_info
  from appscale.common.constants import SCHEMA_CHANGE_TIMEOUT
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from cassandra.cluster import Cluster
  from cassandra.concurrent import execute_concurrent
  from cassandra.query import BatchStatement
  from cassandra.query import ConsistencyLevel
  from cassandra.query import SimpleStatement
  from cassandra.query import ValueSequence
  from .large_batch import (FailedBatch,
                            LargeBatch)
  from .retry_policies import (BASIC_RETRIES,
                               NO_RETRIES)
  from .. import dbconstants
  from ..dbconstants import AppScaleDBConnectionError
  from ..dbconstants import Operations
  from ..dbconstants import TxnActions
  from ..dbinterface import AppDBInterface
  from ..utils import create_key
  from ..utils import get_write_time
  from ..utils import tx_partition
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.api.taskqueue import taskqueue_service_pb
  from google.appengine.datastore import entity_pb
  
  
  # The directory Cassandra is installed to.
  CASSANDRA_INSTALL_DIR = '/opt/cassandra'
  
  # Full path for the nodetool binary.
  NODE_TOOL = '{}/cassandra/bin/nodetool'.format(CASSANDRA_INSTALL_DIR)
  
  # The keyspace used for all tables
  KEYSPACE = "Keyspace1"
  
  # Cassandra watch name.
  CASSANDRA_MONIT_WATCH_NAME = "cassandra"
  
  # The number of times to retry connecting to Cassandra.
  INITIAL_CONNECT_RETRIES = 20
  
  # The data layout version that the datastore expects.
  EXPECTED_DATA_VERSION = 1.0
  
  # The metadata key for the data layout version.
  VERSION_INFO_KEY = 'version'
  
  # The metadata key used to indicate the state of the indexes.
  INDEX_STATE_KEY = 'index_state'
  
  # The metadata key indicating that the database has been primed.
  PRIMED_KEY = 'primed'
  
  # The size in bytes that a batch must be to use the batches table.
  LARGE_BATCH_THRESHOLD = 5 << 10
  
  
  def batch_size(batch):
    """ Calculates the size of a batch.
  
    Args:
      batch: A list of dictionaries representing mutations.
    Returns:
      An integer specifying the size in bytes of the batch.
    """
    size = 0
    for mutation in batch:
      size += len(mutation['key'])
      if 'values' in mutation:
        for value in mutation['values'].values():
          size += len(value)
    return size
  
  
  class ThriftColumn(object):
    """ Columns created by default with thrift interface. """
    KEY = 'key'
    COLUMN_NAME = 'column1'
    VALUE = 'value'
  
  
  class IndexStates(object):
    """ Possible states for datastore indexes. """
    CLEAN = 'clean'
    SCRUB_IN_PROGRESS = 'scrub_in_progress'
  
  
  class DatastoreProxy(AppDBInterface):
    """ 
      Cassandra implementation of the AppDBInterface
    """
    def __init__(self, log_level=logging.INFO, hosts=None):
      """
      Constructor.
      """
      class_name = self.__class__.__name__
      self.logger = logging.getLogger(class_name)
      self.logger.setLevel(log_level)
      self.logger.info('Starting {}'.format(class_name))
  
      if hosts is not None:
        self.hosts = hosts
      else:
        self.hosts = appscale_info.get_db_ips()
  
      remaining_retries = INITIAL_CONNECT_RETRIES
      while True:
        try:
          self.cluster = Cluster(self.hosts, default_retry_policy=BASIC_RETRIES)
          self.session = self.cluster.connect(KEYSPACE)
          break
        except cassandra.cluster.NoHostAvailable as connection_error:
          remaining_retries -= 1
          if remaining_retries < 0:
            raise connection_error
          time.sleep(3)
  
      self.session.default_consistency_level = ConsistencyLevel.QUORUM
      self.prepared_statements = {}
  
    def close(self):
      """ Close all sessions and connections to Cassandra. """
      self.cluster.shutdown()
  
    def batch_get_entity(self, table_name, row_keys, column_names):
      """
      Takes in batches of keys and retrieves their corresponding rows.
      
      Args:
        table_name: The table to access
        row_keys: A list of keys to access
        column_names: A list of columns to access
      Returns:
        A dictionary of rows and columns/values of those rows. The format 
        looks like such: {key:{column_name:value,...}}
      Raises:
        TypeError: If an argument passed in was not of the expected type.
        AppScaleDBConnectionError: If the batch_get could not be performed due to
          an error with Cassandra.
      """
      if not isinstance(table_name, str): raise TypeError("Expected a str")
      if not isinstance(column_names, list): raise TypeError("Expected a list")
      if not isinstance(row_keys, list): raise TypeError("Expected a list")
  
      row_keys_bytes = [bytearray(row_key) for row_key in row_keys]
  
      statement = 'SELECT * FROM "{table}" '\
                  'WHERE {key} IN %s and {column} IN %s'.format(
                    table=table_name,
                    key=ThriftColumn.KEY,
                    column=ThriftColumn.COLUMN_NAME,
                  )
      query = SimpleStatement(statement, retry_policy=BASIC_RETRIES)
      parameters = (ValueSequence(row_keys_bytes), ValueSequence(column_names))
  
      try:
        results = self.session.execute(query, parameters=parameters)
  
        results_dict = {row_key: {} for row_key in row_keys}
        for (key, column, value) in results:
          if key not in results_dict:
            results_dict[key] = {}
          results_dict[key][column] = value
  
        return results_dict
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception during batch_get_entity'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def batch_put_entity(self, table_name, row_keys, column_names, cell_values,
                         ttl=None):
      """
      Allows callers to store multiple rows with a single call. A row can 
      have multiple columns and values with them. We refer to each row as 
      an entity.
     
      Args: 
        table_name: The table to mutate
        row_keys: A list of keys to store on
        column_names: A list of columns to mutate
        cell_values: A dict of key/value pairs
        ttl: The number of seconds to keep the row.
      Raises:
        TypeError: If an argument passed in was not of the expected type.
        AppScaleDBConnectionError: If the batch_put could not be performed due to
          an error with Cassandra.
      """
      if not isinstance(table_name, str):
        raise TypeError("Expected a str")
      if not isinstance(column_names, list):
        raise TypeError("Expected a list")
      if not isinstance(row_keys, list):
        raise TypeError("Expected a list")
      if not isinstance(cell_values, dict):
        raise TypeError("Expected a dict")
  
      insert_str = """
        INSERT INTO "{table}" ({key}, {column}, {value})
        VALUES (?, ?, ?)
      """.format(table=table_name,
                 key=ThriftColumn.KEY,
                 column=ThriftColumn.COLUMN_NAME,
                 value=ThriftColumn.VALUE)
  
      if ttl is not None:
        insert_str += 'USING TTL {}'.format(ttl)
  
      statement = self.session.prepare(insert_str)
  
      statements_and_params = []
      for row_key in row_keys:
        for column in column_names:
          params = (bytearray(row_key), column,
                    bytearray(cell_values[row_key][column]))
          statements_and_params.append((statement, params))
  
      try:
        execute_concurrent(self.session, statements_and_params,
                           raise_on_first_error=True)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception during batch_put_entity'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def prepare_insert(self, table):
      """ Prepare an insert statement.
  
      Args:
        table: A string containing the table name.
      Returns:
        A PreparedStatement object.
      """
      statement = """
        INSERT INTO "{table}" ({key}, {column}, {value})
        VALUES (?, ?, ?)
        USING TIMESTAMP ?
      """.format(table=table,
                 key=ThriftColumn.KEY,
                 column=ThriftColumn.COLUMN_NAME,
                 value=ThriftColumn.VALUE)
  
      if statement not in self.prepared_statements:
        self.prepared_statements[statement] = self.session.prepare(statement)
  
      return self.prepared_statements[statement]
  
    def prepare_delete(self, table):
      """ Prepare a delete statement.
  
      Args:
        table: A string containing the table name.
      Returns:
        A PreparedStatement object.
      """
      statement = """
        DELETE FROM "{table}"
        USING TIMESTAMP ?
        WHERE {key} = ?
      """.format(table=table, key=ThriftColumn.KEY)
  
      if statement not in self.prepared_statements:
        self.prepared_statements[statement] = self.session.prepare(statement)
  
      return self.prepared_statements[statement]
  
    def _normal_batch(self, mutations, txid):
      """ Use Cassandra's native batch statement to apply mutations atomically.
  
      Args:
        mutations: A list of dictionaries representing mutations.
        txid: An integer specifying a transaction ID.
      """
      self.logger.debug('Normal batch: {} mutations'.format(len(mutations)))
      batch = BatchStatement(consistency_level=ConsistencyLevel.QUORUM,
                             retry_policy=BASIC_RETRIES)
      prepared_statements = {'insert': {}, 'delete': {}}
      for mutation in mutations:
        table = mutation['table']
  
        if table == 'group_updates':
          key = mutation['key']
          insert = """
            INSERT INTO group_updates (group, last_update)
            VALUES (%(group)s, %(last_update)s)
            USING TIMESTAMP %(timestamp)s
          """
          parameters = {'group': key, 'last_update': mutation['last_update'],
                        'timestamp': get_write_time(txid)}
          batch.add(insert, parameters)
          continue
  
        if mutation['operation'] == Operations.PUT:
          if table not in prepared_statements['insert']:
            prepared_statements['insert'][table] = self.prepare_insert(table)
          values = mutation['values']
          for column in values:
            batch.add(
              prepared_statements['insert'][table],
              (bytearray(mutation['key']), column, bytearray(values[column]),
               get_write_time(txid))
            )
        elif mutation['operation'] == Operations.DELETE:
          if table not in prepared_statements['delete']:
            prepared_statements['delete'][table] = self.prepare_delete(table)
          batch.add(
            prepared_statements['delete'][table],
            (get_write_time(txid), bytearray(mutation['key']))
          )
  
      try:
        self.session.execute(batch)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception during batch_mutate'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def statements_for_mutations(self, mutations, txid):
      """ Generates Cassandra statements for a list of mutations.
  
      Args:
        mutations: A list of dictionaries representing mutations.
        txid: An integer specifying a transaction ID.
      Returns:
        A list of tuples containing Cassandra statements and parameters.
      """
      prepared_statements = {'insert': {}, 'delete': {}}
      statements_and_params = []
      for mutation in mutations:
        table = mutation['table']
  
        if table == 'group_updates':
          key = mutation['key']
          insert = """
            INSERT INTO group_updates (group, last_update)
            VALUES (%(group)s, %(last_update)s)
            USING TIMESTAMP %(timestamp)s
          """
          parameters = {'group': key, 'last_update': mutation['last_update'],
                        'timestamp': get_write_time(txid)}
          statements_and_params.append((SimpleStatement(insert), parameters))
          continue
  
        if mutation['operation'] == Operations.PUT:
          if table not in prepared_statements['insert']:
            prepared_statements['insert'][table] = self.prepare_insert(table)
          values = mutation['values']
          for column in values:
            params = (bytearray(mutation['key']), column,
                      bytearray(values[column]), get_write_time(txid))
            statements_and_params.append(
              (prepared_statements['insert'][table], params))
        elif mutation['operation'] == Operations.DELETE:
          if table not in prepared_statements['delete']:
            prepared_statements['delete'][table] = self.prepare_delete(table)
          params = (get_write_time(txid), bytearray(mutation['key']))
          statements_and_params.append(
            (prepared_statements['delete'][table], params))
  
      return statements_and_params
  
    def apply_mutations(self, mutations, txid):
      """ Apply mutations across tables.
  
      Args:
        mutations: A list of dictionaries representing mutations.
        txid: An integer specifying a transaction ID.
      """
      statements_and_params = self.statements_for_mutations(mutations, txid)
      execute_concurrent(self.session, statements_and_params,
                         raise_on_first_error=True)
  
    def _large_batch(self, app, mutations, entity_changes, txn):
      """ Insert or delete multiple rows across tables in an atomic statement.
  
      Args:
        app: A string containing the application ID.
        mutations: A list of dictionaries representing mutations.
        entity_changes: A list of changes at the entity level.
        txn: A transaction ID handler.
      Raises:
        FailedBatch if a concurrent process modifies the batch status.
        AppScaleDBConnectionError if a database connection error was encountered.
      """
      self.logger.debug('Large batch: transaction {}, {} mutations'.
                        format(txn, len(mutations)))
      large_batch = LargeBatch(self.session, app, txn)
      try:
        large_batch.start()
      except FailedBatch as batch_error:
        raise AppScaleDBConnectionError(str(batch_error))
  
      insert_item = """
        INSERT INTO batches (app, transaction, namespace, path,
                             old_value, new_value)
        VALUES (?, ?, ?, ?, ?, ?)
      """
      insert_statement = self.session.prepare(insert_item)
  
      statements_and_params = []
      for entity_change in entity_changes:
        old_value = None
        if entity_change['old'] is not None:
          old_value = bytearray(entity_change['old'].Encode())
        new_value = None
        if entity_change['new'] is not None:
          new_value = bytearray(entity_change['new'].Encode())
  
        parameters = (app, txn, entity_change['key'].name_space(),
                      bytearray(entity_change['key'].path().Encode()), old_value,
                      new_value)
        statements_and_params.append((insert_statement, parameters))
  
      try:
        execute_concurrent(self.session, statements_and_params,
                           raise_on_first_error=True)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Unable to write large batch log'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
      try:
        large_batch.set_applied()
      except FailedBatch as batch_error:
        raise AppScaleDBConnectionError(str(batch_error))
  
      try:
        self.apply_mutations(mutations, txn)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception during large batch'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
      try:
        large_batch.cleanup()
      except FailedBatch:
        # This should not raise an exception since the batch is already applied.
        logging.exception('Unable to clear batch status')
  
      clear_batch = """
        DELETE FROM batches
        WHERE app = %(app)s AND transaction = %(transaction)s
      """
      parameters = {'app': app, 'transaction': txn}
      try:
        self.session.execute(clear_batch, parameters)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        logging.exception('Unable to clear batch log')
  
    def batch_mutate(self, app, mutations, entity_changes, txn):
      """ Insert or delete multiple rows across tables in an atomic statement.
  
      Args:
        app: A string containing the application ID.
        mutations: A list of dictionaries representing mutations.
        entity_changes: A list of changes at the entity level.
        txn: A transaction ID handler.
      """
      size = batch_size(mutations)
      if size > LARGE_BATCH_THRESHOLD:
        self._large_batch(app, mutations, entity_changes, txn)
      else:
        self._normal_batch(mutations, txn)
  
    def batch_delete(self, table_name, row_keys, column_names=()):
      """
      Remove a set of rows corresponding to a set of keys.
       
      Args:
        table_name: Table to delete rows from
        row_keys: A list of keys to remove
        column_names: Not used
      Raises:
        TypeError: If an argument passed in was not of the expected type.
        AppScaleDBConnectionError: If the batch_delete could not be performed due
          to an error with Cassandra.
      """ 
      if not isinstance(table_name, str): raise TypeError("Expected a str")
      if not isinstance(row_keys, list): raise TypeError("Expected a list")
  
      row_keys_bytes = [bytearray(row_key) for row_key in row_keys]
  
      statement = 'DELETE FROM "{table}" WHERE {key} IN %s'.\
        format(
          table=table_name,
          key=ThriftColumn.KEY
        )
      query = SimpleStatement(statement, retry_policy=BASIC_RETRIES)
      parameters = (ValueSequence(row_keys_bytes),)
  
      try:
        self.session.execute(query, parameters=parameters)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception during batch_delete'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def delete_table(self, table_name):
      """ 
      Drops a given table (aka column family in Cassandra)
    
      Args:
        table_name: A string name of the table to drop
      Raises:
        TypeError: If an argument passed in was not of the expected type.
        AppScaleDBConnectionError: If the delete_table could not be performed due
          to an error with Cassandra.
      """
      if not isinstance(table_name, str): raise TypeError("Expected a str")
  
      statement = 'DROP TABLE IF EXISTS "{table}"'.format(table=table_name)
      query = SimpleStatement(statement, retry_policy=BASIC_RETRIES)
  
      try:
        self.session.execute(query)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception during delete_table'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def create_table(self, table_name, column_names):
      """ 
      Creates a table if it doesn't already exist.
      
      Args:
        table_name: The column family name
        column_names: Not used but here to match the interface
      Raises:
        TypeError: If an argument passed in was not of the expected type.
        AppScaleDBConnectionError: If the create_table could not be performed due
          to an error with Cassandra.
      """
      if not isinstance(table_name, str): raise TypeError("Expected a str")
      if not isinstance(column_names, list): raise TypeError("Expected a list")
  
      statement = 'CREATE TABLE IF NOT EXISTS "{table}" ('\
          '{key} blob,'\
          '{column} text,'\
          '{value} blob,'\
          'PRIMARY KEY ({key}, {column})'\
        ') WITH COMPACT STORAGE'.format(
          table=table_name,
          key=ThriftColumn.KEY,
          column=ThriftColumn.COLUMN_NAME,
          value=ThriftColumn.VALUE
        )
      query = SimpleStatement(statement, retry_policy=NO_RETRIES)
  
      try:
        self.session.execute(query, timeout=SCHEMA_CHANGE_TIMEOUT)
      except cassandra.OperationTimedOut:
        logging.warning(
          'Encountered an operation timeout while creating a table. Waiting {} '
          'seconds for schema to settle.'.format(SCHEMA_CHANGE_TIMEOUT))
        time.sleep(SCHEMA_CHANGE_TIMEOUT)
        raise AppScaleDBConnectionError('Exception during create_table')
      except (error for error in dbconstants.TRANSIENT_CASSANDRA_ERRORS
              if error != cassandra.OperationTimedOut):
        message = 'Exception during create_table'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def range_query(self,
                    table_name,
                    column_names, 
                    start_key, 
                    end_key, 
                    limit, 
                    offset=0, 
                    start_inclusive=True, 
                    end_inclusive=True,
                    keys_only=False):
      """ 
      Gets a dense range ordered by keys. Returns an ordered list of 
      a dictionary of [key:{column1:value1, column2:value2},...]
      or a list of keys if keys only.
       
      Args:
        table_name: Name of table to access
        column_names: Columns which get returned within the key range
        start_key: String for which the query starts at
        end_key: String for which the query ends at
        limit: Maximum number of results to return
        offset: Cuts off these many from the results [offset:]
        start_inclusive: Boolean if results should include the start_key
        end_inclusive: Boolean if results should include the end_key
        keys_only: Boolean if to only keys and not values
      Raises:
        TypeError: If an argument passed in was not of the expected type.
        AppScaleDBConnectionError: If the range_query could not be performed due
          to an error with Cassandra.
      Returns:
        An ordered list of dictionaries of key=>columns/values
      """
      if not isinstance(table_name, str):
        raise TypeError('table_name must be a string')
      if not isinstance(column_names, list):
        raise TypeError('column_names must be a list')
      if not isinstance(start_key, str):
        raise TypeError('start_key must be a string')
      if not isinstance(end_key, str):
        raise TypeError('end_key must be a string')
      if not isinstance(limit, (int, long)) and limit is not None:
        raise TypeError('limit must be int, long, or NoneType')
      if not isinstance(offset, (int, long)):
        raise TypeError('offset must be int or long')
  
      if start_inclusive:
        gt_compare = '>='
      else:
        gt_compare = '>'
  
      if end_inclusive:
        lt_compare = '<='
      else:
        lt_compare = '<'
  
      query_limit = ''
      if limit is not None:
        query_limit = 'LIMIT {}'.format(len(column_names) * limit)
  
      statement = """
        SELECT * FROM "{table}" WHERE
        token({key}) {gt_compare} %s AND
        token({key}) {lt_compare} %s AND
        {column} IN %s
        {limit}
        ALLOW FILTERING
      """.format(table=table_name,
                 key=ThriftColumn.KEY,
                 gt_compare=gt_compare,
                 lt_compare=lt_compare,
                 column=ThriftColumn.COLUMN_NAME,
                 limit=query_limit)
  
      query = SimpleStatement(statement, retry_policy=BASIC_RETRIES)
      parameters = (bytearray(start_key), bytearray(end_key),
                    ValueSequence(column_names))
  
      try:
        results = self.session.execute(query, parameters=parameters)
  
        results_list = []
        current_item = {}
        current_key = None
        for (key, column, value) in results:
          if keys_only:
            results_list.append(key)
            continue
  
          if key != current_key:
            if current_item:
              results_list.append({current_key: current_item})
            current_item = {}
            current_key = key
  
          current_item[column] = value
        if current_item:
          results_list.append({current_key: current_item})
        return results_list[offset:]
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception during range_query'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def get_metadata(self, key):
      """ Retrieve a value from the datastore metadata table.
  
      Args:
        key: A string containing the key to fetch.
      Returns:
        A string containing the value or None if the key is not present.
      """
      statement = """
        SELECT {value} FROM "{table}"
        WHERE {key} = %s
        AND {column} = %s
      """.format(
        value=ThriftColumn.VALUE,
        table=dbconstants.DATASTORE_METADATA_TABLE,
        key=ThriftColumn.KEY,
        column=ThriftColumn.COLUMN_NAME
      )
      try:
        results = self.session.execute(statement, (bytearray(key), key))
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Unable to fetch {} from datastore metadata'.format(key)
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
      try:
        return results[0].value
      except IndexError:
        return None
  
    def set_metadata(self, key, value):
      """ Set a datastore metadata value.
  
      Args:
        key: A string containing the key to set.
        value: A string containing the value to set.
      """
      if not isinstance(key, str):
        raise TypeError('key should be a string')
  
      if not isinstance(value, str):
        raise TypeError('value should be a string')
  
      statement = """
        INSERT INTO "{table}" ({key}, {column}, {value})
        VALUES (%(key)s, %(column)s, %(value)s)
      """.format(
        table=dbconstants.DATASTORE_METADATA_TABLE,
        key=ThriftColumn.KEY,
        column=ThriftColumn.COLUMN_NAME,
        value=ThriftColumn.VALUE
      )
      parameters = {'key': bytearray(key),
                    'column': key,
                    'value': bytearray(value)}
      try:
        self.session.execute(statement, parameters)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Unable to set datastore metadata for {}'.format(key)
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
      except cassandra.InvalidRequest:
        self.create_table(dbconstants.DATASTORE_METADATA_TABLE,
                          dbconstants.DATASTORE_METADATA_SCHEMA)
        self.session.execute(statement, parameters)
  
    def get_indices(self, app_id):
      """ Gets the indices of the given application.
  
      Args:
        app_id: Name of the application.
      Returns:
        Returns a list of encoded entity_pb.CompositeIndex objects.
      """
      start_key = dbconstants.KEY_DELIMITER.join([app_id, 'index', ''])
      end_key = dbconstants.KEY_DELIMITER.join(
        [app_id, 'index', dbconstants.TERMINATING_STRING])
      result = self.range_query(
        dbconstants.METADATA_TABLE,
        dbconstants.METADATA_SCHEMA,
        start_key,
        end_key,
        dbconstants.MAX_NUMBER_OF_COMPOSITE_INDEXES,
        offset=0,
        start_inclusive=True,
        end_inclusive=True)
      list_result = []
      for list_item in result:
        for key, value in list_item.iteritems():
          list_result.append(value['data'])
      return list_result
  
    def valid_data_version(self):
      """ Checks whether or not the data layout can be used.
  
      Returns:
        A boolean.
      """
      try:
        version = self.get_metadata(VERSION_INFO_KEY)
      except cassandra.InvalidRequest:
        return False
  
      return version is not None and float(version) == EXPECTED_DATA_VERSION
  
    def group_updates(self, groups):
      """ Fetch the latest transaction IDs for each group.
  
      Args:
        groups: An interable containing encoded Reference objects.
      Returns:
        A set of integers specifying transaction IDs.
      """
      futures = []
      for group in groups:
        query = 'SELECT * FROM group_updates WHERE group=%s'
        futures.append(self.session.execute_async(query, [bytearray(group)]))
  
      updates = set()
      for future in futures:
        rows = future.result()
        try:
          result = rows[0]
        except IndexError:
          continue
  
        updates.add(result.last_update)
  
      return updates
  
    def start_transaction(self, app, txid, is_xg, in_progress):
      """ Persist transaction metadata.
  
      Args:
        app: A string containing an application ID.
        txid: An integer specifying the transaction ID.
        is_xg: A boolean specifying that the transaction is cross-group.
        in_progress: An iterable containing transaction IDs.
      """
      if in_progress:
        in_progress_bin = bytearray(
          struct.pack('q' * len(in_progress), *in_progress))
      else:
        in_progress_bin = None
  
      insert = """
        INSERT INTO transactions (txid_hash, operation, namespace, path,
                                  start_time, is_xg, in_progress)
        VALUES (%(txid_hash)s, %(operation)s, %(namespace)s, %(path)s,
                %(start_time)s, %(is_xg)s, %(in_progress)s)
        USING TTL {ttl}
      """.format(ttl=dbconstants.MAX_TX_DURATION * 2)
      parameters = {'txid_hash': tx_partition(app, txid),
                    'operation': TxnActions.START,
                    'namespace': '',
                    'path': bytearray(''),
                    'start_time': datetime.datetime.utcnow(),
                    'is_xg': is_xg,
                    'in_progress': in_progress_bin}
  
      try:
        self.session.execute(insert, parameters)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception while starting a transaction'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def put_entities_tx(self, app, txid, entities):
      """ Update transaction metadata with new put operations.
  
      Args:
        app: A string containing an application ID.
        txid: An integer specifying the transaction ID.
        entities: A list of entities that will be put upon commit.
      """
      batch = BatchStatement(consistency_level=ConsistencyLevel.QUORUM,
                             retry_policy=BASIC_RETRIES)
      insert = self.session.prepare("""
        INSERT INTO transactions (txid_hash, operation, namespace, path, entity)
        VALUES (?, ?, ?, ?, ?)
        USING TTL {ttl}
      """.format(ttl=dbconstants.MAX_TX_DURATION * 2))
  
      for entity in entities:
        args = (tx_partition(app, txid),
                TxnActions.MUTATE,
                entity.key().name_space(),
                bytearray(entity.key().path().Encode()),
                bytearray(entity.Encode()))
        batch.add(insert, args)
  
      try:
        self.session.execute(batch)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception while putting entities in a transaction'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def delete_entities_tx(self, app, txid, entity_keys):
      """ Update transaction metadata with new delete operations.
  
      Args:
        app: A string containing an application ID.
        txid: An integer specifying the transaction ID.
        entity_keys: A list of entity keys that will be deleted upon commit.
      """
      batch = BatchStatement(consistency_level=ConsistencyLevel.QUORUM,
                             retry_policy=BASIC_RETRIES)
      insert = self.session.prepare("""
        INSERT INTO transactions (txid_hash, operation, namespace, path, entity)
        VALUES (?, ?, ?, ?, ?)
        USING TTL {ttl}
      """.format(ttl=dbconstants.MAX_TX_DURATION * 2))
  
      for key in entity_keys:
        # The None value overwrites previous puts.
        args = (tx_partition(app, txid),
                TxnActions.MUTATE,
                key.name_space(),
                bytearray(key.path().Encode()),
                None)
        batch.add(insert, args)
  
      try:
        self.session.execute(batch)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception while deleting entities in a transaction'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def transactional_tasks_count(self, app, txid):
      """ Count the number of existing tasks associated with the transaction.
  
      Args:
        app: A string specifying an application ID.
        txid: An integer specifying a transaction ID.
      Returns:
        An integer specifying the number of existing tasks.
      """
      select = """
        SELECT count(*) FROM transactions
        WHERE txid_hash = %(txid_hash)s
        AND operation = %(operation)s
      """
      parameters = {'txid_hash': tx_partition(app, txid),
                    'operation': TxnActions.ENQUEUE_TASK}
      try:
        return self.session.execute(select, parameters)[0].count
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception while fetching task count'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def add_transactional_tasks(self, app, txid, tasks, service_id, version_id):
      """ Add tasks to be enqueued upon the completion of a transaction.
  
      Args:
        app: A string specifying an application ID.
        txid: An integer specifying a transaction ID.
        tasks: A list of TaskQueueAddRequest objects.
        service_id: A string specifying the client's service ID.
        version_id: A string specifying the client's version ID.
      """
      batch = BatchStatement(consistency_level=ConsistencyLevel.QUORUM,
                             retry_policy=BASIC_RETRIES)
      insert = self.session.prepare("""
        INSERT INTO transactions (txid_hash, operation, namespace, path, task)
        VALUES (?, ?, ?, ?, ?)
        USING TTL {ttl}
      """.format(ttl=dbconstants.MAX_TX_DURATION * 2))
  
      for task in tasks:
        task.clear_transaction()
  
        # The path for the task entry doesn't matter as long as it's unique.
        path = bytearray(str(uuid.uuid4()))
  
        task_payload = '_'.join([service_id, version_id, task.Encode()])
        args = (tx_partition(app, txid),
                TxnActions.ENQUEUE_TASK,
                '',
                path,
                task_payload)
        batch.add(insert, args)
  
      try:
        self.session.execute(batch)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception while adding tasks in a transaction'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def record_reads(self, app, txid, group_keys):
      """ Keep track of which entity groups were read in a transaction.
  
      Args:
        app: A string specifying an application ID.
        txid: An integer specifying a transaction ID.
        group_keys: An iterable containing Reference objects.
      """
      batch = BatchStatement(consistency_level=ConsistencyLevel.QUORUM,
                             retry_policy=BASIC_RETRIES)
      insert = self.session.prepare("""
        INSERT INTO transactions (txid_hash, operation, namespace, path)
        VALUES (?, ?, ?, ?)
        USING TTL {ttl}
      """.format(ttl=dbconstants.MAX_TX_DURATION * 2))
  
      for group_key in group_keys:
        if not isinstance(group_key, entity_pb.Reference):
          group_key = entity_pb.Reference(group_key)
  
        args = (tx_partition(app, txid),
                TxnActions.GET,
                group_key.name_space(),
                bytearray(group_key.path().Encode()))
        batch.add(insert, args)
  
      try:
        self.session.execute(batch)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception while recording reads in a transaction'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
    def get_transaction_metadata(self, app, txid):
      """ Fetch transaction state.
  
      Args:
        app: A string specifying an application ID.
        txid: An integer specifying a transaction ID.
      Returns:
        A dictionary containing transaction state.
      """
      select = """
        SELECT namespace, operation, path, start_time, is_xg, in_progress,
               entity, task
        FROM transactions
        WHERE txid_hash = %(txid_hash)s
      """
      parameters = {'txid_hash': tx_partition(app, txid)}
      try:
        results = self.session.execute(select, parameters)
      except dbconstants.TRANSIENT_CASSANDRA_ERRORS:
        message = 'Exception while inserting entities in a transaction'
        logging.exception(message)
        raise AppScaleDBConnectionError(message)
  
      metadata = {'puts': {}, 'deletes': [], 'tasks': [], 'reads': set()}
      for result in results:
        if result.operation == TxnActions.START:
          metadata['start'] = result.start_time
          metadata['is_xg'] = result.is_xg
          metadata['in_progress'] = set()
          if metadata['in_progress'] is not None:
            metadata['in_progress'] = set(
              struct.unpack('q' * int(len(result.in_progress) / 8),
                            result.in_progress))
        if result.operation == TxnActions.MUTATE:
          key = create_key(app, result.namespace, result.path)
          if result.entity is None:
            metadata['deletes'].append(key)
          else:
            metadata['puts'][key.Encode()] = result.entity
        if result.operation == TxnActions.GET:
          group_key = create_key(app, result.namespace, result.path)
          metadata['reads'].add(group_key.Encode())
        if result.operation == TxnActions.ENQUEUE_TASK:
          service_id, version_id, task_pb = result.task.split('_', 2)
          task_metadata = {
            'service_id': service_id,
            'version_id': version_id,
            'task': taskqueue_service_pb.TaskQueueAddRequest(task_pb)}
          metadata['tasks'].append(task_metadata)
      return metadata
  