From 484fdb3de3952380170ec70a424a2d3ea9b92b74 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Mon, 7 Sep 2026 22:11:50 +0200 Subject: [PATCH] refactor: land dbt column lineage as storage and ingest only MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- ...c718161d95f7ad65a26525c5f43241267608d.json | 17 -- ...fc8ee144b14d564a3e589a60678b3e68effbf.json | 17 -- ...4ff3bc1143dab285e51a9b54b2c2f240b9a8c.json | 16 -- ...eef4f68cc3ecaf680c55a94175f6dfc2e2912.json | 16 -- ...10ba39025d9c9787a71a9456241cf8a9b9d82.json | 16 -- ...f16c20ea06516523bd0f3368a71f067a0af8e.json | 53 ---- ...ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json | 17 -- backend/windmill-api-assets/src/lib.rs | 266 +----------------- .../tests/dbt_pinned_graph.rs | 247 +--------------- backend/windmill-api/openapi.yaml | 116 -------- backend/windmill-api/src/jobs.rs | 86 ++---- .../windmill-worker/src/dbt_column_index.rs | 7 + backend/windmill-worker/src/dbt_executor.rs | 58 +++- docs/dbt-runtime.md | 56 ++-- .../AssetGraph/AssetGraphDetailsPane.svelte | 50 +--- .../AssetGraph/PipelineGraphEditor.svelte | 7 +- .../AssetGraph/columnLineageGraph.test.ts | 76 ----- .../assets/AssetGraph/columnLineageGraph.ts | 69 +---- .../AssetGraph/dbtColumnLineage.svelte.ts | 132 --------- .../src/lib/components/dbt/DbtEditor.svelte | 50 +--- .../lib/components/dbt/DbtModelDetails.svelte | 34 --- .../lib/components/dbt/DbtModelGraph.svelte | 44 +-- .../(logged)/pipeline/[folder]/+page.svelte | 60 +--- 23 files changed, 147 insertions(+), 1363 deletions(-) delete mode 100644 backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json delete mode 100644 backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json delete mode 100644 backend/.sqlx/query-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json delete mode 100644 backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json delete mode 100644 backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json delete mode 100644 backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json delete mode 100644 backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json delete mode 100644 frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts diff --git a/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json b/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json deleted file mode 100644 index db9d674fcf..0000000000 --- a/backend/.sqlx/query-06c1a79bfc24b17acbd79411295c718161d95f7ad65a26525c5f43241267608d.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json b/backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json deleted file mode 100644 index 00b839de6a..0000000000 --- a/backend/.sqlx/query-1008ed150f30b56baf17b3c6e6bfc8ee144b14d564a3e589a60678b3e68effbf.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json b/backend/.sqlx/query-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json deleted file mode 100644 index bdb54cfbb9..0000000000 --- a/backend/.sqlx/query-210441eb7bee09afd27a927e0e44ff3bc1143dab285e51a9b54b2c2f240b9a8c.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json b/backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json deleted file mode 100644 index a498ecccd0..0000000000 --- a/backend/.sqlx/query-895a4feb0b3cc01ad711ce0eff8eef4f68cc3ecaf680c55a94175f6dfc2e2912.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json b/backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json deleted file mode 100644 index 49d01c2b8b..0000000000 --- a/backend/.sqlx/query-9f0979110f86dffc7452d80ea0410ba39025d9c9787a71a9456241cf8a9b9d82.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json b/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json deleted file mode 100644 index 07694284d8..0000000000 --- a/backend/.sqlx/query-c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e.json +++ /dev/null @@ -1,53 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "WITH\n -- The project version that owns the asked-for relation, in the graph\n -- on screen. Not the folder-wide `live` set the graph resolves: one\n -- asset is asked about here, so the version is decided per candidate\n -- row. Usually one row; a relation a second project declares as a\n -- source has two, and each answers for its own lineage.\n owner AS (\n SELECT DISTINCT n.script_path, n.script_hash, n.job_id\n FROM dbt_node n\n WHERE n.workspace_id = $1 AND n.asset_path = $2\n -- The run's snapshot, or the deployed graph when that job stored\n -- none -- a build pins only if it wrote one.\n AND n.job_id = CASE WHEN $5::uuid IS NOT NULL AND EXISTS (\n SELECT 1 FROM dbt_graph_snapshot g\n WHERE g.workspace_id = $1 AND g.job_id = $5)\n THEN $5::uuid\n ELSE '00000000-0000-0000-0000-000000000000'::uuid END\n -- The gate, once, for the whole answer.\n AND ( $6\n OR n.script_path = ANY($7)\n OR EXISTS ( SELECT 1 FROM unnest($8::text[]) AS pfx\n WHERE n.script_path = pfx\n OR left(n.script_path, length(pfx) + 1) = pfx || '/' ) )\n AND CASE\n -- Pinned: which version comes from a job this caller was\n -- already granted, so `script` does not decide THAT — but\n -- it still decides whether the project may be read, the\n -- same second gate `script_visible` is on the graph. Being\n -- entitled to a run is not being entitled to the SQL\n -- behind it, and column lineage is that SQL's shape. A\n -- version-less row is exempt because it is an editor\n -- buffer, which has no `script` row to ask and reaches\n -- this only through the parse job that wrote it.\n WHEN $4::text IS NOT NULL\n THEN n.script_path = $4 AND n.script_hash IS NOT DISTINCT FROM $3::bigint\n AND ($3::bigint IS NULL OR EXISTS (\n SELECT 1 FROM script sc\n WHERE sc.workspace_id = $1 AND sc.path = n.script_path\n AND sc.hash = $3))\n -- A named version, for an editor open on an older one.\n -- `script` is read under RLS, so this is the visibility\n -- check as well as the existence one.\n WHEN $3::bigint IS NOT NULL\n THEN n.script_hash = $3 AND EXISTS (\n SELECT 1 FROM script sc\n WHERE sc.workspace_id = $1 AND sc.path = n.script_path\n AND sc.hash = $3)\n -- Otherwise the version deployed now: an older one's rows\n -- outlive it in `dbt_node` until the sweep, and describe a\n -- project that is no longer what runs. `language` narrows\n -- it the way the graph's own resolution does, so a path\n -- that has since become a script of another kind draws and\n -- explains the same version rather than disagreeing.\n ELSE n.script_hash = (\n SELECT sc.hash FROM script sc\n WHERE sc.workspace_id = $1 AND sc.path = n.script_path\n AND sc.language = 'dbt'\n AND sc.deleted = false AND sc.archived = false\n ORDER BY sc.created_at DESC LIMIT 1)\n END\n )\n -- DIRECT kinds only. `scan` — the column was read to produce the ROW,\n -- not the value — reaches every output column of its model, so it is\n -- most of a project's stored lineage and none of what a trace draws.\n -- It stays in the table for a later view to ask for.\n 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 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": "c742bc04736de8b2a1c47559195f16c20ea06516523bd0f3368a71f067a0af8e" -} diff --git a/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json b/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json deleted file mode 100644 index ccb9863853..0000000000 --- a/backend/.sqlx/query-f6de1512fa3e46883b32d56fd19ffe8aaf6dfe664c330ac8cea51a157a4fe52e.json +++ /dev/null @@ -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" -} diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 08b74c0311..bd08111623 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -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, -} - -#[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, -} - -async fn dbt_column_lineage( - authed: ApiAuthed, - Path(w_id): Path, - Extension(user_db): Extension, - Query(q): Query, -) -> JsonResult { - // `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, -) -> JsonResult { - // 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 = 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, asset_path: &str) -> Vec { - let keep = { - let mut incident: HashMap<(&str, &str), Vec> = 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, diff --git a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs index 5c10446c9c..5fb2c6a088 100644 --- a/backend/windmill-api-assets/tests/dbt_pinned_graph.rs +++ b/backend/windmill-api-assets/tests/dbt_pinned_graph.rs @@ -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, 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, 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) { "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, - authed: &ApiAuthed, - asset_path: &str, - pinned: Option, -) -> 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) { - 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) { - 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) { - 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) { - let job = uuid::Uuid::from_u128(7); - seed(&db, job).await; - // A second family in the same project version, reaching neither of the two - // relations `seed` wired together. - sqlx::query!( - "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id, - resource_type, name, asset_path, tags) - VALUES ($1, $2, $3, $4, 'model.p.stock', 'model', 'stock', - 'u/a/wh/analytics/stock', '{}'), - ($1, $2, $3, $4, 'model.p.stock_daily', 'model', 'stock_daily', - 'u/a/wh/analytics/stock_daily', '{}')", - WS, - PATH, - HASH, - job - ) - .execute(&db) - .await - .unwrap(); - sqlx::query!( - "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id, - parent_unique_id, parent_column, child_unique_id, - child_column, lineage_kind) - VALUES ($1, $2, $3, $4, 'model.p.stock', 'sku', 'model.p.stock_daily', 'sku', 'copy')", - WS, - PATH, - HASH, - job - ) - .execute(&db) - .await - .unwrap(); - - let admin = ApiAuthed { is_admin: true, ..outsider() }; - let pinned = - || PinnedRun { job_id: job, script_path: PATH.to_string(), script_hash: Some(HASH) }; - assert_eq!( - column_lineage(&db, &admin, "u/a/wh/analytics/orders", Some(pinned())).await, - serde_json::json!([{ - "from_asset_path": "u/a/wh/analytics/raw_orders", - "from_column": "id", - "to_asset_path": "u/a/wh/analytics/orders", - "to_column": "order_id", - "kind": "copy", - }]), - "the orders family, and not the stock one beside it in the same project" - ); - assert_eq!( - column_lineage(&db, &admin, "u/a/wh/analytics/stock_daily", Some(pinned())).await, - serde_json::json!([{ - "from_asset_path": "u/a/wh/analytics/stock", - "from_column": "sku", - "to_asset_path": "u/a/wh/analytics/stock_daily", - "to_column": "sku", - "kind": "copy", - }]), - "and the other way round — reached from the child end, which is upstream" - ); -} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 2a6d52fe35..0630b3e262 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -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: >- diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index 7705b7dd20..dbf33cdddc 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -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, } -/// 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> { +/// 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, + Extension(user_db): Extension, + Path((w_id, job_id)): Path<(String, Uuid)>, + Query(q): Query, +) -> error::JsonResult { // 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, - Extension(user_db): Extension, - Path((w_id, job_id)): Path<(String, Uuid)>, - Query(q): Query, -) -> error::JsonResult { - 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, - Extension(user_db): Extension, - Path((w_id, job_id)): Path<(String, Uuid)>, - Query(q): Query, -) -> error::JsonResult { - 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 diff --git a/backend/windmill-worker/src/dbt_column_index.rs b/backend/windmill-worker/src/dbt_column_index.rs index 3a2a4c8a66..b843c95f21 100644 --- a/backend/windmill-worker/src/dbt_column_index.rs +++ b/backend/windmill-worker/src/dbt_column_index.rs @@ -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. diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index 12a092ba71..04376dd189 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -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 { + 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>, +) -> error::Result { + Ok(arg_bool(args, "full_refresh")?.is_some_and(|v| v != descriptor.full_refresh)) +} + fn selection_is_overridden( descriptor: &DbtDescriptor, args: &HashMap>, @@ -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 diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 03ad14f2b4..2920227eb1 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -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` | diff --git a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte index 4b269142a5..2c5e5fd97e 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -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 @@ {/if} - {#if selectionColumnGraph && selectionColumnNodes.length > 0} + {#if selectionColumnGraph && assetColumnNodes(selectionColumnGraph, selection.asset_kind, selection.path).length > 0}
{/key} - {:else if selectionDbt && (selectionDbt.raw_code || selectionColumnNodes.length > 0 || selectionColumnLoading)} + {:else if selectionDbt?.raw_code} - {#if selectionColumnLoading && selectionColumnNodes.length === 0} -
- - Loading column lineage -
- {:else if selectionColumnGraph && selectionColumnNodes.length > 0} -
- -
- {/if} - {#if selectionDbt.raw_code} -
- -
- {/if} {:else}
diff --git a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte index 4064c183a7..3439fa5ad5 100644 --- a/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/PipelineGraphEditor.svelte @@ -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} diff --git a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts index 16c8713c5c..6842dcfd86 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts @@ -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()) diff --git a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts index 7172ec52e9..50fa1c4e7b 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts @@ -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> } -// 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() - const up = new Map>() - const down = new Map>() - 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() - const up = new Map>() - const down = new Map>() - 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>): Set { diff --git a/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts b/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts deleted file mode 100644 index 386399824d..0000000000 --- a/frontend/src/lib/components/assets/AssetGraph/dbtColumnLineage.svelte.ts +++ /dev/null @@ -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 { - 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(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() - - $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 - } - } -} diff --git a/frontend/src/lib/components/dbt/DbtEditor.svelte b/frontend/src/lib/components/dbt/DbtEditor.svelte index ffe2ac9540..1d3de73e58 100644 --- a/frontend/src/lib/components/dbt/DbtEditor.svelte +++ b/frontend/src/lib/components/dbt/DbtEditor.svelte @@ -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(undefined) let selectedAsset = $derived(graphSelection?.kind === 'asset' ? graphSelection : undefined) let selectedDbt = $state(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(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(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 }} /> @@ -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)} /> diff --git a/frontend/src/lib/components/dbt/DbtModelDetails.svelte b/frontend/src/lib/components/dbt/DbtModelDetails.svelte index a1b3629d35..52739ca39c 100644 --- a/frontend/src/lib/components/dbt/DbtModelDetails.svelte +++ b/frontend/src/lib/components/dbt/DbtModelDetails.svelte @@ -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 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 @@
{/if} - {#if columnLoading && columnNodes.length === 0} -
- - Loading column lineage -
- {:else if columnGraph && columnNodes.length > 0} -
- -
- {/if} - {#if showRows && preview} {#if 'error' in preview}
{preview.error}
diff --git a/frontend/src/lib/components/dbt/DbtModelGraph.svelte b/frontend/src/lib/components/dbt/DbtModelGraph.svelte index 702c8242f7..e6d948d926 100644 --- a/frontend/src/lib/components/dbt/DbtModelGraph.svelte +++ b/frontend/src/lib/components/dbt/DbtModelGraph.svelte @@ -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( - 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' }) +
@@ -469,8 +435,8 @@ {#if refreshPending}
- 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. { - 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() - 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}