From d7a0496f3ec534076121632352f44733253e1e16 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 19 Mar 2015 15:59:48 +0000 Subject: Convert storage layer to be mysql compatible --- synapse/storage/room.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'synapse/storage/room.py') diff --git a/synapse/storage/room.py b/synapse/storage/room.py index 549c9af393..3c23f29215 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -114,9 +114,9 @@ class RoomStore(SQLBaseStore): "name": name_subquery, } - c = txn.execute(sql, (is_public,)) + txn.execute(sql, (is_public,)) - return c.fetchall() + return txn.fetchall() rows = yield self.runInteraction( "get_rooms", f -- cgit 1.4.1 From d4f50f3ae5a23886eb271ad8cee92d5fb5aae8b0 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 20 Mar 2015 10:57:08 +0000 Subject: decode_result takes an iterable --- synapse/storage/room.py | 5 ++++- synapse/storage/transactions.py | 2 +- 2 files changed, 5 insertions(+), 2 deletions(-) (limited to 'synapse/storage/room.py') diff --git a/synapse/storage/room.py b/synapse/storage/room.py index 3c23f29215..b5031f5c77 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -68,7 +68,10 @@ class RoomStore(SQLBaseStore): """ query = RoomsTable.select_statement("room_id=?") return self._execute( - "get_room", RoomsTable.decode_single_result, query, room_id, + "get_room", + lambda txn: RoomsTable.decode_single_result(txn.fetchall()), + query, + room_id, ) @defer.inlineCallbacks diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py index b5ed5453d8..92eec69ef4 100644 --- a/synapse/storage/transactions.py +++ b/synapse/storage/transactions.py @@ -122,7 +122,7 @@ class TransactionStore(SQLBaseStore): ) txn.execute(query, (destination,)) - results = SentTransactions.decode_results(txn) + results = SentTransactions.decode_results(txn.fetchall()) prev_txns = [r.transaction_id for r in results] -- cgit 1.4.1 From cb8162d3d17a97574073d49bd6eef51c93b68157 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 20 Mar 2015 13:52:56 +0000 Subject: Rearrange storage modules --- synapse/storage/__init__.py | 466 +------------------------------------------- synapse/storage/_base.py | 7 + synapse/storage/events.py | 387 ++++++++++++++++++++++++++++++++++++ synapse/storage/feedback.py | 47 ----- synapse/storage/room.py | 37 ++++ synapse/storage/state.py | 32 +++ synapse/storage/stream.py | 19 ++ 7 files changed, 486 insertions(+), 509 deletions(-) create mode 100644 synapse/storage/events.py delete mode 100644 synapse/storage/feedback.py (limited to 'synapse/storage/room.py') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 30cba47717..8604746fb9 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -15,12 +15,9 @@ from twisted.internet import defer -from synapse.util.logutils import log_function -from synapse.api.constants import EventTypes - from .appservice import ApplicationServiceStore from .directory import DirectoryStore -from .feedback import FeedbackStore +from .events import EventsStore from .presence import PresenceStore from .profile import ProfileStore from .registration import RegistrationStore @@ -39,11 +36,6 @@ from .state import StateStore from .signatures import SignatureStore from .filtering import FilteringStore -from syutil.base64util import decode_base64 -from syutil.jsonutil import encode_canonical_json - -from synapse.crypto.event_signing import compute_event_reference_hash - import fnmatch import imp @@ -64,15 +56,8 @@ SCHEMA_VERSION = 14 dir_path = os.path.abspath(os.path.dirname(__file__)) -class _RollbackButIsFineException(Exception): - """ This exception is used to rollback a transaction without implying - something went wrong. - """ - pass - - class DataStore(RoomMemberStore, RoomStore, - RegistrationStore, StreamStore, ProfileStore, FeedbackStore, + RegistrationStore, StreamStore, ProfileStore, PresenceStore, TransactionStore, DirectoryStore, KeyStore, StateStore, SignatureStore, ApplicationServiceStore, @@ -81,7 +66,8 @@ class DataStore(RoomMemberStore, RoomStore, RejectionsStore, FilteringStore, PusherStore, - PushRuleStore + PushRuleStore, + EventsStore, ): def __init__(self, hs): @@ -94,412 +80,6 @@ class DataStore(RoomMemberStore, RoomStore, self._next_stream_id_lock = threading.Lock() self._next_stream_id = int(hs.get_clock().time_msec()) * 1000 - @defer.inlineCallbacks - @log_function - def persist_event(self, event, context, backfilled=False, - is_new_state=True, current_state=None): - stream_ordering = None - if backfilled: - if not self.min_token_deferred.called: - yield self.min_token_deferred - self.min_token -= 1 - stream_ordering = self.min_token - - try: - yield self.runInteraction( - "persist_event", - self._persist_event_txn, - event=event, - context=context, - backfilled=backfilled, - stream_ordering=stream_ordering, - is_new_state=is_new_state, - current_state=current_state, - ) - except _RollbackButIsFineException: - pass - - @defer.inlineCallbacks - def get_event(self, event_id, check_redacted=True, - get_prev_content=False, allow_rejected=False, - allow_none=False): - """Get an event from the database by event_id. - - Args: - event_id (str): The event_id of the event to fetch - check_redacted (bool): If True, check if event has been redacted - and redact it. - get_prev_content (bool): If True and event is a state event, - include the previous states content in the unsigned field. - allow_rejected (bool): If True return rejected events. - allow_none (bool): If True, return None if no event found, if - False throw an exception. - - Returns: - Deferred : A FrozenEvent. - """ - event = yield self.runInteraction( - "get_event", self._get_event_txn, - event_id, - check_redacted=check_redacted, - get_prev_content=get_prev_content, - allow_rejected=allow_rejected, - ) - - if not event and not allow_none: - raise RuntimeError("Could not find event %s" % (event_id,)) - - defer.returnValue(event) - - @log_function - def _persist_event_txn(self, txn, event, context, backfilled, - stream_ordering=None, is_new_state=True, - current_state=None): - - # Remove the any existing cache entries for the event_id - self._get_event_cache.pop(event.event_id) - - # We purposefully do this first since if we include a `current_state` - # key, we *want* to update the `current_state_events` table - if current_state: - txn.execute( - "DELETE FROM current_state_events WHERE room_id = ?", - (event.room_id,) - ) - - for s in current_state: - self._simple_insert_txn( - txn, - "current_state_events", - { - "event_id": s.event_id, - "room_id": s.room_id, - "type": s.type, - "state_key": s.state_key, - }, - ) - - if event.is_state() and is_new_state: - if not backfilled and not context.rejected: - self._simple_insert_txn( - txn, - table="state_forward_extremities", - values={ - "event_id": event.event_id, - "room_id": event.room_id, - "type": event.type, - "state_key": event.state_key, - }, - ) - - for prev_state_id, _ in event.prev_state: - self._simple_delete_txn( - txn, - table="state_forward_extremities", - keyvalues={ - "event_id": prev_state_id, - } - ) - - outlier = event.internal_metadata.is_outlier() - - if not outlier: - self._store_state_groups_txn(txn, event, context) - - self._update_min_depth_for_room_txn( - txn, - event.room_id, - event.depth - ) - - self._handle_prev_events( - txn, - outlier=outlier, - event_id=event.event_id, - prev_events=event.prev_events, - room_id=event.room_id, - ) - - have_persisted = self._simple_select_one_onecol_txn( - txn, - table="event_json", - keyvalues={"event_id": event.event_id}, - retcol="event_id", - allow_none=True, - ) - - metadata_json = encode_canonical_json( - event.internal_metadata.get_dict() - ) - - # If we have already persisted this event, we don't need to do any - # more processing. - # The processing above must be done on every call to persist event, - # since they might not have happened on previous calls. For example, - # if we are persisting an event that we had persisted as an outlier, - # but is no longer one. - if have_persisted: - if not outlier: - sql = ( - "UPDATE event_json SET internal_metadata = ?" - " WHERE event_id = ?" - ) - txn.execute( - sql, - (metadata_json.decode("UTF-8"), event.event_id,) - ) - - sql = ( - "UPDATE events SET outlier = 0" - " WHERE event_id = ?" - ) - txn.execute( - sql, - (event.event_id,) - ) - return - - if event.type == EventTypes.Member: - self._store_room_member_txn(txn, event) - elif event.type == EventTypes.Feedback: - self._store_feedback_txn(txn, event) - elif event.type == EventTypes.Name: - self._store_room_name_txn(txn, event) - elif event.type == EventTypes.Topic: - self._store_room_topic_txn(txn, event) - elif event.type == EventTypes.Redaction: - self._store_redaction(txn, event) - - event_dict = { - k: v - for k, v in event.get_dict().items() - if k not in [ - "redacted", - "redacted_because", - ] - } - - self._simple_insert_txn( - txn, - table="event_json", - values={ - "event_id": event.event_id, - "room_id": event.room_id, - "internal_metadata": metadata_json.decode("UTF-8"), - "json": encode_canonical_json(event_dict).decode("UTF-8"), - }, - ) - - content = encode_canonical_json( - event.content - ).decode("UTF-8") - - vals = { - "topological_ordering": event.depth, - "event_id": event.event_id, - "type": event.type, - "room_id": event.room_id, - "content": content, - "processed": True, - "outlier": outlier, - "depth": event.depth, - } - - if stream_ordering is None: - stream_ordering = self.get_next_stream_id() - - - unrec = { - k: v - for k, v in event.get_dict().items() - if k not in vals.keys() and k not in [ - "redacted", - "redacted_because", - "signatures", - "hashes", - "prev_events", - ] - } - - vals["unrecognized_keys"] = encode_canonical_json( - unrec - ).decode("UTF-8") - - sql = ( - "INSERT INTO events" - " (stream_ordering, topological_ordering, event_id, type," - " room_id, content, processed, outlier, depth)" - " VALUES (%s,?,?,?,?,?,?,?,?)" - ) % (stream_ordering,) - - txn.execute( - sql, - (event.depth, event.event_id, event.type, event.room_id, - content, True, outlier, event.depth) - ) - - if context.rejected: - self._store_rejections_txn(txn, event.event_id, context.rejected) - - if event.is_state(): - vals = { - "event_id": event.event_id, - "room_id": event.room_id, - "type": event.type, - "state_key": event.state_key, - } - - # TODO: How does this work with backfilling? - if hasattr(event, "replaces_state"): - vals["prev_state"] = event.replaces_state - - self._simple_insert_txn( - txn, - "state_events", - vals, - ) - - if is_new_state and not context.rejected: - self._simple_insert_txn( - txn, - "current_state_events", - { - "event_id": event.event_id, - "room_id": event.room_id, - "type": event.type, - "state_key": event.state_key, - }, - ) - - for e_id, h in event.prev_state: - self._simple_insert_txn( - txn, - table="event_edges", - values={ - "event_id": event.event_id, - "prev_event_id": e_id, - "room_id": event.room_id, - "is_state": 1, - }, - ) - - for hash_alg, hash_base64 in event.hashes.items(): - hash_bytes = decode_base64(hash_base64) - self._store_event_content_hash_txn( - txn, event.event_id, hash_alg, hash_bytes, - ) - - for prev_event_id, prev_hashes in event.prev_events: - for alg, hash_base64 in prev_hashes.items(): - hash_bytes = decode_base64(hash_base64) - self._store_prev_event_hash_txn( - txn, event.event_id, prev_event_id, alg, hash_bytes - ) - - for auth_id, _ in event.auth_events: - self._simple_insert_txn( - txn, - table="event_auth", - values={ - "event_id": event.event_id, - "room_id": event.room_id, - "auth_id": auth_id, - }, - ) - - (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event) - self._store_event_reference_hash_txn( - txn, event.event_id, ref_alg, ref_hash_bytes - ) - - def _store_redaction(self, txn, event): - # invalidate the cache for the redacted event - self._get_event_cache.pop(event.redacts) - txn.execute( - "INSERT INTO redactions (event_id, redacts) VALUES (?,?)", - (event.event_id, event.redacts) - ) - - @defer.inlineCallbacks - def get_current_state(self, room_id, event_type=None, state_key=""): - del_sql = ( - "SELECT event_id FROM redactions WHERE redacts = e.event_id " - "LIMIT 1" - ) - - sql = ( - "SELECT e.*, (%(redacted)s) AS redacted FROM events as e " - "INNER JOIN current_state_events as c ON e.event_id = c.event_id " - "INNER JOIN state_events as s ON e.event_id = s.event_id " - "WHERE c.room_id = ? " - ) % { - "redacted": del_sql, - } - - if event_type and state_key is not None: - sql += " AND s.type = ? AND s.state_key = ? " - args = (room_id, event_type, state_key) - elif event_type: - sql += " AND s.type = ?" - args = (room_id, event_type) - else: - args = (room_id, ) - - results = yield self._execute_and_decode("get_current_state", sql, *args) - - events = yield self._parse_events(results) - defer.returnValue(events) - - @defer.inlineCallbacks - def get_room_name_and_aliases(self, room_id): - del_sql = ( - "SELECT event_id FROM redactions WHERE redacts = e.event_id " - "LIMIT 1" - ) - - sql = ( - "SELECT e.*, (%(redacted)s) AS redacted FROM events as e " - "INNER JOIN current_state_events as c ON e.event_id = c.event_id " - "INNER JOIN state_events as s ON e.event_id = s.event_id " - "WHERE c.room_id = ? " - ) % { - "redacted": del_sql, - } - - sql += " AND ((s.type = 'm.room.name' AND s.state_key = '')" - sql += " OR s.type = 'm.room.aliases')" - args = (room_id,) - - results = yield self._execute_and_decode("get_current_state", sql, *args) - - events = yield self._parse_events(results) - - name = None - aliases = [] - - for e in events: - if e.type == 'm.room.name': - if 'name' in e.content: - name = e.content['name'] - elif e.type == 'm.room.aliases': - if 'aliases' in e.content: - aliases.extend(e.content['aliases']) - - defer.returnValue((name, aliases)) - - @defer.inlineCallbacks - def _get_min_token(self): - row = yield self._execute( - "_get_min_token", None, "SELECT MIN(stream_ordering) FROM events" - ) - - self.min_token = row[0][0] if row and row[0] and row[0][0] else -1 - self.min_token = min(self.min_token, -1) - - logger.debug("min_token is: %s", self.min_token) - - defer.returnValue(self.min_token) - def insert_client_ip(self, user, access_token, device_id, ip, user_agent): return self._simple_insert( "user_ips", @@ -523,44 +103,6 @@ class DataStore(RoomMemberStore, RoomStore, ], ) - def have_events(self, event_ids): - """Given a list of event ids, check if we have already processed them. - - Returns: - dict: Has an entry for each event id we already have seen. Maps to - the rejected reason string if we rejected the event, else maps to - None. - """ - if not event_ids: - return defer.succeed({}) - - def f(txn): - sql = ( - "SELECT e.event_id, reason FROM events as e " - "LEFT JOIN rejections as r ON e.event_id = r.event_id " - "WHERE e.event_id = ?" - ) - - res = {} - for event_id in event_ids: - txn.execute(sql, (event_id,)) - row = txn.fetchone() - if row: - _, rejected = row - res[event_id] = rejected - - return res - - return self.runInteraction( - "have_events", f, - ) - - def get_next_stream_id(self): - with self._next_stream_id_lock: - i = self._next_stream_id - self._next_stream_id += 1 - return i - def read_schema(path): """ Read the named database schema. diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 24ff872dad..37bb28e6cf 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -791,6 +791,13 @@ class SQLBaseStore(object): return result[0] if result else None +class _RollbackButIsFineException(Exception): + """ This exception is used to rollback a transaction without implying + something went wrong. + """ + pass + + class Table(object): """ A base class used to store information about a particular table. """ diff --git a/synapse/storage/events.py b/synapse/storage/events.py new file mode 100644 index 0000000000..b222dfb4aa --- /dev/null +++ b/synapse/storage/events.py @@ -0,0 +1,387 @@ +# -*- coding: utf-8 -*- +# Copyright 2014, 2015 OpenMarket Ltd +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from _base import SQLBaseStore, _RollbackButIsFineException + +from twisted.internet import defer + +from synapse.util.logutils import log_function +from synapse.api.constants import EventTypes +from synapse.crypto.event_signing import compute_event_reference_hash + +from syutil.base64util import decode_base64 +from syutil.jsonutil import encode_canonical_json + +import logging + +logger = logging.getLogger(__name__) + + +class EventsStore(SQLBaseStore): + @defer.inlineCallbacks + @log_function + def persist_event(self, event, context, backfilled=False, + is_new_state=True, current_state=None): + stream_ordering = None + if backfilled: + if not self.min_token_deferred.called: + yield self.min_token_deferred + self.min_token -= 1 + stream_ordering = self.min_token + + try: + yield self.runInteraction( + "persist_event", + self._persist_event_txn, + event=event, + context=context, + backfilled=backfilled, + stream_ordering=stream_ordering, + is_new_state=is_new_state, + current_state=current_state, + ) + except _RollbackButIsFineException: + pass + + @defer.inlineCallbacks + def get_event(self, event_id, check_redacted=True, + get_prev_content=False, allow_rejected=False, + allow_none=False): + """Get an event from the database by event_id. + + Args: + event_id (str): The event_id of the event to fetch + check_redacted (bool): If True, check if event has been redacted + and redact it. + get_prev_content (bool): If True and event is a state event, + include the previous states content in the unsigned field. + allow_rejected (bool): If True return rejected events. + allow_none (bool): If True, return None if no event found, if + False throw an exception. + + Returns: + Deferred : A FrozenEvent. + """ + event = yield self.runInteraction( + "get_event", self._get_event_txn, + event_id, + check_redacted=check_redacted, + get_prev_content=get_prev_content, + allow_rejected=allow_rejected, + ) + + if not event and not allow_none: + raise RuntimeError("Could not find event %s" % (event_id,)) + + defer.returnValue(event) + + @log_function + def _persist_event_txn(self, txn, event, context, backfilled, + stream_ordering=None, is_new_state=True, + current_state=None): + + # Remove the any existing cache entries for the event_id + self._get_event_cache.pop(event.event_id) + + # We purposefully do this first since if we include a `current_state` + # key, we *want* to update the `current_state_events` table + if current_state: + txn.execute( + "DELETE FROM current_state_events WHERE room_id = ?", + (event.room_id,) + ) + + for s in current_state: + self._simple_insert_txn( + txn, + "current_state_events", + { + "event_id": s.event_id, + "room_id": s.room_id, + "type": s.type, + "state_key": s.state_key, + }, + ) + + if event.is_state() and is_new_state: + if not backfilled and not context.rejected: + self._simple_insert_txn( + txn, + table="state_forward_extremities", + values={ + "event_id": event.event_id, + "room_id": event.room_id, + "type": event.type, + "state_key": event.state_key, + }, + ) + + for prev_state_id, _ in event.prev_state: + self._simple_delete_txn( + txn, + table="state_forward_extremities", + keyvalues={ + "event_id": prev_state_id, + } + ) + + outlier = event.internal_metadata.is_outlier() + + if not outlier: + self._store_state_groups_txn(txn, event, context) + + self._update_min_depth_for_room_txn( + txn, + event.room_id, + event.depth + ) + + self._handle_prev_events( + txn, + outlier=outlier, + event_id=event.event_id, + prev_events=event.prev_events, + room_id=event.room_id, + ) + + have_persisted = self._simple_select_one_onecol_txn( + txn, + table="event_json", + keyvalues={"event_id": event.event_id}, + retcol="event_id", + allow_none=True, + ) + + metadata_json = encode_canonical_json( + event.internal_metadata.get_dict() + ) + + # If we have already persisted this event, we don't need to do any + # more processing. + # The processing above must be done on every call to persist event, + # since they might not have happened on previous calls. For example, + # if we are persisting an event that we had persisted as an outlier, + # but is no longer one. + if have_persisted: + if not outlier: + sql = ( + "UPDATE event_json SET internal_metadata = ?" + " WHERE event_id = ?" + ) + txn.execute( + sql, + (metadata_json.decode("UTF-8"), event.event_id,) + ) + + sql = ( + "UPDATE events SET outlier = 0" + " WHERE event_id = ?" + ) + txn.execute( + sql, + (event.event_id,) + ) + return + + if event.type == EventTypes.Member: + self._store_room_member_txn(txn, event) + elif event.type == EventTypes.Name: + self._store_room_name_txn(txn, event) + elif event.type == EventTypes.Topic: + self._store_room_topic_txn(txn, event) + elif event.type == EventTypes.Redaction: + self._store_redaction(txn, event) + + event_dict = { + k: v + for k, v in event.get_dict().items() + if k not in [ + "redacted", + "redacted_because", + ] + } + + self._simple_insert_txn( + txn, + table="event_json", + values={ + "event_id": event.event_id, + "room_id": event.room_id, + "internal_metadata": metadata_json.decode("UTF-8"), + "json": encode_canonical_json(event_dict).decode("UTF-8"), + }, + ) + + content = encode_canonical_json( + event.content + ).decode("UTF-8") + + vals = { + "topological_ordering": event.depth, + "event_id": event.event_id, + "type": event.type, + "room_id": event.room_id, + "content": content, + "processed": True, + "outlier": outlier, + "depth": event.depth, + } + + if stream_ordering is None: + stream_ordering = self.get_next_stream_id() + + + unrec = { + k: v + for k, v in event.get_dict().items() + if k not in vals.keys() and k not in [ + "redacted", + "redacted_because", + "signatures", + "hashes", + "prev_events", + ] + } + + vals["unrecognized_keys"] = encode_canonical_json( + unrec + ).decode("UTF-8") + + sql = ( + "INSERT INTO events" + " (stream_ordering, topological_ordering, event_id, type," + " room_id, content, processed, outlier, depth)" + " VALUES (%s,?,?,?,?,?,?,?,?)" + ) % (stream_ordering,) + + txn.execute( + sql, + (event.depth, event.event_id, event.type, event.room_id, + content, True, outlier, event.depth) + ) + + if context.rejected: + self._store_rejections_txn(txn, event.event_id, context.rejected) + + if event.is_state(): + vals = { + "event_id": event.event_id, + "room_id": event.room_id, + "type": event.type, + "state_key": event.state_key, + } + + # TODO: How does this work with backfilling? + if hasattr(event, "replaces_state"): + vals["prev_state"] = event.replaces_state + + self._simple_insert_txn( + txn, + "state_events", + vals, + ) + + if is_new_state and not context.rejected: + self._simple_insert_txn( + txn, + "current_state_events", + { + "event_id": event.event_id, + "room_id": event.room_id, + "type": event.type, + "state_key": event.state_key, + }, + ) + + for e_id, h in event.prev_state: + self._simple_insert_txn( + txn, + table="event_edges", + values={ + "event_id": event.event_id, + "prev_event_id": e_id, + "room_id": event.room_id, + "is_state": 1, + }, + ) + + for hash_alg, hash_base64 in event.hashes.items(): + hash_bytes = decode_base64(hash_base64) + self._store_event_content_hash_txn( + txn, event.event_id, hash_alg, hash_bytes, + ) + + for prev_event_id, prev_hashes in event.prev_events: + for alg, hash_base64 in prev_hashes.items(): + hash_bytes = decode_base64(hash_base64) + self._store_prev_event_hash_txn( + txn, event.event_id, prev_event_id, alg, hash_bytes + ) + + for auth_id, _ in event.auth_events: + self._simple_insert_txn( + txn, + table="event_auth", + values={ + "event_id": event.event_id, + "room_id": event.room_id, + "auth_id": auth_id, + }, + ) + + (ref_alg, ref_hash_bytes) = compute_event_reference_hash(event) + self._store_event_reference_hash_txn( + txn, event.event_id, ref_alg, ref_hash_bytes + ) + + def _store_redaction(self, txn, event): + # invalidate the cache for the redacted event + self._get_event_cache.pop(event.redacts) + txn.execute( + "INSERT INTO redactions (event_id, redacts) VALUES (?,?)", + (event.event_id, event.redacts) + ) + + def have_events(self, event_ids): + """Given a list of event ids, check if we have already processed them. + + Returns: + dict: Has an entry for each event id we already have seen. Maps to + the rejected reason string if we rejected the event, else maps to + None. + """ + if not event_ids: + return defer.succeed({}) + + def f(txn): + sql = ( + "SELECT e.event_id, reason FROM events as e " + "LEFT JOIN rejections as r ON e.event_id = r.event_id " + "WHERE e.event_id = ?" + ) + + res = {} + for event_id in event_ids: + txn.execute(sql, (event_id,)) + row = txn.fetchone() + if row: + _, rejected = row + res[event_id] = rejected + + return res + + return self.runInteraction( + "have_events", f, + ) diff --git a/synapse/storage/feedback.py b/synapse/storage/feedback.py deleted file mode 100644 index 8eab769b71..0000000000 --- a/synapse/storage/feedback.py +++ /dev/null @@ -1,47 +0,0 @@ -# -*- coding: utf-8 -*- -# Copyright 2014, 2015 OpenMarket Ltd -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -from twisted.internet import defer - -from ._base import SQLBaseStore - - -class FeedbackStore(SQLBaseStore): - - def _store_feedback_txn(self, txn, event): - self._simple_insert_txn(txn, "feedback", { - "event_id": event.event_id, - "feedback_type": event.content["type"], - "room_id": event.room_id, - "target_event_id": event.content["target_event_id"], - "sender": event.user_id, - }) - - @defer.inlineCallbacks - def get_feedback_for_event(self, event_id): - sql = ( - "SELECT events.* FROM events INNER JOIN feedback " - "ON events.event_id = feedback.event_id " - "WHERE feedback.target_event_id = ? " - ) - - rows = yield self._execute_and_decode("get_feedback_for_event", sql, event_id) - - defer.returnValue( - [ - (yield self._parse_events(r)) - for r in rows - ] - ) diff --git a/synapse/storage/room.py b/synapse/storage/room.py index b5031f5c77..c64f8f53ac 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -161,6 +161,43 @@ class RoomStore(SQLBaseStore): } ) + @defer.inlineCallbacks + def get_room_name_and_aliases(self, room_id): + del_sql = ( + "SELECT event_id FROM redactions WHERE redacts = e.event_id " + "LIMIT 1" + ) + + sql = ( + "SELECT e.*, (%(redacted)s) AS redacted FROM events as e " + "INNER JOIN current_state_events as c ON e.event_id = c.event_id " + "INNER JOIN state_events as s ON e.event_id = s.event_id " + "WHERE c.room_id = ? " + ) % { + "redacted": del_sql, + } + + sql += " AND ((s.type = 'm.room.name' AND s.state_key = '')" + sql += " OR s.type = 'm.room.aliases')" + args = (room_id,) + + results = yield self._execute_and_decode("get_current_state", sql, *args) + + events = yield self._parse_events(results) + + name = None + aliases = [] + + for e in events: + if e.type == 'm.room.name': + if 'name' in e.content: + name = e.content['name'] + elif e.type == 'm.room.aliases': + if 'aliases' in e.content: + aliases.extend(e.content['aliases']) + + defer.returnValue((name, aliases)) + class RoomsTable(Table): table_name = "rooms" diff --git a/synapse/storage/state.py b/synapse/storage/state.py index 888837cd1e..012144302d 100644 --- a/synapse/storage/state.py +++ b/synapse/storage/state.py @@ -15,6 +15,8 @@ from ._base import SQLBaseStore +from twisted.internet import defer + from synapse.util.stringutils import random_string import logging @@ -125,6 +127,36 @@ class StateStore(SQLBaseStore): or_replace=True, ) + @defer.inlineCallbacks + def get_current_state(self, room_id, event_type=None, state_key=""): + del_sql = ( + "SELECT event_id FROM redactions WHERE redacts = e.event_id " + "LIMIT 1" + ) + + sql = ( + "SELECT e.*, (%(redacted)s) AS redacted FROM events as e " + "INNER JOIN current_state_events as c ON e.event_id = c.event_id " + "INNER JOIN state_events as s ON e.event_id = s.event_id " + "WHERE c.room_id = ? " + ) % { + "redacted": del_sql, + } + + if event_type and state_key is not None: + sql += " AND s.type = ? AND s.state_key = ? " + args = (room_id, event_type, state_key) + elif event_type: + sql += " AND s.type = ?" + args = (room_id, event_type) + else: + args = (room_id, ) + + results = yield self._execute_and_decode("get_current_state", sql, *args) + + events = yield self._parse_events(results) + defer.returnValue(events) + def _make_group_id(clock): return str(int(clock.time_msec())) + random_string(5) diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 64adb0c7fa..9565fc77c5 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -419,6 +419,25 @@ class StreamStore(SQLBaseStore): self._get_room_events_max_id_txn ) + @defer.inlineCallbacks + def _get_min_token(self): + row = yield self._execute( + "_get_min_token", None, "SELECT MIN(stream_ordering) FROM events" + ) + + self.min_token = row[0][0] if row and row[0] and row[0][0] else -1 + self.min_token = min(self.min_token, -1) + + logger.debug("min_token is: %s", self.min_token) + + defer.returnValue(self.min_token) + + def get_next_stream_id(self): + with self._next_stream_id_lock: + i = self._next_stream_id + self._next_stream_id += 1 + return i + def _get_room_events_max_id_txn(self, txn): txn.execute( "SELECT MAX(stream_ordering) as m FROM events" -- cgit 1.4.1 From 0ba393924a823698eb5263a29ce6c4b50488ac4b Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 24 Mar 2015 16:31:52 +0000 Subject: Escape non printing ascii character --- synapse/storage/room.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) (limited to 'synapse/storage/room.py') diff --git a/synapse/storage/room.py b/synapse/storage/room.py index 8641033327..501e947ad7 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -102,10 +102,10 @@ class RoomStore(SQLBaseStore): "ON c.event_id = room_names.event_id " ) - # We use non printing ascii character US () as a seperator + # We use non printing ascii character US (\x1F) as a separator sql = ( "SELECT r.room_id, n.name, t.topic, " - "group_concat(a.room_alias, '') " + "group_concat(a.room_alias, '\x1F') " "FROM rooms AS r " "LEFT JOIN (%(topic)s) AS t ON t.room_id = r.room_id " "LEFT JOIN (%(name)s) AS n ON n.room_id = r.room_id " @@ -130,7 +130,7 @@ class RoomStore(SQLBaseStore): "room_id": r[0], "name": r[1], "topic": r[2], - "aliases": r[3].split(""), + "aliases": r[3].split("\x1F"), } for r in rows ] -- cgit 1.4.1 From 0e8f5095c7e7075b249ad53a9f60a4d2fdeeaaed Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 25 Mar 2015 17:15:20 +0000 Subject: Fix unicode database support --- synapse/app/homeserver.py | 47 +++++++++++++--------- synapse/handlers/login.py | 2 +- synapse/rest/client/v1/profile.py | 7 +++- synapse/storage/__init__.py | 12 ++++-- synapse/storage/_base.py | 4 ++ synapse/storage/events.py | 8 ++-- synapse/storage/keys.py | 4 +- synapse/storage/profile.py | 12 +++++- synapse/storage/registration.py | 18 +++++++-- synapse/storage/room.py | 1 + .../schema/full_schemas/11/media_repository.sql | 2 +- .../storage/schema/full_schemas/11/profiles.sql | 2 +- .../schema/full_schemas/11/transactions.sql | 1 - synapse/storage/signatures.py | 10 ++--- synapse/storage/transactions.py | 2 +- 15 files changed, 88 insertions(+), 44 deletions(-) (limited to 'synapse/storage/room.py') diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index 394e93e6c2..beab6ffc7a 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -110,14 +110,12 @@ class SynapseHomeServer(HomeServer): return None def build_db_pool(self): - name = self.db_config.pop("name", None) - if name == "MySQLdb": - return adbapi.ConnectionPool( - name, - **self.db_config - ) + name = self.db_config["name"] - raise RuntimeError("Unsupported database type") + return adbapi.ConnectionPool( + name, + **self.db_config.get("args", {}) + ) def create_resource_tree(self, redirect_root_to_web_client): """Create the resource tree for this Home Server. @@ -323,7 +321,7 @@ def change_resource_limit(soft_file_no): resource.setrlimit(resource.RLIMIT_NOFILE, (soft_file_no, hard)) logger.info("Set file limit to: %d", soft_file_no) - except (ValueError, resource.error) as e: + except ( ValueError, resource.error) as e: logger.warn("Failed to set file limit: %s", e) @@ -363,20 +361,33 @@ def setup(config_options): if config.database_config: with open(config.database_config, 'r') as f: db_config = yaml.safe_load(f) - - name = db_config.get("name", None) - if name == "MySQLdb": - db_config.update({ - "sql_mode": "TRADITIONAL", - "charset": "utf8", - "use_unicode": True, - }) else: db_config = { "name": "sqlite3", "database": config.database_path, } + db_config = { + k: v for k, v in db_config.items() + if not k.startswith("cp_") + } + + name = db_config.get("name", None) + if name in ["MySQLdb", "mysql.connector"]: + db_config.setdefault("args", {}).update({ + "sql_mode": "TRADITIONAL", + "charset": "utf8", + "use_unicode": True, + }) + elif name == "sqlite3": + db_config.setdefault("args", {}).update({ + "cp_min": 1, + "cp_max": 1, + "cp_openfun": prepare_database, + }) + else: + raise RuntimeError("Unsupported database type '%s'" % (name,)) + hs = SynapseHomeServer( config.server_name, domain_with_port=domain_with_port, @@ -401,8 +412,8 @@ def setup(config_options): # with sqlite3.connect(db_name) as db_conn: # prepare_sqlite3_database(db_conn) # prepare_database(db_conn) - import MySQLdb - db_conn = MySQLdb.connect(**db_config) + import mysql.connector + db_conn = mysql.connector.connect(**db_config.get("args", {})) prepare_database(db_conn) except UpgradeDatabaseException: sys.stderr.write( diff --git a/synapse/handlers/login.py b/synapse/handlers/login.py index 7447800460..76647c7941 100644 --- a/synapse/handlers/login.py +++ b/synapse/handlers/login.py @@ -57,7 +57,7 @@ class LoginHandler(BaseHandler): logger.warn("Attempted to login as %s but they do not exist", user) raise LoginError(403, "", errcode=Codes.FORBIDDEN) - stored_hash = user_info[0]["password_hash"] + stored_hash = user_info["password_hash"] if bcrypt.checkpw(password, stored_hash): # generate an access token and store it. token = self.reg_handler._generate_token(user) diff --git a/synapse/rest/client/v1/profile.py b/synapse/rest/client/v1/profile.py index 1e77eb49cf..7387b4adb9 100644 --- a/synapse/rest/client/v1/profile.py +++ b/synapse/rest/client/v1/profile.py @@ -19,9 +19,13 @@ from twisted.internet import defer from .base import ClientV1RestServlet, client_path_pattern from synapse.types import UserID +import logging import simplejson as json +logger = logging.getLogger(__name__) + + class ProfileDisplaynameRestServlet(ClientV1RestServlet): PATTERN = client_path_pattern("/profile/(?P[^/]*)/displayname") @@ -47,7 +51,8 @@ class ProfileDisplaynameRestServlet(ClientV1RestServlet): defer.returnValue((400, "Unable to parse name")) yield self.handlers.profile_handler.set_displayname( - user, auth_user, new_name) + user, auth_user, new_name + ) defer.returnValue((200, {})) diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index e03d55b00d..abde7d0df5 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -410,10 +410,14 @@ def executescript(txn, schema_path): def _get_or_create_schema_state(txn): - schema_path = os.path.join( - dir_path, "schema", "schema_version.sql", - ) - executescript(txn, schema_path) + try: + # Bluntly try creating the schema_version tables. + schema_path = os.path.join( + dir_path, "schema", "schema_version.sql", + ) + executescript(txn, schema_path) + except: + pass txn.execute("SELECT version, upgraded FROM schema_version") row = txn.fetchone() diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 1ea39bc0ad..76ec3ee93f 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -755,6 +755,8 @@ class SQLBaseStore(object): return None internal_metadata, js, redacted, rejected_reason = res + js = js.decode("utf8") + internal_metadata = internal_metadata.decode("utf8") start_time = update_counter("select_event", start_time) @@ -779,9 +781,11 @@ class SQLBaseStore(object): sql_getevents_timer.inc_by(curr_time - last_time, desc) return curr_time + logger.debug("Got js: %r", js) d = json.loads(js) start_time = update_counter("decode_json", start_time) + logger.debug("Got internal_metadata: %r", internal_metadata) internal_metadata = json.loads(internal_metadata) start_time = update_counter("decode_internal", start_time) diff --git a/synapse/storage/events.py b/synapse/storage/events.py index 4d636d3f46..69f598967e 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -294,15 +294,17 @@ class EventsStore(SQLBaseStore): ) if is_new_state and not context.rejected: - self._simple_insert_txn( + self._simple_upsert_txn( txn, "current_state_events", - { - "event_id": event.event_id, + keyvalues={ "room_id": event.room_id, "type": event.type, "state_key": event.state_key, }, + values={ + "event_id": event.event_id, + } ) for e_id, h in event.prev_state: diff --git a/synapse/storage/keys.py b/synapse/storage/keys.py index 25fef79434..e6975a945b 100644 --- a/synapse/storage/keys.py +++ b/synapse/storage/keys.py @@ -64,7 +64,7 @@ class KeyStore(SQLBaseStore): "fingerprint": fingerprint, "from_server": from_server, "ts_added_ms": time_now_ms, - "tls_certificate": buffer(tls_certificate_bytes), + "tls_certificate": tls_certificate_bytes, }, ) @@ -113,6 +113,6 @@ class KeyStore(SQLBaseStore): "key_id": "%s:%s" % (verify_key.alg, verify_key.version), "from_server": from_server, "ts_added_ms": time_now_ms, - "verify_key": buffer(verify_key.encode()), + "verify_key": verify_key.encode(), }, ) diff --git a/synapse/storage/profile.py b/synapse/storage/profile.py index a6e52cb248..09778045bf 100644 --- a/synapse/storage/profile.py +++ b/synapse/storage/profile.py @@ -13,6 +13,8 @@ # See the License for the specific language governing permissions and # limitations under the License. +from twisted.internet import defer + from ._base import SQLBaseStore @@ -24,19 +26,25 @@ class ProfileStore(SQLBaseStore): desc="create_profile", ) + @defer.inlineCallbacks def get_profile_displayname(self, user_localpart): - return self._simple_select_one_onecol( + name = yield self._simple_select_one_onecol( table="profiles", keyvalues={"user_id": user_localpart}, retcol="displayname", desc="get_profile_displayname", ) + if name: + name = name.decode("utf8") + + defer.returnValue(name) + def set_profile_displayname(self, user_localpart, new_displayname): return self._simple_update_one( table="profiles", keyvalues={"user_id": user_localpart}, - updatevalues={"displayname": new_displayname}, + updatevalues={"displayname": new_displayname.encode("utf8")}, desc="set_profile_displayname", ) diff --git a/synapse/storage/registration.py b/synapse/storage/registration.py index fe26d6d62f..7258f7b2a5 100644 --- a/synapse/storage/registration.py +++ b/synapse/storage/registration.py @@ -81,13 +81,23 @@ class RegistrationStore(SQLBaseStore): txn.execute("INSERT INTO access_tokens(user_id, token) " + "VALUES (?,?)", [user_id, token]) + @defer.inlineCallbacks def get_user_by_id(self, user_id): - query = ("SELECT users.name, users.password_hash FROM users" - " WHERE users.name = ?") - return self._execute( - "get_user_by_id", self.cursor_to_dict, query, user_id + user_info = yield self._simple_select_one( + table="users", + keyvalues={ + "name": user_id, + }, + retcols=["name", "password_hash"], + allow_none=True, ) + if user_info: + user_info["password_hash"] = user_info["password_hash"].decode("utf8") + + defer.returnValue(user_info) + + @cached() # TODO(paul): Currently there's no code to invalidate this cache. That # means if/when we ever add internal ways to invalidate access tokens or diff --git a/synapse/storage/room.py b/synapse/storage/room.py index 501e947ad7..a1a76280fe 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -72,6 +72,7 @@ class RoomStore(SQLBaseStore): keyvalues={"room_id": room_id}, retcols=RoomsTable.fields, desc="get_room", + allow_none=True, ) @defer.inlineCallbacks diff --git a/synapse/storage/schema/full_schemas/11/media_repository.sql b/synapse/storage/schema/full_schemas/11/media_repository.sql index 8bc84dc24d..d9559f5902 100644 --- a/synapse/storage/schema/full_schemas/11/media_repository.sql +++ b/synapse/storage/schema/full_schemas/11/media_repository.sql @@ -65,4 +65,4 @@ CREATE TABLE IF NOT EXISTS remote_media_cache_thumbnails ( ) ENGINE = INNODB; CREATE INDEX IF NOT EXISTS remote_media_cache_thumbnails_media_id - ON local_media_repository_thumbnails (media_id); + ON remote_media_cache_thumbnails (media_id); diff --git a/synapse/storage/schema/full_schemas/11/profiles.sql b/synapse/storage/schema/full_schemas/11/profiles.sql index 32defe2f79..552645c56f 100644 --- a/synapse/storage/schema/full_schemas/11/profiles.sql +++ b/synapse/storage/schema/full_schemas/11/profiles.sql @@ -14,7 +14,7 @@ */ CREATE TABLE IF NOT EXISTS profiles( user_id VARCHAR(255) NOT NULL, - displayname VARCHAR(255), + displayname VARBINARY(255), avatar_url VARCHAR(255), UNIQUE(user_id) ) ENGINE = INNODB; diff --git a/synapse/storage/schema/full_schemas/11/transactions.sql b/synapse/storage/schema/full_schemas/11/transactions.sql index 0570bf95d9..bd13bba8c2 100644 --- a/synapse/storage/schema/full_schemas/11/transactions.sql +++ b/synapse/storage/schema/full_schemas/11/transactions.sql @@ -38,7 +38,6 @@ CREATE TABLE IF NOT EXISTS sent_transactions( ) ENGINE = INNODB; CREATE INDEX IF NOT EXISTS sent_transaction_dest ON sent_transactions(destination); -CREATE INDEX IF NOT EXISTS sent_transaction_dest_referenced ON sent_transactions(destination); CREATE INDEX IF NOT EXISTS sent_transaction_txn_id ON sent_transactions(transaction_id); -- So that we can do an efficient look up of all transactions that have yet to be successfully -- sent. diff --git a/synapse/storage/signatures.py b/synapse/storage/signatures.py index 13ce335101..35bba854f9 100644 --- a/synapse/storage/signatures.py +++ b/synapse/storage/signatures.py @@ -54,7 +54,7 @@ class SignatureStore(SQLBaseStore): { "event_id": event_id, "algorithm": algorithm, - "hash": buffer(hash_bytes), + "hash": hash_bytes, }, ) @@ -99,7 +99,7 @@ class SignatureStore(SQLBaseStore): " WHERE event_id = ?" ) txn.execute(query, (event_id, )) - return dict(txn.fetchall()) + return {k: v for k, v in txn.fetchall()} def _store_event_reference_hash_txn(self, txn, event_id, algorithm, hash_bytes): @@ -116,7 +116,7 @@ class SignatureStore(SQLBaseStore): { "event_id": event_id, "algorithm": algorithm, - "hash": buffer(hash_bytes), + "hash": hash_bytes, }, ) @@ -160,7 +160,7 @@ class SignatureStore(SQLBaseStore): "event_id": event_id, "signature_name": signature_name, "key_id": key_id, - "signature": buffer(signature_bytes), + "signature": signature_bytes, }, ) @@ -193,6 +193,6 @@ class SignatureStore(SQLBaseStore): "event_id": event_id, "prev_event_id": prev_event_id, "algorithm": algorithm, - "hash": buffer(hash_bytes), + "hash": hash_bytes, }, ) diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py index 03e1e3b808..e3e484fb2d 100644 --- a/synapse/storage/transactions.py +++ b/synapse/storage/transactions.py @@ -282,7 +282,7 @@ class TransactionStore(SQLBaseStore): query = ( "UPDATE destinations" " SET retry_last_ts = ?, retry_interval = ?" - " WHERE destinations = ?" + " WHERE destination = ?" ) txn.execute( -- cgit 1.4.1 From 58d83399663a080c123d2f112b4f4d84accbc638 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 14 Apr 2015 13:53:20 +0100 Subject: Add support for postgres instead of mysql. Change sql accourdingly. blob + varbinary -> bytea. No support for UNSIGNED or CREATE INDEX IF NOT EXISTS. --- synapse/app/homeserver.py | 2 + synapse/storage/__init__.py | 15 +++----- synapse/storage/_base.py | 2 +- synapse/storage/engines/__init__.py | 2 + synapse/storage/event_federation.py | 10 ++--- synapse/storage/events.py | 4 +- synapse/storage/room.py | 34 ++++++++++++----- .../full_schemas/16/application_services.sql | 10 ++--- .../storage/schema/full_schemas/16/event_edges.sql | 26 ++++++------- .../schema/full_schemas/16/event_signatures.sql | 16 ++++---- synapse/storage/schema/full_schemas/16/im.sql | 44 +++++++++++----------- synapse/storage/schema/full_schemas/16/keys.sql | 8 ++-- .../schema/full_schemas/16/media_repository.sql | 8 ++-- .../storage/schema/full_schemas/16/presence.sql | 4 +- synapse/storage/schema/full_schemas/16/push.sql | 30 +++++++-------- .../storage/schema/full_schemas/16/redactions.sql | 4 +- synapse/storage/schema/full_schemas/16/state.sql | 10 ++--- .../schema/full_schemas/16/transactions.sql | 24 ++++++------ synapse/storage/schema/full_schemas/16/users.sql | 10 ++--- synapse/storage/schema/schema_version.sql | 14 +++---- synapse/storage/stream.py | 16 ++++---- 21 files changed, 153 insertions(+), 140 deletions(-) (limited to 'synapse/storage/room.py') diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index a47e548d66..033011e1d7 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -373,6 +373,8 @@ def setup(config_options): "use_unicode": True, "collation": "utf8mb4_bin", }) + elif name == "psycopg2": + pass elif name == "sqlite3": db_config.setdefault("args", {}).update({ "cp_min": 1, diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index b46cafd25e..272420194d 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -236,7 +236,7 @@ def _setup_new_database(cur, database_engine): cur.execute( database_engine.convert_param_style( - "REPLACE INTO schema_version (version, upgraded)" + "INSERT INTO schema_version (version, upgraded)" " VALUES (?,?)" ), (max_current_ver, False,) @@ -432,14 +432,11 @@ def executescript(txn, schema_path): def _get_or_create_schema_state(txn, database_engine): - try: - # Bluntly try creating the schema_version tables. - schema_path = os.path.join( - dir_path, "schema", "schema_version.sql", - ) - executescript(txn, schema_path) - except: - pass + # Bluntly try creating the schema_version tables. + schema_path = os.path.join( + dir_path, "schema", "schema_version.sql", + ) + executescript(txn, schema_path) txn.execute("SELECT version, upgraded FROM schema_version") row = txn.fetchone() diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index e30514cd5e..fa5199104a 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -330,7 +330,7 @@ class SQLBaseStore(object): continue raise except Exception as e: - logger.debug("[TXN FAIL] {%s}", name, e) + logger.debug("[TXN FAIL] {%s} %s", name, e) raise finally: end = time.time() * 1000 diff --git a/synapse/storage/engines/__init__.py b/synapse/storage/engines/__init__.py index 29702be923..548d4e1b42 100644 --- a/synapse/storage/engines/__init__.py +++ b/synapse/storage/engines/__init__.py @@ -14,6 +14,7 @@ # limitations under the License. from .maria import MariaEngine +from .postgres import PostgresEngine from .sqlite3 import Sqlite3Engine import importlib @@ -22,6 +23,7 @@ import importlib SUPPORTED_MODULE = { "sqlite3": Sqlite3Engine, "mysql.connector": MariaEngine, + "psycopg2": PostgresEngine, } diff --git a/synapse/storage/event_federation.py b/synapse/storage/event_federation.py index 79ad5ddc9c..54a3c9d805 100644 --- a/synapse/storage/event_federation.py +++ b/synapse/storage/event_federation.py @@ -153,7 +153,7 @@ class EventFederationStore(SQLBaseStore): results = self._get_prev_events_and_state( txn, event_id, - is_state=1, + is_state=True, ) return [(e_id, h, ) for e_id, h, _ in results] @@ -164,7 +164,7 @@ class EventFederationStore(SQLBaseStore): } if is_state is not None: - keyvalues["is_state"] = is_state + keyvalues["is_state"] = bool(is_state) res = self._simple_select_list_txn( txn, @@ -259,7 +259,7 @@ class EventFederationStore(SQLBaseStore): "event_id": event_id, "prev_event_id": e_id, "room_id": room_id, - "is_state": 0, + "is_state": False, }, ) @@ -397,7 +397,7 @@ class EventFederationStore(SQLBaseStore): query = ( "SELECT prev_event_id FROM event_edges " - "WHERE room_id = ? AND event_id = ? AND is_state = 0 " + "WHERE room_id = ? AND event_id = ? AND is_state = ? " "LIMIT ?" ) @@ -406,7 +406,7 @@ class EventFederationStore(SQLBaseStore): for event_id in front: txn.execute( query, - (room_id, event_id, limit - len(event_results)) + (room_id, event_id, False, limit - len(event_results)) ) for e_id, in txn.fetchall(): diff --git a/synapse/storage/events.py b/synapse/storage/events.py index a2e87c27ce..9fe2effb4b 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -188,12 +188,12 @@ class EventsStore(SQLBaseStore): ) sql = ( - "UPDATE events SET outlier = 0" + "UPDATE events SET outlier = ?" " WHERE event_id = ?" ) txn.execute( sql, - (event.event_id,) + (False, event.event_id,) ) return diff --git a/synapse/storage/room.py b/synapse/storage/room.py index a1a76280fe..48ebb33057 100644 --- a/synapse/storage/room.py +++ b/synapse/storage/room.py @@ -105,14 +105,12 @@ class RoomStore(SQLBaseStore): # We use non printing ascii character US (\x1F) as a separator sql = ( - "SELECT r.room_id, n.name, t.topic, " - "group_concat(a.room_alias, '\x1F') " - "FROM rooms AS r " - "LEFT JOIN (%(topic)s) AS t ON t.room_id = r.room_id " - "LEFT JOIN (%(name)s) AS n ON n.room_id = r.room_id " - "INNER JOIN room_aliases AS a ON a.room_id = r.room_id " - "WHERE r.is_public = ? " - "GROUP BY r.room_id " + "SELECT r.room_id, max(n.name), max(t.topic)" + " FROM rooms AS r" + " LEFT JOIN (%(topic)s) AS t ON t.room_id = r.room_id" + " LEFT JOIN (%(name)s) AS n ON n.room_id = r.room_id" + " WHERE r.is_public = ?" + " GROUP BY r.room_id" ) % { "topic": topic_subquery, "name": name_subquery, @@ -120,7 +118,22 @@ class RoomStore(SQLBaseStore): txn.execute(sql, (is_public,)) - return txn.fetchall() + rows = txn.fetchall() + + for i, row in enumerate(rows): + room_id = row[0] + aliases = self._simple_select_onecol_txn( + txn, + table="room_aliases", + keyvalues={ + "room_id": room_id + }, + retcol="room_alias", + ) + + rows[i] = list(row) + [aliases] + + return rows rows = yield self.runInteraction( "get_rooms", f @@ -131,9 +144,10 @@ class RoomStore(SQLBaseStore): "room_id": r[0], "name": r[1], "topic": r[2], - "aliases": r[3].split("\x1F"), + "aliases": r[3], } for r in rows + if r[3] # We only return rooms that have at least one alias. ] defer.returnValue(ret) diff --git a/synapse/storage/schema/full_schemas/16/application_services.sql b/synapse/storage/schema/full_schemas/16/application_services.sql index bc709df92d..f08c5bcf76 100644 --- a/synapse/storage/schema/full_schemas/16/application_services.sql +++ b/synapse/storage/schema/full_schemas/16/application_services.sql @@ -14,7 +14,7 @@ */ CREATE TABLE IF NOT EXISTS application_services( - id BIGINT UNSIGNED PRIMARY KEY, + id BIGINT PRIMARY KEY, url VARCHAR(150), token VARCHAR(150), hs_token VARCHAR(150), @@ -23,8 +23,8 @@ CREATE TABLE IF NOT EXISTS application_services( ); CREATE TABLE IF NOT EXISTS application_services_regex( - id BIGINT UNSIGNED PRIMARY KEY, - as_id BIGINT UNSIGNED NOT NULL, + id BIGINT PRIMARY KEY, + as_id BIGINT NOT NULL, namespace INTEGER, /* enum[room_id|room_alias|user_id] */ regex VARCHAR(150), FOREIGN KEY(as_id) REFERENCES application_services(id) @@ -39,10 +39,10 @@ CREATE TABLE IF NOT EXISTS application_services_state( CREATE TABLE IF NOT EXISTS application_services_txns( as_id VARCHAR(150) NOT NULL, txn_id INTEGER NOT NULL, - event_ids LONGBLOB NOT NULL, + event_ids bytea NOT NULL, UNIQUE(as_id, txn_id) ); -CREATE INDEX IF NOT EXISTS application_services_txns_id ON application_services_txns ( +CREATE INDEX application_services_txns_id ON application_services_txns ( as_id ); diff --git a/synapse/storage/schema/full_schemas/16/event_edges.sql b/synapse/storage/schema/full_schemas/16/event_edges.sql index bdb1109094..05d0874f0d 100644 --- a/synapse/storage/schema/full_schemas/16/event_edges.sql +++ b/synapse/storage/schema/full_schemas/16/event_edges.sql @@ -19,8 +19,8 @@ CREATE TABLE IF NOT EXISTS event_forward_extremities( UNIQUE (event_id, room_id) ); -CREATE INDEX IF NOT EXISTS ev_extrem_room ON event_forward_extremities(room_id); -CREATE INDEX IF NOT EXISTS ev_extrem_id ON event_forward_extremities(event_id); +CREATE INDEX ev_extrem_room ON event_forward_extremities(room_id); +CREATE INDEX ev_extrem_id ON event_forward_extremities(event_id); CREATE TABLE IF NOT EXISTS event_backward_extremities( @@ -29,8 +29,8 @@ CREATE TABLE IF NOT EXISTS event_backward_extremities( UNIQUE (event_id, room_id) ); -CREATE INDEX IF NOT EXISTS ev_b_extrem_room ON event_backward_extremities(room_id); -CREATE INDEX IF NOT EXISTS ev_b_extrem_id ON event_backward_extremities(event_id); +CREATE INDEX ev_b_extrem_room ON event_backward_extremities(room_id); +CREATE INDEX ev_b_extrem_id ON event_backward_extremities(event_id); CREATE TABLE IF NOT EXISTS event_edges( @@ -41,8 +41,8 @@ CREATE TABLE IF NOT EXISTS event_edges( UNIQUE (event_id, prev_event_id, room_id, is_state) ); -CREATE INDEX IF NOT EXISTS ev_edges_id ON event_edges(event_id); -CREATE INDEX IF NOT EXISTS ev_edges_prev_id ON event_edges(prev_event_id); +CREATE INDEX ev_edges_id ON event_edges(event_id); +CREATE INDEX ev_edges_prev_id ON event_edges(prev_event_id); CREATE TABLE IF NOT EXISTS room_depth( @@ -51,17 +51,17 @@ CREATE TABLE IF NOT EXISTS room_depth( UNIQUE (room_id) ); -CREATE INDEX IF NOT EXISTS room_depth_room ON room_depth(room_id); +CREATE INDEX room_depth_room ON room_depth(room_id); create TABLE IF NOT EXISTS event_destinations( event_id VARCHAR(150) NOT NULL, destination VARCHAR(150) NOT NULL, - delivered_ts BIGINT UNSIGNED DEFAULT 0, -- or 0 if not delivered + delivered_ts BIGINT DEFAULT 0, -- or 0 if not delivered UNIQUE (event_id, destination) ); -CREATE INDEX IF NOT EXISTS event_destinations_id ON event_destinations(event_id); +CREATE INDEX event_destinations_id ON event_destinations(event_id); CREATE TABLE IF NOT EXISTS state_forward_extremities( @@ -72,10 +72,10 @@ CREATE TABLE IF NOT EXISTS state_forward_extremities( UNIQUE (event_id, room_id) ); -CREATE INDEX IF NOT EXISTS st_extrem_keys ON state_forward_extremities( +CREATE INDEX st_extrem_keys ON state_forward_extremities( room_id, type, state_key ); -CREATE INDEX IF NOT EXISTS st_extrem_id ON state_forward_extremities(event_id); +CREATE INDEX st_extrem_id ON state_forward_extremities(event_id); CREATE TABLE IF NOT EXISTS event_auth( @@ -85,5 +85,5 @@ CREATE TABLE IF NOT EXISTS event_auth( UNIQUE (event_id, auth_id, room_id) ); -CREATE INDEX IF NOT EXISTS evauth_edges_id ON event_auth(event_id); -CREATE INDEX IF NOT EXISTS evauth_edges_auth_id ON event_auth(auth_id); +CREATE INDEX evauth_edges_id ON event_auth(event_id); +CREATE INDEX evauth_edges_auth_id ON event_auth(auth_id); diff --git a/synapse/storage/schema/full_schemas/16/event_signatures.sql b/synapse/storage/schema/full_schemas/16/event_signatures.sql index 09886f607c..4291827368 100644 --- a/synapse/storage/schema/full_schemas/16/event_signatures.sql +++ b/synapse/storage/schema/full_schemas/16/event_signatures.sql @@ -16,40 +16,40 @@ CREATE TABLE IF NOT EXISTS event_content_hashes ( event_id VARCHAR(150), algorithm VARCHAR(150), - hash LONGBLOB, + hash bytea, UNIQUE (event_id, algorithm) ); -CREATE INDEX IF NOT EXISTS event_content_hashes_id ON event_content_hashes(event_id); +CREATE INDEX event_content_hashes_id ON event_content_hashes(event_id); CREATE TABLE IF NOT EXISTS event_reference_hashes ( event_id VARCHAR(150), algorithm VARCHAR(150), - hash LONGBLOB, + hash bytea, UNIQUE (event_id, algorithm) ); -CREATE INDEX IF NOT EXISTS event_reference_hashes_id ON event_reference_hashes(event_id); +CREATE INDEX event_reference_hashes_id ON event_reference_hashes(event_id); CREATE TABLE IF NOT EXISTS event_signatures ( event_id VARCHAR(150), signature_name VARCHAR(150), key_id VARCHAR(150), - signature LONGBLOB, + signature bytea, UNIQUE (event_id, signature_name, key_id) ); -CREATE INDEX IF NOT EXISTS event_signatures_id ON event_signatures(event_id); +CREATE INDEX event_signatures_id ON event_signatures(event_id); CREATE TABLE IF NOT EXISTS event_edge_hashes( event_id VARCHAR(150), prev_event_id VARCHAR(150), algorithm VARCHAR(150), - hash LONGBLOB, + hash bytea, UNIQUE (event_id, prev_event_id, algorithm) ); -CREATE INDEX IF NOT EXISTS event_edge_hashes_id ON event_edge_hashes(event_id); +CREATE INDEX event_edge_hashes_id ON event_edge_hashes(event_id); diff --git a/synapse/storage/schema/full_schemas/16/im.sql b/synapse/storage/schema/full_schemas/16/im.sql index 19f0f34143..a661fc160c 100644 --- a/synapse/storage/schema/full_schemas/16/im.sql +++ b/synapse/storage/schema/full_schemas/16/im.sql @@ -14,33 +14,33 @@ */ CREATE TABLE IF NOT EXISTS events( - stream_ordering BIGINT UNSIGNED PRIMARY KEY, - topological_ordering BIGINT UNSIGNED NOT NULL, + stream_ordering BIGINT PRIMARY KEY, + topological_ordering BIGINT NOT NULL, event_id VARCHAR(150) NOT NULL, type VARCHAR(150) NOT NULL, room_id VARCHAR(150) NOT NULL, - content LONGBLOB NOT NULL, - unrecognized_keys LONGBLOB, + content bytea NOT NULL, + unrecognized_keys bytea, processed BOOL NOT NULL, outlier BOOL NOT NULL, - depth BIGINT UNSIGNED DEFAULT 0 NOT NULL, + depth BIGINT DEFAULT 0 NOT NULL, UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS events_stream_ordering ON events (stream_ordering); -CREATE INDEX IF NOT EXISTS events_topological_ordering ON events (topological_ordering); -CREATE INDEX IF NOT EXISTS events_room_id ON events (room_id); +CREATE INDEX events_stream_ordering ON events (stream_ordering); +CREATE INDEX events_topological_ordering ON events (topological_ordering); +CREATE INDEX events_room_id ON events (room_id); CREATE TABLE IF NOT EXISTS event_json( event_id VARCHAR(150) NOT NULL, room_id VARCHAR(150) NOT NULL, - internal_metadata LONGBLOB NOT NULL, - json LONGBLOB NOT NULL, + internal_metadata bytea NOT NULL, + json bytea NOT NULL, UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS event_json_room_id ON event_json(room_id); +CREATE INDEX event_json_room_id ON event_json(room_id); CREATE TABLE IF NOT EXISTS state_events( @@ -52,9 +52,9 @@ CREATE TABLE IF NOT EXISTS state_events( UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS state_events_room_id ON state_events (room_id); -CREATE INDEX IF NOT EXISTS state_events_type ON state_events (type); -CREATE INDEX IF NOT EXISTS state_events_state_key ON state_events (state_key); +CREATE INDEX state_events_room_id ON state_events (room_id); +CREATE INDEX state_events_type ON state_events (type); +CREATE INDEX state_events_state_key ON state_events (state_key); CREATE TABLE IF NOT EXISTS current_state_events( @@ -66,9 +66,9 @@ CREATE TABLE IF NOT EXISTS current_state_events( UNIQUE (room_id, type, state_key) ); -CREATE INDEX IF NOT EXISTS current_state_events_room_id ON current_state_events (room_id); -CREATE INDEX IF NOT EXISTS current_state_events_type ON current_state_events (type); -CREATE INDEX IF NOT EXISTS current_state_events_state_key ON current_state_events (state_key); +CREATE INDEX current_state_events_room_id ON current_state_events (room_id); +CREATE INDEX current_state_events_type ON current_state_events (type); +CREATE INDEX current_state_events_state_key ON current_state_events (state_key); CREATE TABLE IF NOT EXISTS room_memberships( event_id VARCHAR(150) NOT NULL, @@ -79,8 +79,8 @@ CREATE TABLE IF NOT EXISTS room_memberships( UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS room_memberships_room_id ON room_memberships (room_id); -CREATE INDEX IF NOT EXISTS room_memberships_user_id ON room_memberships (user_id); +CREATE INDEX room_memberships_room_id ON room_memberships (room_id); +CREATE INDEX room_memberships_user_id ON room_memberships (user_id); CREATE TABLE IF NOT EXISTS feedback( event_id VARCHAR(150) NOT NULL, @@ -98,7 +98,7 @@ CREATE TABLE IF NOT EXISTS topics( UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS topics_room_id ON topics(room_id); +CREATE INDEX topics_room_id ON topics(room_id); CREATE TABLE IF NOT EXISTS room_names( event_id VARCHAR(150) NOT NULL, @@ -107,7 +107,7 @@ CREATE TABLE IF NOT EXISTS room_names( UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS room_names_room_id ON room_names(room_id); +CREATE INDEX room_names_room_id ON room_names(room_id); CREATE TABLE IF NOT EXISTS rooms( room_id VARCHAR(150) PRIMARY KEY NOT NULL, @@ -121,4 +121,4 @@ CREATE TABLE IF NOT EXISTS room_hosts( UNIQUE (room_id, host) ); -CREATE INDEX IF NOT EXISTS room_hosts_room_id ON room_hosts (room_id); +CREATE INDEX room_hosts_room_id ON room_hosts (room_id); diff --git a/synapse/storage/schema/full_schemas/16/keys.sql b/synapse/storage/schema/full_schemas/16/keys.sql index 35f141c288..459b510427 100644 --- a/synapse/storage/schema/full_schemas/16/keys.sql +++ b/synapse/storage/schema/full_schemas/16/keys.sql @@ -16,8 +16,8 @@ CREATE TABLE IF NOT EXISTS server_tls_certificates( server_name VARCHAR(150), -- Server name. fingerprint VARCHAR(150), -- Certificate fingerprint. from_server VARCHAR(150), -- Which key server the certificate was fetched from. - ts_added_ms BIGINT UNSIGNED, -- When the certifcate was added. - tls_certificate LONGBLOB, -- DER encoded x509 certificate. + ts_added_ms BIGINT, -- When the certifcate was added. + tls_certificate bytea, -- DER encoded x509 certificate. UNIQUE (server_name, fingerprint) ); @@ -25,7 +25,7 @@ CREATE TABLE IF NOT EXISTS server_signature_keys( server_name VARCHAR(150), -- Server name. key_id VARCHAR(150), -- Key version. from_server VARCHAR(150), -- Which key server the key was fetched form. - ts_added_ms BIGINT UNSIGNED, -- When the key was added. - verify_key LONGBLOB, -- NACL verification key. + ts_added_ms BIGINT, -- When the key was added. + verify_key bytea, -- NACL verification key. UNIQUE (server_name, key_id) ); diff --git a/synapse/storage/schema/full_schemas/16/media_repository.sql b/synapse/storage/schema/full_schemas/16/media_repository.sql index 014bce4aeb..0e819fca38 100644 --- a/synapse/storage/schema/full_schemas/16/media_repository.sql +++ b/synapse/storage/schema/full_schemas/16/media_repository.sql @@ -17,7 +17,7 @@ CREATE TABLE IF NOT EXISTS local_media_repository ( media_id VARCHAR(150), -- The id used to refer to the media. media_type VARCHAR(150), -- The MIME-type of the media. media_length INTEGER, -- Length of the media in bytes. - created_ts BIGINT UNSIGNED, -- When the content was uploaded in ms. + created_ts BIGINT, -- When the content was uploaded in ms. upload_name VARCHAR(150), -- The name the media was uploaded with. user_id VARCHAR(150), -- The user who uploaded the file. UNIQUE (media_id) @@ -35,14 +35,14 @@ CREATE TABLE IF NOT EXISTS local_media_repository_thumbnails ( ) ); -CREATE INDEX IF NOT EXISTS local_media_repository_thumbnails_media_id +CREATE INDEX local_media_repository_thumbnails_media_id ON local_media_repository_thumbnails (media_id); CREATE TABLE IF NOT EXISTS remote_media_cache ( media_origin VARCHAR(150), -- The remote HS the media came from. media_id VARCHAR(150), -- The id used to refer to the media on that server. media_type VARCHAR(150), -- The MIME-type of the media. - created_ts BIGINT UNSIGNED, -- When the content was uploaded in ms. + created_ts BIGINT, -- When the content was uploaded in ms. upload_name VARCHAR(150), -- The name the media was uploaded with. media_length INTEGER, -- Length of the media in bytes. filesystem_id VARCHAR(150), -- The name used to store the media on disk. @@ -64,5 +64,5 @@ CREATE TABLE IF NOT EXISTS remote_media_cache_thumbnails ( ) ); -CREATE INDEX IF NOT EXISTS remote_media_cache_thumbnails_media_id +CREATE INDEX remote_media_cache_thumbnails_media_id ON remote_media_cache_thumbnails (media_id); diff --git a/synapse/storage/schema/full_schemas/16/presence.sql b/synapse/storage/schema/full_schemas/16/presence.sql index fbe5b0af6c..9c41be296e 100644 --- a/synapse/storage/schema/full_schemas/16/presence.sql +++ b/synapse/storage/schema/full_schemas/16/presence.sql @@ -16,7 +16,7 @@ CREATE TABLE IF NOT EXISTS presence( user_id VARCHAR(150) NOT NULL, state VARCHAR(20), status_msg VARCHAR(150), - mtime BIGINT UNSIGNED, -- miliseconds since last state change + mtime BIGINT, -- miliseconds since last state change UNIQUE (user_id) ); @@ -37,4 +37,4 @@ CREATE TABLE IF NOT EXISTS presence_list( UNIQUE (user_id, observed_user_id) ); -CREATE INDEX IF NOT EXISTS presence_list_user_id ON presence_list (user_id); +CREATE INDEX presence_list_user_id ON presence_list (user_id); diff --git a/synapse/storage/schema/full_schemas/16/push.sql b/synapse/storage/schema/full_schemas/16/push.sql index 33300736f9..5c0c7bc201 100644 --- a/synapse/storage/schema/full_schemas/16/push.sql +++ b/synapse/storage/schema/full_schemas/16/push.sql @@ -22,52 +22,52 @@ CREATE TABLE IF NOT EXISTS rejections( -- Push notification endpoints that users have configured CREATE TABLE IF NOT EXISTS pushers ( - id BIGINT UNSIGNED PRIMARY KEY, + id BIGINT PRIMARY KEY, user_name VARCHAR(150) NOT NULL, profile_tag VARCHAR(32) NOT NULL, kind VARCHAR(8) NOT NULL, app_id VARCHAR(64) NOT NULL, app_display_name VARCHAR(64) NOT NULL, device_display_name VARCHAR(128) NOT NULL, - pushkey VARBINARY(512) NOT NULL, - ts BIGINT UNSIGNED NOT NULL, + pushkey bytea NOT NULL, + ts BIGINT NOT NULL, lang VARCHAR(8), - data LONGBLOB, + data bytea, last_token TEXT, - last_success BIGINT UNSIGNED, - failing_since BIGINT UNSIGNED, + last_success BIGINT, + failing_since BIGINT, UNIQUE (app_id, pushkey) ); CREATE TABLE IF NOT EXISTS push_rules ( - id BIGINT UNSIGNED PRIMARY KEY, + id BIGINT PRIMARY KEY, user_name VARCHAR(150) NOT NULL, rule_id VARCHAR(150) NOT NULL, - priority_class TINYINT NOT NULL, + priority_class SMALLINT NOT NULL, priority INTEGER NOT NULL DEFAULT 0, conditions VARCHAR(150) NOT NULL, actions VARCHAR(150) NOT NULL, UNIQUE(user_name, rule_id) ); -CREATE INDEX IF NOT EXISTS push_rules_user_name on push_rules (user_name); +CREATE INDEX push_rules_user_name on push_rules (user_name); CREATE TABLE IF NOT EXISTS user_filters( user_id VARCHAR(150), - filter_id BIGINT UNSIGNED, - filter_json LONGBLOB + filter_id BIGINT, + filter_json bytea ); -CREATE INDEX IF NOT EXISTS user_filters_by_user_id_filter_id ON user_filters( +CREATE INDEX user_filters_by_user_id_filter_id ON user_filters( user_id, filter_id ); CREATE TABLE IF NOT EXISTS push_rules_enable ( - id BIGINT UNSIGNED PRIMARY KEY, + id BIGINT PRIMARY KEY, user_name VARCHAR(150) NOT NULL, rule_id VARCHAR(150) NOT NULL, - enabled TINYINT, + enabled SMALLINT, UNIQUE(user_name, rule_id) ); -CREATE INDEX IF NOT EXISTS push_rules_enable_user_name on push_rules_enable (user_name); +CREATE INDEX push_rules_enable_user_name on push_rules_enable (user_name); diff --git a/synapse/storage/schema/full_schemas/16/redactions.sql b/synapse/storage/schema/full_schemas/16/redactions.sql index b81451eab4..492fd22033 100644 --- a/synapse/storage/schema/full_schemas/16/redactions.sql +++ b/synapse/storage/schema/full_schemas/16/redactions.sql @@ -18,5 +18,5 @@ CREATE TABLE IF NOT EXISTS redactions ( UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS redactions_event_id ON redactions (event_id); -CREATE INDEX IF NOT EXISTS redactions_redacts ON redactions (redacts); +CREATE INDEX redactions_event_id ON redactions (event_id); +CREATE INDEX redactions_redacts ON redactions (redacts); diff --git a/synapse/storage/schema/full_schemas/16/state.sql b/synapse/storage/schema/full_schemas/16/state.sql index 8c51610396..3c54595e64 100644 --- a/synapse/storage/schema/full_schemas/16/state.sql +++ b/synapse/storage/schema/full_schemas/16/state.sql @@ -14,7 +14,7 @@ */ CREATE TABLE IF NOT EXISTS state_groups( - id BIGINT UNSIGNED PRIMARY KEY, + id BIGINT PRIMARY KEY, room_id VARCHAR(150) NOT NULL, event_id VARCHAR(150) NOT NULL ); @@ -33,8 +33,8 @@ CREATE TABLE IF NOT EXISTS event_to_state_groups( UNIQUE (event_id) ); -CREATE INDEX IF NOT EXISTS state_groups_id ON state_groups(id); +CREATE INDEX state_groups_id ON state_groups(id); -CREATE INDEX IF NOT EXISTS state_groups_state_id ON state_groups_state(state_group); -CREATE INDEX IF NOT EXISTS state_groups_state_tuple ON state_groups_state(room_id, type, state_key); -CREATE INDEX IF NOT EXISTS event_to_state_groups_id ON event_to_state_groups(event_id); +CREATE INDEX state_groups_state_id ON state_groups_state(state_group); +CREATE INDEX state_groups_state_tuple ON state_groups_state(room_id, type, state_key); +CREATE INDEX event_to_state_groups_id ON event_to_state_groups(event_id); diff --git a/synapse/storage/schema/full_schemas/16/transactions.sql b/synapse/storage/schema/full_schemas/16/transactions.sql index f381e67603..bc64064936 100644 --- a/synapse/storage/schema/full_schemas/16/transactions.sql +++ b/synapse/storage/schema/full_schemas/16/transactions.sql @@ -16,32 +16,32 @@ CREATE TABLE IF NOT EXISTS received_transactions( transaction_id VARCHAR(150), origin VARCHAR(150), - ts BIGINT UNSIGNED, + ts BIGINT, response_code INTEGER, - response_json LONGBLOB, - has_been_referenced BOOL default 0, -- Whether thishas been referenced by a prev_tx + response_json bytea, + has_been_referenced smallint default 0, -- Whether thishas been referenced by a prev_tx UNIQUE (transaction_id, origin) ); -CREATE INDEX IF NOT EXISTS transactions_have_ref ON received_transactions(origin, has_been_referenced);-- WHERE has_been_referenced = 0; +CREATE INDEX transactions_have_ref ON received_transactions(origin, has_been_referenced);-- WHERE has_been_referenced = 0; -- Stores what transactions we've sent, what their response was (if we got one) and whether we have -- since referenced the transaction in another outgoing transaction CREATE TABLE IF NOT EXISTS sent_transactions( - id BIGINT UNSIGNED PRIMARY KEY, -- This is used to apply insertion ordering + id BIGINT PRIMARY KEY, -- This is used to apply insertion ordering transaction_id VARCHAR(150), destination VARCHAR(150), response_code INTEGER DEFAULT 0, - response_json LONGBLOB, - ts BIGINT UNSIGNED + response_json bytea, + ts BIGINT ); -CREATE INDEX IF NOT EXISTS sent_transaction_dest ON sent_transactions(destination); -CREATE INDEX IF NOT EXISTS sent_transaction_txn_id ON sent_transactions(transaction_id); +CREATE INDEX sent_transaction_dest ON sent_transactions(destination); +CREATE INDEX sent_transaction_txn_id ON sent_transactions(transaction_id); -- So that we can do an efficient look up of all transactions that have yet to be successfully -- sent. -CREATE INDEX IF NOT EXISTS sent_transaction_sent ON sent_transactions(response_code); +CREATE INDEX sent_transaction_sent ON sent_transactions(response_code); -- For sent transactions only. @@ -53,11 +53,11 @@ CREATE TABLE IF NOT EXISTS transaction_id_to_pdu( UNIQUE (transaction_id, destination) ); -CREATE INDEX IF NOT EXISTS transaction_id_to_pdu_dest ON transaction_id_to_pdu(destination); +CREATE INDEX transaction_id_to_pdu_dest ON transaction_id_to_pdu(destination); -- To track destination health CREATE TABLE IF NOT EXISTS destinations( destination VARCHAR(150) PRIMARY KEY, - retry_last_ts BIGINT UNSIGNED, + retry_last_ts BIGINT, retry_interval INTEGER ); diff --git a/synapse/storage/schema/full_schemas/16/users.sql b/synapse/storage/schema/full_schemas/16/users.sql index d0011c04b4..267284d07d 100644 --- a/synapse/storage/schema/full_schemas/16/users.sql +++ b/synapse/storage/schema/full_schemas/16/users.sql @@ -15,17 +15,17 @@ CREATE TABLE IF NOT EXISTS users( name VARCHAR(150), password_hash VARCHAR(150), - creation_ts BIGINT UNSIGNED, - admin BOOL DEFAULT 0 NOT NULL, + creation_ts BIGINT, + admin SMALLINT DEFAULT 0 NOT NULL, UNIQUE(name) ); CREATE TABLE IF NOT EXISTS access_tokens( - id BIGINT UNSIGNED PRIMARY KEY, + id BIGINT PRIMARY KEY, user_id VARCHAR(150) NOT NULL, device_id VARCHAR(150), token VARCHAR(150) NOT NULL, - last_used BIGINT UNSIGNED, + last_used BIGINT, UNIQUE(token) ); @@ -35,7 +35,7 @@ CREATE TABLE IF NOT EXISTS user_ips ( device_id VARCHAR(150), ip VARCHAR(150) NOT NULL, user_agent VARCHAR(150) NOT NULL, - last_seen BIGINT UNSIGNED NOT NULL + last_seen BIGINT NOT NULL ); CREATE INDEX IF NOT EXISTS user_ips_user ON user_ips(user); diff --git a/synapse/storage/schema/schema_version.sql b/synapse/storage/schema/schema_version.sql index e7fa6fe569..d9494611e0 100644 --- a/synapse/storage/schema/schema_version.sql +++ b/synapse/storage/schema/schema_version.sql @@ -14,16 +14,14 @@ */ CREATE TABLE IF NOT EXISTS schema_version( - `Lock` CHAR(1) NOT NULL DEFAULT 'X' UNIQUE, -- Makes sure this table only has one row. - `version` INTEGER NOT NULL, - `upgraded` BOOL NOT NULL, -- Whether we reached this version from an upgrade or an initial schema. - CHECK (`Lock`='X') + Lock CHAR(1) NOT NULL DEFAULT 'X' UNIQUE, -- Makes sure this table only has one row. + version INTEGER NOT NULL, + upgraded BOOL NOT NULL, -- Whether we reached this version from an upgrade or an initial schema. + CHECK (Lock='X') ); CREATE TABLE IF NOT EXISTS applied_schema_deltas( - `version` INTEGER NOT NULL, - `file` VARCHAR(150) NOT NULL, + version INTEGER NOT NULL, + file VARCHAR(150) NOT NULL, UNIQUE(version, file) ); - -CREATE INDEX IF NOT EXISTS schema_deltas_ver ON applied_schema_deltas(version); diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py index 57c2e4dfeb..df6de7cbcd 100644 --- a/synapse/storage/stream.py +++ b/synapse/storage/stream.py @@ -240,7 +240,7 @@ class StreamStore(SQLBaseStore): sql = ( "SELECT e.event_id, e.stream_ordering FROM events AS e WHERE " - "(e.outlier = 0 AND (room_id IN (%(current)s)) OR " + "(e.outlier = ? AND (room_id IN (%(current)s)) OR " "(event_id IN (%(invites)s))) " "AND e.stream_ordering > ? AND e.stream_ordering <= ? " "ORDER BY stream_ordering ASC LIMIT %(limit)d " @@ -251,7 +251,7 @@ class StreamStore(SQLBaseStore): } def f(txn): - txn.execute(sql, (user_id, user_id, from_id.stream, to_id.stream,)) + txn.execute(sql, (False, user_id, user_id, from_id.stream, to_id.stream,)) rows = self.cursor_to_dict(txn) @@ -283,7 +283,7 @@ class StreamStore(SQLBaseStore): # Tokens really represent positions between elements, but we use # the convention of pointing to the event before the gap. Hence # we have a bit of asymmetry when it comes to equalities. - args = [room_id] + args = [False, room_id] if direction == 'b': order = "DESC" bounds = _StreamToken.parse(from_key).upper_bound() @@ -307,7 +307,7 @@ class StreamStore(SQLBaseStore): sql = ( "SELECT * FROM events" - " WHERE outlier = 0 AND room_id = ? AND %(bounds)s" + " WHERE outlier = ? AND room_id = ? AND %(bounds)s" " ORDER BY topological_ordering %(order)s," " stream_ordering %(order)s %(limit)s" ) % { @@ -358,7 +358,7 @@ class StreamStore(SQLBaseStore): sql = ( "SELECT stream_ordering, topological_ordering, event_id" " FROM events" - " WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0" + " WHERE room_id = ? AND stream_ordering <= ? AND outlier = ?" " ORDER BY topological_ordering DESC, stream_ordering DESC" " LIMIT ?" ) @@ -368,17 +368,17 @@ class StreamStore(SQLBaseStore): "SELECT stream_ordering, topological_ordering, event_id" " FROM events" " WHERE room_id = ? AND stream_ordering > ?" - " AND stream_ordering <= ? AND outlier = 0" + " AND stream_ordering <= ? AND outlier = ?" " ORDER BY topological_ordering DESC, stream_ordering DESC" " LIMIT ?" ) def get_recent_events_for_room_txn(txn): if from_token is None: - txn.execute(sql, (room_id, end_token.stream, limit,)) + txn.execute(sql, (room_id, end_token.stream, False, limit,)) else: txn.execute(sql, ( - room_id, from_token.stream, end_token.stream, limit + room_id, from_token.stream, end_token.stream, False, limit )) rows = self.cursor_to_dict(txn) -- cgit 1.4.1