Shashankss1205 · Oct 15, 2026 · Sep 29, 2026
diff --git a/src/codegraphcontext/tools/indexing/persistence/writer.py b/src/codegraphcontext/tools/indexing/persistence/writer.py
@@ -40,6 +40,44 @@ def metadata_priority(row: Dict[str, Any]) -> Tuple[str, int, str, int]:
     return sorted(rows, key=metadata_priority)
 
 
+# Parameter, Module and ExternalClass used to be written as `UNWIND … MERGE (node) … MERGE (a)-[r]->(node)`. The
+# embedded wrapper forces that shape (node MERGE plus a relationship, Kuzu #1605) onto a per-row loop, one query per
+# row. These three writers instead run a node-only UNWIND with distinct node keys, then a relationship-only UNWIND,
+# neither of which the guard touches. Each node phase keeps the old MATCH, so no node is created that the combined
+# query would not have created.
+def _module_node_rows(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    """One row per Module name holding, per field, the first non-null value in `rows` order: what the sequential
+    `SET m.x = coalesce(m.x, row.x)` of the combined query leaves on the node."""
+    nodes: Dict[str, Dict[str, Any]] = {}
+    for r in rows:
+        n = nodes.setdefault(r["name"], {"name": r["name"], "lang": None, "full_import_name": None})
+        for k in ("lang", "full_import_name"):
+            if n[k] is None:
+                n[k] = r.get(k)
+    return list(nodes.values())
+
+
+def _last_row_per_key(rows: List[Dict[str, Any]], key) -> List[Dict[str, Any]]:
+    """Rows sharing a relationship MERGE key are one relationship. Row by row, its SET ends with the last row's
+    values; one batched MERGE+SET keeps the first row's. Keep the last row per key."""
+    last: Dict[Any, Dict[str, Any]] = {}
+    for r in rows:
+        last[key(r)] = r
+    return list(last.values())
+
+
+def _run_node_then_rel(session, node_query: str, node_rows, rel_query: str, rel_rows, combined_query: str,
+                       combined_rows, **params) -> None:
+    """Node phase, then relationship phase. `node_rows` may be a callable reading them from the session. On any error
+    the chunk is rerun through the old combined query, i.e. the wrapper's per-row path with its old skip / retry / drop
+    semantics; every MERGE here is idempotent."""
+    try:
+        session.run(node_query, batch=node_rows(session) if callable(node_rows) else node_rows, **params)
+        session.run(rel_query, batch=rel_rows, **params)
+    except Exception:
+        session.run(combined_query, batch=combined_rows, **params)
+
+
 def _normalize_path(p) -> str:
     """Normalize a path to use forward slashes for cross-platform DB consistency.
 
@@ -639,15 +677,32 @@ def _work(session):
                     if key not in seen_params:
                         seen_params.add(key)
                         unique_params.append(p)
-                session.run(
+                fn_match = "MATCH (fn:Function {name: row.func_name, path: $file_path, line_number: row.line_number, occurrence_index: row.occurrence_index})"
+                _run_node_then_rel(
+                    session,
+                    """
+                    UNWIND $batch AS row
+                    MERGE (p:Parameter {name: row.arg_name, path: $file_path, function_line_number: row.line_number})
+                    SET p.name = row.arg_name, p.path = $file_path, p.function_line_number = row.line_number
+                """,
+                    # one row per Parameter key whose Function exists, as the combined query's MATCH required
+                    lambda s: s.run(f"UNWIND $batch AS row {fn_match} RETURN DISTINCT row.arg_name AS arg_name, "
+                                    "row.line_number AS line_number", batch=unique_params, file_path=file_path_str).data(),
+                    f"""
+                    UNWIND $batch AS row
+                    {fn_match}
+                    MATCH (p:Parameter {{name: row.arg_name, path: $file_path, function_line_number: row.line_number}})
+                    MERGE (fn)-[:HAS_PARAMETER]->(p)
+                """,
+                    unique_params,
                     """
                     UNWIND $batch AS row
                     MATCH (fn:Function {name: row.func_name, path: $file_path, line_number: row.line_number, occurrence_index: row.occurrence_index})
                     MERGE (p:Parameter {name: row.arg_name, path: $file_path, function_line_number: row.line_number})
                     SET p.name = row.arg_name, p.path = $file_path, p.function_line_number = row.line_number
                     MERGE (fn)-[:HAS_PARAMETER]->(p)
                 """,
-                    batch=unique_params,
+                    unique_params,
                     file_path=file_path_str,
                 )
 
@@ -769,19 +824,38 @@ def _work(session):
 
             if other_imports:
                 other_imports = sort_import_rows_for_metadata(other_imports)
-                session.run(
+                _run_node_then_rel(
+                    session,
                     """
                     UNWIND $batch AS row
                     MATCH (f:File {path: $file_path})
                     MERGE (m:Module {name: row.name})
                     SET m.lang = coalesce(m.lang, row.lang),
                         m.full_import_name = coalesce(m.full_import_name, row.full_import_name)
+                """,
+                    _module_node_rows(other_imports),
+                    """
+                    UNWIND $batch AS row
+                    MATCH (f:File {path: $file_path})
+                    MATCH (m:Module {name: row.name})
                     MERGE (f)-[r:IMPORTS {line_number: row.line_number, imported_name: row.imported_name}]->(m)
                     SET r.alias = coalesce(row.alias, ""),
                         r.full_import_name = row.full_import_name,
                         r.lang = row.lang
                 """,
-                    batch=other_imports,
+                    _last_row_per_key(other_imports, lambda r: (r["name"], r["line_number"], r["imported_name"])),
+                    """
+                    UNWIND $batch AS row
+                    MATCH (f:File {path: $file_path})
+                    MERGE (m:Module {name: row.name})
+                    SET m.lang = coalesce(m.lang, row.lang),
+                        m.full_import_name = coalesce(m.full_import_name, row.full_import_name)
+                    MERGE (f)-[r:IMPORTS {line_number: row.line_number, imported_name: row.imported_name}]->(m)
+                    SET r.alias = coalesce(row.alias, ""),
+                        r.full_import_name = row.full_import_name,
+                        r.lang = row.lang
+                """,
+                    other_imports,
                     file_path=file_path_str,
                 )
 
@@ -1244,17 +1318,32 @@ def _run_internal(rows, child_label, parent_label):
                         raise e
 
             def _run_external(rows, child_label):
-                child_cypher = _cypher_label(child_label, backend)
+                child_match = f"MATCH (child:{_cypher_label(child_label, backend)} {{name: row.child_name, path: row.path}})"
+                parents = f"""
+                    UNWIND $batch AS row
+                    {child_match}
+                    WITH DISTINCT row.parent_name AS parent_name
+                    MERGE (:ExternalClass {{name: parent_name}})
+                """
+                links = f"""
+                    UNWIND $batch AS row
+                    {child_match}
+                    MATCH (parent:ExternalClass {{name: row.parent_name}})
+                    MERGE (child)-[r:INHERITS]->(parent)
+                    SET r.confidence_label = coalesce(row.confidence_label, 'INFERRED')
+                """
                 query = f"""
                     UNWIND $batch AS row
-                    MATCH (child:{child_cypher} {{name: row.child_name, path: row.path}})
+                    {child_match}
                     MERGE (parent:ExternalClass {{name: row.parent_name}})
                     MERGE (child)-[r:INHERITS]->(parent)
                     SET r.confidence_label = coalesce(row.confidence_label, 'INFERRED')
                 """
                 for chunk in _chunks(rows):
                     try:
-                        session.run(query, batch=chunk)
+                        # every referenced field present: a key missing from all rows would not bind in a batch
+                        rows_n = [{k: r.get(k) for k in ("child_name", "path", "parent_name", "confidence_label")} for r in chunk]
+                        _run_node_then_rel(session, parents, rows_n, links, rows_n, query, chunk)
                     except Exception as e:
                         if _is_binder_exception(e):
                             return

diff --git a/tests/unit/tools/test_graph_builder_perf_fixes.py b/tests/unit/tools/test_graph_builder_perf_fixes.py
@@ -660,13 +660,19 @@ def test_import_module_metadata_preserves_existing_node_values(self):
 
         gb.add_file_to_graph(file_data, "my_repo", {}, repo_path_str="/repo")
 
+        # Module nodes are written by a node-only query before the IMPORTS relationships
+        module_call = next(
+            c
+            for c in session.calls
+            if "MERGE (m:Module {name: row.name})" in c["query"] and "IMPORTS" not in c["query"]
+        )
+        assert "m.lang = coalesce(m.lang, row.lang)" in module_call["query"]
+        assert "m.full_import_name = coalesce(m.full_import_name, row.full_import_name)" in module_call["query"]
         import_call = next(
             c
             for c in session.calls
             if "MERGE (f)-[r:IMPORTS {line_number: row.line_number" in c["query"]
         )
-        assert "m.lang = coalesce(m.lang, row.lang)" in import_call["query"]
-        assert "m.full_import_name = coalesce(m.full_import_name, row.full_import_name)" in import_call["query"]
         assert "r.full_import_name = row.full_import_name" in import_call["query"]
         assert "r.lang = row.lang" in import_call["query"]
 

diff --git a/tests/unit/tools/test_split_node_rel_batches.py b/tests/unit/tools/test_split_node_rel_batches.py
@@ -0,0 +1,101 @@
+"""Parameter / HAS_PARAMETER, Module / IMPORTS and ExternalClass / INHERITS are written as a node-only UNWIND followed
+by a relationship-only UNWIND, so the embedded wrapper's forced per-row fallback (node MERGE plus a relationship in one
+UNWIND, Kuzu #1605) never fires for them. The graph must be what the combined per-row query wrote: nodes only where
+the old MATCH succeeded, first non-null Module metadata, the last row's values for a repeated IMPORTS key, and the
+label-less external fan-out over every populated child label.
+"""
+import re
+from pathlib import Path
+
+import pytest
+
+import codegraphcontext.core.database_embedded_kuzu as embedded
+from codegraphcontext.core.database_kuzu import KuzuDBManager
+from codegraphcontext.tools.indexing.persistence.writer import GraphWriter
+
+kuzu = pytest.importorskip("kuzu")
+REPO = "C:/fx3"
+A = f"{REPO}/src/a.rs"
+GUARD = re.compile(r"UNWIND\s+\$\w+\s+AS\s+\w+")
+
+
+class _DBM:
+    def get_backend_type(self):
+        return "kuzudb"
+
+
+@pytest.fixture()
+def env(tmp_path: Path, monkeypatch):
+    forced = []
+    original = embedded.EmbeddedSessionWrapper.run
+
+    def run(self, q, **p):
+        if GUARD.search(q) and ("-[" in q or "]->" in q) and re.search(r"MERGE\s*\(\s*\w+\s*:", q):
+            forced.append(q)
+        return original(self, q, **p)
+
+    monkeypatch.setattr(embedded.EmbeddedSessionWrapper, "run", run)
+    manager = KuzuDBManager(str(tmp_path / "db"))
+    driver = manager.get_driver()
+    writer = GraphWriter(driver, _DBM())
+    writer.add_repository_to_graph(Path(REPO))
+    yield writer, driver, forced
+    manager.close_driver()
+
+
+def _q(driver, cypher):
+    with driver.session() as s:
+        return s.run(cypher).data()
+
+
+def _imp(name, full, line, imported, alias=None, lang="rust"):
+    return {"name": name, "full_import_name": full, "line_number": line, "imported_name": imported, "alias": alias, "lang": lang}
+
+
+def _fn(name, line, args):
+    return {"name": name, "line_number": line, "end_line": line + 5, "args": args, "lang": "rust"}
+
+
+def test_parameters_follow_occurrence_identity(env):
+    writer, driver, forced = env
+    writer.add_file_to_graph({"path": A, "repo_path": REPO, "lang": "rust",
+                              "functions": [_fn("f", 10, ["a", "b"]), _fn("f", 10, ["a", "c"]), _fn("g", 20, ["a", "a"])]}, "fx", {})
+    edges = sorted((r["o"], r["p"], r["l"]) for r in _q(driver, "MATCH (f:Function)-[:HAS_PARAMETER]->(p:Parameter) "
+                                                         "RETURN f.occurrence_index AS o, p.name AS p, p.function_line_number AS l"))
+    assert edges == [(0, "a", 10), (0, "a", 20), (0, "b", 10), (1, "a", 10), (1, "c", 10)]
+    assert len(_q(driver, "MATCH (p:Parameter) RETURN p.uid")) == 4        # (a,10) shared by both occurrences
+    assert forced == []
+
+
+def test_imports_metadata_and_repeated_keys(env):
+    writer, driver, forced = env
+    writer.add_file_to_graph({"path": A, "repo_path": REPO, "lang": "rust", "imports": [
+        _imp("M1", "use M1::Baz;", 4, "Baz"), _imp("M1", "use super::*;", 1, "*"), _imp("M1", "pub use M1::Bar;", 2, "Bar", "B"),
+        _imp("M1", "pub use M1::*;", 5, "*"), _imp("M1", "use M1::Bar;", 2, "Bar", "B2"),
+        _imp("M4", "import M4", 1, "M4", lang=None), _imp("M4", "import M4", 2, "M4b")]}, "fx", {})
+    mods = {r["n"]: (r["l"], r["f"]) for r in _q(driver, "MATCH (m:Module) RETURN m.name AS n, m.lang AS l, m.full_import_name AS f")}
+    assert mods == {"M1": ("rust", "use super::*;"), "M4": ("rust", "import M4")}
+    bar = _q(driver, "MATCH (:File)-[r:IMPORTS {line_number: 2, imported_name: 'Bar'}]->(:Module) RETURN r.alias AS a, r.full_import_name AS f")
+    assert bar == [{"a": "B2", "f": "use M1::Bar;"}]                      # last sorted row, as row-by-row SETs leave it
+    assert len(_q(driver, "MATCH (:File)-[r:IMPORTS]->(:Module) RETURN r.line_number")) == 6
+    assert forced == []
+
+
+def test_external_parents_only_for_existing_children(env):
+    writer, driver, forced = env
+    writer.add_file_to_graph({"path": A, "repo_path": REPO, "lang": "rust",
+                              "classes": [{"name": "K", "line_number": 1, "end_line": 2, "lang": "rust"},
+                                          {"name": "Status", "line_number": 3, "end_line": 4, "lang": "rust"}],
+                              "variables": [{"name": "Status", "line_number": 5, "lang": "rust"}]}, "fx", {})
+    x = "__external__"
+    writer.write_inheritance_links([
+        {"child_name": "K", "path": A, "parent_name": "Base", "resolved_parent_file_path": x, "child_label": "Class"},
+        {"child_name": "Status", "path": A, "parent_name": "Legacy", "resolved_parent_file_path": x},
+        {"child_name": "Ghost", "path": A, "parent_name": "Nowhere", "resolved_parent_file_path": x, "child_label": "Class"},
+    ], [], {})
+    assert sorted(r["n"] for r in _q(driver, "MATCH (e:ExternalClass) RETURN e.name AS n")) == ["Base", "Legacy"]
+    rels = sorted((r["lab"], r["c"], r["p"], r["k"]) for r in _q(driver, "MATCH (c)-[r:INHERITS]->(p:ExternalClass) "
+                                                                  "RETURN label(c) AS lab, c.name AS c, p.name AS p, r.confidence_label AS k"))
+    assert rels == [("Class", "K", "Base", "INFERRED"), ("Class", "Status", "Legacy", "INFERRED"),
+                    ("Variable", "Status", "Legacy", "INFERRED")]
+    assert forced == []