Repository: deepset-ai/haystack
Issue: Impossible to combine JoinDocuments and JoinAnswers

Creating a pipeline using JoinDocuments and JoinAnswer, leads to the following error:
```
File "/home/ubuntu/haystack/haystack/nodes/other/join_answers.py", line 45, in <listcomp>
    reader_results = [inp["answers"] for inp in inputs]
KeyError: 'answers'
```

The reason for this is that [these lines](https://github.com/deepset-ai/haystack/blob/b6986ea25d7bbd06158dcb9b2f2b9187266accff/haystack/nodes/base.py#L237-L239) cause the `inputs` parameter of the first join node to be propagated down the pipeline. This eventually leads to [this condition](https://github.com/deepset-ai/haystack/blob/b6986ea25d7bbd06158dcb9b2f2b9187266accff/haystack/pipelines/base.py#L530) being false when constructing the input for the next join node resulting in that error. 

Changed files:
haystack/nodes/base.py, haystack/pipelines/base.py

diff --git a/haystack/nodes/base.py b/haystack/nodes/base.py
index d37fe32b..16a1949d 100644
--- a/haystack/nodes/base.py
+++ b/haystack/nodes/base.py
@@ -251,6 +251,9 @@ class BaseComponent(ABC):
         }
         return component_signature
 
+    def _takes_multiple_inputs(self):
+        return "inputs" in inspect.signature(self.run).parameters
+
 
 class RootNode(BaseComponent):
     """
diff --git a/haystack/pipelines/base.py b/haystack/pipelines/base.py
index bbfd07f7..64a57933 100644
--- a/haystack/pipelines/base.py
+++ b/haystack/pipelines/base.py
@@ -529,24 +529,23 @@ class Pipeline:
                 else:
                     next_nodes = self.get_next_nodes(node_id, stream_id)
                     for n in next_nodes:  # add successor nodes with corresponding inputs to the queue
-                        if queue.get(n):  # concatenate inputs if it's a join node
-                            existing_input = queue[n]
-                            if "inputs" not in existing_input.keys():
-                                updated_input: dict = {"inputs": [existing_input, node_output], "params": params}
+                        if self.graph.nodes[n][
+                            "component"
+                        ]._takes_multiple_inputs():  # concatenate inputs if it's a join node
+                            if n not in queue:
+                                new_input: Dict[str, Any] = {"inputs": [], "params": params}
                                 if query:
-                                    updated_input["query"] = query
+                                    new_input["query"] = query
                                 if file_paths:
-                                    updated_input["file_paths"] = file_paths
+                                    new_input["file_paths"] = file_paths
                                 if labels:
-                                    updated_input["labels"] = labels
+                                    new_input["labels"] = labels
                                 if documents:
-                                    updated_input["documents"] = documents
+                                    new_input["documents"] = documents
                                 if meta:
-                                    updated_input["meta"] = meta
-                            else:
-                                existing_input["inputs"].append(node_output)
-                                updated_input = existing_input
-                            queue[n] = updated_input
+                                    new_input["meta"] = meta
+                            new_input["inputs"].append(node_output)
+                            queue[n] = new_input
                         else:
                             queue[n] = node_output
                 i = 0
@@ -677,24 +676,23 @@ class Pipeline:
                 else:
                     next_nodes = self.get_next_nodes(node_id, stream_id)
                     for n in next_nodes:
-                        if queue.get(n):  # concatenate inputs if it's a join node
-                            existing_input = queue[n]
-                            if "inputs" not in existing_input.keys():
-                                updated_input: Dict = {"inputs": [existing_input, node_output], "params": params}
+                        if self.graph.nodes[n][
+                            "component"
+                        ]._takes_multiple_inputs():  # concatenate inputs if it's a join node
+                            if n not in queue:
+                                new_input: Dict[str, Any] = {"inputs": [], "params": params}
                                 if queries:
-                                    updated_input["queries"] = queries
+                                    new_input["queries"] = queries
                                 if file_paths:
-                                    updated_input["file_paths"] = file_paths
+                                    new_input["file_paths"] = file_paths
                                 if labels:
-                                    updated_input["labels"] = labels
+                                    new_input["labels"] = labels
                                 if documents:
-                                    updated_input["documents"] = documents
+                                    new_input["documents"] = documents
                                 if meta:
-                                    updated_input["meta"] = meta
-                            else:
-                                existing_input["inputs"].append(node_output)
-                                updated_input = existing_input
-                            queue[n] = updated_input
+                                    new_input["meta"] = meta
+                            new_input["inputs"].append(node_output)
+                            queue[n] = new_input
                         else:
                             queue[n] = node_output
                 i = 0