From f7f24ac8e0a8e227fbd3eb1f29baa6b0059eccfa Mon Sep 17 00:00:00 2001 From: WaylandYang Date: Wed, 9 Sep 2026 21:49:40 +0800 Subject: [PATCH] A definition can be written by hand Co-Authored-By: Claude Fable 5.1 Signed-off-by: WaylandYang --- crates/utopia-core/src/models.rs | 3 + .../utopia-server/src/api/mapping_routes.rs | 246 ++++++++++++++++++ crates/utopia-server/src/api/mod.rs | 7 +- crates/utopia-server/src/api/tools.rs | 2 +- crates/utopia-server/src/mappings.rs | 2 +- crates/utopia-store/src/mappings.rs | 58 ++++- .../a_definition_can_be_written_by_hand.rs | 134 ++++++++++ ...46_a_definition_can_be_written_by_hand.sql | 10 + 8 files changed, 456 insertions(+), 6 deletions(-) create mode 100644 crates/utopia-store/tests/a_definition_can_be_written_by_hand.rs create mode 100644 migrations/0046_a_definition_can_be_written_by_hand.sql diff --git a/crates/utopia-core/src/models.rs b/crates/utopia-core/src/models.rs index e7400ec4a..c1c4274b7 100644 --- a/crates/utopia-core/src/models.rs +++ b/crates/utopia-core/src/models.rs @@ -1030,6 +1030,9 @@ pub struct ConceptMapping { /// **状态而不是置信度。** 从前借事实的 confidence 表达「提议 0.6 / 确认 1.0」， /// 那是把二值状态编码成浮点数，还顺带让它落进「低置信事实」那一档 pub status: String, + /// 人从零写的口径记写的人；探索提的为空（#562）。`decided_by` 分不出这件事—— + /// 探索提的经人确认之后同样有 decided_by + pub written_by: Option, } /// 一处公理违规，配好展示所需的三元组文本（见 `axiom_violations`）。 diff --git a/crates/utopia-server/src/api/mapping_routes.rs b/crates/utopia-server/src/api/mapping_routes.rs index af755b6b8..5ee6d3281 100644 --- a/crates/utopia-server/src/api/mapping_routes.rs +++ b/crates/utopia-server/src/api/mapping_routes.rs @@ -146,3 +146,249 @@ pub async fn revisions( let rows = utopia_store::mappings::revisions(&state.pool, kb_id, mapping_id).await?; Ok(Json(json!({ "revisions": rows }))) } + +/// 空白等于没填：前端清空一个输入框传来的是 ""，落库该是 NULL 而不是空串。 +fn clean(s: &Option) -> Option { + s.as_deref() + .map(str::trim) + .filter(|x| !x.is_empty()) + .map(str::to_string) +} + +/// 一条口径至少得有表、表达式、SQL 之一，否则它不是一个可执行的定义。 +/// `create`、`preview` 与 `revise` 判的是同一条。 +fn definition_shape( + table_name: &Option, + expr: &Option, + sql: &Option, +) -> Result<(), AppError> { + if table_name.is_none() && expr.is_none() && sql.is_none() { + return Err(AppError::invalid( + "empty_mapping", + "A mapping needs at least one of table, expression or SQL", + )); + } + Ok(()) +} + +/// 一条口径跑起来是哪句 SQL：给了 `sql` 用 `sql`，否则 `expr` + `table` 拼一句。 +/// 测量台（`scripts/bench/mappings.mjs` 的 `proposalSql`）判分用的是同一个拼法， +/// 所以页面上预览到的数就是测量台会打分的数。 +fn render_sql( + table_name: &Option, + expr: &Option, + sql: &Option, +) -> Option { + if let Some(s) = sql { + return Some(s.trim_end_matches(';').trim().to_string()); + } + match (expr, table_name) { + (Some(e), Some(t)) => Some(format!("SELECT {e} FROM {t}")), + _ => None, + } +} + +#[derive(Deserialize)] +pub struct CreateReq { + concept: String, + /// metric | dimension；缺省 metric + #[serde(default = "default_kind")] + kind: String, + source: String, + table_name: Option, + expr: Option, + sql: Option, + unit: Option, + summary: Option, + #[serde(default)] + derived: bool, +} +fn default_kind() -> String { + "metric".into() +} + +/// 人从零写一条口径（#562）。 +/// +/// 从前这张表只有探索一条来路。一个数据团队手上有自己的指标口径文档，却没有地方 +/// 把它填进去——而口径进了问数的提示词，宽表语料从 1/18 到 17/18（#520）。缺的 +/// 不是结构，是这扇门。 +/// +/// 概念仍落成 Metric / Dimension 类的实体——那是这张表的现状（`concept_id` 非空 +/// 指向实体）。0035 要退役这个形状（#556），但保留这张表并往里渲染，人写的 SQL +/// 搬得过去。 +pub async fn create( + State(state): State, + AuthUser(user): AuthUser, + Path(kb_id): Path, + Json(req): Json, +) -> ApiResult> { + require_kb(&state, &user, kb_id, Role::Editor).await?; + let concept = req.concept.trim(); + if concept.is_empty() { + return Err(AppError::invalid("empty_concept", "A definition needs a concept name").into()); + } + if !matches!(req.kind.as_str(), "metric" | "dimension") { + return Err(AppError::invalid("bad_kind", "kind must be metric or dimension").into()); + } + let (table_name, expr, sql, unit, summary) = ( + clean(&req.table_name), + clean(&req.expr), + clean(&req.sql), + clean(&req.unit), + clean(&req.summary), + ); + definition_shape(&table_name, &expr, &sql)?; + // 源按名字点单，且必须是本库挂载的——与 `query_data` 同一条安全边界 + let mounted = utopia_store::datasources::mounted(&state.pool, kb_id).await?; + let Some(ds) = mounted + .iter() + .find(|d| d.name.eq_ignore_ascii_case(req.source.trim())) + else { + return Err(AppError::invalid( + "source_not_mounted", + "That data source is not mounted on this knowledge base", + ) + .into()); + }; + + crate::mappings::ensure_concept_types(&state.pool, kb_id) + .await + .map_err(AppError::Other)?; + let (type_id,): (Uuid,) = + sqlx::query_as("SELECT id FROM entity_types WHERE kb_id = $1 AND key = $2") + .bind(kb_id) + .bind(&req.kind) + .fetch_one(&state.pool) + .await?; + // 概念实体走消解：同名归并，跟探索提的那些落在一处 + let resolved = utopia_store::resolution::resolve_mention( + &state.pool, + kb_id, + Some(type_id), + concept, + None, + None, + &[], + ) + .await?; + let id = utopia_store::mappings::create( + &state.pool, + kb_id, + resolved.entity_id, + &ds.name, + table_name.as_deref(), + expr.as_deref(), + sql.as_deref(), + unit.as_deref(), + summary.as_deref(), + req.derived, + user.id, + ) + .await?; + let _ = utopia_store::audit::record( + &state.pool, + Some(kb_id), + user.id, + "mapping.written", + "concept_mapping", + Some(id), + json!({ "concept": concept, "source": ds.name }), + ) + .await; + Ok(Json(json!({ "id": id, "concept_id": resolved.entity_id }))) +} + +#[derive(Deserialize)] +pub struct PreviewReq { + source: String, + table_name: Option, + expr: Option, + sql: Option, +} + +/// 把一条口径过只读闸跑一遍，回第一行——写口径的人当场看到它算出来的数。 +/// +/// 这正是测量台判分的那个动作（#501 / #520：把定义跑一遍，跟 gold 比数）， +/// 搬到页面上来。跑不通的定义无害，它失败得很响；跑得通而算错的才是全部风险， +/// 而看一眼数是人能做的唯一一道检查。 +pub async fn preview( + State(state): State, + AuthUser(user): AuthUser, + Path(kb_id): Path, + Json(req): Json, +) -> ApiResult> { + require_kb(&state, &user, kb_id, Role::Editor).await?; + let (table_name, expr, sql) = (clean(&req.table_name), clean(&req.expr), clean(&req.sql)); + definition_shape(&table_name, &expr, &sql)?; + let Some(rendered) = render_sql(&table_name, &expr, &sql) else { + return Err(AppError::invalid( + "empty_mapping", + "A preview needs SQL, or an expression together with a table", + ) + .into()); + }; + let mounted = utopia_store::datasources::mounted(&state.pool, kb_id).await?; + let Some(ds) = mounted + .iter() + .find(|d| d.name.eq_ignore_ascii_case(req.source.trim())) + else { + return Err(AppError::invalid( + "source_not_mounted", + "That data source is not mounted on this knowledge base", + ) + .into()); + }; + // 引擎的报错原样带回：列名拼错、语法错、超时是三件不同的事 + let out = super::tools::run_query(&state, ds.id, &rendered) + .await + .map_err(|e| AppError::invalid("preview_failed", e.to_string()))?; + // 结果是 JSON Lines；第一行就是这条口径算出来的那个数（或那一组数） + let rows: Vec = out + .lines() + .filter_map(|l| serde_json::from_str::(l.trim()).ok()) + .take(5) + .collect(); + Ok(Json( + json!({ "sql": rendered, "row": rows.first(), "rows": rows }), + )) +} + +#[cfg(test)] +mod tests { + use super::{definition_shape, render_sql}; + + #[test] + fn a_definition_needs_something_to_run() { + let none: Option = None; + assert!(definition_shape(&none, &none, &none).is_err()); + assert!(definition_shape(&Some("t".into()), &none, &none).is_ok()); + assert!(definition_shape(&none, &Some("sum(x)".into()), &none).is_ok()); + assert!(definition_shape(&none, &none, &Some("SELECT 1".into())).is_ok()); + } + + #[test] + fn preview_runs_what_the_bench_scores() { + let none: Option = None; + // sql 优先，且去掉结尾的分号——闸门只放行单条语句 + assert_eq!( + render_sql( + &Some("t".into()), + &Some("sum(x)".into()), + &Some("SELECT 2;".into()) + ) + .as_deref(), + Some("SELECT 2") + ); + assert_eq!( + render_sql( + &Some("dw.orders".into()), + &Some("sum(amt) / 100.0".into()), + &none + ) + .as_deref(), + Some("SELECT sum(amt) / 100.0 FROM dw.orders") + ); + // 只有表名不是一个可执行的口径 + assert_eq!(render_sql(&Some("t".into()), &none, &none), None); + } +} diff --git a/crates/utopia-server/src/api/mod.rs b/crates/utopia-server/src/api/mod.rs index fd762c37d..be30680d1 100644 --- a/crates/utopia-server/src/api/mod.rs +++ b/crates/utopia-server/src/api/mod.rs @@ -148,7 +148,12 @@ pub fn router(state: AppState, cfg: &AppConfig) -> Router { "/me/tokens/{token_id}", axum::routing::delete(token_routes::revoke), ) - .route("/kbs/{id}/mappings", get(mapping_routes::list)) + .route( + "/kbs/{id}/mappings", + get(mapping_routes::list).post(mapping_routes::create), + ) + // 静态段排在 `{mapping_id}` 前面：先跑一遍看数，不落库 + .route("/kbs/{id}/mappings/preview", post(mapping_routes::preview)) .route( "/kbs/{id}/mappings/{mapping_id}", axum::routing::patch(mapping_routes::revise), diff --git a/crates/utopia-server/src/api/tools.rs b/crates/utopia-server/src/api/tools.rs index 4aa2ea402..83f80f418 100644 --- a/crates/utopia-server/src/api/tools.rs +++ b/crates/utopia-server/src/api/tools.rs @@ -708,7 +708,7 @@ pub(super) fn charter_source_json(n: usize, h: &utopia_search::DocsSection) -> s } /// 问数执行：安全闸（解析白名单）→ 引擎执行（只读会话 + 强制 LIMIT + 超时）→ JSON 行。 -async fn run_query(state: &AppState, ds_id: Uuid, sql: &str) -> anyhow::Result { +pub(crate) async fn run_query(state: &AppState, ds_id: Uuid, sql: &str) -> anyhow::Result { let (engine, conn) = utopia_store::datasources::engine_and_conn(&state.pool, ds_id).await?; // 闸门按引擎选方言：Databricks 的反引号、Snowflake 的 :: 转型都得先过得了解析 let guarded = crate::query_engine::guard_sql_for(&engine, sql)?; diff --git a/crates/utopia-server/src/mappings.rs b/crates/utopia-server/src/mappings.rs index 8b6e2f1bc..12714f486 100644 --- a/crates/utopia-server/src/mappings.rs +++ b/crates/utopia-server/src/mappings.rs @@ -19,7 +19,7 @@ const MAX_SCHEMA_CHARS: usize = 12_000; /// 内置本体包里——0009 之后建库不再自带类。没有它们，下面的 `type_id` 查不到， /// 每条提议都被 `continue` 吞掉，页面只说"已排队"就再无下文（#223）。 /// 所以探索前把两个类补上：builtin，描述给抽取提示词，本体页可以改 -async fn ensure_concept_types(pool: &sqlx::PgPool, kb_id: Uuid) -> anyhow::Result<()> { +pub(crate) async fn ensure_concept_types(pool: &sqlx::PgPool, kb_id: Uuid) -> anyhow::Result<()> { for (key, label, description) in [ ( "metric", diff --git a/crates/utopia-store/src/mappings.rs b/crates/utopia-store/src/mappings.rs index fb043f1eb..6d8332537 100644 --- a/crates/utopia-store/src/mappings.rs +++ b/crates/utopia-store/src/mappings.rs @@ -76,6 +76,58 @@ pub async fn propose( Ok(id) } +/// 人从零写一条口径（#562）。**落下来就是确认的**：写的人就是表态的人， +/// 不需要再过一遍审。 +/// +/// 从前这张表只有探索一条来路。一个数据团队手上有自己的指标口径文档，却没有 +/// 地方把它填进去——而口径进了问数的提示词，宽表语料从 1/18 到 17/18（#520）。 +/// +/// 同一个 (概念, 源) 已经有一条时报冲突，而不是悄悄覆盖：那一条可能是探索提的、 +/// 人已经确认过的，覆盖等于抹掉一次表态。要改用 `revise`。 +/// 探索反过来也盖不掉这条：`propose` 只刷新 `proposed` 的行。 +#[allow(clippy::too_many_arguments)] +pub async fn create( + pool: &PgPool, + kb_id: Uuid, + concept_id: Uuid, + source: &str, + table_name: Option<&str>, + expr: Option<&str>, + sql: Option<&str>, + unit: Option<&str>, + summary: Option<&str>, + derived: bool, + actor: Uuid, +) -> AppResult { + let id = Uuid::now_v7(); + sqlx::query( + "INSERT INTO concept_mappings + (id, kb_id, concept_id, source, table_name, expr, sql, unit, summary, derived, + status, decided_by, decided_at, written_by) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, 'confirmed', $11, now(), $11)", + ) + .bind(id) + .bind(kb_id) + .bind(concept_id) + .bind(source) + .bind(table_name) + .bind(expr) + .bind(sql) + .bind(unit) + .bind(summary) + .bind(derived) + .bind(actor) + .execute(pool) + .await + .map_err(|e| match &e { + sqlx::Error::Database(db) if db.is_unique_violation() => AppError::Conflict( + "A definition for this concept on this source already exists; revise it instead".into(), + ), + _ => AppError::Db(e), + })?; + Ok(id) +} + /// 还等着人表态的。Review 页读它。 pub async fn proposed( pool: &PgPool, @@ -85,7 +137,7 @@ pub async fn proposed( ) -> AppResult> { Ok(sqlx::query_as( "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source, - m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status + m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by FROM concept_mappings m JOIN entities e ON e.id = m.concept_id WHERE m.kb_id = $1 AND m.status = 'proposed' @@ -104,7 +156,7 @@ pub async fn proposed( pub async fn confirmed(pool: &PgPool, kb_id: Uuid, limit: i64) -> AppResult> { Ok(sqlx::query_as( "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source, - m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status + m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by FROM concept_mappings m JOIN entities e ON e.id = m.concept_id WHERE m.kb_id = $1 AND m.status = 'confirmed' @@ -229,7 +281,7 @@ pub async fn page( OR m.table_name ILIKE '%' || $3 || '%')"; let rows: Vec = sqlx::query_as(&format!( "SELECT m.id, m.concept_id, e.canonical_name AS concept_name, m.source, - m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status + m.table_name, m.expr, m.sql, m.unit, m.summary, m.derived, m.status, m.written_by FROM concept_mappings m JOIN entities e ON e.id = m.concept_id {WHERE} diff --git a/crates/utopia-store/tests/a_definition_can_be_written_by_hand.rs b/crates/utopia-store/tests/a_definition_can_be_written_by_hand.rs new file mode 100644 index 000000000..254ddbdbd --- /dev/null +++ b/crates/utopia-store/tests/a_definition_can_be_written_by_hand.rs @@ -0,0 +1,134 @@ +//! 人从零写的口径——打在真库上（#562）。 +//! +//! 三条性质： +//! 1. **写下来就是确认的**，且记得是谁写的（`written_by`），探索提的那些为空。 +//! 2. **同一个 (概念, 源) 不悄悄覆盖**：已经有一条时报冲突，改要走 `revise`。 +//! 3. **探索盖不掉人写的**：`propose` 只刷新 `proposed` 的行——与它不刷回一条 +//! 拒绝是同一条规则。 + +use sqlx::PgPool; +use uuid::Uuid; + +async fn fixture(pool: &PgPool) -> anyhow::Result<(Uuid, Uuid, Uuid)> { + let (org, ws, kb, ent, user) = ( + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + Uuid::now_v7(), + ); + sqlx::query("INSERT INTO organizations (id, name) VALUES ($1, 'write-test')") + .bind(org) + .execute(pool) + .await?; + sqlx::query("INSERT INTO workspaces (id, org_id, name) VALUES ($1, $2, 'write-test')") + .bind(ws) + .bind(org) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO knowledge_bases (id, workspace_id, name) VALUES ($1, $2, 'write-test')", + ) + .bind(kb) + .bind(ws) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO entities (id, kb_id, canonical_name, aliases) VALUES ($1, $2, 'GMV', '{}')", + ) + .bind(ent) + .bind(kb) + .execute(pool) + .await?; + sqlx::query( + "INSERT INTO users (id, org_id, email, display_name, password_hash) + VALUES ($1, $2, $1 || '@w.test', 'w', 'x')", + ) + .bind(user) + .bind(org) + .execute(pool) + .await?; + Ok((kb, ent, user)) +} + +#[tokio::test] +async fn a_written_definition_is_confirmed_and_signed() -> anyhow::Result<()> { + let Some(url) = utopia_store::test_db::url() else { + return Ok(()); + }; + let pool = PgPool::connect(&url).await?; + let (kb, ent, user) = fixture(&pool).await?; + + let run = async { + let id = utopia_store::mappings::create( + &pool, + kb, + ent, + "warehouse", + Some("dw.dwd_ord_dtl"), + Some("sum(amt_pay) FILTER (WHERE pay_st IN (1,2) AND is_test = 0) / 100.0"), + None, + Some("CNY"), + Some("paid, non-test, in yuan"), + false, + user, + ) + .await?; + + // 落下来就是确认的：问数下一句就读得到，不用再过审 + let confirmed = utopia_store::mappings::confirmed(&pool, kb, 100).await?; + assert_eq!(confirmed.len(), 1); + assert_eq!(confirmed[0].id, id); + assert_eq!(confirmed[0].written_by, Some(user), "人写的记谁写的"); + assert!( + utopia_store::mappings::proposed(&pool, kb, 100, 0).await?.is_empty(), + "不该出现在待审队列里" + ); + + // 同一个 (概念, 源) 再写一条：冲突，而不是悄悄盖掉已有的那条 + let again = utopia_store::mappings::create( + &pool, kb, ent, "warehouse", Some("dw.dwd_ord_dtl"), Some("sum(amt_total)"), + None, None, None, false, user, + ) + .await; + assert!( + matches!(again, Err(utopia_core::AppError::Conflict(_))), + "同键第二条该是 Conflict，得到 {again:?}" + ); + + // 探索盖不掉人写的：propose 同键 → 行原样不动、仍是 confirmed + utopia_store::mappings::propose( + &pool, kb, ent, "warehouse", Some("orders_v2"), Some("sum(amt_total)"), + None, None, None, false, + ) + .await?; + let confirmed = utopia_store::mappings::confirmed(&pool, kb, 100).await?; + assert_eq!(confirmed.len(), 1); + assert_eq!(confirmed[0].table_name.as_deref(), Some("dw.dwd_ord_dtl")); + assert_eq!(confirmed[0].written_by, Some(user)); + // 探索自己提的那些 written_by 为空——页面靠这个分「人写」与「探索提的」 + let ent2 = Uuid::now_v7(); + sqlx::query("INSERT INTO entities (id, kb_id, canonical_name, aliases) VALUES ($1, $2, 'Orders', '{}')") + .bind(ent2) + .bind(kb) + .execute(&pool) + .await?; + utopia_store::mappings::propose( + &pool, kb, ent2, "warehouse", Some("dw.dwd_ord_dtl"), Some("count(distinct ord_id)"), + None, None, None, false, + ) + .await?; + let proposed = utopia_store::mappings::proposed(&pool, kb, 100, 0).await?; + assert_eq!(proposed.len(), 1); + assert_eq!(proposed[0].written_by, None); + Ok::<_, anyhow::Error>(()) + } + .await; + + // 只删知识库，不删 org/user——用户是软删除的，测试也不该造一个产品里不存在的动作 + sqlx::query("DELETE FROM knowledge_bases WHERE id = $1") + .bind(kb) + .execute(&pool) + .await?; + run +} diff --git a/migrations/0046_a_definition_can_be_written_by_hand.sql b/migrations/0046_a_definition_can_be_written_by_hand.sql new file mode 100644 index 000000000..f178251eb --- /dev/null +++ b/migrations/0046_a_definition_can_be_written_by_hand.sql @@ -0,0 +1,10 @@ +-- 一条口径可以由人从零写（#562）。 +-- +-- `concept_mappings` 的行从前只有一条来路：探索提议、人确认。一个数据团队手上有 +-- 自己的指标口径文档，却没有地方把它填进去——而测量台已经量过，口径进了问数的 +-- 提示词，宽表语料从 1/18 到 17/18（#520）。缺的不是结构，是这条入口。 +-- +-- 记下是谁写的。探索提的 `written_by` 为空，人写的记人；页面据此标「人写」还是 +-- 「探索提的」。`decided_by` 分不出这件事：探索提的经人确认之后同样有 decided_by。 +ALTER TABLE concept_mappings + ADD COLUMN written_by UUID REFERENCES users(id);