From 0a9fc81649c642c6c7d21af795054ca111bdde2b Mon Sep 17 00:00:00 2001 From: Tim Jenness Date: Wed, 15 Jul 2020 16:29:10 -0700 Subject: [PATCH 1/2] WIP: Hack up basic MySQL interface Requires pymysql, MariaDB, and testing.mysqld. The `replace` tests all fail. There are some major adjustments of column sizes in butler_attributes to make it fit. --- python/lsst/daf/butler/configs/registry.yaml | 1 + python/lsst/daf/butler/registry/attributes.py | 4 +- .../daf/butler/registry/databases/mysql.py | 119 ++++++++++++ tests/test_mysql.py | 174 ++++++++++++++++++ 4 files changed, 296 insertions(+), 2 deletions(-) create mode 100644 python/lsst/daf/butler/registry/databases/mysql.py create mode 100644 tests/test_mysql.py diff --git a/python/lsst/daf/butler/configs/registry.yaml b/python/lsst/daf/butler/configs/registry.yaml index fee5a7b0ab..304a5e9dbc 100644 --- a/python/lsst/daf/butler/configs/registry.yaml +++ b/python/lsst/daf/butler/configs/registry.yaml @@ -4,6 +4,7 @@ registry: engines: sqlite: lsst.daf.butler.registry.databases.sqlite.SqliteDatabase postgresql: lsst.daf.butler.registry.databases.postgresql.PostgresqlDatabase + mysql: lsst.daf.butler.registry.databases.mysql.MySqlDatabase managers: attributes: lsst.daf.butler.registry.attributes.DefaultButlerAttributeManager opaque: lsst.daf.butler.registry.opaque.ByNameOpaqueTableStorageManager diff --git a/python/lsst/daf/butler/registry/attributes.py b/python/lsst/daf/butler/registry/attributes.py index 230dd83f20..26c9bf6c4d 100644 --- a/python/lsst/daf/butler/registry/attributes.py +++ b/python/lsst/daf/butler/registry/attributes.py @@ -65,8 +65,8 @@ def __init__(self, db: Database, table: sqlalchemy.schema.Table): _TABLE_SPEC: ClassVar[TableSpec] = TableSpec( fields=[ - FieldSpec("name", dtype=sqlalchemy.String, length=1024, primaryKey=True), - FieldSpec("value", dtype=sqlalchemy.String, length=65535, nullable=False), + FieldSpec("name", dtype=sqlalchemy.String, length=512, primaryKey=True), + FieldSpec("value", dtype=sqlalchemy.String, length=12000, nullable=False), ], ) diff --git a/python/lsst/daf/butler/registry/databases/mysql.py b/python/lsst/daf/butler/registry/databases/mysql.py new file mode 100644 index 0000000000..5028275091 --- /dev/null +++ b/python/lsst/daf/butler/registry/databases/mysql.py @@ -0,0 +1,119 @@ +# This file is part of daf_butler. +# +# Developed for the LSST Data Management System. +# This product includes software developed by the LSST Project +# (http://www.lsst.org). +# See the COPYRIGHT file at the top-level directory of this distribution +# for details of code ownership. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . +from __future__ import annotations + +__all__ = ["MySqlDatabase"] + +from contextlib import closing, contextmanager +from typing import Iterator, Optional + +import sqlalchemy + +from ..interfaces import Database, ReadOnlyDatabaseError +from ..nameShrinker import NameShrinker + + +class MySqlDatabase(Database): + """An implementation of the `Database` interface for PostgreSQL. + + Parameters + ---------- + connection : `sqlalchemy.engine.Connection` + An existing connection created by a previous call to `connect`. + origin : `int` + An integer ID that should be used as the default for any datasets, + quanta, or other entities that use a (autoincrement, origin) compound + primary key. + namespace : `str`, optional + The namespace (schema) this database is associated with. If `None`, + the default schema for the connection is used (which may be `None`). + writeable : `bool`, optional + If `True`, allow write operations on the database, including + ``CREATE TABLE``. + + Notes + ----- + This currently requires the mysqlclient driver to be used as the backend + for SQLAlchemy. + """ + + def __init__( + self, + *, + connection: sqlalchemy.engine.Connection, + origin: int, + namespace: Optional[str] = None, + writeable: bool = True, + ): + super().__init__(origin=origin, connection=connection, namespace=namespace) + # Relationship of the database specified in the connection and + # the supplied namespace is uncertain. + if namespace is not None: + # This will warn if the schema/database already exists + connection.execute(f"CREATE SCHEMA IF NOT EXISTS {namespace};") + connection.execute(f"USE {namespace};") + self.namespace = namespace + self.dbname = "No idea" + self._writeable = writeable + self._shrinker = NameShrinker(64) # connection.engine.dialect.max_identifier_length is wrong + + @classmethod + def connect(cls, uri: str, *, writeable: bool = True) -> sqlalchemy.engine.Connection: + return sqlalchemy.engine.create_engine(uri, poolclass=sqlalchemy.pool.NullPool).connect() + + @classmethod + def fromConnection( + cls, + connection: sqlalchemy.engine.Connection, + *, + origin: int, + namespace: Optional[str] = None, + writeable: bool = True, + ) -> Database: + return cls(connection=connection, origin=origin, namespace=namespace, writeable=writeable) + + @contextmanager + def transaction(self, *, interrupting: bool = False) -> Iterator[None]: + with super().transaction(interrupting=interrupting): + if not self.isWriteable(): + with closing(self._connection.connection.cursor()) as cursor: + cursor.execute("SET TRANSACTION READ ONLY") + yield + + def isWriteable(self) -> bool: + return self._writeable + + def __str__(self) -> str: + return f"MySQL@{self.dbname}:{self.namespace}" + + def shrinkDatabaseEntityName(self, original: str) -> str: + return self._shrinker.shrink(original) + + def expandDatabaseEntityName(self, shrunk: str) -> str: + return self._shrinker.expand(shrunk) + + def replace(self, table: sqlalchemy.schema.Table, *rows: dict) -> None: + # This is all wrong + if not self.isWriteable(): + raise ReadOnlyDatabaseError(f"Attempt to replace into read-only database '{self}'.") + if not rows: + return + raise NotImplementedError("No support for replace in MySQL") diff --git a/tests/test_mysql.py b/tests/test_mysql.py new file mode 100644 index 0000000000..6019b85d76 --- /dev/null +++ b/tests/test_mysql.py @@ -0,0 +1,174 @@ +# This file is part of daf_butler. +# +# Developed for the LSST Data Management System. +# This product includes software developed by the LSST Project +# (http://www.lsst.org). +# See the COPYRIGHT file at the top-level directory of this distribution +# for details of code ownership. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +import gc +import os +import secrets +import unittest +from contextlib import contextmanager + +try: + # It's possible but silly to have testing.mysqld installed without + # having the postgresql server installed (because then nothing in + # testing.mysqld would work), so we use the presence of that module + # to test whether we can expect the server to be available. + import testing.mysqld +except ImportError: + testing = None + +import sqlalchemy +from lsst.daf.butler import ddl +from lsst.daf.butler.registry import Registry +from lsst.daf.butler.registry.databases.mysql import MySqlDatabase +from lsst.daf.butler.registry.tests import DatabaseTests, RegistryTests + + +@unittest.skipUnless(testing is not None, "testing.mysqld module not found") +class MySqlDatabaseTestCase(unittest.TestCase, DatabaseTests): + @classmethod + def setUpClass(cls): + # MariaDB requires the anonymous user + cls.server = testing.mysqld.Mysqld(user="", my_cnf={"skip-networking": None}) + + @classmethod + def tearDownClass(cls): + # Clean up any lingering SQLAlchemy engines/connections + # so they're closed before we shut down the server. + gc.collect() + cls.server.stop() + + def makeEmptyDatabase(self, origin: int = 0) -> MySqlDatabase: + namespace = f"namespace_{secrets.token_hex(8).lower()}" + return MySqlDatabase.fromUri(origin=origin, uri=self.server.url(), namespace=namespace) + + def getNewConnection(self, database: MySqlDatabase, *, writeable: bool) -> MySqlDatabase: + return MySqlDatabase.fromUri( + origin=database.origin, uri=self.server.url(), namespace=database.namespace, writeable=writeable + ) + + @contextmanager + def asReadOnly(self, database: MySqlDatabase) -> MySqlDatabase: + yield self.getNewConnection(database, writeable=False) + + def testNameShrinking(self): + """Test that too-long names for database entities other than tables + and columns (which we preserve, and just expect to fit) are shrunk. + """ + db = self.makeEmptyDatabase(origin=1) + with db.declareStaticTables(create=True) as context: + # Table and field names are each below the 63-char limit even when + # accounting for the prefix, but their combination (which will + # appear in sequences and constraints) is not. + tableName = "a_table_with_a_very_very_long_42_char_name" + fieldName1 = "a_column_with_a_very_very_long_43_char_name" + fieldName2 = "another_column_with_a_very_very_long_49_char_name" + context.addTable( + tableName, + ddl.TableSpec( + fields=[ + ddl.FieldSpec( + fieldName1, dtype=sqlalchemy.BigInteger, autoincrement=True, primaryKey=True + ), + ddl.FieldSpec( + fieldName2, + dtype=sqlalchemy.String, + length=16, + nullable=False, + ), + ], + unique={(fieldName2,)}, + ), + ) + # Add another table, this time dynamically, with a foreign key to the + # first table. + db.ensureTableExists( + tableName + "_b", + ddl.TableSpec( + fields=[ + ddl.FieldSpec( + fieldName1 + "_b", dtype=sqlalchemy.BigInteger, autoincrement=True, primaryKey=True + ), + ddl.FieldSpec( + fieldName2 + "_b", + dtype=sqlalchemy.String, + length=16, + nullable=False, + ), + ], + foreignKeys=[ + ddl.ForeignKeySpec(tableName, source=(fieldName2 + "_b",), target=(fieldName2,)), + ], + ), + ) + + +@unittest.skipUnless(testing is not None, "testing.mysqld module not found") +class MySqlRegistryTests(RegistryTests): + """Tests for `Registry` backed by a PostgreSQL database. + + Note + ---- + This is not a subclass of `unittest.TestCase` but to avoid repetition it + defines methods that override `unittest.TestCase` methods. To make this + work sublasses have to have this class first in the bases list. + """ + + @classmethod + def setUpClass(cls): + # MariaDB requires the anonymous user + cls.server = testing.mysqld.Mysqld(user="", my_cnf={"skip-networking": None}) + + @classmethod + def tearDownClass(cls): + cls.server.stop() + + @classmethod + def getDataDir(cls) -> str: + return os.path.normpath(os.path.join(os.path.dirname(__file__), "data", "registry")) + + def makeRegistry(self) -> Registry: + namespace = f"namespace_{secrets.token_hex(8).lower()}" + config = self.makeRegistryConfig() + config["db"] = self.server.url() + config["namespace"] = namespace + return Registry.fromConfig(config, create=True) + + +class MySqlRegistryNameKeyCollMgrTestCase(MySqlRegistryTests, unittest.TestCase): + """Tests for `Registry` backed by a PostgreSQL database. + + This test case uses NameKeyCollectionManager. + """ + + collectionsManager = "lsst.daf.butler.registry.collections.nameKey.NameKeyCollectionManager" + + +class MySqlRegistrySynthIntKeyCollMgrTestCase(MySqlRegistryTests, unittest.TestCase): + """Tests for `Registry` backed by a PostgreSQL database. + + This test case uses SynthIntKeyCollectionManager. + """ + + collectionsManager = "lsst.daf.butler.registry.collections.synthIntKey.SynthIntKeyCollectionManager" + + +if __name__ == "__main__": + unittest.main() From b7611095836415fc3e8815a9775dcc2d7250f675 Mon Sep 17 00:00:00 2001 From: Tim Jenness Date: Thu, 16 Jul 2020 09:22:15 -0700 Subject: [PATCH 2/2] Add on_duplicate_key_update for mysql testCollections still fails. --- .../daf/butler/registry/databases/mysql.py | 19 ++++++++++++++++--- 1 file changed, 16 insertions(+), 3 deletions(-) diff --git a/python/lsst/daf/butler/registry/databases/mysql.py b/python/lsst/daf/butler/registry/databases/mysql.py index 5028275091..ecc21e7fd2 100644 --- a/python/lsst/daf/butler/registry/databases/mysql.py +++ b/python/lsst/daf/butler/registry/databases/mysql.py @@ -23,7 +23,7 @@ __all__ = ["MySqlDatabase"] from contextlib import closing, contextmanager -from typing import Iterator, Optional +from typing import Iterable, Iterator, Optional import sqlalchemy @@ -90,6 +90,11 @@ def fromConnection( ) -> Database: return cls(connection=connection, origin=origin, namespace=namespace, writeable=writeable) + def _lockTables(self, tables: Iterable[sqlalchemy.schema.Table] = ()) -> None: + # Docstring inherited. + for table in tables: + self._connection.execute(f"LOCK TABLE {table.key} IN SHARE MODE") + @contextmanager def transaction(self, *, interrupting: bool = False) -> Iterator[None]: with super().transaction(interrupting=interrupting): @@ -111,9 +116,17 @@ def expandDatabaseEntityName(self, shrunk: str) -> str: return self._shrinker.expand(shrunk) def replace(self, table: sqlalchemy.schema.Table, *rows: dict) -> None: - # This is all wrong if not self.isWriteable(): raise ReadOnlyDatabaseError(f"Attempt to replace into read-only database '{self}'.") if not rows: return - raise NotImplementedError("No support for replace in MySQL") + # This uses special support for UPSERT in MySQL backend: + # https://docs.sqlalchemy.org/en/13/dialects/mysql.html#insert-on-duplicate-key-update-upsert + query = sqlalchemy.dialects.mysql.dml.insert(table) + data = { + column.name: getattr(query.inserted, column.name) + for column in table.columns + if hasattr(query.inserted, column.name) + } + query = query.on_duplicate_key_update(**data) + self._connection.execute(query, *rows)