separate job_args and job_params tables

backend/migrations/20241001112828_create_job_args_table.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TABLE job_args;

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)
+);

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;

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;

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<PushArgs>,
+        &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<PushArgs>,
+        workspace_id,
         raw_code,
         raw_flow as Option<Json<FlowValue>>,
         tag

backend/windmill-worker/src/worker_flow.rs

@@ -62,14 +62,14 @@ type DB = sqlx::Pool<sqlx::Postgres>;
 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<RowArgs, Error> {
         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<bool>, "SELECT raw_flow->'modules'->($1)->'value'->>'type' = 'flow' FROM {} WHERE id = $2",
+                fetch_one_with_fallback!(db, query_scalar, Option<bool>, "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<bool, Error> {
         query_scalar,
         Option<bool>,
         "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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>
             .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),