diff --git a/backend/.sqlx/query-389828f43e638c02757ba37da46b03111a9915a16b53f3e29a09de89210d6af1.json b/backend/.sqlx/query-389828f43e638c02757ba37da46b03111a9915a16b53f3e29a09de89210d6af1.json new file mode 100644 index 0000000000..e27316017d --- /dev/null +++ b/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" +} diff --git a/backend/.sqlx/query-3c5165992c4b8ad3f91627d1d9f6156d3a6b45a7cb2b37a7c166d36d7caa4d2f.json b/backend/.sqlx/query-3c5165992c4b8ad3f91627d1d9f6156d3a6b45a7cb2b37a7c166d36d7caa4d2f.json new file mode 100644 index 0000000000..08009b7556 --- /dev/null +++ b/backend/.sqlx/query-3c5165992c4b8ad3f91627d1d9f6156d3a6b45a7cb2b37a7c166d36d7caa4d2f.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT itered as \"itered: Json>>\" FROM flow_iterator_data WHERE job_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "itered: Json>>", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "3c5165992c4b8ad3f91627d1d9f6156d3a6b45a7cb2b37a7c166d36d7caa4d2f" +} diff --git a/backend/migrations/20251218220000_flow_iterator_data.down.sql b/backend/migrations/20251218220000_flow_iterator_data.down.sql new file mode 100644 index 0000000000..a0a96c4e08 --- /dev/null +++ b/backend/migrations/20251218220000_flow_iterator_data.down.sql @@ -0,0 +1,2 @@ +-- Drop flow_iterator_data table +DROP TABLE IF EXISTS flow_iterator_data; diff --git a/backend/migrations/20251218220000_flow_iterator_data.up.sql b/backend/migrations/20251218220000_flow_iterator_data.up.sql new file mode 100644 index 0000000000..2764043f40 --- /dev/null +++ b/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 diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs index 307d28313e..692567a8f3 100644 --- a/backend/windmill-common/src/flow_status.rs +++ b/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>, + #[serde(skip_serializing_if = "Option::is_none")] + pub itered: Option>>, + #[serde(skip_serializing_if = "Option::is_none")] + pub itered_len: Option, } #[derive(Serialize, Deserialize, Debug, Clone)] diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 40c7065957..d11fc3a40b 100644 --- a/backend/windmill-common/src/worker.rs +++ b/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> = Arc::new(RwLock::new(false)); pub static ref MIN_VERSION_IS_AT_LEAST_1_432: Arc> = Arc::new(RwLock::new(false)); pub static ref MIN_VERSION_IS_AT_LEAST_1_440: Arc> = Arc::new(RwLock::new(false)); + pub static ref MIN_VERSION_IS_AT_LEAST_1_595: Arc> = 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 diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index e9d9010ae7..37b9cddffc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/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, diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index faed45fec8..2a2d789c41 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/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>, +) -> error::Result>>> { + 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>>, + ) + .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>>, +) -> error::Result>> { + // 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>>\" 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>, + itered_len: usize, flow_jobs: Vec, flow_jobs_success: Option>>, flow_jobs_duration: Option, @@ -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, ¶llel, + 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>>, client: &AuthedClient, parallel: &bool, + db: &DB, ) -> Result { 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(), diff --git a/frontend/src/lib/components/FlowStatusViewerInner.svelte b/frontend/src/lib/components/FlowStatusViewerInner.svelte index 96d5f803ca..0d78b6b0e3 100644 --- a/frontend/src/lib/components/FlowStatusViewerInner.svelte +++ b/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} diff --git a/frontend/src/lib/components/flows/FlowProgressBar.svelte b/frontend/src/lib/components/flows/FlowProgressBar.svelte index 4837db5e43..eb010715a2 100644 --- a/frontend/src/lib/components/flows/FlowProgressBar.svelte +++ b/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 diff --git a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte index bf3ff8067e..0dbac28ef7 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleHeader.svelte @@ -128,6 +128,7 @@ {/if} {#if customUi?.tagEdit != false} ('FlowEditorContext') @@ -32,7 +34,7 @@ {#if $workerTags} {#if $workerTags?.length > 0}
- {#if flowStore.val.tag == undefined} + {#if flowStore.val.tag == undefined || isPreprocessor}