merge: resolve conflict in jobs.py reserve method

Keep master's atomic SQL UPDATE for job reservation (prevents race
conditions) while threading connection-scoped config through
_get_job_version().

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: dimitri-yatsenko

src/datajoint/jobs.py

@@ -13,7 +13,7 @@
 import platform
 import subprocess
 
-from .condition import AndList, Not
+from .condition import AndList, Not, make_condition
 from .errors import DataJointError, DuplicateError
 from .heading import Heading
 from .table import Table
@@ -435,8 +435,10 @@ def reserve(self, key: dict) -> bool:
         """
         Attempt to reserve a pending job for processing.
 
-        Updates status to ``'reserved'`` if currently ``'pending'`` and
-        ``scheduled_time <= now``.
+        Atomically updates status to ``'reserved'`` if currently ``'pending'``
+        and ``scheduled_time <= now``, using a single UPDATE with a WHERE clause
+        that includes the status check. This prevents race conditions where
+        multiple workers could reserve the same job simultaneously.
 
         Parameters
         ----------
@@ -448,33 +450,26 @@ def reserve(self, key: dict) -> bool:
         bool
             True if reservation successful, False if job not available.
         """
-        # Check if job is pending and scheduled (use CURRENT_TIMESTAMP(3) for datetime(3) precision)
-        job = (self & key & "status='pending'" & "scheduled_time <= CURRENT_TIMESTAMP(3)").to_dicts()
-
-        if not job:
-            return False
-
-        # Get server time for reserved_time
-        server_now = self.connection.query("SELECT CURRENT_TIMESTAMP").fetchone()[0]
-
-        # Build update row with primary key and new values
         pk = self._get_pk(key)
-        update_row = {
-            **pk,
-            "status": "reserved",
-            "reserved_time": server_now,
-            "host": platform.node(),
-            "pid": os.getpid(),
-            "connection_id": self.connection.connection_id,
-            "user": self.connection.get_user(),
-            "version": _get_job_version(self.connection._config),
-        }
-
-        try:
-            self.update1(update_row)
-            return True
-        except Exception:
-            return False
+        where = make_condition(self, pk, set())
+        qi = self.adapter.quote_identifier
+        assignments = ", ".join(f"{qi(k)}=%s" for k in ("status", "host", "pid", "connection_id", "user", "version"))
+        query = (
+            f"UPDATE {self.full_table_name} "
+            f"SET {assignments}, {qi('reserved_time')}=CURRENT_TIMESTAMP(3) "
+            f"WHERE {where} AND {qi('status')}='pending' "
+            f"AND {qi('scheduled_time')} <= CURRENT_TIMESTAMP(3)"
+        )
+        args = [
+            "reserved",
+            platform.node(),
+            os.getpid(),
+            self.connection.connection_id,
+            self.connection.get_user(),
+            _get_job_version(self.connection._config),
+        ]
+        cursor = self.connection.query(query, args=args)
+        return cursor.rowcount == 1
 
     def complete(self, key: dict, duration: float | None = None) -> None:
         """

src/datajoint/schemas.py

@@ -320,26 +320,6 @@ def _decorate_table(self, table_class: type, context: dict[str, Any], assert_dec
     def __repr__(self):
         return "Schema `{name}`\n".format(name=self.database)
 
-    @property
-    def size_on_disk(self) -> int:
-        """
-        Return the total size of all tables in the schema.
-
-        Returns
-        -------
-        int
-            Size in bytes (data + indices).
-        """
-        self._assert_exists()
-        return int(
-            self.connection.query(
-                """
-            SELECT SUM(data_length + index_length)
-            FROM information_schema.tables WHERE table_schema='{db}'
-            """.format(db=self.database)
-            ).fetchone()[0]
-        )
-
     def make_classes(self, into: dict[str, Any] | None = None) -> None:
         """
         Create Python table classes for tables in the schema.

src/datajoint/table.py

@@ -1250,22 +1250,6 @@ def drop(self, prompt: bool | None = None):
                 FreeTable(self.connection, table).drop_quick()
             logger.info("Tables dropped. Restart kernel.")
 
-    @property
-    def size_on_disk(self):
-        """
-        Return the size of data and indices in bytes on the storage device.
-
-        Returns
-        -------
-        int
-            Size of data and indices in bytes.
-        """
-        ret = self.connection.query(
-            'SHOW TABLE STATUS FROM `{database}` WHERE NAME="{table}"'.format(database=self.database, table=self.table_name),
-            as_dict=True,
-        ).fetchone()
-        return ret["Data_length"] + ret["Index_length"]
-
     def describe(self, context=None, printout=False):
         """
         Return the definition string for the query using DataJoint DDL.

src/datajoint/version.py

@@ -1,4 +1,4 @@
 # version bump auto managed by Github Actions:
 # label_prs.yaml(prep), release.yaml(bump), post_release.yaml(edit)
 # manually set this version will be eventually overwritten by the above actions
-__version__ = "2.1.0"
+__version__ = "2.1.1"

tests/integration/test_relation.py

@@ -282,11 +282,5 @@ def relation_selector(attr):
             ), "Regular expression matches for {name} but should not".format(name=name)
 
 
-def test_table_size(experiment):
-    """test getting the size of the table and its indices in bytes"""
-    number_of_bytes = experiment.size_on_disk
-    assert isinstance(number_of_bytes, int) and number_of_bytes > 100
-
-
 def test_repr_html(ephys):
     assert ephys._repr_html_().strip().startswith("<style")

tests/integration/test_schema.py

@@ -56,11 +56,6 @@ def schema_empty(connection_test, schema_any, prefix):
     # Don't drop the schema since schema_any still needs it
 
 
-def test_schema_size_on_disk(schema_any):
-    number_of_bytes = schema_any.size_on_disk
-    assert isinstance(number_of_bytes, int)
-
-
 def test_schema_list(schema_any):
     schemas = dj.list_schemas(connection=schema_any.connection)
     assert schema_any.database in schemas