Use uniqueness constraint for KG writer temp ids

CHANGELOG.md

@@ -42,6 +42,8 @@
 - Fixed a bug in `AnthropicLLM` where an `http_client` passed via kwargs (whether an `httpx.Client` or `httpx.AsyncClient`) was forwarded to both the sync `anthropic.Anthropic` and async `anthropic.AsyncAnthropic` clients, causing a type mismatch. `http_client` is now routed to the matching sync/async client only; other kwargs remain shared. An `http_client` of an unrecognized type now emits a warning and is ignored instead of raising, matching `OpenAILLM`'s existing behavior.
 - Vector and VectorCypher retrievers on Neo4j 2026+: prefix SEARCH queries with `CYPHER 25` and fall back to procedure-based vector search when SEARCH is unsupported or fails.
 - E2E tests: added exponential backoff retry logic with jitter (5 attempts, 5–60 second waits) to embedding model downloads to handle Hugging Face rate limits. Retries only on transient network errors (connection, timeout) and immediately fails on unrecoverable errors (missing packages, permissions), improving test reliability in parallel CI runs.
+- `Neo4jWriter` now guarantees uniqueness of the temporary `__tmp_internal_id` used to match nodes when writing relationships by using a `UNIQUENESS` constraint instead of a plain range index, which did not prevent duplicate ids.
+- `Neo4jWriter` now runs its temporary-id cleanup against the configured `neo4j_database` rather than the user's home database, matching the database used for setup and upserts.
 
 ## 1.18.0
 

src/neo4j_graphrag/components/kg_writer.py

@@ -215,9 +215,18 @@ def __init__(
         self.is_version_5_24_or_above = is_version_5_24_or_above(version_tuple)
 
     def _db_setup(self) -> None:
-        self.driver.execute_query("""
-        CREATE INDEX __entity__tmp_internal_id IF NOT EXISTS FOR (n:__KGBuilder__) ON (n.__tmp_internal_id)
-        """)
+        self.driver.execute_query(
+            "DROP INDEX __entity__tmp_internal_id IF EXISTS",
+            database_=self.neo4j_database,
+        )
+        self.driver.execute_query(
+            """
+            CREATE CONSTRAINT __entity__tmp_internal_id_unique IF NOT EXISTS
+            FOR (n:__KGBuilder__)
+            REQUIRE n.__tmp_internal_id IS UNIQUE
+            """,
+            database_=self.neo4j_database,
+        )
 
     @staticmethod
     def _nodes_to_rows(
@@ -280,7 +289,7 @@ def _db_cleaning(self) -> None:
             support_variable_scope_clause=self.is_version_5_23_or_above,
             batch_size=self.batch_size,
         )
-        with self.driver.session() as session:
+        with self.driver.session(database=self.neo4j_database) as session:
             session.run(query)
 
     @validate_call

tests/unit/components/test_kg_writer.py

@@ -102,6 +102,34 @@ def test_get_unique_properties_for_node_type_deprecation_warning() -> None:
         ]
 
 
+@mock.patch(
+    "neo4j_graphrag.experimental.components.kg_writer.get_version",
+    return_value=((5, 22, 0), False, False),
+)
+def test_neo4j_writer_db_setup_uses_unique_constraint(
+    _: Mock, driver: MagicMock
+) -> None:
+    neo4j_writer = Neo4jWriter(driver=driver, neo4j_database="my_db")
+    driver.execute_query.reset_mock()
+
+    neo4j_writer._db_setup()
+
+    assert driver.execute_query.call_count == 2
+    drop_call = driver.execute_query.call_args_list[0]
+    assert drop_call.args == ("DROP INDEX __entity__tmp_internal_id IF EXISTS",)
+    assert drop_call.kwargs == {"database_": "my_db"}
+
+    constraint_call = driver.execute_query.call_args_list[1]
+    constraint_query = constraint_call.args[0]
+    assert (
+        "CREATE CONSTRAINT __entity__tmp_internal_id_unique IF NOT EXISTS"
+        in constraint_query
+    )
+    assert "FOR (n:__KGBuilder__)" in constraint_query
+    assert "REQUIRE n.__tmp_internal_id IS UNIQUE" in constraint_query
+    assert constraint_call.kwargs == {"database_": "my_db"}
+
+
 # --- FilenameCollisionHandler tests ---