Repository: deepset-ai/haystack
Pull request: Upgrade `weaviate-client` to `3.3.3` and fix `get_all_documents`
Closes #1893, closes #2357 

The issue is about `get_all_documents` returning always max. 100 documents, regardless of the amount of docs contained by the docstore that match the query. To overcome this limitation, which seems to be somewhat inherent to Weaviate, I made `_get_all_documents_in_index` request the next page until it has the same number of docs as reported by `get_document_count`.

It also required to bump `weaviate-client` to `3.3.0` and Weaviate itself to `1.8.0`. It's therefore a **breaking change**.

---

Notes: 
- Bump for `weaviate-client`: https://weaviate-python-client.readthedocs.io/en/latest/changelog.html?highlight=pagination#version-3-3-0 
- Bump for Weaviate: https://github.com/semi-technologies/weaviate/releases/tag/v1.8.0


Problem statement:
WeaviateDocumentStore.get_all_documents does not return all documents
**Describe the bug**

Doing some experimenting with Weaviate+Haystack and running into what looks like a problem with the way `WeaviateDocumentStore.get_all_documents` is implemented. Does not look like it is actually retrieving all documents, but rather only the first 100.  This results in subsequent calls to `document_store.update_embeddings(retriever, ...)` that does not generate embeddings for all the data, rather just the first 100 documents as well.

**Expected behavior**
`WeaviateDocumentStore.get_all_documents` returns all documents, `document_store.update_embeddings(retriever, ...)` updates embeddings for all documents

**To Reproduce**
Following example shows the problem:
```
    document_store = WeaviateDocumentStore()

    # Write some documents (~500 in this case)
    document_store.write_documents(docs, batch_size=50)

    print(f"get_document_count is {document_store.get_document_count()}")

    docs = document_store.get_all_documents()
    print(f"Actual count of all documents {len(docs)}")

    # only first 100 documents have vectors updated...
    document_store.update_embeddings(retriever, batch_size=200)
```

Output from this shows that all 500 documents are stored, but only the first 100 documents are returned by the method.
```
get_document_count is 502
Actual count of all documents 100
```

Let me know if I'm doing something wrong, but seems like this isn't working currently.  thanks!


**System:**
 - OS: macOS
 - GPU/CPU: CPU
 - Haystack version (commit or version number): 0.10 and verified also occurs in 1.0
 - DocumentStore: Weaviate
 - Reader: N/A
 - Retriever: DensePassageRetriever


Weaviate Version 3.3.0
I'm unsure what the consequences (if any) would be to change the version of Weaviate but while building a demo colab notebook with the folks at Semi Technologies they pointed out that they are on v 3.3.0 and Haystack requires 2.5.0
Would it be a good idea to upgrade?

Hints from issue discussion:
Hello @cjb06776, thank you for this bug report. I'm going to try replicate this and be back with some feedback asap.
Yes, I confirm I can replicate this issue. Unfortunately I don't have a quick fix for it, but I'll let you know when a patch is out. In the meantime I can only recommend you to temporarily choose another document store, if it's feasible.
As I expected, this is a "feature" of Weaviate: https://weaviate-python-client.readthedocs.io/en/latest/weaviate.data.html?highlight=100#weaviate.data.DataObject.get. I'll have to work it around.

If anybody comes across this and knows of a way to ask Weaviate to remove this limit, by all means please reply to this thread, even if closed. For now I'll perform several queries until the results of `get_document_count` and `get_all_documents` match :confused: 
So, quick summary: the issue is solvable, but it will require a bump of both Weaviate and `weaviate-client` to 1.8.0 and 3.3.0, respectively. Unfortunately this means that this is a breaking change, so I'm not sure I will be able to merge it right away.

Commit message: Update Documentation & Code Style

Patch to review:
diff --git a/.github/workflows/linux_ci.yml b/.github/workflows/linux_ci.yml
index 530626e8e5..fb4e090959 100644
--- a/.github/workflows/linux_ci.yml
+++ b/.github/workflows/linux_ci.yml
@@ -261,7 +261,7 @@ jobs:
         sudo docker-compose ps
 
     - name: Run Weaviate
-      run: docker run -d -p 8080:8080 --name haystack_test_weaviate --env AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED='true' --env PERSISTENCE_DATA_PATH='/var/lib/weaviate' semitechnologies/weaviate:1.7.2
+      run: docker run -d -p 8080:8080 --name haystack_test_weaviate --env AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED='true' --env PERSISTENCE_DATA_PATH='/var/lib/weaviate' semitechnologies/weaviate:1.8.0
 
     - name: Run GraphDB
       run: docker run -d -p 7200:7200 --name haystack_test_graphdb deepset/graphdb-free:9.4.1-adoptopenjdk11
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index c13b50fe49..8caf7d02ae 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -48,7 +48,7 @@ You can launch them like this:
 ```
 docker run -d -p 9200:9200 -e "discovery.type=single-node" -e "ES_JAVA_OPTS=-Xms128m -Xmx128m" elasticsearch:7.9.2
 docker run -d -p 19530:19530 -p 19121:19121 milvusdb/milvus:1.1.0-cpu-d050721-5e559c
-docker run -d -p 8080:8080 --name haystack_test_weaviate --env AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED='true' --env PERSISTENCE_DATA_PATH='/var/lib/weaviate' semitechnologies/weaviate:1.7.2
+docker run -d -p 8080:8080 --name haystack_test_weaviate --env AUTHENTICATION_ANONYMOUS_ACCESS_ENABLED='true' --env PERSISTENCE_DATA_PATH='/var/lib/weaviate' semitechnologies/weaviate:1.8.0
 docker run -d -p 7200:7200 --name haystack_test_graphdb deepset/graphdb-free:9.4.1-adoptopenjdk11
 docker run -d -p 9998:9998 -e "TIKA_CHILD_JAVA_OPTS=-JXms128m" -e "TIKA_CHILD_JAVA_OPTS=-JXmx128m" apache/tika:1.24.1
 ```
