diff --git a/src/datajoint/autopopulate.py b/src/datajoint/autopopulate.py index 61b36c5c9..ff2d63836 100644 --- a/src/datajoint/autopopulate.py +++ b/src/datajoint/autopopulate.py @@ -175,6 +175,13 @@ def __get__(self, obj, objtype=None): ------- Job Job management object for this table. + + Raises + ------ + DataJointError + If the job table exists with a primary key other than the one + this table's job table is declared with, as when it was left by + an earlier table of the same name. """ if obj is None: # Accessed on class - instantiate first @@ -182,9 +189,23 @@ def __get__(self, obj, objtype=None): if obj._jobs is None: from .jobs import Job - obj._jobs = Job(obj) - if not obj._jobs.is_declared: - obj._jobs.declare() + jobs = Job(obj) + if not jobs.is_declared: + jobs.declare() + else: + # A job table keyed on other attributes matches the wrong keys, + # so populate(reserve_jobs=True) would skip work without an error + expected = [name for name, _ in jobs._pk_attrs] + if set(jobs.primary_key) != set(expected): + class_name = type(obj).__name__ + raise DataJointError( + f"Jobs table {jobs.full_table_name} has primary key {jobs.primary_key}, " + f"but the jobs table of {obj.full_table_name} needs {expected}: " + f"it was left by an earlier table of the same name. Remove it with " + f"dj.FreeTable({class_name}.connection, {jobs.full_table_name!r}).drop_quick(); " + f"the next access to {class_name}.jobs declares a new one." + ) + obj._jobs = jobs return obj._jobs jobs: Job = _JobsDescriptor() diff --git a/src/datajoint/jobs.py b/src/datajoint/jobs.py index 920d4eab0..1a94b7b36 100644 --- a/src/datajoint/jobs.py +++ b/src/datajoint/jobs.py @@ -111,7 +111,9 @@ def __init__(self, target_table: Table) -> None: base_name = target_name.lstrip("_") self._table_name = f"~~{base_name}" - # Generate definition from target's FK-derived primary key + # The target's FK-derived primary key, computed once: it generates the + # definition and is what an existing jobs table's primary key must match + self._pk_attrs = self._get_fk_derived_pk_attrs() self._definition = self._generate_definition() # Initialize heading and support @@ -147,7 +149,7 @@ def _generate_definition(self) -> str: str DataJoint table definition string. """ - pk_attrs = self._get_fk_derived_pk_attrs() + pk_attrs = self._pk_attrs if not pk_attrs: raise DataJointError( diff --git a/src/datajoint/table.py b/src/datajoint/table.py index 9f6784fb3..299bba636 100644 --- a/src/datajoint/table.py +++ b/src/datajoint/table.py @@ -1224,6 +1224,8 @@ def delete( def drop_quick(self): """ Drops the table without cascading to dependent tables and without user prompt. + + An auto-populated table's jobs table (``~~table_name``) is dropped with it. """ if self.is_declared: # Clean up lineage entries for this table @@ -1256,6 +1258,18 @@ def drop_quick(self): logger.debug("Dropped enum type %s" % enum_type) except Exception: pass # Ignore errors - type may be used by other tables + + # Drop the jobs table of an auto-populated table, else a table later + # declared under the same name inherits it. drop() reaches each table + # as a FreeTable, so the tier is read from the name, as Schema.jobs does; + # a Manual `result` and a Computed `__result` map to the same `~~result`. + from .user_tables import Computed, Imported, _get_tier + + if _get_tier(self.full_table_name) in (Computed, Imported): + jobs_table_name = "~~" + self.table_name.lstrip("_") + jobs = FreeTable(self.connection, self.adapter.make_full_table_name(self.database, jobs_table_name)) + if jobs.is_declared: + jobs.drop_quick() else: logger.info("Nothing to drop: table %s is not declared" % self.full_table_name) diff --git a/tests/integration/test_drop_jobs.py b/tests/integration/test_drop_jobs.py new file mode 100644 index 000000000..9f4a5b907 --- /dev/null +++ b/tests/integration/test_drop_jobs.py @@ -0,0 +1,270 @@ +""" +Tests that dropping an auto-populated table drops its jobs table, and that +``jobs`` rejects a jobs table left by an earlier table of the same name (#1601). +""" + +import logging +import re +import time + +import pytest + +import datajoint as dj +from datajoint.errors import DataJointError +from datajoint.lineage import get_table_lineages + + +@pytest.fixture(scope="function") +def schema_by_backend(connection_by_backend, db_creds_by_backend): + """Create a fresh schema per test, parameterized across backends.""" + backend = db_creds_by_backend["backend"] + test_id = str(int(time.time() * 1000))[-8:] + schema = dj.Schema(f"djtest_dropjobs_{backend}_{test_id}"[:64], connection=connection_by_backend) + yield schema + schema.drop(prompt=False) + + +def _table_names(schema): + """Every table in the schema, hidden ones included.""" + sql = schema.connection.adapter.list_tables_sql(schema.database) + return {row[0] for row in schema.connection.query(sql).fetchall()} + + +def _enum_types(schema, table_name): + """PostgreSQL enum types used by a table's columns (none on MySQL).""" + adapter = schema.connection.adapter + if not hasattr(adapter, "get_table_enum_types_sql"): + return set() + sql = adapter.get_table_enum_types_sql(schema.database, table_name) + return {row[0] for row in schema.connection.query(sql).fetchall()} + + +def _existing_enum_types(schema): + """Enum types defined in the schema's namespace (PostgreSQL only).""" + sql = ( + "SELECT n.nspname || '.' || t.typname FROM pg_catalog.pg_type t " + "JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace " + "WHERE t.typtype = 'e' AND n.nspname = %s" + ) + return {row[0] for row in schema.connection.query(sql, args=(schema.database,)).fetchall()} + + +def test_drop_removes_jobs_table(schema_by_backend): + """drop() removes the jobs table, so a table declared again under the same name populates in full.""" + schema = schema_by_backend + + @schema + class Item(dj.Manual): + definition = """ + item : int32 + """ + + @schema + class Param(dj.Lookup): + definition = """ + param : int32 + """ + contents = [(1,), (2,)] + + @schema + class Result(dj.Computed): + definition = """ + -> Item + --- + value : int32 + """ + + def make(self, key): + raise RuntimeError("not ready") + + Item.insert({"item": i} for i in range(3)) + Result.populate(reserve_jobs=True, suppress_errors=True) + assert len(Result.jobs.errors) == 3 + assert get_table_lineages(schema.connection, schema.database, "~~result") + + Result.drop(prompt=False) + assert "~~result" not in _table_names(schema) + assert not get_table_lineages(schema.connection, schema.database, "~~result") + + @schema + class Result(dj.Computed): # same name, primary key now includes Param + definition = """ + -> Item + -> Param + --- + value : int32 + """ + + def make(self, key): + self.insert1(dict(key, value=1)) + + assert Result.jobs.primary_key == ["item", "param"] + assert len(Result.jobs) == 0 + Result.populate(reserve_jobs=True) + assert len(Result()) == len(Result.key_source) == 6 + + +def test_drop_quick_removes_jobs_table(schema_by_backend): + """drop_quick() removes the jobs table with its ~lineage rows and, on PostgreSQL, its enum type.""" + schema = schema_by_backend + + @schema + class Item(dj.Manual): + definition = """ + item : int32 + """ + + @schema + class Result(dj.Computed): + definition = """ + -> Item + --- + value : int32 + """ + + def make(self, key): + self.insert1(dict(key, value=key["item"])) + + Item.insert({"item": i} for i in range(3)) + Result.jobs.refresh() + assert "~~result" in _table_names(schema) + enum_types = _enum_types(schema, "~~result") + assert bool(enum_types) == (schema.connection.adapter.backend == "postgresql") + + Result.drop_quick() + tables = _table_names(schema) + assert "__result" not in tables + assert "~~result" not in tables + assert not get_table_lineages(schema.connection, schema.database, "~~result") + if enum_types: + assert not enum_types & _existing_enum_types(schema) + + +def test_drop_manual_table_keeps_jobs_table(schema_by_backend): + """Dropping a Manual table leaves the jobs table its name maps to, which belongs to a Computed table.""" + schema = schema_by_backend + + @schema + class Item(dj.Manual): + definition = """ + item : int32 + """ + + @schema + class Result(dj.Computed): + definition = """ + -> Item + --- + value : int32 + """ + + def make(self, key): + self.insert1(dict(key, value=key["item"])) + + computed = Result + Item.insert({"item": i} for i in range(3)) + computed.jobs.refresh() + assert len(computed.jobs) == 3 + + @schema + class Result(dj.Manual): # table `result`; `~~result` belongs to `__result` + definition = """ + result_id : int32 + """ + + Result.drop(prompt=False) + tables = _table_names(schema) + assert "result" not in tables + assert {"__result", "~~result"} <= tables + assert len(computed.jobs) == 3 + + +def test_jobs_rejects_stale_jobs_table(schema_by_backend): + """jobs raises on a jobs table keyed differently from the table's, and its hint removes it.""" + schema = schema_by_backend + + @schema + class Item(dj.Manual): + definition = """ + item : int32 + """ + + @schema + class Param(dj.Lookup): + definition = """ + param : int32 + """ + contents = [(1,), (2,)] + + @schema + class Result(dj.Computed): + definition = """ + -> Item + --- + value : int32 + """ + + def make(self, key): + raise RuntimeError("not ready") + + Item.insert({"item": i} for i in range(3)) + Result.populate(reserve_jobs=True, suppress_errors=True) + jobs_table = Result.jobs.full_table_name + # Earlier releases dropped the table and left its jobs table; drop it the same way + schema.connection.query(f"DROP TABLE {Result.full_table_name}") + + @schema + class Result(dj.Computed): # same name, primary key now includes Param + definition = """ + -> Item + -> Param + --- + value : int32 + """ + + def make(self, key): + self.insert1(dict(key, value=1)) + + result = Result() + with pytest.raises(DataJointError, match="left by an earlier table of the same name") as excinfo: + result.jobs + message = str(excinfo.value) + assert jobs_table in message + assert "['item']" in message and "['item', 'param']" in message + assert result._jobs is None # the mismatched jobs table is not kept + + # The hint runs as written + hint = re.search(r"dj\.FreeTable\(.*\)\.drop_quick\(\)", message).group(0) + eval(hint, {"dj": dj, "Result": Result}) + assert result.jobs.primary_key == ["item", "param"] + Result.populate(reserve_jobs=True) + assert len(Result()) == len(Result.key_source) == 6 + + +def test_jobs_accepts_table_with_native_key_attribute(schema_by_backend, caplog): + """A jobs table keyed on the FK-derived part of the primary key passes the check, warning once.""" + schema = schema_by_backend + + @schema + class Item(dj.Manual): + definition = """ + item : int32 + """ + + with dj.config.override(jobs__allow_new_pk_fields_in_computed_tables=True): + + @schema + class Result(dj.Computed): + definition = """ + -> Item + index : int32 + --- + value : int32 + """ + + Result.jobs.refresh() # declares the jobs table, keyed on item alone + caplog.clear() + with caplog.at_level(logging.WARNING, logger="datajoint"): + assert Result.jobs.primary_key == ["item"] + warnings = [r for r in caplog.records if "Ignoring non-FK primary key attribute 'index'" in r.getMessage()] + assert len(warnings) == 1