fix: entity upsert dedup on (project_id, name) + observability
CI / CI (pull_request) Successful in 11m36s
CI / CI (pull_request) Successful in 11m36s
Entity save now uses ON CONFLICT (project_id, name) DO UPDATE: - Merges description (keep non-empty) - Keeps highest confidence - Increments source_count - Updates t_updated timestamp Prevents duplicate entities across ingests (was 26 rows, now 11). Unique index added to production DB. Compaction (T3.1 exact dedup + T3.2 semantic) already wired at POST /memory/compact endpoint. Cache alignment + chunk optimizer wired through full_pipeline.rs + query_orchestrator.rs. 781 tests pass.
This commit is contained in:
@@ -10,6 +10,7 @@ use uuid::Uuid;
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use pgvector::Vector;
|
use pgvector::Vector;
|
||||||
|
|
||||||
|
|
||||||
/// Ingest worker — processes queued records through entity/fact extraction pipeline
|
/// Ingest worker — processes queued records through entity/fact extraction pipeline
|
||||||
pub struct IngestWorker {
|
pub struct IngestWorker {
|
||||||
pool: PgPool,
|
pool: PgPool,
|
||||||
@@ -135,9 +136,15 @@ impl IngestWorker {
|
|||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
tracing::info!(
|
tracing::info!(
|
||||||
"Ingest completed: {} (entities={}, edges={}, reviews={})",
|
target: "observability",
|
||||||
ingest_id, total_entities, total_edges, total_reviews
|
event = "ingest_complete",
|
||||||
|
ingest_id = ingest_id,
|
||||||
|
entities = total_entities,
|
||||||
|
edges = total_edges,
|
||||||
|
reviews = total_reviews,
|
||||||
|
"Ingest completed"
|
||||||
);
|
);
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -187,7 +194,12 @@ async fn save_entity_to_db(pool: &PgPool, entity: &mem_core::entity::Entity) ->
|
|||||||
sqlx::query(
|
sqlx::query(
|
||||||
"INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence)
|
"INSERT INTO memory_entity (id, project_id, name, entity_type, description, t_created, t_updated, confidence)
|
||||||
VALUES ($1, $2, $3, $4, $5, $6::TIMESTAMPTZ, $7::TIMESTAMPTZ, $8)
|
VALUES ($1, $2, $3, $4, $5, $6::TIMESTAMPTZ, $7::TIMESTAMPTZ, $8)
|
||||||
ON CONFLICT (id) DO NOTHING"
|
ON CONFLICT (project_id, name) DO UPDATE SET
|
||||||
|
entity_type = EXCLUDED.entity_type,
|
||||||
|
description = COALESCE(NULLIF(EXCLUDED.description, ''), memory_entity.description),
|
||||||
|
t_updated = NOW(),
|
||||||
|
confidence = GREATEST(memory_entity.confidence, EXCLUDED.confidence),
|
||||||
|
source_count = memory_entity.source_count + 1"
|
||||||
)
|
)
|
||||||
.bind(&entity.id)
|
.bind(&entity.id)
|
||||||
.bind(&entity.project_id)
|
.bind(&entity.project_id)
|
||||||
|
|||||||
Reference in New Issue
Block a user