How to set duplicate document policy in Haystack

Repeated document IDs are common when an indexing job retries a batch, refreshes an existing source, or receives corrected content. Haystack lets the writer decide whether that second write preserves the stored document, replaces it, or stops with an error.

The DocumentWriter component keeps its configured DuplicatePolicy for ordinary runs, while a runtime override can select another policy for one batch. Choosing SKIP, OVERWRITE, or FAIL explicitly also avoids depending on a document store's interpretation of DuplicatePolicy.NONE instead.

An InMemoryDocumentStore exposes each policy branch without an external database. The same document ID remains unchanged after SKIP, receives new content after OVERWRITE, and raises DuplicateDocumentError under FAIL while the final document count stays at one.

1. Choose the duplicate policy that matches the indexing job's intended response to an existing document ID.

   DuplicatePolicy.SKIP preserves the first document, DuplicatePolicy.OVERWRITE replaces its stored content, and DuplicatePolicy.FAIL rejects the repeated ID. DuplicatePolicy.NONE delegates the choice to the document store.

2. Create the initial duplicate_policy_demo.py section with the imports, in-memory store, and seeded document.

   duplicate_policy_demo.py

   import logging
    
   from haystack import Document
   from haystack.components.writers import DocumentWriter
   from haystack.document_stores.errors import DuplicateDocumentError
   from haystack.document_stores.in_memory import InMemoryDocumentStore
   from haystack.document_stores.types import DuplicatePolicy
    
    
   logging.getLogger("haystack").setLevel(logging.ERROR)
    
   document_store = InMemoryDocumentStore()
   document_id = "release-notes"
    
   seed_writer = DocumentWriter(
       document_store=document_store,
       policy=DuplicatePolicy.FAIL,
   )
   seed_writer.run(
       documents=[Document(id=document_id, content="Version 1 release notes")]
   )

   The explicit FAIL policy prevents an unexpected duplicate during setup from passing silently. The logging threshold keeps the expected SKIP warning out of the comparison transcript.

3. Add the SKIP branch that records the write count and the content retained for the repeated ID.

   writer = DocumentWriter(
       document_store=document_store,
       policy=DuplicatePolicy.SKIP,
   )
    
   skip_result = writer.run(
       documents=[Document(id=document_id, content="Skipped replacement")]
   )
   stored_after_skip = document_store.filter_documents()[0].content

4. Add the OVERWRITE branch that replaces the stored content for one run without changing the writer's default policy.

   overwrite_result = writer.run(
       documents=[Document(id=document_id, content="Version 2 release notes")],
       policy=DuplicatePolicy.OVERWRITE,
   )
   stored_after_overwrite = document_store.filter_documents()[0].content

5. Add the FAIL branch that catches only the duplicate-document exception from the repeated ID.

   try:
       writer.run(
           documents=[Document(id=document_id, content="Rejected replacement")],
           policy=DuplicatePolicy.FAIL,
       )
   except DuplicateDocumentError:
       fail_error = "DuplicateDocumentError"
   else:
       fail_error = "No error"
    
   final_documents = document_store.filter_documents()

6. Add result output and fail-capable assertions for all three duplicate-policy branches.

   print(f"SKIP wrote: {skip_result['documents_written']}")
   print(f"After SKIP: {stored_after_skip}")
   print(f"OVERWRITE wrote: {overwrite_result['documents_written']}")
   print(f"After OVERWRITE: {stored_after_overwrite}")
   print(f"FAIL raised: {fail_error}")
   print(f"Final count: {len(final_documents)}")
    
   assert skip_result["documents_written"] == 0
   assert stored_after_skip == "Version 1 release notes"
   assert overwrite_result["documents_written"] == 1
   assert stored_after_overwrite == "Version 2 release notes"
   assert fail_error == "DuplicateDocumentError"
   assert len(final_documents) == 1
   assert final_documents[0].content == "Version 2 release notes"

7. Assemble the completed duplicate_policy_demo.py from the dependency-ordered sections above.

   duplicate_policy_demo.py

   import logging
    
   from haystack import Document
   from haystack.components.writers import DocumentWriter
   from haystack.document_stores.errors import DuplicateDocumentError
   from haystack.document_stores.in_memory import InMemoryDocumentStore
   from haystack.document_stores.types import DuplicatePolicy
    
    
   logging.getLogger("haystack").setLevel(logging.ERROR)
    
   document_store = InMemoryDocumentStore()
   document_id = "release-notes"
    
   seed_writer = DocumentWriter(
       document_store=document_store,
       policy=DuplicatePolicy.FAIL,
   )
   seed_writer.run(
       documents=[Document(id=document_id, content="Version 1 release notes")]
   )
    
   writer = DocumentWriter(
       document_store=document_store,
       policy=DuplicatePolicy.SKIP,
   )
    
   skip_result = writer.run(
       documents=[Document(id=document_id, content="Skipped replacement")]
   )
   stored_after_skip = document_store.filter_documents()[0].content
    
   overwrite_result = writer.run(
       documents=[Document(id=document_id, content="Version 2 release notes")],
       policy=DuplicatePolicy.OVERWRITE,
   )
   stored_after_overwrite = document_store.filter_documents()[0].content
    
   try:
       writer.run(
           documents=[Document(id=document_id, content="Rejected replacement")],
           policy=DuplicatePolicy.FAIL,
       )
   except DuplicateDocumentError:
       fail_error = "DuplicateDocumentError"
   else:
       fail_error = "No error"
    
   final_documents = document_store.filter_documents()
    
   print(f"SKIP wrote: {skip_result['documents_written']}")
   print(f"After SKIP: {stored_after_skip}")
   print(f"OVERWRITE wrote: {overwrite_result['documents_written']}")
   print(f"After OVERWRITE: {stored_after_overwrite}")
   print(f"FAIL raised: {fail_error}")
   print(f"Final count: {len(final_documents)}")
    
   assert skip_result["documents_written"] == 0
   assert stored_after_skip == "Version 1 release notes"
   assert overwrite_result["documents_written"] == 1
   assert stored_after_overwrite == "Version 2 release notes"
   assert fail_error == "DuplicateDocumentError"
   assert len(final_documents) == 1
   assert final_documents[0].content == "Version 2 release notes"

8. Verify all three duplicate-policy branches by running duplicate_policy_demo.py in the Python environment that provides haystack-ai.

   $ python duplicate_policy_demo.py
   SKIP wrote: 0
   After SKIP: Version 1 release notes
   OVERWRITE wrote: 1
   After OVERWRITE: Version 2 release notes
   FAIL raised: DuplicateDocumentError
   Final count: 1

   A zero SKIP count, the replaced content after OVERWRITE, DuplicateDocumentError from FAIL, and a final count of one prove that every branch acted on the same ID.
   Related: How to install Haystack with pip