diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index 4ad4af081e..6099a2ddf7 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -942,26 +942,6 @@ impl Display for RunnableKind { } } -#[cfg(test)] -mod tests { - use super::*; - #[test] - fn test_build_arg_str() { - let r = build_arg_str( - &[ - ("host", Some("localhost")), - ("port", Some("5432")), - ("password", None), - ("user", Some("postgres")), - ("dbname", Some("test_db")), - ], - " ", - "=", - ); - assert_eq!(r, "host=localhost port=5432 user=postgres dbname=test_db"); - } -} - #[derive(Clone)] pub struct ExpiringCacheEntry { pub value: T, @@ -986,3 +966,280 @@ pub async fn get_custom_pg_instance_password(db: &DB) -> Result { )) ) } + +// Avoid JSON parsing for merging raw JSON values into an object +pub fn merge_raw_values_to_object( + pairs: &[(String, Box)], +) -> Box { + let mut result = String::from("{"); + + for (i, (key, value)) in pairs.iter().enumerate() { + if i > 0 { + result.push(','); + } + // Serialize the key (handles escaping) + result.push_str(&serde_json::to_string(&key).unwrap()); + result.push(':'); + result.push_str(value.get()); + } + + result.push('}'); + + serde_json::value::RawValue::from_string(result).unwrap() +} + +// Avoid JSON parsing for merging raw JSON values into an array +pub fn merge_raw_values_to_array( + values: &[Box], +) -> Box { + let mut result = String::from("["); + + for (i, value) in values.iter().enumerate() { + if i > 0 { + result.push(','); + } + result.push_str(value.get()); + } + + result.push(']'); + + serde_json::value::RawValue::from_string(result).unwrap() +} + +// Optimisation to avoid allocating intermediate strings when merging nested raw JSON values into an array +pub fn merge_nested_raw_values_to_array< + 'a, + It1: Iterator, + It2: Iterator>, +>( + nested_values: It1, +) -> Box { + let mut result = String::from("["); + let mut outer_first = true; + + for inner_iter in nested_values { + if !outer_first { + result.push(','); + } else { + outer_first = false; + } + + result.push('['); + let mut inner_first = true; + + for value in inner_iter { + if !inner_first { + result.push(','); + } else { + inner_first = false; + } + result.push_str(value.get()); + } + + result.push(']'); + } + + result.push(']'); + + serde_json::value::RawValue::from_string(result).unwrap() +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn test_build_arg_str() { + let r = build_arg_str( + &[ + ("host", Some("localhost")), + ("port", Some("5432")), + ("password", None), + ("user", Some("postgres")), + ("dbname", Some("test_db")), + ], + " ", + "=", + ); + assert_eq!(r, "host=localhost port=5432 user=postgres dbname=test_db"); + } + + #[test] + fn test_merge_raw_values_to_object() { + let key1 = "name".to_string(); + let val1 = serde_json::value::RawValue::from_string("\"John\"".to_string()).unwrap(); + let key2 = "age".to_string(); + let val2 = serde_json::value::RawValue::from_string("30".to_string()).unwrap(); + + let pairs = vec![(key1, val1), (key2, val2)]; + let result = merge_raw_values_to_object(&pairs); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed["name"], "John"); + assert_eq!(parsed["age"], 30); + } + + #[test] + fn test_merge_raw_values_to_object_empty() { + let pairs: Vec<(String, Box)> = vec![]; + let result = merge_raw_values_to_object(&pairs); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!({})); + } + + #[test] + fn test_merge_raw_values_to_object_special_chars() { + let key1 = "key with spaces".to_string(); + let val1 = serde_json::value::RawValue::from_string("\"value\"".to_string()).unwrap(); + let key2 = "key\"with\"quotes".to_string(); + let val2 = serde_json::value::RawValue::from_string("42".to_string()).unwrap(); + + let pairs = vec![(key1, val1), (key2, val2)]; + let result = merge_raw_values_to_object(&pairs); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed["key with spaces"], "value"); + assert_eq!(parsed["key\"with\"quotes"], 42); + } + + #[test] + fn test_merge_raw_values_to_array() { + let val1 = serde_json::value::RawValue::from_string("1".to_string()).unwrap(); + let val2 = serde_json::value::RawValue::from_string("\"text\"".to_string()).unwrap(); + let val3 = serde_json::value::RawValue::from_string("true".to_string()).unwrap(); + + let values = vec![val1, val2, val3]; + let result = merge_raw_values_to_array(&values); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([1, "text", true])); + } + + #[test] + fn test_merge_raw_values_to_array_empty() { + let values: Vec> = vec![]; + let result = merge_raw_values_to_array(&values); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([])); + } + + #[test] + fn test_merge_raw_values_to_array_nested_objects() { + let val1 = serde_json::value::RawValue::from_string("{\"a\":1}".to_string()).unwrap(); + let val2 = serde_json::value::RawValue::from_string("{\"b\":2}".to_string()).unwrap(); + + let values = vec![val1, val2]; + let result = merge_raw_values_to_array(&values); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([{"a": 1}, {"b": 2}])); + } + + #[test] + fn test_merge_nested_raw_values_to_array() { + let val1 = serde_json::value::RawValue::from_string("1".to_string()).unwrap(); + let val2 = serde_json::value::RawValue::from_string("2".to_string()).unwrap(); + let val3 = serde_json::value::RawValue::from_string("3".to_string()).unwrap(); + let val4 = serde_json::value::RawValue::from_string("4".to_string()).unwrap(); + + let inner1 = vec![val1, val2]; + let inner2 = vec![val3, val4]; + let nested = vec![inner1.iter(), inner2.iter()]; + + let result = merge_nested_raw_values_to_array(nested.into_iter()); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([[1, 2], [3, 4]])); + } + + #[test] + fn test_merge_nested_raw_values_to_array_empty_outer() { + let nested: Vec>> = vec![]; + let result = merge_nested_raw_values_to_array(nested.into_iter()); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([])); + } + + #[test] + fn test_merge_nested_raw_values_to_array_empty_inner() { + let inner1: Vec> = vec![]; + let val1 = serde_json::value::RawValue::from_string("1".to_string()).unwrap(); + let inner2 = vec![val1]; + let nested = vec![inner1.iter(), inner2.iter()]; + + let result = merge_nested_raw_values_to_array(nested.into_iter()); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([[], [1]])); + } + + #[test] + fn test_merge_nested_raw_values_to_array_all_empty_inner() { + let inner1: Vec> = vec![]; + let inner2: Vec> = vec![]; + let nested = vec![inner1.iter(), inner2.iter()]; + + let result = merge_nested_raw_values_to_array(nested.into_iter()); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([[], []])); + } + + #[test] + fn test_merge_nested_raw_values_to_array_complex_types() { + let val1 = serde_json::value::RawValue::from_string("{\"name\":\"Alice\"}".to_string()).unwrap(); + let val2 = serde_json::value::RawValue::from_string("[1,2,3]".to_string()).unwrap(); + let val3 = serde_json::value::RawValue::from_string("\"text\"".to_string()).unwrap(); + let val4 = serde_json::value::RawValue::from_string("null".to_string()).unwrap(); + + let inner1 = vec![val1, val2]; + let inner2 = vec![val3, val4]; + let nested = vec![inner1.iter(), inner2.iter()]; + + let result = merge_nested_raw_values_to_array(nested.into_iter()); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!( + parsed, + serde_json::json!([[{"name": "Alice"}, [1, 2, 3]], ["text", null]]) + ); + } + + #[test] + fn test_merge_nested_raw_values_to_array_single_inner() { + let val1 = serde_json::value::RawValue::from_string("1".to_string()).unwrap(); + let val2 = serde_json::value::RawValue::from_string("2".to_string()).unwrap(); + let val3 = serde_json::value::RawValue::from_string("3".to_string()).unwrap(); + + let inner1 = vec![val1, val2, val3]; + let nested = vec![inner1.iter()]; + + let result = merge_nested_raw_values_to_array(nested.into_iter()); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([[1, 2, 3]])); + } + + #[test] + fn test_merge_nested_raw_values_to_array_many_inner() { + let val1 = serde_json::value::RawValue::from_string("1".to_string()).unwrap(); + let val2 = serde_json::value::RawValue::from_string("2".to_string()).unwrap(); + let val3 = serde_json::value::RawValue::from_string("3".to_string()).unwrap(); + let val4 = serde_json::value::RawValue::from_string("4".to_string()).unwrap(); + let val5 = serde_json::value::RawValue::from_string("5".to_string()).unwrap(); + + let inner1 = vec![val1]; + let inner2 = vec![val2]; + let inner3 = vec![val3]; + let inner4 = vec![val4]; + let inner5 = vec![val5]; + let nested = vec![inner1.iter(), inner2.iter(), inner3.iter(), inner4.iter(), inner5.iter()]; + + let result = merge_nested_raw_values_to_array(nested.into_iter()); + + let parsed: serde_json::Value = serde_json::from_str(result.get()).unwrap(); + assert_eq!(parsed, serde_json::json!([[1], [2], [3], [4], [5]])); + } +} diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 3597e75f50..c526b8a9be 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -34,7 +34,7 @@ use crate::{ global_settings::CUSTOM_TAGS_SETTING, indexer::TantivyIndexerSettings, server::Smtp, - utils::GIT_SEM_VERSION, + utils::{merge_nested_raw_values_to_array, merge_raw_values_to_array, GIT_SEM_VERSION}, KillpillSender, BASE_INTERNAL_URL, DB, }; @@ -853,7 +853,7 @@ impl SqlResultCollectionStrategy { } } (true, false) => match values.into_iter().last() { - Some(rows) => Ok(to_raw_value(&rows)), + Some(rows) => Ok(merge_raw_values_to_array(rows.as_slice())), None => Ok(null()), }, (false, true) => { @@ -861,9 +861,11 @@ impl SqlResultCollectionStrategy { .into_iter() .map(|rows| rows.into_iter().next().unwrap_or_else(null)) .collect::>(); - Ok(to_raw_value(&values)) + Ok(merge_raw_values_to_array(values.as_slice())) } - (false, false) => Ok(to_raw_value(&values)), + (false, false) => Ok(merge_nested_raw_values_to_array( + values.iter().map(|x| x.iter()), + )), } } } diff --git a/backend/windmill-worker/src/mssql_executor.rs b/backend/windmill-worker/src/mssql_executor.rs index abb2d0087c..6d809969cb 100644 --- a/backend/windmill-worker/src/mssql_executor.rs +++ b/backend/windmill-worker/src/mssql_executor.rs @@ -4,7 +4,7 @@ use futures::StreamExt; use regex::Regex; use serde::Deserialize; use serde_json::value::RawValue; -use serde_json::{Map, Value}; +use serde_json::Value; use tiberius::{ AuthMethod, Client, ColumnData, Config, EncryptionLevel, FromSqlOwned, Query, Row, SqlBrowser, }; @@ -12,6 +12,7 @@ use tokio::net::TcpStream; use tokio_util::compat::TokioAsyncWriteCompatExt; use uuid::Uuid; use windmill_common::s3_helpers::convert_json_line_stream; +use windmill_common::utils::merge_raw_values_to_object; use windmill_common::worker::SqlResultCollectionStrategy; use windmill_common::{ error::{self, to_anyhow, Error}, @@ -228,7 +229,9 @@ pub async fn do_mssql( if let Some(s3) = s3 { let rows_stream = async_stream::stream! { let mut stream = prepared_query.query(&mut client).await.map_err(to_anyhow)?.into_row_stream().map(|row| { - row_to_json(row.map_err(to_anyhow)?).map_err(to_anyhow) + let raw_value = row_to_json(row.map_err(to_anyhow)?).map_err(to_anyhow); + let json = raw_value.and_then(|raw_value| serde_json::to_value(raw_value.get()).map_err(to_anyhow)); + json }); while let Some(row) = stream.next().await { yield row; @@ -250,8 +253,7 @@ pub async fn do_mssql( } let mut json_rows = vec![]; for row in statement_result { - let row = to_raw_value(&row_to_json(row)?); - json_rows.push(row); + json_rows.push(row_to_json(row)?); if collection_strategy.collect_first_row_only() { break; } @@ -345,81 +347,93 @@ fn json_value_to_sql<'a>( Ok(()) } -fn row_to_json(row: Row) -> Result { +fn row_to_json(row: Row) -> Result, Error> { let cols = row .columns() .iter() .map(|x| x.to_owned()) .collect::>(); - let mut map = Map::new(); + let mut entries = Vec::new(); for (col, val) in cols.iter().zip(row.into_iter()) { - map.insert(col.name().to_string(), sql_to_json_value(val)?); + entries.push((col.name().to_string(), sql_to_json_value(val)?)); } - Ok(Value::Object(map)) + Ok(merge_raw_values_to_object(entries.as_slice())) } -fn value_or_null( - val: Option, - convert: impl Fn(T) -> Result, -) -> Result { - val.map_or(Ok(Value::Null), convert) +fn sql_to_json_value(val: ColumnData) -> Result, Error> { + let null = || RawValue::from_string("null".to_string()).unwrap(); + let val = match val { + ColumnData::Bit(x) => x.map(|x| to_raw_value(&x)).unwrap_or_else(null), + ColumnData::U8(x) => x.map(|x| to_raw_value(&x)).unwrap_or_else(null), + ColumnData::I16(x) => x.map(|x| to_raw_value(&x)).unwrap_or_else(null), + ColumnData::I32(x) => x.map(|x| to_raw_value(&x)).unwrap_or_else(null), + ColumnData::I64(x) => x.map(|x| to_raw_value(&x)).unwrap_or_else(null), + ColumnData::String(x) => x.map(|x| to_raw_value(&x.to_string())).unwrap_or_else(null), + ColumnData::Binary(x) => x + .map(|x| to_raw_value(&general_purpose::STANDARD.encode(x.as_ref()))) + .unwrap_or_else(null), + ColumnData::F32(x) => x.map(|x| to_raw_value(&x)).unwrap_or_else(null), + ColumnData::F64(x) => x.map(|x| to_raw_value(&x)).unwrap_or_else(null), + ColumnData::Guid(x) => x.map(|x| to_raw_value(&x.to_string())).unwrap_or_else(null), + ColumnData::Xml(x) => x.map(|x| to_raw_value(&x.to_string())).unwrap_or_else(null), + ColumnData::Numeric(x) => x + .map(|x| numeric_to_raw_value(&x)) + .transpose()? + .unwrap_or_else(null), + ColumnData::DateTime(x) => NaiveDateTime::from_sql_owned(ColumnData::DateTime(x)) + .map_err(to_anyhow)? + .map(|x| to_raw_value(&x.to_string())) + .unwrap_or_else(null), + ColumnData::DateTime2(x) => NaiveDateTime::from_sql_owned(ColumnData::DateTime2(x)) + .map_err(to_anyhow)? + .map(|x| to_raw_value(&x.to_string())) + .unwrap_or_else(null), + ColumnData::SmallDateTime(x) => NaiveDateTime::from_sql_owned(ColumnData::SmallDateTime(x)) + .map_err(to_anyhow)? + .map(|x| to_raw_value(&x.to_string())) + .unwrap_or_else(null), + ColumnData::Time(x) => NaiveTime::from_sql_owned(ColumnData::Time(x)) + .map_err(to_anyhow)? + .map(|x| to_raw_value(&x.to_string())) + .unwrap_or_else(null), + ColumnData::Date(x) => NaiveDate::from_sql_owned(ColumnData::Date(x)) + .map_err(to_anyhow)? + .map(|x| to_raw_value(&x.to_string())) + .unwrap_or_else(null), + ColumnData::DateTimeOffset(x) => { + DateTime::::from_sql_owned(ColumnData::DateTimeOffset(x)) + .map_err(to_anyhow)? + .map(|x| to_raw_value(&x.to_string())) + .unwrap_or_else(null) + } + }; + Ok(val) } -fn sql_to_json_value(val: ColumnData) -> Result { - match val { - ColumnData::Bit(x) => value_or_null(x, |x| Ok(Value::Bool(x))), - ColumnData::U8(x) => value_or_null(x, |x| Ok(Value::Number(x.into()))), - ColumnData::I16(x) => value_or_null(x, |x| Ok(Value::Number(x.into()))), - ColumnData::I32(x) => value_or_null(x, |x| Ok(Value::Number(x.into()))), - ColumnData::I64(x) => value_or_null(x, |x| Ok(Value::Number(x.into()))), - ColumnData::String(x) => value_or_null(x, |x| Ok(Value::String(x.to_string()))), - ColumnData::Binary(x) => value_or_null(x, |x| { - Ok(Value::String(general_purpose::STANDARD.encode(x.as_ref()))) - }), - ColumnData::F32(x) => value_or_null(x, |x| { - Ok(Value::Number( - serde_json::Number::from_f64(x.into()) - .ok_or(anyhow::anyhow!("invalid json-float"))?, - )) - }), - ColumnData::F64(x) => value_or_null(x, |x| { - Ok(Value::Number( - serde_json::Number::from_f64(x).ok_or(anyhow::anyhow!("invalid json-float"))?, - )) - }), - ColumnData::Guid(x) => value_or_null(x, |x| Ok(Value::String(x.to_string()))), - ColumnData::Xml(x) => value_or_null(x, |x| Ok(Value::String(x.to_string()))), - ColumnData::Numeric(x) => value_or_null(x, |x| { - Ok(Value::Number( - serde_json::Number::from_f64(x.into()) - .ok_or(anyhow::anyhow!("invalid json-float"))?, - )) - }), - ColumnData::DateTime(x) => value_or_null( - NaiveDateTime::from_sql_owned(ColumnData::DateTime(x)).map_err(to_anyhow)?, - |x| Ok(Value::String(x.to_string())), - ), - ColumnData::DateTime2(x) => value_or_null( - NaiveDateTime::from_sql_owned(ColumnData::DateTime2(x)).map_err(to_anyhow)?, - |x| Ok(Value::String(x.to_string())), - ), - ColumnData::SmallDateTime(x) => value_or_null( - NaiveDateTime::from_sql_owned(ColumnData::SmallDateTime(x)).map_err(to_anyhow)?, - |x| Ok(Value::String(x.to_string())), - ), - ColumnData::Time(x) => value_or_null( - NaiveTime::from_sql_owned(ColumnData::Time(x)).map_err(to_anyhow)?, - |x| Ok(Value::String(x.to_string())), - ), - ColumnData::Date(x) => value_or_null( - NaiveDate::from_sql_owned(ColumnData::Date(x)).map_err(to_anyhow)?, - |x| Ok(Value::String(x.to_string())), - ), - ColumnData::DateTimeOffset(x) => value_or_null( - DateTime::::from_sql_owned(ColumnData::DateTimeOffset(x)).map_err(to_anyhow)?, - |x| Ok(Value::String(x.to_string())), - ), - } +fn numeric_to_raw_value(numeric: &tiberius::numeric::Numeric) -> Result, Error> { + // tiberius::Numeric::to_string is broken, don't use it + + let sign = if numeric.int_part().is_negative() { + "-" + } else { + "" + }; + let int_part = numeric.int_part().abs(); + let dec_part = numeric.dec_part().abs(); + + let str = if dec_part == 0 { + format!("{}{}", sign, int_part) + } else { + format!( + "{}{}.{:0pad$}", + sign, + int_part, + dec_part, + pad = numeric.scale() as usize + ) + }; + + Ok(RawValue::from_string(str).map_err(to_anyhow)?) } fn deserialize_aad_token<'de, D>(deserializer: D) -> Result, D::Error> @@ -433,3 +447,62 @@ where _ => Ok(None), } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_sql_to_json_value_numeric_null() { + let result = sql_to_json_value(ColumnData::Numeric(None)).unwrap(); + assert_eq!(result.get(), "null"); + } + + #[test] + fn test_sql_to_json_value_numeric_integer() { + use tiberius::numeric::Numeric; + let numeric = Numeric::new_with_scale(12345, 0); + let result = sql_to_json_value(ColumnData::Numeric(Some(numeric))).unwrap(); + assert_eq!(result.get(), "12345"); + } + + #[test] + fn test_sql_to_json_value_numeric_decimal() { + use tiberius::numeric::Numeric; + let numeric = Numeric::new_with_scale(123456, 2); // Represents 1234.56 + let result = sql_to_json_value(ColumnData::Numeric(Some(numeric))).unwrap(); + assert_eq!(result.get(), "1234.56"); + } + + #[test] + fn test_sql_to_json_value_numeric_negative() { + use tiberius::numeric::Numeric; + let numeric = Numeric::new_with_scale(-98765, 2); // Represents -987.65 + let result = sql_to_json_value(ColumnData::Numeric(Some(numeric))).unwrap(); + assert_eq!(result.get(), "-987.65"); + } + + #[test] + fn test_sql_to_json_value_numeric_negative_integer() { + use tiberius::numeric::Numeric; + let numeric = Numeric::new_with_scale(-98765, 0); + let result = sql_to_json_value(ColumnData::Numeric(Some(numeric))).unwrap(); + assert_eq!(result.get(), "-98765"); + } + + #[test] + fn test_sql_to_json_value_numeric_high_precision() { + use tiberius::numeric::Numeric; + let numeric = Numeric::new_with_scale(123456789012345, 10); // High precision + let result = sql_to_json_value(ColumnData::Numeric(Some(numeric))).unwrap(); + assert_eq!(result.get(), "12345.6789012345"); + } + + #[test] + fn test_sql_to_json_value_numeric_7_69() { + use tiberius::numeric::Numeric; + let numeric = Numeric::new_with_scale(769, 2); + let result = sql_to_json_value(ColumnData::Numeric(Some(numeric))).unwrap(); + assert_eq!(result.get(), "7.69"); + } +}