run_inline param

This commit is contained in:
Diego Imbert
2025-11-27 12:27:06 +01:00
parent cc92c15e15
commit 1ffccbaf62
4 changed files with 296 additions and 165 deletions
+18 -13
View File
@@ -38,6 +38,7 @@ pub async fn do_duckdb(
#[allow(unused_variables)] column_order_ref: &mut Option<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
parent_runnable_path: Option<String>,
run_inline: bool,
) -> Result<Box<RawValue>> {
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),
+18 -13
View File
@@ -199,6 +199,7 @@ pub async fn do_postgresql(
column_order: &mut Option<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
parent_runnable_path: Option<String>,
run_inline: bool,
) -> error::Result<Box<RawValue>> {
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);
+258 -137
View File
@@ -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<Vec<String>>,
codebase: &Option<String>,
lock: &Option<String>,
run_inline: bool,
) -> error::Result<Box<RawValue>> {
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
})
+2 -2
View File
@@ -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])),