Repository: radiocosmology/alpenhorn

Issue:
Add schema versioning to the database
Even if we didn't support multiple schema versions in a single alpenhorn version, we could at least have an out-of-date alpenhorn complain and exit rather than screw up the database.

The proposal here: [chime-experiment/Infrastructure#21](https://github.com/chime-experiment/Infrastructure/issues/21#issuecomment-408268954):

> I'd also recommend adding a table for schema versioning, say `schema_version` (Or `schemaversion` since we seem to like smashing words together. Is that what peewee wants?). Given there are multiple extensions in alpenhorn2, maybe have two columns:
> 
> ```
>    component CHAR(60) NOT NULL PRIMARY KEY
>    version INT NOT NULL
> ```
> 
> and then have at least one column for the alpenhorn2 base schema and maybe one for the CHIME extensions:

component | version
-------------- | --------
alpenhorn  | 1
chime        | 1

> 
> Does it need a peewee model? I suppose, if we're using peewee's migrations per #54 )

Discussion:

Solution patch:
diff --git a/alpenhorn/cli/cli.py b/alpenhorn/cli/cli.py
index 1b96258d9..45a486415 100644
--- a/alpenhorn/cli/cli.py
+++ b/alpenhorn/cli/cli.py
@@ -7,13 +7,28 @@
 import click
 
 from ..common.logger import echo
-from ..db import connect as dbconnect  # noqa: F401
+from ..db import connect, schema_version
 
 if TYPE_CHECKING:
     from collections.abc import Callable
 del TYPE_CHECKING
 
 
+def dbconnect(check: bool = True) -> None:
+    """Connect to the database, with schema checking.
+
+    Parameters
+    ----------
+    check:
+        If True (the default), raises ClickException if the
+        database doesn't conform to the current schema version.
+    """
+
+    connect()
+    if check:
+        schema_version(check=True)
+
+
 def check_then_update(
     do_check: bool, do_update: bool, func: Callable, ctx, *, args: list = []
 ) -> None:
diff --git a/alpenhorn/cli/db/__init__.py b/alpenhorn/cli/db/__init__.py
index 9daf6e9aa..be82ccf12 100644
--- a/alpenhorn/cli/db/__init__.py
+++ b/alpenhorn/cli/db/__init__.py
@@ -10,7 +10,8 @@
 def cli():
     """Manage the Data Index."""
 
-    dbconnect()
+    # Don't exit on schema mismatch
+    dbconnect(check=False)
 
 
 cli.add_command(init, "init")
diff --git a/alpenhorn/cli/db/init.py b/alpenhorn/cli/db/init.py
index 6cdb5f856..86afc12cc 100644
--- a/alpenhorn/cli/db/init.py
+++ b/alpenhorn/cli/db/init.py
@@ -3,20 +3,17 @@
 import click
 
 from ...db import (
-    ArchiveAcq,
-    ArchiveFile,
-    ArchiveFileCopy,
-    ArchiveFileCopyRequest,
-    ArchiveFileImportRequest,
-    StorageGroup,
-    StorageNode,
-    StorageTransferAction,
+    DataIndexVersion,
+    current_version,
     database_proxy,
+    gamut,
+    schema_version,
 )
 
 
 @click.command()
-def init():
+@click.pass_context
+def init(ctx):
     """Initialise the Data Index.
 
     This command will create all the database table required
@@ -27,16 +24,33 @@ def init():
     functionality.
     """
 
-    # Create any alpenhorn core tables
-    core_tables = [
-        ArchiveAcq,
-        ArchiveFile,
-        ArchiveFileCopy,
-        ArchiveFileCopyRequest,
-        ArchiveFileImportRequest,
-        StorageGroup,
-        StorageNode,
-        StorageTransferAction,
-    ]
-
-    database_proxy.create_tables(core_tables, safe=True)
+    # Check database schema version
+    vers = schema_version()
+
+    # Are we already good?
+    if vers == current_version:
+        click.echo("Data Index already initialised.")
+        ctx.exit(0)
+
+    # Version mismatch.  This needs to be solved with schema migration.
+    if vers > 0:
+        raise click.ClickException(f"Data Index version {vers} already present.")
+
+    # No schema version, check the database for existing tables
+    dbtables = database_proxy.get_tables()
+
+    # "Data Index version 1" refers to CHIME's (former) data index.  No
+    # one should encounter that anymore.
+    for table in gamut:
+        if table._meta.table_name in dbtables:
+            raise click.ClickException(
+                "Partially initialsed Data Index (or Data Index version 1) found.  "
+                "Manual repair required."
+            )
+
+    # No data index tables exist.  We should be good to create the database.
+    database_proxy.create_tables(gamut)
+
+    # Set schema version
+    DataIndexVersion.create(component="alpenhorn", version=current_version)
+    click.echo(f"Data Index version {current_version} initialised.")
diff --git a/alpenhorn/daemon/entry.py b/alpenhorn/daemon/entry.py
index 9666ece94..acee0c970 100755
--- a/alpenhorn/daemon/entry.py
+++ b/alpenhorn/daemon/entry.py
@@ -72,6 +72,9 @@ def entry(ctx, conf, once, test_isolation):
     # Connect to the database
     db.connect()
 
+    # Check the data index schema.  This doesn't return on mismatch
+    db.schema_version(check=True)
+
     # Start the prometheus client, if appropriate.
     if not once:
         metrics.start_promclient()
diff --git a/alpenhorn/db/__init__.py b/alpenhorn/db/__init__.py
index 4d287f47c..965df1dc8 100644
--- a/alpenhorn/db/__init__.py
+++ b/alpenhorn/db/__init__.py
@@ -37,6 +37,7 @@
 
 # Table models
 from .acquisition import ArchiveAcq, ArchiveFile
+from .data_index import DataIndexVersion, current_version, schema_version
 from .archive import ArchiveFileCopy, ArchiveFileCopyRequest, ArchiveFileImportRequest
 from .storage import StorageGroup, StorageNode, StorageTransferAction
 
@@ -49,3 +50,16 @@
 # Naive-UTC stuff courtesy peewee.  These were originally in datetime
 # but were deprecated in 3.12 as too confusing.
 from peewee import utcnow, utcfromtimestamp
+
+# This contains all tables in the Data Index
+gamut = (
+    ArchiveAcq,
+    ArchiveFile,
+    ArchiveFileCopy,
+    ArchiveFileCopyRequest,
+    ArchiveFileImportRequest,
+    DataIndexVersion,
+    StorageGroup,
+    StorageNode,
+    StorageTransferAction,
+)
diff --git a/alpenhorn/db/data_index.py b/alpenhorn/db/data_index.py
new file mode 100644
index 000000000..c88983906
--- /dev/null
+++ b/alpenhorn/db/data_index.py
@@ -0,0 +1,141 @@
+"""Data Index metadata
+
+This module provides access to metadata about the Data Index
+itself.
+"""
+
+from __future__ import annotations
+
+import logging
+
+import click
+import peewee as pw
+
+from ._base import base_model, database_proxy
+
+log = logging.getLogger(__name__)
+
+# This is the alpenhorn Data Index schema version implemented by this
+# version of the alpenhorn code-base.  In general, only a single schema
+# version (the one specified here) is supported by a particular version
+# of alpenhorn.
+current_version = 2
+
+
+class DataIndexVersion(base_model):
+    """Schema version table for the Data Index.
+
+    Attributes
+    ----------
+    component : str
+        The component of the Data Index being versioned.
+        The primary alpenhorn tables have the component
+        name "alpenhorn".  Extensions may also use this
+        table to track their database schema versions by
+        setting this to the name of the extension.
+    version : int
+        The schema version of the Data Index for the component.
+        Schema versions should start at one and increment by
+        one each time the schema for the specified `component`
+        is changed.
+    """
+
+    component = pw.CharField(max_length=256, unique=True)
+    version = pw.IntegerField()
+
+
+def schema_version(
+    *,
+    check: bool = False,
+    component: str | None = None,
+    component_version: int | None = None,
+) -> int:
+    """Report and optionally check Data Index schema version
+
+    By default, this function returns the schema version of the
+    Data Index in the database.  This function can also be used
+    to check for a particular version and raise an exception if
+    a version mismatch is encountered.
+
+    All parameters must be specified by keyword.
+
+    Parameters
+    ----------
+    check:
+        If True, check that the alpenhorn Data Index schema version
+        is equal to `alpenhorn.db.current_version` and raise an
+        exception if it isn't.  If this is True, the other parameters
+        to this function are ignored.  Setting this to True is equivalent
+        to setting `component` to "alpenhorn" and `component_version` to
+        `alpenhorn.db.current_version`.
+    component:
+        The name of the schema component to report/check.  Setting this
+        to `None` (the default) is equivalent to setting it to "alpenhorn".
+    component_version:
+        If not `None`, raise an exception if the schema version of the
+        component specified by `component` is not equal to this value.
+        If `component` is not present in the DataIndexVersion table, this
+        check always fails.
+
+    Returns
+    -------
+    version : int
+        The schema version of the specified component, or of the alpenhorn
+        Data Index itself, if no other component was specified.  If this
+        function was requested to check the version (because `check` or
+        `component_version` was set), this will always equal the target
+        version (because any other version will result in an exception, instead).
+        If the `component` was not listed in the DataIndexVersion table, zero
+        is returned.
+
+    Raises
+    ------
+    click.ClickException:
+        A check was requested and the check failed.
+    """
+    # Special case for check
+    if check:
+        component = "alpenhorn"
+        component_version = current_version
+    elif not component:
+        component = "alpenhorn"
+
+    # Fetch version for component
+    try:
+        schema = DataIndexVersion.get_or_none(component=component)
+    except pw.OperationalError:
+        # This may be because the table doesn't exist.  Look for it.
+        if DataIndexVersion._meta.table_name in database_proxy.get_tables():
+            # Table exists, but be some sort of other error
+            raise
+
+        # Otherwise, no table
+        schema = None
+
+    if schema:
+        schema_vers = schema.version
+    else:
+        schema_vers = 0
+
+    # Check, if requested
+    if component_version is not None:
+        # Importantly here: if `schema` is None, this check always fails
+        # (i.e. even if the caller were to have passed component_version=0)
+        if not schema or schema_vers != component_version:
+            # There's a special message for alpenhorn itself
+            if component == "alpenhorn":
+                lead = "Data Index version mismatch"
+            else:
+                lead = f'Schema version mismatch for Data Index component "{component}"'
+
+            wanted = f"wanted version {component_version}"
+
+            if schema:
+                found = f"found version {schema_vers}"
+            else:
+                found = "no schema found"
+
+            raise click.ClickException(f"{lead}:  {wanted}; {found}")
+
+    # Check, if performed, succeeded.  Return found version
+    return schema_vers
diff --git a/demo/demo-script.md b/demo/demo-script.md
index e6ba859fc..fff351055 100644
--- a/demo/demo-script.md
+++ b/demo/demo-script.md
@@ -252,10 +252,11 @@ CLI:
   alpenhorn db init
 ```
 Remember that all these alpenhorn commands need to be run inside the `alpen1` container that we
-started in the last section.  The `db init` command outputs nothing when successful:
+started in the last section.  On successful completion, the `db init` command will report the version
+of the database schema used to create the Data Index:
 ```(console)
 root@alpen1:/# alpenhorn db init
-root@alpen1:/#
+Data Index version 2 initialised.
 ```
 
 ### Create the first StorageNode


Test patch:
diff --git a/tests/cli/acq/test_list.py b/tests/cli/acq/test_list.py
index 71431f5c4..f459b1ff0 100644
--- a/tests/cli/acq/test_list.py
+++ b/tests/cli/acq/test_list.py
@@ -18,6 +18,12 @@ def test_no_list(clidb, cli):
     assert "Name" not in result.output
 
 
+def test_schema_mismatch(clidb, cli, cli_wrong_schema):
+    """Test schema mismatch."""
+
+    cli(1, ["acq", "list"])
+
+
 def test_list(clidb, cli, assert_row_present):
     """Test with no constraints."""
 
diff --git a/tests/cli/db/test_init.py b/tests/cli/db/test_init.py
index adffabe63..e42a51384 100644
--- a/tests/cli/db/test_init.py
+++ b/tests/cli/db/test_init.py
@@ -9,9 +9,11 @@
     ArchiveFileCopy,
     ArchiveFileCopyRequest,
     ArchiveFileImportRequest,
+    DataIndexVersion,
     StorageGroup,
     StorageNode,
     StorageTransferAction,
+    current_version,
 )
 
 
@@ -24,6 +26,9 @@ def test_init(clidb_noinit, cli):
 
     cli(0, ["db", "init"])
 
+    # Check schema version
+    assert DataIndexVersion.get(component="alpenhorn").version == current_version
+
     # Now we should have tables
     StorageGroup.create(name="Test")
     assert StorageGroup.get(name="Test").id == 1
@@ -50,8 +55,8 @@ def test_init(clidb_noinit, cli):
     assert ArchiveFileImportRequest.get(path="path", node_id=1).id == 1
 
 
-def test_init_safe(clidb, cli):
-    """Test DB init doesn't overwrite tables"""
+def test_init_existing(clidb, cli):
+    """Test db init with already eixsting data index."""
 
     # Tables are already present
     StorageGroup.create(name="Test")
@@ -65,6 +70,9 @@ def test_init_safe(clidb, cli):
 
     cli(0, ["db", "init"])
 
+    # Check schema version
+    assert DataIndexVersion.get(component="alpenhorn").version == current_version
+
     # Tables weren't overwritten
     assert StorageGroup.get(name="Test").id == 1
     assert StorageNode.get(name="Test").id == 1
@@ -74,3 +82,20 @@ def test_init_safe(clidb, cli):
     assert ArchiveFileCopy.get(file_id=1, node_id=1).id == 1
     assert ArchiveFileCopyRequest.get(file_id=1, node_from_id=1, group_to_id=1).id == 1
     assert ArchiveFileImportRequest.get(path="path", node_id=1).id == 1
+
+
+def test_init_wrong_version(clidb, cli, cli_wrong_schema):
+    """Test init with version mismatch."""
+
+    # Init fails
+    cli(1, ["db", "init"])
+
+
+def test_init_version1(clidb, cli):
+    """Test init with old CHIME data index."""
+
+    # By "old CHIME data index" we mean one without a schema version
+    DataIndexVersion.delete().execute()
+
+    # Init fails
+    cli(1, ["db", "init"])
diff --git a/tests/cli/file/test_list.py b/tests/cli/file/test_list.py
index b634008e6..89fecf4be 100644
--- a/tests/cli/file/test_list.py
+++ b/tests/cli/file/test_list.py
@@ -11,6 +11,12 @@
 )
 
 
+def test_schema_mismatch(clidb, cli, cli_wrong_schema):
+    """Test schema mismatch."""
+
+    cli(1, ["file", "list"])
+
+
 def test_only_from_to(clidb, cli):
     """Can't only use --from or --to"""
 
diff --git a/tests/cli/group/test_list.py b/tests/cli/group/test_list.py
index 546e2052c..d8b3dd1d6 100644
--- a/tests/cli/group/test_list.py
+++ b/tests/cli/group/test_list.py
@@ -3,6 +3,12 @@
 from alpenhorn.db import StorageGroup
 
 
+def test_schema_mismatch(clidb, cli, cli_wrong_schema):
+    """Test schema mismatch."""
+
+    cli(1, ["group", "list"])
+
+
 def test_list(clidb, cli):
     """Test listing groups."""
 
diff --git a/tests/cli/node/test_list.py b/tests/cli/node/test_list.py
index a8ab9b3ff..2019bb5a7 100644
--- a/tests/cli/node/test_list.py
+++ b/tests/cli/node/test_list.py
@@ -32,6 +32,12 @@ def some_nodes(clidb):
     )
 
 
+def test_schema_mismatch(clidb, cli, cli_wrong_schema):
+    """Test schema mismatch."""
+
+    cli(1, ["node", "list"])
+
+
 def test_no_list(clidb, cli):
     """Test listing no nodes."""
 
diff --git a/tests/conftest.py b/tests/conftest.py
index 532ec6be5..69a880ac0 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -11,6 +11,7 @@
 
 import pytest
 import yaml
+from peewee import SqliteDatabase
 
 import alpenhorn.common.logger
 from alpenhorn import db
@@ -22,6 +23,7 @@
     ArchiveFileCopy,
     ArchiveFileCopyRequest,
     ArchiveFileImportRequest,
+    DataIndexVersion,
     StorageGroup,
     StorageNode,
     StorageTransferAction,
@@ -483,18 +485,10 @@ def dbproxy(set_config):
 def dbtables(dbproxy):
     """Create all the usual tables in the database."""
 
-    dbproxy.create_tables(
-        [
-            StorageGroup,
-            StorageNode,
-            StorageTransferAction,
-            ArchiveAcq,
-            ArchiveFile,
-            ArchiveFileCopy,
-            ArchiveFileCopyRequest,
-            ArchiveFileImportRequest,
-        ]
-    )
+    dbproxy.create_tables(db.gamut)
+
+    # Set schema
+    DataIndexVersion.create(component="alpenhorn", version=db.current_version)
 
 
 @pytest.fixture
@@ -668,18 +662,10 @@ def clidb(clidb_noinit):
 
     Yields the connector."""
 
-    clidb_noinit.create_tables(
-        [
-            StorageGroup,
-            StorageNode,
-            StorageTransferAction,
-            ArchiveAcq,
-            ArchiveFile,
-            ArchiveFileCopy,
-            ArchiveFileCopyRequest,
-            ArchiveFileImportRequest,
-        ]
-    )
+    clidb_noinit.create_tables(db.gamut)
+
+    # Set schema
+    DataIndexVersion.create(component="alpenhorn", version=db.current_version)
 
     yield clidb_noinit
 
@@ -690,31 +676,18 @@ def clidb_noinit(clidb_uri):
 
     Yields the connector."""
 
-    import peewee as pw
-
-    from alpenhorn.db import EnumField, database_proxy
-
     # Open
-    db = pw.SqliteDatabase(clidb_uri, uri=True)
-    assert db is not None
-    database_proxy.initialize(db)
-    EnumField.native = False
+    connector = SqliteDatabase(clidb_uri, uri=True)
+    assert connector is not None
+    db.database_proxy.initialize(connector)
+    db.EnumField.native = False
 
-    yield db
+    yield connector
 
     # Drop all the tables after the test
-    for table in [
-        StorageGroup,
-        StorageNode,
-        StorageTransferAction,
-        ArchiveAcq,
-        ArchiveFile,
-        ArchiveFileCopy,
-        ArchiveFileCopyRequest,
-        ArchiveFileImportRequest,
-    ]:
-        db.execute_sql(f"DROP TABLE IF EXISTS {table._meta.table_name};")
-    db.close()
+    for table in db.gamut:
+        connector.execute_sql(f"DROP TABLE IF EXISTS {table._meta.table_name};")
+    connector.close()
 
 
 @pytest.fixture
@@ -926,3 +899,15 @@ def _assert_row_present(text, *cells):
             pytest.fail("Row not found:\n   " + text_row)
 
     return _assert_row_present
+
+
+@pytest.fixture
+def cli_wrong_schema(clidb):
+    """Set the schema of the CLI data index to the wrong version.
+
+    DB must already be initialised.
+    """
+
+    DataIndexVersion.update(version=db.current_version + 1).where(
+        DataIndexVersion.component == "alpenhorn"
+    ).execute()
diff --git a/tests/daemon/test_entry.py b/tests/daemon/test_entry.py
index 4bd77b6c7..2506bfb27 100644
--- a/tests/daemon/test_entry.py
+++ b/tests/daemon/test_entry.py
@@ -27,13 +27,18 @@
 from click.testing import CliRunner
 
 from alpenhorn.daemon.entry import entry
-from alpenhorn.db.acquisition import ArchiveAcq, ArchiveFile
-from alpenhorn.db.archive import (
+from alpenhorn.db import (
+    ArchiveAcq,
+    ArchiveFile,
     ArchiveFileCopy,
     ArchiveFileCopyRequest,
-    ArchiveFileImportRequest,
+    DataIndexVersion,
+    StorageGroup,
+    StorageNode,
+    StorageTransferAction,
+    current_version,
+    gamut,
 )
-from alpenhorn.db.storage import StorageGroup, StorageNode, StorageTransferAction
 
 # Import pattern_importer from the examples directory
 sys.path.append(str(pathlib.Path(__file__).parent.joinpath("..", "..", "examples")))
@@ -46,15 +51,15 @@ def e2e_db(xfs, clidb_noinit, hostname):
 
     db = clidb_noinit
 
+    # Remove tables duplicated by the pattern_importer
+    filtered = (
+        table for table in gamut if table != ArchiveFile and table != ArchiveAcq
+    )
+
     # Create tables
     db.create_tables(
         [
-            ArchiveFileCopy,
-            ArchiveFileCopyRequest,
-            ArchiveFileImportRequest,
-            StorageGroup,
-            StorageNode,
-            StorageTransferAction,
+            *filtered,
             pattern_importer.AcqType,
             pattern_importer.FileType,
             pattern_importer.ExtendedAcq,
@@ -65,6 +70,8 @@ def e2e_db(xfs, clidb_noinit, hostname):
     # Populate tables
     # ---------------
 
+    DataIndexVersion.create(component="alpenhorn", version=current_version)
+
     # A Default-IO group with one node
     dftgrp = StorageGroup.create(name="dftgroup")
     dftnode = StorageNode.create(
diff --git a/tests/db/test_data_index.py b/tests/db/test_data_index.py
new file mode 100644
index 000000000..4c6dee229
--- /dev/null
+++ b/tests/db/test_data_index.py
@@ -0,0 +1,113 @@
+"""test alpenhorn.db.data_index."""
+
+import click
+import peewee as pw
+import pytest
+
+from alpenhorn.db.data_index import DataIndexVersion, current_version, schema_version
+
+
+def test_data_index_version(dbproxy):
+    """Test that the DataIndexVersion table can be created."""
+    dbproxy.create_tables([DataIndexVersion])
+    assert set(dbproxy.get_tables()) == {"dataindexversion"}
+
+
+def test_data_index_version_model(dbproxy):
+    """Test DataIndexVersion table model."""
+
+    dbproxy.create_tables([DataIndexVersion])
+
+    DataIndexVersion.create(component="test", version=3)
+
+    # component is unique
+    with pytest.raises(pw.IntegrityError):
+        DataIndexVersion.create(component="test", version=4)
+
+    # component cannot be NULL
+    with pytest.raises(pw.IntegrityError):
+        DataIndexVersion.create(component=None, version=5)
+
+    # version cannot be NULL
+    with pytest.raises(pw.IntegrityError):
+        DataIndexVersion.create(component="test2", version=None)
+
+    # Check DB
+    assert DataIndexVersion.select().where(
+        DataIndexVersion.component == "test"
+    ).dicts().get() == {
+        "id": 1,
+        "component": "test",
+        "version": 3,
+    }
+
+
+def test_schema_version(dbtables):
+    """schema_version() should return the version."""
+
+    assert schema_version() == current_version
+
+
+def test_schema_version_check_ok(dbtables):
+    """Test successful check in schema_version."""
+
+    # Parameters other than check are ignored
+    assert (
+        schema_version(
+            check=True, component="MISSING", component_version=current_version + 1
+        )
+        == current_version
+    )
+
+
+def test_schema_version_check_bad(dbtables):
+    """Test unsuccessful check in schema_version."""
+
+    # Change version
+    DataIndexVersion.update(version=current_version + 1).where(
+        DataIndexVersion.component == "alpenhorn"
+    ).execute()
+
+    with pytest.raises(click.ClickException):
+        schema_version(check=True)
+
+
+def test_schema_version_check_explicit(dbtables):
+    """Test explicit check with schema_version."""
+
+    # Change version
+    DataIndexVersion.update(version=current_version + 1).where(
+        DataIndexVersion.component == "alpenhorn"
+    ).execute()
+
+    assert (
+        schema_version(component="alpenhorn", component_version=current_version + 1)
+        == current_version + 1
+    )
+
+    with pytest.raises(click.ClickException):
+        schema_version(component="alpenhorn", component_version=current_version + 2)
+
+
+def test_schema_version_missing(dbproxy, dbtables):
+    """Test schema_version with missing data."""
+
+    # Delete all rows
+    DataIndexVersion.delete().execute()
+
+    assert schema_version() == 0
+
+    with pytest.raises(click.ClickException):
+        schema_version(check=True)
+
+    # This should also fail (can't match to zero)
+    with pytest.raises(click.ClickException):
+        schema_version(component="alpenhorn", component_version=0)
+
+
+def test_schema_version_no_table(dbproxy):
+    """Test schema_version with no DataIndexVersion table."""
+
+    # This is one of the few functions that should still work
+    # if the table is not present
+    assert schema_version() == 0