Bug: reserve_jobs=True fails when dj.config.jobs.add_job_metadata = True and dj.config.jobs.keep_completed = True

[arturoptophys]

Bug Report

Description

When testing the job metadata functionality i encountered following problem:
When setting

dj.config.jobs.add_job_metadata = True
dj.config.jobs.keep_completed = True

i cant run populate with reserve_jobs=True.

Reproducibility

import datajoint as dj
dj.config.jobs.add_job_metadata = True
dj.config.jobs.keep_completed = True
dj.conn() # using datajoint.json
schema = dj.Schema("TEST_")

@schema
class Numbers(dj.Lookup):
 definition = """
 number: int64
 """
 contents = [(0,), (1,), (2,), (4,), (8,)]

@schema
class NumbersSquared(dj.Computed):
 definition = """
 -> Numbers
 ---
 squared: int64
 """

 def make(self, key):
 self.insert1({**key, "squared": key["number"] ** 2})

NumbersSquared.populate(reserve_jobs=True)

Fails with:

---------------------------------------------------------------------------
DataJointError Traceback (most recent call last)
Cell In[2], line 1
----> 1 NumbersSquared.populate(reserve_jobs=True)

File ~/datajoint-python/src/datajoint/autopopulate.py:366, in AutoPopulate.populate(self, suppress_errors, return_exception_objects, reserve_jobs, max_calls, display_progress, processes, make_kwargs, priority, refresh, *restrictions)
 363 raise DataJointError("Populate cannot be called during a transaction.")
 365 if reserve_jobs:
--> 366 return self._populate_distributed(
 367 *restrictions,
 368 suppress_errors=suppress_errors,
 369 return_exception_objects=return_exception_objects,
 370 max_calls=max_calls,
 371 display_progress=display_progress,
 372 processes=processes,
 373 make_kwargs=make_kwargs,
 374 priority=priority,
 375 refresh=refresh,
 376 )
 377 else:
 378 return self._populate_direct(
 379 *restrictions,
 380 suppress_errors=suppress_errors,
 (...) 385 make_kwargs=make_kwargs,
 386 )

File ~/datajoint-python/src/datajoint/autopopulate.py:490, in AutoPopulate._populate_distributed(self, suppress_errors, return_exception_objects, max_calls, display_progress, processes, make_kwargs, priority, refresh, *restrictions)
 486 refresh = config.jobs.auto_refresh
 487 if refresh:
 488 # Use delay=-1 to ensure jobs are immediately schedulable
 489 # (avoids race condition with scheduled_time <= CURRENT_TIMESTAMP(3) check)
--> 490 self.jobs.refresh(*restrictions, priority=priority, delay=-1)
 492 # Fetch pending jobs ordered by priority (use CURRENT_TIMESTAMP(3) for datetime(3) precision)
 493 pending_query = self.jobs.pending & "scheduled_time <= CURRENT_TIMESTAMP(3)"

File ~/datajoint-python/src/datajoint/jobs.py:398, in Job.refresh(self, delay, priority, stale_timeout, orphan_timeout, *restrictions)
 394 # 2. Re-pend success jobs if keep_completed=True
 395 if config.jobs.keep_completed:
 396 # Success jobs whose keys are in key_source but not in target
 397 # Disable semantic_check for Job table operations
--> 398 success_to_repend = self.completed.restrict(key_source, semantic_check=False) - self._target
 399 repend_keys = success_to_repend.keys()
 400 for key in repend_keys:

File ~/datajoint-python/src/datajoint/expression.py:282, in QueryExpression.__sub__(self, restriction)
 276 def __sub__(self, restriction):
 277 """
 278 Inverted restriction e.g. ``q1 - q2``.
 279 :return: a restricted copy of the input argument
 280 See QueryExpression.restrict for more detail.
 281 """
--> 282 return self.restrict(Not(restriction))

File ~/datajoint-python/src/datajoint/expression.py:234, in QueryExpression.restrict(self, restriction, semantic_check)
 232 result._top = restriction
 233 return result
--> 234 new_condition = make_condition(self, restriction, attributes, semantic_check=semantic_check)
 235 if new_condition is True:
 236 return self # restriction has no effect, return the same object

File ~/datajoint-python/src/datajoint/condition.py:403, in make_condition(query_expression, condition, columns, semantic_check)
 401 # restrict by another expression
 402 if isinstance(condition, QueryExpression):
--> 403 assert_join_compatibility(query_expression, condition, semantic_check=semantic_check)
 404 # Match on all non-hidden namesakes (hidden attributes excluded)
 405 common_attributes = [q for q in condition.heading.names if q in query_expression.heading.names]

File ~/datajoint-python/src/datajoint/condition.py:256, in assert_join_compatibility(expr1, expr2, semantic_check)
 254 # Semantic match requires both lineages to be non-None and equal
 255 if lineage1 is None or lineage2 is None or lineage1 != lineage2:
--> 256 raise DataJointError(
 257 f"Cannot join on attribute `{name}`: "
 258 f"different lineages ({lineage1} vs {lineage2}). "
 259 f"Use .proj() to rename one of the attributes."
 260 )

