DM-25972: Investigate MySQL/MariaDB

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

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),
         ],
     )
 

python/lsst/daf/butler/registry/databases/mysql.py

@@ -0,0 +1,132 @@
+# 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 <http://www.gnu.org/licenses/>.
+from __future__ import annotations
+
+__all__ = ["MySqlDatabase"]
+
+from contextlib import closing, contextmanager
+from typing import Iterable, 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)
+
+    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):
+            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:
+        if not self.isWriteable():
+            raise ReadOnlyDatabaseError(f"Attempt to replace into read-only database '{self}'.")
+        if not rows:
+            return
+        # 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)

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 <http://www.gnu.org/licenses/>.
+
+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()