draft implementation of autopopulate2.0

Author: dimitri-yatsenko

datajoint/autopopulate.py

@@ -15,7 +15,7 @@
 
 from .errors import DataJointError, LostConnectionError
 from .expression import AndList, QueryExpression
-from .jobs import JobTable2
+from .jobs import JobTable
 
 # noinspection PyExceptionInherit,PyCallingNonCallable
 
@@ -276,8 +276,7 @@ def populate(
             "error_list": the error list that is filled if `suppress_errors` is True
         """
         if self.connection.in_transaction:
-            raise DataJointError(
-                "Populate cannot be called during a transaction.")
+            raise DataJointError("Populate cannot be called during a transaction.")
 
         valid_order = ["original", "reverse", "random"]
         if order not in valid_order:
@@ -349,8 +348,7 @@ def handler(signum, frame):
                 del self.connection._conn.ctx  # SSLContext is not pickleable
                 with (
                     mp.Pool(
-                        processes, _initialize_populate, (
-                            self, jobs, populate_kwargs)
+                        processes, _initialize_populate, (self, jobs, populate_kwargs)
                     ) as pool,
                     (
                         tqdm(desc="Processes: ", total=nkeys)
@@ -396,8 +394,7 @@ def _populate1(
             True if successfully invoke one `make()` call, otherwise False
         """
         # use the legacy `_make_tuples` callback.
-        make = self._make_tuples if hasattr(
-            self, "_make_tuples") else self.make
+        make = self._make_tuples if hasattr(self, "_make_tuples") else self.make
 
         if reserve_jobs and not self.jobs.reserve(key):
             # unable to reserve the key for the job
@@ -467,8 +464,7 @@ def _populate1(
                 return key, error if return_exception_objects else error_message
         else:
             self.connection.commit_transaction()
-            logger.debug(
-                f"Success making {key} -> {self.target.full_table_name}")
+            logger.debug(f"Success making {key} -> {self.target.full_table_name}")
             if jobs is not None:
                 run_duration = time.time() - start_time
                 if jobs is not None:

datajoint/jobs.py

@@ -19,11 +19,7 @@ class Job(Table):
     def __init__(self, computed_table: Table):
         """
         Initialize a job table for a specific computed table.
-
-        :param computed_table: Database connection
-        :param database: Database name
-        :param table_name: Name of the computed table
-        :param primary_key_attrs: List of primary key attribute names
+        :param computed_table: AutoPopulate instance
         """
         self.database = computed_table.database
         self._connection = computed_table.connection
@@ -37,30 +33,29 @@ def __init__(self, computed_table: Table):
                 conn=computed_table.connection,
                 database=computed_table.database,
                 table_name=self.table_name,
-                context=None
+                context=None,
             )
         )
 
         self._definition = f"# job table for {self.computed_table.full_table_name}\n"
 
-        # add references to parent tables to the definition
+        # add primary foreign key references to parent tables to the definition
         parents = computed_table.parents(
-            as_objects=False, primary_key=True, foreign_key_info=True)
+            as_objects=False, primary_key=True, foreign_key_info=True
+        )
         for parent_name, fk_props in parents:
             if not fk_props["aliased"]:
                 # simple foreign key
                 self._definition += f"->{parent_name}\n"
             else:
                 # projected foreign key
