Repository: deepset-ai/haystack
Pull request:  MostSimilarDocumentsPipeline introduced
**Proposed changes**:
- New pipeline `MostSimilarDocumentsPipeline` introduced. which helps users to find similar documents using document ids.

**Status (please check what you already did)**:
- [X] Final code
- [X] Added tests

closes #1091 

Problem statement:
MostSimilarDocumentsPipeline -Functionality to get top n documents by document Id rather than a query 
**What**

Is there an existing functionality to get top n documents by document Id rather than a query or is it possible to combine existing functionality  to achieve this?

Hints from issue discussion:
Hey @monika1800

If you just want to get documents I would not use our query functionality of the retrievers. Retrievers always want to combine a search query to retrieve relevant documents.

What you are looking for is the inside our document store directly.
We have `document_store.get_document_by_id()`
and `document_store.get_all_documents()`.

I guess you are splitting up your documents and want the individual part when referring to "document_id". This you should be able to achieve with:
```
document_store.get_all_documents(filters={"name": ["example_doc1.txt"]}
```
Of course when inserting the documents into the document store you need to add the "name" as a metainformation. If you dont know how to do this please add the code how you convert/load and insert the docs, so I can help better.
@monika1800 Let us know if the above comment already resolved your issue or if you are maybe rather looking for a way to find the n most similar document for a given document with id ABC. That would be a slightly different use case, but definitely achievable with a retriever (e.g. EmbeddingRetriever).
Thanks  @Timoeller. 

@tholor Yes my use case was to find the n most similar document for a given document with id ABC.

I was able to achieve this with get_all_documents filtering for a given Id embedding using query_by_embedding.
Nice. 

Would you like to share the code that you used here on github so others can find it? I think this is a pretty common use case, so people will be interested in your solution.
Can we close the issue then as well?
```
get_docs_params = cfg.get('get_docs_params', {
        		"filters": {"Id": ["a2W3l000001Q6pmEAC"]}, #multiple filters can be added, here I am providing the document Id to fetch the relevant document embedding.
        		"return_embedding": true, #This is set to true to get embedding for documents
        		"index": "sim_es2"
        	})
query_by_embed_params = cfg.get('query_by_embed_params', {"return_embedding": false,
        	"top_k": 2, #we can give top n number of documents to fetch the similar documents from the docstore
        	"index": "sim_es2"})

docs_gad = docstore.get_all_documents(**get_docs_params) #this function will return the data for the list of records given in filters

for d in docs_gad:
    query_emb = d.embedding #to get embedding of each document
    query_by_embed_params.update({"query_emb": query_emb})
    results = docstore.query_by_embedding(**query_by_embed_params)
    result_list.append(results) 
   
#results are in the form of [[],[],[]], where each list contains the similar documents for eahc of the id

```
@Timoeller I have added the relevant piece of code. Thank you.
Thanks for sharing! I will keep this issue open, as I think we should simplify this in Haystack with something like a "MostSimilarDocumentsPipeline" that gets IDs as input and returns for each ID the most similar top_k docs.
@tholor Assign.

Commit message: [pipeline] test cases added.

Patch to review:
diff --git a/haystack/pipeline.py b/haystack/pipeline.py
index 577d095559..09c74053c4 100644
--- a/haystack/pipeline.py
+++ b/haystack/pipeline.py
@@ -28,6 +28,7 @@
 
 from haystack import BaseComponent
 from haystack.generator.base import BaseGenerator
+from haystack.document_store.base import BaseDocumentStore
 from haystack.reader.base import BaseReader
 from haystack.retriever.base import BaseRetriever
 from haystack.summarizer.base import BaseSummarizer
@@ -1287,4 +1288,31 @@ def run(self, query, documents, **kwargs):
         # Pass also the other incoming kwargs so that future nodes still have access to it
         output.update(**kwargs)
 
-        return output, "output_1"
\ No newline at end of file
+        return output, "output_1"
+
+
+class MostSimilarDocumentsPipeline(BaseStandardPipeline):
+    def __init__(self, document_store: BaseDocumentStore):
+        """
+        Initialize a Pipeline for finding similar documents using document ids.
+
+        :param document_store: Document Store instance
+        """
+        self.document_store = document_store
+
+    def run(self, document_ids: List[str], top_k: int = 5):
+        """
+        :param document_ids: document ids
+        :param top_k: How many documents id to return against single document
+        """
+        return_ids: list = []
+        self.document_store.return_embedding = True  # type: ignore
+        for document in self.document_store.get_documents_by_id(ids=document_ids):
+            found_ids: list = []
+            for found_document in self.document_store.query_by_embedding(query_emb=document.embedding,
+                                                                         return_embedding=False,
+                                                                         top_k=top_k):
+                found_ids.append(found_document.id)
+            return_ids.append(found_ids)
+
+        return return_ids
diff --git a/test/test_pipeline.py b/test/test_pipeline.py
index ac8b28e074..e082b249b2 100644
--- a/test/test_pipeline.py
+++ b/test/test_pipeline.py
@@ -14,6 +14,7 @@
     RootNode,
     SklearnQueryClassifier,
     TransformersQueryClassifier,
+    MostSimilarDocumentsPipeline,
 )
 from haystack.retriever.dense import DensePassageRetriever
 from haystack.retriever.sparse import ElasticsearchRetriever
@@ -547,3 +548,45 @@ def run(self, **kwargs):
 
     output = pipeline.run(query="How old is John?")
     assert output["output"] == "question"
