From caddadfc5ac61d1c91fbaf29bf3298f90a140560 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 10 Jan 2017 15:04:57 +0000 Subject: Change device_inbox stream index to include user This makes fetching the nost recently changed users much tricker, and brings it in line with e.g. presence_stream indices. --- synapse/storage/deviceinbox.py | 38 ++++++++++++++++++++++-- synapse/storage/prepare_database.py | 2 +- synapse/storage/schema/delta/40/device_inbox.sql | 20 +++++++++++++ 3 files changed, 57 insertions(+), 3 deletions(-) create mode 100644 synapse/storage/schema/delta/40/device_inbox.sql diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py index 2821eb89c9..b71ac3ae39 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -18,13 +18,29 @@ import ujson from twisted.internet import defer -from ._base import SQLBaseStore +from .background_updates import BackgroundUpdateStore logger = logging.getLogger(__name__) -class DeviceInboxStore(SQLBaseStore): +class DeviceInboxStore(BackgroundUpdateStore): + DEVICE_INBOX_STREAM_ID = "device_inbox_stream_drop" + + def __init__(self, hs): + super(DeviceInboxStore, self).__init__(hs) + + self.register_background_index_update( + "device_inbox_stream_index", + index_name="device_inbox_stream_id_user_id", + table="device_inbox", + columns=["stream_id", "user_id"], + ) + + self.register_background_update_handler( + self.DEVICE_INBOX_STREAM_ID, + self._background_drop_index_device_inbox, + ) @defer.inlineCallbacks def add_messages_to_device_inbox(self, local_messages_by_user_then_device, @@ -368,3 +384,21 @@ class DeviceInboxStore(SQLBaseStore): "delete_device_msgs_for_remote", delete_messages_for_remote_destination_txn ) + + @defer.inlineCallbacks + def _background_drop_index_device_inbox(self, progress, batch_size): + def reindex_txn(conn): + conn.set_session(autocommit=True) + try: + txn = conn.cursor() + txn.execute( + "DROP INDEX IF EXISTS device_inbox_stream_id" + ) + finally: + conn.set_session(autocommit=False) + + yield self.runWithConnection(reindex_txn) + + yield self._end_background_update(self.DEVICE_INBOX_STREAM_ID) + + defer.returnValue(1) diff --git a/synapse/storage/prepare_database.py b/synapse/storage/prepare_database.py index e46ae6502e..b357f22be7 100644 --- a/synapse/storage/prepare_database.py +++ b/synapse/storage/prepare_database.py @@ -25,7 +25,7 @@ logger = logging.getLogger(__name__) # Remember to update this number every time a change is made to database # schema files, so the users will be informed on server restarts. -SCHEMA_VERSION = 39 +SCHEMA_VERSION = 40 dir_path = os.path.abspath(os.path.dirname(__file__)) diff --git a/synapse/storage/schema/delta/40/device_inbox.sql b/synapse/storage/schema/delta/40/device_inbox.sql new file mode 100644 index 0000000000..ce58fe2082 --- /dev/null +++ b/synapse/storage/schema/delta/40/device_inbox.sql @@ -0,0 +1,20 @@ +/* Copyright 2016 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. + */ + +INSERT into background_updates (update_name, progress_json) + VALUES ('device_inbox_stream_index', '{}'); + +INSERT into background_updates (update_name, progress_json, depends_on) + VALUES ('device_inbox_stream_drop', '{}', 'device_inbox_stream_index'); -- cgit 1.4.1 From 5a32e9273ec9759caf09d5b8204dd29e7a007b97 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 10 Jan 2017 15:11:27 +0000 Subject: Don't disable autocommit --- synapse/storage/deviceinbox.py | 12 ++++-------- 1 file changed, 4 insertions(+), 8 deletions(-) diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py index b71ac3ae39..b0ab70bafe 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -388,14 +388,10 @@ class DeviceInboxStore(BackgroundUpdateStore): @defer.inlineCallbacks def _background_drop_index_device_inbox(self, progress, batch_size): def reindex_txn(conn): - conn.set_session(autocommit=True) - try: - txn = conn.cursor() - txn.execute( - "DROP INDEX IF EXISTS device_inbox_stream_id" - ) - finally: - conn.set_session(autocommit=False) + txn = conn.cursor() + txn.execute( + "DROP INDEX IF EXISTS device_inbox_stream_id" + ) yield self.runWithConnection(reindex_txn) -- cgit 1.4.1 From ab655dca339f8d4168079cc2b4529dc50265fc83 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 10 Jan 2017 15:15:25 +0000 Subject: Explicitly close the cursor --- synapse/storage/deviceinbox.py | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/storage/deviceinbox.py b/synapse/storage/deviceinbox.py index b0ab70bafe..bde3b5cbbc 100644 --- a/synapse/storage/deviceinbox.py +++ b/synapse/storage/deviceinbox.py @@ -392,6 +392,7 @@ class DeviceInboxStore(BackgroundUpdateStore): txn.execute( "DROP INDEX IF EXISTS device_inbox_stream_id" ) + txn.close() yield self.runWithConnection(reindex_txn) -- cgit 1.4.1 From c050f493dd53a74206338f9a5e567d7bd24fbd5d Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 13 Jan 2017 15:14:41 +0000 Subject: Add comment --- synapse/storage/schema/delta/40/device_inbox.sql | 1 + 1 file changed, 1 insertion(+) diff --git a/synapse/storage/schema/delta/40/device_inbox.sql b/synapse/storage/schema/delta/40/device_inbox.sql index ce58fe2082..b9fe1f0480 100644 --- a/synapse/storage/schema/delta/40/device_inbox.sql +++ b/synapse/storage/schema/delta/40/device_inbox.sql @@ -13,6 +13,7 @@ * limitations under the License. */ +-- turn the pre-fill startup query into a index-only scan on postgresql. INSERT into background_updates (update_name, progress_json) VALUES ('device_inbox_stream_index', '{}'); -- cgit 1.4.1