External storage, enhancements, bugfixes

datajoint/autopopulate.py

@@ -2,8 +2,9 @@
 import logging
 import datetime
 import random
+from tqdm import tqdm
 from pymysql import OperationalError
-from .relational_operand import RelationalOperand, AndList
+from .relational_operand import RelationalOperand, AndList, U
 from . import DataJointError
 from . import key as KEY
 from .base_relation import FreeRelation
@@ -63,7 +64,7 @@ def _job_key(self, key):
         """
         return key
 
-    def populate(self, *restrictions, suppress_errors=False, reserve_jobs=False, order="original", limit=None):
+    def populate(self, *restrictions, suppress_errors=False, reserve_jobs=False, order="original", limit=None, report_progress=False):
         """
         rel.populate() calls rel._make_tuples(key) for every primary key in self.key_source
         for which there is not already a tuple in rel.
@@ -73,6 +74,7 @@ def populate(self, *restrictions, suppress_errors=False, reserve_jobs=False, ord
         :param reserve_jobs: if true, reserves job to populate in asynchronous fashion
         :param order: "original"|"reverse"|"random"  - the order of execution
         :param limit: if not None, populates at max that many keys
+        :param report_progress: if True, report progress_bar
         """
         if self.connection.in_transaction:
             raise DataJointError('Populate cannot be called during a transaction.')
@@ -84,10 +86,12 @@ def populate(self, *restrictions, suppress_errors=False, reserve_jobs=False, ord
         todo = self.key_source
         if not isinstance(todo, RelationalOperand):
             raise DataJointError('Invalid key_source value')
-        todo = todo.proj() & AndList(restrictions)
+        todo = (todo & AndList(restrictions)).proj()
+        if any(name not in self.target.heading for name in todo.heading):
+            raise DataJointError('The populated target must have all the attributes of the key source')
+        todo -= self.target
 
         error_list = [] if suppress_errors else None
-
         jobs = self.connection.jobs[self.target.database] if reserve_jobs else None
 
         # define and setup signal handler for SIGTERM
@@ -97,15 +101,14 @@ def handler(signum, frame):
                 raise SystemExit('SIGTERM received')
             old_handler = signal.signal(signal.SIGTERM, handler)
 
-        todo -= self.target
         keys = todo.fetch(KEY, limit=limit)
         if order == "reverse":
             keys.reverse()
         elif order == "random":
             random.shuffle(keys)
 
         logger.info('Found %d keys to populate' % len(keys))
-        for key in keys:
+        for key in (tqdm(keys) if report_progress else keys):
             if not reserve_jobs or jobs.reserve(self.target.table_name, self._job_key(key)):
                 self.connection.start_transaction()
                 if key in self.target:  # already populated
@@ -148,6 +151,8 @@ def progress(self, *restrictions, display=True):
         :return: remaining, total -- tuples to be populated
         """
         todo = self.key_source & AndList(restrictions)
+        if any(name not in self.target.heading for name in todo.heading):
+            raise DataJointError('The populated target must have all the attributes of the key source')
         total = len(todo)
         remaining = len(todo.proj() - self.target)
         if display:

datajoint/base_relation.py

@@ -139,11 +139,11 @@ def insert(self, rows, replace=False, ignore_errors=False, skip_duplicates=False
 
         if isinstance(rows, RelationalOperand):
             # INSERT FROM SELECT - build alternate field-narrowing query (only) when needed
-            if ignore_extra_fields and not all(name in self.heading.names for name in rows.heading.names):
+            if ignore_extra_fields and not all(name in self.heading for name in rows.heading):
                 query = 'INSERT{ignore} INTO {table} ({fields}) SELECT {fields} FROM ({select}) as `__alias`'.format(
                 ignore=" IGNORE" if ignore_errors or skip_duplicates else "",
                 table=self.full_table_name,
-                fields='`'+'`,`'.join(self.heading.names)+'`',
+                fields='`'+'`,`'.join(name for name in self.heading if name in rows.heading) + '`',
                 select=rows.make_sql())
             else:
                 query = 'INSERT{ignore} INTO {table} ({fields}) {select}'.format(
@@ -155,6 +155,7 @@ def insert(self, rows, replace=False, ignore_errors=False, skip_duplicates=False
                 self.connection.query(query)
             except pymysql.err.InternalError as err:
                 if err.args[0] == server_error_codes['unknown column']:
+                    print(query)
                     # args[1] -> Unknown column 'extra' in 'field list'
                     raise DataJointError('%s : To ignore extra fields, set ignore_extra_fields=True in insert.' % err.args[1])
                 else:

datajoint/relational_operand.py

@@ -1,5 +1,5 @@
 import collections
-from itertools import zip_longest
+from itertools import zip_longest, count
 import logging
 import numpy as np
 import re
@@ -237,6 +237,12 @@ def __mul__(self, other):
         """
         return other * self if isinstance(other, U) else Join.create(self, other)
 
+    def __add__(self, other):
+        """
+        union of relations
+        """
+        return Union.create(self, other)
+
     def proj(self, *attributes, **named_attributes):
         """
         Relational projection operator.
@@ -606,9 +612,57 @@ def from_clause(self):
             from2=self._arg2.from_clause)
 
 
+class Union(RelationalOperand):
+    """
+    Union is a private DataJoint class that implements relational union.
+    """
+
+    __count = count()
+
+    def __init__(self, arg=None):
+        super().__init__(arg)
+        if arg is not None:
+            assert isinstance(arg, Union), "Union copy constructore requires a Union object"
+            self._connection = arg.connection
+            self._heading = arg.heading
+            self._arg1 = arg._arg1
+            self._arg2 = arg._arg2
+
+    @classmethod
+    def create(cls, arg1, arg2):
+        obj = cls()
+        if not isinstance(arg1, RelationalOperand) or not isinstance(arg2, RelationalOperand):
+            raise DataJointError('a relation can only be unioned with another relation')
+        if arg1.connection != arg2.connection:
+            raise DataJointError("Cannot operate on relations from different connections.")
+        if set(arg1.heading.names) != set(arg2.heading.names):
+            raise DataJointError('Union requires the same attributes in both arguments')
+        if not all(v.in_key for v in arg1.heading.attributes.values()):
+            raise DataJointError('The left argument of union must not have any secondary attributes')
+        obj._connection = arg1.connection
+        obj._heading = arg1.heading
+        obj._arg1 = arg1
+        obj._arg2 = arg2
+        return obj
+
+    def make_sql(self, select_fields=None):
+        return "SELECT {_fields} FROM {_from}{_where}".format(
+            _fields=self.get_select_fields(select_fields),
+            _from=self.from_clause,
+            _where=self.where_clause)
+
+    @property
+    def from_clause(self):
+        return ("(SELECT {fields} FROM {from1}{where1} UNION SELECT {fields} FROM {from2}{where2}) as `_u%x`".format(
+            fields=self.get_select_fields(None), from1=self._arg1.from_clause,
+            where1=self._arg1.where_clause,
+            from2=self._arg2.from_clause,
+            where2=self._arg2.where_clause)) % next(self.__count)
+
+
 class Projection(RelationalOperand):
     """
-    Projection is an private DataJoint class that implements relational projection.
+    Projection is a private DataJoint class that implements relational projection.
     See RelationalOperand.proj() for user interface.
     """
 
@@ -723,7 +777,7 @@ class Subquery(RelationalOperand):
     The attribute list and the WHERE clause are resolved.  Thus, a subquery no longer has any renamed attributes.
     A subquery of a subquery is a just a copy of the subquery with no change in SQL.
     """
-    __counter = 0
+    __count = count()
 
     def __init__(self, arg=None):
         super().__init__(arg)
@@ -745,14 +799,9 @@ def create(cls, arg):
         obj._arg = arg
         return obj
 
-    @property
-    def counter(self):
-        Subquery.__counter += 1
-        return Subquery.__counter
-
     @property
     def from_clause(self):
-        return '(' + self._arg.make_sql() + ') as `_s%x`' % self.counter
+        return '(' + self._arg.make_sql() + ') as `_s%x`' % next(self.__count)
 
     def get_select_fields(self, select_fields=None):
         return '*' if select_fields is None else self.heading.project(select_fields).as_sql

requirements.txt

@@ -2,5 +2,6 @@ numpy
 pymysql>=0.7.2
 pyparsing
 ipython
+tqdm
 networkx~=1.11
 pydotplus

test_requirements.txt

@@ -1,2 +1,3 @@
 matplotlib
 pygraphviz
+tqdm