diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 3a8ff6af01..959cd8bf84 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -845,6 +845,29 @@ dependencies = [ "uuid", ] +[[package]] +name = "aws-sdk-config" +version = "1.68.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7de8bcff88ff2948bf2154204b7eba0ea30698c1c23770c4a917505fb9fcbd5" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 0.2.12", + "once_cell", + "regex-lite", + "tracing", +] + [[package]] name = "aws-sdk-sqs" version = "1.77.0" @@ -10711,6 +10734,7 @@ dependencies = [ "js-sys", "log", "mime", + "mime_guess", "native-tls", "percent-encoding", "pin-project-lite", @@ -15107,6 +15131,8 @@ name = "windmill" version = "1.538.0" dependencies = [ "anyhow", + "aws-sdk-config", + "aws-sigv4", "axum", "base64 0.22.1", "chrono", @@ -15169,10 +15195,12 @@ dependencies = [ "async-trait", "async_zip", "aws-config", + "aws-sdk-config", "aws-sdk-sqs", "aws-sdk-sso", "aws-sdk-ssooidc", "aws-sdk-sts", + "aws-sigv4", "axum", "backon", "base32", @@ -15333,6 +15361,7 @@ dependencies = [ "aws-smithy-types-convert", "axum", "backon", + "base64 0.22.1", "bytes", "chrono", "chrono-tz", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index a64e0d3ec9..661e932a96 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -146,6 +146,8 @@ pep440_rs.workspace = true systemstat.workspace = true size.workspace = true strum.workspace = true +aws-sigv4.workspace = true +aws-sdk-config.workspace = true kube.workspace = true k8s-openapi.workspace = true @@ -237,7 +239,7 @@ mail-send = { version = "0.4.0", features = ["builder"], default-features=false urlencoding = "^2" url = { version = "^2" , features = ["serde"]} async-oauth2 = "^0" -reqwest = { version = "^0.12", features = ["json", "stream", "gzip"] } +reqwest = { version = "^0.12", features = ["json", "stream", "gzip", "multipart"] } time = "^0" serde_urlencoded = "^0" tokio-tar = "^0" @@ -247,7 +249,9 @@ json-pointer = "^0" itertools = "^0" regex = "^1" semver = "^1" -duckdb = { version = "1.3.2", features = ["bundled"] } +duckdb = { version = "^1.3.2", features = ["bundled"] } +aws-sigv4 = "^1.3.4" +aws-sdk-config = "=1.68.0" async-trait = "0.1.88" v8 = "=130.0.7" # Exact version NOTE: Do not forget to update version and hash in flake.nix diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 5645873939..2ee878c68b 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -3547930948d370445984962a035c39a2eae7eba4 \ No newline at end of file +c409ac5e5bd202002648f18e8adb7cdf78b2a83e \ No newline at end of file diff --git a/backend/windmill-api/Cargo.toml b/backend/windmill-api/Cargo.toml index 39b3a4ed5f..5a776e53c7 100644 --- a/backend/windmill-api/Cargo.toml +++ b/backend/windmill-api/Cargo.toml @@ -145,6 +145,8 @@ aws-sdk-ssooidc = { workspace = true, optional = true } aws-sdk-sts = { workspace = true, optional = true } rustls = { workspace = true } +aws-sigv4.workspace = true +aws-sdk-config.workspace = true aws-config = { workspace = true, optional = true } async-trait.workspace = true google-cloud-pubsub = { workspace = true, optional = true } diff --git a/backend/windmill-api/src/job_helpers_oss.rs b/backend/windmill-api/src/job_helpers_oss.rs index 53d4a796a1..61c9f0606b 100644 --- a/backend/windmill-api/src/job_helpers_oss.rs +++ b/backend/windmill-api/src/job_helpers_oss.rs @@ -134,3 +134,36 @@ pub async fn download_s3_file_internal( "Not implemented in Windmill's Open Source repository".to_string(), )) } + +#[allow(dead_code)] +#[cfg(not(feature = "private"))] +pub async fn read_object_streamable( + _s3_client: Arc, + _file_key: &str, +) -> error::Result { + Err(error::Error::internal_err( + "Not implemented in Windmill's Open Source repository".to_string(), + )) +} + +#[allow(dead_code)] +#[cfg(not(feature = "private"))] +pub async fn delete_s3_file_internal( + _authed: &ApiAuthed, + _db: &DB, + _token: &str, + _w_id: &str, + _query: DeleteS3FileQuery, +) -> error::Result<()> { + Err(error::Error::internal_err( + "Not implemented in Windmill's Open Source repository".to_string(), + )) +} + +#[cfg(not(feature = "private"))] +#[derive(Deserialize)] +#[allow(dead_code)] +pub struct DeleteS3FileQuery { + pub file_key: String, + pub storage: Option, +} diff --git a/backend/windmill-api/src/lib.rs b/backend/windmill-api/src/lib.rs index f31764b026..2930c8d283 100644 --- a/backend/windmill-api/src/lib.rs +++ b/backend/windmill-api/src/lib.rs @@ -100,6 +100,10 @@ mod integration; mod live_migrations; #[cfg(feature = "postgres_trigger")] mod postgres_triggers; +#[cfg(all(feature = "private"))] +pub mod s3_proxy_ee; +mod s3_proxy_oss; + mod trigger_helpers; pub mod openapi; @@ -657,6 +661,10 @@ pub async fn run_server( "/w/:workspace_id/capture_u", capture::workspaced_unauthed_service().layer(cors.clone()), ) + .nest( + "/w/:workspace_id/s3_proxy", + s3_proxy_oss::workspaced_unauthed_service(), + ) .nest( "/auth", users::make_unauthed_service().layer(Extension(argon2)), diff --git a/backend/windmill-api/src/s3_proxy_oss.rs b/backend/windmill-api/src/s3_proxy_oss.rs new file mode 100644 index 0000000000..15108a40c3 --- /dev/null +++ b/backend/windmill-api/src/s3_proxy_oss.rs @@ -0,0 +1,11 @@ +#[cfg(feature = "private")] +#[allow(unused)] +pub use crate::s3_proxy_ee::*; + +#[cfg(not(feature = "private"))] +use axum::Router; + +#[cfg(not(feature = "private"))] +pub fn workspaced_unauthed_service() -> Router { + Router::new() +} diff --git a/backend/windmill-common/Cargo.toml b/backend/windmill-common/Cargo.toml index fbdad80b0a..d70c190b49 100644 --- a/backend/windmill-common/Cargo.toml +++ b/backend/windmill-common/Cargo.toml @@ -63,6 +63,7 @@ object_store = { workspace = true, optional = true } prometheus = { workspace = true, optional = true } aws-config = { workspace = true, optional = true } aws-sdk-sts = { workspace = true, optional = true } +base64.workspace = true aws-smithy-types-convert = { workspace = true, optional = true } indexmap.workspace = true diff --git a/backend/windmill-common/src/client.rs b/backend/windmill-common/src/client.rs index 6bc1d12ef9..9507100eb5 100644 --- a/backend/windmill-common/src/client.rs +++ b/backend/windmill-common/src/client.rs @@ -2,11 +2,7 @@ use anyhow::Context; use reqwest::{Body, Response}; use serde::de::DeserializeOwned; -use crate::{ - error::{self, to_anyhow}, - s3_helpers::{DuckdbConnectionSettingsQueryV2, DuckdbConnectionSettingsResponse}, - utils::HTTP_CLIENT, -}; +use crate::utils::HTTP_CLIENT; #[derive(Clone)] pub struct AuthedClient { @@ -210,44 +206,4 @@ impl AuthedClient { _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default()))?, } } - - pub async fn get_duckdb_connection_settings( - &self, - s3: &DuckdbConnectionSettingsQueryV2, - ) -> error::Result { - let url = format!( - "{}/api/w/{}/job_helpers/v2/duckdb_connection_settings", - self.base_internal_url, &self.workspace - ); - let response = self - .force_client - .as_ref() - .unwrap_or(&HTTP_CLIENT) - .post(url) - .header( - reqwest::header::CONTENT_TYPE, - reqwest::header::HeaderValue::from_static("application/json"), - ) - .header( - reqwest::header::ACCEPT, - reqwest::header::HeaderValue::from_static("application/json"), - ) - .header( - reqwest::header::AUTHORIZATION, - reqwest::header::HeaderValue::from_str(&format!("Bearer {}", self.token)) - .map_err(|e| error::Error::BadConfig(e.to_string()))?, - ) - .body(serde_json::to_string(&s3).map_err(to_anyhow)?) - .send() - .await - .context(format!("Sent get_duckdb_connection_settings request",)) - .map_err(error::Error::from)?; - match response.status().as_u16() { - 200u16 => Ok(response - .json::() - .await - .context("decoding duckdb_connection_settings response as json")?), - _ => Err(anyhow::anyhow!(response.text().await.unwrap_or_default()))?, - } - } } diff --git a/backend/windmill-common/src/error.rs b/backend/windmill-common/src/error.rs index 97eb6fbced..1c41bdc9b5 100644 --- a/backend/windmill-common/src/error.rs +++ b/backend/windmill-common/src/error.rs @@ -293,3 +293,42 @@ where Self(err.into()) } } +#[cfg(feature = "parquet")] +impl From for Error { + fn from(err: object_store::Error) -> Self { + use object_store::Error::*; + match err { + Generic { store, source } => Error::Generic( + StatusCode::INTERNAL_SERVER_ERROR, + format!("Generic {} error: {}", store, source), + ), + NotFound { path, source } => Error::NotFound(format!("{}: {}", path, source)), + InvalidPath { source } => Error::BadRequest(format!("Invalid path: {}", source)), + JoinError { source } => Error::InternalErr(format!("Join error: {}", source)), + NotSupported { source } => { + Error::BadRequest(format!("Operation not supported: {}", source)) + } + AlreadyExists { path, source } => { + Error::BadRequest(format!("Object at {} already exists: {}", path, source)) + } + Precondition { path, source } => { + Error::BadRequest(format!("Precondition failed at {}: {}", path, source)) + } + NotModified { path, source } => { + Error::ExecutionErr(format!("Not modified at {}: {}", path, source)) + } + NotImplemented => Error::BadRequest("Operation not yet implemented.".to_string()), + PermissionDenied { path, source } => { + Error::PermissionDenied(format!("Permission denied at {}: {}", path, source)) + } + Unauthenticated { path, source } => { + Error::NotAuthorized(format!("Unauthenticated for {}: {}", path, source)) + } + UnknownConfigurationKey { store, key } => Error::BadConfig(format!( + "Invalid config key '{}' for store '{}'", + key, store + )), + _ => Error::InternalErr(format!("Object store error: {}", err)), + } + } +} diff --git a/backend/windmill-common/src/jwt.rs b/backend/windmill-common/src/jwt.rs index 8ebdb8dacb..30e56cf155 100644 --- a/backend/windmill-common/src/jwt.rs +++ b/backend/windmill-common/src/jwt.rs @@ -1,5 +1,8 @@ use crate::error::{self, to_anyhow, Error}; +use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine}; +use hmac::{Hmac, Mac}; use serde::{de::DeserializeOwned, Serialize}; +use sha2::Sha256; use std::{collections::HashSet, sync::Arc}; use tokio::sync::RwLock; @@ -57,3 +60,18 @@ pub fn decode_without_verify(token: &str) -> anyhow::Result Ok(token_data.claims) } + +// header_and_payload: `{header}.{payload}` +pub async fn generate_signature(header_and_payload: &str) -> anyhow::Result { + let header_and_payload = header_and_payload.trim_start_matches("jwt_ext_"); + let header_and_payload = header_and_payload.trim_start_matches("jwt_"); + let secret = JWT_SECRET.read().await; + + // Create HMAC-SHA256 + let mut mac = Hmac::::new_from_slice(secret.as_bytes())?; + mac.update(header_and_payload.as_bytes()); + + // Finalize and encode + let result = mac.finalize().into_bytes(); + Ok(URL_SAFE_NO_PAD.encode(result)) +} diff --git a/backend/windmill-common/src/s3_helpers.rs b/backend/windmill-common/src/s3_helpers.rs index 5b6e113729..7ffcad3cc8 100644 --- a/backend/windmill-common/src/s3_helpers.rs +++ b/backend/windmill-common/src/s3_helpers.rs @@ -1,4 +1,5 @@ -use crate::error; +use crate::db::Authed; +use crate::error::{self}; #[cfg(feature = "parquet")] use aws_sdk_sts::config::ProvideCredentials; #[cfg(feature = "parquet")] @@ -236,6 +237,16 @@ impl LargeFileStorage { LargeFileStorage::GoogleCloudStorage(gcs_lfs) => &gcs_lfs.gcs_resource_path, } } + pub fn is_public_resource(&self) -> bool { + match self { + LargeFileStorage::S3Storage(lfs) => lfs.public_resource, + LargeFileStorage::S3AwsOidc(lfs) => lfs.public_resource, + LargeFileStorage::AzureBlobStorage(lfs) => lfs.public_resource, + LargeFileStorage::AzureWorkloadIdentity(lfs) => lfs.public_resource, + LargeFileStorage::GoogleCloudStorage(glfs) => glfs.public_resource, + } + .unwrap_or(false) + } } #[derive(Serialize, Deserialize, Debug)] @@ -323,6 +334,18 @@ pub struct AzureBlobResource { pub federated_token_file: Option, } +impl AzureBlobResource { + pub fn get_endpoint_url(&self) -> error::Result { + Ok(render_endpoint( + self.endpoint.clone().unwrap_or_else(|| "".to_string()), + self.use_ssl.unwrap_or(false), + None, + None, + "".to_string(), + )) + } +} + fn as_string<'de, D>(deserializer: D) -> Result where D: serde::de::Deserializer<'de>, @@ -1081,3 +1104,44 @@ pub fn duckdb_connection_settings_internal( }; return Ok(response); } + +impl ObjectStoreResource { + pub fn get_endpoint_url(&self) -> error::Result { + match self { + ObjectStoreResource::S3(s3_resource) => Ok(render_endpoint( + s3_resource.endpoint.clone(), + s3_resource.use_ssl, + s3_resource.port, + s3_resource.path_style, + s3_resource.bucket.clone(), + )), + ObjectStoreResource::Gcs(gcs_resource) => Ok(format!( + "https://storage.googleapis.com/{}", + gcs_resource.bucket + )), + ObjectStoreResource::Azure(az_resource) => az_resource.get_endpoint_url(), + } + } +} + +pub fn check_lfs_object_path_permissions( + lfs: &LargeFileStorage, + _object_path: &str, + authed: &Authed, +) -> error::Result<()> { + if authed.is_admin || lfs.is_public_resource() { + return Ok(()); + } + let _username = authed.username.as_str(); + + // TODO : Extend permission possibilities + + // if lfs.restrict_to_user_paths() { + // if !object_path.starts_with(&format!("u/{username}/")) { + // return Err(error::Error::NotAuthorized(format!( + // "Can only access paths u/{username}/**" + // ))); + // } + // } + return Ok(()); +} diff --git a/backend/windmill-common/src/workspaces.rs b/backend/windmill-common/src/workspaces.rs index c120e3c39a..70be40e28b 100644 --- a/backend/windmill-common/src/workspaces.rs +++ b/backend/windmill-common/src/workspaces.rs @@ -7,10 +7,6 @@ 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, }; @@ -164,7 +160,6 @@ 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( @@ -188,24 +183,6 @@ pub async fn get_ducklake_from_db_unchecked( let ducklake = serde_json::from_value::(ducklake)?; - let lfs = if let Some(storage) = &ducklake.storage.storage { - sqlx::query_scalar!("SELECT large_file_storage->'secondary_storage'->$2 FROM workspace_settings WHERE workspace_id = $1", w_id, storage) - } else { - sqlx::query_scalar!("SELECT large_file_storage FROM workspace_settings WHERE workspace_id = $1", w_id) - }.fetch_optional(db) - .await? - .flatten() - .map(serde_json::from_value::) - .ok_or_else(|| Error::ExecutionErr("Ducklake storage not found".to_string()))??; - - let s3_resource = transform_json_unchecked( - &serde_json::Value::String(lfs.get_s3_resource_path().to_string()), - w_id, - db, - ) - .await?; - let object_store_resource = lfs_to_object_store_resource(&lfs, s3_resource)?; - let catalog_resource = if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance { let pg_creds = parse_postgres_url(&get_database_url().await?)?; @@ -227,7 +204,6 @@ pub async fn get_ducklake_from_db_unchecked( }; let ducklake = DucklakeWithConnData { catalog_resource, - storage_settings: format_duckdb_connection_settings(object_store_resource)?, catalog: ducklake.catalog, storage: ducklake.storage, }; diff --git a/backend/windmill-worker/src/duckdb_executor.rs b/backend/windmill-worker/src/duckdb_executor.rs index 6e321b4e9d..062f4433e8 100644 --- a/backend/windmill-worker/src/duckdb_executor.rs +++ b/backend/windmill-worker/src/duckdb_executor.rs @@ -11,9 +11,7 @@ use serde_json::{json, Value}; 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::s3_helpers::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}; @@ -97,19 +95,18 @@ pub async fn do_duckdb( column_order_ref: &mut Option>, occupancy_metrics: &mut OccupancyMetrics, ) -> Result> { + let token = client.token.clone(); let hidden_passwords = Arc::new(Mutex::new(Vec::::new())); let result_f = async { let mut hidden_passwords = hidden_passwords.clone(); let mut bigquery_credentials = None; - let mut duckdb_connection_settings_cache = - HashMap::, DuckdbConnectionSettingsResponse>::new(); let sig = parse_duckdb_sig(query)?.args; let mut job_args = build_args_values(job, client, conn).await?; let (query, _) = &sanitize_and_interpolate_unsafe_sql_args(query, &sig, &job_args)?; - let query = transform_s3_uris(query, client, &mut duckdb_connection_settings_cache).await?; + let query = transform_s3_uris(query).await?; let job_args = { let mut m: HashMap = HashMap::new(); @@ -123,16 +120,11 @@ pub async fn do_duckdb( let s3_obj = serde_json::from_value::(json_value).map_err(|e| { Error::ExecutionErr(format!("Failed to deserialize S3Object: {}", e)) })?; - let duckdb_conn_settings: DuckdbConnectionSettingsResponse = - get_duckdb_connection_settings( - &s3_obj.storage, - &mut duckdb_connection_settings_cache, - client, - ) - .await?; - - let uri = - duckdb_conn_settings_to_s3_network_uri(&duckdb_conn_settings, &s3_obj.s3)?; + let uri = format!( + "s3://{}/{}", + s3_obj.storage.as_deref().unwrap_or("_default_"), + s3_obj.s3 + ); m.insert(sig_arg.name, duckdb::types::Value::Text(uri)); } else { let duckdb_value = json_value_to_duckdb_value( @@ -177,7 +169,6 @@ pub async fn do_duckdb( None => match transform_attach_ducklake( &query_block, conn, - &mut duckdb_connection_settings_cache, &mut hidden_passwords, &job.workspace_id, ) @@ -191,21 +182,48 @@ pub async fn do_duckdb( v }; + let base_internal_url = client.base_internal_url.clone(); + let w_id = job.workspace_id.clone(); + // 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 - 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 (s3_access_key, s3_secret_key) = token.split_at(token.rfind('.').unwrap_or(0)); + let s3_secret_key = &s3_secret_key[1..]; + let (s3_endpoint_ssl, s3_endpoint) = base_internal_url + .split_once("://") + .unwrap_or(("http", &base_internal_url)); + let s3_endpoint_ssl = match s3_endpoint_ssl { + "https" => true, + _ => false, + }; + + conn.execute_batch(&format!( + "INSTALL httpfs; LOAD httpfs; + INSTALL azure; LOAD azure; + CREATE OR REPLACE SECRET s3_secret ( + TYPE s3, + PROVIDER config, + KEY_ID '{s3_access_key}', + SECRET '{s3_secret_key}', + ENDPOINT '{s3_endpoint}/api/w/{w_id}/s3_proxy', + URL_STYLE path, + USE_SSL {s3_endpoint_ssl} + ); + CREATE OR REPLACE SECRET gcs_secret ( + TYPE gcs, + KEY_ID '{s3_access_key}', + SECRET '{s3_secret_key}', + ENDPOINT '{s3_endpoint}/api/w/{w_id}/s3_proxy', + USE_SSL {s3_endpoint_ssl} + ); + ", + )) + .map_err(|e| { + Error::ExecutionErr(format!("Error setting up S3 secret: {}", e.to_string())) + })?; let mut result: Option> = None; let mut column_order = None; @@ -581,7 +599,6 @@ async fn transform_attach_db_resource_query( async fn transform_attach_ducklake( query: &str, conn: &Connection, - duckdb_connection_settings_cache: &mut DuckDbConnectionSettingsCache, hidden_passwords: &mut Arc>>, w_id: &str, ) -> Result>> { @@ -616,17 +633,11 @@ async fn transform_attach_ducklake( } 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 storage = ducklake.storage.storage.as_deref().unwrap_or("_default_"); + let data_path = ducklake.storage.path; let attach_str = format!( - "ATTACH 'ducklake:{db_type}:{db_conn_str}' AS {alias_name} (DATA_PATH '{s3_network_uri}'{extra_args});", + "ATTACH 'ducklake:{db_type}:{db_conn_str}' AS {alias_name} (DATA_PATH 's3://{storage}/{data_path}'{extra_args});", ); let install_db_ext_str = get_attach_db_install_str(db_type)?; @@ -637,12 +648,7 @@ async fn transform_attach_ducklake( ])) } -// Replaces all s3 URIs in the windmill syntax with the actual S3 network URIs -async fn transform_s3_uris( - query: &str, - client: &AuthedClient, - duckdb_connection_settings_cache: &mut DuckDbConnectionSettingsCache, -) -> Result { +async fn transform_s3_uris(query: &str) -> Result { let mut transformed_query = None; lazy_static::lazy_static! { static ref RE: regex::Regex = regex::Regex::new(r"'s3://([^'/]*)/([^']*)'").unwrap(); @@ -650,46 +656,24 @@ async fn transform_s3_uris( 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(); - let storage = match storage.map(|m| m.as_str()) { - Some("") | None => None, - Some(s) => Some(s.to_string()), - }; - let original_str_lit = - format!("'s3://{}/{}'", storage.as_deref().unwrap_or(""), s3_path); - 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}'"); + let mut storage = storage.map(|m| m.as_str()).unwrap_or(""); + if !storage.is_empty() { + continue; + } + let original_str_lit: String = format!("'s3://{}/{}'", storage, s3_path); + storage = "_default_"; + + let new_s3_lit = format!("'s3://{}/{}'", storage, s3_path); transformed_query = Some( transformed_query - .unwrap_or(query.to_string()) - .replace(&original_str_lit, &url), + .unwrap_or_else(|| query.to_string()) + .replace(&original_str_lit, &new_s3_lit), ); } } Ok(transformed_query.unwrap_or(query.to_string())) } -pub fn duckdb_conn_settings_to_s3_network_uri( - s: &DuckdbConnectionSettingsResponse, - s3_path: &str, -) -> Result { - match &s { - DuckdbConnectionSettingsResponse { s3_bucket: Some(bucket), .. } => { - Ok(format!("s3://{bucket}/{s3_path}")) - } - DuckdbConnectionSettingsResponse { azure_container_path: Some(base), .. } => { - Ok(format!("{base}/{s3_path}")) - } - _ => { - Err(Error::ExecutionErr( - "DuckDB connection settings response must have either s3_bucket or azure_container_path".to_string(), - )) - } - } -} - // BigQuery extension requires a json file as credentials // The file path is set as an env var by do_duckdb // It is created by transform_attach_db_resource_query (when bigquery is detected) @@ -766,25 +750,6 @@ fn remove_comments(stmt: &str) -> &str { return &stmt[start.unwrap_or(0)..end]; } -async fn get_duckdb_connection_settings( - storage: &Option, - cache: &mut DuckDbConnectionSettingsCache, - client: &AuthedClient, -) -> Result { - if let Some(settings) = cache.get(storage) { - return Ok(settings.clone()); - } else { - let settings = client - .get_duckdb_connection_settings(&DuckdbConnectionSettingsQueryV2 { - s3_resource_path: None, - storage: storage.clone(), - }) - .await?; - cache.insert(storage.clone(), settings.clone()); - return Ok(settings); - } -} - #[cfg(test)] mod tests { use super::*; @@ -819,5 +784,3 @@ mod tests { assert_eq!(remove_comments(sql), "SELECT\n\n * FROM\n table\n;"); } } - -type DuckDbConnectionSettingsCache = HashMap, DuckdbConnectionSettingsResponse>; diff --git a/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte b/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte index 70814cbc13..1ca55c4bef 100644 --- a/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte +++ b/frontend/src/lib/components/workspaceSettings/StorageSettings.svelte @@ -1,7 +1,7 @@