From e6f1211d317ca5c5b33ced324bf914b080d3ce2c Mon Sep 17 00:00:00 2001 From: Diego Imbert <70353967+diegoimbert@users.noreply.github.com> Date: Wed, 6 Aug 2025 15:55:36 +0200 Subject: [PATCH] feat: Ducklake native support (#6268) * upgrade duckdb * basic ducklake works * ducklake works with custom db catalogs * fix: pwsh skip already installed modules outside of cache (#6037) * improve query performance of user stats * separate ducklake_catalog db * ducklake settings * DucklakeSettings frontend * Ducklake ws settings saved in backend * fetch ducklake catalog resource * Ducklake works with configured s3 storage * Ducklake as asset * ducklake asset icon * Fix duckdb array and object args not working properly (#6254) * Fix bug with comments in duckdb * Avoid multiple queries when doing ATTACH ducklake * trunc sig no longer needed now that comments are trimmed * cache DuckdbConnectionSettingsResponse * duplicated code * transform_attach_ducklake contributes to duckdb_connection_settings_cache * eliminate the need for used_storages * nit * cleaner management of the bigquery credentials file * DBManagerDrawer refactor to prepare for Ducklake * get ducklake schema * implement delete for ducklake * load column metadata for ducklake * Select query works for ducklake, basic db explorer works ! * duckdb count query * Support all db ops for ducklake * clean migrations * SQL repl for Ducklake * fix broken database studio * nit * assert function * Ducklake in Editor Bar * default ducklake syntax + allow extra args * DucklakeCatalogWizard UI * nit + remove extra $ * modal when databases do not exist * cannot be windmill * Ducklake works safely with instance database * Avoid sending instance db credentials on network * resource leak security * remove fetch_attach_db_conn_str * prevent instance pg password leak * hide asset usage count when not available * case unsensitivity duckdb * warnings * disable instance catalog * use shorthand syntax when inserting with EditorBar * Instance ducklake catalog is now safe to use * use safer argon2 pwd * update package json parsers * update package json * better msgs * tooltips * disable explore button until saved * nit * fix warnings * better ducklake_user password management * nit * Sanitize passwords from errors in ducklake * DisplayResult broken in job result * remove superadmin requirement to check databases_exist * duckdb_connection_settings_v2_inner * Ducklake works on agent worker (finally) * ci * #[allow(dead_code)] * fix openapi missing response * Separate +Database button for DuckDB in EditorBar * Fix dropdown in ducklake settings * Attempt to fix migration race condition in CI * update sqlx failing for some offline queries * avoid temp password for ducklake_user * nits * ducklake settings nits * update duckdb default script * fix sql repl resetting text on refresh * avoid pgcrypto extension --------- Co-authored-by: HugoCasa Co-authored-by: Ruben Fiszel --- ...5c8a1291a5106fff46ceba85fa27d50a0354c.json | 22 + ...8234ca7d1efeee9661f3901f298da375e73f7.json | 8 +- ...cb4db9e0c9aee05596a594f81975a0cdb74d1.json | 23 + ...269f95904b5a1aa3560aa372621154ddbb28c.json | 15 + ...096d01fe7a72d749229f21634b549c6b1241a.json | 22 + ...dcc40f463cbc52d94ed9315cf9a547d4c89f2.json | 8 +- ...cb5e7cc038042a6fa7089f3f73cd137a63956.json | 190 +++++++ ...d548e3adefe5847d2e35331edb61442d0dd92.json | 20 + ...3f702af06663df6ddce22dd7e3d8402eb1d94.json | 23 + ...a54bc215b6fac86d8b92569d0074969530e1a.json | 23 + ...1fd381094070dd7707edac81e1b5e7d1d958b.json | 23 + backend/Cargo.lock | 2 + backend/Cargo.toml | 2 +- ...719095639_add_variable_asset_kind.down.sql | 7 +- .../20250724084100_ducklake.down.sql | 9 + .../migrations/20250724084100_ducklake.up.sql | 4 + ...132157_ducklake_instance_settings.down.sql | 2 + ...31132157_ducklake_instance_settings.up.sql | 23 + .../windmill-parser-sql/src/asset_parser.rs | 15 + .../windmill-parser/src/asset_parser.rs | 3 + .../src/lib.rs | 5 + backend/windmill-api/Cargo.toml | 1 + backend/windmill-api/openapi.yaml | 130 ++++- backend/windmill-api/src/settings.rs | 101 +++- backend/windmill-api/src/utils.rs | 3 + backend/windmill-api/src/workspaces.rs | 187 ++++++- backend/windmill-common/Cargo.toml | 1 + backend/windmill-common/src/assets.rs | 1 + backend/windmill-common/src/lib.rs | 37 ++ backend/windmill-common/src/s3_helpers.rs | 100 +++- backend/windmill-common/src/utils.rs | 45 ++ backend/windmill-common/src/workspaces.rs | 187 +++++++ backend/windmill-worker/src/agent_workers.rs | 18 +- .../windmill-worker/src/duckdb_executor.rs | 527 ++++++++++++------ frontend/package-lock.json | 16 +- frontend/package.json | 4 +- .../src/lib/components/DBManagerDrawer.svelte | 323 +++++------ frontend/src/lib/components/DBTable.svelte | 2 +- .../src/lib/components/DBTableEditor.svelte | 17 +- frontend/src/lib/components/EditorBar.svelte | 85 ++- .../lib/components/ExploreAssetButton.svelte | 18 +- .../components/ExploreAssetButton.svelte.d.ts | 23 - frontend/src/lib/components/SqlRepl.svelte | 20 +- .../display/dbtable/AppDbExplorer.svelte | 12 +- .../display/dbtable/InsertRow.svelte | 8 +- .../display/dbtable/queries/count.ts | 14 +- .../display/dbtable/queries/createTable.ts | 10 +- .../display/dbtable/queries/delete.ts | 7 + .../display/dbtable/queries/insert.ts | 2 + .../display/dbtable/queries/select.ts | 22 + .../display/dbtable/queries/update.ts | 7 + .../apps/components/display/dbtable/utils.ts | 98 ++-- .../assets/AssetsDropdownButton.svelte | 2 +- .../components/assets/JobAssetsViewer.svelte | 4 +- frontend/src/lib/components/assets/lib.ts | 4 + .../ConfirmationModal.svelte | 19 +- .../asyncConfirmationModal.svelte.ts | 57 ++ frontend/src/lib/components/custom_ui.ts | 2 + frontend/src/lib/components/dbOps.ts | 153 +++-- .../graph/renderers/nodes/AssetNode.svelte | 8 +- .../components/icons/AssetDucklakeIcon.svelte | 35 ++ .../components/icons/AssetGenericIcon.svelte | 3 + .../lib/components/icons/DucklakeIcon.svelte | 22 + .../workspaceSettings/DucklakeSettings.svelte | 341 ++++++++++++ frontend/src/lib/consts.ts | 3 +- frontend/src/lib/infer.ts | 3 +- frontend/src/lib/script_helpers.ts | 32 +- frontend/src/lib/stores.ts | 2 +- frontend/src/lib/utils.ts | 12 +- .../(logged)/workspace_settings/+page.svelte | 40 +- openflow.openapi.yaml | 1 + 71 files changed, 2642 insertions(+), 576 deletions(-) create mode 100644 backend/.sqlx/query-02ecdcc882931d5cbb2243e32805c8a1291a5106fff46ceba85fa27d50a0354c.json create mode 100644 backend/.sqlx/query-2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1.json create mode 100644 backend/.sqlx/query-40d372c253bef624ad07010b164269f95904b5a1aa3560aa372621154ddbb28c.json create mode 100644 backend/.sqlx/query-543859cf1c8d9e3bf2c2b23d21d096d01fe7a72d749229f21634b549c6b1241a.json create mode 100644 backend/.sqlx/query-6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956.json create mode 100644 backend/.sqlx/query-a22e166746ee10943c737668aafd548e3adefe5847d2e35331edb61442d0dd92.json create mode 100644 backend/.sqlx/query-b51b1e024f6dc7f30e4fbab26243f702af06663df6ddce22dd7e3d8402eb1d94.json create mode 100644 backend/.sqlx/query-b91b692f070c6f1e651870fbd5aa54bc215b6fac86d8b92569d0074969530e1a.json create mode 100644 backend/.sqlx/query-d1f505524793c03a63967d058ed1fd381094070dd7707edac81e1b5e7d1d958b.json create mode 100644 backend/migrations/20250724084100_ducklake.down.sql create mode 100644 backend/migrations/20250724084100_ducklake.up.sql create mode 100644 backend/migrations/20250731132157_ducklake_instance_settings.down.sql create mode 100644 backend/migrations/20250731132157_ducklake_instance_settings.up.sql delete mode 100644 frontend/src/lib/components/ExploreAssetButton.svelte.d.ts create mode 100644 frontend/src/lib/components/common/confirmationModal/asyncConfirmationModal.svelte.ts create mode 100644 frontend/src/lib/components/icons/AssetDucklakeIcon.svelte create mode 100644 frontend/src/lib/components/icons/DucklakeIcon.svelte create mode 100644 frontend/src/lib/components/workspaceSettings/DucklakeSettings.svelte diff --git a/backend/.sqlx/query-02ecdcc882931d5cbb2243e32805c8a1291a5106fff46ceba85fa27d50a0354c.json b/backend/.sqlx/query-02ecdcc882931d5cbb2243e32805c8a1291a5106fff46ceba85fa27d50a0354c.json new file mode 100644 index 0000000000..7fde9b87c4 --- /dev/null +++ b/backend/.sqlx/query-02ecdcc882931d5cbb2243e32805c8a1291a5106fff46ceba85fa27d50a0354c.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT jsonb_object_keys(ws.ducklake->'ducklakes') AS ducklake_name\n FROM workspace_settings ws\n WHERE ws.workspace_id = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "ducklake_name", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "02ecdcc882931d5cbb2243e32805c8a1291a5106fff46ceba85fa27d50a0354c" +} diff --git a/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json b/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json index 388fd55418..0ef59c340c 100644 --- a/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json +++ b/backend/.sqlx/query-08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7.json @@ -142,6 +142,11 @@ "ordinal": 27, "name": "git_app_installations", "type_info": "Jsonb" + }, + { + "ordinal": 28, + "name": "ducklake", + "type_info": "Jsonb" } ], "parameters": { @@ -177,7 +182,8 @@ true, true, true, - false + false, + true ] }, "hash": "08f288d2781d823e109a9e5b8848234ca7d1efeee9661f3901f298da375e73f7" diff --git a/backend/.sqlx/query-2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1.json b/backend/.sqlx/query-2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1.json new file mode 100644 index 0000000000..9c4b3eb842 --- /dev/null +++ b/backend/.sqlx/query-2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value FROM variable WHERE workspace_id = $1 AND path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value", + "type_info": "Varchar" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "2c0ab7571e1a7c4290315bc3efccb4db9e0c9aee05596a594f81975a0cdb74d1" +} diff --git a/backend/.sqlx/query-40d372c253bef624ad07010b164269f95904b5a1aa3560aa372621154ddbb28c.json b/backend/.sqlx/query-40d372c253bef624ad07010b164269f95904b5a1aa3560aa372621154ddbb28c.json new file mode 100644 index 0000000000..998359d80c --- /dev/null +++ b/backend/.sqlx/query-40d372c253bef624ad07010b164269f95904b5a1aa3560aa372621154ddbb28c.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "UPDATE workspace_settings SET ducklake = $1 WHERE workspace_id = $2", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Jsonb", + "Text" + ] + }, + "nullable": [] + }, + "hash": "40d372c253bef624ad07010b164269f95904b5a1aa3560aa372621154ddbb28c" +} diff --git a/backend/.sqlx/query-543859cf1c8d9e3bf2c2b23d21d096d01fe7a72d749229f21634b549c6b1241a.json b/backend/.sqlx/query-543859cf1c8d9e3bf2c2b23d21d096d01fe7a72d749229f21634b549c6b1241a.json new file mode 100644 index 0000000000..f5285690ec --- /dev/null +++ b/backend/.sqlx/query-543859cf1c8d9e3bf2c2b23d21d096d01fe7a72d749229f21634b549c6b1241a.json @@ -0,0 +1,22 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT elem FROM (SELECT unnest($1::TEXT[]) AS elem)\n WHERE elem NOT IN (SELECT datname FROM pg_catalog.pg_database);", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "elem", + "type_info": "Text" + } + ], + "parameters": { + "Left": [ + "TextArray" + ] + }, + "nullable": [ + null + ] + }, + "hash": "543859cf1c8d9e3bf2c2b23d21d096d01fe7a72d749229f21634b549c6b1241a" +} diff --git a/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json b/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json index 009d9fe5d1..f589df856d 100644 --- a/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json +++ b/backend/.sqlx/query-55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2.json @@ -142,6 +142,11 @@ "ordinal": 27, "name": "git_app_installations", "type_info": "Jsonb" + }, + { + "ordinal": 28, + "name": "ducklake", + "type_info": "Jsonb" } ], "parameters": { @@ -177,7 +182,8 @@ true, true, true, - false + false, + true ] }, "hash": "55cb03040bc2a8c53dd7fbb42bbdcc40f463cbc52d94ed9315cf9a547d4c89f2" diff --git a/backend/.sqlx/query-6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956.json b/backend/.sqlx/query-6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956.json new file mode 100644 index 0000000000..39510e0f99 --- /dev/null +++ b/backend/.sqlx/query-6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956.json @@ -0,0 +1,190 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT workspace_id, slack_team_id, teams_team_id, teams_team_name, slack_name, slack_command_script, teams_command_script, slack_email, auto_invite_domain, auto_invite_operator, auto_add, customer_id, plan, webhook, deploy_to, ai_config, error_handler, error_handler_extra_args, error_handler_muted_on_cancel, large_file_storage, ducklake, git_sync, deploy_ui, default_app, default_scripts, mute_critical_alerts, color, operator_settings, git_app_installations FROM workspace_settings WHERE workspace_id = $1", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "workspace_id", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "slack_team_id", + "type_info": "Varchar" + }, + { + "ordinal": 2, + "name": "teams_team_id", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "teams_team_name", + "type_info": "Text" + }, + { + "ordinal": 4, + "name": "slack_name", + "type_info": "Varchar" + }, + { + "ordinal": 5, + "name": "slack_command_script", + "type_info": "Varchar" + }, + { + "ordinal": 6, + "name": "teams_command_script", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "slack_email", + "type_info": "Varchar" + }, + { + "ordinal": 8, + "name": "auto_invite_domain", + "type_info": "Varchar" + }, + { + "ordinal": 9, + "name": "auto_invite_operator", + "type_info": "Bool" + }, + { + "ordinal": 10, + "name": "auto_add", + "type_info": "Bool" + }, + { + "ordinal": 11, + "name": "customer_id", + "type_info": "Varchar" + }, + { + "ordinal": 12, + "name": "plan", + "type_info": "Varchar" + }, + { + "ordinal": 13, + "name": "webhook", + "type_info": "Text" + }, + { + "ordinal": 14, + "name": "deploy_to", + "type_info": "Varchar" + }, + { + "ordinal": 15, + "name": "ai_config", + "type_info": "Jsonb" + }, + { + "ordinal": 16, + "name": "error_handler", + "type_info": "Varchar" + }, + { + "ordinal": 17, + "name": "error_handler_extra_args", + "type_info": "Json" + }, + { + "ordinal": 18, + "name": "error_handler_muted_on_cancel", + "type_info": "Bool" + }, + { + "ordinal": 19, + "name": "large_file_storage", + "type_info": "Jsonb" + }, + { + "ordinal": 20, + "name": "ducklake", + "type_info": "Jsonb" + }, + { + "ordinal": 21, + "name": "git_sync", + "type_info": "Jsonb" + }, + { + "ordinal": 22, + "name": "deploy_ui", + "type_info": "Jsonb" + }, + { + "ordinal": 23, + "name": "default_app", + "type_info": "Varchar" + }, + { + "ordinal": 24, + "name": "default_scripts", + "type_info": "Jsonb" + }, + { + "ordinal": 25, + "name": "mute_critical_alerts", + "type_info": "Bool" + }, + { + "ordinal": 26, + "name": "color", + "type_info": "Varchar" + }, + { + "ordinal": 27, + "name": "operator_settings", + "type_info": "Jsonb" + }, + { + "ordinal": 28, + "name": "git_app_installations", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text" + ] + }, + "nullable": [ + false, + true, + true, + true, + true, + true, + true, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + true, + false, + true, + true, + true, + true, + true, + true, + true, + true, + true, + false + ] + }, + "hash": "6e60068b582af46afbb93010355cb5e7cc038042a6fa7089f3f73cd137a63956" +} diff --git a/backend/.sqlx/query-a22e166746ee10943c737668aafd548e3adefe5847d2e35331edb61442d0dd92.json b/backend/.sqlx/query-a22e166746ee10943c737668aafd548e3adefe5847d2e35331edb61442d0dd92.json new file mode 100644 index 0000000000..33005ff7a6 --- /dev/null +++ b/backend/.sqlx/query-a22e166746ee10943c737668aafd548e3adefe5847d2e35331edb61442d0dd92.json @@ -0,0 +1,20 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT trim(both '\"' from value::text) FROM global_settings WHERE name = 'ducklake_user_pg_pwd';", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "btrim", + "type_info": "Text" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + null + ] + }, + "hash": "a22e166746ee10943c737668aafd548e3adefe5847d2e35331edb61442d0dd92" +} diff --git a/backend/.sqlx/query-b51b1e024f6dc7f30e4fbab26243f702af06663df6ddce22dd7e3d8402eb1d94.json b/backend/.sqlx/query-b51b1e024f6dc7f30e4fbab26243f702af06663df6ddce22dd7e3d8402eb1d94.json new file mode 100644 index 0000000000..54a41ff62f --- /dev/null +++ b/backend/.sqlx/query-b51b1e024f6dc7f30e4fbab26243f702af06663df6ddce22dd7e3d8402eb1d94.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT ws.ducklake->'ducklakes'->$2 AS config\n FROM workspace_settings ws\n WHERE ws.workspace_id = $1\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "config", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "b51b1e024f6dc7f30e4fbab26243f702af06663df6ddce22dd7e3d8402eb1d94" +} diff --git a/backend/.sqlx/query-b91b692f070c6f1e651870fbd5aa54bc215b6fac86d8b92569d0074969530e1a.json b/backend/.sqlx/query-b91b692f070c6f1e651870fbd5aa54bc215b6fac86d8b92569d0074969530e1a.json new file mode 100644 index 0000000000..db77434998 --- /dev/null +++ b/backend/.sqlx/query-b91b692f070c6f1e651870fbd5aa54bc215b6fac86d8b92569d0074969530e1a.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT value AS \"value!: _\" FROM resource WHERE workspace_id = $1 AND path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "value!: _", + "type_info": "Jsonb" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + true + ] + }, + "hash": "b91b692f070c6f1e651870fbd5aa54bc215b6fac86d8b92569d0074969530e1a" +} diff --git a/backend/.sqlx/query-d1f505524793c03a63967d058ed1fd381094070dd7707edac81e1b5e7d1d958b.json b/backend/.sqlx/query-d1f505524793c03a63967d058ed1fd381094070dd7707edac81e1b5e7d1d958b.json new file mode 100644 index 0000000000..979d328083 --- /dev/null +++ b/backend/.sqlx/query-d1f505524793c03a63967d058ed1fd381094070dd7707edac81e1b5e7d1d958b.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT 1 FROM resource WHERE workspace_id = $1 AND path = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "?column?", + "type_info": "Int4" + } + ], + "parameters": { + "Left": [ + "Text", + "Text" + ] + }, + "nullable": [ + null + ] + }, + "hash": "d1f505524793c03a63967d058ed1fd381094070dd7707edac81e1b5e7d1d958b" +} diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 5cb04c7f78..0a3a0bfbf8 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -14888,6 +14888,7 @@ dependencies = [ "tokio", "tokio-native-tls", "tokio-postgres 0.7.11", + "tokio-postgres 0.7.13", "tokio-stream", "tokio-tar", "tokio-tungstenite", @@ -14961,6 +14962,7 @@ name = "windmill-common" version = "1.518.2" dependencies = [ "anyhow", + "async-recursion", "async-stream", "aws-config", "aws-sdk-sts", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 74180a88d9..0b9e020cc1 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -243,7 +243,7 @@ json-pointer = "^0" itertools = "^0" regex = "^1" semver = "^1" -duckdb = { version = "1.2.2", features = ["bundled"] } +duckdb = { version = "1.3.2", features = ["bundled"] } v8 = "=130.0.7" # Exact version NOTE: Do not forget to update version and hash in flake.nix deno_fetch = "0.214.0" diff --git a/backend/migrations/20250719095639_add_variable_asset_kind.down.sql b/backend/migrations/20250719095639_add_variable_asset_kind.down.sql index d2f607c5b8..f92afec9d8 100644 --- a/backend/migrations/20250719095639_add_variable_asset_kind.down.sql +++ b/backend/migrations/20250719095639_add_variable_asset_kind.down.sql @@ -1 +1,6 @@ --- Add down migration script here +-- Remove 'variable' kind +DELETE FROM asset WHERE kind = 'variable'; +ALTER TABLE asset ALTER column kind TYPE VARCHAR; +DROP TYPE asset_kind; +CREATE TYPE ASSET_KIND AS ENUM ('s3object', 'resource'); +ALTER TABLE asset ALTER column kind TYPE ASSET_KIND using kind::ASSET_KIND; \ No newline at end of file diff --git a/backend/migrations/20250724084100_ducklake.down.sql b/backend/migrations/20250724084100_ducklake.down.sql new file mode 100644 index 0000000000..d26a17c914 --- /dev/null +++ b/backend/migrations/20250724084100_ducklake.down.sql @@ -0,0 +1,9 @@ +ALTER TABLE workspace_settings +DROP COLUMN ducklake; + +-- Remove 'ducklake' kind +DELETE FROM asset WHERE kind = 'ducklake'; +ALTER TABLE asset ALTER column kind TYPE VARCHAR; +DROP TYPE asset_kind; +CREATE TYPE ASSET_KIND AS ENUM ('s3object', 'resource', 'variable'); +ALTER TABLE asset ALTER column kind TYPE ASSET_KIND using kind::ASSET_KIND; \ No newline at end of file diff --git a/backend/migrations/20250724084100_ducklake.up.sql b/backend/migrations/20250724084100_ducklake.up.sql new file mode 100644 index 0000000000..a5e6692d6e --- /dev/null +++ b/backend/migrations/20250724084100_ducklake.up.sql @@ -0,0 +1,4 @@ +ALTER TYPE asset_kind ADD VALUE 'ducklake'; + +ALTER TABLE workspace_settings +ADD COLUMN ducklake JSONB; diff --git a/backend/migrations/20250731132157_ducklake_instance_settings.down.sql b/backend/migrations/20250731132157_ducklake_instance_settings.down.sql new file mode 100644 index 0000000000..50651d6193 --- /dev/null +++ b/backend/migrations/20250731132157_ducklake_instance_settings.down.sql @@ -0,0 +1,2 @@ +DELETE FROM global_settings +WHERE name = 'ducklake_user_pg_pwd'; \ No newline at end of file diff --git a/backend/migrations/20250731132157_ducklake_instance_settings.up.sql b/backend/migrations/20250731132157_ducklake_instance_settings.up.sql new file mode 100644 index 0000000000..3c1707203b --- /dev/null +++ b/backend/migrations/20250731132157_ducklake_instance_settings.up.sql @@ -0,0 +1,23 @@ +INSERT INTO global_settings (name, value) +VALUES ('ducklake_user_pg_pwd', ('"' || gen_random_uuid()::text || '"')::jsonb) +ON CONFLICT DO NOTHING; + +-- Cannot simply create the user because Postgres expect a static string for the password +-- Also we cannot drop the user easily in the down migration because databases will depend on it +-- And we cannot drop databases in transactions (migrations) + +DO $$ +DECLARE + pwd text; +BEGIN + SELECT trim(both '"' from value::text) INTO pwd FROM global_settings WHERE name = 'ducklake_user_pg_pwd'; + IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'ducklake_user') THEN + EXECUTE format('CREATE USER ducklake_user WITH PASSWORD %L', pwd); + ELSE + EXECUTE format('ALTER USER ducklake_user WITH PASSWORD %L', pwd); + END IF; +EXCEPTION + WHEN others THEN + RAISE NOTICE 'ducklake_user migration error, skipping.'; +END +$$; \ No newline at end of file diff --git a/backend/parsers/windmill-parser-sql/src/asset_parser.rs b/backend/parsers/windmill-parser-sql/src/asset_parser.rs index 531efd9c50..a6579798d3 100644 --- a/backend/parsers/windmill-parser-sql/src/asset_parser.rs +++ b/backend/parsers/windmill-parser-sql/src/asset_parser.rs @@ -7,6 +7,8 @@ use nom::{ branch::alt, bytes::complete::{tag, tag_no_case, take_while}, character::complete::{char, multispace0}, + combinator::opt, + sequence::preceded, IResult, Parser, }; @@ -52,6 +54,11 @@ fn parse_asset(input: &str) -> IResult<&str, ParseAssetsResult<&str>> { kind: AssetKind::Resource, access_type: None, }), + parse_ducklake_lit.map(|path| ParseAssetsResult { + path, + kind: AssetKind::Ducklake, + access_type: None, + }), )) .parse(input) } @@ -114,6 +121,14 @@ fn parse_resource_lit(input: &str) -> IResult<&str, &str> { Ok((input, path)) } +fn parse_ducklake_lit(input: &str) -> IResult<&str, &str> { + let (input, _) = quote(input)?; + let (input, _) = tag("ducklake").parse(input)?; + let (input, path) = + opt(preceded(tag("://"), take_while(|c| c != '\'' && c != '"'))).parse(input)?; + let (input, _) = quote(input)?; + Ok((input, path.unwrap_or("main"))) +} fn parse_comment(input: &str) -> IResult<&str, &str> { let (input, _) = tag("--").parse(input)?; let (input, comment) = take_while(|c| c != '\n')(input)?; diff --git a/backend/parsers/windmill-parser/src/asset_parser.rs b/backend/parsers/windmill-parser/src/asset_parser.rs index 6857e8d68e..82cf8b8499 100644 --- a/backend/parsers/windmill-parser/src/asset_parser.rs +++ b/backend/parsers/windmill-parser/src/asset_parser.rs @@ -15,6 +15,7 @@ use AssetUsageAccessType::*; pub enum AssetKind { S3Object, Resource, + Ducklake, } #[derive(Serialize)] @@ -56,6 +57,8 @@ pub fn parse_asset_syntax(s: &str) -> Option<(AssetKind, &str)> { Some((AssetKind::Resource, &s[6..])) } else if s.starts_with("$res:") { Some((AssetKind::Resource, &s[5..])) + } else if s.starts_with("ducklake://") { + Some((AssetKind::Ducklake, &s[11..])) } else { None } diff --git a/backend/parsers/windmill-sql-datatype-parser-wasm/src/lib.rs b/backend/parsers/windmill-sql-datatype-parser-wasm/src/lib.rs index 30103beafd..3517e0c450 100644 --- a/backend/parsers/windmill-sql-datatype-parser-wasm/src/lib.rs +++ b/backend/parsers/windmill-sql-datatype-parser-wasm/src/lib.rs @@ -38,3 +38,8 @@ pub fn parse_snowflake(typ: &str) -> String { pub fn parse_mssql(typ: &str) -> String { to_str(windmill_parser_sql::parse_mssql_typ(typ)) } + +#[wasm_bindgen] +pub fn parse_duckdb(typ: &str) -> String { + to_str(windmill_parser_sql::parse_duckdb_typ(typ)) +} diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 99449b0327..eb65715600 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -92,6 +92,7 @@ magic-crypt.workspace = true tempfile.workspace = true tokio-util.workspace = true tokio-tar.workspace = true +tokio-postgres.workspace = true hmac.workspace = true cookie.workspace = true sha2.workspace = true diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 5eabd44a2c..e7c3ef37b9 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -728,6 +728,50 @@ paths: schema: type: boolean + /settings/databases_exist: + post: + summary: checks that all given databases exist or else return the ones that don't + operationId: databasesExist + tags: + - setting + requestBody: + required: true + content: + application/json: + schema: + type: array + items: + type: string + responses: + "200": + description: databases that do not exist + content: + application/json: + schema: + type: array + items: + type: string + + /settings/create_ducklake_database/{name}: + post: + summary: Runs CREATE DATABASE on the Windmill Postgres and grants access to the ducklake_user + operationId: createDucklakeDatabase + tags: + - setting + parameters: + - in: path + name: name + required: true + schema: + type: string + description: The name of the database to create + responses: + "200": + description: status + content: + application/json: + schema: {} + /settings/global/{key}: get: summary: get global settings @@ -1918,6 +1962,8 @@ paths: type: boolean large_file_storage: $ref: "#/components/schemas/LargeFileStorage" + ducklake: + $ref: "#/components/schemas/DucklakeSettings" git_sync: $ref: "#/components/schemas/WorkspaceGitSyncSettings" deploy_ui: @@ -2421,6 +2467,51 @@ paths: properties: large_file_storage: $ref: "#/components/schemas/LargeFileStorage" + responses: + "200": + description: status + content: + application/json: + schema: {} + + + /w/{workspace}/workspaces/list_ducklakes: + get: + summary: list ducklakes + operationId: listDucklakes + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + responses: + "200": + description: status + content: + application/json: + schema: + type: array + items: + type: string + + /w/{workspace}/workspaces/edit_ducklake_config: + post: + summary: edit ducklake settings + operationId: editDucklakeConfig + tags: + - workspace + parameters: + - $ref: "#/components/parameters/WorkspaceId" + requestBody: + description: Ducklake settings + required: true + content: + application/json: + schema: + type: object + required: [settings] + properties: + settings: + $ref: "#/components/schemas/DucklakeSettings" responses: "200": @@ -13976,10 +14067,7 @@ components: path: type: string kind: - type: string - enum: - - s3object - - resource + $ref: "#/components/schemas/AssetKind" access_type: type: string enum: [r, w, rw] @@ -17133,6 +17221,39 @@ components: public_resource: type: boolean + DucklakeSettings: + type: object + required: [ducklakes] + properties: + ducklakes: + type: object + additionalProperties: + type: object + required: [catalog, storage] + properties: + catalog: + type: object + properties: + resource_type: + type: string + enum: + - postgresql + - mysql + - instance + resource_path: + type: string + required: + - resource_type + storage: + type: object + properties: + storage: + type: string + path: + type: string + required: + - path + WindmillLargeFile: type: object properties: @@ -17711,6 +17832,7 @@ components: enum: - s3object - resource + - ducklake Asset: type: object properties: diff --git a/backend/windmill-api/src/settings.rs b/backend/windmill-api/src/settings.rs index 9dfb687e51..59ec5e3569 100644 --- a/backend/windmill-api/src/settings.rs +++ b/backend/windmill-api/src/settings.rs @@ -30,19 +30,20 @@ use crate::utils::require_devops_role; use serde::Deserialize; #[cfg(feature = "enterprise")] use windmill_common::ee_oss::{send_critical_alert, CriticalAlertKind, CriticalErrorChannel}; +use windmill_common::error::to_anyhow; use windmill_common::{ email_oss::send_email, error::{self, JsonResult, Result}, + get_database_url, global_settings::{ AUTOMATE_USERNAME_CREATION_SETTING, CRITICAL_ALERT_MUTE_UI_SETTING, EMAIL_DOMAIN_SETTING, ENV_SETTINGS, HUB_ACCESSIBLE_URL_SETTING, HUB_BASE_URL_SETTING, }, + parse_postgres_url, server::Smtp, + utils::build_arg_str, }; -#[cfg(feature = "parquet")] -use windmill_common::error::to_anyhow; - pub fn global_service() -> Router { #[warn(unused_mut)] let r = Router::new() @@ -67,6 +68,11 @@ pub fn global_service() -> Router { "/critical_alerts/:id/acknowledge", post(acknowledge_critical_alert), ) + .route("/databases_exist", post(databases_exist)) + .route( + "/create_ducklake_database/:name", + post(create_ducklake_database), + ) .route( "/critical_alerts/acknowledge_all", post(acknowledge_all_critical_alerts), @@ -498,3 +504,92 @@ pub async fn acknowledge_all_critical_alerts( pub async fn acknowledge_all_critical_alerts() -> error::Error { error::Error::NotFound("Critical Alerts require EE".to_string()) } + +async fn databases_exist( + _authed: ApiAuthed, + Extension(db): Extension, + Json(database_names): Json>, +) -> JsonResult> { + let result = sqlx::query_scalar!( + r#"SELECT elem FROM (SELECT unnest($1::TEXT[]) AS elem) + WHERE elem NOT IN (SELECT datname FROM pg_catalog.pg_database);"#, + database_names.as_slice() + ) + .fetch_all(&db) + .await? + .into_iter() + .filter_map(|x| x) + .collect(); + + Ok(Json(result)) +} + +async fn create_ducklake_database( + authed: ApiAuthed, + Extension(db): Extension, + Path(dbname): Path, +) -> Result<()> { + require_super_admin(&db, &authed.email).await?; + + // Validate name to ensure it only contains alphanumeric characters + // Prevents SQL injection on the instance database + let valid_name = regex::Regex::new(r"^[a-zA-Z0-9_]+$") + .map_err(|_| error::Error::internal_err("Failed to compile regex".to_string()))?; + if !valid_name.is_match(&dbname) { + return Err(error::Error::BadRequest( + "Invalid database name".to_string(), + )); + } + + sqlx::query(&format!("CREATE DATABASE \"{dbname}\"")) + .execute(&db) + .await?; + + sqlx::query(&format!( + "GRANT CONNECT ON DATABASE \"{dbname}\" TO ducklake_user" + )) + .execute(&db) + .await?; + + // We have to connect to the newly created database as admin to grant permissions + let pg_creds = parse_postgres_url(&get_database_url().await?)?; + let Some(wm_pg_pwd) = pg_creds.password else { + return Err(error::Error::BadRequest("Password not found".to_string())); + }; + let conn_str: String = build_arg_str( + &[ + ("host", Some(&pg_creds.host)), + ("port", pg_creds.port.map(|p| p.to_string()).as_deref()), + ("password", Some(&wm_pg_pwd)), + ("user", pg_creds.username.as_deref()), + ("dbname", Some(&dbname)), + ], + " ", + "=", + ); + let (client, connection) = tokio::time::timeout( + std::time::Duration::from_secs(20), + tokio_postgres::connect(&conn_str, tokio_postgres::NoTls), + ) + .await + .map_err(to_anyhow)? + .map_err(to_anyhow)?; + + tokio::spawn(async move { + if let Err(e) = connection.await { + eprintln!("connection error: {}", e); + } + }); + + client + .batch_execute(&format!( + "GRANT USAGE ON SCHEMA public TO ducklake_user; + GRANT CREATE ON SCHEMA public TO ducklake_user; + ALTER DEFAULT PRIVILEGES IN SCHEMA public + GRANT SELECT, INSERT, UPDATE, DELETE ON TABLES TO ducklake_user;" + )) + .await + .map_err(to_anyhow)?; + + Ok(()) +} diff --git a/backend/windmill-api/src/utils.rs b/backend/windmill-api/src/utils.rs index 0e3062340c..ebbcad481c 100644 --- a/backend/windmill-api/src/utils.rs +++ b/backend/windmill-api/src/utils.rs @@ -452,3 +452,6 @@ pub async fn update_rw_lock(lock: std::sync::Arc>, val let mut w = lock.write().await; *w = value; } +lazy_static::lazy_static! { + static ref DUCKLAKE_INSTANCE_PG_PASSWORD: std::sync::RwLock> = std::sync::RwLock::new(None); +} diff --git a/backend/windmill-api/src/workspaces.rs b/backend/windmill-api/src/workspaces.rs index 6c61ff9ae8..6f6650713e 100644 --- a/backend/windmill-api/src/workspaces.rs +++ b/backend/windmill-api/src/workspaces.rs @@ -38,17 +38,18 @@ use windmill_common::users::username_to_permissioned_as; use windmill_common::variables::{build_crypt, decrypt, encrypt}; use windmill_common::worker::{to_raw_value, CLOUD_HOSTED}; #[cfg(feature = "enterprise")] +use windmill_common::workspaces::GitRepositorySettings; +#[cfg(feature = "enterprise")] use windmill_common::workspaces::WorkspaceDeploymentUISettings; #[cfg(feature = "enterprise")] use windmill_common::workspaces::WorkspaceGitSyncSettings; +use windmill_common::workspaces::{Ducklake, DucklakeCatalogResourceType}; use windmill_common::{ error::{Error, JsonResult, Result}, global_settings::AUTOMATE_USERNAME_CREATION_SETTING, oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH, utils::{paginate, rd_string, require_admin, Pagination}, }; -#[cfg(feature = "enterprise")] -use windmill_common::workspaces::GitRepositorySettings; use windmill_git_sync::{handle_deployment_metadata, DeployedObject}; #[cfg(feature = "enterprise")] @@ -114,9 +115,14 @@ pub fn workspaced_service() -> Router { "/edit_large_file_storage_config", post(edit_large_file_storage_config), ) + .route("/edit_ducklake_config", post(edit_ducklake_config)) + .route("/list_ducklakes", get(list_ducklakes)) .route("/edit_git_sync_config", post(edit_git_sync_config)) .route("/edit_git_sync_repository", post(edit_git_sync_repository)) - .route("/delete_git_sync_repository", delete(delete_git_sync_repository)) + .route( + "/delete_git_sync_repository", + delete(delete_git_sync_repository), + ) .route("/edit_deploy_ui_config", post(edit_deploy_ui_config)) .route("/edit_default_app", post(edit_default_app)) .route("/default_app", get(get_default_app)) @@ -229,6 +235,8 @@ pub struct WorkspaceSettings { #[serde(skip_serializing_if = "Option::is_none")] pub large_file_storage: Option, // effectively: DatasetsStorage #[serde(skip_serializing_if = "Option::is_none")] + pub ducklake: Option, + #[serde(skip_serializing_if = "Option::is_none")] pub git_sync: Option, // effectively: WorkspaceGitSyncSettings #[serde(skip_serializing_if = "Option::is_none")] pub deploy_ui: Option, // effectively: WorkspaceDeploymentUISettings @@ -296,11 +304,21 @@ struct LargeFileStorageWithSecondary { secondary_storage: HashMap, } +#[derive(Deserialize, Serialize, Debug)] +pub struct DucklakeSettings { + pub ducklakes: HashMap, +} + #[derive(Deserialize, Debug)] struct EditLargeFileStorageConfig { large_file_storage: Option, } +#[derive(Deserialize, Debug)] +struct EditDucklakeConfig { + settings: DucklakeSettings, +} + #[derive(Deserialize)] struct CreateWorkspace { id: String, @@ -439,7 +457,7 @@ async fn get_settings( let mut tx = user_db.begin(&authed).await?; let settings = sqlx::query_as!( WorkspaceSettings, - "SELECT workspace_id, slack_team_id, teams_team_id, teams_team_name, slack_name, slack_command_script, teams_command_script, slack_email, auto_invite_domain, auto_invite_operator, auto_add, customer_id, plan, webhook, deploy_to, ai_config, error_handler, error_handler_extra_args, error_handler_muted_on_cancel, large_file_storage, git_sync, deploy_ui, default_app, default_scripts, mute_critical_alerts, color, operator_settings, git_app_installations FROM workspace_settings WHERE workspace_id = $1", + "SELECT workspace_id, slack_team_id, teams_team_id, teams_team_name, slack_name, slack_command_script, teams_command_script, slack_email, auto_invite_domain, auto_invite_operator, auto_add, customer_id, plan, webhook, deploy_to, ai_config, error_handler, error_handler_extra_args, error_handler_muted_on_cancel, large_file_storage, ducklake, git_sync, deploy_ui, default_app, default_scripts, mute_critical_alerts, color, operator_settings, git_app_installations FROM workspace_settings WHERE workspace_id = $1", &w_id ) .fetch_optional(&mut *tx) @@ -867,6 +885,90 @@ async fn edit_large_file_storage_config( )) } +async fn list_ducklakes( + _authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, +) -> JsonResult> { + let ducklakes = sqlx::query_scalar!( + r#" + SELECT jsonb_object_keys(ws.ducklake->'ducklakes') AS ducklake_name + FROM workspace_settings ws + WHERE ws.workspace_id = $1 + "#, + &w_id + ) + .fetch_all(&db) + .await? + .into_iter() + .filter_map(|s| s) + .collect(); + + Ok(Json(ducklakes)) +} + +async fn edit_ducklake_config( + authed: ApiAuthed, + Extension(db): Extension, + Path(w_id): Path, + ApiAuthed { is_admin, username, .. }: ApiAuthed, + Json(new_config): Json, +) -> Result { + require_admin(is_admin, &username)?; + + let mut tx = db.begin().await?; + + let args_for_audit = format!("{:?}", new_config.settings); + audit_log( + &mut *tx, + &authed, + "workspaces.edit_ducklake_config", + ActionKind::Update, + &w_id, + Some(&authed.email), + Some([("ducklake", args_for_audit.as_str())].into()), + ) + .await?; + + // Check that all ducklake catalog resources exist to prevent + // exploiting the shared property to see any resource + for dl in new_config.settings.ducklakes.values() { + if dl.catalog.resource_type == DucklakeCatalogResourceType::Instance { + continue; + } + let catalog_res = sqlx::query_scalar!( + "SELECT 1 FROM resource WHERE workspace_id = $1 AND path = $2", + &w_id, + &dl.catalog.resource_path + ) + .fetch_optional(&mut *tx) + .await? + .flatten(); + + if catalog_res.is_none() { + return Err(Error::BadRequest(format!( + "Ducklake catalog resource {} not found in workspace {}", + dl.catalog.resource_path, &w_id + ))); + } + } + + let config: serde_json::Value = serde_json::to_value(new_config.settings) + .map_err(|err| Error::internal_err(err.to_string()))?; + + sqlx::query!( + "UPDATE workspace_settings SET ducklake = $1 WHERE workspace_id = $2", + config, + &w_id + ) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + + Ok(format!("Edit ducklake config for workspace {}", &w_id)) +} + #[derive(Deserialize)] pub struct EditGitSyncConfig { #[cfg(feature = "enterprise")] @@ -890,12 +992,16 @@ pub struct DeleteGitSyncRepositoryRequest { fn validate_git_repo_resource_path(path: &str) -> Result<()> { // Resource paths should follow the pattern: $res:f// or $res:u// if path.is_empty() { - return Err(Error::BadRequest("Resource path cannot be empty".to_string())); + return Err(Error::BadRequest( + "Resource path cannot be empty".to_string(), + )); } // Must start with $res: prefix if !path.starts_with("$res:") { - return Err(Error::BadRequest("Resource path must start with '$res:'".to_string())); + return Err(Error::BadRequest( + "Resource path must start with '$res:'".to_string(), + )); } // Extract the actual path after $res: @@ -903,19 +1009,29 @@ fn validate_git_repo_resource_path(path: &str) -> Result<()> { // Basic validation: must start with f/ or u/ and contain at least one slash if !actual_path.starts_with("f/") && !actual_path.starts_with("u/") { - return Err(Error::BadRequest("Resource path must start with '$res:f/' or '$res:u/'".to_string())); + return Err(Error::BadRequest( + "Resource path must start with '$res:f/' or '$res:u/'".to_string(), + )); } // Must have at least 3 parts (type, folder/user, name) let parts: Vec<&str> = actual_path.split('/').collect(); if parts.len() < 3 || parts.iter().any(|part| part.is_empty()) { - return Err(Error::BadRequest("Invalid resource path format".to_string())); + return Err(Error::BadRequest( + "Invalid resource path format".to_string(), + )); } // Resource name validation (last part) let resource_name = parts.last().unwrap(); - if !resource_name.chars().all(|c| c.is_alphanumeric() || c == '_' || c == '-') { - return Err(Error::BadRequest("Resource name can only contain alphanumeric characters, underscores, and hyphens".to_string())); + if !resource_name + .chars() + .all(|c| c.is_alphanumeric() || c == '_' || c == '-') + { + return Err(Error::BadRequest( + "Resource name can only contain alphanumeric characters, underscores, and hyphens" + .to_string(), + )); } Ok(()) @@ -927,11 +1043,15 @@ fn cleanup_legacy_git_sync_settings_in_memory( workspace_id: &str, ) { // Check if all repositories are in new format (have settings field) - let all_repos_migrated = git_sync_settings.repositories.iter() + let all_repos_migrated = git_sync_settings + .repositories + .iter() .all(|repo| repo.settings.is_some()); // If all repos are migrated and we still have legacy workspace-level settings - if all_repos_migrated && (git_sync_settings.include_path.is_some() || git_sync_settings.include_type.is_some()) { + if all_repos_migrated + && (git_sync_settings.include_path.is_some() || git_sync_settings.include_type.is_some()) + { tracing::info!( workspace_id = workspace_id, "All git sync repositories migrated to new format, cleaning up legacy workspace-level settings" @@ -1073,16 +1193,29 @@ async fn edit_git_sync_repository( ActionKind::Update, &w_id, Some(&authed.email), - Some([("repository_path", new_config.git_repo_resource_path.as_str()), ("repository_data", &format!("{:?}", new_config.repository))].into()), + Some( + [ + ( + "repository_path", + new_config.git_repo_resource_path.as_str(), + ), + ("repository_data", &format!("{:?}", new_config.repository)), + ] + .into(), + ), ) .await?; // Check if repository exists before modifying - let repo_exists = git_sync_settings.repositories.iter() + let repo_exists = git_sync_settings + .repositories + .iter() .any(|repo| repo.git_repo_resource_path == new_config.git_repo_resource_path); // Find and update the specific repository, or add it if it doesn't exist - let repo_found = git_sync_settings.repositories.iter_mut() + let repo_found = git_sync_settings + .repositories + .iter_mut() .find(|repo| repo.git_repo_resource_path == new_config.git_repo_resource_path); if let Some(existing_repo) = repo_found { @@ -1117,7 +1250,8 @@ async fn edit_git_sync_repository( &db, &w_id, windmill_git_sync::DeployedObject::Settings { setting_type: "git_sync".to_string() }, - Some(format!("Git sync repository '{}' {}", + Some(format!( + "Git sync repository '{}' {}", new_config.git_repo_resource_path, if repo_exists { "updated" } else { "added" } )), @@ -1125,7 +1259,8 @@ async fn edit_git_sync_repository( ) .await?; - Ok(format!("{} git sync repository '{}' for workspace {}", + Ok(format!( + "{} git sync repository '{}' for workspace {}", if repo_exists { "Updated" } else { "Added" }, new_config.git_repo_resource_path, &w_id @@ -1156,7 +1291,9 @@ async fn delete_git_sync_repository( // For deletion, only validate that path is not empty to allow cleanup of malformed entries if request.git_repo_resource_path.is_empty() { - return Err(Error::BadRequest("Resource path cannot be empty".to_string())); + return Err(Error::BadRequest( + "Resource path cannot be empty".to_string(), + )); } let mut tx = db.begin().await?; @@ -1182,7 +1319,9 @@ async fn delete_git_sync_repository( // Check if repository exists and remove it let original_count = git_sync_settings.repositories.len(); - git_sync_settings.repositories.retain(|repo| repo.git_repo_resource_path != request.git_repo_resource_path); + git_sync_settings + .repositories + .retain(|repo| repo.git_repo_resource_path != request.git_repo_resource_path); if git_sync_settings.repositories.len() == original_count { return Err(Error::BadRequest(format!( @@ -1227,12 +1366,18 @@ async fn delete_git_sync_repository( &db, &w_id, windmill_git_sync::DeployedObject::Settings { setting_type: "git_sync".to_string() }, - Some(format!("Git sync repository '{}' deleted", request.git_repo_resource_path)), + Some(format!( + "Git sync repository '{}' deleted", + request.git_repo_resource_path + )), false, ) .await?; - Ok(format!("Deleted git sync repository '{}' from workspace {}", request.git_repo_resource_path, &w_id)) + Ok(format!( + "Deleted git sync repository '{}' from workspace {}", + request.git_repo_resource_path, &w_id + )) } #[derive(Debug, Deserialize)] diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index 35b6572b51..fbdad80b0a 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -83,6 +83,7 @@ openidconnect = { workspace = true, optional = true } strum.workspace = true strum_macros.workspace = true url.workspace = true +async-recursion.workspace = true semver.workspace = true croner = "2.2.0" diff --git a/backend/windmill-common/src/assets.rs b/backend/windmill-common/src/assets.rs index 20c711c127..80804a3544 100644 --- a/backend/windmill-common/src/assets.rs +++ b/backend/windmill-common/src/assets.rs @@ -11,6 +11,7 @@ pub enum AssetKind { Resource, // Avoid unnexpected crashes when deserializing old assets Variable, // Deprecated + Ducklake, } #[derive(Serialize, Deserialize, Debug, PartialEq, Copy, Clone, Hash, Eq, sqlx::Type)] diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 299e99ebd4..a1bb0babf5 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -275,6 +275,43 @@ async fn reset() -> () { todo!() } +pub struct PostgresUrlComponents { + pub scheme: String, + pub username: Option, + pub password: Option, + pub host: String, + pub port: Option, + pub database: String, + pub ssl_mode: Option, +} + +pub fn parse_postgres_url(url: &str) -> Result { + let parsed_url = url::Url::parse(url).map_err(|_| Error::BadConfig("Invalid PostgreSQL URL".to_string()))?; + + let scheme = parsed_url.scheme().to_string(); + let username = parsed_url.username().to_string(); + let password = parsed_url.password().map(|p| p.to_string()); + let host = parsed_url.host_str().ok_or_else(|| Error::BadConfig("Missing host in PostgreSQL URL".to_string()))?.to_string(); + let port = parsed_url.port(); + let database = parsed_url.path().trim_start_matches('/').to_string(); + let mut ssl_mode = None; + for query in parsed_url.query_pairs() { + if query.0 == "sslmode" { + ssl_mode = Some(query.1.to_string()); + } + } + + Ok(PostgresUrlComponents { + scheme, + username: if username.is_empty() { None } else { Some(username) }, + password, + host, + port, + database, + ssl_mode, + }) +} + pub async fn get_database_url() -> Result { use std::env::var; use tokio::fs::File; diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index f838e2ef48..5b6e113729 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -1,4 +1,3 @@ -#[cfg(feature = "parquet")] use crate::error; #[cfg(feature = "parquet")] use aws_sdk_sts::config::ProvideCredentials; @@ -227,6 +226,18 @@ pub enum LargeFileStorage { // TODO: Add a filesystem type here in the future if needed } +impl LargeFileStorage { + pub fn get_s3_resource_path(&self) -> &str { + match self { + LargeFileStorage::S3Storage(s3_lfs) => &s3_lfs.s3_resource_path, + LargeFileStorage::S3AwsOidc(s3_lfs) => &s3_lfs.s3_resource_path, + LargeFileStorage::AzureBlobStorage(az_lfs) => &az_lfs.azure_blob_resource_path, + LargeFileStorage::AzureWorkloadIdentity(az_lfs) => &az_lfs.azure_blob_resource_path, + LargeFileStorage::GoogleCloudStorage(gcs_lfs) => &gcs_lfs.gcs_resource_path, + } + } +} + #[derive(Serialize, Deserialize, Debug)] pub struct S3Storage { pub s3_resource_path: String, @@ -969,7 +980,7 @@ pub async fn convert_json_line_stream>( Ok(tokio_stream::wrappers::ReceiverStream::new(rx)) } -#[derive(Deserialize, Serialize)] +#[derive(Deserialize, Serialize, Clone, Debug)] pub struct DuckdbConnectionSettingsResponse { pub connection_settings_str: String, #[serde(skip_serializing_if = "Option::is_none")] @@ -985,3 +996,88 @@ pub struct DuckdbConnectionSettingsQueryV2 { #[serde(skip_serializing_if = "Option::is_none")] pub storage: Option, } + +pub fn lfs_to_object_store_resource( + lfs: &LargeFileStorage, + resource_value: serde_json::Value, +) -> error::Result { + match lfs { + LargeFileStorage::S3Storage(_) | LargeFileStorage::S3AwsOidc(_) => { + let s3_resource: S3Resource = serde_json::from_value(resource_value).map_err(|e| { + error::Error::internal_err(format!("Error parsing S3 resource: {}", e)) + })?; + Ok(ObjectStoreResource::S3(s3_resource)) + } + LargeFileStorage::AzureBlobStorage(_) | LargeFileStorage::AzureWorkloadIdentity(_) => { + let azure_blob_resource: AzureBlobResource = serde_json::from_value(resource_value) + .map_err(|e| { + error::Error::internal_err(format!("Error parsing Azure Blob resource: {}", e)) + })?; + Ok(ObjectStoreResource::Azure(azure_blob_resource)) + } + LargeFileStorage::GoogleCloudStorage(_) => { + let gcs_resource: GcsResource = + serde_json::from_value(resource_value).map_err(|e| { + error::Error::internal_err(format!("Error parsing GCS resource: {}", e)) + })?; + Ok(ObjectStoreResource::Gcs(gcs_resource)) + } + } +} + +pub fn format_duckdb_connection_settings( + object_store_resource: ObjectStoreResource, +) -> error::Result { + match object_store_resource { + ObjectStoreResource::S3(s3_resource) => duckdb_connection_settings_internal(s3_resource), + ObjectStoreResource::Azure(azure_resource) => { + let connection_string = format!( + "CREATE SECRET az_secret (TYPE AZURE, CONNECTION_STRING 'DefaultEndpointsProtocol=https;AccountName={};AccountKey={};EndpointSuffix=core.windows.net');", + azure_resource.account_name, + azure_resource.access_key.unwrap_or_default() + ); + let response = DuckdbConnectionSettingsResponse { + connection_settings_str: connection_string, + azure_container_path: Some(format!("az://{}", azure_resource.container_name)), + s3_bucket: None, + }; + Ok(response) + } + ObjectStoreResource::Gcs(_) => { + return Err(error::Error::BadRequest( + "GCS is not supported in DuckDB".to_string(), + )); + } + } +} + +pub fn duckdb_connection_settings_internal( + s3_resource: S3Resource, +) -> error::Result { + let mut duckdb_settings: String = String::new(); + + duckdb_settings.push_str("SET home_directory='./';\n"); // TODO: make this configurable maybe, or point to a temporary folder + duckdb_settings.push_str("INSTALL 'httpfs';\n"); + if s3_resource.path_style.unwrap_or(true) { + duckdb_settings.push_str("SET s3_url_style='path';\n"); + } + duckdb_settings.push_str(format!("SET s3_region='{}';\n", s3_resource.region).as_str()); + duckdb_settings.push_str(format!("SET s3_endpoint='{}';\n", s3_resource.endpoint).as_str()); + if !s3_resource.use_ssl { + duckdb_settings.push_str("SET s3_use_ssl=0;\n"); // default is true for DuckDB + } + if let Some(access_key_id) = s3_resource.access_key { + duckdb_settings.push_str(format!("SET s3_access_key_id='{}';\n", access_key_id).as_str()); + } + if let Some(secret_access_key) = s3_resource.secret_key { + duckdb_settings + .push_str(format!("SET s3_secret_access_key='{}';\n", secret_access_key).as_str()); + } + + let response = DuckdbConnectionSettingsResponse { + connection_settings_str: duckdb_settings, + azure_container_path: None, + s3_bucket: Some(s3_resource.bucket), + }; + return Ok(response); +} diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index d7ad79f6e4..0abed36d9b 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -19,6 +19,7 @@ use git_version::git_version; use chrono::Utc; use croner::Cron; +use itertools::Itertools; use rand::{distr::Alphanumeric, rng, Rng}; use reqwest::Client; use semver::Version; @@ -595,6 +596,30 @@ pub async fn fetch_mute_workspace(_db: &DB, workspace_id: &str) -> Result } } +// build_arg_str(&[("name", Some("value")), ("name2", None)], " ", "=") +pub fn build_arg_str(args: &[(&str, Option<&str>)], sep: &str, eq: &str) -> String { + args.iter() + .filter_map(|(k, v)| { + if let Some(value) = v { + Some(format!("{}{}{}", k, eq, value)) + } else { + None + } + }) + .join(sep) +} + +// Some errors (duckdb) leak the password in the error message +pub fn sanitize_string_from_password(s: &str, passwd: &str) -> Option { + if s.contains(passwd) { + return Some(s.replace(passwd, "******")); + } + // Do NOT check substrings + // In the case the user finds a string and notices that it gets substituted, + // He can very easily find the next character in O(1) and thus the entire password + None +} + pub enum ScheduleType { Croner(Cron), Cron(cron::Schedule), @@ -859,3 +884,23 @@ impl Display for RunnableKind { write!(f, "{}", runnable_kind) } } + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn test_build_arg_str() { + let r = build_arg_str( + &[ + ("host", Some("localhost")), + ("port", Some("5432")), + ("password", None), + ("user", Some("postgres")), + ("dbname", Some("test_db")), + ], + " ", + "=", + ); + assert_eq!(r, "host=localhost port=5432 user=postgres dbname=test_db"); + } +} diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs index 1148af5ee4..5d7d3d9967 100644 --- a/backend/windmill-common/src/workspaces.rs +++ b/backend/windmill-common/src/workspaces.rs @@ -1,5 +1,19 @@ +use async_recursion::async_recursion; use quick_cache::sync::Cache; use serde::{Deserialize, Serialize}; +use serde_json::json; +use strum::AsRefStr; + +use crate::{ + error::{to_anyhow, Error, Result}, + get_database_url, parse_postgres_url, + s3_helpers::{ + format_duckdb_connection_settings, lfs_to_object_store_resource, + DuckdbConnectionSettingsResponse, LargeFileStorage, + }, + variables::{build_crypt, decrypt}, + DB, +}; #[derive(Serialize, Deserialize, Debug, Default)] pub struct WorkspaceGitSyncSettings { @@ -89,3 +103,176 @@ pub async fn is_premium_workspace(_db: &crate::DB, _w_id: &str) -> bool { IS_PREMIUM_CACHE.insert(_w_id.to_string(), premium); premium } + +#[derive(Deserialize, Serialize, Debug)] +pub struct Ducklake { + pub catalog: DucklakeCatalog, + pub storage: DucklakeStorage, +} + +#[derive(Deserialize, Serialize, Debug)] +pub struct DucklakeCatalog { + pub resource_type: DucklakeCatalogResourceType, + pub resource_path: String, +} + +#[derive(Deserialize, Serialize, Debug)] +pub struct DucklakeStorage { + #[serde(skip_serializing_if = "Option::is_none")] + pub storage: Option, + pub path: String, +} + +#[derive(Deserialize, Serialize, Debug, PartialEq)] +#[serde(rename_all = "lowercase")] +#[derive(AsRefStr)] +#[strum(serialize_all = "lowercase")] +pub enum DucklakeCatalogResourceType { + #[strum(serialize = "postgres")] + Postgresql, + Mysql, + Instance, +} + +#[derive(Deserialize, Serialize)] +pub struct DucklakeWithConnData { + pub catalog: DucklakeCatalog, + pub catalog_resource: serde_json::Value, + pub storage: DucklakeStorage, + pub storage_settings: DuckdbConnectionSettingsResponse, +} + +pub async fn get_ducklake_from_db_unchecked( + name: &str, + w_id: &str, + db: &DB, +) -> Result { + let ducklake = sqlx::query_scalar!( + r#" + SELECT ws.ducklake->'ducklakes'->$2 AS config + FROM workspace_settings ws + WHERE ws.workspace_id = $1 + "#, + &w_id, + name + ) + .fetch_one(db) + .await + .map_err(|err| Error::internal_err(format!("getting ducklake {name}: {err}")))? + .ok_or_else(|| Error::internal_err(format!("ducklake {name} not found")))?; + + let ducklake = serde_json::from_value::(ducklake)?; + + let lfs = if let Some(storage) = &ducklake.storage.storage { + sqlx::query_scalar!("SELECT large_file_storage->'secondary_storage'->$2 FROM workspace_settings WHERE workspace_id = $1", w_id, storage) + } else { + sqlx::query_scalar!("SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", w_id) + }.fetch_optional(db) + .await? + .flatten() + .map(serde_json::from_value::) + .ok_or_else(|| Error::ExecutionErr("Ducklake storage not found".to_string()))??; + + let s3_resource = transform_json_unchecked( + &serde_json::Value::String(lfs.get_s3_resource_path().to_string()), + w_id, + db, + ) + .await?; + let object_store_resource = lfs_to_object_store_resource(&lfs, s3_resource)?; + + let catalog_resource = + if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance { + let pg_creds = parse_postgres_url(&get_database_url().await?)?; + json!({ + "dbname": ducklake.catalog.resource_path, + "host": pg_creds.host, + "port": pg_creds.port, + "user": "ducklake_user", + "sslmode": pg_creds.ssl_mode, + "password": get_ducklake_instance_pg_catalog_password(&db).await?, + }) + } else { + transform_json_unchecked( + &serde_json::Value::String(format!("$res:{}", ducklake.catalog.resource_path)), + w_id, + db, + ) + .await? + }; + let ducklake = DucklakeWithConnData { + catalog_resource, + storage_settings: format_duckdb_connection_settings(object_store_resource)?, + catalog: ducklake.catalog, + storage: ducklake.storage, + }; + Ok(ducklake) +} + +pub async fn get_ducklake_instance_pg_catalog_password(db: &DB) -> Result { + sqlx::query_scalar!( + "SELECT trim(both '\"' from value::text) FROM global_settings WHERE name = 'ducklake_user_pg_pwd';" + ) + .fetch_optional(db) + .await? + .flatten().ok_or_else(|| + Error::BadRequest(format!( + "Ducklake instance catalog password not found, did you run migrations ?" + )) + ) +} + +// This does not check for any permission. Should never be displayed to a user. +#[async_recursion] +async fn transform_json_unchecked( + value: &serde_json::Value, + w_id: &str, + db: &DB, +) -> Result { + let value = match value { + serde_json::Value::Object(map) => { + let mut transformed_map = serde_json::Map::new(); + for (key, val) in map { + let transformed_val = transform_json_unchecked(val, w_id, db).await?; + transformed_map.insert(key.clone(), serde_json::to_value(transformed_val)?); + } + serde_json::Value::Object(transformed_map) + } + serde_json::Value::Array(arr) => { + let mut transformed_array = Vec::new(); + for val in arr { + let transformed_val = transform_json_unchecked(val, w_id, db).await?; + transformed_array.push(serde_json::to_value(transformed_val)?); + } + serde_json::Value::Array(transformed_array) + } + serde_json::Value::String(s) if s.starts_with("$res:") => { + let resource = sqlx::query_scalar!( + "SELECT value AS \"value!: _\" FROM resource WHERE workspace_id = $1 AND path = $2", + &w_id, + &s[5..] + ) + .fetch_one(db) + .await + .map_err(to_anyhow)?; + transform_json_unchecked(&resource, w_id, db).await? + } + serde_json::Value::String(s) if s.starts_with("$var:") => { + let variable = sqlx::query_scalar!( + "SELECT value FROM variable WHERE workspace_id = $1 AND path = $2", + &w_id, + &s[5..] + ) + .fetch_one(db) + .await + .map_err(to_anyhow)?; + let mc = build_crypt(&db, &w_id).await?; + let variable = decrypt(&mc, variable)?; + serde_json::Value::String(variable) + } + s @ serde_json::Value::String(_) => s.clone(), + x => x.clone(), + }; + + Ok(value) +} diff --git a/backend/windmill-worker/src/agent_workers.rs b/backend/windmill-worker/src/agent_workers.rs index e8b3ef4300..0aaecdbe1e 100644 --- a/backend/windmill-worker/src/agent_workers.rs +++ b/backend/windmill-worker/src/agent_workers.rs @@ -1,6 +1,8 @@ use reqwest::header::HeaderMap; use uuid::Uuid; -use windmill_common::{agent_workers::QueueInitJob, worker::HttpClient}; +use windmill_common::{ + agent_workers::QueueInitJob, worker::HttpClient, workspaces::DucklakeWithConnData, +}; use windmill_queue::{JobAndPerms, JobCompleted}; pub async fn queue_init_job(client: &HttpClient, content: &str) -> anyhow::Result { @@ -48,4 +50,18 @@ pub async fn send_result(client: &HttpClient, jc: JobCompleted) -> anyhow::Resul .await } +#[allow(dead_code)] +pub async fn get_ducklake_from_agent_http( + client: &HttpClient, + name: &str, + w_id: &str, +) -> anyhow::Result { + client + .get(&format!( + "/api/w/{}/agent_workers/get_ducklake/{}", + w_id, &name + )) + .await +} + pub const UPDATE_PING_URL: &str = "/api/agent_workers/update_ping"; diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 82cd452c59..6e321b4e9d 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::env; +use std::sync::{Arc, Mutex}; use duckdb::types::TimeUnit; use duckdb::{params_from_iter, Row}; @@ -7,17 +8,19 @@ use rust_decimal::prelude::FromPrimitive; use rust_decimal::Decimal; use serde_json::value::RawValue; use serde_json::{json, Value}; -use tokio::fs::remove_file; use tokio::task; use uuid::Uuid; use windmill_common::error::{to_anyhow, Error, Result}; use windmill_common::s3_helpers::{ DuckdbConnectionSettingsQueryV2, DuckdbConnectionSettingsResponse, S3Object, }; +use windmill_common::utils::sanitize_string_from_password; use windmill_common::worker::{to_raw_value, Connection}; +use windmill_common::workspaces::{get_ducklake_from_db_unchecked, DucklakeCatalogResourceType}; use windmill_parser_sql::{parse_duckdb_sig, parse_sql_blocks}; use windmill_queue::{CanceledBy, MiniPulledJob}; +use crate::agent_workers::get_ducklake_from_agent_http; use crate::common::{build_args_values, OccupancyMetrics}; use crate::handle_child::run_future_with_polling_update_job_poller; #[cfg(feature = "mysql")] @@ -94,18 +97,19 @@ pub async fn do_duckdb( column_order_ref: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, ) -> Result> { + let hidden_passwords = Arc::new(Mutex::new(Vec::::new())); + let result_f = async { + let mut hidden_passwords = hidden_passwords.clone(); + let mut bigquery_credentials = None; + let mut duckdb_connection_settings_cache = + HashMap::, DuckdbConnectionSettingsResponse>::new(); + let sig = parse_duckdb_sig(query)?.args; let mut job_args = build_args_values(job, client, conn).await?; let (query, _) = &sanitize_and_interpolate_unsafe_sql_args(query, &sig, &job_args)?; - // Prevent interpolate_named_args from detecting argument identifiers in the signature for - // the first query block - let query = trunc_sig(query); - - let (_query_with_transformed_s3_uris, mut used_storages) = - transform_s3_uris(query, client).await?; - let query = _query_with_transformed_s3_uris.as_deref().unwrap_or(query); + let query = transform_s3_uris(query, client, &mut duckdb_connection_settings_cache).await?; let job_args = { let mut m: HashMap = HashMap::new(); @@ -119,28 +123,17 @@ pub async fn do_duckdb( let s3_obj = serde_json::from_value::(json_value).map_err(|e| { Error::ExecutionErr(format!("Failed to deserialize S3Object: {}", e)) })?; - let duckdb_conn_settings: windmill_common::s3_helpers::DuckdbConnectionSettingsResponse = client - .get_duckdb_connection_settings(&DuckdbConnectionSettingsQueryV2 { - s3_resource_path: None, - storage: s3_obj.storage.clone(), - }) + let duckdb_conn_settings: DuckdbConnectionSettingsResponse = + get_duckdb_connection_settings( + &s3_obj.storage, + &mut duckdb_connection_settings_cache, + client, + ) .await?; - let uri = match ( - &duckdb_conn_settings.s3_bucket, - &duckdb_conn_settings.azure_container_path, - ) { - (Some(s3_bucket), None) => format!("s3://{}/{}", s3_bucket, &s3_obj.s3), - (None, Some(az_container)) => format!("{}/{}", az_container, &s3_obj.s3), - _ => { - return Err(Error::ExecutionErr( - "S3Object must have either s3_bucket or azure_container_path" - .to_string(), - )); - } - }; + let uri = + duckdb_conn_settings_to_s3_network_uri(&duckdb_conn_settings, &s3_obj.s3)?; m.insert(sig_arg.name, duckdb::types::Value::Text(uri)); - used_storages.insert(s3_obj.storage, duckdb_conn_settings); } else { let duckdb_value = json_value_to_duckdb_value( &json_value, @@ -156,36 +149,67 @@ pub async fn do_duckdb( m }; - let query_block_list = parse_sql_blocks(query); + let query_block_list = parse_sql_blocks(&query); // Replace windmill resource ATTACH statements with the real instructions let query_block_list = { let mut v = vec![]; for query_block in query_block_list.iter() { + let query_block = remove_comments(&query_block); match parse_attach_db_resource(query_block) { - Some(parsed) => v.extend( - transform_attach_db_resource_query(&parsed, &job.id, client).await?, - ), - None => v.push(query_block.to_string()), + Some(parsed) => { + v.extend( + transform_attach_db_resource_query( + &parsed, + &job.id, + client, + &mut hidden_passwords, + ) + .await?, + ); + if parsed.db_type == "bigquery" { + bigquery_credentials = Some(UseBigQueryCredentialsFile::new( + job.id, + parsed.resource_path, + )?); + } + } + None => match transform_attach_ducklake( + &query_block, + conn, + &mut duckdb_connection_settings_cache, + &mut hidden_passwords, + &job.workspace_id, + ) + .await? + { + Some(ducklake_query) => v.extend(ducklake_query), + None => v.push(query_block.to_string()), + }, }; } v }; - // duckdb::Connection is not Send so we do it in a single blocking task + // duckdb::Connection is not Send so we run the queries in a single blocking task let (result, column_order) = task::spawn_blocking(move || { let conn = duckdb::Connection::open_in_memory() .map_err(|e| Error::ConnectingToDatabase(e.to_string()))?; - for (_, DuckdbConnectionSettingsResponse { connection_settings_str, .. }) in - used_storages.into_iter() + for DuckdbConnectionSettingsResponse { connection_settings_str, .. } in + duckdb_connection_settings_cache.values() { + hidden_passwords + .lock() + .unwrap() + .push(connection_settings_str.clone()); conn.execute_batch(&connection_settings_str) .map_err(|e| Error::ExecutionErr(e.to_string()))?; } let mut result: Option> = None; let mut column_order = None; + for (query_block_index, query_block) in query_block_list.iter().enumerate() { result = Some( do_duckdb_inner( @@ -204,14 +228,9 @@ pub async fn do_duckdb( .await .map_err(to_anyhow)??; - *column_order_ref = column_order; + drop(bigquery_credentials); - // BigQuery cleanup - let bq_credentials_path = make_bq_credentials_path(&job.id); - env::remove_var("GOOGLE_APPLICATION_CREDENTIALS"); - if matches!(tokio::fs::try_exists(&bq_credentials_path).await, Ok(true)) { - remove_file(&bq_credentials_path).await.map_err(to_anyhow)?; - } + *column_order_ref = column_order; Ok(result) }; @@ -227,9 +246,21 @@ pub async fn do_duckdb( &mut Some(occupancy_metrics), Box::pin(futures::stream::once(async { 0 })), ) - .await?; + .await; - Ok(result) + match result { + Ok(result) => Ok(result), + Err(e) => { + // Passwords might appear in the error message + let mut err_str = e.to_string(); + for pwd in hidden_passwords.lock().unwrap().iter() { + if let Some(sanitized) = sanitize_string_from_password(&err_str, &pwd.clone()) { + err_str = sanitized; + } + } + Err(Error::ExecutionErr(err_str)) + } + } } fn row_to_value(row: &Row<'_>, column_names: &[String]) -> Result> { @@ -398,7 +429,8 @@ fn string_to_duckdb_timestamp(s: &str) -> Result { fn string_to_duckdb_date(s: &str) -> Result { use chrono::Datelike; - let date = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d").unwrap(); + let date = chrono::NaiveDate::parse_from_str(s, "%Y-%m-%d") + .map_err(|e| Error::ExecutionErr(format!("Invalid date format: {}", e)))?; Ok(duckdb::types::Value::Date32(date.num_days_from_ce())) } @@ -419,7 +451,7 @@ struct ParsedAttachDbResource<'a> { } fn parse_attach_db_resource<'a>(query: &'a str) -> Option> { lazy_static::lazy_static! { - static ref RE: regex::Regex = regex::Regex::new(r"ATTACH '(\$res:|res://)([^']+)' AS (\S+) \(TYPE (\w+)(.*)\)").unwrap(); + static ref RE: regex::Regex = regex::Regex::new(r"(?i)ATTACH '(\$res:|res://)([^']+)' AS (\S+) \(TYPE (\w+)(.*)\)").unwrap(); } for cap in RE.captures_iter(query) { @@ -438,25 +470,32 @@ fn parse_attach_db_resource<'a>(query: &'a str) -> Option, - job_id: &Uuid, - client: &AuthedClient, -) -> Result> { - match parsed.db_type.to_lowercase().as_str() { - "postgres" => { - let resource: PgDatabase = client - .get_resource_value_interpolated(parsed.resource_path, Some(job_id.to_string())) - .await?; - - let attach_str = format!( - "ATTACH 'dbname={} {} host={} {} {}' AS {} (TYPE postgres{});", - resource.dbname, - resource - .user - .map(|u| format!("user={}", u)) +fn format_attach_db_conn_str(db_resource: Value, db_type: &str) -> Result { + let s = match db_type.to_lowercase().as_str() { + "postgres" | "postgresql" => { + let res: PgDatabase = serde_json::from_value(db_resource)?; + format!( + "dbname={} {} host={} {} {}", + res.dbname, + res.user.map(|u| format!("user={}", u)).unwrap_or_default(), + res.host, + res.password + .map(|p| format!("password={}", p)) .unwrap_or_default(), + res.port.map(|p| format!("port={}", p)).unwrap_or_default(), + ) + } + #[cfg(feature = "mysql")] + "mysql" => { + let resource: MysqlDatabase = serde_json::from_value(db_resource)?; + format!( + "database={} host={} ssl_mode={} {} {} {}", + resource.database, resource.host, + resource + .ssl + .map(|ssl| if ssl { "required" } else { "disabled" }) + .unwrap_or("preferred"), resource .password .map(|p| format!("password={}", p)) @@ -465,76 +504,15 @@ async fn transform_attach_db_resource_query( .port .map(|p| format!("port={}", p)) .unwrap_or_default(), - parsed.name, - parsed.extra_args.unwrap_or("") - ); - - Ok(vec![ - "INSTALL postgres;".to_string(), - "LOAD postgres;".to_string(), - attach_str, - ]) - } - "mysql" => { - #[cfg(not(feature = "mysql"))] - return Err(Error::ExecutionErr( - "MySQL feature is not enabled".to_string(), - )); - - #[cfg(feature = "mysql")] - { - let resource: MysqlDatabase = client - .get_resource_value_interpolated(parsed.resource_path, Some(job_id.to_string())) - .await?; - - let attach_str = format!( - "ATTACH 'database={} host={} ssl_mode={} {} {} {}' AS {} (TYPE mysql{});", - resource.database, - resource.host, - resource - .ssl - .map(|ssl| if ssl { "required" } else { "disabled" }) - .unwrap_or("preferred"), - resource - .password - .map(|p| format!("password={}", p)) - .unwrap_or_default(), - resource - .port - .map(|p| format!("port={}", p)) - .unwrap_or_default(), - resource - .user - .map(|u| format!("user={}", u)) - .unwrap_or_default(), - parsed.name, - parsed.extra_args.unwrap_or("") - ); - - Ok(vec![ - "INSTALL mysql;".to_string(), - "LOAD mysql;".to_string(), - attach_str, - ]) - } + resource + .user + .map(|u| format!("user={}", u)) + .unwrap_or_default(), + ) } "bigquery" => { - let resource: Value = client - .get_resource_value_interpolated(parsed.resource_path, Some(job_id.to_string())) - .await?; - // duckdb's bigquery extension requires a json file as credentials - let bq_credentials_path = make_bq_credentials_path(job_id); - env::set_var("GOOGLE_APPLICATION_CREDENTIALS", &bq_credentials_path); - tokio::fs::write(&bq_credentials_path, resource.to_string()) - .await - .map_err(|e| { - Error::ExecutionErr(format!( - "Failed to write BigQuery credentials to {}: {}", - &bq_credentials_path, e - )) - })?; let project_id: String = serde_json::from_value( - resource + db_resource .get("project_id") .ok_or_else(|| { Error::ExecutionErr("BigQuery resource must contain project_id".to_string()) @@ -542,39 +520,133 @@ async fn transform_attach_db_resource_query( .to_owned(), ) .map_err(|_e| Error::ExecutionErr("failed project_id deserialize".to_string()))?; - let attach_str = format!( - "ATTACH 'project={}' as {} (TYPE bigquery{});", - project_id, - parsed.name, - parsed.extra_args.unwrap_or("") - ) - .to_string(); - Ok(vec![ - "INSTALL bigquery FROM community;".to_string(), - "LOAD bigquery;".to_string(), - attach_str, - ]) + format!("project={}", project_id,) } + _ => { + return Err(Error::ExecutionErr(format!( + "Unsupported db type in DuckDB ATTACH: {db_type}", + ))) + } + }; + Ok(s) +} + +fn get_attach_db_install_str(db_type: &str) -> Result<&str> { + match db_type.to_lowercase().as_str() { + "postgres" => Ok("INSTALL postgres;"), + "mysql" => { + #[cfg(not(feature = "mysql"))] + return Err(Error::ExecutionErr( + "MySQL feature is not enabled".to_string(), + )); + #[cfg(feature = "mysql")] + Ok("INSTALL mysql;") + } + "bigquery" => Ok("INSTALL bigquery FROM community;"), _ => Err(Error::ExecutionErr(format!( "Unsupported db type in DuckDB ATTACH: {}", - parsed.db_type + db_type ))), } } -// Returns the transformed query and the set of storages used +async fn transform_attach_db_resource_query( + parsed: &ParsedAttachDbResource<'_>, + job_id: &Uuid, + client: &AuthedClient, + hidden_passwords: &mut Arc>>, +) -> Result> { + let db_resource: Value = client + .get_resource_value_interpolated(parsed.resource_path, Some(job_id.to_string())) + .await?; + if let Some(pwd) = db_resource.get("password").and_then(|p| p.as_str()) { + hidden_passwords.lock().unwrap().push(pwd.to_string()); + } + let attach_str = format!( + "ATTACH '{}' as {} (TYPE {}{});", + format_attach_db_conn_str(db_resource, parsed.db_type)?, + parsed.name, + parsed.db_type, + parsed.extra_args.unwrap_or("") + ) + .to_string(); + + Ok(vec![ + get_attach_db_install_str(parsed.db_type)?.to_string(), + format!("LOAD {};", parsed.db_type), + attach_str, + ]) +} + +async fn transform_attach_ducklake( + query: &str, + conn: &Connection, + duckdb_connection_settings_cache: &mut DuckDbConnectionSettingsCache, + hidden_passwords: &mut Arc>>, + w_id: &str, +) -> Result>> { + lazy_static::lazy_static! { + static ref RE: regex::Regex = regex::Regex::new(r"(?i)ATTACH\s*'ducklake(://[^':]+)?'\s*AS\s+([^ ;]+)\s*(\([^)]*\))?").unwrap(); + } + let Some(cap) = RE.captures(query) else { + return Ok(None); + }; + let name = cap.get(1).map(|m| &m.as_str()[3..]).unwrap_or("main"); + let alias_name = cap.get(2).map(|m| m.as_str()).unwrap_or(""); + let extra_args = cap + .get(3) + .map(|m| format!(", {}", &m.as_str()[1..m.as_str().len() - 1])) + .unwrap_or("".to_string()); + + let ducklake = match conn { + Connection::Http(client) => get_ducklake_from_agent_http(client, name, w_id).await?, + Connection::Sql(db) => get_ducklake_from_db_unchecked(name, w_id, db).await?, + }; + let db_type = match ducklake.catalog.resource_type { + DucklakeCatalogResourceType::Instance => "postgres", + _ => ducklake.catalog.resource_type.as_ref(), + }; + + if let Some(pwd) = ducklake + .catalog_resource + .get("password") + .and_then(|p| p.as_str()) + { + hidden_passwords.lock().unwrap().push(pwd.to_string()); + } + + let db_conn_str = format_attach_db_conn_str(ducklake.catalog_resource, db_type)?; + + let storage_settings = ducklake.storage_settings; + let storage = ducklake.storage.storage; + if !duckdb_connection_settings_cache.contains_key(&storage) { + duckdb_connection_settings_cache.insert(storage.clone(), storage_settings.clone()); + }; + let s3_network_uri = + duckdb_conn_settings_to_s3_network_uri(&storage_settings, &ducklake.storage.path)?; + + let attach_str = format!( + "ATTACH 'ducklake:{db_type}:{db_conn_str}' AS {alias_name} (DATA_PATH '{s3_network_uri}'{extra_args});", + ); + + let install_db_ext_str = get_attach_db_install_str(db_type)?; + Ok(Some(vec![ + "INSTALL ducklake;".to_string(), + install_db_ext_str.to_string(), + attach_str, + ])) +} + +// Replaces all s3 URIs in the windmill syntax with the actual S3 network URIs async fn transform_s3_uris( query: &str, client: &AuthedClient, -) -> Result<( - Option, - HashMap, DuckdbConnectionSettingsResponse>, -)> { + duckdb_connection_settings_cache: &mut DuckDbConnectionSettingsCache, +) -> Result { let mut transformed_query = None; lazy_static::lazy_static! { - static ref RE: regex::Regex = regex::Regex::new(r"'s3://([^'/]*)/([^']+)'").unwrap(); + static ref RE: regex::Regex = regex::Regex::new(r"'s3://([^'/]*)/([^']*)'").unwrap(); } - let mut used_storages = HashMap::new(); for cap in RE.captures_iter(query) { if let (storage, Some(s3_path)) = (cap.get(1), cap.get(2)) { let s3_path = s3_path.as_str(); @@ -584,42 +656,63 @@ async fn transform_s3_uris( }; let original_str_lit = format!("'s3://{}/{}'", storage.as_deref().unwrap_or(""), s3_path); - let duckdb_conn_settings = client - .get_duckdb_connection_settings(&DuckdbConnectionSettingsQueryV2 { - s3_resource_path: None, - storage: storage.clone(), - }) - .await?; - let url = match &duckdb_conn_settings { - DuckdbConnectionSettingsResponse { s3_bucket: Some(bucket), .. } => { - format!("'s3://{bucket}/{s3_path}'") - } - DuckdbConnectionSettingsResponse { azure_container_path: Some(base), .. } => { - format!("'{base}/{s3_path}'") - } - _ => { - return Err(Error::ExecutionErr( - "DuckDB connection settings response must have either s3_bucket or azure_container_path".to_string(), - ))?; - } - }; + let duckdb_conn_settings = + get_duckdb_connection_settings(&storage, duckdb_connection_settings_cache, client) + .await?; + let url = duckdb_conn_settings_to_s3_network_uri(&duckdb_conn_settings, s3_path)?; + let url = format!("'{url}'"); transformed_query = Some( transformed_query .unwrap_or(query.to_string()) .replace(&original_str_lit, &url), ); - used_storages.insert(storage, duckdb_conn_settings); } } - Ok((transformed_query, used_storages)) + Ok(transformed_query.unwrap_or(query.to_string())) +} + +pub fn duckdb_conn_settings_to_s3_network_uri( + s: &DuckdbConnectionSettingsResponse, + s3_path: &str, +) -> Result { + match &s { + DuckdbConnectionSettingsResponse { s3_bucket: Some(bucket), .. } => { + Ok(format!("s3://{bucket}/{s3_path}")) + } + DuckdbConnectionSettingsResponse { azure_container_path: Some(base), .. } => { + Ok(format!("{base}/{s3_path}")) + } + _ => { + Err(Error::ExecutionErr( + "DuckDB connection settings response must have either s3_bucket or azure_container_path".to_string(), + )) + } + } } // BigQuery extension requires a json file as credentials // The file path is set as an env var by do_duckdb // It is created by transform_attach_db_resource_query (when bigquery is detected) // and deleted by do_duckdb after the query is executed -fn make_bq_credentials_path(job_id: &Uuid) -> String { - format!("/tmp/service-account-credentials-{}.json", job_id) +pub struct UseBigQueryCredentialsFile { + path: String, +} +impl UseBigQueryCredentialsFile { + fn new(job_id: Uuid, bigquery_resource: &str) -> Result { + let path = format!("/tmp/service-account-credentials-{}.json", job_id); + env::set_var("GOOGLE_APPLICATION_CREDENTIALS", &path); + std::fs::write(&path, bigquery_resource) + .map_err(|e| Error::ExecutionErr(format!("Failed to write BigQuery creds: {e}")))?; + Ok(Self { path }) + } +} +impl Drop for UseBigQueryCredentialsFile { + fn drop(&mut self) { + env::remove_var("GOOGLE_APPLICATION_CREDENTIALS"); + if matches!(std::fs::exists(&self.path), Ok(true)) { + let _ = std::fs::remove_file(&self.path); + } + } } // duckdb-rs does not support named parameters, @@ -644,9 +737,87 @@ fn interpolate_named_args<'a>( (query, values) } -fn trunc_sig(query: &str) -> &str { - let idx = query.rfind("-- $").unwrap_or(query.len()); - // find next \n starting from idx and return everything after it - let idx = query[idx..].find('\n').map(|i| i + idx).unwrap_or(0); - &query[idx..] +// input should contain a single statement. remove all comments before and after it +fn remove_comments(stmt: &str) -> &str { + let mut in_stmt = false; + let mut in_comment = false; + let mut start = None; + let mut end = stmt.len(); + + let mut c = ' '; + for (next_i, next_char) in stmt.char_indices() { + if next_i > 0 { + let i = next_i - 1; + if !in_comment && in_stmt && c == ';' { + end = i + 1; + break; + } else if in_comment && c == '\n' { + in_comment = false; + } else if c == '-' && next_char == '-' { + in_comment = true; + } else if !in_comment && !c.is_whitespace() && start == None { + start = Some(i); + in_stmt = true; + } + } + c = next_char; + } + + return &stmt[start.unwrap_or(0)..end]; } + +async fn get_duckdb_connection_settings( + storage: &Option, + cache: &mut DuckDbConnectionSettingsCache, + client: &AuthedClient, +) -> Result { + if let Some(settings) = cache.get(storage) { + return Ok(settings.clone()); + } else { + let settings = client + .get_duckdb_connection_settings(&DuckdbConnectionSettingsQueryV2 { + s3_resource_path: None, + storage: storage.clone(), + }) + .await?; + cache.insert(storage.clone(), settings.clone()); + return Ok(settings); + } +} + +#[cfg(test)] +mod tests { + use super::*; + #[test] + fn test_remove_comments_single_line() { + let sql = "-- This is a comment\nSELECT * FROM table;"; + assert_eq!(remove_comments(sql), "SELECT * FROM table;"); + } + #[test] + fn test_remove_comments_multi_line() { + let sql = "-- This is a comment\nSELECT * FROM table;\n-- Another comment"; + assert_eq!(remove_comments(sql), "SELECT * FROM table;"); + } + #[test] + fn test_remove_comments_inline_comment() { + let sql = " SELECT * FROM table; -- This is an inline comment "; + assert_eq!(remove_comments(sql), "SELECT * FROM table;"); + } + #[test] + fn test_remove_comments_no_comments() { + let sql = "SELECT * FROM table;"; + assert_eq!(remove_comments(sql), "SELECT * FROM table;"); + } + #[test] + fn test_remove_comments_empty_string() { + let sql = ""; + assert_eq!(remove_comments(sql), ""); + } + #[test] + fn test_remove_comments_with_whitespace() { + let sql = " -- Comment\n -- Comment2\n -- Comment3\n SELECT\n\n * FROM\n table\n;\n\n -- end comment "; + assert_eq!(remove_comments(sql), "SELECT\n\n * FROM\n table\n;"); + } +} + +type DuckDbConnectionSettingsCache = HashMap, DuckdbConnectionSettingsResponse>; diff --git a/frontend/package-lock.json b/frontend/package-lock.json index 3f7ddc9e30..a5268d9b79 100644 --- a/frontend/package-lock.json +++ b/frontend/package-lock.json @@ -80,11 +80,11 @@ "windmill-parser-wasm-nu": "1.510.1", "windmill-parser-wasm-php": "1.510.1", "windmill-parser-wasm-py": "1.510.1", - "windmill-parser-wasm-regex": "1.510.1", + "windmill-parser-wasm-regex": "1.512.0", "windmill-parser-wasm-rust": "1.510.1", "windmill-parser-wasm-ts": "1.514.1", "windmill-parser-wasm-yaml": "1.510.1", - "windmill-sql-datatype-parser-wasm": "1.318.0", + "windmill-sql-datatype-parser-wasm": "1.512.0", "windmill-utils-internal": "^1.0.0", "xterm": "^5.3.0", "xterm-readline": "^1.1.2", @@ -12971,9 +12971,9 @@ "integrity": "sha512-0ES6W1l/j3NjvHgGX0l6wHvioMX3SsTWe4kQIkYP12ISpAIKfSvDpE2sFm6FQ+xoxekzGqYS4luPSmKX4aDYpQ==" }, "node_modules/windmill-parser-wasm-regex": { - "version": "1.510.1", - "resolved": "https://registry.npmjs.org/windmill-parser-wasm-regex/-/windmill-parser-wasm-regex-1.510.1.tgz", - "integrity": "sha512-CUH5+DVramdKD1zVQteWIdwJ2RK9vlfipwYKvXI33DCEAUupQdLdCEyLd8Ics0aZBOrpDoWpgILqkmzPKdRJEg==" + "version": "1.512.0", + "resolved": "https://registry.npmjs.org/windmill-parser-wasm-regex/-/windmill-parser-wasm-regex-1.512.0.tgz", + "integrity": "sha512-mOkuspfjPPhGZwmerBlFOjRKHjycrlzZUpxO7gHy5D6kwKI2bSz+VI3TsJqmK1y9PwAGgO8JnwRmNAR/YLUkzA==" }, "node_modules/windmill-parser-wasm-rust": { "version": "1.510.1", @@ -12991,9 +12991,9 @@ "integrity": "sha512-zQ1imcKrhP3iccJ01BK0+tptguo3Xc+J5ku2lgrZ+YQdDcC2wjGb6gH+kvcsMXDE3WT4aRwV3nL+ecHa7WHSrw==" }, "node_modules/windmill-sql-datatype-parser-wasm": { - "version": "1.318.0", - "resolved": "https://registry.npmjs.org/windmill-sql-datatype-parser-wasm/-/windmill-sql-datatype-parser-wasm-1.318.0.tgz", - "integrity": "sha512-jlRw6abUJi4vDm+7xDSjhb7dvm4tC+lBXv0EEwn52Veadwcl5EB1yGHb9XVqQEfcYr9JU62xfJlN6DoGQmYE/g==" + "version": "1.512.0", + "resolved": "https://registry.npmjs.org/windmill-sql-datatype-parser-wasm/-/windmill-sql-datatype-parser-wasm-1.512.0.tgz", + "integrity": "sha512-uHNL8F72/Tf96xF3hOHnPDjkEyqXw7fNjcPJiUhth9sTQkcwUIoJMOdwm8/cs+j9kKVRJ4tgNYMHEBLylazp6g==" }, "node_modules/windmill-utils-internal": { "version": "1.0.0", diff --git a/frontend/package.json b/frontend/package.json index f5baf1dd19..7b7c426e19 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -147,11 +147,11 @@ "windmill-parser-wasm-nu": "1.510.1", "windmill-parser-wasm-php": "1.510.1", "windmill-parser-wasm-py": "1.510.1", - "windmill-parser-wasm-regex": "1.510.1", + "windmill-parser-wasm-regex": "1.512.0", "windmill-parser-wasm-rust": "1.510.1", "windmill-parser-wasm-ts": "1.514.1", "windmill-parser-wasm-yaml": "1.510.1", - "windmill-sql-datatype-parser-wasm": "1.318.0", + "windmill-sql-datatype-parser-wasm": "1.512.0", "windmill-utils-internal": "^1.0.0", "xterm": "^5.3.0", "xterm-readline": "^1.1.2", diff --git a/frontend/src/lib/components/DBManagerDrawer.svelte b/frontend/src/lib/components/DBManagerDrawer.svelte index b0da47c4ac..583f7f30a6 100644 --- a/frontend/src/lib/components/DBManagerDrawer.svelte +++ b/frontend/src/lib/components/DBManagerDrawer.svelte @@ -11,11 +11,16 @@ getLanguageByResourceType, loadAllTablesMetaData, loadTableMetaData, - type DbType, type TableMetadata } from './apps/components/display/dbtable/utils' import DbManager from './DBManager.svelte' - import { dbDeleteTableActionWithPreviewScript, dbTableOpsWithPreviewScripts } from './dbOps' + import { + dbDeleteTableActionWithPreviewScript, + dbTableOpsWithPreviewScripts, + getDucklakeSchema, + wrapDucklakeQuery, + type DbInput + } from './dbOps' import { makeCreateTableQuery } from './apps/components/display/dbtable/queries/createTable' import { runScriptAndPollResult } from './jobs/utils' import { Pane, Splitpanes } from 'svelte-splitpanes' @@ -23,25 +28,28 @@ import SimpleAgTable from './SimpleAgTable.svelte' import { untrack } from 'svelte' - let resourceType: DbType | undefined = $state(undefined) - let resourcePath: string | undefined = $state(undefined) - let open = $derived(resourcePath && resourceType) + let input: DbInput | undefined = $state() + let open = $derived(!!input) - export function openDrawer(_resourceType: DbType, _resourcePath: string) { - resourceType = _resourceType - resourcePath = _resourcePath + export function openDrawer(nInput: DbInput) { + input = nInput getSchema() } export function closeDrawer() { - resourceType = undefined - resourcePath = undefined + input = undefined refreshCount = 0 refreshing = false } - let dbSchema: DBSchema | undefined = $derived( - resourcePath && resourcePath in $dbSchemas ? $dbSchemas[resourcePath] : undefined - ) + let dbSchema: DBSchema | undefined = $derived(input && $dbSchemas[getDbSchemasPath(input)]) + function getDbSchemasPath(input: DbInput): string { + switch (input.type) { + case 'database': + return input.resourcePath + case 'ducklake': + return 'ducklake://' + input.ducklake + } + } // `refreshCount` is a derived state. `refreshing` is the source of truth let refreshCount = $state(0) @@ -61,26 +69,32 @@ }) async function getSchema() { - if (!resourcePath || !resourceType || ($dbSchemas[resourcePath] && !refreshing)) return - try { - const oldDbSchema = $dbSchemas[resourcePath] + if (!input) return + const dbSchemasPath = getDbSchemasPath(input) + if ($dbSchemas[dbSchemasPath] && !refreshing) return + + const oldDbSchema = $dbSchemas[dbSchemasPath] + if (input.type == 'database') { await getDbSchemas( - resourceType, - resourcePath, + input.resourceType, + input.resourcePath, $workspaceStore, $dbSchemas, (message: string) => { if (open) sendUserToast(message, true) } ) - // avoid infinite loop on error due to the way getDbSchemas is implemented - // and relying on an assignement side effect - if (oldDbSchema !== $dbSchemas[resourcePath]) $dbSchemas = $dbSchemas - } catch (e) { - console.error(e) - } finally { - refreshing = false + } else if (input.type == 'ducklake') { + $dbSchemas[dbSchemasPath] = await getDucklakeSchema({ + workspace: $workspaceStore!, + ducklake: input.ducklake + }) } + + // avoid infinite loop on error due to the way getDbSchemas is implemented + // and relying on an assignement side effect + if (oldDbSchema !== $dbSchemas[dbSchemasPath]) $dbSchemas = $dbSchemas + refreshing = false } let windowWidth = $state(window.innerWidth) @@ -93,26 +107,21 @@ let cachedColDefs: Record = {} let cachedLastRefreshCount = 0 - async function getColDefs(tableKey: string) { - if (!resourcePath || !resourceType) return [] - + async function getColDefs(tableKey: string): Promise { if (cachedLastRefreshCount !== refreshCount) cachedColDefs = {} cachedLastRefreshCount = refreshCount - if (cachedColDefs[tableKey]) { - return cachedColDefs[tableKey] - } + + if (cachedColDefs[tableKey]) return cachedColDefs[tableKey] + if (!input) return [] + try { - cachedColDefs = - (await loadAllTablesMetaData('$res:' + resourcePath, $workspaceStore, resourceType)) ?? - cachedColDefs + cachedColDefs = (await loadAllTablesMetaData($workspaceStore, input)) ?? cachedColDefs return cachedColDefs[tableKey] } catch (e) { - const result = await loadTableMetaData( - '$res:' + resourcePath, - $workspaceStore, - tableKey, - resourceType - ) + if (input?.type == 'ducklake') + throw 'Impossible that loadAllTablesMetaData fails for Ducklake' + // Query is not implemented for all dbs, need a fallback + const result = await loadTableMetaData(input, $workspaceStore, tableKey) if (result) cachedColDefs[tableKey] = result return result ?? [] @@ -137,126 +146,120 @@ preventEscape on:close={closeDrawer} > - {#key [resourceType, resourcePath, dbSchema]} - { - if (replResultData) { - replResultData = undefined - } else { - closeDrawer() - } - }} - CloseIcon={replResultData ? ArrowLeft : undefined} - noPadding - > - {#if dbSchema && $workspaceStore && resourceType && resourcePath} - - - - -
{ - // Only proceed if the click is directly on this div and not on the child elements - if (e.target === e.currentTarget) { - replResultData = undefined - } - }} - > - {#if replResultData} - {#key replResultData} - - {/key} - {/if} -
- - dbTableOpsWithPreviewScripts({ - colDefs, - tableKey, - resourcePath: resourcePath!, - resourceType: resourceType!, - workspace: $workspaceStore - })} - dbTableActionsFactory={[ - dbDeleteTableActionWithPreviewScript({ - resourcePath, - resourceType, - workspace: $workspaceStore - }) - ]} - {refresh} - dbTableEditorPropsFactory={({ selectedSchemaKey }) => ({ - resourceType: resourceType!, - previewSql: (values) => - makeCreateTableQuery(values, resourceType!, selectedSchemaKey), - async onConfirm(values) { - await runScriptAndPollResult({ - workspace: $workspaceStore, - requestBody: { - args: { database: '$res:' + resourcePath }, - content: makeCreateTableQuery(values, resourceType!, selectedSchemaKey), - language: getLanguageByResourceType(resourceType!) - } - }) - refresh() - } + { + if (replResultData) { + replResultData = undefined + } else { + closeDrawer() + } + }} + CloseIcon={replResultData ? ArrowLeft : undefined} + noPadding + > + {#if dbSchema && $workspaceStore && input} + {@const _input = input} + {@const dbType = input.type == 'database' ? input.resourceType : 'duckdb'} + + + + +
{ + // Only proceed if the click is directly on this div and not on the child elements + if (e.target === e.currentTarget) { + replResultData = undefined + } + }} + > + {#if replResultData} + {#key replResultData} + + {/key} + {/if} +
+ + dbTableOpsWithPreviewScripts({ + colDefs, + tableKey, + input: _input, + workspace: $workspaceStore })} - /> - - - { - replResultData = data - }} - placeholderTableName={sortArray( - Object.keys( - dbSchema?.schema[ - 'public' in dbSchema?.schema - ? 'public' - : 'dbo' in dbSchema?.schema - ? 'dbo' - : Object.keys(dbSchema?.schema)?.[0] - ] - ) - )?.[0]} - /> - - - {:else} - - - - - - {/if} - {#snippet actions()} - + dbTableActionsFactory={[ + dbDeleteTableActionWithPreviewScript({ input: _input, workspace: $workspaceStore }) + ]} + {refresh} + dbTableEditorPropsFactory={({ selectedSchemaKey }) => ({ + dbType, + previewSql: (values) => makeCreateTableQuery(values, dbType, selectedSchemaKey), + async onConfirm(values) { + const dbArg = + input?.type === 'database' ? { database: '$res:' + input.resourcePath } : {} + const language = getLanguageByResourceType(dbType) + let query = makeCreateTableQuery(values, dbType, selectedSchemaKey) + if (input?.type === 'ducklake') query = wrapDucklakeQuery(query, input.ducklake) + await runScriptAndPollResult({ + workspace: $workspaceStore, + requestBody: { args: dbArg, content: query, language } + }) + refresh() + } + })} + /> + + + { + replResultData = data + }} + placeholderTableName={sortArray( + Object.keys( + dbSchema?.schema[ + 'public' in dbSchema?.schema + ? 'public' + : 'dbo' in dbSchema?.schema + ? 'dbo' + : Object.keys(dbSchema?.schema)?.[0] + ] + ) + )?.[0]} + /> + + + {:else} + + + + + + {/if} + {#snippet actions()} + - diff --git a/frontend/src/lib/components/ExploreAssetButton.svelte.d.ts b/frontend/src/lib/components/ExploreAssetButton.svelte.d.ts deleted file mode 100644 index e35cd917dc..0000000000 --- a/frontend/src/lib/components/ExploreAssetButton.svelte.d.ts +++ /dev/null @@ -1,23 +0,0 @@ -export declare function assetCanBeExplored(asset: Asset, _resourceMetadata?: { - resource_type?: string; -}): boolean; -import { type Asset } from '$lib/components/assets/lib'; -import { ButtonType } from '$lib/components/common'; -import DbManagerDrawer from '$lib/components/DBManagerDrawer.svelte'; -import S3FilePicker from '$lib/components/S3FilePicker.svelte'; -type $$ComponentProps = { - asset: Asset; - _resourceMetadata?: { - resource_type?: string; - }; - s3FilePicker?: S3FilePicker; - dbManagerDrawer?: DbManagerDrawer; - onClick?: () => void; - class?: string; - noText?: boolean; - buttonVariant?: ButtonType.Variant; - btnClasses?: string; -}; -declare const ExploreAssetButton: import("svelte").Component<$$ComponentProps, {}, "">; -type ExploreAssetButton = ReturnType; -export default ExploreAssetButton; diff --git a/frontend/src/lib/components/SqlRepl.svelte b/frontend/src/lib/components/SqlRepl.svelte index 3f589c86c5..683b35cb1c 100644 --- a/frontend/src/lib/components/SqlRepl.svelte +++ b/frontend/src/lib/components/SqlRepl.svelte @@ -55,14 +55,15 @@ import { getLanguageByResourceType } from './apps/components/display/dbtable/utils' import StepHistory, { type StepHistoryData } from './flows/propPicker/StepHistory.svelte' import { Pane, Splitpanes } from 'svelte-splitpanes' + import { getDbType, wrapDucklakeQuery, type DbInput } from './dbOps' type Props = { - resourceType: string - resourcePath: string + input: DbInput onData: (data: Record[]) => void placeholderTableName?: string } - let { resourcePath, resourceType, onData, placeholderTableName }: Props = $props() + let { input, onData, placeholderTableName }: Props = $props() + let dbType = $derived(getDbType(input)) const DEFAULT_SQL = 'SELECT * FROM _' let code = $state(DEFAULT_SQL) @@ -100,16 +101,19 @@ }) .join(';') } + const dbArg = input?.type === 'database' ? { database: '$res:' + input.resourcePath } : {} + + if (input?.type === 'ducklake') { + transformedCode = wrapDucklakeQuery(transformedCode, input.ducklake) + } let { job, result } = (await runScriptAndPollResult( { workspace: $workspaceStore, requestBody: { - language: getLanguageByResourceType(resourceType), + language: getLanguageByResourceType(dbType), content: transformedCode, - args: { - database: '$res:' + resourcePath - } + args: dbArg } }, { withJobData: true } @@ -142,7 +146,7 @@ else sendUserToast('Query executed') } catch (e) { console.error(e) - if (resourceType === 'postgresql' && !doPostgresRowToJsonFix) { + if (dbType === 'postgresql' && !doPostgresRowToJsonFix) { console.error('Error running query, trying with row_to_json fix') isRunning = false return await run({ doPostgresRowToJsonFix: true }) diff --git a/frontend/src/lib/components/apps/components/display/dbtable/AppDbExplorer.svelte b/frontend/src/lib/components/apps/components/display/dbtable/AppDbExplorer.svelte index b7abd1810d..56271f6ca0 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/AppDbExplorer.svelte +++ b/frontend/src/lib/components/apps/components/display/dbtable/AppDbExplorer.svelte @@ -38,6 +38,7 @@ import RefreshButton from '$lib/components/apps/components/helpers/RefreshButton.svelte' import RunnableWrapper from '../../helpers/RunnableWrapper.svelte' import InsertRowDrawerButton from '../InsertRowDrawerButton.svelte' + import { assert } from '$lib/utils' interface Props { id: string @@ -365,11 +366,16 @@ gridItem.data = gridItem.data $app = $app + let resource = resolvedConfig.type.configuration[selected].resource + assert('resource starts with $res:', resource?.startsWith('$res:'), resource) let tableMetadata = await loadTableMetaData( - resolvedConfig.type.configuration[selected].resource, + { + type: 'database', + resourcePath: resource.substring(5), + resourceType: selected + }, $workspaceStore, - resolvedConfig.type.configuration[selected].table, - selected + resolvedConfig.type.configuration[selected].table ) if (!tableMetadata) return diff --git a/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte b/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte index 773d37e8d0..c085cf97b7 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte +++ b/frontend/src/lib/components/apps/components/display/dbtable/InsertRow.svelte @@ -7,7 +7,8 @@ parse_mysql, parse_bigquery, parse_snowflake, - parse_mssql + parse_mssql, + parse_duckdb } from 'windmill-sql-datatype-parser-wasm' import wasmUrl from 'windmill-sql-datatype-parser-wasm/windmill_sql_datatype_parser_wasm_bg.wasm?url' @@ -46,8 +47,9 @@ case 'ms_sql_server': rawType = parse_mssql(field) break - default: - throw new Error('Language not supported') + case 'duckdb': + rawType = parse_duckdb(field) + break } return rawType diff --git a/frontend/src/lib/components/apps/components/display/dbtable/queries/count.ts b/frontend/src/lib/components/apps/components/display/dbtable/queries/count.ts index 482752b844..02d5d768ef 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/queries/count.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/queries/count.ts @@ -105,7 +105,16 @@ export function makeCountQuery( query += `SELECT COUNT(*) as count FROM \`${table}\`` break } - + case 'duckdb': + if (filteredColumns.length > 0) { + quicksearchCondition += ` ($quicksearch = '' OR CONCAT(' ', ${filteredColumns.join( + ', ' + )}) LIKE CONCAT('%', $quicksearch, '%'))` + } else { + quicksearchCondition += ` ($quicksearch = '' OR 1 = 1)` + } + query += `SELECT COUNT(*) as count FROM ${table}` + break default: throw new Error('Unsupported database type:' + dbType) } @@ -122,7 +131,8 @@ export function makeCountQuery( (dbType === 'mysql' || dbType === 'postgresql' || dbType === 'snowflake' || - dbType === 'bigquery') + dbType === 'bigquery' || + dbType === 'duckdb') ) { query = query.replace(`${andCondition}`, wherePrefix) } diff --git a/frontend/src/lib/components/apps/components/display/dbtable/queries/createTable.ts b/frontend/src/lib/components/apps/components/display/dbtable/queries/createTable.ts index 06a952312a..5685ba24be 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/queries/createTable.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/queries/createTable.ts @@ -23,16 +23,12 @@ type CreateTableValuesColumn = { datatype_length?: number // e.g varchar(255) } -export function makeCreateTableQuery( - values: CreateTableValues, - resourceType: DbType, - schema?: string -) { +export function makeCreateTableQuery(values: CreateTableValues, dbType: DbType, schema?: string) { const pkCount = values.columns.reduce((p, c) => p + (c.primaryKey ? 1 : 0), 0) function transformColumn(c: CreateTableValuesColumn): string { const datatype = c.datatype_length ? `${c.datatype}(${c.datatype_length})` : c.datatype - const defValue = c.defaultValue && formatDefaultValue(c.defaultValue, datatype, resourceType) + const defValue = c.defaultValue && formatDefaultValue(c.defaultValue, datatype, dbType) let str = ` ${c.name} ${datatype}` if (c.not_null) str += ' NOT NULL' @@ -57,7 +53,7 @@ export function makeCreateTableQuery( return l } - const useSchema = dbSupportsSchemas(resourceType) + const useSchema = dbSupportsSchemas(dbType) const lines = values.columns.map(transformColumn) lines.push(...values.foreignKeys.map(transformFk)) diff --git a/frontend/src/lib/components/apps/components/display/dbtable/queries/delete.ts b/frontend/src/lib/components/apps/components/display/dbtable/queries/delete.ts index bf343cb974..5ff2bba2aa 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/queries/delete.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/queries/delete.ts @@ -52,6 +52,13 @@ export function makeDeleteQuery(table: string, columns: ColumnDef[], dbType: DbT query += `\nDELETE FROM ${table} \nWHERE ${conditions}` return query } + case 'duckdb': { + const conditions = columns + .map((c) => `($${c.field} IS NULL AND ${c.field} IS NULL OR ${c.field} = $${c.field})`) + .join('\n AND ') + query += `\nDELETE FROM ${table} \nWHERE ${conditions}` + return query + } default: throw new Error('Unsupported database type') } diff --git a/frontend/src/lib/components/apps/components/display/dbtable/queries/insert.ts b/frontend/src/lib/components/apps/components/display/dbtable/queries/insert.ts index 0b908504f4..c13b6c1bc2 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/queries/insert.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/queries/insert.ts @@ -14,6 +14,8 @@ function formatInsertValues(columns: ColumnDef[], dbType: DbType, startIndex: nu return columns.map(() => `?`).join(', ') case 'bigquery': return columns.map((c) => `@${c.field}`).join(', ') + case 'duckdb': + return columns.map((c) => `$${c.field}`).join(', ') default: throw new Error('Unsupported database type') } diff --git a/frontend/src/lib/components/apps/components/display/dbtable/queries/select.ts b/frontend/src/lib/components/apps/components/display/dbtable/queries/select.ts index f4453360ec..c440be4b7c 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/queries/select.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/queries/select.ts @@ -217,6 +217,28 @@ CASE WHEN :order_by = '${column.field}' AND :is_desc IS true THEN \`${column.fie query += ` LIMIT @limit OFFSET @offset` break } + case 'duckdb': { + const orderBy = ` + ${columnDefs + .map( + (column) => + ` + (CASE WHEN $order_by = '${column.field}' AND $is_desc IS false THEN "${column.field}"::text END), + (CASE WHEN $order_by = '${column.field}' AND $is_desc IS true THEN "${column.field}"::text END) DESC` + ) + .join(',\n')}` + + quicksearchCondition = `($quicksearch = '' OR CONCAT(${filteredColumns.join( + ', ' + )}) ILIKE '%' || $quicksearch || '%')` + + query += `SELECT ${filteredColumns.join(', ')} FROM ${table}\n` + query += ` WHERE ${whereClause ? `${whereClause} AND` : ''} ${quicksearchCondition}\n` + query += ` ORDER BY ${orderBy}\n` + query += ` LIMIT $limit::INT OFFSET $offset::INT` + + break + } default: throw new Error('Unsupported database type') diff --git a/frontend/src/lib/components/apps/components/display/dbtable/queries/update.ts b/frontend/src/lib/components/apps/components/display/dbtable/queries/update.ts index 467922b9d7..38258eeb0b 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/queries/update.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/queries/update.ts @@ -62,6 +62,13 @@ export function makeUpdateQuery( query += `\nUPDATE ${table} SET ${column.field} = @value_to_update \nWHERE ${conditions}` return query } + case 'duckdb': { + const conditions = columns + .map((c) => `($${c.field} IS NULL AND ${c.field} IS NULL OR ${c.field} = $${c.field})`) + .join('\n AND ') + query += `\nUPDATE ${table} SET ${column.field} = $value_to_update \nWHERE ${conditions}` + return query + } default: throw new Error('Unsupported database type') } diff --git a/frontend/src/lib/components/apps/components/display/dbtable/utils.ts b/frontend/src/lib/components/apps/components/display/dbtable/utils.ts index 05b30ec1f2..dd755bc636 100644 --- a/frontend/src/lib/components/apps/components/display/dbtable/utils.ts +++ b/frontend/src/lib/components/apps/components/display/dbtable/utils.ts @@ -1,4 +1,4 @@ -import { JobService, type Preview, ResourceService } from '$lib/gen' +import { JobService, type Preview, ResourceService, type ScriptLang } from '$lib/gen' import type { DBSchema, DBSchemas, GraphqlSchema, SQLSchema } from '$lib/stores' import { buildClientSchema, @@ -9,6 +9,7 @@ import { import { tryEvery } from '$lib/utils' import { stringifySchema } from '$lib/components/copilot/lib' import { runScriptAndPollResult } from '$lib/components/jobs/utils' +import type { DbInput } from '$lib/components/dbOps' export enum ColumnIdentity { ByDefault = 'By Default', @@ -54,22 +55,21 @@ export type ColumnDef = { } & ColumnMetadata export async function loadTableMetaData( - resource: string, + input: DbInput, workspace: string | undefined, - table: string | undefined, - resourceType: string + table: string | undefined ): Promise { - if (!resource || !table || !workspace) { - return undefined - } + if (!input || !table || !workspace) return undefined + + let language = input.type == 'ducklake' ? 'duckdb' : getLanguageByResourceType(input.resourceType) + let content = await makeLoadTableMetaDataQuery(input, workspace, table) + const job = await JobService.runScriptPreview({ workspace: workspace, requestBody: { - language: getLanguageByResourceType(resourceType), - content: await makeLoadTableMetaDataQuery(resource, workspace, table, resourceType), - args: { - database: resource - } + language, + content, + args: input.type === 'database' ? { database: '$res:' + input.resourcePath } : {} } }) @@ -87,7 +87,7 @@ export async function loadTableMetaData( if (testResult.success) { attempts = maxRetries - if (resourceType === 'ms_sql_server') { + if (input.type === 'database' && input.resourceType === 'ms_sql_server') { return testResult.result[0].map(lowercaseKeys) } else { return testResult.result.map(lowercaseKeys) @@ -105,26 +105,23 @@ export async function loadTableMetaData( } export async function loadAllTablesMetaData( - resource: string, workspace: string | undefined, - resourceType: string + input: DbInput ): Promise | undefined> { - if (!resource || !workspace) { - return undefined - } + if (!input || !workspace) return undefined + + let language = input.type == 'ducklake' ? 'duckdb' : getLanguageByResourceType(input.resourceType) try { let result = (await runScriptAndPollResult({ workspace: workspace, requestBody: { - language: getLanguageByResourceType(resourceType), - content: await makeLoadTableMetaDataQuery(resource, workspace, undefined, resourceType), - args: { - database: resource - } + language, + content: await makeLoadTableMetaDataQuery(input, workspace, undefined), + args: input.type === 'database' ? { database: '$res:' + input.resourcePath } : {} } })) as ({ table_name: string; schema_name?: string } & object)[] - if (resourceType === 'ms_sql_server') { + if (input.type === 'database' && input.resourceType === 'ms_sql_server') { result = (result as any)[0] } const map: Record = {} @@ -144,15 +141,27 @@ export async function loadAllTablesMetaData( } async function makeLoadTableMetaDataQuery( - resource: string, + input: DbInput, workspace: string, - table: string | undefined, - resourceType: string + table: string | undefined ): Promise { - if (resourceType === 'mysql') { + if (input.type === 'ducklake') { + return `ATTACH 'ducklake://${input.ducklake}' AS __ducklake__; + SELECT + COLUMN_NAME as field, + DATA_TYPE as DataType, + COLUMN_DEFAULT as DefaultValue, + false as IsPrimaryKey, + false as IsIdentity, + IS_NULLABLE as IsNullable, + false as IsEnum, + TABLE_NAME as table_name + FROM information_schema.columns c + WHERE table_catalog = '__ducklake__' AND table_schema = current_schema()` + } else if (input.resourceType === 'mysql') { const resourceObj = (await ResourceService.getResourceValue({ workspace, - path: resource.split(':')[1] + path: input.resourcePath })) as any return ` SELECT @@ -184,7 +193,7 @@ async function makeLoadTableMetaDataQuery( TABLE_NAME, ORDINAL_POSITION; ` - } else if (resourceType === 'postgresql') { + } else if (input.resourceType === 'postgresql') { return ` SELECT a.attname as field, @@ -231,7 +240,7 @@ async function makeLoadTableMetaDataQuery( ORDER BY ${table ? 'a.attnum' : 'ns.nspname, c.relname, a.attnum'}; ` - } else if (resourceType === 'ms_sql_server') { + } else if (input.resourceType === 'ms_sql_server') { return ` SELECT COLUMN_NAME as field, @@ -257,7 +266,10 @@ WHERE ORDER BY ORDINAL_POSITION; ` - } else if (resourceType === 'snowflake' || resourceType === 'snowflake_oauth') { + } else if ( + input.resourceType === 'snowflake' || + (input.resourceType as any) === 'snowflake_oauth' + ) { return ` select COLUMN_NAME as field, DATA_TYPE as DataType, @@ -282,7 +294,7 @@ ORDER BY } order by ORDINAL_POSITION; ` - } else if (resourceType === 'bigquery') { + } else if (input.resourceType === 'bigquery') { // TODO: find a solution for this (query uses hardcoded dataset name) if (!table) throw new Error('Table name is required for BigQuery') return `SELECT @@ -302,7 +314,7 @@ WHERE c.TABLE_NAME = '${table.split('.')[1]}' order by c.ORDINAL_POSITION;` } else { - throw new Error('Unsupported database type:' + resourceType) + throw new Error('Unsupported database type:' + input.resourceType) } } @@ -656,7 +668,14 @@ export function formatGraphqlSchema(schema: IntrospectionQuery): string { } export type DbType = (typeof dbTypes)[number] -export const dbTypes = ['mysql', 'ms_sql_server', 'postgresql', 'snowflake', 'bigquery'] as const +export const dbTypes = [ + 'mysql', + 'ms_sql_server', + 'postgresql', + 'snowflake', + 'bigquery', + 'duckdb' +] as const export const isDbType = (str?: string): str is DbType => !!str && dbTypes.includes(str as DbType) export function buildVisibleFieldList(columnDefs: ColumnDef[], dbType: DbType) { @@ -675,20 +694,23 @@ export function buildVisibleFieldList(columnDefs: ColumnDef[], dbType: DbType) { return `"${column?.field}"` // Snowflake uses double quotes for identifiers case 'bigquery': return `\`${column?.field}\`` // BigQuery uses backticks + case 'duckdb': + return `"${column?.field}"` // DuckDB uses double quotes for identifiers default: throw new Error('Unsupported database type') } }) } -export function getLanguageByResourceType(name: string): Preview['language'] { +export function getLanguageByResourceType(name: string): ScriptLang { const language = { postgresql: 'postgresql', mysql: 'mysql', ms_sql_server: 'mssql', snowflake: 'snowflake', snowflake_oauth: 'snowflake', - bigquery: 'bigquery' + bigquery: 'bigquery', + duckdb: 'duckdb' } return language[name] } @@ -713,6 +735,8 @@ export function buildParameters( return `-- ? ${column.field} (${column.datatype.split('(')[0]})` case 'bigquery': return `-- @${column.field} (${column.datatype.split('(')[0]})` + case 'duckdb': + return `-- $${column.field} (${column.datatype.split('(')[0]})` } }) .join('\n') diff --git a/frontend/src/lib/components/assets/AssetsDropdownButton.svelte b/frontend/src/lib/components/assets/AssetsDropdownButton.svelte index f451b36bb9..9418fa13c2 100644 --- a/frontend/src/lib/components/assets/AssetsDropdownButton.svelte +++ b/frontend/src/lib/components/assets/AssetsDropdownButton.svelte @@ -163,7 +163,7 @@
- + {asset.path} {asset.path} diff --git a/frontend/src/lib/components/assets/JobAssetsViewer.svelte b/frontend/src/lib/components/assets/JobAssetsViewer.svelte index fcaad2416d..f1977073a0 100644 --- a/frontend/src/lib/components/assets/JobAssetsViewer.svelte +++ b/frontend/src/lib/components/assets/JobAssetsViewer.svelte @@ -3,7 +3,7 @@ import { inferAssets } from '$lib/infer' import { workspaceStore } from '$lib/stores' import { usePromise } from '$lib/svelte5Utils.svelte' - import { pruneNullishArrayWithSet, uniqueBy } from '$lib/utils' + import { pruneNullishArray, uniqueBy } from '$lib/utils' import DbManagerDrawer from '../DBManagerDrawer.svelte' import ResourceEditorDrawer from '../ResourceEditorDrawer.svelte' import S3FilePicker from '../S3FilePicker.svelte' @@ -26,7 +26,7 @@ // TODO : Transitive assets } return uniqueBy( - pruneNullishArrayWithSet([ + pruneNullishArray([ ...(job.raw_flow?.modules.flatMap((m) => getFlowModuleAssets(m, additionalAssetsMap)) ?? []), ...parseInputArgsAssets(job.args ?? {}) diff --git a/frontend/src/lib/components/assets/lib.ts b/frontend/src/lib/components/assets/lib.ts index a58f94de4f..f002beed15 100644 --- a/frontend/src/lib/components/assets/lib.ts +++ b/frontend/src/lib/components/assets/lib.ts @@ -20,6 +20,8 @@ export function formatAsset(asset: Asset): string { return `res://${asset.path}` case 's3object': return `s3://${asset.path}` + case 'ducklake': + return `ducklake://${asset.path}` } } @@ -68,6 +70,8 @@ export function formatAssetKind(asset: { } case 's3object': return 'S3 Object' + case 'ducklake': + return 'Ducklake' } } diff --git a/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte b/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte index 6bb4f1f0e5..fa8bc635cf 100644 --- a/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte +++ b/frontend/src/lib/components/common/confirmationModal/ConfirmationModal.svelte @@ -1,6 +1,6 @@ @@ -277,10 +278,9 @@ {/if}
- Used in {pluralize( - flowGraphAssetsCtx?.val.computeAssetsCount?.(data.asset) ?? -1, - 'step' - )}
+ {#if usageCount !== undefined} + Used in {pluralize(usageCount, 'step')}
+ {/if} + interface Props { + height?: string + width?: string + fill?: string + class?: string + } + + let { height = '24px', width = '24px', fill = 'black', class: className = '' }: Props = $props() + + + + + + + diff --git a/frontend/src/lib/components/icons/AssetGenericIcon.svelte b/frontend/src/lib/components/icons/AssetGenericIcon.svelte index be8f4feb58..16263f6344 100644 --- a/frontend/src/lib/components/icons/AssetGenericIcon.svelte +++ b/frontend/src/lib/components/icons/AssetGenericIcon.svelte @@ -3,6 +3,7 @@ import type { AssetKind } from '../assets/lib' import AssetResIcon from './AssetResIcon.svelte' import AssetS3Icon from './AssetS3Icon.svelte' + import AssetDucklakeIcon from './AssetDucklakeIcon.svelte' interface Props { size?: string @@ -18,6 +19,8 @@ {:else if assetKind == 'resource'} +{:else if assetKind == 'ducklake'} + {:else} {/if} diff --git a/frontend/src/lib/components/icons/DucklakeIcon.svelte b/frontend/src/lib/components/icons/DucklakeIcon.svelte new file mode 100644 index 0000000000..6bf68a3395 --- /dev/null +++ b/frontend/src/lib/components/icons/DucklakeIcon.svelte @@ -0,0 +1,22 @@ + + + + + + diff --git a/frontend/src/lib/components/workspaceSettings/DucklakeSettings.svelte b/frontend/src/lib/components/workspaceSettings/DucklakeSettings.svelte new file mode 100644 index 0000000000..587f2ffa33 --- /dev/null +++ b/frontend/src/lib/components/workspaceSettings/DucklakeSettings.svelte @@ -0,0 +1,341 @@ + + + + +
+
+
Ducklake
+ + Windmill has first class support for Ducklake. You can use and explore ducklakes like a normal + SQL database, even through the data is actually stored in parquet files in S3 ! + +
+
+ +{#if ducklakeSettings.ducklakes.some((d) => d.catalog.resource_type === 'instance')} + + Using an instance catalog is the fastest way to get started with Ducklake. They are public to + the instance and can be re-used in other workspaces' Ducklake settings. + +{/if} + + + + + {#each tableHeadNames as name, i} + + {name} + {#if tableHeadTooltips[name]} + + {@html tableHeadTooltips[name]} + + {/if} + + {/each} + + + + {#if ducklakeSettings.ducklakes.length == 0} + + + No ducklake in this workspace yet + + + {/if} + {#each ducklakeSettings.ducklakes as ducklake, ducklakeIndex} + + + {#if ducklake.name === 'main'} + + The main ducklake can be accessed with the +
+ ATTACH 'ducklake' AS dl; shorthand + + {/if} + + + +
+
+ {#if ducklake.catalog.resource_type === 'instance'} + + Use Windmill's PostgreSQL instance as a catalog + + {/if} + + {/if} +
+
+ + +
+ +
+ + + {#if ducklakeIsDirty[ducklake.name]} + + + + + Please save settings first + + {:else} + + {/if} + + + removeDucklake(ducklakeIndex)} /> + + + {/each} + + +
+ +
+ + + + + + + + diff --git a/frontend/src/lib/consts.ts b/frontend/src/lib/consts.ts index 847bbf42da..32f54f723c 100644 --- a/frontend/src/lib/consts.ts +++ b/frontend/src/lib/consts.ts @@ -259,5 +259,6 @@ export const DB_TYPES: Record = { ms_sql_server: MSSQL_TYPES, mysql: MYSQL_TYPES, postgresql: POSTGRES_TYPES, - snowflake: SNOWFLAKE_TYPES + snowflake: SNOWFLAKE_TYPES, + duckdb: DUCKDB_TYPES } diff --git a/frontend/src/lib/infer.ts b/frontend/src/lib/infer.ts index 8c6599bc06..702d8c7265 100644 --- a/frontend/src/lib/infer.ts +++ b/frontend/src/lib/infer.ts @@ -87,7 +87,8 @@ export async function inferAssets( try { if (language === 'duckdb') { await initWasmRegex() - return JSON.parse(parse_assets_sql(code)) + let r = JSON.parse(parse_assets_sql(code)) + return r } if (language === 'deno' || language === 'nativets' || language === 'bun') { await initWasmTs() diff --git a/frontend/src/lib/script_helpers.ts b/frontend/src/lib/script_helpers.ts index 07c05974dc..819565336a 100644 --- a/frontend/src/lib/script_helpers.ts +++ b/frontend/src/lib/script_helpers.ts @@ -307,17 +307,33 @@ INSERT INTO demo VALUES (@P1, @P2); UPDATE demo SET col2 = @P3 WHERE col2 = @P2; ` -const DUCKDB_INIT_CODE = `-- $friends_csv (s3object) --- $name (text) = Ben +const DUCKDB_INIT_CODE = `-- $name (text) = Ben +-- $age (text) = 20 +-- -- $friends_csv (s3object) -ATTACH '$res:u/demo/amazed_postgresql' AS db (TYPE postgres); -CREATE TABLE IF NOT EXISTS db.public.friends (name text); +-- Click the +Database button to connect to a database +-- https://www.windmill.dev/docs/getting_started/scripts_quickstart/sql#duckdb-1 +-- +-- ATTACH '$res:u/demo/amazed_postgresql' AS db (TYPE postgres); +-- SELECT * FROM db.public.friends; -INSERT INTO db.public.friends - SELECT name FROM read_csv($friends_csv); +-- Click the +Ducklake button to use a ducklake +-- https://www.windmill.dev/docs/core_concepts/ducklake +-- +-- ATTACH 'ducklake' AS dl; +-- USE dl; +-- SELECT * FROM customers; -SELECT 'Hello ' || $name || ', you have ' || - (SELECT COUNT(*) FROM read_csv($friends_csv)) || ' new friends !'; +CREATE TABLE friends ( + name text, + age int +); + +INSERT INTO friends VALUES ($name, $age); +-- INSERT INTO friends +-- SELECT name, age FROM read_csv($friends_csv); + +SELECT * FROM friends; ` const GRAPHQL_INIT_CODE = `query($name4: String, $name2: Int, $name3: [String]) { diff --git a/frontend/src/lib/stores.ts b/frontend/src/lib/stores.ts index e11ca41ca4..c27baa1d1c 100644 --- a/frontend/src/lib/stores.ts +++ b/frontend/src/lib/stores.ts @@ -203,7 +203,7 @@ export const SQLSchemaLanguages = [ ] as const export interface SQLSchema { - lang: (typeof SQLSchemaLanguages)[number] + lang: (typeof SQLSchemaLanguages)[number] | 'ducklake' schema: SQLBaseSchema publicOnly: boolean | undefined stringified: string diff --git a/frontend/src/lib/utils.ts b/frontend/src/lib/utils.ts index 35c1f79273..99b1ccef76 100644 --- a/frontend/src/lib/utils.ts +++ b/frontend/src/lib/utils.ts @@ -1530,6 +1530,16 @@ export function uniqueBy(array: T[], key: (t: T) => any): T[] { }) } -export function pruneNullishArrayWithSet(array: (T | null | undefined)[]): T[] { +export function pruneNullishArray(array: (T | null | undefined)[]): T[] { return array.filter((item): item is T => item !== null && item !== undefined) } + +export function assert(msg: string, condition: boolean, value?: any) { + if (!condition) { + let m = 'Assertion failed: ' + msg + if (value) m += '\nValue: ' + JSON.stringify(value, null, 2) + m += '\nPlease alert the Windmill team about this' + sendUserToast(m, true) + console.error(m) + } +} diff --git a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte index aeb2d1f248..303a385eb5 100644 --- a/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte +++ b/frontend/src/routes/(root)/(logged)/workspace_settings/+page.svelte @@ -29,11 +29,8 @@ isCriticalAlertsUIOpen } from '$lib/stores' import { sendUserToast } from '$lib/toast' - import { emptyString } from '$lib/utils' - import { - RotateCw, - Save - } from 'lucide-svelte' + import { clone, emptyString } from '$lib/utils' + import { RotateCw, Save } from 'lucide-svelte' import PremiumInfo from '$lib/components/settings/PremiumInfo.svelte' import Toggle from '$lib/components/Toggle.svelte' @@ -52,6 +49,10 @@ import StorageSettings from '$lib/components/workspaceSettings/StorageSettings.svelte' import GitSyncSection from '$lib/components/git_sync/GitSyncSection.svelte' import { untrack } from 'svelte' + import DucklakeSettings, { + convertDucklakeSettingsFromBackend, + type DucklakeSettingsType + } from '$lib/components/workspaceSettings/DucklakeSettings.svelte' let slackInitialPath: string = $state('') let slackScriptPath: string = $state('') @@ -84,6 +85,10 @@ secondaryStorage: undefined }) + let ducklakeSettings: DucklakeSettingsType = $state({ + ducklakes: [] + }) + let ducklakeSavedSettings: DucklakeSettingsType = $state(untrack(() => ducklakeSettings)) let workspaceDefaultAppPath: string | undefined = $state(undefined) let workspaceEncryptionKey: string | undefined = $state(undefined) @@ -103,15 +108,8 @@ ) let usingOpenaiClientCredentialsOauth = $state(false) - - let loadedSettings = $state(false) - - - - - async function editWorkspaceCommand(platform: 'slack' | 'teams'): Promise { if (platform === 'slack') { if (slackInitialPath === slackScriptPath) return @@ -166,7 +164,6 @@ } } - async function editWorkspaceDefaultApp(appPath: string | undefined): Promise { if (emptyString(appPath)) { await WorkspaceService.editWorkspaceDefaultApp({ @@ -268,6 +265,8 @@ workspaceDefaultAppPath = settings.default_app s3ResourceSettings = convertBackendSettingsToFrontendSettings(settings.large_file_storage) + ducklakeSettings = convertDucklakeSettingsFromBackend(settings.ducklake) + ducklakeSavedSettings = clone(ducklakeSettings) if (settings.deploy_ui != undefined && settings.deploy_ui != null) { deployUiSettings = { @@ -339,10 +338,6 @@ } } - - - - async function editCriticalAlertMuteSetting() { await SettingService.workspaceMuteCriticalAlertsUi({ workspace: $workspaceStore!, @@ -475,6 +470,14 @@ >
Object Storage (S3)
+ +
Ducklake
+ {:else if tab == 'windmill_lfs'} + {:else if tab == 'git_sync'} {#if $workspaceStore} @@ -918,7 +922,5 @@ {/if} - - diff --git a/openflow.openapi.yaml b/openflow.openapi.yaml index f1b35bc614..3cb5db84bd 100644 --- a/openflow.openapi.yaml +++ b/openflow.openapi.yaml @@ -285,6 +285,7 @@ components: enum: - s3object - resource + - ducklake access_type: type: string enum: [r, w, rw]