diff --git a/backend/.sqlx/query-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json b/backend/.sqlx/query-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json deleted file mode 100644 index 29118c1893..0000000000 --- a/backend/.sqlx/query-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json +++ /dev/null @@ -1,53 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH RECURSIVE\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 -- The owning project's direct edges, resolved to relations once.\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 edge AS (\n SELECT e.script_path, e.script_hash, e.job_id, e.lineage_kind,\n e.parent_unique_id, e.parent_column, p.asset_path AS from_path,\n e.child_unique_id, e.child_column, c.asset_path AS to_path\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\n ),\n -- Both directions of every edge. A trace walks up AND down, and a\n -- recursive term may reference the working table only once, so the\n -- symmetry has to live here rather than in two recursive branches.\n adj AS (\n SELECT script_path, script_hash, job_id,\n parent_unique_id AS a_uid, parent_column AS a_col, from_path AS a_path,\n child_unique_id AS b_uid, child_column AS b_col\n FROM edge\n UNION ALL\n SELECT script_path, script_hash, job_id,\n child_unique_id, child_column, to_path,\n parent_unique_id, parent_column\n FROM edge\n ),\n -- Every column the asked-for relation's columns can reach, either\n -- way. This is exactly what the canvas draws — it lays out the\n -- connected component of the selected relation's columns — so\n -- answering with the whole project would send edges no consumer can\n -- render. The project key travels along: two projects can describe\n -- one relation, and dbt's node ids are per project, so a shared\n -- `unique_id` must not walk from one project's graph into another's.\n reach AS (\n SELECT script_path, script_hash, job_id, a_uid AS uid, a_col AS col\n FROM adj WHERE a_path = $2\n UNION\n SELECT a.script_path, a.script_hash, a.job_id, a.b_uid, a.b_col\n FROM adj a\n JOIN reach r ON r.script_path = a.script_path\n AND r.script_hash IS NOT DISTINCT FROM a.script_hash\n AND r.job_id = a.job_id\n AND r.uid = a.a_uid AND r.col = a.a_col\n )\n -- One endpoint in the component puts the other there too, so matching\n -- the parent alone is the whole component and matches each edge once.\n SELECT e.from_path AS \"from_path!\", e.parent_column AS \"from_column!\",\n e.to_path AS \"to_path!\", e.child_column AS \"to_column!\",\n e.lineage_kind AS \"kind!\"\n FROM edge e\n JOIN reach r ON r.script_path = e.script_path\n AND r.script_hash IS NOT DISTINCT FROM e.script_hash\n AND r.job_id = e.job_id\n AND r.uid = e.parent_unique_id AND r.col = e.parent_column", - "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": "6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6" -} diff --git a/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json b/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json new file mode 100644 index 0000000000..07694284d8 --- /dev/null +++ b/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json @@ -0,0 +1,53 @@ +{ + "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/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index c6b7336a8e..08b74c0311 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -6,6 +6,7 @@ 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, @@ -976,21 +977,23 @@ struct DbtColumnLineageEdge { kind: String, } -/// The column-level lineage of the dbt project one relation belongs to. +/// One dbt relation's column lineage: the connected component its columns sit +/// in, within the project that owns it. /// -/// The PROJECT's, not the relation's own edges: a trace walks transitively, so -/// stopping at the asked-for relation would cut every hop past its neighbours. -/// The asset is what the answer is keyed BY — it names the project and the -/// version — not what it is filtered to. +/// 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. /// -/// Keyed that way rather than carried on the graph, which is folder-wide and +/// 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 -/// needs a bound, and a bound has to be applied after every filter that could -/// drop a row. One project's is already bounded where it is written -/// (`MAX_COLUMN_EDGES` per version, of which only the direct kinds are served), -/// so there is nothing here for a filter to be on the wrong side of: scope and -/// visibility are decided once, for the script that owns the relation. +/// 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. @@ -1003,9 +1006,9 @@ pub struct ColumnLineageQuery { #[derive(Serialize, Debug, PartialEq, Eq, PartialOrd, Ord)] pub struct ColumnLineageResponse { - /// Direct (`copy` / `mod`) column edges of the project this relation belongs - /// to, in the terms the canvas draws. Empty when the project never asked for - /// the analysis pass, which is the ordinary case. + /// 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, } @@ -1050,7 +1053,7 @@ pub async fn dbt_column_lineage_for( 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 RECURSIVE + 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 @@ -1110,77 +1113,31 @@ pub async fn dbt_column_lineage_for( AND sc.deleted = false AND sc.archived = false ORDER BY sc.created_at DESC LIMIT 1) END - ), - -- The owning project's direct edges, resolved to relations once. - -- + ) -- 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. - edge AS ( - SELECT e.script_path, e.script_hash, e.job_id, e.lineage_kind, - e.parent_unique_id, e.parent_column, p.asset_path AS from_path, - e.child_unique_id, e.child_column, c.asset_path AS to_path - 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 - ), - -- Both directions of every edge. A trace walks up AND down, and a - -- recursive term may reference the working table only once, so the - -- symmetry has to live here rather than in two recursive branches. - adj AS ( - SELECT script_path, script_hash, job_id, - parent_unique_id AS a_uid, parent_column AS a_col, from_path AS a_path, - child_unique_id AS b_uid, child_column AS b_col - FROM edge - UNION ALL - SELECT script_path, script_hash, job_id, - child_unique_id, child_column, to_path, - parent_unique_id, parent_column - FROM edge - ), - -- Every column the asked-for relation's columns can reach, either - -- way. This is exactly what the canvas draws — it lays out the - -- connected component of the selected relation's columns — so - -- answering with the whole project would send edges no consumer can - -- render. The project key travels along: two projects can describe - -- one relation, and dbt's node ids are per project, so a shared - -- `unique_id` must not walk from one project's graph into another's. - reach AS ( - SELECT script_path, script_hash, job_id, a_uid AS uid, a_col AS col - FROM adj WHERE a_path = $2 - UNION - SELECT a.script_path, a.script_hash, a.job_id, a.b_uid, a.b_col - FROM adj a - JOIN reach r ON r.script_path = a.script_path - AND r.script_hash IS NOT DISTINCT FROM a.script_hash - AND r.job_id = a.job_id - AND r.uid = a.a_uid AND r.col = a.a_col - ) - -- One endpoint in the component puts the other there too, so matching - -- the parent alone is the whole component and matches each edge once. - SELECT e.from_path AS "from_path!", e.parent_column AS "from_column!", - e.to_path AS "to_path!", e.child_column AS "to_column!", + 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 edge e - JOIN reach r ON r.script_path = e.script_path - AND r.script_hash IS NOT DISTINCT FROM e.script_hash - AND r.job_id = e.job_id - AND r.uid = e.parent_unique_id AND r.col = e.parent_column"#, + 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, @@ -1207,7 +1164,61 @@ pub async fn dbt_column_lineage_for( // Two projects can describe one relation, so the same edge can arrive twice. edges.sort(); edges.dedup(); - Ok(Json(ColumnLineageResponse { edges })) + 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( diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index b1c8a02c6f..ad8ac79540 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -24189,22 +24189,22 @@ paths: /w/{workspace}/assets/column_lineage: get: - summary: Column-level lineage of the dbt project a relation belongs to + summary: Column-level lineage of one dbt relation description: > - The direct (`copy` / `mod`) column-to-column lineage of the dbt project - the given relation belongs to, from the engine's static analysis. The - project's, not the relation's own edges: a column trace walks - transitively, so stopping at the asked-for relation would cut every hop - past its neighbours. The asset names the project and the version — it is - what the answer is keyed by, not what it is filtered to. + 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 needs a cap, and a cap has to come after - every filter; one project's is already bounded where it is written, so - the caller's scope and the project's visibility are simply decided once - for the owning script. + 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` @@ -26062,9 +26062,9 @@ components: DbtColumnLineage: type: object description: >- - The direct column-to-column lineage of the dbt project one relation - belongs to, in the terms the canvas draws — relations and columns, never - dbt's node ids. + 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: diff --git a/backend/windmill-worker/src/dbt_column_index.rs b/backend/windmill-worker/src/dbt_column_index.rs index 15d980397c..3a2a4c8a66 100644 --- a/backend/windmill-worker/src/dbt_column_index.rs +++ b/backend/windmill-worker/src/dbt_column_index.rs @@ -376,18 +376,20 @@ fn read_index_blocking( // budget's worth however the kinds are distributed. let held = out.edges.len() + scan.len(); if is_direct(&edge.lineage_kind) { - // Full of the kind that displaces the other: nothing later in the - // file can be kept, so this is where the read ends. - if out.edges.len() >= MAX_COLUMN_EDGES { - return ControlFlow::Break(()); - } // A direct edge displaces a `scan` one: the budget exists to be // spent on what a trace draws. if held >= MAX_COLUMN_EDGES { scan.pop(); } out.edges.push(edge); - return ControlFlow::Continue(()); + // The edge that FILLS the budget ends the read, not the next one to + // arrive: once the displacing kind is full nothing later in the file + // can be kept, and waiting for another direct edge to say so decodes + // a `scan`-only tail all the way to the backstop for nothing. + return match out.edges.len() >= MAX_COLUMN_EDGES { + true => ControlFlow::Break(()), + false => ControlFlow::Continue(()), + }; } if held < MAX_COLUMN_EDGES { scan.push(edge); diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 9c1fed03c1..03ad14f2b4 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1231,17 +1231,29 @@ table is `ref()` lineage. The typed column list lands in 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. Keyed to the asset -there is no cap for a filter to sit on the wrong side of: the answer is one -project's, already bounded where it is written (`MAX_COLUMN_EDGES` per version, -of which only the direct kinds are served), and the caller's `scripts:read` scope -and the project's visibility are decided once, for the script that owns the -relation. The asset names the project and the version rather than filtering the -edges — a trace walks transitively, so an answer cut to the selected relation's -own edges would stop one hop out. 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`. +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`. + +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. 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 diff --git a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts index 4348bcf728..16c8713c5c 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts @@ -4,6 +4,8 @@ import { buildColumnGraph, buildDbtColumnGraph, colNodeId, + mergeColumnGraphs, + type ColumnLineageGraph, traceColumn, connectedComponent, assetColumnNodes, @@ -160,6 +162,40 @@ describe('buildDbtColumnGraph', () => { }) }) +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 c4396d6380..7172ec52e9 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts @@ -128,6 +128,34 @@ export function buildDbtColumnGraph(edges: DbtColumnEdge[]): ColumnLineageGraph 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 index 7c33dd1c85..0afef7c27d 100644 --- a/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts +++ b/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts @@ -38,6 +38,10 @@ export function useDbtColumnLineage(args: { let graph = $state(EMPTY_COLUMN_GRAPH) let loading = $state(false) + // What the graph in hand describes, so a selection already inside it can be + // recognised without asking again. + let held: { workspace: string; pin: string } | undefined = undefined + $effect(() => { const workspace = args.workspace() const assetPath = args.assetPath() @@ -49,6 +53,20 @@ export function useDbtColumnLineage(args: { loading = false return } + // The answer is one connected component, so every relation inside the one + // already held has the same answer — which is most clicks, since a + // project's models are connected by construction. Keyed to the graph the + // component was fetched against: the same relation under a different pin + // is a different project. + const key = `${workspace}|${jobId ?? ''}|${scriptHash ?? ''}` + if (held?.workspace === workspace && held.pin === key) { + for (const n of graph.nodes.values()) { + if (n.path === assetPath) { + 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 @@ -64,6 +82,7 @@ export function useDbtColumnLineage(args: { (r) => { if (!current) return graph = buildDbtColumnGraph(r?.edges ?? []) + held = { workspace, pin: key } loading = false }, () => { @@ -71,6 +90,7 @@ export function useDbtColumnLineage(args: { // fetch shows no section rather than an error over the model. if (!current) return graph = EMPTY_COLUMN_GRAPH + held = undefined loading = false } ) diff --git a/frontend/src/lib/components/dbt/DbtEditor.svelte b/frontend/src/lib/components/dbt/DbtEditor.svelte index f72e0d25f8..621c0e61ac 100644 --- a/frontend/src/lib/components/dbt/DbtEditor.svelte +++ b/frontend/src/lib/components/dbt/DbtEditor.svelte @@ -32,9 +32,14 @@ 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, @@ -222,6 +227,14 @@ assetPath: () => (selectedDbt ? selectedAsset?.path : undefined), 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 @@ -529,11 +542,12 @@ testRunning={testIsLoading} testResult={testJob?.result} selection={graphSelection} - onSelect={(sel, dbt, buffer, pin) => { + onSelect={(sel, dbt, buffer, pin, producerColumns) => { graphSelection = sel selectedDbt = dbt selectedBuffer = buffer selectionPin = pin + selectionProducerColumns = producerColumns }} /> @@ -555,7 +569,7 @@ {args} fileInBundle={!!selectedDbt.original_file_path && !!modules?.[selectedDbt.original_file_path]} - columnGraph={columnLineage.graph} + columnGraph={selectionColumnGraph} columnLoading={columnLineage.loading} onOpenFile={open} onClose={() => (graphSelection = undefined)} diff --git a/frontend/src/lib/components/dbt/DbtModelGraph.svelte b/frontend/src/lib/components/dbt/DbtModelGraph.svelte index 3c4b367090..702c8242f7 100644 --- a/frontend/src/lib/components/dbt/DbtModelGraph.svelte +++ b/frontend/src/lib/components/dbt/DbtModelGraph.svelte @@ -25,7 +25,14 @@ DbtAssetProvenance } from '$lib/components/assets/AssetGraph/types' import { useDbtRunStatus } from './runStatus.svelte' - import type { DbtGraphPin } from '$lib/components/assets/AssetGraph/dbtColumnLineage.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, @@ -87,7 +94,13 @@ * 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 + 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 ) => void } = $props() @@ -378,6 +391,12 @@ 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. @@ -389,7 +408,8 @@ ? graph?.assets.find((a) => a.kind === sel.asset_kind && a.path === sel.path)?.dbt : undefined, editorParsed ? parsedBuffer : undefined, - pin + pin, + producerColumns ) ) } diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index 4ce94f9b6c..ac9eb1b2ab 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -32,7 +32,12 @@ import PipelineModeToggle from '$lib/components/assets/AssetGraph/PipelineModeToggle.svelte' import MacroExplorerDrawer from '$lib/components/assets/AssetGraph/MacroExplorerDrawer.svelte' import { parsePipelineAnnotations } from '$lib/components/assets/AssetGraph/parsePipelineAnnotations' - import { buildColumnGraph } from '$lib/components/assets/AssetGraph/columnLineageGraph' + import { + assetColumnNodes, + buildColumnGraph, + connectedComponent, + mergeColumnGraphs + } from '$lib/components/assets/AssetGraph/columnLineageGraph' import { EMPTY_COLUMN_GRAPH, useDbtColumnLineage @@ -1988,8 +1993,9 @@ // overlays in edit / show-drafts, deployed-only in plain View. Gated to a // ducklake selection so it isn't rebuilt on every editor keystroke when the // trace UI isn't even shown. - let ducklakeColumnGraph = $derived( - pe.selection?.kind === 'asset' && pe.selection.asset_kind === 'ducklake' + let producerColumnGraph = $derived( + pe.selection?.kind === 'asset' && + (pe.selection.asset_kind === 'ducklake' || pe.selection.asset_kind === 'dbt') ? buildColumnGraph(displayGraph) : EMPTY_COLUMN_GRAPH ) @@ -1997,18 +2003,30 @@ // is stored per relation and only exists if the descriptor asked for it, so // the folder-wide graph does not carry it. A draft is never asked about — // nothing has parsed it, so there is nothing to fetch. + // + // The relation to ask about is the selected one when it IS a dbt relation, + // and otherwise the dbt column a producer feeding this selection names as a + // source — the boundary node above. Asking there is what lets a ducklake + // selection trace back up the dbt project that fed it, rather than stopping + // at the annotation. + let dbtSeedPath = $derived.by(() => { + const sel = pe.selection + if (pe.activeDraft || sel?.kind !== 'asset') return undefined + if (sel.asset_kind === 'dbt') return sel.path + const seeds = assetColumnNodes(producerColumnGraph, sel.asset_kind, sel.path) + for (const id of connectedComponent(seeds, producerColumnGraph)) { + const node = producerColumnGraph.nodes.get(id) + if (node?.kind === 'dbt') return node.path + } + return undefined + }) const dbtColumnLineage = useDbtColumnLineage({ workspace: () => $workspaceStore, - assetPath: () => - !pe.activeDraft && pe.selection?.kind === 'asset' && pe.selection.asset_kind === 'dbt' - ? pe.selection.path - : undefined + assetPath: () => dbtSeedPath }) - let columnGraph = $derived( - pe.selection?.kind === 'asset' && pe.selection.asset_kind === 'dbt' - ? dbtColumnLineage.graph - : ducklakeColumnGraph - ) + // 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`