diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 8acdbb8b07..4a6857ad1a 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -38,6 +38,7 @@ pub async fn do_duckdb( #[allow(unused_variables)] column_order_ref: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, parent_runnable_path: Option, + run_inline: bool, ) -> Result> { let annotations = windmill_common::worker::SqlAnnotations::parse(query); let collection_strategy = @@ -182,19 +183,23 @@ pub async fn do_duckdb( Ok(result) }; - let result = run_future_with_polling_update_job_poller( - job.id, - job.timeout, - conn, - mem_peak, - canceled_by, - result_f, - worker_name, - &job.workspace_id, - &mut Some(occupancy_metrics), - Box::pin(futures::stream::once(async { 0 })), - ) - .await; + let result = if run_inline { + result_f.await + } else { + run_future_with_polling_update_job_poller( + job.id, + job.timeout, + conn, + mem_peak, + canceled_by, + result_f, + worker_name, + &job.workspace_id, + &mut Some(occupancy_metrics), + Box::pin(futures::stream::once(async { 0 })), + ) + .await + }; match result { Ok(result) => Ok(result), diff --git a/backend/windmill-worker/src/pg_executor.rs b/backend/windmill-worker/src/pg_executor.rs index 3e0943111e..beb66b39a5 100644 --- a/backend/windmill-worker/src/pg_executor.rs +++ b/backend/windmill-worker/src/pg_executor.rs @@ -199,6 +199,7 @@ pub async fn do_postgresql( column_order: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, parent_runnable_path: Option, + run_inline: bool, ) -> error::Result> { let pg_args = build_args_values(job, client, conn).await?; @@ -396,19 +397,23 @@ pub async fn do_postgresql( collection_strategy.collect(results) }; - let result = run_future_with_polling_update_job_poller( - job.id, - job.timeout, - conn, - mem_peak, - canceled_by, - result_f, - worker_name, - &job.workspace_id, - &mut Some(occupancy_metrics), - Box::pin(futures::stream::once(async { 0 })), - ) - .await?; + let result = if run_inline { + result_f.await? + } else { + run_future_with_polling_update_job_poller( + job.id, + job.timeout, + conn, + mem_peak, + canceled_by, + result_f, + worker_name, + &job.workspace_id, + &mut Some(occupancy_metrics), + Box::pin(futures::stream::once(async { 0 })), + ) + .await? + }; // drop the mtex to avoid holding the lock for too long, result has been returned drop(mtex); diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index b632e36218..81b0e4c4b4 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -3263,6 +3263,7 @@ async fn handle_code_execution_job( envs, codebase, lock, + false, ) .await } @@ -3289,6 +3290,7 @@ pub async fn run_language_executor( envs: &Option>, codebase: &Option, lock: &Option, + run_inline: bool, ) -> error::Result> { if language == Some(ScriptLang::Postgresql) { return do_postgresql( @@ -3302,6 +3304,7 @@ pub async fn run_language_executor( column_order, occupancy_metrics, parent_runnable_path, + run_inline, ) .await; } else if language == Some(ScriptLang::Mysql) { @@ -3311,19 +3314,26 @@ pub async fn run_language_executor( )); #[cfg(feature = "mysql")] - return do_mysql( - job, - &client, - &code, - conn, - mem_peak, - canceled_by, - worker_name, - column_order, - occupancy_metrics, - parent_runnable_path, - ) - .await; + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + return do_mysql( + job, + &client, + &code, + conn, + mem_peak, + canceled_by, + worker_name, + column_order, + occupancy_metrics, + parent_runnable_path, + ) + .await; + } } else if language == Some(ScriptLang::Bigquery) { #[cfg(not(feature = "enterprise"))] { @@ -3342,6 +3352,11 @@ pub async fn run_language_executor( #[cfg(all(feature = "enterprise", feature = "bigquery"))] { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } return do_bigquery( job, &client, @@ -3366,6 +3381,11 @@ pub async fn run_language_executor( #[cfg(feature = "enterprise")] { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } return do_snowflake( job, &client, @@ -3398,6 +3418,11 @@ pub async fn run_language_executor( #[cfg(all(feature = "enterprise", feature = "mssql"))] { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } return do_mssql( job, &client, @@ -3430,6 +3455,11 @@ pub async fn run_language_executor( #[cfg(all(feature = "enterprise", feature = "oracledb"))] { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } return do_oracledb( job, &client, @@ -3466,10 +3496,16 @@ pub async fn run_language_executor( column_order, occupancy_metrics, parent_runnable_path, + run_inline, ) .await; } } else if language == Some(ScriptLang::Graphql) { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } return do_graphql( job, &client, @@ -3482,6 +3518,11 @@ pub async fn run_language_executor( ) .await; } else if language == Some(ScriptLang::Nativets) { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } append_logs( &job.id, &job.workspace_id, @@ -3568,29 +3609,41 @@ mount {{ )); #[cfg(feature = "python")] - Box::pin(handle_python_job( - lock.as_ref(), - job_dir, - worker_dir, - worker_name, - job, - mem_peak, - canceled_by, - conn, - client, - parent_runnable_path, - &code, - &shared_mount, - base_internal_url, - envs, - new_args, - occupancy_metrics, - precomputed_agent_info, - has_stream, - )) - .await + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + Box::pin(handle_python_job( + lock.as_ref(), + job_dir, + worker_dir, + worker_name, + job, + mem_peak, + canceled_by, + conn, + client, + parent_runnable_path, + &code, + &shared_mount, + base_internal_url, + envs, + new_args, + occupancy_metrics, + precomputed_agent_info, + has_stream, + )) + .await + } } Some(ScriptLang::Deno) => { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } Box::pin(handle_deno_job( lock.as_ref(), mem_peak, @@ -3611,6 +3664,11 @@ mount {{ .await } Some(ScriptLang::Bun) | Some(ScriptLang::Bunnative) => { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } Box::pin(handle_bun_job( lock.as_ref(), codebase.as_ref(), @@ -3634,6 +3692,11 @@ mount {{ .await } Some(ScriptLang::Go) => { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } Box::pin(handle_go_job( mem_peak, canceled_by, @@ -3653,6 +3716,11 @@ mount {{ .await } Some(ScriptLang::Bash) => { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } Box::pin(handle_bash_job( mem_peak, canceled_by, @@ -3672,6 +3740,11 @@ mount {{ .await } Some(ScriptLang::Powershell) => { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } Box::pin(handle_powershell_job( mem_peak, canceled_by, @@ -3696,23 +3769,30 @@ mount {{ )); #[cfg(feature = "php")] - Box::pin(handle_php_job( - lock.as_ref(), - mem_peak, - canceled_by, - job, - conn, - client, - parent_runnable_path, - job_dir, - &code, - base_internal_url, - worker_name, - envs, - &shared_mount, - occupancy_metrics, - )) - .await + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + Box::pin(handle_php_job( + lock.as_ref(), + mem_peak, + canceled_by, + job, + conn, + client, + parent_runnable_path, + job_dir, + &code, + base_internal_url, + worker_name, + envs, + &shared_mount, + occupancy_metrics, + )) + .await + } } Some(ScriptLang::Rust) => { #[cfg(not(feature = "rust"))] @@ -3721,23 +3801,30 @@ mount {{ )); #[cfg(feature = "rust")] - Box::pin(handle_rust_job( - mem_peak, - canceled_by, - job, - conn, - client, - parent_runnable_path, - &code, - job_dir, - lock.as_ref(), - &shared_mount, - base_internal_url, - worker_name, - envs, - occupancy_metrics, - )) - .await + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + Box::pin(handle_rust_job( + mem_peak, + canceled_by, + job, + conn, + client, + parent_runnable_path, + &code, + job_dir, + lock.as_ref(), + &shared_mount, + base_internal_url, + worker_name, + envs, + occupancy_metrics, + )) + .await + } } Some(ScriptLang::Ansible) => { #[cfg(not(feature = "python"))] @@ -3746,26 +3833,38 @@ mount {{ )); #[cfg(feature = "python")] - Box::pin(handle_ansible_job( - lock.as_ref(), - job_dir, - worker_dir, - worker_name, - job, - mem_peak, - canceled_by, - conn, - client, - parent_runnable_path, - &code, - &shared_mount, - base_internal_url, - envs, - occupancy_metrics, - )) - .await + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + Box::pin(handle_ansible_job( + lock.as_ref(), + job_dir, + worker_dir, + worker_name, + job, + mem_peak, + canceled_by, + conn, + client, + parent_runnable_path, + &code, + &shared_mount, + base_internal_url, + envs, + occupancy_metrics, + )) + .await + } } Some(ScriptLang::CSharp) => { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } Box::pin(handle_csharp_job( mem_peak, canceled_by, @@ -3791,23 +3890,30 @@ mount {{ ); #[cfg(feature = "nu")] - Box::pin(handle_nu_job(JobHandlerInputNu { - mem_peak, - canceled_by, - job, - conn, - client, - parent_runnable_path, - inner_content: &code, - job_dir, - requirements_o: lock.as_ref(), - shared_mount: &shared_mount, - base_internal_url, - worker_name, - envs, - occupancy_metrics, - })) - .await + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + Box::pin(handle_nu_job(JobHandlerInputNu { + mem_peak, + canceled_by, + job, + conn, + client, + parent_runnable_path, + inner_content: &code, + job_dir, + requirements_o: lock.as_ref(), + shared_mount: &shared_mount, + base_internal_url, + worker_name, + envs, + occupancy_metrics, + })) + .await + } } Some(ScriptLang::Java) => { #[cfg(not(feature = "java"))] @@ -3817,23 +3923,30 @@ mount {{ .into()); #[cfg(feature = "java")] - Box::pin(handle_java_job(JobHandlerInputJava { - mem_peak, - canceled_by, - job, - conn, - client, - parent_runnable_path, - inner_content: &code, - job_dir, - requirements_o: lock.as_ref(), - shared_mount: &shared_mount, - base_internal_url, - worker_name, - envs, - occupancy_metrics, - })) - .await + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + Box::pin(handle_java_job(JobHandlerInputJava { + mem_peak, + canceled_by, + job, + conn, + client, + parent_runnable_path, + inner_content: &code, + job_dir, + requirements_o: lock.as_ref(), + shared_mount: &shared_mount, + base_internal_url, + worker_name, + envs, + occupancy_metrics, + })) + .await + } } Some(ScriptLang::Ruby) => { #[cfg(not(feature = "ruby"))] @@ -3843,23 +3956,30 @@ mount {{ .into()); #[cfg(feature = "ruby")] - Box::pin(handle_ruby_job(JobHandlerInputRuby { - mem_peak, - canceled_by, - job, - conn, - client, - parent_runnable_path, - inner_content: &code, - job_dir, - requirements_o: lock.as_ref(), - shared_mount: &shared_mount, - base_internal_url, - worker_name, - envs, - occupancy_metrics, - })) - .await + { + if run_inline { + return Err(Error::internal_err( + "Inline execution is not supported for this language".to_string(), + )); + } + Box::pin(handle_ruby_job(JobHandlerInputRuby { + mem_peak, + canceled_by, + job, + conn, + client, + parent_runnable_path, + inner_content: &code, + job_dir, + requirements_o: lock.as_ref(), + shared_mount: &shared_mount, + base_internal_url, + worker_name, + envs, + occupancy_metrics, + })) + .await + } } // for related places search: ADD_NEW_LANG _ => panic!("unreachable, language is not supported: {language:#?}"), @@ -4019,6 +4139,7 @@ pub fn init_worker_internal_server_inline_utils( &None, &None, &None, + true, ) .await }) diff --git a/typescript-client/sqlUtils.ts b/typescript-client/sqlUtils.ts index 4376da5945..b0c1418ee6 100644 --- a/typescript-client/sqlUtils.ts +++ b/typescript-client/sqlUtils.ts @@ -51,12 +51,12 @@ function sqlProviderImpl( ducklake: (i: number) => `-- $arg${i + 1}`, }[provider]; let content = values.map((_, i) => formatArg(i)).join("\n") + "\n"; + if (provider === "ducklake") + content += `ATTACH 'ducklake://${name}' AS dl;USE dl;\n`; for (let i = 0; i < strings.length; i++) { content += strings[i]; if (i !== strings.length - 1) content += `$${i + 1}`; } - if (provider === "ducklake") - content += `ATTACH 'ducklake://${name}' AS dl;USE dl;\n`; const args = { ...Object.fromEntries(values.map((v, i) => [`arg${i + 1}`, v])),