diff --git a/backend/.sqlx/query-7e7cbe03e512096ac65acce50c8be0863f85119f0d49d18553b240855e28a3ad.json b/backend/.sqlx/query-7e7cbe03e512096ac65acce50c8be0863f85119f0d49d18553b240855e28a3ad.json new file mode 100644 index 0000000000..eeb9773476 --- /dev/null +++ b/backend/.sqlx/query-7e7cbe03e512096ac65acce50c8be0863f85119f0d49d18553b240855e28a3ad.json @@ -0,0 +1,66 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH live AS (\n SELECT * FROM (\n SELECT DISTINCT ON (s.path) s.path, s.hash\n FROM script s\n WHERE $5::text IS NULL AND s.workspace_id = $1 AND s.language = 'dbt'\n AND ($3::bigint IS NULL OR s.hash = $3)\n AND ($3::bigint IS NOT NULL OR (s.deleted = false AND s.archived = false))\n ORDER BY s.path, s.created_at DESC\n ) cur\n UNION ALL\n SELECT $5::text, $3::bigint WHERE $5::text IS NOT NULL\n ),\n -- The run's own snapshot when it left one, the version's graph\n -- otherwise. A static descriptor never snapshots, so all of its runs\n -- fall through to the same rows. Existence comes from the marker, not\n -- from a node row: a dynamic run that disabled every model has a\n -- snapshot whose graph is legitimately empty.\n chosen AS (\n -- No visibility check on the job here: reaching this with a job at\n -- all means the caller passed `require_job_read_access` for it, and\n -- re-deciding it under plain RLS can only DISAGREE with that answer\n -- — silently, by falling back to the deployed graph rather than\n -- erroring. A share-link viewer is entitled to the run and would be\n -- shown a different run's model set. See `asset_graph_for`.\n SELECT CASE WHEN $4::uuid IS NOT NULL AND EXISTS (\n SELECT 1 FROM dbt_graph_snapshot g\n WHERE g.workspace_id = $1 AND g.job_id = $4)\n THEN $4::uuid\n ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id\n ),\n -- Both kinds of edge in the terms the joins below need. A column edge\n -- carries three more fields and is otherwise keyed identically, which\n -- is the whole reason one set of joins can serve both.\n edges AS (\n SELECT script_path, script_hash, job_id, workspace_id,\n parent_unique_id, child_unique_id,\n NULL::text AS parent_column, NULL::text AS child_column,\n NULL::text AS lineage_kind\n FROM dbt_edge WHERE workspace_id = $1\n UNION ALL\n SELECT script_path, script_hash, job_id, workspace_id,\n parent_unique_id, child_unique_id,\n parent_column, child_column, lineage_kind\n FROM dbt_column_edge\n WHERE 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 \"show indirect\"\n -- view to ask for.\n AND lineage_kind IN ('copy', 'mod')\n -- Scope filtering in the WHERE, not a retain after the fetch:\n -- the column half is capped below, and a post-fetch filter lets\n -- rows the token may not read consume that cap and return an\n -- allowed project's trace short or empty.\n AND ( $6\n OR script_path = ANY($7)\n OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx\n WHERE script_path = pfx\n OR left(script_path, length(pfx) + 1) = pfx || '/' ) )\n ),\n -- Everything both kinds of edge need, resolved once.\n joined AS (\n SELECT p.asset_path AS from_path, c.asset_path AS to_path,\n e.parent_column, e.child_column, e.lineage_kind,\n e.script_path,\n -- Whether the caller may read the project the edge\n -- describes. A column-level view is the shape of what the\n -- author WROTE, so it takes the model's own gate; the\n -- relation-level `ref()` edges are ungated because they only\n -- connect relations the caller already sees in `asset`.\n --\n -- The NULL arm is not optional: a version-less row has no\n -- `script` row to ask and needs none — it exists because\n -- this caller's own parse job wrote the buffer — and\n -- `sc.hash = NULL` is never true, so without it every editor\n -- parse loses its lineage.\n (e.script_hash IS NULL OR EXISTS (\n SELECT 1 FROM script sc\n WHERE sc.workspace_id = e.workspace_id\n AND sc.path = e.script_path\n AND sc.hash = e.script_hash)) AS script_visible\n FROM edges e\n JOIN live l ON l.path = e.script_path\n AND (e.script_hash = l.hash\n OR ($5::text IS NOT NULL AND l.hash IS NULL\n AND e.script_hash IS NULL))\n JOIN chosen ch ON ch.job_id = e.job_id\n -- Gated the same way as the `live` joins above, and for the same\n -- reason twice over: unpinned this folds to a plain equality the\n -- versioned key can bound, and pinned it is the only way an\n -- editor graph's NULL-to-NULL hashes meet at all.\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 = e.script_hash\n OR ($5::text IS NOT NULL AND e.script_hash IS NULL\n AND p.script_hash IS NULL))\n AND p.job_id = ch.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 = e.script_hash\n OR ($5::text IS NOT NULL AND e.script_hash IS NULL\n AND c.script_hash IS NULL))\n AND c.job_id = ch.job_id\n AND c.unique_id = e.child_unique_id\n WHERE p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL\n -- Tests attach to their model as a badge, not as a lineage edge.\n AND c.resource_type <> 'test'\n -- Scoped by the RELATIONS, like the node provenance above, not\n -- by the producing script's folder: two tables consumed in this\n -- folder but produced by a dbt project outside it would\n -- otherwise both render with their `ref()` edge missing.\n -- Same as the node scope: pinned, `asset` describes the CURRENT\n -- deploy, so gating on it drops the edges of models this version\n -- had and a later one removed. The pinned graph's own edges are\n -- the answer.\n AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL OR EXISTS (\n SELECT 1 FROM asset a\n WHERE a.workspace_id = $1 AND a.kind = 'dbt'\n AND a.path = c.asset_path\n AND ($2::text IS NULL OR a.usage_path LIKE $2)))\n )\n SELECT from_path AS \"from_path!\", to_path AS \"to_path!\",\n parent_column, child_column, lineage_kind,\n script_path AS \"script_path!\", script_visible AS \"script_visible!\"\n FROM (\n -- The `ref()` half needs no cap: it is bounded by the relations\n -- in view, which is what the rest of this response is bounded by.\n SELECT * FROM joined WHERE lineage_kind IS NULL\n UNION ALL\n -- The column half is capped, and capped HERE — after the scope\n -- filter, the visibility check and the graph joins — so no row a\n -- caller may not read, and none outside the graph on screen, can\n -- spend the budget and leave an allowed project's trace short.\n -- Ordered so the rows kept are the same ones on every request.\n (SELECT * FROM joined WHERE lineage_kind IS NOT NULL\n ORDER BY from_path, to_path, parent_column, child_column\n LIMIT $9)\n ) all_edges", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "from_path!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "to_path!", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "parent_column", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "child_column", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "lineage_kind", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "script_path!", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "script_visible!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8", + "Uuid", + "Text", + "Bool", + "TextArray", + "TextArray", + "Int8" + ] + }, + "nullable": [ + null, + null, + null, + null, + null, + null, + null + ] + }, + "hash": "7e7cbe03e512096ac65acce50c8be0863f85119f0d49d18553b240855e28a3ad" +} diff --git a/backend/.sqlx/query-fad10d08b6a7fb4c6974a6dce79028f9bae1b9a9806c9efb2844f7e7e8196f5b.json b/backend/.sqlx/query-fad10d08b6a7fb4c6974a6dce79028f9bae1b9a9806c9efb2844f7e7e8196f5b.json deleted file mode 100644 index 6d0fdd1bd8..0000000000 --- a/backend/.sqlx/query-fad10d08b6a7fb4c6974a6dce79028f9bae1b9a9806c9efb2844f7e7e8196f5b.json +++ /dev/null @@ -1,32 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH live AS (\n SELECT * FROM (\n SELECT DISTINCT ON (s.path) s.path, s.hash\n FROM script s\n WHERE $5::text IS NULL AND s.workspace_id = $1 AND s.language = 'dbt'\n AND ($3::bigint IS NULL OR s.hash = $3)\n AND ($3::bigint IS NOT NULL OR (s.deleted = false AND s.archived = false))\n ORDER BY s.path, s.created_at DESC\n ) cur\n UNION ALL\n SELECT $5::text, $3::bigint WHERE $5::text IS NOT NULL\n ),\n -- The run's own snapshot when it left one, the version's graph\n -- otherwise. A static descriptor never snapshots, so all of its runs\n -- fall through to the same rows. Existence comes from the marker, not\n -- from a node row: a dynamic run that disabled every model has a\n -- snapshot whose graph is legitimately empty.\n chosen AS (\n -- No visibility check on the job here: reaching this with a job at\n -- all means the caller passed `require_job_read_access` for it, and\n -- re-deciding it under plain RLS can only DISAGREE with that answer\n -- — silently, by falling back to the deployed graph rather than\n -- erroring. A share-link viewer is entitled to the run and would be\n -- shown a different run's model set. See `asset_graph_for`.\n SELECT CASE WHEN $4::uuid IS NOT NULL AND EXISTS (\n SELECT 1 FROM dbt_graph_snapshot g\n WHERE g.workspace_id = $1 AND g.job_id = $4)\n THEN $4::uuid\n ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id\n )\n SELECT p.asset_path AS \"from_path!\", c.asset_path AS \"to_path!\"\n FROM dbt_edge e\n JOIN live l ON l.path = e.script_path\n AND (e.script_hash = l.hash\n OR ($5::text IS NOT NULL AND l.hash IS NULL\n AND e.script_hash IS NULL))\n JOIN chosen ch ON ch.job_id = e.job_id\n -- Gated the same way as the `live` joins above, and for the same\n -- reason twice over: unpinned this folds to a plain equality the\n -- versioned key can bound, and pinned it is the only way an editor\n -- graph's NULL-to-NULL hashes meet at all.\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 = e.script_hash\n OR ($5::text IS NOT NULL AND e.script_hash IS NULL\n AND p.script_hash IS NULL))\n AND p.job_id = ch.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 = e.script_hash\n OR ($5::text IS NOT NULL AND e.script_hash IS NULL\n AND c.script_hash IS NULL))\n AND c.job_id = ch.job_id\n AND c.unique_id = e.child_unique_id\n WHERE e.workspace_id = $1\n AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL\n -- Tests attach to their model as a badge, not as a lineage edge.\n AND c.resource_type <> 'test'\n -- Scoped by the RELATIONS, like the node provenance above, not by\n -- the producing script's folder: two tables consumed in this\n -- folder but produced by a dbt project outside it would otherwise\n -- both render with their `ref()` edge missing.\n -- Same as the node scope: pinned, `asset` describes the CURRENT\n -- deploy, so gating on it drops the edges of models this version\n -- had and a later one removed. The pinned graph's own edges are\n -- the answer.\n AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL OR EXISTS (\n SELECT 1 FROM asset a\n WHERE a.workspace_id = $1 AND a.kind = 'dbt'\n AND a.path = c.asset_path\n AND ($2::text IS NULL OR a.usage_path LIKE $2)))", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "from_path!", - "type_info": "Text" - }, - { - "ordinal": 1, - "name": "to_path!", - "type_info": "Text" - } - ], - "parameters": { - "Left": [ - "Text", - "Text", - "Int8", - "Uuid", - "Text" - ] - }, - "nullable": [ - true, - true - ] - }, - "hash": "fad10d08b6a7fb4c6974a6dce79028f9bae1b9a9806c9efb2844f7e7e8196f5b" -} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 5e4290115d..28540d3648 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -13,7 +13,9 @@ use windmill_common::{ utils::escape_ilike_pattern, }; -use windmill_api_auth::{build_scope_path_predicate, ApiAuthed}; +use windmill_api_auth::{ + build_scope_path_filter, build_scope_path_predicate, ApiAuthed, ScopePathFilter, +}; // Partition-range backfill preview. The logic (producer resolution, range // enumeration, status join) is enterprise: the `private` build compiles the @@ -1042,6 +1044,16 @@ pub async fn asset_graph_for( // node's SQL body may be returned, independently of the `assets:read` scope // that authorizes this endpoint. let dbt_source_scope = build_scope_path_predicate(authed, "scripts", "read"); + // The same grant, decomposed for SQL. The column-edge half of the graph + // query is capped, so its scope filter has to run IN the query: a retain + // after the fetch lets rows the token may not read spend that cap and return + // an allowed project's trace short. `ScopePathFilter::allows` mirrors the + // emitted SQL. + let (dbt_scope_all, dbt_scope_exact, dbt_scope_prefix) = + match build_scope_path_filter(authed, "scripts", "read") { + ScopePathFilter::AllowAll => (true, Vec::new(), Vec::new()), + ScopePathFilter::Restricted { exact, prefix } => (false, exact, prefix), + }; // REFUSED, not dropped: a caller that asks for a kind this server does not // know has asked for something, and answering with the other kinds returns a @@ -1413,9 +1425,20 @@ pub async fn asset_graph_for( .fetch_all(&mut *tx) .await?; - // `ref()` lineage between two models, resolved to the relations they - // produce. Joined to `dbt_node` on both key columns because a dbt - // `unique_id` is only unique within its project. + // `ref()` lineage between two models, and the column-to-column lineage + // beneath it, resolved to the relations they produce. + // + // ONE statement over a UNION ALL'd edge source rather than two near-copies. + // The `live`/`chosen` CTEs and — more to the point — the version and + // editor-buffer join conditions are the invariants a second copy has to + // restate, and restating them is what dropped the `script_hash IS NULL` arm + // and hid every buffer parse's column lineage. Written once, they cannot + // disagree. + // + // Column edges are NOT hung off `dbt_edge` rows, which is why the union is + // at the source and not a join: an incremental model reading `{{ this }}` + // has column lineage from itself to itself, and `parent_map` has no + // self-loop, so those pairs have no `dbt_edge` row to attach to. let dbt_edge_rows = sqlx::query!( r#"WITH live AS ( SELECT * FROM ( @@ -1446,146 +1469,125 @@ pub async fn asset_graph_for( WHERE g.workspace_id = $1 AND g.job_id = $4) THEN $4::uuid ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id - ) - SELECT p.asset_path AS "from_path!", c.asset_path AS "to_path!" - FROM dbt_edge e - JOIN live l ON l.path = e.script_path - AND (e.script_hash = l.hash - OR ($5::text IS NOT NULL AND l.hash IS NULL - AND e.script_hash IS NULL)) - JOIN chosen ch ON ch.job_id = e.job_id - -- Gated the same way as the `live` joins above, and for the same - -- reason twice over: unpinned this folds to a plain equality the - -- versioned key can bound, and pinned it is the only way an editor - -- graph's NULL-to-NULL hashes meet at all. - JOIN dbt_node p ON p.workspace_id = e.workspace_id - AND p.script_path = e.script_path - AND (p.script_hash = e.script_hash - OR ($5::text IS NOT NULL AND e.script_hash IS NULL - AND p.script_hash IS NULL)) - AND p.job_id = ch.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 = e.script_hash - OR ($5::text IS NOT NULL AND e.script_hash IS NULL - AND c.script_hash IS NULL)) - AND c.job_id = ch.job_id - AND c.unique_id = e.child_unique_id - WHERE e.workspace_id = $1 - AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL - -- Tests attach to their model as a badge, not as a lineage edge. - AND c.resource_type <> 'test' - -- Scoped by the RELATIONS, like the node provenance above, not by - -- the producing script's folder: two tables consumed in this - -- folder but produced by a dbt project outside it would otherwise - -- both render with their `ref()` edge missing. - -- Same as the node scope: pinned, `asset` describes the CURRENT - -- deploy, so gating on it drops the edges of models this version - -- had and a later one removed. The pinned graph's own edges are - -- the answer. - AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL OR EXISTS ( - SELECT 1 FROM asset a - WHERE a.workspace_id = $1 AND a.kind = 'dbt' - AND a.path = c.asset_path - AND ($2::text IS NULL OR a.usage_path LIKE $2)))"#, - &w_id, - folder_filter.as_deref(), - dbt_script_hash, - dbt_job_id, - pinned_path, - ) - .fetch_all(&mut *tx) - .await?; - - // Column-to-column lineage, resolved to the relations the two nodes produce, - // like the `ref()` edges above and scoped exactly the same way. Present only - // for a project that opted into the analysis pass, so the usual graph pays - // one indexed lookup that finds nothing. - // - // DIRECT edges only. `scan` — the column was read to produce the row, not - // the value — is emitted from every join key and predicate column to every - // output column, so it is most of a project's stored lineage and none of - // what a column trace draws. It stays in `dbt_column_edge`, where a later - // view can ask for it without every project being redeployed, rather than - // riding on the graph endpoint a run page polls. - // - // Bounded for the same reason: every other array here is bounded by the - // relations in view, and this one would be bounded by the size of a - // project's column graph. - let dbt_column_edge_rows = sqlx::query!( - r#"WITH live AS ( - SELECT * FROM ( - SELECT DISTINCT ON (s.path) s.path, s.hash - FROM script s - WHERE $5::text IS NULL AND s.workspace_id = $1 AND s.language = 'dbt' - AND ($3::bigint IS NULL OR s.hash = $3) - AND ($3::bigint IS NOT NULL OR (s.deleted = false AND s.archived = false)) - ORDER BY s.path, s.created_at DESC - ) cur - UNION ALL - SELECT $5::text, $3::bigint WHERE $5::text IS NOT NULL ), - chosen AS ( - SELECT CASE WHEN $4::uuid IS NOT NULL AND EXISTS ( - SELECT 1 FROM dbt_graph_snapshot g - WHERE g.workspace_id = $1 AND g.job_id = $4) - THEN $4::uuid - ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id + -- Both kinds of edge in the terms the joins below need. A column edge + -- carries three more fields and is otherwise keyed identically, which + -- is the whole reason one set of joins can serve both. + edges AS ( + SELECT script_path, script_hash, job_id, workspace_id, + parent_unique_id, child_unique_id, + NULL::text AS parent_column, NULL::text AS child_column, + NULL::text AS lineage_kind + FROM dbt_edge WHERE workspace_id = $1 + UNION ALL + SELECT script_path, script_hash, job_id, workspace_id, + parent_unique_id, child_unique_id, + parent_column, child_column, lineage_kind + FROM dbt_column_edge + WHERE 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 "show indirect" + -- view to ask for. + AND lineage_kind IN ('copy', 'mod') + -- Scope filtering in the WHERE, not a retain after the fetch: + -- the column half is capped below, and a post-fetch filter lets + -- rows the token may not read consume that cap and return an + -- allowed project's trace short or empty. + AND ( $6 + OR script_path = ANY($7) + OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx + WHERE script_path = pfx + OR left(script_path, length(pfx) + 1) = pfx || '/' ) ) + ), + -- Everything both kinds of edge need, resolved once. + joined AS ( + SELECT p.asset_path AS from_path, c.asset_path AS to_path, + e.parent_column, e.child_column, e.lineage_kind, + e.script_path, + -- Whether the caller may read the project the edge + -- describes. A column-level view is the shape of what the + -- author WROTE, so it takes the model's own gate; the + -- relation-level `ref()` edges are ungated because they only + -- connect relations the caller already sees in `asset`. + -- + -- The NULL arm is not optional: a version-less row has no + -- `script` row to ask and needs none — it exists because + -- this caller's own parse job wrote the buffer — and + -- `sc.hash = NULL` is never true, so without it every editor + -- parse loses its lineage. + (e.script_hash IS NULL OR EXISTS ( + SELECT 1 FROM script sc + WHERE sc.workspace_id = e.workspace_id + AND sc.path = e.script_path + AND sc.hash = e.script_hash)) AS script_visible + FROM edges e + JOIN live l ON l.path = e.script_path + AND (e.script_hash = l.hash + OR ($5::text IS NOT NULL AND l.hash IS NULL + AND e.script_hash IS NULL)) + JOIN chosen ch ON ch.job_id = e.job_id + -- Gated the same way as the `live` joins above, and for the same + -- reason twice over: unpinned this folds to a plain equality the + -- versioned key can bound, and pinned it is the only way an + -- editor graph's NULL-to-NULL hashes meet at all. + JOIN dbt_node p ON p.workspace_id = e.workspace_id + AND p.script_path = e.script_path + AND (p.script_hash = e.script_hash + OR ($5::text IS NOT NULL AND e.script_hash IS NULL + AND p.script_hash IS NULL)) + AND p.job_id = ch.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 = e.script_hash + OR ($5::text IS NOT NULL AND e.script_hash IS NULL + AND c.script_hash IS NULL)) + AND c.job_id = ch.job_id + AND c.unique_id = e.child_unique_id + WHERE p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL + -- Tests attach to their model as a badge, not as a lineage edge. + AND c.resource_type <> 'test' + -- Scoped by the RELATIONS, like the node provenance above, not + -- by the producing script's folder: two tables consumed in this + -- folder but produced by a dbt project outside it would + -- otherwise both render with their `ref()` edge missing. + -- Same as the node scope: pinned, `asset` describes the CURRENT + -- deploy, so gating on it drops the edges of models this version + -- had and a later one removed. The pinned graph's own edges are + -- the answer. + AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL OR EXISTS ( + SELECT 1 FROM asset a + WHERE a.workspace_id = $1 AND a.kind = 'dbt' + AND a.path = c.asset_path + AND ($2::text IS NULL OR a.usage_path LIKE $2))) ) - 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 "lineage_kind!", e.script_path AS "script_path!", - -- Same gate as the model's own SQL, applied in Rust beside it: - -- a column graph 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 in - -- `asset`. - -- - -- The NULL arm is not optional, for the reason the node query - -- gives: a version-less row has no `script` row to ask and - -- needs none, and `sc.hash = NULL` is never true — so without - -- it an editor buffer's parse renders its columns and none of - -- their lineage. - (e.script_hash IS NULL OR EXISTS ( - SELECT 1 FROM script sc - WHERE sc.workspace_id = e.workspace_id - AND sc.path = e.script_path - AND sc.hash = e.script_hash)) AS "script_visible!" - FROM dbt_column_edge e - JOIN live l ON l.path = e.script_path - AND (e.script_hash = l.hash - OR ($5::text IS NOT NULL AND l.hash IS NULL - AND e.script_hash IS NULL)) - JOIN chosen ch ON ch.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 = e.script_hash - OR ($5::text IS NOT NULL AND e.script_hash IS NULL - AND p.script_hash IS NULL)) - AND p.job_id = ch.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 = e.script_hash - OR ($5::text IS NOT NULL AND e.script_hash IS NULL - AND c.script_hash IS NULL)) - AND c.job_id = ch.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 - AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL OR EXISTS ( - SELECT 1 FROM asset a - WHERE a.workspace_id = $1 AND a.kind = 'dbt' - AND a.path = c.asset_path - AND ($2::text IS NULL OR a.usage_path LIKE $2))) - LIMIT $6"#, + SELECT from_path AS "from_path!", to_path AS "to_path!", + parent_column, child_column, lineage_kind, + script_path AS "script_path!", script_visible AS "script_visible!" + FROM ( + -- The `ref()` half needs no cap: it is bounded by the relations + -- in view, which is what the rest of this response is bounded by. + SELECT * FROM joined WHERE lineage_kind IS NULL + UNION ALL + -- The column half is capped, and capped HERE — after the scope + -- filter, the visibility check and the graph joins — so no row a + -- caller may not read, and none outside the graph on screen, can + -- spend the budget and leave an allowed project's trace short. + -- Ordered so the rows kept are the same ones on every request. + (SELECT * FROM joined WHERE lineage_kind IS NOT NULL + ORDER BY from_path, to_path, parent_column, child_column + LIMIT $9) + ) all_edges"#, &w_id, folder_filter.as_deref(), dbt_script_hash, dbt_job_id, pinned_path, + dbt_scope_all, + &dbt_scope_exact[..], + &dbt_scope_prefix[..], MAX_RENDERED_COLUMN_EDGES, ) .fetch_all(&mut *tx) @@ -2286,33 +2288,43 @@ pub async fn asset_graph_for( // that is not here has nothing to draw. let rendered: std::collections::HashSet<&str> = assets.iter().map(|a| a.path.as_str()).collect(); - let mut dbt_edges: Vec = dbt_edge_rows - .into_iter() - .filter(|r| { - r.from_path != r.to_path - && rendered.contains(r.from_path.as_str()) - && rendered.contains(r.to_path.as_str()) - }) - .map(|r| DbtLineageEdge { from_asset_path: r.from_path, to_asset_path: r.to_path }) - .collect(); + // One query, two result sets: a row with no `lineage_kind` is a `ref()` edge + // between relations, one with a kind is a column edge beneath it. + let mut dbt_edges: Vec = Vec::new(); + let mut dbt_column_edges: Vec = Vec::new(); + for r in dbt_edge_rows { + if !rendered.contains(r.from_path.as_str()) || !rendered.contains(r.to_path.as_str()) { + continue; + } + match (r.lineage_kind, r.parent_column, r.child_column) { + (Some(kind), Some(from_column), Some(to_column)) => { + // `script_visible` is already the query's, and the scope filter + // ran there too; this only mirrors the same gate for the rows a + // non-scoped path reached. + if !r.script_visible { + continue; + } + dbt_column_edges.push(DbtColumnLineageEdge { + from_asset_path: r.from_path, + from_column, + to_asset_path: r.to_path, + to_column, + kind, + }); + } + // A relation has no `ref()` edge to itself worth drawing, while a + // column edge to itself is real — an incremental reading `{{ this }}` + // carries the previous run's value forward — so the self-pair filter + // belongs on this arm alone. + _ if r.from_path != r.to_path => dbt_edges.push(DbtLineageEdge { + from_asset_path: r.from_path, + to_asset_path: r.to_path, + }), + _ => {} + } + } dbt_edges.sort(); dbt_edges.dedup(); - let mut dbt_column_edges: Vec = dbt_column_edge_rows - .into_iter() - .filter(|r| { - rendered.contains(r.from_path.as_str()) - && rendered.contains(r.to_path.as_str()) - && r.script_visible - && dbt_source_scope(&r.script_path) - }) - .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.lineage_kind, - }) - .collect(); dbt_column_edges.sort(); dbt_column_edges.dedup(); diff --git a/backend/windmill-common/src/dbt_manifest.rs b/backend/windmill-common/src/dbt_manifest.rs index a9f9213a4a..a81faf80c3 100644 --- a/backend/windmill-common/src/dbt_manifest.rs +++ b/backend/windmill-common/src/dbt_manifest.rs @@ -518,10 +518,11 @@ impl IngestedManifest { // passes over one project must not read as two different graphs. // // Direct kinds first, so what the truncation below gives up is `scan` — - // the bulk of a wide project's lineage and the kind nothing renders. - // The worker's own reader already prioritizes them while decoding, since - // the memory bound has to apply there; this is the same order for a - // caller that did not come through it, which is the agent-worker wire. + // the bulk of a wide project's lineage and the kind nothing renders. The + // worker's reader already applies this order while decoding, because the + // memory bound has to; repeating it here is what makes the ordering a + // property of the manifest rather than of one caller's reader, and it is + // the only ordering an index assembled some other way would get. edges.sort_by(|a, b| { is_direct(&b.lineage_kind) .cmp(&is_direct(&a.lineage_kind)) diff --git a/backend/windmill-worker/src/dbt_column_index.rs b/backend/windmill-worker/src/dbt_column_index.rs index 3a3390b6b7..d60591bb20 100644 --- a/backend/windmill-worker/src/dbt_column_index.rs +++ b/backend/windmill-worker/src/dbt_column_index.rs @@ -52,16 +52,18 @@ const NODE_COLUMNS_PARQUET: &str = "dbt.node_columns.parquet"; /// Run the lineage pass and read what it produced. /// -/// Best-effort about the COMPILE and about the artifact: a wrong engine, a -/// failed analysis, a missing or unreadable parquet, or this phase outrunning -/// its budget all return `None` with a line in the job log saying which, because -/// the graph without column lineage is exactly the graph this project had before -/// it asked for any. +/// Two steps with deliberately different contracts, because conflating them is +/// what made a best-effort annotation able to fail the job it annotates: /// -/// NOT best-effort about the job: a cancellation, the job's own deadline or the -/// output ceiling are returned as `Err` and fail it. Swallowing those would let -/// a run that blew its timeout inside an optional annotation go on to publish a -/// graph and report success. +/// - [`compile_index`] runs a subprocess and owns the JOB's semantics. Only a +/// cancellation or the job's own deadline can `Err` out of it; a non-zero exit +/// and an over-long output are outcomes, not failures. +/// - [`read_index`] owns the ARTIFACT's semantics. It is infallible and +/// memory-bounded, and knows nothing about the job. +/// +/// The phase budget wraps the compile alone. A budget around the whole pass +/// would time out with a decode still running on a blocking thread, which is +/// precisely what "the build below gets the rest" must not mean. pub(crate) async fn collect( p: &PreparedProject, descriptor: &DbtDescriptor, @@ -71,43 +73,6 @@ pub(crate) async fn collect( w_id: &str, conn: &Connection, kept: &HashSet<&str>, -) -> error::Result> { - let Some(budget) = phase_budget(ctx) else { - return run_pass(p, descriptor, inv, ctx, job_id, w_id, conn, kept).await; - }; - match tokio::time::timeout( - budget, - run_pass(p, descriptor, inv, ctx, job_id, w_id, conn, kept), - ) - .await - { - Ok(r) => r, - Err(_) => { - append_logs( - job_id, - w_id, - format!( - "\nNo column lineage: the analysis pass did not finish within {}s, half of \ - what was left of this job's time. The build below gets the rest.\n", - budget.as_secs() - ), - conn, - ) - .await; - Ok(None) - } - } -} - -async fn run_pass( - p: &PreparedProject, - descriptor: &DbtDescriptor, - inv: &Invocation, - ctx: &mut JobCtx<'_>, - job_id: &Uuid, - w_id: &str, - conn: &Connection, - kept: &HashSet<&str>, ) -> error::Result> { if !descriptor.column_lineage { return Ok(None); @@ -129,6 +94,90 @@ async fn run_pass( } let index_dir = p.project_dir.join(CLL_ARTIFACTS_DIR).join("index"); + let Some(compiled) = + compile_index(p, descriptor, inv, ctx, job_id, w_id, conn).await? + else { + return Ok(None); + }; + + 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, compiled.analyzed_everything()) { + (Some(_), true) => {} + // The ordinary partial outcome: some models did not analyze, the rest + // did, and their lineage is real. + (Some(_), false) => { + let note = "Column lineage: `--static-analysis strict` rejected part of the project, \ + so the lineage covers only the models it could analyze."; + append_logs(job_id, w_id, format!("\n{note}\n{}", compiled.detail()), conn).await; + } + (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{}", compiled.detail()), conn).await; + } + } + Ok(index) +} + +/// What the analysis compile did, from the pass's point of view. +/// +/// Every way the COMPILE can disappoint is a value here rather than an error: a +/// non-zero exit (strict analysis rejected some SQL), an output ceiling, or the +/// phase budget. An `Err` from `compile_index` is the JOB's — a cancellation or +/// its deadline — and must fail it. +struct Compiled { + /// `None` once the pass gave up on its own terms, which the caller has + /// already been told about. + outcome: Option, +} + +impl Compiled { + /// Whether the engine analyzed the whole project. False means the index, if + /// there is one, covers only part of it. + /// + /// A truncated run is not a rejected project: the ceiling killed a compile + /// that was otherwise doing its job, and whatever it had already written to + /// `target/index/` is still on disk. + fn analyzed_everything(&self) -> bool { + self.outcome + .as_ref() + .is_some_and(|c| c.success || c.truncated) + } + + /// The engine's own diagnostics, bounded, or nothing when the pass never got + /// far enough to have any. + fn detail(&self) -> String { + self.outcome + .as_ref() + .map(|c| diagnostics(&c.stderr)) + .unwrap_or_default() + } +} + +/// Run `dbt compile --static-analysis strict --write-index`, under this phase's +/// share of the job's clock. +/// +/// `Ok(None)` is "the pass gave up and said so"; `Err` is the job's own +/// cancellation or deadline and must propagate. Nothing outlives this function: +/// the budget is a race around the child, and dropping that future kills it +/// through `run_captured`'s `kill_on_drop`. +async fn compile_index( + p: &PreparedProject, + descriptor: &DbtDescriptor, + inv: &Invocation, + ctx: &mut JobCtx<'_>, + job_id: &Uuid, + w_id: &str, + conn: &Connection, +) -> error::Result> { // A previous pass in the same job directory — a retry's second attempt — // would otherwise be read back as this one's answer. tokio::fs::remove_dir_all(p.project_dir.join(CLL_ARTIFACTS_DIR)) @@ -159,13 +208,9 @@ async fn run_pass( // 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. - // - // `run_captured`, so a failed COMPILE is `success == false` and gets - // downgraded below, while an `Err` — a cancellation, the job's deadline, the - // output ceiling — still fails the job. Swallowing those would let a run - // that blew its timeout inside this pass go on to publish a graph and report - // success. - let outcome = crate::dbt_executor::run_captured( + // Read before the future below borrows `ctx` mutably. + let budget = phase_budget(ctx); + let run = crate::dbt_executor::run_captured( cmd, "dbt compile (column lineage)", ctx, @@ -173,47 +218,31 @@ async fn run_pass( w_id, conn, CLL_MAX_OUTPUT_BYTES, - ) - .await?; - - let index = read_index(&index_dir, kept).await; - // What the caller cannot say for itself. The COUNTS are logged where the - // index is folded into the graph, since the graph is what decides how much - // of it is kept; this is the part only the pass knows. - match (&index, outcome.success) { - (Some(_), true) => {} - // The ordinary partial outcome: some models did not analyze, the rest - // did, and their lineage is real. - (Some(_), false) => { - let note = "Column lineage: `--static-analysis strict` rejected part of the project, \ - so the lineage covers only the models it could analyze."; + // The ceiling is this pass's, not the job's: a compile that prints more + // than it than we care to read has still analyzed the project, and the + // index it wrote is on disk either way. + crate::dbt_executor::Overflow::Truncate, + ); + let Some(budget) = budget else { + return Ok(Some(Compiled { outcome: Some(run.await?) })); + }; + match tokio::time::timeout(budget, run).await { + Ok(r) => Ok(Some(Compiled { outcome: Some(r?) })), + Err(_) => { append_logs( job_id, w_id, - format!("\n{note}\n{}", diagnostics(&outcome.stderr)), - conn, - ) - .await; - } - (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(&outcome.stderr)), + format!( + "\nNo column lineage: the analysis pass did not finish within {}s, half of \ + what was left of this job's time. The build below gets the rest.\n", + budget.as_secs() + ), conn, ) .await; + Ok(None) } } - Ok(index) } /// stdout the pass may produce. It is a compile, so this is diagnostics rather @@ -227,12 +256,12 @@ const CLL_MAX_OUTPUT_BYTES: usize = 1 << 20; /// budget and fail the run it exists only to annotate. Half leaves the build at /// least as long as the annotation was allowed to take. /// -/// Spent as a race around the whole pass rather than as a shortened deadline -/// handed to the runner: the runner reports its expiry as an `Err`, which is -/// indistinguishable from a cancellation or the job's own deadline, and those -/// two MUST fail the job. Expiring here is this budget and nothing else, so it -/// degrades to no lineage. Dropping the future kills the child, which -/// `run_captured` spawns with `kill_on_drop`. +/// Spent as a race around the COMPILE rather than as a shortened deadline handed +/// to the runner: the runner reports its expiry as an `Err`, indistinguishable +/// from a cancellation or the job's own deadline, and those two MUST fail the +/// job. Expiring here is this budget and nothing else. The child dies with the +/// dropped future through `run_captured`'s `kill_on_drop`, and the decode is +/// outside the race so nothing survives it. fn phase_budget(ctx: &JobCtx<'_>) -> Option { ctx.timeout() .map(|left| Duration::from_secs((left.max(0) as u64 / 2).max(1))) diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index 5efa8a1db6..12a092ba71 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -3553,10 +3553,25 @@ pub(crate) struct Captured { pub stdout: String, pub stderr: String, /// Whether the child exited zero. Separate from the `Result` on purpose: an - /// `Err` from `run_captured` is never the child's exit status but the JOB's - /// — a cancellation, the deadline, an output ceiling — so a caller that - /// tolerates a failed command must still propagate one. + /// `Err` from `run_captured` is the JOB's — a cancellation or its deadline — + /// so a caller that tolerates a failed command must still propagate one. pub success: bool, + /// Whether the output ceiling cut the child short. Only ever true under + /// [`Overflow::Truncate`]. + pub truncated: bool, +} + +/// What an over-long stdout means to the caller. +/// +/// The ceiling belongs to the PASS, not to the job: a caller that only annotates +/// a job wants to keep what it read and carry on, while one whose whole result +/// is that output has nothing to return without it. +#[derive(PartialEq, Eq, Clone, Copy)] +pub(crate) enum Overflow { + /// Fail the job. For a command whose output IS the answer. + Fail, + /// Stop reading, kill the child, and report `truncated`. + Truncate, } /// Run a command for its stdout under the job's cancellation and timeout. @@ -3579,6 +3594,7 @@ pub(crate) async fn run_captured( w_id: &str, conn: &Connection, max_stdout_bytes: usize, + on_overflow: Overflow, ) -> error::Result { use tokio::io::AsyncReadExt; @@ -3614,6 +3630,7 @@ pub(crate) async fn run_captured( let mut out_buf = vec![0u8; 16 * 1024]; let mut err_buf = vec![0u8; 16 * 1024]; let (mut out_open, mut err_open) = (true, true); + let mut truncated = false; while out_open || err_open { tokio::select! { r = stdout_pipe.read(&mut out_buf[..]), if out_open => match r { @@ -3621,14 +3638,19 @@ pub(crate) async fn run_captured( Ok(n) => { if stdout.len() + n > max_stdout_bytes { // Killed here rather than left to `kill_on_drop` - // so the child is gone before the error unwinds, - // not merely once this future is dropped. + // so the child is gone before this returns, not + // merely once the future is dropped. let _ = child.kill().await; - return Err(Error::ExecutionErr(format!( - "{name} produced more than {} MB of output. Narrow the \ - selection, or query the relation from a SQL script.", - max_stdout_bytes / 1024 / 1024 - ))); + if on_overflow == Overflow::Fail { + return Err(Error::ExecutionErr(format!( + "{name} produced more than {} MB of output. Narrow the \ + selection, or query the relation from a SQL script.", + max_stdout_bytes / 1024 / 1024 + ))); + } + truncated = true; + out_open = false; + continue; } stdout.extend_from_slice(&out_buf[..n]); } @@ -3651,7 +3673,7 @@ pub(crate) async fn run_captured( .wait() .await .map_err(|e| Error::internal_err(format!("{name} failed: {e}")))?; - Ok((status, stdout, stderr)) + Ok((status, stdout, stderr, truncated)) }, ctx.worker_name, w_id, @@ -3661,11 +3683,14 @@ pub(crate) async fn run_captured( })), ) .await?; - let (status, stdout, stderr) = out; + let (status, stdout, stderr, truncated) = out; Ok(Captured { stdout: String::from_utf8_lossy(&stdout).to_string(), stderr: String::from_utf8_lossy(&stderr).to_string(), + // A killed child reports failure; under `Truncate` that is the ceiling's + // doing, not the project's, and the caller reads `truncated` to tell. success: status.success(), + truncated, }) } @@ -3680,7 +3705,17 @@ pub(crate) async fn run_capturing( conn: &Connection, max_stdout_bytes: usize, ) -> error::Result { - let captured = run_captured(cmd, name, ctx, job_id, w_id, conn, max_stdout_bytes).await?; + let captured = run_captured( + cmd, + name, + ctx, + job_id, + w_id, + conn, + max_stdout_bytes, + Overflow::Fail, + ) + .await?; if !captured.success { return Err(Error::ExecutionErr(format!( "{name} failed: {}", diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 63434acb9c..0aae6aea45 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -1169,12 +1169,20 @@ ref(...)` is `UnresolvedIdentifier (dbt0227)` and exit 1 under `strict`, and compiles fine under `baseline` (the default). So this is a separate `dbt compile` with its own `--target-path`, never a flag on the build, and it is opt-in per project: `column_lineage: true` in the descriptor. Off, nothing changes. On, a -project that cannot be analyzed keeps exactly the graph it had. The pass never -fails a deploy or a run. +project that cannot be analyzed keeps exactly the graph it had. + +The pass is best-effort about everything that is ITS: a wrong engine, a rejected +analysis, a missing or unreadable artifact, an over-long output, and outrunning +its own time budget all degrade to no lineage and a line in the job log. It is +not best-effort about the JOB: a cancellation or the job's own deadline fail it, +because swallowing those would let a run that blew its timeout inside an optional +annotation publish a graph and report success. The budget is half the job's +remaining wall clock, spent on the compile alone, so the build that follows +cannot be starved by it. **A failed pass still writes the index**, holding every edge of the models that did analyze, so the artifact is read whatever the exit status and partial lineage -is a normal outcome rather than an error. An unreachable *source* is milder +is a normal outcome. An unreachable *source* is milder still: `RemoteError (dbt1014)` downgrades that model to `static_analysis: off` and the compile succeeds. (Strict analysis queries the warehouse catalog for source schemas; a `ref()`ed model is inferred statically and needs no built @@ -1203,7 +1211,21 @@ what a project's index holds. Keeping it in the table is what lets a later Storage mirrors `dbt_edge` exactly: `dbt_column_edge`, keyed by (path, version, job) with the same composite foreign key to `script`, so a version's column -lineage dies with the version and a run's snapshot with the sweep. The typed +lineage dies with the version and a run's snapshot with the sweep. + +**A table of its own, not `dbt_edge.column_lineage` JSONB.** Hanging the links on +the `ref()` edge they sit beneath would inherit its clone, prune, clear and +cascade paths for free, and it does not work: a model reading `{{ this }}` gets +column lineage from itself to itself, and `parent_map` has no self-loop, because +a model does not `ref()` itself. Those pairs have no `dbt_edge` row to attach to. +The loss is not hypothetical — an incremental that selects from `{{ this }}` +(`coalesce(p.dbl, s.dbl)`, `p.up as prev_up`) yields `up → prev_up` with kind +`copy`, a drawn edge meaning "this column carries the previous run's value". +Inventing self-loop `dbt_edge` rows to hold it is not an option either: that +table is `ref()` lineage. What the separate table DOES inherit is the read: the +column edges come back from the same statement as the `ref()` edges, over a +`UNION ALL`'d edge source, so the `live`/`chosen` CTEs and the version and +editor-buffer join conditions exist once. The typed column list lands in `dbt_node.column_schema`, beside `columns` rather than merged into it — `columns` stays what the author *declared*. Both are gated on being able to read the producing project, like the model's SQL: a column-level