feat: opensearch delete and query operations

Author: zakahan

veadk/database/vector/opensearch_vector_database.py

@@ -186,7 +186,7 @@ def query(self, query: str, **kwargs: Any) -> list[str]:
         assert collection_name is not None, "Collection name is required."
         if not self._opensearch_client.indices.exists(index=collection_name):
             logger.warning(
-                f"querying {query}, but collection {collection_name} does not exist. retun a empty list."
+                f"querying {query}, but collection {collection_name} does not exist. return a empty list."
             )
             return []
         query_vector = self._embedding_client.embed_query(query)
@@ -196,10 +196,43 @@ def query(self, query: str, **kwargs: Any) -> list[str]:
 
     @override
     def delete(self, collection_name: str, **kwargs: Any):
+        """drop index"""
         if not self._opensearch_client.indices.exists(index=collection_name):
             raise ValueError(f"Collection {collection_name} does not exist.")
         self._opensearch_client.indices.delete(index=collection_name)
 
     def is_empty(self, collection_name: str):
         response = self._opensearch_client.count(index=collection_name)
         return response["count"] == 0
+
+    def collection_exists(self, collection_name: str):
+        return self._opensearch_client.indices.exists(index=collection_name)
+
+    def list_all_collection(self):
+        """List all index name of OpenSearch."""
+        response = self._opensearch_client.indices.get_alias()
+        return list(response.keys())
+
+    def get_all_docs(self, collection_name: str, size: int = 10000):
+        """Match all docs in one index of OpenSearch"""
+        if not self.collection_exists(collection_name):
+            logger.warning(
+                f"Get all docs, but collection {collection_name} does not exist. return a empty list."
+            )
+            return []
+
+        query = {"size": size, "query": {"match_all": {}}}
+        response = self._opensearch_client.search(index=collection_name, body=query)
+        return [hit["_source"]["page_content"] for hit in response["hits"]["hits"]]
+
+    def delete_by_query(self, collection_name: str, query: str):
+        """Delete docs by query in one index of OpenSearch"""
+        if not self.collection_exists(collection_name):
+            raise ValueError(f"Collection {collection_name} does not exist.")
+
+        query = {"query": {"match": {"page_content": query}}}
+        response = self._opensearch_client.delete_by_query(
+            index=collection_name, body=query
+        )
+        self._opensearch_client.indices.refresh(index=collection_name)
+        return response