diff --git a/backend/.sqlx/query-178cb9cd8dfda66e878cd924f64a58ec742d7079703d417c886c8cf0a0e767a6.json b/backend/.sqlx/query-178cb9cd8dfda66e878cd924f64a58ec742d7079703d417c886c8cf0a0e767a6.json new file mode 100644 index 0000000000..7bb2cac5db --- /dev/null +++ b/backend/.sqlx/query-178cb9cd8dfda66e878cd924f64a58ec742d7079703d417c886c8cf0a0e767a6.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dbt_column_edge\n WHERE workspace_id = $1 AND script_path = $2 AND script_hash = $3", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "178cb9cd8dfda66e878cd924f64a58ec742d7079703d417c886c8cf0a0e767a6" +} diff --git a/backend/.sqlx/query-9309262d8a37827e5ab0b3534d1595cb6d94118432fd1770f1f89f7cb52a4f12.json b/backend/.sqlx/query-4efca4ff8de0bd297de9eaf0fb7f86f320e4ff58aac8794248925c084707f9b5.json similarity index 70% rename from backend/.sqlx/query-9309262d8a37827e5ab0b3534d1595cb6d94118432fd1770f1f89f7cb52a4f12.json rename to backend/.sqlx/query-4efca4ff8de0bd297de9eaf0fb7f86f320e4ff58aac8794248925c084707f9b5.json index da77dc1de5..134e58ed58 100644 --- a/backend/.sqlx/query-9309262d8a37827e5ab0b3534d1595cb6d94118432fd1770f1f89f7cb52a4f12.json +++ b/backend/.sqlx/query-4efca4ff8de0bd297de9eaf0fb7f86f320e4ff58aac8794248925c084707f9b5.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "WITH live AS (\n -- The graph is stored per deployed VERSION, so this endpoint — which\n -- describes the project as it is now — takes the newest live one per\n -- path. Resolved once here rather than per row: a correlated lookup\n -- on every node is what makes these queries fall over.\n SELECT * FROM (\n SELECT DISTINCT ON (s.path) s.path, s.hash\n FROM script s\n WHERE $5::text IS NULL AND s.workspace_id = $1 AND s.language = 'dbt'\n AND ($3::bigint IS NULL OR s.hash = $3)\n -- A pinned version may be archived by now; that is precisely\n -- the case a historical run needs, so the liveness filter\n -- applies only when picking the current one.\n AND ($3::bigint IS NOT NULL OR (s.deleted = false AND s.archived = false))\n ORDER BY s.path, s.created_at DESC\n ) cur\n UNION ALL\n -- A pinned run names its own version, so `script` is not consulted:\n -- under RLS it would answer for the CALLER's grants on the project,\n -- emptying the graph for a share-link viewer who is entitled to the\n -- run but not the script. A NULL hash here is a job that names no\n -- version at all — an editor buffer parse — and matches only the\n -- version-less rows that parse stored.\n SELECT $5::text, $3::bigint WHERE $5::text IS NOT NULL\n ),\n -- The run's own snapshot when it left one, the version's graph\n -- otherwise. A static descriptor never snapshots, so all of its runs\n -- fall through to the same rows. Existence comes from the marker, not\n -- from a node row: a dynamic run that disabled every model has a\n -- snapshot whose graph is legitimately empty.\n chosen AS (\n -- No visibility check on the job here: reaching this with a job at\n -- all means the caller passed `require_job_read_access` for it, and\n -- re-deciding it under plain RLS can only DISAGREE with that answer\n -- — silently, by falling back to the deployed graph rather than\n -- erroring. A share-link viewer is entitled to the run and would be\n -- shown a different run's model set. See `asset_graph_for`.\n SELECT CASE WHEN $4::uuid IS NOT NULL AND EXISTS (\n SELECT 1 FROM dbt_graph_snapshot g\n WHERE g.workspace_id = $1 AND g.job_id = $4)\n THEN $4::uuid\n ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id\n ),\n scoped AS (\n SELECT n.script_path, n.unique_id FROM dbt_node n\n -- `=` still, with the NULL-to-NULL case spelled out and gated on\n -- the pin: a version-less row's hash is NULL on both sides, which\n -- `=` never matches, but `IS NOT DISTINCT FROM` would cost the\n -- equality its index bound on the UNPINNED workspace graph — the\n -- hot path. Unpinned, `$5` is NULL and the second arm folds away.\n JOIN live l ON l.path = n.script_path\n AND (n.script_hash = l.hash\n OR ($5::text IS NOT NULL AND l.hash IS NULL\n AND n.script_hash IS NULL))\n JOIN chosen ch ON ch.job_id = n.job_id\n WHERE n.workspace_id = $1 AND n.asset_path IS NOT NULL\n -- Unpinned, the scope is the relations in view: `asset` says\n -- which of them this folder touches. Pinned, that table is the\n -- WRONG scope — it holds one row set per path, describing the\n -- current deploy, so a model this version had and the current one\n -- dropped would be filtered out of its own run's graph. The\n -- pinned graph's nodes are the scope. Keyed on the pin rather\n -- than on the hash: an editor parse pins without naming one, and\n -- its models are precisely the ones `asset` does not know yet.\n AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL\n OR n.asset_path IN (\n SELECT path FROM asset\n WHERE workspace_id = $1 AND kind = 'dbt'\n AND ($2::text IS NULL OR usage_path LIKE $2)))\n )\n SELECT n.script_path AS \"script_path!\", n.unique_id AS \"unique_id!\",\n n.resource_type AS \"resource_type!\", n.name AS \"name!\", n.asset_path,\n n.materialized, n.materialize_strategy, n.tags AS \"tags!\", n.description,\n n.test_kind, n.test_column, n.test_args, n.severity, n.attached_node,\n n.columns, n.freshness,\n n.raw_code, n.original_file_path,\n -- Whether the caller may read the project this row describes.\n -- The query deliberately reaches outside the requested folder\n -- so an in-scope consumer can explain the relation it reads,\n -- and `dbt_node` carries no RLS of its own; the relation's\n -- SHAPE is fine to answer that way, everything the project's\n -- author WROTE is not. Applied in Rust, over one predicate, so\n -- the fields it covers are named in one place. This runs in the\n -- authed transaction, so `script`'s RLS answers it. Matched on\n -- the HASH as well: `extra_perms` is per row, so a path\n -- recreated with narrower ones leaves the archived version\n -- readable, and a path-only probe would answer for THAT grant\n -- while returning this version's source.\n --\n -- A version-less row has no `script` row to ask, and needs\n -- none: it exists only because this caller's own parse job\n -- created it from a buffer they wrote, and the unpinned `live`\n -- branch — fed from `script` — can never join to one.\n (n.script_hash IS NULL OR EXISTS (\n SELECT 1 FROM script sc\n WHERE sc.workspace_id = n.workspace_id AND sc.path = n.script_path\n AND sc.hash = n.script_hash\n )) AS \"script_visible!\"\n FROM dbt_node n\n JOIN live l ON l.path = n.script_path\n AND (n.script_hash = l.hash\n OR ($5::text IS NOT NULL AND l.hash IS NULL\n AND n.script_hash IS NULL))\n -- Every join onto `dbt_node` needs this, not just the scoping CTE:\n -- `job_id` is part of the key, so without it each model comes back\n -- once per retained snapshot plus once for the version's graph.\n JOIN chosen ch ON ch.job_id = n.job_id\n WHERE n.workspace_id = $1\n -- Joined on BOTH columns: a dbt `unique_id` is project-local, so\n -- two projects with the same model name would otherwise pull each\n -- other's rows.\n AND (EXISTS (SELECT 1 FROM scoped s\n WHERE s.script_path = n.script_path\n AND s.unique_id = n.unique_id)\n OR EXISTS (SELECT 1 FROM scoped s\n WHERE s.script_path = n.script_path\n AND s.unique_id = n.attached_node))\n ORDER BY n.script_path, n.unique_id", + "query": "WITH live AS (\n -- The graph is stored per deployed VERSION, so this endpoint — which\n -- describes the project as it is now — takes the newest live one per\n -- path. Resolved once here rather than per row: a correlated lookup\n -- on every node is what makes these queries fall over.\n SELECT * FROM (\n SELECT DISTINCT ON (s.path) s.path, s.hash\n FROM script s\n WHERE $5::text IS NULL AND s.workspace_id = $1 AND s.language = 'dbt'\n AND ($3::bigint IS NULL OR s.hash = $3)\n -- A pinned version may be archived by now; that is precisely\n -- the case a historical run needs, so the liveness filter\n -- applies only when picking the current one.\n AND ($3::bigint IS NOT NULL OR (s.deleted = false AND s.archived = false))\n ORDER BY s.path, s.created_at DESC\n ) cur\n UNION ALL\n -- A pinned run names its own version, so `script` is not consulted:\n -- under RLS it would answer for the CALLER's grants on the project,\n -- emptying the graph for a share-link viewer who is entitled to the\n -- run but not the script. A NULL hash here is a job that names no\n -- version at all — an editor buffer parse — and matches only the\n -- version-less rows that parse stored.\n SELECT $5::text, $3::bigint WHERE $5::text IS NOT NULL\n ),\n -- The run's own snapshot when it left one, the version's graph\n -- otherwise. A static descriptor never snapshots, so all of its runs\n -- fall through to the same rows. Existence comes from the marker, not\n -- from a node row: a dynamic run that disabled every model has a\n -- snapshot whose graph is legitimately empty.\n chosen AS (\n -- No visibility check on the job here: reaching this with a job at\n -- all means the caller passed `require_job_read_access` for it, and\n -- re-deciding it under plain RLS can only DISAGREE with that answer\n -- — silently, by falling back to the deployed graph rather than\n -- erroring. A share-link viewer is entitled to the run and would be\n -- shown a different run's model set. See `asset_graph_for`.\n SELECT CASE WHEN $4::uuid IS NOT NULL AND EXISTS (\n SELECT 1 FROM dbt_graph_snapshot g\n WHERE g.workspace_id = $1 AND g.job_id = $4)\n THEN $4::uuid\n ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id\n ),\n scoped AS (\n SELECT n.script_path, n.unique_id FROM dbt_node n\n -- `=` still, with the NULL-to-NULL case spelled out and gated on\n -- the pin: a version-less row's hash is NULL on both sides, which\n -- `=` never matches, but `IS NOT DISTINCT FROM` would cost the\n -- equality its index bound on the UNPINNED workspace graph — the\n -- hot path. Unpinned, `$5` is NULL and the second arm folds away.\n JOIN live l ON l.path = n.script_path\n AND (n.script_hash = l.hash\n OR ($5::text IS NOT NULL AND l.hash IS NULL\n AND n.script_hash IS NULL))\n JOIN chosen ch ON ch.job_id = n.job_id\n WHERE n.workspace_id = $1 AND n.asset_path IS NOT NULL\n -- Unpinned, the scope is the relations in view: `asset` says\n -- which of them this folder touches. Pinned, that table is the\n -- WRONG scope — it holds one row set per path, describing the\n -- current deploy, so a model this version had and the current one\n -- dropped would be filtered out of its own run's graph. The\n -- pinned graph's nodes are the scope. Keyed on the pin rather\n -- than on the hash: an editor parse pins without naming one, and\n -- its models are precisely the ones `asset` does not know yet.\n AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL\n OR n.asset_path IN (\n SELECT path FROM asset\n WHERE workspace_id = $1 AND kind = 'dbt'\n AND ($2::text IS NULL OR usage_path LIKE $2)))\n )\n SELECT n.script_path AS \"script_path!\", n.unique_id AS \"unique_id!\",\n n.resource_type AS \"resource_type!\", n.name AS \"name!\", n.asset_path,\n n.materialized, n.materialize_strategy, n.tags AS \"tags!\", n.description,\n n.test_kind, n.test_column, n.test_args, n.severity, n.attached_node,\n n.columns, n.column_schema, n.freshness,\n n.raw_code, n.original_file_path,\n -- Whether the caller may read the project this row describes.\n -- The query deliberately reaches outside the requested folder\n -- so an in-scope consumer can explain the relation it reads,\n -- and `dbt_node` carries no RLS of its own; the relation's\n -- SHAPE is fine to answer that way, everything the project's\n -- author WROTE is not. Applied in Rust, over one predicate, so\n -- the fields it covers are named in one place. This runs in the\n -- authed transaction, so `script`'s RLS answers it. Matched on\n -- the HASH as well: `extra_perms` is per row, so a path\n -- recreated with narrower ones leaves the archived version\n -- readable, and a path-only probe would answer for THAT grant\n -- while returning this version's source.\n --\n -- A version-less row has no `script` row to ask, and needs\n -- none: it exists only because this caller's own parse job\n -- created it from a buffer they wrote, and the unpinned `live`\n -- branch — fed from `script` — can never join to one.\n (n.script_hash IS NULL OR EXISTS (\n SELECT 1 FROM script sc\n WHERE sc.workspace_id = n.workspace_id AND sc.path = n.script_path\n AND sc.hash = n.script_hash\n )) AS \"script_visible!\"\n FROM dbt_node n\n JOIN live l ON l.path = n.script_path\n AND (n.script_hash = l.hash\n OR ($5::text IS NOT NULL AND l.hash IS NULL\n AND n.script_hash IS NULL))\n -- Every join onto `dbt_node` needs this, not just the scoping CTE:\n -- `job_id` is part of the key, so without it each model comes back\n -- once per retained snapshot plus once for the version's graph.\n JOIN chosen ch ON ch.job_id = n.job_id\n WHERE n.workspace_id = $1\n -- Joined on BOTH columns: a dbt `unique_id` is project-local, so\n -- two projects with the same model name would otherwise pull each\n -- other's rows.\n AND (EXISTS (SELECT 1 FROM scoped s\n WHERE s.script_path = n.script_path\n AND s.unique_id = n.unique_id)\n OR EXISTS (SELECT 1 FROM scoped s\n WHERE s.script_path = n.script_path\n AND s.unique_id = n.attached_node))\n ORDER BY n.script_path, n.unique_id", "describe": { "columns": [ { @@ -80,21 +80,26 @@ }, { "ordinal": 15, - "name": "freshness", + "name": "column_schema", "type_info": "Jsonb" }, { "ordinal": 16, + "name": "freshness", + "type_info": "Jsonb" + }, + { + "ordinal": 17, "name": "raw_code", "type_info": "Text" }, { - "ordinal": 17, + "ordinal": 18, "name": "original_file_path", "type_info": "Text" }, { - "ordinal": 18, + "ordinal": 19, "name": "script_visible!", "type_info": "Bool" } @@ -127,8 +132,9 @@ true, true, true, + true, null ] }, - "hash": "9309262d8a37827e5ab0b3534d1595cb6d94118432fd1770f1f89f7cb52a4f12" + "hash": "4efca4ff8de0bd297de9eaf0fb7f86f320e4ff58aac8794248925c084707f9b5" } diff --git a/backend/.sqlx/query-97fd4f9939d96176f4cfc4a07e69466279aad62f33c57f392835e8265ed40f26.json b/backend/.sqlx/query-97fd4f9939d96176f4cfc4a07e69466279aad62f33c57f392835e8265ed40f26.json new file mode 100644 index 0000000000..b98a71fb8a --- /dev/null +++ b/backend/.sqlx/query-97fd4f9939d96176f4cfc4a07e69466279aad62f33c57f392835e8265ed40f26.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT count(*) FROM dbt_column_edge WHERE workspace_id = $1 AND script_hash = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "97fd4f9939d96176f4cfc4a07e69466279aad62f33c57f392835e8265ed40f26" +} diff --git a/backend/.sqlx/query-e5417b36319dafd5fcb2e42d812f778b2d3cf8b0f13cf4620192604cc3824a77.json b/backend/.sqlx/query-e5417b36319dafd5fcb2e42d812f778b2d3cf8b0f13cf4620192604cc3824a77.json new file mode 100644 index 0000000000..797fee956b --- /dev/null +++ b/backend/.sqlx/query-e5417b36319dafd5fcb2e42d812f778b2d3cf8b0f13cf4620192604cc3824a77.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dbt_column_edge\n WHERE job_id <> '00000000-0000-0000-0000-000000000000'\n AND ingested_at < now() - make_interval(days => $1)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Int4" + ] + }, + "nullable": [] + }, + "hash": "e5417b36319dafd5fcb2e42d812f778b2d3cf8b0f13cf4620192604cc3824a77" +} diff --git a/backend/.sqlx/query-f49cd9b5ea1e94d7cd10dcd1706cad62b5c1eff191f06905a3a78eb600e3bb4b.json b/backend/.sqlx/query-f49cd9b5ea1e94d7cd10dcd1706cad62b5c1eff191f06905a3a78eb600e3bb4b.json new file mode 100644 index 0000000000..75526d2db8 --- /dev/null +++ b/backend/.sqlx/query-f49cd9b5ea1e94d7cd10dcd1706cad62b5c1eff191f06905a3a78eb600e3bb4b.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dbt_column_edge WHERE workspace_id = $1 AND script_path = $2\n AND script_hash = $3 AND job_id = $4", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "f49cd9b5ea1e94d7cd10dcd1706cad62b5c1eff191f06905a3a78eb600e3bb4b" +} diff --git a/backend/.sqlx/query-f73676fe0138004a73cbf16eea680a0434da121dc2c81f59cdb63d651273c463.json b/backend/.sqlx/query-f73676fe0138004a73cbf16eea680a0434da121dc2c81f59cdb63d651273c463.json new file mode 100644 index 0000000000..f5d4c76896 --- /dev/null +++ b/backend/.sqlx/query-f73676fe0138004a73cbf16eea680a0434da121dc2c81f59cdb63d651273c463.json @@ -0,0 +1,50 @@ +{ + "db_name": "PostgreSQL", + "query": "WITH live AS (\n SELECT * FROM (\n SELECT DISTINCT ON (s.path) s.path, s.hash\n FROM script s\n WHERE $5::text IS NULL AND s.workspace_id = $1 AND s.language = 'dbt'\n AND ($3::bigint IS NULL OR s.hash = $3)\n AND ($3::bigint IS NOT NULL OR (s.deleted = false AND s.archived = false))\n ORDER BY s.path, s.created_at DESC\n ) cur\n UNION ALL\n SELECT $5::text, $3::bigint WHERE $5::text IS NOT NULL\n ),\n chosen AS (\n SELECT CASE WHEN $4::uuid IS NOT NULL AND EXISTS (\n SELECT 1 FROM dbt_graph_snapshot g\n WHERE g.workspace_id = $1 AND g.job_id = $4)\n THEN $4::uuid\n ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id\n )\n SELECT p.asset_path AS \"from_path!\", e.parent_column AS \"from_column!\",\n c.asset_path AS \"to_path!\", e.child_column AS \"to_column!\",\n e.lineage_kind AS \"lineage_kind!\"\n FROM dbt_column_edge e\n JOIN live l ON l.path = e.script_path\n AND (e.script_hash = l.hash\n OR ($5::text IS NOT NULL AND l.hash IS NULL\n AND e.script_hash IS NULL))\n JOIN chosen ch ON ch.job_id = e.job_id\n JOIN dbt_node p ON p.workspace_id = e.workspace_id\n AND p.script_path = e.script_path\n AND (p.script_hash = e.script_hash\n OR ($5::text IS NOT NULL AND e.script_hash IS NULL\n AND p.script_hash IS NULL))\n AND p.job_id = ch.job_id\n AND p.unique_id = e.parent_unique_id\n JOIN dbt_node c ON c.workspace_id = e.workspace_id\n AND c.script_path = e.script_path\n AND (c.script_hash = e.script_hash\n OR ($5::text IS NOT NULL AND e.script_hash IS NULL\n AND c.script_hash IS NULL))\n AND c.job_id = ch.job_id\n AND c.unique_id = e.child_unique_id\n WHERE e.workspace_id = $1\n AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL\n AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL OR EXISTS (\n SELECT 1 FROM asset a\n WHERE a.workspace_id = $1 AND a.kind = 'dbt'\n AND a.path = c.asset_path\n AND ($2::text IS NULL OR a.usage_path LIKE $2)))", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "from_path!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "from_column!", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "to_path!", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "to_column!", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "lineage_kind!", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8", + "Uuid", + "Text" + ] + }, + "nullable": [ + true, + false, + true, + false, + false + ] + }, + "hash": "f73676fe0138004a73cbf16eea680a0434da121dc2c81f59cdb63d651273c463" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index d314168b35..543ca5f194 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -16694,6 +16694,7 @@ dependencies = [ "opentelemetry 0.30.0", "opentelemetry-proto 0.30.0", "oracle", + "parquet", "pem 3.0.6", "pep440_rs", "postgres-native-tls 0.5.3", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index f432f022c5..75ca03758e 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -660,6 +660,12 @@ process-wrap = { version = "8.2.1", features = ["tokio1"] } systemstat = "0.2.4" datafusion = "47.0.0" +# The row API only: a dbt engine's parquet index is six string columns, so this +# needs no arrow and no writer. `parquet` is already in the tree with `arrow` for +# every shipped edition (`oss_core`), and cargo unifies the features there; this +# set is what a build WITHOUT object storage compiles. ZSTD is what the engine +# writes today, snap what parquet writers most often default to. +parquet = { version = "55.2.0", default-features = false, features = ["snap", "zstd"] } object_store = { git = "https://github.com/apache/arrow-rs-object-store", rev = "36752c975d4f29e20b57c91f81a10872dcd48ae7", features = ["aws", "azure", "gcp"] } openidconnect = { version = "4.0.0-rc.1" } aws-config = "^1" diff --git a/backend/migrations/20260904143633_dbt_column_lineage.down.sql b/backend/migrations/20260904143633_dbt_column_lineage.down.sql new file mode 100644 index 0000000000..a053a94291 --- /dev/null +++ b/backend/migrations/20260904143633_dbt_column_lineage.down.sql @@ -0,0 +1,2 @@ +DROP TABLE IF EXISTS dbt_column_edge; +ALTER TABLE dbt_node DROP COLUMN IF EXISTS column_schema; diff --git a/backend/migrations/20260904143633_dbt_column_lineage.up.sql b/backend/migrations/20260904143633_dbt_column_lineage.up.sql new file mode 100644 index 0000000000..c725fc757c --- /dev/null +++ b/backend/migrations/20260904143633_dbt_column_lineage.up.sql @@ -0,0 +1,62 @@ +-- Column-level lineage, from the engine's own static analysis. +-- +-- `manifest.json` carries none, which is why decision 14 recorded the feature as +-- unavailable. The edges exist in a different artifact: an engine that does +-- static analysis writes `target/index/dbt.column_lineage.parquet` under +-- `dbt compile --static-analysis strict --write-index`. That pass is opt-in per +-- project (`column_lineage: true`), because strict analysis rejects SQL the +-- default accepts and must never become a silent requirement of running a build. + +-- One column-to-column edge, keyed exactly like `dbt_edge`: a version's graph +-- dies with its version through the composite foreign key, a run's snapshot is +-- keyed by `job_id` with the zero UUID meaning "the version's own graph", and an +-- editor buffer's parse carries a NULL `script_hash` keyed to its preview job. +CREATE TABLE IF NOT EXISTS dbt_column_edge ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + script_path VARCHAR(255) NOT NULL, + script_hash BIGINT, + job_id UUID NOT NULL DEFAULT '00000000-0000-0000-0000-000000000000', + parent_unique_id TEXT NOT NULL, + parent_column TEXT NOT NULL, + child_unique_id TEXT NOT NULL, + child_column TEXT NOT NULL, + -- dbt's own word for how the value travelled: `copy` (passthrough), `mod` + -- (transformed), `scan` (the column was read to produce the ROW rather than + -- the value -- a join key, a `where` predicate, a `group by`). TEXT rather + -- than an enum because the engine treats the set as open: its own reader maps + -- those three and returns anything else verbatim. + lineage_kind TEXT NOT NULL, + ingested_at TIMESTAMPTZ NOT NULL DEFAULT now(), + -- Two partial unique indexes rather than a primary key, for the reason + -- 20260801121717 gives: a versioned graph is keyed by its version, a buffer + -- parse by its job alone. + CONSTRAINT dbt_column_edge_script_fkey FOREIGN KEY (workspace_id, script_hash) + REFERENCES script (workspace_id, hash) ON DELETE CASCADE ON UPDATE CASCADE +); + +CREATE UNIQUE INDEX IF NOT EXISTS dbt_column_edge_versioned_key + ON dbt_column_edge (workspace_id, script_path, script_hash, job_id, + parent_unique_id, parent_column, child_unique_id, child_column) + WHERE script_hash IS NOT NULL; +CREATE UNIQUE INDEX IF NOT EXISTS dbt_column_edge_editor_key + ON dbt_column_edge (workspace_id, job_id, + parent_unique_id, parent_column, child_unique_id, child_column) + WHERE script_hash IS NULL; + +-- Same age sweep as the other per-run rows, and the same reason there is no +-- foreign key to `v2_job`. +CREATE INDEX IF NOT EXISTS idx_dbt_column_edge_run_age ON dbt_column_edge (ingested_at) + WHERE job_id <> '00000000-0000-0000-0000-000000000000'; + +-- The real column schema of a node, which only static analysis knows: an +-- ordered `[{"name": …, "type": …}]`, from `dbt.node_columns.parquet`. +-- +-- Beside `columns` rather than folded into it. `columns` is the DECLARED +-- metadata `manifest.json` carries -- the names an author wrote in `schema.yml` +-- and the prose against them -- and stays exactly that, so a project that +-- documents two of forty columns keeps saying so. This is the other forty, +-- typed, in the order the model produces them. +ALTER TABLE dbt_node ADD COLUMN IF NOT EXISTS column_schema JSONB; + +GRANT ALL ON dbt_column_edge TO windmill_user; +GRANT ALL ON dbt_column_edge TO windmill_admin; diff --git a/backend/parsers/windmill-parser-yaml/src/dbt.rs b/backend/parsers/windmill-parser-yaml/src/dbt.rs index 24f1e2c39d..1637a9ca04 100644 --- a/backend/parsers/windmill-parser-yaml/src/dbt.rs +++ b/backend/parsers/windmill-parser-yaml/src/dbt.rs @@ -58,6 +58,19 @@ impl DbtEngine { pub fn emits_node_events(&self) -> bool { matches!(self, DbtEngine::DbtCore1x) } + + /// Whether the engine's CLI has `--write-index`, the flag that writes the + /// parquet index column lineage lives in. False for 1.x, whose Python CLI + /// has no such option. + /// + /// True is not a promise that the artifact appears: `dbt-core` 2.0.0-alpha.5 + /// accepts the flag, declares the views over `dbt.column_lineage` in its own + /// `views.sql`, and writes neither that parquet nor `dbt.node_columns`. Only + /// Fusion does today. Attempting the pass on both is what lets a later 2.x + /// release pick the feature up with no change here. + pub fn writes_column_index(&self) -> bool { + !matches!(self, DbtEngine::DbtCore1x) + } } /// How the warehouse connection is supplied. Both paths are supported @@ -127,6 +140,18 @@ pub struct DbtDescriptor { pub selector: Option, #[serde(default)] pub test_behavior: DbtTestBehavior, + /// Ingest column-to-column lineage and the real column schemas, from the + /// engine's static analysis. + /// + /// Opt-in, and it has to be: the artifact only appears under + /// `--static-analysis strict`, which rejects SQL the default accepts (an + /// unresolvable identifier is an error there and compiles fine otherwise). + /// Turning it on for everyone would make a stricter dialect the price of + /// deploying a dbt project. It is a separate `dbt compile` pass, so nothing + /// it decides can change what a build does; a project it cannot analyze + /// keeps the graph it has today. + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + pub column_lineage: bool, /// `--vars`. dbt vars are typed — numbers, booleans, lists and objects are /// all normal — so values keep their YAML type; only string leaves carry /// `{{ arg }}` placeholders the worker substitutes from job args. Coercing diff --git a/backend/windmill-api-assets/src/lib.rs b/backend/windmill-api-assets/src/lib.rs index 09720b2203..2f3272977e 100644 --- a/backend/windmill-api-assets/src/lib.rs +++ b/backend/windmill-api-assets/src/lib.rs @@ -663,10 +663,18 @@ struct DbtAssetProvenance { description: Option, #[serde(skip_serializing_if = "Vec::is_empty")] data_tests: Vec, - /// Declared column metadata (name -> description). NOT column lineage — - /// `manifest.json` carries none (docs/dbt-runtime.md, decision 14). + /// Declared column metadata (name -> description): what `manifest.json` + /// carries, which is only the columns an author wrote down. #[serde(skip_serializing_if = "Option::is_none")] columns: Option, + /// Every column of the relation, typed and in order — + /// `[{"name": …, "type": …}]` — from the engine's static analysis. Present + /// only for a project that opted into it. Not gated on `script_visible`, + /// like the column edges and the `ref()` graph: the SHAPE of a relation is + /// what an in-scope consumer needs in order to explain what it reads, and + /// the prose about it is what `columns` above gates. + #[serde(skip_serializing_if = "Option::is_none")] + column_schema: Option, /// A source's declared freshness policy, for the staleness chip. #[serde(skip_serializing_if = "Option::is_none")] freshness: Option, @@ -921,6 +929,11 @@ pub struct AssetGraphResponse { /// instead of the project's actual shape. #[serde(skip_serializing_if = "Vec::is_empty", default)] dbt_edges: Vec, + /// Column-to-column lineage between two dbt models, from the engine's static + /// analysis. Empty unless the project opted into it — `manifest.json` + /// carries none. + #[serde(skip_serializing_if = "Vec::is_empty", default)] + dbt_column_edges: Vec, /// The job whose snapshot the dbt half was resolved from, when one was /// asked for and found. A run page polls the graph while its job runs /// because a dynamic descriptor's snapshot is written mid-run, and this is @@ -946,6 +959,22 @@ 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, +} + async fn asset_graph( authed: ApiAuthed, Path(w_id): Path, @@ -1323,7 +1352,7 @@ pub async fn asset_graph_for( n.resource_type AS "resource_type!", n.name AS "name!", n.asset_path, n.materialized, n.materialize_strategy, n.tags AS "tags!", n.description, n.test_kind, n.test_column, n.test_args, n.severity, n.attached_node, - n.columns, n.freshness, + n.columns, n.column_schema, n.freshness, n.raw_code, n.original_file_path, -- Whether the caller may read the project this row describes. -- The query deliberately reaches outside the requested folder @@ -1461,6 +1490,69 @@ pub async fn asset_graph_for( .fetch_all(&mut *tx) .await?; + // Column-to-column lineage, resolved to the relations the two nodes produce, + // like the `ref()` edges above and scoped exactly the same way. Present only + // for a project that opted into the analysis pass, so the usual graph pays + // one indexed lookup that finds nothing. + let dbt_column_edge_rows = sqlx::query!( + r#"WITH live AS ( + SELECT * FROM ( + SELECT DISTINCT ON (s.path) s.path, s.hash + FROM script s + WHERE $5::text IS NULL AND s.workspace_id = $1 AND s.language = 'dbt' + AND ($3::bigint IS NULL OR s.hash = $3) + AND ($3::bigint IS NOT NULL OR (s.deleted = false AND s.archived = false)) + ORDER BY s.path, s.created_at DESC + ) cur + UNION ALL + SELECT $5::text, $3::bigint WHERE $5::text IS NOT NULL + ), + chosen AS ( + SELECT CASE WHEN $4::uuid IS NOT NULL AND EXISTS ( + SELECT 1 FROM dbt_graph_snapshot g + WHERE g.workspace_id = $1 AND g.job_id = $4) + THEN $4::uuid + ELSE '00000000-0000-0000-0000-000000000000'::uuid END AS job_id + ) + SELECT p.asset_path AS "from_path!", e.parent_column AS "from_column!", + c.asset_path AS "to_path!", e.child_column AS "to_column!", + e.lineage_kind AS "lineage_kind!" + FROM dbt_column_edge e + JOIN live l ON l.path = e.script_path + AND (e.script_hash = l.hash + OR ($5::text IS NOT NULL AND l.hash IS NULL + AND e.script_hash IS NULL)) + JOIN chosen ch ON ch.job_id = e.job_id + JOIN dbt_node p ON p.workspace_id = e.workspace_id + AND p.script_path = e.script_path + AND (p.script_hash = e.script_hash + OR ($5::text IS NOT NULL AND e.script_hash IS NULL + AND p.script_hash IS NULL)) + AND p.job_id = ch.job_id + AND p.unique_id = e.parent_unique_id + JOIN dbt_node c ON c.workspace_id = e.workspace_id + AND c.script_path = e.script_path + AND (c.script_hash = e.script_hash + OR ($5::text IS NOT NULL AND e.script_hash IS NULL + AND c.script_hash IS NULL)) + AND c.job_id = ch.job_id + AND c.unique_id = e.child_unique_id + WHERE e.workspace_id = $1 + AND p.asset_path IS NOT NULL AND c.asset_path IS NOT NULL + AND ($3::bigint IS NOT NULL OR $5::text IS NOT NULL OR EXISTS ( + SELECT 1 FROM asset a + WHERE a.workspace_id = $1 AND a.kind = 'dbt' + AND a.path = c.asset_path + AND ($2::text IS NULL OR a.usage_path LIKE $2)))"#, + &w_id, + folder_filter.as_deref(), + dbt_script_hash, + dbt_job_id, + pinned_path, + ) + .fetch_all(&mut *tx) + .await?; + // The marker `chosen` resolved to, answered once for the caller: its // EXISTENCE is what lets a run page stop polling, and its timestamp is what // an editor labels the graph's provenance with — "parsed from the editor at @@ -1604,6 +1696,7 @@ pub async fn asset_graph_for( description: r.description.clone().filter(|_| source_allowed), data_tests: vec![], columns: r.columns.clone().filter(|_| source_allowed), + column_schema: r.column_schema.clone(), freshness: r.freshness.clone().filter(|_| source_allowed), }; // One relation can carry rows from several projects — typically a model @@ -2166,6 +2259,21 @@ pub async fn asset_graph_for( .collect(); dbt_edges.sort(); dbt_edges.dedup(); + let mut dbt_column_edges: Vec = dbt_column_edge_rows + .into_iter() + .filter(|r| { + rendered.contains(r.from_path.as_str()) && rendered.contains(r.to_path.as_str()) + }) + .map(|r| DbtColumnLineageEdge { + from_asset_path: r.from_path, + from_column: r.from_column, + to_asset_path: r.to_path, + to_column: r.to_column, + kind: r.lineage_kind, + }) + .collect(); + dbt_column_edges.sort(); + dbt_column_edges.dedup(); Ok(Json(AssetGraphResponse { assets, @@ -2175,6 +2283,7 @@ pub async fn asset_graph_for( macro_edges, test_edges, dbt_edges, + dbt_column_edges, dbt_snapshot_job, dbt_graph_ingested_at, })) diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index a27be7d295..373c819872 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -25952,6 +25952,33 @@ components: type: string to_asset_path: type: string + dbt_column_edges: + type: array + description: >- + Column-to-column lineage between two dbt models, from the engine's + own static analysis: `manifest.json` carries none, and the artifact + that does is only written when the project opts into the analysis + pass. Omitted when empty, which is the usual case. + 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), `mod` (transformed), `scan` (read to produce the + ROW rather than the value: a join key, a predicate, a `group + by`). Not an enum: the engine's own reader maps those three and + passes anything else through. dbt_snapshot_job: type: string format: uuid @@ -26018,7 +26045,23 @@ components: columns: type: object additionalProperties: true - description: Declared column metadata (name -> description). NOT column lineage — `manifest.json` carries none. + description: Declared column metadata (name -> description) — what `manifest.json` carries, which is only the columns an author wrote down. Omitted when the caller cannot read the script. + column_schema: + type: array + description: >- + Every column of the relation, typed and in the order the model + produces them, from the engine's static analysis. Present only for a + project that opted into it. Unlike `columns` this is the relation's + SHAPE, so it is not gated on being able to read the project. + items: + type: object + required: [name] + properties: + name: + type: string + type: + type: string + description: The declared type where `schema.yml` gives one, else the inferred one. Omitted when neither is known. freshness: type: object additionalProperties: true diff --git a/backend/windmill-common/src/dbt_manifest.rs b/backend/windmill-common/src/dbt_manifest.rs index 8fb124a33f..28d57bd10c 100644 --- a/backend/windmill-common/src/dbt_manifest.rs +++ b/backend/windmill-common/src/dbt_manifest.rs @@ -188,6 +188,12 @@ fn graph_digest(ingested: &IngestedManifest, relation_root: &str) -> String { .unwrap_or_default() .as_bytes(), ); + h.update(b"\0"); + h.update( + serde_json::to_string(&ingested.column_edges) + .unwrap_or_default() + .as_bytes(), + ); format!("{:x}", h.finalize()) } @@ -297,6 +303,14 @@ pub async fn prune_dbt_run_graphs( ) .execute(db) .await?; + sqlx::query!( + "DELETE FROM dbt_column_edge + WHERE job_id <> '00000000-0000-0000-0000-000000000000' + AND ingested_at < now() - make_interval(days => $1)", + RUN_GRAPH_RETENTION_DAYS, + ) + .execute(db) + .await?; // In ONE transaction with the orphan sweep: a restart in the gap leaves graph // rows whose marker is gone, and since the sweep runs only when a marker went, // every later call computes `retired == 0` and skips them for good. @@ -325,7 +339,7 @@ pub async fn prune_dbt_run_graphs( // partial index here — all of them `WHERE job_id <> DEPLOYED` — and past the // keep-count is rare, so the ordinary run should pay for neither. if retired > 0 { - for table in ["dbt_node", "dbt_edge"] { + for table in ["dbt_node", "dbt_edge", "dbt_column_edge"] { sqlx::query(&format!( "DELETE FROM {table} t WHERE t.workspace_id = $1 AND t.script_path = $2 @@ -381,6 +395,11 @@ pub struct IngestedNode { pub severity: Option, pub attached_node: Option, pub columns: Option, + /// The node's real columns, typed and ordered — `[{"name": …, "type": …}]`, + /// from the engine's static analysis. `None` when the project did not ask + /// for it or the engine wrote none. Beside `columns` rather than merged into + /// it: that one is what the author DECLARED, and stays that. + pub column_schema: Option, pub freshness: Option, /// The transform itself, for the graph to render. The copy taken at /// deploy: the file itself is in the script's module bundle. @@ -388,6 +407,40 @@ pub struct IngestedNode { pub original_file_path: Option, } +/// One column-to-column edge of the ingested graph. +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Default)] +#[serde(default)] +pub struct IngestedColumnEdge { + pub parent_unique_id: String, + pub parent_column: String, + pub child_unique_id: String, + pub child_column: String, + /// dbt's own word: `copy`, `mod` or `scan`. Kept verbatim — the engine's own + /// reader maps those three and passes anything else through, so the set is + /// open. + pub lineage_kind: String, +} + +/// One column of a node, as the engine's static analysis resolved it. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct IndexedColumn { + pub name: String, + /// The declared type where `schema.yml` gives one, else the inferred one. + /// Empty when neither is known; the column still belongs to the relation, so + /// only the type is left out. + pub column_type: String, + /// Position in the relation, which is the order the panel lists them in. + pub index: i64, +} + +/// What one `--write-index` pass produced: the column edges of the whole +/// project and the real column schema per node. +#[derive(Debug, Default)] +pub struct ColumnIndex { + pub edges: Vec, + pub columns: HashMap>, +} + // Serde: an agent worker cannot write these tables directly, so it posts the // whole manifest to the server, which stores it with the same function the SQL // path uses. @@ -398,6 +451,9 @@ pub struct IngestedNode { pub struct IngestedManifest { pub nodes: Vec, pub edges: Vec<(String, String)>, + /// Column-to-column lineage, when the project asked for it and the engine + /// produced it. Empty is the normal case — see `attach_column_index`. + pub column_edges: Vec, /// The `asset` rows the owning script produces (models) and consumes /// (sources) — what the lineage graph is drawn from. pub assets: Vec, @@ -405,6 +461,66 @@ pub struct IngestedManifest { pub adapter_type: String, } +/// The most column edges one graph stores. +/// +/// A `scan` edge — the column was read to produce the row, not the value — is +/// emitted from every join key and every predicate column to every output +/// column, so one wide model over a multi-column join contributes columns times +/// predicates edges on its own. The cap is what keeps a project shaped like that +/// from turning one deploy into a multi-million-row insert; past it the lineage +/// is truncated and the rest of the graph is unaffected. +pub const MAX_COLUMN_EDGES: usize = 200_000; + +impl IngestedManifest { + /// Fold one `--write-index` pass into the graph. + /// + /// Both halves are scoped to the nodes this graph already kept: the index + /// describes the whole project, while the graph describes what this script's + /// selection builds plus the parents anchoring its edges, and an edge whose + /// endpoint is absent has nothing to draw. + pub fn attach_column_index(&mut self, index: ColumnIndex) { + let kept: std::collections::HashSet<&str> = + self.nodes.iter().map(|n| n.unique_id.as_str()).collect(); + let mut edges: Vec = index + .edges + .into_iter() + .filter(|e| { + kept.contains(e.parent_unique_id.as_str()) + && kept.contains(e.child_unique_id.as_str()) + }) + .collect(); + // Sorted and deduplicated for the digest, which decides whether a run + // stores a snapshot at all: parquet row order is the engine's and two + // passes over one project must not read as two different graphs. + edges.sort(); + edges.dedup(); + edges.truncate(MAX_COLUMN_EDGES); + self.column_edges = edges; + + let mut columns = index.columns; + for node in self.nodes.iter_mut() { + let Some(mut cols) = columns.remove(&node.unique_id) else { + continue; + }; + if cols.is_empty() { + continue; + } + cols.sort_by_key(|c| c.index); + node.column_schema = Some(serde_json::Value::Array( + cols.into_iter() + // A column the analysis typed as nothing still belongs in + // the list — that it exists is the half `manifest.json` + // could not answer. + .map(|c| match c.column_type.is_empty() { + true => serde_json::json!({ "name": c.name }), + false => serde_json::json!({ "name": c.name, "type": c.column_type }), + }) + .collect(), + )); + } + } +} + /// dbt's `materialized` mapped onto Windmill's write strategy. /// /// The mapping is exact for the four strategies Windmill has, and deliberately @@ -655,6 +771,9 @@ pub fn ingest_manifest( .map(|(k, v)| (k.clone(), v.description.clone().unwrap_or_default())) .collect::>()) }), + // Filled by `attach_column_index` when the project asked for it: + // the manifest carries declared columns only. + column_schema: None, freshness: node.freshness.clone(), // The transform the graph renders. Capped: a project can hold // thousands of models and this is duplicated per deploy, so a @@ -826,6 +945,16 @@ pub async fn replace_dbt_manifest( ) .execute(&mut **tx) .await?; + sqlx::query!( + "DELETE FROM dbt_column_edge WHERE workspace_id = $1 AND script_path = $2 + AND script_hash = $3 AND job_id = $4", + workspace_id, + script_path, + script_hash, + job_id + ) + .execute(&mut **tx) + .await?; // The marker, before the rows: a graph with no nodes at all is a legitimate // answer for a dynamic run that disabled every model, and the reader must be // able to tell it from a run that stored nothing. @@ -881,7 +1010,7 @@ async fn insert_graph_rows( "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id, \ resource_type, name, asset_path, materialized, materialize_strategy, unique_key, \ tags, description, test_kind, test_column, test_args, severity, attached_node, \ - columns, freshness, raw_code, original_file_path) ", + columns, column_schema, freshness, raw_code, original_file_path) ", ); q.push_values(chunk, |mut b, n| { b.push_bind(workspace_id) @@ -903,6 +1032,7 @@ async fn insert_graph_rows( .push_bind(&n.severity) .push_bind(&n.attached_node) .push_bind(&n.columns) + .push_bind(&n.column_schema) .push_bind(&n.freshness) .push_bind(&n.raw_code) .push_bind(&n.original_file_path); @@ -926,6 +1056,26 @@ async fn insert_graph_rows( q.push(" ON CONFLICT DO NOTHING"); q.build().execute(&mut **tx).await?; } + + for chunk in ingested.column_edges.chunks(COLUMN_EDGE_INSERT_CHUNK) { + let mut q = sqlx::QueryBuilder::new( + "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) ", + ); + q.push_values(chunk, |mut b, e| { + b.push_bind(workspace_id) + .push_bind(script_path) + .push_bind(script_hash) + .push_bind(job_id) + .push_bind(&e.parent_unique_id) + .push_bind(&e.parent_column) + .push_bind(&e.child_unique_id) + .push_bind(&e.child_column) + .push_bind(&e.lineage_kind); + }); + q.push(" ON CONFLICT DO NOTHING"); + q.build().execute(&mut **tx).await?; + } Ok(()) } @@ -966,7 +1116,7 @@ pub async fn replace_dbt_editor_graph( ) -> Result<()> { // By job alone, so re-executing one — a zombie recovered onto another // worker — replaces its rows rather than colliding with them. - for table in ["dbt_node", "dbt_edge", "dbt_graph_snapshot"] { + for table in ["dbt_node", "dbt_edge", "dbt_column_edge", "dbt_graph_snapshot"] { sqlx::query(&format!( "DELETE FROM {table} WHERE workspace_id = $1 AND job_id = $2 AND script_hash IS NULL" )) @@ -1016,7 +1166,7 @@ pub async fn replace_dbt_editor_graph( .fetch_all(&mut **tx) .await?; if !retired.is_empty() { - for table in ["dbt_node", "dbt_edge"] { + for table in ["dbt_node", "dbt_edge", "dbt_column_edge"] { sqlx::query(&format!( "DELETE FROM {table} WHERE workspace_id = $1 AND job_id = ANY($2) \ AND script_hash IS NULL" @@ -1035,6 +1185,8 @@ pub async fn replace_dbt_editor_graph( const NODE_INSERT_CHUNK: usize = 2000; /// Six columns, so the same ceiling allows far more. const EDGE_INSERT_CHUNK: usize = 8000; +/// Nine columns, and by far the most numerous rows of the three. +const COLUMN_EDGE_INSERT_CHUNK: usize = 6000; /// Clear one VERSION's graph: the delete-by-hash route, which only soft-deletes /// its `script` row and so fires no cascade, and the ingest that finds no @@ -1070,6 +1222,15 @@ pub async fn clear_dbt_manifest_version( ) .execute(&mut **tx) .await?; + sqlx::query!( + "DELETE FROM dbt_column_edge + WHERE workspace_id = $1 AND script_path = $2 AND script_hash = $3", + workspace_id, + script_path, + script_hash + ) + .execute(&mut **tx) + .await?; // The marker too, and every job's: a marker left standing for rows that are // gone is read as a snapshot, and its digest still answers the suppression // check — so an identical run would write nothing and then render an empty @@ -1105,7 +1266,7 @@ pub async fn clear_dbt_editor_graphs( workspace_id: &str, script_path: &str, ) -> Result<()> { - for table in ["dbt_node", "dbt_edge", "dbt_graph_snapshot"] { + for table in ["dbt_node", "dbt_edge", "dbt_column_edge", "dbt_graph_snapshot"] { sqlx::query(&format!( "DELETE FROM {table} WHERE workspace_id = $1 AND script_path = $2 AND script_hash IS NULL" @@ -1765,6 +1926,69 @@ mod tests { assert_eq!(back.assets.len(), ingested.assets.len()); assert_eq!(back.assets[0].path, ingested.assets[0].path); } + + // The index describes the whole PROJECT while the graph describes what this + // script's selection builds, so an edge whose endpoint the graph does not + // hold has nothing to draw and must not be stored. + #[test] + fn column_lineage_is_scoped_to_the_nodes_the_graph_kept() { + let mut i = ingested(); + let kept = "model.jaffle_shop.customers"; + let dropped = "model.other_project.elsewhere"; + i.attach_column_index(ColumnIndex { + edges: vec![ + edge("model.jaffle_shop.orders_daily", "id", kept, "id", "copy"), + edge(dropped, "id", kept, "id", "copy"), + edge(kept, "id", dropped, "id", "copy"), + ], + columns: [ + ( + kept.to_string(), + vec![ + col("total", "Float64", 1), + col("id", "Int32", 0), + col("untyped", "", 2), + ], + ), + (dropped.to_string(), vec![col("id", "Int32", 0)]), + ] + .into(), + }); + assert_eq!( + i.column_edges + .iter() + .map(|e| (e.parent_unique_id.as_str(), e.child_unique_id.as_str())) + .collect::>(), + vec![("model.jaffle_shop.orders_daily", kept)] + ); + // In `column_index` order, and a column the analysis could not type still + // belongs to the relation. + assert_eq!( + node(&i, kept).column_schema, + Some(serde_json::json!([ + {"name": "id", "type": "Int32"}, + {"name": "total", "type": "Float64"}, + {"name": "untyped"}, + ])) + ); + assert!(node(&i, "model.jaffle_shop.orders_daily") + .column_schema + .is_none()); + } + + fn edge(from: &str, from_col: &str, to: &str, to_col: &str, kind: &str) -> IngestedColumnEdge { + IngestedColumnEdge { + parent_unique_id: from.into(), + parent_column: from_col.into(), + child_unique_id: to.into(), + child_column: to_col.into(), + lineage_kind: kind.into(), + } + } + + fn col(name: &str, column_type: &str, index: i64) -> IndexedColumn { + IndexedColumn { name: name.into(), column_type: column_type.into(), index } + } } /// Record one model's state for THIS RUN. diff --git a/backend/windmill-common/tests/dbt_graph_storage.rs b/backend/windmill-common/tests/dbt_graph_storage.rs index 7b4971312a..cb87fdc81d 100644 --- a/backend/windmill-common/tests/dbt_graph_storage.rs +++ b/backend/windmill-common/tests/dbt_graph_storage.rs @@ -8,8 +8,8 @@ use sqlx::{Pool, Postgres}; use windmill_common::dbt_manifest::{ clear_dbt_editor_graphs, clear_dbt_manifest_version, prune_dbt_run_graphs, - replace_dbt_editor_graph, replace_dbt_manifest, IngestedManifest, IngestedNode, - DBT_EDITOR_GRAPHS_KEPT, DEPLOYED_GRAPH, DEPLOYED_GRAPH_VERSIONS_KEPT, + replace_dbt_editor_graph, replace_dbt_manifest, IngestedColumnEdge, IngestedManifest, + IngestedNode, DBT_EDITOR_GRAPHS_KEPT, DEPLOYED_GRAPH, DEPLOYED_GRAPH_VERSIONS_KEPT, }; const WS: &str = "test-workspace"; @@ -52,10 +52,36 @@ fn manifest(names: &[&str]) -> IngestedManifest { .windows(2) .map(|w| (format!("model.p.{}", w[0]), format!("model.p.{}", w[1]))) .collect(), + // Same reason: a project that opted into the analysis pass has these, and + // a fixture without them leaves every column-edge insert and sweep in + // this file unexecuted. + column_edges: names + .windows(2) + .map(|w| IngestedColumnEdge { + parent_unique_id: format!("model.p.{}", w[0]), + parent_column: w[0].to_string(), + child_unique_id: format!("model.p.{}", w[1]), + child_column: w[1].to_string(), + lineage_kind: "copy".to_string(), + }) + .collect(), ..Default::default() } } +/// Column edges of one version, so the sweeps can be shown to reach them. +async fn column_edges_for(db: &Pool, hash: i64) -> i64 { + sqlx::query_scalar!( + "SELECT count(*) FROM dbt_column_edge WHERE workspace_id = $1 AND script_hash = $2", + WS, + hash + ) + .fetch_one(db) + .await + .unwrap() + .unwrap_or(0) +} + /// Edges for one version, so a test can assert the batched insert ran at all. async fn edges_for(db: &Pool, hash: i64) -> i64 { sqlx::query_scalar!( @@ -275,6 +301,8 @@ async fn deleting_the_script_cascades_to_every_sidecar(db: Pool) { assert_eq!(nodes_for(&db, 2, DEPLOYED_GRAPH).await, 0); assert_eq!(edges_for(&db, 1).await, 0); assert_eq!(edges_for(&db, 2).await, 0); + assert_eq!(column_edges_for(&db, 1).await, 0); + assert_eq!(column_edges_for(&db, 2).await, 0); assert_eq!(markers_for_path(&db).await, 0); } @@ -315,7 +343,7 @@ async fn the_sweep_takes_old_snapshots_and_spares_the_version(db: Pool tx.commit().await.unwrap(); // Age one snapshot past the window, rows and marker together. - for t in ["dbt_node", "dbt_edge", "dbt_graph_snapshot"] { + for t in ["dbt_node", "dbt_edge", "dbt_column_edge", "dbt_graph_snapshot"] { sqlx::query(&format!( "UPDATE {t} SET ingested_at = now() - interval '400 days' WHERE job_id = $1" )) diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index 0390e1c366..a15deb6ca5 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -114,6 +114,10 @@ rust_decimal.workspace = true jsonwebtoken.workspace = true sha2.workspace = true hmac.workspace = true +# Reads the dbt engine's `target/index/*.parquet`, which is where column-level +# lineage lives. Unconditional rather than behind the `parquet` FEATURE: that one +# is object storage, and a build without it still runs dbt jobs. +parquet.workspace = true pem = { workspace = true, optional = true } rsa = { workspace = true, optional = true } urlencoding.workspace = true diff --git a/backend/windmill-worker/src/dbt_column_index.rs b/backend/windmill-worker/src/dbt_column_index.rs new file mode 100644 index 0000000000..4994e1d1e9 --- /dev/null +++ b/backend/windmill-worker/src/dbt_column_index.rs @@ -0,0 +1,283 @@ +//! Column-level lineage and real column schemas, from the engine's own static +//! analysis. +//! +//! `manifest.json` carries neither. What does is the parquet index an engine +//! writes under `dbt compile --static-analysis strict --write-index`: +//! `dbt.column_lineage.parquet` (column-to-column edges, each labelled `copy`, +//! `mod` or `scan`) and `dbt.node_columns.parquet` (every column of every node, +//! typed and ordered, rather than only the ones an author documented). +//! +//! Three properties shape everything here, all of them measured against the real +//! engines rather than assumed: +//! +//! - **Strict analysis rejects SQL the default accepts.** An unresolvable +//! identifier is an error under `strict` and compiles fine otherwise, so this +//! is a SEPARATE pass with its own `--target-path`, never a flag on the build, +//! and it is opt-in per project. +//! - **A failed pass still writes the index**, holding every edge of the models +//! that did analyze. So the artifact is read whatever the exit status. +//! - **The flag is not the capability.** `dbt-core` 2.0.0-alpha.5 accepts +//! `--write-index`, declares the views over these two tables in its own +//! `views.sql`, and writes neither file; only Fusion does today. Nothing here +//! asks which engine it is beyond "has the flag" — a release that starts +//! writing them is picked up with no change. + +use std::path::Path; + +use parquet::file::reader::{FileReader, SerializedFileReader}; +use parquet::record::{Field, Row}; +use uuid::Uuid; +use windmill_common::dbt_manifest::{ColumnIndex, IndexedColumn, IngestedColumnEdge}; +use windmill_common::error; +use windmill_common::worker::Connection; +use windmill_parser_yaml::dbt::DbtDescriptor; +use windmill_queue::append_logs; + +use crate::dbt_executor::{dbt_command, Invocation, PreparedProject}; +use crate::handle_child::JobCtx; + +/// Where the lineage pass writes, relative to the project directory. +/// +/// Its own tree, not the runtime's `wm_target`: a `dbt compile` writes +/// `manifest.json` and `run_results.json` like any other invocation, and after a +/// build those two are what the graph ingest and `dbt retry` read. +const CLL_ARTIFACTS_DIR: &str = "wm_target_cll"; + +const COLUMN_LINEAGE_PARQUET: &str = "dbt.column_lineage.parquet"; +const NODE_COLUMNS_PARQUET: &str = "dbt.node_columns.parquet"; + +/// Run the lineage pass and read what it produced. +/// +/// Best-effort throughout: every failure returns `None` with a line in the job +/// log saying which one, because the graph without column lineage is exactly the +/// graph this project had before it asked for any. +pub(crate) async fn collect( + p: &PreparedProject, + descriptor: &DbtDescriptor, + inv: &Invocation, + ctx: &mut JobCtx<'_>, + job_id: &Uuid, + w_id: &str, + conn: &Connection, +) -> Option { + if !descriptor.column_lineage { + return None; + } + if !p.engine.engine.writes_column_index() { + append_logs( + job_id, + w_id, + format!( + "\n`column_lineage` is set, but the {} engine has no `--write-index`: column \ + lineage needs an engine that does static analysis. The rest of the graph is \ + unaffected.\n", + p.engine.engine.as_str() + ), + conn, + ) + .await; + return None; + } + + let index_dir = p.project_dir.join(CLL_ARTIFACTS_DIR).join("index"); + // A previous pass in the same job directory — a retry's second attempt — + // would otherwise be read back as this one's answer. + tokio::fs::remove_dir_all(p.project_dir.join(CLL_ARTIFACTS_DIR)) + .await + .ok(); + + let mut cmd = dbt_command( + p, + &[ + "compile", + "--static-analysis", + "strict", + "--write-index", + // Documented as what builds the CLL graph, and `--write-index` alone + // happens to imply it on the engine probed. Passed explicitly so the + // pass does not depend on which of the two is doing the work. + "--write-lineage", + "--target-path", + CLL_ARTIFACTS_DIR, + ], + ); + if let Err(e) = crate::dbt_executor::add_vars(&mut cmd, descriptor, inv) { + tracing::warn!("dbt column lineage: {e:#}"); + return None; + } + // 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. + let outcome = crate::dbt_executor::run_capturing( + cmd, + "dbt compile (column lineage)", + ctx, + job_id, + w_id, + conn, + CLL_MAX_OUTPUT_BYTES, + ) + .await; + + let index = read_index(&index_dir).await; + // What the caller cannot say for itself. The COUNTS are logged where the + // index is folded into the graph, since the graph is what decides how much + // of it is kept; this is the part only the pass knows. + match (&index, &outcome) { + // The ordinary partial outcome: some models did not analyze, the rest + // did, and their lineage is real. + (Some(_), Err(e)) => { + let note = "Column lineage: `--static-analysis strict` rejected part of the project, \ + so the lineage covers only the models it could analyze."; + append_logs(job_id, w_id, format!("\n{note}\n{}", diagnostics(&e.to_string())), conn) + .await; + } + (Some(_), Ok(_)) => {} + (None, outcome) => { + let note = format!( + "No column lineage: the analysis pass wrote no `{COLUMN_LINEAGE_PARQUET}`. Only \ + an engine that computes it does, and only for the warehouses it analyzes \ + natively — the flag alone is not the capability." + ); + // The engine's own diagnostics. They are how a reader learns that + // this adapter turned static analysis off, which it reports as a + // warning on a SUCCESSFUL compile that nothing else would show. + let detail = match outcome { + Ok(c) => diagnostics(&c.stderr), + Err(e) => diagnostics(&e.to_string()), + }; + append_logs(job_id, w_id, format!("\n{note}\n{detail}"), conn).await; + } + } + index +} + +/// stdout the pass may produce. It is a compile, so this is diagnostics rather +/// than data. +const CLL_MAX_OUTPUT_BYTES: usize = 1 << 20; + +/// The tail of what the engine said, bounded. The whole of it is every rendered +/// model on a large project, which is not what a job log is for. +const DIAGNOSTIC_LINES: usize = 40; + +fn diagnostics(out: &str) -> String { + let lines: Vec<&str> = out.lines().filter(|l| !l.trim().is_empty()).collect(); + let tail = &lines[lines.len().saturating_sub(DIAGNOSTIC_LINES)..]; + match tail.is_empty() { + true => String::new(), + false => format!("{}\n", tail.join("\n")), + } +} + +/// Read both parquets, if the lineage one is there. +/// +/// The column schemas alone are not worth a graph: they arrive with the lineage +/// or not at all, and a node's declared columns already answer for the case +/// where the pass never ran. +async fn read_index(index_dir: &Path) -> Option { + let lineage = index_dir.join(COLUMN_LINEAGE_PARQUET); + if !tokio::fs::try_exists(&lineage).await.unwrap_or(false) { + return None; + } + let columns = index_dir.join(NODE_COLUMNS_PARQUET); + // Decompressing and decoding a parquet is CPU work on a file the engine just + // wrote, so it does not belong on the runtime's poll thread. + tokio::task::spawn_blocking(move || read_index_blocking(&lineage, &columns)) + .await + .map_err(|e| tracing::warn!("reading the dbt column index: {e:#}")) + .ok()? + .map_err(|e| tracing::warn!("reading the dbt column index: {e:#}")) + .ok() +} + +fn read_index_blocking(lineage: &Path, columns: &Path) -> error::Result { + let mut out = ColumnIndex::default(); + for row in rows(lineage)? { + let parent_unique_id = string(&row, "from_node_unique_id"); + let child_unique_id = string(&row, "to_node_unique_id"); + let parent_column = string(&row, "from_column_name"); + let child_column = string(&row, "to_column_name"); + // A column of a node the analysis could not name is not an endpoint the + // graph can draw. + if parent_unique_id.is_empty() + || child_unique_id.is_empty() + || parent_column.is_empty() + || child_column.is_empty() + { + continue; + } + out.edges.push(IngestedColumnEdge { + parent_unique_id, + parent_column, + child_unique_id, + child_column, + lineage_kind: string(&row, "lineage_kind"), + }); + } + // Absent is normal — an engine can write the lineage table and not this one — + // and unreadable is not worth losing the lineage over. + if let Ok(rows) = rows(columns) { + for row in rows { + let unique_id = string(&row, "unique_id"); + let name = string(&row, "column_name"); + if unique_id.is_empty() || name.is_empty() { + continue; + } + // The author's `data_type` where `schema.yml` gives one, since that + // is what the project calls the column; the analysis's own inference + // otherwise. + let column_type = match string(&row, "declared_type") { + t if !t.is_empty() => t, + _ => string(&row, "inferred_type"), + }; + out.columns + .entry(unique_id) + .or_default() + .push(IndexedColumn { + name, + column_type, + index: int(&row, "column_index").unwrap_or(i64::MAX), + }); + } + } + Ok(out) +} + +fn rows(path: &Path) -> error::Result> { + let file = std::fs::File::open(path) + .map_err(|e| error::Error::internal_err(format!("opening {}: {e}", path.display())))?; + let reader = SerializedFileReader::new(file) + .map_err(|e| error::Error::internal_err(format!("reading {}: {e}", path.display())))?; + reader + .get_row_iter(None) + .map_err(|e| error::Error::internal_err(format!("reading {}: {e}", path.display())))? + .collect::, _>>() + .map_err(|e| error::Error::internal_err(format!("reading {}: {e}", path.display()))) +} + +/// By NAME, not by position: these tables are the engine's own schema and it +/// adds columns to them between releases. +fn field<'a>(row: &'a Row, name: &str) -> Option<&'a Field> { + row.get_column_iter() + .find(|(k, _)| k.as_str() == name) + .map(|(_, v)| v) +} + +fn string(row: &Row, name: &str) -> String { + match field(row, name) { + Some(Field::Str(s)) => s.clone(), + Some(Field::Bytes(b)) => String::from_utf8_lossy(b.data()).into_owned(), + _ => String::new(), + } +} + +fn int(row: &Row, name: &str) -> Option { + match field(row, name) { + Some(Field::Long(v)) => Some(*v), + Some(Field::Int(v)) => Some(*v as i64), + Some(Field::Short(v)) => Some(*v as i64), + Some(Field::UInt(v)) => Some(*v as i64), + Some(Field::ULong(v)) => i64::try_from(*v).ok(), + _ => None, + } +} diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index 33ddd6c32e..bd9668ba40 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -653,12 +653,23 @@ pub(crate) async fn dbt_dep( None => GraphPublisher::Unversioned, }; let superseded = if let Some(warehouse) = prepared.warehouse.as_deref() { - let ingested = windmill_common::dbt_manifest::ingest_manifest( + let mut ingested = windmill_common::dbt_manifest::ingest_manifest( &manifest, warehouse, prepared.default_database.as_deref(), selected.as_ref(), ); + attach_column_index( + &mut ingested, + &prepared, + &descriptor, + &inv, + &mut ctx, + job_id, + w_id, + &conn, + ) + .await; let published = persist_ingest( db, w_id, @@ -2882,7 +2893,8 @@ async fn run_show( conn, SHOW_MAX_OUTPUT_BYTES, ) - .await?; + .await? + .stdout; // dbt frames the rows as `{"node": …, "show": [ … ]}`, pretty-printed, with a // banner before and a deprecation summary after — so neither "the line starting // with `{`" nor "first `{` to the end" parses. A streaming deserializer stops at @@ -3064,12 +3076,23 @@ async fn run_parse_only( // manifest and the selection while the warehouse only keys them — so a project // with no warehouse identity still reports what dbt found. The placeholder // reaches no row: the guard below returns before anything is written. - let ingested = windmill_common::dbt_manifest::ingest_manifest( + let mut ingested = windmill_common::dbt_manifest::ingest_manifest( &manifest, p.warehouse.as_deref().unwrap_or("unkeyed"), p.default_database.as_deref(), selected.as_ref(), ); + attach_column_index( + &mut ingested, + p, + descriptor, + inv, + ctx, + &job.id, + &job.workspace_id, + conn, + ) + .await; result.nodes = ingested.nodes.len(); result.edges = ingested.edges.len(); for n in &ingested.nodes { @@ -3143,6 +3166,51 @@ async fn run_parse_only( Ok(to_raw_value(&result)) } +/// Fold this project's column lineage into the graph about to be stored, when +/// the descriptor asked for it. +/// +/// One helper for all three ingests — deploy, editor parse, per-run refresh — +/// because a graph that carries column lineage in one provenance and not another +/// reads as the lineage having disappeared. +async fn attach_column_index( + ingested: &mut windmill_common::dbt_manifest::IngestedManifest, + p: &PreparedProject, + descriptor: &DbtDescriptor, + inv: &Invocation, + ctx: &mut JobCtx<'_>, + job_id: &Uuid, + w_id: &str, + conn: &Connection, +) { + let Some(index) = + crate::dbt_column_index::collect(p, descriptor, inv, ctx, job_id, w_id, conn).await + else { + return; + }; + let found = index.edges.len(); + ingested.attach_column_index(index); + let kept = ingested.column_edges.len(); + let typed: usize = ingested + .nodes + .iter() + .filter(|n| n.column_schema.is_some()) + .count(); + // Counted here rather than at the pass: the index describes the whole + // project and this graph describes one selection of it, so `found` is what + // dbt produced and `kept` is what the graph can draw. + let dropped = match found.saturating_sub(kept) { + 0 => String::new(), + n => format!(" ({n} outside this graph or past the cap)"), + }; + append_logs( + job_id, + w_id, + format!("\nIngested {kept} column lineage edges{dropped} and typed {typed} nodes\n"), + conn, + ) + .await; +} + /// Refresh the stored graph from the manifest this run produced. async fn ingest_from_run( p: &PreparedProject, @@ -3164,12 +3232,23 @@ async fn ingest_from_run( // filter this run's manifest by a different node set than it built. let selected = resolve_selection(p, descriptor, inv, ctx, &job.id, &job.workspace_id, conn).await?; - let ingested = windmill_common::dbt_manifest::ingest_manifest( + let mut ingested = windmill_common::dbt_manifest::ingest_manifest( &manifest, warehouse, p.default_database.as_deref(), selected.as_ref(), ); + attach_column_index( + &mut ingested, + p, + descriptor, + inv, + ctx, + &job.id, + &job.workspace_id, + conn, + ) + .await; // Only a run whose models are its own snapshots per run. A static // descriptor at a moved profile re-ingests the VERSION's graph, since the // move outlives the run; one that neither drifted nor overrode anything @@ -3428,7 +3507,9 @@ async fn resolve_selection( // through the job-log writer, which `NO_LOGS_AT_ALL` discards — the selection // would resolve to the empty set and the ingest would wipe the script's assets // while dbt went on building the descriptor's models. - let stdout = run_capturing(cmd, "dbt ls", ctx, job_id, w_id, conn, LS_MAX_OUTPUT_BYTES).await?; + let stdout = run_capturing(cmd, "dbt ls", ctx, job_id, w_id, conn, LS_MAX_OUTPUT_BYTES) + .await? + .stdout; let mut set = std::collections::HashSet::new(); for line in stdout.lines() { let line = line.trim(); @@ -3471,7 +3552,15 @@ const CAPTURE_MAX_STDERR_BYTES: usize = 64 * 1024; /// never holds more than it, so it has to be enforced while reading. Both pipes /// are drained concurrently because a child that fills the one nobody reads /// blocks forever. -async fn run_capturing( +/// What a captured invocation produced. `stderr` is where dbt writes its +/// diagnostics — the errors and warnings block — so a caller that has to explain +/// a SUCCESSFUL run needs it as much as a failing one does. +pub(crate) struct Captured { + pub stdout: String, + pub stderr: String, +} + +pub(crate) async fn run_capturing( mut cmd: Command, name: &str, ctx: &mut JobCtx<'_>, @@ -3479,7 +3568,7 @@ async fn run_capturing( w_id: &str, conn: &Connection, max_stdout_bytes: usize, -) -> error::Result { +) -> error::Result { use tokio::io::AsyncReadExt; let mut child = cmd @@ -3568,7 +3657,10 @@ async fn run_capturing( String::from_utf8_lossy(&stderr) ))); } - Ok(String::from_utf8_lossy(&stdout).to_string()) + Ok(Captured { + stdout: String::from_utf8_lossy(&stdout).to_string(), + stderr: String::from_utf8_lossy(&stderr).to_string(), + }) } /// Run a preparation command through the same child handler the build uses, so @@ -4518,7 +4610,11 @@ fn has_retryable_node(run_results: &str) -> bool { } /// Append `--vars` if the descriptor (or the run) declares any. -fn add_vars(cmd: &mut Command, descriptor: &DbtDescriptor, inv: &Invocation) -> error::Result<()> { +pub(crate) fn add_vars( + cmd: &mut Command, + descriptor: &DbtDescriptor, + inv: &Invocation, +) -> error::Result<()> { let vars = resolved_vars(descriptor, &inv.args, inv.strict)?; if !vars.is_empty() { cmd.args(["--vars", &serde_json::to_string(&vars).unwrap_or_default()]); diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index e3e7868548..6e1d4ff8d7 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -33,6 +33,7 @@ pub mod common; mod config; mod csharp_executor; +mod dbt_column_index; mod dbt_engine; mod dbt_executor; mod dbt_profiles; diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index dc06a22fa0..00d909faf4 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -36,7 +36,7 @@ the dominant way dbt is orchestrated today. | 11 | Asset kind | `dbt:////` — keyed on the relation, not on dbt's node id. See below | | 12 | Graph refresh | Deploy-time, re-ingested per run only when the descriptor is dynamic, plus an explicit `parse` of the editor's buffer. See below | | 13 | Manifest storage | Sidecar table for nodes/edges. Full manifest **not** stored — see below | -| 14 | Metadata depth | Tests, strategy, tags, freshness, column descriptions. Column **lineage** is not in the manifest — see below | +| 14 | Metadata depth | Tests, strategy, tags, freshness, column descriptions. Column **lineage** and real column schemas come from the engine's parquet index, opt-in per project — see below | | 15 | Node rendering | Asset nodes per model plus one runnable node for the script | | 16 | Progress | Live, from the JSON event stream | | 17 | Test failures | Honor dbt's own `severity` | @@ -857,6 +857,8 @@ profile: select: ["tag:nightly+"] exclude: [] test_behavior: build # build | after_all | none +column_lineage: false # opt in to the static-analysis pass that + # produces column-level lineage (decision 14) vars: # typed: numbers/bools/lists keep their type, run_date: "{{ run_date }}" # and string leaves take job arguments strict: false @@ -1151,11 +1153,57 @@ would be an unread copy of data that is already reproducible by redeploying (or, for a dynamic descriptor, by the next run). Worth adding the day something needs the parts the sidecar drops — compiled SQL, macro definitions — and not before. -**Decision 14 — column lineage is not available.** The decision assumed -`manifest.json` carries column-to-column edges; it does not, in either core -engine. What it does carry is declared column *descriptions*, which are -ingested. Real column lineage would need Fusion (which does static analysis) or -a SQL-AST pass of our own, so `columnLineageGraph.ts` is not wired up for dbt. +**Decision 14 — column lineage comes from the parquet index, not the manifest.** +`manifest.json` carries no column-to-column edges, in any engine, and its +`columns` are the ones an author declared in `schema.yml`. Both halves exist in a +different artifact: `dbt compile --static-analysis strict --write-index` writes +`target/index/`, and two of its tables are `dbt.column_lineage.parquet` +(`from_node_unique_id`, `from_column_name`, `to_node_unique_id`, +`to_column_name`, `lineage_kind`) and `dbt.node_columns.parquet` (every column of +every node, with its declared type, its inferred type and its description). + +Three measured properties decide the shape of the ingest. + +**Strict analysis is a stricter dialect.** `select no_such_column from +ref(...)` is `UnresolvedIdentifier (dbt0227)` and exit 1 under `strict`, and +compiles fine under `baseline` (the default). So this is a separate `dbt compile` +with its own `--target-path`, never a flag on the build, and it is opt-in per +project: `column_lineage: true` in the descriptor. Off, nothing changes. On, a +project that cannot be analyzed keeps exactly the graph it had. The pass never +fails a deploy or a run. + +**A failed pass still writes the index**, holding every edge of the models that +did analyze, so the artifact is read whatever the exit status and partial lineage +is a normal outcome rather than an error. An unreachable *source* is milder +still: `RemoteError (dbt1014)` downgrades that model to `static_analysis: off` +and the compile succeeds. (Strict analysis queries the warehouse catalog for +source schemas; a `ref()`ed model is inferred statically and needs no built +table.) + +**The flag is not the capability.** `dbt-core` 2.0.0-alpha.5 — the version +`DBT_CORE_2X_VERSION` pins — accepts `--write-index` and `--write-lineage`, and +its own `views.sql` declares views over both tables, but it writes neither +parquet; only Fusion does today. The ADAPTER decides too: an experimental one +(postgres under `DBT_ALLOW_EXPERIMENTAL_ADAPTERS`) turns static analysis off and +says so only in a warning on an otherwise successful compile. The gate is +therefore "the engine has the flag" (everything but 1.x, whose Python CLI has no +such option) plus "the file appeared", so a later release picking the feature up +needs no change here — and the job log carries the engine's own stderr whenever +no index appears, since without it "no column lineage" has no explanation. + +`lineage_kind` is stored as TEXT, not an enum. Three values exist — `copy` +(passthrough), `mod` (transformed) and `scan` (the column was read to produce the +ROW rather than the value: a join key, a `where` predicate, a `group by`) — and +the engine's own reader maps those three and passes anything else through, so the +set is the engine's to extend. All three are stored and sent; the column-trace +diagram draws `copy` and `mod`, because a `scan` edge reaches every output column +of its model and would render as a complete bipartite graph. + +Storage mirrors `dbt_edge` exactly: `dbt_column_edge`, keyed by (path, version, +job) with the same composite foreign key to `script`, so a version's column +lineage dies with the version and a run's snapshot with the sweep. The typed +column list lands in `dbt_node.column_schema`, beside `columns` rather than +merged into it — `columns` stays what the author *declared*. ## Concept mapping @@ -1167,7 +1215,9 @@ a SQL-AST pass of our own, so `columnLineageGraph.ts` is not wired up for dbt. | `materialized: incremental` | `append` or `merge` (by `unique_key`) | same | | `{% snapshot %}` | `scd2` | same, incl. `_current` handling | | `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; see the note below | +| 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 | | 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 2c5e5fd97e..be2141ce3c 100644 --- a/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte +++ b/frontend/src/lib/components/assets/AssetGraph/AssetGraphDetailsPane.svelte @@ -446,6 +446,15 @@ 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 @@ -1221,7 +1230,7 @@ {/if} - {#if selectionColumnGraph && assetColumnNodes(selectionColumnGraph, selection.asset_kind, selection.path).length > 0} + {#if selectionColumnGraph && selectionColumnNodes.length > 0}
{/key} - {:else if selectionDbt?.raw_code} + {:else if selectionDbt && (selectionDbt.raw_code || selectionColumnNodes.length > 0)} + {#if selectionColumnGraph && selectionColumnNodes.length > 0} +
+ +
+ {/if} + {#if selectionDbt.raw_code} +
+ +
+ {/if} {:else}
diff --git a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts index 6842dcfd86..0783875fe4 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.test.ts @@ -109,6 +109,48 @@ describe('buildColumnGraph', () => { expect(colNodeId('ducklake', 'a:b', 'c')).not.toBe(colNodeId('ducklake', 'a', 'b:c')) }) + it('takes dbt column edges but not the indirect `scan` ones', () => { + // dbt arrives already resolved to two relations rather than anchored to a + // producer. `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 the lineage a column trace means. + const g = buildColumnGraph({ + assets: [], + runnables: [], + edges: [], + triggers: [], + dbt_column_edges: [ + { + 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')]) + ) + }) + it('skips producers with no ducklake output asset (columns unanchorable)', () => { const graph = chainGraph() graph.edges = graph.edges.filter((e) => e.runnable_path !== 's1') // s1 loses its output edge diff --git a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts index 50fa1c4e7b..01111b1446 100644 --- a/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts +++ b/frontend/src/lib/components/assets/AssetGraph/columnLineageGraph.ts @@ -23,11 +23,24 @@ export type ColumnLineageGraph = { down: Map> } +// Direct value flow, as dbt's static analysis labels it: `copy` passes a column +// through, `mod` transforms it. A `scan` edge is the third kind and 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. Kept on the +// wire (the API sends all three) so a later "show indirect" control needs no +// backend change; kept out of the lineage a column trace means. +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 // from its write-edge. Producers without a known ducklake output are skipped // (their columns can't be anchored to an asset node). +// +// dbt contributes the same edges from the other side: a dbt project is one +// runnable with many output assets, so its lineage arrives already resolved to +// the two relations rather than anchored to a producer. export function buildColumnGraph(graph: AssetGraphResponse): ColumnLineageGraph { const nodes = new Map() const up = new Map>() @@ -86,6 +99,14 @@ export function buildColumnGraph(graph: AssetGraphResponse): ColumnLineageGraph } } + for (const e of graph.dbt_column_edges ?? []) { + if (!DIRECT_DBT_LINEAGE.has(e.kind)) continue + addEdge( + addNode({ kind: 'dbt', path: e.from_asset_path, column: e.from_column }), + addNode({ kind: 'dbt', path: e.to_asset_path, column: e.to_column }) + ) + } + return { nodes, up, down } } diff --git a/frontend/src/lib/components/assets/AssetGraph/types.ts b/frontend/src/lib/components/assets/AssetGraph/types.ts index 4ada1feec3..639fd31135 100644 --- a/frontend/src/lib/components/assets/AssetGraph/types.ts +++ b/frontend/src/lib/components/assets/AssetGraph/types.ts @@ -36,9 +36,14 @@ export interface DbtAssetProvenance { tags?: string[] description?: string data_tests?: DbtDataTest[] - /** Declared column metadata (name -> description). NOT column lineage: - * `manifest.json` carries none (docs/dbt-runtime.md, decision 14). */ + /** Declared column metadata (name -> description): what `manifest.json` + * carries, which is only the columns an author wrote down. */ columns?: Record + /** Every column of the relation, typed and in the order the model produces + * them, from the engine's static analysis. Present only for a project that + * opted into it (`column_lineage: true`); `manifest.json` has no such + * thing. Lockstep with Rust `DbtAssetProvenance.column_schema`. */ + column_schema?: { name: string; type?: string }[] /** A source's declared freshness policy. */ freshness?: unknown /** The model's SQL as written — the transform behind the node. Read-only: @@ -225,6 +230,7 @@ export interface AssetGraphResponse { macro_edges?: AssetGraphMacroEdge[] test_edges?: AssetGraphTestEdge[] dbt_edges?: AssetGraphDbtEdge[] + dbt_column_edges?: AssetGraphDbtColumnEdge[] /** The job whose snapshot the dbt half was resolved from, when one was * asked for and found. A run page polls the graph until this is its own * job, which is how it knows a dynamic descriptor's ingest has landed. */ @@ -243,6 +249,21 @@ export interface AssetGraphDbtEdge { to_asset_path: string } +// Column-to-column lineage between two dbt models, from the engine's own static +// analysis. Empty unless the project opted into it — `manifest.json` carries no +// column edges at all. +export interface AssetGraphDbtColumnEdge { + from_asset_path: string + from_column: string + to_asset_path: string + to_column: string + // dbt's own word: `copy` (passthrough), `mod` (transformed), `scan` (read to + // produce the ROW rather than the value — a join key, a predicate, a + // `group by`). Left open: the engine's own reader maps those three and + // passes anything else through. + kind: string +} + export type AssetGraphNodeData = | { kind: 'asset' diff --git a/frontend/src/lib/components/dbt/DbtEditor.svelte b/frontend/src/lib/components/dbt/DbtEditor.svelte index f12e510c61..0e53c86f7b 100644 --- a/frontend/src/lib/components/dbt/DbtEditor.svelte +++ b/frontend/src/lib/components/dbt/DbtEditor.svelte @@ -31,6 +31,7 @@ AssetGraphNodeData, DbtAssetProvenance } from '$lib/components/assets/AssetGraph/types' + import type { ColumnLineageGraph } from '$lib/components/assets/AssetGraph/columnLineageGraph' import { DBT_DESCRIPTOR, DBT_MODULE_EXTENSIONS, @@ -207,6 +208,7 @@ let graphSelection = $state(undefined) let selectedAsset = $derived(graphSelection?.kind === 'asset' ? graphSelection : undefined) let selectedDbt = $state(undefined) + let selectedColumnGraph = $state(undefined) // 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 @@ -514,10 +516,11 @@ testRunning={testIsLoading} testResult={testJob?.result} selection={graphSelection} - onSelect={(sel, dbt, buffer) => { + onSelect={(sel, dbt, buffer, columnGraph) => { graphSelection = sel selectedDbt = dbt selectedBuffer = buffer + selectedColumnGraph = columnGraph }} /> @@ -539,6 +542,7 @@ {args} fileInBundle={!!selectedDbt.original_file_path && !!modules?.[selectedDbt.original_file_path]} + columnGraph={selectedColumnGraph} onOpenFile={open} onClose={() => (graphSelection = undefined)} /> diff --git a/frontend/src/lib/components/dbt/DbtModelDetails.svelte b/frontend/src/lib/components/dbt/DbtModelDetails.svelte index 2d25dbdc00..6d6ded71ef 100644 --- a/frontend/src/lib/components/dbt/DbtModelDetails.svelte +++ b/frontend/src/lib/components/dbt/DbtModelDetails.svelte @@ -12,6 +12,11 @@ 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' @@ -34,6 +39,10 @@ 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. + * Resolved from the same graph response the canvas draws, so the trace + * and the nodes above it describe one parse. */ + columnGraph, onOpenFile, onClose }: { @@ -45,6 +54,7 @@ buffer?: DbtPreviewBuffer args?: Record fileInBundle?: boolean + columnGraph?: ColumnLineageGraph onOpenFile?: (path: string) => void onClose?: () => void } = $props() @@ -111,7 +121,29 @@ return typeof v === 'object' ? JSON.stringify(v) : String(v) } - let columns = $derived(Object.entries(dbt.columns ?? {})) + // The real columns where the analysis pass produced them — typed and in the + // order the model emits them — and the declared ones otherwise. The + // description comes from `columns` either way: that is the only place an + // author's prose lives, and a project documents a handful of forty. + let columns = $derived( + dbt.column_schema?.length + ? dbt.column_schema.map((c) => ({ + name: c.name, + type: c.type, + description: dbt.columns?.[c.name] ?? '' + })) + : Object.entries(dbt.columns ?? {}).map(([name, description]) => ({ + name, + type: undefined, + description + })) + ) + 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. @@ -238,20 +270,29 @@
{#if columns.length > 0}
-
columns declared
+
+ {columnsAreAnalyzed ? 'columns' : 'columns declared'} +
- {#each columns as [name, desc] (name)} + {#each columns as col (col.name)}
- {name} - {desc} + {col.name} + {#if col.type} + {col.type} + {/if} + {col.description}
{/each}
- -
- Declared metadata — dbt reports no column-level lineage. -
+ + {#if !columnsAreAnalyzed} +
+ Declared metadata. Set `column_lineage: true` in the descriptor for the real + column schema and column-level lineage. +
+ {/if}
{/if} {#if (dbt.data_tests?.length ?? 0) > 0} @@ -269,6 +310,17 @@
{/if} + {#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 e6d948d926..e4033a4c99 100644 --- a/frontend/src/lib/components/dbt/DbtModelGraph.svelte +++ b/frontend/src/lib/components/dbt/DbtModelGraph.svelte @@ -25,6 +25,10 @@ DbtAssetProvenance } from '$lib/components/assets/AssetGraph/types' import { useDbtRunStatus } from './runStatus.svelte' + import { + buildColumnGraph, + type ColumnLineageGraph + } from '$lib/components/assets/AssetGraph/columnLineageGraph' let { workspace, @@ -80,7 +84,12 @@ * 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 + buffer: DbtPreviewBuffer | undefined, + /** The project's column-level lineage, when the descriptor asked for + * it. Travels with the selection for the same reason the provenance + * does: it is resolved from the graph response, which is this + * component's. */ + columnGraph: ColumnLineageGraph | undefined ) => void } = $props() @@ -364,6 +373,13 @@ // graph that actually came back. let editorParsed = $derived(refreshJob != undefined && raw?.dbt_snapshot_job === refreshJob) + // Built lazily off the graph on screen, so the trace in the details pane + // describes the same parse the canvas does. Empty for a project that never + // asked for the analysis pass, which is the ordinary case. + let columnGraph = $derived( + graph?.dbt_column_edges?.length ? buildColumnGraph(graph) : undefined + ) + // `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. @@ -374,7 +390,8 @@ sel?.kind === 'asset' ? graph?.assets.find((a) => a.kind === sel.asset_kind && a.path === sel.path)?.dbt : undefined, - editorParsed ? parsedBuffer : undefined + editorParsed ? parsedBuffer : undefined, + sel?.kind === 'asset' ? columnGraph : undefined ) ) } diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index 76ecfa59d8..ac791557a8 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -1381,6 +1381,11 @@ full_refresh: false # resolved to that Windmill variable, so secrets stay out of this file. # env: # DBT_PASSWORD: $var:u/user/my_warehouse_password +# Column-level lineage and real column schemas, from the engine's static +# analysis. Opt-in because it runs a separate dbt compile under +# --static-analysis strict, which rejects SQL the default accepts; a project it +# cannot analyze keeps the graph it has. Needs an engine that computes it. +# column_lineage: true ` // for related places search: ADD_NEW_LANG export const INITIAL_CODE = { diff --git a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte index cda7a4718d..5d8ccd5e00 100644 --- a/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/pipeline/[folder]/+page.svelte @@ -1989,14 +1989,16 @@ 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` + // (inferred + annotated) `column_lineage`, the asset write-edges and the + // column edges a dbt project's static analysis produced. 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 a - // ducklake-asset selection so it isn't rebuilt on every editor keystroke when - // the trace UI isn't even shown. + // 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 columnGraph = $derived( - pe.selection?.kind === 'asset' && pe.selection.asset_kind === 'ducklake' + pe.selection?.kind === 'asset' && + (pe.selection.asset_kind === 'ducklake' || pe.selection.asset_kind === 'dbt') ? buildColumnGraph(displayGraph) : EMPTY_COLUMN_GRAPH )