Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 24 additions & 3 deletions src/datajoint/autopopulate.py
Original file line number Diff line number Diff line change
Expand Up @@ -175,16 +175,37 @@ 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
obj = objtype()
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()
Expand Down
6 changes: 4 additions & 2 deletions src/datajoint/jobs.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down
14 changes: 14 additions & 0 deletions src/datajoint/table.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand Down
270 changes: 270 additions & 0 deletions tests/integration/test_drop_jobs.py
Original file line number Diff line number Diff line change
@@ -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
Loading