From c4f064de531a67e062187ef536610b7d108215a6 Mon Sep 17 00:00:00 2001 From: Dimitri Yatsenko Date: Sat, 17 Oct 2026 17:10:15 -0500 Subject: [PATCH] fix(drop): drop an auto-populated table's jobs table with it drop_quick() dropped only the table itself, so drop() left an auto-populated table's jobs table, ~~, in place with its records and its primary key. A table later declared under the same name picked it up through its jobs property. When the new primary key was wider, populate(reserve_jobs=True) computed nothing and raised nothing: the old records, keyed on part of the new primary key, covered every new key. drop_quick() now drops the jobs table of a Computed or Imported table after the table itself. drop() reaches each table as a FreeTable, so the tier is read from the table name, as Schema.jobs does. A Manual or Lookup table never drops a jobs table, though a Manual `result` and a Computed `__result` map to the same `~~result`. The jobs table is itself dropped through drop_quick(), so its ~lineage rows and, on PostgreSQL, its enum type go with it. Jobs tables left by earlier releases remain, so the jobs property now checks an existing jobs table's primary key against the one it would be declared with, and raises if they differ, naming the jobs table and the command that removes it. Job computes its foreign-key-derived primary key once, so a table with a primary key attribute of its own still logs the "Ignoring non-FK primary key attribute" warning once per access. The tests run on MySQL and PostgreSQL. In the issue's reproduction, drop() removes ~~result and its lineage rows, and the table declared again with a wider primary key populates all six keys. drop_quick() removes ~~result too, with its enum type on PostgreSQL. Dropping a Manual table leaves the Computed table's ~~result in place. A jobs table left behind by a plain DROP TABLE makes jobs raise; the command in the message removes it, and populate then completes. A table with a primary key attribute of its own passes the check and warns once. Closes #1601. --- src/datajoint/autopopulate.py | 27 ++- src/datajoint/jobs.py | 6 +- src/datajoint/table.py | 14 ++ tests/integration/test_drop_jobs.py | 270 ++++++++++++++++++++++++++++ 4 files changed, 312 insertions(+), 5 deletions(-) create mode 100644 tests/integration/test_drop_jobs.py 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