Repository: datahq/dataflows

Issue:
Don't sort keys before adding in `join` processor
Hi,

I know the new fields object inputed to join is a dictionary so order is never guaranteed. However, more often than not the order from the pipeline-spec.yaml is preserved when it is loaded in and passed on the dataflows. Is it possible to have an option in dataflows `join` that does NOT sort the keys (here https://github.com/datahq/dataflows/blob/master/dataflows/processors/join.py#L280)?

Discussion:
roll: @akariv @cschloer
Please take a look

Patch:
diff --git a/PROCESSORS.md b/PROCESSORS.md
index 8ee2244..fbbae46 100644
--- a/PROCESSORS.md
+++ b/PROCESSORS.md
@@ -480,11 +480,16 @@ def set_type(name, resources=-1, regex=True, on_error=None, **options):
   - The index of the resource in the package
 - `regex` - if set to `False` field names will be interpreted as strings not as regular expressions (`True` by default)
 - `on_error` - callback function to be called when a validation error occurs.
-  Function has the signature `callback(resource_name, row, row_index, exception)`.
-  It could raise an exception, or return `True` (for keeping the row anyway) or `False` (for dropping it).
+
+  Function can have the signature
+  - `callback(resource_name, row, row_index, exception)` - will receive the resource name, the contents of the offending row, its index and the raised exception
+  - `callback(resource_name, row, row_index, exception, field)` - will receive the resource name, the contents of the offending row, its index and the raised exception and the offending field (as a Field object)
+  
+  This function may raise an exception, return `True` for keeping the row anyway or `False` for dropping it.
+
   A few predefined options are:
   - `dataflows.base.schema_validator.raise_exception` - the default behaviour, will raise a `dataflows.ValidationError` exception.
-  - `dataflows.base.schema_validator.ignore` - drop invalid rows
+  - `dataflows.base.schema_validator.drop` - drop invalid rows
   - `dataflows.base.schema_validator.ignore` - ignore all errors
 - `options` - options to set for the field. Most common ones would be:
   - `type` - set the data type (e.g. `string`, `integer`, `number` etc.)
diff --git a/data/cities_metadata.csv b/data/cities_metadata.csv
new file mode 100644
index 0000000..337a8bb
--- /dev/null
+++ b/data/cities_metadata.csv
@@ -0,0 +1,4 @@
+id,key2,key1
+1,val2,val1
+2,val2,val1
+3,val2,val1
diff --git a/dataflows/VERSION b/dataflows/VERSION
index 6e8bf73..17e51c3 100644
--- a/dataflows/VERSION
+++ b/dataflows/VERSION
@@ -1,1 +1,1 @@
-0.1.0
+0.1.1
diff --git a/dataflows/base/schema_validator.py b/dataflows/base/schema_validator.py
index 7cb643f..cc4fba5 100644
--- a/dataflows/base/schema_validator.py
+++ b/dataflows/base/schema_validator.py
@@ -1,3 +1,5 @@
+from inspect import signature
+
 from datapackage import Resource
 from tableschema import Schema
 from tableschema.exceptions import CastError
@@ -30,10 +32,22 @@ def drop(res_name, row, i, e):
     return False
 
 
+def wrap_handler(on_error):
+    assert callable(on_error)
+    if len(list(signature(on_error).parameters)) > 4:
+        return on_error
+
+    def func(res_name, row, i, e, _):
+        return on_error(res_name, row, i, e)
+    return func
+
+
 def schema_validator(resource, iterator,
                      field_names=None, on_error=None):
     if on_error is None:
         on_error = raise_exception
+    on_error = wrap_handler(on_error)
+
     if isinstance(resource, Resource):
         schema: Schema = resource.schema
         assert schema is not None
@@ -44,11 +58,12 @@ def schema_validator(resource, iterator,
         field_names = [f.name for f in schema.fields]
     schema_fields = [f for f in schema.fields if f.name in field_names]
     for i, row in enumerate(iterator):
+        field = None
         try:
-            for f in schema_fields:
-                row[f.name] = f.cast_value(row.get(f.name))
+            for field in schema_fields:
+                row[field.name] = field.cast_value(row.get(field.name))
         except CastError as e:
-            if not on_error(resource['name'], row, i, e):
+            if not on_error(resource['name'], row, i, e, field):
                 continue
 
         yield row
diff --git a/dataflows/processors/join.py b/dataflows/processors/join.py
index 1853223..37dbd00 100644
--- a/dataflows/processors/join.py
+++ b/dataflows/processors/join.py
@@ -124,6 +124,7 @@ AGGREGATORS = {
 
 
 # Input helpers
+
 def fix_fields(fields):
     for field in sorted(fields.keys()):
         spec = fields[field]
@@ -147,6 +148,17 @@ def expand_fields(fields, schema_fields):
                 fields[sf_name]['name'] = sf_name
 
 
+def order_fields(fields, schema_fields):
+    ordered_fields = collections.OrderedDict()
+    for descriptor in schema_fields:
+        name = descriptor['name']
+        if name in fields:
+            ordered_fields[name] = fields.pop(name)
+    for name in sorted(fields.keys()):
+        ordered_fields[name] = fields[name]
+    return ordered_fields
+
+
 def concatenator(resources, all_target_fields, field_mapping):
     for resource_ in resources:
         for row in resource_:
@@ -277,9 +289,7 @@ def join_aux(source_name, source_key, source_delete,  # noqa: C901
     def process_target_resource(source_spec, resource):
         target_fields = \
             resource.setdefault('schema', {}).setdefault('fields', [])
-        added_fields = sorted(fields.keys())
-        for field in added_fields:
-            spec = fields[field]
+        for name, spec in fields.items():
             agg = spec['aggregate']
             data_type = AGGREGATORS[agg].dataType
             copy_properties = AGGREGATORS[agg].copyProperties
@@ -297,13 +307,13 @@ def join_aux(source_name, source_key, source_delete,  # noqa: C901
                 data_type = source_field['type']
             try:
                 existing_field = next(iter(filter(
-                    lambda f: f['name'] == field,
+                    lambda f: f['name'] == name,
                     target_fields)))
                 assert existing_field['type'] == data_type, \
-                    'Reusing %s but with different data types: %s != %s' % (field, existing_field['type'], data_type)
+                    'Reusing %s but with different data types: %s != %s' % (name, existing_field['type'], data_type)
             except StopIteration:
                 to_copy.update({
-                    'name': field,
+                    'name': name,
                     'type': data_type
                 })
                 target_fields.append(to_copy)
@@ -326,8 +336,11 @@ def join_aux(source_name, source_key, source_delete,  # noqa: C901
         for resource in datapackage['resources']:
 
             if resource['name'] == source_name:
+                nonlocal fields
                 source_spec = resource
-                expand_fields(fields, source_spec.get('schema', {}).get('fields', []))
+                schema_fields = source_spec.get('schema', {}).get('fields', [])
+                expand_fields(fields, schema_fields)
+                fields = order_fields(fields, schema_fields)
                 if not source_delete:
                     new_resources.append(resource)
                 if deduplication:
diff --git a/dataflows/processors/validate.py b/dataflows/processors/validate.py
index 89ed01a..75a7cfc 100644
--- a/dataflows/processors/validate.py
+++ b/dataflows/processors/validate.py
@@ -1,7 +1,7 @@
 from inspect import isfunction
 
 from .. import DataStreamProcessor, schema_validator, ResourceWrapper
-from ..base.schema_validator import raise_exception
+from ..base.schema_validator import raise_exception, wrap_handler
 from ..helpers import ResourceMatcher
 
 
@@ -11,7 +11,7 @@ class validate(DataStreamProcessor):
         super(validate, self).__init__()
         if on_error is None:
             on_error = raise_exception
-        self.on_error = on_error
+        self.on_error = wrap_handler(on_error)
         self.resources = resources
         if len(args) == 2:
             field, validator = args
@@ -40,7 +40,7 @@ class validate(DataStreamProcessor):
             res_name = rows.res.name
             for i, row in enumerate(rows):
                 if not row_validator(row):
-                    if not self.on_error(res_name, row, i, None):
+                    if not self.on_error(res_name, row, i, None, None):
                         continue
                 yield row
         return func
diff --git a/tox.ini b/tox.ini
index 22f82fe..22c2308 100644
--- a/tox.ini
+++ b/tox.ini
@@ -10,6 +10,7 @@ deps=
   pytest
   pytest-cov
   coverage
+  lxml
   py37-plyvel: plyvel
 passenv=
   CI


Test patch:
diff --git a/tests/test_examples.py b/tests/test_examples.py
index 82b11fb..0cce914 100644
--- a/tests/test_examples.py
+++ b/tests/test_examples.py
@@ -35,10 +35,11 @@ def test_example_2():
 
 
 def country_population():
-    from xml.etree import ElementTree
+    from lxml import etree
     from urllib.request import urlopen
     page = urlopen('https://en.wikipedia.org/wiki/List_of_countries_and_dependencies_by_population').read()
-    tree = ElementTree.fromstring(page)
+    parser = etree.HTMLParser()
+    tree = etree.fromstring(page, parser=parser)
     tables = tree.findall('.//table')
     for table in tables:
         if 'wikitable' in table.attrib.get('class', ''):
diff --git a/tests/test_lib.py b/tests/test_lib.py
index f382f59..91bfbf9 100644
--- a/tests/test_lib.py
+++ b/tests/test_lib.py
@@ -1,6 +1,7 @@
 import pytest
 from dataflows import Flow
 
+
 data = [
     dict(x=1, y='a'),
     dict(x=2, y='b'),
@@ -1146,6 +1147,15 @@ def test_validate():
             self.bad_row, self.bad_index = row, i
             return False
 
+    class on_error_fields():
+        def __init__(self):
+            self.bad_row, self.bad_index, self.bad_field = None, None, None
+
+        def __call__(self, name, row, i, e, field):
+            self.bad_row, self.bad_index = row, i
+            self.bad_field = field
+            return False
+
     # Schema validator
     handler = on_error()
     res, *_ = Flow(
@@ -1157,6 +1167,18 @@ def test_validate():
     assert handler.bad_row == {'a': 4, 'b': 'a'}
     assert handler.bad_index == 3
 
+    # Schema validator with fields
+    handler = on_error_fields()
+    res, *_ = Flow(
+        data,
+        set_type('b', type='integer', on_error=ignore),
+        validate(on_error=handler)
+    ).results()
+    assert len(res[0]) == 3
+    assert handler.bad_row == {'a': 4, 'b': 'a'}
+    assert handler.bad_index == 3
+    assert handler.bad_field.name == 'b'
+
     # Field validator
     handler = on_error()
     res, *_ = Flow(
@@ -1515,6 +1537,33 @@ def test_join_row_number_format_string():
     ]]
 
 
+def test_join_preserve_source_fields_order():
+    from dataflows import load, join
+    flow = Flow(
+        load('data/cities_metadata.csv'),
+        load('data/cities.csv'),
+        join(
+            source_name='cities_metadata',
+            source_key='{id}',
+            target_name='cities',
+            target_key='{id}',
+            fields={'key1': {'name': 'key1'}, 'key2': {'name': 'key2'}}
+        ),
+    )
+    data, package, stats = flow.results()
+    assert package.descriptor['resources'][0]['schema']['fields'] == [
+        {'name': 'id', 'type': 'integer', 'format': 'default'},
+        {'name': 'city', 'type': 'string', 'format': 'default'},
+        {'name': 'key2', 'type': 'string', 'format': 'default'},
+        {'name': 'key1', 'type': 'string', 'format': 'default'}
+    ]
+    assert data == [[
+        {'id': 1, 'city': 'london', 'key1': 'val1', 'key2': 'val2'},
+        {'id': 2, 'city': 'paris', 'key1': 'val1', 'key2': 'val2'},
+        {'id': 3, 'city': 'rome', 'key1': 'val1', 'key2': 'val2'},
+    ]]
+
+
 def test_load_duplicate_headers():
     from dataflows import load, exceptions
     flow = Flow(