diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py
index d6ec446bd2..a08c74fac1 100644
--- a/synapse/storage/__init__.py
+++ b/synapse/storage/__init__.py
@@ -45,8 +45,10 @@ from syutil.jsonutil import encode_canonical_json
from synapse.crypto.event_signing import compute_event_reference_hash
+import imp
import logging
import os
+import sqlite3
logger = logging.getLogger(__name__)
@@ -610,49 +612,115 @@ class UpgradeDatabaseException(PrepareDatabaseException):
def prepare_database(db_conn):
- """ Set up all the dbs. Since all the *.sql have IF NOT EXISTS, so we
- don't have to worry about overwriting existing content.
+ """Prepares a database for usage. Will either create all necessary tables
+ or upgrade from an older schema version.
"""
- c = db_conn.cursor()
- c.execute("PRAGMA user_version")
- row = c.fetchone()
+ cur = db_conn.cursor()
+ version_info = get_schema_state(cur)
- if row and row[0]:
- user_version = row[0]
+ if version_info:
+ user_version, delta_files = version_info
+ _upgrade_existing_database(cur, user_version, delta_files)
+ else:
+ _setup_new_database(cur)
+
+ cur.execute("PRAGMA user_version = %d" % (SCHEMA_VERSION,))
+ db_conn.commit()
+
+ cur.close()
+
+
+def _setup_new_database(cur):
+ sql_script = "BEGIN TRANSACTION;\n"
+ for sql_loc in SCHEMAS:
+ logger.debug("Applying schema %r", sql_loc)
+ sql_script += read_schema(sql_loc)
+ sql_script += "\n"
+ sql_script += "COMMIT TRANSACTION;"
+ cur.executescript(sql_script)
+
+
+def _upgrade_existing_database(cur, user_version, delta_files):
+ """Upgrades an existing database.
+
+ Delta files can either be SQL stored in *.sql files, or python modules
+ in *.py.
+
+ There can be multiple delta files per version. Synapse will keep track of
+ which delta files have been applied, and will apply any that haven't been
+ even if there has been no version bump. This is useful for development
+ where orthogonal schema changes may happen on separate branches.
+ """
+
+ if user_version > SCHEMA_VERSION:
+ raise ValueError(
+ "Cannot use this database as it is too " +
+ "new for the server to understand"
+ )
- if user_version > SCHEMA_VERSION:
- raise ValueError(
- "Cannot use this database as it is too " +
- "new for the server to understand"
+ for v in range(user_version, SCHEMA_VERSION + 1):
+ delta_dir = os.path.join(dir_path, "schema", "delta", v)
+ directory_entries = os.listdir(delta_dir)
+
+ for file_name in directory_entries:
+ relative_path = os.path.join(v, file_name)
+ if relative_path in delta_files:
+ continue
+
+ absolute_path = os.path.join(
+ dir_path, "schema", "delta", relative_path,
)
- elif user_version < SCHEMA_VERSION:
- logger.info(
- "Upgrading database from version %d",
- user_version
+ root_name, ext = os.path.splitext(file_name)
+ if ext == ".py":
+ module_name = "synapse.storage.schema.v%d_%s" % (
+ v, root_name
+ )
+ with open(absolute_path) as schema_file:
+ module = imp.load_source(
+ module_name, absolute_path, schema_file
+ )
+ module.run_upgrade(cur)
+ elif ext == ".sql":
+ with open(absolute_path) as schema_file:
+ delta_schema = schema_file.read()
+ cur.executescript(delta_schema)
+ else:
+ # Not a valid delta file.
+ logger.warn(
+ "Found directory entry that did not end in .py or"
+ " .sql: %s",
+ relative_path,
+ )
+ continue
+
+ # Mark as done.
+ cur.execute(
+ "INSERT INTO schema_version (version, file)"
+ " VALUES (?,?)",
+ (v, relative_path)
)
- # Run every version since after the current version.
- for v in range(user_version + 1, SCHEMA_VERSION + 1):
- if v in (10, 14,):
- raise UpgradeDatabaseException(
- "No delta for version 10"
- )
- sql_script = read_schema("delta/v%d" % (v,))
- c.executescript(sql_script)
- db_conn.commit()
- else:
- logger.info("Database is at version %r", user_version)
+def get_schema_state(txn):
+ sql = (
+ "SELECT MAX(version), file FROM schema_version"
+ " WHERE version = (SELECT MAX(version) FROM schema_version)"
+ )
- else:
- sql_script = "BEGIN TRANSACTION;\n"
- for sql_loc in SCHEMAS:
- logger.debug("Applying schema %r", sql_loc)
- sql_script += read_schema(sql_loc)
- sql_script += "\n"
- sql_script += "COMMIT TRANSACTION;"
- c.executescript(sql_script)
- db_conn.commit()
- c.execute("PRAGMA user_version = %d" % SCHEMA_VERSION)
-
- c.close()
+ try:
+ txn.execute(sql)
+ res = txn.fetchall()
+
+ if res:
+ current_verison = max(r[0] for r in res)
+ applied_delta = [r[1] for r in res]
+
+ return current_verison, applied_delta
+ except sqlite3.OperationalError:
+ txn.execute("PRAGMA user_version")
+ row = txn.fetchone()
+ if row and row[0]:
+ # FIXME: We need to create schema_version table!
+ return row[0], []
+
+ return None
|
