1use crate::errors::AppError;
11use crate::i18n::errors_msg;
12use crate::output::{self, OutputFormat};
13use crate::paths::AppPaths;
14use crate::storage::connection::open_rw;
15use crate::storage::entities;
16use rusqlite::params;
17use serde::Serialize;
18
19#[derive(clap::Args)]
20#[command(after_long_help = "EXAMPLES:\n \
21 # Merge two source entities into a target\n \
22 sqlite-graphrag merge-entities --names auth,authentication --into auth-service\n\n \
23 # Merge three sources into one target across a namespace\n \
24 sqlite-graphrag merge-entities --names svc-a,svc-b,old-svc --into canonical-service --namespace my-project\n\n \
25 # Merge by ID (unambiguous when homonyms exist across namespaces)\n \
26 sqlite-graphrag merge-entities --ids 12,17 --into-id 3\n\n\
27NOTE:\n \
28 --names is a comma-separated list of source entity names.\n \
29 --into is the target entity name and must already exist.\n \
30 --ids / --into-id select entities by ID; IDs are globally unique so they\n \
31 disambiguate homonyms. They conflict with --names / --into respectively\n \
32 and must belong to the resolved namespace.\n \
33 Source entities are deleted after the merge; the target is preserved.\n \
34 Duplicate relationships (same endpoints + relation) are removed automatically.\n \
35 Run `sqlite-graphrag cleanup-orphans` afterwards if sources had no other links.")]
36pub struct MergeEntitiesArgs {
37 #[arg(
39 long,
40 value_delimiter = ',',
41 value_name = "NAMES",
42 required_unless_present = "ids",
43 conflicts_with = "ids"
44 )]
45 pub names: Vec<String>,
46 #[arg(long, value_delimiter = ',', value_name = "IDS")]
50 pub ids: Vec<i64>,
51 #[arg(
53 long,
54 value_name = "TARGET",
55 required_unless_present = "into_id",
56 conflicts_with = "into_id"
57 )]
58 pub into: Option<String>,
59 #[arg(long, value_name = "TARGET_ID")]
61 pub into_id: Option<i64>,
62 #[arg(long)]
63 pub namespace: Option<String>,
64 #[arg(long, value_enum, default_value = "json")]
65 pub format: OutputFormat,
66 #[arg(long, hide = true, help = "No-op; JSON is always emitted on stdout")]
67 pub json: bool,
68 #[arg(long, env = "SQLITE_GRAPHRAG_DB_PATH")]
69 pub db: Option<String>,
70 #[arg(long, default_value_t = false, hide = false)]
75 pub cross_namespace: bool,
76}
77
78#[derive(Serialize)]
79struct MergeEntitiesResponse {
80 action: String,
81 sources: Vec<String>,
82 target: String,
83 namespace: String,
84 target_id: i64,
86 relationships_moved: usize,
87 entities_removed: usize,
88 elapsed_ms: u64,
90}
91
92fn find_entity_name_by_id(
99 conn: &rusqlite::Connection,
100 namespace: &str,
101 id: i64,
102 enforce_namespace: bool,
103) -> Result<(String, String), AppError> {
104 let mut stmt = if enforce_namespace {
105 conn.prepare_cached(
106 "SELECT name, namespace FROM entities WHERE id = ?1 AND namespace = ?2",
107 )?
108 } else {
109 conn.prepare_cached("SELECT name, namespace FROM entities WHERE id = ?1")?
110 };
111 let row = if enforce_namespace {
112 stmt.query_row(params![id, namespace], |r| {
113 Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
114 })
115 } else {
116 stmt.query_row(params![id], |r| {
117 Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))
118 })
119 };
120 match row {
121 Ok((name, ns_actual)) => Ok((name, ns_actual)),
122 Err(rusqlite::Error::QueryReturnedNoRows) => Err(AppError::NotFound(format!(
123 "entity id={id} not found in namespace '{namespace}'"
124 ))),
125 Err(e) => Err(AppError::Database(e)),
126 }
127}
128
129pub fn run(args: MergeEntitiesArgs) -> Result<(), AppError> {
130 let inicio = std::time::Instant::now();
131
132 if args.names.is_empty() && args.ids.is_empty() {
133 return Err(AppError::Validation(
134 "--names or --ids must contain at least one source entity".to_string(),
135 ));
136 }
137
138 let namespace = crate::namespace::resolve_namespace(args.namespace.as_deref())?;
139 let paths = AppPaths::resolve(args.db.as_deref())?;
140
141 crate::storage::connection::ensure_db_ready(&paths)?;
142
143 let mut conn = open_rw(&paths.db)?;
144
145 let (target_id, target_name) = match args.into_id {
148 Some(id) => {
149 let (name, _ns_actual) = find_entity_name_by_id(&conn, &namespace, id, true)?;
152 (id, name)
153 }
154 None => {
155 let Some(name) = args.into.clone() else {
156 return Err(AppError::Validation(
157 "--into or --into-id is required".to_string(),
158 ));
159 };
160 let id = entities::find_entity_id(&conn, &namespace, &name)?.ok_or_else(|| {
161 AppError::NotFound(errors_msg::entity_not_found(&name, &namespace))
162 })?;
163 (id, name)
164 }
165 };
166
167 let mut source_ids: Vec<i64> = Vec::with_capacity(args.names.len() + args.ids.len());
170 let mut source_names: Vec<String> = Vec::with_capacity(source_ids.capacity());
171 if !args.ids.is_empty() {
172 for &id in &args.ids {
173 if id == target_id {
174 return Err(AppError::Validation(format!(
175 "source entity id={id} equals target id={target_id} — \
176 self-referential merge is not allowed"
177 )));
178 }
179 let (name, ns_actual) =
183 find_entity_name_by_id(&conn, &namespace, id, !args.cross_namespace)?;
184 if args.cross_namespace && ns_actual != namespace {
185 tracing::warn!(
186 target: "merge_entities",
187 from_id = id,
188 from_namespace = %ns_actual,
189 to_namespace = %namespace,
190 "cross-namespace merge"
191 );
192 }
193 if !source_ids.contains(&id) {
194 source_ids.push(id);
195 source_names.push(name);
196 }
197 }
198 } else {
199 for name in &args.names {
200 if name == &target_name {
201 return Err(AppError::Validation(format!(
202 "source entity '{name}' equals target '{target_name}' — \
203 self-referential merge is not allowed"
204 )));
205 }
206 let id = entities::find_entity_id(&conn, &namespace, name)?.ok_or_else(|| {
207 AppError::NotFound(errors_msg::entity_not_found(name, &namespace))
208 })?;
209 if id == target_id {
210 return Err(AppError::Validation(format!(
211 "source entity '{name}' resolves to the target (id={target_id}) — \
212 self-referential merge is not allowed"
213 )));
214 }
215 if !source_ids.contains(&id) {
216 source_ids.push(id);
217 source_names.push(name.clone());
218 }
219 }
220 }
221
222 if source_ids.is_empty() {
223 return Err(AppError::Validation(
224 "no valid source entities to merge (all names equal the target or were duplicates)"
225 .to_string(),
226 ));
227 }
228
229 let tx = conn.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?;
230
231 let mut relationships_moved: usize = 0;
232
233 for &src_id in &source_ids {
234 let moved_src = tx.execute(
236 "UPDATE OR IGNORE relationships SET source_id = ?1 WHERE source_id = ?2",
237 params![target_id, src_id],
238 )?;
239 tx.execute(
240 "DELETE FROM relationships WHERE source_id = ?1",
241 params![src_id],
242 )?;
243 let moved_tgt = tx.execute(
245 "UPDATE OR IGNORE relationships SET target_id = ?1 WHERE target_id = ?2",
246 params![target_id, src_id],
247 )?;
248 tx.execute(
249 "DELETE FROM relationships WHERE target_id = ?1",
250 params![src_id],
251 )?;
252 relationships_moved += moved_src + moved_tgt;
253 }
254
255 tx.execute("DELETE FROM relationships WHERE source_id = target_id", [])?;
257
258 tx.execute(
261 "DELETE FROM relationships
262 WHERE id NOT IN (
263 SELECT MIN(id)
264 FROM relationships
265 GROUP BY source_id, target_id, relation
266 )",
267 [],
268 )?;
269
270 for &src_id in &source_ids {
275 tx.execute(
276 "UPDATE OR IGNORE memory_entities SET entity_id = ?1 WHERE entity_id = ?2",
277 params![target_id, src_id],
278 )?;
279 tx.execute(
280 "DELETE FROM memory_entities WHERE entity_id = ?1",
281 params![src_id],
282 )?;
283 }
284
285 tx.execute(
287 "DELETE FROM memory_entities
288 WHERE rowid NOT IN (
289 SELECT MIN(rowid)
290 FROM memory_entities
291 GROUP BY memory_id, entity_id
292 )",
293 [],
294 )?;
295
296 let mut entities_removed: usize = 0;
299 for &src_id in &source_ids {
300 let removed = tx.execute("DELETE FROM entities WHERE id = ?1", params![src_id])?;
301 entities_removed += removed;
302 }
303
304 let adjacent_ids: Vec<i64> = {
306 let mut stmt = tx.prepare(
307 "SELECT DISTINCT CASE WHEN source_id = ?1 THEN target_id ELSE source_id END
308 FROM relationships WHERE source_id = ?1 OR target_id = ?1",
309 )?;
310 let ids: Vec<i64> = stmt
311 .query_map(params![target_id], |r| r.get(0))?
312 .collect::<Result<Vec<_>, _>>()?;
313 ids
314 };
315 entities::recalculate_degree(&tx, target_id)?;
316 for &adj_id in &adjacent_ids {
317 entities::recalculate_degree(&tx, adj_id)?;
318 }
319
320 tx.commit()?;
321
322 conn.execute_batch("PRAGMA wal_checkpoint(TRUNCATE);")?;
323
324 let response = MergeEntitiesResponse {
325 action: "merged".to_string(),
326 sources: source_names,
327 target: target_name,
328 namespace: namespace.clone(),
329 target_id,
330 relationships_moved,
331 entities_removed,
332 elapsed_ms: inicio.elapsed().as_millis() as u64,
333 };
334
335 match args.format {
336 OutputFormat::Json => output::emit_json(&response)?,
337 OutputFormat::Text | OutputFormat::Markdown => {
338 output::emit_text(&format!(
339 "merged: {} sources into '{}' (relationships_moved={}, entities_removed={}) [{}]",
340 response.sources.len(),
341 response.target,
342 response.relationships_moved,
343 response.entities_removed,
344 response.namespace
345 ));
346 }
347 }
348
349 Ok(())
350}
351
352#[cfg(test)]
353mod tests {
354 use super::*;
355
356 #[test]
359 fn find_entity_name_by_id_disambiguates_homonyms_across_namespaces() {
360 let conn = rusqlite::Connection::open_in_memory().unwrap();
361 conn.execute_batch(
362 "CREATE TABLE entities (
363 id INTEGER PRIMARY KEY,
364 namespace TEXT NOT NULL,
365 name TEXT NOT NULL,
366 UNIQUE(namespace, name)
367 );",
368 )
369 .unwrap();
370 conn.execute(
371 "INSERT INTO entities (id, namespace, name)
372 VALUES (1, 'ns-a', 'auth'), (2, 'ns-b', 'auth')",
373 [],
374 )
375 .unwrap();
376
377 assert_eq!(
379 find_entity_name_by_id(&conn, "ns-a", 1, true).unwrap(),
380 ("auth".to_string(), "ns-a".to_string())
381 );
382 assert_eq!(
383 find_entity_name_by_id(&conn, "ns-b", 2, true).unwrap(),
384 ("auth".to_string(), "ns-b".to_string())
385 );
386 let err = find_entity_name_by_id(&conn, "ns-a", 2, true).unwrap_err();
387 assert_eq!(err.exit_code(), 4, "cross-namespace ID must be NotFound");
388 assert!(err.to_string().contains("id=2"), "obtido: {err}");
389
390 let (name, ns_actual) = find_entity_name_by_id(&conn, "ns-a", 2, false).unwrap();
393 assert_eq!(name, "auth");
394 assert_eq!(ns_actual, "ns-b");
395 }
396
397 #[test]
398 fn find_entity_name_by_id_missing_id_is_not_found() {
399 let conn = rusqlite::Connection::open_in_memory().unwrap();
400 conn.execute_batch(
401 "CREATE TABLE entities (
402 id INTEGER PRIMARY KEY,
403 namespace TEXT NOT NULL,
404 name TEXT NOT NULL
405 );",
406 )
407 .unwrap();
408 let err = find_entity_name_by_id(&conn, "global", 99, true).unwrap_err();
409 assert_eq!(err.exit_code(), 4);
410 }
411
412 #[derive(clap::Parser)]
415 struct TestCli {
416 #[command(flatten)]
417 args: MergeEntitiesArgs,
418 }
419
420 #[test]
421 fn clap_rejects_names_combined_with_ids() {
422 use clap::Parser;
423 let err =
424 match TestCli::try_parse_from(["t", "--names", "a,b", "--ids", "1,2", "--into", "tgt"])
425 {
426 Ok(_) => panic!("expected argument conflict"),
427 Err(e) => e,
428 };
429 assert_eq!(err.kind(), clap::error::ErrorKind::ArgumentConflict);
430 }
431
432 #[test]
433 fn clap_rejects_into_combined_with_into_id() {
434 use clap::Parser;
435 let err =
436 match TestCli::try_parse_from(["t", "--names", "a", "--into", "tgt", "--into-id", "3"])
437 {
438 Ok(_) => panic!("expected argument conflict"),
439 Err(e) => e,
440 };
441 assert_eq!(err.kind(), clap::error::ErrorKind::ArgumentConflict);
442 }
443
444 #[test]
445 fn clap_requires_a_source_and_a_target_selector() {
446 use clap::Parser;
447 assert!(TestCli::try_parse_from(["t", "--into", "tgt"]).is_err());
448 assert!(TestCli::try_parse_from(["t", "--names", "a"]).is_err());
449 let ok = match TestCli::try_parse_from(["t", "--ids", "1,2", "--into-id", "3"]) {
450 Ok(cli) => cli,
451 Err(e) => panic!("expected successful parse: {e}"),
452 };
453 assert_eq!(ok.args.ids, vec![1, 2]);
454 assert_eq!(ok.args.into_id, Some(3));
455 assert!(ok.args.names.is_empty());
456 assert!(ok.args.into.is_none());
457 }
458
459 #[test]
460 fn merge_entities_response_serializes_all_fields() {
461 let resp = MergeEntitiesResponse {
462 action: "merged".to_string(),
463 sources: vec!["auth".to_string(), "authentication".to_string()],
464 target: "auth-service".to_string(),
465 namespace: "global".to_string(),
466 target_id: 1,
467 relationships_moved: 7,
468 entities_removed: 2,
469 elapsed_ms: 15,
470 };
471 let json = serde_json::to_value(&resp).expect("serialization failed");
472 assert_eq!(json["action"], "merged");
473 assert_eq!(json["target"], "auth-service");
474 assert_eq!(json["namespace"], "global");
475 assert_eq!(json["relationships_moved"], 7);
476 assert_eq!(json["entities_removed"], 2);
477 let sources = json["sources"].as_array().expect("must be array");
478 assert_eq!(sources.len(), 2);
479 assert!(json["elapsed_ms"].is_number());
480 }
481
482 #[test]
483 fn merge_entities_response_action_is_merged() {
484 let resp = MergeEntitiesResponse {
485 action: "merged".to_string(),
486 sources: vec!["src".to_string()],
487 target: "tgt".to_string(),
488 namespace: "ns".to_string(),
489 target_id: 1,
490 relationships_moved: 0,
491 entities_removed: 1,
492 elapsed_ms: 0,
493 };
494 assert_eq!(resp.action, "merged");
495 }
496
497 #[test]
498 fn merge_entities_response_empty_sources_serializes() {
499 let resp = MergeEntitiesResponse {
500 action: "merged".to_string(),
501 sources: vec![],
502 target: "target".to_string(),
503 namespace: "global".to_string(),
504 target_id: 1,
505 relationships_moved: 0,
506 entities_removed: 0,
507 elapsed_ms: 1,
508 };
509 let json = serde_json::to_value(&resp).expect("serialization failed");
510 let sources = json["sources"].as_array().expect("must be array");
511 assert_eq!(sources.len(), 0);
512 }
513
514 #[test]
515 fn merge_entities_response_with_zero_relationships_moved() {
516 let resp = MergeEntitiesResponse {
517 action: "merged".to_string(),
518 sources: vec!["src-a".to_string()],
519 target: "tgt".to_string(),
520 namespace: "global".to_string(),
521 target_id: 1,
522 relationships_moved: 0,
523 entities_removed: 1,
524 elapsed_ms: 5,
525 };
526 let json = serde_json::to_value(&resp).expect("serialization failed");
527 assert_eq!(json["relationships_moved"], 0);
528 assert_eq!(json["entities_removed"], 1);
529 }
530
531 #[test]
532 fn merge_entities_response_multiple_sources() {
533 let resp = MergeEntitiesResponse {
534 action: "merged".to_string(),
535 sources: vec!["a".into(), "b".into(), "c".into()],
536 target: "canonical".to_string(),
537 namespace: "proj".to_string(),
538 target_id: 1,
539 relationships_moved: 12,
540 entities_removed: 3,
541 elapsed_ms: 42,
542 };
543 let json = serde_json::to_value(&resp).expect("serialization failed");
544 assert_eq!(json["entities_removed"], 3);
545 let sources = json["sources"].as_array().unwrap();
546 assert_eq!(sources.len(), 3);
547 }
548
549 fn setup_migrated_db_on_disk() -> (tempfile::TempDir, rusqlite::Connection, std::path::PathBuf)
553 {
554 crate::storage::connection::register_vec_extension();
555 let tmp = tempfile::TempDir::new().expect("tempdir");
556 let db_path = tmp.path().join("test.db");
557 let mut conn = rusqlite::Connection::open(&db_path).expect("open");
558 crate::migrations::runner().run(&mut conn).expect("migrate");
559 (tmp, conn, db_path)
560 }
561
562 #[test]
566 fn cross_namespace_merges_source_from_other_namespace() {
567 let (_tmp, conn, db_path) = setup_migrated_db_on_disk();
568 conn.execute(
570 "INSERT INTO entities (namespace, name, type) VALUES ('global','tgt','concept')",
571 [],
572 )
573 .unwrap();
574 let tgt_id = conn.last_insert_rowid();
575 conn.execute(
577 "INSERT INTO entities (namespace, name, type) VALUES ('ai-sdd','dup-src','concept')",
578 [],
579 )
580 .unwrap();
581 let src_id = conn.last_insert_rowid();
582 conn.execute(
585 "INSERT INTO entities (namespace, name, type) VALUES ('global','third','concept')",
586 [],
587 )
588 .unwrap();
589 let third_id = conn.last_insert_rowid();
590 conn.execute(
591 "INSERT INTO relationships (namespace, source_id, target_id, relation, weight) \
592 VALUES ('ai-sdd', ?1, ?2, 'related', 0.5)",
593 params![src_id, third_id],
594 )
595 .unwrap();
596 drop(conn);
598
599 let args = MergeEntitiesArgs {
600 names: vec![],
601 ids: vec![src_id],
602 into: None,
603 into_id: Some(tgt_id),
604 namespace: Some("global".to_string()),
605 format: OutputFormat::Json,
606 json: false,
607 db: Some(db_path.to_string_lossy().into_owned()),
608 cross_namespace: true,
609 };
610 run(args).expect("cross-namespace merge must succeed");
611
612 let conn = rusqlite::Connection::open(&db_path).unwrap();
614 let src_remaining: i64 = conn
615 .query_row(
616 "SELECT COUNT(*) FROM entities WHERE id = ?1",
617 params![src_id],
618 |r| r.get(0),
619 )
620 .unwrap();
621 assert_eq!(src_remaining, 0, "cross-namespace source must be deleted");
622 let moved: i64 = conn
623 .query_row(
624 "SELECT COUNT(*) FROM relationships WHERE source_id = ?1 AND target_id = ?2",
625 params![tgt_id, third_id],
626 |r| r.get(0),
627 )
628 .unwrap();
629 assert!(
630 moved > 0,
631 "relationship must be retargeted to the target entity"
632 );
633 }
634
635 #[test]
639 fn cross_namespace_default_false_rejects_cross_id() {
640 let (_tmp, conn, db_path) = setup_migrated_db_on_disk();
641 conn.execute(
642 "INSERT INTO entities (namespace, name, type) VALUES ('global','tgt','concept')",
643 [],
644 )
645 .unwrap();
646 let tgt_id = conn.last_insert_rowid();
647 conn.execute(
648 "INSERT INTO entities (namespace, name, type) VALUES ('ai-sdd','dup-src','concept')",
649 [],
650 )
651 .unwrap();
652 let src_id = conn.last_insert_rowid();
653 drop(conn);
654
655 let args = MergeEntitiesArgs {
656 names: vec![],
657 ids: vec![src_id],
658 into: None,
659 into_id: Some(tgt_id),
660 namespace: Some("global".to_string()),
661 format: OutputFormat::Json,
662 json: false,
663 db: Some(db_path.to_string_lossy().into_owned()),
664 cross_namespace: false,
665 };
666 let err = run(args).expect_err("default must reject cross-namespace ID");
667 assert_eq!(err.exit_code(), 4, "cross-namespace ID must be NotFound");
668 }
669
670 #[test]
673 fn cross_namespace_target_must_still_be_in_resolved_namespace() {
674 let (_tmp, conn, db_path) = setup_migrated_db_on_disk();
675 conn.execute(
677 "INSERT INTO entities (namespace, name, type) VALUES ('ai-sdd','tgt','concept')",
678 [],
679 )
680 .unwrap();
681 let tgt_id = conn.last_insert_rowid();
682 conn.execute(
683 "INSERT INTO entities (namespace, name, type) VALUES ('global','src','concept')",
684 [],
685 )
686 .unwrap();
687 let src_id = conn.last_insert_rowid();
688 drop(conn);
689
690 let args = MergeEntitiesArgs {
691 names: vec![],
692 ids: vec![src_id],
693 into: None,
694 into_id: Some(tgt_id),
695 namespace: Some("global".to_string()),
696 format: OutputFormat::Json,
697 json: false,
698 db: Some(db_path.to_string_lossy().into_owned()),
699 cross_namespace: true,
700 };
701 let err = run(args).expect_err("target in wrong namespace must fail");
702 assert_eq!(
703 err.exit_code(),
704 4,
705 "target must still be NotFound in the resolved namespace"
706 );
707 }
708}