+
+
+@pytest.mark.elasticsearch
+@pytest.mark.parametrize(
+        "retriever,document_store",
+        [
+            ("embedding", "faiss"),
+            ("embedding", "milvus"),
+            ("embedding", "elasticsearch"),
+        ],
+        indirect=True,
+)
+def test_document_search_pipeline(retriever, document_store):
+    documents = [
+        {"text": "Sample text for document-1", "meta": {"source": "wiki1"}},
+        {"text": "Sample text for document-2", "meta": {"source": "wiki2"}},
+        {"text": "Sample text for document-3", "meta": {"source": "wiki3"}},
+        {"text": "Sample text for document-4", "meta": {"source": "wiki4"}},
+        {"text": "Sample text for document-5", "meta": {"source": "wiki5"}},
+    ]
+
+    document_store.write_documents(documents)
+    document_store.update_embeddings(retriever)
+
+    docs_id: list = []
+    for index, doc in enumerate(document_store.get_all_documents(return_embedding=False)):
+        if index == 2:
+            break
+        docs_id.append(doc.id)
+
+    pipeline = MostSimilarDocumentsPipeline(document_store=document_store)
+    output = pipeline.run(document_ids=docs_id)
+
+    assert isinstance(output, list)
+    assert isinstance(output, list)
+    assert len(output) == len(docs_id)
+
+    for item in output:
+        assert isinstance(item, list)
+        if len(item) > 0:
+            for _id in item:
+                assert isinstance(_id, str)


Code review comments:
diff_hunk: @@ -1287,4 +1288,31 @@ def run(self, query, documents, **kwargs):
         # Pass also the other incoming kwargs so that future nodes still have access to it
         output.update(**kwargs)
 
-        return output, "output_1"
\ No newline at end of file
+        return output, "output_1"
+
+
+class MostSimilarDocumentsPipeline(BaseStandardPipeline):
+    def __init__(self, document_store: BaseDocumentStore):
+        """
+        Initialize a Pipeline for finding similar documents using document ids.
+
+        :param document_store: Document Store instance
+        """
+        self.document_store = document_store
+
+    def run(self, document_ids: List[str], top_k: int = 5):
+        """
+        :param document_ids: document ids
+        :param top_k: How many documents id to return against single document
+        """
+        return_ids: list = []
+        self.document_store.return_embedding = True  # type: ignore
+        for document in self.document_store.get_documents_by_id(ids=document_ids):
+            found_ids: list = []
+            for found_document in self.document_store.query_by_embedding(query_emb=document.embedding,
+                                                                         return_embedding=False,
+                                                                         top_k=top_k):
+                found_ids.append(found_document.id)
+            return_ids.append(found_ids)
path: haystack/pipeline.py
text: @user1:
Let's rather return a list of Documents here than just the ids

@author:
Done.

diff_hunk: @@ -547,3 +548,45 @@ def run(self, **kwargs):
 
     output = pipeline.run(query="How old is John?")
     assert output["output"] == "question"
+
+
+@pytest.mark.elasticsearch
+@pytest.mark.parametrize(
+        "retriever,document_store",
+        [
+            ("embedding", "faiss"),
+            ("embedding", "milvus"),
+            ("embedding", "elasticsearch"),
+        ],
+        indirect=True,
+)
+def test_document_search_pipeline(retriever, document_store):
+    documents = [
+        {"text": "Sample text for document-1", "meta": {"source": "wiki1"}},
+        {"text": "Sample text for document-2", "meta": {"source": "wiki2"}},
+        {"text": "Sample text for document-3", "meta": {"source": "wiki3"}},
+        {"text": "Sample text for document-4", "meta": {"source": "wiki4"}},
+        {"text": "Sample text for document-5", "meta": {"source": "wiki5"}},
+    ]
+
+    document_store.write_documents(documents)
+    document_store.update_embeddings(retriever)
+
+    docs_id: list = []
+    for index, doc in enumerate(document_store.get_all_documents(return_embedding=False)):
path: test/test_pipeline.py
text: @user1:
I believe this for loop is not really needed. We can define the ids above together with the rest of the document content `{"id"="a","text": "Sample text for document-1", "meta": {"source": "wiki1"}}`

@author:
That totally slipped my mind. Done.

diff_hunk: @@ -547,3 +548,45 @@ def run(self, **kwargs):
 
     output = pipeline.run(query="How old is John?")
     assert output["output"] == "question"
+
+
+@pytest.mark.elasticsearch
+@pytest.mark.parametrize(
+        "retriever,document_store",
+        [
+            ("embedding", "faiss"),
+            ("embedding", "milvus"),
+            ("embedding", "elasticsearch"),
+        ],
+        indirect=True,
+)
+def test_document_search_pipeline(retriever, document_store):
+    documents = [
+        {"text": "Sample text for document-1", "meta": {"source": "wiki1"}},
+        {"text": "Sample text for document-2", "meta": {"source": "wiki2"}},
+        {"text": "Sample text for document-3", "meta": {"source": "wiki3"}},
+        {"text": "Sample text for document-4", "meta": {"source": "wiki4"}},
+        {"text": "Sample text for document-5", "meta": {"source": "wiki5"}},
+    ]
+
+    document_store.write_documents(documents)
+    document_store.update_embeddings(retriever)
+
+    docs_id: list = []
+    for index, doc in enumerate(document_store.get_all_documents(return_embedding=False)):
+        if index == 2:
+            break
+        docs_id.append(doc.id)
+
+    pipeline = MostSimilarDocumentsPipeline(document_store=document_store)
+    output = pipeline.run(document_ids=docs_id)
+
+    assert isinstance(output, list)
+    assert isinstance(output, list)
path: test/test_pipeline.py
text: @user1:
Twice the same assert

@author:
Oops.