diff --git a/backend/.sqlx/query-6bdddcaa371df626c17b4d70d5e1b7a09c4c4ed7435fa2b23aa1a95acf0476e0.json b/backend/.sqlx/query-63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b.json similarity index 90% rename from backend/.sqlx/query-6bdddcaa371df626c17b4d70d5e1b7a09c4c4ed7435fa2b23aa1a95acf0476e0.json rename to backend/.sqlx/query-63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b.json index 73b5db5e4c..82af09d6a6 100644 --- a/backend/.sqlx/query-6bdddcaa371df626c17b4d70d5e1b7a09c4c4ed7435fa2b23aa1a95acf0476e0.json +++ b/backend/.sqlx/query-63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "SELECT s.path AS \"path!\", s.language AS \"language!: ScriptLang\"\n FROM asset a\n JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path\n AND s.archived = false AND s.deleted = false\n WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2\n AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')", + "query": "SELECT s.path AS \"path!\", s.language AS \"language!: ScriptLang\"\n FROM asset a\n JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path\n AND s.archived = false AND s.deleted = false\n WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2\n AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')\n AND a.usage_path <> $3", "describe": { "columns": [ { @@ -49,6 +49,7 @@ ], "parameters": { "Left": [ + "Text", "Text", "Text" ] @@ -58,5 +59,5 @@ false ] }, - "hash": "6bdddcaa371df626c17b4d70d5e1b7a09c4c4ed7435fa2b23aa1a95acf0476e0" + "hash": "63c0b60d1042ef74e69b11119355c26d8ca6e5e94902450c11ab7816a385fd9b" } diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 7211b8299d..27583726fd 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -29,10 +29,10 @@ pub enum AssetKind { Ducklake, DataTable, Volume, - /// A warehouse relation a dbt project builds or reads, - /// `dbt:////`, the warehouse named as the - /// workspace configures it. The scheme names the producer, the path stays - /// the relation — see `windmill_types::AssetKind::Dbt`. + /// A warehouse relation, `dbt:////`, the warehouse + /// named as the workspace configures it. The scheme names the namespace dbt + /// made — a script of any language can declare a write to one — and the path + /// stays the relation. See `windmill_types::AssetKind::Dbt`. Dbt, } @@ -1742,10 +1742,7 @@ mod pipeline_annotation_tests { // just stop being the same node and the cross-boundary cascade never fires. #[test] fn table_paths_from_every_spelling_canonicalize_to_one_key() { - let canonical = Some(( - AssetKind::Dbt, - Cow::Owned("main/analytics/orders".into()), - )); + let canonical = Some((AssetKind::Dbt, Cow::Owned("main/analytics/orders".into()))); for spelling in [ // Hand-written annotation. "dbt://main/analytics/orders", @@ -1793,10 +1790,7 @@ mod pipeline_annotation_tests { // database qualifier. assert_eq!( parse_asset_syntax("dbt://main/\"sales.v2\"/orders", false), - Some(( - AssetKind::Dbt, - Cow::Owned("main/sales.v2/orders".into()) - )) + Some((AssetKind::Dbt, Cow::Owned("main/sales.v2/orders".into()))) ); } @@ -1815,14 +1809,8 @@ mod pipeline_annotation_tests { "dbt://main/analytics/\"order\"\"s\"", "main/analytics/order\"s", ), - ( - "dbt://main/`da``ta`/`orders`", - "main/da`ta/orders", - ), - ( - "dbt://main/[my]]schema]/[orders]", - "main/my]schema/orders", - ), + ("dbt://main/`da``ta`/`orders`", "main/da`ta/orders"), + ("dbt://main/[my]]schema]/[orders]", "main/my]schema/orders"), // And in one half of a database-qualified segment. ( "dbt://main/\"arch\"\"ive\".\"sales\"/orders", @@ -1841,8 +1829,14 @@ mod pipeline_annotation_tests { // apart. A lone delimiter treated as opening a quote would be dropped — // `sa"les` filed as `sales` — and the two derivations would split. for (decoded, spelled) in [ - ("dbt://main/sa\"les/orders", "dbt://main/\"sa\"\"les\"/orders"), - ("dbt://main/analytics/order\"s", "dbt://main/analytics/\"order\"\"s\""), + ( + "dbt://main/sa\"les/orders", + "dbt://main/\"sa\"\"les\"/orders", + ), + ( + "dbt://main/analytics/order\"s", + "dbt://main/analytics/\"order\"\"s\"", + ), ( "dbt://main/arch\"ive.sales/orders", "dbt://main/\"arch\"\"ive\".\"sales\"/orders", diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 5ae7bc5cc3..ee2043fe97 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -2441,7 +2441,8 @@ async fn create_script_internal<'c>( ))); } if let Some(dbt_owner) = - windmill_common::assets::sole_dbt_producer(&db, &w_id, relation).await? + windmill_common::assets::sole_dbt_producer(&db, &w_id, relation, &ns.path) + .await? { return Err(Error::BadRequest(format!( "`{trigger_ref}` cannot be subscribed to: it is built by the dbt project at \ diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 114f24f9e6..96a7573620 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -258,13 +258,20 @@ pub fn derive_pipeline_asset_trigger_refs( /// for every `dbt://` node anyway (the source that script wrote stays gated). /// Callers must therefore already be scoped to `workspace_id`. /// -/// Reads committed rows only, so a caller inside a deploy transaction does not -/// see its own not-yet-committed writes: a script subscribing to a relation it -/// also materializes is left to the dispatcher's self-loop skip. +/// `subscriber_path` is excluded from the producer set, and has to be: reading +/// committed rows means the deploying script's own are the version being +/// replaced, so one that just dropped its `// materialize` would still count as a +/// producer and let a now-dormant subscription through. Excluding it is free of +/// the opposite error, because a script never wakes its own subscription — the +/// dispatcher skips that as a self-loop. +/// +/// A producer another deploy is committing concurrently is still invisible, so +/// that race resolves toward refusing with a message the user can retry past. pub async fn sole_dbt_producer<'e>( executor: impl PgExecutor<'e>, workspace_id: &str, asset_path: &str, + subscriber_path: &str, ) -> error::Result> { use crate::scripts::ScriptLang; let producers = sqlx::query!( @@ -273,9 +280,11 @@ pub async fn sole_dbt_producer<'e>( JOIN script s ON s.workspace_id = a.workspace_id AND s.path = a.usage_path AND s.archived = false AND s.deleted = false WHERE a.workspace_id = $1 AND a.kind = 'dbt' AND a.path = $2 - AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw')"#, + AND a.usage_kind = 'script' AND a.usage_access_type IN ('w', 'rw') + AND a.usage_path <> $3"#, workspace_id, - asset_path + asset_path, + subscriber_path ) .fetch_all(executor) .await?; diff --git a/backend/windmill-common/tests/dbt_producer_rules.rs b/backend/windmill-common/tests/dbt_producer_rules.rs index 07484c131f..6b57f4e4b4 100644 --- a/backend/windmill-common/tests/dbt_producer_rules.rs +++ b/backend/windmill-common/tests/dbt_producer_rules.rs @@ -15,6 +15,7 @@ use windmill_common::assets::{dormant_dbt_subscriptions, sole_dbt_producer}; const WS: &str = "test-workspace"; const RELATION: &str = "main/analytics/orders"; +const SUBSCRIBER: &str = "u/test-user/consumer"; async fn plant_producer(db: &Pool, path: &str, language: &str, hash: i64) { sqlx::query( @@ -58,7 +59,9 @@ async fn plant_subscriber(db: &Pool, path: &str) { #[sqlx::test(migrations = "../migrations", fixtures("base"))] async fn no_producer_is_not_dormant(db: Pool) { assert_eq!( - sole_dbt_producer(&db, WS, RELATION).await.unwrap(), + sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + .await + .unwrap(), None, "a relation nothing produces yet must not refuse the subscription" ); @@ -68,7 +71,9 @@ async fn no_producer_is_not_dormant(db: Pool) { async fn dbt_only_producer_is_dormant(db: Pool) { plant_producer(&db, "u/test-user/project", "dbt", 1).await; assert_eq!( - sole_dbt_producer(&db, WS, RELATION).await.unwrap(), + sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + .await + .unwrap(), Some("u/test-user/project".to_string()) ); } @@ -77,7 +82,12 @@ async fn dbt_only_producer_is_dormant(db: Pool) { async fn a_native_producer_beside_dbt_is_not_dormant(db: Pool) { plant_producer(&db, "u/test-user/project", "dbt", 1).await; plant_producer(&db, "u/test-user/ingest", "postgresql", 2).await; - assert_eq!(sole_dbt_producer(&db, WS, RELATION).await.unwrap(), None); + assert_eq!( + sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + .await + .unwrap(), + None + ); } /// `asset` is keyed by path while `script` holds every version of it, so the @@ -92,7 +102,25 @@ async fn a_superseded_native_version_does_not_count(db: Pool) { .expect("archive the old version"); plant_producer(&db, "u/test-user/project", "dbt", 2).await; assert_eq!( - sole_dbt_producer(&db, WS, RELATION).await.unwrap(), + sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + .await + .unwrap(), + Some("u/test-user/project".to_string()) + ); +} + +/// The rows of the script being deployed describe the version it replaces, so a +/// script dropping its `// materialize` while adding a subscription would +/// otherwise count itself as the producer that wakes it — and commit a dormant +/// edge. It can never be that producer anyway: the dispatcher skips self-loops. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn the_subscriber_is_never_its_own_producer(db: Pool) { + plant_producer(&db, "u/test-user/project", "dbt", 1).await; + plant_producer(&db, SUBSCRIBER, "postgresql", 2).await; + assert_eq!( + sole_dbt_producer(&db, WS, RELATION, SUBSCRIBER) + .await + .unwrap(), Some("u/test-user/project".to_string()) ); } diff --git a/backend/windmill-types/src/assets.rs b/backend/windmill-types/src/assets.rs index 463a365b06..d5b5354755 100644 --- a/backend/windmill-types/src/assets.rs +++ b/backend/windmill-types/src/assets.rs @@ -14,17 +14,19 @@ pub enum AssetKind { Ducklake, DataTable, Volume, - /// A warehouse relation a dbt project builds or reads, - /// `dbt:////`, where `` is the name the - /// workspace configures it under. + /// A warehouse relation, `dbt:////`, where + /// `` is the name the workspace configures it under. /// - /// The SCHEME names the producer — dbt is the only thing that creates one — - /// while the PATH stays the physical relation, because that is what two - /// projects agree on: a mart one builds is a `source` the next reads, and - /// their dbt `unique_id`s differ (`model.a.orders` vs - /// `source.b.analytics.orders`) where the relation does not - /// (docs/dbt-runtime.md, decision 11). A dbt run does not trigger that - /// reader — the shared node is lineage, not a cascade edge. + /// The SCHEME names the namespace dbt made rather than an exclusive + /// producer: dbt is what derives these relations from a project, and a + /// script of any language can DECLARE one it writes + /// (`// materialize manual dbt://…`). The PATH stays the physical relation, + /// because that is what two producers agree on: a mart one builds is a + /// `source` the next reads, and their dbt `unique_id`s differ + /// (`model.a.orders` vs `source.b.analytics.orders`) where the relation does + /// not (docs/dbt-runtime.md, decision 11). A dbt run does not trigger the + /// readers of what it built — that shared node is lineage, not a cascade + /// edge — while a declared write does (decision 25). Dbt, } diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 449ea6fba7..ded2cd2a28 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -5583,6 +5583,15 @@ async fn record_declared_warehouse_write( use windmill_common::materialization::{ MaterializationStatus, RecordMaterializationRequest, UNPARTITIONED, }; + // A DEPLOYED script only. The annotation is a deploy-time contract — `manual`, + // a three-segment relation, a configured warehouse — checked in + // `create_script_internal`, which also required write access to the path. A + // preview, hub or inline-flow body reaches this function without any of that, + // so honouring it there would let `jobs:run` alone restamp any relation's last + // writer from a script that never touched it. + if job.kind != JobKind::Script { + return; + } // Cheap guard: the annotation scan is skipped for the overwhelming majority // of jobs, which carry no `materialize` line at all. if !code.contains("materialize") { diff --git a/docs/dbt-runtime.md b/docs/dbt-runtime.md index 22b6dcf3fc..54048d1fe8 100644 --- a/docs/dbt-runtime.md +++ b/docs/dbt-runtime.md @@ -715,11 +715,17 @@ deploy-order-independent syncs. A dbt script may not subscribe at all: its graph ingest clears its own `dbt://` trigger rows, so accepting one would deploy an edge the dependency job then silently removes. -That leaves one ordering the deploy cannot catch: a subscriber accepted while the -relation had no producer, and a dbt project deployed afterwards that claims it. So -a dbt deploy names those edges in its own log rather than leaving them silently -dormant — the same "an edge that can never fire is worse than saying so" the -refusal is for, at the only other point where it is knowable. +The producer set is read as it stands committed, minus the deploying script's own +rows — those describe the version being replaced, so a script dropping its +`// materialize` while adding a subscription would otherwise count itself as the +producer that wakes it, which it could not be anyway (the dispatcher skips +self-loops). + +What that leaves is a subscription accepted while it was live and later orphaned: +a dbt project deployed afterwards that claims the relation, or a native producer +that stops writing it. A dbt deploy names those edges in its own log rather than +leaving them silently dormant — the same "an edge that can never fire is worse +than saying so" the refusal is for, at the only other point where it is knowable. A plain READ still renders the consumer beside the model, which is what makes the lineage one graph — but it is written in the script's own code, not in a