fix: replace in-place dataclass mutations in document stores (#3114)

* fix: replace in-place dataclass mutations in document stores

Replaces direct field assignments on Document instances with
dataclasses.replace(), removing DeprecationWarnings from haystack#10650.
Affected stores:

- elasticsearch: doc.score in BM25 score scaling (sync + async)
- opensearch: doc.score in BM25 score scaling
- pinecone: doc.score in filter_documents (sync + async),
  document.meta in _update_meta_for_documents and _discard_invalid_meta
- qdrant: document.score in score scaling

Part of deepset-ai/haystack#10956

* fix(pinecone): fix PLW2901 and update tests for non-mutating _discard_invalid_meta

---------

Co-authored-by: David S. Batista <dsbatista@gmail.com>

Author: SyedShahmeerAli12

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

@@ -8,6 +8,7 @@
 
 
 from collections.abc import Mapping
+from dataclasses import replace
 from typing import Any, Literal
 
 import numpy as np
@@ -904,10 +905,12 @@ def _bm25_retrieval(
         documents = self._search_documents(**body)
 
         if scale_score:
-            for doc in documents:
-                if doc.score is None:
-                    continue
-                doc.score = float(1 / (1 + np.exp(-np.asarray(doc.score / BM25_SCALING_FACTOR))))
+            documents = [
+                replace(doc, score=float(1 / (1 + np.exp(-np.asarray(doc.score / BM25_SCALING_FACTOR)))))
+                if doc.score is not None
+                else doc
+                for doc in documents
+            ]
 
         return documents
 
@@ -962,9 +965,12 @@ async def _bm25_retrieval_async(
         documents = await self._search_documents_async(**search_body)
 
         if scale_score:
-            for doc in documents:
-                if doc.score is not None:
-                    doc.score = float(1 / (1 + np.exp(-(doc.score / float(BM25_SCALING_FACTOR)))))
+            documents = [
+                replace(doc, score=float(1 / (1 + np.exp(-(doc.score / float(BM25_SCALING_FACTOR))))))
+                if doc.score is not None
+                else doc
+                for doc in documents
+            ]
 
         return documents
 

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

@@ -5,6 +5,7 @@
 # ruff: noqa: FBT001, FBT002   boolean-type-hint-positional-argument and boolean-default-value-positional-argument
 
 from collections.abc import Generator, Mapping
+from dataclasses import replace
 from math import exp
 from typing import Any, Literal
 
@@ -940,10 +941,12 @@ def _postprocess_bm25_search_results(*, results: list[Document], scale_score: bo
         if not scale_score:
             return
 
-        for doc in results:
-            if doc.score is None:
-                continue
-            doc.score = float(1 / (1 + exp(-(doc.score / float(BM25_SCALING_FACTOR)))))
+        results = [
+            replace(doc, score=float(1 / (1 + exp(-(doc.score / float(BM25_SCALING_FACTOR))))))
+            if doc.score is not None
+            else doc
+            for doc in results
+        ]
 
     def _bm25_retrieval(
         self,

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

@@ -3,6 +3,7 @@
 # SPDX-License-Identifier: Apache-2.0
 
 from copy import copy
+from dataclasses import replace
 from typing import Any, Literal
 
 from haystack import default_from_dict, default_to_dict, logging
@@ -310,8 +311,7 @@ def filter_documents(self, filters: dict[str, Any] | None = None) -> list[Docume
 
         # when simply filtering, we don't want to return any scores
         # furthermore, we are querying with a dummy vector, so the scores are meaningless
-        for doc in documents:
-            doc.score = None
+        documents = [replace(doc, score=None) for doc in documents]
 
         if len(documents) == TOP_K_LIMIT:
             logger.warning(
@@ -336,8 +336,7 @@ async def filter_documents_async(self, filters: dict[str, Any] | None = None) ->
             query_embedding=self._dummy_vector, filters=filters, top_k=TOP_K_LIMIT
         )
 
-        for doc in documents:
-            doc.score = None
+        documents = [replace(doc, score=None) for doc in documents]
 
         if len(documents) == TOP_K_LIMIT:
             logger.warning(
@@ -399,10 +398,8 @@ def _update_documents_metadata(documents: list[Document], meta: dict[str, Any])
         :param documents: List of documents to update.
         :param meta: Metadata fields to merge into each document's existing metadata.
         """
-        for document in documents:
-            if document.meta is None:
-                document.meta = {}
-            document.meta.update(meta)
+        for i, document in enumerate(documents):
+            documents[i] = replace(document, meta={**(document.meta or {}), **meta})
 
     def delete_by_filter(self, filters: dict[str, Any]) -> int:
         """
@@ -671,7 +668,7 @@ def _convert_query_result_to_documents(self, query_result: Any) -> list[Document
         return documents
 
     @staticmethod
-    def _discard_invalid_meta(document: Document) -> None:
+    def _discard_invalid_meta(document: Document) -> Document:
         """
         Remove metadata fields with unsupported types from the document.
         """
@@ -697,7 +694,9 @@ def valid_type(value: Any) -> bool:
                 )
                 logger.warning(msg)
 
-            document.meta = new_meta
+            return replace(document, meta=new_meta)
+
+        return document
 
     def _convert_documents_to_pinecone_format(
         self, documents: list[Document]
@@ -712,10 +711,9 @@ def _convert_documents_to_pinecone_format(
                 )
                 embedding = self._dummy_vector
 
-            if document.meta:
-                self._discard_invalid_meta(document)
+            filtered_meta = self._discard_invalid_meta(document).meta if document.meta else {}
 
-            metadata = dict(document.meta) if document.meta else {}
+            metadata = dict(filtered_meta) if filtered_meta else {}
 
             # we save content as metadata
             if document.content is not None:

integrations/pinecone/tests/test_document_store.py

@@ -177,13 +177,13 @@ def test_discard_invalid_meta_invalid():
             ],
         },
     )
-    PineconeDocumentStore._discard_invalid_meta(invalid_metadata_doc)
+    result = PineconeDocumentStore._discard_invalid_meta(invalid_metadata_doc)
 
-    assert invalid_metadata_doc.meta["source_id"] == "62049ba1d1e1d5ebb1f6230b0b00c5356b8706c56e0b9c36b1dfc86084cd75f0"
-    assert invalid_metadata_doc.meta["page_number"] == 1
-    assert invalid_metadata_doc.meta["split_id"] == 0
-    assert invalid_metadata_doc.meta["split_idx_start"] == 0
-    assert "_split_overlap" not in invalid_metadata_doc.meta
+    assert result.meta["source_id"] == "62049ba1d1e1d5ebb1f6230b0b00c5356b8706c56e0b9c36b1dfc86084cd75f0"
+    assert result.meta["page_number"] == 1
+    assert result.meta["split_id"] == 0
+    assert result.meta["split_idx_start"] == 0
+    assert "_split_overlap" not in result.meta
 
 
 def test_discard_invalid_meta_valid():
@@ -194,10 +194,10 @@ def test_discard_invalid_meta_valid():
             "page_number": 1,
         },
     )
-    PineconeDocumentStore._discard_invalid_meta(valid_metadata_doc)
+    result = PineconeDocumentStore._discard_invalid_meta(valid_metadata_doc)
 
-    assert valid_metadata_doc.meta["source_id"] == "62049ba1d1e1d5ebb1f6230b0b00c5356b8706c56e0b9c36b1dfc86084cd75f0"
-    assert valid_metadata_doc.meta["page_number"] == 1
+    assert result.meta["source_id"] == "62049ba1d1e1d5ebb1f6230b0b00c5356b8706c56e0b9c36b1dfc86084cd75f0"
+    assert result.meta["page_number"] == 1
 
 
 def test_convert_meta_to_int():

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

@@ -1,5 +1,6 @@
 import inspect
 from collections.abc import AsyncGenerator, Generator
+from dataclasses import replace
 from itertools import islice
 from typing import Any, ClassVar, cast
 
@@ -2474,15 +2475,17 @@ def _process_query_point_results(
         ]
 
         if scale_score:
-            for document in documents:
-                score = document.score
-                if score is None:
-                    continue
-                if self.similarity == "cosine":
-                    score = (score + 1) / 2
-                else:
-                    score = float(1 / (1 + exp(-score / 100)))
-                document.score = score
+            documents = [
+                replace(
+                    document,
+                    score=(document.score + 1) / 2
+                    if self.similarity == "cosine"
+                    else float(1 / (1 + exp(-document.score / 100))),
+                )
+                if document.score is not None
+                else document
+                for document in documents
+            ]
 
         return documents