Merge branch 'main' into fix/valkey-async-mixin-tests

Author: davidsbatista

.github/workflows/faiss.yml

@@ -89,8 +89,17 @@ jobs:
           SUBPROJECT_ID: faiss
           COMMENT_ARTIFACT_NAME: coverage-comment-faiss
 
-      # No integration tests yet — add integration-cov-append-retry + combined coverage step when needed
+      - name: Run integration tests
+        run: hatch run test:integration-cov-append-retry
 
+      - name: Store combined coverage
+        if: github.event_name == 'push'
+        uses: py-cov-action/python-coverage-comment-action@7188638f871f721a365d644f505d1ff3df20d683 # v3.40
+        with:
+          GITHUB_TOKEN: ${{ github.token }}
+          COVERAGE_PATH: integrations/faiss
+          SUBPROJECT_ID: faiss-combined
+          COMMENT_ARTIFACT_NAME: coverage-comment-faiss-combined
       - name: Run unit tests with lowest direct dependencies
         if: matrix.python-version == '3.10' && runner.os == 'Linux'
         run: |

.github/workflows/qdrant.yml

@@ -93,8 +93,18 @@ jobs:
           SUBPROJECT_ID: qdrant
           COMMENT_ARTIFACT_NAME: coverage-comment-qdrant
 
-      # No integration tests yet — add integration-cov-append-retry + combined coverage step when needed
+      - name: Run integration tests
+        run: hatch run test:integration-cov-append-retry
 
+      - name: Store combined coverage
+        if: github.event_name == 'push'
+        uses: py-cov-action/python-coverage-comment-action@7188638f871f721a365d644f505d1ff3df20d683 # v3.40
+        with:
+          GITHUB_TOKEN: ${{ github.token }}
+          COVERAGE_PATH: integrations/qdrant
+          SUBPROJECT_ID: qdrant-combined
+          COMMENT_ARTIFACT_NAME: coverage-comment-qdrant-combined
+          
       - name: Run unit tests with lowest direct dependencies
         if: github.event_name != 'push'
         run: |

README.md

@@ -22,6 +22,7 @@
 from haystack_integrations.document_stores.faiss import FAISSDocumentStore
 
 
+@pytest.mark.integration
 class TestFAISSDocumentStore(
     CountDocumentsTest,
     DeleteDocumentsTest,

integrations/faiss/tests/test_document_store.py

@@ -31,6 +31,12 @@ def populated_store(document_store):
     return document_store
 
 
+def test_invalid_document_store_type():
+    with pytest.raises(ValueError, match="document_store must be an instance of FAISSDocumentStore"):
+        FAISSEmbeddingRetriever(document_store="not_a_store")  # type: ignore[arg-type]
+
+
+@pytest.mark.integration
 class TestFAISSEmbeddingRetriever:
     def test_run_with_query_embedding_only(self, populated_store):
         retriever = FAISSEmbeddingRetriever(document_store=populated_store, top_k=2)
@@ -112,10 +118,6 @@ def test_filter_policy_merge(self, populated_store):
         assert len(result["documents"]) >= 1
         assert all(d.meta["category"] == "A" for d in result["documents"])
 
-    def test_invalid_document_store_type(self):
-        with pytest.raises(ValueError, match="document_store must be an instance of FAISSDocumentStore"):
-            FAISSEmbeddingRetriever(document_store="not_a_store")  # type: ignore[arg-type]
-
     def test_run_in_pipeline(self, populated_store):
         """End-to-end: FAISSEmbeddingRetriever wired into a Haystack Pipeline."""
         retriever = FAISSEmbeddingRetriever(document_store=populated_store, top_k=2)

integrations/faiss/tests/test_embedding_retriever.py

@@ -14,7 +14,7 @@
 from haystack.document_stores.types import DuplicatePolicy
 from haystack.utils.auth import Secret
 from haystack.utils.misc import _normalize_metadata_field_name
-from opensearchpy import AsyncHttpConnection, AsyncOpenSearch, OpenSearch
+from opensearchpy import AsyncHttpConnection, AsyncOpenSearch, OpenSearch, TransportError
 from opensearchpy.helpers import async_bulk, bulk
 
 from haystack_integrations.document_stores.opensearch.auth import AsyncAWSAuth, AWSAuth
@@ -979,7 +979,27 @@ def _bm25_retrieval(
             all_terms_must_match=all_terms_must_match,
             custom_query=custom_query,
         )
-        documents = self._search_documents(search_params)
+        try:
+            documents = self._search_documents(search_params)
+        except TransportError as e:
+            if "too_many_clauses" in f"{e.info} {e.error}" and fuzziness not in (0, "0") and custom_query is None:
+                logger.warning(
+                    "BM25 query with fuzziness='{fuzziness}' exceeded OpenSearch's clause limit. "
+                    "Retrying with fuzziness=0 (exact matching). Consider reducing query length or "
+                    "setting fuzziness=0 explicitly if this occurs frequently.",
+                    fuzziness=fuzziness,
+                )
+                search_params = self._prepare_bm25_search_request(
+                    query=query,
+                    filters=filters,
+                    fuzziness=0,
+                    top_k=top_k,
+                    all_terms_must_match=all_terms_must_match,
+                    custom_query=custom_query,
+                )
+                documents = self._search_documents(search_params)
+            else:
+                raise
         OpenSearchDocumentStore._postprocess_bm25_search_results(results=documents, scale_score=scale_score)
         return documents
 
@@ -1019,7 +1039,27 @@ async def _bm25_retrieval_async(
             all_terms_must_match=all_terms_must_match,
             custom_query=custom_query,
         )
-        documents = await self._search_documents_async(search_params)
+        try:
+            documents = await self._search_documents_async(search_params)
+        except TransportError as e:
+            if "too_many_clauses" in f"{e.info} {e.error}" and fuzziness not in (0, "0") and custom_query is None:
+                logger.warning(
+                    "BM25 query with fuzziness='{fuzziness}' exceeded OpenSearch's clause limit. "
+                    "Retrying with fuzziness=0 (exact matching). Consider reducing query length or "
+                    "setting fuzziness=0 explicitly if this occurs frequently.",
+                    fuzziness=fuzziness,
+                )
+                search_params = self._prepare_bm25_search_request(
+                    query=query,
+                    filters=filters,
+                    fuzziness=0,
+                    top_k=top_k,
+                    all_terms_must_match=all_terms_must_match,
+                    custom_query=custom_query,
+                )
+                documents = await self._search_documents_async(search_params)
+            else:
+                raise
         OpenSearchDocumentStore._postprocess_bm25_search_results(results=documents, scale_score=scale_score)
         return documents
 

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

@@ -3,7 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import random
-from unittest.mock import patch
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from haystack.dataclasses.document import Document
@@ -17,7 +17,7 @@
     GetMetadataFieldsInfoTest,
     GetMetadataFieldUniqueValuesTest,
 )
