diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 23945e3d32..ba3ba77eec 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -2651,6 +2651,20 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "jsonwebtoken" +version = "8.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6971da4d9c3aa03c3d8f3ff0f4155b534aad021292003895a469716b2a230378" +dependencies = [ + "base64 0.21.2", + "pem 1.1.1", + "ring", + "serde", + "serde_json", + "simple_asn1", +] + [[package]] name = "konst" version = "0.2.19" @@ -3132,7 +3146,7 @@ dependencies = [ "mysql_common", "native-tls", "once_cell", - "pem", + "pem 2.0.1", "percent-encoding", "pin-project", "priority-queue", @@ -3465,6 +3479,15 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19b17cddbe7ec3f8bc800887bab5e717348c95ea2ca0b1bf0837fb964dc67099" +[[package]] +name = "pem" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a8835c273a76a90455d7344889b0964598e3316e2a79ede8e36f16bdcf2228b8" +dependencies = [ + "base64 0.13.1", +] + [[package]] name = "pem" version = "2.0.1" @@ -3475,6 +3498,16 @@ dependencies = [ "serde", ] +[[package]] +name = "pem" +version = "3.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed3127afbfc30b4cad60c34aeb741fb562a808642b81142bcf4afb73142da960" +dependencies = [ + "base64 0.21.2", + "serde", +] + [[package]] name = "pem-rfc7468" version = "0.6.0" @@ -5082,6 +5115,18 @@ version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f27f6278552951f1f2b8cf9da965d10969b2efdea95a6ec47987ab46edfe263a" +[[package]] +name = "simple_asn1" +version = "0.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adc4e5204eb1910f40f9cfa375f6f05b68c3abac4b6fd879c8ff5e7ae8a0a085" +dependencies = [ + "num-bigint", + "num-traits", + "thiserror", + "time 0.3.24", +] + [[package]] name = "siphasher" version = "0.3.10" @@ -7341,10 +7386,12 @@ dependencies = [ "gcp_auth", "git-version", "itertools 0.11.0", + "jsonwebtoken", "lazy_static", "mysql_async", "native-tls", "once_cell", + "pem 3.0.1", "postgres-native-tls", "prometheus", "rand 0.8.5", @@ -7353,6 +7400,7 @@ dependencies = [ "rust_decimal", "serde", "serde_json", + "sha2 0.10.7", "sqlx", "tokio", "tokio-postgres", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 61e1b8cf14..e80b46bb45 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -184,4 +184,6 @@ postgres-native-tls = "^0" native-tls = "^0" samael = { version = "0.0.12", features = ["xmlsec"] } gcp_auth = "0.9.0" -rust_decimal = {version = "1.31.0", features = ["db-postgres"]} \ No newline at end of file +rust_decimal = {version = "1.31.0", features = ["db-postgres"]} +jsonwebtoken = "8.3.0" +pem = "3.0.1" \ No newline at end of file diff --git a/backend/migrations/20230801103814_add_snowflake_lang.down.sql b/backend/migrations/20230801103814_add_snowflake_lang.down.sql new file mode 100644 index 0000000000..d2f607c5b8 --- /dev/null +++ b/backend/migrations/20230801103814_add_snowflake_lang.down.sql @@ -0,0 +1 @@ +-- Add down migration script here diff --git a/backend/migrations/20230801103814_add_snowflake_lang.up.sql b/backend/migrations/20230801103814_add_snowflake_lang.up.sql new file mode 100644 index 0000000000..59a9d8d635 --- /dev/null +++ b/backend/migrations/20230801103814_add_snowflake_lang.up.sql @@ -0,0 +1,2 @@ +-- Add up migration script here +ALTER TYPE SCRIPT_LANG ADD VALUE IF NOT EXISTS 'snowflake'; \ 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 b73f524af7..356de250b9 100644 --- a/backend/parsers/windmill-parser-sql/src/lib.rs +++ b/backend/parsers/windmill-parser-sql/src/lib.rs @@ -37,6 +37,16 @@ pub fn parse_bigquery_sig(code: &str) -> anyhow::Result { } } +pub fn parse_snowflake_sig(code: &str) -> anyhow::Result { + let parsed = parse_snowflake_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(); @@ -48,6 +58,8 @@ lazy_static::lazy_static! { // -- @name (type) = default static ref RE_ARG_BIGQUERY: Regex = Regex::new(r#"(?m)^-- @(\w+) \((\w+(?:\[\])?)\)(?: ?\= ?(.+))? *[\r\n$]"#).unwrap(); + static ref RE_ARG_SNOWFLAKE: Regex = Regex::new(r#"(?m)^-- \? (\w+) \((\w+)\)(?: ?\= ?(.+))? *[\r\n$]"#).unwrap(); + } fn parse_mysql_file(code: &str) -> anyhow::Result>> { @@ -158,6 +170,36 @@ fn parse_bigquery_file(code: &str) -> anyhow::Result>> { Ok(Some(args)) } +fn parse_snowflake_file(code: &str) -> anyhow::Result>> { + let mut args: Vec = vec![]; + + for cap in RE_ARG_SNOWFLAKE.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_snowflake_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" => { @@ -207,6 +249,18 @@ pub fn parse_bigquery_typ(typ: &str) -> Typ { } } +pub fn parse_snowflake_typ(typ: &str) -> Typ { + match typ { + "varchar" => Typ::Str(None), + "binary" => Typ::Bytes, + "date" | "time" | "timestamp" => Typ::Datetime, + "int" => Typ::Int, + "float" => Typ::Float, + "boolean" => Typ::Bool, + _ => Typ::Str(None), + } +} + #[cfg(test)] mod tests { diff --git a/backend/parsers/windmill-parser-wasm/pkg/package.json b/backend/parsers/windmill-parser-wasm/pkg/package.json index 99897334c1..e9d5d2709a 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.134.2", + "version": "1.138.1", "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 d9da9d77a7..40d3f8e5ac 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 @@ -35,6 +35,11 @@ export function parse_mysql(code: string): string; * @returns {string} */ export function parse_bigquery(code: string): string; +/** +* @param {string} code +* @returns {string} +*/ +export function parse_snowflake(code: string): string; export type InitInput = RequestInfo | URL | Response | BufferSource | WebAssembly.Module; @@ -47,6 +52,7 @@ export interface InitOutput { 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 parse_snowflake: (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 c28033a55f..66f8267a18 100644 --- a/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.js +++ b/backend/parsers/windmill-parser-wasm/pkg/windmill_parser_wasm.js @@ -358,6 +358,29 @@ export function parse_bigquery(code) { } } +/** +* @param {string} code +* @returns {string} +*/ +export function parse_snowflake(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_snowflake(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); @@ -400,6 +423,10 @@ async function __wbg_load(module, imports) { function __wbg_get_imports() { const imports = {}; imports.wbg = {}; + imports.wbg.__wbg_eval_d972bbef37d2cd5a = function(arg0, arg1) { + const ret = eval(getStringFromWasm0(arg0, arg1)); + return addHeapObject(ret); + }; imports.wbg.__wbindgen_object_drop_ref = function(arg0) { takeObject(arg0); }; @@ -451,10 +478,6 @@ 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 e0a10e02de..e6966037fb 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 eede785741..fb8cc99df7 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 @@ -8,6 +8,7 @@ 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 parse_snowflake(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 14bb547212..e22812a34c 100644 --- a/backend/parsers/windmill-parser-wasm/src/lib.rs +++ b/backend/parsers/windmill-parser-wasm/src/lib.rs @@ -44,3 +44,8 @@ pub fn parse_mysql(code: &str) -> String { pub fn parse_bigquery(code: &str) -> String { wrap_sig(windmill_parser_sql::parse_bigquery_sig(code)) } + +#[wasm_bindgen] +pub fn parse_snowflake(code: &str) -> String { + wrap_sig(windmill_parser_sql::parse_snowflake_sig(code)) +} diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index fd2860f9c8..ac3f948160 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -5672,18 +5672,7 @@ components: language: type: string enum: - [ - python3, - deno, - go, - bash, - postgresql, - mysql, - bigquery, - graphql, - nativets, - bun, - ] + [python3, deno, go, bash, postgresql, mysql, bigquery, snowflake, graphql, nativets, bun] kind: type: string enum: [script, failure, trigger, command, approval] @@ -5748,18 +5737,7 @@ components: language: type: string enum: - [ - python3, - deno, - go, - bash, - postgresql, - mysql, - bigquery, - graphql, - nativets, - bun, - ] + [python3, deno, go, bash, postgresql, mysql, bigquery, snowflake, graphql, nativets, bun] kind: type: string enum: [script, failure, trigger, command, approval] @@ -5929,18 +5907,7 @@ components: language: type: string enum: - [ - python3, - deno, - go, - bash, - postgresql, - mysql, - bigquery, - graphql, - nativets, - bun, - ] + [python3, deno, go, bash, postgresql, mysql, bigquery, snowflake, graphql, nativets, bun] email: type: string visible_to_owner: @@ -6030,18 +5997,7 @@ components: language: type: string enum: - [ - python3, - deno, - go, - bash, - postgresql, - mysql, - bigquery, - graphql, - nativets, - bun, - ] + [python3, deno, go, bash, postgresql, mysql, bigquery, snowflake, graphql, nativets, bun] is_skipped: type: boolean email: @@ -6470,18 +6426,7 @@ components: language: type: string enum: - [ - python3, - deno, - go, - bash, - postgresql, - mysql, - bigquery, - graphql, - nativets, - bun, - ] + [python3, deno, go, bash, postgresql, mysql, bigquery, snowflake, graphql, nativets, bun] tag: type: string kind: diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index f7bae77a74..4563b20e2a 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -1477,6 +1477,7 @@ async fn tarball_workspace( ScriptLang::Postgresql => "pg.sql", ScriptLang::Mysql => "my.sql", ScriptLang::Bigquery => "bq.sql", + ScriptLang::Snowflake => "sf.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 ca52e0f270..7f3a40828c 100644 --- a/backend/windmill-common/src/scripts.rs +++ b/backend/windmill-common/src/scripts.rs @@ -34,6 +34,7 @@ pub enum ScriptLang { Bun, Mysql, Bigquery, + Snowflake, } impl ScriptLang { @@ -48,6 +49,7 @@ impl ScriptLang { ScriptLang::Postgresql => "postgresql", ScriptLang::Mysql => "mysql", ScriptLang::Bigquery => "bigquery", + ScriptLang::Snowflake => "snowflake", } } } diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index ae9dd38d6c..c012fedf22 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -75,6 +75,7 @@ lazy_static::lazy_static! { "bun".to_string(), "postgresql".to_string(), "bigquery".to_string(), + "snowflake".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 f7e04355aa..9bb1c7cad1 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", "dep:gcp_auth"] +enterprise = ["windmill-queue/enterprise", "dep:gcp_auth", "dep:jsonwebtoken", "dep:pem", "dep:sha2"] [dependencies] windmill-queue.workspace = true @@ -65,6 +65,9 @@ mysql_async.workspace = true base64.workspace = true gcp_auth = { workspace = true, optional = true } rust_decimal.workspace = true +jsonwebtoken = { workspace = true, optional = true } +sha2 = { workspace = true, optional = true } +pem = { 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 index 7f4d074855..bd79dfc83c 100644 --- a/backend/windmill-worker/src/bigquery_executor.rs +++ b/backend/windmill-worker/src/bigquery_executor.rs @@ -122,7 +122,7 @@ pub async fn do_bigquery( .unwrap_or(&vec![]) .iter() .map(|x| { - convert_val(base_type.to_string(), x.clone()).ok().unwrap() + convert_val(base_type.to_string(), x.clone()) }) .collect::>() } @@ -134,7 +134,7 @@ pub async fn do_bigquery( "type": arg_t.to_uppercase() }, "parameterValue": { - "value": convert_val(arg_t, arg_v)?, + "value": convert_val(arg_t, arg_v), } }) }; @@ -239,7 +239,7 @@ pub async fn do_bigquery( } } -fn convert_val(arg_t: String, arg_v: Value) -> Result { +fn convert_val(arg_t: String, arg_v: Value) -> Value { match arg_t.as_str() { "timestamp" | "datetime" | "date" | "time" => { let mut v: String = arg_v.as_str().unwrap_or("").to_owned(); @@ -269,7 +269,7 @@ fn convert_val(arg_t: String, arg_v: Value) -> Result { _ => {} } - Ok(json!({ "value": json!(v) })) + json!({ "value": json!(v) }) } _ => { let mut v = arg_v; @@ -279,10 +279,10 @@ fn convert_val(arg_t: String, arg_v: Value) -> Result { v = json!(v.to_string()); } - Ok(json!({ + json!({ "value": v, } - )) + ) } } } diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index c778dad4ca..a172374861 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/backend/windmill-worker/src/lib.rs @@ -1,5 +1,7 @@ #[cfg(feature = "enterprise")] mod bigquery_executor; +#[cfg(feature = "enterprise")] +mod snowflake_executor; mod common; mod global_cache; diff --git a/backend/windmill-worker/src/snowflake_executor.rs b/backend/windmill-worker/src/snowflake_executor.rs new file mode 100644 index 0000000000..d2a5328345 --- /dev/null +++ b/backend/windmill-worker/src/snowflake_executor.rs @@ -0,0 +1,296 @@ +use base64::{engine, Engine as _}; +use core::fmt::Write; +use jsonwebtoken::{encode, Algorithm, EncodingKey, Header}; +use pem; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use windmill_common::error::Error; +use windmill_common::jobs::QueuedJob; +use windmill_parser_sql::parse_snowflake_sig; +use windmill_queue::HTTP_CLIENT; + +use serde::{Deserialize, Serialize}; + +use crate::{get_content, transform_json_value, AuthedClient, JobCompleted}; + +#[derive(Serialize)] +struct Claims { + iss: String, + sub: String, + iat: i64, + exp: i64, +} + +#[derive(Deserialize)] +struct SnowflakeDatabase { + account_identifier: String, + public_key: String, + private_key: String, + username: String, + database: Option, + schema: Option, + warehouse: Option, + role: Option, +} + +#[derive(Deserialize)] +#[allow(non_snake_case)] +struct SnowflakeResponse { + data: Vec>, + resultSetMetaData: SnowflakeResultSetMetaData, +} + +#[derive(Deserialize)] +#[allow(non_snake_case)] +struct SnowflakeResultSetMetaData { + numRows: i64, + rowType: Vec, +} + +#[derive(Deserialize)] +struct SnowflakeRowType { + name: String, + r#type: String, +} + +#[allow(non_snake_case)] +#[derive(Deserialize)] +struct SnowflakeError { + message: String, +} + +pub async fn do_snowflake( + 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 snowflake_args: Value = serde_json::from_value(args.unwrap_or_else(|| json!({}))) + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + let database = serde_json::from_value::( + snowflake_args.get("database").unwrap_or(&json!({})).clone(), + ) + .map_err(|e| Error::ExecutionErr(e.to_string()))?; + + let qualified_username = + format!("{}.{}", database.account_identifier, database.username).to_uppercase(); + + let public_key = pem::parse(database.public_key.as_bytes()).map_err(|e| { + Error::ExecutionErr(format!("Failed to parse public key: {}", e.to_string())) + })?; + let mut public_key_hash = Sha256::new(); + public_key_hash.update(public_key.contents()); + + let public_key_fp = engine::general_purpose::STANDARD.encode(public_key_hash.finalize()); + + let iss = format!("{}.SHA256:{}", qualified_username, public_key_fp); + + let claims = Claims { + iss: iss, + sub: qualified_username, + iat: chrono::Utc::now().timestamp(), + exp: (chrono::Utc::now() + chrono::Duration::hours(1)).timestamp(), + }; + + let private_key = EncodingKey::from_rsa_pem(database.private_key.as_bytes()).map_err(|e| { + Error::ExecutionErr(format!("Failed to parse private key: {}", e.to_string())) + })?; + + let token = encode(&Header::new(Algorithm::RS256), &claims, &private_key) + .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 query = get_content(&job, db).await?; + + let mut bindings = serde_json::Map::new(); + let sig = parse_snowflake_sig(&query) + .map_err(|x| Error::ExecutionErr(x.to_string()))? + .args; + + let mut i = 1; + for arg in &sig { + let arg_t = arg.otyp.clone().unwrap_or_else(|| "string".to_string()); + let arg_v = args.get(&arg.name).cloned().unwrap_or(json!("")); + let snowflake_v = convert_typ_val(arg_t, arg_v); + + bindings.insert(i.to_string(), snowflake_v); + i += 1; + } + + let mut body = serde_json::Map::new(); + if database.schema.is_some() { + body.insert( + "schema".to_string(), + json!(database.schema.unwrap().to_uppercase()), + ); + } + if database.warehouse.is_some() { + body.insert( + "warehouse".to_string(), + json!(database.warehouse.unwrap().to_uppercase()), + ); + } + if database.role.is_some() { + body.insert( + "role".to_string(), + json!(database.role.unwrap().to_uppercase()), + ); + } + if database.database.is_some() { + body.insert( + "database".to_string(), + json!(database.database.unwrap().to_uppercase()), + ); + } + body.insert("statement".to_string(), json!(query)); + body.insert("timeout".to_string(), json!(10)); // in seconds + + if i > 1 { + body.insert("bindings".to_string(), json!(bindings)); + } + + let response = HTTP_CLIENT + .post(format!( + "https://{}.snowflakecomputing.com/api/v2/statements/", + database.account_identifier.to_uppercase() + )) + .bearer_auth(token) + .header("X-Snowflake-Authorization-Token-Type", "KEYPAIR_JWT") + .json(&body) + .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.resultSetMetaData.numRows > 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 + .data + .iter() + .map(|row| { + let mut row_map = serde_json::Map::new(); + row.iter() + .zip(result.resultSetMetaData.rowType.iter()) + .for_each(|(val, row_type)| { + row_map + .insert(row_type.name.clone(), parse_val(&val, &row_type.r#type)); + }); + Value::from(row_map) + }) + .collect(); + + Ok(JobCompleted { job: job, result: rows, logs: "".to_string(), success: true }) + } + Err(e) => { + let resp = response.text().await.unwrap_or("".to_string()); + match serde_json::from_str::(&resp) { + Ok(sf_err) => Err(Error::ExecutionErr(sf_err.message)), + Err(_) => Err(Error::ExecutionErr(e.to_string())), + } + } + } +} + +fn convert_typ_val(arg_t: String, arg_v: Value) -> Value { + match arg_t.as_str() { + "date" => { + let arr = arg_v + .as_str() + .unwrap_or("") + .split("T") + .collect::>(); + json!({ + "type": "TEXT", + "value": match arr.as_slice() { + [date, _] => { + json!(date) + } + _ => { + arg_v + } + } + }) + } + "time" => { + let arr = arg_v + .as_str() + .unwrap_or("") + .split("T") + .collect::>(); + json!({ + "type": "TEXT", + "value": match arr.as_slice() { + [_, time] => { + json!(time) + } + _ => { + arg_v + } + } + }) + } + "binary" => { + // convert base64 to hex as expected by snowflake + let bytes = engine::general_purpose::STANDARD + .decode(arg_v.as_str().unwrap_or("")) + .unwrap_or(vec![]); + let mut hex = String::with_capacity(bytes.len() * 2); + for byte in bytes { + write!(hex, "{:02X}", byte).unwrap_or(()); + } + json!({ + "type": "TEXT", + "value": hex + }) + } + _ => { + let mut v = arg_v; + + if !v.is_string() { + // if not string, convert to string for api request + v = json!(v.to_string()); + } + + json!({ + "type": "TEXT", // snowflake infer type from schema + "value": v + }) + } + } +} + +fn parse_val(value: &Value, typ: &str) -> Value { + let str_value = value.as_str().unwrap_or("").to_string(); + match typ.to_lowercase().as_str() { + "boolean" => json!(str_value.parse::().ok().unwrap_or(false)), + "real" | "time" | "timestamp_ltz" | "timestamp_ntz" => { + json!(str_value.parse::().ok().unwrap_or(0.0)) + } + "fixed" | "date" | "number" => json!(str_value.parse::().ok().unwrap_or(0)), + _ => value.clone(), + } +} diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index f480a8cdfa..bacdaf4a5c 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/backend/windmill-worker/src/worker.rs @@ -70,7 +70,7 @@ use crate::{ }; #[cfg(feature = "enterprise")] -use crate::bigquery_executor::do_bigquery; +use crate::{bigquery_executor::do_bigquery, snowflake_executor::do_snowflake}; pub async fn create_token_for_owner_in_bg(db: &Pool, job: &QueuedJob) -> Arc> { let rw_lock = Arc::new(RwLock::new(String::new())); @@ -351,6 +351,7 @@ pub async fn run_worker = all_langs.clone().into_iter().map(|x| (x.clone(), prometheus::register_histogram!( @@ -1336,6 +1337,17 @@ async fn handle_code_execution_job( let jc = do_bigquery(job.clone(), &client.get_authed().await, &db).await?; return Ok(jc.result) } + } else if language == Some(ScriptLang::Snowflake) { + #[cfg(not(feature = "enterprise"))] + { + return Err(Error::ExecutionErr("Snowflake is only available with an enterprise license".to_string())) + } + + #[cfg(feature = "enterprise")] + { + let jc = do_snowflake(job.clone(), &client.get_authed().await, &db).await?; + return Ok(jc.result) + } } else if language == Some(ScriptLang::Nativets) { logs.push_str("\n--- FETCH TS EXECUTION ---\n"); let jc = do_nativets(job.clone(), logs.clone(), &client.get_authed().await, inner_content).await?; @@ -2211,6 +2223,7 @@ async fn capture_dependency_job( ScriptLang::Postgresql => Ok("".to_owned()), ScriptLang::Mysql => Ok("".to_owned()), ScriptLang::Bigquery => Ok("".to_owned()), + ScriptLang::Snowflake => Ok("".to_owned()), ScriptLang::Bash => Ok("".to_owned()), ScriptLang::Nativets => Ok("".to_owned()), diff --git a/cli/dev.ts b/cli/dev.ts index 016299afc3..c98c72f998 100644 --- a/cli/dev.ts +++ b/cli/dev.ts @@ -59,8 +59,10 @@ 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" + : splitted.length > 2 && splitted[splitted.length - 2] == "bq" ? "bigquery" + : splitted.length > 2 && splitted[splitted.length - 2] == "sf" + ? "snowflake" : "postgresql" : "unknown"; currentLastEdit = { diff --git a/cli/sync.ts b/cli/sync.ts index 6d6f05be72..b63b8d920d 100644 --- a/cli/sync.ts +++ b/cli/sync.ts @@ -136,6 +136,7 @@ function ZipFSElement(zip: JSZip, useYaml: boolean): DynFSElement { else if (language == "postgresql") ext = "pg.sql"; else if (language == "mysql") ext = "my.sql"; else if (language == "bigquery") ext = "bq.sql"; + else if (language == "snowflake") ext = "sf.sql"; else if (language == "bun") ext = "bun.ts"; else if (language == "nativets") ext = "native.ts"; diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 136be49cca..e6969a43ee 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -41,7 +41,7 @@ "svelte-timezone-picker": "^2.0.3", "tailwind-merge": "^1.13.2", "vscode-ws-jsonrpc": "3.0.0", - "windmill-parser-wasm": "^1.134.2", + "windmill-parser-wasm": "^1.138.1", "y-monaco": "^0.1.4", "y-websocket": "^1.5.0", "yjs": "^13.6.7" @@ -9777,9 +9777,9 @@ } }, "node_modules/windmill-parser-wasm": { - "version": "1.134.2", - "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.134.2.tgz", - "integrity": "sha512-XxAPj/JDfQVmvJCJ+8rktlAOsesvznARTPcLsBEKvN8MvG5jt2V3Xqh66MvmDVKl6QvHqOWFVetpd5LVrCDBSQ==" + "version": "1.138.1", + "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.138.1.tgz", + "integrity": "sha512-KGbiKKk7i8xRV+kcGXNRChIVXRQRAQ6eAGVJo7c2qjq+Sft3cGteWbkgeh8gjU5+KdmnP866jGwsRtw+zW2+5A==" }, "node_modules/wordwrap": { "version": "1.0.0", @@ -16766,9 +16766,9 @@ } }, "windmill-parser-wasm": { - "version": "1.134.2", - "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.134.2.tgz", - "integrity": "sha512-XxAPj/JDfQVmvJCJ+8rktlAOsesvznARTPcLsBEKvN8MvG5jt2V3Xqh66MvmDVKl6QvHqOWFVetpd5LVrCDBSQ==" + "version": "1.138.1", + "resolved": "https://registry.npmjs.org/windmill-parser-wasm/-/windmill-parser-wasm-1.138.1.tgz", + "integrity": "sha512-KGbiKKk7i8xRV+kcGXNRChIVXRQRAQ6eAGVJo7c2qjq+Sft3cGteWbkgeh8gjU5+KdmnP866jGwsRtw+zW2+5A==" }, "wordwrap": { "version": "1.0.0", diff --git a/frontend/package.json b/frontend/package.json index a988fd8526..9233a5fba1 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -104,7 +104,7 @@ "svelte-timezone-picker": "^2.0.3", "tailwind-merge": "^1.13.2", "vscode-ws-jsonrpc": "3.0.0", - "windmill-parser-wasm": "^1.134.2", + "windmill-parser-wasm": "^1.138.1", "y-monaco": "^0.1.4", "y-websocket": "^1.5.0", "yjs": "^13.6.7" diff --git a/frontend/src/lib/components/DBSchemaExplorer.svelte b/frontend/src/lib/components/DBSchemaExplorer.svelte index 4af1c10033..cbbebae501 100644 --- a/frontend/src/lib/components/DBSchemaExplorer.svelte +++ b/frontend/src/lib/components/DBSchemaExplorer.svelte @@ -1,7 +1,6 @@ -{#if resource_type == 'postgresql' || resource_type == 'mysql'} +{#if Object.keys(scripts).includes(resource_type || '')}