diff --git a/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json b/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json deleted file mode 100644 index db9d674fcf..0000000000 --- a/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id,\n resource_type, name, asset_path, tags)\n VALUES ($1, $2, $3, $4, 'model.p.stock', 'model', 'stock',\n 'u/a/wh/analytics/stock', '{}'),\n ($1, $2, $3, $4, 'model.p.stock_daily', 'model', 'stock_daily',\n 'u/a/wh/analytics/stock_daily', '{}')", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Int8", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d" -} diff --git a/backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json b/backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json deleted file mode 100644 index 00b839de6a..0000000000 --- a/backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,\n parent_unique_id, parent_column, child_unique_id,\n child_column, lineage_kind)\n VALUES ($1, $2, $3, $4, 'model.p.stock', 'sku', 'model.p.stock_daily', 'sku', 'copy')", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Int8", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf" -} diff --git a/backend/.sqlx/query-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json b/backend/.sqlx/query-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json deleted file mode 100644 index bdb54cfbb9..0000000000 --- a/backend/.sqlx/query-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,\n parent_unique_id, parent_column, child_unique_id,\n child_column, lineage_kind)\n VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000',\n 'model.p.raw_orders', 'id', 'model.p.orders', 'order_id', 'copy'),\n ($1, $2, $3, '00000000-0000-0000-0000-000000000000',\n 'model.p.raw_orders', 'status', 'model.p.orders', 'order_id', 'scan')", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Int8" - ] - }, - "nullable": [] - }, - "hash": "210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c" -} diff --git a/backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json b/backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json deleted file mode 100644 index a498ecccd0..0000000000 --- a/backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,\n parent_unique_id, parent_column, child_unique_id,\n child_column, lineage_kind)\n VALUES ($1, $2, NULL, $3, 'model.p.draft_src', 'raw', 'model.p.draft', 'clean', 'mod')", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912" -} diff --git a/backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json b/backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json deleted file mode 100644 index 49d01c2b8b..0000000000 --- a/backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json +++ /dev/null @@ -1,16 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id,\n resource_type, name, asset_path, tags)\n VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.raw_orders',\n 'model', 'raw_orders', 'u/a/wh/analytics/raw_orders', '{}'),\n ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.orders',\n 'model', 'orders', 'u/a/wh/analytics/orders', '{}')", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Int8" - ] - }, - "nullable": [] - }, - "hash": "9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82" -} diff --git a/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json b/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json deleted file mode 100644 index 07694284d8..0000000000 --- a/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json +++ /dev/null @@ -1,53 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH\n -- The project version that owns the asked-for relation, in the graph\n -- on screen. Not the folder-wide `live` set the graph resolves: one\n -- asset is asked about here, so the version is decided per candidate\n -- row. Usually one row; a relation a second project declares as a\n -- source has two, and each answers for its own lineage.\n owner AS (\n SELECT DISTINCT n.script_path, n.script_hash, n.job_id\n FROM dbt_node n\n WHERE n.workspace_id = $1 AND n.asset_path = $2\n -- The run's snapshot, or the deployed graph when that job stored\n -- none -- a build pins only if it wrote one.\n AND n.job_id = CASE WHEN $5::uuid IS NOT NULL AND EXISTS (\n SELECT 1 FROM dbt_graph_snapshot g\n WHERE g.workspace_id = $1 AND g.job_id = $5)\n THEN $5::uuid\n ELSE '00000000-0000-0000-0000-000000000000'::uuid END\n -- The gate, once, for the whole answer.\n AND ( $6\n OR n.script_path = ANY($7)\n OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx\n WHERE n.script_path = pfx\n OR left(n.script_path, length(pfx) + 1) = pfx || '/' ) )\n AND CASE\n -- Pinned: which version comes from a job this caller was\n -- already granted, so `script` does not decide THAT — but\n -- it still decides whether the project may be read, the\n -- same second gate `script_visible` is on the graph. Being\n -- entitled to a run is not being entitled to the SQL\n -- behind it, and column lineage is that SQL's shape. A\n -- version-less row is exempt because it is an editor\n -- buffer, which has no `script` row to ask and reaches\n -- this only through the parse job that wrote it.\n WHEN $4::text IS NOT NULL\n THEN n.script_path = $4 AND n.script_hash IS NOT DISTINCT FROM $3::bigint\n AND ($3::bigint IS NULL OR EXISTS (\n SELECT 1 FROM script sc\n WHERE sc.workspace_id = $1 AND sc.path = n.script_path\n AND sc.hash = $3))\n -- A named version, for an editor open on an older one.\n -- `script` is read under RLS, so this is the visibility\n -- check as well as the existence one.\n WHEN $3::bigint IS NOT NULL\n THEN n.script_hash = $3 AND EXISTS (\n SELECT 1 FROM script sc\n WHERE sc.workspace_id = $1 AND sc.path = n.script_path\n AND sc.hash = $3)\n -- Otherwise the version deployed now: an older one's rows\n -- outlive it in `dbt_node` until the sweep, and describe a\n -- project that is no longer what runs. `language` narrows\n -- it the way the graph's own resolution does, so a path\n -- that has since become a script of another kind draws and\n -- explains the same version rather than disagreeing.\n ELSE n.script_hash = (\n SELECT sc.hash FROM script sc\n WHERE sc.workspace_id = $1 AND sc.path = n.script_path\n AND sc.language = 'dbt'\n AND sc.deleted = false AND sc.archived = false\n ORDER BY sc.created_at DESC LIMIT 1)\n END\n )\n -- DIRECT kinds only. `scan` — the column was read to produce the ROW,\n -- not the value — reaches every output column of its model, so it is\n -- most of a project's stored lineage and none of what a trace draws.\n -- It stays in the table for a later view to ask for.\n SELECT p.asset_path AS \"from_path!\", e.parent_column AS \"from_column!\",\n c.asset_path AS \"to_path!\", e.child_column AS \"to_column!\",\n e.lineage_kind AS \"kind!\"\n FROM dbt_column_edge e\n JOIN owner o ON o.script_path = e.script_path\n AND o.script_hash IS NOT DISTINCT FROM e.script_hash\n AND o.job_id = e.job_id\n JOIN dbt_node p ON p.workspace_id = e.workspace_id\n AND p.script_path = e.script_path\n AND p.script_hash IS NOT DISTINCT FROM e.script_hash\n AND p.job_id = e.job_id\n AND p.unique_id = e.parent_unique_id\n JOIN dbt_node c ON c.workspace_id = e.workspace_id\n AND c.script_path = e.script_path\n AND c.script_hash IS NOT DISTINCT FROM e.script_hash\n AND c.job_id = e.job_id\n AND c.unique_id = e.child_unique_id\n WHERE e.workspace_id = $1\n AND e.lineage_kind IN ('copy', 'mod')\n AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "from_path!", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "from_column!", - "type_info": "Text" - }, - { - "ordinal": 2, - "name": "to_path!", - "type_info": "Text" - }, - { - "ordinal": 3, - "name": "to_column!", - "type_info": "Text" - }, - { - "ordinal": 4, - "name": "kind!", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Int8", - "Text", - "Uuid", - "Bool", - "TextArray", - "TextArray" - ] - }, - "nullable": [ - true, - false, - true, - false, - false - ] - }, - "hash": "c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e" -} diff --git a/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json b/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json deleted file mode 100644 index ccb9863853..0000000000 --- a/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json +++ /dev/null @@ -1,17 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,\n parent_unique_id, parent_column, child_unique_id,\n child_column, lineage_kind)\n VALUES ($1, $2, $3, $4, 'model.p.raw_orders', 'id', 'model.p.orders', 'order_id',\n 'copy')", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Varchar", - "Int8", - "Uuid" - ] - }, - "nullable": [] - }, - "hash": "f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e" -} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 08b74c0311..bd08111623 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -6,7 +6,6 @@ use axum::{ use serde::{Deserialize, Serialize}; use serde_json::Value; use sqlx::Row; -use std::collections::{HashMap, HashSet}; use windmill_common::{ assets::{parse_asset_trigger_ref, AssetKind, AssetUsageKind}, db::UserDB, @@ -14,9 +13,7 @@ use windmill_common::{ utils::escape_ilike_pattern, }; -use windmill_api_auth::{ - build_scope_path_filter, build_scope_path_predicate, ApiAuthed, ScopePathFilter, -}; +use windmill_api_auth::{build_scope_path_predicate, ApiAuthed}; // Partition-range backfill preview. The logic (producer resolution, range // enumeration, status join) is enterprise: the `private` build compiles the @@ -36,7 +33,6 @@ pub fn workspaced_service() -> Router { .route("/list_by_usages", post(list_assets_by_usages)) .route("/list_favorites", get(list_favorites)) .route("/graph", get(asset_graph)) - .route("/column_lineage", get(dbt_column_lineage)) .route("/pipelines", get(list_pipeline_folders)) .route("/partitions", get(list_partitions)) .route("/partitions_in_range", get(list_partitions_in_range)) @@ -961,266 +957,6 @@ struct DbtLineageEdge { to_asset_path: String, } -/// One column-to-column edge, in the same terms: the two relations and the two -/// columns, never dbt's node ids. -#[derive(Serialize, Debug, PartialEq, Eq, PartialOrd, Ord)] -struct DbtColumnLineageEdge { - from_asset_path: String, - from_column: String, - to_asset_path: String, - to_column: String, - /// dbt's own word for how the value travelled: `copy` (passthrough), `mod` - /// (transformed), `scan` (read to produce the ROW rather than the value — a - /// join key, a predicate, a `group by`). Sent verbatim, including a kind - /// this engine version invented, because the renderer decides what a kind - /// means and the set is the engine's. - kind: String, -} - -/// One dbt relation's column lineage: the connected component its columns sit -/// in, within the project that owns it. -/// -/// Not the relation's own edges, which would stop one hop out — a trace walks -/// transitively — and not the whole project's, which carries families the -/// selected relation cannot reach. The component is what the canvas lays out, -/// so it is exactly what a consumer can draw. -/// -/// Its own endpoint rather than a field on the graph, which is folder-wide and -/// polled by a run page while this is rendered for a single selection. A -/// folder's worth of edges spans many projects and many callers' access, so it -/// would need a cap, and a cap has to be applied after every filter that could -/// drop a row — which is the ordering this shape removes rather than gets -/// right. Here the filters ARE the answer: scope and visibility are decided -/// once in SQL for the script that owns the relation, the component is walked -/// over what that returns, and the size is bounded at ingest -/// (`MAX_COLUMN_EDGES` per version, of which only the direct kinds are served). -#[derive(Deserialize)] -pub struct ColumnLineageQuery { - /// The `dbt://` relation whose lineage to return. - pub asset_path: String, - /// A deployed version other than the current one, for an editor open on an - /// older one. A run's or an editor buffer's graph is NOT reachable from - /// here: it pins to a job, and that costs the job-read gate. - pub dbt_script_hash: Option, -} - -#[derive(Serialize, Debug, PartialEq, Eq, PartialOrd, Ord)] -pub struct ColumnLineageResponse { - /// Direct (`copy` / `mod`) column edges of the component this relation's - /// columns sit in, in the terms the canvas draws. Empty when the project - /// never asked for the analysis pass, which is the ordinary case. - edges: Vec, -} - -async fn dbt_column_lineage( - authed: ApiAuthed, - Path(w_id): Path, - Extension(user_db): Extension, - Query(q): Query, -) -> JsonResult { - // `None`: pinning to one run is job-scoped and this endpoint is authorized - // as `assets:read`. See `dbt_column_lineage_for`. - dbt_column_lineage_for(&authed, &w_id, user_db, q, None).await -} - -/// One dbt relation's column lineage, optionally as one run saw it. -/// -/// AUTHORIZES NOTHING BY ITSELF, on the same contract as `asset_graph_for`: -/// `assets:read` always, and the job-read gate for `Some(pinned)`, whose path -/// and hash are then taken from that job's row rather than from the caller. -pub async fn dbt_column_lineage_for( - authed: &ApiAuthed, - w_id: &str, - user_db: UserDB, - q: ColumnLineageQuery, - pinned: Option, -) -> JsonResult { - // A column-level view is the shape of what the author WROTE, so it takes the - // model's own gate rather than the relation's. Decided ONCE here, for the - // script that owns the asset, instead of per row against a cap. - let (scope_all, scope_exact, scope_prefix) = - match build_scope_path_filter(authed, "scripts", "read") { - ScopePathFilter::AllowAll => (true, Vec::new(), Vec::new()), - ScopePathFilter::Restricted { exact, prefix } => (false, exact, prefix), - }; - let (pinned_path, script_hash) = match pinned.as_ref() { - // The job's own version, so a pin cannot name one project's run while - // claiming another's version — including when it names NONE, which is - // the editor buffer. - Some(p) => (Some(p.script_path.as_str()), p.script_hash), - None => (None, q.dbt_script_hash.map(|h| h.0)), - }; - let pinned_job_id = pinned.as_ref().map(|p| p.job_id); - let mut tx = user_db.begin(authed).await?; - let rows = sqlx::query!( - r#"WITH - -- The project version that owns the asked-for relation, in the graph - -- on screen. Not the folder-wide `live` set the graph resolves: one - -- asset is asked about here, so the version is decided per candidate - -- row. Usually one row; a relation a second project declares as a - -- source has two, and each answers for its own lineage. - owner AS ( - SELECT DISTINCT n.script_path, n.script_hash, n.job_id - FROM dbt_node n - WHERE n.workspace_id = $1 AND n.asset_path = $2 - -- The run's snapshot, or the deployed graph when that job stored - -- none -- a build pins only if it wrote one. - AND n.job_id = CASE WHEN $5::uuid IS NOT NULL AND EXISTS ( - SELECT 1 FROM dbt_graph_snapshot g - WHERE g.workspace_id = $1 AND g.job_id = $5) - THEN $5::uuid - ELSE '00000000-0000-0000-0000-000000000000'::uuid END - -- The gate, once, for the whole answer. - AND ( $6 - OR n.script_path = ANY($7) - OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx - WHERE n.script_path = pfx - OR left(n.script_path, length(pfx) + 1) = pfx || '/' ) ) - AND CASE - -- Pinned: which version comes from a job this caller was - -- already granted, so `script` does not decide THAT — but - -- it still decides whether the project may be read, the - -- same second gate `script_visible` is on the graph. Being - -- entitled to a run is not being entitled to the SQL - -- behind it, and column lineage is that SQL's shape. A - -- version-less row is exempt because it is an editor - -- buffer, which has no `script` row to ask and reaches - -- this only through the parse job that wrote it. - WHEN $4::text IS NOT NULL - THEN n.script_path = $4 AND n.script_hash IS NOT DISTINCT FROM $3::bigint - AND ($3::bigint IS NULL OR EXISTS ( - SELECT 1 FROM script sc - WHERE sc.workspace_id = $1 AND sc.path = n.script_path - AND sc.hash = $3)) - -- A named version, for an editor open on an older one. - -- `script` is read under RLS, so this is the visibility - -- check as well as the existence one. - WHEN $3::bigint IS NOT NULL - THEN n.script_hash = $3 AND EXISTS ( - SELECT 1 FROM script sc - WHERE sc.workspace_id = $1 AND sc.path = n.script_path - AND sc.hash = $3) - -- Otherwise the version deployed now: an older one's rows - -- outlive it in `dbt_node` until the sweep, and describe a - -- project that is no longer what runs. `language` narrows - -- it the way the graph's own resolution does, so a path - -- that has since become a script of another kind draws and - -- explains the same version rather than disagreeing. - ELSE n.script_hash = ( - SELECT sc.hash FROM script sc - WHERE sc.workspace_id = $1 AND sc.path = n.script_path - AND sc.language = 'dbt' - AND sc.deleted = false AND sc.archived = false - ORDER BY sc.created_at DESC LIMIT 1) - END - ) - -- DIRECT kinds only. `scan` — the column was read to produce the ROW, - -- not the value — reaches every output column of its model, so it is - -- most of a project's stored lineage and none of what a trace draws. - -- It stays in the table for a later view to ask for. - SELECT p.asset_path AS "from_path!", e.parent_column AS "from_column!", - c.asset_path AS "to_path!", e.child_column AS "to_column!", - e.lineage_kind AS "kind!" - FROM dbt_column_edge e - JOIN owner o ON o.script_path = e.script_path - AND o.script_hash IS NOT DISTINCT FROM e.script_hash - AND o.job_id = e.job_id - JOIN dbt_node p ON p.workspace_id = e.workspace_id - AND p.script_path = e.script_path - AND p.script_hash IS NOT DISTINCT FROM e.script_hash - AND p.job_id = e.job_id - AND p.unique_id = e.parent_unique_id - JOIN dbt_node c ON c.workspace_id = e.workspace_id - AND c.script_path = e.script_path - AND c.script_hash IS NOT DISTINCT FROM e.script_hash - AND c.job_id = e.job_id - AND c.unique_id = e.child_unique_id - WHERE e.workspace_id = $1 - AND e.lineage_kind IN ('copy', 'mod') - AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL"#, - w_id, - q.asset_path, - script_hash, - pinned_path, - pinned_job_id, - scope_all, - &scope_exact[..], - &scope_prefix[..], - ) - .fetch_all(&mut *tx) - .await?; - tx.commit().await?; - - let mut edges: Vec = rows - .into_iter() - .map(|r| DbtColumnLineageEdge { - from_asset_path: r.from_path, - from_column: r.from_column, - to_asset_path: r.to_path, - to_column: r.to_column, - kind: r.kind, - }) - .collect(); - // Two projects can describe one relation, so the same edge can arrive twice. - edges.sort(); - edges.dedup(); - Ok(Json(ColumnLineageResponse { - edges: component(edges, &q.asset_path), - })) -} - -/// Keep the edges of the connected component the asked-for relation sits in. -/// -/// The canvas lays out the component of the selected relation's columns, so a -/// project's other model families are edges nothing it draws can reach. Walked -/// here rather than in SQL: a recursive CTE has no index to walk, so it rescans -/// the whole edge set once per level — measured at 1.24s against 59ms for the -/// query alone on a 3000-model project, for a walk that is microseconds over a -/// map. Columns are keyed by relation, not by project, which is how the canvas -/// keys them too: two projects describing one relation draw one node. -fn component(mut edges: Vec, asset_path: &str) -> Vec { - let keep = { - let mut incident: HashMap<(&str, &str), Vec> = HashMap::new(); - for (i, e) in edges.iter().enumerate() { - let from = (e.from_asset_path.as_str(), e.from_column.as_str()); - let to = (e.to_asset_path.as_str(), e.to_column.as_str()); - incident.entry(from).or_default().push(i); - incident.entry(to).or_default().push(i); - } - let mut stack: Vec<(&str, &str)> = incident - .keys() - .filter(|(path, _)| *path == asset_path) - .copied() - .collect(); - let mut seen_node: HashSet<(&str, &str)> = stack.iter().copied().collect(); - let mut seen_edge = vec![false; edges.len()]; - while let Some(node) = stack.pop() { - for &i in incident.get(&node).map(Vec::as_slice).unwrap_or_default() { - if std::mem::replace(&mut seen_edge[i], true) { - continue; - } - let e = &edges[i]; - let ends = [ - (e.from_asset_path.as_str(), e.from_column.as_str()), - (e.to_asset_path.as_str(), e.to_column.as_str()), - ]; - for end in ends { - if seen_node.insert(end) { - stack.push(end); - } - } - } - } - seen_edge - }; - let mut i = 0; - edges.retain(|_| { - i += 1; - keep[i - 1] - }); - edges -} - async fn asset_graph( authed: ApiAuthed, Path(w_id): Path, diff --git a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs index 5c10446c9c..5fb2c6a088 100644 --- a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs +++ b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs @@ -8,9 +8,7 @@ //! whole dbt half, and every later fix in this area re-touched one of the two. use sqlx::{Pool, Postgres}; -use windmill_api_assets::{ - asset_graph_for, dbt_column_lineage_for, ColumnLineageQuery, GraphQuery, PinnedRun, -}; +use windmill_api_assets::{asset_graph_for, GraphQuery, PinnedRun}; use windmill_api_auth::ApiAuthed; use windmill_common::db::UserDB; @@ -113,20 +111,6 @@ async fn seed(db: &Pool, job: uuid::Uuid) { .execute(db) .await .unwrap(); - sqlx::query!( - "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id, - parent_unique_id, parent_column, child_unique_id, - child_column, lineage_kind) - VALUES ($1, $2, $3, $4, 'model.p.raw_orders', 'id', 'model.p.orders', 'order_id', - 'copy')", - WS, - PATH, - HASH, - job - ) - .execute(db) - .await - .unwrap(); // A test node, for the arguments it carries: `accepted_values` spells out a // column's domain. sqlx::query!( @@ -426,22 +410,6 @@ async fn seed_editor_graph(db: &Pool, job: uuid::Uuid) { .execute(db) .await .unwrap(); - // A version-less row's `script_hash` is NULL on both sides of every join and - // every visibility check, and `= NULL` is never true — so the column edges - // need the same NULL arm the node query has, or a buffer parse renders its - // columns and none of their lineage. - sqlx::query!( - "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id, - parent_unique_id, parent_column, child_unique_id, - child_column, lineage_kind) - VALUES ($1, $2, NULL, $3, 'model.p.draft_src', 'raw', 'model.p.draft', 'clean', 'mod')", - WS, - PATH, - job - ) - .execute(db) - .await - .unwrap(); } /// A buffer parse is the third provenance: no deployed version behind it, and @@ -520,216 +488,3 @@ async fn an_editor_graph_renders_only_through_its_own_job(db: Pool) { "nor of a run of the deployed version: {deployed_run}" ); } - -fn column_query(asset_path: &str) -> ColumnLineageQuery { - ColumnLineageQuery { asset_path: asset_path.to_string(), dbt_script_hash: None } -} - -async fn column_lineage( - db: &Pool, - authed: &ApiAuthed, - asset_path: &str, - pinned: Option, -) -> serde_json::Value { - let res = dbt_column_lineage_for( - authed, - WS, - UserDB::new(db.clone()), - column_query(asset_path), - pinned, - ) - .await - .unwrap(); - serde_json::to_value(&res.0).unwrap()["edges"].clone() -} - -/// The buffer parse's own lineage, which is the case the versionless rows exist -/// for. Its `script_hash` is NULL on both sides of every join and every -/// visibility check, and `= NULL` is never true — so the versionless arm has to -/// be written for it, or a parse renders its columns and none of their lineage. -#[sqlx::test(migrations = "../migrations", fixtures("base"))] -async fn an_editor_buffers_column_lineage_answers_through_its_job(db: Pool) { - let parse = uuid::Uuid::from_u128(9); - seed(&db, uuid::Uuid::from_u128(7)).await; - seed_editor_graph(&db, parse).await; - let admin = ApiAuthed { is_admin: true, ..outsider() }; - - let pinned = PinnedRun { job_id: parse, script_path: PATH.to_string(), script_hash: None }; - assert_eq!( - column_lineage(&db, &admin, "u/a/wh/analytics/draft", Some(pinned)).await, - serde_json::json!([{ - "from_asset_path": "u/a/wh/analytics/draft_src", - "from_column": "raw", - "to_asset_path": "u/a/wh/analytics/draft", - "to_column": "clean", - "kind": "mod", - }]), - ); - // Unpinned, the same relation resolves through the deployed version, which - // never heard of the buffer's models. - assert_eq!( - column_lineage(&db, &admin, "u/a/wh/analytics/draft", None).await, - serde_json::json!([]), - "a buffer's lineage is reachable only through the job that parsed it" - ); -} - -/// Being entitled to a RUN is not being entitled to the SQL behind it, and -/// column lineage is that SQL's shape. The pinned graph draws the relations for -/// a share-link viewer and redacts what the author wrote; the lineage is the -/// second, and resolving the version from the job must not be mistaken for -/// deciding that too. -#[sqlx::test(migrations = "../migrations", fixtures("base"))] -async fn a_pinned_run_does_not_hand_over_the_projects_column_lineage(db: Pool) { - let job = uuid::Uuid::from_u128(7); - seed(&db, job).await; - let pinned = - || PinnedRun { job_id: job, script_path: PATH.to_string(), script_hash: Some(HASH) }; - - assert_eq!( - column_lineage(&db, &outsider(), "u/a/wh/analytics/orders", Some(pinned())).await, - serde_json::json!([]), - "the run renders for them, its column-level shape does not" - ); - assert_eq!( - column_lineage( - &db, - &ApiAuthed { is_admin: true, ..outsider() }, - "u/a/wh/analytics/orders", - Some(pinned()) - ) - .await, - serde_json::json!([{ - "from_asset_path": "u/a/wh/analytics/raw_orders", - "from_column": "id", - "to_asset_path": "u/a/wh/analytics/orders", - "to_column": "order_id", - "kind": "copy", - }]), - "while a reader of the project gets it" - ); -} - -/// A column-level view is the shape of what the author WROTE, so it takes the -/// script's own gate — the same one that keeps `raw_code` behind access to the -/// project, applied here once for the script that owns the relation. -#[sqlx::test(migrations = "../migrations", fixtures("base"))] -async fn column_lineage_takes_the_scripts_gate_and_only_the_direct_kinds(db: Pool) { - let job = uuid::Uuid::from_u128(7); - seed(&db, job).await; - // A second relation of the deployed version, feeding the first. `scan` says - // the column was read to produce the ROW rather than the value, so it - // reaches every output column of its model and is never served. - sqlx::query!( - "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id, - resource_type, name, asset_path, tags) - VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.raw_orders', - 'model', 'raw_orders', 'u/a/wh/analytics/raw_orders', '{}'), - ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.orders', - 'model', 'orders', 'u/a/wh/analytics/orders', '{}')", - WS, - PATH, - HASH, - ) - .execute(&db) - .await - .unwrap(); - sqlx::query!( - "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id, - parent_unique_id, parent_column, child_unique_id, - child_column, lineage_kind) - VALUES ($1, $2, $3, '00000000-0000-0000-0000-000000000000', - 'model.p.raw_orders', 'id', 'model.p.orders', 'order_id', 'copy'), - ($1, $2, $3, '00000000-0000-0000-0000-000000000000', - 'model.p.raw_orders', 'status', 'model.p.orders', 'order_id', 'scan')", - WS, - PATH, - HASH, - ) - .execute(&db) - .await - .unwrap(); - - let admin = ApiAuthed { is_admin: true, ..outsider() }; - assert_eq!( - column_lineage(&db, &admin, "u/a/wh/analytics/orders", None).await, - serde_json::json!([{ - "from_asset_path": "u/a/wh/analytics/raw_orders", - "from_column": "id", - "to_asset_path": "u/a/wh/analytics/orders", - "to_column": "order_id", - "kind": "copy", - }]), - "the direct edge, and not the `scan` one beside it" - ); - assert_eq!( - column_lineage(&db, &outsider(), "u/a/wh/analytics/orders", None).await, - serde_json::json!([]), - "and nothing at all for a caller who cannot read the project" - ); -} - -/// One project routinely holds model families that share no column, and the -/// canvas lays out the connected component of the selected relation's columns. -/// Answering with the project's other components sends edges nothing can draw. -#[sqlx::test(migrations = "../migrations", fixtures("base"))] -async fn column_lineage_stops_at_the_selected_relations_component(db: Pool) { - let job = uuid::Uuid::from_u128(7); - seed(&db, job).await; - // A second family in the same project version, reaching neither of the two - // relations `seed` wired together. - sqlx::query!( - "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id, - resource_type, name, asset_path, tags) - VALUES ($1, $2, $3, $4, 'model.p.stock', 'model', 'stock', - 'u/a/wh/analytics/stock', '{}'), - ($1, $2, $3, $4, 'model.p.stock_daily', 'model', 'stock_daily', - 'u/a/wh/analytics/stock_daily', '{}')", - WS, - PATH, - HASH, - job - ) - .execute(&db) - .await - .unwrap(); - sqlx::query!( - "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id, - parent_unique_id, parent_column, child_unique_id, - child_column, lineage_kind) - VALUES ($1, $2, $3, $4, 'model.p.stock', 'sku', 'model.p.stock_daily', 'sku', 'copy')", - WS, - PATH, - HASH, - job - ) - .execute(&db) - .await - .unwrap(); - - let admin = ApiAuthed { is_admin: true, ..outsider() }; - let pinned = - || PinnedRun { job_id: job, script_path: PATH.to_string(), script_hash: Some(HASH) }; - assert_eq!( - column_lineage(&db, &admin, "u/a/wh/analytics/orders", Some(pinned())).await, - serde_json::json!([{ - "from_asset_path": "u/a/wh/analytics/raw_orders", - "from_column": "id", - "to_asset_path": "u/a/wh/analytics/orders", - "to_column": "order_id", - "kind": "copy", - }]), - "the orders family, and not the stock one beside it in the same project" - ); - assert_eq!( - column_lineage(&db, &admin, "u/a/wh/analytics/stock_daily", Some(pinned())).await, - serde_json::json!([{ - "from_asset_path": "u/a/wh/analytics/stock", - "from_column": "sku", - "to_asset_path": "u/a/wh/analytics/stock_daily", - "to_column": "sku", - "kind": "copy", - }]), - "and the other way round — reached from the child end, which is upstream" - ); -} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 2a6d52fe35..0630b3e262 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -24364,57 +24364,6 @@ paths: schema: $ref: "#/components/schemas/AssetGraph" - /w/{workspace}/assets/column_lineage: - get: - summary: Column-level lineage of one dbt relation - description: > - The direct (`copy` / `mod`) column-to-column lineage a relation's columns - sit in — the connected component within the dbt project that owns it, - from the engine's static analysis. Not the relation's own edges, which - would stop one hop out since a column trace walks transitively, and not - the whole project's, which carries model families the selected relation - cannot reach. The component is what the canvas lays out. - - Its own endpoint rather than a field on the asset graph: the graph is - folder-wide and polled by a run page, while this is rendered for one - selected asset at a time. A folder's worth of edges spans many projects - and many callers' access, so it would need a cap, and a cap has to come - after every filter that can drop a row. Here the filters are the answer: - the caller's scope and the project's visibility are decided once for the - owning script, and the size is bounded where the index is ingested. - - Empty for a project that did not opt into the analysis pass - (`column_lineage: true`), which is the ordinary case. The indirect `scan` - kind is stored but never served: it reaches every output column of its - model. - operationId: getDbtColumnLineage - tags: - - asset - parameters: - - $ref: "#/components/parameters/WorkspaceId" - - name: asset_path - in: query - required: true - description: The `dbt://` relation whose lineage to return - schema: - type: string - - name: dbt_script_hash - in: query - description: > - Pin to one deployed version, for an editor open on an older one. A - run's or an editor buffer's own graph is not reachable here: that - pins to a job, and costs the job-read gate — see - `jobs/dbt_column_lineage/{id}`. - schema: - type: string - responses: - "200": - description: the relation's column-level lineage - content: - application/json: - schema: - $ref: "#/components/schemas/DbtColumnLineage" - /w/{workspace}/assets/macros: get: summary: List every workspace DuckDB macro (deployed `// macros` libraries) @@ -24606,43 +24555,6 @@ paths: schema: $ref: "#/components/schemas/AssetGraph" - /w/{workspace}/jobs/dbt_column_lineage/{id}: - get: - summary: Get a relation's project column lineage as one run saw it - description: > - The same answer as `assets/column_lineage`, for the project version a - single job ran — including the dbt editor's parse of its own buffer, - whose graph belongs to that job and is reachable no other way. - Authorized through the job, the same gate as `dbt_graph`. Reaching the - run is not on its own enough to read the project: a caller with no access - to the script gets its relations and `ref()` edges from `dbt_graph` and - an empty answer here, exactly as that endpoint redacts the model's SQL. - operationId: getDbtRunColumnLineage - tags: - - job - parameters: - - $ref: "#/components/parameters/WorkspaceId" - - name: id - in: path - required: true - description: The job whose graph the lineage is read from - schema: - type: string - format: uuid - - name: asset_path - in: query - required: true - description: The `dbt://` relation whose lineage to return - schema: - type: string - responses: - "200": - description: the relation's column-level lineage - content: - application/json: - schema: - $ref: "#/components/schemas/DbtColumnLineage" - /w/{workspace}/jobs/run_progress/{id}: get: summary: List the per-relation progress one job has recorded so far @@ -26236,34 +26148,6 @@ components: drawn identically and the ambiguity would otherwise just move into the editor. Omitted for the unpinned workspace graph, which spans every project and so has no one time. - DbtColumnLineage: - type: object - description: >- - The direct column-to-column lineage a relation's columns sit in — the - connected component within the project that owns it — in the terms the - canvas draws: relations and columns, never dbt's node ids. - required: [edges] - properties: - edges: - type: array - items: - type: object - required: [from_asset_path, from_column, to_asset_path, to_column, kind] - properties: - from_asset_path: - type: string - from_column: - type: string - to_asset_path: - type: string - to_column: - type: string - kind: - type: string - description: >- - dbt's own word for how the value travelled — `copy` - (passthrough) or `mod` (transformed). Not an enum: the engine - treats the set as open. DbtAssetProvenance: type: object description: >- diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 7705b7dd20..dbf33cdddc 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -139,7 +139,6 @@ pub fn workspaced_service() -> Router { .route("/run_progress/{id}", get(get_run_progress)) .route("/run_assets/{id}", get(list_run_assets)) .route("/dbt_graph/{id}", get(get_dbt_run_graph)) - .route("/dbt_column_lineage/{id}", get(get_dbt_run_column_lineage)) .route("/dbt_resumable/{id}", get(get_dbt_resumable)) .route( "/dbt_resumable_script/p/{*script_path}", @@ -892,27 +891,21 @@ struct AssetProgress { error: Option, } -/// Which project version a dbt view pins to for this job, once the caller has -/// been shown to be entitled to it. -/// -/// `Ok(None)` is "answer unpinned", not a refusal: a job that stored no graph of -/// its own — and one that has aged out of retention — is served the deployed -/// version rather than an error, so a run page keeps drawing after the run is -/// gone. Pinning needs the full job-read contract, which is why it lives on -/// `require_job_read_access` here rather than as a parameter on `/assets/*`. -/// See docs/dbt-runtime.md. -async fn dbt_pinned_run( - authed: &ApiAuthed, - db: &DB, - user_db: &UserDB, - w_id: &str, - job_id: Uuid, - view_token: Option<&str>, -) -> error::Result> { +/// The asset graph as one run saw it. Pinning to a job needs the full job-read +/// contract, so it lives on `require_job_read_access` here rather than as a +/// parameter on `/assets/graph`. See docs/dbt-runtime.md. +async fn get_dbt_run_graph( + authed: ApiAuthed, + OptViewToken(view_token): OptViewToken, + Extension(db): Extension, + Extension(user_db): Extension, + Path((w_id, job_id)): Path<(String, Uuid)>, + Query(q): Query, +) -> error::JsonResult { // The scope domain comes from the URL segment, so `/jobs` asks a scoped token // for `jobs:read` alone while the body returned is asset data. Both are // required: the job gate below reaches this run, this reaches assets at all. - check_scopes(authed, || "assets:read".to_string())?; + check_scopes(&authed, || "assets:read".to_string())?; let job = sqlx::query!( r#"SELECT created_by, runnable_path, CASE WHEN kind = 'script' THEN runnable_id END AS script_hash, @@ -925,69 +918,42 @@ async fn dbt_pinned_run( AND g.script_hash IS NULL) AS "editor_graph!" FROM v2_job WHERE id = $1 AND workspace_id = $2"#, job_id, - w_id + &w_id ) - .fetch_optional(db) + .fetch_optional(&db) .await?; - // Unpinned rather than 404 for a job that is gone. Reachable only with - // `assets:read`, which is exactly what the unpinned route would have cost - // for the same answer. + // No such job: answer the unpinned graph rather than 404, so a run page whose + // job has aged out of retention still draws the deployed version instead of + // an error. Reachable only with `assets:read`, which is exactly what + // `/assets/graph` would have cost for the same answer. let Some(job) = job else { - return Ok(None); + return windmill_api_assets::asset_graph_for(&authed, &w_id, user_db, db, q, None).await; }; require_job_read_access( - db, - user_db, - authed, - w_id, + &db, + &user_db, + &authed, + &w_id, &job_id, &job.created_by, - view_token, + view_token.as_deref(), ) .await?; // A preview or flow job names no deployed version, so there is usually no // graph to pin to and the workspace one answers. The exception is a job that // parsed one itself, which is what the dbt editor's refresh is: its graph // belongs to that job alone and nothing else can reach it. - Ok(job + let pinned = job .runnable_path .filter(|_| job.script_hash.is_some() || job.editor_graph) .map(|path| windmill_api_assets::PinnedRun { job_id, script_path: path, script_hash: job.script_hash, - })) -} - -/// The asset graph as one run saw it. -async fn get_dbt_run_graph( - authed: ApiAuthed, - OptViewToken(view_token): OptViewToken, - Extension(db): Extension, - Extension(user_db): Extension, - Path((w_id, job_id)): Path<(String, Uuid)>, - Query(q): Query, -) -> error::JsonResult { - let pinned = - dbt_pinned_run(&authed, &db, &user_db, &w_id, job_id, view_token.as_deref()).await?; + }); windmill_api_assets::asset_graph_for(&authed, &w_id, user_db, db, q, pinned).await } -/// One relation's column lineage as one run saw it — the same pin as -/// `get_dbt_run_graph`, for the trace drawn beside a node of that graph. -async fn get_dbt_run_column_lineage( - authed: ApiAuthed, - OptViewToken(view_token): OptViewToken, - Extension(db): Extension, - Extension(user_db): Extension, - Path((w_id, job_id)): Path<(String, Uuid)>, - Query(q): Query, -) -> error::JsonResult { - let pinned = - dbt_pinned_run(&authed, &db, &user_db, &w_id, job_id, view_token.as_deref()).await?; - windmill_api_assets::dbt_column_lineage_for(&authed, &w_id, user_db, q, pinned).await -} - /// Whether a `dbt retry` submitted by this caller would resume THIS run. /// /// One failure is saved per script per execution principal, so a page showing an diff --git a/backend/windmill-worker/src/dbt_column_index.rs b/backend/windmill-worker/src/dbt_column_index.rs index 3a2a4c8a66..b843c95f21 100644 --- a/backend/windmill-worker/src/dbt_column_index.rs +++ b/backend/windmill-worker/src/dbt_column_index.rs @@ -226,6 +226,13 @@ async fn compile_index( // `run_results.json` a `dbt retry` resumes from. cmd.env("DBT_TARGET_PATH", CLL_ARTIFACTS_DIR); crate::dbt_executor::add_vars(&mut cmd, descriptor, inv)?; + // The BUILD's answer, not the descriptor's default: `is_incremental()` + // branches on it, so a model reading `{{ this }}` compiles its self-join — + // and any `ref()` inside that branch — only when this is absent. Guessing + // here stores lineage for SQL the run never executed. + if crate::dbt_executor::full_refresh(descriptor, inv)? { + cmd.arg("--full-refresh"); + } // Captured rather than streamed: a strict-analysis failure is a wall of // diagnostics about SQL the build itself accepts, and this pass decides // nothing about whether that build runs. diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index 12a092ba71..04376dd189 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -877,6 +877,9 @@ impl GraphRefresh { if selection_is_overridden(descriptor, args)? { self.per_run_models = true; } + if full_refresh_is_overridden(descriptor, args)? { + self.per_run_models = true; + } Ok(()) } } @@ -2243,8 +2246,7 @@ async fn run_dbt( if let Some(t) = descriptor.threads { cmd.args(["--threads", &t.to_string()]); } - let full_refresh = arg_bool(&inv.args, "full_refresh")?.unwrap_or(descriptor.full_refresh); - if full_refresh && command != "test" { + if full_refresh(descriptor, inv)? && command != "test" { cmd.arg("--full-refresh"); } } @@ -3184,10 +3186,14 @@ async fn attach_column_index( ) -> error::Result<()> { // The nodes this graph kept, so the pass reads only rows it could store: the // index describes the whole project, this graph one selection of it. - let kept: std::collections::HashSet<&str> = - ingested.nodes.iter().map(|n| n.unique_id.as_str()).collect(); + let kept: std::collections::HashSet<&str> = ingested + .nodes + .iter() + .map(|n| n.unique_id.as_str()) + .collect(); let index = - crate::dbt_column_index::collect(p, descriptor, inv, ctx, job_id, w_id, conn, &kept).await?; + crate::dbt_column_index::collect(p, descriptor, inv, ctx, job_id, w_id, conn, &kept) + .await?; drop(kept); let Some(index) = index else { return Ok(()); @@ -4896,6 +4902,27 @@ fn add_selection( /// /// A run that wants the whole project despite a descriptor selector names a /// selection that differs — `["*"]`. +/// Whether this invocation rebuilds incremental models from scratch: the run +/// form's answer when it gave one, else the descriptor's. +/// +/// Shared with the column-lineage pass rather than recomputed there, because +/// `is_incremental()` branches on it: the same model compiles to different SQL — +/// a `{{ this }}` self-join, and any `ref()` inside the incremental branch — so a +/// pass that guessed would describe a build that never ran. +pub(crate) fn full_refresh(descriptor: &DbtDescriptor, inv: &Invocation) -> error::Result { + Ok(arg_bool(&inv.args, "full_refresh")?.unwrap_or(descriptor.full_refresh)) +} + +/// Whether this run answered `full_refresh` differently from the deployed +/// descriptor. Like a selection override it changes what the graph describes, +/// since an incremental branch can carry its own `ref()`. +fn full_refresh_is_overridden( + descriptor: &DbtDescriptor, + args: &HashMap>, +) -> error::Result { + Ok(arg_bool(args, "full_refresh")?.is_some_and(|v| v != descriptor.full_refresh)) +} + fn selection_is_overridden( descriptor: &DbtDescriptor, args: &HashMap>, @@ -5896,6 +5923,27 @@ mod tests { .unwrap(); assert!(untouched.publishes_ownership()); assert_eq!(untouched.snapshot_job(job), None); + + // `full_refresh` decides whether `is_incremental()` is true, so an + // incremental model's self-join — and any `ref()` inside that branch — + // exists in one answer and not the other. A run that flips it describes + // a different graph, and gets its own. + let mut refreshed = GraphRefresh::default(); + refreshed + .add_caller_args(&descriptor, &arg("full_refresh", "true")) + .unwrap(); + assert!(refreshed.needed()); + assert_eq!(refreshed.snapshot_job(job), Some(job)); + + // The same echo rule: the form posts the descriptor's own value back on + // every run, and reading that as an override would make each one + // caller-scoped. + let always = DbtDescriptor { full_refresh: true, ..Default::default() }; + let mut echoed_flag = GraphRefresh { profile_drift: true, ..Default::default() }; + echoed_flag + .add_caller_args(&always, &arg("full_refresh", "true")) + .unwrap(); + assert_eq!(echoed_flag.snapshot_job(job), None); } // `dbt retry` restores the previous run's target/ from this directory, so two diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 03ad14f2b4..2920227eb1 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1227,44 +1227,26 @@ table is `ref()` lineage. The typed column list lands in `dbt_node.column_schema`, beside `columns` rather than merged into it — `columns` stays what the author *declared*. -**Served from `assets/column_lineage`, keyed to one relation, not as a field on -the asset graph.** The graph is folder-wide and a run page polls it, while a -column trace is drawn for one selected node; carried on the graph the edges would -need a cap, and a cap has to be applied after every filter that can drop a row — -scope, project visibility, the asset set actually rendered. That ordering is what -the separate endpoint removes rather than gets right: here the filters *are* the -answer. The caller's `scripts:read` scope and the project's visibility are decided -once in SQL, for the script that owns the relation; the size is bounded at ingest -(`MAX_COLUMN_EDGES` per version, of which only the direct kinds are served); and -what comes back is the **connected component** the relation's columns sit in, -which is exactly what the canvas lays out. Neither the relation's own edges (a -trace walks transitively, so that stops one hop out) nor the whole project's -(model families the selection cannot reach). The component is walked in Rust over -the rows the gated query returns, not by a recursive CTE: a CTE has no index to -walk, so the recursive term rescans the whole edge set once per level — measured -at 1.24s against 59ms for the query alone on a 3000-model project. Pinning to a -run's snapshot or to the editor's parse of its own buffer costs the job-read gate, -so that form is `jobs/dbt_column_lineage/{id}`, exactly as `jobs/dbt_graph/{id}` -is to `assets/graph`. +**Stored now, served later.** This change lands the ingest and the storage; the +endpoint that draws a column trace is a follow-up. What is user-visible today is +`column_schema` — every column of a relation, typed and in the order the model +emits them — which rides the asset graph the details pane already fetches, and +replaces a panel that could only list the columns an author happened to document. +The edges sit in `dbt_column_edge` waiting for their surface. -The two halves of a column trace are fetched separately and merged in the -browser: the producer half — what a DuckDB script's `// column` annotations and -inferred SQL lineage say — rides on the asset graph, and dbt's rides on this -endpoint. They meet at shared node ids, since `// column total <- -dbt://wh/analytics/orders.amount` mints the same `(dbt, path, column)` node dbt's -own lineage does, so a trace crosses the boundary in both directions rather than -ending at it. +`column_schema` is gated on being able to read the producing project, like the +model's SQL: a column-level view is the shape of what the author wrote, one level +finer than the `ref()` graph, which is ungated only because it draws relations the +caller already sees. A share-link viewer entitled to a dbt run therefore gets its +relations and `ref()` edges, and neither the SQL nor the columns. -Both the lineage and `column_schema` are gated on being able to read the -producing project, like the model's SQL: a column-level view is the shape of what -the author wrote, one level finer than the `ref()` graph, which is ungated only -because it draws relations the caller already sees. **That gate is separate from -the pin**, and the pinned read is where the two are easiest to conflate: a run -resolves WHICH version answers, and never whether the caller may read it. A -share-link viewer entitled to a dbt run gets its relations and `ref()` edges and -an empty lineage, the same split `dbt_graph` already makes by redacting -`raw_code`. The one exemption is a version-less row — an editor buffer, which has -no `script` row to ask and is reachable only through the parse job that wrote it. +**The analysis pass takes the build's own `--full-refresh`.** `is_incremental()` +branches on it, so an incremental model reading `{{ this }}` compiles its +self-join — and any `ref()` inside that branch — only when the flag is absent. A +pass that used the descriptor's default while the run overrode it would store +lineage for SQL that run never executed. For the same reason an invocation that +overrides the flag counts as `per_run_models`: its graph is its own, keyed to the +job, rather than standing as the version's. ## Concept mapping @@ -1278,7 +1260,7 @@ no `script` row to ask and is reachable only through the parse job that wrote it | `unique`/`not_null`/`accepted_values`/`relationships` | `data_tests` | exact 1:1 with the four `// data_test` kinds | | declared column metadata | `columns` on the asset node | descriptions only, from the manifest | | analyzed column schema | `column_schema` on the asset node | `dbt.node_columns.parquet`, opt-in | -| column-to-column lineage | column-lineage trace | `dbt.column_lineage.parquet`, opt-in | +| column-to-column lineage | `dbt_column_edge` rows (no view yet) | `dbt.column_lineage.parquet`, opt-in | | model `tags` | node badge | `tag` | | source freshness | `freshness` | `last_success_at` chip | | `run_results.json` | materialization records | `record_materialization` | diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 4b269142a5..2c5e5fd97e 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -156,9 +156,6 @@ // resolved graph). Drives the transitive column-lineage trace shown for a // selected materialized asset. selectionColumnGraph?: ColumnLineageGraph - /** That graph still being fetched — a dbt relation's lineage is a request - * of its own, so it arrives after the selection does. */ - selectionColumnLoading?: boolean /** dbt provenance of the selected relation, when a dbt project * materializes it — carries the model's own SQL. */ selectionDbt?: DbtAssetProvenance @@ -292,7 +289,6 @@ onScriptRemoved, selectionProducers = [], selectionColumnGraph, - selectionColumnLoading = false, selectionDbt, schemaCanEvolve = true, selectionForkMaterialization = undefined, @@ -450,15 +446,6 @@ return scripts.length === 1 ? `${scripts[0].path}__dbt/${file}` : file }) - // The selected relation's own column nodes, which is what decides whether - // there is a trace to draw at all: a producer that declares no column lineage - // — or a dbt project that never asked for the analysis pass — has none. - let selectionColumnNodes = $derived( - selection?.kind === 'asset' && selectionColumnGraph - ? assetColumnNodes(selectionColumnGraph, selection.asset_kind, selection.path) - : [] - ) - // Bound from ScriptEditor — populated by inferAssets on every code // change. Forwarded to the page so the canvas can re-derive write // edges as the user edits the body (e.g. renaming a CREATE TABLE @@ -1234,7 +1221,7 @@ {/if} - {#if selectionColumnGraph && selectionColumnNodes.length > 0} + {#if selectionColumnGraph && assetColumnNodes(selectionColumnGraph, selection.asset_kind, selection.path).length > 0}
{/key} - {:else if selectionDbt && (selectionDbt.raw_code || selectionColumnNodes.length > 0 || selectionColumnLoading)} + {:else if selectionDbt?.raw_code} - {#if selectionColumnLoading && selectionColumnNodes.length === 0} -
- - Loading column lineage -
- {:else if selectionColumnGraph && selectionColumnNodes.length > 0} -
- -
- {/if} - {#if selectionDbt.raw_code} -
- -
- {/if} {:else}
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte index 4064c183a7..3439fa5ad5 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte @@ -76,7 +76,6 @@ localScriptsVersion, selectionProducers = [], selectionColumnGraph, - selectionColumnLoading = false, selectionDbt, schemaCanEvolve = true, selectionForkMaterialization = undefined, @@ -181,11 +180,8 @@ * the selected node's source on live-reload. */ localScriptsVersion?: unknown selectionProducers?: Array<{ kind: 'script' | 'flow'; path: string; unsaved?: boolean }> - /** Transitive column-lineage trace for the selected asset (route page). */ + /** Transitive column-lineage trace for a selected ducklake asset (route page). */ selectionColumnGraph?: ColumnLineageGraph - /** That trace still being fetched — a dbt relation's is a request of its - * own, so it arrives after the selection does. */ - selectionColumnLoading?: boolean /** dbt provenance of the selected relation — carries its SQL. */ selectionDbt?: DbtAssetProvenance schemaCanEvolve?: boolean @@ -518,7 +514,6 @@ selection={activeDraft ? undefined : editor.selection} selectionProducers={activeDraft ? [] : selectionProducers} {selectionColumnGraph} - {selectionColumnLoading} {selectionDbt} {schemaCanEvolve} {selectionForkMaterialization} diff --git a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts index 16c8713c5c..6842dcfd86 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts @@ -2,10 +2,7 @@ import { describe, expect, it } from 'vitest' import type { AssetGraphResponse } from './types' import { buildColumnGraph, - buildDbtColumnGraph, colNodeId, - mergeColumnGraphs, - type ColumnLineageGraph, traceColumn, connectedComponent, assetColumnNodes, @@ -123,79 +120,6 @@ describe('buildColumnGraph', () => { }) }) -describe('buildDbtColumnGraph', () => { - it('takes the direct kinds and drops any other', () => { - // `scan` means the column was read to produce the ROW — a join key, a - // predicate, a `group by` — so it reaches every output column of its model - // and is not what a column trace means. The server filters it out; this - // filters again, because the kind set is the engine's and an unknown one - // must not become an edge the trace calls data flow. - const g = buildDbtColumnGraph([ - { - from_asset_path: 'main/s/stg', - from_column: 'raw_name', - to_asset_path: 'main/s/mart', - to_column: 'clean_name', - kind: 'mod' - }, - { - from_asset_path: 'main/s/stg', - from_column: 'id', - to_asset_path: 'main/s/mart', - to_column: 'id', - kind: 'copy' - }, - { - from_asset_path: 'main/s/stg', - from_column: 'id', - to_asset_path: 'main/s/mart', - to_column: 'clean_name', - kind: 'scan' - } - ]) - expect(g.up.get(colNodeId('dbt', 'main/s/mart', 'clean_name'))).toEqual( - new Set([colNodeId('dbt', 'main/s/stg', 'raw_name')]) - ) - expect(g.up.get(colNodeId('dbt', 'main/s/mart', 'id'))).toEqual( - new Set([colNodeId('dbt', 'main/s/stg', 'id')]) - ) - }) -}) - -describe('mergeColumnGraphs', () => { - it('chains a dbt column into what a producer derives from it', () => { - // The two halves arrive separately — the producer's from the asset graph, - // dbt's from its own request — and meet at the dbt node a `// column` - // annotation names. A trace has to cross that, or a dbt selection stops - // before the script consuming it. - const dbt = buildDbtColumnGraph([ - { - from_asset_path: 'main/s/stg', - from_column: 'raw', - to_asset_path: 'main/s/mart', - to_column: 'clean', - kind: 'copy' - } - ]) - const producer: ColumnLineageGraph = { - nodes: new Map(), - up: new Map(), - down: new Map() - } - const src = colNodeId('dbt', 'main/s/mart', 'clean') - const out = colNodeId('ducklake', 'wh/report', 'total') - producer.nodes.set(src, { kind: 'dbt', path: 'main/s/mart', column: 'clean' }) - producer.nodes.set(out, { kind: 'ducklake', path: 'wh/report', column: 'total' }) - producer.up.set(out, new Set([src])) - producer.down.set(src, new Set([out])) - - const merged = mergeColumnGraphs(dbt, producer) - expect(traceColumn(colNodeId('dbt', 'main/s/stg', 'raw'), merged)).toEqual( - new Set([colNodeId('dbt', 'main/s/stg', 'raw'), src, out]) - ) - }) -}) - describe('traceColumn', () => { it('returns the full upstream + downstream impact set of a source column', () => { const g = buildColumnGraph(chainGraph()) diff --git a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts index 7172ec52e9..50fa1c4e7b 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts @@ -1,11 +1,6 @@ -import type { AssetKind, DbtColumnLineage } from '$lib/gen' +import type { AssetKind } from '$lib/gen' import type { AssetGraphResponse } from './types' -// One column-to-column edge of a dbt project's static analysis, as the API -// serves it. Taken from the generated client rather than restated: unlike the -// asset graph, this response is fetched through it. -export type DbtColumnEdge = DbtColumnLineage['edges'][number] - // A node in the column-level lineage graph: one column of one asset. export type ColumnNode = { kind: AssetKind; path: string; column: string } export type ColumnNodeId = string @@ -28,15 +23,6 @@ export type ColumnLineageGraph = { down: Map> } -// Direct value flow, as dbt's static analysis labels it: `copy` passes a column -// through, `mod` transforms it. The API serves only those two — the third kind, -// `scan`, means the column was read to produce the ROW rather than the value (a -// join key, a `where` predicate, a `group by`), so it reaches EVERY output -// column of the model and would draw the diagram as a complete bipartite graph. -// Filtered here as well so a kind the engine invents cannot silently become an -// edge the trace claims is data flow. -const DIRECT_DBT_LINEAGE = new Set(['copy', 'mod']) - // Build the column graph from a resolved asset graph. A producer's // `column_lineage` describes the columns of the asset it materializes; that // output asset is the ducklake target it writes (v1 materialize target), found @@ -103,59 +89,6 @@ export function buildColumnGraph(graph: AssetGraphResponse): ColumnLineageGraph return { nodes, up, down } } -// The same graph, from one dbt relation's column lineage. dbt arrives already -// resolved to two relations rather than anchored to a producer, and the API -// serves only the direct kinds, so this is a straight edge list. -export function buildDbtColumnGraph(edges: DbtColumnEdge[]): ColumnLineageGraph { - const nodes = new Map() - const up = new Map>() - const down = new Map>() - const addNode = (n: ColumnNode): ColumnNodeId => { - const id = colNodeId(n.kind, n.path, n.column) - if (!nodes.has(id)) nodes.set(id, n) - return id - } - for (const e of edges) { - // Belt and braces: the API filters to `copy`/`mod`, and a kind an engine - // invents must not silently become an edge the trace calls data flow. - if (!DIRECT_DBT_LINEAGE.has(e.kind)) continue - const src = addNode({ kind: 'dbt', path: e.from_asset_path, column: e.from_column }) - const out = addNode({ kind: 'dbt', path: e.to_asset_path, column: e.to_column }) - if (src === out) continue - ;(up.get(out) ?? up.set(out, new Set()).get(out)!).add(src) - ;(down.get(src) ?? down.set(src, new Set()).get(src)!).add(out) - } - return { nodes, up, down } -} - -// One graph out of several, so a trace crosses the boundary between them. -// -// The two halves reach each other through shared node ids: a producer's -// `// column out <- dbt://wh/schema/model.col` puts a `('dbt', path, column)` -// node in the producer graph under the same `colNodeId` the dbt lineage mints -// for it, so the union chains a dbt model's columns into the script that -// consumes them and on into what that script writes. Kept separate up to here -// because they are fetched separately — the producer half rides on the asset -// graph, the dbt half is asked for per selection. -export function mergeColumnGraphs(...graphs: ColumnLineageGraph[]): ColumnLineageGraph { - const nodes = new Map() - const up = new Map>() - const down = new Map>() - for (const g of graphs) { - for (const [id, n] of g.nodes) if (!nodes.has(id)) nodes.set(id, n) - for (const [dir, into] of [ - [g.up, up], - [g.down, down] - ] as const) { - for (const [id, adj] of dir) { - const target = into.get(id) ?? into.set(id, new Set()).get(id)! - for (const m of adj) target.add(m) - } - } - } - return { nodes, up, down } -} - // Every node reachable from `start` by following `adj` (transitive closure, // excluding `start` itself). Iterative to avoid deep-recursion limits. function reach(start: ColumnNodeId, adj: Map>): Set { diff --git a/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts deleted file mode 100644 index 386399824d..0000000000 --- a/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts +++ /dev/null @@ -1,132 +0,0 @@ -import { untrack } from 'svelte' -import { AssetService, JobService } from '$lib/gen' -import { - buildDbtColumnGraph, - mergeColumnGraphs, - type ColumnLineageGraph -} from './columnLineageGraph' - -export const EMPTY_COLUMN_GRAPH: ColumnLineageGraph = { - nodes: new Map(), - up: new Map(), - down: new Map() -} - -/** Which stored dbt graph a view is drawing. A job — a run's snapshot, or the - * editor's parse of its own buffer — is asked through the job route, the only - * way to reach a graph that names no deployed version; otherwise the deployed - * version by hash, or the current one when there is no hash. */ -export type DbtGraphPin = { jobId?: string; scriptHash?: string | number } - -/** What a dbt relation's column lineage is doing right now. `loading` is - * separate because a project still being fetched and one that never asked for - * the analysis pass are the same empty graph otherwise. */ -export type DbtColumnLineageState = { - readonly graph: ColumnLineageGraph - readonly loading: boolean -} - -function pinKey(workspace: string, pin: DbtGraphPin | undefined): string { - return `${workspace}|${pin?.jobId ?? ''}|${pin?.scriptHash ?? ''}` -} - -function fetchComponent( - workspace: string, - assetPath: string, - pin: DbtGraphPin | undefined -): Promise { - const req = pin?.jobId - ? JobService.getDbtRunColumnLineage({ workspace, id: pin.jobId, assetPath }) - : AssetService.getDbtColumnLineage({ - workspace, - assetPath, - dbtScriptHash: pin?.scriptHash != undefined ? String(pin.scriptHash) : undefined - }) - return req.then( - (r) => buildDbtColumnGraph(r?.edges ?? []), - // Lineage annotates a graph that renders without it, so a failed fetch - // leaves that branch unexpanded rather than putting an error over the - // model — and one failed boundary does not lose the others. - () => EMPTY_COLUMN_GRAPH - ) -} - -/** Follow the selection, fetching the dbt column lineage it reaches. - * - * Per asset rather than off the graph response: the graph is folder-wide and a - * run page polls it, while this is drawn for one selection. It also means the - * request is never made for a project that did not opt into the analysis pass - * — the pane simply never shows the section. - */ -export function useDbtColumnLineage(args: { - workspace: () => string | undefined - /** The dbt relations to expand. The selection itself when it is one; for a - * selection of another kind, every dbt relation its own lineage reaches — - * a ducklake table can be derived from several, and expanding only the - * first would leave the rest as leaves. */ - assetPaths: () => string[] - /** The graph on screen, so the lineage describes the same project. */ - pin?: () => DbtGraphPin | undefined -}): DbtColumnLineageState { - let graph = $state(EMPTY_COLUMN_GRAPH) - let loading = $state(false) - - // Which pin the graph in hand was fetched against, and which relations were - // actually ASKED about under it. - let heldPin: string | undefined = undefined - let asked = new Set() - - $effect(() => { - const workspace = args.workspace() - const paths = args.assetPaths() - const pin = args.pin?.() - if (!workspace || paths.length === 0) { - graph = EMPTY_COLUMN_GRAPH - heldPin = undefined - asked = new Set() - loading = false - return - } - const key = pinKey(workspace, pin) - // `untrack`: this effect writes `graph`, so reading it as a dependency - // would make it retrigger itself forever. - const fresh = heldPin !== key - const base = untrack(() => (fresh ? EMPTY_COLUMN_GRAPH : graph)) - if (fresh) asked = new Set() - // Only a relation this pin has ASKED about is skipped, not every relation - // present in what came back. A relation two projects describe has an owner - // row in each, and a component fetched for one of them carries that - // relation as an endpoint without the other project's half — so treating - // "appears in the graph" as "resolved" would hide exactly the cross-project - // edges the server's relation-keyed walk exists to merge. - const missing = paths.filter((p) => !asked.has(p)) - if (missing.length === 0) { - graph = base - loading = false - return - } - // A selection changes faster than a request completes, so an answer is - // applied only while it is still the one being asked for. - let current = true - loading = true - Promise.all(missing.map((p) => fetchComponent(workspace, p, pin))).then((parts) => { - if (!current) return - graph = mergeColumnGraphs(base, ...parts) - heldPin = key - for (const p of missing) asked.add(p) - loading = false - }) - return () => { - current = false - } - }) - - return { - get graph() { - return graph - }, - get loading() { - return loading - } - } -} diff --git a/frontend/src/lib/components/dbt/DbtEditor.svelte b/frontend/src/lib/components/dbt/DbtEditor.svelte index ffe2ac9540..1d3de73e58 100644 --- a/frontend/src/lib/components/dbt/DbtEditor.svelte +++ b/frontend/src/lib/components/dbt/DbtEditor.svelte @@ -8,7 +8,7 @@ // single file, so the arguments, the run and the graph are all the project's // whichever file happens to be open. import { untrack } from 'svelte' - import { createEventDispatcher, onDestroy } from 'svelte' + import { createEventDispatcher, onDestroy, onMount } from 'svelte' import type { Schema, SupportedLanguage } from '$lib/common' import type { Preview, ScriptModule } from '$lib/gen' import { workspaceStore } from '$lib/stores' @@ -31,15 +31,6 @@ AssetGraphNodeData, DbtAssetProvenance } from '$lib/components/assets/AssetGraph/types' - import { - EMPTY_COLUMN_GRAPH, - useDbtColumnLineage, - type DbtGraphPin - } from '$lib/components/assets/AssetGraph/dbtColumnLineage.svelte' - import { - mergeColumnGraphs, - type ColumnLineageGraph - } from '$lib/components/assets/AssetGraph/columnLineageGraph' import { DBT_DESCRIPTOR, DBT_MODULE_EXTENSIONS, @@ -106,6 +97,17 @@ } }) + // The stored schema can predate the parser (a CLI push, an older version), and + // the autosave baseline is taken as the mounted editor holds it + // (`schemaAsEditorMounts`), so the descriptor is re-inferred on mount rather + // than only on its first edit — otherwise the two only agree once edited. + onMount(async () => { + await inferSchema(code) + // Same single retry as ScriptEditor: the first parse can lose a transient + // wasm init race, and the baseline (`schemaAsEditorMounts`) retries too. + if (!validDescriptor && code) await inferSchema(code) + }) + function flushOpenFile() { if (openFile !== null && modules?.[openFile]) { modules[openFile] = { ...modules[openFile], content: editorCode } @@ -216,28 +218,6 @@ let graphSelection = $state(undefined) let selectedAsset = $derived(graphSelection?.kind === 'asset' ? graphSelection : undefined) let selectedDbt = $state(undefined) - // Which graph the selection came from, so the lineage fetched below is the - // selected node's own project rather than whatever is deployed. - let selectionPin = $state(undefined) - // The selected model's column lineage, fetched on selection. Its own request - // rather than a field on the graph: only a project that opted into the - // analysis pass has any, and it is drawn for one model at a time. - const columnLineage = useDbtColumnLineage({ - workspace: () => opWs, - assetPaths: () => { - const path = selectedDbt ? selectedAsset?.path : undefined - return path ? [path] : [] - }, - pin: () => selectionPin - }) - // What the scripts around this project declare about its columns, off the - // same graph response the canvas drew. Merged rather than chosen between: a - // model's column and the ducklake column a script derives from it are one - // chain, and the trace has to cross that boundary. - let selectionProducerColumns = $state(EMPTY_COLUMN_GRAPH) - let selectionColumnGraph = $derived( - mergeColumnGraphs(columnLineage.graph, selectionProducerColumns) - ) // Set when the selected node came from a buffer parse: the project that parse // ran on, which is the one its rows must come from. Undefined for a node off // the deployed graph, which previews by version instead. Either way the rows @@ -545,12 +525,10 @@ testRunning={testIsLoading} testResult={testJob?.result} selection={graphSelection} - onSelect={(sel, dbt, buffer, pin, producerColumns) => { + onSelect={(sel, dbt, buffer) => { graphSelection = sel selectedDbt = dbt selectedBuffer = buffer - selectionPin = pin - selectionProducerColumns = producerColumns }} /> @@ -572,8 +550,6 @@ {args} fileInBundle={!!selectedDbt.original_file_path && !!modules?.[selectedDbt.original_file_path]} - columnGraph={selectionColumnGraph} - columnLoading={columnLineage.loading} onOpenFile={open} onClose={() => (graphSelection = undefined)} /> diff --git a/frontend/src/lib/components/dbt/DbtModelDetails.svelte b/frontend/src/lib/components/dbt/DbtModelDetails.svelte index a1b3629d35..52739ca39c 100644 --- a/frontend/src/lib/components/dbt/DbtModelDetails.svelte +++ b/frontend/src/lib/components/dbt/DbtModelDetails.svelte @@ -12,11 +12,6 @@ import { ClipboardCopy, Code2, FileCode2, Loader2, TableProperties, X } from 'lucide-svelte' import { copyToClipboard } from '$lib/utils' import type { DbtAssetProvenance } from '$lib/components/assets/AssetGraph/types' - import ColumnLineageTrace from '$lib/components/assets/AssetGraph/ColumnLineageTrace.svelte' - import { - assetColumnNodes, - type ColumnLineageGraph - } from '$lib/components/assets/AssetGraph/columnLineageGraph' import { previewDbtRows, type DbtPreview, type DbtPreviewBuffer } from './previewRows' import { nodeSelector } from './parseDbtRun' @@ -39,14 +34,6 @@ args, /** Whether this model's file is in the bundle being edited. */ fileInBundle = false, - /** The project's column-level lineage, when the descriptor asked for it. - * Fetched for this relation against the same graph the canvas draws, so - * the trace and the nodes above it describe one parse. */ - columnGraph, - /** That fetch still in flight. Distinguished from an empty graph: a - * project without the analysis pass shows nothing at all, and a slow - * answer must not read as that. */ - columnLoading = false, onOpenFile, onClose }: { @@ -58,8 +45,6 @@ buffer?: DbtPreviewBuffer args?: Record fileInBundle?: boolean - columnGraph?: ColumnLineageGraph - columnLoading?: boolean onOpenFile?: (path: string) => void onClose?: () => void } = $props() @@ -144,9 +129,6 @@ })) ) let columnsAreAnalyzed = $derived(!!dbt.column_schema?.length) - // The selected relation's own column nodes: empty for a project that never - // asked for the analysis pass, which is the ordinary case. - let columnNodes = $derived(columnGraph ? assetColumnNodes(columnGraph, 'dbt', assetPath) : []) // `dbt show` SELECTs from the node's own relation and the worker intersects // the selector with `resource_type:model`, so offering it on a seed, snapshot // or source only ever produces a failed job. @@ -313,22 +295,6 @@
{/if} - {#if columnLoading && columnNodes.length === 0} -
- - Loading column lineage -
- {:else if columnGraph && columnNodes.length > 0} -
- -
- {/if} - {#if showRows && preview} {#if 'error' in preview}
{preview.error}
diff --git a/frontend/src/lib/components/dbt/DbtModelGraph.svelte b/frontend/src/lib/components/dbt/DbtModelGraph.svelte index 702c8242f7..e6d948d926 100644 --- a/frontend/src/lib/components/dbt/DbtModelGraph.svelte +++ b/frontend/src/lib/components/dbt/DbtModelGraph.svelte @@ -25,14 +25,6 @@ DbtAssetProvenance } from '$lib/components/assets/AssetGraph/types' import { useDbtRunStatus } from './runStatus.svelte' - import { - EMPTY_COLUMN_GRAPH, - type DbtGraphPin - } from '$lib/components/assets/AssetGraph/dbtColumnLineage.svelte' - import { - buildColumnGraph, - type ColumnLineageGraph - } from '$lib/components/assets/AssetGraph/columnLineageGraph' let { workspace, @@ -88,19 +80,7 @@ * buffer rather than a deployed version — as submitted, not as the * editor holds it now. Sent with the selection rather than exposed on * its own so it can never disagree with the SQL the parent shows. */ - buffer: DbtPreviewBuffer | undefined, - /** Which graph this node was taken from, so anything else fetched - * about it describes the same project: the editor's own parse job - * when the panel is pinned to one, else the deployed version. Sent - * with the selection for the same reason the buffer is — it must not - * be able to disagree with the node on screen. */ - pin: DbtGraphPin, - /** Column lineage the CONSUMERS of this project declare — a script - * reading a model's column and writing a ducklake one. It comes off - * the same graph response, and the details pane merges it with the - * project's own so a trace crosses that boundary instead of ending - * at it. */ - producerColumns: ColumnLineageGraph + buffer: DbtPreviewBuffer | undefined ) => void } = $props() @@ -384,19 +364,6 @@ // graph that actually came back. let editorParsed = $derived(refreshJob != undefined && raw?.dbt_snapshot_job === refreshJob) - // Which stored graph is on screen. Anything the details pane fetches about a - // selected node asks for this one, so it cannot describe a node parsed from - // the buffer with the deployed version's answer. - let pin = $derived( - editorParsed && refreshJob ? { jobId: refreshJob } : { scriptHash: deployedHash } - ) - - // What the scripts around this project declare about its columns. Empty for - // the ordinary project nothing downstream annotates. - let producerColumns = $derived( - graph ? buildColumnGraph(graph) : EMPTY_COLUMN_GRAPH - ) - // `untrack`, because the effect that reloads the graph clears the selection // through here: reading the graph to describe a selection would subscribe that // effect to the very state its own fetch writes, and it would reload forever. @@ -407,9 +374,7 @@ sel?.kind === 'asset' ? graph?.assets.find((a) => a.kind === sel.asset_kind && a.path === sel.path)?.dbt : undefined, - editorParsed ? parsedBuffer : undefined, - pin, - producerColumns + editorParsed ? parsedBuffer : undefined ) ) } @@ -440,6 +405,7 @@ if (deployedHash != undefined) return 'as of last deploy' return 'never parsed' }) +
@@ -469,8 +435,8 @@ {#if refreshPending}
- Still parsing. A cold worker provisions the dbt engine before it starts; a project pinned to a - worker tag nothing serves waits here indefinitely. + Still parsing. A cold worker provisions the dbt engine before it starts; a project + pinned to a worker tag nothing serves waits here indefinitely. { - const sel = pe.selection - if (pe.activeDraft || sel?.kind !== 'asset') return [] - if (sel.asset_kind === 'dbt') return [sel.path] - // EVERY dbt relation this selection reaches, not the first: one output can - // be derived from several, and expanding one would leave the others as - // leaves on the canvas. - const seeds = assetColumnNodes(producerColumnGraph, sel.asset_kind, sel.path) - const paths = new Set() - for (const id of connectedComponent(seeds, producerColumnGraph)) { - const node = producerColumnGraph.nodes.get(id) - if (node?.kind === 'dbt') paths.add(node.path) - } - return [...paths] - }) - const dbtColumnLineage = useDbtColumnLineage({ - workspace: () => $workspaceStore, - assetPaths: () => dbtSeedPaths - }) - // One graph across both, so a trace crosses the dbt/ducklake boundary in - // either direction rather than stopping at it. - let columnGraph = $derived(mergeColumnGraphs(producerColumnGraph, dbtColumnLineage.graph)) // Producer-side facts for the editor's live schema-contract diagnostics: // which assets are muted (`on_schema_change=ignore`) and which `_current` @@ -2633,7 +2602,6 @@ {selectionProducers} {selectionDbt} selectionColumnGraph={pe.activeDraft ? EMPTY_COLUMN_GRAPH : columnGraph} - selectionColumnLoading={dbtColumnLineage.loading} {schemaCanEvolve} {selectionForkMaterialization} {schemaContractContext}