starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-07 08:02:40 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add bigquery (#1934)

Browse Source 
* feat: add bigquery

* fix: remove debug logs

* fix: add records number limit

* fix: revert unwanted changes

* feat: bigquery enterprise only

* fix: google auth only when enterprise

* fix: rename bigquery scripts
This commit is contained in:
HugoCasa
2023-07-27 14:24:27 +02:00
committed by GitHub
parent 6db3fce76d
commit c9110575cd
34 changed files with 711 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/Cargo.lock
Generated
+37
View File
@@ -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",
backend/Cargo.toml
+2 -1
View File
@@ -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"] }
samael = { version = "0.0.12", features = ["xmlsec"] }
gcp_auth = "0.9.0"
backend/migrations/20230721135049_add_bigquery_lang.down.sql
+1
View File
@@ -0,0 +1 @@
-- Add down migration script here
backend/migrations/20230721135049_add_bigquery_lang.up.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add up migration script here
ALTER TYPE SCRIPT_LANG ADD VALUE IF NOT EXISTS 'bigquery';
backend/parsers/windmill-parser-sql/src/lib.rs
+97
View File
@@ -27,6 +27,16 @@ pub fn parse_pgsql_sig(code: &str) -> anyhow::Result<MainArgSignature> {
}
}
pub fn parse_bigquery_sig(code: &str) -> anyhow::Result<MainArgSignature> {
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<Option<Vec<Arg>>> {
@@ -115,6 +128,36 @@ fn parse_pg_file(code: &str) -> anyhow::Result<Option<Vec<Arg>>> {
Ok(Some(args))
}
fn parse_bigquery_file(code: &str) -> anyhow::Result<Option<Vec<Arg>>> {
let mut args: Vec<Arg> = 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::<i64>().ok().map(|x| json!(x)),
Typ::Float => x.parse::<f64>().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(())
}
}
backend/parsers/windmill-parser-wasm/pkg/package.json
+1 -1
View File
@@ -3,7 +3,7 @@
"collaborators": [
"Ruben Fiszel <ruben@windmill.dev>"
],
"version": "1.127.0",
"version": "1.134.2",
"files": [
"windmill_parser_wasm_bg.wasm",
"windmill_parser_wasm.js",
backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.d.ts
Vendored
+6
View File
@@ -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;
backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.js
+27 -4
View File
@@ -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;
backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm
BIN
View File
Binary file not shown.
backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm_bg.wasm.d.ts
Vendored
+1
View File
@@ -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;
backend/parsers/windmill-parser-wasm/src/lib.rs
+5
View File
@@ -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))
}
backend/windmill-api/openapi.yaml
+5 -5
View File
@@ -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:
backend/windmill-api/src/workspaces.rs
+1
View File
@@ -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",
};
backend/windmill-common/src/scripts.rs
+2
View File
@@ -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",
}
}
}
backend/windmill-queue/src/jobs.rs
+1
View File
@@ -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(),
backend/windmill-worker/Cargo.toml
+2 -1
View File
@@ -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
backend/windmill-worker/src/bigquery_executor.rs
+332
View File
@@ -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<Vec<BigqueryResponseRow>>,
totalRows: Option<Value>,
schema: Option<BigqueryResponseSchema>,
jobComplete: bool,
}
#[derive(Deserialize)]
struct BigqueryResponseRow {
f: Vec<BigqueryResponseValue>,
}
#[derive(Deserialize)]
struct BigqueryResponseValue {
v: Value,
}
#[derive(Deserialize)]
struct BigqueryResponseSchema {
fields: Vec<BigqueryResponseSchemaField>,
}
#[derive(Deserialize)]
struct BigqueryResponseSchemaField {
name: String,
r#type: String,
fields: Option<Vec<BigqueryResponseSchemaField>>,
}
#[derive(Deserialize)]
struct BigqueryErrorResponse {
error: BigqueryError,
}
#[derive(Deserialize)]
struct BigqueryError {
message: String,
}
pub async fn do_bigquery(
job: QueuedJob,
client: &AuthedClient,
db: &sqlx::Pool<sqlx::Postgres>,
) -> windmill_common::error::Result<JobCompleted> {
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<Value> = 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::<Vec<Value>>()
}
})
} 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::<BigqueryResponse>()
.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::<i64>()
.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::<BigqueryErrorResponse>().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<Value, Error> {
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::<Vec<&str>>();
match arr.as_slice() {
[date, _] => {
v = date.to_string();
}
_ => {}
}
}
"time" => {
let arr = v.split("T").collect::<Vec<&str>>();
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::<BigqueryResponseValue>(x.clone())
.ok()
.unwrap_or(BigqueryResponseValue { v: json!({}) })
.v,
typ,
schema,
)
})
.collect::<Vec<Value>>(),
);
}
match typ.to_lowercase().as_str() {
"struct" | "record" => {
let mut nested_row_map = serde_json::Map::new();
serde_json::from_value::<BigqueryResponseRow>(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::<bool>().ok().unwrap_or(false)),
"float" | "float64" => json!(str_value.parse::<f64>().ok().unwrap_or(0.0)),
"int64" | "integer" | "timestamp" => json!(str_value.parse::<i64>().ok().unwrap_or(0)),
"json" => serde_json::from_str(&str_value).ok().unwrap_or(json!({})),
_ => value.clone(),
}
}
backend/windmill-worker/src/lib.rs
+3
View File
@@ -1,3 +1,6 @@
#[cfg(feature = "enterprise")]
mod bigquery_executor;
mod common;
mod global_cache;
mod go_executor;
backend/windmill-worker/src/worker.rs
+51 -2
View File
@@ -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<Postgres>, job: &QueuedJob) -> Arc<RwLock<String>> {
let rw_lock = Arc::new(RwLock::new(String::new()));
// skipping test runs
@@ -352,6 +355,7 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
Some(ScriptLang::Nativets),
Some(ScriptLang::Postgresql),
Some(ScriptLang::Mysql),
Some(ScriptLang::Bigquery),
Some(ScriptLang::Bun)];
let worker_execution_duration: HashMap<_, _> = all_langs.clone().into_iter().map(|x| (x.clone(), prometheus::register_histogram!(
@@ -1107,6 +1111,39 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
};
});
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<Postgres> = db.clone();
tokio::task::spawn(async move {
let jc: std::result::Result<JobCompleted, Error> = 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<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
};
});
return Ok(());
}
}
};
@@ -1228,6 +1265,17 @@ async fn handle_queued_job<R: rsmq_async::RsmqConnection + Send + Sync + Clone>(
} 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()),
cli/dev.ts
+2
View File
@@ -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 = {
cli/sync.ts
+1
View File
@@ -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}`;
frontend/package-lock.json
Generated
+7 -7
View File
@@ -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",
frontend/package.json
+1 -1
View File
@@ -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"
frontend/src/lib/components/ScriptBuilder.svelte
+24 -15
View File
@@ -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 @@
<div class=" grid grid-cols-3 gap-2">
{#each langs as [label, lang]}
{@const isPicked = script.language == lang && template == 'script'}
<Button
size="sm"
variant="border"
color={isPicked ? 'blue' : 'light'}
btnClasses={isPicked ? '!border-2 !bg-blue-50/75 dark:!bg-frost-900/75' : 'm-[1px]'}
on:click={() => {
template = 'script'
initContent(lang, script.kind, template)
script.language = lang
}}
disabled={lockedLanguage}
>
<LanguageIcon {lang} />
<span class="ml-2 py-2 truncate">{label}</span>
</Button>
<Popover disablePopup={!enterpriseLangs.includes(lang) || !!$enterpriseLicense}>
<Button
size="sm"
variant="border"
color={isPicked ? 'blue' : 'light'}
btnClasses={isPicked ? '!border-2 !bg-blue-50/75 dark:!bg-frost-900/75' : 'm-[1px]'}
on:click={() => {
template = 'script'
initContent(lang, script.kind, template)
script.language = lang
}}
disabled={lockedLanguage || (enterpriseLangs.includes(lang) && !$enterpriseLicense)}
>
<LanguageIcon {lang} />
<span class="ml-2 py-2 truncate">{label}</span>
</Button>
<svelte:fragment slot="text"
>{label} is only available with an enterprise license</svelte:fragment
>
</Popover>
{/each}
<Button
size="sm"
frontend/src/lib/components/apps/editor/inlineScriptsPanel/EmptyInlineScript.svelte
+2 -1
View File
@@ -100,7 +100,8 @@
'bash',
'nativets',
'postgresql',
'mysql'
'mysql',
'bigquery'
] as Script.language[]
</script>
frontend/src/lib/components/common/languageIcons/LanguageIcon.svelte
+2
View File
@@ -10,6 +10,7 @@
import RestIcon from '$lib/components/icons/RestIcon.svelte'
import { Script } from '$lib/gen'
import PowershellIcon from '$lib/components/icons/PowershellIcon.svelte'
import BigQueryIcon from '$lib/components/icons/BigQueryIcon.svelte'
export let lang:
| SupportedLanguage
@@ -46,6 +47,7 @@
bash: BashIcon,
pgsql: PostgresIcon,
mysql: MySQLIcon,
bigquery: BigQueryIcon,
javascript: JavaScript,
fetch: FetchIcon,
docker: DockerIcon,
frontend/src/lib/components/flows/content/FlowInputs.svelte
+11
View File
@@ -165,6 +165,17 @@
})
}}
/>
<FlowScriptPicker
label="BigQuery"
lang={Script.language.BIGQUERY}
on:click={() => {
dispatch('new', {
language: RawScript.language.BIGQUERY,
kind,
subkind: 'flow'
})
}}
/>
<FlowScriptPicker
label={`Docker`}
frontend/src/lib/components/flows/pickers/FlowScriptPicker.svelte
+28 -20
View File
@@ -1,7 +1,9 @@
<script lang="ts">
import type { SupportedLanguage } from '$lib/common'
import Popover from '$lib/components/Popover.svelte'
import Button from '$lib/components/common/button/Button.svelte'
import LanguageIcon from '$lib/components/common/languageIcons/LanguageIcon.svelte'
import { enterpriseLicense } from '$lib/stores'
import type { IconDefinition } from '@fortawesome/free-solid-svg-icons'
export let disabled: boolean = false
@@ -17,25 +19,31 @@
| undefined = undefined
export let icon: IconDefinition | undefined = undefined
export let iconColor: string | undefined = undefined
const enterpriseLangs = ['bigquery']
</script>
<Button
{disabled}
btnClasses="w-24 truncate"
on:click
size="sm"
spacingSize="md"
variant="border"
color="light"
startIcon={{
icon,
classes: iconColor
}}
>
<div class="flex justify-center flex-col items-center gap-2">
{#if lang}
<LanguageIcon {lang} />
{/if}
<span class="text-xs">{label}</span>
</div>
</Button>
<Popover disablePopup={!enterpriseLangs.includes(lang || '') || !!$enterpriseLicense}>
<Button
btnClasses="w-24 truncate"
on:click
size="sm"
spacingSize="md"
variant="border"
color="light"
startIcon={{
icon,
classes: iconColor
}}
disabled={disabled || (enterpriseLangs.includes(lang || '') && !$enterpriseLicense)}
>
<div class="flex justify-center flex-col items-center gap-2">
{#if lang}
<LanguageIcon {lang} />
{/if}
<span class="text-xs">{label}</span>
</div>
</Button>
<svelte:fragment slot="text">{label} is only available with an enterprise license</svelte:fragment
>
</Popover>
frontend/src/lib/components/icons/BigQueryIcon.svelte
+38
View File
@@ -0,0 +1,38 @@
<script lang="ts">
export let height = '24px'
export let width = '24px'
</script>
<svg xmlns="http://www.w3.org/2000/svg" {width} {height} viewBox="0 0 24 24"
><defs
><style>
.cls-1 {
fill: #aecbfa;
}
.cls-1,
.cls-2,
.cls-3 {
fill-rule: evenodd;
}
.cls-2 {
fill: #669df6;
}
.cls-3 {
fill: #4285f4;
}
</style></defs
><title>Icon_24px_BigQuery_Color</title><g data-name="Product Icons"
><g
><path class="cls-1" d="M6.73,10.83v2.63A4.91,4.91,0,0,0,8.44,15.2V10.83Z" /><path
class="cls-2"
d="M9.89,8.41v7.53A7.62,7.62,0,0,0,11,16,8,8,0,0,0,12,16V8.41Z"
/><path class="cls-1" d="M13.64,11.86v3.29a5,5,0,0,0,1.7-1.82V11.86Z" /><path
class="cls-3"
d="M17.74,16.32l-1.42,1.42a.42.42,0,0,0,0,.6l3.54,3.54a.42.42,0,0,0,.59,0l1.43-1.43a.42.42,0,0,0,0-.59l-3.54-3.54a.42.42,0,0,0-.6,0"
/><path
class="cls-2"
d="M11,2a9,9,0,1,0,9,9,9,9,0,0,0-9-9m0,15.69A6.68,6.68,0,1,1,17.69,11,6.68,6.68,0,0,1,11,17.69"
/></g
></g
></svg
>
frontend/src/lib/infer.ts
+8 -1
View File
@@ -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') {
frontend/src/lib/script_helpers.ts
+7
View File
@@ -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 {
frontend/src/lib/scripts.ts
+2
View File
@@ -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') {
frontend/src/routes/(root)/(logged)/run/[...run]/+page.svelte
+1 -1
View File
@@ -291,7 +291,7 @@
<Badge color="blue">{job.job_kind}</Badge>
</div>
{/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)}
<div>
<Badge color="indigo">Worker group: {job.tag}</Badge>
</div>
openflow.openapi.yaml
+1
View File
@@ -194,6 +194,7 @@ components:
- bash
- postgresql
- mysql
- bigquery
- graphql
- nativets
path: