adding sync methods + tests

Author: davidsbatista

integrations/qdrant/src/haystack_integrations/document_stores/qdrant/document_store.py

@@ -53,9 +53,8 @@ def get_batches_from_generator(iterable: List, n: int) -> Generator:
 
 class QdrantDocumentStore:
     """
-    A QdrantDocumentStore implementation that you
-    can use with any Qdrant instance: in-memory, disk-persisted, Docker-based,
-    and Qdrant Cloud Cluster deployments.
+    A QdrantDocumentStore implementation that you can use with any Qdrant instance: in-memory, disk-persisted,
+    Docker-based, and Qdrant Cloud Cluster deployments.
 
     Usage example by creating an in-memory instance:
 
@@ -65,7 +64,8 @@ class QdrantDocumentStore:
 
     document_store = QdrantDocumentStore(
         ":memory:",
-        recreate_index=True
+        recreate_index=True,
+        embedding_dim=5
     )
     document_store.write_documents([
         Document(content="This is first", embedding=[0.0]*5),
@@ -135,7 +135,7 @@ def __init__(
         payload_fields_to_index: Optional[List[dict]] = None,
     ) -> None:
         """
-        Initialize a QdrantDocumentStore.
+        Initializes a QdrantDocumentStore.
 
         :param location:
             If `":memory:"` - use in-memory Qdrant instance.
@@ -347,8 +347,7 @@ def filter_documents(
         [documentation](https://docs.haystack.deepset.ai/docs/metadata-filtering)
 
         :param filters: The filters to apply to the document list.
-        :returns:
-            A list of documents that match the given filters.
+        :returns: A list of documents that match the given filters.
         """
         # No need to initialize client here as _get_documents_generator
         # will handle client initialization internally
@@ -380,7 +379,6 @@ def write_documents(
     ) -> int:
         """
         Writes documents to Qdrant using the specified policy.
-
         The QdrantDocumentStore can handle duplicate documents based on the given policy.
         The available policies are:
         - `FAIL`: The operation will raise an error if any document already exists.
@@ -390,8 +388,7 @@ def write_documents(
         :param documents: A list of Document objects to write to Qdrant.
         :param policy: The policy for handling duplicate documents.
 
-        :returns:
-            The number of documents written to the document store.
+        :returns: The number of documents written to the document store.
         """
 
         self._initialize_client()
@@ -435,7 +432,6 @@ async def write_documents_async(
     ) -> int:
         """
         Asynchronously writes documents to Qdrant using the specified policy.
-
         The QdrantDocumentStore can handle duplicate documents based on the given policy.
         The available policies are:
         - `FAIL`: The operation will raise an error if any document already exists.
@@ -527,39 +523,69 @@ async def delete_documents_async(self, document_ids: List[str]) -> None:
                 "Called QdrantDocumentStore.delete_documents_async() on a non-existing ID",
             )
 
-    def delete_all_documents(self, recreate_index: bool = False) -> None:
+    def delete_all_documents(self, recreate_index: bool = False):
         """
-        Deletes all documents in the document store.
+        Deletes all documents from the document store.
 
-        :param recreate_index: If `True`, the index will be recreated after deletion.
-
-        It deletes the collection and recreates it to ensure all documents are removed.
+        :param recreate_index: Whether to recreate the index after deleting all documents.
         """
 
+        self._initialize_client()
+        assert self._client is not None
+
         if recreate_index:
-            self.recreate_collection(
-                self.index,
-                self.get_distance(self.similarity),
-                self.embedding_dim,
-                self.on_disk,
-                self.use_sparse_embeddings,
-                self.sparse_idf,
+
+            # get current collection config
+            collection_info = self._client.get_collection(collection_name=self.index)
+
+            """
+            x = CollectionInfo(
+                status= < CollectionStatus.GREEN: 'green' >,
+                optimizer_status = < OptimizersStatusOneOf.OK: 'ok' >, vectors_count = None, indexed_vectors_count = 0, points_count = 5, segments_count = 1, config = CollectionConfig(
+                params=CollectionParams(vectors=VectorParams(size=768, distance= < Distance.COSINE: 'Cosine' >, hnsw_config = None, quantization_config = None, on_disk = False, datatype = None, multivector_config = None), shard_number = None, sharding_method = None, replication_factor = None, write_consistency_factor = None, read_fan_out_factor = None, on_disk_payload = None, sparse_vectors = None), hnsw_config = HnswConfig(
+                m=16, ef_construct=100, full_scan_threshold=10000, max_indexing_threads=0, on_disk=None,
+                payload_m=None), optimizer_config = OptimizersConfig(deleted_threshold=0.2,
+                                                                     vacuum_min_vector_number=1000,
+                                                                     default_segment_number=0, max_segment_size=None,
+                                                                     memmap_threshold=None, indexing_threshold=20000,
+                                                                     flush_interval_sec=5,
+                                                                     max_optimization_threads=1), wal_config = WalConfig(
+                wal_capacity_mb=32,
+                wal_segments_ahead=0), quantization_config = None, strict_mode_config = None), payload_schema = {})
+            """
+
+            # recreate collection
+            self._set_up_collection(
+                collection_name=self.index,
+                embedding_dim=collection_info.config.params.vectors.size,
+                recreate_collection=True,
+                similarity=collection_info.config.params.vectors.distance.value,
+                use_sparse_embeddings=collection_info.config.params.sparse_vectors == SPARSE_VECTORS_NAME,
+                sparse_idf=(collection_info.config.params.vectors.name == SPARSE_VECTORS_NAME) and
+                           collection_info.config.params.vectors.config.hnsw_config is not None,
+                on_disk=collection_info.config.params.vectors.config.on_disk,
+                # ToDo: investigate
+                #   - CollectionInfo has payload_schema as Optional[Dict[str, PayloadSchemaType]],
+                #   - self._set_up_collection expects Optional[List[dict]]
+                payload_fields_to_index=None
             )
-        else:
-            self._initialize_client()  # _initialize_client assures the client is initialized
-            self._client.delete(  # type: ignore
+
+        try:
+            self._client.delete(
                 collection_name=self.index,
-                points_selector=rest.PointsSelectorAll(),
+                points_selector=rest.FilterSelector(
+                    filter=rest.Filter(
+                        must=[],
+                    )
+                ),
                 wait=self.wait_result_from_api,
             )
+        except Exception as e:
+            logger.warning(
+                f"Error {e} when calling QdrantDocumentStore.delete_all_documents()",
+            )
 
-    async def del_all_documents_async(self) -> None:
-        """
-        Asynchronously deletes all documents in the document store.
 
-        It deletes the collection and recreates it to ensure all documents are removed.
-        """
-        # ToDo
 
     @classmethod
     def from_dict(cls, data: Dict[str, Any]) -> "QdrantDocumentStore":
@@ -672,8 +698,10 @@ def get_documents_by_id(
         """
         Retrieves documents from Qdrant by their IDs.
 
-        :param ids: A list of document IDs to retrieve.
-        :returns: A list of documents.
+        :param ids:
+            A list of document IDs to retrieve.
+        :returns:
+            A list of documents.
         """
         documents: List[Document] = []
 
@@ -701,7 +729,8 @@ async def get_documents_by_id_async(
         """
         Retrieves documents from Qdrant by their IDs.
 
-        :param ids: A list of document IDs to retrieve.
+        :param ids:
+            A list of document IDs to retrieve.
         :returns:
             A list of documents.
         """
@@ -892,7 +921,7 @@ def _query_hybrid(
              value, all values will be used for grouping. One point can be in multiple groups.
         :param group_size: Maximum amount of points to return per group. Default is 3.
 
-        :returns: A list of Document that are most similar to `query_embedding` and `query_sparse_embedding`.
+        :returns: List of Document that are most similar to `query_embedding` and `query_sparse_embedding`.
 
         :raises QdrantStoreError:
             If the Document Store was initialized with `use_sparse_embeddings=False`.
@@ -1002,7 +1031,7 @@ async def _query_by_sparse_async(
              value, all values will be used for grouping. One point can be in multiple groups.
         :param group_size: Maximum amount of points to return per group. Default is 3.
 
-        :returns: A list of documents that are most similar to `query_sparse_embedding`.
+        :returns: List of documents that are most similar to `query_sparse_embedding`.
 
         :raises QdrantStoreError:
             If the Document Store was initialized with `use_sparse_embeddings=False`.
@@ -1082,7 +1111,7 @@ async def _query_by_embedding_async(
              value, all values will be used for grouping. One point can be in multiple groups.
         :param group_size: Maximum amount of points to return per group. Default is 3.
 
-        :returns: A list of documents that are most similar to `query_embedding`.
+        :returns: List of documents that are most similar to `query_embedding`.
         """
         await self._initialize_async_client()
         assert self._async_client is not None
@@ -1147,7 +1176,7 @@ async def _query_hybrid_async(
              value, all values will be used for grouping. One point can be in multiple groups.
         :param group_size: Maximum amount of points to return per group. Default is 3.
 
-        :returns: A list of Document that are most similar to `query_embedding` and `query_sparse_embedding`.
+        :returns: List of Document that are most similar to `query_embedding` and `query_sparse_embedding`.
 
         :raises QdrantStoreError:
             If the Document Store was initialized with `use_sparse_embeddings=False`.
@@ -1322,6 +1351,7 @@ def _set_up_collection(
             If the collection exists with incompatible settings.
         :raises ValueError:
             If the collection exists with a different similarity measure or embedding dimension.
+
         """
 
         self._initialize_client()
@@ -1519,8 +1549,9 @@ async def _handle_duplicate_documents_async(
         policy: Optional[DuplicatePolicy] = None,
     ) -> List[Document]:
         """
-        Asynchronously checks whether any of the passed documents is already existing in the chosen index and returns a
-        list of documents that are not in the index yet.
+        Asynchronously checks whether any of the passed documents is already existing
+        in the chosen index and returns a list of
+        documents that are not in the index yet.
 
         :param documents: A list of Haystack Document objects.
         :param policy: The duplicate policy to use when writing documents.
@@ -1544,9 +1575,6 @@ def _drop_duplicate_documents(self, documents: List[Document]) -> List[Document]
         """
         Drop duplicate documents based on same hash ID.
 
-        :param documents: A list of Haystack Document objects.
-
-        :returns: A list of Haystack Document objects with unique IDs.
         """
         _hash_ids: Set = set()
         _documents: List[Document] = []
@@ -1583,6 +1611,7 @@ def _prepare_collection_params(self) -> Dict[str, Any]:
     def _prepare_client_params(self) -> Dict[str, Any]:
         """
         Prepares the common parameters for client initialization.
+
         """
         return {
             "location": self.location,
@@ -1685,6 +1714,7 @@ def _process_query_point_results(
     def _process_group_results(self, groups: List[rest.PointGroup]) -> List[Document]:
         """
         Processes grouped query results from Qdrant.
+
         """
         if not groups:
             return []

integrations/qdrant/tests/test_document_store.py

@@ -5,7 +5,7 @@
 from haystack import Document
 from haystack.dataclasses import SparseEmbedding
 from haystack.document_stores.errors import DuplicateDocumentError
-from haystack.document_stores.types import DuplicatePolicy
+from haystack.document_stores.types import DuplicatePolicy, DocumentStore
 from haystack.testing.document_store import (
     CountDocumentsTest,
     DeleteDocumentsTest,
@@ -297,3 +297,41 @@ def test_set_up_collection_with_dimension_mismatch(self):
         ):
             with pytest.raises(ValueError, match="different vector size"):
                 document_store._set_up_collection("test_collection", 768, False, "cosine", False, False)
+
+    def test_delete_all_documents_no_index_recreation(self, document_store):
+        document_store._initialize_client()
+
+        # write some documents
+        docs = [Document(id=str(i)) for i in range(5)]
+        document_store.write_documents(docs)
+
+        # delete all documents without recreating the index
+        document_store.delete_all_documents(recreate_index=False)
+        assert document_store.count_documents() == 0
+
+        # ensure the collection still exists by writing documents again
+        document_store.write_documents(docs)
+        assert document_store.count_documents() == 5
+
+    def test_delete_all_documents_index_recreation(self, document_store):
+        document_store._initialize_client()
+
+        # write some documents
+        docs = [Document(id=str(i)) for i in range(5)]
+        document_store.write_documents(docs)
+
+        # get the current document_store config
+        config_before = document_store._client.get_collection(document_store.index)
+
+        # delete all documents with recreating the index
+        document_store.delete_all_documents(recreate_index=True)
+        assert document_store.count_documents() == 0
+
+        # assure that with the same config
+        config_after = document_store._client.get_collection(document_store.index)
+
+        assert config_before.config == config_after.config
+
+        # ensure the collection still exists by writing documents again
+        document_store.write_documents(docs)
+        assert document_store.count_documents() == 5