salim0986/graph-bug-ai
0
1"""2Incremental Ingestion System with Performance Optimizations3 4Features:5- Git diff detection for changed files only6- Batch processing for embeddings (50+ at once)7- Parallel file parsing with asyncio8- Smart caching and upsert operations9- Memory-efficient processing10"""11 12import os13import git14import asyncio15from typing import List, Dict, Set, Tuple, Optional16from pathlib import Path17from .logger import setup_logger18from .parser import UniversalParser, EXTENSION_MAP, ParseReport19from .graph_builder import GraphBuilder20from .vector_builder import VectorBuilder21from .chunker import SemanticChunker22 23logger = setup_logger(__name__)24 25 26class IncrementalIngester:27 """28 High-performance incremental ingestion engine29 30 Optimizations:31 1. Only processes changed files (git diff)32 2. Batches embeddings (50+ functions per API call)33 3. Parallel file parsing with asyncio34 4. Upserts instead of full deletion35 5. Memory-efficient streaming36 """37 38 def __init__(39 self,40 parser: UniversalParser,41 graph_db: GraphBuilder,42 vector_db: VectorBuilder,43 ignore_dirs: Set[str]44 ):45 self.parser = parser46 self.graph_db = graph_db47 self.vector_db = vector_db48 self.ignore_dirs = ignore_dirs49 self.batch_size = 50 # Embed 50 functions at once50 self.chunker = SemanticChunker(max_tokens=512, overlap_tokens=64)51 52 async def ingest_incremental(53 self,54 repo_id: str,55 local_path: str,56 last_commit: Optional[str] = None,57 current_commit: str = "HEAD"58 ) -> Dict[str, int]:59 """60 Incrementally ingest only changed files61 62 Args:63 repo_id: Unique repository identifier64 local_path: Path to local git repository65 last_commit: Previous commit SHA (None for full ingestion)66 current_commit: Current commit SHA (default: HEAD)67 68 Returns:69 Dict with stats: files_processed, nodes_added, nodes_deleted, vectors_updated70 """71 logger.info("=" * 80)72 logger.info(f"🔄 INCREMENTAL INGESTION STARTING")73 logger.info(f" Repo ID: {repo_id}")74 logger.info(f" Local path: {local_path}")75 logger.info(f" Range: {last_commit or 'initial'} → {current_commit}")76 logger.info("=" * 80)77 78 stats = {79 "files_processed": 0,80 "files_deleted": 0,81 "nodes_added": 0,82 "nodes_deleted": 0,83 "vectors_updated": 084 }85 86 try:87 repo = git.Repo(local_path)88 89 # Get changed files90 if last_commit:91 changed_files, deleted_files = self._get_changed_files(92 repo, last_commit, current_commit93 )94 else:95 # Full ingestion - process all files96 logger.info("📁 Full ingestion mode - scanning all files...")97 changed_files = self._get_all_files(local_path)98 deleted_files = set()99 100 logger.info(f"📊 Changed files: {len(changed_files)}, Deleted files: {len(deleted_files)}")101 102 # Process deleted files first103 if deleted_files:104 logger.info(f"🗑️ Processing {len(deleted_files)} deleted files...")105 for file_path in deleted_files:106 await self._delete_file_data(repo_id, file_path)107 stats["files_deleted"] += 1108 stats["nodes_deleted"] += 1 # Approximate109 110 # Process changed files in parallel batches111 if changed_files:112 logger.info(f"⚙️ Processing {len(changed_files)} changed files in parallel batches...")113 batch_size = 10 # Parse 10 files in parallel114 for i in range(0, len(changed_files), batch_size):115 batch = changed_files[i:i + batch_size]116 logger.info(f" Batch {i//batch_size + 1}/{(len(changed_files) + batch_size - 1)//batch_size}: Processing {len(batch)} files...")117 batch_stats = await self._process_file_batch(118 repo_id, local_path, batch119 )120 stats["files_processed"] += batch_stats["files_processed"]121 stats["nodes_added"] += batch_stats["nodes_added"]122 stats["vectors_updated"] += batch_stats["vectors_updated"]123 124 # Rebuild dependencies for changed files only125 logger.info("🔗 Rebuilding dependencies...")126 self.graph_db.build_dependencies(repo_id)127 128 logger.info("=" * 80)129 logger.info(f"✅ INCREMENTAL INGESTION COMPLETE")130 logger.info(f" Files processed: {stats['files_processed']}")131 logger.info(f" Files deleted: {stats['files_deleted']}")132 logger.info(f" Parse failures: {stats.get('parse_failures', 0)}")133 logger.info(f" Nodes added: {stats['nodes_added']}")134 logger.info(f" Nodes deleted: {stats['nodes_deleted']}")135 logger.info(f" Vectors updated: {stats['vectors_updated']}")136 logger.info(f" [M1] Calls found: {stats.get('calls_extracted', 0)}")137 logger.info(f" [M1] Imports found: {stats.get('imports_extracted', 0)}")138 logger.info(f" [M1] Inherit edges: {stats.get('inheritances_extracted', 0)}")139 logger.info("=" * 80)140 return stats141 142 except Exception as e:143 logger.error("=" * 80)144 logger.error(f"❌ INCREMENTAL INGESTION FAILED")145 logger.error(f" Repo ID: {repo_id}")146 logger.error(f" Error: {e}")147 logger.error("=" * 80)148 logger.error(f"Full traceback:", exc_info=True)149 raise150 151 def _get_changed_files(152 self,153 repo: git.Repo,154 from_commit: str,155 to_commit: str156 ) -> Tuple[List[str], Set[str]]:157 """158 Get list of changed and deleted files between two commits159 160 Returns:161 (changed_files, deleted_files)162 """163 try:164 # Get diff between commits165 diff = repo.commit(from_commit).diff(to_commit)166 167 changed_files = []168 deleted_files = set()169 170 for change in diff:171 # change.a_path is the file path172 file_path = change.a_path or change.b_path173 174 # Filter by extension175 if not self._is_valid_file(file_path):176 continue177 178 if change.deleted_file:179 deleted_files.add(file_path)180 else:181 # Modified or added file182 changed_files.append(file_path)183 184 return changed_files, deleted_files185 186 except Exception as e:187 logger.error(f"Error getting changed files: {e}")188 return [], set()189 190 def _get_all_files(self, local_path: str) -> List[str]:191 """Get all valid files for full ingestion"""192 all_files = []193 194 for root, dirs, files in os.walk(local_path):195 # Filter ignored directories196 dirs[:] = [d for d in dirs if d not in self.ignore_dirs]197 198 for file in files:199 file_path = os.path.join(root, file)200 rel_path = os.path.relpath(file_path, local_path)201 202 if self._is_valid_file(rel_path):203 all_files.append(rel_path)204 205 return all_files206 207 def _is_valid_file(self, file_path: str) -> bool:208 """Check if file should be processed"""209 filename = os.path.basename(file_path)210 _, ext = os.path.splitext(filename)211 return filename in EXTENSION_MAP or ext in EXTENSION_MAP212 213 async def _delete_file_data(self, repo_id: str, file_path: str):214 """Delete all data for a file"""215 try:216 # Delete from graph217 file_uid = f"{repo_id}::{file_path}"218 self.graph_db.delete_file(file_uid)219 220 # Delete from vectors221 self.vector_db.delete_by_file(repo_id, file_path)222 223 logger.debug(f"Deleted data for {file_path}")224 225 except Exception as e:226 logger.error(f"Error deleting file data for {file_path}: {e}")227 228 async def _process_file_batch(229 self,230 repo_id: str,231 base_path: str,232 file_paths: List[str]233 ) -> Dict[str, int]:234 """235 Process a batch of files in parallel236 237 Returns stats for the batch238 """239 stats = {240 "files_processed": 0,241 "nodes_added": 0,242 "vectors_updated": 0,243 # M1 relationship stats244 "parse_failures": 0,245 "calls_extracted": 0,246 "imports_extracted": 0,247 "inheritances_extracted": 0,248 }249 250 # Parse files in parallel (two tasks per file: legacy + relationships)251 parse_tasks = [252 self._parse_file(base_path, file_path)253 for file_path in file_paths254 ]255 rel_tasks = [256 self._extract_relationships(base_path, file_path)257 for file_path in file_paths258 ]259 ast_tasks = [260 self._parse_file_ast(base_path, file_path)261 for file_path in file_paths262 ]263 264 parse_results = await asyncio.gather(*parse_tasks, return_exceptions=True)265 rel_results = await asyncio.gather(*rel_tasks, return_exceptions=True)266 ast_results = await asyncio.gather(*ast_tasks, return_exceptions=True)267 268 # Collect all chunks for batch embedding269 all_chunks_for_embedding = []270 271 for file_path, result, rel_result, ast_result in zip(file_paths, parse_results, rel_results, ast_results):272 if isinstance(result, Exception):273 logger.error(f"Error parsing {file_path}: {result}")274 stats["parse_failures"] += 1275 continue276 277 if result is None:278 continue279 280 captures, code_bytes = result281 282 if not captures:283 continue284 285 stats["files_processed"] += 1286 287 # Update graph (legacy path — declarations/nodes)288 try:289 self.graph_db.process_file_nodes(290 repo_id, file_path, captures, code_bytes291 )292 stats["nodes_added"] += len(captures)293 except Exception as e:294 logger.error(f"Error updating graph for {file_path}: {e}")295 continue296 297 # M4: store relationship edges (CALLS / IMPORTS / INHERITS) in Neo4j298 if isinstance(rel_result, ParseReport):299 if rel_result.status == "failed" and rel_result.errors:300 stats["parse_failures"] += 1301 logger.warning(302 f"[M4] Relationship extraction failed for {file_path}: "303 f"{rel_result.errors[0]}"304 )305 else:306 stats["calls_extracted"] += len(rel_result.calls)307 stats["imports_extracted"] += len(rel_result.imports)308 stats["inheritances_extracted"] += len(rel_result.inheritances)309 try:310 rel_stats = self.graph_db.process_relationships(311 repo_id, file_path, rel_result312 )313 logger.debug(314 f"[M4] {file_path}: linked {rel_stats['calls_linked']} calls "315 f"({rel_stats['calls_external']} ext), "316 f"{rel_stats['imports_linked']} imports, "317 f"{rel_stats['inheritances_linked']} inherits "318 f"({rel_stats['inheritances_external']} ext)"319 )320 except Exception as e:321 logger.error(f"[M4] Failed to store relationships for {file_path}: {e}")322 elif isinstance(rel_result, Exception):323 logger.error(f"[M4] Relationship extraction exception for {file_path}: {rel_result}")324 325 # Collect chunks using SemanticChunker for embedding (M2)326 if not isinstance(ast_result, Exception) and ast_result is not None:327 tree, code_bytes, language = ast_result328 if tree is not None:329 try:330 chunks = self.chunker.chunk_ast(tree, code_bytes, file_path, language)331 for chunk in chunks:332 chunk["repo_id"] = repo_id333 all_chunks_for_embedding.append(chunk)334 except Exception as e:335 logger.warning(f"Error chunking file {file_path}: {e}")336 337 # Batch embed all collected chunks338 if all_chunks_for_embedding:339 embedded_count = await self._batch_embed_nodes(all_chunks_for_embedding)340 stats["vectors_updated"] = embedded_count341 342 return stats343 344 async def _parse_file(345 self,346 base_path: str,347 file_path: str348 ) -> Optional[Tuple]:349 """Parse a single file asynchronously (legacy captures interface)."""350 try:351 full_path = os.path.join(base_path, file_path)352 loop = asyncio.get_event_loop()353 return await loop.run_in_executor(354 None,355 self.parser.parse_file,356 full_path357 )358 except Exception as e:359 logger.error(f"Parse error for {file_path}: {e}")360 return None361 362 async def _parse_file_ast(363 self,364 base_path: str,365 file_path: str366 ) -> Optional[Tuple]:367 """Get the raw AST for M2 Semantic Chunker."""368 try:369 full_path = os.path.join(base_path, file_path)370 loop = asyncio.get_event_loop()371 return await loop.run_in_executor(372 None,373 self.parser.parse_file_ast,374 full_path375 )376 except Exception as e:377 logger.error(f"AST parse error for {file_path}: {e}")378 return None379 380 async def _extract_relationships(381 self,382 base_path: str,383 file_path: str384 ) -> "ParseReport":385 """Extract typed relationship records (M1) for a single file."""386 try:387 full_path = os.path.join(base_path, file_path)388 loop = asyncio.get_event_loop()389 return await loop.run_in_executor(390 None,391 self.parser.extract_relationships_from_file,392 full_path393 )394 except Exception as e:395 logger.error(f"[M1] Relationship extraction error for {file_path}: {e}")396 from .parser import ParseReport397 report = ParseReport(file=file_path, language=None, status="failed")398 report.errors.append(str(e))399 return report400 401 async def _batch_embed_nodes(self, nodes: List[Dict]) -> int:402 """403 Embed nodes in batches for performance404 405 Args:406 nodes: List of node dictionaries with code and metadata407 408 Returns:409 Number of vectors created410 """411 embedded_count = 0412 413 try:414 # Process in batches of self.batch_size415 for i in range(0, len(nodes), self.batch_size):416 batch = nodes[i:i + self.batch_size]417 418 # Extract code snippets for batch encoding419 code_snippets = [node["raw_code"] for node in batch]420 421 # Batch embed (this is much faster than one-by-one)422 embeddings = self.vector_db.embed_model.encode(423 code_snippets,424 batch_size=self.batch_size,425 show_progress_bar=False426 )427 428 # Store vectors with metadata429 for node, embedding in zip(batch, embeddings):430 try:431 self.vector_db.upsert_function_vector(432 repo_id=node["repo_id"],433 func_name=node.get("parent_function") or node.get("parent_class") or "chunk",434 embedding=embedding,435 file_path=node["file"],436 start_line=node["start_line"],437 raw_code=node["raw_code"],438 language=node.get("language"),439 parent_function=node.get("parent_function"),440 parent_class=node.get("parent_class")441 )442 embedded_count += 1443 444 except Exception as e:445 logger.warning(f"Error storing vector: {e}")446 447 logger.debug(f"Batch embedded {len(batch)} nodes")448 449 except Exception as e:450 logger.error(f"Batch embedding failed: {e}")451 452 return embedded_count453 454 455async def ingest_repo_incremental(456 repo_id: str,457 repo_url: str,458 local_path: str,459 parser: UniversalParser,460 graph_db: GraphBuilder,461 vector_db: VectorBuilder,462 ignore_dirs: Set[str],463 last_commit: Optional[str] = None464) -> Dict[str, int]:465 """466 High-level function to perform incremental ingestion467 468 Args:469 repo_id: Repository identifier470 repo_url: Git repository URL471 local_path: Local clone path472 parser: UniversalParser instance473 graph_db: GraphBuilder instance474 vector_db: VectorBuilder instance475 ignore_dirs: Set of directories to ignore476 last_commit: Previous commit SHA (None for full ingestion)477 478 Returns:479 Stats dictionary480 """481 ingester = IncrementalIngester(parser, graph_db, vector_db, ignore_dirs)482 483 # Ensure collection exists484 vector_db.ensure_collection()485 486 # Run incremental ingestion487 stats = await ingester.ingest_incremental(488 repo_id=repo_id,489 local_path=local_path,490 last_commit=last_commit491 )492 493 return stats494