adding get_field_unique_values async

Author: davidsbatista

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

@@ -1352,5 +1352,66 @@ def get_field_unique_values(
 
         return unique_values, total_count
 
+    async def get_field_unique_values_async(
+        self, metadata_field: str, search_term: str | None, from_: int, size: int
+    ) -> tuple[list[str], int]:
+        """
+        Asynchronously returns unique values for a metadata field, optionally filtered by a search term in the content.
+
+        :param metadata_field: The metadata field to get unique values for.
+        :param search_term: Optional search term to filter documents by matching in the content field.
+        :param from_: The starting index for pagination.
+        :param size: The number of unique values to return.
+        :returns: A tuple containing (list of unique values, total count of unique values).
+        """
+        await self._ensure_initialized_async()
+        assert self._async_client is not None
+
+        field_name = self._normalize_metadata_field_name(metadata_field)
+
+        # filter by search_term if provided
+        query = {"match_all": {}}
+        if search_term:
+            # Use match_phrase for exact phrase matching to avoid tokenization issues
+            query = {"match_phrase": {"content": search_term}}
+
+        # Build aggregations
+        # Terms aggregation for paginated unique values
+        # Note: Terms aggregation doesn't support 'from' parameter directly,
+        # so we fetch from_ + size results and slice them
+        # Cardinality aggregation for total count
+        terms_size = from_ + size if from_ > 0 else size
+        body = {
+            "query": query,
+            "aggs": {
+                "unique_values": {
+                    "terms": {
+                        "field": field_name,
+                        "size": terms_size,
+                    }
+                },
+                "total_count": {
+                    "cardinality": {
+                        "field": field_name,
+                    }
+                },
+            },
+            "size": 0,  # we only need aggregations, not documents
+        }
+
+        result = await self._async_client.search(index=self._index, body=body)
+        aggregations = result.get("aggregations", {})
+
+        # Extract unique values from terms aggregation buckets
+        unique_values_buckets = aggregations.get("unique_values", {}).get("buckets", [])
+        # Apply pagination by slicing the results
+        paginated_buckets = unique_values_buckets[from_ : from_ + size]
+        unique_values = [str(bucket["key"]) for bucket in paginated_buckets]
+
+        # Extract total count from cardinality aggregation
+        total_count = int(aggregations.get("total_count", {}).get("value", 0))
+
+        return unique_values, total_count
+
     def query_sql(self, query: str):
         pass

integrations/opensearch/tests/test_document_store_async.py

@@ -499,4 +499,66 @@ async def test_get_field_min_max(self, document_store: OpenSearchDocumentStore):
         assert min_max_score["min"] == pytest.approx(5.2)
         assert min_max_score["max"] == pytest.approx(20.3)
 
+    @pytest.mark.asyncio
+    async def test_get_field_unique_values(self, document_store: OpenSearchDocumentStore):
+        # Test with string values
+        docs = [
+            Document(content="Python programming", meta={"category": "A", "language": "Python"}),
+            Document(content="Java programming", meta={"category": "B", "language": "Java"}),
+            Document(content="Python scripting", meta={"category": "A", "language": "Python"}),
+            Document(content="JavaScript development", meta={"category": "C", "language": "JavaScript"}),
+            Document(content="Python data science", meta={"category": "A", "language": "Python"}),
+            Document(content="Java backend", meta={"category": "B", "language": "Java"}),
+        ]
+        await document_store.write_documents_async(docs)
+
+        # Test getting all unique values without search term
+        unique_values, total_count = await document_store.get_field_unique_values_async("meta.category", None, 0, 10)
+        assert set(unique_values) == {"A", "B", "C"}
+        assert total_count == 3
+
+        # Test with "meta." prefix
+        unique_languages, lang_count = await document_store.get_field_unique_values_async("meta.language", None, 0, 10)
+        assert set(unique_languages) == {"Python", "Java", "JavaScript"}
+        assert lang_count == 3
+
+        # Test pagination - first page
+        unique_values_page1, total_count = await document_store.get_field_unique_values_async("meta.category", None, 0, 2)
+        assert len(unique_values_page1) == 2
+        assert total_count == 3
+        assert all(val in ["A", "B", "C"] for val in unique_values_page1)
+
+        # Test pagination - second page
+        unique_values_page2, total_count = await document_store.get_field_unique_values_async("meta.category", None, 2, 2)
+        assert len(unique_values_page2) == 1
+        assert total_count == 3
+        assert unique_values_page2[0] in ["A", "B", "C"]
+
+        # Test with search term - filter by content matching "Python"
+        unique_values_filtered, total_count = await document_store.get_field_unique_values_async("meta.category", "Python", 0, 10)
+        assert set(unique_values_filtered) == {"A"}  # Only category A has documents with "Python" in content
+        assert total_count == 1
+
+        # Test with search term - filter by content matching "Java"
+        unique_values_java, total_count = await document_store.get_field_unique_values_async("meta.category", "Java", 0, 10)
+        assert set(unique_values_java) == {"B"}  # Only category B has documents with "Java" in content
+        assert total_count == 1
+
+        # Test with integer values
+        int_docs = [
+            Document(content="Doc 1", meta={"priority": 1}),
+            Document(content="Doc 2", meta={"priority": 2}),
+            Document(content="Doc 3", meta={"priority": 1}),
+            Document(content="Doc 4", meta={"priority": 3}),
+        ]
+        await document_store.write_documents_async(int_docs)
+        unique_priorities, priority_count = await document_store.get_field_unique_values_async("meta.priority", None, 0, 10)
+        assert set(unique_priorities) == {"1", "2", "3"}
+        assert priority_count == 3
+
+        # Test with search term on integer field
+        unique_priorities_filtered, priority_count = await document_store.get_field_unique_values_async("meta.priority", "Doc 1", 0, 10)
+        assert set(unique_priorities_filtered) == {"1"}
+        assert priority_count == 1
+