mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-08 16:03:27 +00:00
feat: ingest dbt column lineage and real column schemas from the engine's parquet index (#10977)
* 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 * fix: address review findings on the dbt column-lineage pass - The workspace fork copied every other dbt sidecar table and not this one, so a fork lost its column lineage silently and could not recover it: the cloned digest covers the column edges, so a dynamic run in the fork matched it and stored nothing. - The parquet was collected whole before the edge cap applied, which is exactly the input the cap exists for — a project whose `scan` lineage is quadratic in its widest model could take the worker process down. Decoded a row at a time with the bound enforced during the decode. - The pass swallowed every error from the runner, including the job poller's cancellation and deadline, so a run that blew its timeout inside an optional annotation could still publish a graph and report success. `run_captured` now carries the exit status in its value, so only a failed COMPILE is downgraded, and the pass may spend at most half the remaining wall clock so it cannot starve the build that follows it. - `scan` edges are stored but no longer served: they are most of a project's lineage, nothing renders them, and the graph endpoint is polled by the run page. They are also the first thing the storage cap gives up now, rather than evicting the direct edges the trace draws. - `column_schema` and the column edges take the same gate as the model's SQL. A column-level view is the shape of what the author wrote, one level finer than the `ref()` graph, which is ungated only because it draws relations the caller already sees. - `graph_digest` hashes the new section only when it has edges, so a project that never asked for the pass keeps the digest it has instead of re-snapshotting on every dynamic run until it is redeployed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PRtsPQ3Ck69Fu9DNr7bMcJ * fix: the editor buffer's column lineage, and three bounds that were wrong Round-2 review found four defects, all of them introduced by the round-1 fixes. - The `script_visible` gate on the column edges was copied from the node query without its `script_hash IS NULL` arm. `= NULL` is never true, so every version-less row was filtered out and an editor buffer's parse rendered its typed columns and none of their lineage — the one place the feature is meant to be used. Pinned by an assertion in `dbt_pinned_graph.rs`, which is where this class of bug already had a home. - The phase budget was handed to the poller, whose expiry is an `Err` indistinguishable from a cancellation or the job's own deadline, so a slow but valid analysis aborted the build it exists to annotate. The runner gets the full deadline again — those two must still fail the job — and the budget is a race around the whole pass, where expiring is this budget and nothing else. - The decode cap counted parquet ROWS, so `scan` and out-of-graph rows could spend it before a single drawn edge was read. It now counts what is kept, takes direct kinds in a first pass, and is handed the graph's own nodes so the budget cannot go on rows that could never be stored. - Hashing the new digest section conditionally did not preserve old digests, because an absent `column_schema` still serialized as `null` inside the nodes. It is skipped when absent instead. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PRtsPQ3Ck69Fu9DNr7bMcJ * refactor: split the lineage pass by error contract, and read it in one query Round 3's findings were all consequences of round 1 and 2's fixes, clustered in the same two files, so this reshapes those two seams rather than patching again. The worker pass was one function being three things at once — a subprocess runner with job-lifecycle error semantics, a bounded decoder, and a best-effort degrader — which is why each fix to one perturbed another. It is now `compile_index`, which owns the JOB's semantics (only a cancellation or the job's deadline can `Err`; a non-zero exit, the output ceiling and the phase budget are outcomes), and `read_index`, which owns the ARTIFACT's and knows nothing about the job. The budget wraps the compile alone, so a decode can no longer outlive the timeout that reported the build would get the rest. The output ceiling likewise becomes a value rather than a job error, for the caller that can carry on without the tail of a compile's stdout. The column edges were read by a fourth hand-written copy of the `live`/`chosen` CTEs and the version/editor-buffer join conditions, and copying them is what dropped the `script_hash IS NULL` arm and hid every buffer parse's lineage. Both kinds of edge now come from ONE statement over a `UNION ALL`'d edge source, so those conditions exist once. The union is at the source rather than a join because column lineage can name a node pair `dbt_edge` has no row for: a model reading `{{ this }}` gets edges from itself to itself, and `parent_map` has no self-loop. The cap on the column half now sits after the scope filter, the visibility check and the graph joins — the scope moved into SQL via the existing `ScopePathFilter` — so a row the caller may not read can no longer spend it and leave an allowed project's trace short. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PRtsPQ3Ck69Fu9DNr7bMcJ * refactor: serve dbt column lineage from its own endpoint The column edges rode on the folder-wide asset graph, which a run page polls, while the trace is drawn for one selected relation. That needed a cap, and a cap has to be applied after every filter that can drop a row. Keyed to the asset there is no cap: `assets/column_lineage` answers for one relation, and the caller's `scripts:read` scope and the project's visibility are decided once, for the script that owns it. Pinning to a run's snapshot or the editor's parse of its buffer costs the job-read gate, so that form is `jobs/dbt_column_lineage/{id}` — the same shape `jobs/dbt_graph/{id}` has. The worker's decode now bounds work and memory separately, and a compile stopped by the output ceiling reports as truncated rather than complete. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: resolve the owning dbt version the way the graph does The unpinned arm picked the newest live version at the path without narrowing to dbt, so a path since redeployed in another language answered with no lineage while the graph beside it still drew that project's stale nodes. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: gate pinned column lineage on reading the project, and answer the component Four things round 5 found, three of them in code this branch rewrote: - The pinned arm resolved the version from the job and stopped there, so a share-link viewer entitled to a run got the project's column names and edges while the graph beside it still redacted `raw_code` and `column_schema`. Resolving WHICH version answers is not deciding whether the caller may read it; the version-less editor buffer keeps its exemption, having no `script` row to ask. - The answer was the whole owning project's edges. The canvas lays out the connected component of the selected relation's columns, so the rest was unrenderable weight; a recursive walk over both directions returns exactly what is drawn, and the project key travels with it so a `unique_id` two projects share cannot walk from one graph into the other. - The decode had no exit but the 4M-row backstop once its buckets were full, spending wall clock the build below does not get. - An unreadable index was reported as a missing one, sending the reader to look at their engine rather than at the file. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: stitch the two column graphs, and walk the component in Rust Round 6's two findings, both regressions this branch introduced: - The decode returned `Continue` on the edge that FILLED the direct-edge budget, so a `scan`-only tail after it decoded to the 4M-row backstop with nowhere to put anything. The read now ends on that edge. - Seam 3 made the pipeline page choose between the dbt graph and the producer one. They share node ids — `// column total <- dbt://wh/analytics/orders.amount` mints the same `(dbt, path, column)` node dbt's own lineage does — so choosing ended a trace at the boundary in both directions. They are merged again, and a ducklake selection asks about the dbt relation its producers name so the chain continues past it. The dbt editor gets the same merge. Also: the component is walked in Rust rather than by a recursive CTE. A CTE has no index, so the recursive term rescanned the doubled edge set once per level — 1243ms against 59ms for the query alone on a 3000-model project, 11.7M rows in the plan. Same answers, same tests; end to end 1.48s to 0.73s there and 1.60s to 0.26s on a 1000-deep chain. The client stops re-asking for a component it already holds, which is most clicks within one project. The four doc sites that described a whole-project answer are rewritten around what it now is, rather than edited where they disagreed. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: expand every dbt boundary a selection reaches, and only skip what was asked Round 7's findings, all in the frontend seam this branch added: - A ducklake selection seeded the dbt fetch from the FIRST boundary relation it found, so a table derived from two unconnected dbt relations expanded one and left the other a leaf — the same "stops at the boundary" symptom the round-6 fix removed, one hop further along. Every distinct boundary is fetched now and the components merged. - The component cache skipped a relation merely PRESENT in the graph in hand. A relation two projects describe has an owner row in each, and a component fetched for one carries it as an endpoint without the other's half, so that skipped the request that would have resolved the second owner. Only a relation actually asked about under this pin is skipped. - A comment still called the producer graph gated to ducklake selections after it was widened to dbt. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * refactor: land dbt column lineage as storage and ingest only The API surface that draws a column trace moves to a follow-up PR, on `dbt-column-lineage-surface`. It kept generating findings — a client cache whose premise was wrong for a two-owner relation, then staleness and a lost retry from tightening it, and a seed walk that stopped at the first boundary — and the fix for the last of them is a transitive owner expansion, which has to re-apply the caller's gate to every newly discovered project. That is the same shape as the leak four reviewers caught in the pinned arm, and it wants its own review rather than being the fourth fix at the end of this one. What lands here stands on its own: the analysis pass, `dbt_column_edge`, `dbt_node.column_schema`, the engine gating and the error-contract split — plus the one user-visible half, the typed and ordered column list, which rides the asset graph the details pane already fetches and replaces a panel that could only show the columns an author had documented. Also fixes a real bug in the pass, found in review: it compiled without the build's `--full-refresh`. `is_incremental()` branches on that flag, so an incremental model reading `{{ this }}` compiles its self-join — and any `ref()` inside that branch — only when the flag is absent, and the pass was storing lineage for SQL a full-refresh run never executed. The flag now comes from one place shared with the build, and a run that overrides it gets its own graph rather than standing as the version's. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs: say why direct kinds get the budget without naming a view The bucketing comments explained the priority by what a trace draws, which is a forward reference now that the surface moved out. The reason stands on its own: `copy`/`mod` say the value travelled, `scan` says the column was read to produce the row and so reaches every output column of its model. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: round-9 findings on the descoped PR - The `full_refresh` helper was inserted between `selection_is_overridden` and its doc comment, so thirteen lines about `select`/`exclude` echoes documented the wrong function and the one they were written for had none. Moved below it. - The parse path ran the analysis compile and the parquet decode BEFORE the guard that returns when there is no warehouse identity, paying for both and dropping the result. Moved after it. - Three sites still described a `/column_lineage` endpoint this branch no longer has, and two user-facing strings promised a column trace it no longer renders: the panel's hint and the descriptor template now say what the flag actually buys, which is the typed column schema. - Dropped test scaffolding the removed suite left behind: a `raw_orders` node and `dbt_edge` whose only assertion re-tested pre-existing graph behaviour, and a second editor-buffer node nothing asserts on. Documented rather than fixed: an incremental model has two shapes, and which one the index holds depends on whether the target existed when the pass ran. `is_incremental()` is false with no target as well as under `--full-refresh`, and dbt has no mode that emits both — so a version's graph describes the compile that produced it, and only a re-ingesting run describes its own run. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: keep lineage_kind in the edge key, and one answer for --full-refresh - Both unique indexes omitted `lineage_kind`, so a column that is projected AND used as a predicate for the same output column — an ordinary shape — had its `copy` and `scan` edges collapse under `ON CONFLICT DO NOTHING`, while the digest counted both. The kind is part of the fact, so it is part of the key. Edited in the migration rather than added as a second one: it has not landed. - `full_refresh` was shared between the build and the analysis pass without the `command != "test"` condition that sat at the build's call site, so the two disagreed for exactly the runs that build nothing. The condition moved inside the function, which is the point of sharing it, and the command is threaded to the pass. - The "what a trace draws" rewrite missed the copy in `dbt_manifest.rs`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: drop the unreachable full_refresh threading, test the uniqueness key `DBT_COMMANDS` is `["build", "retry", "show", "parse"]` and `default_command` returns `build` in every arm, so `command == "test"` cannot happen — the guard the last commit moved into `full_refresh` was already inert where it came from. Threading the command through five signatures to preserve it bought nothing, so it is gone; the build and the pass call one function of the descriptor and the invocation, which is what the sharing was for. The uniqueness-key fix now has a test: a column projected AND used as a predicate for the same output column stores both its `copy` and its `scan` row. Verified against the old key, where it returns 1 instead of 2. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: restore the dbt test --full-refresh guard I removed on a wrong premise The previous commit removed it after reading `DBT_COMMANDS` and concluding `"test"` was unreachable. That is only true of the command a CALLER can name: `run_dbt` is invoked with `"test"` directly for the `after_all` test phase, so an `after_all` project with `full_refresh: true` reached it — and dbt rejects `--full-refresh` on `test`, failing the phase. Both reviewers caught it. The guard is back inside the shared function, where the build and the pass get one answer, and its doc now records why reading the allowlist alone is misleading. The test covering the `test` case is restored with it. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: notice a job that ended during the decode, and name truncation as the cause - The parquet decode runs on a blocking thread with no poller watching it, so a cancellation or an expired deadline during it was invisible: `dbt_dep` went on to publish the graph and the job returned success. The job's state is checked once the decode returns, before the caller publishes anything, and an ended job `Err`s — which this module may always do for the job's own semantics. - A compile stopped by the output ceiling could leave no artifact, and the log then blamed the engine's capability, sending the reader to check their adapter rather than the ceiling. Truncation now names itself in the missing and unreadable branches too. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: read cancellation from the DB after the decode, not from a poller's field `ctx.canceled_by` is only ever written by a poller, and no poller runs during the blocking decode — which is the exact window the check was added for. So the guard caught only a cancellation already observed before it, and the comment beside it claimed more than it did. It now queries `v2_job_queue` directly, the same probe `worker_lockfiles` uses before it overwrites a flow. A failed probe answers "still running": this decides whether to discard work already done, so an unreachable database must not be the reason a healthy deploy loses its graph. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * refactor: reuse job_is_canceled rather than a second copy of it The probe added last round was `job_is_canceled` from the same file, retyped — same query, same `Connection::Http` behaviour. Reused instead. Its doc said a non-database connection was "a failed probe", which reads as an error path. It is not: it is the agent worker, and on one there is no database to ask, so only the deadline answers and a cancel issued during the decode is not observable. The retry path avoids that by refusing to run on an agent worker at all — which an optional annotation has no business doing — so the gap is recorded at both ends instead. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: close the agent-worker cancellation gap instead of documenting it The previous commit said a cancel issued during the decode is not observable on an agent worker. It is: `ping_job_status` returns `canceled_by` over both connection kinds, and is how the poller itself notices one there. So the check asks through the ping rather than querying `v2_job_queue` directly, and holds on an agent worker, where a direct query reaches no database at all. `job_is_canceled` goes back to private and its doc to what it said before — the retry that calls it still refuses to run on an agent worker for its own reasons. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: decode the index under the job poller instead of checking after it Two findings with one cause: the decode was the only phase of this pass with no subprocess behind it, so nothing heartbeated while it ran. A large index left the worker silent for as long as it took, which the zombie sweep reads as a dead job and restarts — and the cancellation check bolted on afterwards could only ever report what had already happened, while dropping the ping's `already_completed`, so a force-cancelled deploy still published its graph. Running it under `run_future_with_polling_update_job_poller` answers all of it: the poller pings throughout, and ends the phase with an `Err` on cancellation, `AlreadyCompleted` or the phase timeout. The bespoke probe is gone with it. Verified on a live deploy: 32 edges and 4 typed schemas ingested through the polled decode. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: stop a cancelled decode, and say what the read phase can now do Putting the decode under the poller heartbeats it and ends the phase when the job does, but dropping a `JoinHandle` detaches a blocking task rather than cancelling it — so a cancelled job left a thread decoding up to four million rows for a job that was over. The row loop reads an abandonment flag that a drop guard on the awaiting future sets, so the decode stops at its next row. That same change made the read phase able to `Err`, and three places still said it could not — decision 14 in as many words. The distinction that holds is narrower: nothing the ARTIFACT does or fails to do can fail a job, so absent, unreadable and partial are all values; the JOB can still end the phase the read runs in. Stated that way in the module doc, the `Artifact` doc, `MAX_INDEX_ROWS` and the decision. Verified on a live deploy: 32 edges and 4 typed schemas. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * refactor: share AbortOnDrop, and stop citing a hazard that is now handled `Abandon` was `ansible_executor`'s `AbortOnDrop` retyped — same struct, same reason, same `spawn_blocking` shape. Moved to `common` and used from both. The paragraph explaining why the phase budget wraps the compile alone gave as its reason "a decode still running on a blocking thread", which is exactly what the abandonment flag now prevents. The reason that survives is the one that was always the point: the budget exists to leave the build its share of the clock, and only the compile can spend that share unboundedly. The decode's end is the job's, through the poller it runs under. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * fix: put both doc comments back on the items they describe Moving AbortOnDrop orphaned a doc at each end: it landed between `raw_to_string`'s doc and `raw_to_string`, and the doc of the struct it replaced stayed behind to prefix `fetch_repo_archive`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> * docs: name the binding the row loop actually reads `Abandoned` was neither the type nor the binding; the flag is `abandoned`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5
Claude Opus 5
parent
621fac55ab
commit
0139467b01
+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"
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id,\n resource_type, name, asset_path, materialized, materialize_strategy, unique_key,\n tags, description, test_kind, test_column, test_args, severity, attached_node,\n columns, column_schema, freshness, raw_code, original_file_path, ingested_at)\n SELECT $2, script_path, script_hash, job_id, unique_id,\n resource_type, name, asset_path, materialized, materialize_strategy, unique_key,\n tags, description, test_kind, test_column, test_args, severity, attached_node,\n columns, column_schema, freshness, raw_code, original_file_path, ingested_at\n FROM dbt_node\n WHERE workspace_id = $1 AND job_id = '00000000-0000-0000-0000-000000000000'",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Varchar"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "2e562fb35cfa71702e5fdbc505d2fa6f3114a151b66b0a116b1d98dd84115188"
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,\n parent_unique_id, parent_column, child_unique_id, child_column, lineage_kind,\n ingested_at)\n SELECT $2, script_path, script_hash, job_id, parent_unique_id, parent_column,\n child_unique_id, child_column, lineage_kind, ingested_at\n FROM dbt_column_edge\n WHERE workspace_id = $1 AND job_id = '00000000-0000-0000-0000-000000000000'",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Varchar"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "3b7858e47e4e3f31f861e114c7ff26b467185edcfde151f22c6e903d15722ad4"
|
||||
}
|
||||
+11
-5
File diff suppressed because one or more lines are too long
-15
@@ -1,15 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO dbt_node (workspace_id, script_path, script_hash, job_id, unique_id,\n resource_type, name, asset_path, materialized, materialize_strategy, unique_key,\n tags, description, test_kind, test_column, test_args, severity, attached_node,\n columns, freshness, raw_code, original_file_path, ingested_at)\n SELECT $2, script_path, script_hash, job_id, unique_id,\n resource_type, name, asset_path, materialized, materialize_strategy, unique_key,\n tags, description, test_kind, test_column, test_args, severity, attached_node,\n columns, freshness, raw_code, original_file_path, ingested_at\n FROM dbt_node\n WHERE workspace_id = $1 AND job_id = '00000000-0000-0000-0000-000000000000'",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Varchar"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "94c0aae349172b9295b81b9e61e2f8f0ca36920ab66e8922815157a0c147bc47"
|
||||
}
|
||||
+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"
|
||||
}
|
||||
Generated
+1
@@ -16699,6 +16699,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",
|
||||
|
||||
@@ -665,6 +665,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,68 @@
|
||||
-- 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. `lineage_kind` is part of the key because it is part
|
||||
-- of the fact: a column that is both projected and used as a predicate for the
|
||||
-- same output column has a `copy` edge AND a `scan` one, and the digest counts
|
||||
-- both. Leaving it out let `ON CONFLICT DO NOTHING` drop the second while the
|
||||
-- digest still claimed it was stored.
|
||||
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,
|
||||
lineage_kind)
|
||||
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,
|
||||
lineage_kind)
|
||||
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;
|
||||
@@ -59,6 +59,19 @@ impl DbtEngine {
|
||||
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)
|
||||
}
|
||||
|
||||
/// Whether the engine has `--defer-state`, the deferral-only half of
|
||||
/// `--state`.
|
||||
///
|
||||
@@ -139,6 +152,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
|
||||
|
||||
@@ -70,13 +70,15 @@ ci_test_reference: workspace_id(char), test_script_path(char), test_script_hash(
|
||||
concurrency_settings: hash(bigint), concurrency_key(char), concurrent_limit(int), concurrency_time_window_s(int)
|
||||
config: name(char), config(jsonb)
|
||||
custom_concurrency_key_ended: key(char), ended_at(ts)
|
||||
dbt_column_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), parent_unique_id(text), parent_column(text), child_unique_id(text), child_column(text), lineage_kind(text), ingested_at(ts)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), parent_unique_id(text), child_unique_id(text), ingested_at(ts)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_graph_snapshot: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), digest(text), relation_root_at_last_ingest(text), ingested_at(ts), permissioned_as(char)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_environment_state: workspace_id(char), script_path(char), environment(text), job_id(uuid), manifest(text), manifest_key(text), run_results(text), run_results_key(text), updated_at(ts)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
dbt_node: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), unique_id(text), resource_type(text), name(text), asset_path(text), materialized(text), materialize_strategy(text), unique_key(text), tags(text[]), description(text), test_kind(text), test_column(text), test_args(jsonb), severity(text), attached_node(text), columns(jsonb), freshness(jsonb), raw_code(text), original_file_path(text), ingested_at(ts)
|
||||
dbt_node: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), unique_id(text), resource_type(text), name(text), asset_path(text), materialized(text), materialize_strategy(text), unique_key(text), tags(text[]), description(text), test_kind(text), test_column(text), test_args(jsonb), severity(text), attached_node(text), columns(jsonb), column_schema(jsonb), freshness(jsonb), raw_code(text), original_file_path(text), ingested_at(ts)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_run_progress: workspace_id(char), job_id(uuid), asset_kind(asset_kind), asset_path(char), status(materialization_status), row_count(bigint), error(text), updated_at(ts)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
|
||||
@@ -663,10 +663,21 @@ 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.
|
||||
///
|
||||
/// Gated exactly like `columns` and the model's SQL: a full column list is
|
||||
/// the shape of what the author WROTE, one level finer than the `ref()`
|
||||
/// graph, which is ungated only because it draws relations the caller
|
||||
/// already sees in `asset`. Widening that boundary has to be a decision, not
|
||||
/// a consequence of a project turning the analysis pass on.
|
||||
#[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>,
|
||||
@@ -1323,7 +1334,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
|
||||
@@ -1379,6 +1390,10 @@ pub async fn asset_graph_for(
|
||||
// `ref()` lineage between two models, resolved to the relations they
|
||||
// produce. Joined to `dbt_node` on both key columns because a dbt
|
||||
// `unique_id` is only unique within its project.
|
||||
//
|
||||
// Column lineage is NOT here. It is stored per relation and per column, and
|
||||
// this response is folder-wide and polled by a run page, so it carries only
|
||||
// what the canvas draws for every node at once.
|
||||
let dbt_edge_rows = sqlx::query!(
|
||||
r#"WITH live AS (
|
||||
SELECT * FROM (
|
||||
@@ -1604,6 +1619,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().filter(|_| source_allowed),
|
||||
freshness: r.freshness.clone().filter(|_| source_allowed),
|
||||
};
|
||||
// One relation can carry rows from several projects — typically a model
|
||||
|
||||
@@ -6627,7 +6627,7 @@ async fn clone_scripts(
|
||||
}
|
||||
|
||||
/// The parsed dbt graph a deployed script carries: its models, their SQL and
|
||||
/// tests, and the `ref()` lineage between them.
|
||||
/// tests, and the `ref()` and column-level lineage between them.
|
||||
///
|
||||
/// Keyed on (workspace_id, script_path, script_hash), and the fork keeps every
|
||||
/// script's hash, so each row moves across as itself.
|
||||
@@ -6645,11 +6645,11 @@ async fn clone_dbt_graph(
|
||||
"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, ingested_at)
|
||||
columns, column_schema, freshness, raw_code, original_file_path, ingested_at)
|
||||
SELECT $2, 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, ingested_at
|
||||
columns, column_schema, freshness, raw_code, original_file_path, ingested_at
|
||||
FROM dbt_node
|
||||
WHERE workspace_id = $1 AND job_id = '00000000-0000-0000-0000-000000000000'",
|
||||
source_workspace_id,
|
||||
@@ -6669,6 +6669,24 @@ async fn clone_dbt_graph(
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
// Column lineage travels with the rest of the graph, and it has to: the
|
||||
// snapshot's digest covers it, so a fork missing these rows recomputes the
|
||||
// digest the source stored, matches, and stores nothing — leaving the
|
||||
// lineage gone until someone redeploys, which is the failure this whole
|
||||
// function exists to prevent.
|
||||
sqlx::query!(
|
||||
"INSERT INTO dbt_column_edge (workspace_id, script_path, script_hash, job_id,
|
||||
parent_unique_id, parent_column, child_unique_id, child_column, lineage_kind,
|
||||
ingested_at)
|
||||
SELECT $2, script_path, script_hash, job_id, parent_unique_id, parent_column,
|
||||
child_unique_id, child_column, lineage_kind, ingested_at
|
||||
FROM dbt_column_edge
|
||||
WHERE workspace_id = $1 AND job_id = '00000000-0000-0000-0000-000000000000'",
|
||||
source_workspace_id,
|
||||
target_workspace_id
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
sqlx::query!(
|
||||
"INSERT INTO dbt_graph_snapshot (workspace_id, script_path, script_hash, job_id,
|
||||
digest, relation_root_at_last_ingest, ingested_at)
|
||||
|
||||
@@ -26276,7 +26276,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, and gated like `columns` and the model's
|
||||
SQL: a full column list is the shape of what the author wrote.
|
||||
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
|
||||
|
||||
@@ -190,6 +190,19 @@ fn graph_digest(ingested: &IngestedManifest, relation_root: &str) -> String {
|
||||
.unwrap_or_default()
|
||||
.as_bytes(),
|
||||
);
|
||||
// Only when there are any, so a project that never asked for the analysis
|
||||
// pass keeps the digest it already has. Hashing an empty section
|
||||
// unconditionally would change every stored digest at once, and every
|
||||
// dynamic run would then store a full snapshot until its script is
|
||||
// redeployed — which reads exactly like the suppression above never working.
|
||||
if !ingested.column_edges.is_empty() {
|
||||
h.update(b"\0");
|
||||
h.update(
|
||||
serde_json::to_string(&ingested.column_edges)
|
||||
.unwrap_or_default()
|
||||
.as_bytes(),
|
||||
);
|
||||
}
|
||||
format!("{:x}", h.finalize())
|
||||
}
|
||||
|
||||
@@ -299,6 +312,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.
|
||||
@@ -327,7 +348,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
|
||||
@@ -383,6 +404,18 @@ 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.
|
||||
///
|
||||
/// Skipped when absent, unlike its neighbours, because `graph_digest`
|
||||
/// serializes these nodes: emitting `"column_schema":null` would change
|
||||
/// every stored digest at once, and every dynamic run of a project that
|
||||
/// never asked for the pass would store a full snapshot until its script is
|
||||
/// redeployed.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
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.
|
||||
@@ -390,6 +423,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.
|
||||
@@ -400,6 +467,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>,
|
||||
@@ -407,6 +477,88 @@ 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;
|
||||
|
||||
/// Whether the value travelled along this edge, as opposed to the column merely
|
||||
/// being read to produce the row.
|
||||
///
|
||||
/// A `scan` edge reaches every output column of its model, so it is most of what
|
||||
/// a wide project's index holds and the first thing `MAX_COLUMN_EDGES` gives up.
|
||||
/// It is still stored, for a view that wants indirect influence.
|
||||
pub fn is_direct(lineage_kind: &str) -> bool {
|
||||
matches!(lineage_kind, "copy" | "mod")
|
||||
}
|
||||
|
||||
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.
|
||||
//
|
||||
// Direct kinds first, so what the truncation below gives up is `scan` —
|
||||
// the bulk of a wide project's lineage, and the kind that says the column
|
||||
// was read to produce the row rather than the value. The
|
||||
// worker's reader already applies this order while decoding, because the
|
||||
// memory bound has to; repeating it here is what makes the ordering a
|
||||
// property of the manifest rather than of one caller's reader, and it is
|
||||
// the only ordering an index assembled some other way would get.
|
||||
edges.sort_by(|a, b| {
|
||||
is_direct(&b.lineage_kind)
|
||||
.cmp(&is_direct(&a.lineage_kind))
|
||||
.then_with(|| a.cmp(b))
|
||||
});
|
||||
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
|
||||
@@ -657,6 +809,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
|
||||
@@ -828,6 +983,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.
|
||||
@@ -883,7 +1048,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)
|
||||
@@ -905,6 +1070,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);
|
||||
@@ -928,6 +1094,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(())
|
||||
}
|
||||
|
||||
@@ -968,7 +1154,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"
|
||||
))
|
||||
@@ -1018,7 +1204,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"
|
||||
@@ -1037,6 +1223,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
|
||||
@@ -1072,6 +1260,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
|
||||
@@ -1107,7 +1304,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"
|
||||
@@ -1812,6 +2009,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.
|
||||
|
||||
@@ -9,8 +9,8 @@ use sqlx::{Pool, Postgres};
|
||||
use windmill_common::dbt_manifest::{
|
||||
clear_dbt_editor_graphs, clear_dbt_manifest_version, clear_dbt_script_state,
|
||||
clear_dbt_script_state_if_path_retired, move_dbt_script_state, 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";
|
||||
@@ -53,10 +53,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!(
|
||||
@@ -136,6 +162,33 @@ async fn an_identical_run_stores_no_snapshot(db: Pool<Postgres>) {
|
||||
assert_eq!(markers(&db, 1).await, 1, "and leaves no marker of its own");
|
||||
}
|
||||
|
||||
/// A column that is projected AND used as a predicate for the same output column
|
||||
/// has both a `copy` edge and a `scan` one. They are two facts, and the digest
|
||||
/// counts both — so the uniqueness key has to carry `lineage_kind`, or the
|
||||
/// second is dropped by `ON CONFLICT DO NOTHING` while the digest still claims
|
||||
/// it was stored.
|
||||
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
||||
async fn both_kinds_of_one_column_pair_are_stored(db: Pool<Postgres>) {
|
||||
deploy_script(&db, 1).await;
|
||||
let pair = |kind: &str| IngestedColumnEdge {
|
||||
parent_unique_id: "model.p.a".to_string(),
|
||||
parent_column: "id".to_string(),
|
||||
child_unique_id: "model.p.b".to_string(),
|
||||
child_column: "id".to_string(),
|
||||
lineage_kind: kind.to_string(),
|
||||
};
|
||||
let mut m = manifest(&["a", "b"]);
|
||||
m.column_edges = vec![pair("copy"), pair("scan")];
|
||||
|
||||
let mut tx = db.begin().await.unwrap();
|
||||
replace_dbt_manifest(&mut tx, WS, PATH, 1, None, &m, "root")
|
||||
.await
|
||||
.unwrap();
|
||||
tx.commit().await.unwrap();
|
||||
|
||||
assert_eq!(column_edges_for(&db, 1).await, 2, "both kinds survive");
|
||||
}
|
||||
|
||||
/// A run whose model set differs keeps its own, and the version's is untouched:
|
||||
/// this is what lets an older run page render the project that run built.
|
||||
#[sqlx::test(migrations = "../migrations", fixtures("base"))]
|
||||
@@ -280,6 +333,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);
|
||||
}
|
||||
|
||||
@@ -320,7 +375,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
|
||||
|
||||
@@ -408,16 +408,6 @@ pub fn create_empty_dir(path: &PathBuf) -> std::io::Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
/// Signals a detached `spawn_blocking` task that the future awaiting it is
|
||||
/// gone, so it can stop instead of running to completion in the background.
|
||||
struct AbortOnDrop(std::sync::Arc<std::sync::atomic::AtomicBool>);
|
||||
|
||||
impl Drop for AbortOnDrop {
|
||||
fn drop(&mut self) {
|
||||
self.0.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// Lay down the tree of an app-backed repository, which git can't clone
|
||||
/// because its URL carries no credential.
|
||||
///
|
||||
@@ -490,7 +480,7 @@ async fn fetch_repo_archive(
|
||||
// stopping it, so the flag is what a cancelled job uses to reach the
|
||||
// extraction loop. The guard sets it when this future is dropped.
|
||||
let aborted = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
|
||||
let _abort_on_drop = AbortOnDrop(aborted.clone());
|
||||
let _abort_on_drop = crate::common::AbortOnDrop(aborted.clone());
|
||||
let unpack_archive = download_archive.clone();
|
||||
tokio::task::spawn_blocking(move || {
|
||||
unpack_repo_archive(&unpack_archive, &download_target, &aborted)
|
||||
|
||||
@@ -67,6 +67,20 @@ mount {
|
||||
#[cfg(not(debug_assertions))]
|
||||
pub const DEV_CONF_NSJAIL: &str = "";
|
||||
|
||||
/// Tells a `spawn_blocking` task to stop when the future awaiting it goes away.
|
||||
///
|
||||
/// Dropping a `JoinHandle` detaches the task rather than cancelling it, so a
|
||||
/// cancelled or timed-out phase otherwise leaves the blocking pool working on an
|
||||
/// answer nobody will read. Hold one of these beside the handle and have the
|
||||
/// blocking loop check the flag.
|
||||
pub(crate) struct AbortOnDrop(pub(crate) std::sync::Arc<std::sync::atomic::AtomicBool>);
|
||||
|
||||
impl Drop for AbortOnDrop {
|
||||
fn drop(&mut self) {
|
||||
self.0.store(true, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// Turn a JSON value into the string a shell/CLI arg should receive: a JSON string
|
||||
/// becomes its inner value, anything else is re-serialized compactly.
|
||||
pub(crate) fn raw_to_string(x: &str) -> String {
|
||||
|
||||
@@ -0,0 +1,589 @@
|
||||
//! 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).
|
||||
//!
|
||||
//! Four 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.
|
||||
//! - **An incremental model has two shapes, and one ingest holds one of them.**
|
||||
//! `is_incremental()` is false when the target does not exist or the build
|
||||
//! is `--full-refresh`, so the `{{ this }}` self-join — and any `ref()` inside
|
||||
//! that branch — compiles only in the other case. What this stores is
|
||||
//! therefore what the compile in front of it saw: at DEPLOY, before the first
|
||||
//! build, that is the cold shape, and a project deployed again after its
|
||||
//! tables exist stores the incremental one for the same source. Nothing here
|
||||
//! can reconcile that; dbt has no mode that emits both. The flag is taken from
|
||||
//! the build so a per-run ingest matches its own run, and the version's graph
|
||||
//! is honest about the compile that produced it rather than about every run
|
||||
//! that will follow.
|
||||
|
||||
use std::collections::HashSet;
|
||||
use std::ops::ControlFlow;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::path::Path;
|
||||
use std::time::Duration;
|
||||
|
||||
use parquet::file::reader::{FileReader, SerializedFileReader};
|
||||
use parquet::record::{Field, Row};
|
||||
use uuid::Uuid;
|
||||
use windmill_common::dbt_manifest::{
|
||||
is_direct, ColumnIndex, IndexedColumn, IngestedColumnEdge, MAX_COLUMN_EDGES,
|
||||
};
|
||||
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.
|
||||
///
|
||||
/// Two steps with deliberately different contracts, because conflating them is
|
||||
/// what made a best-effort annotation able to fail the job it annotates:
|
||||
///
|
||||
/// - [`compile_index`] runs a subprocess and owns the JOB's semantics. Only a
|
||||
/// cancellation or the job's own deadline can `Err` out of it; a non-zero exit
|
||||
/// and an over-long output are outcomes, not failures.
|
||||
/// - [`read_index`] owns the ARTIFACT's semantics. Reading it never fails the
|
||||
/// job on the artifact's account: an absent, unreadable or partial index is a
|
||||
/// value, not an error. It runs UNDER the poller all the same, so the job can
|
||||
/// still end the phase — a cancel, a completion or the phase timeout — which
|
||||
/// is the job's semantics reaching in, not the artifact's reaching out.
|
||||
///
|
||||
/// The phase budget wraps the compile alone, because it exists to leave the
|
||||
/// BUILD its share of the clock and only the compile can spend that share
|
||||
/// unboundedly. The decode's own end is the job's: the poller it runs under
|
||||
/// stops it when the job stops.
|
||||
pub(crate) async fn collect(
|
||||
p: &PreparedProject,
|
||||
descriptor: &DbtDescriptor,
|
||||
inv: &Invocation,
|
||||
// The dbt subcommand the job runs, which decides the effective
|
||||
// `--full-refresh` — see `dbt_executor::full_refresh`.
|
||||
command: &str,
|
||||
ctx: &mut JobCtx<'_>,
|
||||
job_id: &Uuid,
|
||||
w_id: &str,
|
||||
conn: &Connection,
|
||||
kept: &HashSet<&str>,
|
||||
) -> error::Result<Option<ColumnIndex>> {
|
||||
if !descriptor.column_lineage {
|
||||
return Ok(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 Ok(None);
|
||||
}
|
||||
|
||||
let index_dir = p.project_dir.join(CLL_ARTIFACTS_DIR).join("index");
|
||||
let Some(compiled) = compile_index(p, descriptor, inv, command, ctx, job_id, w_id, conn).await? else {
|
||||
return Ok(None);
|
||||
};
|
||||
let coverage = Coverage::of(&compiled);
|
||||
|
||||
// Decoded UNDER the poller, not followed by a check of its own. The decode is
|
||||
// the one phase of this pass with no subprocess behind it, so nothing else
|
||||
// heartbeats while it runs: left alone, a large index is a silent worker for
|
||||
// as long as it takes, which the zombie sweep reads as a dead job and
|
||||
// restarts. The poller pings throughout and ends this with an `Err` if the
|
||||
// job is cancelled or completed meanwhile — the job's own semantics, which
|
||||
// this module may always propagate.
|
||||
let artifact = crate::handle_child::run_future_with_polling_update_job_poller(
|
||||
*job_id,
|
||||
ctx.timeout(),
|
||||
conn,
|
||||
ctx.mem_peak,
|
||||
ctx.canceled_by,
|
||||
async { Ok(read_index(&index_dir, kept).await) },
|
||||
ctx.worker_name,
|
||||
w_id,
|
||||
&mut Some(ctx.occupancy_metrics),
|
||||
Box::pin(futures::stream::empty()),
|
||||
)
|
||||
.await?;
|
||||
|
||||
// What only the pass knows. The COUNTS are logged where the index is folded
|
||||
// into the graph, since the graph decides how much of it is kept.
|
||||
let note = match artifact {
|
||||
Artifact::Read(index) => {
|
||||
if let Some(note) = coverage.caveat() {
|
||||
log(job_id, w_id, note, &compiled.stderr, conn).await;
|
||||
}
|
||||
return Ok(Some(index));
|
||||
}
|
||||
// The truncated arms come first: a compile stopped part-way explains an
|
||||
// absent or unreadable artifact, and blaming the engine's capability
|
||||
// for it sends the reader to check the wrong thing entirely.
|
||||
Artifact::Missing if matches!(coverage, Coverage::Truncated) => format!(
|
||||
"No column lineage: the analysis pass printed more than this runtime reads and was \
|
||||
stopped before it wrote `{COLUMN_LINEAGE_PARQUET}`."
|
||||
),
|
||||
Artifact::Unreadable(why) if matches!(coverage, Coverage::Truncated) => format!(
|
||||
"No column lineage: the analysis pass was stopped for printing more than this \
|
||||
runtime reads, and the `{COLUMN_LINEAGE_PARQUET}` it had written could not be read \
|
||||
({why})."
|
||||
),
|
||||
// Said apart from the one below, because it sends the reader somewhere
|
||||
// else entirely: the engine did its job and this runtime could not read
|
||||
// what it wrote.
|
||||
Artifact::Unreadable(why) => format!(
|
||||
"No column lineage: `{COLUMN_LINEAGE_PARQUET}` was written but could not be read \
|
||||
({why}). The graph is unaffected."
|
||||
),
|
||||
Artifact::Missing => format!(
|
||||
"No column lineage: the analysis pass wrote no `{COLUMN_LINEAGE_PARQUET}`. Only an \
|
||||
engine that computes it does, and only for the warehouses it analyzes natively — \
|
||||
the flag alone is not the capability."
|
||||
),
|
||||
};
|
||||
// The engine's own diagnostics come along. They are how a reader learns that
|
||||
// this adapter turned static analysis off, which it reports as a warning on
|
||||
// a SUCCESSFUL compile that nothing else would show.
|
||||
log(job_id, w_id, ¬e, &compiled.stderr, conn).await;
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
async fn log(job_id: &Uuid, w_id: &str, note: &str, stderr: &str, conn: &Connection) {
|
||||
append_logs(
|
||||
job_id,
|
||||
w_id,
|
||||
format!("\n{note}\n{}", diagnostics(stderr)),
|
||||
conn,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
/// How completely the analysis compile covered the project.
|
||||
///
|
||||
/// Every way the COMPILE can disappoint is a value here rather than an error. An
|
||||
/// `Err` from `compile_index` is the JOB's — a cancellation or its deadline —
|
||||
/// and must fail it; the pass giving up on its own terms is `Ok(None)` and has
|
||||
/// already been logged.
|
||||
enum Coverage {
|
||||
/// Every model analyzed.
|
||||
Whole,
|
||||
/// `--static-analysis strict` rejected part of the project. Whatever it did
|
||||
/// analyze is still in the index.
|
||||
Partial,
|
||||
/// The output ceiling killed the compile mid-run. Distinct from `Partial`:
|
||||
/// nothing rejected the project, but the index is however far it had got, so
|
||||
/// it is not `Whole` either.
|
||||
Truncated,
|
||||
}
|
||||
|
||||
impl Coverage {
|
||||
fn of(c: &crate::dbt_executor::Captured) -> Self {
|
||||
match (c.truncated, c.success) {
|
||||
(true, _) => Coverage::Truncated,
|
||||
(false, true) => Coverage::Whole,
|
||||
(false, false) => Coverage::Partial,
|
||||
}
|
||||
}
|
||||
|
||||
/// What to tell the reader when an index WAS produced. `None` for a run that
|
||||
/// covered everything, which needs no caveat.
|
||||
fn caveat(&self) -> Option<&'static str> {
|
||||
match self {
|
||||
Coverage::Whole => None,
|
||||
Coverage::Partial => Some(
|
||||
"Column lineage: `--static-analysis strict` rejected part of the project, so \
|
||||
the lineage covers only the models it could analyze.",
|
||||
),
|
||||
Coverage::Truncated => Some(
|
||||
"Column lineage: the analysis pass printed more than this runtime reads and was \
|
||||
stopped, so the lineage covers only the models it had reached.",
|
||||
),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Run `dbt compile --static-analysis strict --write-index`, under this phase's
|
||||
/// share of the job's clock.
|
||||
///
|
||||
/// `Ok(None)` is "the pass gave up and said so"; `Err` is the job's own
|
||||
/// cancellation or deadline and must propagate. Nothing outlives this function:
|
||||
/// the budget is a race around the child, and dropping that future kills it
|
||||
/// through `run_captured`'s `kill_on_drop`.
|
||||
async fn compile_index(
|
||||
p: &PreparedProject,
|
||||
descriptor: &DbtDescriptor,
|
||||
inv: &Invocation,
|
||||
command: &str,
|
||||
ctx: &mut JobCtx<'_>,
|
||||
job_id: &Uuid,
|
||||
w_id: &str,
|
||||
conn: &Connection,
|
||||
) -> error::Result<Option<crate::dbt_executor::Captured>> {
|
||||
// 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,
|
||||
],
|
||||
);
|
||||
// The flag already wins over the env var dbt_command sets, but setting both
|
||||
// means this pass cannot write into the runtime's artifacts even if that
|
||||
// precedence ever changes — and what is in there after a build is the
|
||||
// `run_results.json` a `dbt retry` resumes from.
|
||||
cmd.env("DBT_TARGET_PATH", CLL_ARTIFACTS_DIR);
|
||||
crate::dbt_executor::add_vars(&mut cmd, descriptor, inv)?;
|
||||
// The BUILD's answer, not the descriptor's default: `is_incremental()`
|
||||
// branches on it, so a model reading `{{ this }}` compiles its self-join —
|
||||
// and any `ref()` inside that branch — only when this is absent. Guessing
|
||||
// here stores lineage for SQL the run never executed.
|
||||
if crate::dbt_executor::full_refresh(descriptor, inv, command)? {
|
||||
cmd.arg("--full-refresh");
|
||||
}
|
||||
// Captured rather than streamed: a strict-analysis failure is a wall of
|
||||
// diagnostics about SQL the build itself accepts, and this pass decides
|
||||
// nothing about whether that build runs.
|
||||
// Read before the future below borrows `ctx` mutably.
|
||||
let budget = phase_budget(ctx);
|
||||
let run = crate::dbt_executor::run_captured(
|
||||
cmd,
|
||||
"dbt compile (column lineage)",
|
||||
ctx,
|
||||
job_id,
|
||||
w_id,
|
||||
conn,
|
||||
CLL_MAX_OUTPUT_BYTES,
|
||||
// The ceiling is this pass's, not the job's: a compile that prints more
|
||||
// than it than we care to read has still analyzed the project, and the
|
||||
// index it wrote is on disk either way.
|
||||
crate::dbt_executor::Overflow::Truncate,
|
||||
);
|
||||
let Some(budget) = budget else {
|
||||
return Ok(Some(run.await?));
|
||||
};
|
||||
match tokio::time::timeout(budget, run).await {
|
||||
Ok(r) => Ok(Some(r?)),
|
||||
Err(_) => {
|
||||
append_logs(
|
||||
job_id,
|
||||
w_id,
|
||||
format!(
|
||||
"\nNo column lineage: the analysis pass did not finish within {}s, half of \
|
||||
what was left of this job's time. The build below gets the rest.\n",
|
||||
budget.as_secs()
|
||||
),
|
||||
conn,
|
||||
)
|
||||
.await;
|
||||
Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 share of the job's remaining wall clock this pass may spend.
|
||||
///
|
||||
/// A per-run refresh ingests BEFORE the build and shares the job's one deadline,
|
||||
/// so an unbounded pass on a slow project would hand `dbt build` an expired
|
||||
/// budget and fail the run it exists only to annotate. Half leaves the build at
|
||||
/// least as long as the annotation was allowed to take.
|
||||
///
|
||||
/// Spent as a race around the COMPILE rather than as a shortened deadline handed
|
||||
/// to the runner: the runner reports its expiry as an `Err`, indistinguishable
|
||||
/// from a cancellation or the job's own deadline, and those two MUST fail the
|
||||
/// job. Expiring here is this budget and nothing else. The child dies with the
|
||||
/// dropped future through `run_captured`'s `kill_on_drop`; the decode is outside
|
||||
/// this race and answers to the poller instead.
|
||||
fn phase_budget(ctx: &JobCtx<'_>) -> Option<Duration> {
|
||||
ctx.timeout()
|
||||
.map(|left| Duration::from_secs((left.max(0) as u64 / 2).max(1)))
|
||||
}
|
||||
|
||||
/// 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")),
|
||||
}
|
||||
}
|
||||
|
||||
/// What came back from the artifact. Never an `Err`: nothing the file does or
|
||||
/// fails to do is a reason to fail a job. `Unreadable` is separate from
|
||||
/// `Missing` because the two send a reader looking in different places — one at
|
||||
/// their engine and adapter, the other at a file that exists.
|
||||
enum Artifact {
|
||||
Read(ColumnIndex),
|
||||
Missing,
|
||||
Unreadable(String),
|
||||
}
|
||||
|
||||
/// Read both parquets, if the lineage one is there.
|
||||
///
|
||||
/// The column schemas alone are not worth a graph: they arrive with the lineage
|
||||
/// or not at all, and a node's declared columns already answer for the case
|
||||
/// where the pass never ran.
|
||||
async fn read_index(index_dir: &Path, kept: &HashSet<&str>) -> Artifact {
|
||||
let lineage = index_dir.join(COLUMN_LINEAGE_PARQUET);
|
||||
if !tokio::fs::try_exists(&lineage).await.unwrap_or(false) {
|
||||
return Artifact::Missing;
|
||||
}
|
||||
let columns = index_dir.join(NODE_COLUMNS_PARQUET);
|
||||
// Owned, because the decode moves to a blocking thread. The index describes
|
||||
// the whole project while this graph describes one selection of it, so
|
||||
// scoping HERE is what keeps the bound below from being spent on rows the
|
||||
// graph would discard anyway.
|
||||
let kept: HashSet<String> = kept.iter().map(|s| (*s).to_string()).collect();
|
||||
// Dropping the handle of a blocking task does NOT stop it: the poller
|
||||
// cancelling this phase would otherwise leave a thread decoding millions of
|
||||
// rows for a job that is over. `abandoned` is set when this future is
|
||||
// dropped, and the row loop reads it.
|
||||
let abandoned = Arc::new(AtomicBool::new(false));
|
||||
let _stop = crate::common::AbortOnDrop(abandoned.clone());
|
||||
// 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.
|
||||
let read = tokio::task::spawn_blocking(move || {
|
||||
read_index_blocking(&lineage, &columns, &kept, &abandoned)
|
||||
})
|
||||
.await;
|
||||
match read {
|
||||
Ok(Ok(index)) => Artifact::Read(index),
|
||||
Ok(Err(e)) => Artifact::Unreadable(e.to_string()),
|
||||
Err(e) => Artifact::Unreadable(e.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn read_index_blocking(
|
||||
lineage: &Path,
|
||||
columns: &Path,
|
||||
kept: &HashSet<String>,
|
||||
abandoned: &AtomicBool,
|
||||
) -> error::Result<ColumnIndex> {
|
||||
let mut out = ColumnIndex::default();
|
||||
// ONE pass, with the two kinds bucketed as they arrive. `copy` and `mod` say
|
||||
// the value itself travelled, so they get the whole budget; `scan` — the
|
||||
// column was read to produce the ROW, which reaches every output column of
|
||||
// its model and is the bulk of a wide project's index — fills only what is
|
||||
// left over at the end. Reading the file twice to get that ordering would
|
||||
// double the decode of exactly the large index this bound exists for.
|
||||
let mut scan: Vec<IngestedColumnEdge> = Vec::new();
|
||||
for_each_row(lineage, abandoned, |row| {
|
||||
let lineage_kind = string(row, "lineage_kind");
|
||||
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, and neither is one outside this graph's nodes.
|
||||
if parent_column.is_empty()
|
||||
|| child_column.is_empty()
|
||||
|| !kept.contains(&parent_unique_id)
|
||||
|| !kept.contains(&child_unique_id)
|
||||
{
|
||||
return ControlFlow::Continue(());
|
||||
}
|
||||
let edge = IngestedColumnEdge {
|
||||
parent_unique_id,
|
||||
parent_column,
|
||||
child_unique_id,
|
||||
child_column,
|
||||
lineage_kind,
|
||||
};
|
||||
// The bound covers BOTH buckets, so the pass never holds more than one
|
||||
// budget's worth however the kinds are distributed.
|
||||
let held = out.edges.len() + scan.len();
|
||||
if is_direct(&edge.lineage_kind) {
|
||||
// A direct edge displaces a `scan` one: the budget is spent on
|
||||
// value flow first.
|
||||
if held >= MAX_COLUMN_EDGES {
|
||||
scan.pop();
|
||||
}
|
||||
out.edges.push(edge);
|
||||
// The edge that FILLS the budget ends the read, not the next one to
|
||||
// arrive: once the displacing kind is full nothing later in the file
|
||||
// can be kept, and waiting for another direct edge to say so decodes
|
||||
// a `scan`-only tail all the way to the backstop for nothing.
|
||||
return match out.edges.len() >= MAX_COLUMN_EDGES {
|
||||
true => ControlFlow::Break(()),
|
||||
false => ControlFlow::Continue(()),
|
||||
};
|
||||
}
|
||||
if held < MAX_COLUMN_EDGES {
|
||||
scan.push(edge);
|
||||
}
|
||||
// Not a stopping point even when full: a direct edge still to come takes
|
||||
// a `scan` entry's place.
|
||||
ControlFlow::Continue(())
|
||||
})?;
|
||||
out.edges.append(&mut scan);
|
||||
// Absent is normal — an engine can write the lineage table and not this one —
|
||||
// and unreadable is not worth losing the lineage over.
|
||||
let mut held = 0usize;
|
||||
let _ = for_each_row(columns, abandoned, |row| {
|
||||
let unique_id = string(row, "unique_id");
|
||||
let name = string(row, "column_name");
|
||||
if held >= MAX_INDEXED_COLUMNS {
|
||||
return ControlFlow::Break(());
|
||||
}
|
||||
if name.is_empty() || !kept.contains(&unique_id) {
|
||||
return ControlFlow::Continue(());
|
||||
}
|
||||
held += 1;
|
||||
// The author's `data_type` where `schema.yml` gives one, since that is
|
||||
// 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),
|
||||
});
|
||||
ControlFlow::Continue(())
|
||||
});
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
/// The most rows of `dbt.node_columns.parquet` one pass keeps. One per column of
|
||||
/// the project, so the same bound as the edges is far more than any project
|
||||
/// reaches; it exists for the same reason.
|
||||
const MAX_INDEXED_COLUMNS: usize = MAX_COLUMN_EDGES;
|
||||
|
||||
/// The most rows of an index one pass DECODES, whatever it keeps of them.
|
||||
///
|
||||
/// A bound on work rather than on memory, and the two are separate because the
|
||||
/// input this defends against is the one that cannot be collected: `scan`
|
||||
/// lineage is emitted from every predicate and join column to every output
|
||||
/// column, so a project shaped that way writes an index whose row count is
|
||||
/// quadratic in its widest model. This pass runs outside the phase budget, on a
|
||||
/// blocking thread, and nothing the file contains may fail a deploy or a run —
|
||||
/// so the file it walks needs an end even when almost nothing in it is
|
||||
/// retained. The abandonment flag ends it sooner when the job is over; this is
|
||||
/// the bound for a job that is not.
|
||||
const MAX_INDEX_ROWS: usize = 4_000_000;
|
||||
|
||||
/// Decode a parquet a row at a time, handing each to `f` and never holding two.
|
||||
///
|
||||
/// Collecting first would put a `Vec<Row>` — each row carrying its own copy of
|
||||
/// every column NAME — in front of the caller's own bound, which is what would
|
||||
/// take the worker process down on the index described above.
|
||||
///
|
||||
/// `f` says when it has all it will take, and that is the ordinary end: this
|
||||
/// runs outside the phase budget, so every row decoded past the point of being
|
||||
/// able to keep one is wall clock the build below does not get.
|
||||
fn for_each_row(
|
||||
path: &Path,
|
||||
abandoned: &AtomicBool,
|
||||
mut f: impl FnMut(&Row) -> ControlFlow<()>,
|
||||
) -> error::Result<()> {
|
||||
let fail = |e: parquet::errors::ParquetError| {
|
||||
error::Error::internal_err(format!("reading {}: {e}", path.display()))
|
||||
};
|
||||
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(fail)?;
|
||||
for (n, row) in reader.get_row_iter(None).map_err(fail)?.enumerate() {
|
||||
// Nobody is waiting for this any more — the job was cancelled, completed
|
||||
// or ran out of time while it decoded.
|
||||
if abandoned.load(Ordering::Relaxed) {
|
||||
break;
|
||||
}
|
||||
if n >= MAX_INDEX_ROWS {
|
||||
tracing::warn!(
|
||||
"dbt column index: {} holds more than {MAX_INDEX_ROWS} rows; the rest is dropped",
|
||||
path.display()
|
||||
);
|
||||
break;
|
||||
}
|
||||
if f(&row.map_err(fail)?).is_break() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// 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,
|
||||
}
|
||||
}
|
||||
@@ -491,7 +491,7 @@ pub(crate) async fn handle_dbt_job(
|
||||
// For a retry the restored manifest already describes the invocation
|
||||
// being resumed, so only the ingest runs — with that invocation's
|
||||
// arguments, which the selection resolver needs to interpolate.
|
||||
ingest_from_run(&prepared, &descriptor, &inv, &mut ctx, job, conn).await?;
|
||||
ingest_from_run(&prepared, &descriptor, &inv, &command, &mut ctx, job, conn).await?;
|
||||
}
|
||||
|
||||
// A read-only command prints rows to stdout, so it is captured rather than
|
||||
@@ -826,12 +826,26 @@ 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,
|
||||
// A deploy resolves the project by parsing it; nothing is built, so
|
||||
// the pass takes the descriptor's own answer.
|
||||
"parse",
|
||||
&mut ctx,
|
||||
job_id,
|
||||
w_id,
|
||||
&conn,
|
||||
)
|
||||
.await?;
|
||||
let published = persist_ingest(
|
||||
db,
|
||||
w_id,
|
||||
@@ -1040,6 +1054,9 @@ impl GraphRefresh {
|
||||
if selection_is_overridden(descriptor, args)? {
|
||||
self.per_run_models = true;
|
||||
}
|
||||
if full_refresh_is_overridden(descriptor, args)? {
|
||||
self.per_run_models = true;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -2507,8 +2524,7 @@ async fn run_dbt(
|
||||
if let Some(t) = descriptor.threads {
|
||||
cmd.args(["--threads", &t.to_string()]);
|
||||
}
|
||||
let full_refresh = arg_bool(&inv.args, "full_refresh")?.unwrap_or(descriptor.full_refresh);
|
||||
if full_refresh && command != "test" {
|
||||
if full_refresh(descriptor, inv, command)? {
|
||||
cmd.arg("--full-refresh");
|
||||
}
|
||||
}
|
||||
@@ -3160,7 +3176,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
|
||||
@@ -3343,7 +3360,7 @@ 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(),
|
||||
@@ -3363,6 +3380,20 @@ async fn run_parse_only(
|
||||
else {
|
||||
return Ok(to_raw_value(&result));
|
||||
};
|
||||
// AFTER the guard: the pass is a second `dbt compile` and a parquet decode,
|
||||
// and a parse that stores nothing has nowhere to put what it would produce.
|
||||
attach_column_index(
|
||||
&mut ingested,
|
||||
p,
|
||||
descriptor,
|
||||
inv,
|
||||
"parse",
|
||||
ctx,
|
||||
&job.id,
|
||||
&job.workspace_id,
|
||||
conn,
|
||||
)
|
||||
.await?;
|
||||
match conn {
|
||||
Connection::Sql(db) => match job.runnable_id.map(|h| h.0) {
|
||||
Some(script_hash) => {
|
||||
@@ -3422,11 +3453,70 @@ 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,
|
||||
command: &str,
|
||||
ctx: &mut JobCtx<'_>,
|
||||
job_id: &Uuid,
|
||||
w_id: &str,
|
||||
conn: &Connection,
|
||||
) -> error::Result<()> {
|
||||
// The nodes this graph kept, so the pass reads only rows it could store: the
|
||||
// index describes the whole project, this graph one selection of it.
|
||||
let kept: std::collections::HashSet<&str> = ingested
|
||||
.nodes
|
||||
.iter()
|
||||
.map(|n| n.unique_id.as_str())
|
||||
.collect();
|
||||
let index =
|
||||
crate::dbt_column_index::collect(
|
||||
p, descriptor, inv, command, ctx, job_id, w_id, conn, &kept,
|
||||
)
|
||||
.await?;
|
||||
drop(kept);
|
||||
let Some(index) = index else {
|
||||
return Ok(());
|
||||
};
|
||||
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;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Refresh the stored graph from the manifest this run produced.
|
||||
async fn ingest_from_run(
|
||||
p: &PreparedProject,
|
||||
descriptor: &DbtDescriptor,
|
||||
inv: &Invocation,
|
||||
command: &str,
|
||||
ctx: &mut JobCtx<'_>,
|
||||
job: &MiniPulledJob,
|
||||
conn: &Connection,
|
||||
@@ -3443,12 +3533,24 @@ 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,
|
||||
command,
|
||||
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
|
||||
@@ -3773,7 +3875,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();
|
||||
@@ -3826,6 +3930,34 @@ async fn resolve_selection(
|
||||
/// what is kept is the TAIL, because dbt prints its error summary last.
|
||||
const CAPTURE_MAX_STDERR_BYTES: usize = 64 * 1024;
|
||||
|
||||
/// 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,
|
||||
/// Whether the child exited zero. Separate from the `Result` on purpose: an
|
||||
/// `Err` from `run_captured` is the JOB's — a cancellation or its deadline —
|
||||
/// so a caller that tolerates a failed command must still propagate one.
|
||||
pub success: bool,
|
||||
/// Whether the output ceiling cut the child short. Only ever true under
|
||||
/// [`Overflow::Truncate`].
|
||||
pub truncated: bool,
|
||||
}
|
||||
|
||||
/// What an over-long stdout means to the caller.
|
||||
///
|
||||
/// The ceiling belongs to the PASS, not to the job: a caller that only annotates
|
||||
/// a job wants to keep what it read and carry on, while one whose whole result
|
||||
/// is that output has nothing to return without it.
|
||||
#[derive(PartialEq, Eq, Clone, Copy)]
|
||||
pub(crate) enum Overflow {
|
||||
/// Fail the job. For a command whose output IS the answer.
|
||||
Fail,
|
||||
/// Stop reading, kill the child, and report `truncated`.
|
||||
Truncate,
|
||||
}
|
||||
|
||||
/// Run a command for its stdout under the job's cancellation and timeout.
|
||||
/// The same poller `handle_child` uses drives them, so a cancel or a deadline
|
||||
/// drops the wait future — which owns the child, and `kill_on_drop` then
|
||||
@@ -3838,7 +3970,7 @@ 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(
|
||||
pub(crate) async fn run_captured(
|
||||
mut cmd: Command,
|
||||
name: &str,
|
||||
ctx: &mut JobCtx<'_>,
|
||||
@@ -3846,7 +3978,8 @@ async fn run_capturing(
|
||||
w_id: &str,
|
||||
conn: &Connection,
|
||||
max_stdout_bytes: usize,
|
||||
) -> error::Result<String> {
|
||||
on_overflow: Overflow,
|
||||
) -> error::Result<Captured> {
|
||||
use tokio::io::AsyncReadExt;
|
||||
|
||||
let mut child = cmd
|
||||
@@ -3881,6 +4014,7 @@ async fn run_capturing(
|
||||
let mut out_buf = vec![0u8; 16 * 1024];
|
||||
let mut err_buf = vec![0u8; 16 * 1024];
|
||||
let (mut out_open, mut err_open) = (true, true);
|
||||
let mut truncated = false;
|
||||
while out_open || err_open {
|
||||
tokio::select! {
|
||||
r = stdout_pipe.read(&mut out_buf[..]), if out_open => match r {
|
||||
@@ -3888,14 +4022,19 @@ async fn run_capturing(
|
||||
Ok(n) => {
|
||||
if stdout.len() + n > max_stdout_bytes {
|
||||
// Killed here rather than left to `kill_on_drop`
|
||||
// so the child is gone before the error unwinds,
|
||||
// not merely once this future is dropped.
|
||||
// so the child is gone before this returns, not
|
||||
// merely once the future is dropped.
|
||||
let _ = child.kill().await;
|
||||
return Err(Error::ExecutionErr(format!(
|
||||
"{name} produced more than {} MB of output. Narrow the \
|
||||
selection, or query the relation from a SQL script.",
|
||||
max_stdout_bytes / 1024 / 1024
|
||||
)));
|
||||
if on_overflow == Overflow::Fail {
|
||||
return Err(Error::ExecutionErr(format!(
|
||||
"{name} produced more than {} MB of output. Narrow the \
|
||||
selection, or query the relation from a SQL script.",
|
||||
max_stdout_bytes / 1024 / 1024
|
||||
)));
|
||||
}
|
||||
truncated = true;
|
||||
out_open = false;
|
||||
continue;
|
||||
}
|
||||
stdout.extend_from_slice(&out_buf[..n]);
|
||||
}
|
||||
@@ -3918,7 +4057,7 @@ async fn run_capturing(
|
||||
.wait()
|
||||
.await
|
||||
.map_err(|e| Error::internal_err(format!("{name} failed: {e}")))?;
|
||||
Ok((status, stdout, stderr))
|
||||
Ok((status, stdout, stderr, truncated))
|
||||
},
|
||||
ctx.worker_name,
|
||||
w_id,
|
||||
@@ -3928,14 +4067,46 @@ async fn run_capturing(
|
||||
})),
|
||||
)
|
||||
.await?;
|
||||
let (status, stdout, stderr) = out;
|
||||
if !status.success() {
|
||||
let (status, stdout, stderr, truncated) = out;
|
||||
Ok(Captured {
|
||||
stdout: String::from_utf8_lossy(&stdout).to_string(),
|
||||
stderr: String::from_utf8_lossy(&stderr).to_string(),
|
||||
// A killed child reports failure; under `Truncate` that is the ceiling's
|
||||
// doing, not the project's, and the caller reads `truncated` to tell.
|
||||
success: status.success(),
|
||||
truncated,
|
||||
})
|
||||
}
|
||||
|
||||
/// `run_captured`, with a non-zero exit folded into the error — what a caller
|
||||
/// that needs the command to have WORKED wants.
|
||||
pub(crate) async fn run_capturing(
|
||||
cmd: Command,
|
||||
name: &str,
|
||||
ctx: &mut JobCtx<'_>,
|
||||
job_id: &Uuid,
|
||||
w_id: &str,
|
||||
conn: &Connection,
|
||||
max_stdout_bytes: usize,
|
||||
) -> error::Result<Captured> {
|
||||
let captured = run_captured(
|
||||
cmd,
|
||||
name,
|
||||
ctx,
|
||||
job_id,
|
||||
w_id,
|
||||
conn,
|
||||
max_stdout_bytes,
|
||||
Overflow::Fail,
|
||||
)
|
||||
.await?;
|
||||
if !captured.success {
|
||||
return Err(Error::ExecutionErr(format!(
|
||||
"{name} failed: {}",
|
||||
String::from_utf8_lossy(&stderr)
|
||||
captured.stderr
|
||||
)));
|
||||
}
|
||||
Ok(String::from_utf8_lossy(&stdout).to_string())
|
||||
Ok(captured)
|
||||
}
|
||||
|
||||
/// Run a preparation command through the same child handler the build uses, so
|
||||
@@ -4888,7 +5059,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()]);
|
||||
@@ -5243,6 +5418,41 @@ fn selection_is_overridden(
|
||||
Ok(differs("select", &descriptor.select)? || differs("exclude", &descriptor.exclude)?)
|
||||
}
|
||||
|
||||
/// Whether this invocation rebuilds incremental models from scratch: the run
|
||||
/// form's answer when it gave one, else the descriptor's — and never for a
|
||||
/// `test`, which builds nothing whatever the form said.
|
||||
///
|
||||
/// Shared with the column-lineage pass rather than recomputed there, because
|
||||
/// `is_incremental()` branches on it: the same model compiles to different SQL —
|
||||
/// a `{{ this }}` self-join, and any `ref()` inside the incremental branch — so a
|
||||
/// pass that guessed would describe a build that never ran.
|
||||
///
|
||||
/// `test` returns false because `dbt test` rejects `--full-refresh` outright.
|
||||
/// It never arrives as a caller's `dbt_command` — the allowlist has no such
|
||||
/// value — so reading only that allowlist suggests this branch is dead. It is
|
||||
/// not: `run_dbt` is invoked with `"test"` directly for the `after_all` test
|
||||
/// phase, and an `after_all` project with `full_refresh: true` reaches here.
|
||||
pub(crate) fn full_refresh(
|
||||
descriptor: &DbtDescriptor,
|
||||
inv: &Invocation,
|
||||
command: &str,
|
||||
) -> error::Result<bool> {
|
||||
if command == "test" {
|
||||
return Ok(false);
|
||||
}
|
||||
Ok(arg_bool(&inv.args, "full_refresh")?.unwrap_or(descriptor.full_refresh))
|
||||
}
|
||||
|
||||
/// Whether this run answered `full_refresh` differently from the deployed
|
||||
/// descriptor. Like a selection override it changes what the graph describes,
|
||||
/// since an incremental branch can carry its own `ref()`.
|
||||
fn full_refresh_is_overridden(
|
||||
descriptor: &DbtDescriptor,
|
||||
args: &HashMap<String, Box<RawValue>>,
|
||||
) -> error::Result<bool> {
|
||||
Ok(arg_bool(args, "full_refresh")?.is_some_and(|v| v != descriptor.full_refresh))
|
||||
}
|
||||
|
||||
/// The descriptor's named selector, unless this run named its own selection.
|
||||
///
|
||||
/// dbt resolves `--selector` INSTEAD of `--select`, so passing both makes the
|
||||
@@ -6254,6 +6464,58 @@ mod tests {
|
||||
"an overridden selection must not publish ownership"
|
||||
);
|
||||
}
|
||||
|
||||
// `full_refresh` decides whether `is_incremental()` is true, so an
|
||||
// incremental model's self-join — and any `ref()` inside that branch —
|
||||
// exists in one answer and not the other. A run that flips it describes
|
||||
// a different graph, and gets its own.
|
||||
let mut refreshed = GraphRefresh::default();
|
||||
refreshed
|
||||
.add_caller_args(&descriptor, &arg("full_refresh", "true"))
|
||||
.unwrap();
|
||||
assert!(refreshed.needed());
|
||||
assert_eq!(refreshed.snapshot_job(job), Some(job));
|
||||
|
||||
// The same echo rule: the form posts the descriptor's own value back on
|
||||
// every run, and reading that as an override would make each one
|
||||
// caller-scoped.
|
||||
let always = DbtDescriptor { full_refresh: true, ..Default::default() };
|
||||
let mut echoed_flag = GraphRefresh { profile_drift: true, ..Default::default() };
|
||||
echoed_flag
|
||||
.add_caller_args(&always, &arg("full_refresh", "true"))
|
||||
.unwrap();
|
||||
assert_eq!(echoed_flag.snapshot_job(job), None);
|
||||
}
|
||||
|
||||
/// The build and the analysis pass read this through one function, so they
|
||||
/// cannot disagree about which SQL the run compiles — including for `test`,
|
||||
/// which rebuilds nothing whatever the descriptor or the form said.
|
||||
#[test]
|
||||
fn full_refresh_is_one_answer_for_the_build_and_the_pass() {
|
||||
let inv = |args: HashMap<String, Box<RawValue>>| Invocation {
|
||||
args,
|
||||
raw_args: Default::default(),
|
||||
envs: Default::default(),
|
||||
strict: true,
|
||||
deferral: None,
|
||||
};
|
||||
let always = DbtDescriptor { full_refresh: true, ..Default::default() };
|
||||
let never = DbtDescriptor::default();
|
||||
let on = HashMap::from([(
|
||||
"full_refresh".to_string(),
|
||||
RawValue::from_string("true".to_string()).unwrap(),
|
||||
)]);
|
||||
|
||||
assert!(full_refresh(&always, &inv(Default::default()), "build").unwrap());
|
||||
assert!(!full_refresh(&never, &inv(Default::default()), "build").unwrap());
|
||||
assert!(
|
||||
full_refresh(&never, &inv(on), "build").unwrap(),
|
||||
"the form's answer wins over the descriptor's"
|
||||
);
|
||||
assert!(
|
||||
!full_refresh(&always, &inv(Default::default()), "test").unwrap(),
|
||||
"a test builds nothing, so neither the build nor the pass may pass the flag"
|
||||
);
|
||||
}
|
||||
|
||||
// `dbt retry` restores the previous run's target/ from this directory, so two
|
||||
|
||||
@@ -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;
|
||||
|
||||
+115
-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; the whole manifest is kept once per environment, for deferral — 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` |
|
||||
@@ -943,6 +943,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
|
||||
@@ -1567,11 +1569,115 @@ answers for a machine's history rather than for the environment. So exactly one
|
||||
manifest is kept per (script, environment), replaced by each successful run,
|
||||
rather than one per version (see "Durable state per environment" above).
|
||||
|
||||
**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 is best-effort about everything that is ITS: a wrong engine, a rejected
|
||||
analysis, a missing or unreadable artifact, an over-long output and outrunning its
|
||||
own time budget all degrade to partial lineage or none, plus a line in the job
|
||||
log saying which. It is not best-effort about the JOB: a cancellation or the job's
|
||||
own deadline fail it, because swallowing those would let a run that blew its
|
||||
timeout inside an optional annotation publish a graph and report success. That
|
||||
split is why the two halves have separate error contracts — the compile owns the
|
||||
job's semantics and may `Err`; nothing the artifact does or fails to do is a
|
||||
reason to fail a job, so an absent, unreadable or partial index is a value. The
|
||||
decode still runs under the job poller, which both heartbeats through it and
|
||||
ends it if the job is cancelled or completed meanwhile: the job reaching in, not
|
||||
the artifact reaching out. The
|
||||
budget is half the job's remaining wall clock, spent on the compile alone, so the
|
||||
build that follows cannot be starved by it.
|
||||
|
||||
**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. 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 `copy`/`mod` are kept
|
||||
first when the bound bites: a `scan` edge reaches every output column of its
|
||||
model, so it is most of what a project's index holds and would draw as a complete
|
||||
bipartite graph. Keeping it in the table anyway is what lets a later "show
|
||||
indirect" view ask for it without every project being redeployed.
|
||||
|
||||
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.
|
||||
|
||||
**A table of its own, not `dbt_edge.column_lineage` JSONB.** Hanging the links on
|
||||
the `ref()` edge they sit beneath would inherit its clone, prune, clear and
|
||||
cascade paths for free, and it does not work: a model reading `{{ this }}` gets
|
||||
column lineage from itself to itself, and `parent_map` has no self-loop, because
|
||||
a model does not `ref()` itself. Those pairs have no `dbt_edge` row to attach to.
|
||||
The loss is not hypothetical — an incremental that selects from `{{ this }}`
|
||||
(`coalesce(p.dbl, s.dbl)`, `p.up as prev_up`) yields `up → prev_up` with kind
|
||||
`copy`, a drawn edge meaning "this column carries the previous run's value".
|
||||
Inventing self-loop `dbt_edge` rows to hold it is not an option either: that
|
||||
table is `ref()` lineage. The typed column list lands in
|
||||
`dbt_node.column_schema`, beside `columns` rather than merged into it —
|
||||
`columns` stays what the author *declared*.
|
||||
|
||||
**Stored now, served later.** This change lands the ingest and the storage; the
|
||||
endpoint that draws a column trace is a follow-up. What is user-visible today is
|
||||
`column_schema` — every column of a relation, typed and in the order the model
|
||||
emits them — which rides the asset graph the details pane already fetches, and
|
||||
replaces a panel that could only list the columns an author happened to document.
|
||||
The edges sit in `dbt_column_edge` waiting for their surface.
|
||||
|
||||
`column_schema` is gated on being able to read the producing project, like the
|
||||
model's SQL: a column-level view is the shape of what the author wrote, one level
|
||||
finer than the `ref()` graph, which is ungated only because it draws relations the
|
||||
caller already sees. A share-link viewer entitled to a dbt run therefore gets its
|
||||
relations and `ref()` edges, and neither the SQL nor the columns.
|
||||
|
||||
**The analysis pass takes the build's own `--full-refresh`.** `is_incremental()`
|
||||
branches on it, so an incremental model reading `{{ this }}` compiles its
|
||||
self-join — and any `ref()` inside that branch — only when the flag is absent. A
|
||||
pass that used the descriptor's default while the run overrode it would store
|
||||
lineage for SQL that run never executed. For the same reason an invocation that
|
||||
overrides the flag counts as `per_run_models`: its graph is its own, keyed to the
|
||||
job, rather than standing as the version's.
|
||||
|
||||
That flag is not the whole of it, and the rest is a property rather than a bug to
|
||||
fix. `is_incremental()` is also false when the target table does not exist, so an
|
||||
incremental model has **two shapes and one ingest holds one of them**: a deploy
|
||||
before the first build compiles the cold shape, and the same project deployed
|
||||
again once its tables exist compiles the incremental one. A static descriptor
|
||||
re-ingests on neither runs nor time, so what is stored stays whatever the compile
|
||||
in front of it saw. dbt has no mode that emits both, and re-analyzing per run
|
||||
would buy a second `dbt compile` on every build to keep a graph nobody asked to
|
||||
refresh. The contract is therefore the honest one: a version's graph describes
|
||||
the compile that produced it, and a run that re-ingests describes its own run.
|
||||
|
||||
## Concept mapping
|
||||
|
||||
@@ -1583,7 +1689,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 | `dbt_column_edge` rows (no view yet) | `dbt.column_lineage.parquet`, opt-in |
|
||||
| model `tags` | node badge | `tag` |
|
||||
| source freshness | `freshness` | `last_success_at` chip |
|
||||
| `run_results.json` | materialization records | `record_materialization` |
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -111,7 +111,24 @@
|
||||
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)
|
||||
// `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.
|
||||
@@ -202,15 +219,15 @@
|
||||
|
||||
{#if stalePlaceholders}
|
||||
<div class="shrink-0 px-2 py-1 border-b text-2xs text-secondary bg-surface-secondary">
|
||||
The run arguments have changed since this graph was parsed, so these rows need not
|
||||
describe the models on screen — arguments reach schemas, aliases and which models exist
|
||||
at all. Refresh the models to draw and preview them under the current ones.
|
||||
The run arguments have changed since this graph was parsed, so these rows need not describe
|
||||
the models on screen — arguments reach schemas, aliases and which models exist at all. Refresh
|
||||
the models to draw and preview them under the current ones.
|
||||
</div>
|
||||
{:else if staleVars}
|
||||
<div class="shrink-0 px-2 py-1 border-b text-2xs text-secondary bg-surface-secondary">
|
||||
The run form's vars have changed since this graph was parsed. Rows are previewed under
|
||||
the vars it was parsed with, so they still describe the models on screen — refresh the
|
||||
models to draw and preview them under the current ones.
|
||||
The run form's vars have changed since this graph was parsed. Rows are previewed under the
|
||||
vars it was parsed with, so they still describe the models on screen — refresh the models to
|
||||
draw and preview them under the current ones.
|
||||
</div>
|
||||
{/if}
|
||||
|
||||
@@ -238,20 +255,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, typed and in the order the model produces it.
|
||||
</div>
|
||||
{/if}
|
||||
</div>
|
||||
{/if}
|
||||
{#if (dbt.data_tests?.length ?? 0) > 0}
|
||||
|
||||
@@ -1386,6 +1386,12 @@ defer: false
|
||||
# resolved to that Windmill variable, so secrets stay out of this file.
|
||||
# env:
|
||||
# DBT_PASSWORD: $var:u/user/my_warehouse_password
|
||||
# Real column schemas — every column typed and in the order the model produces
|
||||
# it — from the engine's static analysis, which also records column-level
|
||||
# lineage for a later view. 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 = {
|
||||
|
||||
Reference in New Issue
Block a user