-from opensearchpy.exceptions import RequestError
+from opensearchpy.exceptions import RequestError, TransportError
 
 from haystack_integrations.document_stores.opensearch import OpenSearchDocumentStore
 from haystack_integrations.document_stores.opensearch.document_store import DEFAULT_MAX_CHUNK_BYTES
@@ -224,6 +224,155 @@ def test_routing_in_delete(mock_bulk, _mock_opensearch_client):
     assert "_routing" not in actions[2]
 
 
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+def test_bm25_retrieval_retries_with_fuzziness_zero_on_too_many_clauses(_mock_opensearch_client, caplog):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._client = MagicMock()
+
+    too_many_clauses_error = TransportError(
+        500, "search_phase_execution_exception", "too_many_clauses: maxClauseCount is set to 1024"
+    )
+    store._client.search.side_effect = [
+        too_many_clauses_error,
+        {"hits": {"hits": []}},
+    ]
+
+    results = store._bm25_retrieval("a very long query", fuzziness="AUTO")
+
+    assert results == []
+    assert store._client.search.call_count == 2
+    # Verify the retry used fuzziness=0
+    second_call_body = store._client.search.call_args_list[1].kwargs["body"]
+    assert second_call_body["query"]["bool"]["must"][0]["multi_match"]["fuzziness"] == 0
+    assert "Retrying with fuzziness=0" in caplog.text
+
+
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+def test_bm25_retrieval_no_retry_when_fuzziness_already_zero(_mock_opensearch_client):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._client = MagicMock()
+
+    too_many_clauses_error = TransportError(
+        500, "search_phase_execution_exception", "too_many_clauses: maxClauseCount is set to 1024"
+    )
+    store._client.search.side_effect = too_many_clauses_error
+
+    with pytest.raises(TransportError):
+        store._bm25_retrieval("a very long query", fuzziness=0)
+
+    assert store._client.search.call_count == 1
+
+
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+def test_bm25_retrieval_no_retry_with_custom_query(_mock_opensearch_client):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._client = MagicMock()
+
+    too_many_clauses_error = TransportError(
+        500, "search_phase_execution_exception", "too_many_clauses: maxClauseCount is set to 1024"
+    )
+    store._client.search.side_effect = too_many_clauses_error
+
+    custom_query = {"query": {"match": {"content": "$query"}}}
+    with pytest.raises(TransportError):
+        store._bm25_retrieval("a very long query", fuzziness="AUTO", custom_query=custom_query)
+
+    assert store._client.search.call_count == 1
+
+
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+def test_bm25_retrieval_reraises_other_transport_errors(_mock_opensearch_client):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._client = MagicMock()
+
+    other_error = TransportError(500, "parsing_exception", {"error": {"reason": "some other error"}})
+    store._client.search.side_effect = other_error
+
+    with pytest.raises(TransportError):
+        store._bm25_retrieval("some query", fuzziness="AUTO")
+
+    assert store._client.search.call_count == 1
+
+
+@pytest.mark.asyncio
+@patch("haystack_integrations.document_stores.opensearch.document_store.AsyncOpenSearch")
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+async def test_bm25_retrieval_async_retries_with_fuzziness_zero_on_too_many_clauses(
+    _mock_opensearch_client, _mock_async_client, caplog
+):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._async_client = AsyncMock()
+
+    too_many_clauses_error = TransportError(
+        500, "search_phase_execution_exception", "too_many_clauses: maxClauseCount is set to 1024"
+    )
+    store._async_client.search.side_effect = [
+        too_many_clauses_error,
+        {"hits": {"hits": []}},
+    ]
+
+    results = await store._bm25_retrieval_async("a very long query", fuzziness="AUTO")
+
+    assert results == []
+    assert store._async_client.search.call_count == 2
+    second_call_body = store._async_client.search.call_args_list[1].kwargs["body"]
+    assert second_call_body["query"]["bool"]["must"][0]["multi_match"]["fuzziness"] == 0
+    assert "Retrying with fuzziness=0" in caplog.text
+
+
+@pytest.mark.asyncio
+@patch("haystack_integrations.document_stores.opensearch.document_store.AsyncOpenSearch")
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+async def test_bm25_retrieval_async_no_retry_when_fuzziness_already_zero(_mock_opensearch_client, _mock_async_client):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._async_client = AsyncMock()
+
+    too_many_clauses_error = TransportError(
+        500, "search_phase_execution_exception", "too_many_clauses: maxClauseCount is set to 1024"
+    )
+    store._async_client.search.side_effect = too_many_clauses_error
+
+    with pytest.raises(TransportError):
+        await store._bm25_retrieval_async("a very long query", fuzziness=0)
+
+    assert store._async_client.search.call_count == 1
+
+
+@pytest.mark.asyncio
+@patch("haystack_integrations.document_stores.opensearch.document_store.AsyncOpenSearch")
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+async def test_bm25_retrieval_async_no_retry_with_custom_query(_mock_opensearch_client, _mock_async_client):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._async_client = AsyncMock()
+
+    too_many_clauses_error = TransportError(
+        500, "search_phase_execution_exception", "too_many_clauses: maxClauseCount is set to 1024"
+    )
+    store._async_client.search.side_effect = too_many_clauses_error
+
+    custom_query = {"query": {"match": {"content": "$query"}}}
+    with pytest.raises(TransportError):
+        await store._bm25_retrieval_async("a very long query", fuzziness="AUTO", custom_query=custom_query)
+
+    assert store._async_client.search.call_count == 1
+
+
+@pytest.mark.asyncio
+@patch("haystack_integrations.document_stores.opensearch.document_store.AsyncOpenSearch")
+@patch("haystack_integrations.document_stores.opensearch.document_store.OpenSearch")
+async def test_bm25_retrieval_async_reraises_other_transport_errors(_mock_opensearch_client, _mock_async_client):
+    store = OpenSearchDocumentStore(hosts="testhost")
+    store._async_client = AsyncMock()
+
+    other_error = TransportError(500, "parsing_exception", {"error": {"reason": "some other error"}})
+    store._async_client.search.side_effect = other_error
+
+    with pytest.raises(TransportError):
+        await store._bm25_retrieval_async("some query", fuzziness="AUTO")
+
+    assert store._async_client.search.call_count == 1
+
+
 @pytest.mark.integration
 class TestDocumentStore(
     CountDocumentsByFilterTest,
@@ -333,6 +482,33 @@ def test_bm25_retrieval_with_fuzziness(
         assert "functional" in res[1].content
         assert "functional" in res[2].content
 
+    def test_bm25_retrieval_with_fuzziness_overflow(self, document_store: OpenSearchDocumentStore, caplog):
+        """
+        Test that a long query with fuzziness="AUTO" that exceeds OpenSearch's maxClauseCount
+        is automatically retried with fuzziness=0 instead of raising an error.
+        """
+        # Build an index vocabulary of similar 5-character words. With fuzziness="AUTO",
+        # 5-char words get edit distance 1, so each query term fuzzy-matches many similar
+        # indexed terms, causing clause expansion beyond the default maxClauseCount (1024).
+        # With fuzziness=0, each term produces exactly 1 clause, staying well under the limit.
+        words = [f"foo{chr(97 + i)}{chr(97 + j)}" for i in range(20) for j in range(26)]  # 520 words
+
+        chunk_size = 52
+        docs = [
+            Document(content=" ".join(words[i : i + chunk_size]), id=str(idx))
+            for idx, i in enumerate(range(0, len(words), chunk_size))
+        ]
+        document_store.write_documents(docs)
+
+        # Query with a subset of words. With fuzziness="AUTO", each 5-char term expands
+        # to match ~45 similar indexed terms, pushing total clauses well above 1024.
+        long_query = " ".join(words[:100])
+
+        # This should not raise: the too_many_clauses error is caught and retried with fuzziness=0
+        res = document_store._bm25_retrieval(long_query, top_k=3, fuzziness="AUTO")
+        assert isinstance(res, list)
+        assert "Retrying with fuzziness=0" in caplog.text
+
     def test_bm25_retrieval_with_filters(self, document_store: OpenSearchDocumentStore, test_documents: list[Document]):
         document_store.write_documents(test_documents)
         res = document_store._bm25_retrieval(