Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion crates/utopia-cli/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 = 106;
const CURRENT_SCHEMA_VERSION: u32 = 107;

fn main() -> anyhow::Result<()> {
dotenvy::dotenv().ok();
Expand Down
2 changes: 2 additions & 0 deletions crates/utopia-core/src/models.rs
Original file line number Diff line number Diff line change
Expand Up @@ -662,6 +662,8 @@ pub struct StatementQualifier {
pub value: Option<serde_json::Value>,
pub entity_id: Option<Uuid>,
pub entity_name: Option<String>,
/// 说这个值的那几块(0070):两块活着的块说法不同,各自一条
pub chunk_ids: Vec<Uuid>,
}

/// 一次 OWL 导入的记录。原文按内容寻址存在 blob 里,这行只是账。
Expand Down
8 changes: 5 additions & 3 deletions crates/utopia-server/src/api/mcp_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1008,10 +1008,11 @@ async fn entity_facts_reads_back_a_statements_role_words() -> anyhow::Result<()>
return Ok(());
};
sqlx::query(
"INSERT INTO statement_qualifiers(fact_id,role,value)
VALUES ($1,'lines','{\"value\":\"10-42\"}'::jsonb)",
"INSERT INTO statement_qualifiers(fact_id,role,value,chunk_id)
VALUES ($1,'lines','{\"value\":\"10-42\"}'::jsonb,$2)",
)
.bind(f.corrected)
.bind(f.chunk)
.execute(&f.state.pool)
.await?;
let result = f
Expand All @@ -1027,7 +1028,8 @@ async fn entity_facts_reads_back_a_statements_role_words() -> anyhow::Result<()>
.unwrap();
assert_eq!(
fact["statement_qualifiers"],
json!([{"role":"lines","value":{"value":"10-42"},"entity_id":null,"entity_name":null}])
json!([{"role":"lines","value":{"value":"10-42"},"entity_id":null,"entity_name":null,
"chunk_ids":[f.chunk]}])
);
f.clean().await
}
Expand Down
36 changes: 36 additions & 0 deletions crates/utopia-server/src/api/sources_statements_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -437,6 +437,42 @@ async fn a_second_push_under_the_same_identity_updates_in_place() -> anyhow::Res
f.cleanup().await
}

/// 0070 的例子:再推一次把 `lines` 从 40-52 改成 58-70,读回来只有新的范围
#[tokio::test]
async fn a_re_push_that_moves_lines_to_58_70_reads_back_only_58_70() -> anyhow::Result<()> {
let Some(f) = Fixture::new().await? else {
return Ok(());
};
for lines in ["40-52", "58-70"] {
let mut body = observation("08:14:03", "kitchen table");
body["s"][0][5] = json!({ "lines": lines });
let (status, body) = f.push(f.source, &f.token, &body).await?;
assert_eq!(status, StatusCode::OK, "{body}");
let doc = documents::find_by_external_key(&f.pool, f.source, "statements:obs-000412")
.await?
.expect("the push has a document");
crate::pipeline::process_document(&f.state, doc.id).await?;
let proposer = Proposer {
user_id: None,
token_id: None,
};
crate::extraction::extract_document(&f.state, doc.id, proposer).await?;
}
let (cup,): (Uuid,) =
sqlx::query_as("SELECT id FROM entities WHERE kb_id = $1 AND canonical_name = 'cup-7'")
.bind(f.kb)
.fetch_one(&f.pool)
.await?;
let (_, facts) = utopia_store::graph::entity_detail(&f.pool, f.kb, cup, None, None).await?;
let said: Vec<_> = facts
.iter()
.flat_map(|x| &x.statement_qualifiers)
.map(|q| (q.role.as_str(), q.value.clone()))
.collect();
assert_eq!(said, vec![("lines", Some(json!("58-70")))]);
f.cleanup().await
}