-                self._definition += (
-                    "->{parent_name}.proj({proj_list})\n".format(
-                        parent_name=parent_name,
-                        proj_list=",".join(
-                            '{}="{}"'.format(attr, ref)
-                            for attr, ref in fk_props["attr_map"].items()
-                            if ref != attr
-                        )
-                    )
+                self._definition += "->{parent_name}.proj({proj_list})\n".format(
+                    parent_name=parent_name,
+                    proj_list=",".join(
+                        '{}="{}"'.format(attr, ref)
+                        for attr, ref in fk_props["attr_map"].items()
+                        if ref != attr
+                    ),
                 )
         self._definition += """
         ---
@@ -105,22 +100,22 @@ def refresh(self, key_source):
         :param key_source: QueryExpression that defines available jobs
         """
         # Get current jobs that should be preserved (reserved, error, ignore)
-        preserve_statuses = ('reserved', 'error', 'ignore')
-        existing_jobs = (self & f'status in {preserve_statuses}').fetch('KEY')
+        preserve_statuses = ("reserved", "error", "ignore")
+        existing_jobs = (self & f"status in {preserve_statuses}").fetch("KEY")
 
         # Get all keys from key_source
-        available_keys = key_source.fetch('KEY')
+        available_keys = key_source.fetch("KEY")
 
         # Remove jobs that are no longer in key_source
         for job_key in existing_jobs:
             if job_key not in available_keys:
                 (self & job_key).delete_quick()
 
         # Add new jobs that aren't already in the table
-        existing_all = self.fetch('KEY')
+        existing_all = self.fetch("KEY")
         for key in available_keys:
             if key not in existing_all:
-                self.insert1(dict(key, status='scheduled', priority=3))
+                self.insert1(dict(**key, status="scheduled", priority=3))
 
     def reserve(self, key):
         """
@@ -130,7 +125,7 @@ def reserve(self, key):
         :return: True if reserved job successfully. False = the job is already taken
         """
         job = dict(
-            key,
+            **key,
             status="reserved",
             host=platform.node(),
             pid=os.getpid(),
@@ -151,7 +146,7 @@ def ignore(self, key):
         :return: True if ignore job successfully. False = the job is already taken
         """
         job = dict(
-            key,
+            **key,
             status="ignore",
             host=platform.node(),
             pid=os.getpid(),
@@ -173,7 +168,7 @@ def complete(self, key, run_duration=None, run_version=None):
         :param run_version: version information (e.g., git commit hash)
         """
         job = dict(
-            key,
+            **key,
             status="success",
             run_duration=run_duration,
             run_version=json.dumps(run_version) if run_version else None,
@@ -194,13 +189,12 @@ def error(self, key, error_message, error_stack=None):
         """
         if len(error_message) > ERROR_MESSAGE_LENGTH:
             error_message = (
-                error_message[: ERROR_MESSAGE_LENGTH -
-                              len(TRUNCATION_APPENDIX)]
+                error_message[: ERROR_MESSAGE_LENGTH - len(TRUNCATION_APPENDIX)]
                 + TRUNCATION_APPENDIX
             )
 
         job = dict(
-            key,
+            **key,
             status="error",
             error_message=error_message,
             error_stack=error_stack,
@@ -227,12 +221,12 @@ def get_scheduled_jobs(self, limit=None, order_by=None):
             query = query.fetch(order_by=order_by)
         else:
             # Default ordering: priority (ascending), then timestamp (ascending)
-            query = query.fetch(order_by=['priority', 'timestamp'])
+            query = query.fetch(order_by=["priority", "timestamp"])
 
         if limit:
             query = query[:limit]
 
-        return query.fetch('KEY')
+        return query.fetch("KEY")
 
     def set_priority(self, key, priority):
         """
@@ -241,4 +235,4 @@ def set_priority(self, key, priority):
         :param key: the dict of the job's primary key
         :param priority: priority value (lower = higher priority)
         """
-        self.update1(dict(key, priority=priority))
+        self.update1(dict(**key, priority=priority))

docs/src/compute/distributed.md

@@ -2,12 +2,9 @@
 
 ## Job reservations
 
-Running `populate` on the same table on multiple computers will causes them to attempt
+Running `populate` on the same table on multiple computers will cause them to attempt
 to compute the same data all at once.
 This will not corrupt the data since DataJoint will reject any duplication.
-One solution could be to cause the different computing nodes to populate the tables in
-random order.
-This would reduce some collisions but not completely prevent them.
 
 To allow efficient distributed computing, DataJoint provides a built-in job reservation
 process.

docs/src/concepts/data-model.md

@@ -36,23 +36,22 @@ objects in memory with properties and methods for transformations of such data.
 ## Relational data model
 
 The **relational model** is a way of thinking about data as sets and operations on sets.
-Formalized almost a half-century ago ([Codd,
-1969](https://dl.acm.org/citation.cfm?doid=362384.362685)). The relational data model is
-one of the most powerful and precise ways to store and manage structured data. At its
-core, this model organizes all data into tables--representing mathematical
-relations---where each table consists of rows (representing mathematical tuples) and
-columns (often called attributes).
+Formalized half a century ago by Edgar F. Codd (@10.1145/362384.36268), the relational data model is
+one of the most powerful and precise ways to define and manage structured data. At its
+core, this model organizes all data into tables---representing mathematical
+relations---where each table consists of rows (mathematical tuples) and
+columns (attributes).
 
 ### Core principles of the relational data model
 
 **Data representation:**
-  Data are represented and manipulated in the form of relations.
+  Data are represented and manipulated in the form of relations (mathematical sets).
   A relation is a set (i.e. an unordered collection) of entities of values for each of
   the respective named attributes of the relation.
   Base relations represent stored data while derived relations are formed from base
-  relations through query expressions.
+  relations through relational operators.
   A collection of base relations with their attributes, domain constraints, uniqueness
-  constraints, and referential constraints is called a schema.
+  constraints, and referential constraints is called a [schema](../design/schema.md).
 
 **Domain constraints:**
   Each attribute (column) in a table is associated with a specific attribute domain (or