DataJointError: Cannot join on attribute `number`: different lineages (TEST_.~~numbers_squared.number vs TEST_.#numbers.number). Use .proj() to rename one of the attributes.

Ubuntu 24.04, python 3.12.11
MySQL - remotely accessed myslq-latest docker (MySQL Server 9.6.0-1.el9)
DataJoint Version : DataJoint 2.0.0a27

[arturoptophys]

Also: related to dj.config.jobs.add_job_metadata = True

When creating and populating tables (with keep_completed = False to get around the bug above) as above in jupyter notebook, then dropping the tables, then restarting kernel the tables cannot be created

---------------------------------------------------------------------------
DataJointError Traceback (most recent call last)
Cell In[1], line 24
 21 def make(self, key):
 22 self.insert1({**key, "squared": key["number"] ** 2})
---> 24 NumbersSquared.populate(reserve_jobs=True)

File ~/datajoint-python/src/datajoint/autopopulate.py:366, in AutoPopulate.populate(self, suppress_errors, return_exception_objects, reserve_jobs, max_calls, display_progress, processes, make_kwargs, priority, refresh, *restrictions)
 363 raise DataJointError("Populate cannot be called during a transaction.")
 365 if reserve_jobs:
--> 366 return self._populate_distributed(
 367 *restrictions,
 368 suppress_errors=suppress_errors,
 369 return_exception_objects=return_exception_objects,
 370 max_calls=max_calls,
 371 display_progress=display_progress,
 372 processes=processes,
 373 make_kwargs=make_kwargs,
 374 priority=priority,
 375 refresh=refresh,
 376 )
 377 else:
 378 return self._populate_direct(
 379 *restrictions,
 380 suppress_errors=suppress_errors,
 (...) 385 make_kwargs=make_kwargs,
 386 )

File ~/datajoint-python/src/datajoint/autopopulate.py:490, in AutoPopulate._populate_distributed(self, suppress_errors, return_exception_objects, max_calls, display_progress, processes, make_kwargs, priority, refresh, *restrictions)
 486 refresh = config.jobs.auto_refresh
 487 if refresh:
 488 # Use delay=-1 to ensure jobs are immediately schedulable
 489 # (avoids race condition with scheduled_time <= CURRENT_TIMESTAMP(3) check)
--> 490 self.jobs.refresh(*restrictions, priority=priority, delay=-1)
 492 # Fetch pending jobs ordered by priority (use CURRENT_TIMESTAMP(3) for datetime(3) precision)
 493 pending_query = self.jobs.pending & "scheduled_time <= CURRENT_TIMESTAMP(3)"

File ~/datajoint-python/src/datajoint/autopopulate.py:124, in AutoPopulate._JobsDescriptor.__get__(self, obj, objtype)
 121 if obj._jobs is None:
 122 from .jobs import Job
--> 124 obj._jobs = Job(obj)
 125 if not obj._jobs.is_declared:
 126 obj._jobs.declare()

File ~/datajoint-python/src/datajoint/jobs.py:109, in Job.__init__(self, target_table)
 106 self._table_name = f"~~{base_name}"
 108 # Generate definition from target's FK-derived primary key
--> 109 self._definition = self._generate_definition()
 111 # Initialize heading and support
 112 self._heading = Heading(
 113 table_info=dict(
 114 conn=self._connection,
 (...) 118 )
 119 )

File ~/datajoint-python/src/datajoint/jobs.py:144, in Job._generate_definition(self)
 135 def _generate_definition(self) -> str:
 136 """
 137 Generate jobs table definition from target's FK-derived primary key.
 138 
 (...) 142 DataJoint table definition string.
 143 """
--> 144 pk_attrs = self._get_fk_derived_pk_attrs()
 146 if not pk_attrs:
 147 raise DataJointError(
 148 f"Cannot create jobs table for {self._target.full_table_name}: no FK-derived primary key attributes found."
 149 )

File ~/datajoint-python/src/datajoint/jobs.py:186, in Job._get_fk_derived_pk_attrs(self)
 174 """
 175 Extract FK-derived primary key attributes using the dependency graph.
 176 
 (...) 183 List of (attribute_name, datatype) tuples in target PK order.
 184 """
 185 heading = self._target.heading
--> 186 target_pk = heading.primary_key
 188 # Load dependency graph if not already loaded
 189 self._connection.dependencies.load()

File ~/datajoint-python/src/datajoint/heading.py:241, in Heading.primary_key(self)
 238 @property
 239 def primary_key(self) -> list[str]:
 240 """List of primary key attribute names."""
--> 241 return [k for k, v in self.attributes.items() if v.in_key]

File ~/datajoint-python/src/datajoint/heading.py:230, in Heading.attributes(self)
 224 """
 225 Mapping of attribute names to Attribute objects.
 226 
 227 Excludes hidden attributes (names starting with ``_``).
 228 """
 229 if self._attributes is None:
--> 230 self._init_from_database() # lazy loading from database
 231 return {k: v for k, v in self._attributes.items() if not v.is_hidden}

File ~/work_geneva/datajoint-python/src/datajoint/heading.py:358, in Heading._init_from_database(self)
 353 info = conn.query(
 354 'SHOW TABLE STATUS FROM `{database}` WHERE name="{table_name}"'.format(table_name=table_name, database=database),
 355 as_dict=True,
 356 ).fetchone()
 357 if info is None:
--> 358 raise DataJointError(
 359 "The table `{database}`.`{table_name}` is not defined.".format(table_name=table_name, database=database)
 360 )
 361 self._table_status = {k.lower(): v for k, v in info.items()}
 362 cur = conn.query(
 363 "SHOW FULL COLUMNS FROM `{table_name}` IN `{database}`".format(table_name=table_name, database=database),
 364 as_dict=True,
 365 )

DataJointError: The table `TEST_`.`__numbers_squared` is not defined.

My guess is there are some hanging jobs_metadata tables still in mysql?