diff --git a/crates/utopia-cli/src/main.rs b/crates/utopia-cli/src/main.rs index 99316eee4..f12de8858 100644 --- a/crates/utopia-cli/src/main.rs +++ b/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<()> { diff --git a/crates/utopia-store/src/name_vectors.rs b/crates/utopia-store/src/name_vectors.rs index 14700c80a..4980a8c19 100644 --- a/crates/utopia-store/src/name_vectors.rs +++ b/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(()) } diff --git a/crates/utopia-store/src/name_vectors_plan_tests.rs b/crates/utopia-store/src/name_vectors_plan_tests.rs index 651522f77..37983dc30 100644 --- a/crates/utopia-store/src/name_vectors_plan_tests.rs +++ b/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::()) + .unwrap_or(0) +} diff --git a/migrations/0122_fact_evidence_is_found_by_document.sql b/migrations/0122_fact_evidence_is_found_by_document.sql new file mode 100644 index 000000000..8f4c28dd2 --- /dev/null +++ b/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);