From c0eeea9c833f9be3981389a19d0964400fd2bda8 Mon Sep 17 00:00:00 2001 From: hugocasa Date: Tue, 28 Apr 2026 22:01:06 +0200 Subject: [PATCH] feat: support S3Object input args in native SQL scripts (#8954) * feat: support S3Object input args in native SQL scripts Co-Authored-By: Claude Opus 4.5 * fix: review fixes from local-review Co-Authored-By: Claude Opus 4.5 * update parser --------- Co-authored-by: Claude Opus 4.5 --- .../parsers/windmill-parser-sql/src/lib.rs | 61 ++++++ backend/windmill-object-store/src/lib.rs | 74 +++++++ .../windmill-worker/src/bigquery_executor.rs | 34 +++- backend/windmill-worker/src/lib.rs | 1 + backend/windmill-worker/src/mssql_executor.rs | 31 ++- backend/windmill-worker/src/mysql_executor.rs | 30 ++- backend/windmill-worker/src/pg_executor.rs | 57 +++++- .../windmill-worker/src/snowflake_executor.rs | 40 +++- backend/windmill-worker/src/sql_s3_input.rs | 191 ++++++++++++++++++ frontend/package-lock.json | 58 +----- frontend/package.json | 2 +- frontend/src/lib/script_helpers.ts | 17 ++ 12 files changed, 536 insertions(+), 60 deletions(-) create mode 100644 backend/windmill-worker/src/sql_s3_input.rs diff --git a/backend/parsers/windmill-parser-sql/src/lib.rs b/backend/parsers/windmill-parser-sql/src/lib.rs index 77979541f8..1c56bec175 100644 --- a/backend/parsers/windmill-parser-sql/src/lib.rs +++ b/backend/parsers/windmill-parser-sql/src/lib.rs @@ -854,6 +854,7 @@ pub fn parse_mysql_typ(typ: &str) -> Typ { "bool" | "bit" => Typ::Bool, "double precision" | "float" | "real" | "dec" | "fixed" => Typ::Float, "date" | "datetime" | "timestamp" | "time" => Typ::Datetime, + "s3object" => Typ::Resource("S3Object".to_string()), _ => Typ::Str(None), } } @@ -901,6 +902,7 @@ pub fn parse_pg_typ(typ: &str) -> Typ { | "timestamp with time zone" | "timestamp without time zone" => Typ::Datetime, "bytea" => Typ::Bytes, + "s3object" => Typ::Resource("S3Object".to_string()), _ => Typ::Str(None), } } @@ -919,6 +921,7 @@ pub fn parse_bigquery_typ(typ: &str) -> Typ { "integer" | "int64" => Typ::Int, "float" | "float64" | "numeric" | "bignumeric" => Typ::Float, "boolean" | "bool" => Typ::Bool, + "s3object" => Typ::Resource("S3Object".to_string()), _ => Typ::Str(None), } } @@ -959,6 +962,7 @@ pub fn parse_snowflake_typ(typ: &str) -> Typ { "int" => Typ::Int, "float" => Typ::Float, "boolean" => Typ::Bool, + "s3object" => Typ::Resource("S3Object".to_string()), _ => Typ::Str(None), } } @@ -973,6 +977,7 @@ pub fn parse_mssql_typ(typ: &str) -> Typ { "bigint" | "int" | "tinyint" | "smallint" => Typ::Int, "float" | "real" | "numeric" | "decimal" => Typ::Float, "bit" => Typ::Bool, + "s3object" => Typ::Resource("S3Object".to_string()), _ => Typ::Str(None), } } @@ -1698,6 +1703,62 @@ SELECT $1::integer; Ok(()) } + #[test] + fn test_parse_s3object_arg_per_dialect() -> anyhow::Result<()> { + // Confirms that `(s3object)` is recognised as a resource-typed arg in every + // native SQL dialect that opts in (PG, MySQL, MSSQL, BigQuery, Snowflake). + // The frontend uses `Typ::Resource("S3Object")` to render the S3 picker, and + // the worker dispatches on `otyp == "s3object"` to fetch + bind the file. + let s3 = || Typ::Resource("S3Object".to_string()); + + assert_eq!( + parse_pgsql_sig("-- $1 myfile (s3object)\nSELECT $1::jsonb;")? + .args + .into_iter() + .map(|a| (a.name, a.typ, a.otyp)) + .collect::>(), + vec![("myfile".to_string(), s3(), Some("s3object".to_string()))] + ); + + assert_eq!( + parse_mssql_sig("-- @P1 myfile (s3object)\nSELECT @P1;")? + .args + .into_iter() + .map(|a| (a.name, a.typ, a.otyp)) + .collect::>(), + vec![("myfile".to_string(), s3(), Some("s3object".to_string()))] + ); + + assert_eq!( + parse_mysql_sig("-- :myfile (s3object)\nSELECT :myfile;")? + .args + .into_iter() + .map(|a| (a.name, a.typ, a.otyp)) + .collect::>(), + vec![("myfile".to_string(), s3(), Some("s3object".to_string()))] + ); + + assert_eq!( + parse_bigquery_sig("-- @myfile (s3object)\nSELECT @myfile;")? + .args + .into_iter() + .map(|a| (a.name, a.typ, a.otyp)) + .collect::>(), + vec![("myfile".to_string(), s3(), Some("s3object".to_string()))] + ); + + assert_eq!( + parse_snowflake_sig("-- ? myfile (s3object)\nSELECT ?;")? + .args + .into_iter() + .map(|a| (a.name, a.typ, a.otyp)) + .collect::>(), + vec![("myfile".to_string(), s3(), Some("s3object".to_string()))] + ); + + Ok(()) + } + #[test] fn test_parse_pgsql_safe_interpolated_args() -> anyhow::Result<()> { // There was a bug where enum would be "angrycreative"/"bishop"/"test SELECT x" diff --git a/backend/windmill-object-store/src/lib.rs b/backend/windmill-object-store/src/lib.rs index 1a25513e54..aea5a0e4cd 100644 --- a/backend/windmill-object-store/src/lib.rs +++ b/backend/windmill-object-store/src/lib.rs @@ -1210,6 +1210,80 @@ where )) } +/// Decode the bytes of a Parquet file into a JSON array text (`[ {...}, {...} ]`) +/// suitable for binding as a single SQL parameter and consuming with `OPENJSON`, +/// `jsonb_to_recordset`, `JSON_TABLE`, etc. +/// +/// Runs the synchronous Arrow parquet reader on a `spawn_blocking` thread, which is +/// fine for the ~500 MB ceiling we target. Larger files should use a streaming +/// path (out of scope for the s3-input feature). +#[cfg(feature = "parquet")] +pub async fn decode_parquet_bytes_to_json_array(bytes: bytes::Bytes) -> anyhow::Result { + use datafusion::parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + + task::spawn_blocking(move || { + let builder = ParquetRecordBatchReaderBuilder::try_new(bytes).map_err(to_anyhow)?; + let reader = builder.build().map_err(to_anyhow)?; + + let mut out: Vec = Vec::new(); + let mut writer = json::Writer::<_, JsonArray>::new(&mut out); + for batch in reader { + let batch = batch.map_err(to_anyhow)?; + writer.write(&batch).map_err(to_anyhow)?; + } + writer.finish().map_err(to_anyhow)?; + drop(writer); + String::from_utf8(out).map_err(to_anyhow) + }) + .await + .map_err(to_anyhow)? +} + +#[cfg(not(feature = "parquet"))] +pub async fn decode_parquet_bytes_to_json_array(_bytes: bytes::Bytes) -> anyhow::Result { + anyhow::bail!("Parquet S3 input requires the `parquet` feature to be enabled on this build") +} + +/// Decode the bytes of a CSV file into a JSON array text using the first row as headers. +/// Same blocking-thread pattern as the parquet decoder. +#[cfg(feature = "parquet")] +pub async fn decode_csv_bytes_to_json_array(bytes: bytes::Bytes) -> anyhow::Result { + use datafusion::arrow::csv::ReaderBuilder; + use std::io::Cursor; + + task::spawn_blocking(move || { + let cursor = Cursor::new(bytes); + // Two-pass: infer schema from the bytes, then build the reader. The infer step + // rewinds the underlying reader for us. + let (schema, _) = datafusion::arrow::csv::reader::Format::default() + .with_header(true) + .infer_schema(Cursor::new(&cursor.get_ref()[..]), Some(1024)) + .map_err(to_anyhow)?; + + let reader = ReaderBuilder::new(Arc::new(schema)) + .with_header(true) + .build(cursor) + .map_err(to_anyhow)?; + + let mut out: Vec = Vec::new(); + let mut writer = json::Writer::<_, JsonArray>::new(&mut out); + for batch in reader { + let batch = batch.map_err(to_anyhow)?; + writer.write(&batch).map_err(to_anyhow)?; + } + writer.finish().map_err(to_anyhow)?; + drop(writer); + String::from_utf8(out).map_err(to_anyhow) + }) + .await + .map_err(to_anyhow)? +} + +#[cfg(not(feature = "parquet"))] +pub async fn decode_csv_bytes_to_json_array(_bytes: bytes::Bytes) -> anyhow::Result { + anyhow::bail!("CSV S3 input requires the `parquet` feature to be enabled on this build") +} + lazy_static::lazy_static! { pub static ref S3_PROXY_LAST_ERRORS_CACHE: Cache = Cache::new(4); } diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs index 84028ba4b9..bfe0874f83 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -327,7 +327,7 @@ pub async fn do_bigquery( occupancy_metrics: &mut OccupancyMetrics, parent_runnable_path: Option, ) -> windmill_common::error::Result> { - let bigquery_args = build_args_values(job, client, conn).await?; + let mut bigquery_args = build_args_values(job, client, conn).await?; let inline_db_res_path = parse_db_resource(&query); let s3 = parse_s3_mode(&query)?.map(|s3| s3_mode_args_to_worker_data(s3, client.clone(), job)); @@ -378,10 +378,40 @@ pub async fn do_bigquery( .await .map_err(|e| Error::ExecutionErr(e.to_string()))?; - let sig = parse_bigquery_sig(&query) + let mut sig = parse_bigquery_sig(&query) .map_err(|x| Error::ExecutionErr(x.to_string()))? .args; + // Materialize any `(s3object)` args into JSON text and rewrite the arg type to + // STRING. The user wraps the parameter with `JSON_EXTRACT_ARRAY(@p)` (or similar) + // in their SQL. + for arg in sig.iter_mut() { + if arg.otyp.as_deref() != Some("s3object") { + continue; + } + let raw = bigquery_args.remove(&arg.name).unwrap_or(Value::Null); + if matches!(raw, Value::Null) { + return Err(Error::BadRequest(format!( + "Missing S3Object value for arg `{}`", + arg.name + ))); + } + let s3_obj: windmill_types::s3::S3Object = serde_json::from_value(raw).map_err(|e| { + Error::ExecutionErr(format!("Invalid S3Object for arg `{}`: {e}", arg.name)) + })?; + let json_text = + crate::sql_s3_input::fetch_s3object_as_json_text(client, &job.workspace_id, &s3_obj) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "Failed to fetch S3 object for arg `{}`: {e}", + arg.name + )) + })?; + bigquery_args.insert(arg.name.clone(), Value::String(json_text)); + arg.otyp = Some("string".to_string()); + } + let reserved_variables = get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?; diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index b87c20ac05..2634824d5c 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -71,6 +71,7 @@ pub mod result_processor; mod rust_executor; mod sanitized_sql_params; mod schema; +mod sql_s3_input; pub mod sql_utils; mod universal_pkg_installer; #[cfg(feature = "private")] diff --git a/backend/windmill-worker/src/mssql_executor.rs b/backend/windmill-worker/src/mssql_executor.rs index f1e8d93171..f3df3ea3b4 100644 --- a/backend/windmill-worker/src/mssql_executor.rs +++ b/backend/windmill-worker/src/mssql_executor.rs @@ -28,7 +28,9 @@ use crate::common::{ }; use crate::handle_child::run_future_with_polling_update_job_poller; use crate::sanitized_sql_params::sanitize_and_interpolate_unsafe_sql_args; +use crate::sql_s3_input::fetch_s3object_as_json_text; use windmill_common::client::AuthedClient; +use windmill_types::s3::S3Object; use serde::Deserializer; @@ -71,7 +73,7 @@ pub async fn do_mssql( job_dir: &str, parent_runnable_path: Option, ) -> error::Result> { - let mssql_args = build_args_values(job, authed_client, conn).await?; + let mut mssql_args = build_args_values(job, authed_client, conn).await?; let inline_db_res_path = parse_db_resource(&query); let s3 = parse_s3_mode(&query)? @@ -219,6 +221,33 @@ pub async fn do_mssql( .map_err(|x| Error::ExecutionErr(x.to_string()))? .args; + // Materialize any `(s3object)` args into JSON text. tiberius binds the resulting + // String as nvarchar(max), which is exactly the input type for `OPENJSON(@P)`. + for arg in sig.iter() { + if arg.otyp.as_deref() != Some("s3object") { + continue; + } + let raw = mssql_args.remove(&arg.name).unwrap_or(Value::Null); + if matches!(raw, Value::Null) { + return Err(Error::BadRequest(format!( + "Missing S3Object value for arg `{}`", + arg.name + ))); + } + let s3_obj: S3Object = serde_json::from_value(raw).map_err(|e| { + Error::ExecutionErr(format!("Invalid S3Object for arg `{}`: {e}", arg.name)) + })?; + let json_text = fetch_s3object_as_json_text(authed_client, &job.workspace_id, &s3_obj) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "Failed to fetch S3 object for arg `{}`: {e}", + arg.name + )) + })?; + mssql_args.insert(arg.name.clone(), Value::String(json_text)); + } + let reserved_variables = get_reserved_variables(job, &authed_client.token, conn, parent_runnable_path).await?; diff --git a/backend/windmill-worker/src/mysql_executor.rs b/backend/windmill-worker/src/mysql_executor.rs index 301c3d7aa2..b7234b7912 100644 --- a/backend/windmill-worker/src/mysql_executor.rs +++ b/backend/windmill-worker/src/mysql_executor.rs @@ -173,7 +173,7 @@ pub async fn do_mysql( occupancy_metrics: &mut OccupancyMetrics, parent_runnable_path: Option, ) -> windmill_common::error::Result> { - let job_args = build_args_values(job, client, conn).await?; + let mut job_args = build_args_values(job, client, conn).await?; let inline_db_res_path = parse_db_resource(&query); let s3 = parse_s3_mode(&query)?.map(|s3| s3_mode_args_to_worker_data(s3, client.clone(), job)); @@ -226,6 +226,34 @@ pub async fn do_mysql( .map_err(|x| Error::ExecutionErr(x.to_string()))? .args; + // Materialize any `(s3object)` args into JSON text. mysql_async binds strings as + // VARBINARY/TEXT, which MySQL's `JSON_TABLE`/`JSON_EXTRACT` accept directly. + for arg in sig.iter() { + if arg.otyp.as_deref() != Some("s3object") { + continue; + } + let raw = job_args.remove(&arg.name).unwrap_or(Value::Null); + if matches!(raw, Value::Null) { + return Err(Error::BadRequest(format!( + "Missing S3Object value for arg `{}`", + arg.name + ))); + } + let s3_obj: windmill_types::s3::S3Object = serde_json::from_value(raw).map_err(|e| { + Error::ExecutionErr(format!("Invalid S3Object for arg `{}`: {e}", arg.name)) + })?; + let json_text = + crate::sql_s3_input::fetch_s3object_as_json_text(client, &job.workspace_id, &s3_obj) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "Failed to fetch S3 object for arg `{}`: {e}", + arg.name + )) + })?; + job_args.insert(arg.name.clone(), Value::String(json_text)); + } + let reserved_variables = get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?; diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 71332eca71..ab27f0244c 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -43,11 +43,13 @@ use crate::common::{ }; use crate::handle_child::run_future_with_polling_update_job_poller; use crate::sanitized_sql_params::sanitize_and_interpolate_unsafe_sql_args; +use crate::sql_s3_input::fetch_s3object_as_json_text; use crate::sql_utils::remove_comments; use crate::MAX_RESULT_SIZE; use bytes::Buf; use lazy_static::lazy_static; use windmill_common::client::AuthedClient; +use windmill_types::s3::S3Object; lazy_static! { pub static ref CONNECTION_CACHE: Arc>> = @@ -281,7 +283,7 @@ pub async fn do_postgresql( parent_runnable_path: Option, run_inline: bool, ) -> error::Result> { - let pg_args = build_args_values(job, client, conn).await?; + let mut pg_args = build_args_values(job, client, conn).await?; let inline_db_res_path = parse_db_resource(&query); @@ -385,9 +387,14 @@ pub async fn do_postgresql( new_client = Some(new_pg_connection(&database, use_iam_auth, conn.as_sql()).await?); } - let (sig, typed_schema) = parse_pgsql_sig_with_typed_schema(&query) + let (mut sig, typed_schema) = parse_pgsql_sig_with_typed_schema(&query) .map_err(|x| Error::ExecutionErr(x.to_string()))?; + // Materialize any `(s3object)` args into JSON text and rebind them as `jsonb` so + // `otyp_to_pg_type` picks the right binding. Must run before the param map is + // built below. + materialize_s3object_args(&mut sig.args, &mut pg_args, client, &job.workspace_id).await?; + let reserved_variables = get_reserved_variables(job, &client.token, conn, parent_runnable_path).await?; @@ -584,6 +591,52 @@ async fn increment_connection_counter(database_string: &str) { *counter_map.entry(database_string.to_string()).or_insert(0) += 1; } +/// For each `(s3object)` arg in `sig_args`: download the referenced file, decode it +/// to JSON text, then rewrite the arg to bind as `jsonb`. Mutates `args_map` in place +/// so the existing bind path picks up the materialized payload. +async fn materialize_s3object_args( + sig_args: &mut [Arg], + args_map: &mut HashMap, + client: &AuthedClient, + workspace_id: &str, +) -> error::Result<()> { + for arg in sig_args.iter_mut() { + if arg.otyp.as_deref() != Some("s3object") { + continue; + } + let raw = args_map.remove(&arg.name).unwrap_or(Value::Null); + if matches!(raw, Value::Null) { + return Err(Error::BadRequest(format!( + "Missing S3Object value for arg `{}`", + arg.name + ))); + } + let s3_obj: S3Object = serde_json::from_value(raw).map_err(|e| { + Error::ExecutionErr(format!("Invalid S3Object for arg `{}`: {e}", arg.name)) + })?; + let json_text = fetch_s3object_as_json_text(client, workspace_id, &s3_obj) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "Failed to fetch S3 object for arg `{}`: {e}", + arg.name + )) + })?; + // Parse to a Value so `convert_val`'s Array/Object → JSONB branches bind it + // correctly. A bare String would mismatch the JSONB param type. + let parsed: Value = serde_json::from_str(&json_text).map_err(|e| { + Error::ExecutionErr(format!( + "S3 object for arg `{}` is not valid JSON after decoding: {e}", + arg.name + )) + })?; + args_map.insert(arg.name.clone(), parsed); + arg.otyp = Some("jsonb".to_string()); + arg.typ = Typ::Object(windmill_parser::ObjectType::new(None, Some(vec![]))); + } + Ok(()) +} + /// Parse a date string in formats produced by chrono's Display or JS frontends. fn parse_naive_date(s: &str) -> Result { chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") diff --git a/backend/windmill-worker/src/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs index 8b30c3b35a..839609d9da 100644 --- a/backend/windmill-worker/src/snowflake_executor.rs +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -541,7 +541,45 @@ pub async fn do_snowflake( occupancy_metrics: &mut OccupancyMetrics, parent_runnable_path: Option, ) -> windmill_common::error::Result> { - let snowflake_args = build_args_values(job, client, conn).await?; + let mut snowflake_args = build_args_values(job, client, conn).await?; + + // Materialize any `(s3object)` args into JSON text. The catch-all branch in + // `convert_typ_val` binds a String value as `{type: "TEXT", value: ...}`, which + // the user wraps with `PARSE_JSON(?)` in their SQL. + { + let sig = parse_snowflake_sig(query) + .map_err(|x| Error::ExecutionErr(x.to_string()))? + .args; + for arg in sig.iter() { + if arg.otyp.as_deref() != Some("s3object") { + continue; + } + let raw = snowflake_args.remove(&arg.name).unwrap_or(Value::Null); + if matches!(raw, Value::Null) { + return Err(Error::BadRequest(format!( + "Missing S3Object value for arg `{}`", + arg.name + ))); + } + let s3_obj: windmill_types::s3::S3Object = + serde_json::from_value(raw).map_err(|e| { + Error::ExecutionErr(format!("Invalid S3Object for arg `{}`: {e}", arg.name)) + })?; + let json_text = crate::sql_s3_input::fetch_s3object_as_json_text( + client, + &job.workspace_id, + &s3_obj, + ) + .await + .map_err(|e| { + Error::ExecutionErr(format!( + "Failed to fetch S3 object for arg `{}`: {e}", + arg.name + )) + })?; + snowflake_args.insert(arg.name.clone(), Value::String(json_text)); + } + } let inline_db_res_path = parse_db_resource(&query); let s3 = parse_s3_mode(&query)?.map(|s3| s3_mode_args_to_worker_data(s3, client.clone(), job)); diff --git a/backend/windmill-worker/src/sql_s3_input.rs b/backend/windmill-worker/src/sql_s3_input.rs new file mode 100644 index 0000000000..937a21fa20 --- /dev/null +++ b/backend/windmill-worker/src/sql_s3_input.rs @@ -0,0 +1,191 @@ +//! Helpers for feeding S3Object files into native SQL scripts (PG, MSSQL, MySQL, +//! BigQuery, Snowflake) as bound parameters. +//! +//! Native SQL executors that already support `(s3object)` args call +//! [`fetch_s3object_as_json_text`]: it downloads the file via the authed-client S3 +//! endpoint, infers the format from the object's extension, and returns a JSON-text +//! payload (`[{...}, {...}]`) that the user's SQL consumes via the dialect's JSON +//! parser (`OPENJSON`, `jsonb_to_recordset`, `JSON_TABLE`, ...). +//! +//! DuckDB does not go through this path — DuckDB's engine reads S3 natively, so its +//! executor binds the bare `s3://...` URI instead. + +use anyhow::Context; +use windmill_common::client::AuthedClient; +use windmill_types::s3::S3Object; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum InputFormat { + Json, + Parquet, + Csv, +} + +fn detect_format(key: &str) -> InputFormat { + let lower = key.to_ascii_lowercase(); + if lower.ends_with(".parquet") { + InputFormat::Parquet + } else if lower.ends_with(".csv") { + InputFormat::Csv + } else { + // .json, .jsonl, .ndjson, no extension, anything else → treat as JSON text + InputFormat::Json + } +} + +/// Fetch an [`S3Object`] and materialise it as a JSON text payload ready to bind as +/// a SQL parameter. +/// +/// Behaviour by detected format: +/// - **Parquet** / **CSV**: decoded into a JSON array via the helpers in +/// `windmill-object-store` (require the `parquet` feature on the build). +/// - **JSON / JSONL**: returned as-is; if the bytes look like newline-delimited JSON +/// (multiple top-level values), they are rewrapped as a JSON array so user SQL can +/// uniformly assume an array shape. +pub async fn fetch_s3object_as_json_text( + client: &AuthedClient, + workspace_id: &str, + obj: &S3Object, +) -> anyhow::Result { + let bytes = client + .download_s3_file(workspace_id, &obj.s3, obj.storage.clone()) + .await + .with_context(|| format!("Failed to download S3 object `{}`", obj.s3))?; + + match detect_format(&obj.s3) { + InputFormat::Parquet => { + windmill_object_store::decode_parquet_bytes_to_json_array(bytes).await + } + InputFormat::Csv => windmill_object_store::decode_csv_bytes_to_json_array(bytes).await, + InputFormat::Json => normalise_json_or_jsonl(&bytes), + } +} + +/// Accept either: +/// - a single JSON value (object/array/scalar) — passed through verbatim; +/// - newline-delimited JSON (JSONL / NDJSON) — repackaged as a `[v1, v2, ...]` array. +/// +/// The "is JSONL" heuristic is: bytes parse as a newline-separated sequence of +/// independently-valid JSON values, with at least one newline boundary. This avoids +/// misclassifying a pretty-printed JSON object that happens to contain newlines. +fn normalise_json_or_jsonl(bytes: &[u8]) -> anyhow::Result { + let text = std::str::from_utf8(bytes) + .context("S3 input is not valid UTF-8 (expected JSON or JSONL)")?; + let trimmed = text.trim(); + if trimmed.is_empty() { + return Ok("[]".to_string()); + } + + if let Ok(values) = parse_jsonl(trimmed) { + if values.len() > 1 { + // genuine JSONL (≥2 records on separate lines): wrap as array. + return Ok(serde_json::to_string(&values)?); + } + if values.len() == 1 { + // Single record on a single line — could be either a one-record JSONL + // or a single JSON value. Either way, returning the line as-is gives + // the user the most flexibility (preserves shape). + return Ok(values.into_iter().next().unwrap().get().to_string()); + } + } + + // Fall back to validating the whole blob as a single JSON value. + let parsed: Box = + serde_json::from_str(trimmed).context("S3 input could not be parsed as JSON or JSONL")?; + Ok(parsed.get().to_string()) +} + +fn parse_jsonl(text: &str) -> anyhow::Result>> { + let mut out = Vec::new(); + for line in text.lines() { + let line = line.trim(); + if line.is_empty() { + continue; + } + let v: Box = serde_json::from_str(line) + .with_context(|| format!("Invalid JSONL line: {}", truncate_for_log(line)))?; + out.push(v); + } + Ok(out) +} + +fn truncate_for_log(s: &str) -> String { + const MAX_CHARS: usize = 80; + if s.chars().count() <= MAX_CHARS { + s.to_string() + } else { + let head: String = s.chars().take(MAX_CHARS).collect(); + format!("{}…", head) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn detect_format_by_extension() { + assert_eq!(detect_format("path/file.parquet"), InputFormat::Parquet); + assert_eq!(detect_format("path/file.PARQUET"), InputFormat::Parquet); + assert_eq!(detect_format("path/file.csv"), InputFormat::Csv); + assert_eq!(detect_format("path/file.json"), InputFormat::Json); + assert_eq!(detect_format("path/file.jsonl"), InputFormat::Json); + assert_eq!(detect_format("path/file.ndjson"), InputFormat::Json); + assert_eq!(detect_format("path/with-no-ext"), InputFormat::Json); + } + + #[test] + fn jsonl_two_records_becomes_array() { + let body = "{\"id\":1}\n{\"id\":2}\n"; + let out = normalise_json_or_jsonl(body.as_bytes()).unwrap(); + let v: serde_json::Value = serde_json::from_str(&out).unwrap(); + assert_eq!(v, serde_json::json!([{"id":1},{"id":2}])); + } + + #[test] + fn jsonl_single_record_passes_through() { + let body = "{\"id\":1}"; + let out = normalise_json_or_jsonl(body.as_bytes()).unwrap(); + let v: serde_json::Value = serde_json::from_str(&out).unwrap(); + assert_eq!(v, serde_json::json!({"id":1})); + } + + #[test] + fn json_array_passes_through() { + let body = "[{\"id\":1},{\"id\":2}]"; + let out = normalise_json_or_jsonl(body.as_bytes()).unwrap(); + let v: serde_json::Value = serde_json::from_str(&out).unwrap(); + assert_eq!(v, serde_json::json!([{"id":1},{"id":2}])); + } + + #[test] + fn pretty_printed_json_object_is_not_misclassified_as_jsonl() { + // Each line on its own is not valid JSON, so the JSONL parser bails and we + // fall through to the single-value path. + let body = "{\n \"id\": 1,\n \"name\": \"x\"\n}"; + let out = normalise_json_or_jsonl(body.as_bytes()).unwrap(); + let v: serde_json::Value = serde_json::from_str(&out).unwrap(); + assert_eq!(v, serde_json::json!({"id":1,"name":"x"})); + } + + #[test] + fn empty_body_yields_empty_array() { + let out = normalise_json_or_jsonl(b"").unwrap(); + assert_eq!(out, "[]"); + } + + #[test] + fn malformed_payload_errors() { + let err = normalise_json_or_jsonl(b"not valid json").unwrap_err(); + assert!(err.to_string().contains("could not be parsed")); + } + + #[test] + fn truncate_for_log_handles_multibyte_at_boundary() { + // 80 narrow chars then a 4-byte emoji = byte index 80 falls inside the emoji. + // Naive `&s[..80]` would panic; chars-aware truncation must not. + let s = format!("{}{}", "a".repeat(80), "🦀"); + let out = truncate_for_log(&s); + assert!(out.ends_with('…')); + } +} diff --git a/frontend/package-lock.json b/frontend/package-lock.json index b4e35305c9..39d721e207 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -83,7 +83,7 @@ "windmill-parser-wasm-php": "1.647.1", "windmill-parser-wasm-py": "1.657.2", "windmill-parser-wasm-r": "1.668.1", - "windmill-parser-wasm-regex": "1.688.0", + "windmill-parser-wasm-regex": "1.692.0", "windmill-parser-wasm-ruby": "1.526.1", "windmill-parser-wasm-rust": "1.647.1", "windmill-parser-wasm-ts": "1.657.2", @@ -844,7 +844,6 @@ "version": "1.9.0", "resolved": "https://registry.npmjs.org/@emnapi/core/-/core-1.9.0.tgz", "integrity": "sha512-0DQ98G9ZQZOxfUcQn1waV2yS8aWdZ6kJMbYCJB3oUBecjWYO1fqJ+a1DRfPF3O5JEkwqwP1A9QEN/9mYm2Yd0w==", - "dev": true, "license": "MIT", "optional": true, "dependencies": { @@ -856,7 +855,6 @@ "version": "1.9.0", "resolved": "https://registry.npmjs.org/@emnapi/runtime/-/runtime-1.9.0.tgz", "integrity": "sha512-QN75eB0IH2ywSpRpNddCRfQIhmJYBCJ1x5Lb3IscKAL8bMnVAKnRg8dCoXbHzVLLH7P38N2Z3mtulB7W0J0FKw==", - "dev": true, "license": "MIT", "optional": true, "dependencies": { @@ -867,7 +865,6 @@ "version": "1.2.0", "resolved": "https://registry.npmjs.org/@emnapi/wasi-threads/-/wasi-threads-1.2.0.tgz", "integrity": "sha512-N10dEJNSsUx41Z6pZsXU8FjPjpBEplgH24sfkmITrBED1/U2Esum9F3lfLrMjKHHjmi557zQn7kR9R+XWXu5Rg==", - "dev": true, "license": "MIT", "optional": true, "dependencies": { @@ -1357,7 +1354,6 @@ "version": "1.1.1", "resolved": "https://registry.npmjs.org/@napi-rs/wasm-runtime/-/wasm-runtime-1.1.1.tgz", "integrity": "sha512-p64ah1M1ld8xjWv3qbvFwHiFVWrq1yFvV4f7w+mzaqiR4IlSgkqhcRdHwsGgomwzBH51sRY4NEowLxnaBjcW/A==", - "dev": true, "license": "MIT", "optional": true, "dependencies": { @@ -1514,7 +1510,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1531,7 +1526,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1548,7 +1542,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1565,7 +1558,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1582,7 +1574,6 @@ "cpu": [ "arm" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1599,7 +1590,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1616,7 +1606,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1633,7 +1622,6 @@ "cpu": [ "ppc64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1650,7 +1638,6 @@ "cpu": [ "s390x" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1667,7 +1654,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1684,7 +1670,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1701,7 +1686,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1718,7 +1702,6 @@ "cpu": [ "wasm32" ], - "dev": true, "license": "MIT", "optional": true, "dependencies": { @@ -1735,7 +1718,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -1752,7 +1734,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MIT", "optional": true, "os": [ @@ -2058,7 +2039,6 @@ "version": "0.10.1", "resolved": "https://registry.npmjs.org/@tybys/wasm-util/-/wasm-util-0.10.1.tgz", "integrity": "sha512-9tTaPJLSiejZKx+Bmog4uSubteqTvFrVrURwkmHixBo0G4seD0zUxp98E1DzUBJxLQ3NPwXrGKDiVjwx/DpPsg==", - "dev": true, "license": "MIT", "optional": true, "dependencies": { @@ -6834,7 +6814,7 @@ "version": "1.21.7", "resolved": "https://registry.npmjs.org/jiti/-/jiti-1.21.7.tgz", "integrity": "sha512-/imKNG4EbWNrVjoNC/1H5/9GFy+tqjGBHCaSsN+P2RnPqjsLmv6UD3Ej+Kj8nBWaRAwyk7kK5ZUc+OEatnTR3A==", - "dev": true, + "devOptional": true, "license": "MIT", "bin": { "jiti": "bin/jiti.js" @@ -7333,7 +7313,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7354,7 +7333,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7375,7 +7353,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7396,7 +7373,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7417,7 +7393,6 @@ "cpu": [ "arm" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7438,7 +7413,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7459,7 +7433,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7480,7 +7453,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7501,7 +7473,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7522,7 +7493,6 @@ "cpu": [ "arm64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -7543,7 +7513,6 @@ "cpu": [ "x64" ], - "dev": true, "license": "MPL-2.0", "optional": true, "os": [ @@ -12112,21 +12081,6 @@ } } }, - "node_modules/svelte-check/node_modules/picomatch": { - "version": "4.0.4", - "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.4.tgz", - "integrity": "sha512-QP88BAKvMam/3NxH6vj2o21R6MjxZUAd6nlwAS/pnGvN9IVLocLHxGYIzFhg6fUQ+5th6P4dv4eW9jX3DSIj7A==", - "dev": true, - "license": "MIT", - "optional": true, - "peer": true, - "engines": { - "node": ">=12" - }, - "funding": { - "url": "https://github.com/sponsors/jonschlinkert" - } - }, "node_modules/svelte-eslint-parser": { "version": "0.43.0", "resolved": "https://registry.npmjs.org/svelte-eslint-parser/-/svelte-eslint-parser-0.43.0.tgz", @@ -12857,7 +12811,7 @@ "version": "5.9.3", "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", - "dev": true, + "devOptional": true, "license": "Apache-2.0", "bin": { "tsc": "bin/tsc", @@ -13671,9 +13625,9 @@ "integrity": "sha512-5YNeUibxpNBvYrxCgQcz1PxGhTFx2CyEpg2udtIhq7bx0d4gF/KDZVupMeQmAObmrEtTSFGUWNRJ4zXSWNrSpQ==" }, "node_modules/windmill-parser-wasm-regex": { - "version": "1.688.0", - "resolved": "https://registry.npmjs.org/windmill-parser-wasm-regex/-/windmill-parser-wasm-regex-1.688.0.tgz", - "integrity": "sha512-TB6ysy8nRcWDPRR79ujDihJc3S4oJTqawWSjUF837JDxaG595P9osZErxR4MOO5Ye5MecKESeIgP3UDK6O6bhg==" + "version": "1.692.0", + "resolved": "https://registry.npmjs.org/windmill-parser-wasm-regex/-/windmill-parser-wasm-regex-1.692.0.tgz", + "integrity": "sha512-BHGTxrinZJ9ef6hFxbKiBqBEr5uqgG/QySOgMA5r1LswO9n/8fyGswr8JcPT2kGaoeoweV6/RQ+RHVaOhosnKw==" }, "node_modules/windmill-parser-wasm-ruby": { "version": "1.526.1", diff --git a/frontend/package.json b/frontend/package.json index b30a5fd725..9c1ae3f9fb 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -156,7 +156,7 @@ "windmill-parser-wasm-php": "1.647.1", "windmill-parser-wasm-py": "1.657.2", "windmill-parser-wasm-r": "1.668.1", - "windmill-parser-wasm-regex": "1.688.0", + "windmill-parser-wasm-regex": "1.692.0", "windmill-parser-wasm-ruby": "1.526.1", "windmill-parser-wasm-rust": "1.647.1", "windmill-parser-wasm-ts": "1.657.2", diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index 6a9c56993b..0cdae5afef 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -260,6 +260,9 @@ export async function main(message: string, name: string, step_id: string) { const POSTGRES_INIT_CODE = `-- result_collection=last_statement_all_rows -- to pin the database use '-- database f/your/path' -- to stream a large query result to your workspace storage use '-- s3' +-- to feed an S3Object (json/jsonl/parquet/csv) as a parameter, declare it as (s3object): +-- -- $5 input_file (s3object) +-- INSERT INTO demo SELECT * FROM jsonb_to_recordset(\$5::jsonb) AS x(id INT, name TEXT); -- $1 name1 = default arg -- $2 name2 -- $3 name3 @@ -271,6 +274,9 @@ UPDATE demo SET col2 = \$4::INT WHERE col2 = \$2::INT; const MYSQL_INIT_CODE = `-- result_collection=last_statement_all_rows -- to pin the database use '-- database f/your/path' -- to stream a large query result to your workspace storage use '-- s3' +-- to feed an S3Object (json/jsonl/parquet/csv) as a parameter, declare it as (s3object): +-- -- :input_file (s3object) +-- INSERT INTO demo SELECT * FROM JSON_TABLE(:input_file, '$[*]' COLUMNS (id INT PATH '$.id', name VARCHAR(255) PATH '$.name')) AS x; -- :name1 (text) = default arg -- :name2 (int) -- :name3 (int) @@ -281,6 +287,9 @@ UPDATE demo SET col2 = :name3 WHERE col2 = :name2; const BIGQUERY_INIT_CODE = `-- result_collection=last_statement_all_rows -- to pin the database use '-- database f/your/path' -- to stream a large query result to your workspace storage use '-- s3' +-- to feed an S3Object (json/jsonl/parquet/csv) as a parameter, declare it as (s3object): +-- -- @input_file (s3object) +-- SELECT * FROM UNNEST(JSON_QUERY_ARRAY(@input_file)) AS row; -- @name1 (string) = default arg -- @name2 (integer) -- @name3 (string[]) @@ -302,6 +311,10 @@ UPDATE demo SET col2 = :name3 WHERE col2 = :name2; const SNOWFLAKE_INIT_CODE = `-- result_collection=last_statement_all_rows -- to pin the database use '-- database f/your/path' -- to stream a large query result to your workspace storage use '-- s3' +-- to feed an S3Object (json/jsonl/parquet/csv) as a parameter, declare it as (s3object): +-- -- ? input_file (s3object) +-- SELECT v.value:id::int AS id, v.value:name::string AS name +-- FROM TABLE(FLATTEN(input => PARSE_JSON(?))) v; -- ? name1 (varchar) = default arg -- ? name2 (int) INSERT INTO demo VALUES (?, ?); @@ -313,6 +326,10 @@ UPDATE demo SET col2 = ? WHERE col2 = ?; const MSSQL_INIT_CODE = `-- result_collection=last_statement_all_rows -- to pin the database use '-- database f/your/path' -- to stream a large query result to your workspace storage use '-- s3' +-- to feed an S3Object (json/jsonl/parquet/csv) as a parameter, declare it as (s3object): +-- -- @P4 input_file (s3object) +-- INSERT INTO demo +-- SELECT id, name FROM OPENJSON(@P4) WITH (id INT '$.id', name NVARCHAR(255) '$.name'); -- @P1 name1 (varchar) = default arg -- @P2 name2 (int) -- @P3 name3 (int)