Drop unused columns from event_edges

changelog.d/13217.misc

@@ -0,0 +1 @@
+Simplify the database schema for `event_edges`.

synapse/storage/databases/main/events_bg_updates.py

@@ -66,6 +66,7 @@ class _BackgroundUpdates:
 
     EVENT_EDGES_DROP_INVALID_ROWS = "event_edges_drop_invalid_rows"
     EVENT_EDGES_REPLACE_INDEX = "event_edges_replace_index"
+    EVENT_EDGES_DROP_OLD_COLS = "event_edges_drop_old_cols"
 
 
 @attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -237,6 +238,8 @@ def __init__(
         )
 
         ################################################################################
+        #
+        # event_edges cleanups
 
         self.db_pool.updates.register_background_update_handler(
             _BackgroundUpdates.EVENT_EDGES_DROP_INVALID_ROWS,
@@ -253,6 +256,11 @@ def __init__(
             replaces_index="ev_edges_id",
         )
 
+        self.db_pool.updates.register_background_update_handler(
+            _BackgroundUpdates.EVENT_EDGES_DROP_OLD_COLS,
+            self._background_event_edges_drop_old_cols,
+        )
+
     async def _background_reindex_fields_sender(
         self, progress: JsonDict, batch_size: int
     ) -> int:
@@ -1399,3 +1407,30 @@ def drop_invalid_event_edges_txn(txn: LoggingTransaction) -> bool:
             )
 
         return batch_size
+
+    async def _background_event_edges_drop_old_cols(
+        self, progress: JsonDict, batch_size: int
+    ) -> int:
+        """Drop unused columns from event_edges
+
+        This only runs for postgres. For SQLite, it all happens synchronously.
+        """
+
+        def _drop_txn(txn: LoggingTransaction):
+            txn.execute(
+                """
+                ALTER TABLE event_edges
+                    DROP COLUMN room_id,
+                    DROP COLUMN is_state
+                """
+            )
+
+        await self.db_pool.runInteraction(
+            desc="drop_invalid_event_edges", func=_drop_txn
+        )
+
+        await self.db_pool.updates._end_background_update(
+            _BackgroundUpdates.EVENT_EDGES_DROP_OLD_COLS
+        )
+
+        return 1

synapse/storage/schema/__init__.py

@@ -73,14 +73,14 @@
     - Tables related to groups are no longer accessed.
 
 Changes in SCHEMA_VERSION = 72:
-    - event_edges.(room_id, is_state) are no longer written to.
+    - event_edges.(room_id, is_state) are removed.
 """
 
 
 SCHEMA_COMPAT_VERSION = (
-    # We no longer maintain `event_edges.room_id`, so synapses with SCHEMA_VERSION < 71
+    # 'event_edges.room_id` no longer exists, so synapses with SCHEMA_VERSION < 72
     # will break.
-    71
+    72
 )
 """Limit on how far the synapse codebase can be rolled back without breaking db compat
 

synapse/storage/schema/main/delta/72/03drop_event_edges_cols.py

@@ -0,0 +1,66 @@
+# Copyright 2022 The Matrix.org Foundation C.I.C.
+#
+# 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 synapse.storage.engines import BaseDatabaseEngine, PostgresEngine, Sqlite3Engine
+from synapse.storage.types import Cursor
+
+# event_edges.room_id and event_edges.is_state are no longer used, so we can drop them.
+
+
+def run_create(cur: Cursor, database_engine: BaseDatabaseEngine, *args, **kwargs):
+    if isinstance(database_engine, PostgresEngine):
+        _update_postgres(cur)
+    elif isinstance(database_engine, Sqlite3Engine):
+        _update_sqlite(cur)
+    else:
+        raise NotImplementedError("Unknown database engine")
+
+
+def _update_postgres(cur: Cursor):
+    # for postgres, we have to wait for the background update which drops bad rows to
+    # complete before we can actually drop the bad columns.
+    cur.execute(
+        """
+        INSERT INTO background_updates (ordering, update_name, progress_json, depends_on)
+            VALUES
+            (7203, 'event_edges_drop_old_cols', '{}', 'event_edges_drop_invalid_rows')
+        """
+    )
+
+
+def _update_sqlite(cur: Cursor):
+    # sqlite is easier in one way, in that there was no background update to worry
+    # about. However, there is no ALTER TABLE DROP COLUMN (at least until 3.33), so it'a
+    # rather more fiddly.
+    #
+    # However, we recently (in 71/01rebuild_event_edges.sql.sqlite) rebuilt event_edges,
+    # so we know that room_id and is_state are the last columns in the schema, which
+    # means that we can just tell sqlite to change the schema without needing to change
+    # the data.
+    sql = (
+        'CREATE TABLE "event_edges" (event_id TEXT NOT NULL, '
+        "prev_event_id TEXT NOT NULL, "
+        "FOREIGN KEY(event_id) REFERENCES events(event_id))"
+    )
+
+    cur.execute("PRAGMA schema_version")
+    (oldver,) = cur.fetchone()
+    cur.execute("PRAGMA writable_schema=ON")
+    cur.execute(
+        "UPDATE sqlite_master SET sql=? WHERE tbl_name='event_edges' AND type='table'",
+        (sql,),
+    )
+    cur.execute("PRAGMA schema_version=%i" % (oldver + 1,))
+    cur.execute("PRAGMA writable_schema=OFF")