starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-09 16:05:42 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat keep sql columns ordering (#3444)

Browse Source 
* feat: keep pg columns ordering

* feat: pass column order through flow status

* feat: sql ordering for all sql langs

* fix: sqlx build on mac + rename to _metadata

* Update JobPreview.svelte

* Update +page.svelte

* Update +page.svelte

* Update LogPanel.svelte

* Update +page.svelte

* Update LogPanel.svelte

* Update JobPreview.svelte

* fix: build

---------

Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
HugoCasa
2024-03-22 10:34:32 +01:00
committed by GitHub
co-authored by Ruben Fiszel
parent 03fd4a7468
commit 522f32c6c0
17 changed files with 282 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-4b866ea0b902a2c27d6817935b1ceb1d0590959c4e23995a27aac12ef34cc071.json → backend/.sqlx/query-b08bf73ca7da6af302eeb5a5f443d71e03478e642a7999157a631a7ff0b7c63e.json
Generated
+2 -2
View File
@@ -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"
}
backend/Cargo.lock
Generated
+1
View File
@@ -9778,6 +9778,7 @@ dependencies = [
"hex",
"hmac",
"hyper 1.2.0",
"indexmap 2.2.5",
"itertools 0.12.1",
"lazy_static",
"magic-crypt",
backend/Cargo.toml
+2 -1
View File
@@ -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"] }
backend/windmill-api/src/jobs.rs
+91 -27
View File
@@ -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<FlowStatus> {
@@ -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<sqlx::types::Json<Box<RawValue>>>,
pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
pub language: Option<ScriptLang>,
}
#[derive(FromRow)]
pub struct RawResultWithSuccess<'a> {
pub result: &'a JsonRawValue,
pub struct RawResultWithSuccess {
pub result: Option<sqlx::types::Json<Box<RawValue>>>,
pub flow_status: Option<sqlx::types::Json<Box<RawValue>>>,
pub language: Option<ScriptLang>,
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<DB>,
Path((w_id, id)): Path<(String, Uuid)>,
@@ -3540,7 +3586,7 @@ async fn get_completed_job_result(
) -> error::Result<Response> {
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<bool>,
success: Option<bool>,
completed: bool,
result: Option<&'c JsonRawValue>,
result: Option<FormattedResult>,
}
#[derive(Deserialize)]
@@ -3583,7 +3638,7 @@ async fn get_completed_job_result_maybe(
Query(GetCompletedJobQuery { get_started }): Query<GetCompletedJobQuery>,
) -> error::Result<Response> {
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)
}
backend/windmill-common/Cargo.toml
+2 -1
View File
@@ -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 }
aws-sdk-sts = { workspace = true, optional = true }
indexmap.workspace = true
backend/windmill-common/src/jobs.rs
+90
View File
@@ -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<Box<RawValue>>),
Vec(Vec<Box<RawValue>>),
}
#[derive(Serialize, Debug)]
pub struct CompletedJobWithFormattedResult {
#[serde(flatten)]
pub cj: CompletedJob,
pub result: Option<FormattedResult>,
}
#[derive(Deserialize)]
struct FlowStatusMetadata {
column_order: Vec<String>,
}
#[derive(Deserialize)]
struct FlowStatusWithMetadataOnly {
_metadata: FlowStatusMetadata,
}
pub fn order_columns(
rows: Option<Vec<Box<RawValue>>>,
column_order: Vec<String>,
) -> Option<Vec<Box<RawValue>>> {
if let Some(mut rows) = rows {
if let Some(first_row) = rows.get(0) {
let first_row = serde_json::from_str::<HashMap<String, Box<RawValue>>>(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<Box<RawValue>>,
result: Option<Box<RawValue>>,
) -> 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::<FlowStatusWithMetadataOnly>(x.get()))
{
if let Some(result) = result {
let rows = serde_json::from_str::<Vec<Box<RawValue>>>(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) }
}
backend/windmill-queue/src/jobs.rs
+4 -2
View File
@@ -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
)
backend/windmill-worker/src/bigquery_executor.rs
+12
View File
@@ -65,6 +65,7 @@ pub async fn do_bigquery(
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
) -> windmill_common::error::Result<Box<RawValue>> {
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::<Vec<String>>(),
);
let rows = result
.rows
.unwrap()
backend/windmill-worker/src/mysql_executor.rs
+13
View File
@@ -37,6 +37,7 @@ pub async fn do_mysql(
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
) -> windmill_common::error::Result<Box<RawValue>> {
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::<Vec<String>>()
})
.unwrap_or_default(),
);
Ok(rows
.into_iter()
.map(|x| convert_row_to_value(x))
backend/windmill-worker/src/pg_executor.rs
+17 -2
View File
@@ -63,6 +63,7 @@ pub async fn do_postgresql(
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
) -> error::Result<Box<RawValue>> {
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::<Vec<Row>>().await.map_err(to_anyhow)?;
Ok(rows
*column_order = Some(
rows.first()
.map(|x| {
x.columns()
.iter()
.map(|x| x.name().to_string())
.collect::<Vec<String>>()
})
.unwrap_or_default(),
);
let result = rows
.into_iter()
.map(|x: Row| postgres_row_to_json_value(x))
.collect::<Result<Vec<_>, _>>()?) as anyhow::Result<Vec<serde_json::Value>>
.collect::<Result<Vec<_>, _>>()?;
Ok(result)
};
let result = run_future_with_polling_update_job_poller(
backend/windmill-worker/src/snowflake_executor.rs
+11 -1
View File
@@ -71,6 +71,7 @@ pub async fn do_snowflake(
mem_peak: &mut i32,
canceled_by: &mut Option<CanceledBy>,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
) -> windmill_common::error::Result<Box<RawValue>> {
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::<Vec<String>>(),
);
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::<Vec<_>>(),
);
backend/windmill-worker/src/worker.rs
+29 -1
View File
@@ -2721,6 +2721,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
);
append_logs(job.id, job.workspace_id.clone(), logs, db).await;
let mut column_order: Option<Vec<String>> = None;
let result = match job.job_kind {
JobKind::Dependencies => {
handle_dependency_job(
@@ -2785,6 +2786,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
&mut canceled_by,
base_internal_url,
worker_name,
&mut column_order,
)
.await;
#[cfg(feature = "prometheus")]
@@ -2806,6 +2808,7 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
canceled_by,
cached_res_path,
client.get_token().await,
column_order,
db,
)
.await?;
@@ -2822,13 +2825,33 @@ async fn process_result(
canceled_by: Option<CanceledBy>,
cached_res_path: Option<String>,
token: String,
column_order: Option<Vec<String>>,
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<CanceledBy>,
base_internal_url: &str,
worker_name: &str,
column_order: &mut Option<Vec<String>>,
) -> error::Result<Box<RawValue>> {
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;
}
frontend/src/lib/components/DisplayResult.svelte
+4 -1
View File
@@ -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)
)
)
}
frontend/src/lib/components/runs/JobPreview.svelte
+1 -1
View File
@@ -99,7 +99,7 @@
{/if}
<div class=" w-full rounded-md min-h-full">
{#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)}
<WorkflowTimeline
flow_status={asWorkflowStatus(job.flow_status)}
flowDone={job.type == 'CompletedJob'}
frontend/src/lib/components/scriptEditor/LogPanel.svelte
+1 -1
View File
@@ -89,7 +89,7 @@
{#if selectedTab === 'logs'}
<SplitPanesWrapper>
<Splitpanes horizontal>
{#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)}
<Pane class="relative">
<WorkflowTimeline
flow_status={asWorkflowStatus(previewJob.flow_status)}
frontend/src/lib/components/table/AutoDataTable.svelte
+1 -1
View File
@@ -363,7 +363,7 @@
on:change={() => handleCheckboxChange(_id)}
/>
</Cell>
{#each Object.keys(rowData ?? {}) ?? [] as key, index}
{#each Object.keys(data[0].rowData ?? {}) ?? [] as key, index}
{@const value = rowData[key]}
<Cell last={index == Object.values(rowData ?? {}).length - 1}>
{#if hiddenColumns.includes(key)}
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+1 -1
View File
@@ -605,7 +605,7 @@
/>
</div>
{: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)}
<div class="mt-10" />
<WorkflowTimeline
flow_status={asWorkflowStatus(job.flow_status)}