fix(backend): put for loop itered in a separate table (#7419)

* fix(backend): put for loop itered in a separate table

* Update SQLx metadata

---------

Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

backend/.sqlx/query-389828f43e638c02757ba37da46b03111a9915a16b53f3e29a09de89210d6af1.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "INSERT INTO flow_iterator_data (job_id, itered) VALUES ($1, $2)\n             ON CONFLICT (job_id) DO UPDATE SET itered = $2",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Jsonb"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "389828f43e638c02757ba37da46b03111a9915a16b53f3e29a09de89210d6af1"
+}

backend/.sqlx/query-3c5165992c4b8ad3f91627d1d9f6156d3a6b45a7cb2b37a7c166d36d7caa4d2f.json

@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT itered as \"itered: Json<Vec<Box<RawValue>>>\" FROM flow_iterator_data WHERE job_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "itered: Json<Vec<Box<RawValue>>>",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Uuid"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "3c5165992c4b8ad3f91627d1d9f6156d3a6b45a7cb2b37a7c166d36d7caa4d2f"
+}

backend/migrations/20251218220000_flow_iterator_data.down.sql

@@ -0,0 +1,2 @@
+-- Drop flow_iterator_data table
+DROP TABLE IF EXISTS flow_iterator_data;

backend/migrations/20251218220000_flow_iterator_data.up.sql

@@ -0,0 +1,9 @@
+-- Create separate table for storing flow iterator data (itered arrays)
+-- This avoids expensive JSONB_SET operations on large itered arrays during parallel loop execution
+CREATE TABLE IF NOT EXISTS flow_iterator_data (
+    job_id      UUID PRIMARY KEY REFERENCES v2_job_queue (id) ON DELETE CASCADE NOT NULL,
+    itered      JSONB NOT NULL,
+    created_at  TIMESTAMP WITH TIME ZONE DEFAULT now()
+);
+
+-- Index not needed beyond primary key since all lookups are by job_id

backend/windmill-common/src/flow_status.rs

