diff --git a/backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json b/backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json new file mode 100644 index 0000000000..1268656e34 --- /dev/null +++ b/backend/.sqlx/query-3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3.json @@ -0,0 +1,21 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n manifest, manifest_key, run_results,\n run_results_key, updated_at)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now())\n ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET\n job_id = EXCLUDED.job_id, manifest = EXCLUDED.manifest,\n manifest_key = EXCLUDED.manifest_key, run_results = EXCLUDED.run_results,\n run_results_key = EXCLUDED.run_results_key, updated_at = now()", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Text", + "Uuid", + "Text", + "Text", + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "3fc12582cbae2ebc77ecfcaeed4bce43c749a44cbc4e7b8f719bf2a598ce57f3" +} diff --git a/backend/.sqlx/query-446909206f289fc3f9201a6ad025cc73eef1e3d01825031ba263881b7cfd5ed4.json b/backend/.sqlx/query-446909206f289fc3f9201a6ad025cc73eef1e3d01825031ba263881b7cfd5ed4.json new file mode 100644 index 0000000000..7f950346ee --- /dev/null +++ b/backend/.sqlx/query-446909206f289fc3f9201a6ad025cc73eef1e3d01825031ba263881b7cfd5ed4.json @@ -0,0 +1,14 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM script WHERE workspace_id = $1 AND hash = 1", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [] + }, + "hash": "446909206f289fc3f9201a6ad025cc73eef1e3d01825031ba263881b7cfd5ed4" +} diff --git a/backend/.sqlx/query-4ce90ff818e5058a7e31da24abd938eb135d6eb9714ca115359221d7c02861f0.json b/backend/.sqlx/query-4ce90ff818e5058a7e31da24abd938eb135d6eb9714ca115359221d7c02861f0.json new file mode 100644 index 0000000000..913c16c31d --- /dev/null +++ b/backend/.sqlx/query-4ce90ff818e5058a7e31da24abd938eb135d6eb9714ca115359221d7c02861f0.json @@ -0,0 +1,17 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n manifest)\n SELECT $1::varchar, $2::varchar, 'main||analytics|wh'::text, $3::uuid, '{}'::text\n WHERE EXISTS (SELECT 1 FROM script\n WHERE workspace_id = $1 AND path = $2\n AND deleted = false AND archived = false\n AND language = 'dbt'\n AND (hash = $4 OR $4 = ANY(parent_hashes)))\n ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET job_id = EXCLUDED.job_id", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Uuid", + "Int8" + ] + }, + "nullable": [] + }, + "hash": "4ce90ff818e5058a7e31da24abd938eb135d6eb9714ca115359221d7c02861f0" +} diff --git a/backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json b/backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json new file mode 100644 index 0000000000..b2a0d728cd --- /dev/null +++ b/backend/.sqlx/query-58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT 1 FROM script\n WHERE workspace_id = $1 AND path = $2\n AND deleted = false AND archived = false AND language = 'dbt'\n AND (hash = $3 OR $3 = ANY(parent_hashes))\n FOR SHARE", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "58ec340c78af046a40296b15543beeedc63b4a941e5b17ab7bb5d3c259f05147" +} diff --git a/backend/.sqlx/query-5c7260785ebcece2ddb04dc85d09b6a36c50d70e20c5a10310f75a39930db521.json b/backend/.sqlx/query-5c7260785ebcece2ddb04dc85d09b6a36c50d70e20c5a10310f75a39930db521.json new file mode 100644 index 0000000000..09efdc7340 --- /dev/null +++ b/backend/.sqlx/query-5c7260785ebcece2ddb04dc85d09b6a36c50d70e20c5a10310f75a39930db521.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id,\n manifest)\n VALUES ($1, $2, 'main||analytics|wh', $3, '{}')", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Varchar", + "Varchar", + "Uuid" + ] + }, + "nullable": [] + }, + "hash": "5c7260785ebcece2ddb04dc85d09b6a36c50d70e20c5a10310f75a39930db521" +} diff --git a/backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json b/backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json new file mode 100644 index 0000000000..c5740cfb88 --- /dev/null +++ b/backend/.sqlx/query-7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba.json @@ -0,0 +1,16 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE dbt_environment_state SET script_path = $3\n WHERE workspace_id = $1 AND script_path = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text", + "Varchar" + ] + }, + "nullable": [] + }, + "hash": "7f421bdf3dc4d47b36457af703ce69ef1e0784e9f25ada27e1b0f5cc0944e9ba" +} diff --git a/backend/.sqlx/query-8165a447f458d62f7bafb9827d87d192f5e926b932c3dd0e1bef4822cf722630.json b/backend/.sqlx/query-8165a447f458d62f7bafb9827d87d192f5e926b932c3dd0e1bef4822cf722630.json new file mode 100644 index 0000000000..f4bd63c718 --- /dev/null +++ b/backend/.sqlx/query-8165a447f458d62f7bafb9827d87d192f5e926b932c3dd0e1bef4822cf722630.json @@ -0,0 +1,48 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT job_id, manifest, manifest_key, run_results, run_results_key\n FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "manifest", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "manifest_key", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "run_results", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "run_results_key", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true + ] + }, + "hash": "8165a447f458d62f7bafb9827d87d192f5e926b932c3dd0e1bef4822cf722630" +} diff --git a/backend/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json b/backend/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json new file mode 100644 index 0000000000..909e6ad42d --- /dev/null +++ b/backend/.sqlx/query-a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT pg_advisory_xact_lock($1)", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "pg_advisory_xact_lock", + "type_info": "Void" + } + ], + "parameters": { + "Left": [ + "Int8" + ] + }, + "nullable": [ + null + ] + }, + "hash": "a06e1d9f6f95e4c4c2b98310ebddcc9d963cc033582bf2e945e8bf3a301b4247" +} diff --git a/backend/.sqlx/query-b820bf7b0a93476fb7762e8ff2819b2c2b7df2788c9eb0115abc60ea7b733d64.json b/backend/.sqlx/query-b820bf7b0a93476fb7762e8ff2819b2c2b7df2788c9eb0115abc60ea7b733d64.json new file mode 100644 index 0000000000..7476ae9871 --- /dev/null +++ b/backend/.sqlx/query-b820bf7b0a93476fb7762e8ff2819b2c2b7df2788c9eb0115abc60ea7b733d64.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2\n AND NOT EXISTS (SELECT 1 FROM script\n WHERE workspace_id = $1 AND path = $2\n AND deleted = false AND archived = false)", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "b820bf7b0a93476fb7762e8ff2819b2c2b7df2788c9eb0115abc60ea7b733d64" +} diff --git a/backend/.sqlx/query-d2c02f1c7f4555fa849f5cd86169eb8e43d384084b68dc1bbb3b961206ff4b6c.json b/backend/.sqlx/query-d2c02f1c7f4555fa849f5cd86169eb8e43d384084b68dc1bbb3b961206ff4b6c.json new file mode 100644 index 0000000000..9c75f7ac97 --- /dev/null +++ b/backend/.sqlx/query-d2c02f1c7f4555fa849f5cd86169eb8e43d384084b68dc1bbb3b961206ff4b6c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [] + }, + "hash": "d2c02f1c7f4555fa849f5cd86169eb8e43d384084b68dc1bbb3b961206ff4b6c" +} diff --git a/backend/.sqlx/query-d8b7241518ce0822288fb123b48205427cc50af031a84d334f65f870d86302ac.json b/backend/.sqlx/query-d8b7241518ce0822288fb123b48205427cc50af031a84d334f65f870d86302ac.json new file mode 100644 index 0000000000..748ca6d9ec --- /dev/null +++ b/backend/.sqlx/query-d8b7241518ce0822288fb123b48205427cc50af031a84d334f65f870d86302ac.json @@ -0,0 +1,48 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT job_id, manifest, manifest_key, run_results, run_results_key\n FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "job_id", + "type_info": "Uuid" + }, + { + "ordinal": 1, + "name": "manifest", + "type_info": "Text" + }, + { + "ordinal": 2, + "name": "manifest_key", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "run_results", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "run_results_key", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true + ] + }, + "hash": "d8b7241518ce0822288fb123b48205427cc50af031a84d334f65f870d86302ac" +} diff --git a/backend/.sqlx/query-f5e061eed38d483980ee3691aea821dde2c093bb926b77ec3f99a1c0d9283a14.json b/backend/.sqlx/query-f5e061eed38d483980ee3691aea821dde2c093bb926b77ec3f99a1c0d9283a14.json new file mode 100644 index 0000000000..1c79a53ab0 --- /dev/null +++ b/backend/.sqlx/query-f5e061eed38d483980ee3691aea821dde2c093bb926b77ec3f99a1c0d9283a14.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT count(*) FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "count", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "f5e061eed38d483980ee3691aea821dde2c093bb926b77ec3f99a1c0d9283a14" +} diff --git a/backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json b/backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json new file mode 100644 index 0000000000..540d2e0e08 --- /dev/null +++ b/backend/.sqlx/query-f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT manifest_key, run_results_key FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "manifest_key", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "run_results_key", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "f963ea31d1744ff0d76ff86140f01cca0bcbc832f01eb3340050504b8b3875a2" +} diff --git a/backend/.sqlx/query-ffde6e45841090440bc99e22788d313db5945faa0b93c127833e44d38665a392.json b/backend/.sqlx/query-ffde6e45841090440bc99e22788d313db5945faa0b93c127833e44d38665a392.json new file mode 100644 index 0000000000..c1eb5fa826 --- /dev/null +++ b/backend/.sqlx/query-ffde6e45841090440bc99e22788d313db5945faa0b93c127833e44d38665a392.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT manifest_key, run_results_key FROM dbt_environment_state\n WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "manifest_key", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "run_results_key", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text", + "Text", + "Text" + ] + }, + "nullable": [ + true, + true + ] + }, + "hash": "ffde6e45841090440bc99e22788d313db5945faa0b93c127833e44d38665a392" +} diff --git a/backend/migrations/20260904135713_dbt_environment_state.down.sql b/backend/migrations/20260904135713_dbt_environment_state.down.sql new file mode 100644 index 0000000000..dab9025417 --- /dev/null +++ b/backend/migrations/20260904135713_dbt_environment_state.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS dbt_environment_state; diff --git a/backend/migrations/20260904135713_dbt_environment_state.up.sql b/backend/migrations/20260904135713_dbt_environment_state.up.sql new file mode 100644 index 0000000000..a49df5b3d1 --- /dev/null +++ b/backend/migrations/20260904135713_dbt_environment_state.up.sql @@ -0,0 +1,52 @@ +-- The dbt state one project last built into one environment: the `manifest.json` +-- (and the `run_results.json` beside it) that `dbt --defer --state ` resolves +-- an unbuilt `ref()` through. +-- +-- Separate from `dbt_run_state`, which answers a different question. That one is +-- keyed by the executing principal and holds the LAST run whatever its outcome, +-- so `dbt retry` can resume its failures; this one is keyed by environment and +-- holds the last SUCCESSFUL run, because a relation a later run defers to has to +-- exist. Merging them would make a retry resume a run that is not the last one, +-- or a deferral point at relations a failed run never wrote. +CREATE TABLE IF NOT EXISTS dbt_environment_state ( + workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE, + script_path VARCHAR(255) NOT NULL, + -- The workspace warehouse, the dbt target, and the database and schema that + -- target resolves to. All four, because deferring is resolving a relation + -- NAME: a repointed warehouse or a moved schema makes the stored manifest + -- describe relations that are not where this run would look for them, and the + -- run has no other way to notice. A move therefore reads as an environment + -- with no state yet rather than as state that silently no longer fits. + -- + -- TEXT rather than VARCHAR(255): a project bringing its own `profiles.yml` + -- spells its own schema and database, so the length is the project's. + environment TEXT NOT NULL, + -- The run that published it, so a deferring run can say what it deferred to. + job_id UUID NOT NULL, + -- Exactly one home each. A manifest grows with the project and passes a few + -- hundred KB on a handful of models, so a large one goes to the INSTANCE's + -- object storage and this row keeps the key; a small one stays here, where it + -- costs no round trip and works on an instance that has configured no storage + -- at all. The instance's and not the workspace's, because a member can write + -- the workspace bucket under a key of their choosing, and a manifest is what a + -- later run resolves every unbuilt `ref()` through. `run_results.json` is a + -- tenth of the size and takes the same two homes rather than a rule of its own. + manifest TEXT, + manifest_key TEXT, + run_results TEXT, + run_results_key TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (workspace_id, script_path, environment), + CONSTRAINT dbt_environment_state_manifest_one_home + CHECK (num_nonnulls(manifest, manifest_key) = 1), + CONSTRAINT dbt_environment_state_run_results_one_home + CHECK (num_nonnulls(run_results, run_results_key) <= 1) +); + +-- No age sweep, unlike the per-run graph rows next door: this table holds one +-- row per script per environment and replaces it in place, so it does not grow +-- with runs, and its reader is every later run of that script — a project that +-- runs monthly must still find last month's state. It goes with the script +-- instead, alongside `dbt_run_state`. +GRANT ALL ON dbt_environment_state TO windmill_user; +GRANT ALL ON dbt_environment_state TO windmill_admin; diff --git a/backend/parsers/windmill-parser-yaml/src/dbt.rs b/backend/parsers/windmill-parser-yaml/src/dbt.rs index 24f1e2c39d..980b547c68 100644 --- a/backend/parsers/windmill-parser-yaml/src/dbt.rs +++ b/backend/parsers/windmill-parser-yaml/src/dbt.rs @@ -58,6 +58,18 @@ impl DbtEngine { pub fn emits_node_events(&self) -> bool { matches!(self, DbtEngine::DbtCore1x) } + + /// Whether the engine has `--defer-state`, the deferral-only half of + /// `--state`. + /// + /// It matters on one command. `dbt retry` reads the run it resumes from + /// `--state`, so an engine with only that flag cannot be told to defer and + /// to resume from the job's own results at once: handed the deferral's + /// directory, it resumes the all-green run stored there and rebuilds + /// nothing. Only dbt-core 1.x separates the two. + pub fn has_defer_state_flag(&self) -> bool { + matches!(self, DbtEngine::DbtCore1x) + } } /// How the warehouse connection is supplied. Both paths are supported @@ -137,6 +149,16 @@ pub struct DbtDescriptor { pub threads: Option, #[serde(default)] pub full_refresh: bool, + /// Resolve a `ref()` a run does not build through the state the last + /// successful run of this environment published, rather than through the + /// schema that run writes into. + /// + /// Only the default for the `build` block's own `defer`, since the choice is + /// per run: the run that publishes an environment's state and the run that + /// defers to it are two invocations of ONE script (decision 6), so a project + /// that could only defer by descriptor could never populate what it reads. + #[serde(default)] + pub defer: bool, /// Automatic in-job retry of the nodes a build failed on. /// /// dbt already confines a failure to its own subtree, and `dbt retry` @@ -258,6 +280,7 @@ pub const RESERVED_ARG_NAMES: &[&str] = &[ "exclude", "vars", "full_refresh", + "defer", "dbt_command", "dbt_retry_job", "model", @@ -345,15 +368,26 @@ fn command_variants(d: &DbtDescriptor) -> Vec<(&'static str, Vec)> { "build", selection() .into_iter() - .chain([Arg { - name: "full_refresh".to_string(), - otyp: None, - typ: Typ::Bool, - has_default: true, - default: Some(serde_json::json!(d.full_refresh)), - oidx: None, - otyp_inferred: false, - }]) + .chain([ + Arg { + name: "full_refresh".to_string(), + otyp: None, + typ: Typ::Bool, + has_default: true, + default: Some(serde_json::json!(d.full_refresh)), + oidx: None, + otyp_inferred: false, + }, + Arg { + name: "defer".to_string(), + otyp: None, + typ: Typ::Bool, + has_default: true, + default: Some(serde_json::json!(d.defer)), + oidx: None, + otyp_inferred: false, + }, + ]) .collect(), ), ( @@ -546,7 +580,9 @@ fn property_of(arg: &Arg) -> serde_json::Value { ), "select" => Some( "dbt selection syntax, e.g. `tag:nightly`, `stg_orders+`, \ - `config.materialized:incremental`. Empty runs the descriptor's own selection.", + `config.materialized:incremental`. `state:modified+` and `result:error+` \ + compare against the state a previous run published, so they need `defer` on. \ + Empty runs the descriptor's own selection.", ), "exclude" => Some("Nodes to leave out of the selection above, same syntax."), "vars" => Some( @@ -554,6 +590,11 @@ fn property_of(arg: &Arg) -> serde_json::Value { exist makes this run store its own graph rather than the deployed one.", ), "full_refresh" => Some("Rebuild incremental models from scratch instead of appending."), + "defer" => Some( + "Resolve a `ref()` this run does not build to the relation the last successful \ + run of this warehouse and target published, instead of to the schema this run \ + writes into.", + ), "model" => Some( "The model to preview, by name — `stg_orders`, or `my_package.stg_orders` when \ two packages share a name. Any dbt selector resolving to ONE node works.", @@ -687,8 +728,15 @@ full_refresh: true }; let (build, build_args) = of("build"); - assert_eq!(build_args, ["exclude", "full_refresh", "select", "vars"]); + assert_eq!( + build_args, + ["defer", "exclude", "full_refresh", "select", "vars"] + ); assert_eq!(build["properties"]["full_refresh"]["type"], "boolean"); + // `defer` is a per-run toggle rather than a descriptor-only setting: the + // run that publishes an environment's state and the run that defers to + // it are two invocations of ONE script. + assert_eq!(build["properties"]["defer"]["type"], "boolean"); // Defaults come from the descriptor, so an untouched run reproduces it. assert_eq!( build["properties"]["select"]["default"], diff --git a/backend/summarized_schema.txt b/backend/summarized_schema.txt index 61f8a66f85..5ab10d1e02 100644 --- a/backend/summarized_schema.txt +++ b/backend/summarized_schema.txt @@ -74,6 +74,8 @@ dbt_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uui 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) 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) diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index d24b3c3a5c..97e3019835 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -2454,27 +2454,28 @@ async fn create_script_internal<'c>( // while its own finished runs still render from them. Clearing by path // would empty those run pages for good. if ns.language != ScriptLang::Dbt { - // The saved retry state does go: nothing regenerates it, it is keyed by - // path alone, and it carries one user's failed invocation and its - // arguments. No dbt version is live at this path any more to resume it. - windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, &ns.path).await?; + // The saved run and environment state do go: nothing regenerates them, + // both are keyed by path alone, and they carry one user's failed + // invocation with its arguments and the project's own manifest. No dbt + // version is live at this path any more to resume or defer to. + windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, &ns.path).await?; } if let Some(ref old) = p_path_opt { if old != &ns.path { clear_script_triggers(&mut *tx, &w_id, old, AssetUsageKind::Script).await?; clear_static_asset_usage(&mut *tx, &w_id, old, AssetUsageKind::Script).await?; - // The saved retry state travels rather than being cleared: nothing + // The saved state travels rather than being cleared: nothing // regenerates it, so dropping it would throw away a resumable - // failure for what is only a rename. Only while the destination is - // still dbt — a rename that also converts the language would - // otherwise reinstate at the new path the state the branch above - // just cleared, leaving one user's arguments and results under a - // path no dbt script occupies. + // failure and every deferral until the next full run, for what is + // only a rename. Only while the destination is still dbt — a rename + // that also converts the language would otherwise reinstate at the + // new path the state the branch above just cleared, leaving one + // user's arguments and results under a path no dbt script occupies. if ns.language == ScriptLang::Dbt { - windmill_common::dbt_manifest::move_dbt_run_state(&mut tx, &w_id, old, &ns.path) + windmill_common::dbt_manifest::move_dbt_script_state(&mut tx, &w_id, old, &ns.path) .await?; } else { - windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, old).await?; + windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, old).await?; } } } @@ -3712,7 +3713,11 @@ async fn archive_script_by_path( path, &w_id ) - .fetch_one(&db) + // In the SAME transaction as the cleanup below, as the by-hash routes are: + // committed on its own, a cleanup that then fails leaves dbt state at a path + // no live version occupies, for whatever is created there next to defer + // through. + .fetch_one(&mut *tx) .await .map_err(|e| Error::internal_err(format!("archiving script in {w_id}: {e:#}")))?; @@ -3720,9 +3725,10 @@ async fn archive_script_by_path( // The graph stays: the pinned read resolves versions through a CTE that // already skips archived rows, so it stops answering for current relations // either way, while deleting it would empty the Models panel of every - // completed run of the project. Retry state does go — nothing may resume a - // script that is no longer live. - windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, path).await?; + // completed run of the project. The saved run and environment state do go — + // nothing may resume a script that is no longer live, and nothing may defer + // through what it last built. + windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, path).await?; // Pipeline event hygiene: an archived script must not be triggered by // anything. Wipe declared `// on ...` edges (asset-event subscribers // look these up). @@ -3807,7 +3813,7 @@ async fn archive_script_by_hash( clear_static_asset_usage_by_script_hash(&mut *tx, &w_id, hash).await?; // The version's graph stays: its finished runs still render from it, and // the live-version CTE already skips archived rows. Deletion clears it. - windmill_common::dbt_manifest::clear_dbt_run_state_if_path_retired( + windmill_common::dbt_manifest::clear_dbt_script_state_if_path_retired( &mut tx, &w_id, &script.path, @@ -3870,7 +3876,12 @@ async fn delete_script_by_hash( ) .bind(&hash.0) .bind(&w_id) - .fetch_one(&db) + // In the SAME transaction as the cleanup below, as `archive_script_by_hash` + // already does. Committed on its own, it opens a window where the path has + // no live version and a concurrent deploy can take it — and the retirement + // guard below then finds that new script live, keeps the old project's dbt + // state, and leaves the replacement able to defer through its manifest. + .fetch_one(&mut *tx) .await .map_err(|e| Error::internal_err(format!("deleting script by hash {w_id}: {e:#}")))?; @@ -3883,7 +3894,7 @@ async fn delete_script_by_hash( windmill_common::dbt_manifest::clear_dbt_manifest_version(&mut tx, &w_id, &script.path, hash.0) .await?; clear_static_asset_usage_by_script_hash(&mut *tx, &w_id, hash).await?; - windmill_common::dbt_manifest::clear_dbt_run_state_if_path_retired( + windmill_common::dbt_manifest::clear_dbt_script_state_if_path_retired( &mut tx, &w_id, &script.path, @@ -3984,11 +3995,11 @@ async fn delete_script_by_path( // After the DELETE, never before: every dbt writer locks the `script` row // first, so taking a sidecar ahead of it deadlocks one of the pair. The - // VERSIONED graph needs no clear at all, cascading off `script`; the retry - // state does, being keyed by path alone and so inherited by whatever is - // created here next, and so do the editor's own graphs, whose NULL + // VERSIONED graph needs no clear at all, cascading off `script`; the saved + // run and environment state do, being keyed by path alone and so inherited + // by whatever is created here next, and so do the editor's own graphs, whose NULL // `script_hash` satisfies that foreign key without riding its cascade. - windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, path).await?; + windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, path).await?; windmill_common::dbt_manifest::clear_dbt_editor_graphs(&mut tx, &w_id, path).await?; if !trash_scripts.is_empty() { @@ -4157,7 +4168,7 @@ async fn delete_scripts_bulk( // Same reason as the single-path delete, over every requested path rather // than the deleted ones: a path that had no script left can still hold state. for p in &request.paths { - windmill_common::dbt_manifest::clear_dbt_run_state(&mut tx, &w_id, p).await?; + windmill_common::dbt_manifest::clear_dbt_script_state(&mut tx, &w_id, p).await?; windmill_common::dbt_manifest::clear_dbt_editor_graphs(&mut tx, &w_id, p).await?; } diff --git a/backend/windmill-common/src/dbt_manifest.rs b/backend/windmill-common/src/dbt_manifest.rs index e85dfc1910..22ac6ddf33 100644 --- a/backend/windmill-common/src/dbt_manifest.rs +++ b/backend/windmill-common/src/dbt_manifest.rs @@ -35,8 +35,8 @@ //! Every `pub` mutator in this module — the manifest ones //! (`replace_dbt_manifest`, `clear_dbt_manifest_version`, //! `clear_dbt_editor_graphs`), -//! the snapshot sweep, and the retry-state ones (`move_dbt_run_state`, -//! `clear_dbt_run_state`, `clear_dbt_run_state_if_path_retired`) — takes the +//! the snapshot sweep, and the script-state ones (`move_dbt_script_state`, +//! `clear_dbt_script_state`, `clear_dbt_script_state_if_path_retired`) — takes the //! workspace and the script to act on as plain arguments and enforces nothing: //! **the caller must already have verified write access to that script**, //! exactly like the sibling `assets::replace_static_asset_usage` each is called @@ -1120,22 +1120,27 @@ pub async fn clear_dbt_editor_graphs( Ok(()) } -/// Move a dbt script's saved retry state to its new path. +/// Move a dbt script's saved state to its new path: the run `dbt retry` resumes, +/// and the state each environment's deferrals resolve through. /// -/// Keyed by path like the sidecar, but unlike the sidecar it is not -/// regenerated by anything: the deploy re-ingests a manifest, while these are -/// the results of a run that already happened. Clearing on rename would throw -/// away a resumable failure for a cosmetic change, so it travels instead. +/// Keyed by path like the sidecar, but unlike the sidecar neither is regenerated +/// by anything: the deploy re-ingests a manifest, while these are the results of +/// runs that already happened. Clearing on rename would throw away a resumable +/// failure, and every deferral until the next full run, for a cosmetic change — +/// so they travel instead. An artifact too large for its row is unaffected: its +/// key is that publication's own, and the moved row is what names it. /// /// See the mutator contract above: this authorizes nothing. -pub async fn move_dbt_run_state( +pub async fn move_dbt_script_state( tx: &mut Transaction<'_, Postgres>, workspace_id: &str, old_path: &str, new_path: &str, ) -> Result<()> { // The destination may already hold state from a script that lived there - // before; the incoming row is the newer truth for this project. + // before; the incoming row is the newer truth for this project. What the + // displaced row named in object storage is left there, as a cleared one's is + // — see `clear_dbt_script_state`. sqlx::query!( "DELETE FROM dbt_run_state WHERE workspace_id = $1 AND script_path = $2", workspace_id, @@ -1151,22 +1156,38 @@ pub async fn move_dbt_run_state( ) .execute(&mut **tx) .await?; + sqlx::query!( + "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2", + workspace_id, + new_path + ) + .execute(&mut **tx) + .await?; + sqlx::query!( + "UPDATE dbt_environment_state SET script_path = $3 + WHERE workspace_id = $1 AND script_path = $2", + workspace_id, + old_path, + new_path + ) + .execute(&mut **tx) + .await?; Ok(()) } -/// Drop the saved retry state, but only once NO live version of the path is -/// left. +/// Drop the saved state, but only once NO live version of the path is left. /// -/// `dbt_run_state`'s key is the path and the principal — one saved run per script -/// per identity it executes as, not -/// one per version — so archiving or deleting a single version must not take it -/// with them: the live version's `dbt retry` would be refused and the -/// partial-failure resume lost. It does not need to be version-scoped either, -/// because `identity` already refuses a resume whose project, warehouse or -/// engine moved. +/// Neither table is keyed by version — `dbt_run_state` by path and principal, +/// `dbt_environment_state` by path and environment — so archiving or deleting a +/// single version must not take them with it: the live version's `dbt retry` +/// would be refused, its partial-failure resume lost, and every deferral would +/// have to wait for another full run to republish. Neither needs to be +/// version-scoped either: `identity` already refuses a resume whose project, +/// warehouse or engine moved, and a deferral resolves relation names, which a +/// new version of the same project spells the same way. /// /// See the mutator contract above: this authorizes nothing. -pub async fn clear_dbt_run_state_if_path_retired( +pub async fn clear_dbt_script_state_if_path_retired( tx: &mut Transaction<'_, Postgres>, workspace_id: &str, script_path: &str, @@ -1181,17 +1202,34 @@ pub async fn clear_dbt_run_state_if_path_retired( ) .execute(&mut **tx) .await?; + sqlx::query!( + "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2 + AND NOT EXISTS (SELECT 1 FROM script + WHERE workspace_id = $1 AND path = $2 + AND deleted = false AND archived = false)", + workspace_id, + script_path + ) + .execute(&mut **tx) + .await?; Ok(()) } -/// Drop a dbt script's saved retry state. +/// Drop a dbt script's saved state, both halves. /// -/// Archive and delete: `run_results` is not small, the invocation arguments it -/// carries are the user's, and a script later created at the same path would -/// otherwise inherit a stranger's resumable failure. +/// Archive and delete: neither is small, the invocation arguments and manifest +/// they carry are the user's, and a script later created at the same path would +/// otherwise inherit a stranger's resumable failure and defer to a project it +/// has nothing to do with. +/// +/// An artifact too large for its row lives in the instance's object storage, and +/// this leaves it there — as a deleted script leaves its bundle. Reaching it from +/// here would mean an object-store client in this crate and a delete that has to +/// land after the caller's transaction commits, for one object per environment of +/// a script that is gone. /// /// See the mutator contract above: this authorizes nothing. -pub async fn clear_dbt_run_state( +pub async fn clear_dbt_script_state( tx: &mut Transaction<'_, Postgres>, workspace_id: &str, script_path: &str, @@ -1203,6 +1241,13 @@ pub async fn clear_dbt_run_state( ) .execute(&mut **tx) .await?; + sqlx::query!( + "DELETE FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2", + workspace_id, + script_path + ) + .execute(&mut **tx) + .await?; Ok(()) } diff --git a/backend/windmill-common/tests/dbt_graph_storage.rs b/backend/windmill-common/tests/dbt_graph_storage.rs index 7b4971312a..b6c1a1287a 100644 --- a/backend/windmill-common/tests/dbt_graph_storage.rs +++ b/backend/windmill-common/tests/dbt_graph_storage.rs @@ -7,7 +7,8 @@ use sqlx::{Pool, Postgres}; use windmill_common::dbt_manifest::{ - clear_dbt_editor_graphs, clear_dbt_manifest_version, prune_dbt_run_graphs, + 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, }; @@ -235,7 +236,11 @@ async fn clearing_one_version_leaves_the_others(db: Pool) { // this is where two versions coexist: it pins the batched edge insert // against a real database as well as the version scoping. assert_eq!(edges_for(&db, 1).await, 0, "the cleared version's edges go"); - assert_eq!(edges_for(&db, 2).await, 1, "the other version keeps its own"); + assert_eq!( + edges_for(&db, 2).await, + 1, + "the other version keeps its own" + ); } /// The routes that hard-delete a path clear no graph rows: they delete the @@ -363,7 +368,11 @@ async fn only_the_newest_deploys_keep_their_graph(db: Pool) { // The newest is always among them: losing the live version's graph would // empty the page of every run of it. assert_eq!(nodes_for(&db, over, DEPLOYED_GRAPH).await, 1); - assert_eq!(nodes_for(&db, 1, DEPLOYED_GRAPH).await, 0, "the oldest is reclaimed"); + assert_eq!( + nodes_for(&db, 1, DEPLOYED_GRAPH).await, + 0, + "the oldest is reclaimed" + ); } /// The third provenance: a `parse` of the EDITOR's buffer, which names no @@ -480,17 +489,27 @@ async fn a_version_clear_spares_editor_graphs_and_a_path_clear_does_not(db: Pool replace_dbt_editor_graph(&mut tx, WS, PATH, job, ME, &manifest(&["a"]), "root") .await .unwrap(); - clear_dbt_manifest_version(&mut tx, WS, PATH, 1).await.unwrap(); + clear_dbt_manifest_version(&mut tx, WS, PATH, 1) + .await + .unwrap(); tx.commit().await.unwrap(); assert_eq!(nodes_for(&db, 1, DEPLOYED_GRAPH).await, 0); - assert_eq!(editor_nodes(&db, job).await, 1, "the buffer's graph survives"); + assert_eq!( + editor_nodes(&db, job).await, + 1, + "the buffer's graph survives" + ); let mut tx = db.begin().await.unwrap(); clear_dbt_editor_graphs(&mut tx, WS, PATH).await.unwrap(); tx.commit().await.unwrap(); - assert_eq!(editor_nodes(&db, job).await, 0, "retiring the path takes it"); + assert_eq!( + editor_nodes(&db, job).await, + 0, + "retiring the path takes it" + ); } /// A preview names its own PATH and needs only `jobs:run`, so a bound over the @@ -552,3 +571,157 @@ async fn editor_markers(db: &Pool) -> i64 { .unwrap() .unwrap_or(0) } + +/// A deferral resolves a `ref()` through the manifest of the last successful run +/// at this path, so that state has to follow the script the way the retry state +/// does: a rename must not strand it, and a path no live dbt version occupies +/// must not hand its manifest to whatever is created there next. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn environment_state_follows_the_script(db: Pool) { + const MOVED: &str = "f/test/renamed"; + deploy_script(&db, 1).await; + publish_environment_state(&db, PATH).await; + + let mut tx = db.begin().await.unwrap(); + move_dbt_script_state(&mut tx, WS, PATH, MOVED) + .await + .unwrap(); + tx.commit().await.unwrap(); + assert_eq!(environment_states(&db, PATH).await, 0); + assert_eq!(environment_states(&db, MOVED).await, 1); + + let mut tx = db.begin().await.unwrap(); + clear_dbt_script_state(&mut tx, WS, MOVED).await.unwrap(); + tx.commit().await.unwrap(); + assert_eq!(environment_states(&db, MOVED).await, 0); +} + +/// Archiving or deleting ONE version must not take the path's state with it — +/// the live version's next deferral still needs it — while the last one leaving +/// must, or a script later created at that path inherits the previous project's +/// manifest. The condition is a `NOT EXISTS` in raw SQL, so both directions are +/// pinned against a real database. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn state_goes_only_once_no_live_version_is_left(db: Pool) { + deploy_script(&db, 1).await; + deploy_script(&db, 2).await; + publish_environment_state(&db, PATH).await; + + retire(&db, 1).await; + let mut tx = db.begin().await.unwrap(); + clear_dbt_script_state_if_path_retired(&mut tx, WS, PATH) + .await + .unwrap(); + tx.commit().await.unwrap(); + assert_eq!( + environment_states(&db, PATH).await, + 1, + "another version is still live here" + ); + + retire(&db, 2).await; + let mut tx = db.begin().await.unwrap(); + clear_dbt_script_state_if_path_retired(&mut tx, WS, PATH) + .await + .unwrap(); + tx.commit().await.unwrap(); + assert_eq!( + environment_states(&db, PATH).await, + 0, + "the last one leaving takes it" + ); +} + +async fn retire(db: &Pool, hash: i64) { + sqlx::query!( + "UPDATE script SET archived = true WHERE workspace_id = $1 AND hash = $2", + WS, + hash + ) + .execute(db) + .await + .unwrap(); +} + +/// The worker publishes under a guard naming the version that ran, and the whole +/// point of it is a job that finishes late: its script can be renamed away and an +/// unrelated one created at the same path while it runs, and that project must +/// not inherit this one's manifest as its deferral state. Enforced in raw SQL, +/// where a refactor can drop a predicate with no type error, so it is pinned +/// against a real database — the same shape `dbt_state::publish` issues. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn a_late_job_cannot_publish_for_a_path_it_no_longer_owns(db: Pool) { + deploy_script(&db, 1).await; + assert_eq!(guarded_publish(&db, PATH, 1).await, 1, "its own version"); + assert_eq!( + guarded_publish(&db, PATH, 2).await, + 0, + "a version that never lived here" + ); + + // The script is gone from this path and another one takes it. + sqlx::query!( + "DELETE FROM script WHERE workspace_id = $1 AND hash = 1", + WS + ) + .execute(&db) + .await + .unwrap(); + deploy_script(&db, 3).await; + assert_eq!( + guarded_publish(&db, PATH, 1).await, + 0, + "the late job's version does not own this path any more" + ); +} + +/// The predicate `dbt_state::publish` locks the script row on, reduced to what it +/// decides. Keep the two in step — this file cannot call `publish` itself, which +/// is `pub(crate)` in `windmill-worker`. +async fn guarded_publish(db: &Pool, path: &str, ran: i64) -> u64 { + sqlx::query!( + "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id, + manifest) + SELECT $1::varchar, $2::varchar, 'main||analytics|wh'::text, $3::uuid, '{}'::text + WHERE EXISTS (SELECT 1 FROM script + WHERE workspace_id = $1 AND path = $2 + AND deleted = false AND archived = false + AND language = 'dbt' + AND (hash = $4 OR $4 = ANY(parent_hashes))) + ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET job_id = EXCLUDED.job_id", + WS, + path, + uuid::Uuid::from_u128(9), + ran, + ) + .execute(db) + .await + .unwrap() + .rows_affected() +} + +async fn publish_environment_state(db: &Pool, path: &str) { + sqlx::query!( + "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id, + manifest) + VALUES ($1, $2, 'main||analytics|wh', $3, '{}')", + WS, + path, + uuid::Uuid::from_u128(9), + ) + .execute(db) + .await + .unwrap(); +} + +async fn environment_states(db: &Pool, path: &str) -> i64 { + sqlx::query_scalar!( + "SELECT count(*) FROM dbt_environment_state WHERE workspace_id = $1 AND script_path = $2", + WS, + path + ) + .fetch_one(db) + .await + .unwrap() + .unwrap_or(0) +} diff --git a/backend/windmill-worker/src/dbt_executor.rs b/backend/windmill-worker/src/dbt_executor.rs index f302df6a1a..727ed78c5c 100644 --- a/backend/windmill-worker/src/dbt_executor.rs +++ b/backend/windmill-worker/src/dbt_executor.rs @@ -19,12 +19,13 @@ use tokio::process::Command; use uuid::Uuid; use windmill_common::client::AuthedClient; use windmill_common::error::{self, Error}; +use windmill_common::jobs::JobKind; use windmill_common::materialization::{ record_materialization, MaterializationStatus, RecordMaterializationRequest, }; use windmill_common::worker::{to_raw_value, write_file, Connection}; use windmill_parser_yaml::{ - parse_dbt_descriptor, DbtDescriptor, DbtTestBehavior, DBT_COMMANDS, DBT_COMMAND_ARG, + parse_dbt_descriptor, DbtDescriptor, DbtEngine, DbtTestBehavior, DBT_COMMANDS, DBT_COMMAND_ARG, DBT_COMMAND_LABEL, DBT_DEFAULT_WAREHOUSE, }; use windmill_queue::{append_logs, CanceledBy, MiniPulledJob}; @@ -37,6 +38,9 @@ use crate::dbt_engine::{provision_engine, ProvisionedEngine, DBT_CACHE_DIR}; use crate::dbt_profiles::{ ensure_adapter_licensed, render_dbt_profile, render_profile, DbtAdapter, KnownAdapter, }; +use crate::dbt_state::{ + environment_label, prepare_deferral, write_state_dir, Deferral, StateManifest, STATE_DIR, +}; use crate::handle_child::{ get_mem_peak, handle_child, run_future_with_polling_update_job_poller, JobCtx, JobDeadline, }; @@ -121,6 +125,12 @@ pub struct DbtRunResult { /// the same project — cannot get them from the job. #[serde(skip_serializing_if = "std::collections::HashMap::is_empty")] pub invocation_args: std::collections::HashMap>, + /// The run whose stored state this one resolved its unbuilt `ref()`s + /// through, absent when it deferred to none. What a deferring run built + /// against is otherwise unrecoverable: the state is replaced by the next + /// successful run of that environment. + #[serde(skip_serializing_if = "Option::is_none")] + pub deferred_to: Option, } #[derive(Serialize, Debug, Default)] @@ -189,7 +199,13 @@ pub(crate) async fn handle_dbt_job( // result publishes, and both describe an invocation of this script, not one // executor's view of it. let raw_args = job.args.as_ref().map(|a| a.0.clone()).unwrap_or_default(); - let inv = Invocation { args: args.clone(), raw_args, envs: envs.clone(), strict: true }; + let inv = Invocation { + args: args.clone(), + raw_args, + envs: envs.clone(), + deferral: None, + strict: true, + }; // One wall clock for the whole job. A dbt job is a sequence of // subprocesses — provision, deps, parse, ls, build, then the // `after_all` tests — and each would otherwise resolve the job's full @@ -262,6 +278,16 @@ pub(crate) async fn handle_dbt_job( // applies — nothing is built, so there is no test phase, no materialization, // no retry state and no ownership to publish. if command == "parse" { + // Checked here rather than at the seam below, which cannot tell a parse + // from a run that simply left `defer` off: a parse never reaches the + // deferral at all, so it is the one caller for which "turn `defer` on" + // would be advice that leads nowhere. + check_state_selectors( + &effective_select(&descriptor, &inv)?, + &effective_exclude(&descriptor, &inv)?, + StateAccess::Never(&command), + !selection_is_overridden(&descriptor, &inv.args)?, + )?; return run_parse_only( &prepared, &descriptor, @@ -364,6 +390,84 @@ pub(crate) async fn handle_dbt_job( inv }; + // Read AFTER the retry restore, so a retry defers exactly as the run it + // resumes did: a retry's own arguments are the command block alone, and the + // relations its unbuilt `ref()`s resolve to must not depend on that. + let defer = arg_bool(&inv.args, "defer")?.unwrap_or(descriptor.defer); + // Before the state is fetched, not only at the seam where the selection + // reaches dbt: a selector that cannot work whatever the state says would + // otherwise be masked by the "nothing published yet" refusal, which sends the + // caller to publish a state that will not help. + check_state_selectors( + &effective_select(&descriptor, &inv)?, + &effective_exclude(&descriptor, &inv)?, + if defer { + StateAccess::Given + } else { + StateAccess::OnRequest + }, + !selection_is_overridden(&descriptor, &inv.args)?, + )?; + // A `show` defers too, and every engine takes the flags on it: it COMPILES + // the model it previews, so a model whose upstream this environment built and + // this run did not is exactly the case a deferral exists for. + let inv = if defer { + // Refused before anything runs. `dbt retry` reads the run it resumes + // from `--state`, the flag a deferral needs, so an engine without + // `--defer-state` can be given one or the other: told to defer, it + // resumes the stored state's own (successful) results and rebuilds + // nothing, and left alone it rebuilds the failed nodes with every + // `ref()` resolving into the schema THIS run writes — which for the + // narrowed run a deferral exists to serve is not where those models go. + if command == "retry" && !prepared.engine.engine.has_defer_state_flag() { + return Err(Error::BadRequest(format!( + "`{}` cannot resume a run that deferred: `dbt retry` takes the run it resumes \ + from `--state`, which is also where a deferral reads its manifest, and this \ + engine has no `--defer-state` to tell the two apart. Run the script again \ + instead of resuming it, or move the project to dbt-core-1x", + prepared.engine.engine.as_str() + ))); + } + let deferral = prepare_deferral(&prepared, &job.workspace_id, job_dir, conn).await?; + // Only answerable once the state is loaded: `defer` is enough for a + // `state:` method, which reads the manifest every publication carries, + // but a `result:` one reads `run_results.json` — and a build recovered by + // node retry publishes without it, since the results it holds describe + // only the nodes the retry rebuilt. dbt-core then raises an INTERNAL + // error and the Rust engines match nothing and exit 0. + if !deferral.has_run_results + && selection_names( + &effective_select(&descriptor, &inv)?, + &effective_exclude(&descriptor, &inv)?, + &["result"], + ) + { + return Err(Error::BadRequest(format!( + "a `result:` selector reads `run_results.json` out of the published state, and \ + the state for this environment ({}) carries only the manifest run {} \ + published: a build recovered by node retry stores none, its results describing \ + the retried nodes rather than the whole build. Run this script once without \ + `defer` and without overrides to publish a complete state, or drop the selector", + environment_label(&prepared), + deferral.published_by + ))); + } + append_logs( + &job.id, + &job.workspace_id, + format!( + "\nDeferring unbuilt refs to the dbt state published by run {}; this run \ + publishes none of its own\n", + deferral.published_by + ), + conn, + ) + .await; + Invocation { deferral: Some(deferral), ..inv } + } else { + inv + }; + // Ingested BEFORE the build, from a `dbt parse` with this run's vars, so the // models shown are the ones about to be built. Rows are keyed by path, version // AND job so no two runs collide; the path-keyed `asset` usage belongs to one @@ -428,9 +532,28 @@ pub(crate) async fn handle_dbt_job( // previous attempt's `run_results.json` is still in the job directory. Never on // an agent worker, which cannot read `v2_job_queue` — the wait below would be // uninterruptible, so a cancelled job would hold its slot and then start dbt. + // And never where the engine cannot be told to defer on a `retry`: the + // rebuild would resolve this run's unbuilt refs into the schema it writes + // into, so the nodes it "recovered" would read from the wrong relations. + // Said out loud below rather than silently skipped. + let retry_would_lose_the_deferral = + inv.deferral.is_some() && !prepared.engine.engine.has_defer_state_flag(); let node_retry = descriptor .retry_failed_nodes - .filter(|_| matches!(conn, Connection::Sql(_))); + .filter(|_| matches!(conn, Connection::Sql(_))) + .filter(|_| !retry_would_lose_the_deferral); + if descriptor.retry_failed_nodes.is_some() && retry_would_lose_the_deferral { + append_logs( + &job.id, + &job.workspace_id, + format!( + "\nSkipping the automatic node retry: `{}` cannot defer on a `dbt retry`\n", + prepared.engine.engine.as_str() + ), + conn, + ) + .await; + } let mut retries_left = node_retry.map(|p| p.attempts()).unwrap_or(0); if let Some(policy) = node_retry.filter(|_| run.is_err()) { retry_failed_nodes( @@ -519,6 +642,56 @@ pub(crate) async fn handle_dbt_job( { tracing::warn!("dbt: could not save retry state for job {}: {e:#}", job.id); } + // What a later run defers to, published by the runs whose relations are the + // SCRIPT's — the same condition that decides whether a run's graph becomes + // what the script owns, and for the same reason: an invocation that scoped + // its own models has no standing to say where this project's relations live. + // Success is the other half, because a relation a deferral resolves to has + // to exist. A `retry` is excluded: its `run_results.json` names only the + // nodes it redid, so publishing it would leave the environment claiming a + // run of a handful of models. + // + // And never a run that DEFERRED, whatever narrowed it. A deferring run built + // some of the relations its manifest names and resolved the rest out of the + // state it read, so publishing that manifest would record relations nothing + // built — and a model renamed since would be recorded under a name only a + // full build creates, breaking every later deferral until one repairs it. + // `publishes_ownership` cannot see this on its own: it reads the caller's + // overrides, and a descriptor that already narrows `select` needs none. + if run.is_ok() + && command == "build" + && inv.deferral.is_none() + // A run of the DEPLOYED version, by kind. A preview carries a + // caller-supplied `script_hash` into `runnable_id` + // (`run_preview_script`), so the version guard alone would let anyone who + // may run a job publish arbitrary content as a deployed script's state. + && job.kind == JobKind::Script + && prepared.graph_refresh.publishes_ownership() + { + // Losing it costs the next deferral, not the run that just finished — + // but silently, so the one actionable case (an artifact too large for + // the database on an instance with no object storage) says so. + if let Err(e) = crate::dbt_state::publish( + &prepared, + &job.workspace_id, + &job.id, + job.runnable_id.map(|h| h.0), + // An attempt was spent, so `run_results.json` on disk is the one + // `dbt retry` left: the nodes it redid, not the build. + node_retry.is_some_and(|p| retries_left < p.attempts()), + conn, + ) + .await + { + append_logs( + &job.id, + &job.workspace_id, + format!("\nCould not publish this run as the environment's dbt state: {e}\n"), + conn, + ) + .await; + } + } let reconciled = reconcile_materializations(&prepared, &results, job, conn, client).await; terminalize_running_relations(job, &reconciled, conn).await; @@ -895,6 +1068,17 @@ pub struct PreparedProject { /// The descriptor's `profile.target`, passed as `--target` so it applies to /// a project-owned `profiles.yml` as well as a rendered one. pub target: Option, + /// The target dbt actually runs, which is the above only when the descriptor + /// names one: otherwise it is the workspace warehouse's, or the project's own + /// `profiles.yml` default. Half of an environment's identity, since a + /// `target.name` macro decides where a model is built. + pub effective_target: Option, + /// Whether the profile templates where its relations go — a project-owned + /// `profiles.yml`, a `dbt_profile` resource's block, or `profile.schema`, + /// all of which reach dbt as written. Two renderings then share one + /// `relation_root` and an environment cannot be told apart, so such a + /// project neither publishes state nor defers to any. + pub templated_location: bool, /// The profile target's database. Nodes that override it qualify their /// `dbt://` schema segment so two databases cannot collapse onto one node. pub default_database: Option, @@ -932,7 +1116,7 @@ impl PreparedProject { /// Where this run's relations live: the resolved schema and database. Drift /// here since the deploy means the stored graph names relations that no /// longer exist. - fn relation_root(&self) -> String { + pub(crate) fn relation_root(&self) -> String { format!( "{}|{}", self.default_schema.as_deref().unwrap_or(""), @@ -1064,8 +1248,8 @@ pub(crate) async fn prepare_project( .chain(invocation_env.iter().map(|(k, v)| (k.clone(), v.clone()))) .collect(); - let (profiles_dir, warehouse, adapter, default_database, default_schema, profile_digest) = - write_profiles(descriptor, &project_dir, job_dir, client, &template_env).await?; + let profile = write_profiles(descriptor, &project_dir, job_dir, client, &template_env).await?; + let adapter = profile.adapter.clone(); // The lockfile's version, when it pinned one for this same engine — a // descriptor edited to another engine invalidates the pin. let pinned_version = locks @@ -1186,18 +1370,20 @@ pub(crate) async fn prepare_project( h.finish() }, sandbox_config, - profile_digest, + profile_digest: profile.digest, project_dir, - profiles_dir, + profiles_dir: profile.dir, engine, graph_refresh, - warehouse, + warehouse: profile.warehouse, target: descriptor.profile.target.clone(), + effective_target: profile.target, + templated_location: profile.templated_location, descriptor_content: descriptor_content.to_string(), descriptor_env, - default_database, - default_schema, + default_database: profile.database, + default_schema: profile.schema, script_path: script_path.to_string(), env, }; @@ -1512,6 +1698,24 @@ async fn strip_git_remote(dir: &Path) -> std::io::Result<()> { tokio::fs::write(&config, out).await } +/// What resolving the run's connection settled, beyond the file itself. +struct ResolvedProfile { + dir: PathBuf, + /// The workspace warehouse's NAME, when this project belongs to one. + warehouse: Option, + adapter: DbtAdapter, + database: Option, + schema: Option, + /// The target dbt actually runs, which is not always the descriptor's: it + /// falls back to the workspace warehouse's, and to the project's own + /// `profiles.yml` default. Resolved because it is half of an environment's + /// identity and a `target.name` macro can move every relation. + target: Option, + /// Whether a project-owned `profiles.yml` templates where its relations go. + templated_location: bool, + digest: String, +} + /// Write `profiles.yml`, either rendered from a Windmill resource or taken from /// the project itself. Both paths are supported (decision 8): the workspace /// warehouse is the ergonomic one, the project's own file is what makes an @@ -1522,14 +1726,7 @@ async fn write_profiles( job_dir: &str, client: &AuthedClient, template_env: &HashMap, -) -> error::Result<( - PathBuf, - Option, - DbtAdapter, - Option, - Option, - String, -)> { +) -> error::Result { // The workspace's warehouse, always: a descriptor names one by NAME or takes // `main`, and cannot name a resource at all. The NAME is what asset identity // keys on, so every project on one warehouse shares its nodes while the @@ -1610,14 +1807,16 @@ async fn write_profiles( } None => None, }; - return Ok(( + return Ok(ResolvedProfile { dir, - identity, + warehouse: identity, adapter, - target.database, - target.schema, - profile_digest, - )); + database: target.database, + schema: target.schema, + target: Some(target.name), + templated_location: target.templated_location, + digest: profile_digest, + }); } use windmill_common::workspaces::DBT_PROFILE_RESOURCE_TYPE; @@ -1712,14 +1911,22 @@ async fn write_profiles( rendered.root_certificate_pem.as_deref(), &client.token, ); - Ok(( + Ok(ResolvedProfile { dir, - Some(warehouse.to_string()), + warehouse: Some(warehouse.to_string()), adapter, - rendered.database, - rendered.schema, - profile_digest, - )) + // A `dbt_profile` resource is one block of the user's own + // `profiles.yml`, copied through unchanged, and `profile.schema` is + // written as given — so either can be a template dbt renders and this + // runtime does not, exactly as a project-owned file can. + templated_location: [rendered.database.as_deref(), rendered.schema.as_deref()] + .iter() + .any(|v| v.is_some_and(is_jinja)), + database: rendered.database, + schema: rendered.schema, + target: Some(target.to_string()), + digest: profile_digest, + }) } /// Where a workspace warehouse name points: its resource path and, if the @@ -1860,13 +2067,45 @@ async fn adapter_from_profiles_yml( // identically to one on a workspace warehouse, which is what lets the two // meet on the same node when they are on the same relation. let (database_key, schema_key) = adapter.target_identity_keys(); - let read = |k: &str| { + let raw = |k: &str| { out.get(k) .and_then(|v| v.as_str()) - .map(|v| v.to_string()) - .filter(|v| !v.is_empty() && !v.contains("{{")) + .filter(|v| !v.is_empty()) }; - Ok(ProfileTarget { adapter, database: read(database_key), schema: read(schema_key) }) + let read = |k: &str| raw(k).filter(|v| !v.contains("{{")).map(|v| v.to_string()); + Ok(ProfileTarget { + adapter, + database: read(database_key), + schema: read(schema_key), + // A TEMPLATED location is one dbt renders and this runtime does not, so + // two renderings of this file resolve to one `relation_root` and would + // share one environment — `{{ }}` because `read` drops it and it reads + // as absent, `{% %}` because the raw block is kept and reads the same + // for every rendering. Distinguished from plainly absent, which is the + // adapter's default and does not move. + templated_location: [database_key, schema_key] + .iter() + .any(|k| raw(k).is_some_and(is_jinja)), + // The output actually chosen, which for a templated `target:` is the sole + // one rather than the template text no output answers to. + name: match ( + templated_target, + outputs.as_mapping().and_then(|m| m.keys().next()), + ) { + (true, Some(only)) => only.as_str().unwrap_or(target).to_string(), + _ => target.to_string(), + }, + }) +} + +/// Whether dbt would RENDER this value rather than take it literally. +/// +/// Both delimiters, because dbt renders a profile through Jinja: `{{ … }}` +/// substitutes and `{% … %}` branches, and a schema spelled +/// `{% if env_var('ENV') == 'prod' %}analytics{% else %}dev{% endif %}` moves +/// every relation exactly as an `env_var()` does. +fn is_jinja(v: &str) -> bool { + v.contains("{{") || v.contains("{%") } /// What a project-owned `profiles.yml` target says, for the two things Windmill @@ -1877,6 +2116,14 @@ struct ProfileTarget { adapter: DbtAdapter, database: Option, schema: Option, + /// The output this resolved to, by name. + name: String, + /// Whether its database or schema is a template rather than a literal. The + /// fields above cannot say: a `{{ }}` value is dropped and reads as absent, + /// a `{% %}` block is kept and reads the same for every rendering. So this + /// is what separates "the adapter's default, which does not move" from + /// "wherever this run's environment renders it to". + templated_location: bool, } lazy_static::lazy_static! { @@ -2192,6 +2439,29 @@ async fn retry_failed_nodes( } } +/// The flags that point a deferring invocation at its state directory. +/// +/// `--state` is where a deferred `ref()` resolves through — except on a `retry`, +/// which reads the run it RESUMES from that same flag: handed the deferral's +/// directory, dbt resumes the successful run stored there and rebuilds nothing. +/// dbt-core 1.x has `--defer-state` for exactly this split; the Rust engines do +/// not, and a run that defers is refused a retry there rather than rebuilt with +/// its refs resolving into the schema it writes into (`handle_dbt_job`), which +/// is why the last arm never fires in practice. +/// +/// The directory is relative because dbt records the invocation's flags into +/// `run_results.json`: an absolute path would name the job directory of the run +/// being resumed, gone by the time anything reads it back. +fn defer_flags(command: &str, engine: DbtEngine) -> &'static [&'static str] { + match command { + // `--defer` itself is restored with the rest of the resumed + // invocation's arguments and cannot be set from here. + "retry" if engine.has_defer_state_flag() => &["--defer-state", STATE_DIR], + "retry" => &[], + _ => &["--defer", "--state", STATE_DIR], + } +} + #[allow(clippy::too_many_arguments)] async fn run_dbt( p: &PreparedProject, @@ -2213,6 +2483,10 @@ async fn run_dbt( .args(["--log-format-file", "json"]) .args(["--log-level-file", p.engine.engine.progress_log_level()]); + if inv.deferral.is_some() { + cmd.args(defer_flags(command, p.engine.engine)); + } + if with_selection && command != "retry" { add_selection(&mut cmd, descriptor, inv)?; } @@ -2864,6 +3138,9 @@ async fn run_show( ))); } let mut cmd = dbt_command(p, &["show"]); + if inv.deferral.is_some() { + cmd.args(defer_flags("show", p.engine.engine)); + } add_vars(&mut cmd, descriptor, inv)?; // Intersected with `resource_type:model`, because `show` is only read-only // for models: dbt dispatches a selected SEED through its seed runner and @@ -2957,6 +3234,7 @@ fn build_result( totals, nodes, invocation_args: inv.raw_args.clone(), + deferred_to: inv.deferral.as_ref().map(|d| d.published_by), } } @@ -3470,6 +3748,12 @@ async fn resolve_selection( return Ok(None); } let mut cmd = dbt_command(p, &["ls"]); + // The same state the build resolves through, or a `result:` selector — which + // reads `run_results.json` out of it, and which `select` passes to dbt + // verbatim — fails here, before the build that would have honoured it. + if inv.deferral.is_some() { + cmd.args(defer_flags("ls", p.engine.engine)); + } // A project whose models call `var()` without a default fails to parse // without these, so the selection resolver needs them exactly as the run // does. Placeholders that only a run can fill are dropped rather than @@ -3483,6 +3767,8 @@ async fn resolve_selection( } cmd.args(["--output", "json", "--quiet"]); add_selection(&mut cmd, descriptor, inv)?; + let select = effective_select(descriptor, inv)?; + let exclude = effective_exclude(descriptor, inv)?; // Captured directly, not through `handle_child`: its `pipe_stdout` path goes // 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 @@ -3500,15 +3786,37 @@ async fn resolve_selection( } } } - if set.is_empty() { - // A selection that matches nothing would be ingested as "this script - // owns no relations", wiping its graph and cascade edges — the same - // outcome a failed capture produces, and indistinguishable from it. - // Refuse rather than silently un-wire the script. + // Empty is a real answer from a `state:` or `result:` method and from nothing + // else: `state:modified+` matches nothing exactly when nothing changed since + // the published state, and a run with no work to do is a successful one. Any + // other selection matching nothing is a selector that names nothing — a + // misspelled model, say — which must not pass as a build that did its job. + // Exempting by ORIGIN rather than by method would let every such typo through. + // + // What makes the exemption safe is that the empty set is never ingested as + // ownership, and that now holds through `check_state_selectors`: a `state:` + // or `result:` method survives it only from a run's OWN selection, which + // makes `add_caller_args` set `per_run_models`, which makes + // `publishes_ownership()` false, so the run stores a snapshot of its own. + // Relax the descriptor arm there and a descriptor-narrowed `state:modified+` + // reaches here on an unchanged project and wipes the graph the `else` below + // guards, with nothing failing. + if set.is_empty() && !selection_names(&select, &exclude, &["state", "result"]) { return Err(Error::ExecutionErr( - "the descriptor's `select`/`exclude` matched no dbt nodes; fix the selection rather \ - than deploying a script that owns nothing" - .to_string(), + if selection_is_overridden(descriptor, &inv.args)? { + "this run's `select`/`exclude` matched no dbt nodes, so it would build nothing; \ + check the selector. Only a `state:` or `result:` selector may match nothing, \ + its empty answer being a real one" + .to_string() + } else { + // The descriptor's is also ingested as "this script owns no + // relations", wiping its graph and cascade edges — the same + // outcome a failed capture produces, and indistinguishable from + // it. Refuse rather than silently un-wire the script. + "the descriptor's `select`/`exclude` matched no dbt nodes; fix the selection \ + rather than deploying a script that owns nothing" + .to_string() + }, )); } Ok(Some(set)) @@ -3820,7 +4128,7 @@ async fn save_run_state( if let Connection::Sql(db) = conn { { // Only while a live dbt version stays at this path — the test - // `clear_dbt_run_state_if_path_retired` retires state by, plus the + // `clear_dbt_script_state_if_path_retired` retires state by, plus the // language, since a rename leaves the old path archived rather than // deleted and a path can come back as another language. A job already // running finishes after those move or clear the row: writing then @@ -3982,6 +4290,12 @@ pub struct Invocation { /// what it pointed at must not. pub raw_args: HashMap>, pub envs: HashMap, + /// The stored dbt state this invocation resolves an unbuilt `ref()` through, + /// materialised into the job directory. Carried here rather than passed to + /// each phase: the model phase, the `after_all` tests and every in-job node + /// retry must all resolve a `ref()` the same way, or the tests assert against + /// relations the models never read. + pub deferral: Option, /// A run must fail on a `{{ }}` placeholder it cannot fill; a deploy, which /// has no arguments at all, tolerates them. Declared rather than inferred /// from the argument count: a run submitted with `{}` is still a run, and @@ -4134,11 +4448,12 @@ async fn restore_from_db( if !has_retryable_node(&row.run_results) { return Err(nothing_to_retry()); } - let target = p.project_dir.join(ARTIFACTS_DIR); - tokio::fs::create_dir_all(&target).await.ok(); - tokio::fs::write(target.join("run_results.json"), &row.run_results) - .await - .map_err(|e| Error::internal_err(format!("restoring run_results.json: {e}")))?; + write_state_dir( + &p.project_dir.join(ARTIFACTS_DIR), + Some(&row.run_results), + StateManifest::None, + ) + .await?; // No manifest came with the row, so one has to be re-derived — but not here: // these arguments are as SUBMITTED, and a `$var:` in them shapes the graph // only once resolved. The caller resolves, then parses. @@ -4370,20 +4685,16 @@ async fn restore_run_state( return Err(different_project()); } let saved_args_digest = saved_args_digest.map(str::to_string); - let target = p.project_dir.join(ARTIFACTS_DIR); - tokio::fs::create_dir_all(&target).await.ok(); - // From the bytes already read, not by copying the file again: a burst of saves - // can prune this generation mid-restore, and a `dbt retry` whose + // The results go from the bytes already read, not by copying the file again: a + // burst of saves can prune this generation mid-restore, and a `dbt retry` whose // `run_results.json` went missing rebuilds nothing and reports success. The // manifest has no such copy, so a failure there falls back to a `dbt parse`. - tokio::fs::write(target.join("run_results.json"), &saved_results) - .await - .map_err(|e| { - Error::internal_err(format!("could not restore the previous run's results: {e}")) - })?; - let needs_parse = tokio::fs::copy(snapshot.join("manifest.json"), target.join("manifest.json")) - .await - .is_err(); + let needs_parse = !write_state_dir( + &p.project_dir.join(ARTIFACTS_DIR), + Some(&saved_results), + StateManifest::CopyOf(snapshot.join("manifest.json")), + ) + .await?; // The generation was chosen from a row read before the file work above. A run // finishing in that window publishes a newer one, and resuming the superseded // generation redoes nodes it has already rebuilt — appending to an incremental @@ -4772,10 +5083,26 @@ fn add_selection( descriptor: &DbtDescriptor, inv: &Invocation, ) -> error::Result<()> { - for s in effective_select(descriptor, inv)? { + let select = effective_select(descriptor, inv)?; + let exclude = effective_exclude(descriptor, inv)?; + // The seam itself, which the DEPLOY reaches without going through a run: it + // resolves the descriptor's selection to decide what the script owns, and + // never computes a `defer`. A run has been checked earlier, where the message + // can still come before the state fetch. + check_state_selectors( + &select, + &exclude, + if inv.deferral.is_some() { + StateAccess::Given + } else { + StateAccess::OnRequest + }, + !selection_is_overridden(descriptor, &inv.args)?, + )?; + for s in select { cmd.args(["--select", &s]); } - for s in effective_exclude(descriptor, inv)? { + for s in exclude { cmd.args(["--exclude", &s]); } if let Some(sel) = effective_selector(descriptor, inv)? { @@ -4784,6 +5111,115 @@ fn add_selection( Ok(()) } +/// The method a selection token names, with the graph operators that can +/// surround a node stripped (`@model`, `+model`, `2+model`, `model+`). +fn selector_method(token: &str) -> Option<&str> { + token + .trim_start_matches('@') + .trim_start_matches(|c: char| c.is_ascii_digit()) + .trim_start_matches('+') + .split_once(':') + .map(|(method, _)| method) +} + +/// Every method a selection names. Each entry is a union of whitespace-separated +/// tokens, and each of those an intersection of comma-separated ones. +fn selection_methods<'a>(entries: &'a [String]) -> impl Iterator { + entries + .iter() + .flat_map(|entry| entry.split([' ', '\t', ','])) + .filter_map(selector_method) +} + +/// Whether the run being checked has the state directory a `state:` or `result:` +/// method reads, or could be given one. +#[derive(Clone, Copy)] +enum StateAccess<'a> { + /// Deferring, so the directory is there. + Given, + /// Not deferring, and `defer` is what would hand it one. + OnRequest, + /// This command resolves a selection without ever deferring, so no setting + /// gives it a state and "turn `defer` on" would be advice that leads nowhere. + Never(&'a str), +} + +/// Whether a selection names any of these methods. +fn selection_names(select: &[String], exclude: &[String], methods: &[&str]) -> bool { + selection_methods(select) + .chain(selection_methods(exclude)) + .any(|method| methods.contains(&method)) +} + +/// Refuse a selection dbt cannot resolve, before it silently resolves to the +/// wrong thing. +/// +/// `state:` and `result:` compare against the artifacts in `--state`, which only +/// a deferring run is given. The engines do not agree on what happens without +/// one: dbt-core 1.x raises, but dbt-sa-cli and fusion read a missing state as an +/// EMPTY one and exit 0, so `state:modified` builds nothing and `state:new` +/// builds the whole project, each as a run that reports success. +/// +/// From the DESCRIPTOR they are refused whether or not the run defers, because +/// that selection also decides which nodes the script owns, and "whatever changed +/// last" is not an ownership answer — the deploy resolves it with no state at all. +/// They describe one run, so they belong in a run's own `select`. +/// +/// `source_status:` compares `sources.json`, which `dbt source freshness` writes +/// and no run publishes here, so it has nothing to compare against under any +/// setting. +/// +/// Only what `select` and `exclude` spell directly: a method reached through a +/// `selectors.yml` definition is named nowhere the worker can read, and dbt's +/// own behaviour is what stands there. +fn check_state_selectors( + select: &[String], + exclude: &[String], + access: StateAccess<'_>, + from_descriptor: bool, +) -> error::Result<()> { + for method in selection_methods(select).chain(selection_methods(exclude)) { + match method { + "source_status" => { + return Err(Error::BadRequest( + "a `source_status:` selector compares the source freshness recorded in \ + `sources.json`, which `dbt source freshness` writes and no run stores \ + here, so there is nothing for it to compare against. Drop the selector" + .to_string(), + )) + } + "state" | "result" if from_descriptor => { + return Err(Error::BadRequest(format!( + "a `{method}:` selector describes what ONE run builds, but the descriptor's \ + selection also decides which nodes this script owns, which a deploy \ + resolves with no state to compare against. Move it to the `select` of a \ + run with `defer` on" + ))) + } + "state" | "result" => match access { + StateAccess::Given => {} + StateAccess::OnRequest => { + return Err(Error::BadRequest(format!( + "a `{method}:` selector compares against the dbt state a previous run \ + of this environment published, and only a run with `defer` on is given \ + that state. Turn `defer` on, or drop the selector" + ))) + } + StateAccess::Never(command) => { + return Err(Error::BadRequest(format!( + "a `{method}:` selector compares against the dbt state a previous run \ + of this environment published, and `{command}` resolves its selection \ + without building and never defers, so no setting hands it that state. \ + Drop the selector" + ))) + } + }, + _ => {} + } + } + Ok(()) +} + /// Whether this invocation chose its own `select`/`exclude`. /// /// DIFFERENT from the descriptor's, not merely present: `parse_dbt_sig` gives @@ -5797,6 +6233,27 @@ mod tests { .unwrap(); assert!(untouched.publishes_ownership()); assert_eq!(untouched.snapshot_job(job), None); + + // `resolve_selection` lets a selection match nothing on exactly this + // predicate, because a run that scoped its own selection stores a + // snapshot instead of publishing ownership. Should the two ever drift + // apart, an empty caller selection would wipe the script's graph and + // cascade edges, which is the outcome that guard exists to prevent. + // One-directional: a `vars` override also withholds ownership without + // touching the selection, which is why this is an implication and not an + // equivalence. + for args in [ + arg("select", r#"["state:modified+"]"#), + arg("exclude", r#"["tag:nightly"]"#), + ] { + assert!(selection_is_overridden(&descriptor, &args).unwrap()); + let mut g = GraphRefresh::default(); + g.add_caller_args(&descriptor, &args).unwrap(); + assert!( + !g.publishes_ownership(), + "an overridden selection must not publish ownership" + ); + } } // `dbt retry` restores the previous run's target/ from this directory, so two @@ -5825,6 +6282,128 @@ mod tests { ); } + // A profile whose location dbt renders cannot be told apart from another + // rendering of itself, so it neither publishes state nor defers. Both + // delimiters count: a conditional block moves a schema exactly as an + // `env_var()` substitution does. + #[test] + fn a_rendered_profile_location_is_recognised_by_either_delimiter() { + assert!(is_jinja("{{ env_var('DBT_SCHEMA') }}")); + assert!(is_jinja( + "{% if env_var('ENV') == 'prod' %}analytics{% else %}dev{% endif %}" + )); + assert!(!is_jinja("analytics")); + assert!(!is_jinja("")); + } + + // dbt-sa-cli and fusion exit 0 on a state selector with no state, so nothing + // downstream would report this: the graph operators have to be stripped for + // the method to be seen at all. + #[test] + fn a_state_selector_is_found_under_any_graph_operator() { + // A run's own selection, which is the only place these belong. + let refused = |sel: &str, access: StateAccess<'_>| { + check_state_selectors(&[sel.to_string()], &[], access, false).is_err() + }; + for sel in [ + "state:modified", + "state:modified+", + "+state:new", + "@state:modified", + "2+state:modified+3", + "tag:nightly,state:modified", + "stg_orders+ result:error+", + ] { + assert!( + refused(sel, StateAccess::OnRequest), + "{sel} should need `defer`" + ); + assert!( + !refused(sel, StateAccess::Given), + "{sel} should pass while deferring" + ); + // A parse resolves a selection without ever deferring, so it is + // refused where a run would have been told to turn `defer` on. + assert!( + refused(sel, StateAccess::Never("parse")), + "{sel} cannot parse" + ); + // The descriptor's selection also decides what the script owns, and + // the deploy resolves it with no state, so deferring cannot save it. + assert!( + check_state_selectors(&[sel.to_string()], &[], StateAccess::Given, true).is_err(), + "{sel} should never be a descriptor selection" + ); + } + // A node whose name merely starts with a method's letters is not one. + for sel in ["stg_orders+", "tag:nightly", "stateful_model+"] { + assert!( + !refused(sel, StateAccess::OnRequest), + "{sel} is not a state selector" + ); + } + // No run publishes `sources.json`, so deferring does not help. + for access in [ + StateAccess::Given, + StateAccess::OnRequest, + StateAccess::Never("parse"), + ] { + assert!(refused("source_status:fresher+", access)); + } + // `exclude` reaches dbt the same way `select` does. + assert!(check_state_selectors( + &[], + &["state:modified".to_string()], + StateAccess::OnRequest, + false + ) + .is_err()); + + // The same recognition decides which empty selections `resolve_selection` + // lets through. Only these two answer "nothing" meaningfully; a selector + // naming nothing must not pass as a build that did its work. + const STATE_BACKED: &[&str] = &["state", "result"]; + for sel in ["state:modified+", "result:error+", "tag:x,state:new"] { + assert!( + selection_names(&[sel.to_string()], &[], STATE_BACKED), + "{sel}" + ); + } + for sel in [ + "mispelled_model", + "tag:nightly", + "stg_orders+", + "source_status:fresher+", + ] { + assert!( + !selection_names(&[sel.to_string()], &[], STATE_BACKED), + "{sel}" + ); + } + } + + // The one flag choice that is silently wrong rather than loudly wrong: a + // `retry` handed `--state` resumes the SUCCESSFUL run stored there and + // rebuilds nothing, reporting a green retry of a failed run. + #[test] + fn a_retry_is_never_handed_the_deferral_as_its_state() { + assert_eq!( + defer_flags("build", DbtEngine::DbtCore1x), + ["--defer", "--state", crate::dbt_state::STATE_DIR] + ); + assert_eq!( + defer_flags("test", DbtEngine::Fusion), + ["--defer", "--state", crate::dbt_state::STATE_DIR] + ); + assert_eq!( + defer_flags("retry", DbtEngine::DbtCore1x), + ["--defer-state", crate::dbt_state::STATE_DIR] + ); + for engine in [DbtEngine::DbtCore2x, DbtEngine::Fusion] { + assert!(defer_flags("retry", engine).is_empty()); + } + } + #[test] fn events_without_a_relation_are_not_materializations() { // A test node has no relation of its own. diff --git a/backend/windmill-worker/src/dbt_state.rs b/backend/windmill-worker/src/dbt_state.rs new file mode 100644 index 0000000000..ecf3d0dbe3 --- /dev/null +++ b/backend/windmill-worker/src/dbt_state.rs @@ -0,0 +1,748 @@ +//! The dbt state a project last built into one environment, and the state +//! directory a run reads it back through. +//! +//! `dbt --defer --state ` resolves a `ref()` the run does not build to the +//! relation the manifest in `` names, instead of to the schema this run +//! writes into. That makes the state a durable, per-environment artifact rather +//! than a cache: the next run of a project usually lands on a worker holding +//! neither the manifest nor the results, so anything worker-local answers for +//! one machine's history rather than for the environment. +//! +//! Two artifacts live in that directory and both are stored: `manifest.json`, +//! which is what a deferral resolves through, and `run_results.json`, which +//! `select`'s `result:` selectors read — and `select` reaches dbt verbatim, so a +//! state directory missing it fails a selection a user may legitimately write. + +use std::path::{Path, PathBuf}; + +use uuid::Uuid; +use windmill_common::error::{self, Error}; +use windmill_common::worker::Connection; + +use crate::dbt_executor::{digest, PreparedProject, ARTIFACTS_DIR}; + +lazy_static::lazy_static! { + /// Above this, an artifact goes to the instance's object storage instead of + /// into the row. A manifest passes a few hundred KB on a handful of models + /// and grows with the project, so this ceiling is what decides whether a + /// large project needs storage configured at all; a small one stays in the + /// database, where it costs no round trip and needs nothing configured. + static ref DBT_STATE_INLINE_MAX_BYTES: usize = std::env::var("DBT_STATE_INLINE_MAX_BYTES") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(8 * 1024 * 1024); +} + +/// The directory `--state` points at. Inside the job directory, so it sits in +/// the sandbox's one writable bind and goes away with the job, and prefixed like +/// the artifacts directory beside it so a project carrying a directory of this +/// name is not overwritten. +/// +/// Passed to dbt RELATIVE, and that is load-bearing rather than tidiness. dbt +/// records the invocation's flags into `run_results.json` and a later +/// `dbt retry` restores them, so an absolute path would name the job directory +/// of the run being resumed — gone by then, leaving the retry to resolve a +/// deferred `ref()` against nothing. Relative, it resolves against the project +/// root, which is whichever job directory the retry landed in. +pub(crate) const STATE_DIR: &str = "wm_dbt_state"; + +/// Where this run's relations live, which is the only thing a deferral is about. +pub(crate) fn environment(p: &PreparedProject) -> String { + environment_key( + p.warehouse.as_deref(), + // The target dbt RUNS, not the descriptor's: it falls back to the + // workspace warehouse's and to the project's own default, so reading the + // descriptor's would put two inherited targets under one empty name. + p.effective_target.as_deref(), + // The pair `relation_root` reports to the graph's drift check, taken + // apart so neither can absorb the other's delimiter below. + p.default_schema.as_deref(), + p.default_database.as_deref(), + ) +} + +/// The warehouse and the target name the environment; the database and schema +/// they resolve to are in the key because a repointed warehouse resource or a +/// moved schema keeps both names while putting the relations somewhere else — +/// and a manifest is a list of relation names, so a deferral has no other way to +/// notice. A move therefore reads as an environment nothing has published yet. +/// +/// Length-prefixed rather than joined on a separator. Every component but the +/// warehouse is spelled by the user — a dbt target name and a schema are both +/// arbitrary strings a profile may quote — so `prod|analytics` + `scratch` and +/// `prod` + `analytics|scratch` would otherwise be one key, and a profile moving +/// between them would read as the same environment rather than as one nothing +/// has published. Same reasoning as `stable_digest`, and still legible in a row: +/// `4:main|4:prod|9:analytics|12:dbt_wh_defer`. What a MESSAGE names is +/// `environment_label`, since this encoding is for storage. +fn environment_key( + warehouse: Option<&str>, + target: Option<&str>, + schema: Option<&str>, + database: Option<&str>, +) -> String { + [warehouse, target, schema, database] + .iter() + .map(|v| { + let v = v.unwrap_or(""); + format!("{}:{v}", v.len()) + }) + .collect::>() + .join("|") +} + +/// The environment as a message names it: the key above is length-prefixed for +/// storage, which is not something to put in front of a caller. +pub(crate) fn environment_label(p: &PreparedProject) -> String { + format!( + "warehouse `{}`, target `{}`, relations in `{}`", + p.warehouse.as_deref().unwrap_or("(none)"), + p.effective_target + .as_deref() + .unwrap_or("(the profile's default)"), + match (p.default_database.as_deref(), p.default_schema.as_deref()) { + (Some(db), Some(schema)) => format!("{db}.{schema}"), + (None, Some(schema)) => schema.to_string(), + _ => "(the adapter's default)".to_string(), + } + ) +} + +/// The state one environment last published. +pub(crate) struct StoredState { + pub manifest: String, + pub run_results: Option, + /// The run that published it, so a deferring run can say what it deferred to. + pub job_id: Uuid, +} + +/// Publish this run's artifacts as the environment's state. +/// +/// Called for a run that BUILT what the script's own descriptor selects and +/// succeeded (see `handle_dbt_job`). Best-effort in the same sense as the retry +/// state: losing it costs the next deferral, not the run that just finished. +/// +/// **What the artifacts may carry follows from that condition.** A publishing run +/// added nothing of its own — no `select` or `vars` override, and a descriptor +/// interpolating a `{{ }}` placeholder into `vars` never publishes at all — so +/// dbt's `run_results.json` records the descriptor's own arguments, which are the +/// script's content. That is why this is keyed by environment where +/// `dbt_run_state` is keyed by principal: the retry state holds whatever a caller +/// submitted, this holds what the script says. Widen the publish condition and +/// that stops being true. +pub(crate) async fn publish( + p: &PreparedProject, + w_id: &str, + job_id: &Uuid, + // The version this job ran. `None` for a preview, which publishes nothing. + script_hash: Option, + // A build recovered by the automatic in-job node retry has a + // `run_results.json` naming only the nodes that retry redid. The manifest is + // unaffected — it is a function of the project, not of what ran — so the + // state is published without results rather than with a set describing some + // other slice of the build. + results_are_partial: bool, + conn: &Connection, +) -> error::Result<()> { + let Connection::Sql(db) = conn else { + // An agent worker reaches the database only through the API, which does + // not expose this table. + return Ok(()); + }; + if p.script_path.is_empty() { + // A preview has no path to key state on, and an empty one would be + // shared by every dbt script in the workspace. + return Ok(()); + } + if p.templated_location { + // Refused on this side too, not only where a deferral reads. A template + // renders to one location per environment while the key sees the + // template, so publishing would file this run's manifest under a key a + // literal profile shares — and de-templating later would make that stale + // manifest readable as the new location's. + return Ok(()); + } + let artifacts = p.project_dir.join(ARTIFACTS_DIR); + // The manifest is what a deferral resolves through, so there is no state + // without one. Every engine writes it beside the results of a build, so this + // is the invocation that built nothing rather than a case to report. + let Ok(manifest) = tokio::fs::read_to_string(artifacts.join("manifest.json")).await else { + return Ok(()); + }; + let run_results = match results_are_partial { + true => None, + false => tokio::fs::read_to_string(artifacts.join("run_results.json")) + .await + .ok(), + }; + let environment = environment(p); + // Uploaded BEFORE the transaction, and to this publication's own keys, so two + // publishers cannot collide on them and nothing here can overwrite an + // artifact a committed row still names. A failure below has only its own + // objects to drop. + let nonce = Uuid::new_v4(); + let (manifest, manifest_key) = store( + manifest, + "manifest.json", + &environment, + &p.script_path, + w_id, + job_id, + &nonce, + ) + .await?; + let (run_results, run_results_key) = match run_results { + Some(r) => match store( + r, + "run_results.json", + &environment, + &p.script_path, + w_id, + job_id, + &nonce, + ) + .await + { + Ok(stored) => stored, + Err(e) => { + forget_objects(&[manifest_key, None]).await; + return Err(e); + } + }, + None => (None, None), + }; + let mine = [manifest_key.clone(), run_results_key.clone()]; + // One publisher per environment at a time, so the row and the objects it + // displaces are settled by one of them at a time. An advisory lock rather + // than the row's, because the first publish of an environment has no row to + // lock and is exactly when two runs of a newly deployed script are most + // likely to race. + let mut tx = match db.begin().await { + Ok(tx) => tx, + Err(e) => { + forget_objects(&mine).await; + return Err(e.into()); + } + }; + let staged = async { + sqlx::query_scalar!( + "SELECT pg_advisory_xact_lock($1)", + publication_lock(w_id, &p.script_path, &environment) + ) + .execute(&mut *tx) + .await?; + // The script row FIRST, and held, so a rename, archive or delete of this + // path either waits for this publication or is seen by it. Reading it + // unlocked leaves a window where lifecycle cleanup finds no row to clear, + // finishes, and this transaction then commits state at a path a new + // script goes on to occupy. Script row before sidecar is also the order + // every other dbt writer takes, which is what keeps the two off a + // deadlock. + // + // The version, not just the path: "some live dbt script is here" is also + // satisfied by a script created at a path this one was renamed away from. + // A preview names no version, so `script_hash` is NULL and nothing + // matches — right for a run of content that was never deployed. + let owns_path = sqlx::query_scalar!( + "SELECT 1 FROM script + WHERE workspace_id = $1 AND path = $2 + AND deleted = false AND archived = false AND language = 'dbt' + AND (hash = $3 OR $3 = ANY(parent_hashes)) + FOR SHARE", + w_id, + &p.script_path, + script_hash, + ) + .fetch_optional(&mut *tx) + .await? + .is_some(); + if !owns_path { + return error::Result::Ok(None); + } + // What the row points at NOW, so those objects can go once this one is + // committed in their place — never before, since a reader that has + // already read the row is about to fetch them. + let displaced = sqlx::query!( + "SELECT manifest_key, run_results_key FROM dbt_environment_state + WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + w_id, + &p.script_path, + environment + ) + .fetch_optional(&mut *tx) + .await? + .map(|r| [r.manifest_key, r.run_results_key]) + .unwrap_or_default() + // Never a key this publication is about to commit. The keys carry a + // per-execution nonce so the two cannot coincide, and this is what says + // so rather than leaving it to be re-derived. + .map(|k| k.filter(|k| !mine.iter().flatten().any(|m| m == k))); + sqlx::query!( + "INSERT INTO dbt_environment_state (workspace_id, script_path, environment, job_id, + manifest, manifest_key, run_results, + run_results_key, updated_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now()) + ON CONFLICT (workspace_id, script_path, environment) DO UPDATE SET + job_id = EXCLUDED.job_id, manifest = EXCLUDED.manifest, + manifest_key = EXCLUDED.manifest_key, run_results = EXCLUDED.run_results, + run_results_key = EXCLUDED.run_results_key, updated_at = now()", + w_id, + &p.script_path, + environment, + job_id, + manifest, + manifest_key, + run_results, + run_results_key, + ) + .execute(&mut *tx) + .await?; + error::Result::Ok(Some(displaced)) + } + .await; + let displaced = match staged { + // Refused by the guard, or the write failed: nothing is committed and + // what was uploaded above has no row naming it. + Ok(None) | Err(_) => { + forget_objects(&mine).await; + return staged.map(|_| ()); + } + Ok(Some(displaced)) => displaced, + }; + // A commit that reports an error may still have committed — what was lost can + // be the acknowledgement. Dropping this run's objects would then leave the + // committed row naming objects that are gone, and every deferral would fail + // until the next publication; an orphan costs storage instead. + tx.commit().await?; + forget_objects(&displaced).await; + Ok(()) +} + +/// The environment's state, or `None` where nothing has published one. +pub(crate) async fn load( + p: &PreparedProject, + w_id: &str, + conn: &Connection, +) -> error::Result> { + let Connection::Sql(db) = conn else { + return Err(Error::BadRequest( + "`defer` resolves a `ref()` through the dbt state stored for this environment, which \ + an agent worker cannot read: it reaches the database only through the API. Run this \ + script on a worker of the main group, or without `defer`" + .to_string(), + )); + }; + let environment = environment(p); + // A publication committing between the row and the objects it named drops + // those objects, so a miss is re-read rather than reported as a state that is + // not there. Re-read for as long as the row keeps MOVING: a reader takes no + // lock, so back-to-back publications can each overtake it, and a fixed one + // retry would report the second as missing. An unmoved row is the other + // answer — nothing republished, so the object really is gone. + let mut tried: Option<(Uuid, Option, Option)> = None; + for _ in 0..PUBLICATIONS_OUTRUN { + let Some(row) = sqlx::query!( + "SELECT job_id, manifest, manifest_key, run_results, run_results_key + FROM dbt_environment_state + WHERE workspace_id = $1 AND script_path = $2 AND environment = $3", + w_id, + &p.script_path, + environment + ) + .fetch_optional(db) + .await? + else { + return Ok(None); + }; + let seen = ( + row.job_id, + row.manifest_key.clone(), + row.run_results_key.clone(), + ); + let fetched = async { + let manifest = fetch(row.manifest, row.manifest_key).await?; + let run_results = fetch(row.run_results, row.run_results_key).await?; + error::Result::Ok((manifest, run_results)) + } + .await; + match fetched { + Ok((Some(manifest), run_results)) => { + return Ok(Some(StoredState { manifest, run_results, job_id: seen.0 })) + } + Ok((None, _)) => return Ok(None), + Err(e) => { + if tried.as_ref() == Some(&seen) { + return Err(e); + } + tried = Some(seen); + } + } + } + Err(Error::internal_err( + "the dbt state for this environment was replaced faster than it could be read; run this \ + script again" + .to_string(), + )) +} + +/// How many publications a read may lose to before it gives up. Each one costs a +/// re-read, and a project publishing this often while another run defers is +/// already contending for the same relations. +const PUBLICATIONS_OUTRUN: usize = 5; + +/// A `manifest.json` for a state directory, whichever side it comes from. +/// +/// One enum because the three restores — a deferral's stored state, a retry's +/// worker-local generation, a retry's database row — differ only in where the +/// bytes are, and a second copy of the directory layout is a second chance for +/// one of them to write a directory dbt reads differently. +pub(crate) enum StateManifest { + Bytes(String), + /// A file on this worker, copied rather than read into memory: a manifest + /// grows with the project. + CopyOf(PathBuf), + None, +} + +/// Write the artifacts a dbt state directory holds, creating it if needed. +/// +/// Returns whether a `manifest.json` ended up there — a worker-local generation +/// can be pruned out from under a restore, and the caller then owes a +/// `dbt parse` for one. +pub(crate) async fn write_state_dir( + dir: &Path, + run_results: Option<&str>, + manifest: StateManifest, +) -> error::Result { + tokio::fs::create_dir_all(dir) + .await + .map_err(|e| Error::internal_err(format!("preparing the dbt state directory: {e}")))?; + if let Some(run_results) = run_results { + tokio::fs::write(dir.join("run_results.json"), run_results) + .await + .map_err(|e| Error::internal_err(format!("writing run_results.json: {e}")))?; + } + Ok(match manifest { + StateManifest::Bytes(m) => { + tokio::fs::write(dir.join("manifest.json"), m) + .await + .map_err(|e| Error::internal_err(format!("writing manifest.json: {e}")))?; + true + } + StateManifest::CopyOf(from) => tokio::fs::copy(from, dir.join("manifest.json")) + .await + .is_ok(), + StateManifest::None => false, + }) +} + +/// The advisory lock one environment's publishers take, so only one of them +/// settles the row and the objects it displaces at a time. +/// +/// Derived from the same three components as the row's key. Two environments +/// whose digests collide wait for each other, which costs a moment and nothing +/// else. +fn publication_lock(w_id: &str, script_path: &str, environment: &str) -> i64 { + let d = digest(&format!("{w_id}|{script_path}|{environment}")); + // Parsed unsigned and reinterpreted: half of all digests set the top bit, + // and read as `i64` those overflow and would collapse onto one key. + u64::from_str_radix(&d[..16], 16).unwrap_or_default() as i64 +} + +/// The object-storage key an artifact takes. +/// +/// One key per PUBLICATION, so an upload never overwrites an artifact the +/// committed row still names: a run that fails between its two uploads, or +/// between them and its row, leaves the state pointing at the pair it already +/// had. The row switches to these in one statement and the objects it displaced +/// are dropped afterwards. The path and environment are only a prefix — the row +/// is what says where an artifact is, so state that moves with a renamed script +/// keeps naming objects under the old one. Digested because a Windmill path and a +/// schema name may both carry characters an object key gives meaning to. +/// +/// The `nonce` is per EXECUTION rather than per job, because zombie recovery +/// re-runs a job under its own id: keyed on that alone, the second attempt would +/// overwrite the objects the first attempt's committed row still names, and then +/// read those same keys back as displaced and drop them. +fn object_key( + w_id: &str, + script_path: &str, + environment: &str, + job_id: &Uuid, + nonce: &Uuid, + artifact: &str, +) -> String { + format!( + "wmill_dbt_state/{w_id}/{}/{job_id}.{nonce}/{artifact}", + digest(&format!("{script_path}|{environment}")) + ) +} + +/// Put an artifact where its size says it belongs: `(inline, key)`, exactly one +/// of which is set. +#[allow(clippy::too_many_arguments)] +async fn store( + value: String, + artifact: &str, + environment: &str, + script_path: &str, + w_id: &str, + job_id: &Uuid, + nonce: &Uuid, +) -> error::Result<(Option, Option)> { + if value.len() <= *DBT_STATE_INLINE_MAX_BYTES { + return Ok((Some(value), None)); + } + let key = object_key(w_id, script_path, environment, job_id, nonce, artifact); + let size = value.len(); + if put_object(&key, value).await? { + return Ok((None, Some(key))); + } + Err(Error::BadRequest(format!( + "this project's {artifact} is {}, past the {} this instance keeps in the database, and \ + this instance has no object storage configured to hold it. Configure instance object \ + storage, or raise DBT_STATE_INLINE_MAX_BYTES", + mib(size), + mib(*DBT_STATE_INLINE_MAX_BYTES), + ))) +} + +fn mib(bytes: usize) -> String { + format!("{:.1} MiB", bytes as f64 / (1024.0 * 1024.0)) +} + +/// Read an artifact back from whichever home the row names. +async fn fetch(inline: Option, key: Option) -> error::Result> { + match (inline, key) { + (Some(inline), _) => Ok(Some(inline)), + (None, Some(key)) => get_object(&key).await.map(Some), + (None, None) => Ok(None), + } +} + +/// Drop the objects nothing points at any more. Best-effort: an object left +/// behind costs storage, and there is nothing useful to do about it in the path +/// of a run that has already finished. +async fn forget_objects(keys: &[Option; 2]) { + for key in keys.iter().flatten() { + delete_object(key).await; + } +} + +/// Whether the artifact was stored. `false` means this instance has no object +/// storage to put it in. +/// +/// The INSTANCE store, where every other internal worker artifact lives — bun +/// bundles, python wheels, job logs, the global cache. Not the workspace's: +/// that bucket is the one workspace members read and write through +/// `job_helpers/*` and `wmill.write_s3_file`, so a manifest there is one any +/// member could replace, and the next deferring run would hand dbt an +/// attacker-chosen `defer_relation` for every unbuilt `ref()` while holding the +/// script's warehouse credentials. Its compiled SQL would be readable there too, +/// for a project the reader may have no access to. +#[cfg(all(feature = "enterprise", feature = "parquet"))] +async fn put_object(key: &str, value: String) -> error::Result { + use windmill_object_store::object_store_reexports::Path as ObjectPath; + let Some(store) = windmill_object_store::get_object_store().await else { + return Ok(false); + }; + store + .put(&ObjectPath::from(key), bytes::Bytes::from(value).into()) + .await + .map_err(|e| Error::internal_err(format!("storing the dbt state at {key}: {e:#}")))?; + Ok(true) +} + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +async fn get_object(key: &str) -> error::Result { + let Some(store) = windmill_object_store::get_object_store().await else { + return Err(missing_storage()); + }; + let bytes = windmill_object_store::attempt_fetch_bytes(store, key).await?; + String::from_utf8(bytes.to_vec()) + .map_err(|e| Error::internal_err(format!("the stored dbt state is not valid UTF-8: {e}"))) +} + +#[cfg(all(feature = "enterprise", feature = "parquet"))] +async fn delete_object(key: &str) { + use windmill_object_store::object_store_reexports::Path as ObjectPath; + let Some(store) = windmill_object_store::get_object_store().await else { + return; + }; + if let Err(e) = store.delete(&ObjectPath::from(key)).await { + tracing::warn!("dbt: could not drop the superseded state object {key}: {e:#}"); + } +} + +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +async fn delete_object(_key: &str) {} + +/// A build without the instance store carries no client at all, so an oversized +/// artifact has nowhere but the row, and a row naming a key was written by a +/// worker that did have one. +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +async fn put_object(_key: &str, _value: String) -> error::Result { + Ok(false) +} + +#[cfg(not(all(feature = "enterprise", feature = "parquet")))] +async fn get_object(_key: &str) -> error::Result { + Err(missing_storage()) +} + +fn missing_storage() -> Error { + Error::BadRequest( + "the dbt state for this environment is in the instance's object storage, which this \ + worker cannot reach: it is no longer configured, or this worker was built without \ + object-storage support" + .to_string(), + ) +} + +/// The stored state this run resolves its unbuilt `ref()`s through, materialised +/// into the job directory at `STATE_DIR`. +#[derive(Clone, Debug)] +pub(crate) struct Deferral { + /// The run that published the state, so the job log and the result can say + /// what this one deferred to. + pub published_by: Uuid, + /// Whether the state carries `run_results.json` beside its manifest. A build + /// recovered by node retry publishes without one, and that is the only file a + /// `result:` selector reads. + pub has_run_results: bool, +} + +/// Materialise the environment's state so `--state` has a directory to read. +/// +/// Refused rather than run without deferral where nothing is published: the run +/// would build against a `ref()` resolving into the schema it writes, and fail +/// deep inside dbt with a relation-not-found the caller has no way to connect +/// back to a missing state. +pub(crate) async fn prepare_deferral( + p: &PreparedProject, + w_id: &str, + job_dir: &str, + conn: &Connection, +) -> error::Result { + if p.script_path.is_empty() { + return Err(Error::BadRequest( + "`defer` resolves a `ref()` through the state a previous run of this script \ + published, so it needs a deployed script; a preview run has no environment to have \ + published one" + .to_string(), + )); + } + // An environment is the warehouse, the target and where they RESOLVE to, and + // a `profiles.yml` that templates its schema or database resolves somewhere + // this runtime does not render. Two renderings would then share one + // environment, and a deferral after the value changed would resolve every + // unbuilt `ref()` through the previous location's manifest. + if p.templated_location { + return Err(Error::BadRequest( + "this project's profile selects its schema or database with a template, which dbt \ + renders and Windmill does not — so two environments cannot be told apart and a \ + deferral could resolve through the wrong one's manifest. Spell the target's schema \ + and database literally to use `defer`" + .to_string(), + )); + } + let Some(state) = load(p, w_id, conn).await? else { + return Err(Error::BadRequest(format!( + "no dbt state is stored for this environment ({}), so a `ref()` this run does not \ + build has no relation to resolve to. It is published by a successful run that adds \ + nothing of its own: one overriding `select` or `vars` does not publish, and neither \ + does any run of a descriptor that interpolates a `{{{{ }}}}` placeholder into `vars` \ + or a `$var:` into `env` — those describe a model set the caller's arguments decided. \ + Run this script once without `defer` and without overrides", + environment_label(p) + ))); + }; + let has_run_results = state.run_results.is_some(); + write_state_dir( + &PathBuf::from(job_dir).join(STATE_DIR), + state.run_results.as_deref(), + StateManifest::Bytes(state.manifest), + ) + .await?; + Ok(Deferral { published_by: state.job_id, has_run_results }) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Every component earns its place: a deferral resolves relation NAMES, so + // state published where those names meant something else has to read as no + // state at all rather than as state that silently no longer fits. + #[test] + fn a_moved_profile_is_another_environment() { + let here = environment_key(Some("main"), Some("prod"), Some("analytics"), Some("wh")); + assert_eq!( + here, + environment_key(Some("main"), Some("prod"), Some("analytics"), Some("wh")) + ); + assert_ne!( + here, + environment_key(Some("other"), Some("prod"), Some("analytics"), Some("wh")) + ); + assert_ne!( + here, + environment_key(Some("main"), Some("dev"), Some("analytics"), Some("wh")) + ); + assert_ne!( + here, + environment_key(Some("main"), Some("prod"), Some("marts"), Some("wh")) + ); + assert_ne!( + here, + environment_key( + Some("main"), + Some("prod"), + Some("analytics"), + Some("other_db") + ) + ); + } + + // A target name and a schema are both the user's own strings, so a component + // carrying the separator must not be able to spell another tuple's key: a + // profile moving between the two would read as the same environment and + // defer through the manifest of relations that are somewhere else. + #[test] + fn a_component_cannot_spell_another_environments_key() { + assert_ne!( + environment_key(Some("main"), Some("prod|analytics"), Some("scratch"), None), + environment_key(Some("main"), Some("prod"), Some("analytics|scratch"), None) + ); + assert_ne!( + environment_key(Some("main"), Some("prod"), Some("a"), Some("b|c")), + environment_key(Some("main"), Some("prod"), Some("a|b"), Some("c")) + ); + // A component the profile leaves out is the same environment as one it + // spells empty: there is no target named "". + assert_eq!( + environment_key(Some("main"), None, Some("a"), None), + environment_key(Some("main"), Some(""), Some("a"), Some("")) + ); + } + + // Two environments must not queue behind one advisory lock, which is what a + // digest folded through a signed parse did for every one whose top bit is + // set — half of them. + #[test] + fn each_environment_gets_its_own_publication_lock() { + let mut seen = std::collections::HashSet::new(); + for i in 0..64 { + seen.insert(publication_lock( + "ws", + "f/a/p", + &format!("main|prod|s{i}|db"), + )); + } + assert_eq!(seen.len(), 64); + assert_eq!( + publication_lock("ws", "f/a/p", "e"), + publication_lock("ws", "f/a/p", "e") + ); + } +} diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index e3e7868548..8aa739ecd8 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -36,6 +36,7 @@ mod csharp_executor; mod dbt_engine; mod dbt_executor; mod dbt_profiles; +mod dbt_state; #[cfg(feature = "private")] mod dedicated_worker_ee; mod dedicated_worker_oss; diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 3749bda3ea..acb4621b27 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -13,8 +13,8 @@ the dominant way dbt is orchestrated today. - **In**: run an unmodified dbt project synced into Windmill, one Windmill job per invocation, live per-model observability, dbt models as first-class assets in the existing asset graph. -- **Out**: one Windmill job per dbt model, `state:modified` / slim CI, - `dbt docs` hosting, semantic layer, dbt platform integration. +- **Out**: one Windmill job per dbt model, slim CI orchestration, `dbt docs` + hosting, semantic layer, dbt platform integration. - **CE**: the runtime, the manifest ingest, the asset graph and every piece of UI ship in CE, as do all adapters except two. Only the `mssql` and `oracle` adapters are EE, mirroring the native `ScriptLang` boundary (decision 21). @@ -35,7 +35,7 @@ the dominant way dbt is orchestrated today. | 10 | Private repo auth | Not applicable: the project is synced, not fetched | | 11 | Asset kind | `dbt:////` — keyed on the relation, not on dbt's node id. See below | | 12 | Graph refresh | Deploy-time, re-ingested per run only when the descriptor is dynamic, plus an explicit `parse` of the editor's buffer. See below | -| 13 | Manifest storage | Sidecar table for nodes/edges. Full manifest **not** stored — see below | +| 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 | | 15 | Node rendering | Asset nodes per model plus one runnable node for the script | | 16 | Progress | Live, from the JSON event stream | @@ -48,6 +48,7 @@ the dominant way dbt is orchestrated today. | 23 | Descriptor | `wm_dbt.yaml` inside the project, OPTIONAL. See below | | 24 | Warehouse | Configured on the workspace by name, `main` by default. See below | | 25 | Cascade direction | Into a relation, not out of a run: `// materialize manual dbt://…` declares a write from any language but dbt's own and wakes `# on dbt://…` subscribers; a finished dbt run still does not dispatch. See "No cascade *from* dbt" | +| 26 | Deferral | A durable state per environment, published by the runs whose relations are the script's; `defer` is a per-run toggle. See below | ## Decision 1: engine toggle, and why the shipped default is not Fusion yet @@ -1228,13 +1229,343 @@ block, since that is what `dbt_run_state` saves and `invocation_args` publishes. without failing. Overriding this would make the same project behave differently on Windmill than locally, breaking the core promise. +## Durable state per environment, and what defers to it + +`dbt --defer --state ` resolves a `ref()` the run does not build to the +relation the manifest in `` names, instead of to the schema this run writes +into. That is what lets one model be rebuilt into a scratch schema without +rebuilding everything above it, and it needs a manifest of the environment the +project actually lives in. + +Nothing that already existed could supply one. `dbt_run_state` answers a +different question — it holds the LAST run whatever its outcome, keyed by the +principal, so `dbt retry` can resume its failures — and the worker-local +generations behind it are a cache: the next run of a project usually lands on a +worker holding neither artifact. So the state is its own table, +`dbt_environment_state`, one row per (workspace, script path, environment), +holding `manifest.json` and `run_results.json` from the last SUCCESSFUL run. +Success is half of the contract: a relation a later run defers to has to exist. + +### The environment is the warehouse, the target and where they resolve to + +The workspace warehouse's name, the target dbt actually runs, and the database +and schema that target resolves to — the pair `relation_root` reports to the +graph's drift check. Each component is length-prefixed rather than joined on a +separator — `|||`, each written `:`, +so `main`/`prod`/`analytics`/`dbt_wh_defer` is stored as +`4:main|4:prod|9:analytics|12:dbt_wh_defer`. A target name and a schema are both +the user's own strings, so `prod|analytics` + `scratch` and `prod` + +`analytics|scratch` would otherwise be one key, and a profile moving between them +would read as the same environment rather than as one nothing has published. What +a message names is spelled out instead, never the encoded key. + +The target is the EFFECTIVE one, not the descriptor's `profile.target`: a +descriptor naming none inherits the workspace warehouse's, or the default in the +project's own `profiles.yml`, so reading the descriptor's would file every +inherited target under one empty name — and a `target.name` macro decides where a +model is built. + +The last two are in the key because deferring is resolving a relation NAME. A +warehouse repointed at another database, or a `profile.schema` moved by a +redeploy, keeps the first two while putting every relation somewhere else, and a +manifest is a list of relation names — there is no other way to notice. Keyed on +the first two alone, such a move would hand the next deferring run the names of +relations that are no longer there. Keyed on all four, it reads as an +environment nothing has published yet, which is what it is. + +What the key deliberately does NOT carry is the resolved connection. That is the +`profile_digest` a retry is held to, and it moves when a password is rotated, +which moves no relation; a warehouse pointing somewhere else entirely is +decision 11's accepted limitation, spelled the same way here as everywhere else. + +Today one script has one environment, because a descriptor fixes both the +warehouse and the target and a run cannot override either. The key is what makes +the *later* item — fork and preview environments — an addition rather than a +migration, and what makes a profile move detectable now. + +### Which runs publish it + +A successful `build` that did not itself defer, and whose graph becomes what the +script owns (`GraphRefresh::publishes_ownership`) — the same condition as the +graph's and the same reason: an invocation that scoped its own model set — a +`vars` or `select` override, or a descriptor dynamic by construction — describes +where the CALLER put those relations, not where this project's models live. +Publishing it would point every later deferral at one caller's scratch schema. + +**A run that deferred never publishes, whatever narrowed it**, and that is a +separate condition rather than a consequence of the first. A deferring run built +some of the relations its manifest names and resolved the rest out of the state +it read, so recording that manifest would claim relations nothing built — and a +model renamed since would be recorded under a name only a full build creates, +breaking every later deferral until one repairs it. `publishes_ownership` cannot +see this: it reads the caller's overrides, and a descriptor that already narrows +`select` needs none. + +A `retry` publishes nothing. Its `run_results.json` names only the nodes it +redid, so the environment would come to claim a run of a handful of models. The +environment's state is therefore the last full successful build, exactly as dbt +Cloud's "last successful run" is, and a run recovered by a retry leaves it at +the previous one. + +The AUTOMATIC in-job node retry is the same artifact under a different name: a +build it recovers is a successful build, but the `run_results.json` on disk is +the retry's. Such a run publishes the manifest **without** results, rather than +with a set describing some other slice of the build — the manifest is a function +of the project rather than of what ran, so deferral is unaffected. A `result:` +selector is the one thing left with nothing to read, and it is refused by name +against such a publication rather than passed to dbt (see "Selectors that read +the state" below). + +Under `test_behavior: after_all` the stored `run_results.json` is the test +phase's, because that is what the second invocation leaves in the target +directory — the same artifact a local `dbt run && dbt test` leaves behind. + +**What that condition means for what the artifacts may carry**, and why this +table is keyed by environment where `dbt_run_state` is keyed by principal. dbt +records the invocation's flags into `run_results.json`, and Windmill resolves +`$var:` / `$res:` references before dbt sees them — which is exactly why the +retry state is per-principal, so one caller's resolved `select` and `vars` are +not restorable by the next. Here they cannot be one caller's: a publishing run +added nothing of its own, and a descriptor that interpolates a `{{ }}` +placeholder into `vars` never publishes at all, so what is recorded is the +descriptor's own arguments — the script's content, which anyone entitled to run +it may already read. Widen the publish condition and that stops being true. + +### Where the blob goes + +`run_results.json` is small; `manifest.json` is not, and grows with the project +(535 KB on a two-model fixture). Each takes the same two homes: inline in the row +under `DBT_STATE_INLINE_MAX_BYTES` (8 MiB), and the INSTANCE's object storage +above it, with the row keeping the key. Inline is what makes the feature work on +an instance that has configured no storage at all; the ceiling is what stops one +project's manifest from becoming a multi-megabyte row rewritten by every run. A +project past the ceiling with no storage configured is told so, in the job log, +naming the setting and the variable — the run itself still succeeds, since +losing the state costs the next deferral rather than the build that just ran. + +**The instance store, not the workspace's**, which is where every other internal +worker artifact already lives (bun bundles, python wheels, job logs, the global +cache). The workspace bucket is the one members read and write through +`job_helpers/*` and `wmill.write_s3_file` with a caller-supplied key, and only +`volumes/` is reserved there — so a manifest under it is one any member could +replace, and the next deferring run would hand dbt an attacker-chosen +`defer_relation` for every unbuilt `ref()` while holding the script's warehouse +credentials. Its compiled SQL would be readable there too, for a project the +reader may have no access to. The consequence to know: a project past the ceiling +needs the instance store configured, which is an EE feature, so on CE the ceiling +is the limit and `DBT_STATE_INLINE_MAX_BYTES` is how it moves. + +Each publication writes its OWN keys +(`wmill_dbt_state///./`) +and the row switches to them in one statement, so an upload never overwrites an +artifact the committed row still names: a run that fails between its two uploads, +or between them and its row, leaves the state pointing at the pair it already +had. The objects the commit displaced are dropped afterwards, never before, since +a reader that has already read the row is about to fetch them; a reader that +loses that race re-reads for as long as the row keeps MOVING, rather than +reporting a state that is there. A reader takes no lock, so successive +publications can each overtake one; an unmoved row whose objects are gone is the +error that means what it says, and a bound on the re-reads is the other, for a +project republishing faster than a run can read. What a publication uploaded and then could not commit is dropped on the +way out — except after a commit that REPORTED an error, where what was lost may +be only the acknowledgement: dropping then would leave a committed row naming +objects that are gone, so an orphan is the cheaper side to take. + +The path and the environment are only a prefix of that key. The row is what says +where an artifact is, which is why state can travel with a renamed script and go +on naming objects under the old path's digest. The rest of the key is the job and +a per-EXECUTION nonce — zombie recovery re-runs a job under its own id, so keyed +on that alone a second attempt would overwrite the objects the first attempt's +committed row still names, then read those keys back as displaced and drop them. + +Publishers of one environment serialize on `pg_advisory_xact_lock`, so only one +of them settles the row and the objects it displaces at a time — an advisory lock +rather than the row's, because the first publish of an environment has no row to +lock and is exactly when two runs of a newly deployed script are most likely to +race. + +### Retention + +None, deliberately, and this is where it differs from the graph tables next +door. Those are pruned by age by the dbt runs themselves because their reader is +a transient run page. This one holds a single row per script per environment, +replaced in place, so it does not grow with runs — and its reader is every later +run of that script, so a project that runs monthly must still find last month's +state. It goes with the script instead: a path no live dbt version occupies any +more clears it, alongside `dbt_run_state` (`clear_dbt_script_state`, +`clear_dbt_script_state_if_path_retired`). + +The write carries a guard of its own, and it names the VERSION rather than the +path: the live dbt script there must be the one this job ran, or a later version +of it (`hash = $n OR $n = ANY(parent_hashes)`). "Some live dbt script is here" — +which is what the retry state settles for — is also satisfied by a script created +at a path this one was renamed away from, and this job's manifest would then +become that project's deferral state. A preview names no version and so publishes +nothing, which is right for a run of content that was never deployed. + +The job's KIND is checked beside it, because a preview carries a caller-supplied +`script_hash` into `runnable_id` (`run_preview_script`): the version alone would +let anyone who may run a job publish arbitrary content as a deployed script's +state. A flow or app step naming a deployed dbt script by path is an ordinary +`script` job carrying that script's own hash, so it publishes like any other run; +only INLINE flow code is a `FlowScript`, and that has no deployed version to +publish for. + +That guard HOLDS the script row (`FOR SHARE`) for the rest of the publication, so +a rename, archive or delete of the path either waits for it or is seen by it. +Read unlocked, it leaves a window where the lifecycle clear finds no row to take, +finishes, and the publication then commits state at a path a new script goes on +to occupy. The script row is taken before the sidecar, which is the order every +other dbt writer takes and what keeps the two off a deadlock. + +An artifact too large for its row is left in the store when the row is cleared, +as a deleted script leaves its bundle: reaching it from the delete would mean an +object-store client in `windmill-common` and a delete that has to land after the +caller's transaction commits, for one object per environment of a script that is +gone. + +### Asking for it + +`defer` is a field on the `build` command block, defaulting to the descriptor's +own `defer:`. A per-run toggle rather than a descriptor-only setting, because the +run that publishes an environment's state and the run that defers to it are two +invocations of ONE script (decision 6: N scripts means N projects): a project +that could only defer by descriptor could never populate the state it reads. + +A project whose profile selects its schema or database with a TEMPLATE — either +delimiter, since dbt renders `{% … %}` blocks as well as `{{ … }}` — is refused a +deferral outright, and publishes no state either: dbt renders those and Windmill +does not, so two renderings resolve to one `relation_root`, and a +deferral after the value changed would resolve every unbuilt `ref()` through the +previous location's manifest. Both sides, because a published template would sit +under a key a literal profile shares, and de-templating later would make that +stale manifest readable as the new location's. It covers a project-owned +`profiles.yml`, a `dbt_profile` resource — one block of the user's own file, +copied through unchanged — and a `profile.schema` written as given. Plainly +absent is different: that is the adapter's default, which does not move. + +A run that asks to defer with nothing published is refused, naming the +environment and the runs that cannot publish one. The alternative — running +without deferral — fails deep inside dbt with a relation-not-found the caller has +no way to connect back to a missing state. An agent worker is refused the same +way and for a reason it can act on: it reaches the database only through the API, +which does not expose this table. + +A `show` defers too, and every engine takes the flags on it. It compiles the +model it previews, so a model whose upstream this environment built and this run +did not is exactly the case a deferral exists for. So does the `dbt ls` that +resolves what a run's selection owns, without which a `result:` selector — which +reads `run_results.json` out of the state directory, and which `select` passes to +dbt verbatim — would fail before the build that would have honoured it. + +The result carries `deferred_to`, the run whose state was used. Without it what +a deferring run built against is unrecoverable, since the next successful run of +that environment replaces the state. + +### Selectors that read the state, and why they are refused rather than passed + +`--state` also feeds dbt's own selector methods, so publishing the state is what +makes `state:modified+`, `state:new` and `result:error+` resolve at all. Only a +deferring run is handed the directory, so a `state:` or `result:` method in +`select` or `exclude` without `defer` is refused before dbt starts. + +Refused, rather than left to dbt, because the engines disagree about it and two +of the three disagree silently. Given a state selector and no `--state`, +dbt-core 1.x raises (`Got a state selector method, but no comparison manifest`, +exit 2), but dbt-sa-cli 2.x and fusion read a MISSING state as an EMPTY one and +exit 0: `state:modified` then selects nothing and the run reports success having +built nothing, while `state:new` selects everything, because against an empty +state every node is new. A scheduled run that quietly stops doing work, or +quietly rebuilds the project, is the failure this state exists to prevent. + +From the DESCRIPTOR they are refused whether or not the run defers, and the +message says so. That selection is also what decides which nodes the script owns, +and the deploy resolves it before any run exists, with no state to compare +against. "Whatever changed last" is not an ownership answer. They describe one +run, so they belong in a run's own `select`. + +`source_status:` is refused under any setting: it compares `sources.json`, which +`dbt source freshness` writes and no run publishes here, so there is nothing to +compare against even while deferring. + +Two more refusals follow from the same argument, that a selector with nothing to +read must say so rather than resolve to a silent answer: + +- A `result:` method while deferring to a state that carries **no** + `run_results.json`. Publishing that is deliberate — a build recovered by + automatic node retry stores the manifest alone, its results describing the + retried nodes rather than the build ("Which runs publish it") — so `defer` + being on is not enough to know the file is there. Answerable only once the + state is loaded, so it is checked right after, naming the run that published. +- Any of them on a `parse`. A parse resolves a selection to store the graph and + never defers, so `defer` would not hand it a state at any setting, and the + remedy the other refusal offers would lead nowhere. It says that instead. + +Matching nothing is then an ordinary outcome for these methods, and for no +others. `state:modified+` selects the empty set exactly when nothing changed +since the published state, which is the answer a CI run wants, so a selection +naming a `state:` or `result:` method may resolve to no nodes. Such a run scoped +its own selection, so what it stores is a snapshot of its own and never what the +script owns, and nothing is un-wired by the empty set. + +The exemption is by METHOD, not by who chose the selection. Exempting every +caller-chosen one would take a misspelled model name, which resolves to nothing +just as surely, and report it as a build that did its work. An ordinary selection +matching nothing stays refused, from a run as from the descriptor — from the +descriptor because that one also decides ownership. + +Only what `select` and `exclude` spell directly. A method reached through a +`selectors.yml` definition is named nowhere the worker reads, and dbt's own +behaviour — including the silent one — is what stands there. + +### `--state` is also a retry's own argument, and that is a trap + +`dbt retry` reads the run it RESUMES from `--state`. Handed the deferral's +directory it resumes the successful run stored there, finds nothing failed, and +reports a green retry having rebuilt nothing — silently, on dbt-core 1.x, which +warns and exits 0. + +dbt-core 1.x has `--defer-state`, the deferral-only half of the pair, so a retry +there passes that and leaves `--state` alone. The Rust engines do not have it, +and a run that deferred is refused a retry on them, before the build: the +alternative is rebuilding the failed nodes with every `ref()` resolving into the +schema this run writes into, which for the narrowed run a deferral exists to +serve means writing them somewhere they do not belong. The automatic in-job node +retry is dropped for the same reason and says so in the log. + +The state directory is passed RELATIVE (`wm_dbt_state`, beside `wm_target` in the +job directory). dbt records the invocation's flags into `run_results.json` and a +later `dbt retry` restores them, so an absolute path would name the job directory +of the run being resumed, which is gone by then. Relative, it resolves against +the project root — whichever job directory the retry landed in. + +Three engine facts found while wiring this up, all worth knowing before filing a +bug against the feature. `dbt retry` on dbt-core 2.x restores **neither** the +resumed invocation's `--vars` nor its deferral: it re-parses with the current +(empty) ones, so a retry of a run that overrode `vars` rebuilds into the +descriptor's schema rather than the run's. That is independent of deferral and +predates it; the refusal above stops the deferring case from being the way it is +discovered. `dbt show` on either Rust engine prints a bare JSON array where +dbt-core frames it as `{"node": …, "show": […]}`, which `run_show` is written +against — so a preview there fails to parse whether or not it defers, and the +deferral itself resolves correctly under it. And neither Rust engine reached +dbt's own service-backed State (`--manage-state`) on any run measured here, so no +flag is passed to disable it. + +Because `select` reaches dbt verbatim, a deferring run also has a `--state` +directory for `result:` selectors, which is why `run_results.json` is stored +beside the manifest rather than the manifest alone. + ## Two decisions the implementation narrowed -**Decision 13 — no S3 copy of the manifest.** The sidecar holds every field the -graph renders; nothing reads a stored `manifest.json`, so writing one to S3 -would be an unread copy of data that is already reproducible by redeploying (or, -for a dynamic descriptor, by the next run). Worth adding the day something needs the -parts the sidecar drops — compiled SQL, macro definitions — and not before. +**Decision 13 — the manifest is stored once per environment, not per version.** +The sidecar holds every field the graph renders, so a copy of `manifest.json` +bought the graph nothing: it is reproducible by redeploying, or for a dynamic +descriptor by the next run. Deferral is the reader that changed that — it +resolves an unbuilt `ref()` through a manifest, and one on worker-local disk +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 @@ -1280,7 +1611,10 @@ render through the existing `RunnableNode.svelte` / `AssetNode.svelte` / on the canvas mid-run. `record_materialization` per model. Profile and select pickers in the editor. Per-model failure triage in the run view. -**Phase 4 (not in this PR).** `--defer` and `state:modified`. Partition and +**Phase 4 (not in this PR).** Slim CI: the fork and preview environments a +deferral would name instead of its own. The selectors themselves are here, since +`state:` and `result:` read the published state like any deferral does; what is +missing is a per-branch environment to compare a CI run against. Partition and backfill integration so `BackfillRangeDialog.svelte` works on dbt models. `wmill dbt import ` reading `DbtDag(...)` kwargs. @@ -1317,6 +1651,16 @@ Against a real dbt project (jaffle_shop shape) and the local Postgres: own `profiles.yml` with env-var injection. 11. **Caching**: a second run reuses the cached `dbt_packages/` with no network fetch. +12. **Deferral**: a full run publishes the environment's state; a second run + that builds one downstream model into another schema resolves its unbuilt + `ref()` to the relation the state names, where the same run without `defer` + fails with relation-not-found. +13. **State selectors**: with a state published, `state:modified+` selects + nothing while the project is unchanged and exactly the changed model and its + children after one is edited. Without `defer` it is refused rather than + passed, and a `result:` selector against a state published by a + node-retry-recovered build is refused too, that one carrying no + `run_results.json`. Keep only tests that pin behavior a future change could break. Per AGENTS.md, delete development scaffolding before marking the PR ready. diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index 76ecfa59d8..c253bc64e7 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -1370,6 +1370,11 @@ test_behavior: build vars: {} threads: 4 full_refresh: false +# Resolve a ref() this run does not build through the state the last successful +# run of this environment published, instead of through the schema it writes +# into. The default for the run form's toggle: the run that publishes the state +# and the run that defers to it are two invocations of this one script. +defer: false # Rebuild the nodes a failed build left failed or skipped, in this same job, # before reporting failure. dbt confines a failure to its own subtree, so a # transient warehouse error costs those nodes rather than the whole project.