From f4df3affa32caa4140c2cad6e3586e5da03c88a5 Mon Sep 17 00:00:00 2001 From: yacine Bouraroui Date: Tue, 1 Oct 2024 15:17:08 +0200 Subject: [PATCH] separate job_args and job_params tables --- ...241001112828_create_job_args_table.down.sql | 2 ++ ...20241001112828_create_job_args_table.up.sql | 7 +++++++ ...41001113254_alter_job_params_table.down.sql | 5 +++++ ...0241001113254_alter_job_params_table.up.sql | 5 +++++ backend/windmill-queue/src/jobs.rs | 18 ++++++++++++++++-- backend/windmill-worker/src/worker_flow.rs | 11 +++++++---- 6 files changed, 42 insertions(+), 6 deletions(-) create mode 100644 backend/migrations/20241001112828_create_job_args_table.down.sql create mode 100644 backend/migrations/20241001112828_create_job_args_table.up.sql create mode 100644 backend/migrations/20241001113254_alter_job_params_table.down.sql create mode 100644 backend/migrations/20241001113254_alter_job_params_table.up.sql diff --git a/backend/migrations/20241001112828_create_job_args_table.down.sql b/backend/migrations/20241001112828_create_job_args_table.down.sql new file mode 100644 index 0000000000..3ff2b010ef --- /dev/null +++ b/backend/migrations/20241001112828_create_job_args_table.down.sql @@ -0,0 +1,2 @@ +-- Add down migration script here +DROP TABLE job_args; \ No newline at end of file diff --git a/backend/migrations/20241001112828_create_job_args_table.up.sql b/backend/migrations/20241001112828_create_job_args_table.up.sql new file mode 100644 index 0000000000..22bd182c7d --- /dev/null +++ b/backend/migrations/20241001112828_create_job_args_table.up.sql @@ -0,0 +1,7 @@ +-- Add up migration script here +CREATE TABLE job_args ( + id UUID PRIMARY KEY, + args JSONB, + tag VARCHAR(50), + workspace_id VARCHAR(50) +); \ No newline at end of file diff --git a/backend/migrations/20241001113254_alter_job_params_table.down.sql b/backend/migrations/20241001113254_alter_job_params_table.down.sql new file mode 100644 index 0000000000..f19592ab0c --- /dev/null +++ b/backend/migrations/20241001113254_alter_job_params_table.down.sql @@ -0,0 +1,5 @@ +-- Add down migration script here +ALTER TABLE job_params DROP COLUMN workspace_id; + +-- args has been moved to job_params table +ALTER TABLE job_params ADD COLUMN args JSONB; \ No newline at end of file diff --git a/backend/migrations/20241001113254_alter_job_params_table.up.sql b/backend/migrations/20241001113254_alter_job_params_table.up.sql new file mode 100644 index 0000000000..58ca7fca44 --- /dev/null +++ b/backend/migrations/20241001113254_alter_job_params_table.up.sql @@ -0,0 +1,5 @@ +-- Add up migration script here +ALTER TABLE job_params ADD COLUMN workspace_id VARCHAR(50); + +-- args has been moved to job_params table +ALTER TABLE job_params DROP COLUMN args; \ No newline at end of file diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 292027dd8c..98b0a004b4 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -2528,6 +2528,7 @@ async fn extract_result_from_job_result( Ok(to_raw_value(&result)) } }, + JobResult::SingleJob(x) => Ok(sqlx::query_as::<_, ResultR>( "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2", ) @@ -3959,11 +3960,24 @@ pub async fn push<'c, 'd, R: rsmq_async::RsmqConnection + Send + 'c>( // insert into args queue sqlx::query!( r#" - INSERT INTO job_params (id, args, raw_code, raw_flow, tag) + INSERT INTO job_args (id, workspace_id, args, tag) + VALUES ($1, $2, $3, $4) + "#, + uuid, + workspace_id, + args as Json, + &tag + ) + .execute(&mut tx) + .await?; + + sqlx::query!( + r#" + INSERT INTO job_params (id, workspace_id, raw_code, raw_flow, tag) VALUES ($1, $2, $3, $4, $5) "#, uuid, - args as Json, + workspace_id, raw_code, raw_flow as Option>, tag diff --git a/backend/windmill-worker/src/worker_flow.rs b/backend/windmill-worker/src/worker_flow.rs index a257e50ea9..dd5e50af2a 100644 --- a/backend/windmill-worker/src/worker_flow.rs +++ b/backend/windmill-worker/src/worker_flow.rs @@ -62,14 +62,14 @@ type DB = sqlx::Pool; use windmill_queue::{canceled_job_to_result, get_queued_job_tx, push, QueueTransaction}; macro_rules! fetch_one_with_fallback { - ($db:expr, $fetch_method:ident, $row_type:ty, $query:literal, $( $param:expr ),* ) => {{ - let primary_query = sqlx::$fetch_method::<_, $row_type>(const_format::formatcp!($query, "job_params")) + ($db:expr, $fetch_method:ident, $row_type:ty, $query:literal, $table:literal ,$( $param:expr ),* ) => {{ + let primary_query = sqlx::$fetch_method::<_, $row_type>(const_format::formatcp!($query, $table)) $(.bind($param))* .fetch_one($db) .await; if let Err(sqlx::Error::RowNotFound) = primary_query { - // Fallback to the second query if the first query returns no result + tracing::info!("Data not found in job_params, falling back to fetching from queue"); sqlx::$fetch_method::<_, $row_type>(const_format::formatcp!($query, "queue")) $(.bind($param))* .fetch_one($db) @@ -86,6 +86,7 @@ async fn get_args_from_job_id(db: &DB, id: &Uuid) -> Result { query_as, RowArgs, "SELECT args FROM {} WHERE id = $1", + "job_args", id ); @@ -333,7 +334,7 @@ pub async fn update_flow_status_after_job_completion_internal< }; let is_flow = if let Some(step) = step { - fetch_one_with_fallback!(db, query_scalar, Option, "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM {} WHERE id = $2", + fetch_one_with_fallback!(db, query_scalar, Option, "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM {} WHERE id = $2","job_params", step as i32, &flow).map_err(|e| { Error::InternalErr(format!("error during retrieval of step's type: {e:#}")) @@ -1284,6 +1285,7 @@ async fn has_failure_module<'c>(flow: Uuid, db: &DB) -> Result { query_scalar, Option, "SELECT raw_flow->'failure_module' != 'null'::jsonb FROM {} WHERE id = $1", + "job_params", flow ) .map_err(|e| { @@ -2238,6 +2240,7 @@ async fn push_next_flow_job .bind(&flow_job.workspace_id) .fetch_optional(db) .await?; + if let Some(raw_args) = row { Ok(Marc::new( raw_args.args.map(|x| x.0).unwrap_or_else(HashMap::new),