From 2f7be29f59f4e321ee6c30bac9fb5553a7c7eb57 Mon Sep 17 00:00:00 2001 From: abhigyanpatwari Date: Tue, 17 Feb 2026 01:01:14 +0530 Subject: [PATCH] perf: hybrid incremental - workerData fast path + cross-file edge preservation Co-authored-by: Cursor --- gitnexus/src/core/ingestion/pipeline.ts | 98 ++++++++++++++++--------- gitnexus/src/core/kuzu/kuzu-adapter.ts | 23 +++--- 2 files changed, 78 insertions(+), 43 deletions(-) diff --git a/gitnexus/src/core/ingestion/pipeline.ts b/gitnexus/src/core/ingestion/pipeline.ts index bf95acbe6..33f5a7c16 100644 --- a/gitnexus/src/core/ingestion/pipeline.ts +++ b/gitnexus/src/core/ingestion/pipeline.ts @@ -455,10 +455,7 @@ export const runIncrementalPipeline = async ( console.log(`πŸ“ Incremental: parsed ${changedFiles.length} changed files, skipped ${unchangedCount} unchanged`); } - // ── Phase 5: Imports (ALL files β€” full re-resolution for cross-file correctness) ── - // We always use the full parser path here (not workerData fast path) because: - // - Pre-loaded relationships only include DEFINES (intra-file) - // - Cross-file IMPORTS must be freshly resolved to capture unchangedβ†’changed file edges + // ── Phase 5: Imports (changed files via workerData, unchanged from pre-loaded) ── onProgress({ phase: 'imports', percent: 60, @@ -466,20 +463,29 @@ export const runIncrementalPipeline = async ( stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, }); - // Resize AST cache for all files (import/call/heritage parse on-the-fly) - astCache = createASTCache(files.length); + if (workerData) { + await processImportsFromExtracted(graph, files, workerData.imports, importMap, (current, total) => { + const importProgress = 60 + ((current / total) * 12); + onProgress({ + phase: 'imports', + percent: Math.round(importProgress), + message: 'Resolving imports...', + stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + }); + }, repoPath); + } else { + await processImports(graph, files, astCache, importMap, (current, total) => { + const importProgress = 60 + ((current / total) * 12); + onProgress({ + phase: 'imports', + percent: Math.round(importProgress), + message: 'Resolving imports...', + stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + }); + }, repoPath); + } - await processImports(graph, files, astCache, importMap, (current, total) => { - const importProgress = 60 + ((current / total) * 12); - onProgress({ - phase: 'imports', - percent: Math.round(importProgress), - message: 'Resolving imports...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, - }); - }, repoPath); - - // ── Phase 6: Calls (ALL files) ──────────────────────────────────── + // ── Phase 6: Calls (changed files via workerData, unchanged from pre-loaded) ── onProgress({ phase: 'calls', percent: 72, @@ -487,17 +493,29 @@ export const runIncrementalPipeline = async ( stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, }); - await processCalls(graph, files, astCache, symbolTable, importMap, (current, total) => { - const callProgress = 72 + ((current / total) * 10); - onProgress({ - phase: 'calls', - percent: Math.round(callProgress), - message: 'Tracing function calls...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + if (workerData) { + await processCallsFromExtracted(graph, workerData.calls, symbolTable, importMap, (current, total) => { + const callProgress = 72 + ((current / total) * 10); + onProgress({ + phase: 'calls', + percent: Math.round(callProgress), + message: 'Tracing function calls...', + stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + }); }); - }); + } else { + await processCalls(graph, files, astCache, symbolTable, importMap, (current, total) => { + const callProgress = 72 + ((current / total) * 10); + onProgress({ + phase: 'calls', + percent: Math.round(callProgress), + message: 'Tracing function calls...', + stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + }); + }); + } - // ── Phase 7: Heritage (ALL files) ───────────────────────────────── + // ── Phase 7: Heritage (changed files via workerData, unchanged from pre-loaded) ── onProgress({ phase: 'heritage', percent: 82, @@ -505,15 +523,27 @@ export const runIncrementalPipeline = async ( stats: { filesProcessed: 0, totalFiles: files.length, nodesCreated: graph.nodeCount }, }); - await processHeritage(graph, files, astCache, symbolTable, (current, total) => { - const heritageProgress = 82 + ((current / total) * 6); - onProgress({ - phase: 'heritage', - percent: Math.round(heritageProgress), - message: 'Extracting class inheritance...', - stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + if (workerData) { + await processHeritageFromExtracted(graph, workerData.heritage, symbolTable, (current, total) => { + const heritageProgress = 82 + ((current / total) * 6); + onProgress({ + phase: 'heritage', + percent: Math.round(heritageProgress), + message: 'Extracting class inheritance...', + stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + }); }); - }); + } else { + await processHeritage(graph, files, astCache, symbolTable, (current, total) => { + const heritageProgress = 82 + ((current / total) * 6); + onProgress({ + phase: 'heritage', + percent: Math.round(heritageProgress), + message: 'Extracting class inheritance...', + stats: { filesProcessed: current, totalFiles: total, nodesCreated: graph.nodeCount }, + }); + }); + } // ── Phase 8: Communities ─────────────────────────────────────────── onProgress({ diff --git a/gitnexus/src/core/kuzu/kuzu-adapter.ts b/gitnexus/src/core/kuzu/kuzu-adapter.ts index f39d15b0d..732649acd 100644 --- a/gitnexus/src/core/kuzu/kuzu-adapter.ts +++ b/gitnexus/src/core/kuzu/kuzu-adapter.ts @@ -570,7 +570,11 @@ export const loadExistingGraph = async ( } catch { /* table may not exist or is empty */ } } - // Load relationships (exclude any involving changed/deleted files) + // Load relationships for unchanged files. + // Filter: SOURCE must be in retained nodes (unchanged files). + // TARGET can be any node β€” edges to changed/deleted nodes that no longer exist + // will naturally fail during KuzuDB COPY (gracefully skipped). + // This preserves cross-file edges (unchanged β†’ changed) from the old index. const relationships: GraphRelationship[] = []; const nodeIdSet = new Set(nodes.map(n => n.id)); try { @@ -581,15 +585,16 @@ export const loadExistingGraph = async ( for (const row of await result.getAll()) { const fromId = str(row.fromId ?? row[0]); const toId = str(row.toId ?? row[1]); - if (!nodeIdSet.has(fromId) || !nodeIdSet.has(toId)) continue; - // All relationships are regenerated by the incremental pipeline: - // CONTAINS (processStructure), IMPORTS/CALLS/EXTENDS/IMPLEMENTS - // (full import/call/heritage), MEMBER_OF/STEP_IN_PROCESS (communities/processes). - // Only keep DEFINES for unchanged nodes (File -> code element). + // Source must be from an unchanged file + if (!nodeIdSet.has(fromId)) continue; const type = str(row.type ?? row[2]); - if (type !== 'DEFINES') continue; - // For DEFINES: use the canonical ID format to avoid duplicates with freshly parsed results - const canonicalId = `DEFINES:${fromId}->${toId}`; + // Skip types that are regenerated globally by the incremental pipeline: + // - CONTAINS: recomputed by processStructure on all files + // - MEMBER_OF: recomputed by community detection + // - STEP_IN_PROCESS: recomputed by process detection + if (type === 'CONTAINS' || type === 'MEMBER_OF' || type === 'STEP_IN_PROCESS') continue; + // Use canonical ID format (matching generateId output) to avoid duplicates + const canonicalId = `${type}:${fromId}->${toId}`; relationships.push({ id: canonicalId, sourceId: fromId,