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 <hugo@casademont.ch>
Co-authored-by: Ruben Fiszel <ruben@windmill.dev>
This commit is contained in:
Diego Imbert
2025-08-06 15:55:36 +02:00
committed by GitHub
parent 24febaead3
commit e6f1211d31
71 changed files with 2642 additions and 576 deletions
@@ -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"
}
@@ -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"
@@ -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"
}
@@ -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"
}
@@ -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"
}
@@ -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"
@@ -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"
}
@@ -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"
}
@@ -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"
}
@@ -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"
}
@@ -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"
}
+2
View File
@@ -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",
+1 -1
View File
@@ -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"
@@ -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;
@@ -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;
@@ -0,0 +1,4 @@
ALTER TYPE asset_kind ADD VALUE 'ducklake';
ALTER TABLE workspace_settings
ADD COLUMN ducklake JSONB;
@@ -0,0 +1,2 @@
DELETE FROM global_settings
WHERE name = 'ducklake_user_pg_pwd';
@@ -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
$$;
@@ -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)?;
@@ -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
}
@@ -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))
}
+1
View File
@@ -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
+126 -4
View File
@@ -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:
+98 -3
View File
@@ -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<DB>,
Json(database_names): Json<Vec<String>>,
) -> JsonResult<Vec<String>> {
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<DB>,
Path(dbname): Path<String>,
) -> 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(())
}
+3
View File
@@ -452,3 +452,6 @@ pub async fn update_rw_lock<T>(lock: std::sync::Arc<tokio::sync::RwLock<T>>, val
let mut w = lock.write().await;
*w = value;
}
lazy_static::lazy_static! {
static ref DUCKLAKE_INSTANCE_PG_PASSWORD: std::sync::RwLock<Option<String>> = std::sync::RwLock::new(None);
}
+166 -21
View File
@@ -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<serde_json::Value>, // effectively: DatasetsStorage
#[serde(skip_serializing_if = "Option::is_none")]
pub ducklake: Option<serde_json::Value>,
#[serde(skip_serializing_if = "Option::is_none")]
pub git_sync: Option<serde_json::Value>, // effectively: WorkspaceGitSyncSettings
#[serde(skip_serializing_if = "Option::is_none")]
pub deploy_ui: Option<serde_json::Value>, // effectively: WorkspaceDeploymentUISettings
@@ -296,11 +304,21 @@ struct LargeFileStorageWithSecondary {
secondary_storage: HashMap<String, LargeFileStorage>,
}
#[derive(Deserialize, Serialize, Debug)]
pub struct DucklakeSettings {
pub ducklakes: HashMap<String, Ducklake>,
}
#[derive(Deserialize, Debug)]
struct EditLargeFileStorageConfig {
large_file_storage: Option<LargeFileStorageWithSecondary>,
}
#[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<DB>,
Path(w_id): Path<String>,
) -> JsonResult<Vec<String>> {
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<DB>,
Path(w_id): Path<String>,
ApiAuthed { is_admin, username, .. }: ApiAuthed,
Json(new_config): Json<EditDucklakeConfig>,
) -> Result<String> {
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/<folder>/<name> or $res:u/<username>/<name>
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)]
+1
View File
@@ -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"
+1
View File
@@ -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)]
+37
View File
@@ -275,6 +275,43 @@ async fn reset() -> () {
todo!()
}
pub struct PostgresUrlComponents {
pub scheme: String,
pub username: Option<String>,
pub password: Option<String>,
pub host: String,
pub port: Option<u16>,
pub database: String,
pub ssl_mode: Option<String>,
}
pub fn parse_postgres_url(url: &str) -> Result<PostgresUrlComponents, Error> {
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<String, Error> {
use std::env::var;
use tokio::fs::File;
+98 -2
View File
@@ -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<E: Into<anyhow::Error>>(
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<String>,
}
pub fn lfs_to_object_store_resource(
lfs: &LargeFileStorage,
resource_value: serde_json::Value,
) -> error::Result<ObjectStoreResource> {
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<DuckdbConnectionSettingsResponse> {
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<DuckdbConnectionSettingsResponse> {
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);
}
+45
View File
@@ -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<bool>
}
}
// 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<String> {
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");
}
}
+187
View File
@@ -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<String>,
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<DucklakeWithConnData> {
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>(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::<LargeFileStorage>)
.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<String> {
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<serde_json::Value> {
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)
}
+17 -1
View File
@@ -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<Uuid> {
@@ -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<DucklakeWithConnData> {
client
.get(&format!(
"/api/w/{}/agent_workers/get_ducklake/{}",
w_id, &name
))
.await
}
pub const UPDATE_PING_URL: &str = "/api/agent_workers/update_ping";
+349 -178
View File
@@ -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<Vec<String>>,
occupancy_metrics: &mut OccupancyMetrics,
) -> Result<Box<RawValue>> {
let hidden_passwords = Arc::new(Mutex::new(Vec::<String>::new()));
let result_f = async {
let mut hidden_passwords = hidden_passwords.clone();
let mut bigquery_credentials = None;
let mut duckdb_connection_settings_cache =
HashMap::<Option<String>, 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<String, duckdb::types::Value> = HashMap::new();
@@ -119,28 +123,17 @@ pub async fn do_duckdb(
let s3_obj = serde_json::from_value::<S3Object>(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<Box<RawValue>> = 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<Box<RawValue>> {
@@ -398,7 +429,8 @@ fn string_to_duckdb_timestamp(s: &str) -> Result<duckdb::types::Value> {
fn string_to_duckdb_date(s: &str) -> Result<duckdb::types::Value> {
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<ParsedAttachDbResource<'a>> {
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<ParsedAttachDbResource
None
}
async fn transform_attach_db_resource_query(
parsed: &ParsedAttachDbResource<'_>,
job_id: &Uuid,
client: &AuthedClient,
) -> Result<Vec<String>> {
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<String> {
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<Mutex<Vec<String>>>,
) -> Result<Vec<String>> {
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<Mutex<Vec<String>>>,
w_id: &str,
) -> Result<Option<Vec<String>>> {
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<String>,
HashMap<Option<String>, DuckdbConnectionSettingsResponse>,
)> {
duckdb_connection_settings_cache: &mut DuckDbConnectionSettingsCache,
) -> Result<String> {
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<String> {
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<Self> {
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<String>,
cache: &mut DuckDbConnectionSettingsCache,
client: &AuthedClient,
) -> Result<DuckdbConnectionSettingsResponse> {
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<Option<String>, DuckdbConnectionSettingsResponse>;
+8 -8
View File
@@ -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",
+2 -2
View File
@@ -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",
+163 -160
View File
@@ -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<string, TableMetadata> = {}
let cachedLastRefreshCount = 0
async function getColDefs(tableKey: string) {
if (!resourcePath || !resourceType) return []
async function getColDefs(tableKey: string): Promise<TableMetadata> {
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]}
<DrawerContent
title={replResultData ? 'Query Result' : 'Database Manager'}
on:close={() => {
if (replResultData) {
replResultData = undefined
} else {
closeDrawer()
}
}}
CloseIcon={replResultData ? ArrowLeft : undefined}
noPadding
>
{#if dbSchema && $workspaceStore && resourceType && resourcePath}
<Splitpanes horizontal>
<Pane class="relative">
<!-- svelte-ignore a11y_click_events_have_key_events -->
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class={'absolute inset-0 z-10 p-8 ' +
(replResultData
? 'bg-surface/90'
: 'transition-colors bg-transparent pointer-events-none select-none')}
onclick={(e) => {
// 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}
<SimpleAgTable data={replResultData} class="animate-zoom-in" />
{/key}
{/if}
</div>
<DbManager
dbSupportsSchemas={dbSupportsSchemas(resourceType)}
{dbSchema}
{getColDefs}
dbTableOpsFactory={({ colDefs, tableKey }) =>
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()
}
<DrawerContent
title={replResultData ? 'Query Result' : 'Database Manager'}
on:close={() => {
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'}
<Splitpanes horizontal>
<Pane class="relative">
<!-- svelte-ignore a11y_click_events_have_key_events -->
<!-- svelte-ignore a11y_no_static_element_interactions -->
<div
class={'absolute inset-0 z-10 p-8 ' +
(replResultData
? 'bg-surface/90'
: 'transition-colors bg-transparent pointer-events-none select-none')}
onclick={(e) => {
// 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}
<SimpleAgTable data={replResultData} class="animate-zoom-in" />
{/key}
{/if}
</div>
<DbManager
dbSupportsSchemas={input.type == 'database' && dbSupportsSchemas(input.resourceType)}
{dbSchema}
{getColDefs}
dbTableOpsFactory={({ colDefs, tableKey }) =>
dbTableOpsWithPreviewScripts({
colDefs,
tableKey,
input: _input,
workspace: $workspaceStore
})}
/>
</Pane>
<Pane bind:size={replPanelSize} minSize={REPL_MIN_SIZE} class="relative">
<SqlRepl
{resourcePath}
{resourceType}
onData={(data) => {
replResultData = data
}}
placeholderTableName={sortArray(
Object.keys(
dbSchema?.schema[
'public' in dbSchema?.schema
? 'public'
: 'dbo' in dbSchema?.schema
? 'dbo'
: Object.keys(dbSchema?.schema)?.[0]
]
)
)?.[0]}
/>
</Pane>
</Splitpanes>
{:else}
<Splitpanes>
<Pane class="relative flex justify-center items-center">
<Loader2 class="animate-spin" size={32} />
</Pane>
</Splitpanes>
{/if}
{#snippet actions()}
<Button
loading={refreshing}
on:click={() => refresh()}
startIcon={{ icon: RefreshCcw }}
size="xs"
color="light"
>
Refresh
</Button>
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()
}
})}
/>
</Pane>
<Pane bind:size={replPanelSize} minSize={REPL_MIN_SIZE} class="relative">
<SqlRepl
{input}
onData={(data) => {
replResultData = data
}}
placeholderTableName={sortArray(
Object.keys(
dbSchema?.schema[
'public' in dbSchema?.schema
? 'public'
: 'dbo' in dbSchema?.schema
? 'dbo'
: Object.keys(dbSchema?.schema)?.[0]
]
)
)?.[0]}
/>
</Pane>
</Splitpanes>
{:else}
<Splitpanes>
<Pane class="relative flex justify-center items-center">
<Loader2 class="animate-spin" size={32} />
</Pane>
</Splitpanes>
{/if}
{#snippet actions()}
<Button
loading={refreshing}
on:click={() => refresh()}
startIcon={{ icon: RefreshCcw }}
size="xs"
color="light"
>
Refresh
</Button>
<Button
on:click={() => (expand = !expand)}
startIcon={{ icon: expand ? Minimize : Expand }}
size="xs"
color="light"
/>
{/snippet}
</DrawerContent>
{/key}
<Button
on:click={() => (expand = !expand)}
startIcon={{ icon: expand ? Minimize : Expand }}
size="xs"
color="light"
/>
{/snippet}
</DrawerContent>
</Drawer>
+1 -1
View File
@@ -141,7 +141,7 @@
{#if dbTableOps.onInsert}
<InsertRowDrawerButton
columnDefs={dbTableOps.colDefs ?? []}
dbType={dbTableOps.resourceType}
dbType={dbTableOps.dbType}
onInsert={(values) => {
if (!$workspaceStore) return
dbTableOps.onInsert?.({ values }).then((result) => {
@@ -40,7 +40,7 @@
export type DBTableEditorProps = {
onConfirm: (values: CreateTableValues) => void | Promise<void>
previewSql?: (values: CreateTableValues) => string
resourceType: DbType
dbType: DbType
dbSchema?: DBSchema
currentSchema?: string
}
@@ -74,19 +74,19 @@
import Select from './select/Select.svelte'
import { safeSelectItems } from './select/utils.svelte'
const { onConfirm, resourceType, previewSql, dbSchema, currentSchema }: DBTableEditorProps =
$props()
const { onConfirm, dbType, previewSql, dbSchema, currentSchema }: DBTableEditorProps = $props()
const columnTypes = DB_TYPES[resourceType]
const columnTypes = DB_TYPES[dbType]
const defaultColumnType = (
{
postgresql: 'VARCHAR',
snowflake: 'varchar',
ms_sql_server: 'varchar',
bigquery: 'string',
mysql: 'varchar'
mysql: 'varchar',
duckdb: 'string'
} satisfies Record<DbType, string>
)[resourceType]
)[dbType]
const values: CreateTableValues = $state({
name: '',
@@ -104,7 +104,7 @@
...(primaryKey && { primaryKey })
})
}
addColumn({ name: 'id', primaryKey: true })
addColumn({ name: 'id', primaryKey: dbType !== 'duckdb' })
const errors: ReturnType<typeof validate> = $derived(validate(values, dbSchema))
@@ -258,8 +258,7 @@
items={getFlatTableNamesFromSchema(dbSchema).map((o) => ({
value: o,
label:
(currentSchema && o.startsWith(currentSchema)) ||
!dbSupportsSchemas(resourceType)
(currentSchema && o.startsWith(currentSchema)) || !dbSupportsSchemas(dbType)
? o.split('.')[1]
: o
}))}
+82 -3
View File
@@ -5,7 +5,7 @@
<script lang="ts">
import { run } from 'svelte/legacy'
import { ResourceService, VariableService, type Script } from '$lib/gen'
import { ResourceService, VariableService, WorkspaceService, type Script } from '$lib/gen'
import { workspaceStore } from '$lib/stores'
import { base } from '$lib/base'
@@ -27,6 +27,7 @@
import Toggle from './Toggle.svelte'
import {
DatabaseIcon,
DiffIcon,
DollarSign,
File,
@@ -50,6 +51,7 @@
import type { EditorBarUi } from './custom_ui'
import EditorSettings from './EditorSettings.svelte'
import S3FilePicker from './S3FilePicker.svelte'
import DucklakeIcon from './icons/DucklakeIcon.svelte'
interface Props {
lang: SupportedLanguage | 'bunnative' | undefined
@@ -112,6 +114,8 @@
let variableEditor: VariableEditor | undefined = $state()
let resourceEditor: ResourceEditorDrawer | undefined = $state()
let s3FilePicker: S3FilePicker | undefined = $state()
let ducklakePicker: ItemPicker | undefined = $state()
let databasePicker: ItemPicker | undefined = $state()
let showContextVarPicker = $derived(
[
@@ -165,8 +169,7 @@
'rust',
'csharp',
'nu',
'java',
'duckdb'
'java'
// for related places search: ADD_NEW_LANG
].includes(lang ?? '')
)
@@ -175,6 +178,8 @@
['duckdb', 'python3'].includes(lang ?? '') ||
['typescript', 'javascript'].includes(scriptLangToEditorLang(lang))
)
let showDucklakePicker = $derived(['duckdb'].includes(lang ?? ''))
let showDatabasePicker = $derived(['duckdb'].includes(lang ?? ''))
let showResourceTypePicker = $derived(
['typescript', 'javascript'].includes(scriptLangToEditorLang(lang)) ||
@@ -644,6 +649,48 @@ JsonNode ${windmillPathToCamelCaseName(path)} = JsonNode.Parse(await client.GetS
<ResourceEditorDrawer bind:this={resourceEditor} on:refresh={resourcePicker.openDrawer} />
<VariableEditor bind:this={variableEditor} on:create={variablePicker.openDrawer} />
{#if showDucklakePicker}
<ItemPicker
bind:this={ducklakePicker}
pickCallback={async (_, name) => {
const connStr = name == 'main' ? 'ducklake' : `ducklake://${name}`
editor?.insertAtCursor(`ATTACH '${connStr}' AS dl; USE dl;\n`)
}}
tooltip="Attach a Ducklake in your DuckDB script. Ducklake allows you to manipulate large data on S3 blob files through a traditional SQL interface."
documentationLink="https://www.windmill.dev/docs/core_concepts/ducklake"
itemName="ducklake"
loadItems={async () =>
(await WorkspaceService.listDucklakes({ workspace: $workspaceStore ?? 'NO_W' })).map(
(path) => ({ path })
)}
/>
{/if}
{#if showDatabasePicker}
<ItemPicker
bind:this={databasePicker}
pickCallback={(path, _, resType) => {
if (!editor) return
if (lang == 'duckdb') {
let t = { postgresql: 'postgres', mysql: 'mysql', bigquery: 'bigquery' }[resType]
editor.insertAtCursor(`ATTACH 'res://${path}' AS db (TYPE ${t});`)
}
sendUserToast(`${path} inserted at cursor`)
}}
tooltip="Attach a database resource in your script. This allows you to query data from the database using SQL."
documentationLink="https://www.windmill.dev/docs/core_concepts/resources_and_types"
itemName="Database"
buttons={{ 'Edit/View': (x) => resourceEditor?.initEdit(x) }}
extraField="description"
extraField2="resource_type"
loadItems={async () =>
await ResourceService.listResource({
workspace: $workspaceStore ?? 'NO_W',
resourceType: 'postgresql,mysql,bigquery'
})}
></ItemPicker>
{/if}
<S3FilePicker
bind:this={s3FilePicker}
readOnlyMode={false}
@@ -757,6 +804,38 @@ JsonNode ${windmillPathToCamelCaseName(path)} = JsonNode.Parse(await client.GetS
</Button>
{/if}
{#if showDatabasePicker && customUi?.database != false}
<Button
aiId="editor-bar-add-database"
aiDescription="Add database"
title="Add database"
color="light"
on:click={() => databasePicker?.openDrawer()}
size="xs"
btnClasses="!font-medium text-tertiary"
spacingSize="md"
startIcon={{ icon: DatabaseIcon }}
{iconOnly}
>+Database
</Button>
{/if}
{#if showDucklakePicker && customUi?.ducklake != false}
<Button
aiId="editor-bar-use-ducklake"
aiDescription="Use Ducklake"
title="Use Ducklake"
color="light"
on:click={() => ducklakePicker?.openDrawer()}
size="xs"
btnClasses="!font-medium text-tertiary"
spacingSize="md"
startIcon={{ icon: DucklakeIcon }}
{iconOnly}
>+Ducklake
</Button>
{/if}
{#if customUi?.reset != false}
<Button
aiId="editor-bar-reset-content"
@@ -4,6 +4,7 @@
_resourceMetadata?: { resource_type?: string }
): boolean {
return (
asset.kind === 'ducklake' ||
asset.kind === 's3object' ||
(asset.kind === 'resource' && isDbType(_resourceMetadata?.resource_type))
)
@@ -19,6 +20,7 @@
import { userStore } from '$lib/stores'
import { isS3Uri } from '$lib/utils'
import { Database, File } from 'lucide-svelte'
import DucklakeIcon from './icons/DucklakeIcon.svelte'
const {
asset,
@@ -29,7 +31,8 @@
class: className = '',
noText = false,
buttonVariant = 'border',
btnClasses = ''
btnClasses = '',
disabled = false
}: {
asset: Asset
_resourceMetadata?: { resource_type?: string }
@@ -40,12 +43,13 @@
noText?: boolean
buttonVariant?: ButtonType.Variant
btnClasses?: string
disabled?: boolean
} = $props()
const assetUri = $derived(formatAsset(asset))
</script>
<Button
disabled={$userStore?.operator}
disabled={$userStore?.operator || disabled}
size="xs"
variant={buttonVariant}
spacingSize="xs2"
@@ -53,9 +57,15 @@
{btnClasses}
on:click={async () => {
if (asset.kind === 'resource' && isDbType(_resourceMetadata?.resource_type)) {
dbManagerDrawer?.openDrawer(_resourceMetadata.resource_type, asset.path)
dbManagerDrawer?.openDrawer({
type: 'database',
resourceType: _resourceMetadata.resource_type,
resourcePath: asset.path
})
} else if (asset.kind === 's3object' && isS3Uri(assetUri)) {
s3FilePicker?.open(assetUri)
} else if (asset.kind === 'ducklake') {
dbManagerDrawer?.openDrawer({ type: 'ducklake', ducklake: asset.path })
}
onClick?.()
}}
@@ -64,5 +74,7 @@
<span class:hidden={noText}>Explore</span> <File size={18} />
{:else if asset.kind === 'resource'}
<span class:hidden={noText}>Manage</span> <Database size={18} />
{:else if asset.kind === 'ducklake'}
<span class:hidden={noText}>Manage</span> <DucklakeIcon size={18} />
{/if}
</Button>
@@ -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<typeof ExploreAssetButton>;
export default ExploreAssetButton;
+12 -8
View File
@@ -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<string, any>[]) => 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 })
@@ -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
@@ -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
@@ -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)
}
@@ -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))
@@ -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')
}
@@ -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')
}
@@ -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')
@@ -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')
}
@@ -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<TableMetadata | undefined> {
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<Record<string, TableMetadata> | 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<string, TableMetadata> = {}
@@ -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<string> {
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')
@@ -163,7 +163,7 @@
</svelte:fragment>
</Popover>
<div class="flex flex-col flex-1">
<Tooltip class="select-none max-w-48 truncate" disablePopup={disableLiTooltip}>
<Tooltip class="select-none w-48 truncate" disablePopup={disableLiTooltip}>
{asset.path}
<svelte:fragment slot="text">
{asset.path}
@@ -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 ?? {})
@@ -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'
}
}
@@ -1,6 +1,6 @@
<script lang="ts">
import { classNames } from '$lib/utils'
import { createEventDispatcher } from 'svelte'
import { createEventDispatcher, type Snippet } from 'svelte'
import { fade } from 'svelte/transition'
import Button from '../button/Button.svelte'
import { AlertTriangle, CornerDownLeft, Loader2, RefreshCcw } from 'lucide-svelte'
@@ -14,6 +14,9 @@
open?: boolean
type?: 'danger' | 'reload'
showIcon?: boolean
children?: Snippet
onConfirmed?: () => void | Promise<void>
onCanceled?: () => void
}
const {
@@ -23,7 +26,10 @@
loading = false,
open = false,
type: _type,
showIcon = true
showIcon = true,
children,
onConfirmed,
onCanceled
}: Props = $props()
const type = $derived(_type ?? 'danger')
@@ -36,9 +42,11 @@
switch (event.key) {
case 'Enter':
dispatch('confirmed')
onConfirmed?.()
break
case 'Escape':
dispatch('canceled')
onCanceled?.()
break
}
}
@@ -107,15 +115,14 @@
{title}
</h3>
<div class="mt-2 text-sm text-secondary">
<!-- svelte-ignore slot_element_deprecated -->
<slot />
{@render children?.()}
</div>
</div>
</div>
<div class="flex items-center space-x-2 flex-row-reverse space-x-reverse mt-4">
<Button
disabled={loading}
on:click={() => dispatch('confirmed')}
on:click={() => (dispatch('confirmed'), onConfirmed?.())}
color={theme[type].color}
size="sm"
shortCut={{ Icon: CornerDownLeft, hide: !keyListen, withoutModifier: true }}
@@ -127,7 +134,7 @@
</Button>
<Button
disabled={loading}
on:click={() => dispatch('canceled')}
on:click={() => (dispatch('canceled'), onCanceled?.())}
color="light"
size="sm"
shortCut={{ key: 'Esc', hide: !keyListen, withoutModifier: true }}
@@ -0,0 +1,57 @@
import { createRawSnippet, type ComponentProps } from 'svelte'
import type ConfirmationModal from './ConfirmationModal.svelte'
import { sendUserToast } from '$lib/toast'
/**
* This allows asking for confirmation while maintaining a linear imperative flow,
* and avoiding unnecessary states and callback hopping
*
* @example
* let confirmationModal = createAsyncConfirmationModal()
* // ...
* let confirmed = await confirmationModal.ask({
* title: "The following objects do not exist",
* confirmationText: `Yes`,
* children: `Do you want to create the objects ?`
* })
* if (!confirmed) return
* // ...
* <ConfirmationModal {...confirmationModal.props} />
*/
export function createAsyncConfirmationModal(): {
props: ComponentProps<ConfirmationModal>
ask: (props: Params) => Promise<boolean>
} {
// Create a new instance of ConfirmationModal with the provided props
const o: ReturnType<typeof createAsyncConfirmationModal> = $state({
props: {
confirmationText: '',
title: ''
},
ask: (params) =>
new Promise<boolean>((resolve) => {
o.props = {
...params,
children: createRawSnippet(() => ({ render: () => params.children })),
open: true,
loading: false,
onCanceled: () => (resolve(false), (o.props.open = false)),
onConfirmed: async () => {
o.props.loading = true
try {
await params.onConfirmed?.()
o.props.loading = false
resolve(true)
o.props.open = false
} catch (e) {
sendUserToast('Error : ' + JSON.stringify(e), true)
o.props.loading = false
}
}
}
})
})
return o
}
type Params = Omit<ComponentProps<ConfirmationModal>, 'children'> & { children: string }
+2
View File
@@ -71,6 +71,8 @@ export type EditorBarUi = {
useVsCode?: boolean
diffMode?: boolean
s3object?: boolean
database?: boolean
ducklake?: boolean
}
export type EditableSchemaFormUi = {
+103 -50
View File
@@ -11,10 +11,19 @@ import { makeDeleteQuery } from './apps/components/display/dbtable/queries/delet
import { makeInsertQuery } from './apps/components/display/dbtable/queries/insert'
import { Trash2 } from 'lucide-svelte'
import { makeDeleteTableQuery } from './apps/components/display/dbtable/queries/deleteTable'
import type { DBSchema, SQLSchema } from '$lib/stores'
import { stringifySchema } from './copilot/lib'
export type DbInput =
| {
type: 'database'
resourceType: DbType
resourcePath: string
}
| { type: 'ducklake'; ducklake: string }
export type IDbTableOps = {
resourcePath: string
resourceType: DbType
dbType: DbType
tableKey: string
colDefs: ColumnDef[]
@@ -36,89 +45,73 @@ export type IDbTableOps = {
}
export function dbTableOpsWithPreviewScripts({
resourcePath,
resourceType,
input,
tableKey,
colDefs,
workspace
}: {
resourcePath: string
resourceType: DbType
input: DbInput
tableKey: string
colDefs: ColumnDef[]
workspace: string
}): IDbTableOps {
const dbType = getDbType(input)
const language = getLanguageByResourceType(dbType)
const dbArg = input?.type === 'database' ? { database: '$res:' + input.resourcePath } : {}
return {
resourcePath,
resourceType,
dbType,
tableKey,
colDefs,
getCount: async ({ quicksearch }) => {
const countQuery = makeCountQuery(resourceType, tableKey, undefined, colDefs)
let countQuery = makeCountQuery(dbType, tableKey, undefined, colDefs)
if (input.type === 'ducklake') countQuery = wrapDucklakeQuery(countQuery, input.ducklake)
const result = await runScriptAndPollResult({
workspace,
requestBody: {
args: { database: '$res:' + resourcePath, quicksearch },
language: getLanguageByResourceType(resourceType),
content: countQuery
}
requestBody: { args: { ...dbArg, quicksearch }, language, content: countQuery }
})
const count = result?.[0].count as number
return count
},
getRows: async (params) => {
const query = makeSelectQuery(tableKey, colDefs, undefined, resourceType as DbType)
let query = makeSelectQuery(tableKey, colDefs, undefined, dbType)
if (input.type === 'ducklake') query = wrapDucklakeQuery(query, input.ducklake)
let items = (await runScriptAndPollResult({
workspace,
requestBody: {
args: { database: '$res:' + resourcePath, ...params },
language: getLanguageByResourceType(resourceType),
content: query
}
requestBody: { args: { ...dbArg, ...params }, language, content: query }
})) as unknown[]
if (resourceType === 'ms_sql_server') items = items?.[0] as unknown[]
if (input.type === 'database' && input.resourceType === 'ms_sql_server')
items = items?.[0] as unknown[]
if (!items || !Array.isArray(items)) {
throw 'items is not an array'
}
return items
},
onUpdate: async ({ values }, colDef, newValue) => {
const updateQuery = makeUpdateQuery(tableKey, colDef, colDefs, resourceType)
let updateQuery = makeUpdateQuery(tableKey, colDef, colDefs, dbType)
if (input.type === 'ducklake') updateQuery = wrapDucklakeQuery(updateQuery, input.ducklake)
await runScriptAndPollResult({
workspace,
requestBody: {
args: {
database: '$res:' + resourcePath,
value_to_update: newValue,
...values
},
language: getLanguageByResourceType(resourceType),
args: { ...dbArg, value_to_update: newValue, ...values },
language,
content: updateQuery
}
})
},
onDelete: async ({ values }) => {
const deleteQuery = makeDeleteQuery(tableKey, colDefs, resourceType)
let deleteQuery = makeDeleteQuery(tableKey, colDefs, dbType)
if (input.type === 'ducklake') deleteQuery = wrapDucklakeQuery(deleteQuery, input.ducklake)
await runScriptAndPollResult({
workspace,
requestBody: {
args: { database: '$res:' + resourcePath, ...values },
language: getLanguageByResourceType(resourceType),
content: deleteQuery
}
requestBody: { args: { ...dbArg, ...values }, language, content: deleteQuery }
})
},
onInsert: async ({ values }) => {
const insertQuery = makeInsertQuery(tableKey, colDefs, resourceType)
runScriptAndPollResult({
let insertQuery = makeInsertQuery(tableKey, colDefs, dbType)
if (input.type === 'ducklake') insertQuery = wrapDucklakeQuery(insertQuery, input.ducklake)
await runScriptAndPollResult({
workspace,
requestBody: {
args: { database: '$res:' + resourcePath, ...values },
language: getLanguageByResourceType(resourceType),
content: insertQuery
}
requestBody: { args: { ...dbArg, ...values }, language, content: insertQuery }
})
}
}
@@ -140,13 +133,13 @@ export type DbTableActionFactory = (params: {
export function dbDeleteTableActionWithPreviewScript({
workspace,
resourcePath,
resourceType
input
}: {
workspace: string
resourcePath: string
resourceType: DbType
input: DbInput
}): DbTableActionFactory {
const dbArg = input?.type === 'database' ? { database: '$res:' + input.resourcePath } : {}
return ({ tableKey, refresh }) => ({
confirmTitle: `Are you sure you want to delete '${tableKey}' ? This action is irreversible`,
displayName: 'Delete',
@@ -154,12 +147,15 @@ export function dbDeleteTableActionWithPreviewScript({
icon: Trash2,
successText: `Table '${tableKey}' deleted successfully`,
action: async () => {
const deleteQuery = makeDeleteTableQuery(tableKey, resourceType)
const dbType = getDbType(input)
const language = getLanguageByResourceType(dbType)
let deleteQuery = makeDeleteTableQuery(tableKey, dbType)
if (input.type === 'ducklake') deleteQuery = wrapDucklakeQuery(deleteQuery, input.ducklake)
await runScriptAndPollResult({
workspace,
requestBody: {
args: { database: '$res:' + resourcePath },
language: getLanguageByResourceType(resourceType),
args: { ...dbArg },
language,
content: deleteQuery
}
})
@@ -167,3 +163,60 @@ export function dbDeleteTableActionWithPreviewScript({
}
})
}
export async function getDucklakeSchema({
workspace,
ducklake
}: {
workspace: string
ducklake: string
}): Promise<DBSchema> {
let result = await runScriptAndPollResult({
workspace,
requestBody: {
language: 'duckdb',
content: `ATTACH 'ducklake://${ducklake}' AS __ducklake__; ${DUCKLAKE_GET_SCHEMA_QUERY}`,
args: {}
}
})
const stringified = Array.isArray(result) && result.length && (result?.[0]?.['result'] ?? '[]')
if (!stringified) throw new Error('Failed to get Ducklake schema: ' + JSON.stringify(result))
let schema: Omit<SQLSchema, 'stringified'> = {
schema: { main: JSON.parse(stringified) },
publicOnly: true,
lang: 'ducklake'
}
return { ...schema, stringified: stringifySchema(schema) }
}
const DUCKLAKE_GET_SCHEMA_QUERY = `
SELECT json_group_object(table_name, table_data) AS result FROM (
SELECT
table_name,
json_group_object(
c.column_name,
json_object(
'type', c.data_type,
'default', c.column_default,
'required', c.is_nullable == 'NO'
)
) AS table_data
FROM information_schema.columns c
WHERE table_catalog = '__ducklake__' AND table_schema = current_schema()
GROUP BY c.table_name
)`
export function getDbType(input: DbInput): DbType {
switch (input.type) {
case 'database':
return input.resourceType
case 'ducklake':
return 'duckdb'
}
}
export function wrapDucklakeQuery(query: string, ducklake: string): string {
let attach = `ATTACH 'ducklake://${ducklake}' AS dl;USE dl;\n`
return query.replace(/^(--.*\n)*/, (match) => match + attach)
}
@@ -234,6 +234,7 @@
const cachedResourceMetadata = $derived(
flowGraphAssetsCtx?.val.resourceMetadataCache[data.asset.path]
)
const usageCount = $derived(flowGraphAssetsCtx?.val.computeAssetsCount?.(data.asset))
</script>
<NodeWrapper>
@@ -277,10 +278,9 @@
{/if}
</div>
<svelte:fragment slot="text">
Used in {pluralize(
flowGraphAssetsCtx?.val.computeAssetsCount?.(data.asset) ?? -1,
'step'
)}<br />
{#if usageCount !== undefined}
Used in {pluralize(usageCount, 'step')}<br />
{/if}
<a
href={undefined}
class={twMerge(
@@ -0,0 +1,35 @@
<script lang="ts">
interface Props {
height?: string
width?: string
fill?: string
class?: string
}
let { height = '24px', width = '24px', fill = 'black', class: className = '' }: Props = $props()
</script>
<svg
{width}
{height}
class={className}
viewBox="0 0 22 22"
fill="none"
xmlns="http://www.w3.org/2000/svg"
>
<path
d="M11.0586 0.306641C11.3106 0.315989 11.5582 0.378277 11.7822 0.495117C12.0247 0.621655 12.2331 0.805033 12.3897 1.0293L12.3916 1.03223L17.9033 9H16.2022L11.6963 2.4873V9H10.2959V2.4873L1.75196 14.8457C1.72867 14.8796 1.71294 14.9186 1.70508 14.959C1.69725 14.9995 1.69803 15.0418 1.70704 15.082C1.71605 15.1222 1.73351 15.1601 1.75782 15.1934C1.78213 15.2266 1.81309 15.2546 1.84864 15.2754H1.84766L8.75001 19.2158V20.8281L1.14844 16.4883L1.14258 16.4844C0.940872 16.3666 0.765826 16.2081 0.627936 16.0195C0.489969 15.8308 0.391887 15.6158 0.340827 15.3877C0.289772 15.1595 0.286682 14.9229 0.331061 14.6934C0.375447 14.4641 0.466395 14.2462 0.59864 14.0537L0.599616 14.0518L9.59962 1.03223L9.60157 1.0293C9.75823 0.804889 9.9673 0.62167 10.21 0.495117C10.4336 0.378564 10.6802 0.316161 10.9316 0.306641C10.9528 0.304686 10.9744 0.299805 10.9961 0.299805C11.0172 0.299841 11.038 0.304797 11.0586 0.306641Z"
{fill}
stroke="none"
/>
<path
d="M14.036 10C16.4558 10 18.4768 11.6945 18.9614 13.9535H17.2654C16.8219 12.6035 15.5438 11.6279 14.036 11.6279C12.1603 11.6279 10.6396 13.1376 10.6396 15C10.6396 16.8624 12.1603 18.3721 14.036 18.3721C15.7121 18.3721 17.1044 17.1667 17.3819 15.5814H19.038C18.7478 18.0689 16.6193 20 14.036 20C11.2547 20 9 17.7614 9 15C9 12.2386 11.2547 10 14.036 10Z"
{fill}
stroke="none"
/>
<path
d="M22 14.7674C22 16.0338 20.9049 16.9767 19.6618 16.9767L18.018 16.9767V15.5814L19.6618 15.5814C20.2251 15.5814 20.5946 15.1708 20.5946 14.7674C20.5946 14.3641 20.2251 13.9535 19.6618 13.9535L18.018 13.9535V12.5581L19.6618 12.5581C20.9049 12.5581 22 13.5011 22 14.7674Z"
{fill}
stroke="none"
/>
</svg>
@@ -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 @@
<AssetS3Icon {fill} width={size} height={size} class={className} />
{:else if assetKind == 'resource'}
<AssetResIcon {fill} width={size} height={size} class={className} />
{:else if assetKind == 'ducklake'}
<AssetDucklakeIcon {fill} width={size} height={size} class={className} />
{:else}
<Pyramid {size} color={fill} class={'!fill-none ' + className} />
{/if}
@@ -0,0 +1,22 @@
<script lang="ts">
interface Props {
size?: number
}
let { size = 24 }: Props = $props()
</script>
<svg
width={size}
height={size}
viewBox="0 0 13 13"
style="fill:currentcolor;"
xmlns="http://www.w3.org/2000/svg"
>
<path
d="M5.03604 1.5C7.45583 1.5 9.4768 3.19448 9.96136 5.45349H8.26545C7.82193 4.10348 6.54384 3.12791 5.03604 3.12791C3.16026 3.12791 1.63964 4.63764 1.63964 6.5C1.63964 8.36236 3.16026 9.87209 5.03604 9.87209C6.71211 9.87209 8.10437 8.66667 8.38188 7.0814H10.038C9.74784 9.56889 7.61927 11.5 5.03604 11.5C2.25471 11.5 0 9.26142 0 6.5C0 3.73858 2.25471 1.5 5.03604 1.5Z"
/>
<path
d="M13 6.26744C13 7.53379 11.9049 8.47674 10.6618 8.47674L9.01802 8.47674V7.0814L10.6618 7.0814C11.2251 7.0814 11.5946 6.67079 11.5946 6.26744C11.5946 5.86409 11.2251 5.45349 10.6618 5.45349L9.01802 5.45349V4.05814L10.6618 4.05814C11.9049 4.05814 13 5.00109 13 6.26744Z"
/>
</svg>
@@ -0,0 +1,341 @@
<script module lang="ts">
import { _ } from 'ag-grid-community'
export type DucklakeSettingsType = {
ducklakes: {
name: string
catalog: {
resource_type: 'postgresql' | 'mysql' | 'instance'
resource_path?: string // Name of the database when resource_type is instance
}
storage: {
storage?: string
path: string
}
}[]
}
export function convertDucklakeSettingsFromBackend(
settings: GetSettingsResponse['ducklake']
): DucklakeSettingsType {
const s: DucklakeSettingsType = { ducklakes: [] }
if (settings?.ducklakes) {
for (const [name, rest] of Object.entries(settings.ducklakes)) {
s.ducklakes.push({ name, ...rest })
}
}
return s
}
export function convertDucklakeSettingsToBackend(
settings: DucklakeSettingsType
): NonNullable<GetSettingsResponse['ducklake']> {
const s: GetSettingsResponse['ducklake'] = { ducklakes: {} }
for (const ducklake of settings.ducklakes) {
const catalog = ducklake.catalog
if (ducklake.name in s.ducklakes)
throw 'Settings contain duplicate ducklake name: ' + ducklake.name
if (!catalog.resource_path) throw 'No resource selected for ' + ducklake.name
if (catalog.resource_type === 'instance' && catalog.resource_path === 'windmill')
throw ducklake.name + ' catalog cannot be called "windmill"'
if (ducklake.storage.path.startsWith('/'))
ducklake.storage.path = ducklake.storage.path.slice(1)
s.ducklakes[ducklake.name] = {
catalog: ducklake.catalog,
storage: ducklake.storage
}
}
return s
}
</script>
<script>
import { Plus } from 'lucide-svelte'
import Button from '../common/button/Button.svelte'
import Description from '../Description.svelte'
import { random_adj } from '../random_positive_adjetive'
import { DataTable, Cell, Row } from '../table'
import Head from '../table/Head.svelte'
import CloseButton from '../common/CloseButton.svelte'
import Select from '../select/Select.svelte'
import ResourcePicker from '../ResourcePicker.svelte'
import { usePromise } from '$lib/svelte5Utils.svelte'
import { SettingService, WorkspaceService, type GetSettingsResponse } from '$lib/gen'
import { superadmin, workspaceStore } from '$lib/stores'
import { sendUserToast } from '$lib/toast'
import ExploreAssetButton from '../ExploreAssetButton.svelte'
import DbManagerDrawer from '../DBManagerDrawer.svelte'
import Tooltip from '../Tooltip.svelte'
import { isCloudHosted } from '$lib/cloud'
import ConfirmationModal from '../common/confirmationModal/ConfirmationModal.svelte'
import { createAsyncConfirmationModal } from '../common/confirmationModal/asyncConfirmationModal.svelte'
import { clone, pluralize } from '$lib/utils'
import Alert from '../common/alert/Alert.svelte'
import { deepEqual } from 'fast-equals'
import Popover from '../meltComponents/Popover.svelte'
const DEFAULT_DUCKLAKE_CATALOG_NAME = 'ducklake_catalog'
type Props = {
ducklakeSettings: DucklakeSettingsType
ducklakeSavedSettings: DucklakeSettingsType
}
let { ducklakeSettings = $bindable(), ducklakeSavedSettings = $bindable() }: Props = $props()
let isWmDbEnabled = $derived($superadmin && !isCloudHosted())
function onNewDucklake() {
const name = ducklakeSettings.ducklakes.some((d) => d.name === 'main')
? `${random_adj()}_ducklake`
: 'main'
ducklakeSettings.ducklakes.push({
name,
catalog: {
resource_type: isWmDbEnabled ? 'instance' : 'postgresql',
resource_path: isWmDbEnabled ? DEFAULT_DUCKLAKE_CATALOG_NAME : undefined
},
storage: {
storage: undefined,
path: ''
}
})
}
function removeDucklake(index: number) {
ducklakeSettings.ducklakes.splice(index, 1)
}
const windmillDbNames = $derived(
ducklakeSettings.ducklakes
.filter((d) => d.catalog.resource_type === 'instance')
.map((d) => d.catalog.resource_path ?? '')
)
const ducklakeIsDirty: Record<string, boolean> = $derived(
Object.fromEntries(
ducklakeSettings.ducklakes.map((d) => {
const saved = ducklakeSavedSettings.ducklakes.find((saved) => saved.name === d.name)
return [d.name, !deepEqual(saved, d)] as const
})
)
)
async function onSave() {
try {
if (windmillDbNames.length) {
// Ensure that all instance dbs exist
const nonExistentDbs = await SettingService.databasesExist({ requestBody: windmillDbNames })
if (nonExistentDbs.length) {
let confirmed = await confirmationModal.ask({
title: "The following databases do not exist in Windmill's Postgres instance",
confirmationText: `Create ${pluralize(nonExistentDbs.length, 'database')}`,
children: `<span>
Confirm running the following in the instance's Postgres :<br />
${nonExistentDbs.map((db) => `<pre class='border mt-1 p-2 rounded-md'>CREATE DATABASE "${db}";</pre>`).join('\n')}
</span>`
})
if (!confirmed) return
await Promise.all(
nonExistentDbs.map((name) => SettingService.createDucklakeDatabase({ name }))
)
}
}
const settings = convertDucklakeSettingsToBackend(ducklakeSettings)
await WorkspaceService.editDucklakeConfig({
workspace: $workspaceStore!,
requestBody: { settings }
})
ducklakeSavedSettings = clone(ducklakeSettings)
sendUserToast('Ducklake settings saved successfully')
} catch (e) {
sendUserToast(e, true)
console.error('Error saving ducklake settings', e)
}
}
let secondaryStorageNames = usePromise(
() => SettingService.getSecondaryStorageNames({ workspace: $workspaceStore! }),
{ loadInit: false }
)
$effect(() => {
$workspaceStore
secondaryStorageNames.refresh()
})
let tableHeadNames = ['Name', 'Catalog', 'Workspace storage', '', ''] as const
let tableHeadTooltips: Partial<Record<(typeof tableHeadNames)[number], string | undefined>> = {
Name: "Ducklakes are referenced in DuckDB scripts with the <code class='px-1 py-0.5 border rounded-md'>ATTACH 'ducklake://name' AS dl;</code> syntax",
Catalog: 'Ducklake needs an SQL database to store metadata about the data',
'Workspace storage':
'Where the data is actually stored, in parquet format. You need to configure a workspace storage first'
}
let dbManagerDrawer: DbManagerDrawer | undefined = $state()
let confirmationModal = createAsyncConfirmationModal()
</script>
<div class="flex flex-col gap-4 mb-8 mt-20">
<div class="flex flex-col gap-1">
<div class="text-primary text-lg font-semibold">Ducklake</div>
<Description link="https://www.windmill.dev/docs/core_concepts/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 !
</Description>
</div>
</div>
{#if ducklakeSettings.ducklakes.some((d) => d.catalog.resource_type === 'instance')}
<Alert title="Instance catalogs use the Windmill database" class="mb-4" type="info">
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.
</Alert>
{/if}
<DataTable>
<Head>
<tr>
{#each tableHeadNames as name, i}
<Cell head first={i == 0} last={i == tableHeadNames.length - 1}>
{name}
{#if tableHeadTooltips[name]}
<Tooltip>
{@html tableHeadTooltips[name]}
</Tooltip>
{/if}
</Cell>
{/each}
</tr>
</Head>
<tbody class="divide-y bg-surface">
{#if ducklakeSettings.ducklakes.length == 0}
<Row>
<Cell colspan={tableHeadNames.length} class="text-center">
No ducklake in this workspace yet
</Cell>
</Row>
{/if}
{#each ducklakeSettings.ducklakes as ducklake, ducklakeIndex}
<Row>
<Cell first class="w-48 relative">
{#if ducklake.name === 'main'}
<Tooltip wrapperClass="absolute mt-2.5 right-4" placement="bottom-start">
The <i>main</i> ducklake can be accessed with the
<br />
<code class="px-1 py-0.5 border rounded-md">ATTACH 'ducklake' AS dl;</code> shorthand
</Tooltip>
{/if}
<input bind:value={ducklake.name} placeholder="Name" />
</Cell>
<Cell>
<div class="flex gap-4">
<div class="relative">
{#if ducklake.catalog.resource_type === 'instance'}
<Tooltip wrapperClass="absolute mt-2.5 right-2 z-20" placement="bottom-start">
Use Windmill's PostgreSQL instance as a catalog
</Tooltip>
{/if}
<Select
items={[
{ value: 'postgresql', label: 'PostgreSQL' },
{ value: 'mysql', label: 'MySQL' },
...(isWmDbEnabled ? [{ value: 'instance', label: 'Instance' }] : [])
]}
bind:value={
() => ducklake.catalog.resource_type,
(resource_type) => {
ducklake.catalog = {
resource_type,
resource_path:
resource_type === 'instance' ? DEFAULT_DUCKLAKE_CATALOG_NAME : undefined
}
}
}
class="w-28"
/>
</div>
<div class="flex items-center gap-1 w-80">
{#if ducklake.catalog.resource_type !== 'instance'}
<ResourcePicker
bind:value={ducklake.catalog.resource_path}
resourceType={ducklake.catalog.resource_type}
/>
{:else}
<input
bind:value={ducklake.catalog.resource_path}
placeholder="PostgreSQL database name"
/>
{/if}
</div>
</div>
</Cell>
<Cell>
<div class="flex gap-4">
<Select
placeholder="Default storage"
items={[
{ value: undefined, label: 'Default storage' },
...(secondaryStorageNames.value?.map((value) => ({ value })) ?? [])
]}
bind:value={
() => ducklake.storage.storage,
(s) => {
if (s) ducklake.storage.storage = s
else delete ducklake.storage.storage
}
}
class="w-48"
inputClass="!placeholder-primary"
/>
<input placeholder="Data path (defaults to /)" bind:value={ducklake.storage.path} />
</div>
</Cell>
<Cell class="w-12">
{#if ducklakeIsDirty[ducklake.name]}
<Popover
openOnHover
contentClasses="p-2 text-sm text-secondary italic"
class="cursor-not-allowed"
>
<svelte:fragment slot="trigger">
<ExploreAssetButton
asset={{ kind: 'ducklake', path: ducklake.name }}
{dbManagerDrawer}
disabled
/>
</svelte:fragment>
<svelte:fragment slot="content">Please save settings first</svelte:fragment>
</Popover>
{:else}
<ExploreAssetButton
asset={{ kind: 'ducklake', path: ducklake.name }}
{dbManagerDrawer}
/>
{/if}
</Cell>
<Cell class="w-12">
<CloseButton small on:close={() => removeDucklake(ducklakeIndex)} />
</Cell>
</Row>
{/each}
<Row class="!border-0">
<Cell colspan={tableHeadNames.length} class="pt-0 pb-2">
<div class="flex justify-center">
<Button size="sm" btnClasses="max-w-fit" variant="border" on:click={onNewDucklake}>
<Plus /> New ducklake
</Button>
</div>
</Cell>
</Row>
</tbody>
</DataTable>
<Button
wrapperClasses="mt-4 mb-44 max-w-fit"
on:click={onSave}
disabled={Object.values(ducklakeIsDirty).every((v) => v === false)}>Save ducklake settings</Button
>
<DbManagerDrawer bind:this={dbManagerDrawer} />
<ConfirmationModal {...confirmationModal.props} />
+2 -1
View File
@@ -259,5 +259,6 @@ export const DB_TYPES: Record<DbType, string[]> = {
ms_sql_server: MSSQL_TYPES,
mysql: MYSQL_TYPES,
postgresql: POSTGRES_TYPES,
snowflake: SNOWFLAKE_TYPES
snowflake: SNOWFLAKE_TYPES,
duckdb: DUCKDB_TYPES
}
+2 -1
View File
@@ -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()
+24 -8
View File
@@ -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]) {
+1 -1
View File
@@ -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
+11 -1
View File
@@ -1530,6 +1530,16 @@ export function uniqueBy<T>(array: T[], key: (t: T) => any): T[] {
})
}
export function pruneNullishArrayWithSet<T>(array: (T | null | undefined)[]): T[] {
export function pruneNullishArray<T>(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)
}
}
@@ -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<void> {
if (platform === 'slack') {
if (slackInitialPath === slackScriptPath) return
@@ -166,7 +164,6 @@
}
}
async function editWorkspaceDefaultApp(appPath: string | undefined): Promise<void> {
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 @@
>
<div class="flex gap-2 items-center my-1"> Object Storage (S3)</div>
</Tab>
<Tab
size="xs"
value="ducklake"
aiId="workspace-settings-ducklake"
aiDescription="Ducklake workspace settings"
>
<div class="flex gap-2 items-center my-1">Ducklake</div>
</Tab>
<Tab
size="xs"
value="default_app"
@@ -816,6 +819,7 @@
/>
{:else if tab == 'windmill_lfs'}
<StorageSettings bind:s3ResourceSettings />
<DucklakeSettings bind:ducklakeSettings bind:ducklakeSavedSettings />
{:else if tab == 'git_sync'}
{#if $workspaceStore}
<GitSyncSection />
@@ -918,7 +922,5 @@
{/if}
</CenteredPage>
<style>
</style>
+1
View File
@@ -285,6 +285,7 @@ components:
enum:
- s3object
- resource
- ducklake
access_type:
type: string
enum: [r, w, rw]