files: """
  This module contains shared encoding and decoding utilities for translating
  datastore types to bytestrings and vice versa.
  """
  import bisect
  import logging
  import struct
  import sys
  
  import six
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.dbconstants import BadRequest, InternalError
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.datastore import entity_pb
  
  # The number of bytes used to encode a read version in a way that is comparable
  # to commit versionstamps.
  READ_VERSION_SIZE = 8
  
  # Byte values that signify a data type. A couple values are left between each
  # one just in case they need adjustment.
  NULL_CODE = 0x01
  # 17 values are reserved for integers in order to allow variable size in both a
  # negative and positive direction.
  INT64_ZERO_CODE = 0x0C
  FALSE_CODE = 0x17
  TRUE_CODE = 0x18
  BYTES_CODE = 0x1B
  DOUBLE_CODE = 0x1E
  POINT_CODE = 0x21
  USER_CODE = 0x24
  REFERENCE_CODE = 0x27
  
  # These are defined for clarity when selecting property type limits.
  MIN_INT64_CODE = INT64_ZERO_CODE - 8
  MAX_INT64_CODE = INT64_ZERO_CODE + 8
  
  # Ensures the shorter of two variable-length values (with identical prefixes)
  # is placed before the longer one. Otherwise, the following byte(s) could
  # determine the sort order. It also allows a decoder to find the end of the
  # value.
  TERMINATOR = 0x00
  
  logger = logging.getLogger(__name__)
  
  
  def reverse_bits(blob):
    return b''.join(map(lambda x: six.int2byte(x ^ 0xFF), six.iterbytes(blob)))
  
  
  def encode_marker(marker, reverse):
    return six.int2byte(marker ^ 0xFF) if reverse else six.int2byte(marker)
  
  
  def decode_str(string):
    """ Converts byte strings to unicode strings. """
    if isinstance(string, six.text_type):
      return string
  
    return string.decode('utf-8')
  
  
  class Int64(object):
    """
    Handles encoding and decoding operations for 64-bit integers. The encoded
    value is variable-length. The value of the marker indicates how many bytes
    were used to encode the value.
    """
    LIMITS = tuple((1 << (i * 8)) - 1 for i in range(9))
  
    @classmethod
    def encode(cls, value, reverse=False):
      if value == 0:
        code = INT64_ZERO_CODE
        packed = b''
      elif value > 0:
        encoded_size = bisect.bisect_left(cls.LIMITS, value)
        adjusted_value = cls.LIMITS[encoded_size] - value if reverse else value
        code = INT64_ZERO_CODE + encoded_size
        packed = struct.pack('>Q', adjusted_value)[-encoded_size:]
      else:
        encoded_size = bisect.bisect_left(cls.LIMITS, -value)
        # Shift negative values to an unsigned space.
        adjusted_value = -value if reverse else cls.LIMITS[encoded_size] + value
        code = INT64_ZERO_CODE - encoded_size
        packed = struct.pack('>Q', adjusted_value)[-encoded_size:]
  
      return encode_marker(code, reverse) + packed
  
    @classmethod
    def decode(cls, marker, blob, pos, reverse=False):
      code = ord(marker) ^ 0xFF if reverse else ord(marker)
      if code == INT64_ZERO_CODE:
        return 0, pos
  
      encoded_size = abs(code - INT64_ZERO_CODE)
      packed = b'\x00' * (8 - encoded_size) + blob[pos:pos + encoded_size]
      pos += encoded_size
      value = struct.unpack('>Q', packed)[0]
      if code > INT64_ZERO_CODE:
        value = cls.LIMITS[encoded_size] - value if reverse else value
      else:
        # Shift values back to their original, negative state.
        value = -value if reverse else value - cls.LIMITS[encoded_size]
  
      return value, pos
  
    @classmethod
    def encode_bare(cls, value, byte_count):
      """
      Encodes an integer without a prefix using the specified number of bytes.
      """
      encoded = struct.pack('>Q', value)
      if any(byte != b'\x00' for byte in encoded[:-byte_count]):
        raise InternalError(u'Value exceeds maximum size')
  
      return encoded[-byte_count:]
  
    @classmethod
    def decode_bare(cls, encoded_value):
      """ Decodes a byte string back to an integer. """
      encoded_value = b'\x00' * (8 - len(encoded_value)) + encoded_value
      return struct.unpack('>Q', encoded_value)[0]
  
  
  class Bytes(object):
    """
    Handles encoding and decoding operations for arbitrary blobs. Since the
    length of the value is variable and there is usually data that follows the
    encoded value, a terminator byte is used to indicate the end of the value.
    Any occurrence of the terminator byte within the value is escaped in order to
    preserve the ordering and allow the parser to find the real terminator.
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(BYTES_CODE), reverse=False):
      packed = reverse_bits(value) if reverse else value
      terminator = encode_marker(TERMINATOR, reverse)
      # Replace each occurrence of the terminator with a sequence that maintains
      # the sort order. In order for the parser to be able to find the end of the
      # value, directories that use this codec must ensure that the first byte
      # that follows the encoded Bytes object must not be 0x00 or 0xFF.
      if reverse:
        # The preceding 0xFE ensures a terminator character within the value
        # compares as less than the real terminator. For example,
        # 0x0200 -> 0xFDFF -> 0xFDFEFF00FF should come before
        # 0x02   -> 0xFD   -> 0xFDFF
        # The trailing 0x00 ensures that the parser is able to differentiate
        # between a replaced value and the real terminator.
        packed = packed.replace(terminator, b'\xFE\xFF\x00')
      else:
        # A trailing 0xFF is enough for the parser to find the terminator and
        # to maintain sort order despite the values that follow. For example
        # (assuming 0x10 is the value that follows),
        # 0x00, 0x10   -> 0x00FF0010     (not 0x000010) should come before
        # 0x0000, 0x10 -> 0x00FF00FF0010 (not 0x00000010)
        packed = packed.replace(terminator, b'\x00\xFF')
  
      return prefix + packed + terminator
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      end = cls._find_terminator(blob, pos, reverse)
      packed = blob[pos:end]
      terminator = encode_marker(TERMINATOR, reverse)
      if reverse:
        packed = packed.replace(b'\xFE\xFF\x00', terminator)
      else:
        packed = packed.replace(b'\x00\xFF', terminator)
  
      value = reverse_bits(packed) if reverse else packed
      return value, end + 1
  
    @staticmethod
    def _find_terminator(blob, pos, reverse=False):
      """ Finds the position of the terminator. """
      terminator = encode_marker(TERMINATOR, reverse)
      escape_byte = b'\x00' if reverse else b'\xFF'
      while True:
        pos = blob.find(terminator, pos)
        if pos < 0:
          raise InternalError(u'Byte string is missing terminator')
  
        if blob[pos + 1:pos + 2] != escape_byte:
          return pos
  
        pos += 2
  
  
  class Double(object):
    """
    Handles encoding and decoding operations for floating point values. 8 bytes
    (plus the marker byte) are always used.
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(DOUBLE_CODE), reverse=False):
      packed = struct.pack('>d', value)
      # The first bit of the packed value indicates whether it's negative or not.
      positive = six.indexbytes(packed, 0) & 0x80 == 0x00
      if positive:
        # Flip the sign bit.
        packed = six.int2byte(six.indexbytes(packed, 0) ^ 0x80) + packed[1:]
        if reverse:
          packed = reverse_bits(packed)
      else:
        # Flip all the bits unless the sort order is reversed.
        packed = packed if reverse else reverse_bits(packed)
  
      return prefix + packed
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      packed = blob[pos:pos + 8]
      pos += 8
      first_bit = six.indexbytes(packed, 0) & 0x80 == 0x80
      positive = first_bit if not reverse else not first_bit
      if positive:
        if reverse:
          packed = reverse_bits(packed)
  
        # Restore the sign bit.
        packed = six.int2byte(six.indexbytes(packed, 0) ^ 0x80) + packed[1:]
      else:
        # Restore all the original bits for ascending values.
        packed = packed if reverse else reverse_bits(packed)
  
      value = struct.unpack('>d', packed)[0]
      return value, pos
  
  
  class Point(object):
    """
    Handles encoding and decoding operations for point values. The Double codec
    is used for each coordinate (x and y).
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(POINT_CODE), reverse=False):
      x_packed = Double.encode(value.x(), prefix=b'', reverse=reverse)
      y_packed = Double.encode(value.y(), prefix=b'', reverse=reverse)
      return prefix + x_packed + y_packed
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      x, pos = Double.decode(blob, pos, reverse)
      y, pos = Double.decode(blob, pos, reverse)
      point_val = entity_pb.PropertyValue_PointValue()
      point_val.set_x(x)
      point_val.set_y(y)
      return point_val, pos
  
  
  class Text(object):
    """
    Handles encoding and decoding operations for unicode strings. Since the
    length of the value is variable and there is usually data that follows the
    encoded value, a terminator byte is used to indicate the end of the value.
    """
    @classmethod
    def encode(cls, unicode_string, prefix=b'', reverse=False):
      # Ensure the encoded value does not contain the terminator. UTF-8 does not
      # use 0xFF, so this can be done without exceeding the largest byte value.
      if reverse:
        encode_byte = lambda x: six.int2byte(x + 1 ^ 0xFF)
      else:
        encode_byte = lambda x: six.int2byte(x + 1)
  
      encoded = b''.join(
        map(encode_byte, six.iterbytes(unicode_string.encode('utf-8'))))
      terminator = encode_marker(TERMINATOR, reverse)
      return prefix + encoded + terminator
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      terminator = TERMINATOR ^ 0xFF if reverse else TERMINATOR
      byte_array = bytearray()
      while pos < len(blob):
        byte_value = six.indexbytes(blob, pos)
        pos += 1
        if byte_value == terminator:
          break
  
        if reverse:
          byte_value ^= 0xFF
  
        # Shift the byte back to its original value.
        byte_array.append(byte_value - 1)
  
      return byte_array.decode('utf-8'), pos
  
  
  class User(object):
    """
    Handles encoding and decoding operations for User values. The Double codec
    is used for each of the required fields (email and auth_domain).
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(POINT_CODE), reverse=False):
      packed = (Text.encode(decode_str(value.email()), reverse=reverse) +
                Text.encode(decode_str(value.auth_domain()), reverse=reverse))
      return prefix + packed
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      email, pos = Text.decode(blob, pos, reverse)
      auth_domain, pos = Text.decode(blob, pos, reverse)
      user_val = entity_pb.PropertyValue_UserValue()
      user_val.set_email(email)
      user_val.set_auth_domain(auth_domain)
      return user_val, pos
  
  
  class Path(object):
    """
    Handles encoding and decoding operations for entity paths. The Text codec
    is used for kinds and names. The Int64 codec is used for IDs.
    """
    KIND_MARKER = 0x1C
  
    MIN_ID_MARKER = INT64_ZERO_CODE - 8
  
    # Marks an item as a name rather than an ID. These should be be placed after
    # entries with IDs.
    NAME_MARKER = 0x1D
  
    @classmethod
    def pack(cls, path, prefix=b'', omit_terminator=False, reverse=False):
      if not isinstance(path, tuple):
        path = cls.flatten(path)
  
      encoded_items = []
      kind_marker = encode_marker(cls.KIND_MARKER, reverse)
      for index in range(0, len(path), 2):
        kind = path[index]
        id_or_name = path[index + 1]
        encoded_items.append(Text.encode(kind, kind_marker, reverse))
        encoded_items.append(cls.encode_id_or_name(id_or_name, reverse))
  
      terminator = b'' if omit_terminator else encode_marker(TERMINATOR, reverse)
      return b''.join([prefix] + encoded_items + [terminator])
  
    @classmethod
    def encode_id_or_name(cls, id_or_name, reverse=False):
      if isinstance(id_or_name, six.text_type):
        name_marker = encode_marker(cls.NAME_MARKER, reverse)
        return Text.encode(id_or_name, name_marker, reverse)
      elif isinstance(id_or_name, int):
        return Int64.encode(id_or_name, reverse)
      else:
        raise BadRequest(u'Invalid path element type')
  
    @classmethod
    def unpack(cls, blob, pos, reverse=False):
      items = []
      terminator = encode_marker(TERMINATOR, reverse)
      kind_marker = encode_marker(cls.KIND_MARKER, reverse)
      name_marker = encode_marker(cls.NAME_MARKER, reverse)
      while pos < len(blob):
        marker = blob[pos]
        pos += 1
        if marker == terminator:
          break
  
        if marker != kind_marker:
          raise InternalError(u'Encoded path is missing kind')
  
        kind, pos = Text.decode(blob, pos, reverse)
        items.append(kind)
  
        marker = blob[pos]
        pos += 1
        if marker == name_marker:
          elem_name, pos = Text.decode(blob, pos, reverse)
          items.append(elem_name)
        else:
          elem_id, pos = Int64.decode(marker, blob, pos, reverse)
          items.append(elem_id)
  
      return tuple(items), pos
  
    @staticmethod
    def flatten(path):
      """ Converts a key path protobuf object to a tuple. """
      if isinstance(path, entity_pb.PropertyValue):
        element_list = path.referencevalue().pathelement_list()
      elif isinstance(path, entity_pb.PropertyValue_ReferenceValue):
        element_list = path.pathelement_list()
      else:
        element_list = path.element_list()
  
      return tuple(item for element in element_list
                   for item in Path.encode_element(element))
  
    @staticmethod
    def decode(flat_path, reference_value=False):
      """ Converts a tuple to a key path protobuf object. """
      if reference_value:
        path = entity_pb.PropertyValue_ReferenceValue()
      else:
        path = entity_pb.Path()
  
      for index in range(0, len(flat_path), 2):
        if reference_value:
          element = path.add_pathelement()
        else:
          element = path.add_element()
  
        element.set_type(flat_path[index])
        id_or_name = flat_path[index + 1]
        if isinstance(id_or_name, int):
          element.set_id(id_or_name)
        else:
          element.set_name(id_or_name.encode('utf-8'))
  
      return path
  
    @staticmethod
    def encode_element(element):
      """ Converts a path element protobuf object to a tuple. """
      if element.has_id():
        id_or_name = int(element.id())
      elif element.has_name():
        id_or_name = decode_str(element.name())
      else:
        raise BadRequest(u'All path elements must either have a name or ID')
  
      return decode_str(element.type()), id_or_name
  
    @staticmethod
    def decode_element(element_tuple):
      """ Converts a tuple to a path element protobuf object. """
      path_element = entity_pb.Path_Element()
      path_element.set_type(element_tuple[0])
      if isinstance(element_tuple[1], int):
        path_element.set_id(element_tuple[1])
      else:
        path_element.set_name(element_tuple[1])
  
      return path_element
  
  
  class Reference(object):
    """
    Handles encoding and decoding operations for entity keys. The Text codec is
    used for the project ID and namespace. The Path codec is used for the key's
    path.
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(REFERENCE_CODE), reverse=False):
      return b''.join([
        prefix,
        Text.encode(decode_str(value.app()), reverse=reverse),
        Text.encode(decode_str(value.name_space()), reverse=reverse),
        Path.pack(Path.flatten(value), reverse=reverse)
      ])
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      project_id, pos = Text.decode(blob, pos, reverse)
      namespace, pos = Text.decode(blob, pos, reverse)
      flat_path, pos = Path.unpack(blob, pos, reverse)
      reference_val = entity_pb.PropertyValue_ReferenceValue()
      reference_val.set_app(project_id)
      reference_val.set_name_space(namespace)
      reference_val.MergeFrom(Path.decode(flat_path, reference_value=True))
      return reference_val, pos
  
  
  def encode_value(value, reverse=False):
    if isinstance(value, six.integer_types):
      return Int64.encode(value, reverse)
  
    if value.has_int64value():
      return Int64.encode(value.int64value(), reverse)
  
    if value.has_booleanvalue():
      to_encode = TRUE_CODE if value.booleanvalue() else FALSE_CODE
      return encode_marker(to_encode, reverse)
  
    if value.has_stringvalue():
      prefix = encode_marker(BYTES_CODE, reverse)
      return Bytes.encode(value.stringvalue(), prefix, reverse)
  
    if value.has_doublevalue():
      prefix = encode_marker(DOUBLE_CODE, reverse)
      return Double.encode(value.doublevalue(), prefix, reverse)
  
    if value.has_pointvalue():
      prefix = encode_marker(POINT_CODE, reverse)
      return Point.encode(value.pointvalue(), prefix, reverse)
  
    if value.has_uservalue():
      prefix = encode_marker(USER_CODE, reverse)
      return User.encode(value.uservalue(), prefix, reverse)
  
    if value.has_referencevalue():
      prefix = encode_marker(REFERENCE_CODE, reverse)
      return Reference.encode(value.referencevalue(), prefix, reverse)
  
    return encode_marker(NULL_CODE, reverse)
  
  
  def decode_value(blob, pos, reverse=False):
    prop_value = entity_pb.PropertyValue()
    marker = blob[pos]
    pos += 1
    code = ord(marker) ^ 0xFF if reverse else ord(marker)
    if code == NULL_CODE:
      pass
    elif INT64_ZERO_CODE - 8 <= code <= INT64_ZERO_CODE + 8:
      int_val, pos = Int64.decode(marker, blob, pos, reverse)
      prop_value.set_int64value(int_val)
    elif code in (TRUE_CODE, FALSE_CODE):
      prop_value.set_booleanvalue(code == TRUE_CODE)
    elif code == BYTES_CODE:
      bytes_val, pos = Bytes.decode(blob, pos, reverse)
      prop_value.set_stringvalue(bytes_val)
    elif code == DOUBLE_CODE:
      double_val, pos = Double.decode(blob, pos, reverse)
      prop_value.set_doublevalue(double_val)
    elif code == POINT_CODE:
      point_val, pos = Point.decode(blob, pos, reverse)
      prop_value.mutable_pointvalue().MergeFrom(point_val)
    elif code == USER_CODE:
      user_val, pos = User.decode(blob, pos, reverse)
      prop_value.mutable_uservalue().MergeFrom(user_val)
    elif code == REFERENCE_CODE:
      ref_val, pos = Reference.decode(blob, pos, reverse)
      prop_value.mutable_referencevalue().MergeFrom(ref_val)
  
    return prop_value, pos
  
  
  class TransactionID(object):
    """
    Handles encoding and decoding operations for transaction IDs. Since this
    cannot handle every commit versionstamp value, it will be removed as soon as
    the interface can handle larger values for transaction handles.
    """
    COMMIT_VERSION_BITS = 8 * 7 - 4
  
    BATCH_ORDER_BITS = 8
  
    @classmethod
    def encode(cls, scatter_val, commit_versionstamp):
      commit_version = struct.unpack('>Q', commit_versionstamp[:8])[0]
      batch_order = struct.unpack('>H', commit_versionstamp[8:])[0]
      if not 0 <= scatter_val <= 15:
        raise InternalError(u'Invalid scatter value')
  
      if commit_version >= 2 ** cls.COMMIT_VERSION_BITS:
        raise InternalError(u'Commit version too high')
  
      if batch_order >= 2 ** cls.BATCH_ORDER_BITS:
        raise InternalError(u'Batch order too high')
  
      return (commit_version << 12) + (batch_order << 4) + scatter_val
  
    @classmethod
    def decode(cls, txid):
      commit_version_bytes = struct.pack('>Q', txid >> 12)
      batch_order_bytes = struct.pack('>H', (txid & 0x0FF0) >> 4)
      scatter_val = txid & 0x0F
      return scatter_val, commit_version_bytes + batch_order_bytes
  
  
  encode_read_version = lambda read_version: Int64.encode_bare(
    read_version, READ_VERSION_SIZE)
  
  
  def encode_versionstamp_index(versionstamp_position):
    """ Encodes an FDB key index position. """
    return struct.pack(u'<L', versionstamp_position)
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/fdb/codecs.py
files: """
  This module contains shared encoding and decoding utilities for translating
  datastore types to bytestrings and vice versa.
  """
  import bisect
  import logging
  import struct
  import sys
  
  import six
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.dbconstants import BadRequest, InternalError
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.datastore import entity_pb
  
  # The number of bytes used to encode a read version in a way that is comparable
  # to commit versionstamps.
  READ_VERSION_SIZE = 8
  
  # Byte values that signify a data type. A couple values are left between each
  # one just in case they need adjustment.
  NULL_CODE = 0x01
  # 17 values are reserved for integers in order to allow variable size in both a
  # negative and positive direction.
  INT64_ZERO_CODE = 0x0C
  FALSE_CODE = 0x17
  TRUE_CODE = 0x18
  BYTES_CODE = 0x1B
  DOUBLE_CODE = 0x1E
  POINT_CODE = 0x21
  USER_CODE = 0x24
  REFERENCE_CODE = 0x27
  
  # Ensures the shorter of two variable-length values (with identical prefixes)
  # is placed before the longer one. Otherwise, the following byte(s) could
  # determine the sort order. It also allows a decoder to find the end of the
  # value.
  TERMINATOR = 0x00
  
  logger = logging.getLogger(__name__)
  
  
  def reverse_bits(blob):
    return b''.join(map(lambda x: six.int2byte(x ^ 0xFF), six.iterbytes(blob)))
  
  
  def encode_marker(marker, reverse):
    return six.int2byte(marker ^ 0xFF) if reverse else six.int2byte(marker)
  
  
  def decode_str(string):
    """ Converts byte strings to unicode strings. """
    if isinstance(string, six.text_type):
      return string
  
    return string.decode('utf-8')
  
  
  class Int64(object):
    """
    Handles encoding and decoding operations for 64-bit integers. The encoded
    value is variable-length. The value of the marker indicates how many bytes
    were used to encode the value.
    """
    LIMITS = tuple((1 << (i * 8)) - 1 for i in range(9))
  
    @classmethod
    def encode(cls, value, reverse=False):
      if value == 0:
        code = INT64_ZERO_CODE
        packed = b''
      elif value > 0:
        encoded_size = bisect.bisect_left(cls.LIMITS, value)
        adjusted_value = cls.LIMITS[encoded_size] - value if reverse else value
        code = INT64_ZERO_CODE + encoded_size
        packed = struct.pack('>Q', adjusted_value)[-encoded_size:]
      else:
        encoded_size = bisect.bisect_left(cls.LIMITS, -value)
        # Shift negative values to an unsigned space.
        adjusted_value = -value if reverse else cls.LIMITS[encoded_size] + value
        code = INT64_ZERO_CODE - encoded_size
        packed = struct.pack('>Q', adjusted_value)[-encoded_size:]
  
      return encode_marker(code, reverse) + packed
  
    @classmethod
    def decode(cls, marker, blob, pos, reverse=False):
      code = ord(marker) ^ 0xFF if reverse else ord(marker)
      if code == INT64_ZERO_CODE:
        return 0, pos
  
      encoded_size = abs(code - INT64_ZERO_CODE)
      packed = b'\x00' * (8 - encoded_size) + blob[pos:pos + encoded_size]
      pos += encoded_size
      value = struct.unpack('>Q', packed)[0]
      if code > INT64_ZERO_CODE:
        value = cls.LIMITS[encoded_size] - value if reverse else value
      else:
        # Shift values back to their original, negative state.
        value = -value if reverse else value - cls.LIMITS[encoded_size]
  
      return value, pos
  
    @classmethod
    def encode_bare(cls, value, byte_count):
      """
      Encodes an integer without a prefix using the specified number of bytes.
      """
      encoded = struct.pack('>Q', value)
      if any(byte != b'\x00' for byte in encoded[:-byte_count]):
        raise InternalError(u'Value exceeds maximum size')
  
      return encoded[-byte_count:]
  
    @classmethod
    def decode_bare(cls, encoded_value):
      """ Decodes a byte string back to an integer. """
      encoded_value = b'\x00' * (8 - len(encoded_value)) + encoded_value
      return struct.unpack('>Q', encoded_value)[0]
  
  
  class Bytes(object):
    """
    Handles encoding and decoding operations for arbitrary blobs. Since the
    length of the value is variable and there is usually data that follows the
    encoded value, a terminator byte is used to indicate the end of the value.
    Any occurrence of the terminator byte within the value is escaped in order to
    preserve the ordering and allow the parser to find the real terminator.
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(BYTES_CODE), reverse=False):
      packed = reverse_bits(value) if reverse else value
      terminator = encode_marker(TERMINATOR, reverse)
      # Replace each occurrence of the terminator with a sequence that maintains
      # the sort order. In order for the parser to be able to find the end of the
      # value, directories that use this codec must ensure that the first byte
      # that follows the encoded Bytes object must not be 0x00 or 0xFF.
      if reverse:
        # The preceding 0xFE ensures a terminator character within the value
        # compares as less than the real terminator. For example,
        # 0x0200 -> 0xFDFF -> 0xFDFEFF00FF should come before
        # 0x02   -> 0xFD   -> 0xFDFF
        # The trailing 0x00 ensures that the parser is able to differentiate
        # between a replaced value and the real terminator.
        packed = packed.replace(terminator, b'\xFE\xFF\x00')
      else:
        # A trailing 0xFF is enough for the parser to find the terminator and
        # to maintain sort order despite the values that follow. For example
        # (assuming 0x10 is the value that follows),
        # 0x00, 0x10   -> 0x00FF0010     (not 0x000010) should come before
        # 0x0000, 0x10 -> 0x00FF00FF0010 (not 0x00000010)
        packed = packed.replace(terminator, b'\x00\xFF')
  
      return prefix + packed + terminator
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      end = cls._find_terminator(blob, pos, reverse)
      packed = blob[pos:end]
      terminator = encode_marker(TERMINATOR, reverse)
      if reverse:
        packed = packed.replace(b'\xFE\xFF\x00', terminator)
      else:
        packed = packed.replace(b'\x00\xFF', terminator)
  
      value = reverse_bits(packed) if reverse else packed
      return value, end + 1
  
    @staticmethod
    def _find_terminator(blob, pos, reverse=False):
      """ Finds the position of the terminator. """
      terminator = encode_marker(TERMINATOR, reverse)
      escape_byte = b'\x00' if reverse else b'\xFF'
      while True:
        pos = blob.find(terminator, pos)
        if pos < 0:
          raise InternalError(u'Byte string is missing terminator')
  
        if blob[pos + 1:pos + 2] != escape_byte:
          return pos
  
        pos += 2
  
  
  class Double(object):
    """
    Handles encoding and decoding operations for floating point values. 8 bytes
    (plus the marker byte) are always used.
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(DOUBLE_CODE), reverse=False):
      packed = struct.pack('>d', value)
      # The first bit of the packed value indicates whether it's negative or not.
      positive = six.indexbytes(packed, 0) & 0x80 == 0x00
      if positive:
        # Flip the sign bit.
        packed = six.int2byte(six.indexbytes(packed, 0) ^ 0x80) + packed[1:]
        if reverse:
          packed = reverse_bits(packed)
      else:
        # Flip all the bits unless the sort order is reversed.
        packed = packed if reverse else reverse_bits(packed)
  
      return prefix + packed
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      packed = blob[pos:pos + 8]
      pos += 8
      first_bit = six.indexbytes(packed, 0) & 0x80 == 0x80
      positive = first_bit if not reverse else not first_bit
      if positive:
        if reverse:
          packed = reverse_bits(packed)
  
        # Restore the sign bit.
        packed = six.int2byte(six.indexbytes(packed, 0) ^ 0x80) + packed[1:]
      else:
        # Restore all the original bits for ascending values.
        packed = packed if reverse else reverse_bits(packed)
  
      value = struct.unpack('>d', packed)[0]
      return value, pos
  
  
  class Point(object):
    """
    Handles encoding and decoding operations for point values. The Double codec
    is used for each coordinate (x and y).
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(POINT_CODE), reverse=False):
      x_packed = Double.encode(value.x(), prefix=b'', reverse=reverse)
      y_packed = Double.encode(value.y(), prefix=b'', reverse=reverse)
      return prefix + x_packed + y_packed
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      x, pos = Double.decode(blob, pos, reverse)
      y, pos = Double.decode(blob, pos, reverse)
      point_val = entity_pb.PropertyValue_PointValue()
      point_val.set_x(x)
      point_val.set_y(y)
      return point_val, pos
  
  
  class Text(object):
    """
    Handles encoding and decoding operations for unicode strings. Since the
    length of the value is variable and there is usually data that follows the
    encoded value, a terminator byte is used to indicate the end of the value.
    """
    @classmethod
    def encode(cls, unicode_string, prefix=b'', reverse=False):
      # Ensure the encoded value does not contain the terminator. UTF-8 does not
      # use 0xFF, so this can be done without exceeding the largest byte value.
      if reverse:
        encode_byte = lambda x: six.int2byte(x + 1 ^ 0xFF)
      else:
        encode_byte = lambda x: six.int2byte(x + 1)
  
      encoded = b''.join(
        map(encode_byte, six.iterbytes(unicode_string.encode('utf-8'))))
      terminator = encode_marker(TERMINATOR, reverse)
      return prefix + encoded + terminator
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      terminator = TERMINATOR ^ 0xFF if reverse else TERMINATOR
      byte_array = bytearray()
      while pos < len(blob):
        byte_value = six.indexbytes(blob, pos)
        pos += 1
        if byte_value == terminator:
          break
  
        if reverse:
          byte_value ^= 0xFF
  
        # Shift the byte back to its original value.
        byte_array.append(byte_value - 1)
  
      return byte_array.decode('utf-8'), pos
  
  
  class User(object):
    """
    Handles encoding and decoding operations for User values. The Double codec
    is used for each of the required fields (email and auth_domain).
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(POINT_CODE), reverse=False):
      packed = (Text.encode(decode_str(value.email()), reverse=reverse) +
                Text.encode(decode_str(value.auth_domain()), reverse=reverse))
      return prefix + packed
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      email, pos = Text.decode(blob, pos, reverse)
      auth_domain, pos = Text.decode(blob, pos, reverse)
      user_val = entity_pb.PropertyValue_UserValue()
      user_val.set_email(email)
      user_val.set_auth_domain(auth_domain)
      return user_val, pos
  
  
  class Path(object):
    """
    Handles encoding and decoding operations for entity paths. The Text codec
    is used for kinds and names. The Int64 codec is used for IDs.
    """
    KIND_MARKER = 0x1C
  
    MIN_ID_MARKER = INT64_ZERO_CODE - 8
  
    # Marks an item as a name rather than an ID. These should be be placed after
    # entries with IDs.
    NAME_MARKER = 0x1D
  
    @classmethod
    def pack(cls, path, prefix=b'', omit_terminator=False, reverse=False):
      if not isinstance(path, tuple):
        path = cls.flatten(path)
  
      encoded_items = []
      kind_marker = encode_marker(cls.KIND_MARKER, reverse)
      for index in range(0, len(path), 2):
        kind = path[index]
        id_or_name = path[index + 1]
        encoded_items.append(Text.encode(kind, kind_marker, reverse))
        encoded_items.append(cls.encode_id_or_name(id_or_name, reverse))
  
      terminator = b'' if omit_terminator else encode_marker(TERMINATOR, reverse)
      return b''.join([prefix] + encoded_items + [terminator])
  
    @classmethod
    def encode_id_or_name(cls, id_or_name, reverse=False):
      if isinstance(id_or_name, six.text_type):
        name_marker = encode_marker(cls.NAME_MARKER, reverse)
        return Text.encode(id_or_name, name_marker, reverse)
      elif isinstance(id_or_name, int):
        return Int64.encode(id_or_name, reverse)
      else:
        raise BadRequest(u'Invalid path element type')
  
    @classmethod
    def unpack(cls, blob, pos, reverse=False):
      items = []
      terminator = encode_marker(TERMINATOR, reverse)
      kind_marker = encode_marker(cls.KIND_MARKER, reverse)
      name_marker = encode_marker(cls.NAME_MARKER, reverse)
      while pos < len(blob):
        marker = blob[pos]
        pos += 1
        if marker == terminator:
          break
  
        if marker != kind_marker:
          raise InternalError(u'Encoded path is missing kind')
  
        kind, pos = Text.decode(blob, pos, reverse)
        items.append(kind)
  
        marker = blob[pos]
        pos += 1
        if marker == name_marker:
          elem_name, pos = Text.decode(blob, pos, reverse)
          items.append(elem_name)
        else:
          elem_id, pos = Int64.decode(marker, blob, pos, reverse)
          items.append(elem_id)
  
      return tuple(items), pos
  
    @staticmethod
    def flatten(path):
      """ Converts a key path protobuf object to a tuple. """
      if isinstance(path, entity_pb.PropertyValue):
        element_list = path.referencevalue().pathelement_list()
      elif isinstance(path, entity_pb.PropertyValue_ReferenceValue):
        element_list = path.pathelement_list()
      else:
        element_list = path.element_list()
  
      return tuple(item for element in element_list
                   for item in Path.encode_element(element))
  
    @staticmethod
    def decode(flat_path, reference_value=False):
      """ Converts a tuple to a key path protobuf object. """
      if reference_value:
        path = entity_pb.PropertyValue_ReferenceValue()
      else:
        path = entity_pb.Path()
  
      for index in range(0, len(flat_path), 2):
        if reference_value:
          element = path.add_pathelement()
        else:
          element = path.add_element()
  
        element.set_type(flat_path[index])
        id_or_name = flat_path[index + 1]
        if isinstance(id_or_name, int):
          element.set_id(id_or_name)
        else:
          element.set_name(id_or_name.encode('utf-8'))
  
      return path
  
    @staticmethod
    def encode_element(element):
      """ Converts a path element protobuf object to a tuple. """
      if element.has_id():
        id_or_name = int(element.id())
      elif element.has_name():
        id_or_name = decode_str(element.name())
      else:
        raise BadRequest(u'All path elements must either have a name or ID')
  
      return decode_str(element.type()), id_or_name
  
    @staticmethod
    def decode_element(element_tuple):
      """ Converts a tuple to a path element protobuf object. """
      path_element = entity_pb.Path_Element()
      path_element.set_type(element_tuple[0])
      if isinstance(element_tuple[1], int):
        path_element.set_id(element_tuple[1])
      else:
        path_element.set_name(element_tuple[1])
  
      return path_element
  
  
  class Reference(object):
    """
    Handles encoding and decoding operations for entity keys. The Text codec is
    used for the project ID and namespace. The Path codec is used for the key's
    path.
    """
    @classmethod
    def encode(cls, value, prefix=six.int2byte(REFERENCE_CODE), reverse=False):
      return b''.join([
        prefix,
        Text.encode(decode_str(value.app()), reverse=reverse),
        Text.encode(decode_str(value.name_space()), reverse=reverse),
        Path.pack(Path.flatten(value), reverse=reverse)
      ])
  
    @classmethod
    def decode(cls, blob, pos, reverse=False):
      project_id, pos = Text.decode(blob, pos, reverse)
      namespace, pos = Text.decode(blob, pos, reverse)
      flat_path, pos = Path.unpack(blob, pos, reverse)
      reference_val = entity_pb.PropertyValue_ReferenceValue()
      reference_val.set_app(project_id)
      reference_val.set_name_space(namespace)
      reference_val.MergeFrom(Path.decode(flat_path, reference_value=True))
      return reference_val, pos
  
  
  def encode_value(value, reverse=False):
    if isinstance(value, six.integer_types):
      return Int64.encode(value, reverse)
  
    if value.has_int64value():
      return Int64.encode(value.int64value(), reverse)
  
    if value.has_booleanvalue():
      to_encode = TRUE_CODE if value.booleanvalue() else FALSE_CODE
      return encode_marker(to_encode, reverse)
  
    if value.has_stringvalue():
      prefix = encode_marker(BYTES_CODE, reverse)
      return Bytes.encode(value.stringvalue(), prefix, reverse)
  
    if value.has_doublevalue():
      prefix = encode_marker(DOUBLE_CODE, reverse)
      return Double.encode(value.doublevalue(), prefix, reverse)
  
    if value.has_pointvalue():
      prefix = encode_marker(POINT_CODE, reverse)
      return Point.encode(value.pointvalue(), prefix, reverse)
  
    if value.has_uservalue():
      prefix = encode_marker(USER_CODE, reverse)
      return User.encode(value.uservalue(), prefix, reverse)
  
    if value.has_referencevalue():
      prefix = encode_marker(REFERENCE_CODE, reverse)
      return Reference.encode(value.referencevalue(), prefix, reverse)
  
    return encode_marker(NULL_CODE, reverse)
  
  
  def decode_value(blob, pos, reverse=False):
    prop_value = entity_pb.PropertyValue()
    marker = blob[pos]
    pos += 1
    code = ord(marker) ^ 0xFF if reverse else ord(marker)
    if code == NULL_CODE:
      pass
    elif INT64_ZERO_CODE - 8 <= code <= INT64_ZERO_CODE + 8:
      int_val, pos = Int64.decode(marker, blob, pos, reverse)
      prop_value.set_int64value(int_val)
    elif code in (TRUE_CODE, FALSE_CODE):
      prop_value.set_booleanvalue(code == TRUE_CODE)
    elif code == BYTES_CODE:
      bytes_val, pos = Bytes.decode(blob, pos, reverse)
      prop_value.set_stringvalue(bytes_val)
    elif code == DOUBLE_CODE:
      double_val, pos = Double.decode(blob, pos, reverse)
      prop_value.set_doublevalue(double_val)
    elif code == POINT_CODE:
      point_val, pos = Point.decode(blob, pos, reverse)
      prop_value.mutable_pointvalue().MergeFrom(point_val)
    elif code == USER_CODE:
      user_val, pos = User.decode(blob, pos, reverse)
      prop_value.mutable_uservalue().MergeFrom(user_val)
    elif code == REFERENCE_CODE:
      ref_val, pos = Reference.decode(blob, pos, reverse)
      prop_value.mutable_referencevalue().MergeFrom(ref_val)
  
    return prop_value, pos
  
  
  class TransactionID(object):
    """
    Handles encoding and decoding operations for transaction IDs. Since this
    cannot handle every commit versionstamp value, it will be removed as soon as
    the interface can handle larger values for transaction handles.
    """
    COMMIT_VERSION_BITS = 8 * 7 - 4
  
    BATCH_ORDER_BITS = 8
  
    @classmethod
    def encode(cls, scatter_val, commit_versionstamp):
      commit_version = struct.unpack('>Q', commit_versionstamp[:8])[0]
      batch_order = struct.unpack('>H', commit_versionstamp[8:])[0]
      if not 0 <= scatter_val <= 15:
        raise InternalError(u'Invalid scatter value')
  
      if commit_version >= 2 ** cls.COMMIT_VERSION_BITS:
        raise InternalError(u'Commit version too high')
  
      if batch_order >= 2 ** cls.BATCH_ORDER_BITS:
        raise InternalError(u'Batch order too high')
  
      return (commit_version << 12) + (batch_order << 4) + scatter_val
  
    @classmethod
    def decode(cls, txid):
      commit_version_bytes = struct.pack('>Q', txid >> 12)
      batch_order_bytes = struct.pack('>H', (txid & 0x0FF0) >> 4)
      scatter_val = txid & 0x0F
      return scatter_val, commit_version_bytes + batch_order_bytes
  
  
  encode_read_version = lambda read_version: Int64.encode_bare(
    read_version, READ_VERSION_SIZE)
  
  
  def encode_versionstamp_index(versionstamp_position):
    """ Encodes an FDB key index position. """
    return struct.pack(u'<L', versionstamp_position)
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/fdb/codecs.py
files: """
  This module stores and retrieves entity data as well as the metadata needed to
  achieve snapshot isolation during transactions. The DataManager is the main
  interface that clients can use to interact with the data layer. See its
  documentation for implementation details.
  """
  from __future__ import division
  import logging
  import math
  import sys
  
  import six
  import six.moves as sm
  from tornado import gen
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.fdb.codecs import (
    decode_str, encode_versionstamp_index, Int64, Path, Text)
  from appscale.datastore.fdb.utils import (
    ABSENT_VERSION, DS_ROOT, fdb, hash_tuple, ResultIterator, VERSIONSTAMP_SIZE)
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.datastore import entity_pb
  
  logger = logging.getLogger(__name__)
  
  first_gt_or_equal = fdb.KeySelector.first_greater_or_equal
  
  
  class VersionEntry(object):
    """ Encapsulates details for an entity version. """
    INCOMPLETE = 1
    __slots__ = ['project_id', 'namespace', 'path', 'commit_versionstamp',
                 'version', '_encoded_entity', '_decoded_entity']
  
    def __init__(self, project_id, namespace, path, commit_versionstamp=None,
                 version=None, encoded_entity=None):
      self.project_id = project_id
      self.namespace = namespace
      self.path = path
      self.commit_versionstamp = commit_versionstamp
      self.version = ABSENT_VERSION if version is None else version
      self._encoded_entity = encoded_entity
      self._decoded_entity = None
  
    def __repr__(self):
      # Since the encoded entity can be large, it probably does not make sense to
      # include it in the string representation.
      blob_repr = self._encoded_entity
      if blob_repr is not None:
        blob_repr = u'<bytes object with length={}>'.format(len(blob_repr))
  
      return u'VersionEntry({!r}, {!r}, {!r}, {!r}, {!r}, {})'.format(
        self.project_id, self.namespace, self.path, self.commit_versionstamp,
        self.version, blob_repr)
  
    @property
    def present(self):
      return self.commit_versionstamp is not None
  
    @property
    def complete(self):
      return self._encoded_entity != self.INCOMPLETE
  
    @property
    def has_entity(self):
      return bool(self._encoded_entity)
  
    @property
    def key(self):
      key = entity_pb.Reference()
      key.set_app(self.project_id)
      if self.namespace is not None:
        key.set_name_space(self.namespace)
  
      key.mutable_path().MergeFrom(Path.decode(self.path))
      return key
  
    @property
    def encoded(self):
      if not self.complete:
        raise ValueError(u'Version entry is not complete')
  
      return self._encoded_entity
  
    @property
    def decoded(self):
      if not self.has_entity:
        return None
  
      if self._decoded_entity is None:
        self._decoded_entity = entity_pb.EntityProto(self.encoded)
  
      return self._decoded_entity
  
    @classmethod
    def from_key(cls, key):
      project_id = decode_str(key.app())
      namespace = None
      if key.has_name_space():
        namespace = decode_str(key.name_space())
  
      path = Path.flatten(key.path())
      return cls(project_id, namespace, path)
  
  
  class DataNamespace(object):
    """
    A DataNamespace handles the encoding and decoding details for entity data
    for a specific project_id/namespace combination.
  
    The directory path looks like (<project-dir>, 'data', <namespace>).
  
    Within this directory, keys are encoded as
    <scatter-byte> + <path> + <commit-versionstamp> + <index>.
  
    The <scatter-byte> is a single byte determined by hashing the entity path.
    Its purpose is to spread writes more evenly across the cluster and minimize
    hotspots.
  
    The <path> contains the entity path. See codecs.Path for encoding details.
  
    The <commit-versionstamp> is a 10-byte versionstamp that specifies the commit version
    of the transaction that wrote the entity data.
  
    The <index> is a single byte specifying the position of the value's chunk.
  
    Values are encoded as <entity-encoding> + <entity> + <entity-version>.
  
    The <entity-encoding> is a single byte specifying the encoding scheme of the
    entity to follow.
  
    The <entity> is an encoded protobuffer value.
  
    The <entity-version> is an integer specifying the approximate insert
    timestamp in microseconds (according to the client performing the insert).
    Though there is a one-to-one mapping of commit versionstamps to entity
    versions, the datastore uses a different value for the entity version in
    order to satisfy the 8-byte constraint and to follow the GAE convention of
    the value representing a timestamp. It is encoded using 7 bytes.
  
    Since encoded values can exceed the size limit imposed by FoundationDB,
    values encoded values are split into chunks. Each chunk is stored as a
    Key-Value and ordered by a unique <index> byte.
    """
    DIR_NAME = u'data'
  
    # The max number of bytes for each FDB value.
    _CHUNK_SIZE = 10000
  
    # The number of bytes used to store an entity version.
    _VERSION_SIZE = 7
  
    # The number of bytes used to encode the chunk index.
    _INDEX_SIZE = 1
  
    # Indicates the encoded blob is a V3 entity object.
    _V3_MARKER = 0x01
  
    def __init__(self, directory):
      self.directory = directory
  
    @property
    def project_id(self):
      return self.directory.get_path()[len(DS_ROOT)]
  
    @property
    def namespace(self):
      return self.directory.get_path()[len(DS_ROOT) + 2]
  
    @property
    def path_slice(self):
      """ The portion of keys that contain the encoded path. """
      return slice(len(self.directory.rawPrefix) + 1,
                   -1 * (VERSIONSTAMP_SIZE + self._INDEX_SIZE))
  
    @property
    def versionstamp_slice(self):
      """ The portion of keys that contain the commit versionstamp. """
      return slice(self.path_slice.stop,
                   self.path_slice.stop + VERSIONSTAMP_SIZE)
  
    @property
    def index_slice(self):
      """ The portion of keys that contain the chunk index. """
      return slice(-1 * self._INDEX_SIZE, None)
  
    @property
    def encoding_slice(self):
      """ The portion of values that specify the entity encoding type. """
      return slice(0, 1)
  
    @property
    def entity_slice(self):
      """ The portion of values that contain the encoded entity. """
      return slice(1, -self._VERSION_SIZE)
  
    @property
    def version_slice(self):
      """ The portion of values that contain the entity version. """
      return slice(-self._VERSION_SIZE, None)
  
    @classmethod
    def directory_path(cls, project_id, namespace):
      return project_id, cls.DIR_NAME, namespace
  
    def encode(self, path, entity, version):
      """ Encodes a tuple of Key-Value tuples for a given version entry.
  
      Args:
        path: A tuple or protobuf path object.
        entity: An encoded entity or protobuf object.
        version: An integer specifying the new entity version.
      Returns:
        A tuple of Key-Value tuples suitable for writing in an FDB transaction.
      """
      if isinstance(entity, entity_pb.EntityProto):
        entity = entity.Encode()
  
      encoded_version = Int64.encode_bare(version, self._VERSION_SIZE)
      if not entity:
        return ((self.encode_key(path, commit_versionstamp=None, index=0),
                 encoded_version),)
  
      value = b''.join([six.int2byte(self._V3_MARKER), entity])
      chunk_count = int(math.ceil(len(value) / self._CHUNK_SIZE))
      chunks = [
        value[slice(index * self._CHUNK_SIZE, (index + 1) * self._CHUNK_SIZE)]
        for index in sm.range(chunk_count)]
      # Place the version at the end of the last chunk. Though this allows the
      # last chunk to exceed the chunk size, it ensures that the entity version
      # can always be retrieved from the last chunk.
      chunks[-1] += encoded_version
      return tuple(
        (self.encode_key(path, commit_versionstamp=None, index=index), chunk)
        for index, chunk in enumerate(chunks))
  
    def encode_key(self, path, commit_versionstamp, index):
      """ Encodes a key for the given version entry.
  
      Args:
        path: A tuple or protobuf path object.
        commit_versionstamp: A 10-byte string specifying the version's commit
          versionstamp or None.
        index: An integer specifying the chunk index.
      Returns:
        A string containing an FDB key. If commit_versionstamp was None, the key
        should be used with set_versionstamped_key.
      """
      encoded_key = b''.join([self._encode_path_prefix(path),
                              commit_versionstamp or b'\x00' * VERSIONSTAMP_SIZE,
                              six.int2byte(index)])
      if not commit_versionstamp:
        versionstamp_index = (len(encoded_key) -
                              (VERSIONSTAMP_SIZE + self._INDEX_SIZE))
        encoded_key += encode_versionstamp_index(versionstamp_index)
  
      return encoded_key
  
    def decode(self, kvs):
      """ Decodes Key-Values to a version entry.
  
      Args:
        kvs: An iterable containing KeyValue objects.
      Returns:
        A VersionEntry object.
      """
      path = Path.unpack(kvs[0].key, self.path_slice.start)[0]
      commit_versionstamp = kvs[0].key[self.versionstamp_slice]
      first_index = ord(kvs[0].key[self.index_slice])
  
      version = Int64.decode_bare(kvs[-1].value[self.version_slice])
      if first_index == 0:
        encoded_entity = b''.join([kv.value for kv in kvs])[self.entity_slice]
      else:
        encoded_entity = VersionEntry.INCOMPLETE
  
      return VersionEntry(self.project_id, self.namespace, path,
                          commit_versionstamp, version, encoded_entity)
  
    def get_slice(self, path, commit_versionstamp=None, read_versionstamp=None):
      """ Gets the range of keys relevant to the given constraints.
  
      Args:
        path: A tuple or protobuf path object.
        commit_versionstamp: The commit versionstamp for a specific entity
          version.
        read_versionstamp: The transaction's read versionstamp. All newer entity
          versions are ignored.
      Returns:
        A slice specifying the start and stop keys.
      """
      path_prefix = self._encode_path_prefix(path)
      if commit_versionstamp is not None:
        # All chunks for a given version.
        prefix = path_prefix + commit_versionstamp
        return slice(first_gt_or_equal(prefix + b'\x00'),
                     first_gt_or_equal(prefix + b'\xFF'))
  
      if read_versionstamp is not None:
        # All versions for a given path except those written after the
        # read_versionstamp.
        version_prefix = path_prefix + read_versionstamp
        return slice(first_gt_or_equal(path_prefix + b'\x00'),
                     first_gt_or_equal(version_prefix + b'\xFF'))
  
      # All versions for a given path.
      return slice(first_gt_or_equal(path_prefix + b'\x00'),
                   first_gt_or_equal(path_prefix + b'\xFF'))
  
    def _encode_path_prefix(self, path):
      """ Encodes the portion of the key up to and including the path.
  
      Args:
        path: A tuple or protobuf path object.
      Returns:
        A string containing the path prefix.
      """
      if not isinstance(path, tuple):
        path = Path.flatten(path)
  
      return b''.join([self.directory.rawPrefix, hash_tuple(path),
                       Path.pack(path)])
  
  
  class GroupUpdatesNS(object):
    """
    A GroupUpdatesNS handles the encoding and decoding details for commit
    versionstamps for each entity group. These are used to materialize conflicts
    for transactions that involve ancestory queries on the same entity groups.
  
    The directory path looks like (<project-dir>, 'group-updates', <namespace>).
  
    Within this directory, keys are encoded as <scatter-byte> + <group-path>.
  
    The <scatter-byte> is a single byte determined by hashing the group path.
    Its purpose is to spread writes more evenly across the cluster and minimize
    hotspots.
  
    The <group-path> contains the entity group path.
  
    Values are 10-byte strings that specify the latest commit version for the
    entity group.
    """
    DIR_NAME = u'group-updates'
  
    def __init__(self, directory):
      self.directory = directory
  
    @classmethod
    def directory_path(cls, project_id, namespace):
      return project_id, cls.DIR_NAME, namespace
  
    def encode(self, path):
      """ Creates a Key-Value tuple for updating a group's commit versionstamp.
  
      Args:
        path: A tuple or protobuf path object.
      Returns:
        A (key, value) tuple suitable for set_versionstamped_value.
      """
      if not isinstance(path, tuple):
        path = Path.flatten(path)
  
      group_path = path[:2]
      return (self.encode_key(group_path),
              b'\x00' * VERSIONSTAMP_SIZE + encode_versionstamp_index(0))
  
    def encode_key(self, group_path):
      """ Encodes a key for a given entity group.
  
      Args:
        group_path: A tuple containing path elements.
      Returns:
        A byte string containing the relevant FDB key.
      """
      return b''.join([
        self.directory.rawPrefix, hash_tuple(group_path),
        Text.encode(group_path[0]) + Path.encode_id_or_name(group_path[1])])
  
  
  class DataManager(object):
    """
    The DataManager is the main interface that clients can use to interact with
    the data layer. It makes use of the DataNamespace and GroupUpdateNS
    namespaces to handle the encoding and decoding details when satisfying
    requests. When a client requests data, the DataManager encapsulates entity
    data in a VersionEntry object.
  
    See the DataNamespace and GroupUpdateNS classes for implementation details
    about how data is stored and retrieved.
    """
    def __init__(self, tornado_fdb, directory_cache):
      self._tornado_fdb = tornado_fdb
      self._directory_cache = directory_cache
  
    @gen.coroutine
    def get_latest(self, tr, key, read_versionstamp=None, include_data=True,
                   snapshot=False):
      """ Gets the newest entity version for the given read versionstamp.
  
      Args:
        tr: An FDB transaction.
        key: A protubuf reference object.
        read_versionstamp: A 10-byte string specifying the FDB read versionstamp.
          Newer versionstamps are ignored.
        include_data: A boolean specifying whether or not to fetch all of the
          entity's Key-Values.
        snapshot: If True, the read will not cause a transaction conflict.
      """
      data_ns = yield self._data_ns_from_key(tr, key)
      desired_slice = data_ns.get_slice(
        key.path(), read_versionstamp=read_versionstamp)
      last_entry = yield self._last_version(
        tr, data_ns, desired_slice, include_data, snapshot=snapshot)
      if last_entry is None:
        last_entry = VersionEntry(data_ns.project_id, data_ns.namespace,
                                  Path.flatten(key.path()))
  
      raise gen.Return(last_entry)
  
    @gen.coroutine
    def get_entry(self, tr, index_entry, snapshot=False):
      """ Gets the entity data from an index entry.
  
      Args:
        tr: An FDB transaction.
        index_entry: An IndexEntry object.
        snapshot: If True, the read will not cause a transaction conflict.
      Returns:
        A VersionEntry or None.
      """
      if index_entry.kind in (u'__namespace__', u'__kind__'):
        entity = entity_pb.EntityProto()
        entity.mutable_key().MergeFrom(index_entry.key)
        entity.mutable_entity_group()
        version_entry = VersionEntry(
          index_entry.project_id, index_entry.namespace, index_entry.path,
          encoded_entity=entity.Encode())
        raise gen.Return(version_entry)
      elif index_entry.kind == u'__property__':
        entity = index_entry.prop_result()
        entity.clear_entity_group()
        entity.mutable_entity_group()
        version_entry = VersionEntry(
          index_entry.project_id, index_entry.namespace, index_entry.path,
          encoded_entity=entity.Encode())
        raise gen.Return(version_entry)
  
      version_entry = yield self.get_version_from_path(
        tr, index_entry.project_id, index_entry.namespace, index_entry.path,
        index_entry.commit_versionstamp, snapshot)
      raise gen.Return(version_entry)
  
    @gen.coroutine
    def get_version_from_path(self, tr, project_id, namespace, path,
                              commit_versionstamp, snapshot=False):
      """ Gets the entity data for a specific version.
  
      Args:
        tr: An FDB transaction.
        project_id: A string specifying the project ID.
        namespace: A string specifying the namespace.
        path: A tuple or protobuf path object.
        commit_versionstamp: A 10-byte string specyfing the FDB commit
          versionstamp.
        snapshot: If True, the read will not cause a transaction conflict.
      Returns:
        A VersionEntry or None.
      """
      data_ns = yield self._data_ns(tr, project_id, namespace)
      desired_slice = data_ns.get_slice(path, commit_versionstamp)
      results = yield ResultIterator(tr, self._tornado_fdb, desired_slice,
                                     snapshot=snapshot).list()
      raise gen.Return(data_ns.decode(results))
  
    @gen.coroutine
    def last_group_versionstamp(self, tr, project_id, namespace, group_path):
      """ Gets the most recent commit versionstamp for the entity group.
  
      Args:
        tr: An FDB transaction.
        project_id: A string specifying the project ID.
        namespace: A string specifying the namespace.
        group_path: A tuple containing the group's path elements.
      Returns:
        A 10-byte string specifying the versionstamp or None.
      """
      group_ns = yield self._group_updates_ns(tr, project_id, namespace)
      last_updated_versionstamp = yield self._tornado_fdb.get(
        tr, group_ns.encode_key(group_path))
      if not last_updated_versionstamp.present():
        return
  
      raise gen.Return(last_updated_versionstamp.value)
  
    @gen.coroutine
    def put(self, tr, key, version, encoded_entity):
      """ Writes a new version entry and updates the entity group versionstamp.
  
      Args:
        tr: An FDB transaction.
        key: A protobuf reference object.
        version: An integer specifying the new entity version.
        encoded_entity: A string specifying the encoded entity data.
      """
      data_ns = yield self._data_ns_from_key(tr, key)
      for fdb_key, val in data_ns.encode(key.path(), encoded_entity, version):
        tr.set_versionstamped_key(fdb_key, val)
  
      group_ns = yield self._group_updates_ns(
        tr, data_ns.project_id, data_ns.namespace)
      tr.set_versionstamped_value(*group_ns.encode(key.path()))
  
    @gen.coroutine
    def hard_delete(self, tr, version_entry):
      """ Deletes a version entry. Only the GC should use this.
  
      Args:
        tr: An FDB transaction.
        version_entry: A VersionEntry object.
      """
      data_ns = yield self._data_ns(
        tr, version_entry.project_id, version_entry.namespace)
      version_slice = data_ns.get_slice(version_entry.path,
                                        version_entry.commit_versionstamp)
      del tr[version_slice.start.key:version_slice.stop.key]
  
    @gen.coroutine
    def _data_ns(self, tr, project_id, namespace):
      directory = yield self._directory_cache.get(
        tr, DataNamespace.directory_path(project_id, namespace))
      raise gen.Return(DataNamespace(directory))
  
    @gen.coroutine
    def _data_ns_from_key(self, tr, key):
      project_id = decode_str(key.app())
      namespace = decode_str(key.name_space())
      data_ns = yield self._data_ns(tr, project_id, namespace)
      raise gen.Return(data_ns)
  
    @gen.coroutine
    def _group_updates_ns(self, tr, project_id, namespace):
      directory = yield self._directory_cache.get(
        tr, GroupUpdatesNS.directory_path(project_id, namespace))
      raise gen.Return(GroupUpdatesNS(directory))
  
    @gen.coroutine
    def _last_version(self, tr, data_ns, desired_slice, include_data=True,
                      snapshot=False):
      """ Gets the most recent entity data for a given slice.
  
      Args:
        tr: An FDB transaction.
        data_ns: A DataNamespace.
        desired_slice: A slice specifying the start and stop keys.
        include_data: A boolean indicating that all chunks should be fetched.
        snapshot: If True, the read will not cause a transaction conflict.
      Returns:
        A VersionEntry or None.
      """
      results, count, more_results = yield self._tornado_fdb.get_range(
        tr, desired_slice, limit=1, reverse=True, snapshot=snapshot)
  
      if not results:
        return
  
      last_chunk = results[0]
      entry = data_ns.decode([last_chunk])
      if not include_data or not entry.present or entry.complete:
        raise gen.Return(entry)
  
      # Retrieve the remaining chunks.
      version_slice = data_ns.get_slice(entry.path, entry.commit_versionstamp)
      remaining = slice(version_slice.start, first_gt_or_equal(last_chunk.key))
      results = yield ResultIterator(tr, self._tornado_fdb, remaining).list()
      raise gen.Return(data_ns.decode(results + [last_chunk]))
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/fdb/data.py
files: """
  This module stores and retrieves entity data as well as the metadata needed to
  achieve snapshot isolation during transactions. The DataManager is the main
  interface that clients can use to interact with the data layer. See its
  documentation for implementation details.
  """
  from __future__ import division
  import logging
  import math
  import sys
  
  import six
  import six.moves as sm
  from tornado import gen
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.fdb.codecs import (
    decode_str, encode_versionstamp_index, Int64, Path, Text)
  from appscale.datastore.fdb.utils import (
    ABSENT_VERSION, DS_ROOT, fdb, hash_tuple, ResultIterator, VERSIONSTAMP_SIZE)
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.datastore import entity_pb
  
  logger = logging.getLogger(__name__)
  
  first_gt_or_equal = fdb.KeySelector.first_greater_or_equal
  
  
  class VersionEntry(object):
    """ Encapsulates details for an entity version. """
    INCOMPLETE = 1
    __slots__ = ['project_id', 'namespace', 'path', 'commit_versionstamp',
                 'version', '_encoded_entity', '_decoded_entity']
  
    def __init__(self, project_id, namespace, path, commit_versionstamp=None,
                 version=None, encoded_entity=None):
      self.project_id = project_id
      self.namespace = namespace
      self.path = path
      self.commit_versionstamp = commit_versionstamp
      self.version = ABSENT_VERSION if version is None else version
      self._encoded_entity = encoded_entity
      self._decoded_entity = None
  
    def __repr__(self):
      # Since the encoded entity can be large, it probably does not make sense to
      # include it in the string representation.
      blob_repr = self._encoded_entity
      if blob_repr is not None:
        blob_repr = u'<bytes object with length={}>'.format(len(blob_repr))
  
      return u'VersionEntry({!r}, {!r}, {!r}, {!r}, {!r}, {})'.format(
        self.project_id, self.namespace, self.path, self.commit_versionstamp,
        self.version, blob_repr)
  
    @property
    def present(self):
      return self.commit_versionstamp is not None
  
    @property
    def complete(self):
      return self._encoded_entity != self.INCOMPLETE
  
    @property
    def has_entity(self):
      return bool(self._encoded_entity)
  
    @property
    def key(self):
      key = entity_pb.Reference()
      key.set_app(self.project_id)
      if self.namespace is not None:
        key.set_name_space(self.namespace)
  
      key.mutable_path().MergeFrom(Path.decode(self.path))
      return key
  
    @property
    def encoded(self):
      if not self.complete:
        raise ValueError(u'Version entry is not complete')
  
      return self._encoded_entity
  
    @property
    def decoded(self):
      if not self.has_entity:
        return None
  
      if self._decoded_entity is None:
        self._decoded_entity = entity_pb.EntityProto(self.encoded)
  
      return self._decoded_entity
  
    @classmethod
    def from_key(cls, key):
      project_id = decode_str(key.app())
      namespace = None
      if key.has_name_space():
        namespace = decode_str(key.name_space())
  
      path = Path.flatten(key.path())
      return cls(project_id, namespace, path)
  
  
  class DataNamespace(object):
    """
    A DataNamespace handles the encoding and decoding details for entity data
    for a specific project_id/namespace combination.
  
    The directory path looks like (<project-dir>, 'data', <namespace>).
  
    Within this directory, keys are encoded as
    <scatter-byte> + <path> + <commit-versionstamp> + <index>.
  
    The <scatter-byte> is a single byte determined by hashing the entity path.
    Its purpose is to spread writes more evenly across the cluster and minimize
    hotspots.
  
    The <path> contains the entity path. See codecs.Path for encoding details.
  
    The <commit-versionstamp> is a 10-byte versionstamp that specifies the commit version
    of the transaction that wrote the entity data.
  
    The <index> is a single byte specifying the position of the value's chunk.
  
    Values are encoded as <entity-encoding> + <entity> + <entity-version>.
  
    The <entity-encoding> is a single byte specifying the encoding scheme of the
    entity to follow.
  
    The <entity> is an encoded protobuffer value.
  
    The <entity-version> is an integer specifying the approximate insert
    timestamp in microseconds (according to the client performing the insert).
    Though there is a one-to-one mapping of commit versionstamps to entity
    versions, the datastore uses a different value for the entity version in
    order to satisfy the 8-byte constraint and to follow the GAE convention of
    the value representing a timestamp. It is encoded using 7 bytes.
  
    Since encoded values can exceed the size limit imposed by FoundationDB,
    values encoded values are split into chunks. Each chunk is stored as a
    Key-Value and ordered by a unique <index> byte.
    """
    DIR_NAME = u'data'
  
    # The max number of bytes for each FDB value.
    _CHUNK_SIZE = 10000
  
    # The number of bytes used to store an entity version.
    _VERSION_SIZE = 7
  
    # The number of bytes used to encode the chunk index.
    _INDEX_SIZE = 1
  
    # Indicates the encoded blob is a V3 entity object.
    _V3_MARKER = 0x01
  
    def __init__(self, directory):
      self.directory = directory
  
    @property
    def project_id(self):
      return self.directory.get_path()[len(DS_ROOT)]
  
    @property
    def namespace(self):
      return self.directory.get_path()[len(DS_ROOT) + 2]
  
    @property
    def path_slice(self):
      """ The portion of keys that contain the encoded path. """
      return slice(len(self.directory.rawPrefix) + 1,
                   -1 * (VERSIONSTAMP_SIZE + self._INDEX_SIZE))
  
    @property
    def versionstamp_slice(self):
      """ The portion of keys that contain the commit versionstamp. """
      return slice(self.path_slice.stop,
                   self.path_slice.stop + VERSIONSTAMP_SIZE)
  
    @property
    def index_slice(self):
      """ The portion of keys that contain the chunk index. """
      return slice(-1 * self._INDEX_SIZE, None)
  
    @property
    def encoding_slice(self):
      """ The portion of values that specify the entity encoding type. """
      return slice(0, 1)
  
    @property
    def entity_slice(self):
      """ The portion of values that contain the encoded entity. """
      return slice(1, -self._VERSION_SIZE)
  
    @property
    def version_slice(self):
      """ The portion of values that contain the entity version. """
      return slice(-self._VERSION_SIZE, None)
  
    @classmethod
    def directory_path(cls, project_id, namespace):
      return project_id, cls.DIR_NAME, namespace
  
    def encode(self, path, entity, version):
      """ Encodes a tuple of Key-Value tuples for a given version entry.
  
      Args:
        path: A tuple or protobuf path object.
        entity: An encoded entity or protobuf object.
        version: An integer specifying the new entity version.
      Returns:
        A tuple of Key-Value tuples suitable for writing in an FDB transaction.
      """
      if isinstance(entity, entity_pb.EntityProto):
        entity = entity.Encode()
  
      encoded_version = Int64.encode_bare(version, self._VERSION_SIZE)
      if not entity:
        return ((self.encode_key(path, commit_versionstamp=None, index=0),
                 encoded_version),)
  
      value = b''.join([six.int2byte(self._V3_MARKER), entity])
      chunk_count = int(math.ceil(len(value) / self._CHUNK_SIZE))
      chunks = [
        value[slice(index * self._CHUNK_SIZE, (index + 1) * self._CHUNK_SIZE)]
        for index in sm.range(chunk_count)]
      # Place the version at the end of the last chunk. Though this allows the
      # last chunk to exceed the chunk size, it ensures that the entity version
      # can always be retrieved from the last chunk.
      chunks[-1] += encoded_version
      return tuple(
        (self.encode_key(path, commit_versionstamp=None, index=index), chunk)
        for index, chunk in enumerate(chunks))
  
    def encode_key(self, path, commit_versionstamp, index):
      """ Encodes a key for the given version entry.
  
      Args:
        path: A tuple or protobuf path object.
        commit_versionstamp: A 10-byte string specifying the version's commit
          versionstamp or None.
        index: An integer specifying the chunk index.
      Returns:
        A string containing an FDB key. If commit_versionstamp was None, the key
        should be used with set_versionstamped_key.
      """
      encoded_key = b''.join([self._encode_path_prefix(path),
                              commit_versionstamp or b'\x00' * VERSIONSTAMP_SIZE,
                              six.int2byte(index)])
      if not commit_versionstamp:
        versionstamp_index = (len(encoded_key) -
                              (VERSIONSTAMP_SIZE + self._INDEX_SIZE))
        encoded_key += encode_versionstamp_index(versionstamp_index)
  
      return encoded_key
  
    def decode(self, kvs):
      """ Decodes Key-Values to a version entry.
  
      Args:
        kvs: An iterable containing KeyValue objects.
      Returns:
        A VersionEntry object.
      """
      path = Path.unpack(kvs[0].key, self.path_slice.start)[0]
      commit_versionstamp = kvs[0].key[self.versionstamp_slice]
      first_index = ord(kvs[0].key[self.index_slice])
  
      version = Int64.decode_bare(kvs[-1].value[self.version_slice])
      if first_index == 0:
        encoded_entity = b''.join([kv.value for kv in kvs])[self.entity_slice]
      else:
        encoded_entity = VersionEntry.INCOMPLETE
  
      return VersionEntry(self.project_id, self.namespace, path,
                          commit_versionstamp, version, encoded_entity)
  
    def get_slice(self, path, commit_versionstamp=None, read_versionstamp=None):
      """ Gets the range of keys relevant to the given constraints.
  
      Args:
        path: A tuple or protobuf path object.
        commit_versionstamp: The commit versionstamp for a specific entity
          version.
        read_versionstamp: The transaction's read versionstamp. All newer entity
          versions are ignored.
      Returns:
        A slice specifying the start and stop keys.
      """
      path_prefix = self._encode_path_prefix(path)
      if commit_versionstamp is not None:
        # All chunks for a given version.
        prefix = path_prefix + commit_versionstamp
        return slice(first_gt_or_equal(prefix + b'\x00'),
                     first_gt_or_equal(prefix + b'\xFF'))
  
      if read_versionstamp is not None:
        # All versions for a given path except those written after the
        # read_versionstamp.
        version_prefix = path_prefix + read_versionstamp
        return slice(first_gt_or_equal(path_prefix + b'\x00'),
                     first_gt_or_equal(version_prefix + b'\xFF'))
  
      # All versions for a given path.
      return slice(first_gt_or_equal(path_prefix + b'\x00'),
                   first_gt_or_equal(path_prefix + b'\xFF'))
  
    def _encode_path_prefix(self, path):
      """ Encodes the portion of the key up to and including the path.
  
      Args:
        path: A tuple or protobuf path object.
      Returns:
        A string containing the path prefix.
      """
      if not isinstance(path, tuple):
        path = Path.flatten(path)
  
      return b''.join([self.directory.rawPrefix, hash_tuple(path),
                       Path.pack(path)])
  
  
  class GroupUpdatesNS(object):
    """
    A GroupUpdatesNS handles the encoding and decoding details for commit
    versionstamps for each entity group. These are used to materialize conflicts
    for transactions that involve ancestory queries on the same entity groups.
  
    The directory path looks like (<project-dir>, 'group-updates', <namespace>).
  
    Within this directory, keys are encoded as <scatter-byte> + <group-path>.
  
    The <scatter-byte> is a single byte determined by hashing the group path.
    Its purpose is to spread writes more evenly across the cluster and minimize
    hotspots.
  
    The <group-path> contains the entity group path.
  
    Values are 10-byte strings that specify the latest commit version for the
    entity group.
    """
    DIR_NAME = u'group-updates'
  
    def __init__(self, directory):
      self.directory = directory
  
    @classmethod
    def directory_path(cls, project_id, namespace):
      return project_id, cls.DIR_NAME, namespace
  
    def encode(self, path):
      """ Creates a Key-Value tuple for updating a group's commit versionstamp.
  
      Args:
        path: A tuple or protobuf path object.
      Returns:
        A (key, value) tuple suitable for set_versionstamped_value.
      """
      if not isinstance(path, tuple):
        path = Path.flatten(path)
  
      group_path = path[:2]
      return (self.encode_key(group_path),
              b'\x00' * VERSIONSTAMP_SIZE + encode_versionstamp_index(0))
  
    def encode_key(self, group_path):
      """ Encodes a key for a given entity group.
  
      Args:
        group_path: A tuple containing path elements.
      Returns:
        A byte string containing the relevant FDB key.
      """
      return b''.join([
        self.directory.rawPrefix, hash_tuple(group_path),
        Text.encode(group_path[0]) + Path.encode_id_or_name(group_path[1])])
  
  
  class DataManager(object):
    """
    The DataManager is the main interface that clients can use to interact with
    the data layer. It makes use of the DataNamespace and GroupUpdateNS
    namespaces to handle the encoding and decoding details when satisfying
    requests. When a client requests data, the DataManager encapsulates entity
    data in a VersionEntry object.
  
    See the DataNamespace and GroupUpdateNS classes for implementation details
    about how data is stored and retrieved.
    """
    def __init__(self, tornado_fdb, directory_cache):
      self._tornado_fdb = tornado_fdb
      self._directory_cache = directory_cache
  
    @gen.coroutine
    def get_latest(self, tr, key, read_versionstamp=None, include_data=True,
                   snapshot=False):
      """ Gets the newest entity version for the given read versionstamp.
  
      Args:
        tr: An FDB transaction.
        key: A protubuf reference object.
        read_versionstamp: A 10-byte string specifying the FDB read versionstamp.
          Newer versionstamps are ignored.
        include_data: A boolean specifying whether or not to fetch all of the
          entity's Key-Values.
        snapshot: If True, the read will not cause a transaction conflict.
      """
      data_ns = yield self._data_ns_from_key(tr, key)
      desired_slice = data_ns.get_slice(
        key.path(), read_versionstamp=read_versionstamp)
      last_entry = yield self._last_version(
        tr, data_ns, desired_slice, include_data, snapshot=snapshot)
      if last_entry is None:
        last_entry = VersionEntry(data_ns.project_id, data_ns.namespace,
                                  Path.flatten(key.path()))
  
      raise gen.Return(last_entry)
  
    @gen.coroutine
    def get_entry(self, tr, index_entry, snapshot=False):
      """ Gets the entity data from an index entry.
  
      Args:
        tr: An FDB transaction.
        index_entry: An IndexEntry object.
        snapshot: If True, the read will not cause a transaction conflict.
      Returns:
        A VersionEntry or None.
      """
      if index_entry.kind in (u'__namespace__', u'__kind__'):
        entity = entity_pb.EntityProto()
        entity.mutable_key().MergeFrom(index_entry.key)
        entity.mutable_entity_group()
        version_entry = VersionEntry(
          index_entry.project_id, index_entry.namespace, index_entry.path,
          encoded_entity=entity.Encode())
        raise gen.Return(version_entry)
  
      version_entry = yield self.get_version_from_path(
        tr, index_entry.project_id, index_entry.namespace, index_entry.path,
        index_entry.commit_versionstamp, snapshot)
      raise gen.Return(version_entry)
  
    @gen.coroutine
    def get_version_from_path(self, tr, project_id, namespace, path,
                              commit_versionstamp, snapshot=False):
      """ Gets the entity data for a specific version.
  
      Args:
        tr: An FDB transaction.
        project_id: A string specifying the project ID.
        namespace: A string specifying the namespace.
        path: A tuple or protobuf path object.
        commit_versionstamp: A 10-byte string specyfing the FDB commit
          versionstamp.
        snapshot: If True, the read will not cause a transaction conflict.
      Returns:
        A VersionEntry or None.
      """
      data_ns = yield self._data_ns(tr, project_id, namespace)
      desired_slice = data_ns.get_slice(path, commit_versionstamp)
      results = yield ResultIterator(tr, self._tornado_fdb, desired_slice,
                                     snapshot=snapshot).list()
      raise gen.Return(data_ns.decode(results))
  
    @gen.coroutine
    def last_group_versionstamp(self, tr, project_id, namespace, group_path):
      """ Gets the most recent commit versionstamp for the entity group.
  
      Args:
        tr: An FDB transaction.
        project_id: A string specifying the project ID.
        namespace: A string specifying the namespace.
        group_path: A tuple containing the group's path elements.
      Returns:
        A 10-byte string specifying the versionstamp or None.
      """
      group_ns = yield self._group_updates_ns(tr, project_id, namespace)
      last_updated_versionstamp = yield self._tornado_fdb.get(
        tr, group_ns.encode_key(group_path))
      if not last_updated_versionstamp.present():
        return
  
      raise gen.Return(last_updated_versionstamp.value)
  
    @gen.coroutine
    def put(self, tr, key, version, encoded_entity):
      """ Writes a new version entry and updates the entity group versionstamp.
  
      Args:
        tr: An FDB transaction.
        key: A protobuf reference object.
        version: An integer specifying the new entity version.
        encoded_entity: A string specifying the encoded entity data.
      """
      data_ns = yield self._data_ns_from_key(tr, key)
      for fdb_key, val in data_ns.encode(key.path(), encoded_entity, version):
        tr.set_versionstamped_key(fdb_key, val)
  
      group_ns = yield self._group_updates_ns(
        tr, data_ns.project_id, data_ns.namespace)
      tr.set_versionstamped_value(*group_ns.encode(key.path()))
  
    @gen.coroutine
    def hard_delete(self, tr, version_entry):
      """ Deletes a version entry. Only the GC should use this.
  
      Args:
        tr: An FDB transaction.
        version_entry: A VersionEntry object.
      """
      data_ns = yield self._data_ns(
        tr, version_entry.project_id, version_entry.namespace)
      version_slice = data_ns.get_slice(version_entry.path,
                                        version_entry.commit_versionstamp)
      del tr[version_slice.start.key:version_slice.stop.key]
  
    @gen.coroutine
    def _data_ns(self, tr, project_id, namespace):
      directory = yield self._directory_cache.get(
        tr, DataNamespace.directory_path(project_id, namespace))
      raise gen.Return(DataNamespace(directory))
  
    @gen.coroutine
    def _data_ns_from_key(self, tr, key):
      project_id = decode_str(key.app())
      namespace = decode_str(key.name_space())
      data_ns = yield self._data_ns(tr, project_id, namespace)
      raise gen.Return(data_ns)
  
    @gen.coroutine
    def _group_updates_ns(self, tr, project_id, namespace):
      directory = yield self._directory_cache.get(
        tr, GroupUpdatesNS.directory_path(project_id, namespace))
      raise gen.Return(GroupUpdatesNS(directory))
  
    @gen.coroutine
    def _last_version(self, tr, data_ns, desired_slice, include_data=True,
                      snapshot=False):
      """ Gets the most recent entity data for a given slice.
  
      Args:
        tr: An FDB transaction.
        data_ns: A DataNamespace.
        desired_slice: A slice specifying the start and stop keys.
        include_data: A boolean indicating that all chunks should be fetched.
        snapshot: If True, the read will not cause a transaction conflict.
      Returns:
        A VersionEntry or None.
      """
      results, count, more_results = yield self._tornado_fdb.get_range(
        tr, desired_slice, limit=1, reverse=True, snapshot=snapshot)
  
      if not results:
        return
  
      last_chunk = results[0]
      entry = data_ns.decode([last_chunk])
      if not include_data or not entry.present or entry.complete:
        raise gen.Return(entry)
  
      # Retrieve the remaining chunks.
      version_slice = data_ns.get_slice(entry.path, entry.commit_versionstamp)
      remaining = slice(version_slice.start, first_gt_or_equal(last_chunk.key))
      results = yield ResultIterator(tr, self._tornado_fdb, remaining).list()
      raise gen.Return(data_ns.decode(results + [last_chunk]))
  
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/fdb/data.py
files: """
  This module stores and queries entity indexes. The IndexManager is the main
  interface that clients can use to interact with the index layer. See its
  documentation for implementation details.
  """
  from __future__ import division
  
  import itertools
  import logging
  import monotonic
  import sys
  
  import six
  from tornado import gen
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.fdb import codecs
  from appscale.datastore.fdb.codecs import (
    decode_str, decode_value, encode_value, encode_versionstamp_index, Path)
  from appscale.datastore.fdb.sdk import FindIndexToUse, ListCursor
  from appscale.datastore.fdb.utils import (
    format_prop_val, DS_ROOT, fdb, get_scatter_val, MAX_FDB_TX_DURATION,
    ResultIterator, SCATTER_PROP, VERSIONSTAMP_SIZE)
  from appscale.datastore.dbconstants import BadRequest, InternalError
  from appscale.datastore.index_manager import IndexInaccessible
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.datastore import datastore_pb, entity_pb
  from google.appengine.datastore.datastore_pb import Query_Filter, Query_Order
  
  logger = logging.getLogger(__name__)
  
  KEY_PROP = u'__key__'
  
  first_gt_or_equal = fdb.KeySelector.first_greater_or_equal
  
  
  class FilterProperty(object):
    """ Encapsulates details for a FilterProperty that came from a query. """
    __slots__ = ['name', 'filters']
  
    def __init__(self, prop_name, filters):
      self.name = prop_name
      self.filters = filters
  
    @property
    def equality(self):
      return all(op == Query_Filter.EQUAL for op, _ in self.filters)
  
    def __repr__(self):
      return u'FilterProperty(%r, %r)' % (self.name, self.filters)
  
  
  def group_filters(query):
    filter_props = []
    for query_filter in query.filter_list():
      if query_filter.property_size() != 1:
        raise BadRequest(u'Each filter must have exactly one property')
  
      prop = query_filter.property(0)
      prop_name = decode_str(prop.name())
      filter_info = (query_filter.op(), prop.value())
      if filter_props and filter_props[-1].name == prop_name:
        filter_props[-1].filters.append(filter_info)
      else:
        filter_props.append(FilterProperty(prop_name, [filter_info]))
  
    # Since the filter list can come in any order, put inequality filters last.
    inequality_index = None
    for index, filter_prop in enumerate(filter_props):
      if not filter_prop.equality:
        inequality_index = index
        break
  
    if inequality_index is not None:
      inequality_prop = filter_props.pop(inequality_index)
      filter_props.append(inequality_prop)
  
    # Put key filters last.
    key_index = None
    for index, filter_prop in enumerate(filter_props):
      if filter_prop.name == KEY_PROP:
        key_index = index
        break
  
    if key_index is not None:
      key_prop = filter_props.pop(key_index)
      filter_props.append(key_prop)
  
    for filter_prop in filter_props[:-1]:
      if filter_prop.name == KEY_PROP:
        raise BadRequest(
          u'Only the last filter property can be on {}'.format(KEY_PROP))
  
      if not filter_prop.equality:
        raise BadRequest(u'All but the last property must be equality filters')
  
    return tuple(filter_props)
  
  
  def get_order_info(query):
    filter_props = group_filters(query)
  
    # Orders on equality filters can be ignored.
    equality_props = [prop.name for prop in filter_props if prop.equality]
    relevant_orders = [order for order in query.order_list()
                       if order.property() not in equality_props]
  
    order_info = []
    for filter_prop in filter_props:
      if filter_prop.equality:
        continue
  
      direction = next(
        (order.direction() for order in relevant_orders
         if order.property() == filter_prop.name), Query_Order.ASCENDING)
      order_info.append((filter_prop.name, direction))
  
    filter_prop_names = [prop.name for prop in filter_props]
    order_info.extend(
      [(decode_str(order.property()), order.direction())
       for order in relevant_orders
       if order.property() not in filter_prop_names])
  
    return tuple(order_info)
  
  
  def get_scan_direction(query, index):
    order_info = get_order_info(query)
    if not order_info:
      return Query_Order.ASCENDING
  
    first_property, first_direction = order_info[0]
    if first_property == KEY_PROP or isinstance(index, SinglePropIndex):
      return first_direction
  
    index_direction = next(direction for prop_name, direction in index.order_info
                           if prop_name == first_property)
    if index_direction == first_direction:
      return Query_Order.ASCENDING
    else:
      return Query_Order.DESCENDING
  
  
  class IndexEntry(object):
    """ Encapsulates details for an index entry. """
    __slots__ = ['project_id', 'namespace', 'path', 'commit_versionstamp',
                 'deleted_versionstamp']
  
    def __init__(self, project_id, namespace, path, commit_versionstamp,
                 deleted_versionstamp):
      self.project_id = project_id
      self.namespace = namespace
      self.path = path
      self.commit_versionstamp = commit_versionstamp
      self.deleted_versionstamp = deleted_versionstamp
  
    @property
    def kind(self):
      return self.path[-2]
  
    @property
    def key(self):
      key = entity_pb.Reference()
      key.set_app(self.project_id)
      key.set_name_space(self.namespace)
      key.mutable_path().MergeFrom(Path.decode(self.path))
      return key
  
    @property
    def group(self):
      group = entity_pb.Path()
      group.add_element().MergeFrom(Path.decode_element(self.path[:2]))
      return group
  
    def __repr__(self):
      return u'IndexEntry(%r, %r, %r, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.commit_versionstamp,
        self.deleted_versionstamp)
  
    def key_result(self):
      entity = entity_pb.EntityProto()
      entity.mutable_key().MergeFrom(self.key)
      entity.mutable_entity_group()
      return entity
  
    def cursor_result(self, ordered_props):
      compiled_cursor = datastore_pb.CompiledCursor()
      position = compiled_cursor.add_position()
      position.mutable_key().MergeFrom(self.key)
      position.set_start_inclusive(False)
      return compiled_cursor
  
  
  class PropertyEntry(IndexEntry):
    """ Encapsulates details for a single-property index entry. """
    __slots__ = ['prop_name', 'value']
  
    def __init__(self, project_id, namespace, path, prop_name, value,
                 commit_versionstamp, deleted_versionstamp):
      super(PropertyEntry, self).__init__(
        project_id, namespace, path, commit_versionstamp, deleted_versionstamp)
      self.prop_name = prop_name
      self.value = value
  
    def __repr__(self):
      return u'PropertyEntry(%r, %r, %r, %r, %r, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.prop_name, self.value,
        self.commit_versionstamp, self.deleted_versionstamp)
  
    def __str__(self):
      return u'PropertyEntry(%s, %r, %s, %s, %s, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.prop_name,
        format_prop_val(self.value), self.commit_versionstamp,
        self.deleted_versionstamp)
  
    def prop_result(self):
      entity = entity_pb.EntityProto()
      entity.mutable_key().MergeFrom(self.key)
      entity.mutable_entity_group().MergeFrom(self.group)
      prop = entity.add_property()
      prop.set_name(self.prop_name)
      prop.set_meaning(entity_pb.Property.INDEX_VALUE)
      prop.set_multiple(False)
      prop.mutable_value().MergeFrom(self.value)
      return entity
  
    def cursor_result(self, ordered_props):
      compiled_cursor = datastore_pb.CompiledCursor()
      position = compiled_cursor.add_position()
      position.mutable_key().MergeFrom(self.key)
      position.set_start_inclusive(False)
      if self.prop_name in ordered_props:
        index_value = position.add_indexvalue()
        index_value.set_property(self.prop_name)
        index_value.mutable_value().MergeFrom(self.value)
  
      return compiled_cursor
  
  
  class CompositeEntry(IndexEntry):
    """ Encapsulates details for a composite index entry. """
    __slots__ = ['properties']
  
    def __init__(self, project_id, namespace, path, properties,
                 commit_versionstamp, deleted_versionstamp):
      super(CompositeEntry, self).__init__(
        project_id, namespace, path, commit_versionstamp, deleted_versionstamp)
      self.properties = properties
  
    def __repr__(self):
      return u'CompositeEntry(%r, %r, %r, %r, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.properties,
        self.commit_versionstamp, self.deleted_versionstamp)
  
    def prop_result(self):
      entity = entity_pb.EntityProto()
      entity.mutable_key().MergeFrom(self.key)
      entity.mutable_entity_group().MergeFrom(self.group)
      prop_names = list(zip(*self.properties))[0]
      for prop_name, value in self.properties:
        prop = entity.add_property()
        prop.set_name(prop_name)
        prop.set_meaning(entity_pb.Property.INDEX_VALUE)
        prop.set_multiple(prop_names.count(prop_name) > 1)
        prop.mutable_value().MergeFrom(value)
  
      return entity
  
    def cursor_result(self, ordered_props):
      compiled_cursor = datastore_pb.CompiledCursor()
      position = compiled_cursor.add_position()
      position.mutable_key().MergeFrom(self.key)
      position.set_start_inclusive(False)
      for prop_name, value in self.properties:
        if prop_name not in ordered_props:
          continue
  
        index_value = position.add_indexvalue()
        index_value.set_property(prop_name)
        index_value.mutable_value().MergeFrom(value)
  
      return compiled_cursor
  
  
  class IndexIterator(object):
    """
    Returns pages of index entry results. It ignores Key-Values that do not apply
    to the given read_versionstamp. It converts usable Key-Values to IndexEntry
    objects.
    """
    def __init__(self, tr, tornado_fdb, index, key_slice, fetch_limit, reverse,
                 read_versionstamp=None, snapshot=False):
      self.index = index
      self._result_iterator = ResultIterator(
        tr, tornado_fdb, key_slice, fetch_limit, reverse, snapshot=snapshot)
      self._read_versionstamp = read_versionstamp
      self._done = False
  
    @property
    def prop_names(self):
      return self.index.prop_names
  
    @property
    def start_key(self):
      return self._result_iterator.slice.start.key
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      results, more_results = yield self._result_iterator.next_page()
      usable_entries = []
      for result in results:
        entry = self.index.decode(result)
        if not self._usable(entry):
          self._result_iterator.increase_limit()
          more_results = not self._result_iterator.done_with_range
          continue
  
        usable_entries.append(entry)
  
      if not more_results:
        self._done = True
  
      raise gen.Return((usable_entries, more_results))
  
    def _usable(self, entry):
      if self._read_versionstamp and entry.deleted_versionstamp:
        return (entry.commit_versionstamp < self._read_versionstamp <
                entry.deleted_versionstamp)
      elif self._read_versionstamp:
        return entry.commit_versionstamp < self._read_versionstamp
      else:
        return entry.deleted_versionstamp is None
  
  
  class NamespaceIterator(object):
    def __init__(self, tr, project_dir):
      self._tr = tr
      self._project_dir = project_dir
      self._done = False
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      # TODO: This can be made async.
      ns_dir = self._project_dir.open(self._tr, (KindIndex.DIR_NAME,))
      namespaces = ns_dir.list(self._tr)
  
      # The API uses an ID of 1 to label the default namespace.
      results = [IndexEntry(self._project_dir.get_path()[-1], u'',
                            (u'__namespace__', namespace or 1), None, None)
                 for namespace in namespaces]
  
      self._done = True
      raise gen.Return((results, False))
  
  
  class KindIterator(object):
    def __init__(self, tr, tornado_fdb, project_dir, namespace):
      self._tr = tr
      self._tornado_fdb = tornado_fdb
      self._project_dir = project_dir
      self._namespace = namespace
      self._done = False
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      # TODO: This can be made async.
      ns_dir = self._project_dir.open(
        self._tr, (KindIndex.DIR_NAME, self._namespace))
      kinds = ns_dir.list(self._tr)
      populated_kinds = [
        kind for kind, populated in zip(
          kinds, (yield [self._populated(ns_dir, kind) for kind in kinds]))
        if populated]
  
      results = [IndexEntry(self._project_dir.get_path()[-1], self._namespace,
                            (u'__kind__', kind), None, None)
                 for kind in populated_kinds]
  
      self._done = True
      raise gen.Return((results, False))
  
    @gen.coroutine
    def _populated(self, ns_dir, kind):
      """ Checks if at least one entity exists for a given kind. """
      kind_dir = ns_dir.open(self._tr, (kind,))
      kind_index = KindIndex(kind_dir)
      # TODO: Check if the presence of stat entities should mark a kind as being
      #  populated.
      index_slice = kind_index.get_slice(())
      # This query is reversed to increase the likelihood of getting a relevant
      # (not marked for GC) entry.
      iterator = IndexIterator(self._tr, self._tornado_fdb, kind_index,
                               index_slice, fetch_limit=1, reverse=True,
                               snapshot=True)
      while True:
        results, more_results = yield iterator.next_page()
        if results:
          raise gen.Return(True)
  
        if not more_results:
          raise gen.Return(False)
  
  
  class PropertyIterator(object):
    """ Iterates over a list of indexed property names for a kind. """
    PROPERTY_TYPES = (u'NULL', u'INT64', u'BOOLEAN', u'STRING', u'DOUBLE',
                      u'POINT', u'USER', u'REFERENCE')
  
    def __init__(self, tr, tornado_fdb, project_dir, namespace):
      self._tr = tr
      self._tornado_fdb = tornado_fdb
      self._project_dir = project_dir
      self._namespace = namespace
      self._done = False
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      # TODO: This can be made async.
      ns_dir = self._project_dir.open(
        self._tr, (SinglePropIndex.DIR_NAME, self._namespace))
      kinds = ns_dir.list(self._tr)
      # TODO: Check if stat entities belong in kinds.
      kind_dirs = [ns_dir.open(self._tr, (kind,)) for kind in kinds]
      results = []
      for kind, kind_dir in zip(kinds, kind_dirs):
        # TODO: This can be made async.
        prop_names = kind_dir.list(self._tr)
        for prop_name in prop_names:
          prop_dir = kind_dir.open(self._tr, (prop_name,))
          index = SinglePropIndex(prop_dir)
          populated_map = yield [self._populated(index, type_name)
                                 for type_name in self.PROPERTY_TYPES]
          populated_types = tuple(
            type_ for type_, populated in zip(self.PROPERTY_TYPES, populated_map)
            if populated)
          if not populated_types:
            continue
  
          project_id = self._project_dir.get_path()[-1]
          path = (u'__kind__', kind, u'__property__', prop_name)
          properties = []
          for prop_type in populated_types:
            prop_value = entity_pb.PropertyValue()
            prop_value.set_stringvalue(prop_type)
            properties.append((u'property_representation', prop_value))
  
          results.append(CompositeEntry(project_id, self._namespace, path,
                                        properties, None, None))
  
      self._done = True
      raise gen.Return((results, False))
  
    @gen.coroutine
    def _populated(self, prop_index, type_name):
      """ Checks if at least one entity exists for a given type name. """
      index_slice = prop_index.type_range(type_name)
      # This query is reversed to increase the likelihood of getting a relevant
      # (not marked for GC) entry.
      iterator = IndexIterator(self._tr, self._tornado_fdb, prop_index,
                               index_slice, fetch_limit=1, reverse=True,
                               snapshot=True)
      while True:
        results, more_results = yield iterator.next_page()
        if results:
          raise gen.Return(True)
  
        if not more_results:
          raise gen.Return(False)
  
  
  class MergeJoinIterator(object):
    """
    Returns pages of index entry results from multiple ranges. It ignores
    Key-Values that do not apply to the given read_versionstamp. It converts
    usable Key-Values to IndexEntry objects.
    """
    def __init__(self, tr, tornado_fdb, filter_props, indexes, fetch_limit,
                 read_versionstamp=None, ancestor_path=None, snapshot=False):
      self.indexes = indexes
      self._filter_props = filter_props
      self._read_versionstamp = read_versionstamp
      self._tr = tr
      self._tornado_fdb = tornado_fdb
      self._fetch_limit = fetch_limit
      self._fetched = 0
      self._snapshot = snapshot
      self._done = False
      self._candidate_path = None
      self._candidate_entries = []
      self._ancestor_path = ancestor_path
  
    @property
    def prop_names(self):
      prop_names = set()
      for index, _, _, _ in self.indexes:
        prop_names.update(index.prop_names)
  
      return tuple(prop_names)
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      composite_entry = None
      for i, (index, key_slice, prop_name, value) in enumerate(self.indexes):
        usable_entry = None
        # TODO: Keep cache of ranges to reduce unnecessary lookups.
        index_exhausted = False
        while True:
          results, count, more = yield self._tornado_fdb.get_range(
            self._tr, key_slice, 0, fdb.StreamingMode.small, 1,
            snapshot=self._snapshot)
          if not count:
            index_exhausted = True
            break
  
          key_slice = slice(fdb.KeySelector.first_greater_than(results[-1].key),
                            key_slice.stop)
          for result in results:
            entry = index.decode(result)
            if self._usable(entry):
              usable_entry = entry
              break
  
          if usable_entry is not None:
            break
  
        if index_exhausted:
          self._done = True
          break
  
        if usable_entry.path == self._candidate_path:
          self._candidate_entries.append(usable_entry)
        else:
          self._candidate_path = usable_entry.path
          self._candidate_entries = [usable_entry]
  
        next_index_op = Query_Filter.GREATER_THAN_OR_EQUAL
        if len(self._candidate_entries) == len(self.indexes):
          properties = []
          for partial_entry in self._candidate_entries:
            if isinstance(partial_entry, PropertyEntry):
              properties.append((partial_entry.prop_name, partial_entry.value))
            else:
              for property in partial_entry.properties:
                if property not in properties:
                  properties.append(property)
  
          composite_entry = CompositeEntry(
            usable_entry.project_id, usable_entry.namespace,
            self._candidate_path, properties, usable_entry.commit_versionstamp,
            usable_entry.deleted_versionstamp)
          self._candidate_entries = []
          next_index_op = Query_Filter.GREATER_THAN
  
        next_index_i = (i + 1) % len(self.indexes)
        next_index, next_slice, next_prop_name, next_value =\
          self.indexes[next_index_i]
  
        # TODO: This probably doesn't work for all cases.
        last_prop = None
        if isinstance(next_index, CompositeIndex):
          last_prop = next_index.prop_names[-1]
  
        tmp_filter_props = []
        for filter_prop in self._filter_props:
          if (filter_prop.name in (next_prop_name, last_prop, KEY_PROP) or
              filter_prop.name not in next_index.prop_names):
            continue
          else:
            tmp_filter_props.append(filter_prop)
  
        tmp_filter_props.append(
          FilterProperty(next_prop_name, [(Query_Filter.EQUAL, next_value)]))
  
        if last_prop is not None:
          val = next(value for prop_name, value in usable_entry.properties
                     if prop_name == last_prop)
          tmp_filter_props.append(
            FilterProperty(last_prop, [(Query_Filter.EQUAL, val)]))
  
        tmp_filter_props.append(
          FilterProperty(KEY_PROP, [(next_index_op, usable_entry.path)]))
  
        new_slice = next_index.get_slice(tmp_filter_props,
                                         ancestor_path=self._ancestor_path)
        self.indexes[next_index_i][1] = new_slice
  
      entries = [composite_entry] if composite_entry is not None else []
      self._fetched += len(entries)
      if self._fetched == self._fetch_limit:
        self._done = True
  
      raise gen.Return((entries, not self._done))
  
    def _usable(self, entry):
      if self._read_versionstamp and entry.deleted_versionstamp:
        return (entry.commit_versionstamp < self._read_versionstamp <
                entry.deleted_versionstamp)
      elif self._read_versionstamp:
        return entry.commit_versionstamp < self._read_versionstamp
      else:
        return entry.deleted_versionstamp is None
  
  
  class IndexSlice(object):
    """ Encapsulates details about an index range in a way that's mutable. """
    __slots__ = ['_directory_prefix', '_order_info', '_ancestor', '_start_parts',
                 '_stop_parts']
  
    def __init__(self, directory_prefix, order_info, ancestor=False):
      self._directory_prefix = directory_prefix
      self._order_info = order_info
      self._ancestor = ancestor
  
      self._start_parts = [self._directory_prefix]
      self._stop_parts = [self._directory_prefix, b'\xFF']
  
    @property
    def start(self):
      return first_gt_or_equal(b''.join(self._
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/fdb/indexes.py
files: """
  This module stores and queries entity indexes. The IndexManager is the main
  interface that clients can use to interact with the index layer. See its
  documentation for implementation details.
  """
  from __future__ import division
  
  import itertools
  import logging
  import monotonic
  import sys
  
  import six
  from tornado import gen
  
  from appscale.common.unpackaged import APPSCALE_PYTHON_APPSERVER
  from appscale.datastore.fdb.codecs import (
    decode_str, decode_value, encode_value, encode_versionstamp_index, Path)
  from appscale.datastore.fdb.sdk import FindIndexToUse, ListCursor
  from appscale.datastore.fdb.utils import (
    format_prop_val, DS_ROOT, fdb, get_scatter_val, MAX_FDB_TX_DURATION,
    ResultIterator, SCATTER_PROP, VERSIONSTAMP_SIZE)
  from appscale.datastore.dbconstants import BadRequest, InternalError
  from appscale.datastore.index_manager import IndexInaccessible
  
  sys.path.append(APPSCALE_PYTHON_APPSERVER)
  from google.appengine.datastore import datastore_pb, entity_pb
  from google.appengine.datastore.datastore_pb import Query_Filter, Query_Order
  
  logger = logging.getLogger(__name__)
  
  KEY_PROP = u'__key__'
  
  first_gt_or_equal = fdb.KeySelector.first_greater_or_equal
  
  
  class FilterProperty(object):
    """ Encapsulates details for a FilterProperty that came from a query. """
    __slots__ = ['name', 'filters']
  
    def __init__(self, prop_name, filters):
      self.name = prop_name
      self.filters = filters
  
    @property
    def equality(self):
      return all(op == Query_Filter.EQUAL for op, _ in self.filters)
  
    def __repr__(self):
      return u'FilterProperty(%r, %r)' % (self.name, self.filters)
  
  
  def group_filters(query):
    filter_props = []
    for query_filter in query.filter_list():
      if query_filter.property_size() != 1:
        raise BadRequest(u'Each filter must have exactly one property')
  
      prop = query_filter.property(0)
      prop_name = decode_str(prop.name())
      filter_info = (query_filter.op(), prop.value())
      if filter_props and filter_props[-1].name == prop_name:
        filter_props[-1].filters.append(filter_info)
      else:
        filter_props.append(FilterProperty(prop_name, [filter_info]))
  
    # Since the filter list can come in any order, put inequality filters last.
    inequality_index = None
    for index, filter_prop in enumerate(filter_props):
      if not filter_prop.equality:
        inequality_index = index
        break
  
    if inequality_index is not None:
      inequality_prop = filter_props.pop(inequality_index)
      filter_props.append(inequality_prop)
  
    # Put key filters last.
    key_index = None
    for index, filter_prop in enumerate(filter_props):
      if filter_prop.name == KEY_PROP:
        key_index = index
        break
  
    if key_index is not None:
      key_prop = filter_props.pop(key_index)
      filter_props.append(key_prop)
  
    for filter_prop in filter_props[:-1]:
      if filter_prop.name == KEY_PROP:
        raise BadRequest(
          u'Only the last filter property can be on {}'.format(KEY_PROP))
  
      if not filter_prop.equality:
        raise BadRequest(u'All but the last property must be equality filters')
  
    return tuple(filter_props)
  
  
  def get_order_info(query):
    filter_props = group_filters(query)
  
    # Orders on equality filters can be ignored.
    equality_props = [prop.name for prop in filter_props if prop.equality]
    relevant_orders = [order for order in query.order_list()
                       if order.property() not in equality_props]
  
    order_info = []
    for filter_prop in filter_props:
      if filter_prop.equality:
        continue
  
      direction = next(
        (order.direction() for order in relevant_orders
         if order.property() == filter_prop.name), Query_Order.ASCENDING)
      order_info.append((filter_prop.name, direction))
  
    filter_prop_names = [prop.name for prop in filter_props]
    order_info.extend(
      [(decode_str(order.property()), order.direction())
       for order in relevant_orders
       if order.property() not in filter_prop_names])
  
    return tuple(order_info)
  
  
  def get_scan_direction(query, index):
    order_info = get_order_info(query)
    if not order_info:
      return Query_Order.ASCENDING
  
    first_property, first_direction = order_info[0]
    if first_property == KEY_PROP or isinstance(index, SinglePropIndex):
      return first_direction
  
    index_direction = next(direction for prop_name, direction in index.order_info
                           if prop_name == first_property)
    if index_direction == first_direction:
      return Query_Order.ASCENDING
    else:
      return Query_Order.DESCENDING
  
  
  class IndexEntry(object):
    """ Encapsulates details for an index entry. """
    __slots__ = ['project_id', 'namespace', 'path', 'commit_versionstamp',
                 'deleted_versionstamp']
  
    def __init__(self, project_id, namespace, path, commit_versionstamp,
                 deleted_versionstamp):
      self.project_id = project_id
      self.namespace = namespace
      self.path = path
      self.commit_versionstamp = commit_versionstamp
      self.deleted_versionstamp = deleted_versionstamp
  
    @property
    def kind(self):
      return self.path[-2]
  
    @property
    def key(self):
      key = entity_pb.Reference()
      key.set_app(self.project_id)
      key.set_name_space(self.namespace)
      key.mutable_path().MergeFrom(Path.decode(self.path))
      return key
  
    @property
    def group(self):
      group = entity_pb.Path()
      group.add_element().MergeFrom(Path.decode_element(self.path[:2]))
      return group
  
    def __repr__(self):
      return u'IndexEntry(%r, %r, %r, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.commit_versionstamp,
        self.deleted_versionstamp)
  
    def key_result(self):
      entity = entity_pb.EntityProto()
      entity.mutable_key().MergeFrom(self.key)
      entity.mutable_entity_group()
      return entity
  
    def cursor_result(self, ordered_props):
      compiled_cursor = datastore_pb.CompiledCursor()
      position = compiled_cursor.add_position()
      position.mutable_key().MergeFrom(self.key)
      position.set_start_inclusive(False)
      return compiled_cursor
  
  
  class PropertyEntry(IndexEntry):
    """ Encapsulates details for a single-property index entry. """
    __slots__ = ['prop_name', 'value']
  
    def __init__(self, project_id, namespace, path, prop_name, value,
                 commit_versionstamp, deleted_versionstamp):
      super(PropertyEntry, self).__init__(
        project_id, namespace, path, commit_versionstamp, deleted_versionstamp)
      self.prop_name = prop_name
      self.value = value
  
    def __repr__(self):
      return u'PropertyEntry(%r, %r, %r, %r, %r, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.prop_name, self.value,
        self.commit_versionstamp, self.deleted_versionstamp)
  
    def __str__(self):
      return u'PropertyEntry(%s, %r, %s, %s, %s, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.prop_name,
        format_prop_val(self.value), self.commit_versionstamp,
        self.deleted_versionstamp)
  
    def prop_result(self):
      entity = entity_pb.EntityProto()
      entity.mutable_key().MergeFrom(self.key)
      entity.mutable_entity_group().MergeFrom(self.group)
      prop = entity.add_property()
      prop.set_name(self.prop_name)
      prop.set_meaning(entity_pb.Property.INDEX_VALUE)
      prop.set_multiple(False)
      prop.mutable_value().MergeFrom(self.value)
      return entity
  
    def cursor_result(self, ordered_props):
      compiled_cursor = datastore_pb.CompiledCursor()
      position = compiled_cursor.add_position()
      position.mutable_key().MergeFrom(self.key)
      position.set_start_inclusive(False)
      if self.prop_name in ordered_props:
        index_value = position.add_indexvalue()
        index_value.set_property(self.prop_name)
        index_value.mutable_value().MergeFrom(self.value)
  
      return compiled_cursor
  
  
  class CompositeEntry(IndexEntry):
    """ Encapsulates details for a composite index entry. """
    __slots__ = ['properties']
  
    def __init__(self, project_id, namespace, path, properties,
                 commit_versionstamp, deleted_versionstamp):
      super(CompositeEntry, self).__init__(
        project_id, namespace, path, commit_versionstamp, deleted_versionstamp)
      self.properties = properties
  
    def __repr__(self):
      return u'CompositeEntry(%r, %r, %r, %r, %r, %r)' % (
        self.project_id, self.namespace, self.path, self.properties,
        self.commit_versionstamp, self.deleted_versionstamp)
  
    def prop_result(self):
      entity = entity_pb.EntityProto()
      entity.mutable_key().MergeFrom(self.key)
      entity.mutable_entity_group().MergeFrom(self.group)
      prop_names = list(zip(*self.properties))[0]
      for prop_name, value in self.properties:
        prop = entity.add_property()
        prop.set_name(prop_name)
        prop.set_meaning(entity_pb.Property.INDEX_VALUE)
        prop.set_multiple(prop_names.count(prop_name) > 1)
        prop.mutable_value().MergeFrom(value)
  
      return entity
  
    def cursor_result(self, ordered_props):
      compiled_cursor = datastore_pb.CompiledCursor()
      position = compiled_cursor.add_position()
      position.mutable_key().MergeFrom(self.key)
      position.set_start_inclusive(False)
      for prop_name, value in self.properties:
        if prop_name not in ordered_props:
          continue
  
        index_value = position.add_indexvalue()
        index_value.set_property(prop_name)
        index_value.mutable_value().MergeFrom(value)
  
      return compiled_cursor
  
  
  class IndexIterator(object):
    """
    Returns pages of index entry results. It ignores Key-Values that do not apply
    to the given read_versionstamp. It converts usable Key-Values to IndexEntry
    objects.
    """
    def __init__(self, tr, tornado_fdb, index, key_slice, fetch_limit, reverse,
                 read_versionstamp=None, snapshot=False):
      self.index = index
      self._result_iterator = ResultIterator(
        tr, tornado_fdb, key_slice, fetch_limit, reverse, snapshot=snapshot)
      self._read_versionstamp = read_versionstamp
      self._done = False
  
    @property
    def prop_names(self):
      return self.index.prop_names
  
    @property
    def start_key(self):
      return self._result_iterator.slice.start.key
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      results, more_results = yield self._result_iterator.next_page()
      usable_entries = []
      for result in results:
        entry = self.index.decode(result)
        if not self._usable(entry):
          self._result_iterator.increase_limit()
          more_results = not self._result_iterator.done_with_range
          continue
  
        usable_entries.append(entry)
  
      if not more_results:
        self._done = True
  
      raise gen.Return((usable_entries, more_results))
  
    def _usable(self, entry):
      if self._read_versionstamp and entry.deleted_versionstamp:
        return (entry.commit_versionstamp < self._read_versionstamp <
                entry.deleted_versionstamp)
      elif self._read_versionstamp:
        return entry.commit_versionstamp < self._read_versionstamp
      else:
        return entry.deleted_versionstamp is None
  
  
  class NamespaceIterator(object):
    def __init__(self, tr, project_dir):
      self._tr = tr
      self._project_dir = project_dir
      self._done = False
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      # TODO: This can be made async.
      ns_dir = self._project_dir.open(self._tr, (KindIndex.DIR_NAME,))
      namespaces = ns_dir.list(self._tr)
  
      # The API uses an ID of 1 to label the default namespace.
      results = [IndexEntry(self._project_dir.get_path()[-1], u'',
                            (u'__namespace__', namespace or 1), None, None)
                 for namespace in namespaces]
  
      self._done = True
      raise gen.Return((results, False))
  
  
  class KindIterator(object):
    def __init__(self, tr, tornado_fdb, project_dir, namespace):
      self._tr = tr
      self._tornado_fdb = tornado_fdb
      self._project_dir = project_dir
      self._namespace = namespace
      self._done = False
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      # TODO: This can be made async.
      ns_dir = self._project_dir.open(
        self._tr, (KindIndex.DIR_NAME, self._namespace))
      kinds = ns_dir.list(self._tr)
      populated_kinds = [
        kind for kind, populated in zip(
          kinds, (yield [self._populated(ns_dir, kind) for kind in kinds]))
        if populated]
  
      results = [IndexEntry(self._project_dir.get_path()[-1], self._namespace,
                            (u'__kind__', kind), None, None)
                 for kind in populated_kinds]
  
      self._done = True
      raise gen.Return((results, False))
  
    @gen.coroutine
    def _populated(self, ns_dir, kind):
      """ Checks if at least one entity exists for a given kind. """
      kind_dir = ns_dir.open(self._tr, (kind,))
      kind_index = KindIndex(kind_dir)
      # TODO: Check if the presence of stat entities should mark a kind as being
      #  populated.
      index_slice = kind_index.get_slice(())
      # This query is reversed to increase the likelihood of getting a relevant
      # (not marked for GC) entry.
      iterator = IndexIterator(self._tr, self._tornado_fdb, kind_index,
                               index_slice, fetch_limit=1, reverse=True,
                               snapshot=True)
      while True:
        results, more_results = yield iterator.next_page()
        if results:
          raise gen.Return(True)
  
        if not more_results:
          raise gen.Return(False)
  
  
  class MergeJoinIterator(object):
    """
    Returns pages of index entry results from multiple ranges. It ignores
    Key-Values that do not apply to the given read_versionstamp. It converts
    usable Key-Values to IndexEntry objects.
    """
    def __init__(self, tr, tornado_fdb, filter_props, indexes, fetch_limit,
                 read_versionstamp=None, ancestor_path=None, snapshot=False):
      self.indexes = indexes
      self._filter_props = filter_props
      self._read_versionstamp = read_versionstamp
      self._tr = tr
      self._tornado_fdb = tornado_fdb
      self._fetch_limit = fetch_limit
      self._fetched = 0
      self._snapshot = snapshot
      self._done = False
      self._candidate_path = None
      self._candidate_entries = []
      self._ancestor_path = ancestor_path
  
    @property
    def prop_names(self):
      prop_names = set()
      for index, _, _, _ in self.indexes:
        prop_names.update(index.prop_names)
  
      return tuple(prop_names)
  
    @gen.coroutine
    def next_page(self):
      if self._done:
        raise gen.Return(([], False))
  
      composite_entry = None
      for i, (index, key_slice, prop_name, value) in enumerate(self.indexes):
        usable_entry = None
        # TODO: Keep cache of ranges to reduce unnecessary lookups.
        index_exhausted = False
        while True:
          results, count, more = yield self._tornado_fdb.get_range(
            self._tr, key_slice, 0, fdb.StreamingMode.small, 1,
            snapshot=self._snapshot)
          if not count:
            index_exhausted = True
            break
  
          key_slice = slice(fdb.KeySelector.first_greater_than(results[-1].key),
                            key_slice.stop)
          for result in results:
            entry = index.decode(result)
            if self._usable(entry):
              usable_entry = entry
              break
  
          if usable_entry is not None:
            break
  
        if index_exhausted:
          self._done = True
          break
  
        if usable_entry.path == self._candidate_path:
          self._candidate_entries.append(usable_entry)
        else:
          self._candidate_path = usable_entry.path
          self._candidate_entries = [usable_entry]
  
        next_index_op = Query_Filter.GREATER_THAN_OR_EQUAL
        if len(self._candidate_entries) == len(self.indexes):
          properties = []
          for partial_entry in self._candidate_entries:
            if isinstance(partial_entry, PropertyEntry):
              properties.append((partial_entry.prop_name, partial_entry.value))
            else:
              for property in partial_entry.properties:
                if property not in properties:
                  properties.append(property)
  
          composite_entry = CompositeEntry(
            usable_entry.project_id, usable_entry.namespace,
            self._candidate_path, properties, usable_entry.commit_versionstamp,
            usable_entry.deleted_versionstamp)
          self._candidate_entries = []
          next_index_op = Query_Filter.GREATER_THAN
  
        next_index_i = (i + 1) % len(self.indexes)
        next_index, next_slice, next_prop_name, next_value =\
          self.indexes[next_index_i]
  
        # TODO: This probably doesn't work for all cases.
        last_prop = None
        if isinstance(next_index, CompositeIndex):
          last_prop = next_index.prop_names[-1]
  
        tmp_filter_props = []
        for filter_prop in self._filter_props:
          if (filter_prop.name in (next_prop_name, last_prop, KEY_PROP) or
              filter_prop.name not in next_index.prop_names):
            continue
          else:
            tmp_filter_props.append(filter_prop)
  
        tmp_filter_props.append(
          FilterProperty(next_prop_name, [(Query_Filter.EQUAL, next_value)]))
  
        if last_prop is not None:
          val = next(value for prop_name, value in usable_entry.properties
                     if prop_name == last_prop)
          tmp_filter_props.append(
            FilterProperty(last_prop, [(Query_Filter.EQUAL, val)]))
  
        tmp_filter_props.append(
          FilterProperty(KEY_PROP, [(next_index_op, usable_entry.path)]))
  
        new_slice = next_index.get_slice(tmp_filter_props,
                                         ancestor_path=self._ancestor_path)
        self.indexes[next_index_i][1] = new_slice
  
      entries = [composite_entry] if composite_entry is not None else []
      self._fetched += len(entries)
      if self._fetched == self._fetch_limit:
        self._done = True
  
      raise gen.Return((entries, not self._done))
  
    def _usable(self, entry):
      if self._read_versionstamp and entry.deleted_versionstamp:
        return (entry.commit_versionstamp < self._read_versionstamp <
                entry.deleted_versionstamp)
      elif self._read_versionstamp:
        return entry.commit_versionstamp < self._read_versionstamp
      else:
        return entry.deleted_versionstamp is None
  
  
  class IndexSlice(object):
    """ Encapsulates details about an index range in a way that's mutable. """
    __slots__ = ['_directory_prefix', '_order_info', '_ancestor', '_start_parts',
                 '_stop_parts']
  
    def __init__(self, directory_prefix, order_info, ancestor=False):
      self._directory_prefix = directory_prefix
      self._order_info = order_info
      self._ancestor = ancestor
  
      self._start_parts = [self._directory_prefix]
      self._stop_parts = [self._directory_prefix, b'\xFF']
  
    @property
    def start(self):
      return first_gt_or_equal(b''.join(self._start_parts))
  
    @property
    def stop(self):
      return first_gt_or_equal(b''.join(self._stop_parts))
  
    @property
    def _expected_parts(self):
      total = 1  # directory prefix
      if self._ancestor:
        total += 1
  
      total += len(self._order_info)
      total += 1  # path
      total += 1  # commit versionstamp
      return total
  
    def set_ancestor(self, ancestor_path):
      if not ancestor_path:
        return
  
      index = 1 if self._ancestor else -2
      if self._ancestor:
        self._set_start(index, Path.pack(ancestor_path))
        self._set_stop(index, Path.pack(ancestor_path))
        self._set_stop(index + 1, b'\xFF')
      else:
        prefix = Path.pack(ancestor_path, omit_terminator=True)
        self._set_start(index, prefix)
        self._set_stop(index, prefix + b'\xFF')
  
    def apply_prop_filter(self, prop_name, op, value):
      index, direction = self._prop_details(prop_name)
      prop_reverse = direction == Query_Order.DESCENDING
      encoded_value = encode_value(value, prop_reverse)
      if op == Query_Filter.EQUAL:
        self._set_start(index, encoded_value)
        self._set_stop(index, encoded_value)
        self._set_stop(index + 1, b'\xFF')
        return
  
      if (op == Query_Filter.GREATER_THAN_OR_EQUAL and not prop_reverse or
          op == Query_Filter.LESS_THAN_OR_EQUAL and prop_reverse):
        self._set_start(index, encoded_value)
      elif (op == Query_Filter.GREATER_THAN and not prop_reverse or
            op == Query_Filter.LESS_THAN and prop_reverse):
        self._set_start(index, encoded_value + b'\xFF')
      elif (op == Query_Filter.LESS_THAN_OR_EQUAL and not prop_reverse or
            op == Query_Filter.GREATER_THAN_OR_EQUAL and prop_reverse):
        self._set_stop(index, encoded_value)
        self._set_stop(index + 1, b'\xFF')
      elif (op == Query_Filter.LESS_THAN and not prop_reverse or
            op == Query_Filter.GREATER_THAN and prop_reverse):
        self._set_stop(index, encoded_value)
      else:
        raise BadRequest(u'Unexpected filter operation')
  
    def apply_path_filter(self, op, path, ancestor_path=()):
      if not isinstance(path, tuple):
        path = Path.flatten(path)
  
      remaining_path = path[len(ancestor_path):] if self._ancestor else path
      if not remaining_path:
        raise InternalError(u'Path filter must be within ancestor')
  
      start = Path.pack(remaining_path, omit_terminator=True)
      # Since the commit versionstamp could potentially start with 0
  
  [... the rest is omitted ...]
files: AppDB/appscale/datastore/fdb/indexes.py