Dropping an identity document's name vectors reads only its evidence

crates/utopia-cli/src/main.rs

@@ -82,7 +82,7 @@ struct ManifestDataDir {
 /// not a side effect of a code change.
 // 是迁移文件的**个数**，不是最大的编号（守卫 `schema_version_policy_compares_against_current`
 // 按个数比）：编号有空缺时两者不同——0071 由一个开放 PR 占着，0072 先落，个数是 71
-const CURRENT_SCHEMA_VERSION: u32 = 107;
+const CURRENT_SCHEMA_VERSION: u32 = 108;
 
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {

crates/utopia-store/src/name_vectors.rs

@@ -79,31 +79,31 @@ pub async fn pending(
     Ok(rows)
 }
 
+const DROP_FOR_SOURCE_SQL: &str = "DELETE FROM name_vectors v
+          USING fact_evidence fe JOIN documents d ON d.id = fe.document_id
+          WHERE fe.fact_id = v.fact_id AND d.source_id = $1";
+
+const DROP_FOR_DOCUMENT_SQL: &str = "DELETE FROM name_vectors v USING fact_evidence fe
+          WHERE fe.fact_id = v.fact_id AND fe.document_id = $1";
+
 /// 来源刚开了身份名（0071 决定 6）：删掉有它文档证据的名字向量，与改配置同一个事务
 pub async fn drop_for_source(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     source_id: Uuid,
 ) -> AppResult<()> {
-    sqlx::query(
-        "DELETE FROM name_vectors v
-          USING fact_evidence fe JOIN documents d ON d.id = fe.document_id
-          WHERE fe.fact_id = v.fact_id AND d.source_id = $1",
-    )
-    .bind(source_id)
-    .execute(&mut **tx)
-    .await?;
+    sqlx::query(DROP_FOR_SOURCE_SQL)
+        .bind(source_id)
+        .execute(&mut **tx)
+        .await?;
     Ok(())
 }
 
 /// 身份来源的文档抽完（0071 决定 5）：散文先写下的名字向量，这篇给了它身份证据就删
 pub async fn drop_for_document(pool: &PgPool, document_id: Uuid) -> AppResult<()> {
-    sqlx::query(
-        "DELETE FROM name_vectors v USING fact_evidence fe
-          WHERE fe.fact_id = v.fact_id AND fe.document_id = $1",
-    )
-    .bind(document_id)
-    .execute(pool)
-    .await?;
+    sqlx::query(DROP_FOR_DOCUMENT_SQL)
+        .bind(document_id)
+        .execute(pool)
+        .await?;
     Ok(())
 }
 

crates/utopia-store/src/name_vectors_plan_tests.rs

@@ -4,7 +4,7 @@
 //! 规划器按统计选索引：库是空的时候几个索引代价一样，选哪个看运气（别的测试同时在写 facts，
 //! 选择就会翻）。所以这里在回滚的事务里造一个两千条事实、十分之一是名字的库并 ANALYZE，
 //! 比的是有数据时它走哪条路。
-use super::PENDING_SQL;
+use super::{DROP_FOR_DOCUMENT_SQL, DROP_FOR_SOURCE_SQL, PENDING_SQL};
 use sqlx::PgPool;
 use uuid::Uuid;
 
@@ -68,3 +68,80 @@ async fn pending_reads_the_name_facts_through_the_predicate_index() -> anyhow::R
     assert!(!plan.contains("Sort"), "{plan}");
     Ok(())
 }
+
+/// 0071 的清向量每篇身份文档都走一次，不能读完别的文档的证据才挑中这一篇（#1408）。
+/// 临时表复制迁移后的列和索引，统计只来自本例，避免别的测试的写入改变查询计划。
+/// 这里约束选择性高的小来源；一次覆盖很多文档时，顺扫可能更便宜，应由规划器选择。
+#[tokio::test]
+async fn dropping_vectors_reads_only_the_selected_documents_evidence() -> anyhow::Result<()> {
+    let Some(url) = crate::test_db::url() else {
+        return Ok(());
+    };
+    let pool = PgPool::connect(&url).await?;
+    let mut tx = pool.begin().await?;
+    sqlx::raw_sql(
+        "CREATE TEMP TABLE documents (LIKE public.documents INCLUDING ALL) ON COMMIT DROP;
+         CREATE TEMP TABLE fact_evidence (LIKE public.fact_evidence INCLUDING ALL) ON COMMIT DROP;
+         CREATE TEMP TABLE name_vectors (LIKE public.name_vectors INCLUDING ALL) ON COMMIT DROP;
+         INSERT INTO documents (id, kb_id, source_id, filename, sha256, external_key)
+         SELECT gen_random_uuid(), md5('kb')::uuid,
+                md5(CASE WHEN n <= 2 THEN 'selected' ELSE 'other' END)::uuid,
+                'document-' || n, md5(n::text), n::text
+           FROM generate_series(1, 1000) n;
+         INSERT INTO fact_evidence (fact_id, chunk_id, document_id)
+         SELECT gen_random_uuid(), gen_random_uuid(), d.id
+           FROM documents d CROSS JOIN generate_series(1, 20) n;
+         INSERT INTO name_vectors (fact_id, kb_id, entity_id, embedding)
+         SELECT fact_id, md5('kb')::uuid, md5('entity')::uuid, '[1,0,0]'::vector
+           FROM fact_evidence;
+         ANALYZE pg_temp.documents;
+         ANALYZE pg_temp.fact_evidence;
+         ANALYZE pg_temp.name_vectors;",
+    )
+    .execute(&mut *tx)
+    .await?;
+    let (document, source): (Uuid, Uuid) =
+        sqlx::query_as("SELECT id, source_id FROM documents WHERE external_key = '1'")
+            .fetch_one(&mut *tx)
+            .await?;
+
+    for (query, id, expected_rows) in [
+        (DROP_FOR_DOCUMENT_SQL, document, 20),
+        (DROP_FOR_SOURCE_SQL, source, 40),
+    ] {
+        // 真正执行删除，但每次都回到相同的向量集合。比访问行数，不把机器快慢变成断言。
+        sqlx::query("SAVEPOINT plan_case").execute(&mut *tx).await?;
+        let plan: serde_json::Value =
+            sqlx::query_scalar(&format!("EXPLAIN (ANALYZE, FORMAT JSON) {query}"))
+                .bind(id)
+                .fetch_one(&mut *tx)
+                .await?;
+        sqlx::query("ROLLBACK TO SAVEPOINT plan_case")
+            .execute(&mut *tx)
+            .await?;
+        let visited = evidence_rows_visited(&plan[0]["Plan"]);
+        assert!(visited > 0 && visited <= expected_rows, "{plan}");
+    }
+    tx.rollback().await?;
+    Ok(())
+}
+
+fn evidence_rows_visited(node: &serde_json::Value) -> u64 {
+    let own_rows = if node["Relation Name"] == "fact_evidence"
+        && node["Node Type"]
+            .as_str()
+            .is_some_and(|kind| kind.ends_with("Scan"))
+    {
+        (node["Actual Rows"].as_u64().unwrap_or(0)
+            + node["Rows Removed by Filter"].as_u64().unwrap_or(0)
+            + node["Rows Removed by Index Recheck"].as_u64().unwrap_or(0))
+            * node["Actual Loops"].as_u64().unwrap_or(0)
+    } else {
+        0
+    };
+    own_rows
+        + node["Plans"]
+            .as_array()
+            .map(|children| children.iter().map(evidence_rows_visited).sum::<u64>())
+            .unwrap_or(0)
+}

migrations/0122_fact_evidence_is_found_by_document.sql

@@ -0,0 +1,6 @@
+-- 身份文档每次抽取都按 document_id 清理名字向量，来源开启身份名时也按文档找证据。
+-- 原主键 (fact_id, chunk_id) 服务的是从事实溯源，不能定位一篇文档的证据；没有这个
+-- 反向入口，一批逐篇推送会反复扫描整张 fact_evidence（#1408）。
+-- 有下游分支早已建过同名同形的索引，所以 IF NOT EXISTS，那里是空操作；
+-- 不用 CONCURRENTLY：理由同 0094；而且并发建失败会留下无效索引，下次 IF NOT EXISTS 会把它跳过。
+CREATE INDEX IF NOT EXISTS fact_evidence_document_idx ON fact_evidence (document_id);