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

fix: gate pinned column lineage on reading the project, and answer the component

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-05 04:36:36 +02:00
co-authored by Claude Opus 5
parent c3335b9ee3
commit 1491aefbb2
10 changed files with 422 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json
Generated
+17
View File
@@ -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"
}
backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json
Generated
+17
View File
@@ -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"
}
backend/.sqlx/query-6efbc5a934d1b880de034cc1e5aa582e7b06e083dd75445eeff365e8541c2ac6.json
Generated
+53
View File
File diff suppressed because one or more lines are too long
backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json
Generated
+17
View File
@@ -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"
}
backend/.sqlx/query-fb2320986fa4f05663f734e694ddd4b5528b92319bfe274f12e147ce0882fcb8.json
Generated
-53
View File
@@ -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"
}
backend/windmill-api-assets/src/lib.rs
+96 -32
View File
@@ -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,
backend/windmill-api-assets/tests/dbt_pinned_graph.rs
+115
View File
@@ -112,6 +112,20 @@ async fn seed(db: &Pool<Postgres>, 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<Postg
);
}
/// Being entitled to a RUN is not being entitled to the SQL behind it, and
/// column lineage is that SQL's shape. The pinned graph draws the relations for
/// a share-link viewer and redacts what the author wrote; the lineage is the
/// second, and resolving the version from the job must not be mistaken for
/// deciding that too.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn a_pinned_run_does_not_hand_over_the_projects_column_lineage(db: Pool<Postgres>) {
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<Postgres>) {
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"
);
}
backend/windmill-api/openapi.yaml
+16 -8
View File
@@ -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
backend/windmill-worker/src/dbt_column_index.rs
+77 -46
View File
@@ -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, &note, &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<ColumnIndex> {
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<String> = 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<Row>` — 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(())
}
docs/dbt-runtime.md
+14 -4
View File
@@ -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