fix: keep the highest-scoring duplicate in merge and RRF joins

haystack/components/joiners/document_joiner.py

@@ -16,6 +16,11 @@
 logger = logging.getLogger(__name__)
 
 
+def _score_or_neg_inf(document: Document) -> float:
+    """Sort key that ranks a Document without a score below every scored one."""
+    return document.score if document.score is not None else -inf
+
+
 class JoinMode(Enum):
     """
     Enum for join mode.
@@ -188,7 +193,7 @@ def _concatenate(document_lists: list[list[Document]]) -> list[Document]:
         for doc in itertools.chain.from_iterable(document_lists):
             docs_per_id[doc.id].append(doc)
         for docs in docs_per_id.values():
-            doc_with_best_score = max(docs, key=lambda doc: doc.score if doc.score is not None else -inf)
+            doc_with_best_score = max(docs, key=_score_or_neg_inf)
             output.append(doc_with_best_score)
         return output
 
@@ -207,7 +212,14 @@ def _merge(self, document_lists: list[list[Document]]) -> list[Document]:
         for documents, weight in zip(document_lists, weights, strict=True):
             for doc in documents:
                 scores_map[doc.id] += (doc.score if doc.score is not None else 0) * weight
-                documents_map[doc.id] = doc
+                # Duplicates share an id but not necessarily a payload: two retrievers can return
+                # the same document with different content or meta. Keep the highest-scoring copy,
+                # as `_concatenate` and `_distribution_based_rank_fusion` do -- overwriting
+                # unconditionally would keep whichever list happened to come last, making the
+                # returned payload depend on connection order even though the fused score does not.
+                incumbent = documents_map.get(doc.id)
+                if incumbent is None or _score_or_neg_inf(doc) > _score_or_neg_inf(incumbent):
+                    documents_map[doc.id] = doc
 
         return [replace(doc, score=scores_map[doc.id]) for doc in documents_map.values()]
 

haystack/utils/misc.py

@@ -180,7 +180,15 @@ def _reciprocal_rank_fusion(
     for documents, weight in zip(document_lists, resolved_weights, strict=True):
         for rank, doc in enumerate(documents):
             scores_map[doc.id] += (weight * len(document_lists)) / (k + rank)
-            documents_map[doc.id] = doc
+            # Duplicates share an id but not necessarily a payload, so keep the highest-scoring
+            # copy the way `_deduplicate_documents` does. Overwriting unconditionally would keep
+            # whichever list came last, making the returned payload depend on list order even
+            # though the fused score does not.
+            incumbent = documents_map.get(doc.id)
+            if incumbent is None or (doc.score if doc.score is not None else -inf) > (
+                incumbent.score if incumbent.score is not None else -inf
+            ):
+                documents_map[doc.id] = doc
 
     for _id in scores_map:
         scores_map[_id] /= len(document_lists) / k

releasenotes/notes/document-joiner-keep-highest-scoring-duplicate-3f8c21ab6d4e7a19.yaml

@@ -0,0 +1,11 @@
+---
+fixes:
+  - |
+    `DocumentJoiner` now keeps the highest-scoring copy of a duplicate document in `merge` and
+    `reciprocal_rank_fusion` mode, matching what `concatenate` and
+    `distribution_based_rank_fusion` already do. Both modes previously overwrote the stored
+    document unconditionally, so when two retrievers returned the same id with different content
+    or meta, the copy that survived was whichever list happened to come last. The fused score is
+    order-independent in both modes, so the returned payload silently depended on connection
+    order while the score did not. The shared `_reciprocal_rank_fusion` helper is fixed too, so
+    every caller benefits.

test/components/joiners/test_document_joiner.py

@@ -188,6 +188,35 @@ def test_run_with_merge_join_mode_handles_zero_score(self):
         output = joiner.run([documents_1, documents_2])
         assert output["documents"][0].score == 0.0
 
+    def test_run_with_merge_join_mode_keeps_highest_scoring_duplicate(self):
+        # Duplicates share an id but not necessarily a payload. The fused score is a weighted sum
+        # and so is order-independent; the kept copy must be too, and must match what
+        # `concatenate` and `distribution_based_rank_fusion` keep -- the highest-scoring one.
+        high = Document(id="d1", content="full text", score=0.9)
+        low = Document(id="d1", content="stub", score=0.1)
+
+        for lists in ([high], [low]), ([low], [high]):
+            output = DocumentJoiner(join_mode="merge", weights=[0.5, 0.5]).run(list(lists))
+            assert len(output["documents"]) == 1
+            assert output["documents"][0].content == "full text"
+
+    def test_run_with_rrf_join_mode_keeps_highest_scoring_duplicate(self):
+        high = Document(id="d1", content="full text", score=0.9)
+        low = Document(id="d1", content="stub", score=0.1)
+
+        for lists in ([high], [low]), ([low], [high]):
+            output = DocumentJoiner(join_mode="reciprocal_rank_fusion").run(list(lists))
+            assert len(output["documents"]) == 1
+            assert output["documents"][0].content == "full text"
+
+    def test_run_with_merge_join_mode_keeps_scored_duplicate_over_unscored(self):
+        scored = Document(id="d1", content="scored", score=0.2)
+        unscored = Document(id="d1", content="unscored", score=None)
+
+        for lists in ([scored], [unscored]), ([unscored], [scored]):
+            output = DocumentJoiner(join_mode="merge", weights=[0.5, 0.5]).run(list(lists))
+            assert output["documents"][0].content == "scored"
+
     def test_run_with_merge_join_mode(self):
         joiner = DocumentJoiner(join_mode="merge", weights=[1.5, 0.5])
         documents_1 = [Document(content="a", score=1.0), Document(content="b", score=2.0)]