fix: ElasticSearch match search_term against metadata field value, not content (#3646)

Co-authored-by: Sebastian Husch Lee <10526848+sjrl@users.noreply.github.com>

Author: davidsbatista

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

@@ -1845,19 +1845,22 @@ def get_metadata_field_unique_values(
         self,
         metadata_field: str,
         search_term: str | None = None,
-        size: int | None = 10000,
+        size: int | None = 10,
         after: dict[str, Any] | None = None,
     ) -> tuple[list[str], dict[str, Any] | None]:
         """
-        Returns unique values for a metadata field, optionally filtered by a search term in the content.
+        Returns unique values for a metadata field, optionally filtered by a substring match on the field's value.
 
         Uses composite aggregations for proper pagination beyond 10k results.
 
         See: https://www.elastic.co/docs/reference/aggregations/search-aggregations-bucket-composite-aggregation
 
         :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 size: The number of unique values to return per page. Defaults to 10000.
+        :param search_term: Optional term to filter the returned values by, matching as a case-insensitive substring
+            of the metadata field's own value (not the document content). NOTE: The matching is done with a server-side
+            script to accomplish the substring matching on the value of the field and this operation is quite expensive
+            for a large corpus.
+        :param size: The number of unique values to return per page. Defaults to 10.
         :param after: Optional pagination key from the previous response. Use None for the first page.
             For subsequent pages, pass the `after_key` from the previous response.
         :returns: A tuple containing (list of unique values, after_key for pagination).
@@ -1868,12 +1871,6 @@ def get_metadata_field_unique_values(
 
         field_name = _normalize_metadata_field_name(metadata_field)
 
-        # filter by search_term if provided
-        query: dict[str, Any] = {"match_all": {}}
-        if search_term:
-            # Use match_phrase for exact phrase matching to avoid tokenization issues
-            query = {"match_phrase": {"content": search_term}}
-
         # Build composite aggregation for proper pagination
         composite_agg: dict[str, Any] = {
             "size": size,
@@ -1882,15 +1879,32 @@ def get_metadata_field_unique_values(
         if after is not None:
             composite_agg["after"] = after
 
-        body = {
-            "query": query,
+        body: dict[str, Any] = {
             "aggs": {
                 "unique_values": {
                     "composite": composite_agg,
                 }
             },
             "size": 0,  # we only need aggregations, not documents
         }
+        if search_term:
+            # Composite aggregation terms sources don't support `include`/`exclude` (that's only valid on
+            # standalone `terms` aggregations), and a `regexp` query only works on keyword/text fields, not
+            # numeric ones. A doc-value script query works uniformly across field types by stringifying the
+            # field's value, matching the case-insensitive substring semantics documented above. The term is
+            # lower-cased once here rather than per-document in the script.
+            body["query"] = {
+                "script": {
+                    "script": {
+                        "source": (
+                            "def v = doc[params.field]; "
+                            "if (v.size() == 0) { return false; } "
+                            "return v.value.toString().toLowerCase().contains(params.term)"
+                        ),
+                        "params": {"field": field_name, "term": search_term.lower()},
+                    }
+                }
+            }
 
         result = self.client.search(index=self._index, body=body)
         aggregations = result.get("aggregations", {})
@@ -1912,19 +1926,23 @@ async def get_metadata_field_unique_values_async(
         self,
         metadata_field: str,
         search_term: str | None = None,
-        size: int | None = 10000,
+        size: int | None = 10,
         after: dict[str, Any] | None = None,
     ) -> tuple[list[str], dict[str, Any] | None]:
         """
-        Asynchronously returns unique values for a metadata field, optionally filtered by a search term in the content.
+        Asynchronously returns unique values for a metadata field.
 
+        Optionally filtered by a substring match on the field's own value.
         Uses composite aggregations for proper pagination beyond 10k results.
 
         See: https://www.elastic.co/docs/reference/aggregations/search-aggregations-bucket-composite-aggregation
 
         :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 size: The number of unique values to return per page. Defaults to 10000.
+        :param search_term: Optional term to filter the returned values by, matching as a case-insensitive substring
+            of the metadata field's own value (not the document content). NOTE: The matching is done with a server-side
+            script to accomplish the substring matching on the value of the field and this operation is quite expensive
+            for a large corpus.
+        :param size: The number of unique values to return per page. Defaults to 10.
         :param after: Optional pagination key from the previous response. Use None for the first page.
             For subsequent pages, pass the `after_key` from the previous response.
         :returns: A tuple containing (list of unique values, after_key for pagination).
@@ -1935,12 +1953,6 @@ async def get_metadata_field_unique_values_async(
 
         field_name = _normalize_metadata_field_name(metadata_field)
 
-        # filter by search_term if provided
-        query: dict[str, Any] = {"match_all": {}}
-        if search_term:
-            # Use match_phrase for exact phrase matching to avoid tokenization issues
-            query = {"match_phrase": {"content": search_term}}
-
         # Build composite aggregation for proper pagination
         composite_agg: dict[str, Any] = {
             "size": size,
@@ -1949,15 +1961,32 @@ async def get_metadata_field_unique_values_async(
         if after is not None:
             composite_agg["after"] = after
 
-        body = {
-            "query": query,
+        body: dict[str, Any] = {
             "aggs": {
                 "unique_values": {
                     "composite": composite_agg,
                 }
             },
             "size": 0,  # we only need aggregations, not documents
         }
+        if search_term:
+            # Composite aggregation terms sources don't support `include`/`exclude` (that's only valid on
+            # standalone `terms` aggregations), and a `regexp` query only works on keyword/text fields, not
+            # numeric ones. A doc-value script query works uniformly across field types by stringifying the
+            # field's value, matching the case-insensitive substring semantics documented above. The term is
+            # lower-cased once here rather than per-document in the script.
+            body["query"] = {
+                "script": {
+                    "script": {
+                        "source": (
+                            "def v = doc[params.field]; "
+                            "if (v.size() == 0) { return false; } "
+                            "return v.value.toString().toLowerCase().contains(params.term)"
+                        ),
+                        "params": {"field": field_name, "term": search_term.lower()},
+                    }
+                }
+            }
 
         result = await self.async_client.search(index=self._index, body=body)
         aggregations = result.get("aggregations", {})

integrations/elasticsearch/tests/test_document_store.py

@@ -1290,13 +1290,13 @@ def test_get_metadata_field_unique_values(self, document_store: ElasticsearchDoc
         # Should have no more results
         assert after_key_page2 is None
 
-        # Test with search term - filter by content matching "Python"
-        unique_values_filtered, _ = document_store.get_metadata_field_unique_values("meta.category", "Python", 10)
-        assert set(unique_values_filtered) == {"A"}  # Only category A has documents with "Python" in content
+        # Test with search term - substring match against the target field's own value (not the content)
+        # "Java" is a substring of both "Java" and "JavaScript"
+        unique_languages_filtered, _ = document_store.get_metadata_field_unique_values("meta.language", "Java", 10)
+        assert set(unique_languages_filtered) == {"Java", "JavaScript"}
 
-        # Test with search term - filter by content matching "Java"
-        unique_values_java, _ = document_store.get_metadata_field_unique_values("meta.category", "Java", 10)
-        assert set(unique_values_java) == {"B"}  # Only category B has documents with "Java" in content
+        unique_languages_python, _ = document_store.get_metadata_field_unique_values("meta.language", "Python", 10)
+        assert set(unique_languages_python) == {"Python"}
 
         # Test with integer values
         int_docs = [
@@ -1309,10 +1309,43 @@ def test_get_metadata_field_unique_values(self, document_store: ElasticsearchDoc
         unique_priorities, _ = document_store.get_metadata_field_unique_values("meta.priority", None, 10)
         assert set(unique_priorities) == {"1", "2", "3"}
 
-        # Test with search term on integer field
-        unique_priorities_filtered, _ = document_store.get_metadata_field_unique_values("meta.priority", "Doc 1", 10)
+        # Test with search term on integer field - substring match against the field's own (stringified) value
+        unique_priorities_filtered, _ = document_store.get_metadata_field_unique_values("meta.priority", "1", 10)
         assert set(unique_priorities_filtered) == {"1"}
 
+    def test_get_metadata_field_unique_values_search_term_matches_field_value_not_content(
+        self, document_store: ElasticsearchDocumentStore
+    ):
+        """
+        `search_term` must filter by substring match on the metadata field's own value, not by matching
+        against the document's `content`.
+        """
+        docs = [
+            # "Python" appears in the content but NOT in the category value -> must be EXCLUDED
+            Document(content="Python programming guide", meta={"category": "Backend"}),
+            # "Python" appears in the category value but NOT in the content -> must be INCLUDED
+            Document(content="General purpose scripting language", meta={"category": "Python-based"}),
+        ]
+        document_store.write_documents(docs)
+
+        unique_values, _ = document_store.get_metadata_field_unique_values("meta.category", "Python", 10)
+
+        assert unique_values == ["Python-based"]
+
+    def test_get_metadata_field_unique_values_search_term_case_insensitive(
+        self, document_store: ElasticsearchDocumentStore
+    ):
+        docs = [
+            Document(content="n/a", meta={"category": "Python-based"}),
+            Document(content="n/a", meta={"category": "Java-based"}),
+        ]
+        document_store.write_documents(docs)
+
+        unique_values, _ = document_store.get_metadata_field_unique_values("meta.category", "PYTHON", 10)
+
+        assert unique_values == ["Python-based"]
+        assert "Backend" not in unique_values
+
     def test_query_sql(self, document_store: ElasticsearchDocumentStore):
         docs = [
             Document(content="Python programming", meta={"category": "A", "status": "active", "priority": 1}),

integrations/elasticsearch/tests/test_document_store_async.py

@@ -364,3 +364,24 @@ async def test_query_sql_async_error_handling(self, document_store: Elasticsearc
         invalid_query = "SELECT * FROM non_existent_index"
         with pytest.raises(DocumentStoreError, match="Failed to execute SQL query"):
             await document_store._query_sql_async(invalid_query)
+
+    @pytest.mark.asyncio
+    async def test_get_metadata_field_unique_values_async_search_term_matches_field_value_not_content(
+        self, document_store: ElasticsearchDocumentStore
+    ):
+        """
+        `search_term` must filter by substring match on the metadata field's own value, not by matching
+        against the document's `content`.
+        """
+        docs = [
+            # "Python" appears in the content but NOT in the category value -> must be EXCLUDED
+            Document(content="Python programming guide", meta={"category": "Backend"}),
+            # "Python" appears in the category value but NOT in the content -> must be INCLUDED
+            Document(content="General purpose scripting language", meta={"category": "Python-based"}),
+        ]
+        await document_store.write_documents_async(docs)
+
+        unique_values, _ = await document_store.get_metadata_field_unique_values_async("meta.category", "Python", 10)
+
+        assert unique_values == ["Python-based"]
+        assert "Backend" not in unique_values