Finding a base's name facts without vectors reads only its name facts (#1318, ahead of upstream)

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 = 103;
+const CURRENT_SCHEMA_VERSION: u32 = 104;
 
 fn main() -> anyhow::Result<()> {
     dotenvy::dotenv().ok();

crates/utopia-store/src/name_vectors.rs

@@ -25,6 +25,21 @@ pub const TOP_K: usize = 8;
 /// 0.4 上下的余弦——所以这条线比画像的 `SIM_ATTACH` 高。同样是待测量的临时值
 pub const SIM_FLOOR: f32 = 0.60;
 
+/// `known_as` 的 id 用子查询取，不要改回 JOIN：子查询先算出一个值，`predicate_id = 它` 才成了
+/// `facts_live_predicate_idx`（0115）的索引条件，扫描只碰这个库的名字事实，且仍按 id 走，
+/// `ORDER BY ... LIMIT` 不用排序。写成 JOIN，规划器沿 `facts_kb_id_key` 把整个库按 id 读一遍，
+/// 谓词只能逐行筛（#1296：20 万条事实的库上读 20 万行，这样读 2 万行）
+const PENDING_SQL: &str = "SELECT f.id, f.subject_id, f.object_value->>'value'
+       FROM facts f
+       LEFT JOIN name_vectors v ON v.fact_id = f.id
+      WHERE f.kb_id = $1
+        AND f.predicate_id = (SELECT nr.id FROM relation_types nr
+                               WHERE nr.kb_id = $1 AND nr.builtin AND nr.key = $2)
+        AND f.invalidated_at IS NULL AND f.object_value IS NOT NULL
+        AND v.fact_id IS NULL
+      ORDER BY f.id
+      LIMIT $3";
+
 /// 还没有向量的名字事实：现行的、`known_as` 上的、`name_vectors` 里没有它的。
 /// 按写入先后取（`facts.id` 是 uuid v7，按它排即按写入排；这张表没有 created_at——
 /// 端到端跑出来的：按一个不存在的列排，补向量每篇都静默失败），一次取一批
@@ -33,22 +48,12 @@ pub async fn pending(
     kb_id: Uuid,
     limit: i64,
 ) -> AppResult<Vec<(Uuid, Uuid, String)>> {
-    let rows: Vec<(Uuid, Uuid, String)> = sqlx::query_as(
-        "SELECT f.id, f.subject_id, f.object_value->>'value'
-           FROM facts f
-           JOIN relation_types nr ON nr.id = f.predicate_id
-           LEFT JOIN name_vectors v ON v.fact_id = f.id
-          WHERE f.kb_id = $1 AND nr.kb_id = $1 AND nr.builtin AND nr.key = $2
-            AND f.invalidated_at IS NULL AND f.object_value IS NOT NULL
-            AND v.fact_id IS NULL
-          ORDER BY f.id
-          LIMIT $3",
-    )
-    .bind(kb_id)
-    .bind(crate::names::KNOWN_AS)
-    .bind(limit)
-    .fetch_all(pool)
-    .await?;
+    let rows: Vec<(Uuid, Uuid, String)> = sqlx::query_as(PENDING_SQL)
+        .bind(kb_id)
+        .bind(crate::names::KNOWN_AS)
+        .bind(limit)
+        .fetch_all(pool)
+        .await?;
     Ok(rows)
 }
 
@@ -135,3 +140,7 @@ pub async fn nearest(pool: &PgPool, kb_id: Uuid, query: &[f32], k: usize) -> App
         .await?;
     Ok(rows)
 }
+
+#[cfg(test)]
+#[path = "name_vectors_plan_tests.rs"]
+mod plan_tests;

crates/utopia-store/src/name_vectors_plan_tests.rs

