Improve foreign key checks performance(#254)

* An alternative method to test relations
Depends on a commit in datapackage-py
Proof of concept to be discussed

* refacto, passing FK index opti into tableschema

* adding documentation on new method and params

Author: paulgirard

README.md

@@ -206,21 +206,22 @@ Constructor to instantiate `Table` class. If `references` argument is provided,
 
 - `(Schema)` - returns schema class instance
 
-#### `table.iter(keyed=Fase, extended=False, cast=True, relations=False)`
+#### `table.iter(keyed=Fase, extended=False, cast=True, relations=False, foreign_keys_values=False)`
 
 Iterates through the table data and emits rows cast based on table schema. Data casting can be disabled.
 
 - `keyed (bool)` - iterate keyed rows
 - `extended (bool)` - iterate extended rows
 - `cast (bool)` - disable data casting if false
-- `relations (dict)` - dictionary of foreign key references in a form of `{resource1: [{field1: value1, field2: value2}, ...], ...}`. If provided, foreign key fields will checked and resolved to their references
+- `relations (dict)` - dictionary of foreign key references in a form of `{resource1: [{field1: value1, field2: value2}, ...], ...}`. If provided, foreign key fields will checked and resolved to one of their references (/!\ one-to-many fk are not completely resolved).
+- `foreign_keys_values (dict)` - three-level dictionary of foreign key references optimized to speed up validation process in a form of `{resource1: { (foreign_key_field1, foreign_key_field2) : { (value1, value2) : {one_keyedrow}, ... }}}`. If not provided but relations is true, it will be created before the validation process by *index_foreign_keys_values* method
 - `(exceptions.TableSchemaException)` - raises any error that occurs during this process
 - `(any[]/any{})` - yields rows:
   - `[value1, value2]` - base
   - `{header1: value1, header2: value2}` - keyed
   - `[rowNumber, [header1, header2], [value1, value2]]` - extended
 
-#### `table.read(keyed=False, extended=False, cast=True, relations=False, limit=None)`
+#### `table.read(keyed=False, extended=False, cast=True, relations=False, limit=None, foreign_keys_values=False)`
 
 Read the whole table and returns as array of rows. Count of rows could be limited.
 
@@ -229,6 +230,7 @@ Read the whole table and returns as array of rows. Count of rows could be limite
 - `cast (bool)` - flag to disable data casting if false
 - `relations (dict)` - dict of foreign key references in a form of `{resource1: [{field1: value1, field2: value2}, ...], ...}`. If provided foreign key fields will checked and resolved to its references
 - `limit (int)` - integer limit of rows to return
+- `foreign_keys_values (dict)` - three-level dictionary of foreign key references optimized to speed up validation process in a form of `{resource1: { (foreign_key_field1, foreign_key_field2) : { (value1, value2) : {one_keyedrow}, ... }}}`
 - `(exceptions.TableSchemaException)` - raises any error that occurs during this process
 - `(list[])` - returns array of rows (see `table.iter`)
 
@@ -252,6 +254,18 @@ Save data source to file locally in CSV format with `,` (comma) delimiter
 - `(exceptions.TableSchemaException)` - raises an error if there is saving problem
 - `(True/Storage)` - returns true or storage instance
 
+#### `table.index_foreign_keys_values(relations)`
+
+Creates a three-level dictionary of foreign key references optimized to speed up validation process in a form of `{resource1: { (foreign_key_field1, foreign_key_field2) : { (value1, value2) : {one_keyedrow}, ... }}}`. 
+For each foreign key of the schema it will iterate through the corresponding `relations['resource']` to create an index (i.e. a dict) of existing values for the foreign fields and store on keyed row for each value combination.
+The optimization relies on the indexation of possible values for one foreign key in a hashmap to later speed up resolution.
+This method is public to allow creating the index once to apply it on multiple tables charing the same schema (typically [grouped resources in datapackage](https://github.com/frictionlessdata/datapackage-py#group))
+Note 1: the second key of the output is a tuple of the foreign fields, a proxy identifier of the foreign key
+Note 2: the same relation resource can be indexed multiple times as a schema can contain more than one Foreign Keys pointing to the same resource
+
+- `relations (dict)` - dict of foreign key references in a form of `{resource1: [{field1: value1, field2: value2}, ...], ...}`. It must contain all resources pointed in the foreign keys schema definition.
+- `({resource1: { (foreign_key_field1, foreign_key_field2) : { (value1, value2) : {one_keyedrow}, ... }}})` - returns a three-level dictionary of foreign key references optimized to speed up validation process
+
 ### Schema
 
 A model of a schema with helpful methods for working with the schema and supported data. Schema instances can be initialized with a schema source as a url to a JSON file or a JSON object. The schema is initially validated (see [validate](#validate) below). By default validation errors will be stored in `schema.errors` but in a strict mode it will be instantly raised.

tableschema/table.py

@@ -11,6 +11,7 @@
 from .storage import Storage
 from .schema import Schema
 from . import exceptions
+from collections import defaultdict
 
 
 # Module API
@@ -65,7 +66,8 @@ def schema(self):
         """
         return self.__schema
 
-    def iter(self, keyed=False, extended=False, cast=True, relations=False):
+    def iter(self, keyed=False, extended=False, cast=True, relations=False,
+             foreign_keys_values=False):
         """https://github.com/frictionlessdata/tableschema-py#schema
         """
 
@@ -74,6 +76,11 @@ def iter(self, keyed=False, extended=False, cast=True, relations=False):
             unique_fields_cache = {}
             if self.schema:
                 unique_fields_cache = _create_unique_fields_cache(self.schema)
+        # Prepare relation checks
+        if relations and not foreign_keys_values:
+            # we have to test relations but the index has not been precomputed
+            # prepare the index to boost validation process
+            foreign_keys_values = self.index_foreign_keys_values(relations)
 
         # Open/iterate stream
         self.__stream.open()
@@ -110,11 +117,18 @@ def iter(self, keyed=False, extended=False, cast=True, relations=False):
                 if self.schema:
                     row_with_relations = dict(zip(headers, copy(row)))
                     for foreign_key in self.schema.foreign_keys:
-                        refValue = _resolve_relations(row, headers, relations, foreign_key)
+                        refValue = _resolve_relations(row, headers, foreign_keys_values,
+                                                      foreign_key)
                         if refValue is None:
                             self.__stream.close()
-                            message = 'Foreign key "%s" violation in row "%s"'
-                            message = message % (foreign_key['fields'], row_number)
+                            keyed_row = OrderedDict(zip(headers, row))
+                            # local values of the FK
+                            local_values = tuple(keyed_row[f] for f in foreign_key['fields'])
+                            message = 'Foreign key "%s" violation in row "%s": %s not found in %s'
+                            message = message % (foreign_key['fields'],
+                                                row_number,
+                                                local_values,
+                                                foreign_key['reference']['resource'])
                             raise exceptions.RelationError(message)
                         elif type(refValue) is dict:
                             for field in foreign_key['fields']:
@@ -124,6 +138,11 @@ def iter(self, keyed=False, extended=False, cast=True, relations=False):
                                 else:
                                     # alreayd one ref, merging
                                     row_with_relations[field].update(refValue)
+                        else:
+                            # case when all original value of the FK are empty
+                            # refValue == row, there is nothing to do
+                            # an empty dict might be a better returned value for this case ?
+                            pass
 
                     #  mutate row now that we are done, in the right order
                     row = [row_with_relations[f] for f in headers]
@@ -139,11 +158,13 @@ def iter(self, keyed=False, extended=False, cast=True, relations=False):
         # Close stream
         self.__stream.close()
 
-    def read(self, keyed=False, extended=False, cast=True, relations=False, limit=None):
+    def read(self, keyed=False, extended=False, cast=True, relations=False, limit=None,
+             foreign_keys_values=False):
         """https://github.com/frictionlessdata/tableschema-py#schema
         """
         result = []
-        rows = self.iter(keyed=keyed, extended=extended, cast=cast, relations=relations)
+        rows = self.iter(keyed=keyed, extended=extended, cast=cast, relations=relations,
+                         foreign_keys_values=foreign_keys_values)
         for count, row in enumerate(rows, start=1):
             result.append(row)
             if count == limit:
@@ -194,6 +215,32 @@ def save(self, target, storage=None, **options):
             storage.write(target, self.iter(cast=False))
             return storage
 
+    def index_foreign_keys_values(self, relations):
+        # we dont need to load the complete reference table to test relations
+        # we can lower payload AND optimize testing foreign keys
+        # by preparing the right index based on the foreign key definition
+        # foreign_keys are sets of tuples of all possible values in the foreign table
+        # foreign keys =
+        # [reference] [foreign_keys tuple] = { (foreign_keys_values, ) : one_keyedrow, ... }
+        foreign_keys = defaultdict(dict)
+        if self.schema:
+            for fk in self.schema.foreign_keys:
+                # load relation data
+                relation = fk['reference']['resource']
+
+                # create a set of foreign keys
+                # to optimize we prepare index of existing values
+                # this index should use reference + foreign_keys as key
+                # cause many foreign keys may use the same reference
+                foreign_keys[relation][tuple(fk['reference']['fields'])] = {}
+                for row in relations[relation]:
+                    key = tuple([row[foreign_field] for foreign_field in fk['reference']['fields']])
+                    # here we should chose to pick the first or nth row which match
+                    # previous implementation picked the first, so be it
+                    if key not in foreign_keys[relation][tuple(fk['reference']['fields'])]:
+                        foreign_keys[relation][tuple(fk['reference']['fields'])][key] = row
+        return foreign_keys
+
     # Private
 
     def __apply_processors(self, iterator, cast=True):
@@ -237,34 +284,21 @@ def _create_unique_fields_cache(schema):
     return cache
 
 
-def _resolve_relations(row, headers, relations, foreign_key):
+def _resolve_relations(row, headers, foreign_keys_values, foreign_key):
 
     # Prepare helpers - needed data structures
     keyed_row = OrderedDict(zip(headers, row))
-    fields = list(zip(foreign_key['fields'], foreign_key['reference']['fields']))
-    reference = relations.get(foreign_key['reference']['resource'])
-    if not reference:
-        # should an exception beeing raised here ?
-        return None
-
-    # Collect values - valid if all None
-    values = {}
-    empty_row = True
-    for field, ref_field in fields:
-        if field and ref_field:
-            values[ref_field] = keyed_row[field]
-            if keyed_row[field] is not None:
-                empty_row = False
-
-    # Resolve values - valid if match found
-    if not empty_row:
-        for refValues in reference:
-            if set(values.items()).issubset(set(refValues.items())):
-                # return the correct reference values
-                return refValues
-
-    if empty_row:
-        # return the orignal row if empty
-        return row
+    # local values of the FK
+    local_values = tuple(keyed_row[f] for f in foreign_key['fields'])
+    if len([l for l in local_values if l]) > 0:
+        # test existence into the foreign
+        relation = foreign_key['reference']['resource']
+        keys = tuple(foreign_key['reference']['fields'])
+        foreign_values = foreign_keys_values[relation][keys]
+        if local_values in foreign_values:
+            return foreign_values[local_values]
+        else:
+            return None
     else:
-        return None
+        # empty values for all keys, return original values
+        return row