Repository: jina-ai/serve
Pull request: feat: pass executor specific params


┆Issue is synchronized with this [Asana task](https://app.asana.com/0/1200479679796904/1200498158412714) by [Unito](https://www.unito.io)


Problem statement:
Define parameters on executor level
Assume two encoders share parameter names - like batch_size or traversal_path.
When sending the following request, it is not possible to have different batch sizes or traversal_path for both of the  encoders.


```
results = f.post(
    on='/search',
    parameters={'batch_size': 32},
...
```



How to best pass parameters to Executors?
In Jina 2.0 the traversal path (and other parameters) are moved into the hands of the Executor developers.

But we still want to provide the best way to pass these parameters, either to the `flow.post()`, or as `__init__` arguments. Optionally, these could also be some decorators on the Executor methods.

This issue is meant as a place to gather feedback from you (our community) about how you are using parameters in your Executors.

A common pattern I have been using is 

1 **Defaults in constructor and per-request overrides**

```python

class MyExecutor(Executor):

    def __init__(
            self,
            traversal_path: list = 'r',
            **kwargs,
    ):
        super().__init__(**kwargs)
        self.default_traversal_path = traversal_path

...

   @requests(on='/index')
   def index(self, docs, params, ...):
       docs_to_index = docs.traverse_flat(params.get('traversal_path', self.default_traversal_path))
       ...
```

Another option would be

2 **Constructor-time traversal paths**

```python

class MyExecutor(Executor):

    def __init__(
            self,
            paths: dict = {'index': 'r', 'search': 'm'}
            **kwargs,
    ):
        super().__init__(**kwargs)
        self.paths = paths

...

   @requests(on='/index')
   def index(self, docs, params, ...):
       docs_to_index = docs.traverse_flat(self.paths['index'])
       ...
```

Other options would be (NOT currently supported)

3 **Decorators**

```python

   @requests(on='/index')
   @traverse(path='c')
   def index(self, docs, params, ...):
       # docs will be chunks here
       ...
```

This would require some discussions on how to do properly.

Any other patterns you have been using / would like to see us support?

Hints from issue discussion:
This is specially `important` for `public shared executors`, where the user won't have capacity to rewrite its behavior

Commit message: test: add integration test

Patch to review:
diff --git a/jina/peapods/runtimes/zmq/zed.py b/jina/peapods/runtimes/zmq/zed.py
index 658842c3445d3..cd1de0d5bd5e8 100644
--- a/jina/peapods/runtimes/zmq/zed.py
+++ b/jina/peapods/runtimes/zmq/zed.py
@@ -171,6 +171,15 @@ def _post_hook(self, msg: 'Message') -> 'ZEDRuntime':
         msg.update_timestamp()
         return self
 
+    @staticmethod
+    def _parse_params(parameters: Dict, executor_name: str):
+        parsed_params = parameters
+        specific_parameters = parameters.get(executor_name, None)
+        if specific_parameters:
+            parsed_params.update(**specific_parameters)
+            del parsed_params[executor_name]
+        return parsed_params
+
     def _handle(self, msg: 'Message') -> 'ZEDRuntime':
         """Register the current message to this pea, so that all message-related properties are up-to-date, including
         :attr:`request`, :attr:`prev_requests`, :attr:`message`, :attr:`prev_messages`. And then call the executor to handle
@@ -201,11 +210,13 @@ def _handle(self, msg: 'Message') -> 'ZEDRuntime':
         ):
             return self
 
+        params = self._parse_params(self.request.parameters, self._executor.metas.name)
+
         # executor logic
         r_docs = self._executor(
             req_endpoint=self.envelope.header.exec_endpoint,
             docs=self.docs,
-            parameters=self.request.parameters,
+            parameters=params,
             docs_matrix=self.docs_matrix,
             groundtruths=self.groundtruths,
             groundtruths_matrix=self.groundtruths_matrix,
diff --git a/tests/integration/override_executor_specific_params/__init__.py b/tests/integration/override_executor_specific_params/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/override_executor_specific_params/test_specific_params.py b/tests/integration/override_executor_specific_params/test_specific_params.py
new file mode 100644
index 0000000000000..b215a473a711f
--- /dev/null
+++ b/tests/integration/override_executor_specific_params/test_specific_params.py
@@ -0,0 +1,40 @@
+from typing import Dict
+
+from jina import Flow, DocumentArray, Document, Executor, requests
+from tests import validate_callback
+
+
+class DummyOverrideParams(Executor):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    @requests()
+    def bar(self, docs: 'DocumentArray', parameters: Dict, *args, **kwargs):
+        for doc in docs:
+            doc.tags = parameters
+
+
+def test_override_params(mocker):
+    def validate(response):
+        assert len(response.docs) == 1
+        for doc in response.docs:
+            assert doc.tags['param1'] == 'changed'
+            assert doc.tags['param2'] == 60
+
+    f = Flow().add(
+        uses={'jtype': 'DummyOverrideParams', 'metas': {'name': 'exec_name'}},
+    )
+
+    mock = mocker.Mock()
+    error_mock = mocker.Mock()
+
+    with f:
+        f.index(
+            inputs=DocumentArray([Document()]),
+            parameters={'param1': 50, 'param2': 60, 'exec_name': {'param1': 'changed'}},
+            on_done=mock,
+            on_error=error_mock,
+        )
+    mock.assert_called_once()
+    validate_callback(mock, validate)
+    error_mock.assert_not_called()
diff --git a/tests/unit/peapods/runtimes/zmq/__init__.py b/tests/unit/peapods/runtimes/zmq/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/unit/peapods/runtimes/zmq/test_zed_runtime.py b/tests/unit/peapods/runtimes/zmq/test_zed_runtime.py
new file mode 100644
index 0000000000000..535a35aecf7f1
--- /dev/null
+++ b/tests/unit/peapods/runtimes/zmq/test_zed_runtime.py
@@ -0,0 +1,15 @@
+from jina.peapods.runtimes.zmq.zed import ZEDRuntime
+
+
+def test_zed_runtime_parse_params():
+    params = {
+        'traversal_path': 'r',
+        'param1': 5,
+        'executor_name': {'traversal_path': 'c'},
+    }
+
+    parsed_params = ZEDRuntime._parse_params(params, 'executor_name')
+
+    assert parsed_params['traversal_path'] == 'c'
+    assert parsed_params['param1'] == 5
+    assert 'executor_name' not in parsed_params


Code review comments:
diff_hunk: @@ -0,0 +1,41 @@
+from typing import Dict
+
+from jina import Flow, DocumentArray, Document, Executor, requests
+from tests import validate_callback
+
+
+class DummyOverrideParams(Executor):
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+    @requests()
+    def bar(self, docs: 'DocumentArray', parameters: Dict, *args, **kwargs):
+        for doc in docs:
+            doc.tags = parameters
+
+
+def test_override_params(mocker):
+    def validate(response):
+        assert len(response.docs) == 1
+        for doc in response.docs:
+            assert doc.tags['param1'] == 'changed'
+            assert doc.tags['param2'] == 60
+            assert doc.tags['exec_name']['param1'] == 'changed'
+
+    f = Flow().add(
+        uses={'jtype': 'DummyOverrideParams', 'metas': {'name': 'exec_name'}},
+    )
+
+    mock = mocker.Mock()
+    error_mock = mocker.Mock()
+
+    with f:
+        f.index(
+            inputs=DocumentArray([Document()]),
+            parameters={'param1': 50, 'param2': 60, 'exec_name': {'param1': 'changed'}},
+            on_done=mock,
path: tests/integration/override_executor_specific_params/test_specific_params.py
text: @user1:
can u use `return_results=True`? it's easier to read

@user1:
and do the asserts on the returned valued.

```python
resp = f.index(..., return_results=True)
validate(resp)
```

diff_hunk: @@ -171,6 +171,15 @@ def _post_hook(self, msg: 'Message') -> 'ZEDRuntime':
         msg.update_timestamp()
         return self
 
+    @staticmethod
+    def _parse_params(parameters: Dict, executor_name: str):
+        parsed_params = parameters
+        specific_parameters = parameters.get(executor_name, None)
+        if specific_parameters:
+            parsed_params.update(**specific_parameters)
+            del parsed_params[executor_name]
path: jina/peapods/runtimes/zmq/zed.py
text: @user1:
Why deleting?

@author:
I did not have a clear view. I found it more clear to have a single source of truth, but can remove

@author:
Should I change this @user1 , it may be more conservative yes

@user1:
Yes, please don't delete it.

@author:
Done!