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

refactor: land dbt column lineage as storage and ingest only

Browse Source 
The API surface that draws a column trace moves to a follow-up PR, on
`dbt-column-lineage-surface`. It kept generating findings — a client cache
whose premise was wrong for a two-owner relation, then staleness and a lost
retry from tightening it, and a seed walk that stopped at the first boundary —
and the fix for the last of them is a transitive owner expansion, which has to
re-apply the caller's gate to every newly discovered project. That is the same
shape as the leak four reviewers caught in the pinned arm, and it wants its own
review rather than being the fourth fix at the end of this one.

What lands here stands on its own: the analysis pass, `dbt_column_edge`,
`dbt_node.column_schema`, the engine gating and the error-contract split — plus
the one user-visible half, the typed and ordered column list, which rides the
asset graph the details pane already fetches and replaces a panel that could
only show the columns an author had documented.

Also fixes a real bug in the pass, found in review: it compiled without the
build's `--full-refresh`. `is_incremental()` branches on that flag, so an
incremental model reading `{{ this }}` compiles its self-join — and any `ref()`
inside that branch — only when the flag is absent, and the pass was storing
lineage for SQL a full-refresh run never executed. The flag now comes from one
place shared with the build, and a run that overrides it gets its own graph
rather than standing as the version's.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-07 22:11:50 +02:00
co-authored by Claude Opus 5
parent 281f992eb8
commit 484fdb3de3
23 changed files with 147 additions and 1363 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json
Generated
-17
View File
@@ -1,17 +0,0 @@
{
"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
@@ -1,17 +0,0 @@
{
"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-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json
Generated
-16
View File
@@ -1,16 +0,0 @@
{
"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, '00000000-0000-0000-0000-000000000000',\n 'model.p.raw_orders', 'id', 'model.p.orders', 'order_id', 'copy'),\n ($1, $2, $3, '00000000-0000-0000-0000-000000000000',\n 'model.p.raw_orders', 'status', 'model.p.orders', 'order_id', 'scan')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Int8"
]
},
"nullable": []
},
"hash": "210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c"
}
backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json
Generated
-16
View File
@@ -1,16 +0,0 @@
{
"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, NULL, $3, 'model.p.draft_src', 'raw', 'model.p.draft', 'clean', 'mod')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Uuid"
]
},
"nullable": []
},
"hash": "895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912"
}
backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json
Generated
-16
View File
@@ -1,16 +0,0 @@
{
"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, '00000000-0000-0000-0000-000000000000', 'model.p.raw_orders',\n 'model', 'raw_orders', 'u/a/wh/analytics/raw_orders', '{}'),\n ($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.orders',\n 'model', 'orders', 'u/a/wh/analytics/orders', '{}')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Int8"
]
},
"nullable": []
},
"hash": "9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82"
}
backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.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
@@ -1,17 +0,0 @@
{
"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/windmill-api-assets/src/lib.rs
+1 -265
View File
@@ -6,7 +6,6 @@ use axum::{
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sqlx::Row;
use std::collections::{HashMap, HashSet};
use windmill_common::{
assets::{parse_asset_trigger_ref, AssetKind, AssetUsageKind},
db::UserDB,
@@ -14,9 +13,7 @@ use windmill_common::{
utils::escape_ilike_pattern,
};
use windmill_api_auth::{
build_scope_path_filter, build_scope_path_predicate, ApiAuthed, ScopePathFilter,
};
use windmill_api_auth::{build_scope_path_predicate, ApiAuthed};
// Partition-range backfill preview. The logic (producer resolution, range
// enumeration, status join) is enterprise: the `private` build compiles the
@@ -36,7 +33,6 @@ pub fn workspaced_service() -> Router {
.route("/list_by_usages", post(list_assets_by_usages))
.route("/list_favorites", get(list_favorites))
.route("/graph", get(asset_graph))
.route("/column_lineage", get(dbt_column_lineage))
.route("/pipelines", get(list_pipeline_folders))
.route("/partitions", get(list_partitions))
.route("/partitions_in_range", get(list_partitions_in_range))
@@ -961,266 +957,6 @@ struct DbtLineageEdge {
to_asset_path: String,
}
/// One column-to-column edge, in the same terms: the two relations and the two
/// columns, never dbt's node ids.
#[derive(Serialize, Debug, PartialEq, Eq, PartialOrd, Ord)]
struct DbtColumnLineageEdge {
from_asset_path: String,
from_column: String,
to_asset_path: String,
to_column: String,
/// dbt's own word for how the value travelled: `copy` (passthrough), `mod`
/// (transformed), `scan` (read to produce the ROW rather than the value — a
/// join key, a predicate, a `group by`). Sent verbatim, including a kind
/// this engine version invented, because the renderer decides what a kind
/// means and the set is the engine's.
kind: String,
}
/// One dbt relation's column lineage: the connected component its columns sit
/// in, within the project that owns it.
///
/// Not the relation's own edges, which would stop one hop out — a trace walks
/// transitively — and not the whole project's, which carries families the
/// selected relation cannot reach. The component is what the canvas lays out,
/// so it is exactly what a consumer can draw.
///
/// Its own endpoint rather than a field on the graph, which is folder-wide and
/// polled by a run page while this is rendered for a single selection. A
/// folder's worth of edges spans many projects and many callers' access, so it
/// would need a cap, and a cap has to be applied after every filter that could
/// drop a row — which is the ordering this shape removes rather than gets
/// right. Here the filters ARE the answer: scope and visibility are decided
/// once in SQL for the script that owns the relation, the component is walked
/// over what that returns, and the size is bounded at ingest
/// (`MAX_COLUMN_EDGES` per version, of which only the direct kinds are served).
#[derive(Deserialize)]
pub struct ColumnLineageQuery {
/// The `dbt://` relation whose lineage to return.
pub asset_path: String,
/// A deployed version other than the current one, for an editor open on an
/// older one. A run's or an editor buffer's graph is NOT reachable from
/// here: it pins to a job, and that costs the job-read gate.
pub dbt_script_hash: Option<windmill_common::scripts::ScriptHash>,
}
#[derive(Serialize, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct ColumnLineageResponse {
/// Direct (`copy` / `mod`) column edges of the component this relation's
/// columns sit in, in the terms the canvas draws. Empty when the project
/// never asked for the analysis pass, which is the ordinary case.
edges: Vec<DbtColumnLineageEdge>,
}
async fn dbt_column_lineage(
authed: ApiAuthed,
Path(w_id): Path<String>,
Extension(user_db): Extension<UserDB>,
Query(q): Query<ColumnLineageQuery>,
) -> JsonResult<ColumnLineageResponse> {
// `None`: pinning to one run is job-scoped and this endpoint is authorized
// as `assets:read`. See `dbt_column_lineage_for`.
dbt_column_lineage_for(&authed, &w_id, user_db, q, None).await
}
/// One dbt relation's column lineage, optionally as one run saw it.
///
/// AUTHORIZES NOTHING BY ITSELF, on the same contract as `asset_graph_for`:
/// `assets:read` always, and the job-read gate for `Some(pinned)`, whose path
/// and hash are then taken from that job's row rather than from the caller.
pub async fn dbt_column_lineage_for(
authed: &ApiAuthed,
w_id: &str,
user_db: UserDB,
q: ColumnLineageQuery,
pinned: Option<PinnedRun>,
) -> JsonResult<ColumnLineageResponse> {
// A column-level view is the shape of what the author WROTE, so it takes the
// model's own gate rather than the relation's. Decided ONCE here, for the
// script that owns the asset, instead of per row against a cap.
let (scope_all, scope_exact, scope_prefix) =
match build_scope_path_filter(authed, "scripts", "read") {
ScopePathFilter::AllowAll => (true, Vec::new(), Vec::new()),
ScopePathFilter::Restricted { exact, prefix } => (false, exact, prefix),
};
let (pinned_path, script_hash) = match pinned.as_ref() {
// The job's own version, so a pin cannot name one project's run while
// claiming another's version — including when it names NONE, which is
// the editor buffer.
Some(p) => (Some(p.script_path.as_str()), p.script_hash),
None => (None, q.dbt_script_hash.map(|h| h.0)),
};
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
-- 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
-- row. Usually one row; a relation a second project declares as a
-- source has two, and each answers for its own lineage.
owner AS (
SELECT DISTINCT n.script_path, n.script_hash, n.job_id
FROM dbt_node n
WHERE n.workspace_id = $1 AND n.asset_path = $2
-- The run's snapshot, or the deployed graph when that job stored
-- none -- a build pins only if it wrote one.
AND n.job_id = CASE WHEN $5::uuid IS NOT NULL AND EXISTS (
SELECT 1 FROM dbt_graph_snapshot g
WHERE g.workspace_id = $1 AND g.job_id = $5)
THEN $5::uuid
ELSE '00000000-0000-0000-0000-000000000000'::uuid END
-- The gate, once, for the whole answer.
AND ( $6
OR n.script_path = ANY($7)
OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx
WHERE n.script_path = pfx
OR left(n.script_path, length(pfx) + 1) = pfx || '/' ) )
AND CASE
-- 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.
WHEN $3::bigint IS NOT NULL
THEN n.script_hash = $3 AND EXISTS (
SELECT 1 FROM script sc
WHERE sc.workspace_id = $1 AND sc.path = n.script_path
AND sc.hash = $3)
-- Otherwise the version deployed now: an older one's rows
-- outlive it in `dbt_node` until the sweep, and describe a
-- project that is no longer what runs. `language` narrows
-- it the way the graph's own resolution does, so a path
-- that has since become a script of another kind draws and
-- explains the same version rather than disagreeing.
ELSE n.script_hash = (
SELECT sc.hash FROM script sc
WHERE sc.workspace_id = $1 AND sc.path = n.script_path
AND sc.language = 'dbt'
AND sc.deleted = false AND sc.archived = false
ORDER BY sc.created_at DESC LIMIT 1)
END
)
-- 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.
SELECT p.asset_path AS "from_path!", e.parent_column AS "from_column!",
c.asset_path AS "to_path!", e.child_column AS "to_column!",
e.lineage_kind AS "kind!"
FROM dbt_column_edge e
JOIN owner o ON o.script_path = e.script_path
AND o.script_hash IS NOT DISTINCT FROM e.script_hash
AND o.job_id = e.job_id
JOIN dbt_node p ON p.workspace_id = e.workspace_id
AND p.script_path = e.script_path
AND p.script_hash IS NOT DISTINCT FROM e.script_hash
AND p.job_id = e.job_id
AND p.unique_id = e.parent_unique_id
JOIN dbt_node c ON c.workspace_id = e.workspace_id
AND c.script_path = e.script_path
AND c.script_hash IS NOT DISTINCT FROM e.script_hash
AND c.job_id = e.job_id
AND c.unique_id = e.child_unique_id
WHERE e.workspace_id = $1
AND e.lineage_kind IN ('copy', 'mod')
AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL"#,
w_id,
q.asset_path,
script_hash,
pinned_path,
pinned_job_id,
scope_all,
&scope_exact[..],
&scope_prefix[..],
)
.fetch_all(&mut *tx)
.await?;
tx.commit().await?;
let mut edges: Vec<DbtColumnLineageEdge> = rows
.into_iter()
.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.kind,
})
.collect();
// Two projects can describe one relation, so the same edge can arrive twice.
edges.sort();
edges.dedup();
Ok(Json(ColumnLineageResponse {
edges: component(edges, &q.asset_path),
}))
}
/// Keep the edges of the connected component the asked-for relation sits in.
///
/// The canvas lays out the component of the selected relation's columns, so a
/// project's other model families are edges nothing it draws can reach. Walked
/// here rather than in SQL: a recursive CTE has no index to walk, so it rescans
/// the whole edge set once per level — measured at 1.24s against 59ms for the
/// query alone on a 3000-model project, for a walk that is microseconds over a
/// map. Columns are keyed by relation, not by project, which is how the canvas
/// keys them too: two projects describing one relation draw one node.
fn component(mut edges: Vec<DbtColumnLineageEdge>, asset_path: &str) -> Vec<DbtColumnLineageEdge> {
let keep = {
let mut incident: HashMap<(&str, &str), Vec<usize>> = HashMap::new();
for (i, e) in edges.iter().enumerate() {
let from = (e.from_asset_path.as_str(), e.from_column.as_str());
let to = (e.to_asset_path.as_str(), e.to_column.as_str());
incident.entry(from).or_default().push(i);
incident.entry(to).or_default().push(i);
}
let mut stack: Vec<(&str, &str)> = incident
.keys()
.filter(|(path, _)| *path == asset_path)
.copied()
.collect();
let mut seen_node: HashSet<(&str, &str)> = stack.iter().copied().collect();
let mut seen_edge = vec![false; edges.len()];
while let Some(node) = stack.pop() {
for &i in incident.get(&node).map(Vec::as_slice).unwrap_or_default() {
if std::mem::replace(&mut seen_edge[i], true) {
continue;
}
let e = &edges[i];
let ends = [
(e.from_asset_path.as_str(), e.from_column.as_str()),
(e.to_asset_path.as_str(), e.to_column.as_str()),
];
for end in ends {
if seen_node.insert(end) {
stack.push(end);
}
}
}
}
seen_edge
};
let mut i = 0;
edges.retain(|_| {
i += 1;
keep[i - 1]
});
edges
}
async fn asset_graph(
authed: ApiAuthed,
Path(w_id): Path<String>,
backend/windmill-api-assets/tests/dbt_pinned_graph.rs
+1 -246
View File
@@ -8,9 +8,7 @@
//! whole dbt half, and every later fix in this area re-touched one of the two.
use sqlx::{Pool, Postgres};
use windmill_api_assets::{
asset_graph_for, dbt_column_lineage_for, ColumnLineageQuery, GraphQuery, PinnedRun,
};
use windmill_api_assets::{asset_graph_for, GraphQuery, PinnedRun};
use windmill_api_auth::ApiAuthed;
use windmill_common::db::UserDB;
@@ -113,20 +111,6 @@ 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!(
@@ -426,22 +410,6 @@ async fn seed_editor_graph(db: &Pool<Postgres>, job: uuid::Uuid) {
.execute(db)
.await
.unwrap();
// A version-less row's `script_hash` is NULL on both sides of every join and
// every visibility check, and `= NULL` is never true — so the column edges
// need the same NULL arm the node query has, or a buffer parse renders its
// columns and none of their lineage.
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, NULL, $3, 'model.p.draft_src', 'raw', 'model.p.draft', 'clean', 'mod')",
WS,
PATH,
job
)
.execute(db)
.await
.unwrap();
}
/// A buffer parse is the third provenance: no deployed version behind it, and
@@ -520,216 +488,3 @@ async fn an_editor_graph_renders_only_through_its_own_job(db: Pool<Postgres>) {
"nor of a run of the deployed version: {deployed_run}"
);
}
fn column_query(asset_path: &str) -> ColumnLineageQuery {
ColumnLineageQuery { asset_path: asset_path.to_string(), dbt_script_hash: None }
}
async fn column_lineage(
db: &Pool<Postgres>,
authed: &ApiAuthed,
asset_path: &str,
pinned: Option<PinnedRun>,
) -> serde_json::Value {
let res = dbt_column_lineage_for(
authed,
WS,
UserDB::new(db.clone()),
column_query(asset_path),
pinned,
)
.await
.unwrap();
serde_json::to_value(&res.0).unwrap()["edges"].clone()
}
/// The buffer parse's own lineage, which is the case the versionless rows exist
/// for. Its `script_hash` is NULL on both sides of every join and every
/// visibility check, and `= NULL` is never true — so the versionless arm has to
/// be written for it, or a parse renders its columns and none of their lineage.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn an_editor_buffers_column_lineage_answers_through_its_job(db: Pool<Postgres>) {
let parse = uuid::Uuid::from_u128(9);
seed(&db, uuid::Uuid::from_u128(7)).await;
seed_editor_graph(&db, parse).await;
let admin = ApiAuthed { is_admin: true, ..outsider() };
let pinned = PinnedRun { job_id: parse, script_path: PATH.to_string(), script_hash: None };
assert_eq!(
column_lineage(&db, &admin, "u/a/wh/analytics/draft", Some(pinned)).await,
serde_json::json!([{
"from_asset_path": "u/a/wh/analytics/draft_src",
"from_column": "raw",
"to_asset_path": "u/a/wh/analytics/draft",
"to_column": "clean",
"kind": "mod",
}]),
);
// Unpinned, the same relation resolves through the deployed version, which
// never heard of the buffer's models.
assert_eq!(
column_lineage(&db, &admin, "u/a/wh/analytics/draft", None).await,
serde_json::json!([]),
"a buffer's lineage is reachable only through the job that parsed it"
);
}
/// 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.
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
async fn column_lineage_takes_the_scripts_gate_and_only_the_direct_kinds(db: Pool<Postgres>) {
let job = uuid::Uuid::from_u128(7);
seed(&db, job).await;
// A second relation of the deployed version, feeding the first. `scan` says
// the column was read to produce the ROW rather than the value, so it
// reaches every output column of its model and is never served.
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, '00000000-0000-0000-0000-000000000000', 'model.p.raw_orders',
'model', 'raw_orders', 'u/a/wh/analytics/raw_orders', '{}'),
($1, $2, $3, '00000000-0000-0000-0000-000000000000', 'model.p.orders',
'model', 'orders', 'u/a/wh/analytics/orders', '{}')",
WS,
PATH,
HASH,
)
.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, '00000000-0000-0000-0000-000000000000',
'model.p.raw_orders', 'id', 'model.p.orders', 'order_id', 'copy'),
($1, $2, $3, '00000000-0000-0000-0000-000000000000',
'model.p.raw_orders', 'status', 'model.p.orders', 'order_id', 'scan')",
WS,
PATH,
HASH,
)
.execute(&db)
.await
.unwrap();
let admin = ApiAuthed { is_admin: true, ..outsider() };
assert_eq!(
column_lineage(&db, &admin, "u/a/wh/analytics/orders", None).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 direct edge, and not the `scan` one beside it"
);
assert_eq!(
column_lineage(&db, &outsider(), "u/a/wh/analytics/orders", None).await,
serde_json::json!([]),
"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
-116
View File
@@ -24364,57 +24364,6 @@ paths:
schema:
$ref: "#/components/schemas/AssetGraph"
/w/{workspace}/assets/column_lineage:
get:
summary: Column-level lineage of one dbt relation
description: >
The direct (`copy` / `mod`) column-to-column lineage a relation's columns
sit in — the connected component within the dbt project that owns it,
from the engine's static analysis. Not the relation's own edges, which
would stop one hop out since a column trace walks transitively, and not
the whole project's, which carries model families the selected relation
cannot reach. The component is what the canvas lays out.
Its own endpoint rather than a field on the asset graph: the graph is
folder-wide and polled by a run page, while this is rendered for one
selected asset at a time. A folder's worth of edges spans many projects
and many callers' access, so it would need a cap, and a cap has to come
after every filter that can drop a row. Here the filters are the answer:
the caller's scope and the project's visibility are decided once for the
owning script, and the size is bounded where the index is ingested.
Empty for a project that did not opt into the analysis pass
(`column_lineage: true`), which is the ordinary case. The indirect `scan`
kind is stored but never served: it reaches every output column of its
model.
operationId: getDbtColumnLineage
tags:
- asset
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- name: asset_path
in: query
required: true
description: The `dbt://` relation whose lineage to return
schema:
type: string
- name: dbt_script_hash
in: query
description: >
Pin to one deployed version, for an editor open on an older one. A
run's or an editor buffer's own graph is not reachable here: that
pins to a job, and costs the job-read gate — see
`jobs/dbt_column_lineage/{id}`.
schema:
type: string
responses:
"200":
description: the relation's column-level lineage
content:
application/json:
schema:
$ref: "#/components/schemas/DbtColumnLineage"
/w/{workspace}/assets/macros:
get:
summary: List every workspace DuckDB macro (deployed `// macros` libraries)
@@ -24606,43 +24555,6 @@ paths:
schema:
$ref: "#/components/schemas/AssetGraph"
/w/{workspace}/jobs/dbt_column_lineage/{id}:
get:
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`. 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
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- name: id
in: path
required: true
description: The job whose graph the lineage is read from
schema:
type: string
format: uuid
- name: asset_path
in: query
required: true
description: The `dbt://` relation whose lineage to return
schema:
type: string
responses:
"200":
description: the relation's column-level lineage
content:
application/json:
schema:
$ref: "#/components/schemas/DbtColumnLineage"
/w/{workspace}/jobs/run_progress/{id}:
get:
summary: List the per-relation progress one job has recorded so far
@@ -26236,34 +26148,6 @@ components:
drawn identically and the ambiguity would otherwise just move into
the editor. Omitted for the unpinned workspace graph, which spans
every project and so has no one time.
DbtColumnLineage:
type: object
description: >-
The direct column-to-column lineage a relation's columns sit in — the
connected component within the project that owns it — in the terms the
canvas draws: relations and columns, never dbt's node ids.
required: [edges]
properties:
edges:
type: array
items:
type: object
required: [from_asset_path, from_column, to_asset_path, to_column, kind]
properties:
from_asset_path:
type: string
from_column:
type: string
to_asset_path:
type: string
to_column:
type: string
kind:
type: string
description: >-
dbt's own word for how the value travelled — `copy`
(passthrough) or `mod` (transformed). Not an enum: the engine
treats the set as open.
DbtAssetProvenance:
type: object
description: >-
backend/windmill-api/src/jobs.rs
+26 -60
View File
@@ -139,7 +139,6 @@ pub fn workspaced_service() -> Router {
.route("/run_progress/{id}", get(get_run_progress))
.route("/run_assets/{id}", get(list_run_assets))
.route("/dbt_graph/{id}", get(get_dbt_run_graph))
.route("/dbt_column_lineage/{id}", get(get_dbt_run_column_lineage))
.route("/dbt_resumable/{id}", get(get_dbt_resumable))
.route(
"/dbt_resumable_script/p/{*script_path}",
@@ -892,27 +891,21 @@ struct AssetProgress {
error: Option<String>,
}
/// Which project version a dbt view pins to for this job, once the caller has
/// been shown to be entitled to it.
///
/// `Ok(None)` is "answer unpinned", not a refusal: a job that stored no graph of
/// its own — and one that has aged out of retention — is served the deployed
/// version rather than an error, so a run page keeps drawing after the run is
/// gone. Pinning needs the full job-read contract, which is why it lives on
/// `require_job_read_access` here rather than as a parameter on `/assets/*`.
/// See docs/dbt-runtime.md.
async fn dbt_pinned_run(
authed: &ApiAuthed,
db: &DB,
user_db: &UserDB,
w_id: &str,
job_id: Uuid,
view_token: Option<&str>,
) -> error::Result<Option<windmill_api_assets::PinnedRun>> {
/// The asset graph as one run saw it. Pinning to a job needs the full job-read
/// contract, so it lives on `require_job_read_access` here rather than as a
/// parameter on `/assets/graph`. See docs/dbt-runtime.md.
async fn get_dbt_run_graph(
authed: ApiAuthed,
OptViewToken(view_token): OptViewToken,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, job_id)): Path<(String, Uuid)>,
Query(q): Query<windmill_api_assets::GraphQuery>,
) -> error::JsonResult<windmill_api_assets::AssetGraphResponse> {
// The scope domain comes from the URL segment, so `/jobs` asks a scoped token
// for `jobs:read` alone while the body returned is asset data. Both are
// required: the job gate below reaches this run, this reaches assets at all.
check_scopes(authed, || "assets:read".to_string())?;
check_scopes(&authed, || "assets:read".to_string())?;
let job = sqlx::query!(
r#"SELECT created_by, runnable_path,
CASE WHEN kind = 'script' THEN runnable_id END AS script_hash,
@@ -925,69 +918,42 @@ async fn dbt_pinned_run(
AND g.script_hash IS NULL) AS "editor_graph!"
FROM v2_job WHERE id = $1 AND workspace_id = $2"#,
job_id,
w_id
&w_id
)
.fetch_optional(db)
.fetch_optional(&db)
.await?;
// Unpinned rather than 404 for a job that is gone. Reachable only with
// `assets:read`, which is exactly what the unpinned route would have cost
// for the same answer.
// No such job: answer the unpinned graph rather than 404, so a run page whose
// job has aged out of retention still draws the deployed version instead of
// an error. Reachable only with `assets:read`, which is exactly what
// `/assets/graph` would have cost for the same answer.
let Some(job) = job else {
return Ok(None);
return windmill_api_assets::asset_graph_for(&authed, &w_id, user_db, db, q, None).await;
};
require_job_read_access(
db,
user_db,
authed,
w_id,
&db,
&user_db,
&authed,
&w_id,
&job_id,
&job.created_by,
view_token,
view_token.as_deref(),
)
.await?;
// A preview or flow job names no deployed version, so there is usually no
// graph to pin to and the workspace one answers. The exception is a job that
// parsed one itself, which is what the dbt editor's refresh is: its graph
// belongs to that job alone and nothing else can reach it.
Ok(job
let pinned = job
.runnable_path
.filter(|_| job.script_hash.is_some() || job.editor_graph)
.map(|path| windmill_api_assets::PinnedRun {
job_id,
script_path: path,
script_hash: job.script_hash,
}))
}
/// The asset graph as one run saw it.
async fn get_dbt_run_graph(
authed: ApiAuthed,
OptViewToken(view_token): OptViewToken,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, job_id)): Path<(String, Uuid)>,
Query(q): Query<windmill_api_assets::GraphQuery>,
) -> error::JsonResult<windmill_api_assets::AssetGraphResponse> {
let pinned =
dbt_pinned_run(&authed, &db, &user_db, &w_id, job_id, view_token.as_deref()).await?;
});
windmill_api_assets::asset_graph_for(&authed, &w_id, user_db, db, q, pinned).await
}
/// One relation's column lineage as one run saw it — the same pin as
/// `get_dbt_run_graph`, for the trace drawn beside a node of that graph.
async fn get_dbt_run_column_lineage(
authed: ApiAuthed,
OptViewToken(view_token): OptViewToken,
Extension(db): Extension<DB>,
Extension(user_db): Extension<UserDB>,
Path((w_id, job_id)): Path<(String, Uuid)>,
Query(q): Query<windmill_api_assets::ColumnLineageQuery>,
) -> error::JsonResult<windmill_api_assets::ColumnLineageResponse> {
let pinned =
dbt_pinned_run(&authed, &db, &user_db, &w_id, job_id, view_token.as_deref()).await?;
windmill_api_assets::dbt_column_lineage_for(&authed, &w_id, user_db, q, pinned).await
}
/// Whether a `dbt retry` submitted by this caller would resume THIS run.
///
/// One failure is saved per script per execution principal, so a page showing an
backend/windmill-worker/src/dbt_column_index.rs
+7
View File
@@ -226,6 +226,13 @@ async fn compile_index(
// `run_results.json` a `dbt retry` resumes from.
cmd.env("DBT_TARGET_PATH", CLL_ARTIFACTS_DIR);
crate::dbt_executor::add_vars(&mut cmd, descriptor, inv)?;
// The BUILD's answer, not the descriptor's default: `is_incremental()`
// branches on it, so a model reading `{{ this }}` compiles its self-join —
// and any `ref()` inside that branch — only when this is absent. Guessing
// here stores lineage for SQL the run never executed.
if crate::dbt_executor::full_refresh(descriptor, inv)? {
cmd.arg("--full-refresh");
}
// 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.
backend/windmill-worker/src/dbt_executor.rs
+53 -5
View File
@@ -877,6 +877,9 @@ impl GraphRefresh {
if selection_is_overridden(descriptor, args)? {
self.per_run_models = true;
}
if full_refresh_is_overridden(descriptor, args)? {
self.per_run_models = true;
}
Ok(())
}
}
@@ -2243,8 +2246,7 @@ async fn run_dbt(
if let Some(t) = descriptor.threads {
cmd.args(["--threads", &t.to_string()]);
}
let full_refresh = arg_bool(&inv.args, "full_refresh")?.unwrap_or(descriptor.full_refresh);
if full_refresh && command != "test" {
if full_refresh(descriptor, inv)? && command != "test" {
cmd.arg("--full-refresh");
}
}
@@ -3184,10 +3186,14 @@ async fn attach_column_index(
) -> error::Result<()> {
// The nodes this graph kept, so the pass reads only rows it could store: the
// index describes the whole project, this graph one selection of it.
let kept: std::collections::HashSet<&str> =
ingested.nodes.iter().map(|n| n.unique_id.as_str()).collect();
let kept: std::collections::HashSet<&str> = ingested
.nodes
.iter()
.map(|n| n.unique_id.as_str())
.collect();
let index =
crate::dbt_column_index::collect(p, descriptor, inv, ctx, job_id, w_id, conn, &kept).await?;
crate::dbt_column_index::collect(p, descriptor, inv, ctx, job_id, w_id, conn, &kept)
.await?;
drop(kept);
let Some(index) = index else {
return Ok(());
@@ -4896,6 +4902,27 @@ fn add_selection(
///
/// A run that wants the whole project despite a descriptor selector names a
/// selection that differs — `["*"]`.
/// Whether this invocation rebuilds incremental models from scratch: the run
/// form's answer when it gave one, else the descriptor's.
///
/// Shared with the column-lineage pass rather than recomputed there, because
/// `is_incremental()` branches on it: the same model compiles to different SQL —
/// a `{{ this }}` self-join, and any `ref()` inside the incremental branch — so a
/// pass that guessed would describe a build that never ran.
pub(crate) fn full_refresh(descriptor: &DbtDescriptor, inv: &Invocation) -> error::Result<bool> {
Ok(arg_bool(&inv.args, "full_refresh")?.unwrap_or(descriptor.full_refresh))
}
/// Whether this run answered `full_refresh` differently from the deployed
/// descriptor. Like a selection override it changes what the graph describes,
/// since an incremental branch can carry its own `ref()`.
fn full_refresh_is_overridden(
descriptor: &DbtDescriptor,
args: &HashMap<String, Box<RawValue>>,
) -> error::Result<bool> {
Ok(arg_bool(args, "full_refresh")?.is_some_and(|v| v != descriptor.full_refresh))
}
fn selection_is_overridden(
descriptor: &DbtDescriptor,
args: &HashMap<String, Box<RawValue>>,
@@ -5896,6 +5923,27 @@ mod tests {
.unwrap();
assert!(untouched.publishes_ownership());
assert_eq!(untouched.snapshot_job(job), None);
// `full_refresh` decides whether `is_incremental()` is true, so an
// incremental model's self-join — and any `ref()` inside that branch —
// exists in one answer and not the other. A run that flips it describes
// a different graph, and gets its own.
let mut refreshed = GraphRefresh::default();
refreshed
.add_caller_args(&descriptor, &arg("full_refresh", "true"))
.unwrap();
assert!(refreshed.needed());
assert_eq!(refreshed.snapshot_job(job), Some(job));
// The same echo rule: the form posts the descriptor's own value back on
// every run, and reading that as an override would make each one
// caller-scoped.
let always = DbtDescriptor { full_refresh: true, ..Default::default() };
let mut echoed_flag = GraphRefresh { profile_drift: true, ..Default::default() };
echoed_flag
.add_caller_args(&always, &arg("full_refresh", "true"))
.unwrap();
assert_eq!(echoed_flag.snapshot_job(job), None);
}
// `dbt retry` restores the previous run's target/ from this directory, so two
docs/dbt-runtime.md
+19 -37
View File
@@ -1227,44 +1227,26 @@ table is `ref()` lineage. The typed column list lands in
`dbt_node.column_schema`, beside `columns` rather than merged into it —
`columns` stays what the author *declared*.
**Served from `assets/column_lineage`, keyed to one relation, not as a field on
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. That ordering is what
the separate endpoint removes rather than gets right: here the filters *are* the
answer. The caller's `scripts:read` scope and the project's visibility are decided
once in SQL, for the script that owns the relation; the size is bounded at ingest
(`MAX_COLUMN_EDGES` per version, of which only the direct kinds are served); and
what comes back is the **connected component** the relation's columns sit in,
which is exactly what the canvas lays out. Neither the relation's own edges (a
trace walks transitively, so that stops one hop out) nor the whole project's
(model families the selection cannot reach). The component is walked in Rust over
the rows the gated query returns, not by a recursive CTE: a CTE has no index to
walk, so the recursive term rescans the whole edge set once per level — measured
at 1.24s against 59ms for the query alone on a 3000-model project. Pinning to a
run's snapshot or to the editor's parse of its own buffer costs the job-read gate,
so that form is `jobs/dbt_column_lineage/{id}`, exactly as `jobs/dbt_graph/{id}`
is to `assets/graph`.
**Stored now, served later.** This change lands the ingest and the storage; the
endpoint that draws a column trace is a follow-up. What is user-visible today is
`column_schema` — every column of a relation, typed and in the order the model
emits them — which rides the asset graph the details pane already fetches, and
replaces a panel that could only list the columns an author happened to document.
The edges sit in `dbt_column_edge` waiting for their surface.
The two halves of a column trace are fetched separately and merged in the
browser: the producer half — what a DuckDB script's `// column` annotations and
inferred SQL lineage say — rides on the asset graph, and dbt's rides on this
endpoint. They meet at shared node ids, since `// column total <-
dbt://wh/analytics/orders.amount` mints the same `(dbt, path, column)` node dbt's
own lineage does, so a trace crosses the boundary in both directions rather than
ending at it.
`column_schema` is 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. A share-link viewer entitled to a dbt run therefore gets its
relations and `ref()` edges, and neither the SQL nor the columns.
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. **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.
**The analysis pass takes the build's own `--full-refresh`.** `is_incremental()`
branches on it, so an incremental model reading `{{ this }}` compiles its
self-join — and any `ref()` inside that branch — only when the flag is absent. A
pass that used the descriptor's default while the run overrode it would store
lineage for SQL that run never executed. For the same reason an invocation that
overrides the flag counts as `per_run_models`: its graph is its own, keyed to the
job, rather than standing as the version's.
## Concept mapping
@@ -1278,7 +1260,7 @@ no `script` row to ask and is reachable only through the parse job that wrote it
| `unique`/`not_null`/`accepted_values`/`relationships` | `data_tests` | exact 1:1 with the four `// data_test` kinds |
| declared column metadata | `columns` on the asset node | descriptions only, from the manifest |
| analyzed column schema | `column_schema` on the asset node | `dbt.node_columns.parquet`, opt-in |
| column-to-column lineage | column-lineage trace | `dbt.column_lineage.parquet`, opt-in |
| column-to-column lineage | `dbt_column_edge` rows (no view yet) | `dbt.column_lineage.parquet`, opt-in |
| model `tags` | node badge | `tag` |
| source freshness | `freshness` | `last_success_at` chip |
| `run_results.json` | materialization records | `record_materialization` |
frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte
+6 -44
View File
@@ -156,9 +156,6 @@
// resolved graph). Drives the transitive column-lineage trace shown for a
// selected materialized asset.
selectionColumnGraph?: ColumnLineageGraph
/** That graph still being fetched — a dbt relation's lineage is a request
* of its own, so it arrives after the selection does. */
selectionColumnLoading?: boolean
/** dbt provenance of the selected relation, when a dbt project
* materializes it — carries the model's own SQL. */
selectionDbt?: DbtAssetProvenance
@@ -292,7 +289,6 @@
onScriptRemoved,
selectionProducers = [],
selectionColumnGraph,
selectionColumnLoading = false,
selectionDbt,
schemaCanEvolve = true,
selectionForkMaterialization = undefined,
@@ -450,15 +446,6 @@
return scripts.length === 1 ? `${scripts[0].path}__dbt/${file}` : file
})
// The selected relation's own column nodes, which is what decides whether
// there is a trace to draw at all: a producer that declares no column lineage
// — or a dbt project that never asked for the analysis pass — has none.
let selectionColumnNodes = $derived(
selection?.kind === 'asset' && selectionColumnGraph
? assetColumnNodes(selectionColumnGraph, selection.asset_kind, selection.path)
: []
)
// Bound from ScriptEditor — populated by inferAssets on every code
// change. Forwarded to the page so the canvas can re-derive write
// edges as the user edits the body (e.g. renaming a CREATE TABLE
@@ -1234,7 +1221,7 @@
</span>
</div>
{/if}
{#if selectionColumnGraph && selectionColumnNodes.length > 0}
{#if selectionColumnGraph && assetColumnNodes(selectionColumnGraph, selection.asset_kind, selection.path).length > 0}
<div class="border-b shrink-0">
<ColumnLineageTrace
graph={selectionColumnGraph}
@@ -1249,7 +1236,7 @@
</div>
</div>
{/key}
{:else if selectionDbt && (selectionDbt.raw_code || selectionColumnNodes.length > 0 || selectionColumnLoading)}
{:else if selectionDbt?.raw_code}
<!-- The transform behind the node. Read-only on purpose: dbt
development is a local loop (`dbt run --select`, `dbt test`
against a dev target), and a browser textarea over one file
@@ -1262,36 +1249,11 @@
<DbtIcon width={11} height={11} />
<span class="font-mono truncate">{dbtBundlePath ?? selectionDbt.unique_id}</span
>
{#if selectionDbt.raw_code}
<span class="ml-auto shrink-0 opacity-70">read-only · edit locally</span>
{/if}
<span class="ml-auto shrink-0 opacity-70">read-only · edit locally</span>
</div>
<div class="flex-1 min-h-0 overflow-auto">
<HighlightCode language="sql" code={selectionDbt.raw_code} />
</div>
<!-- Above the SQL rather than beside it: the columns are what
the SQL below produces, so reading them in that order is
the model's own shape. Same trace component the ducklake
assets use — the graph is one graph across both. -->
{#if selectionColumnLoading && selectionColumnNodes.length === 0}
<div
class="border-b shrink-0 flex items-center gap-2 px-3 py-1.5 text-2xs text-secondary"
>
<Loader2 size={12} class="animate-spin" />
Loading column lineage
</div>
{:else if selectionColumnGraph && selectionColumnNodes.length > 0}
<div class="border-b shrink-0 overflow-auto max-h-64">
<ColumnLineageTrace
graph={selectionColumnGraph}
assetKind={selection.asset_kind}
assetPath={selection.path}
targetLabel={selectionDbt.unique_id}
/>
</div>
{/if}
{#if selectionDbt.raw_code}
<div class="flex-1 min-h-0 overflow-auto">
<HighlightCode language="sql" code={selectionDbt.raw_code} />
</div>
{/if}
</div>
{:else}
<div class="p-3 text-xs text-secondary">
frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte
+1 -6
View File
@@ -76,7 +76,6 @@
localScriptsVersion,
selectionProducers = [],
selectionColumnGraph,
selectionColumnLoading = false,
selectionDbt,
schemaCanEvolve = true,
selectionForkMaterialization = undefined,
@@ -181,11 +180,8 @@
* the selected node's source on live-reload. */
localScriptsVersion?: unknown
selectionProducers?: Array<{ kind: 'script' | 'flow'; path: string; unsaved?: boolean }>
/** Transitive column-lineage trace for the selected asset (route page). */
/** Transitive column-lineage trace for a selected ducklake asset (route page). */
selectionColumnGraph?: ColumnLineageGraph
/** That trace still being fetched — a dbt relation's is a request of its
* own, so it arrives after the selection does. */
selectionColumnLoading?: boolean
/** dbt provenance of the selected relation — carries its SQL. */
selectionDbt?: DbtAssetProvenance
schemaCanEvolve?: boolean
@@ -518,7 +514,6 @@
selection={activeDraft ? undefined : editor.selection}
selectionProducers={activeDraft ? [] : selectionProducers}
{selectionColumnGraph}
{selectionColumnLoading}
{selectionDbt}
{schemaCanEvolve}
{selectionForkMaterialization}
frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts
-76
View File
@@ -2,10 +2,7 @@ import { describe, expect, it } from 'vitest'
import type { AssetGraphResponse } from './types'
import {
buildColumnGraph,
buildDbtColumnGraph,
colNodeId,
mergeColumnGraphs,
type ColumnLineageGraph,
traceColumn,
connectedComponent,
assetColumnNodes,
@@ -123,79 +120,6 @@ describe('buildColumnGraph', () => {
})
})
describe('buildDbtColumnGraph', () => {
it('takes the direct kinds and drops any other', () => {
// `scan` means the column was read to produce the ROW — a join key, a
// predicate, a `group by` — so it reaches every output column of its model
// and is not what a column trace means. The server filters it out; this
// filters again, because the kind set is the engine's and an unknown one
// must not become an edge the trace calls data flow.
const g = buildDbtColumnGraph([
{
from_asset_path: 'main/s/stg',
from_column: 'raw_name',
to_asset_path: 'main/s/mart',
to_column: 'clean_name',
kind: 'mod'
},
{
from_asset_path: 'main/s/stg',
from_column: 'id',
to_asset_path: 'main/s/mart',
to_column: 'id',
kind: 'copy'
},
{
from_asset_path: 'main/s/stg',
from_column: 'id',
to_asset_path: 'main/s/mart',
to_column: 'clean_name',
kind: 'scan'
}
])
expect(g.up.get(colNodeId('dbt', 'main/s/mart', 'clean_name'))).toEqual(
new Set([colNodeId('dbt', 'main/s/stg', 'raw_name')])
)
expect(g.up.get(colNodeId('dbt', 'main/s/mart', 'id'))).toEqual(
new Set([colNodeId('dbt', 'main/s/stg', 'id')])
)
})
})
describe('mergeColumnGraphs', () => {
it('chains a dbt column into what a producer derives from it', () => {
// The two halves arrive separately — the producer's from the asset graph,
// dbt's from its own request — and meet at the dbt node a `// column`
// annotation names. A trace has to cross that, or a dbt selection stops
// before the script consuming it.
const dbt = buildDbtColumnGraph([
{
from_asset_path: 'main/s/stg',
from_column: 'raw',
to_asset_path: 'main/s/mart',
to_column: 'clean',
kind: 'copy'
}
])
const producer: ColumnLineageGraph = {
nodes: new Map(),
up: new Map(),
down: new Map()
}
const src = colNodeId('dbt', 'main/s/mart', 'clean')
const out = colNodeId('ducklake', 'wh/report', 'total')
producer.nodes.set(src, { kind: 'dbt', path: 'main/s/mart', column: 'clean' })
producer.nodes.set(out, { kind: 'ducklake', path: 'wh/report', column: 'total' })
producer.up.set(out, new Set([src]))
producer.down.set(src, new Set([out]))
const merged = mergeColumnGraphs(dbt, producer)
expect(traceColumn(colNodeId('dbt', 'main/s/stg', 'raw'), merged)).toEqual(
new Set([colNodeId('dbt', 'main/s/stg', 'raw'), src, out])
)
})
})
describe('traceColumn', () => {
it('returns the full upstream + downstream impact set of a source column', () => {
const g = buildColumnGraph(chainGraph())
frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts
+1 -68
View File
@@ -1,11 +1,6 @@
import type { AssetKind, DbtColumnLineage } from '$lib/gen'
import type { AssetKind } from '$lib/gen'
import type { AssetGraphResponse } from './types'
// One column-to-column edge of a dbt project's static analysis, as the API
// serves it. Taken from the generated client rather than restated: unlike the
// asset graph, this response is fetched through it.
export type DbtColumnEdge = DbtColumnLineage['edges'][number]
// A node in the column-level lineage graph: one column of one asset.
export type ColumnNode = { kind: AssetKind; path: string; column: string }
export type ColumnNodeId = string
@@ -28,15 +23,6 @@ export type ColumnLineageGraph = {
down: Map<ColumnNodeId, Set<ColumnNodeId>>
}
// Direct value flow, as dbt's static analysis labels it: `copy` passes a column
// through, `mod` transforms it. The API serves only those two — the third kind,
// `scan`, means the column was read to produce the ROW rather than the value (a
// join key, a `where` predicate, a `group by`), so it reaches EVERY output
// column of the model and would draw the diagram as a complete bipartite graph.
// Filtered here as well so a kind the engine invents cannot silently become an
// edge the trace claims is data flow.
const DIRECT_DBT_LINEAGE = new Set(['copy', 'mod'])
// Build the column graph from a resolved asset graph. A producer's
// `column_lineage` describes the columns of the asset it materializes; that
// output asset is the ducklake target it writes (v1 materialize target), found
@@ -103,59 +89,6 @@ export function buildColumnGraph(graph: AssetGraphResponse): ColumnLineageGraph
return { nodes, up, down }
}
// The same graph, from one dbt relation's column lineage. dbt arrives already
// resolved to two relations rather than anchored to a producer, and the API
// serves only the direct kinds, so this is a straight edge list.
export function buildDbtColumnGraph(edges: DbtColumnEdge[]): ColumnLineageGraph {
const nodes = new Map<ColumnNodeId, ColumnNode>()
const up = new Map<ColumnNodeId, Set<ColumnNodeId>>()
const down = new Map<ColumnNodeId, Set<ColumnNodeId>>()
const addNode = (n: ColumnNode): ColumnNodeId => {
const id = colNodeId(n.kind, n.path, n.column)
if (!nodes.has(id)) nodes.set(id, n)
return id
}
for (const e of edges) {
// Belt and braces: the API filters to `copy`/`mod`, and a kind an engine
// invents must not silently become an edge the trace calls data flow.
if (!DIRECT_DBT_LINEAGE.has(e.kind)) continue
const src = addNode({ kind: 'dbt', path: e.from_asset_path, column: e.from_column })
const out = addNode({ kind: 'dbt', path: e.to_asset_path, column: e.to_column })
if (src === out) continue
;(up.get(out) ?? up.set(out, new Set()).get(out)!).add(src)
;(down.get(src) ?? down.set(src, new Set()).get(src)!).add(out)
}
return { nodes, up, down }
}
// One graph out of several, so a trace crosses the boundary between them.
//
// The two halves reach each other through shared node ids: a producer's
// `// column out <- dbt://wh/schema/model.col` puts a `('dbt', path, column)`
// node in the producer graph under the same `colNodeId` the dbt lineage mints
// for it, so the union chains a dbt model's columns into the script that
// consumes them and on into what that script writes. Kept separate up to here
// because they are fetched separately — the producer half rides on the asset
// graph, the dbt half is asked for per selection.
export function mergeColumnGraphs(...graphs: ColumnLineageGraph[]): ColumnLineageGraph {
const nodes = new Map<ColumnNodeId, ColumnNode>()
const up = new Map<ColumnNodeId, Set<ColumnNodeId>>()
const down = new Map<ColumnNodeId, Set<ColumnNodeId>>()
for (const g of graphs) {
for (const [id, n] of g.nodes) if (!nodes.has(id)) nodes.set(id, n)
for (const [dir, into] of [
[g.up, up],
[g.down, down]
] as const) {
for (const [id, adj] of dir) {
const target = into.get(id) ?? into.set(id, new Set()).get(id)!
for (const m of adj) target.add(m)
}
}
}
return { nodes, up, down }
}
// Every node reachable from `start` by following `adj` (transitive closure,
// excluding `start` itself). Iterative to avoid deep-recursion limits.
function reach(start: ColumnNodeId, adj: Map<ColumnNodeId, Set<ColumnNodeId>>): Set<ColumnNodeId> {
frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts
-132
View File
@@ -1,132 +0,0 @@
import { untrack } from 'svelte'
import { AssetService, JobService } from '$lib/gen'
import {
buildDbtColumnGraph,
mergeColumnGraphs,
type ColumnLineageGraph
} from './columnLineageGraph'
export const EMPTY_COLUMN_GRAPH: ColumnLineageGraph = {
nodes: new Map(),
up: new Map(),
down: new Map()
}
/** Which stored dbt graph a view is drawing. A job — a run's snapshot, or the
* editor's parse of its own buffer — is asked through the job route, the only
* way to reach a graph that names no deployed version; otherwise the deployed
* version by hash, or the current one when there is no hash. */
export type DbtGraphPin = { jobId?: string; scriptHash?: string | number }
/** What a dbt relation's column lineage is doing right now. `loading` is
* separate because a project still being fetched and one that never asked for
* the analysis pass are the same empty graph otherwise. */
export type DbtColumnLineageState = {
readonly graph: ColumnLineageGraph
readonly loading: boolean
}
function pinKey(workspace: string, pin: DbtGraphPin | undefined): string {
return `${workspace}|${pin?.jobId ?? ''}|${pin?.scriptHash ?? ''}`
}
function fetchComponent(
workspace: string,
assetPath: string,
pin: DbtGraphPin | undefined
): Promise<ColumnLineageGraph> {
const req = pin?.jobId
? JobService.getDbtRunColumnLineage({ workspace, id: pin.jobId, assetPath })
: AssetService.getDbtColumnLineage({
workspace,
assetPath,
dbtScriptHash: pin?.scriptHash != undefined ? String(pin.scriptHash) : undefined
})
return req.then(
(r) => buildDbtColumnGraph(r?.edges ?? []),
// Lineage annotates a graph that renders without it, so a failed fetch
// leaves that branch unexpanded rather than putting an error over the
// model — and one failed boundary does not lose the others.
() => EMPTY_COLUMN_GRAPH
)
}
/** Follow the selection, fetching the dbt column lineage it reaches.
*
* Per asset rather than off the graph response: the graph is folder-wide and a
* run page polls it, while this is drawn for one selection. It also means the
* request is never made for a project that did not opt into the analysis pass
* — the pane simply never shows the section.
*/
export function useDbtColumnLineage(args: {
workspace: () => string | undefined
/** The dbt relations to expand. The selection itself when it is one; for a
* selection of another kind, every dbt relation its own lineage reaches —
* a ducklake table can be derived from several, and expanding only the
* first would leave the rest as leaves. */
assetPaths: () => string[]
/** The graph on screen, so the lineage describes the same project. */
pin?: () => DbtGraphPin | undefined
}): DbtColumnLineageState {
let graph = $state<ColumnLineageGraph>(EMPTY_COLUMN_GRAPH)
let loading = $state(false)
// Which pin the graph in hand was fetched against, and which relations were
// actually ASKED about under it.
let heldPin: string | undefined = undefined
let asked = new Set<string>()
$effect(() => {
const workspace = args.workspace()
const paths = args.assetPaths()
const pin = args.pin?.()
if (!workspace || paths.length === 0) {
graph = EMPTY_COLUMN_GRAPH
heldPin = undefined
asked = new Set()
loading = false
return
}
const key = pinKey(workspace, pin)
// `untrack`: this effect writes `graph`, so reading it as a dependency
// would make it retrigger itself forever.
const fresh = heldPin !== key
const base = untrack(() => (fresh ? EMPTY_COLUMN_GRAPH : graph))
if (fresh) asked = new Set()
// Only a relation this pin has ASKED about is skipped, not every relation
// present in what came back. A relation two projects describe has an owner
// row in each, and a component fetched for one of them carries that
// relation as an endpoint without the other project's half — so treating
// "appears in the graph" as "resolved" would hide exactly the cross-project
// edges the server's relation-keyed walk exists to merge.
const missing = paths.filter((p) => !asked.has(p))
if (missing.length === 0) {
graph = base
loading = false
return
}
// A selection changes faster than a request completes, so an answer is
// applied only while it is still the one being asked for.
let current = true
loading = true
Promise.all(missing.map((p) => fetchComponent(workspace, p, pin))).then((parts) => {
if (!current) return
graph = mergeColumnGraphs(base, ...parts)
heldPin = key
for (const p of missing) asked.add(p)
loading = false
})
return () => {
current = false
}
})
return {
get graph() {
return graph
},
get loading() {
return loading
}
}
}
frontend/src/lib/components/dbt/DbtEditor.svelte
+13 -37
View File
@@ -8,7 +8,7 @@
// single file, so the arguments, the run and the graph are all the project's
// whichever file happens to be open.
import { untrack } from 'svelte'
import { createEventDispatcher, onDestroy } from 'svelte'
import { createEventDispatcher, onDestroy, onMount } from 'svelte'
import type { Schema, SupportedLanguage } from '$lib/common'
import type { Preview, ScriptModule } from '$lib/gen'
import { workspaceStore } from '$lib/stores'
@@ -31,15 +31,6 @@
AssetGraphNodeData,
DbtAssetProvenance
} from '$lib/components/assets/AssetGraph/types'
import {
EMPTY_COLUMN_GRAPH,
useDbtColumnLineage,
type DbtGraphPin
} from '$lib/components/assets/AssetGraph/dbtColumnLineage.svelte'
import {
mergeColumnGraphs,
type ColumnLineageGraph
} from '$lib/components/assets/AssetGraph/columnLineageGraph'
import {
DBT_DESCRIPTOR,
DBT_MODULE_EXTENSIONS,
@@ -106,6 +97,17 @@
}
})
// The stored schema can predate the parser (a CLI push, an older version), and
// the autosave baseline is taken as the mounted editor holds it
// (`schemaAsEditorMounts`), so the descriptor is re-inferred on mount rather
// than only on its first edit — otherwise the two only agree once edited.
onMount(async () => {
await inferSchema(code)
// Same single retry as ScriptEditor: the first parse can lose a transient
// wasm init race, and the baseline (`schemaAsEditorMounts`) retries too.
if (!validDescriptor && code) await inferSchema(code)
})
function flushOpenFile() {
if (openFile !== null && modules?.[openFile]) {
modules[openFile] = { ...modules[openFile], content: editorCode }
@@ -216,28 +218,6 @@
let graphSelection = $state<AssetGraphNodeData | undefined>(undefined)
let selectedAsset = $derived(graphSelection?.kind === 'asset' ? graphSelection : undefined)
let selectedDbt = $state<DbtAssetProvenance | undefined>(undefined)
// Which graph the selection came from, so the lineage fetched below is the
// selected node's own project rather than whatever is deployed.
let selectionPin = $state<DbtGraphPin | undefined>(undefined)
// The selected model's column lineage, fetched on selection. Its own request
// rather than a field on the graph: only a project that opted into the
// analysis pass has any, and it is drawn for one model at a time.
const columnLineage = useDbtColumnLineage({
workspace: () => opWs,
assetPaths: () => {
const path = selectedDbt ? selectedAsset?.path : undefined
return path ? [path] : []
},
pin: () => selectionPin
})
// What the scripts around this project declare about its columns, off the
// same graph response the canvas drew. Merged rather than chosen between: a
// model's column and the ducklake column a script derives from it are one
// chain, and the trace has to cross that boundary.
let selectionProducerColumns = $state<ColumnLineageGraph>(EMPTY_COLUMN_GRAPH)
let selectionColumnGraph = $derived(
mergeColumnGraphs(columnLineage.graph, selectionProducerColumns)
)
// Set when the selected node came from a buffer parse: the project that parse
// ran on, which is the one its rows must come from. Undefined for a node off
// the deployed graph, which previews by version instead. Either way the rows
@@ -545,12 +525,10 @@
testRunning={testIsLoading}
testResult={testJob?.result}
selection={graphSelection}
onSelect={(sel, dbt, buffer, pin, producerColumns) => {
onSelect={(sel, dbt, buffer) => {
graphSelection = sel
selectedDbt = dbt
selectedBuffer = buffer
selectionPin = pin
selectionProducerColumns = producerColumns
}}
/>
</Pane>
@@ -572,8 +550,6 @@
{args}
fileInBundle={!!selectedDbt.original_file_path &&
!!modules?.[selectedDbt.original_file_path]}
columnGraph={selectionColumnGraph}
columnLoading={columnLineage.loading}
onOpenFile={open}
onClose={() => (graphSelection = undefined)}
/>
frontend/src/lib/components/dbt/DbtModelDetails.svelte
-34
View File
@@ -12,11 +12,6 @@
import { ClipboardCopy, Code2, FileCode2, Loader2, TableProperties, X } from 'lucide-svelte'
import { copyToClipboard } from '$lib/utils'
import type { DbtAssetProvenance } from '$lib/components/assets/AssetGraph/types'
import ColumnLineageTrace from '$lib/components/assets/AssetGraph/ColumnLineageTrace.svelte'
import {
assetColumnNodes,
type ColumnLineageGraph
} from '$lib/components/assets/AssetGraph/columnLineageGraph'
import { previewDbtRows, type DbtPreview, type DbtPreviewBuffer } from './previewRows'
import { nodeSelector } from './parseDbtRun'
@@ -39,14 +34,6 @@
args,
/** Whether this model's file is in the bundle being edited. */
fileInBundle = false,
/** The project's column-level lineage, when the descriptor asked for it.
* Fetched for this relation against the same graph the canvas draws, so
* the trace and the nodes above it describe one parse. */
columnGraph,
/** That fetch still in flight. Distinguished from an empty graph: a
* project without the analysis pass shows nothing at all, and a slow
* answer must not read as that. */
columnLoading = false,
onOpenFile,
onClose
}: {
@@ -58,8 +45,6 @@
buffer?: DbtPreviewBuffer
args?: Record<string, unknown>
fileInBundle?: boolean
columnGraph?: ColumnLineageGraph
columnLoading?: boolean
onOpenFile?: (path: string) => void
onClose?: () => void
} = $props()
@@ -144,9 +129,6 @@
}))
)
let columnsAreAnalyzed = $derived(!!dbt.column_schema?.length)
// The selected relation's own column nodes: empty for a project that never
// asked for the analysis pass, which is the ordinary case.
let columnNodes = $derived(columnGraph ? assetColumnNodes(columnGraph, 'dbt', assetPath) : [])
// `dbt show` SELECTs from the node's own relation and the worker intersects
// the selector with `resource_type:model`, so offering it on a seed, snapshot
// or source only ever produces a failed job.
@@ -313,22 +295,6 @@
</div>
{/if}
{#if columnLoading && columnNodes.length === 0}
<div class="border-b flex items-center gap-2 p-2 text-2xs text-secondary">
<Loader2 size={12} class="animate-spin" />
Loading column lineage
</div>
{:else if columnGraph && columnNodes.length > 0}
<div class="border-b overflow-auto max-h-64">
<ColumnLineageTrace
graph={columnGraph}
assetKind="dbt"
{assetPath}
targetLabel={dbt.unique_id}
/>
</div>
{/if}
{#if showRows && preview}
{#if 'error' in preview}
<div class="p-2 text-2xs text-secondary">{preview.error}</div>
frontend/src/lib/components/dbt/DbtModelGraph.svelte
+5 -39
View File
@@ -25,14 +25,6 @@
DbtAssetProvenance
} from '$lib/components/assets/AssetGraph/types'
import { useDbtRunStatus } from './runStatus.svelte'
import {
EMPTY_COLUMN_GRAPH,
type DbtGraphPin
} from '$lib/components/assets/AssetGraph/dbtColumnLineage.svelte'
import {
buildColumnGraph,
type ColumnLineageGraph
} from '$lib/components/assets/AssetGraph/columnLineageGraph'
let {
workspace,
@@ -88,19 +80,7 @@
* buffer rather than a deployed version — as submitted, not as the
* editor holds it now. Sent with the selection rather than exposed on
* its own so it can never disagree with the SQL the parent shows. */
buffer: DbtPreviewBuffer | undefined,
/** Which graph this node was taken from, so anything else fetched
* about it describes the same project: the editor's own parse job
* when the panel is pinned to one, else the deployed version. Sent
* with the selection for the same reason the buffer is — it must not
* be able to disagree with the node on screen. */
pin: DbtGraphPin,
/** Column lineage the CONSUMERS of this project declare — a script
* reading a model's column and writing a ducklake one. It comes off
* the same graph response, and the details pane merges it with the
* project's own so a trace crosses that boundary instead of ending
* at it. */
producerColumns: ColumnLineageGraph
buffer: DbtPreviewBuffer | undefined
) => void
} = $props()
@@ -384,19 +364,6 @@
// graph that actually came back.
let editorParsed = $derived(refreshJob != undefined && raw?.dbt_snapshot_job === refreshJob)
// Which stored graph is on screen. Anything the details pane fetches about a
// selected node asks for this one, so it cannot describe a node parsed from
// the buffer with the deployed version's answer.
let pin = $derived<DbtGraphPin>(
editorParsed && refreshJob ? { jobId: refreshJob } : { scriptHash: deployedHash }
)
// What the scripts around this project declare about its columns. Empty for
// the ordinary project nothing downstream annotates.
let producerColumns = $derived(
graph ? buildColumnGraph(graph) : EMPTY_COLUMN_GRAPH
)
// `untrack`, because the effect that reloads the graph clears the selection
// through here: reading the graph to describe a selection would subscribe that
// effect to the very state its own fetch writes, and it would reload forever.
@@ -407,9 +374,7 @@
sel?.kind === 'asset'
? graph?.assets.find((a) => a.kind === sel.asset_kind && a.path === sel.path)?.dbt
: undefined,
editorParsed ? parsedBuffer : undefined,
pin,
producerColumns
editorParsed ? parsedBuffer : undefined
)
)
}
@@ -440,6 +405,7 @@
if (deployedHash != undefined) return 'as of last deploy'
return 'never parsed'
})
</script>
<div class="flex flex-col h-full min-h-0">
@@ -469,8 +435,8 @@
{#if refreshPending}
<div class="shrink-0 px-2 py-1.5 border-b text-2xs text-secondary">
Still parsing. A cold worker provisions the dbt engine before it starts; a project pinned to a
worker tag nothing serves waits here indefinitely.
Still parsing. A cold worker provisions the dbt engine before it starts; a project
pinned to a worker tag nothing serves waits here indefinitely.
<a
class="text-blue-500 hover:underline"
href="{base}/run/{refreshPending}?workspace={workspace}"
frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte
+14 -46
View File
@@ -33,15 +33,9 @@
import MacroExplorerDrawer from '$lib/components/assets/AssetGraph/MacroExplorerDrawer.svelte'
import { parsePipelineAnnotations } from '$lib/components/assets/AssetGraph/parsePipelineAnnotations'
import {
assetColumnNodes,
buildColumnGraph,
connectedComponent,
mergeColumnGraphs
type ColumnLineageGraph
} from '$lib/components/assets/AssetGraph/columnLineageGraph'
import {
EMPTY_COLUMN_GRAPH,
useDbtColumnLineage
} from '$lib/components/assets/AssetGraph/dbtColumnLineage.svelte'
import { resolveGraph } from '$lib/components/assets/AssetGraph/resolveGraph'
import { normalizePipelineFolder } from '$lib/utils/pipelineFolder'
import { hideDbtRunnables } from '$lib/components/assets/AssetGraph/hideDbtRunnables'
@@ -1986,51 +1980,26 @@
?.dbt
})
// Empty graph reused when the trace isn't shown (no ducklake-asset selection,
// or a draft is actively edited) so the pane blanks out like the other
// selection overlays and `buildColumnGraph` doesn't run.
const EMPTY_COLUMN_GRAPH: ColumnLineageGraph = {
nodes: new Map(),
up: new Map(),
down: new Map()
}
// Pipeline-wide column-lineage graph, stitched across every producer's
// (inferred + annotated) `column_lineage` and the asset write-edges. Drives
// the transitive column trace in the details pane. Built from `displayGraph`
// — the exact graph the canvas renders — so the trace matches it: draft
// overlays in edit / show-drafts, deployed-only in plain View. Gated to the
// two asset kinds that can carry column lineage so it isn't rebuilt on every
// editor keystroke when the trace UI isn't even shown.
let producerColumnGraph = $derived(
pe.selection?.kind === 'asset' &&
(pe.selection.asset_kind === 'ducklake' || pe.selection.asset_kind === 'dbt')
// overlays in edit / show-drafts, deployed-only in plain View. Gated to a
// ducklake-asset selection so it isn't rebuilt on every editor keystroke when
// the trace UI isn't even shown.
let columnGraph = $derived(
pe.selection?.kind === 'asset' && pe.selection.asset_kind === 'ducklake'
? buildColumnGraph(displayGraph)
: EMPTY_COLUMN_GRAPH
)
// dbt's half comes from its own request instead: a project's static analysis
// is stored per relation and only exists if the descriptor asked for it, so
// the folder-wide graph does not carry it. A draft is never asked about —
// nothing has parsed it, so there is nothing to fetch.
//
// The relation to ask about is the selected one when it IS a dbt relation,
// and otherwise the dbt column a producer feeding this selection names as a
// source — the boundary node above. Asking there is what lets a ducklake
// selection trace back up the dbt project that fed it, rather than stopping
// at the annotation.
let dbtSeedPaths = $derived.by(() => {
const sel = pe.selection
if (pe.activeDraft || sel?.kind !== 'asset') return []
if (sel.asset_kind === 'dbt') return [sel.path]
// EVERY dbt relation this selection reaches, not the first: one output can
// be derived from several, and expanding one would leave the others as
// leaves on the canvas.
const seeds = assetColumnNodes(producerColumnGraph, sel.asset_kind, sel.path)
const paths = new Set<string>()
for (const id of connectedComponent(seeds, producerColumnGraph)) {
const node = producerColumnGraph.nodes.get(id)
if (node?.kind === 'dbt') paths.add(node.path)
}
return [...paths]
})
const dbtColumnLineage = useDbtColumnLineage({
workspace: () => $workspaceStore,
assetPaths: () => dbtSeedPaths
})
// One graph across both, so a trace crosses the dbt/ducklake boundary in
// either direction rather than stopping at it.
let columnGraph = $derived(mergeColumnGraphs(producerColumnGraph, dbtColumnLineage.graph))
// Producer-side facts for the editor's live schema-contract diagnostics:
// which assets are muted (`on_schema_change=ignore`) and which `_current`
@@ -2633,7 +2602,6 @@
{selectionProducers}
{selectionDbt}
selectionColumnGraph={pe.activeDraft ? EMPTY_COLUMN_GRAPH : columnGraph}
selectionColumnLoading={dbtColumnLineage.loading}
{schemaCanEvolve}
{selectionForkMaterialization}
{schemaContractContext}