@@ -0,0 +1,70 @@
+//! 补向量每篇文档都跑一次 `pending`（#1296）：它得以谓词为索引条件走 `facts_live_predicate_idx`，
+//! 只读这个库的名字事实、按 id 顺序读，不用排序。改回 JOIN 时规划器会沿 `facts_kb_id_key` 读整个库。
+//!
+//! 规划器按统计选索引：库是空的时候几个索引代价一样，选哪个看运气（别的测试同时在写 facts，
+//! 选择就会翻）。所以这里在回滚的事务里造一个两千条事实、十分之一是名字的库并 ANALYZE，
+//! 比的是有数据时它走哪条路。
+use super::PENDING_SQL;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+#[tokio::test]
+async fn pending_reads_the_name_facts_through_the_predicate_index() -> anyhow::Result<()> {
+    let Some(url) = crate::test_db::url() else {
+        return Ok(());
+    };
+    let pool = PgPool::connect(&url).await?;
+    let (org, ws, kb) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7());
+    let (named, other, thing) = (Uuid::now_v7(), Uuid::now_v7(), Uuid::now_v7());
+    let mut tx = pool.begin().await?;
+    for (sql, ids) in [
+        ("SET LOCAL enable_seqscan = off", vec![]),
+        (
+            "INSERT INTO organizations (id, name) VALUES ($1, 'plan')",
+            vec![org],
+        ),
+        (
+            "INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'plan')",
+            vec![ws, org],
+        ),
+        (
+            "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'plan')",
+            vec![kb, ws],
+        ),
+        (
+            "INSERT INTO relation_types (id, kb_id, key, label, builtin)
+             VALUES ($1, $3, 'known_as', 'known as', true), ($2, $3, 'mentions', 'mentions', false)",
+            vec![named, other, kb],
+        ),
+        (
+            "INSERT INTO entities (id, kb_id, canonical_name) VALUES ($1, $2, 'thing')",
+            vec![thing, kb],
+        ),
+        (
+            "INSERT INTO facts (id, kb_id, subject_id, predicate_id, object_value)
+             SELECT gen_random_uuid(), $1, $2, CASE WHEN n % 10 = 0 THEN $3 ELSE $4 END,
+                    jsonb_build_object('value', 'v' || n)
+               FROM generate_series(1, 2000) n",
+            vec![kb, thing, named, other],
+        ),
+        ("ANALYZE facts", vec![]),
+    ] {
+        let mut q = sqlx::query(sql);
+        for id in ids {
+            q = q.bind(id);
+        }
+        q.execute(&mut *tx).await?;
+    }
+    let plan: Vec<String> = sqlx::query_scalar(&format!("EXPLAIN {PENDING_SQL}"))
+        .bind(kb)
+        .bind(crate::names::KNOWN_AS)
+        .bind(64_i64)
+        .fetch_all(&mut *tx)
+        .await?;
+    tx.rollback().await?;
+
+    let plan = plan.join("\n");
+    assert!(plan.contains("facts_live_predicate_idx"), "{plan}");
+    assert!(!plan.contains("Sort"), "{plan}");
+    Ok(())
+}

crates/utopia-store/tests/store/a_similar_name_is_proposed_not_merged.rs

@@ -341,6 +341,42 @@ async fn a_new_name_is_pending_until_its_vector_is_set() -> anyhow::Result<()> {
     run
 }
 
+/// 上一篇文档没补上的名字，下一篇照样补：`pending` 取整个库里没有向量的名字，先记的在前，
+/// 不只是这篇刚记的。有向量的名字不在里面（#1296 改这条查询时守住这次重试）
+#[tokio::test]
+async fn a_name_left_without_a_vector_earlier_is_pending_before_a_new_one() -> anyhow::Result<()> {
+    let Some(url) = utopia_store::test_db::url() else {
+        return Ok(());
+    };
+    let pool = PgPool::connect(&url).await?;
+    let f = seed(&pool, "name-vector-backlog").await?;
+    let run = async {
+        let mut recorded = Vec::new();
+        for name in ["远洋重工", "深蓝船舶"] {
+            let entity = Uuid::now_v7();
+            sqlx::query(
+                "INSERT INTO entities (id, kb_id, type_id, canonical_name) VALUES ($1, $2, $3, $4)",
+            )
+            .bind(entity)
+            .bind(f.kb)
+            .bind(f.device)
+            .bind(name)
+            .execute(&pool)
+            .await?;
+            let fact = names::record(&pool, f.kb, entity, name, None, None)
+                .await?
+                .expect("a name fact");
+            recorded.push((fact, entity, name.to_string()));
+        }
+
+        assert_eq!(name_vectors::pending(&pool, f.kb, 100).await?, recorded);
+        Ok::<_, anyhow::Error>(())
+    }
+    .await;
+    teardown(&pool, &f).await?;
+    run
+}
+
 /// 判过「不是一个」的一对，实体再被提到时不再排（#1104）。名字向量通道每次提到都跑，
 /// 而审核表只对 pending 的行去重：从前裁决器分开过的一对，下一篇文档一提到就又是一条
 /// 新的 pending。还没判的那一对不受影响，照常提议

migrations/9001_a_name_fact_is_found_by_its_predicate.sql

@@ -0,0 +1,13 @@
+-- re:cinq fork only until deeplethe/utopia#1318 merges (RE-CINQ.md: fork migrations are numbered from 9000 and
+-- idempotent). Upstream's copy is 0115_a_name_fact_is_found_by_its_predicate.sql; whichever runs second is a no-op.
+--
+-- 按谓词找现行的名字事实（#1296）。
+--
+-- 补向量每篇文档都跑一次 `name_vectors::pending`：现行的、`known_as` 上的、还没有向量的名字
+-- 事实，按 id 取一批。从前没有带谓词的索引，它只能沿 `facts_kb_id_key` 把整个库的事实按 id
+-- 读一遍再筛谓词——20 万条事实的库里每篇文档读 20 万行，为的是剩下的三五个名字。
+--
+-- `id` 列让索引按谓词定位之后仍按 id 排好，`ORDER BY f.id LIMIT` 顺着走、不用排序，只读这个库
+-- 的名字事实。谓词在查询里是一个子查询的值，见 `pending` 上的说明。
+CREATE INDEX IF NOT EXISTS facts_live_predicate_idx ON facts (kb_id, predicate_id, id)
+  WHERE invalidated_at IS NULL;