starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-20 08:01:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: improve perf of get completed flow node (#5418)

Browse Source 
* perf: improve perf of get completed flow node

* better

* nit

* nits

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
HugoCasa
2025-03-19 10:51:30 +01:00
committed by GitHub
parent bf938fdfae
commit 884aba4811
4 changed files with 84 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-433da02be85347333323480f7f279e9bd7e1b8348e91b33030df7181a55798a6.json
Generated
+30
View File
@@ -0,0 +1,30 @@
{
"db_name": "PostgreSQL",
"query": "SELECT j.id, jc.flow_status AS \"flow_status!: Json<FlowStatus>\"\n FROM v2_job j\n JOIN v2_job_completed jc ON j.id = jc.id\n WHERE j.parent_job = $1 AND j.workspace_id = $2 AND j.created_at >= $3 AND jc.flow_status IS NOT NULL",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "flow_status!: Json<FlowStatus>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text",
"Timestamptz"
]
},
"nullable": [
false,
true
]
},
"hash": "433da02be85347333323480f7f279e9bd7e1b8348e91b33030df7181a55798a6"
}
backend/.sqlx/query-77f12f22d7c9da1799e5720c9a8fab3b6f26ccbd822fc05b7a0fb3d6a17c5435.json
Generated
+35
View File
@@ -0,0 +1,35 @@
{
"db_name": "PostgreSQL",
"query": "SELECT jc.id, jc.flow_status AS \"flow_status!: Json<FlowStatus>\", j.created_at\n FROM v2_job_completed jc\n JOIN v2_job j ON j.id = jc.id\n WHERE jc.id = $1 AND jc.workspace_id = $2 AND jc.flow_status IS NOT NULL",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "flow_status!: Json<FlowStatus>",
"type_info": "Jsonb"
},
{
"ordinal": 2,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
false,
true,
false
]
},
"hash": "77f12f22d7c9da1799e5720c9a8fab3b6f26ccbd822fc05b7a0fb3d6a17c5435"
}
backend/.sqlx/query-8123ba05f6e7b9bd395175ee4ec0c36c3726b2da9ff3592589ac7e83df1c537c.json
Generated
-29
View File
@@ -1,29 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, flow_status AS \"flow_status!: Json<FlowStatus>\"\n FROM v2_job_completed WHERE id = $1 AND workspace_id = $2",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "flow_status!: Json<FlowStatus>",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Text"
]
},
"nullable": [
false,
true
]
},
"hash": "8123ba05f6e7b9bd395175ee4ec0c36c3726b2da9ff3592589ac7e83df1c537c"
}
backend/windmill-queue/src/jobs.rs
+19 -14
View File
@@ -2591,7 +2591,8 @@ pub async fn get_result_by_id_from_running_flow_inner(
async fn get_completed_flow_node_result_rec(
db: &Pool<Postgres>,
w_id: &str,
subflows: impl std::iter::Iterator<Item = (Uuid, FlowStatus)>,
created_at: DateTime<Utc>,
subflows: Vec<(Uuid, FlowStatus)>,
node_id: &str,
) -> error::Result<Option<JobResult>> {
for (id, flow_status) in subflows {
@@ -2611,19 +2612,19 @@ async fn get_completed_flow_node_result_rec(
};
} else {
let subflows = sqlx::query!(
"SELECT v2_job_completed.id AS \"id!\", flow_status AS \"flow_status!: Json<FlowStatus>\"
FROM v2_job_completed
INNER JOIN v2_job ON (v2_job_completed.id = v2_job.id)
WHERE parent_job = $1 AND v2_job_completed.workspace_id = $2 AND flow_status IS NOT NULL",
"SELECT j.id, jc.flow_status AS \"flow_status!: Json<FlowStatus>\"
FROM v2_job j
JOIN v2_job_completed jc ON j.id = jc.id
WHERE j.parent_job = $1 AND j.workspace_id = $2 AND j.created_at >= $3 AND jc.flow_status IS NOT NULL",
id,
w_id
w_id,
created_at
)
.map(|record| (record.id, record.flow_status.0))
.fetch_all(db)
.await?
.into_iter();
.await?;
match Box::pin(get_completed_flow_node_result_rec(
db, w_id, subflows, node_id,
db, w_id, created_at, subflows, node_id,
))
.await?
{
@@ -2643,22 +2644,26 @@ async fn get_result_by_id_from_original_flow_inner(
node_id: &str,
) -> error::Result<JobResult> {
let flow_job = sqlx::query!(
"SELECT id, flow_status AS \"flow_status!: Json<FlowStatus>\"
FROM v2_job_completed WHERE id = $1 AND workspace_id = $2",
"SELECT jc.id, jc.flow_status AS \"flow_status!: Json<FlowStatus>\", j.created_at
FROM v2_job_completed jc
JOIN v2_job j ON j.id = jc.id
WHERE jc.id = $1 AND jc.workspace_id = $2 AND jc.flow_status IS NOT NULL",
completed_flow_id,
w_id
)
.map(|record| (record.id, record.flow_status.0))
.map(|record| (record.id, record.flow_status.0, record.created_at))
.fetch_optional(db)
.await?;
let flow_job = not_found_if_none(
let (id, flow_status, created_at) = not_found_if_none(
flow_job,
"Root completed job",
format!("root: {}, id: {}", completed_flow_id, node_id),
)?;
match get_completed_flow_node_result_rec(db, w_id, [flow_job].into_iter(), node_id).await? {
match get_completed_flow_node_result_rec(db, w_id, created_at, vec![(id, flow_status)], node_id)
.await?
{
Some(res) => Ok(res),
None => Err(Error::NotFound(format!(
"Flow result by id not found going top-down from {}, (id: {})",