mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-26 00:03:59 +00:00
feat: column-level lineage for dbt from the engine's parquet index
`manifest.json` carries no column-to-column edges, which is why decision 14 recorded column lineage as unavailable. The edges live in a different artifact: `dbt compile --static-analysis strict --write-index` writes `target/index/`, whose `dbt.column_lineage.parquet` holds them and whose `dbt.node_columns.parquet` holds every column of every node, typed and ordered rather than only the ones an author documented. Strict analysis rejects SQL the default accepts, so this is a separate compile with its own `--target-path`, opt-in per project via `column_lineage: true`, and best-effort throughout: a project it cannot analyze keeps exactly the graph it had, with the engine's own diagnostics in the job log. Storage mirrors `dbt_edge`: `dbt_column_edge` keyed by (path, version, job) with the same composite FK to `script` and the same sweeps. The typed column list lands in `dbt_node.column_schema`, beside `columns` rather than merged into it, so `columns` stays what the author declared. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PRtsPQ3Ck69Fu9DNr7bMcJ
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5
Claude Opus 5
parent
11138284ac
commit
39fa2b10d6
+16
@@ -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"
|
||||
}
|
||||
+11
-5
File diff suppressed because one or more lines are too long
+23
@@ -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"
|
||||
}
|
||||
+14
@@ -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"
|
||||
}
|
||||
+17
@@ -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"
|
||||
}
|
||||
+50
@@ -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"
|
||||
}
|
||||
Generated
+1
@@ -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",
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -0,0 +1,2 @@
|
||||
DROP TABLE IF EXISTS dbt_column_edge;
|
||||
ALTER TABLE dbt_node DROP COLUMN IF EXISTS column_schema;
|
||||
@@ -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;
|
||||
@@ -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<String>,
|
||||
#[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
|
||||
|
||||
@@ -663,10 +663,18 @@ struct DbtAssetProvenance {
|
||||
description: Option<String>,
|
||||
#[serde(skip_serializing_if = "Vec::is_empty")]
|
||||
data_tests: Vec<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.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
columns: Option<serde_json::Value>,
|
||||
/// 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<serde_json::Value>,
|
||||
/// A source's declared freshness policy, for the staleness chip.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
freshness: Option<serde_json::Value>,
|
||||
@@ -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<DbtLineageEdge>,
|
||||
/// 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<DbtColumnLineageEdge>,
|
||||
/// 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<String>,
|
||||
@@ -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<DbtColumnLineageEdge> = 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,
|
||||
}))
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<String>,
|
||||
pub attached_node: Option<String>,
|
||||
pub columns: Option<serde_json::Value>,
|
||||
/// 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<serde_json::Value>,
|
||||
pub freshness: Option<serde_json::Value>,
|
||||
/// 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<String>,
|
||||
}
|
||||
|
||||
/// 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<IngestedColumnEdge>,
|
||||
pub columns: HashMap<String, Vec<IndexedColumn>>,
|
||||
}
|
||||
|
||||
// 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<IngestedNode>,
|
||||
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<IngestedColumnEdge>,
|
||||
/// The `asset` rows the owning script produces (models) and consumes
|
||||
/// (sources) — what the lineage graph is drawn from.
|
||||
pub assets: Vec<AssetWithAltAccessType>,
|
||||
@@ -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<IngestedColumnEdge> = 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::<BTreeMap<_, _>>())
|
||||
}),
|
||||
// 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<_>>(),
|
||||
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.
|
||||
|
||||
@@ -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<Postgres>, 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<Postgres>, hash: i64) -> i64 {
|
||||
sqlx::query_scalar!(
|
||||
@@ -275,6 +301,8 @@ async fn deleting_the_script_cascades_to_every_sidecar(db: Pool<Postgres>) {
|
||||
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<Postgres>
|
||||
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"
|
||||
))
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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<ColumnIndex> {
|
||||
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<ColumnIndex> {
|
||||
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<ColumnIndex> {
|
||||
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<Vec<Row>> {
|
||||
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::<Result<Vec<_>, _>>()
|
||||
.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<i64> {
|
||||
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,
|
||||
}
|
||||
}
|
||||
@@ -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<String> {
|
||||
) -> error::Result<Captured> {
|
||||
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()]);
|
||||
|
||||
@@ -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;
|
||||
|
||||
+57
-7
@@ -36,7 +36,7 @@ the dominant way dbt is orchestrated today.
|
||||
| 11 | Asset kind | `dbt://<warehouse>/<schema>/<name>` — 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. `<dim>_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` |
|
||||
|
||||
@@ -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 @@
|
||||
</span>
|
||||
</div>
|
||||
{/if}
|
||||
{#if selectionColumnGraph && assetColumnNodes(selectionColumnGraph, selection.asset_kind, selection.path).length > 0}
|
||||
{#if selectionColumnGraph && selectionColumnNodes.length > 0}
|
||||
<div class="border-b shrink-0">
|
||||
<ColumnLineageTrace
|
||||
graph={selectionColumnGraph}
|
||||
@@ -1236,7 +1245,7 @@
|
||||
</div>
|
||||
</div>
|
||||
{/key}
|
||||
{:else if selectionDbt?.raw_code}
|
||||
{:else if selectionDbt && (selectionDbt.raw_code || selectionColumnNodes.length > 0)}
|
||||
<!-- The transform behind the node. Read-only on purpose: dbt
|
||||
development is a local loop (`dbt run --select`, `dbt test`
|
||||
against a dev target), and a browser textarea over one file
|
||||
@@ -1249,11 +1258,29 @@
|
||||
<DbtIcon width={11} height={11} />
|
||||
<span class="font-mono truncate">{dbtBundlePath ?? selectionDbt.unique_id}</span
|
||||
>
|
||||
<span class="ml-auto shrink-0 opacity-70">read-only · edit locally</span>
|
||||
</div>
|
||||
<div class="flex-1 min-h-0 overflow-auto">
|
||||
<HighlightCode language="sql" code={selectionDbt.raw_code} />
|
||||
{#if selectionDbt.raw_code}
|
||||
<span class="ml-auto shrink-0 opacity-70">read-only · edit locally</span>
|
||||
{/if}
|
||||
</div>
|
||||
<!-- Above the SQL rather than beside it: the columns are what
|
||||
the SQL below produces, so reading them in that order is
|
||||
the model's own shape. Same trace component the ducklake
|
||||
assets use — the graph is one graph across both. -->
|
||||
{#if selectionColumnGraph && selectionColumnNodes.length > 0}
|
||||
<div class="border-b shrink-0 overflow-auto max-h-64">
|
||||
<ColumnLineageTrace
|
||||
graph={selectionColumnGraph}
|
||||
assetKind={selection.asset_kind}
|
||||
assetPath={selection.path}
|
||||
targetLabel={selectionDbt.unique_id}
|
||||
/>
|
||||
</div>
|
||||
{/if}
|
||||
{#if selectionDbt.raw_code}
|
||||
<div class="flex-1 min-h-0 overflow-auto">
|
||||
<HighlightCode language="sql" code={selectionDbt.raw_code} />
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
{:else}
|
||||
<div class="p-3 text-xs text-secondary">
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -23,11 +23,24 @@ export type ColumnLineageGraph = {
|
||||
down: Map<ColumnNodeId, Set<ColumnNodeId>>
|
||||
}
|
||||
|
||||
// Direct value flow, as dbt's static analysis labels it: `copy` passes a column
|
||||
// through, `mod` transforms it. 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<ColumnNodeId, ColumnNode>()
|
||||
const up = new Map<ColumnNodeId, Set<ColumnNodeId>>()
|
||||
@@ -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 }
|
||||
}
|
||||
|
||||
|
||||
@@ -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<string, string>
|
||||
/** 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'
|
||||
|
||||
@@ -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<AssetGraphNodeData | undefined>(undefined)
|
||||
let selectedAsset = $derived(graphSelection?.kind === 'asset' ? graphSelection : undefined)
|
||||
let selectedDbt = $state<DbtAssetProvenance | undefined>(undefined)
|
||||
let selectedColumnGraph = $state<ColumnLineageGraph | undefined>(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
|
||||
}}
|
||||
/>
|
||||
</Pane>
|
||||
@@ -539,6 +542,7 @@
|
||||
{args}
|
||||
fileInBundle={!!selectedDbt.original_file_path &&
|
||||
!!modules?.[selectedDbt.original_file_path]}
|
||||
columnGraph={selectedColumnGraph}
|
||||
onOpenFile={open}
|
||||
onClose={() => (graphSelection = undefined)}
|
||||
/>
|
||||
|
||||
@@ -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<string, unknown>
|
||||
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 @@
|
||||
<div class="px-2 py-1.5 border-b flex flex-col gap-1.5">
|
||||
{#if columns.length > 0}
|
||||
<div class="text-2xs">
|
||||
<div class="text-tertiary mb-0.5">columns declared</div>
|
||||
<div class="text-tertiary mb-0.5">
|
||||
{columnsAreAnalyzed ? 'columns' : 'columns declared'}
|
||||
</div>
|
||||
<div class="flex flex-col gap-0.5">
|
||||
{#each columns as [name, desc] (name)}
|
||||
{#each columns as col (col.name)}
|
||||
<div class="flex gap-2">
|
||||
<span class="font-mono text-primary shrink-0">{name}</span>
|
||||
<span class="text-secondary truncate">{desc}</span>
|
||||
<span class="font-mono text-primary shrink-0">{col.name}</span>
|
||||
{#if col.type}
|
||||
<span class="font-mono text-tertiary shrink-0">{col.type}</span>
|
||||
{/if}
|
||||
<span class="text-secondary truncate">{col.description}</span>
|
||||
</div>
|
||||
{/each}
|
||||
</div>
|
||||
<!-- dbt's manifest carries no column-to-column edges, so this is a
|
||||
declared column SET rather than lineage. -->
|
||||
<div class="text-tertiary mt-0.5">
|
||||
Declared metadata — dbt reports no column-level lineage.
|
||||
</div>
|
||||
<!-- `manifest.json` carries declared columns only, so without the
|
||||
analysis pass this list is what an author wrote down rather than
|
||||
what the model produces. -->
|
||||
{#if !columnsAreAnalyzed}
|
||||
<div class="text-tertiary mt-0.5">
|
||||
Declared metadata. Set `column_lineage: true` in the descriptor for the real
|
||||
column schema and column-level lineage.
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
{/if}
|
||||
{#if (dbt.data_tests?.length ?? 0) > 0}
|
||||
@@ -269,6 +310,17 @@
|
||||
</div>
|
||||
{/if}
|
||||
|
||||
{#if columnGraph && columnNodes.length > 0}
|
||||
<div class="border-b overflow-auto max-h-64">
|
||||
<ColumnLineageTrace
|
||||
graph={columnGraph}
|
||||
assetKind="dbt"
|
||||
assetPath={assetPath}
|
||||
targetLabel={dbt.unique_id}
|
||||
/>
|
||||
</div>
|
||||
{/if}
|
||||
|
||||
{#if showRows && preview}
|
||||
{#if 'error' in preview}
|
||||
<div class="p-2 text-2xs text-secondary">{preview.error}</div>
|
||||
|
||||
@@ -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
|
||||
)
|
||||
)
|
||||
}
|
||||
|
||||
@@ -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 = {
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user