starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-09 08:03:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: split the lineage pass by error contract, and read it in one query

Browse Source 
Round 3's findings were all consequences of round 1 and 2's fixes, clustered in
the same two files, so this reshapes those two seams rather than patching again.

The worker pass was one function being three things at once — a subprocess
runner with job-lifecycle error semantics, a bounded decoder, and a best-effort
degrader — which is why each fix to one perturbed another. It is now
`compile_index`, which owns the JOB's semantics (only a cancellation or the
job's deadline can `Err`; a non-zero exit, the output ceiling and the phase
budget are outcomes), and `read_index`, which owns the ARTIFACT's and knows
nothing about the job. The budget wraps the compile alone, so a decode can no
longer outlive the timeout that reported the build would get the rest. The
output ceiling likewise becomes a value rather than a job error, for the caller
that can carry on without the tail of a compile's stdout.

The column edges were read by a fourth hand-written copy of the `live`/`chosen`
CTEs and the version/editor-buffer join conditions, and copying them is what
dropped the `script_hash IS NULL` arm and hid every buffer parse's lineage. Both
kinds of edge now come from ONE statement over a `UNION ALL`'d edge source, so
those conditions exist once. The union is at the source rather than a join
because column lineage can name a node pair `dbt_edge` has no row for: a model
reading `{{ this }}` gets edges from itself to itself, and `parent_map` has no
self-loop.

The cap on the column half now sits after the scope filter, the visibility
check and the graph joins — the scope moved into SQL via the existing
`ScopePathFilter` — so a row the caller may not read can no longer spend it and
leave an allowed project's trace short.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01PRtsPQ3Ck69Fu9DNr7bMcJ
This commit is contained in:
Ruben Fiszel
2026-09-04 23:09:10 +02:00
co-authored by Claude Opus 5
parent c0aeed2d18
commit 542135f52b
7 changed files with 440 additions and 307 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-7e7cbe03e512096ac65acce50c8be0863f85119f0d49d18553b240855e28a3ad.json
Generated
+66
View File
File diff suppressed because one or more lines are too long
backend/.sqlx/query-fad10d08b6a7fb4c6974a6dce79028f9bae1b9a9806c9efb2844f7e7e8196f5b.json
Generated
-32
View File
@@ -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"
}
backend/windmill-api-assets/src/lib.rs
+174 -162
View File
@@ -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<DbtLineageEdge> = 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<DbtLineageEdge> = Vec::new();
let mut dbt_column_edges: Vec<DbtColumnLineageEdge> = 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<DbtColumnLineageEdge> = 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();
backend/windmill-common/src/dbt_manifest.rs
+5 -4
View File
@@ -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))
backend/windmill-worker/src/dbt_column_index.rs
+121 -92
View File
@@ -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<Option<ColumnIndex>> {
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<Option<ColumnIndex>> {
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<crate::dbt_executor::Captured>,
}
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<Option<Compiled>> {
// 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<Duration> {
ctx.timeout()
.map(|left| Duration::from_secs((left.max(0) as u64 / 2).max(1)))
backend/windmill-worker/src/dbt_executor.rs
+48 -13
View File
@@ -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<Captured> {
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<Captured> {
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: {}",
docs/dbt-runtime.md
+26 -4
View File
@@ -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