@@ -78,7 +78,10 @@ pub struct RestartedFrom {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct Iterator {
     pub index: usize,
-    pub itered: Vec<Box<serde_json::value::RawValue>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub itered: Option<Vec<Box<serde_json::value::RawValue>>>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub itered_len: Option<usize>,
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]

backend/windmill-common/src/worker.rs

@@ -278,6 +278,7 @@ lazy_static::lazy_static! {
     pub static ref MIN_VERSION_IS_AT_LEAST_1_427: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
     pub static ref MIN_VERSION_IS_AT_LEAST_1_432: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
     pub static ref MIN_VERSION_IS_AT_LEAST_1_440: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
+    pub static ref MIN_VERSION_IS_AT_LEAST_1_595: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
 
     // Features flags:
     pub static ref DISABLE_FLOW_SCRIPT: bool = std::env::var("DISABLE_FLOW_SCRIPT").ok().is_some_and(|x| x == "1" || x == "true");
@@ -1308,6 +1309,7 @@ pub async fn update_min_version(conn: &Connection) -> bool {
     *MIN_VERSION_IS_AT_LEAST_1_427.write().await = min_version >= Version::new(1, 427, 0);
     *MIN_VERSION_IS_AT_LEAST_1_432.write().await = min_version >= Version::new(1, 432, 0);
     *MIN_VERSION_IS_AT_LEAST_1_440.write().await = min_version >= Version::new(1, 440, 0);
+    *MIN_VERSION_IS_AT_LEAST_1_595.write().await = min_version >= Version::new(1, 595, 0);
 
     *MIN_VERSION.write().await = min_version.clone();
     min_version >= cur_version

backend/windmill-queue/src/jobs.rs

@@ -5612,7 +5612,11 @@ fn create_restarted_module(
             Ok(FlowStatusModule::InProgress {
                 id: module.id(),
                 job: new_flow_jobs[new_flow_jobs.len() - 1],
-                iterator: Some(FlowIterator { index: branch_or_iteration_n - 1, itered: vec![] }),
+                iterator: Some(FlowIterator {
+                    index: branch_or_iteration_n - 1,
+                    itered: None,
+                    itered_len: None,
+                }),
                 flow_jobs: Some(new_flow_jobs),
                 flow_jobs_success: new_flow_jobs_success,
                 flow_jobs_duration: new_flow_jobs_timeline,

backend/windmill-worker/src/worker_flow.rs

@@ -61,6 +61,7 @@ use windmill_common::{
         MAX_RETRY_ATTEMPTS, MAX_RETRY_INTERVAL,
     },
     flows::{FlowModule, FlowModuleValue, FlowValue, InputTransform, Retry, Step, Suspend},
+    worker::MIN_VERSION_IS_AT_LEAST_1_595,
 };
 use windmill_queue::schedule::get_schedule_opt;
 use windmill_queue::{
@@ -73,6 +74,58 @@ use windmill_audit::audit_oss::audit_log;
 use windmill_audit::ActionKind;
 use windmill_common::audit::AuditAuthor;
 use windmill_queue::{canceled_job_to_result, push};
+
+/// Helper function to write itered data to separate table
+/// Returns None if data was written to separate table, Some(itered) if it should be stored in JSONB
+async fn write_itered_to_db(
+    db: &DB,
+    job_id: Uuid,
+    itered: &Vec<Box<RawValue>>,
+) -> error::Result<Option<Vec<Box<RawValue>>>> {
+    if *MIN_VERSION_IS_AT_LEAST_1_595.read().await {
+        // Write to separate table
+        sqlx::query!(
+            "INSERT INTO flow_iterator_data (job_id, itered) VALUES ($1, $2)
+             ON CONFLICT (job_id) DO UPDATE SET itered = $2",
+            job_id,
+            Json(itered) as Json<&Vec<Box<RawValue>>>,
+        )
+        .execute(db)
+        .await?;
+        // Return None to indicate itered should not be stored in JSONB
+        Ok(None)
+    } else {
+        // Return Some(itered) to indicate it should be stored in JSONB for backwards compatibility
+        Ok(Some(itered.clone()))
+    }
+}
+
+/// Helper function to read itered data from separate table
+/// Falls back to reading from JSONB flow_status if not found in separate table or version too old
+async fn read_itered_from_db(
+    db: &DB,
+    job_id: Uuid,
+    itered_from_status: &Option<Vec<Box<RawValue>>>,
+) -> error::Result<Vec<Box<RawValue>>> {
+    // Only try to read from separate table if version supports it
+    if *MIN_VERSION_IS_AT_LEAST_1_595.read().await {
+        let result = sqlx::query_scalar!(
+            "SELECT itered as \"itered: Json<Vec<Box<RawValue>>>\" FROM flow_iterator_data WHERE job_id = $1",
+            job_id,
+        )
+        .fetch_optional(db)
+        .await?;
+
+        if let Some(Json(itered)) = result {
+            // Found in separate table
+            return Ok(itered);
+        }
+    }
+
+    // Fall back to reading from JSONB flow_status (backwards compatibility or not found in table)
+    Ok(itered_from_status.clone().unwrap_or_default()) // can be none for restarted flows, in which case we return an empty vector
+}
+
 // #[instrument(level = "trace", skip_all)]
 pub async fn update_flow_status_after_job_completion(
     db: &DB,
@@ -548,6 +601,9 @@ pub async fn update_flow_status_after_job_completion_internal(
             } if *parallel => {
                 let (nindex, len) = match (iterator, branchall) {
                     (Some(FlowIterator { itered, .. }), _) => {
+                        // Read itered from separate table or fallback to JSONB
+                        let itered = read_itered_from_db(db, flow, itered).await?;
+
                         let position = if flow_jobs_success.is_some() {
                             find_flow_job_index(jobs, job_id_for_status)
                         } else {
@@ -876,14 +932,21 @@ pub async fn update_flow_status_after_job_completion_internal(
                 }
             }
             FlowStatusModule::InProgress {
-                iterator: Some(FlowIterator { index, itered, .. }),
+                iterator: Some(FlowIterator { index, itered_len, itered }),
                 flow_jobs_success,
                 flow_jobs,
                 while_loop,
                 ..
-            } if (*while_loop
-                || (*index + 1 < itered.len()) && (success || skip_loop_failures))
-                && !stop_early =>
+            } if {
+                let itered_len = if let Some(itered_len) = itered_len {
+                    *itered_len
+                } else {
+                    // backwards compatibility
+                    itered.as_ref().map(|itered| itered.len()).unwrap_or(0)
+                };
+                (*while_loop || (*index + 1 < itered_len) && (success || skip_loop_failures))
+                    && !stop_early
+            } =>
             {
                 if let Some(jobs) = flow_jobs {
                     set_success_and_duration_in_flow_job_success(
@@ -3177,6 +3240,9 @@ async fn push_next_flow_job(
                 simple_input_transforms,
             } => {
                 if let Ok(args) = args.as_ref() {
+                    // Read itered from separate table or fallback to JSONB
+                    let itered = read_itered_from_db(db, flow_job.id, itered).await?;
+
                     let mut hm = HashMap::new();
                     for (k, v) in args.iter() {
                         hm.insert(k.to_string(), v.to_owned());
@@ -3455,6 +3521,7 @@ async fn push_next_flow_job(
                 ForloopNextIteration {
                     index,
                     itered,
+                    itered_len,
                     mut flow_jobs,
                     while_loop,
                     mut flow_jobs_success,
@@ -3473,9 +3540,18 @@ async fn push_next_flow_job(
             if let Some(flow_jobs_duration) = &mut flow_jobs_duration {
                 flow_jobs_duration.push(&None);
             }
+
+            // Conditionally write itered to separate table if all workers support it
+            // Returns None if written to table, Some(itered) if should be in JSONB
+            let itered_for_status = write_itered_to_db(db, flow_job.id, &itered).await?;
+
             FlowStatusModule::InProgress {
                 job: uuid,
-                iterator: Some(FlowIterator { index, itered }),
+                iterator: Some(FlowIterator {
+                    index,
+                    itered: itered_for_status,
+                    itered_len: Some(itered_len),
+                }),
                 flow_jobs: Some(flow_jobs),
                 flow_jobs_success,
                 flow_jobs_duration,
@@ -3489,21 +3565,34 @@ async fn push_next_flow_job(
                 agent_actions_success: None,
             }
         }
-        NextStatus::AllFlowJobs { iterator, branchall, .. } => FlowStatusModule::InProgress {
-            job: flow_job.id,
-            iterator,
-            flow_jobs_success: Some(vec![None; uuids.len()]),
-            flow_jobs: Some(uuids.clone()),
-            flow_jobs_duration: Some(FlowJobsDuration::new(uuids.len())),
-            branch_chosen: None,
-            branchall,
-            id: status_module.id(),
-            parallel: true,
-            while_loop: false,
-            progress: None,
-            agent_actions: None,
-            agent_actions_success: None,
-        },
+        NextStatus::AllFlowJobs { iterator, branchall, .. } => {
+            // Conditionally write itered to separate table if all workers support it
+            // write_itered_to_db returns None if written to table, Some(itered) if should be in JSONB
+            let iterator_for_status = if let Some(mut iter) = iterator {
+                if let Some(ref itered) = iter.itered {
+                    iter.itered = write_itered_to_db(db, flow_job.id, itered).await?;
+                }
+                Some(iter)
+            } else {
+                None
+            };
+
+            FlowStatusModule::InProgress {
+                job: flow_job.id,
+                iterator: iterator_for_status,
+                flow_jobs_success: Some(vec![None; uuids.len()]),
+                flow_jobs: Some(uuids.clone()),
+                flow_jobs_duration: Some(FlowJobsDuration::new(uuids.len())),
+                branch_chosen: None,
+                branchall,
+                id: status_module.id(),
+                parallel: true,
+                while_loop: false,
+                progress: None,
+                agent_actions: None,
+                agent_actions_success: None,
+            }
+        }
         NextStatus::NextBranchStep(NextBranch {
             mut flow_jobs,
             status,
@@ -3750,6 +3839,7 @@ async fn push_next_flow_job(
 struct ForloopNextIteration {
     index: usize,
     itered: Vec<Box<RawValue>>,
+    itered_len: usize,
     flow_jobs: Vec<Uuid>,
     flow_jobs_success: Option<Vec<Option<bool>>>,
     flow_jobs_duration: Option<FlowJobsDuration>,
@@ -4039,6 +4129,7 @@ async fn compute_next_flow_transform(
                 ForloopNextIteration {
                     index: next_loop_idx,
                     itered: vec![],
+                    itered_len: 0,
                     flow_jobs: flow_jobs,
                     flow_jobs_success: flow_jobs_success,
                     flow_jobs_duration: flow_jobs_duration,
@@ -4093,6 +4184,7 @@ async fn compute_next_flow_transform(
                 flow_env,
                 client,
                 &parallel,
+                db,
             )
             .await?;
 
@@ -4159,7 +4251,11 @@ async fn compute_next_flow_transform(
                         ContinuePayload::ParallelJobs(payloads),
                         NextStatus::AllFlowJobs {
                             branchall: None,
-                            iterator: Some(FlowIterator { index: 0, itered }),
+                            iterator: Some(FlowIterator {
+                                index: 0,
+                                itered_len: Some(itered.len()),
+                                itered: Some(itered),
+                            }),
                             // we removed the is_simple_case for simple_input_transforms
                             // if is_simple {
                             //     match value {
@@ -4457,6 +4553,7 @@ async fn next_forloop_status(
     flow_env: Option<&HashMap<String, Box<RawValue>>>,
     client: &AuthedClient,
     parallel: &bool,
+    db: &DB,
 ) -> Result<ForLoopStatus, Error> {
     let next_loop_status = match status_module {
         FlowStatusModule::WaitingForPriorSteps { .. }
@@ -4511,6 +4608,7 @@ async fn next_forloop_status(
                 let iter = Iter { index: 0 as i32, value: first.to_owned() };
                 ForLoopStatus::NextIteration(ForloopNextIteration {
                     index: 0,
+                    itered_len: itered.len(),
                     itered,
                     flow_jobs: vec![],
                     flow_jobs_success: Some(vec![]),
@@ -4524,12 +4622,15 @@ async fn next_forloop_status(
         }
 
         FlowStatusModule::InProgress {
-            iterator: Some(FlowIterator { itered, index }),
+            iterator: Some(FlowIterator { itered, index, .. }),
             flow_jobs: Some(flow_jobs),
             flow_jobs_success,
             flow_jobs_duration,
             ..
         } if !*parallel => {
+            // Read itered from separate table or fallback to JSONB
+            let itered = read_itered_from_db(db, flow_job.id, itered).await?;
+
             let itered_new = if itered.is_empty() {
                 // it's possible we need to re-compute the iterator Input Transforms here, in particular if the flow is being restarted inside the loop
                 let by_id = if let Some(x) = by_id {
@@ -4582,6 +4683,7 @@ async fn next_forloop_status(
             ForLoopStatus::NextIteration(ForloopNextIteration {
                 index,
                 itered: itered_new.clone(),
+                itered_len: itered_new.len(),
                 flow_jobs: flow_jobs.clone(),
                 flow_jobs_success: flow_jobs_success.clone(),
                 flow_jobs_duration: flow_jobs_duration.clone(),

frontend/src/lib/components/FlowStatusViewerInner.svelte

@@ -570,7 +570,7 @@
 						flow_jobs_success: mod.flow_jobs_success,
 						flow_jobs_duration: mod.flow_jobs_duration,
 						flow_jobs: mod.flow_jobs,
-						iteration_total: mod.iterator?.itered?.length ?? mod.flow_jobs?.length
+						iteration_total: mod.iterator?.itered_len ?? mod.flow_jobs?.length
 					})
 				}
 
@@ -892,7 +892,7 @@
 						flow_jobs: mod.flow_jobs,
 						flow_jobs_success: mod.flow_jobs_success,
 						flow_jobs_duration: mod.flow_jobs_duration,
-						iteration_total: mod.iterator?.itered?.length,
+						iteration_total: mod.iterator?.itered_len,
 						retries: mod?.failed_retries?.length,
 						skipped: mod.skipped,
 						agent_actions: mod.agent_actions,
@@ -1584,7 +1584,7 @@
 													flowJobs: mod.flow_jobs,
 													flowJobsSuccess: mod.flow_jobs_success ?? [],
 													flowJobsDuration: mod.flow_jobs_duration,
-													length: mod.iterator?.itered?.length ?? mod.flow_jobs.length,
+													length: mod.iterator?.itered_len ?? mod.flow_jobs.length,
 													branchall: job?.raw_flow?.modules?.[i]?.value?.type == 'branchall'
 												}
 											: undefined}

frontend/src/lib/components/flows/FlowProgressBar.svelte

@@ -55,7 +55,7 @@
 		// Loop is still iterating
 		if (module?.iterator) {
 			const stepIndex = module.iterator.index || 0
-			const stepLength = module.iterator.itered?.length || 0
+			const stepLength = module.iterator.itered_len || 0
 			if (module.iterator.index != undefined) {
 				subStepIndex = stepIndex
 				subStepLength = stepLength

frontend/src/lib/components/flows/content/FlowModuleHeader.svelte

@@ -128,6 +128,7 @@
 		{/if}
 		{#if customUi?.tagEdit != false}
 			<FlowModuleWorkerTagSelect
+				isPreprocessor={module.id == 'preprocessor'}
 				placeholder={customUi?.tagSelectPlaceholder}
 				noLabel={customUi?.tagSelectNoLabel}
 				nullTag={tag}
@@ -175,6 +176,7 @@
 
 	{#if module.value.type === 'rawscript'}
 		<FlowModuleWorkerTagSelect
+			isPreprocessor={module.id == 'preprocessor'}
 			placeholder={customUi?.tagSelectPlaceholder}
 			noLabel={customUi?.tagSelectNoLabel}
 			nullTag={tag}

frontend/src/lib/components/flows/content/FlowModuleWorkerTagSelect.svelte

@@ -9,12 +9,14 @@
 		tag = $bindable(),
 		nullTag,
 		placeholder,
-		noLabel
+		noLabel,
+		isPreprocessor
 	}: {
 		tag: string | undefined
 		nullTag?: string | undefined
 		placeholder?: string
 		noLabel?: boolean
+		isPreprocessor: boolean
 	} = $props()
 
 	const { flowStore, selectionManager } = getContext<FlowEditorContext>('FlowEditorContext')
@@ -32,7 +34,7 @@
 {#if $workerTags}
 	{#if $workerTags?.length > 0}
 		<div class="w-40">
-			{#if flowStore.val.tag == undefined}
+			{#if flowStore.val.tag == undefined || isPreprocessor}
 				<WorkerTagSelect
 					{noLabel}
 					{placeholder}

openflow.openapi.yaml

@@ -852,6 +852,8 @@ components:
             itered:
               type: array
               items: {}
+            itered_len:
+              type: integer
             args: {}
         flow_jobs:
           type: array