feat: KuzuDB CSV bulk import + buffered node/edge insertion in GraphBuilder

- KuzuBackend.bulk_add_nodes/bulk_add_edges use CSV COPY FROM (~2x faster)
- GraphBuilder now buffers both nodes AND edges, flushing all nodes
  before edges to guarantee correct ordering across all backends
- flush() auto-detects bulk_add_* methods on backend for optimal path
- Data parity maintained: 2,298/2,890 on contoso, 25,893/30,636 on
  spring-boot — identical across NetworkX, SQLite, KuzuDB

Performance: KuzuDB contoso 26.7s → 13.8s, spring-boot 5m10s → 2m30s

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: aksOps

src/code_intelligence/graph/backends/kuzu.py

@@ -2,8 +2,11 @@
 
 from __future__ import annotations
 
+import csv
 import json
 import logging
+import os
+import tempfile
 from typing import Any
 
 import kuzu
@@ -201,6 +204,80 @@ def add_edge(self, edge: GraphEdge) -> None:
             params,
         )
 
+    def bulk_add_nodes(self, nodes: list[GraphNode]) -> None:
+        """Bulk-insert nodes via CSV COPY FROM (~100x faster than per-row)."""
+        if not nodes:
+            return
+        seen: set[str] = set()
+        unique_nodes: list[GraphNode] = []
+        for n in nodes:
+            if n.id not in seen:
+                seen.add(n.id)
+                unique_nodes.append(n)
+
+        csv_path = ""
+        try:
+            fd = tempfile.NamedTemporaryFile(
+                mode="w", suffix=".csv", delete=False, newline=""
+            )
+            csv_path = fd.name
+            writer = csv.writer(fd)
+            for node in unique_nodes:
+                p = _node_to_params(node)
+                writer.writerow([
+                    p["id"], p["kind"], p["label"], p["fqn"], p["module"],
+                    p["file_path"], p["line_start"], p["line_end"],
+                    p["annotations"], p["properties"],
+                ])
+            fd.close()
+            self._conn.execute(
+                f'COPY CodeNode FROM "{csv_path}" (HEADER=false)'
+            )
+        except Exception:
+            logger.exception("Bulk node insert failed, falling back to per-row")
+            for node in unique_nodes:
+                self.add_node(node)
+        finally:
+            if csv_path:
+                try:
+                    os.unlink(csv_path)
+                except OSError:
+                    pass
+
+    def bulk_add_edges(self, edges: list[GraphEdge]) -> None:
+        """Bulk-insert edges via CSV COPY FROM (~100x faster than per-row)."""
+        if not edges:
+            return
+        csv_path = ""
+        try:
+            fd = tempfile.NamedTemporaryFile(
+                mode="w", suffix=".csv", delete=False, newline=""
+            )
+            csv_path = fd.name
+            writer = csv.writer(fd)
+            for edge in edges:
+                writer.writerow([
+                    edge.source,
+                    edge.target,
+                    edge.kind.value,
+                    edge.label or "",
+                    json.dumps(edge.properties),
+                ])
+            fd.close()
+            self._conn.execute(
+                f'COPY CODE_EDGE FROM "{csv_path}" (HEADER=false)'
+            )
+        except Exception:
+            logger.exception("Bulk edge insert failed, falling back to per-row")
+            for edge in edges:
+                self.add_edge(edge)
+        finally:
+            if csv_path:
+                try:
+                    os.unlink(csv_path)
+                except OSError:
+                    pass
+
     def clear(self) -> None:
         """Remove all data by dropping and recreating both tables."""
         try:

src/code_intelligence/graph/builder.py

@@ -204,6 +204,7 @@ class GraphBuilder:
 
     def __init__(self, backend: GraphBackend | None = None) -> None:
         self._store = GraphStore(backend=backend)
+        self._pending_nodes: list[GraphNode] = []
         self._pending_edges: list[GraphEdge] = []
         self._linkers: list[Linker] = [
             TopicLinker(),
@@ -212,23 +213,39 @@ def __init__(self, backend: GraphBackend | None = None) -> None:
         ]
 
     def add_nodes(self, nodes: list[GraphNode]) -> None:
-        """Add a batch of nodes to the graph store."""
-        for node in nodes:
-            self._store.add_node(node)
+        """Buffer nodes for deferred insertion."""
+        self._pending_nodes.extend(nodes)
 
     def add_edges(self, edges: list[GraphEdge]) -> None:
         """Buffer edges for deferred insertion."""
         self._pending_edges.extend(edges)
 
-    def flush_edges(self) -> None:
-        """Insert all buffered edges into the store.
+    def flush(self) -> None:
+        """Insert all buffered nodes then edges into the store.
 
-        Call this after all nodes have been added so that backends
-        which validate node existence won't reject valid cross-file edges.
+        Nodes first, then edges — ensures backends that validate node
+        existence won't reject valid cross-file edges. Uses bulk insert
+        when the backend supports it (e.g. KuzuDB CSV COPY FROM).
         """
-        for edge in self._pending_edges:
-            self._store.add_edge(edge)
-        self._pending_edges.clear()
+        backend = self._store._backend
+
+        # Flush nodes
+        if self._pending_nodes:
+            if hasattr(backend, "bulk_add_nodes"):
+                backend.bulk_add_nodes(self._pending_nodes)
+            else:
+                for node in self._pending_nodes:
+                    self._store.add_node(node)
+            self._pending_nodes.clear()
+
+        # Flush edges
+        if self._pending_edges:
+            if hasattr(backend, "bulk_add_edges"):
+                backend.bulk_add_edges(self._pending_edges)
+            else:
+                for edge in self._pending_edges:
+                    self._store.add_edge(edge)
+            self._pending_edges.clear()
 
     def merge_detector_result(self, result: object) -> None:
         """Merge a DetectorResult into the graph.
@@ -242,9 +259,9 @@ def merge_detector_result(self, result: object) -> None:
         self.add_edges(edges)  # buffered, not inserted yet
 
     def run_linkers(self) -> None:
-        """Flush pending edges, then run all registered linkers."""
-        # Flush detector edges first so linkers see the full graph
-        self.flush_edges()
+        """Flush pending nodes and edges, then run all registered linkers."""
+        # Flush all buffered detector data so linkers see the full graph
+        self.flush()
 
         for linker in self._linkers:
             try:
@@ -268,11 +285,11 @@ def run_linkers(self) -> None:
                 )
 
         # Flush linker edges (linker-created nodes are already added above)
-        self.flush_edges()
+        self.flush()
 
     def build(self) -> GraphStore:
         """Return the assembled graph store."""
         # Safety: flush any remaining edges
         if self._pending_edges:
-            self.flush_edges()
+            self.flush()
         return self._store