diff --git a/docs/_src/api/api/document_store.md b/docs/_src/api/api/document_store.md
index 5fec82f662..390c856204 100644
--- a/docs/_src/api/api/document_store.md
+++ b/docs/_src/api/api/document_store.md
@@ -3180,6 +3180,22 @@ def get_all_documents(index: Optional[str] = None, filters: Optional[Dict[str, U
 
 Get documents from the document store.
 
+Note this limitation from the changelog of Weaviate 1.8.0:
+
+.. quote::
+    Due to the increasing cost of each page outlined above, there is a limit to
+    how many objects can be retrieved using pagination. By default setting the sum
+    of offset and limit to higher than 10,000 objects, will lead to an error.
+    If you must retrieve more than 10,000 objects, you can increase this limit by
+    setting the environment variable `QUERY_MAXIMUM_RESULTS=<desired-value>`.
+
+    Warning: Setting this to arbitrarily high values can make the memory consumption
+    of a single query explode and single queries can slow down the entire cluster.
+    We recommend setting this value to the lowest possible value that does not
+    interfere with your users' expectations.
+
+(https://github.com/semi-technologies/weaviate/releases/tag/v1.8.0)
+
 **Arguments**:
 
 - `index`: Name of the index to get the documents from. If None, the
@@ -3226,6 +3242,22 @@ Get documents from the document store. Under-the-hood, documents are fetched in
 document store and yielded as individual documents. This method can be used to iteratively process
 a large number of documents without having to load all documents in memory.
 
+Note this limitation from the changelog of Weaviate 1.8.0:
+
+.. quote::
+    Due to the increasing cost of each page outlined above, there is a limit to
+    how many objects can be retrieved using pagination. By default setting the sum
+    of offset and limit to higher than 10,000 objects, will lead to an error.
+    If you must retrieve more than 10,000 objects, you can increase this limit by
+    setting the environment variable `QUERY_MAXIMUM_RESULTS=<desired-value>`.
+
+    Warning: Setting this to arbitrarily high values can make the memory consumption
+    of a single query explode and single queries can slow down the entire cluster.
+    We recommend setting this value to the lowest possible value that does not
+    interfere with your users' expectations.
+
+(https://github.com/semi-technologies/weaviate/releases/tag/v1.8.0)
+
 **Arguments**:
 
 - `index`: Name of the index to get the documents from. If None, the
diff --git a/haystack/document_stores/weaviate.py b/haystack/document_stores/weaviate.py
index d096399f62..a0b35cee94 100644
--- a/haystack/document_stores/weaviate.py
+++ b/haystack/document_stores/weaviate.py
@@ -17,7 +17,7 @@
 
 try:
     from weaviate import client, AuthClientPassword
-    from weaviate import ObjectsBatchRequest
+    from weaviate.batch import Batch
 except (ImportError, ModuleNotFoundError) as ie:
     from haystack.utils.import_utils import _optional_component_not_installed
 
@@ -458,7 +458,7 @@ def write_documents(
         batched_documents = get_batches_from_generator(document_objects, batch_size)
         with tqdm(total=len(document_objects), disable=not self.progress_bar) as progress_bar:
             for document_batch in batched_documents:
-                docs_batch = ObjectsBatchRequest()
+                # docs_batch = Batch(self.weaviate_client)
                 for idx, doc in enumerate(document_batch):
                     _doc = {**doc.to_dict(field_map=self._create_document_field_map())}
                     _ = _doc.pop("score", None)
@@ -495,7 +495,7 @@ def write_documents(
                     docs_batch.add(_doc, class_name=index, uuid=doc_id, vector=vector)
 
                 # Ingest a batch of documents
-                results = self.weaviate_client.batch.create(docs_batch)
+                results = self.weaviate_client.batch.create_objects()
                 # Weaviate returns errors for every failed document in the batch
                 if results is not None:
                     for result in results:
@@ -571,7 +571,8 @@ def get_document_count(
 
         if "data" in result:
             if "Aggregate" in result.get("data"):
-                doc_count = result.get("data").get("Aggregate").get(index)[0]["meta"]["count"]
+                if result.get("data").get("Aggregate").get(index):
+                    doc_count = result.get("data").get("Aggregate").get(index)[0]["meta"]["count"]
 
         return doc_count
 
@@ -586,6 +587,22 @@ def get_all_documents(
         """
         Get documents from the document store.
 
+        Note this limitation from the changelog of Weaviate 1.8.0:
+
+        .. quote::
+            Due to the increasing cost of each page outlined above, there is a limit to
+            how many objects can be retrieved using pagination. By default setting the sum
+            of offset and limit to higher than 10,000 objects, will lead to an error.
+            If you must retrieve more than 10,000 objects, you can increase this limit by
+            setting the environment variable `QUERY_MAXIMUM_RESULTS=<desired-value>`.
+
+            Warning: Setting this to arbitrarily high values can make the memory consumption
+            of a single query explode and single queries can slow down the entire cluster.
+            We recommend setting this value to the lowest possible value that does not
+            interfere with your users' expectations.
+
+        (https://github.com/semi-technologies/weaviate/releases/tag/v1.8.0)
+
         :param index: Name of the index to get the documents from. If None, the
                       DocumentStore's default index (self.index) will be used.
         :param filters: Optional filters to narrow down the search space to documents whose metadata fulfill certain
@@ -651,10 +668,28 @@ def _get_all_documents_in_index(
         else:
             result = self.weaviate_client.query.get(class_name=index, properties=properties).do()
 
-        all_docs = {}
-        if result and "data" in result and "Get" in result.get("data"):
-            if result.get("data").get("Get").get(index):
-                all_docs = result.get("data").get("Get").get(index)
+        # Inherent Weaviate limitation to 100 elements forces us to loop here:
+        #   https://weaviate-python-client.readthedocs.io/en/latest/weaviate.data.html?highlight=100#weaviate.data.DataObject.get
+        base_query = self.weaviate_client.query.get(class_name=index, properties=properties)
+        while len(all_docs) < num_of_documents:
+
+            query = base_query
+            if filters:
+                filter_dict = self._build_filter_clause(filters=filters)
+                query = query.with_where(filter_dict)
+
+            if all_docs:
+                # .with_limit() must be used with .with_offset, of the latter won't work properly
+                #   https://weaviate-python-client.readthedocs.io/en/latest/weaviate.gql.html?highlight=offset#weaviate.gql.get.GetBuilder.with_offset
+                query = query.with_limit(100).with_offset(offset=len(all_docs))
+
+            result = query.do()
+
+            if result and "data" in result and "Get" in result.get("data"):
+                if result.get("data").get("Get").get(index):
+                    all_docs += result.get("data").get("Get").get(index)
+            else:
+                raise ValueError(f"Weaviate returned ad exception: {result}")
 
         yield from all_docs
 
@@ -671,6 +706,22 @@ def get_all_documents_generator(
         document store and yielded as individual documents. This method can be used to iteratively process
         a large number of documents without having to load all documents in memory.
 
+        Note this limitation from the changelog of Weaviate 1.8.0:
+
+        .. quote::
+            Due to the increasing cost of each page outlined above, there is a limit to
+            how many objects can be retrieved using pagination. By default setting the sum
+            of offset and limit to higher than 10,000 objects, will lead to an error.
+            If you must retrieve more than 10,000 objects, you can increase this limit by
+            setting the environment variable `QUERY_MAXIMUM_RESULTS=<desired-value>`.
+
+            Warning: Setting this to arbitrarily high values can make the memory consumption
+            of a single query explode and single queries can slow down the entire cluster.
+            We recommend setting this value to the lowest possible value that does not
+            interfere with your users' expectations.
+
+        (https://github.com/semi-technologies/weaviate/releases/tag/v1.8.0)
+
         :param index: Name of the index to get the documents from. If None, the
                       DocumentStore's default index (self.index) will be used.
         :param filters: Optional filters to narrow down the search space to documents whose metadata fulfill certain
diff --git a/test/test_document_store.py b/test/test_document_store.py
index 62102f6d5e..087ca55c56 100644
--- a/test/test_document_store.py
+++ b/test/test_document_store.py
@@ -159,6 +159,19 @@ def test_get_all_documents_without_filters(document_store_with_docs):
     assert {d.meta["meta_field"] for d in documents} == {"test1", "test2", "test3", "test4", "test5"}
 
 
+def test_get_all_documents_large_quantities(document_store):
+    # Test to exclude situations like Weaviate not returning more than 100 docs by default
+    #   https://github.com/deepset-ai/haystack/issues/1893
+    docs_to_write = [
+        {"meta": {"name": f"name_{i}"}, "content": f"text_{i}", "embedding": np.random.rand(768).astype(np.float32)}
+        for i in range(1000)
+    ]
+    document_store.write_documents(docs_to_write)
+    documents = document_store.get_all_documents()
+    assert all(isinstance(d, Document) for d in documents)
+    assert len(documents) == len(docs_to_write)
+
+
 def test_get_all_document_filter_duplicate_text_value(document_store):
     documents = [
         Document(content="Doc1", meta={"f1": "0"}, id_hash_keys=["meta"]),


Code review comments:
diff_hunk: @@ -458,7 +458,7 @@ def write_documents(
         batched_documents = get_batches_from_generator(document_objects, batch_size)
         with tqdm(total=len(document_objects), disable=not self.progress_bar) as progress_bar:
             for document_batch in batched_documents:
-                docs_batch = ObjectsBatchRequest()
+                # docs_batch = Batch(self.weaviate_client)
path: haystack/document_stores/weaviate.py
text: @user1:
If you plan using `weaviate-client==3.3.3`, you do not need this line too.

diff_hunk: @@ -17,7 +17,7 @@
 
 try:
     from weaviate import client, AuthClientPassword
-    from weaviate import ObjectsBatchRequest
+    from weaviate.batch import Batch
path: haystack/document_stores/weaviate.py
text: @user1:
If you plan using `weaviate-client==3.3.3`, you do not need this line.

diff_hunk: @@ -495,7 +495,7 @@ def write_documents(
                     docs_batch.add(_doc, class_name=index, uuid=doc_id, vector=vector)
path: haystack/document_stores/weaviate.py
text: @user1:
If you plan using `weaviate-client==3.3.3`, instead of `docs_batch.add` use `self.weaviate_client.batch.add_data_object`. So it should look like this:
```python
self.weaviate_client.batch.add_data_object(
    data_object=_doc,
    class_name=index,
    uuid=doc_id,
    vector=vector,
)
```