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 <noreply@anthropic.com>

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"
 }

backend/migrations/20260516190545_add_script_trigger_join_all.down.sql

@@ -0,0 +1,2 @@
+ALTER TABLE script_trigger
+  DROP COLUMN join_all;

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;

backend/parsers/windmill-parser-py-asset/src/lib.rs

@@ -35,6 +35,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         triggers: pipeline.triggers,
         partition: pipeline.partition,
         freshness: pipeline.freshness,
+        join_mode: pipeline.join_mode,
         ..Default::default()
     })
 }

backend/parsers/windmill-parser-sql-asset/src/asset_parser.rs

@@ -40,6 +40,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         triggers: pipeline.triggers,
         partition: pipeline.partition,
         freshness: pipeline.freshness,
+        join_mode: pipeline.join_mode,
         ..Default::default()
     })
 }

backend/parsers/windmill-parser-ts-asset/src/lib.rs

@@ -46,6 +46,7 @@ pub fn parse_assets(code: &str) -> anyhow::Result<ParseAssetsOutput> {
         triggers: pipeline.triggers,
         partition: pipeline.partition,
         freshness: pipeline.freshness,
+        join_mode: pipeline.join_mode,
     })
 }
 

backend/parsers/windmill-parser-yaml/src/asset_parser.rs

@@ -47,6 +47,7 @@ pub fn parse_assets(input: &str) -> anyhow::Result<ParseAssetsOutput> {
         triggers: pipeline.triggers,
         partition: pipeline.partition,
         freshness: pipeline.freshness,
+        join_mode: pipeline.join_mode,
         ..Default::default()
     })
 }

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<FreshnessSpec>,
+    // `// 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 <asset>` 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<TriggerSpec>,
     pub partition: Option<PartitionSpec>,
     pub freshness: Option<FreshnessSpec>,
+    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\"";

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?;
     }

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?;