/// 门口的拒绝走到 HTTP 是 422(`AppError::Validation`,与 `api` 推送被拒时同一个码),
/// 并且这次推送留在 run 历史里;`api` 来源不认这条路由(404),钥匙不对是 401
#[tokio::test]
Expand Down
1 change: 1 addition & 0 deletions crates/utopia-server/src/extraction_open.rs
Original file line number Diff line number Diff line change
Expand Up @@ -665,6 +665,7 @@ async fn apply_open_extraction(
utopia_store::graph::add_statement_qualifier(
pool,
fact_id,
chunk.id,
role,
value.as_ref(),
*entity,
Expand Down
8 changes: 8 additions & 0 deletions crates/utopia-store/src/export_provenance_integrity.sql
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,14 @@ SELECT edge, kind, COUNT(*) AS rows FROM (
LEFT JOIN entities e ON e.id = q.entity_id
WHERE f.kb_id = $1 AND q.entity_id IS NOT NULL
UNION ALL
-- @edge statement_qualifiers.chunk_id -> chunks.id
-- 角色词记在哪一块上:归属同样按所属 fact 的库判
SELECT 'squalifier.chunk', 'cross_kb', c.kb_id IS DISTINCT FROM f.kb_id
FROM statement_qualifiers q
JOIN facts f ON f.id = q.fact_id
LEFT JOIN chunks c ON c.id = q.chunk_id
WHERE f.kb_id = $1
UNION ALL
-- @edge time_mentions.fact_id -> facts.id
-- 时间提及以行自己的 kb 归属:指的事实与段落都必须同库
SELECT 'timemention.fact', 'cross_kb', f.kb_id IS DISTINCT FROM t.kb_id
Expand Down
43 changes: 26 additions & 17 deletions crates/utopia-store/src/graph.rs
Original file line number Diff line number Diff line change
Expand Up @@ -199,12 +199,12 @@ pub async fn fact_qualifiers_for(
Ok(out)
}

/// 往一条开放陈述上写一个按角色词记的属性。**先写者留着**:同一条陈述同一个角色词
/// 再听到一次不覆盖(与 [`upsert_fact_qualifier`] 的 Conflict 一条规矩——两次观察
/// 不一致该另立一行加一条冲突,这一刀还没走到那一步)。值与实体必须恰好给一个
/// 往一条开放陈述上写一个按角色词记的属性,记在说它的那一块上(0070 决定 1)。
/// 同一块再说一次不写新行;另一块说的另记一行,旧行不动。值与实体必须恰好给一个
pub async fn add_statement_qualifier(
pool: &PgPool,
fact_id: Uuid,
chunk_id: Uuid,
role: &str,
value: Option<&serde_json::Value>,
entity_id: Option<Uuid>,
Expand All @@ -223,22 +223,25 @@ pub async fn add_statement_qualifier(
));
}
sqlx::query(
"INSERT INTO statement_qualifiers (fact_id, role, value, entity_id)
VALUES ($1, $2, $3, $4) ON CONFLICT (fact_id, role) DO NOTHING",
"INSERT INTO statement_qualifiers (fact_id, role, value, entity_id, chunk_id)
VALUES ($1, $2, $3, $4, $5) ON CONFLICT (fact_id, role, chunk_id) DO NOTHING",
)
.bind(fact_id)
.bind(role)
.bind(value)
.bind(entity_id)
.bind(chunk_id)
.execute(pool)
.await?;
Ok(())
}

/// 一批开放陈述各自的角色词属性,按事实 id 取回;实体那一格连上显示名
/// 一批开放陈述各自的角色词属性,按事实 id 取回;实体那一格连上显示名。
/// 只取 `as_of` 时刻说话的行(0070 决定 2、3);同一个值记在几块上返回一次,带上那几块
pub async fn statement_qualifiers_for(
pool: &PgPool,
fact_ids: &[Uuid],
as_of: Option<chrono::DateTime<chrono::Utc>>,
) -> AppResult<HashMap<Uuid, Vec<StatementQualifier>>> {
let mut out: HashMap<Uuid, Vec<StatementQualifier>> = HashMap::new();
if fact_ids.is_empty() {
Expand All @@ -251,23 +254,29 @@ pub async fn statement_qualifiers_for(
value: Option<serde_json::Value>,
entity_id: Option<Uuid>,
entity_name: Option<String>,
chunk_ids: Vec<Uuid>,
}
let rows: Vec<Row> = sqlx::query_as(
"SELECT q.fact_id, q.role, q.value, q.entity_id, e.canonical_name AS entity_name
let sql = format!(
"SELECT q.fact_id, q.role, q.value, q.entity_id, e.canonical_name AS entity_name,
array_agg(q.chunk_id ORDER BY q.chunk_id) AS chunk_ids
FROM statement_qualifiers q
LEFT JOIN entities e ON e.id = q.entity_id
WHERE q.fact_id = ANY($1)
ORDER BY q.fact_id, q.role",
)
.bind(fact_ids)
.fetch_all(pool)
.await?;
for r in rows {
WHERE q.fact_id = ANY($1) AND {speaks}
GROUP BY q.fact_id, q.role, q.value, q.entity_id, e.canonical_name
ORDER BY q.fact_id, q.role, chunk_ids",
speaks = crate::record_axis::role_word_speaks_at("q", as_of.map(|_| 2)),
);
let mut query = sqlx::query_as::<_, Row>(&sql).bind(fact_ids);
if let Some(at) = as_of {
query = query.bind(at);
}
for r in query.fetch_all(pool).await? {
out.entry(r.fact_id).or_default().push(StatementQualifier {
role: r.role,
value: r.value,
entity_id: r.entity_id,
entity_name: r.entity_name,
chunk_ids: r.chunk_ids,
});
}
Ok(out)
Expand Down Expand Up @@ -1517,7 +1526,7 @@ async fn edges_among(
let ids: Vec<Uuid> = edges.iter().map(|x| x.id).collect();
let (mut typed, mut worded) = tokio::try_join!(
fact_qualifiers_for(pool, &ids),
statement_qualifiers_for(pool, &ids)
statement_qualifiers_for(pool, &ids, as_of)
)?;
for x in edges.iter_mut() {
x.qualifiers = typed.remove(&x.id).unwrap_or_default();
Expand Down Expand Up @@ -1866,7 +1875,7 @@ pub async fn entity_detail(
let ids: Vec<Uuid> = facts.iter().map(|x| x.id).collect();
let (mut typed, mut worded) = tokio::try_join!(
fact_qualifiers_for(pool, &ids),
statement_qualifiers_for(pool, &ids)
statement_qualifiers_for(pool, &ids, as_of)
)?;
for x in facts.iter_mut() {
x.qualifiers = typed.remove(&x.id).unwrap_or_default();
Expand Down
4 changes: 2 additions & 2 deletions crates/utopia-store/src/materialize.rs
Original file line number Diff line number Diff line change
Expand Up @@ -424,8 +424,8 @@ async fn materialize_in_tx(
.execute(&mut **tx)
.await?;
sqlx::query(
"INSERT INTO statement_qualifiers (fact_id, role, value, entity_id)
SELECT $1, role, value, entity_id FROM statement_qualifiers WHERE fact_id = $2
"INSERT INTO statement_qualifiers (fact_id, role, value, entity_id, chunk_id)
SELECT $1, role, value, entity_id, chunk_id FROM statement_qualifiers WHERE fact_id = $2
ON CONFLICT DO NOTHING",
)
.bind(fact)
Expand Down
3 changes: 2 additions & 1 deletion crates/utopia-store/src/pending.rs
Original file line number Diff line number Diff line change
Expand Up @@ -499,7 +499,8 @@ async fn confirm_open(
} else {
q.get("value")
};
crate::graph::add_statement_qualifier(pool, fact_id, role, value, entity_id).await?;
crate::graph::add_statement_qualifier(pool, fact_id, v.chunk_id, role, value, entity_id)
.await?;
}
for t in json_items(v.time_words.as_ref()) {
let (Some(text), Some(char_start)) = (
Expand Down
40 changes: 40 additions & 0 deletions crates/utopia-store/src/record_axis.rs
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,46 @@ pub fn chunk_live_at(alias: &str, as_of: Option<usize>) -> String {
}
}

/// `statement_qualifiers`:这一行角色词在 T 时刻替它的事实说话吗(0070 决定 2、3)。
///
/// 它那块与那块的文档在 T 时刻都还活着,就说话;活着的几块值不同,都说话。
/// 事实在 T 时刻一块活证据都没有时,最后被取代的那几块说话(同一时刻取代的都算):
/// 块的终点是 `superseded_at` 与文档 `deleted_at` 里早的那个,只看 T 时刻已有的块。
/// 别名都带 `rw_` 前缀,不和调用方的撞
pub fn role_word_speaks_at(q: &str, as_of: Option<usize>) -> String {
let live = |c: &str, d: &str| {
format!(
"{} AND {}",
chunk_live_at(c, as_of),
document_live_at(d, as_of)
)
};
let existed = |c: &str| match as_of {
None => String::new(),
Some(param) => format!(" AND {c}.created_at <= coalesce(${param}, now())"),
};
format!(
"EXISTS (SELECT 1 FROM chunks rw_c JOIN documents rw_d ON rw_d.id = rw_c.document_id
WHERE rw_c.id = {q}.chunk_id
AND (({live_own})
OR (NOT EXISTS (SELECT 1 FROM fact_evidence rw_fe
JOIN chunks rw_lc ON rw_lc.id = rw_fe.chunk_id
JOIN documents rw_ld ON rw_ld.id = rw_lc.document_id
WHERE rw_fe.fact_id = {q}.fact_id AND {live_any})
{existed_own}
AND least(rw_c.superseded_at, rw_d.deleted_at) =
(SELECT max(least(rw_ec.superseded_at, rw_ed.deleted_at))
FROM fact_evidence rw_ee
JOIN chunks rw_ec ON rw_ec.id = rw_ee.chunk_id
JOIN documents rw_ed ON rw_ed.id = rw_ec.document_id
WHERE rw_ee.fact_id = {q}.fact_id{existed_any}))))",
live_own = live("rw_c", "rw_d"),
live_any = live("rw_lc", "rw_ld"),
existed_own = existed("rw_c"),
existed_any = existed("rw_ec"),
)
}

/// `entity_merges`:这次合并在 T 时刻**生效着**吗(0019 第二刀 / #336)。
///
/// 实体身上没有记录轴——`merged_into` 只说合并发生过,不说何时。时刻在这张表上,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -133,21 +133,26 @@ async fn a_bound_statement_becomes_a_typed_fact() -> anyhow::Result<()> {
.bind(doc)
.execute(&pool)
.await?;
anyhow::Ok(())
anyhow::Ok(chunk)
}
};
evidence_of(stated, "bakery.txt").await?;
let stated_chunk = evidence_of(stated, "bakery.txt").await?;
evidence_of(again, "bakery-2.txt").await?;
let planned_chunk = evidence_of(planned, "bakery-plan.txt").await?;
sqlx::query(
"INSERT INTO statement_qualifiers (fact_id, role, value) VALUES ($1, 'since', '\"2019\"'::jsonb)",
"INSERT INTO statement_qualifiers (fact_id, role, value, chunk_id)
VALUES ($1, 'since', '\"2019\"'::jsonb, $2)",
)
.bind(stated)
.bind(stated_chunk)
.execute(&pool)
.await?;
sqlx::query(
"INSERT INTO statement_qualifiers (fact_id, role, value) VALUES ($1, 'mood', '\"will\"'::jsonb)",
"INSERT INTO statement_qualifiers (fact_id, role, value, chunk_id)
VALUES ($1, 'mood', '\"will\"'::jsonb, $2)",
)
.bind(planned)
.bind(planned_chunk)
.execute(&pool)
.await?;

Expand Down
Loading
Loading