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 <noreply@anthropic.com>

* fix: review fixes from local-review

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* update parser

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

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<_>>(),
+            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<_>>(),
+            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<_>>(),
+            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<_>>(),
+            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<_>>(),
+            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"

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<String> {
+    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<u8> = 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<String> {
+    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<String> {
+    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<u8> = 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<String> {
+    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<String, String> = Cache::new(4);
 }

backend/windmill-worker/src/bigquery_executor.rs

@@ -327,7 +327,7 @@ pub async fn do_bigquery(
     occupancy_metrics: &mut OccupancyMetrics,
     parent_runnable_path: Option<String>,
 ) -> windmill_common::error::Result<Box<RawValue>> {
-    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?;
 

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")]

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<String>,
 ) -> error::Result<Box<RawValue>> {
-    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?;
 

backend/windmill-worker/src/mysql_executor.rs

@@ -173,7 +173,7 @@ pub async fn do_mysql(
     occupancy_metrics: &mut OccupancyMetrics,
     parent_runnable_path: Option<String>,
 ) -> windmill_common::error::Result<Box<RawValue>> {
-    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?;
 

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<Mutex<Option<(String, tokio_postgres::Client)>>> =
@@ -281,7 +283,7 @@ pub async fn do_postgresql(
     parent_runnable_path: Option<String>,
     run_inline: bool,
 ) -> error::Result<Box<RawValue>> {
-    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<String, Value>,
+    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, chrono::ParseError> {
     chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d")

backend/windmill-worker/src/snowflake_executor.rs

@@ -541,7 +541,45 @@ pub async fn do_snowflake(
     occupancy_metrics: &mut OccupancyMetrics,
     parent_runnable_path: Option<String>,
 ) -> windmill_common::error::Result<Box<RawValue>> {
-    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));

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<String> {
+    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<String> {
+    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::value::RawValue> =
+        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<Vec<Box<serde_json::value::RawValue>>> {
+    let mut out = Vec::new();
+    for line in text.lines() {
+        let line = line.trim();
+        if line.is_empty() {
+            continue;
+        }
+        let v: Box<serde_json::value::RawValue> = 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('…'));
+    }
+}

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",

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",

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)