From 9e157868bf686da97201368d3e3e88e774a600de Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Sat, 16 May 2026 19:29:07 +0000 Subject: [PATCH] feat: parser join-mode (`// trigger all`) + script_trigger.join_all Stage A: JoinMode{Any(default),All} + `// trigger any|all` directive in parse_pipeline_annotations; TriggerSpec::is_partition_bearing() (path contains {partition}); join_mode threaded through all 4 asset-parser crates (ts/py/sql/yaml). Stage B: reversible migration adds script_trigger.join_all; insert_script_trigger writes it; deploy path sets it from the parsed annotation. No reader yet (AND-join dispatch is the next stage) so runtime behaviour is unchanged. Co-Authored-By: Claude Opus 4.7 --- ...7c736aa8a8eb2f1db500c8a4adfe39a26112.json} | 7 +- ...90545_add_script_trigger_join_all.down.sql | 2 + ...6190545_add_script_trigger_join_all.up.sql | 9 ++ .../windmill-parser-py-asset/src/lib.rs | 1 + .../src/asset_parser.rs | 1 + .../windmill-parser-ts-asset/src/lib.rs | 1 + .../windmill-parser-yaml/src/asset_parser.rs | 1 + .../windmill-parser/src/asset_parser.rs | 103 ++++++++++++++++++ backend/windmill-api-scripts/src/scripts.rs | 3 + backend/windmill-common/src/assets.rs | 9 +- 10 files changed, 132 insertions(+), 5 deletions(-) rename backend/.sqlx/{query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json => query-28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112.json} (80%) create mode 100644 backend/migrations/20260516190545_add_script_trigger_join_all.down.sql create mode 100644 backend/migrations/20260516190545_add_script_trigger_join_all.up.sql diff --git a/backend/.sqlx/query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json b/backend/.sqlx/query-28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112.json similarity index 80% rename from backend/.sqlx/query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json rename to backend/.sqlx/query-28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112.json index dece68de29..c10709ce12 100644 --- a/backend/.sqlx/query-7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b.json +++ b/backend/.sqlx/query-28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "INSERT INTO script_trigger\n (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref)\n VALUES ($1, $2, $3, $4, $5)", + "query": "INSERT INTO script_trigger\n (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, join_all)\n VALUES ($1, $2, $3, $4, $5, $6)", "describe": { "columns": [], "parameters": { @@ -38,10 +38,11 @@ } } }, - "Text" + "Text", + "Bool" ] }, "nullable": [] }, - "hash": "7cee1d98c68d900d57ac5539fd10f054270fe998411ffb93c7372e2fab2acd4b" + "hash": "28e4b886452f77af784e966649d07c736aa8a8eb2f1db500c8a4adfe39a26112" } diff --git a/backend/migrations/20260516190545_add_script_trigger_join_all.down.sql b/backend/migrations/20260516190545_add_script_trigger_join_all.down.sql new file mode 100644 index 0000000000..b69f7abca6 --- /dev/null +++ b/backend/migrations/20260516190545_add_script_trigger_join_all.down.sql @@ -0,0 +1,2 @@ +ALTER TABLE script_trigger + DROP COLUMN join_all; diff --git a/backend/migrations/20260516190545_add_script_trigger_join_all.up.sql b/backend/migrations/20260516190545_add_script_trigger_join_all.up.sql new file mode 100644 index 0000000000..b3aa5b31df --- /dev/null +++ b/backend/migrations/20260516190545_add_script_trigger_join_all.up.sql @@ -0,0 +1,9 @@ +-- AND-join barrier (`// trigger all`). When true, the subscriber runs only +-- once every partition-bearing input has materialized at the same partition +-- (plus every reference input exists), rather than firing on any input (OR, +-- the default). Stored per script_trigger row — it is a script-level +-- property so every row for a given (workspace, runnable) carries the same +-- value; this matches the wipe-and-reinsert-on-deploy pattern and keeps the +-- subscriber lookup a single query. +ALTER TABLE script_trigger + ADD COLUMN join_all BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/backend/parsers/windmill-parser-py-asset/src/lib.rs b/backend/parsers/windmill-parser-py-asset/src/lib.rs index fc046e03c3..4014f9e1c5 100644 --- a/backend/parsers/windmill-parser-py-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-py-asset/src/lib.rs @@ -35,6 +35,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result { triggers: pipeline.triggers, partition: pipeline.partition, freshness: pipeline.freshness, + join_mode: pipeline.join_mode, ..Default::default() }) } diff --git a/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs b/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs index 0c95c6ec01..4ab03c344b 100644 --- a/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs @@ -40,6 +40,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result { triggers: pipeline.triggers, partition: pipeline.partition, freshness: pipeline.freshness, + join_mode: pipeline.join_mode, ..Default::default() }) } diff --git a/backend/parsers/windmill-parser-ts-asset/src/lib.rs b/backend/parsers/windmill-parser-ts-asset/src/lib.rs index 61834c451e..639c8cf147 100644 --- a/backend/parsers/windmill-parser-ts-asset/src/lib.rs +++ b/backend/parsers/windmill-parser-ts-asset/src/lib.rs @@ -46,6 +46,7 @@ pub fn parse_assets(code: &str) -> anyhow::Result { triggers: pipeline.triggers, partition: pipeline.partition, freshness: pipeline.freshness, + join_mode: pipeline.join_mode, }) } diff --git a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs index 1f4a7b4156..f7b757d184 100644 --- a/backend/parsers/windmill-parser-yaml/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-yaml/src/asset_parser.rs @@ -47,6 +47,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result { triggers: pipeline.triggers, partition: pipeline.partition, freshness: pipeline.freshness, + join_mode: pipeline.join_mode, ..Default::default() }) } diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 221543b59a..f63ebfa5cb 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -83,6 +83,11 @@ pub struct ParseAssetsOutput { // applies regardless of which trigger last fired. #[serde(skip_serializing_if = "Option::is_none", default)] pub freshness: Option, + // `// trigger all` → AND join barrier; default (`any`) = OR (current + // behaviour). Threaded to the deploy path which persists it on the + // subscriber's trigger rows. + #[serde(skip_serializing_if = "JoinMode::is_any", default)] + pub join_mode: JoinMode, } #[derive(Serialize, Debug, PartialEq, Clone)] @@ -109,6 +114,17 @@ pub enum TriggerSpec { Gcp { path: String }, } +impl TriggerSpec { + // A `// on ` whose declared path contains the `{partition}` + // token is *partition-bearing*: in an AND join its concrete partition + // value is the join key. Non-asset triggers and assets without the + // token are reference/presence-only inputs that never define the + // partition (the case-3 guard). + pub fn is_partition_bearing(&self) -> bool { + matches!(self, TriggerSpec::Asset { path, .. } if path.contains(PARTITION_TOKEN)) + } +} + // Partitioning declaration for a pipeline script. `daily`/`hourly`/`weekly`/ // `monthly` are time-based with the runtime supplying the current // partition value derived from the trigger context (schedule fire time, @@ -147,6 +163,24 @@ pub struct FreshnessSpec { pub duration: String, } +// `// trigger any` (default) vs `// trigger all`. `Any` = OR: any trigger +// firing runs the script (current behaviour). `All` = AND: the script +// runs only once every partition-bearing input has materialized at the +// same partition (plus every reference input exists) — the join barrier. +#[derive(Serialize, Debug, PartialEq, Eq, Clone, Copy, Default)] +#[serde(rename_all = "lowercase")] +pub enum JoinMode { + #[default] + Any, + All, +} + +impl JoinMode { + pub fn is_any(&self) -> bool { + matches!(self, JoinMode::Any) + } +} + // All pipeline-level annotations parsed off a script's source. Returned by // `parse_pipeline_annotations` and forwarded into `ParseAssetsOutput`. #[derive(Default, Debug, PartialEq, Clone)] @@ -155,6 +189,7 @@ pub struct PipelineAnnotations { pub triggers: Vec, pub partition: Option, pub freshness: Option, + pub join_mode: JoinMode, } #[derive(Debug, Clone, Serialize)] @@ -412,6 +447,19 @@ pub fn parse_pipeline_annotations(code: &str) -> PipelineAnnotations { continue; } + if let Some(after_kw) = rest.strip_prefix("trigger") { + if !after_kw.starts_with(|c: char| c.is_whitespace()) { + continue; + } + match after_kw.trim() { + "all" => out.join_mode = JoinMode::All, + "any" => out.join_mode = JoinMode::Any, + // Unknown value — leave the default rather than guess. + _ => {} + } + continue; + } + if let Some(after_kw) = rest.strip_prefix("on") { if !after_kw.starts_with(|c: char| c.is_whitespace()) { continue; @@ -587,6 +635,61 @@ mod pipeline_annotation_tests { assert!(out.triggers.is_empty()); } + #[test] + fn join_mode_defaults_to_any() { + let out = parse_pipeline_annotations("// on s3://a/b"); + assert_eq!(out.join_mode, JoinMode::Any); + assert!(out.join_mode.is_any()); + } + + #[test] + fn trigger_all_sets_and_mode() { + let out = parse_pipeline_annotations( + "// pipeline\n// on s3://lake/{partition}/x\n// trigger all", + ); + assert_eq!(out.join_mode, JoinMode::All); + assert!(!out.join_mode.is_any()); + } + + #[test] + fn trigger_any_explicit_and_other_prefixes() { + // explicit `any`, plus `#` / `--` comment prefixes are accepted. + assert_eq!( + parse_pipeline_annotations("# trigger all\n-- trigger any").join_mode, + JoinMode::Any + ); + assert_eq!( + parse_pipeline_annotations("-- trigger all").join_mode, + JoinMode::All + ); + } + + #[test] + fn trigger_unknown_or_glued_keeps_default() { + // unknown value, and `triggerall` (no whitespace) must not match. + assert_eq!( + parse_pipeline_annotations("// trigger bogus").join_mode, + JoinMode::Any + ); + assert_eq!( + parse_pipeline_annotations("// triggerall").join_mode, + JoinMode::Any + ); + } + + #[test] + fn is_partition_bearing_only_for_tokened_assets() { + let out = parse_pipeline_annotations( + "// on s3://lake/raw/{partition}/events.parquet\n\ + // on s3://lake/dim/customers.parquet\n\ + // schedule \"@daily\"", + ); + assert_eq!(out.triggers.len(), 3); + assert!(out.triggers[0].is_partition_bearing()); + assert!(!out.triggers[1].is_partition_bearing()); + assert!(!out.triggers[2].is_partition_bearing()); + } + #[test] fn partitioned_daily() { let code = "// partitioned daily tz=\"UTC\" format=\"YYYY-MM-DD\" start=\"2024-01-01\""; diff --git a/backend/windmill-api-scripts/src/scripts.rs b/backend/windmill-api-scripts/src/scripts.rs index 958df312b2..c07ad22e7f 100644 --- a/backend/windmill-api-scripts/src/scripts.rs +++ b/backend/windmill-api-scripts/src/scripts.rs @@ -1240,6 +1240,8 @@ async fn create_script_internal<'c>( // them in auto_kind itself). let pipeline_annotations = parse_pipeline_annotations(&ns.content); let in_pipeline = pipeline_annotations.in_pipeline; + // `// trigger all` → AND join barrier (else OR, the default). + let pipeline_join_all = !pipeline_annotations.join_mode.is_any(); let pipeline_triggers = pipeline_annotations.triggers; let auto_kind = if in_pipeline { Some("pipeline".to_string()) @@ -1566,6 +1568,7 @@ async fn create_script_internal<'c>( &ns.path, trigger_kind, &trigger_ref, + pipeline_join_all, ) .await?; } diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 37b8189420..daaeda87fc 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -196,6 +196,9 @@ pub async fn delete_managed_pipeline_schedule<'e>( } // Insert a single trigger declaration. Caller is expected to wipe first. +// `join_all` is the script-level `// trigger all` flag (AND join barrier); +// it is the same for every row of a given runnable but stored per-row to +// keep the wipe-and-reinsert pattern and a single-query subscriber lookup. pub async fn insert_script_trigger<'e>( executor: impl PgExecutor<'e>, workspace_id: &str, @@ -203,16 +206,18 @@ pub async fn insert_script_trigger<'e>( runnable_path: &str, trigger_kind: ScriptTriggerKind, trigger_ref: &str, + join_all: bool, ) -> error::Result<()> { sqlx::query!( r#"INSERT INTO script_trigger - (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref) - VALUES ($1, $2, $3, $4, $5)"#, + (workspace_id, runnable_kind, runnable_path, trigger_kind, trigger_ref, join_all) + VALUES ($1, $2, $3, $4, $5, $6)"#, workspace_id, runnable_kind as AssetUsageKind, runnable_path, trigger_kind as ScriptTriggerKind, trigger_ref, + join_all, ) .execute(executor) .await?;