Repository: jina-ai/serve
Pull request: feat: pass `docs_map` to Executor
**Goals:**
Pass `docs_map` to `Executor` as a dictionary with keys as  the `last Executors` that processed a Request, and values as the `DocumentArray` in that specific Request.


Problem statement:
Enforce deterministic results order on 'needs' in topology graph
When making requests to pods running in parallel, the topology graph should await all tasks together to have a deterministic ordering (similar to asyncio.gather in the head runtime)

feat: find out which docs are which in docs_matrix
When pairing with @violenil, we were hitting some limitations of the core. 
It is currently not possible to find out what DocArrays are coming from what executor.
Please implement a mechanism to let us know what the previous executor was for each DocArray.

Example:
Two encoders in parallel and a merger in sequence that has to know what embedding came from what executor

Hints from issue discussion:
@jina-ai/product This issue is stale because it has been open 90 days with no activity. Remove stale label or comment or this will be closed in 14 days
![flow_x](https://user-images.githubusercontent.com/55404563/198677306-000a263b-b2fe-4c26-a82b-40e83d16c2f3.png)

You could have the `disable_reduce` flag and have a `DocumentArrayMatrix` which will allow u to know where the DocArray comes from.
I would like to clarify the scope of this before proceeding. 

If the intention when passing DocumentArray to Executors is to flag Executors, I would say this has to be done with these 2 options:

1 - Either responsability left to the Executor developer to flag the DocArray
2 - The WorkerRuntime flag the Executor in some `__flag__` reserved tag of `DocArray`.

The alternative is to use the `disable_reduce` feature and have the Indexer handle the DocumentArray without flattening, I would assume the Matrix is created in a deterministic order.
I think this issue depends on resolving this
https://github.com/jina-ai/jina/issues/4084
Once the order becomes deterministic, the document array matrix can be used to identify each executor used
What about the following?:
```python
class MergeExecutor(Executor):
    def __init__(self, encoder_names):
        self.encoder_names = encoder_names

    @requests
    def foo(self, docs, parameters, docs_matrix):
        for encoder_name in encoder_names:
             docs = docs_matrix[encoder_name]
             ...
```
Advantages:
- user friendly
- not much effort for the developer of the executor
- does not rely on the order of executors in topology
- if executor is skipped, then it would be hard to know which one if we rely on order

Commit message: style: fix overload and cli autocomplete

Patch to review:
diff --git a/docs/fundamentals/executor/executor-methods.md b/docs/fundamentals/executor/executor-methods.md
index b321b697915a2..7fc5012589ba4 100644
--- a/docs/fundamentals/executor/executor-methods.md
+++ b/docs/fundamentals/executor/executor-methods.md
@@ -50,11 +50,14 @@ with f:
 ```
 
 ```shell
-           Flow@18048[I]:🎉 Flow is ready to use!                                                   
-	🔗 Protocol: 		GRPC
-	🏠 Local access:	0.0.0.0:52255
-	🔒 Private network:	192.168.1.187:52255
-	🌐 Public address:	212.231.186.65:52255
+────────────────────────── 🎉 Flow is ready to serve! ──────────────────────────
+╭────────────── 🔗 Endpoint ───────────────╮
+│  ⛓     Protocol                    GRPC  │
+│  🏠       Local           0.0.0.0:55925  │
+│  🔒     Private     192.168.1.187:55925  │
+│  🌍      Public    212.231.186.65:55925  │
+╰──────────────────────────────────────────╯
+
 Calling foo
 Calling bar
 Calling foo
@@ -93,20 +96,20 @@ All Executor methods decorated by `@requests` need to follow the signature below
 The `async` definition is optional.
 
 ```python
-from typing import Dict, Union, List
+from typing import Dict, Union, List, Optional
 from jina import Executor, requests, DocumentArray
 
 
 class MyExecutor(Executor):
     @requests
     async def foo(
-        self, docs: DocumentArray, parameters: Dict, docs_matrix: List[DocumentArray]
+        self, docs: DocumentArray, parameters: Dict, docs_matrix: Optional[List[DocumentArray]], docs_map: Optional[Dict[str, DocumentArray]]
     ) -> Union[DocumentArray, Dict, None]:
         pass
 
     @requests
     def bar(
-        self, docs: DocumentArray, parameters: Dict, docs_matrix: List[DocumentArray]
+        self, docs: DocumentArray, parameters: Dict, docs_matrix: Optional[List[DocumentArray]], docs_map: Optional[Dict[str, DocumentArray]]
     ) -> Union[DocumentArray, Dict, None]:
         pass
 ```
@@ -118,10 +121,11 @@ any other `list`-like object in a Python function.
 
 - `parameters`: A Dict object that passes extra parameters to Executor functions.
 
-- `docs_matrix`:  This is the least common parameter to be used for an Executor. This is needed when an Executor is used inside a Flow to merge or reduce the output of more than one other Executor.
-
+- `docs_matrix`:  This is one of the least common parameter to be used for an Executor. This is needed when an Executor is used inside a Flow to merge or reduce the output of more than one other Executor.
  
+- `docs_map`:  This is one of the least common parameter to be used for an Executor. It has the same utility as `docs_matrix` but the information comes as a dict with previous Executor names as keys. If onl
 
+- `tracing_context`: Context needed if you want to add custom traces. Check {ref}`how to add custom traces in your Executor <instrumenting-executor>`
 
 
 ````{admonition} Hint
@@ -156,11 +160,14 @@ class MyExecutor(Executor):
 
 ### Multiple DocumentArrays as input argument
 
-You have seen that `Executor` methods can receive three types of parameters: `docs`, `parameters` and `docs_matrix`.
+You have seen that {class}`~jina.Executor` methods can receive multiple parameters.
+
+`docs_matrix` and `docs_map` are only used in some special cases.
 
-`docs_matrix` is only used in some special cases.
+One case is when an Executor receives messages from more than one upstream {class}`~jina.Executor` in the {class}`~jina.Flow`:
 
-One case is when an Executor receives messages from more than one upstream Executor in the Flow:
+If you set `disable_reduce` to True and the Executor has more than one incoming Executor, the Executor will receive all the DocumentArrays coming from previous Executors independently under `docs_matrix` and `docs_map`.
+If `disable_reduce` is not set or set to False, `docs_map` and `docs_matrix` will be None and the Executor will receive a single DocumentArray resulting from the reducing of all the incoming ones.
 
 ```python
 from jina import Flow, Executor, requests, Document, DocumentArray
@@ -202,22 +209,75 @@ f = (
 )
 
 with f:
-    returned_docs = f.post(on='/', Document())
+    returned_docs = f.post(on='/', inputs=Document())
 
 print(f'Resulting documents {returned_docs[0].text}')
 ```
 
 
 ```shell
-           Flow@1244[I]:🎉 Flow is ready to use!
-	🔗 Protocol: 		GRPC
-	🏠 Local access:	0.0.0.0:54550
-	🔒 Private network:	192.168.1.187:54550
-	🌐 Public address:	212.231.186.65:54550
+────────────────────────── 🎉 Flow is ready to serve! ──────────────────────────
+╭────────────── 🔗 Endpoint ───────────────╮
+│  ⛓     Protocol                    GRPC  │
+│  🏠       Local           0.0.0.0:55761  │
+│  🔒     Private     192.168.1.187:55761  │
+│  🌍      Public    212.231.186.65:55761  │
+╰──────────────────────────────────────────╯
+
 MergeExec processing pairs of Documents "Exec1" and "Exec2"
 Resulting documents Document merging from "Exec1" and "Exec2"
 ```
 
+When merging Documents from more than one upstream {class}`~jina.Executor`, sometimes you want to control which Documents come from which Executor.
+Executor will receive the `docs_map` as a dictionary where the key will be the last Executor processing that previous request and the DocumentArray of the request as the values.
+
+```python
+from jina import Flow, Executor, requests, Document
+
+
+class Exec1(Executor):
+    @requests
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'Exec1'
+
+
+class Exec2(Executor):
+    @requests
+    def foo(self, docs, **kwargs):
+        for doc in docs:
+            doc.text = 'Exec2'
+
+
+class MergeExec(Executor):
+    @requests
+    def foo(self, docs_map, **kwargs):
+        print(docs_map)
+
+f = (
+    Flow()
+    .add(uses=Exec1, name='exec1')
+    .add(uses=Exec2, name='exec2')
+    .add(uses=MergeExec, needs=['exec1', 'exec2'], disable_reduce=True)
+)
+
+with f:
+    f.post(on='/', Document())
+```
+
+
+```shell
+────────────────────────── 🎉 Flow is ready to serve! ──────────────────────────
+╭────────────── 🔗 Endpoint ───────────────╮
+│  ⛓     Protocol                    GRPC  │
+│  🏠       Local           0.0.0.0:56286  │
+│  🔒     Private     192.168.1.187:56286  │
+│  🌍      Public    212.231.186.65:56286  │
+╰──────────────────────────────────────────╯
+
+{'exec1': <DocumentArray (length=1) at 140270975034640>, 'exec2': <DocumentArray (length=1) at 140270975034448>}
+```
+
 (async-executors)=
 ## Async coroutines
 
@@ -227,7 +287,7 @@ Python to write concurrent code.
 
 
 ```python
-from jina import Executor, requests, Flow
+from jina import Executor, requests
 
 
 class MyExecutor(Executor):
@@ -242,7 +302,6 @@ This example has a heavy lifting API which we call several times, and we leverag
 async Python features to speed up the {class}`~jina.Executor`'s call by calling the API multiple times concurrently. As a counterpart, in an example without `coroutines`, all 50 API calls are queued and nothing is done concurrently.
 
 
-
 ````{tab} Async coroutines
 ```python
 import asyncio
@@ -269,12 +328,16 @@ with f:
 ```
 
 ```shell
-           Flow@20588[I]:🎉 Flow is ready to use!
-	🔗 Protocol: 		GRPC
-	🏠 Local access:	0.0.0.0:62598
-	🔒 Private network:	192.168.1.187:62598
-	🌐 Public address:	212.231.186.65:62598
-⠙       DONE ━╸━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0:00:01 100% ETA: 0 seconds 41 steps done in 1 second
+────────────────────────── 🎉 Flow is ready to serve! ──────────────────────────
+╭────────────── 🔗 Endpoint ───────────────╮
+│  ⛓     Protocol                    GRPC  │
+│  🏠       Local           0.0.0.0:54153  │
+│  🔒     Private     192.168.1.187:54153  │
+│  🌍      Public    212.231.186.65:54153  │
+╰──────────────────────────────────────────╯
+
+  DONE ━━━━━━━━━━━━━━━━━━━━━━ 0:00:01 100% ETA: 0:00:00 50 steps done in 1      
+                                                        second  
 ```
 
 ````
@@ -305,12 +368,16 @@ with f:
 ```
 
 ```shell
-           Flow@20394[I]:🎉 Flow is ready to use!
-	🔗 Protocol: 		GRPC
-	🏠 Local access:	0.0.0.0:52592
-	🔒 Private network:	192.168.1.187:52592
-	🌐 Public address:	212.231.186.65:52592
-⠏       DONE ━╸━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ 0:00:50 100% ETA: 0 seconds 41 steps done in 50 seconds
+────────────────────────── 🎉 Flow is ready to serve! ──────────────────────────
+╭────────────── 🔗 Endpoint ───────────────╮
+│  ⛓     Protocol                    GRPC  │
+│  🏠       Local           0.0.0.0:52340  │
+│  🔒     Private     192.168.1.187:52340  │
+│  🌍      Public    212.231.186.65:52340  │
+╰──────────────────────────────────────────╯
+
+  DONE ━━━━━━━━━━━━━━━━━━━━━━ 0:00:50 100% ETA: 0:00:00 50 steps done in 50     
+                                                        seconds        
 ```
 ````
 
@@ -335,9 +402,6 @@ class DummyExecutor(Executor):
 ```
 
 
-
-
-
 ## Returns
 
 Every Executor method can `return` in three ways: 
@@ -457,11 +521,14 @@ with f:
 
 
 ```shell
-           Flow@23300[I]:🎉 Flow is ready to use!                                                   
-	🔗 Protocol: 		GRPC
-	🏠 Local access:	0.0.0.0:61855
-	🔒 Private network:	192.168.1.187:61855
-	🌐 Public address:	212.231.186.65:61855
+────────────────────────── 🎉 Flow is ready to serve! ──────────────────────────
+╭────────────── 🔗 Endpoint ───────────────╮
+│  ⛓     Protocol                    GRPC  │
+│  🏠       Local           0.0.0.0:58746  │
+│  🔒     Private     192.168.1.187:58746  │
+│  🌍      Public    212.231.186.65:58746  │
+╰──────────────────────────────────────────╯
+
  ProcessDocuments: received document with text "request1"
  PrintExecutor: received document with text: "I changed the executor in place"
  ProcessDocuments: received document with text: "request2"
diff --git a/docs/fundamentals/flow/executor-args.md b/docs/fundamentals/flow/executor-args.md
index 38bb50da17cc2..cb461d10ed813 100644
--- a/docs/fundamentals/flow/executor-args.md
+++ b/docs/fundamentals/flow/executor-args.md
@@ -17,6 +17,7 @@
 | `py_modules` | The customized python modules need to be imported before loading the executor<br><br>Note that the recommended way is to only import a single module - a simple python file, if your<br>executor can be defined in a single file, or an ``__init__.py`` file if you have multiple files,<br>which should be structured as a python package. For more details, please see the<br>`Executor cookbook <https://docs.jina.ai/fundamentals/executor/executor-files/>`__ | `array` | `None` |
 | `output_array_type` | The type of array `tensor` and `embedding` will be serialized to.<br><br>Supports the same types as `docarray.to_protobuf(.., ndarray_type=...)`, which can be found <br>`here <https://docarray.jina.ai/fundamentals/document/serialization/#from-to-protobuf>`.<br>Defaults to retaining whatever type is returned by the Executor. | `string` | `None` |
 | `exit_on_exceptions` | List of exceptions that will cause the Executor to shut down. | `array` | `[]` |
+| `disable_reduce` | Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor, if the Executor is expected to handle `docs_matrix` | `boolean` | `False` |
 | `port` | The port for input data to bind to, default is a random port between [49152, 65535]. In the case of an external Executor (`--external` or `external=True`) this can be a list of ports, separated by commas. Then, every resulting address will be considered as one replica of the Executor. | `string` | `random in [49152, 65535]` |
 | `host_in` | The host address for binding to, by default it is 0.0.0.0 | `string` | `0.0.0.0` |
 | `grpc_server_options` | Dictionary of kwargs arguments that will be passed to the grpc server as options when starting the server, example : {'grpc.max_send_message_length': -1} | `object` | `None` |
@@ -47,5 +48,4 @@
 | `uses_before_address` | The address of the uses-before runtime | `string` | `None` |
 | `uses_after_address` | The address of the uses-before runtime | `string` | `None` |
 | `connection_list` | dictionary JSON with a list of connections to configure | `string` | `None` |
-| `disable_reduce` | Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor connected to this Head | `boolean` | `False` |
 | `timeout_send` | The timeout in milliseconds used when sending data requests to Executors, -1 means no timeout, disabled by default | `number` | `None` |
\ No newline at end of file
diff --git a/jina/orchestrate/flow/base.py b/jina/orchestrate/flow/base.py
index b672d9f1407fe..8360d07bebd6f 100644
--- a/jina/orchestrate/flow/base.py
+++ b/jina/orchestrate/flow/base.py
@@ -892,7 +892,7 @@ def add(
         :param compression: The compression mechanism used when sending requests from the Head to the WorkerRuntimes. For more details, check https://grpc.github.io/grpc/python/grpc.html#compression.
         :param connection_list: dictionary JSON with a list of connections to configure
         :param disable_auto_volume: Do not automatically mount a volume for dockerized Executors.
-        :param disable_reduce: Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor connected to this Head
+        :param disable_reduce: Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor, if the Executor is expected to handle `docs_matrix`
         :param docker_kwargs: Dictionary of kwargs arguments that will be passed to Docker SDK when starting the docker '
           container.
 
@@ -1048,7 +1048,7 @@ def add(
         :param compression: The compression mechanism used when sending requests from the Head to the WorkerRuntimes. For more details, check https://grpc.github.io/grpc/python/grpc.html#compression.
         :param connection_list: dictionary JSON with a list of connections to configure
         :param disable_auto_volume: Do not automatically mount a volume for dockerized Executors.
-        :param disable_reduce: Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor connected to this Head
+        :param disable_reduce: Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor, if the Executor is expected to handle `docs_matrix`
         :param docker_kwargs: Dictionary of kwargs arguments that will be passed to Docker SDK when starting the docker '
           container.
 
diff --git a/jina/parsers/orchestrate/runtimes/head.py b/jina/parsers/orchestrate/runtimes/head.py
index 76878e538c728..352f654ec0e7b 100644
--- a/jina/parsers/orchestrate/runtimes/head.py
+++ b/jina/parsers/orchestrate/runtimes/head.py
@@ -33,13 +33,6 @@ def mixin_head_parser(parser):
         help='dictionary JSON with a list of connections to configure',
     )
 
-    gp.add_argument(
-        '--disable-reduce',
-        action='store_true',
-        default=False,
-        help='Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor connected to this Head',
-    )
-
     gp.add_argument(
         '--timeout-send',
         type=int,
diff --git a/jina/parsers/orchestrate/runtimes/worker.py b/jina/parsers/orchestrate/runtimes/worker.py
index 328fff4df4465..afb3f872dd5f4 100644
--- a/jina/parsers/orchestrate/runtimes/worker.py
+++ b/jina/parsers/orchestrate/runtimes/worker.py
@@ -92,4 +92,12 @@ def mixin_worker_runtime_parser(parser):
         nargs='*',
         help='List of exceptions that will cause the Executor to shut down.',
     )
+
+    gp.add_argument(
+        '--disable-reduce',
+        action='store_true',
+        default=False,
+        help='Disable the built-in reduce mechanism, set this if the reduction is to be handled by the Executor, if the Executor is expected to handle `docs_matrix`',
+    )
+
     mixin_base_runtime_parser(gp)
diff --git a/jina/serve/runtimes/request_handlers/worker_request_handler.py b/jina/serve/runtimes/request_handlers/worker_request_handler.py
index b37deb3656a9f..5c7f4adcf1d14 100644
--- a/jina/serve/runtimes/request_handlers/worker_request_handler.py
+++ b/jina/serve/runtimes/request_handlers/worker_request_handler.py
@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Dict, List, Optional
+from typing import TYPE_CHECKING, Dict, List, Optional, Tuple
 
 from docarray import DocumentArray
 
@@ -43,7 +43,6 @@ def __init__(
         """
         super().__init__()
         self.args = args
-        self.args.parallel = self.args.shards
         self.logger = logger
         self._is_closed = False
         self._load_executor(
@@ -285,14 +284,13 @@ async def handle(
         )
 
         # executor logic
+        docs_matrix, docs_map = WorkerRequestHandler._get_docs_matrix_from_request(requests)
         return_data = await self._executor.__acall__(
             req_endpoint=requests[0].header.exec_endpoint,
             docs=docs,
             parameters=params,
-            docs_matrix=WorkerRequestHandler.get_docs_matrix_from_request(
-                requests,
-                field='docs',
-            ),
+            docs_matrix=docs_matrix,
+            docs_map=docs_map,
             tracing_context=tracing_context,
         )
 
@@ -346,26 +344,32 @@ def close(self):
             self._is_closed = True
 
     @staticmethod
-    def get_docs_matrix_from_request(
+    def _get_docs_matrix_from_request(
         requests: List['DataRequest'],
-        field: str,
-    ) -> List['DocumentArray']:
+        return_map: bool = True,
+    ) -> Tuple[Optional[List['DocumentArray']], Optional[Dict[str, 'DocumentArray']]]:
         """
         Returns a docs matrix from a list of DataRequest objects.
+
         :param requests: List of DataRequest objects
-        :param field: field to be retrieved
-        :return: docs matrix: list of DocumentArray objects
+        :param return_map: bool indicating if a map should be returned
+        :return: docs matrix and doc: list of DocumentArray objects
         """
+        docs_map = None
         if len(requests) > 1:
-            result = [getattr(request, field) for request in requests]
+            docs_matrix = [request.docs for request in requests]
+            if return_map:
+                docs_map = {req.last_executor: req.docs for req in requests}
         else:
-            result = [getattr(requests[0], field)]
+            docs_matrix = [requests[0].docs]
 
         # to unify all length=0 DocumentArray (or any other results) will simply considered as None
         # otherwise, the executor has to handle [None, None, None] or [DocArray(0), DocArray(0), DocArray(0)]
-        len_r = sum(len(r) for r in result)
-        if len_r:
-            return result
+        len_r = sum(len(r) for r in docs_matrix)
+        if len_r == 0:
+            docs_matrix = None
+
+        return docs_matrix, docs_map
 
     @staticmethod
     def get_parameters_dict_from_request(
@@ -454,8 +458,8 @@ def reduce_requests(requests: List['DataRequest']) -> 'DataRequest':
         :param requests: List of DataRequest objects
         :return: the resulting DataRequest
         """
-        docs_matrix = WorkerRequestHandler.get_docs_matrix_from_request(
-            requests, field='docs'
+        docs_matrix, _ = WorkerRequestHandler._get_docs_matrix_from_request(
+            requests
         )
 
         # Reduction is applied in-place to the first DocumentArray in the matrix
diff --git a/jina/serve/runtimes/worker/__init__.py b/jina/serve/runtimes/worker/__init__.py
index d50e975218e75..c39616c70869f 100644
--- a/jina/serve/runtimes/worker/__init__.py
+++ b/jina/serve/runtimes/worker/__init__.py
@@ -36,6 +36,7 @@ def __init__(
         """
         self._health_servicer = health.HealthServicer(experimental_non_blocking=True)
         super().__init__(args, **kwargs)
+        self.deployment_name = self.name.split('/')[0]
 
     async def async_setup(self):
         """
@@ -225,6 +226,7 @@ async def process_data(self, requests: List[DataRequest], context) -> DataReques
                 result = await self._worker_request_handler.handle(
                     requests=requests, tracing_context=tracing_context
                 )
+                result.add_executor(self.deployment_name)
                 if self._successful_requests_metrics:
                     self._successful_requests_metrics.inc()
                 if self._successful_requests_counter:
diff --git a/jina/types/request/data.py b/jina/types/request/data.py
index a21d6314d7f78..bf376dadcd671 100644
--- a/jina/types/request/data.py
+++ b/jina/types/request/data.py
@@ -54,7 +54,7 @@ def docs(self, value: DocumentArray):
             self.set_docs_convert_arrays(value, None)
 
         def set_docs_convert_arrays(
-            self, value: DocumentArray, ndarray_type: Optional[str] = None
+                self, value: DocumentArray, ndarray_type: Optional[str] = None
         ):
             """ " Convert embedding and tensor to given type, then set DocumentArray
 
@@ -100,8 +100,8 @@ def docs_bytes(self, value: bytes):
     """
 
     def __init__(
-        self,
-        request: Optional[RequestSourceType] = None,
+            self,
+            request: Optional[RequestSourceType] = None,
     ):
         self.buffer = None
         self._pb_body = None
@@ -159,7 +159,7 @@ def is_decompressed_wo_data(self) -> bool:
 
     @property
     def proto_wo_data(
-        self,
+            self,
     ) -> Union['jina_pb2.DataRequestProtoWoData', 'jina_pb2.DataRequestProto']:
         """
         Transform the current buffer to a :class:`jina_pb2.DataRequestProtoWoData` unless the full proto has already
@@ -173,7 +173,7 @@ def proto_wo_data(
 
     @property
     def proto(
-        self,
+            self,
     ) -> Union['jina_pb2.DataRequestProto', 'jina_pb2.DataRequestProtoWoData']:
         """
         Cast ``self`` to a :class:`jina_pb2.DataRequestProto` or a :class:`jina_pb2.DataRequestProto`. Laziness will be broken and serialization will be recomputed when calling.
@@ -187,7 +187,7 @@ def proto(
 
     @property
     def proto_with_data(
-        self,
+            self,
     ) -> 'jina_pb2.DataRequestProto':
         """
         Cast ``self`` to a :class:`jina_pb2.DataRequestProto`. Laziness will be broken and serialization will be recomputed when calling.
@@ -259,6 +259,7 @@ def data(self) -> 'DataRequest._DataContent':
     @property
     def parameters(self) -> Dict:
         """Return the `parameters` field of this DataRequest as a Python dict
+
         :return: a Python dict view of the parameters.
         """
         # if u get this u need to have it decompressed
@@ -267,6 +268,7 @@ def parameters(self) -> Dict:
     @parameters.setter
     def parameters(self, value: Dict):
         """Set the `parameters` field of this Request to a Python dict
+
         :param value: a Python dict
         """
         self.proto_wo_data.parameters.Clear()
@@ -290,6 +292,35 @@ def status(self):
         """
         return self.proto_wo_data.header.status
 
+    @property
+    def last_executor(self):
+        """
+        Returns the name of the last Executor that has processed this Request
+
+        :return: the name of the last Executor that processed this Request
+        """
+        if len(self.proto_wo_data.routes) > 0:
+            return self.proto_wo_data.routes[-1].executor
+
+    def add_executor(self, executor_name: str):
+        """
+        Returns the name of the last Executor that has processed this Request
+
+        :param executor_name: name of the Executor processing the Request
+        """
+        route_proto = jina_pb2.RouteProto()
+        route_proto.executor = executor_name
+        self.proto_wo_data.routes.append(route_proto)
+
+    @property
+    def routes(self):
+        """
+        Returns the routes from the request
+
+        :return: the routes object of this request
+        """
+        return self.proto_wo_data.routes
+
     @property
     def request_id(self):
         """
diff --git a/jina_cli/autocomplete.py b/jina_cli/autocomplete.py
index f80a7d35c7371..7e8f11de1c52f 100644
--- a/jina_cli/autocomplete.py
+++ b/jina_cli/autocomplete.py
@@ -40,6 +40,7 @@
             '--py-modules',
             '--output-array-type',
             '--exit-on-exceptions',
+            '--disable-reduce',
             '--port-in',
             '--host-in',
             '--grpc-server-options',
@@ -75,7 +76,6 @@
             '--uses-before-address',
             '--uses-after-address',
             '--connection-list',
-            '--disable-reduce',
             '--timeout-send',
         ],
         'flow': [
@@ -252,6 +252,7 @@
             '--py-modules',
             '--output-array-type',
             '--exit-on-exceptions',
+            '--disable-reduce',
             '--port-in',
             '--host-in',
             '--grpc-server-options',
@@ -287,7 +288,6 @@
             '--uses-before-address',
             '--uses-after-address',
             '--connection-list',
-            '--disable-reduce',
             '--timeout-send',
         ],
         'deployment': [
@@ -312,6 +312,7 @@
             '--py-modules',
             '--output-array-type',
             '--exit-on-exceptions',
+            '--disable-reduce',
             '--port-in',
             '--host-in',
             '--grpc-server-options',
@@ -347,7 +348,6 @@
             '--uses-before-address',
             '--uses-after-address',
             '--connection-list',
-            '--disable-reduce',
             '--timeout-send',
             '--uses-before',
             '--uses-after',
diff --git a/tests/integration/needs-merge/__init__.py b/tests/integration/needs-merge/__init__.py
new file mode 100644
index 0000000000000..e69de29bb2d1d
diff --git a/tests/integration/needs-merge/test_needs_merge.py b/tests/integration/needs-merge/test_needs_merge.py
new file mode 100644
index 0000000000000..d8c00e26532b4
--- /dev/null
+++ b/tests/integration/needs-merge/test_needs_merge.py
@@ -0,0 +1,19 @@
+from jina import Flow, Executor, requests, DocumentArray
+
+
+def test_needs_docs_map():
+    class TestMergeDictDocMatrixExecutor(Executor):
+
+        @requests()
+        def foo(self, docs_map, **kwargs):
+            assert {'exec0', 'exec1'} == set(docs_map.keys())
+
+    f = Flow().add(name='exec0'). \
+        add(name='exec1', replicas=2, shards=2, needs=['gateway']). \
+        add(name='exec2',
+            needs=['exec0', 'exec1'],
+            uses=TestMergeDictDocMatrixExecutor,
+            disable_reduce=True)
+
+    with f:
+        f.post(on='/', inputs=DocumentArray.empty(2))
diff --git a/tests/unit/types/request/test_request.py b/tests/unit/types/request/test_request.py
index cf1eaa4c59707..0162e117b065b 100644
--- a/tests/unit/types/request/test_request.py
+++ b/tests/unit/types/request/test_request.py
@@ -295,3 +295,14 @@ def test_proto_wo_data_docs():  # check if we can access the docs after deserial
     new_data_request = DataRequest(bytes_)
 
     assert new_data_request.docs == r.docs
+
+
+def test_req_add_get_executors():
+    r = DataRequest()
+    r.add_executor('one')
+    assert r.last_executor == 'one'
+    r.add_executor('two')
+    assert r.last_executor == 'two'
+
+    r2 = DataRequest.from_proto(r.proto)
+    assert r2.last_executor == 'two'


Code review comments:
diff_hunk: @@ -290,6 +292,35 @@ def status(self):
         """
         return self.proto_wo_data.header.status
 
+    @property
+    def last_executor(self):
+        """
+        Returns the name of the last Executor that has processed this Request
+
+        :return: the name of the last Executor that processed this Request
+        """
+        if len(self.proto_wo_data.routes) > 0:
+            return self.proto_wo_data.routes[-1].executor
+
+    def add_executor(self, executor_name: str):
+        """
+        Returns the name of the last Executor that has processed this Request
+
+        :param executor_name: name of the Executor processing the Request
path: jina/types/request/data.py
text: @author:
```suggestion
        :param executor_name: name of the Executor processing the Request to be added to the routes
```