diff --git a/backend/Cargo.lock b/backend/Cargo.lock index d3ca4a8b0e..763a0354fa 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -2087,6 +2087,31 @@ dependencies = [ "slab", ] +[[package]] +name = "gcp_auth" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d3b20d3058763d26d88e6e7a49998841e5296735b00dbfb064ff7cb142933dd" +dependencies = [ + "async-trait", + "base64 0.21.2", + "dirs-next", + "hyper", + "hyper-rustls", + "ring", + "rustls", + "rustls-pemfile", + "serde", + "serde_json", + "thiserror", + "time 0.3.23", + "tokio", + "tracing", + "tracing-futures", + "url", + "which", +] + [[package]] name = "generic-array" version = "0.14.7" @@ -2435,6 +2460,7 @@ dependencies = [ "http", "hyper", "rustls", + "rustls-native-certs", "tokio", "tokio-rustls", ] @@ -6355,6 +6381,16 @@ dependencies = [ "valuable", ] +[[package]] +name = "tracing-futures" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2" +dependencies = [ + "pin-project", + "tracing", +] + [[package]] name = "tracing-log" version = "0.1.3" @@ -7277,6 +7313,7 @@ dependencies = [ "dotenv", "dyn-iter", "futures", + "gcp_auth", "git-version", "itertools 0.11.0", "lazy_static", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 4b7062a859..ec171c41e2 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -182,4 +182,5 @@ tokio-postgres = {version = "^0.7", features = ["array-impls", "with-serde_json- mysql_async = { version = "*", default-features = false, features = ["minimal", "default", "native-tls-tls"]} postgres-native-tls = "^0" native-tls = "^0" -samael = { version = "0.0.12", features = ["xmlsec"] } \ No newline at end of file +samael = { version = "0.0.12", features = ["xmlsec"] } +gcp_auth = "0.9.0" \ No newline at end of file diff --git a/backend/migrations/20230721135049_add_bigquery_lang.down.sql b/backend/migrations/20230721135049_add_bigquery_lang.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230721135049_add_bigquery_lang.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230721135049_add_bigquery_lang.up.sql b/backend/migrations/20230721135049_add_bigquery_lang.up.sql new file mode 100644 index 0000000000..1ddebe49f5 --- /dev/null +++ b/backend/migrations/20230721135049_add_bigquery_lang.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TYPE SCRIPT_LANG ADD VALUE IF NOT EXISTS 'bigquery'; \ No newline at end of file diff --git a/backend/parsers/windmill-parser-sql/src/lib.rs b/backend/parsers/windmill-parser-sql/src/lib.rs index adb82f73af..b73f524af7 100644 --- a/backend/parsers/windmill-parser-sql/src/lib.rs +++ b/backend/parsers/windmill-parser-sql/src/lib.rs @@ -27,6 +27,16 @@ pub fn parse_pgsql_sig(code: &str) -> anyhow::Result { } } +pub fn parse_bigquery_sig(code: &str) -> anyhow::Result { + let parsed = parse_bigquery_file(&code)?; + if let Some(x) = parsed { + let args = x; + Ok(MainArgSignature { star_args: false, star_kwargs: false, args }) + } else { + Err(anyhow!("Error parsing sql".to_string())) + } +} + lazy_static::lazy_static! { static ref RE_CODE_PGSQL: Regex = Regex::new(r#"(?m)\$(\d+)(?:::(\w+))?"#).unwrap(); @@ -35,6 +45,9 @@ lazy_static::lazy_static! { static ref RE_ARG_PGSQL: Regex = Regex::new(r#"(?m)^-- \$(\d+) (\w+)(?: ?\= ?(.+))? *[\r\n$]"#).unwrap(); + // -- @name (type) = default + static ref RE_ARG_BIGQUERY: Regex = Regex::new(r#"(?m)^-- @(\w+) \((\w+(?:\[\])?)\)(?: ?\= ?(.+))? *[\r\n$]"#).unwrap(); + } fn parse_mysql_file(code: &str) -> anyhow::Result>> { @@ -115,6 +128,36 @@ fn parse_pg_file(code: &str) -> anyhow::Result>> { Ok(Some(args)) } +fn parse_bigquery_file(code: &str) -> anyhow::Result>> { + let mut args: Vec = vec![]; + + for cap in RE_ARG_BIGQUERY.captures_iter(code) { + let name = cap.get(1).map(|x| x.as_str().to_string()).unwrap(); + let typ = cap + .get(2) + .map(|x| x.as_str().to_string().to_lowercase()) + .unwrap(); + let default = cap.get(3).map(|x| x.as_str().to_string()); + let has_default = default.is_some(); + let parsed_typ = parse_bigquery_typ(typ.as_str()); + + let parsed_default = default.and_then(|x| match parsed_typ { + Typ::Int => x.parse::().ok().map(|x| json!(x)), + Typ::Float => x.parse::().ok().map(|x| json!(x)), + _ => Some(json!(x)), + }); + args.push(Arg { + name, + typ: parsed_typ, + default: parsed_default, + otyp: Some(typ), + has_default, + }); + } + + Ok(Some(args)) +} + pub fn parse_mysql_typ(typ: &str) -> Typ { match typ { "varchar" | "char" | "binary" | "varbinary" | "blob" | "text" | "enum" | "set" => { @@ -145,6 +188,25 @@ pub fn parse_pg_typ(typ: &str) -> Typ { _ => Typ::Str(None), } } + +pub fn parse_bigquery_typ(typ: &str) -> Typ { + if typ.ends_with("[]") { + let base_typ = parse_bigquery_typ(typ.strip_suffix("[]").unwrap_or(typ)); + Typ::List(Box::new(base_typ)) + } else { + match typ { + "string" => Typ::Str(None), + "bytes" => Typ::Bytes, + "json" => Typ::Object(vec![]), + "timestamp" | "date" | "time" | "datetime" => Typ::Datetime, + "integer" | "int64" => Typ::Int, + "float" | "float64" | "numeric" | "bignumeric" => Typ::Float, + "boolean" | "bool" => Typ::Bool, + _ => Typ::Str(None), + } + } +} + #[cfg(test)] mod tests { @@ -182,4 +244,39 @@ SELECT * FROM table WHERE token=$1::TEXT AND image=$2::BIGINT Ok(()) } + + #[test] + fn test_parse_bigquery_sig() -> anyhow::Result<()> { + let code = r#" +-- @token (string) +-- @image (int64) +SELECT * FROM table WHERE token=@token AND image=@image +"#; + //println!("{}", serde_json::to_string()?); + assert_eq!( + parse_bigquery_sig(code)?, + MainArgSignature { + star_args: false, + star_kwargs: false, + args: vec![ + Arg { + otyp: Some("string".to_string()), + name: "token".to_string(), + typ: Typ::Str(None), + default: None, + has_default: false + }, + Arg { + otyp: Some("int64".to_string()), + name: "image".to_string(), + typ: Typ::Int, + default: None, + has_default: false + }, + ] + } + ); + + Ok(()) + } } diff --git a/backend/parsers/windmill-parser-wasm/pkg/package.json b/backend/parsers/windmill-parser-wasm/pkg/package.json index 4e32534603..99897334c1 100644 --- a/backend/parsers/windmill-parser-wasm/pkg/package.json +++ b/backend/parsers/windmill-parser-wasm/pkg/package.json @@ -3,7 +3,7 @@ "collaborators": [ "Ruben Fiszel " ], - "version": "1.127.0", + "version": "1.134.2", "files": [ "windmill_parser_wasm_bg.wasm", "windmill_parser_wasm.js", diff --git a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.d.ts b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.d.ts index cb497f10ef..d9da9d77a7 100644 --- a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.d.ts +++ b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.d.ts @@ -30,6 +30,11 @@ export function parse_sql(code: string): string; * @returns {string} */ export function parse_mysql(code: string): string; +/** +* @param {string} code +* @returns {string} +*/ +export function parse_bigquery(code: string): string; export type InitInput = RequestInfo | URL | Response | BufferSource | WebAssembly.Module; @@ -41,6 +46,7 @@ export interface InitOutput { readonly parse_python: (a: number, b: number, c: number) => void; readonly parse_sql: (a: number, b: number, c: number) => void; readonly parse_mysql: (a: number, b: number, c: number) => void; + readonly parse_bigquery: (a: number, b: number, c: number) => void; readonly __wbindgen_malloc: (a: number, b: number) => number; readonly __wbindgen_realloc: (a: number, b: number, c: number, d: number) => number; readonly __wbindgen_add_to_stack_pointer: (a: number) => number; diff --git a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.js b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.js index 23ae77d36d..c28033a55f 100644 --- a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.js +++ b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.js @@ -335,6 +335,29 @@ export function parse_mysql(code) { } } +/** +* @param {string} code +* @returns {string} +*/ +export function parse_bigquery(code) { + let deferred2_0; + let deferred2_1; + try { + const retptr = wasm.__wbindgen_add_to_stack_pointer(-16); + const ptr0 = passStringToWasm0(code, wasm.__wbindgen_malloc, wasm.__wbindgen_realloc); + const len0 = WASM_VECTOR_LEN; + wasm.parse_bigquery(retptr, ptr0, len0); + var r0 = getInt32Memory0()[retptr / 4 + 0]; + var r1 = getInt32Memory0()[retptr / 4 + 1]; + deferred2_0 = r0; + deferred2_1 = r1; + return getStringFromWasm0(r0, r1); + } finally { + wasm.__wbindgen_add_to_stack_pointer(16); + wasm.__wbindgen_free(deferred2_0, deferred2_1, 1); + } +} + function handleError(f, args) { try { return f.apply(this, args); @@ -377,10 +400,6 @@ async function __wbg_load(module, imports) { function __wbg_get_imports() { const imports = {}; imports.wbg = {}; - imports.wbg.__wbg_eval_5b4d65f6480422c2 = function(arg0, arg1) { - const ret = eval(getStringFromWasm0(arg0, arg1)); - return addHeapObject(ret); - }; imports.wbg.__wbindgen_object_drop_ref = function(arg0) { takeObject(arg0); }; @@ -432,6 +451,10 @@ function __wbg_get_imports() { const ret = new Error(getStringFromWasm0(arg0, arg1)); return addHeapObject(ret); }; + imports.wbg.__wbg_eval_bb7d5dc518fdea6d = function(arg0, arg1) { + const ret = eval(getStringFromWasm0(arg0, arg1)); + return addHeapObject(ret); + }; imports.wbg.__wbindgen_jsval_loose_eq = function(arg0, arg1) { const ret = getObject(arg0) == getObject(arg1); return ret; diff --git a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm index c03e13bf60..e0a10e02de 100644 Binary files a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm and b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm differ diff --git a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm.d.ts b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm.d.ts index 9142247cee..eede785741 100644 --- a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm.d.ts +++ b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm.d.ts @@ -7,6 +7,7 @@ export function parse_go(a: number, b: number, c: number): void; export function parse_python(a: number, b: number, c: number): void; export function parse_sql(a: number, b: number, c: number): void; export function parse_mysql(a: number, b: number, c: number): void; +export function parse_bigquery(a: number, b: number, c: number): void; export function __wbindgen_malloc(a: number, b: number): number; export function __wbindgen_realloc(a: number, b: number, c: number, d: number): number; export function __wbindgen_add_to_stack_pointer(a: number): number; diff --git a/backend/parsers/windmill-parser-wasm/src/lib.rs b/backend/parsers/windmill-parser-wasm/src/lib.rs index cb1257aaf9..14bb547212 100644 --- a/backend/parsers/windmill-parser-wasm/src/lib.rs +++ b/backend/parsers/windmill-parser-wasm/src/lib.rs @@ -39,3 +39,8 @@ pub fn parse_sql(code: &str) -> String { pub fn parse_mysql(code: &str) -> String { wrap_sig(windmill_parser_sql::parse_mysql_sig(code)) } + +#[wasm_bindgen] +pub fn parse_bigquery(code: &str) -> String { + wrap_sig(windmill_parser_sql::parse_bigquery_sig(code)) +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index c9908318f0..764f556ceb 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5672,7 +5672,7 @@ components: language: type: string enum: - [python3, deno, go, bash, postgresql, mysql, graphql, nativets, bun] + [python3, deno, go, bash, postgresql, mysql, bigquery, graphql, nativets, bun] kind: type: string enum: [script, failure, trigger, command, approval] @@ -5737,7 +5737,7 @@ components: language: type: string enum: - [python3, deno, go, bash, postgresql, mysql, graphql, nativets, bun] + [python3, deno, go, bash, postgresql, mysql, bigquery, graphql, nativets, bun] kind: type: string enum: [script, failure, trigger, command, approval] @@ -5908,7 +5908,7 @@ components: language: type: string enum: - [python3, deno, go, bash, postgresql, mysql, graphql, nativets, bun] + [python3, deno, go, bash, postgresql, mysql, bigquery, graphql, nativets, bun] email: type: string visible_to_owner: @@ -5999,7 +5999,7 @@ components: language: type: string enum: - [python3, deno, go, bash, postgresql, mysql, graphql, nativets, bun] + [python3, deno, go, bash, postgresql, mysql, bigquery, graphql, nativets, bun] is_skipped: type: boolean email: @@ -6428,7 +6428,7 @@ components: language: type: string enum: - [python3, deno, go, bash, postgresql, mysql, graphql, nativets, bun] + [python3, deno, go, bash, postgresql, mysql, bigquery, graphql, nativets, bun] tag: type: string kind: diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 55191b79be..f7bae77a74 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -1476,6 +1476,7 @@ async fn tarball_workspace( ScriptLang::Bash => "sh", ScriptLang::Postgresql => "pg.sql", ScriptLang::Mysql => "my.sql", + ScriptLang::Bigquery => "bq.sql", ScriptLang::Nativets => "fetch.ts", ScriptLang::Bun => "bun.ts", }; diff --git a/backend/windmill-common/src/scripts.rs b/backend/windmill-common/src/scripts.rs index 38612d3822..ca52e0f270 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -33,6 +33,7 @@ pub enum ScriptLang { Postgresql, Bun, Mysql, + Bigquery, } impl ScriptLang { @@ -46,6 +47,7 @@ impl ScriptLang { ScriptLang::Bash => "bash", ScriptLang::Postgresql => "postgresql", ScriptLang::Mysql => "mysql", + ScriptLang::Bigquery => "bigquery", } } } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 86b19ad507..4875cc54d8 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -74,6 +74,7 @@ lazy_static::lazy_static! { "graphql".to_string(), "bun".to_string(), "postgresql".to_string(), + "bigquery".to_string(), "dependency".to_string(), "flow".to_string(), "hub".to_string(), diff --git a/backend/windmill-worker/Cargo.toml b/backend/windmill-worker/Cargo.toml index ad6de0fda2..10160f7d96 100644 --- a/backend/windmill-worker/Cargo.toml +++ b/backend/windmill-worker/Cargo.toml @@ -11,7 +11,7 @@ path = "src/lib.rs" [features] default = [] deno-lock = [] -enterprise = ["windmill-queue/enterprise"] +enterprise = ["windmill-queue/enterprise", "dep:gcp_auth"] [dependencies] windmill-queue.workspace = true @@ -63,6 +63,7 @@ postgres-native-tls.workspace = true native-tls.workspace = true mysql_async.workspace = true base64.workspace = true +gcp_auth = { workspace = true, optional = true } [build-dependencies] deno_fetch.workspace = true diff --git a/backend/windmill-worker/src/bigquery_executor.rs b/backend/windmill-worker/src/bigquery_executor.rs new file mode 100644 index 0000000000..7f4d074855 --- /dev/null +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -0,0 +1,332 @@ +use serde_json::{json, Value}; +use windmill_common::error::Error; +use windmill_common::jobs::QueuedJob; +use windmill_parser_sql::parse_bigquery_sig; +use windmill_queue::HTTP_CLIENT; + +use serde::Deserialize; + +use crate::{get_content, transform_json_value, AuthedClient, JobCompleted}; + +use gcp_auth::{AuthenticationManager, CustomServiceAccount}; + +#[allow(non_snake_case)] +#[derive(Deserialize)] +struct BigqueryResponse { + rows: Option>, + totalRows: Option, + schema: Option, + jobComplete: bool, +} + +#[derive(Deserialize)] +struct BigqueryResponseRow { + f: Vec, +} + +#[derive(Deserialize)] +struct BigqueryResponseValue { + v: Value, +} + +#[derive(Deserialize)] +struct BigqueryResponseSchema { + fields: Vec, +} + +#[derive(Deserialize)] +struct BigqueryResponseSchemaField { + name: String, + r#type: String, + fields: Option>, +} + +#[derive(Deserialize)] +struct BigqueryErrorResponse { + error: BigqueryError, +} + +#[derive(Deserialize)] +struct BigqueryError { + message: String, +} + +pub async fn do_bigquery( + job: QueuedJob, + client: &AuthedClient, + db: &sqlx::Pool, +) -> windmill_common::error::Result { + let args = if let Some(args) = &job.args { + Some(transform_json_value("args", client, &job.workspace_id, args.clone()).await?) + } else { + None + }; + + let bigquery_args: Value = serde_json::from_value(args.unwrap_or_else(|| json!({}))) + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + let database = bigquery_args + .get("database") + .unwrap_or(&json!({})) + .to_string(); + + if database == "{}" { + return Err(Error::ExecutionErr("Invalid database".to_string())); + } + + let service_account = CustomServiceAccount::from_json(&database) + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + let authentication_manager = AuthenticationManager::from(service_account); + let scopes = &["https://www.googleapis.com/auth/bigquery"]; + let token = authentication_manager + .get_token(scopes) + .await + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + let args = &job + .args + .clone() + .unwrap_or_else(|| json!({})) + .as_object() + .map(|x| x.to_owned()) + .unwrap_or_else(|| json!({}).as_object().unwrap().to_owned()); + + let mut statement_values: Vec = vec![]; + + let query = get_content(&job, db).await?; + + let sig = parse_bigquery_sig(&query) + .map_err(|x| Error::ExecutionErr(x.to_string()))? + .args; + + for arg in &sig { + let arg_t = arg.otyp.clone().unwrap_or_else(|| "string".to_string()); + let arg_n = arg.clone().name; + let arg_v = args.get(&arg.name).cloned().unwrap_or(json!("")); + let bigquery_v = if arg_t.ends_with("[]") { + let base_type = arg_t.strip_suffix("[]").unwrap_or(&arg_t); + json!({ + "name": arg.name, + "parameterType": { + "type": "ARRAY", + "arrayType": { + "type": base_type.to_uppercase() + } + }, + "parameterValue": { + "arrayValues": args + .get(&arg.name) + .unwrap_or(&json!([])) + .as_array() + .unwrap_or(&vec![]) + .iter() + .map(|x| { + convert_val(base_type.to_string(), x.clone()).ok().unwrap() + }) + .collect::>() + } + }) + } else { + json!({ + "name": arg_n, + "parameterType": { + "type": arg_t.to_uppercase() + }, + "parameterValue": { + "value": convert_val(arg_t, arg_v)?, + } + }) + }; + + statement_values.push(bigquery_v); + } + + let response = HTTP_CLIENT + .post( + "https://bigquery.googleapis.com/bigquery/v2/projects/".to_string() + + authentication_manager + .project_id() + .await + .map_err(|e| Error::ExecutionErr(e.to_string()))? + .as_str() + + "/queries", + ) + .bearer_auth(token.as_str()) + .json(&json!({ + "query": query, + "useLegacySql": false, + "maxResults": 10000, + "timeoutMs": 10000, // default + "queryParameters": statement_values, + })) + .send() + .await + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + match response.error_for_status_ref() { + Ok(_) => { + let result = response + .json::() + .await + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + if !result.jobComplete { + return Err(Error::ExecutionErr( + "BigQuery API did not answer query in time".to_string(), + )); + } + + if result.rows.is_none() || result.rows.as_ref().unwrap().len() == 0 { + return Ok(JobCompleted { + job: job, + result: Value::Array(vec![]), + logs: "".to_string(), + success: true, + }); + } + + if result.schema.is_none() { + return Err(Error::ExecutionErr( + "Incomplete response from BigQuery API".to_string(), + )); + } + + if result + .totalRows + .unwrap_or(json!("")) + .as_str() + .unwrap_or("") + .parse::() + .unwrap_or(0) + > 10000 + { + return Err(Error::ExecutionErr( + "More than 10000 rows were requested, use LIMIT 10000 to limit the number of rows".to_string(), + )); + } + + let rows = result + .rows + .unwrap() + .iter() + .map(|row| { + let mut row_map = serde_json::Map::new(); + row.f + .iter() + .zip(result.schema.as_ref().unwrap().fields.iter()) + .for_each(|(field, schema)| { + row_map.insert( + schema.name.clone(), + parse_val(&field.v, &schema.r#type, &schema), + ); + }); + Value::from(row_map) + }) + .collect(); + + return Ok(JobCompleted { + job: job, + result: rows, + logs: "".to_string(), + success: true, + }); + } + Err(e) => match response.json::().await { + Ok(bq_err) => return Err(Error::ExecutionErr(bq_err.error.message)), + Err(_) => return Err(Error::ExecutionErr(e.to_string())), + }, + } +} + +fn convert_val(arg_t: String, arg_v: Value) -> Result { + match arg_t.as_str() { + "timestamp" | "datetime" | "date" | "time" => { + let mut v: String = arg_v.as_str().unwrap_or("").to_owned(); + + match arg_t.as_str() { + "timestamp" | "datetime" => { + v = v + ":00"; + } + "date" => { + let arr = v.split("T").collect::>(); + match arr.as_slice() { + [date, _] => { + v = date.to_string(); + } + _ => {} + } + } + "time" => { + let arr = v.split("T").collect::>(); + match arr.as_slice() { + [_, time] => { + v = time.to_string() + ":00"; + } + _ => {} + } + } + _ => {} + } + + Ok(json!({ "value": json!(v) })) + } + _ => { + let mut v = arg_v; + + if !v.is_string() { + // if not string, convert to string for api request + v = json!(v.to_string()); + } + + Ok(json!({ + "value": v, + } + )) + } + } +} + +fn parse_val(value: &Value, typ: &str, schema: &BigqueryResponseSchemaField) -> Value { + let str_value = value.as_str().unwrap_or("").to_string(); + + if value.is_array() { + return Value::Array( + value + .as_array() + .unwrap_or(&vec![]) + .iter() + .map(|x| { + parse_val( + &serde_json::from_value::(x.clone()) + .ok() + .unwrap_or(BigqueryResponseValue { v: json!({}) }) + .v, + typ, + schema, + ) + }) + .collect::>(), + ); + } + match typ.to_lowercase().as_str() { + "struct" | "record" => { + let mut nested_row_map = serde_json::Map::new(); + serde_json::from_value::(value.clone()) + .ok() + .unwrap_or(BigqueryResponseRow { f: vec![] }) + .f + .iter() + .zip(schema.fields.as_ref().clone().unwrap_or(&vec![]).iter()) + .for_each(|(f, s)| { + nested_row_map.insert(s.name.clone(), parse_val(&f.v, &s.r#type, &s)); + }); + Value::from(nested_row_map) + } + "bool" | "boolean" => json!(str_value.parse::().ok().unwrap_or(false)), + "float" | "float64" => json!(str_value.parse::().ok().unwrap_or(0.0)), + "int64" | "integer" | "timestamp" => json!(str_value.parse::().ok().unwrap_or(0)), + "json" => serde_json::from_str(&str_value).ok().unwrap_or(json!({})), + _ => value.clone(), + } +} diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index f046d2e36a..c778dad4ca 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -1,3 +1,6 @@ +#[cfg(feature = "enterprise")] +mod bigquery_executor; + mod common; mod global_cache; mod go_executor; diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index e719eac3c8..097913a843 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -67,9 +67,12 @@ use windmill_queue::{add_completed_job, add_completed_job_error,IDLE_WORKERS}; use crate::{ worker_flow::{ handle_flow, update_flow_status_after_job_completion, update_flow_status_in_progress, - }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, + }, python_executor::{create_dependencies_dir, pip_compile, handle_python_job, handle_python_reqs}, common::{read_result, set_logs}, go_executor::{handle_go_job, install_go_dependencies}, js_eval::{transpile_ts, eval_fetch_timeout}, pg_executor::do_postgresql, mysql_executor::do_mysql, }; +#[cfg(feature = "enterprise")] +use crate::bigquery_executor::do_bigquery; + pub async fn create_token_for_owner_in_bg(db: &Pool, job: &QueuedJob) -> Arc> { let rw_lock = Arc::new(RwLock::new(String::new())); // skipping test runs @@ -352,6 +355,7 @@ pub async fn run_worker = all_langs.clone().into_iter().map(|x| (x.clone(), prometheus::register_histogram!( @@ -1107,6 +1111,39 @@ async fn handle_queued_job( }; }); return Ok(()); + } else if job.language == Some(ScriptLang::Bigquery) { + + #[cfg(not(feature = "enterprise"))] + { + return Err(Error::ExecutionErr("Bigquery is only available with an enterprise license".to_string())); + } + + #[cfg(feature = "enterprise")] + { + wait_available_worker_for_native_job(parallel_count.clone(), &job).await; + let client = client.get_authed().await; + let db: Pool = db.clone(); + + tokio::task::spawn(async move { + let jc: std::result::Result = do_bigquery(job.clone(), &client, &db).await; + parallel_count.fetch_sub(1, Ordering::SeqCst); + + match jc { + Ok(jc) => job_completed_tx.send(jc).await.expect("send job completed"), + Err(e) => job_completed_tx.send(JobCompleted { + job: job, + result: json!({"error": { + "name": "ExecutionError", + "message": e.to_string() + }}), + logs: "".to_string(), + success: false + }).await.expect("send job completed"), + }; + }); + return Ok(()); + } + } else if job.language == Some(ScriptLang::Nativets) { wait_available_worker_for_native_job(parallel_count.clone(), &job).await; logs.push_str("\n--- FETCH TS EXECUTION ---\n"); @@ -1132,7 +1169,7 @@ async fn handle_queued_job( }; }); return Ok(()); - } + } }; @@ -1228,6 +1265,17 @@ async fn handle_queued_job( } else if job.language == Some(ScriptLang::Mysql) { let jc = do_mysql(job.clone(), &client.get_authed().await, &db).await?; Ok(jc.result) + } else if job.language == Some(ScriptLang::Bigquery) { + #[cfg(not(feature = "enterprise"))] + { + Err(Error::ExecutionErr("Bigquery is only available with an enterprise license".to_string())) + } + + #[cfg(feature = "enterprise")] + { + let jc = do_bigquery(job.clone(), &client.get_authed().await, &db).await?; + Ok(jc.result) + } } else if job.language == Some(ScriptLang::Nativets) { logs.push_str("\n--- FETCH TS EXECUTION ---\n"); let jc = do_nativets(job.clone(), logs.clone(), &client.get_authed().await, &db).await?; @@ -2337,6 +2385,7 @@ async fn capture_dependency_job( }, ScriptLang::Postgresql => Ok("".to_owned()), ScriptLang::Mysql => Ok("".to_owned()), + ScriptLang::Bigquery => Ok("".to_owned()), ScriptLang::Bash => Ok("".to_owned()), ScriptLang::Nativets => Ok("".to_owned()), diff --git a/cli/dev.ts b/cli/dev.ts index 29b72836ba..016299afc3 100644 --- a/cli/dev.ts +++ b/cli/dev.ts @@ -59,6 +59,8 @@ async function dev(opts: GlobalOptions & { filter?: string }) { : ext == "sql" ? splitted.length > 2 && splitted[splitted.length - 2] == "my" ? "mysql" + : splitted.length > 2 && splitted[splitted.length - 2] == "bigquery" + ? "bigquery" : "postgresql" : "unknown"; currentLastEdit = { diff --git a/cli/sync.ts b/cli/sync.ts index cd7a186792..d64f958bbe 100644 --- a/cli/sync.ts +++ b/cli/sync.ts @@ -135,6 +135,7 @@ function ZipFSElement(zip: JSZip, useYaml: boolean): DynFSElement { else if (language == "bash") ext = "sh"; else if (language == "postgresql") ext = "pg.sql"; else if (language == "mysql") ext = "my.sql"; + else if (language == "bigquery") ext = "bq.sql"; else if (language == "bun") ext = "bun.ts"; return `${name}.inline_script.${ext}`; diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 5978f45036..c26fdc7c52 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -39,7 +39,7 @@ "svelte-timezone-picker": "^2.0.3", "tailwind-merge": "^1.13.2", "vscode-ws-jsonrpc": "3.0.0", - "windmill-parser-wasm": "^1.127.0", + "windmill-parser-wasm": "^1.134.2", "y-monaco": "^0.1.4", "y-websocket": "^1.5.0", "yjs": "^13.6.7" @@ -8297,9 +8297,9 @@ } }, "node_modules/windmill-parser-wasm": { - "version": "1.127.0", - "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.127.0.tgz", - "integrity": "sha512-/zB/+Bjw+6w0CHuHrR7vLuXHlb1oW03p+jSAfuOIiM9MnOHk09B2qY0LokGTITuoEF1r1FxSmb8v63DIuynZGw==" + "version": "1.134.2", + "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.134.2.tgz", + "integrity": "sha512-XxAPj/JDfQVmvJCJ+8rktlAOsesvznARTPcLsBEKvN8MvG5jt2V3Xqh66MvmDVKl6QvHqOWFVetpd5LVrCDBSQ==" }, "node_modules/wordwrap": { "version": "1.0.0", @@ -14407,9 +14407,9 @@ } }, "windmill-parser-wasm": { - "version": "1.127.0", - "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.127.0.tgz", - "integrity": "sha512-/zB/+Bjw+6w0CHuHrR7vLuXHlb1oW03p+jSAfuOIiM9MnOHk09B2qY0LokGTITuoEF1r1FxSmb8v63DIuynZGw==" + "version": "1.134.2", + "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.134.2.tgz", + "integrity": "sha512-XxAPj/JDfQVmvJCJ+8rktlAOsesvznARTPcLsBEKvN8MvG5jt2V3Xqh66MvmDVKl6QvHqOWFVetpd5LVrCDBSQ==" }, "wordwrap": { "version": "1.0.0", diff --git a/frontend/package.json b/frontend/package.json index 9647f4850d..a6d62e6984 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -102,7 +102,7 @@ "svelte-timezone-picker": "^2.0.3", "tailwind-merge": "^1.13.2", "vscode-ws-jsonrpc": "3.0.0", - "windmill-parser-wasm": "^1.127.0", + "windmill-parser-wasm": "^1.134.2", "y-monaco": "^0.1.4", "y-websocket": "^1.5.0", "yjs": "^13.6.7" diff --git a/frontend/src/lib/components/ScriptBuilder.svelte b/frontend/src/lib/components/ScriptBuilder.svelte index 61149d7dcf..3e78e99c24 100644 --- a/frontend/src/lib/components/ScriptBuilder.svelte +++ b/frontend/src/lib/components/ScriptBuilder.svelte @@ -26,6 +26,7 @@ import Awareness from './Awareness.svelte' import { Icon } from 'svelte-awesome' import { fade } from 'svelte/transition' + import Popover from './Popover.svelte' export let script: NewScript export let initialPath: string = '' @@ -45,6 +46,8 @@ let editor: Editor | undefined = undefined let scriptEditor: ScriptEditor | undefined = undefined + const enterpriseLangs = ['bigquery'] + loadWorkerGroups() async function loadWorkerGroups() { @@ -66,6 +69,7 @@ } langs.push(['PostgreSQL', Script.language.POSTGRESQL]) langs.push(['MySQL', Script.language.MYSQL]) + langs.push(['BigQuery', Script.language.BIGQUERY]) if (SCRIPT_SHOW_GO) { langs.push(['Go', Script.language.GO]) } @@ -306,21 +310,26 @@
{#each langs as [label, lang]} {@const isPicked = script.language == lang && template == 'script'} - + + + {label} is only available with an enterprise license + {/each} + + + {label} is only available with an enterprise license + diff --git a/frontend/src/lib/components/icons/BigQueryIcon.svelte b/frontend/src/lib/components/icons/BigQueryIcon.svelte new file mode 100644 index 0000000000..a12761bd19 --- /dev/null +++ b/frontend/src/lib/components/icons/BigQueryIcon.svelte @@ -0,0 +1,38 @@ + + +Icon_24px_BigQuery_Color diff --git a/frontend/src/lib/infer.ts b/frontend/src/lib/infer.ts index 77fdd71c8c..fe0ed97712 100644 --- a/frontend/src/lib/infer.ts +++ b/frontend/src/lib/infer.ts @@ -9,7 +9,8 @@ import init, { parse_go, parse_python, parse_sql, - parse_mysql + parse_mysql, + parse_bigquery } from 'windmill-parser-wasm' import wasmUrl from 'windmill-parser-wasm/windmill_parser_wasm_bg.wasm?url' @@ -48,6 +49,12 @@ export async function inferArgs( } else if (language == 'mysql') { inferedSchema = JSON.parse(parse_mysql(code)) inferedSchema.args = [{ name: 'database', typ: { resource: 'mysql' } }, ...inferedSchema.args] + } else if (language == 'bigquery') { + inferedSchema = JSON.parse(parse_bigquery(code)) + inferedSchema.args = [ + { name: 'database', typ: { resource: 'gcp_service_account' } }, + ...inferedSchema.args + ] } else if (language == 'go') { inferedSchema = JSON.parse(parse_go(code)) } else if (language == 'bash') { diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index f51376b70a..ed29f5e37f 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -128,6 +128,11 @@ export const MYSQL_INIT_CODE = `-- ? name1 (text) = default arg INSERT INTO demo VALUES (?, ?) ` +export const BIGQUERY_INIT_CODE = `-- @name1 (string) = default arg +-- @name2 (integer) +INSERT INTO \`demodb.demo\` VALUES (@name1, @name2) +` + export const FETCH_INIT_CODE = `export async function main( url: string | undefined, method: 'GET' | 'POST' | 'PUT' | 'DELETE' | 'PATCH' | 'OPTIONS' = 'GET', @@ -319,6 +324,8 @@ export function initialCode( return POSTGRES_INIT_CODE } else if (language == 'mysql') { return MYSQL_INIT_CODE + } else if (language == 'bigquery') { + return BIGQUERY_INIT_CODE } else if (language == 'bun') { return BUN_INIT_CODE } else { diff --git a/frontend/src/lib/scripts.ts b/frontend/src/lib/scripts.ts index 737943651a..4782304c2d 100644 --- a/frontend/src/lib/scripts.ts +++ b/frontend/src/lib/scripts.ts @@ -18,6 +18,8 @@ export function scriptLangToEditorLang(lang: Script.language) { return 'sql' } else if (lang == 'mysql') { return 'sql' + } else if (lang == 'bigquery') { + return 'sql' } else if (lang == 'python3') { return 'python' } else if (lang == 'bash') { diff --git a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte index 60885ca3ef..7baf37748e 100644 --- a/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte @@ -291,7 +291,7 @@ {job.job_kind}
{/if} - {#if job.tag && !['deno', 'python3', 'flow', 'other', 'go', 'postgresql', 'nativets', 'bash', 'other', 'dependency'].includes(job.tag)} + {#if job.tag && !['deno', 'python3', 'flow', 'other', 'go', 'postgresql', 'mysql', 'bigquery', 'nativets', 'bash', 'other', 'dependency'].includes(job.tag)}
Worker group: {job.tag}
diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index fa27f2d58f..3b3b57acd3 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -194,6 +194,7 @@ components: - bash - postgresql - mysql + - bigquery - graphql - nativets path: