diff --git a/backend/.sqlx/query-4b866ea0b902a2c27d6817935b1ceb1d0590959c4e23995a27aac12ef34cc071.json b/backend/.sqlx/query-b08bf73ca7da6af302eeb5a5f443d71e03478e642a7999157a631a7ff0b7c63e.json similarity index 65% rename from backend/.sqlx/query-4b866ea0b902a2c27d6817935b1ceb1d0590959c4e23995a27aac12ef34cc071.json rename to backend/.sqlx/query-b08bf73ca7da6af302eeb5a5f443d71e03478e642a7999157a631a7ff0b7c63e.json index 778d155730..53fc18f0fc 100644 --- a/backend/.sqlx/query-4b866ea0b902a2c27d6817935b1ceb1d0590959c4e23995a27aac12ef34cc071.json +++ b/backend/.sqlx/query-b08bf73ca7da6af302eeb5a5f443d71e03478e642a7999157a631a7ff0b7c63e.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "UPDATE completed_job SET flow_status = q.flow_status FROM queue q WHERE completed_job.id = $1 AND q.id = $1 AND q.workspace_id = $2 AND completed_job.workspace_id = $2", + "query": "UPDATE completed_job SET flow_status = q.flow_status FROM queue q WHERE completed_job.id = $1 AND q.id = $1 AND q.workspace_id = $2 AND completed_job.workspace_id = $2 AND q.flow_status IS NOT NULL", "describe": { "columns": [], "parameters": { @@ -11,5 +11,5 @@ }, "nullable": [] }, - "hash": "4b866ea0b902a2c27d6817935b1ceb1d0590959c4e23995a27aac12ef34cc071" + "hash": "b08bf73ca7da6af302eeb5a5f443d71e03478e642a7999157a631a7ff0b7c63e" } diff --git a/backend/Cargo.lock b/backend/Cargo.lock index dfc0c6161c..0158497137 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -9778,6 +9778,7 @@ dependencies = [ "hex", "hmac", "hyper 1.2.0", + "indexmap 2.2.5", "itertools 0.12.1", "lazy_static", "magic-crypt", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 20f7ed9c00..49998e8463 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -211,7 +211,7 @@ mysql_async = { version = "*", default-features = false, features = ["minimal", postgres-native-tls = "^0" native-tls = "^0" # samael will break compilation on MacOS. Use this fork instead to make it work -# samael = { git="https://github.com/gbouv/samael", rev="2344211ed0ac041a86222b38b928adfc1030cb94", features = ["xmlsec"] } +# samael = { git="https://github.com/njaremko/samael", rev="464d015e3ae393e4b5dd00b4d6baa1b617de0dd6", features = ["xmlsec"] } samael = { version="0.0.14", features = ["xmlsec"] } gcp_auth = "0.9.0" rust_decimal = { version = "^1", features = ["db-postgres"]} @@ -226,6 +226,7 @@ candle-transformers = "0.3.0" candle-nn = "0.3.0" tiberius = { version = "0.12.2", default-features = false, features = ["rustls", "tds73", "chrono", "sql-browser-tokio"] } pin-project = "1" +indexmap = { version = "2.2.5", features = ["serde"]} polars = { version = "0.38.3", features = ["lazy", "parquet", "aws", "azure", "csv", "dtype-full", "serde", "strings", "extract_groups"] } polars-io = { version = "0.38.3", features = ["csv"] } diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs index bc0799db58..a991c88bc9 100644 --- a/backend/windmill-api/src/jobs.rs +++ b/backend/windmill-api/src/jobs.rs @@ -14,7 +14,10 @@ use std::sync::atomic::Ordering; #[cfg(feature = "prometheus")] use tokio::time::Instant; use windmill_common::flow_status::{JobResult, RestartedFrom}; -use windmill_common::jobs::ENTRYPOINT_OVERRIDE; +use windmill_common::jobs::{ + format_completed_job_result, format_result, CompletedJobWithFormattedResult, FormattedResult, + ENTRYPOINT_OVERRIDE, +}; use windmill_common::variables::get_workspace_key; use crate::db::ApiAuthed; @@ -559,7 +562,12 @@ async fn get_job_internal(db: &DB, workspace_id: &str, job_id: Uuid) -> error::R .await? .map(Job::CompletedJob); if let Some(cjob) = cjob_maybe { - Ok(cjob) + Ok(match cjob { + Job::CompletedJob(cjob) => { + Job::CompletedJobWithFormattedResult(format_completed_job_result(cjob)) + } + cjob => cjob, + }) } else { let job_o = sqlx::query_as::<_, QueuedJob>( "SELECT id, queue.workspace_id, parent_job, created_by, queue.created_at, started_at, scheduled_for, running, @@ -1203,6 +1211,7 @@ async fn resume_suspended_job_internal( let trigger_email = match &parent_flow { Job::CompletedJob(job) => &job.email, Job::QueuedJob(job) => &job.email, + Job::CompletedJobWithFormattedResult(job) => &job.cj.email, }; conditionally_require_authed_user(authed.clone(), flow_status, trigger_email)?; @@ -1459,6 +1468,7 @@ pub async fn get_suspended_job_flow( let trigger_email = match &flow { Job::CompletedJob(job) => &job.email, Job::QueuedJob(job) => &job.email, + Job::CompletedJobWithFormattedResult(job) => &job.cj.email, }; conditionally_require_authed_user(authed, flow_status.clone(), trigger_email)?; @@ -1627,6 +1637,8 @@ pub async fn get_resume_urls( pub enum Job { QueuedJob(QueuedJob), CompletedJob(CompletedJob), + #[serde(rename = "CompletedJob")] + CompletedJobWithFormattedResult(CompletedJobWithFormattedResult), } impl Job { @@ -1642,6 +1654,12 @@ impl Job { .as_ref() .map(|rf| serde_json::from_str(rf.0.get()).ok()) .flatten(), + Job::CompletedJobWithFormattedResult(job) => job + .cj + .raw_flow + .as_ref() + .map(|rf| serde_json::from_str(rf.0.get()).ok()) + .flatten(), } } @@ -1661,6 +1679,13 @@ impl Job { job.logs = Some(logs.to_string()); } } + Job::CompletedJobWithFormattedResult(job) => { + if let Some(ref mut l) = job.cj.logs { + l.push_str(logs); + } else { + job.cj.logs = Some(logs.to_string()); + } + } } } @@ -1668,6 +1693,7 @@ impl Job { match self { Job::QueuedJob(job) => job.logs.as_ref().map(|l| l.len()), Job::CompletedJob(job) => job.logs.as_ref().map(|l| l.len()), + Job::CompletedJobWithFormattedResult(job) => job.cj.logs.as_ref().map(|l| l.len()), } } @@ -1675,6 +1701,7 @@ impl Job { match self { Job::QueuedJob(job) => job.logs.clone(), Job::CompletedJob(job) => job.logs.clone(), + Job::CompletedJobWithFormattedResult(job) => job.cj.logs.clone(), } } pub fn flow_status(&self) -> Option { @@ -1689,12 +1716,19 @@ impl Job { .as_ref() .map(|rf| serde_json::from_str(rf.0.get()).ok()) .flatten(), + Job::CompletedJobWithFormattedResult(job) => job + .cj + .flow_status + .as_ref() + .map(|rf| serde_json::from_str(rf.0.get()).ok()) + .flatten(), } } pub fn is_flow_step(&self) -> bool { match self { Job::QueuedJob(job) => job.is_flow_step, Job::CompletedJob(job) => job.is_flow_step, + Job::CompletedJobWithFormattedResult(job) => job.cj.is_flow_step, } } @@ -1702,6 +1736,7 @@ impl Job { match self { Job::QueuedJob(job) => &job.job_kind, Job::CompletedJob(job) => &job.job_kind, + Job::CompletedJobWithFormattedResult(job) => &job.cj.job_kind, } } @@ -1709,6 +1744,7 @@ impl Job { match self { Job::QueuedJob(job) => job.id, Job::CompletedJob(job) => job.id, + Job::CompletedJobWithFormattedResult(job) => job.cj.id, } } } @@ -2310,14 +2346,23 @@ async fn run_wait_result( .await .ok(); } else { - let row = - sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2") - .bind(uuid) - .bind(&w_id) - .fetch_optional(db) - .await?; + let row = sqlx::query( + "SELECT result, language, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2", + ) + .bind(uuid) + .bind(&w_id) + .fetch_optional(db) + .await?; if let Some(row) = row { - result = Some(RawResult::from_row(&row)?.result.to_owned()); + let raw_result = RawResult::from_row(&row)?; + result = match format_result( + raw_result.language.as_ref(), + raw_result.flow_status.map(|x| x.0), + raw_result.result.map(|x| x.0), + ) { + FormattedResult::RawValue(rv) => rv, + FormattedResult::Vec(v) => Some(to_raw_value(&v)), + }; } else { result = None; } @@ -3502,8 +3547,11 @@ async fn get_completed_job<'a>( .await?; let job = not_found_if_none(job_o, "Completed Job", id.to_string())?; + let cj = CompletedJob::from_row(&job)?; + let cj = format_completed_job_result(cj); + let response = Json(cj).into_response(); // let extra_log = query_scalar!( // "SELECT substr(logs, $1) as logs FROM large_logs WHERE workspace_id = $2 AND job_id = $3", @@ -3517,22 +3565,20 @@ async fn get_completed_job<'a>( } #[derive(FromRow)] -pub struct RawResult<'a> { - pub result: &'a JsonRawValue, +pub struct RawResult { + pub result: Option>>, + pub flow_status: Option>>, + pub language: Option, } #[derive(FromRow)] -pub struct RawResultWithSuccess<'a> { - pub result: &'a JsonRawValue, +pub struct RawResultWithSuccess { + pub result: Option>>, + pub flow_status: Option>>, + pub language: Option, pub success: bool, } -impl<'a> IntoResponse for RawResult<'a> { - fn into_response(self) -> Response { - Json(self.result).into_response() - } -} - async fn get_completed_job_result( Extension(db): Extension, Path((w_id, id)): Path<(String, Uuid)>, @@ -3540,7 +3586,7 @@ async fn get_completed_job_result( ) -> error::Result { let result_o = if let Some(json_path) = json_path { sqlx::query( - "SELECT result #> $3 as result FROM completed_job WHERE id = $1 AND workspace_id = $2", + "SELECT result #> $3 as result, flow_status, language FROM completed_job WHERE id = $1 AND workspace_id = $2", ) .bind(id) .bind(w_id) @@ -3553,7 +3599,7 @@ async fn get_completed_job_result( .fetch_optional(&db) .await? } else { - sqlx::query("SELECT result FROM completed_job WHERE id = $1 AND workspace_id = $2") + sqlx::query("SELECT result, flow_status, language FROM completed_job WHERE id = $1 AND workspace_id = $2") .bind(id) .bind(w_id) .fetch_optional(&db) @@ -3561,15 +3607,24 @@ async fn get_completed_job_result( }; let result = not_found_if_none(result_o, "Completed Job", id.to_string())?; - Ok(RawResult::from_row(&result)?.into_response()) + + let raw_result = RawResult::from_row(&result)?; + + let result = format_result( + raw_result.language.as_ref(), + raw_result.flow_status.map(|x| x.0), + raw_result.result.map(|x| x.0), + ); + + Ok(Json(result).into_response()) } #[derive(Serialize)] -struct CompletedJobResult<'c> { +struct CompletedJobResult { started: Option, success: Option, completed: bool, - result: Option<&'c JsonRawValue>, + result: Option, } #[derive(Deserialize)] @@ -3583,7 +3638,7 @@ async fn get_completed_job_result_maybe( Query(GetCompletedJobQuery { get_started }): Query, ) -> error::Result { let result_o = sqlx::query( - "SELECT result, success FROM completed_job WHERE id = $1 AND workspace_id = $2", + "SELECT result, success, language, flow_status FROM completed_job WHERE id = $1 AND workspace_id = $2", ) .bind(id) .bind(&w_id) @@ -3592,11 +3647,16 @@ async fn get_completed_job_result_maybe( if let Some(result) = result_o { let res = RawResultWithSuccess::from_row(&result)?; + let result = format_result( + res.language.as_ref(), + res.flow_status.map(|x| x.0), + res.result.map(|x| x.0), + ); Ok(Json(CompletedJobResult { started: Some(true), success: Some(res.success), completed: true, - result: Some(res.result), + result: Some(result), }) .into_response()) } else if get_started.is_some_and(|x| x) { @@ -3663,6 +3723,10 @@ async fn delete_completed_job<'a>( .await?; tx.commit().await?; - let response = Json(CompletedJob::from_row(&job)?).into_response(); + let cj = CompletedJob::from_row(&job)?; + + let cj = format_completed_job_result(cj); + + let response = Json(cj).into_response(); Ok(response) } diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 3cfbc6092d..5e5c8f4673 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -45,4 +45,5 @@ magic-crypt.workspace = true object_store = { workspace = true, optional = true } prometheus = { workspace = true, optional = true } aws-config = { workspace = true, optional = true } -aws-sdk-sts = { workspace = true, optional = true } \ No newline at end of file +aws-sdk-sts = { workspace = true, optional = true } +indexmap.workspace = true \ No newline at end of file diff --git a/backend/windmill-common/src/jobs.rs b/backend/windmill-common/src/jobs.rs index 3f5bb6dacc..b6a4bb7b7b 100644 --- a/backend/windmill-common/src/jobs.rs +++ b/backend/windmill-common/src/jobs.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; +use indexmap::IndexMap; use serde::{Deserialize, Serialize}; use serde_json::value::RawValue; use sqlx::{types::Json, Pool, Postgres, Transaction}; @@ -13,6 +14,7 @@ use crate::{ flows::{FlowValue, Retry}, get_latest_deployed_hash_for_path, scripts::{ScriptHash, ScriptLang}, + worker::to_raw_value, }; #[derive(sqlx::Type, Serialize, Deserialize, Debug, PartialEq, Clone)] @@ -465,3 +467,91 @@ pub async fn get_payload_tag_from_prefixed_path( }; Ok((payload, tag)) } + +#[derive(Serialize, Debug)] +#[serde(untagged)] +pub enum FormattedResult { + RawValue(Option>), + Vec(Vec>), +} + +#[derive(Serialize, Debug)] +pub struct CompletedJobWithFormattedResult { + #[serde(flatten)] + pub cj: CompletedJob, + pub result: Option, +} + +#[derive(Deserialize)] +struct FlowStatusMetadata { + column_order: Vec, +} + +#[derive(Deserialize)] +struct FlowStatusWithMetadataOnly { + _metadata: FlowStatusMetadata, +} + +pub fn order_columns( + rows: Option>>, + column_order: Vec, +) -> Option>> { + if let Some(mut rows) = rows { + if let Some(first_row) = rows.get(0) { + let first_row = serde_json::from_str::>>(first_row.get()); + if let Ok(first_row) = first_row { + let mut new_first_row = IndexMap::new(); + for col in column_order { + if let Some(val) = first_row.get(&col) { + new_first_row.insert(col.clone(), val.clone()); + } + } + let new_row_as_raw_value = to_raw_value(&new_first_row); + + rows[0] = new_row_as_raw_value; + + return Some(rows); + } + } + } + + None +} + +pub fn format_result( + language: Option<&ScriptLang>, + flow_status: Option>, + result: Option>, +) -> FormattedResult { + match language { + Some(&ScriptLang::Postgresql) + | Some(&ScriptLang::Mysql) + | Some(&ScriptLang::Snowflake) + | Some(&ScriptLang::Bigquery) => { + if let Some(Ok(flow_status)) = + flow_status.map(|x| serde_json::from_str::(x.get())) + { + if let Some(result) = result { + let rows = serde_json::from_str::>>(result.get()).ok(); + match order_columns(rows, flow_status._metadata.column_order) { + Some(rows) => return FormattedResult::Vec(rows), + None => return FormattedResult::RawValue(Some(result)), + } + } + } + } + _ => {} + } + + FormattedResult::RawValue(result) +} + +pub fn format_completed_job_result(mut cj: CompletedJob) -> CompletedJobWithFormattedResult { + let sql_result = format_result( + cj.language.as_ref(), + cj.flow_status.clone().map(|x| x.0), + cj.result.map(|x| x.0), + ); + cj.result = None; // very important to avoid sending the result twice + CompletedJobWithFormattedResult { cj, result: Some(sql_result) } +} diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 7b1e1fbd69..8f4be9b8fe 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -596,9 +596,11 @@ pub async fn add_completed_job< // tracing::error!("2 {:?}", start.elapsed()); if !queued_job.is_flow_step { - if _duration > 500 { + if _duration > 500 + && (queued_job.job_kind == JobKind::Script || queued_job.job_kind == JobKind::Preview) + { if let Err(e) = sqlx::query!( - "UPDATE completed_job SET flow_status = q.flow_status FROM queue q WHERE completed_job.id = $1 AND q.id = $1 AND q.workspace_id = $2 AND completed_job.workspace_id = $2", + "UPDATE completed_job SET flow_status = q.flow_status FROM queue q WHERE completed_job.id = $1 AND q.id = $1 AND q.workspace_id = $2 AND completed_job.workspace_id = $2 AND q.flow_status IS NOT NULL", &queued_job.id, &queued_job.workspace_id ) diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs index 5ac454e462..15523211a2 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -65,6 +65,7 @@ pub async fn do_bigquery( mem_peak: &mut i32, canceled_by: &mut Option, worker_name: &str, + column_order: &mut Option>, ) -> windmill_common::error::Result> { let bigquery_args = build_args_values(job, client, db).await?; @@ -221,6 +222,17 @@ pub async fn do_bigquery( )); } + *column_order = Some( + result + .schema + .as_ref() + .unwrap() + .fields + .iter() + .map(|x| x.name.clone()) + .collect::>(), + ); + let rows = result .rows .unwrap() diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index 5f4b2e425e..e9855381ae 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -37,6 +37,7 @@ pub async fn do_mysql( mem_peak: &mut i32, canceled_by: &mut Option, worker_name: &str, + column_order: &mut Option>, ) -> windmill_common::error::Result> { let args = build_args_map(job, client, db).await?.map(Json); let job_args = if args.is_some() { @@ -173,6 +174,18 @@ pub async fn do_mysql( ) .await .map_err(to_anyhow)?; + + *column_order = Some( + rows.first() + .map(|x| { + x.columns() + .iter() + .map(|x| x.name_str().to_string()) + .collect::>() + }) + .unwrap_or_default(), + ); + Ok(rows .into_iter() .map(|x| convert_row_to_value(x)) diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 215d660651..36e1b1c379 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -63,6 +63,7 @@ pub async fn do_postgresql( mem_peak: &mut i32, canceled_by: &mut Option, worker_name: &str, + column_order: &mut Option>, ) -> error::Result> { let pg_args = build_args_values(job, client, db).await?; @@ -217,10 +218,24 @@ pub async fn do_postgresql( .map_err(to_anyhow)?; let rows = rows.try_collect::>().await.map_err(to_anyhow)?; - Ok(rows + + *column_order = Some( + rows.first() + .map(|x| { + x.columns() + .iter() + .map(|x| x.name().to_string()) + .collect::>() + }) + .unwrap_or_default(), + ); + + let result = rows .into_iter() .map(|x: Row| postgres_row_to_json_value(x)) - .collect::, _>>()?) as anyhow::Result> + .collect::, _>>()?; + + Ok(result) }; let result = run_future_with_polling_update_job_poller( diff --git a/backend/windmill-worker/src/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs index a79adcb465..c3d8cb4fb4 100644 --- a/backend/windmill-worker/src/snowflake_executor.rs +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -71,6 +71,7 @@ pub async fn do_snowflake( mem_peak: &mut i32, canceled_by: &mut Option, worker_name: &str, + column_order: &mut Option>, ) -> windmill_common::error::Result> { let snowflake_args = build_args_values(job, client, db).await?; @@ -204,6 +205,15 @@ pub async fn do_snowflake( )); } + *column_order = Some( + result + .resultSetMetaData + .rowType + .iter() + .map(|x| x.name.clone()) + .collect::>(), + ); + let rows = to_raw_value( &result .data @@ -218,7 +228,7 @@ pub async fn do_snowflake( parse_val(&val, &row_type.r#type), ); }); - Value::from(row_map) + row_map }) .collect::>(), ); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index 04de0190da..1f3ecba0b7 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -2721,6 +2721,7 @@ async fn handle_queued_job( ); append_logs(job.id, job.workspace_id.clone(), logs, db).await; + let mut column_order: Option> = None; let result = match job.job_kind { JobKind::Dependencies => { handle_dependency_job( @@ -2785,6 +2786,7 @@ async fn handle_queued_job( &mut canceled_by, base_internal_url, worker_name, + &mut column_order, ) .await; #[cfg(feature = "prometheus")] @@ -2806,6 +2808,7 @@ async fn handle_queued_job( canceled_by, cached_res_path, client.get_token().await, + column_order, db, ) .await?; @@ -2822,13 +2825,33 @@ async fn process_result( canceled_by: Option, cached_res_path: Option, token: String, + column_order: Option>, db: &DB, ) -> error::Result<()> { match result { Ok(r) => { + let job = if let Some(column_order) = column_order { + let mut job_with_column_order = (*job).clone(); + match job_with_column_order.flow_status { + Some(_) => { + tracing::warn!("flow_status was expected to be none"); + } + None => { + job_with_column_order.flow_status = + Some(sqlx::types::Json(to_raw_value(&serde_json::json!({ + "_metadata": { + "column_order": column_order + } + })))); + } + } + Arc::new(job_with_column_order) + } else { + job + }; job_completed_tx .send(JobCompleted { - job: job, + job, result: r, mem_peak, canceled_by, @@ -3011,6 +3034,7 @@ async fn handle_code_execution_job( canceled_by: &mut Option, base_internal_url: &str, worker_name: &str, + column_order: &mut Option>, ) -> error::Result> { let ContentReqLangEnvs { content: inner_content, lockfile: requirements_o, language, envs } = match job.job_kind { @@ -3051,6 +3075,7 @@ async fn handle_code_execution_job( mem_peak, canceled_by, worker_name, + column_order, ) .await; } else if language == Some(ScriptLang::Mysql) { @@ -3062,6 +3087,7 @@ async fn handle_code_execution_job( mem_peak, canceled_by, worker_name, + column_order, ) .await; } else if language == Some(ScriptLang::Bigquery) { @@ -3082,6 +3108,7 @@ async fn handle_code_execution_job( mem_peak, canceled_by, worker_name, + column_order, ) .await; } @@ -3103,6 +3130,7 @@ async fn handle_code_execution_job( mem_peak, canceled_by, worker_name, + column_order, ) .await; } diff --git a/frontend/src/lib/components/DisplayResult.svelte b/frontend/src/lib/components/DisplayResult.svelte index 230cc10943..e2c1ba5ab7 100644 --- a/frontend/src/lib/components/DisplayResult.svelte +++ b/frontend/src/lib/components/DisplayResult.svelte @@ -194,7 +194,10 @@ return ( Array.isArray(json) && json.length > 0 && - json.every((item) => item && typeof item === 'object' && Object.keys(item).length > 0) + json.every( + (item) => + item && typeof item === 'object' && Object.keys(item).length > 0 && !Array.isArray(item) + ) ) } diff --git a/frontend/src/lib/components/runs/JobPreview.svelte b/frontend/src/lib/components/runs/JobPreview.svelte index 8d7635f690..026538e82f 100644 --- a/frontend/src/lib/components/runs/JobPreview.svelte +++ b/frontend/src/lib/components/runs/JobPreview.svelte @@ -99,7 +99,7 @@ {/if}
- {#if job?.is_flow_step == false && job?.flow_status && (job?.job_kind == 'preview' || job?.job_kind == 'script')} + {#if job?.is_flow_step == false && job?.flow_status && (job?.job_kind == 'preview' || job?.job_kind == 'script') && !(typeof job.flow_status == 'object' && '_metadata' in job.flow_status)} - {#if previewJob?.is_flow_step == false && previewJob?.flow_status} + {#if previewJob?.is_flow_step == false && previewJob?.flow_status && !(typeof previewJob.flow_status == 'object' && '_metadata' in previewJob.flow_status)} handleCheckboxChange(_id)} /> - {#each Object.keys(rowData ?? {}) ?? [] as key, index} + {#each Object.keys(data[0].rowData ?? {}) ?? [] as key, index} {@const value = rowData[key]} {#if hiddenColumns.includes(key)} diff --git a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte index e6da117cbf..0e31ec8689 100644 --- a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte @@ -605,7 +605,7 @@ />
{:else if job?.job_kind !== 'flow' && job?.job_kind !== 'flowpreview' && job?.job_kind !== 'singlescriptflow'} - {#if job?.flow_status} + {#if job?.flow_status && typeof job.flow_status == 'object' && !('_metadata' in job.flow_status)}