From 1491aefbb28efd1debfc04b0d7259228d91260f0 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 5 Sep 2026 04:36:36 +0200 Subject: [PATCH] fix: gate pinned column lineage on reading the project, and answer the component Four things round 5 found, three of them in code this branch rewrote: - The pinned arm resolved the version from the job and stopped there, so a share-link viewer entitled to a run got the project's column names and edges while the graph beside it still redacted `raw_code` and `column_schema`. Resolving WHICH version answers is not deciding whether the caller may read it; the version-less editor buffer keeps its exemption, having no `script` row to ask. - The answer was the whole owning project's edges. The canvas lays out the connected component of the selected relation's columns, so the rest was unrenderable weight; a recursive walk over both directions returns exactly what is drawn, and the project key travels with it so a `unique_id` two projects share cannot walk from one graph into the other. - The decode had no exit but the 4M-row backstop once its buckets were full, spending wall clock the build below does not get. - An unreadable index was reported as a missing one, sending the reader to look at their engine rather than at the file. Co-Authored-By: Claude Opus 5 (1M context) --- ...c718161d95f7ad65a26525c5f43241267608d.json | 17 +++ ...fc8ee144b14d564a3e589a60678b3e68effbf.json | 17 +++ ...a582e7b06e083dd75445eeff365e8541c2ac6.json | 53 ++++++++ ...ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json | 17 +++ ...dd4b5528b92319bfe274f12e147ce0882fcb8.json | 53 -------- backend/windmill-api-assets/src/lib.rs | 128 +++++++++++++----- .../tests/dbt_pinned_graph.rs | 115 ++++++++++++++++ backend/windmill-api/openapi.yaml | 24 ++-- .../windmill-worker/src/dbt_column_index.rs | 123 ++++++++++------- docs/dbt-runtime.md | 18 ++- 10 files changed, 422 insertions(+), 143 deletions(-) create mode 100644 backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json create mode 100644 backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json create mode 100644 backend/.sqlx/query-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json create mode 100644 backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json delete mode 100644 backend/.sqlx/query-fb2320986fa4f05663f734e694ddd4b5528b92319bfe274f12e147ce0882fcb8.json diff --git a/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json b/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json new file mode 100644 index 0000000000..db9d674fcf --- /dev/null +++ b/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json @@ -0,0 +1,17 @@ +{ + "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 new file mode 100644 index 0000000000..00b839de6a --- /dev/null +++ b/backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json @@ -0,0 +1,17 @@ +{ + "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-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json b/backend/.sqlx/query-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json new file mode 100644 index 0000000000..29118c1893 --- /dev/null +++ b/backend/.sqlx/query-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json @@ -0,0 +1,53 @@ +{ + "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-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json b/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json new file mode 100644 index 0000000000..ccb9863853 --- /dev/null +++ b/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json @@ -0,0 +1,17 @@ +{ + "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/.sqlx/query-fb2320986fa4f05663f734e694ddd4b5528b92319bfe274f12e147ce0882fcb8.json b/backend/.sqlx/query-fb2320986fa4f05663f734e694ddd4b5528b92319bfe274f12e147ce0882fcb8.json deleted file mode 100644 index f230622ac7..0000000000 --- a/backend/.sqlx/query-fb2320986fa4f05663f734e694ddd4b5528b92319bfe274f12e147ce0882fcb8.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: path and version come from a job this caller was\n -- already granted, which is what makes the versionless\n -- editor buffer reachable and needs no `script` row.\n WHEN $4::text IS NOT NULL\n THEN n.script_path = $4 AND n.script_hash IS NOT DISTINCT FROM $3::bigint\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 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 -- DIRECT kinds only. `scan` — the column was read to produce the\n -- ROW, not the value — reaches every output column of its model,\n -- so it is most of a project's stored lineage and none of what a\n -- trace draws. It stays in the table for a later view to ask for.\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": "fb2320986fa4f05663f734e694ddd4b5528b92319bfe274f12e147ce0882fcb8" -} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 2600b7ac21..c6b7336a8e 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -976,13 +976,21 @@ struct DbtColumnLineageEdge { kind: String, } -/// One dbt relation's column-level lineage. +/// The column-level lineage of the dbt project one relation belongs to. /// -/// Keyed to one asset rather than carried on the graph, which is folder-wide and +/// 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. +/// +/// Keyed that way rather than carried 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 column edges needs a bound, and a bound has to be applied -/// after every filter that could drop a row; one relation's does not, so scope -/// and visibility are simply decided once, for the script that owns it. +/// 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. #[derive(Deserialize)] pub struct ColumnLineageQuery { /// The `dbt://` relation whose lineage to return. @@ -1042,7 +1050,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 + r#"WITH RECURSIVE -- 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 @@ -1066,11 +1074,21 @@ pub async fn dbt_column_lineage_for( WHERE n.script_path = pfx OR left(n.script_path, length(pfx) + 1) = pfx || '/' ) ) AND CASE - -- Pinned: path and version come from a job this caller was - -- already granted, which is what makes the versionless - -- editor buffer reachable and needs no `script` row. + -- 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. @@ -1092,31 +1110,77 @@ 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 ) - 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!", + -- 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!", 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 - -- 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. - AND e.lineage_kind IN ('copy', 'mod') - AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL"#, + 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"#, w_id, q.asset_path, script_hash, diff --git a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs index 05d3c3a3dc..a27a5c9235 100644 --- a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs +++ b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs @@ -112,6 +112,20 @@ 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!( @@ -559,6 +573,42 @@ async fn an_editor_buffers_column_lineage_answers_through_its_job(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. @@ -617,3 +667,68 @@ async fn column_lineage_takes_the_scripts_gate_and_only_the_direct_kinds(db: Poo "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 50040c382c..b1c8a02c6f 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -24189,17 +24189,22 @@ paths: /w/{workspace}/assets/column_lineage: get: - summary: Column-level lineage of one dbt relation + summary: Column-level lineage of the dbt project a relation belongs to 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 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. 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 column edges would need a - cap, and a cap has to come after every filter; one relation's does not, - so the caller's scope and the project's visibility are simply decided - once for the owning script. + 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. Empty for a project that did not opt into the analysis pass (`column_lineage: true`), which is the ordinary case. The indirect `scan` @@ -24426,12 +24431,15 @@ paths: /w/{workspace}/jobs/dbt_column_lineage/{id}: get: - summary: Get one relation's column lineage as one run saw it + 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`. + 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 diff --git a/backend/windmill-worker/src/dbt_column_index.rs b/backend/windmill-worker/src/dbt_column_index.rs index 842e15e4c6..15d980397c 100644 --- a/backend/windmill-worker/src/dbt_column_index.rs +++ b/backend/windmill-worker/src/dbt_column_index.rs @@ -23,6 +23,7 @@ //! writing them is picked up with no change. use std::collections::HashSet; +use std::ops::ControlFlow; use std::path::Path; use std::time::Duration; @@ -99,39 +100,43 @@ pub(crate) async fn collect( }; let coverage = Coverage::of(&compiled); - let index = read_index(&index_dir, kept).await; // What only the pass knows. The COUNTS are logged where the index is folded // into the graph, since the graph decides how much of it is kept. - match (&index, coverage.caveat()) { - (Some(_), None) => {} - (Some(_), Some(note)) => { - append_logs( - job_id, - w_id, - format!("\n{note}\n{}", diagnostics(&compiled.stderr)), - conn, - ) - .await; + let note = match read_index(&index_dir, kept).await { + Artifact::Read(index) => { + if let Some(note) = coverage.caveat() { + log(job_id, w_id, note, &compiled.stderr, conn).await; + } + return Ok(Some(index)); } - (None, _) => { - let note = format!( - "No column lineage: the analysis pass wrote no `{COLUMN_LINEAGE_PARQUET}`. Only \ - an engine that computes it does, and only for the warehouses it analyzes \ - natively — the flag alone is not the capability." - ); - // The engine's own diagnostics. They are how a reader learns that - // this adapter turned static analysis off, which it reports as a - // warning on a SUCCESSFUL compile that nothing else would show. - append_logs( - job_id, - w_id, - format!("\n{note}\n{}", diagnostics(&compiled.stderr)), - conn, - ) - .await; - } - } - Ok(index) + // Said apart from the one above, because it sends the reader somewhere + // else entirely: the engine did its job and this runtime could not read + // what it wrote. + Artifact::Unreadable(why) => format!( + "No column lineage: `{COLUMN_LINEAGE_PARQUET}` was written but could not be read \ + ({why}). The graph is unaffected." + ), + Artifact::Missing => format!( + "No column lineage: the analysis pass wrote no `{COLUMN_LINEAGE_PARQUET}`. Only an \ + engine that computes it does, and only for the warehouses it analyzes natively — \ + the flag alone is not the capability." + ), + }; + // The engine's own diagnostics come along. They are how a reader learns that + // this adapter turned static analysis off, which it reports as a warning on + // a SUCCESSFUL compile that nothing else would show. + log(job_id, w_id, ¬e, &compiled.stderr, conn).await; + Ok(None) +} + +async fn log(job_id: &Uuid, w_id: &str, note: &str, stderr: &str, conn: &Connection) { + append_logs( + job_id, + w_id, + format!("\n{note}\n{}", diagnostics(stderr)), + conn, + ) + .await; } /// How completely the analysis compile covered the project. @@ -296,30 +301,41 @@ fn diagnostics(out: &str) -> String { } } +/// What came back from the artifact. Never an `Err`: this half owns none of the +/// job's semantics. `Unreadable` is separate from `Missing` because the two send +/// a reader looking in different places — one at their engine and adapter, the +/// other at a file that exists. +enum Artifact { + Read(ColumnIndex), + Missing, + Unreadable(String), +} + /// Read both parquets, if the lineage one is there. /// /// The column schemas alone are not worth a graph: they arrive with the lineage /// or not at all, and a node's declared columns already answer for the case /// where the pass never ran. -async fn read_index(index_dir: &Path, kept: &HashSet<&str>) -> Option { +async fn read_index(index_dir: &Path, kept: &HashSet<&str>) -> Artifact { let lineage = index_dir.join(COLUMN_LINEAGE_PARQUET); if !tokio::fs::try_exists(&lineage).await.unwrap_or(false) { - return None; + return Artifact::Missing; } let columns = index_dir.join(NODE_COLUMNS_PARQUET); // Owned, because the decode moves to a blocking thread. The index describes // the whole project while this graph describes one selection of it, so - // scoping HERE is what keeps the cap below from being spent on rows the + // scoping HERE is what keeps the bound below from being spent on rows the // graph would discard anyway. let kept: HashSet = kept.iter().map(|s| (*s).to_string()).collect(); // Decompressing and decoding a parquet is CPU work on a file the engine just // wrote, so it does not belong on the runtime's poll thread. - tokio::task::spawn_blocking(move || read_index_blocking(&lineage, &columns, &kept)) - .await - .map_err(|e| tracing::warn!("reading the dbt column index: {e:#}")) - .ok()? - .map_err(|e| tracing::warn!("reading the dbt column index: {e:#}")) - .ok() + let read = + tokio::task::spawn_blocking(move || read_index_blocking(&lineage, &columns, &kept)).await; + match read { + Ok(Ok(index)) => Artifact::Read(index), + Ok(Err(e)) => Artifact::Unreadable(e.to_string()), + Err(e) => Artifact::Unreadable(e.to_string()), + } } fn read_index_blocking( @@ -347,7 +363,7 @@ fn read_index_blocking( || !kept.contains(&parent_unique_id) || !kept.contains(&child_unique_id) { - return; + return ControlFlow::Continue(()); } let edge = IngestedColumnEdge { parent_unique_id, @@ -360,8 +376,10 @@ 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; + return ControlFlow::Break(()); } // A direct edge displaces a `scan` one: the budget exists to be // spent on what a trace draws. @@ -369,11 +387,14 @@ fn read_index_blocking( scan.pop(); } out.edges.push(edge); - return; + return ControlFlow::Continue(()); } if held < MAX_COLUMN_EDGES { scan.push(edge); } + // Not a stopping point even when full: a direct edge still to come takes + // a `scan` entry's place. + ControlFlow::Continue(()) })?; out.edges.append(&mut scan); // Absent is normal — an engine can write the lineage table and not this one — @@ -382,8 +403,11 @@ fn read_index_blocking( let _ = for_each_row(columns, |row| { let unique_id = string(row, "unique_id"); let name = string(row, "column_name"); - if name.is_empty() || !kept.contains(&unique_id) || held >= MAX_INDEXED_COLUMNS { - return; + if held >= MAX_INDEXED_COLUMNS { + return ControlFlow::Break(()); + } + if name.is_empty() || !kept.contains(&unique_id) { + return ControlFlow::Continue(()); } held += 1; // The author's `data_type` where `schema.yml` gives one, since that is @@ -401,6 +425,7 @@ fn read_index_blocking( column_type, index: int(row, "column_index").unwrap_or(i64::MAX), }); + ControlFlow::Continue(()) }); Ok(out) } @@ -427,7 +452,11 @@ const MAX_INDEX_ROWS: usize = 4_000_000; /// Collecting first would put a `Vec` — each row carrying its own copy of /// every column NAME — in front of the caller's own bound, which is what would /// take the worker process down on the index described above. -fn for_each_row(path: &Path, mut f: impl FnMut(&Row)) -> error::Result<()> { +/// +/// `f` says when it has all it will take, and that is the ordinary end: this +/// runs outside the phase budget, so every row decoded past the point of being +/// able to keep one is wall clock the build below does not get. +fn for_each_row(path: &Path, mut f: impl FnMut(&Row) -> ControlFlow<()>) -> error::Result<()> { let fail = |e: parquet::errors::ParquetError| { error::Error::internal_err(format!("reading {}: {e}", path.display())) }; @@ -442,7 +471,9 @@ fn for_each_row(path: &Path, mut f: impl FnMut(&Row)) -> error::Result<()> { ); break; } - f(&row.map_err(fail)?); + if f(&row.map_err(fail)?).is_break() { + break; + } } Ok(()) } diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 775f35fb15..9c1fed03c1 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1232,9 +1232,13 @@ 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 caller's -`scripts:read` scope and the project's visibility are decided once, for the -script that owns the relation. Pinning to a run's snapshot or to the editor's +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`. @@ -1242,7 +1246,13 @@ parse of its own buffer costs the job-read gate, so that form is 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. +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. ## Concept mapping