roll: @akariv @cschloer
Please take a look
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

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)?
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(

pip install -e .[develop]
parse_log_pytest
requirements.txt
pytest
apt-get update apt-get install -y build-essential
3.7
requirements/base.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs==24.2.0
awesome-slugify==1.6.5
bitarray==2.9.3
bitstring==4.1.4
blessed==1.20.0
boto3==1.33.13
botocore==1.33.13
cached-property==1.5.2
cachetools==5.5.2
certifi @ file:///croot/certifi_1671487769961/work/certifi
cffi==1.15.1
chardet==5.2.0
charset-normalizer==3.4.1
click==8.1.8
colorama==0.4.6
cryptography==44.0.2
-e git+https://github.com/datahq/dataflows.git@13a848926c8416f5f976466752918d5d1163c47e#egg=dataflows
dataflows-tabulator==1.54.3
datapackage==1.15.4
distlib==0.3.9
et-xmlfile==1.1.0
exceptiongroup==1.2.2
filelock==3.12.2
greenlet==3.1.1
idna==3.10
ijson==3.3.0
importlib-metadata==6.7.0
importlib-resources==5.12.0
iniconfig==2.0.0
inquirer==2.10.1
isodate==0.7.2
Jinja2==3.1.6
jmespath==1.0.1
jsonlines==3.1.0
jsonpointer==3.0.0
jsonschema==4.17.3
kvfile==1.1.2
linear-tsv==1.1.0
MarkupSafe==2.1.5
mccabe==0.7.0
openpyxl==3.1.3
packaging==24.0
pkgutil_resolve_name==1.3.10
platformdirs==4.0.0
pluggy==1.2.0
pybloom-live==4.0.0
pycodestyle==2.10.0
pycparser==2.21
pydocstyle==6.1.1
pyflakes==3.0.1
pylama==8.4.1
pyproject-api==1.5.3
pyrsistent==0.19.3
pytest==7.4.4
python-dateutil==2.9.0.post0
python-editor==1.0.4
readchar==4.0.5
regex==2024.4.16
requests==2.31.0
rfc3986==2.0.0
s3transfer==0.8.2
six==1.17.0
snowballstemmer==2.2.0
SQLAlchemy==2.0.40
tableschema==1.21.0
tableschema-sql==2.0.1
tabulate==0.9.0
tabulator==1.53.5
tomli==2.0.1
tox==4.8.0
typing_extensions==4.7.1
unicodecsv==0.14.1
Unidecode==0.4.21
urllib3==1.26.20
virtualenv==20.26.6
wcwidth==0.2.13
xlrd==2.0.1
xmljson==0.2.1
xxhash==3.5.0
zipp==3.15.0

name: dataflows
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2022.12.7=py37h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=22.3.1=py37h06a4308_0
  - python=3.7.16=h7a1cb2a_0
  - readline=8.2=h5eee18b_0
  - setuptools=65.6.3=py37h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.38.4=py37h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - attrs==24.2.0
      - awesome-slugify==1.6.5
      - bitarray==2.9.3
      - bitstring==4.1.4
      - blessed==1.20.0
      - boto3==1.33.13
      - botocore==1.33.13
      - cached-property==1.5.2
      - cachetools==5.5.2
      - cffi==1.15.1
      - chardet==5.2.0
      - charset-normalizer==3.4.1
      - click==8.1.8
      - colorama==0.4.6
      - cryptography==44.0.2
      - dataflows-tabulator==1.54.3
      - datapackage==1.15.4
      - distlib==0.3.9
      - et-xmlfile==1.1.0
      - exceptiongroup==1.2.2
      - filelock==3.12.2
      - greenlet==3.1.1
      - idna==3.10
      - ijson==3.3.0
      - importlib-metadata==6.7.0
      - importlib-resources==5.12.0
      - iniconfig==2.0.0
      - inquirer==2.10.1
      - isodate==0.7.2
      - jinja2==3.1.6
      - jmespath==1.0.1
      - jsonlines==3.1.0
      - jsonpointer==3.0.0
      - jsonschema==4.17.3
      - kvfile==1.1.2
      - linear-tsv==1.1.0
      - markupsafe==2.1.5
      - mccabe==0.7.0
      - openpyxl==3.1.3
      - packaging==24.0
      - pkgutil-resolve-name==1.3.10
      - platformdirs==4.0.0
      - pluggy==1.2.0
      - pybloom-live==4.0.0
      - pycodestyle==2.10.0
      - pycparser==2.21
      - pydocstyle==6.1.1
      - pyflakes==3.0.1
      - pylama==8.4.1
      - pyproject-api==1.5.3
      - pyrsistent==0.19.3
      - pytest==7.4.4
      - python-dateutil==2.9.0.post0
      - python-editor==1.0.4
      - readchar==4.0.5
      - regex==2024.4.16
      - requests==2.31.0
      - rfc3986==2.0.0
      - s3transfer==0.8.2
      - six==1.17.0
      - snowballstemmer==2.2.0
      - sqlalchemy==2.0.40
      - tableschema==1.21.0
      - tableschema-sql==2.0.1
      - tabulate==0.9.0
      - tabulator==1.53.5
      - tomli==2.0.1
      - tox==4.8.0
      - typing-extensions==4.7.1
      - unicodecsv==0.14.1
      - unidecode==0.04.21
      - urllib3==1.26.20
      - virtualenv==20.26.6
      - wcwidth==0.2.13
      - xlrd==2.0.1
      - xmljson==0.2.1
      - xxhash==3.5.0
      - zipp==3.15.0
prefix: /opt/conda/envs/dataflows

tests/test_lib.py::test_validate tests/test_lib.py::test_join_preserve_source_fields_order
tests/test_examples.py::test_example_3 tests/test_examples.py::test_example_4 tests/test_examples.py::test_example_5 tests/test_lib.py::test_dump_to_sql
tests/test_examples.py::test_example_1 tests/test_examples.py::test_example_2 tests/test_examples.py::test_example_6 tests/test_examples.py::test_validate tests/test_examples.py::test_example_7 tests/test_examples.py::test_example_75 tests/test_examples.py::test_example_8 tests/test_examples.py::test_example_9 tests/test_examples.py::test_rename_resource tests/test_examples.py::test_rename_resource2 tests/test_lib.py::test_add_computed_field tests/test_lib.py::test_add_computed_field_func tests/test_lib.py::test_add_metadata tests/test_lib.py::test_delete_field tests/test_lib.py::test_select_field tests/test_lib.py::test_find_replace tests/test_lib.py::test_unpivot tests/test_lib.py::test_unpivot_any_resources tests/test_lib.py::test_concatenate tests/test_lib.py::test_concatenate_multifield tests/test_lib.py::test_filter_rows tests/test_lib.py::test_filter_rows_callable tests/test_lib.py::test_sort_rows tests/test_lib.py::test_sort_reverse_many_rows tests/test_lib.py::test_sort_rows_number tests/test_lib.py::test_sort_rows_decimal tests/test_lib.py::test_sort_rows_datetime tests/test_lib.py::test_deduplicate tests/test_lib.py::test_duplicate tests/test_lib.py::test_duplicate_many_rows tests/test_lib.py::test_flow_as_step tests/test_lib.py::test_load_from_package tests/test_lib.py::test_load_from_package_zip tests/test_lib.py::test_load_from_env_var tests/test_lib.py::test_load_from_package_resource_matching tests/test_lib.py::test_load_strategies tests/test_lib.py::test_load_strategy_infer_strings_from_native_types tests/test_lib.py::test_load_name_path tests/test_lib.py::test_load_from_package_resources tests/test_lib.py::test_checkpoint tests/test_lib.py::test_load_from_checkpoint tests/test_lib.py::test_update_resource tests/test_lib.py::test_update_schema tests/test_lib.py::test_set_type_resources tests/test_lib.py::test_set_type_errors tests/test_lib.py::test_dump_to_path_use_titles tests/test_lib.py::test_load_dates tests/test_lib.py::test_load_dates_timezones tests/test_lib.py::test_add_field tests/test_lib.py::test_load_empty_headers tests/test_lib.py::test_load_xml tests/test_lib.py::test_save_load_dates tests/test_lib.py::test_stream_simple tests/test_lib.py::test_set_primary_key tests/test_lib.py::test_join tests/test_lib.py::test_load_limit_rows tests/test_lib.py::test_set_type_regex tests/test_lib.py::test_load_override_schema tests/test_lib.py::test_load_override_schema_and_fields tests/test_lib.py::test_delete_fields_regex tests/test_lib.py::test_join_full_outer tests/test_lib.py::test_join_row_number tests/test_lib.py::test_join_row_number_readme_example tests/test_lib.py::test_join_row_number_format_string tests/test_lib.py::test_load_duplicate_headers tests/test_lib.py::test_load_duplicate_headers_with_deduplicate_headers_flag tests/test_lib.py::test_force_temporal_format tests/test_lib.py::test_extract_missing_values tests/test_lib.py::test_extract_missing_values_options tests/test_lib.py::test_extract_missing_values_options_source_is_list tests/test_lib.py::test_conditional tests/test_lib.py::test_finalizer
data/cities_metadata.csv PROCESSORS.md dataflows/base/schema_validator.py tox.ini dataflows/processors/join.py dataflows/processors/validate.py dataflows/VERSION
data/cities_metadata.csv PROCESSORS.md dataflows/base/schema_validator.py tox.ini dataflows/processors/join.py dataflows/processors/validate.py dataflows/VERSION