grant usage on the table schema

docs/developers-guide/driver-changelog.md

@@ -8,6 +8,27 @@ title: Driver interface changelog
 
 - Added `sql-jdbc.execute/db-type-name` multimethod. Override this if something more than the default is needed in your sql-jdbc-based driver. See the `:mysql` implementation as an example.
 
+- Added driver feature `:isolation` for drivers that support workspace isolation. This feature lets workspaces
+  operate in isolated database environments with their own schemas and credentials.
+
+- Added driver multimethod `metabase.driver/init-workspace-isolation!` to initialize database isolation for a workspace.
+  Creates an isolated schema/database and user credentials with appropriate permissions. Returns a map with `:schema`
+  and `:database_details` keys.
+
+- Added driver multimethod `metabase.driver/destroy-workspace-isolation!` to destroy all database resources created
+  for workspace isolation, including schemas, users, and roles.
+
+- Added driver multimethod `metabase.driver/grant-workspace-read-access!` to grant read access on specified tables
+  to a workspace's isolated user. Takes a sequence of table maps with `:schema` and `:name` keys.
+
+- Added driver multimethod `metabase.driver/check-isolation-permissions` to check if a database connection has
+  sufficient permissions for workspace isolation. Performs actual isolation operations in a test workspace and
+  returns nil on success or an error message string on failure.
+
+- Added `metabase.driver/with-swapped-connection-details` macro for temporarily swapping connection details for a
+  specific database within a dynamic scope. This lets workspaces use alternate credentials without mutating
+  the database record.
+
 ## Metabase 0.58.0
 
 - Added a `:collate` feature for drivers that support collation settings on text fields

modules/drivers/bigquery-cloud-sdk/deps.edn

@@ -10,6 +10,13 @@
                                                         ;; netty is updated because of CVE
                                                         io.netty/netty-common
                                                         io.netty/netty-buffer]}
+  ;; Required for service account impersonation in workspace isolation
+  ;; Version matches what google-cloud-bigquery depends on
+  com.google.auth/google-auth-library-oauth2-http {:mvn/version "1.40.0"}
+  ;; Required for granting project-level IAM roles to workspace service accounts
+  com.google.cloud/google-cloud-resourcemanager {:mvn/version "1.82.0"}
+  ;; Required for creating service accounts for workspace isolation
+  com.google.cloud/google-iam-admin {:mvn/version "3.54.0"}
   ;; pin arrow formats to 17 for cve about parquet files
   org.apache.arrow/arrow-format          {:mvn/version "18.3.0"}
   org.apache.arrow/arrow-memory-core     {:mvn/version "18.3.0"}

modules/drivers/bigquery-cloud-sdk/src/metabase/driver/bigquery_cloud_sdk.clj

@@ -19,6 +19,7 @@
    [metabase.driver.sql.query-processor :as sql.qp]
    [metabase.driver.sql.util :as sql.u]
    [metabase.driver.sync :as driver.s]
+   [metabase.driver.util :as driver.u]
    [metabase.util :as u]
    [metabase.util.date-2 :as u.date]
    [metabase.util.i18n :refer [tru]]
@@ -30,16 +31,22 @@
   (:import
    (clojure.lang PersistentList)
    (com.google.api.gax.rpc FixedHeaderProvider)
+   (com.google.auth.oauth2 ImpersonatedCredentials ServiceAccountCredentials)
+   (com.google.cloud Identity Role)
    (com.google.cloud.bigquery
+    Acl Acl$Role Acl$User
     BigQuery
     BigQuery$DatasetListOption
+    BigQuery$DatasetOption
+    BigQuery$IAMOption
     BigQuery$JobOption
     BigQuery$QueryResultsOption
     BigQuery$TableDataListOption
     BigQuery$TableOption
     BigQueryException
-    BigQueryOptions
+    BigQueryOptions BigQueryOptions$Builder
     Dataset
+    DatasetId DatasetInfo
     Field
     Field$Mode
     FieldValue
@@ -53,8 +60,13 @@
     TableDefinition$Type
     TableId
     TableResult)
+   (com.google.cloud.iam.admin.v1 IAMClient IAMSettings)
+   (com.google.cloud.resourcemanager.v3 ProjectsClient ProjectsSettings)
    (com.google.common.collect ImmutableMap)
    (com.google.gson JsonParser)
+   (com.google.iam.admin.v1 CreateServiceAccountRequest DeleteServiceAccountRequest ServiceAccount)
+   (com.google.iam.v1 Binding Policy SetIamPolicyRequest GetIamPolicyRequest)
+   (java.io ByteArrayInputStream)
    (java.util Iterator)))
 
 (set! *warn-on-reflection* true)
@@ -76,15 +88,29 @@
 
 (mu/defn- database-details->client
   ^BigQuery [details :- :map]
-  (let [creds   (bigquery.common/database-details->service-account-credential details)
-        mb-version (:tag driver-api/mb-version-info)
-        run-mode   (name driver-api/run-mode)
-        user-agent (format "Metabase/%s (GPN:Metabase; %s)" mb-version run-mode)
+  (let [base-creds   (bigquery.common/database-details->service-account-credential details)
+        ;; Check if we should impersonate a different service account
+        ;; ImpersonatedCredentials automatically refreshes tokens before expiration,
+        ;; so the 1-hour lifetime is just the initial token validity period.
+        ;; Each query creates a fresh client, and the credentials handle refresh internally.
+        final-creds  (if-let [target-sa (:impersonate-service-account details)]
+                       (do
+                         (log/debugf "Creating impersonated credentials for service account: %s" target-sa)
+                         (ImpersonatedCredentials/create
+                          (.createScoped base-creds bigquery-scopes)
+                          target-sa
+                          nil  ;; delegates (not needed)
+                          (java.util.ArrayList. bigquery-scopes)
+                          3600))  ;; 1 hour token lifetime
+                       (.createScoped base-creds bigquery-scopes))
+        mb-version   (:tag driver-api/mb-version-info)
+        run-mode     (name driver-api/run-mode)
+        user-agent   (format "Metabase/%s (GPN:Metabase; %s)" mb-version run-mode)
         header-provider (FixedHeaderProvider/create
                          (ImmutableMap/of "user-agent" user-agent))
-        bq-bldr (doto (BigQueryOptions/newBuilder)
-                  (.setCredentials (.createScoped creds bigquery-scopes))
-                  (.setHeaderProvider header-provider))]
+        bq-bldr      (doto (BigQueryOptions/newBuilder)
+                       (.setCredentials final-creds)
+                       (.setHeaderProvider header-provider))]
     (when-let [host (not-empty (:host details))]
       (.setHost bq-bldr host))
     (.. bq-bldr build getService)))
@@ -745,13 +771,15 @@
   {:pre [(map? database) (map? (:details database))]}
   ;; automatically retry the query if it times out or otherwise fails. This is on top of the auto-retry added by
   ;; `execute`
-  (let [thunk (fn []
-                (execute-bigquery
-                 respond
-                 (:details database)
-                 sql
-                 parameters
-                 cancel-chan))]
+  ;; Apply any swapped connection details (e.g., for workspace isolation)
+  (let [details (driver/maybe-swap-details (:id database) (:details database))
+        thunk   (fn []
+                  (execute-bigquery
+                   respond
+                   details
+                   sql
+                   parameters
+                   cancel-chan))]
     (try
       (thunk)
       (catch Throwable e
@@ -801,7 +829,9 @@
                               :set-timezone                     true
                               :split-part                       true
                               :transforms/python                true
-                              :transforms/table                 true}]
+                              :transforms/table                 true
+                              ;; Workspace isolation using service account impersonation
+                              :isolation                        true}]
   (defmethod driver/database-supports? [:bigquery-cloud-sdk feature] [_driver _feature _db] supported?))
 
 ;; BigQuery is always in UTC
@@ -1009,7 +1039,7 @@
 (defmethod driver/connection-spec :bigquery-cloud-sdk
   [_driver database]
   ;; Return the database details directly since we don't use a JDBC spec for bigquery
-  (:details database))
+  (driver/maybe-swap-details (:id database) (:details database)))
 
 (defmethod driver.sql/default-schema :bigquery-cloud-sdk
   [_]
@@ -1034,8 +1064,17 @@
 
 (defmethod driver/create-schema-if-needed! :bigquery-cloud-sdk
   [driver conn-spec schema]
-  (let [sql [[(format "CREATE SCHEMA IF NOT EXISTS `%s`;" schema)]]]
-    (driver/execute-raw-queries! driver conn-spec sql)))
+  ;; Check if dataset exists using the BigQuery API before trying to create.
+  ;; This is important for workspace isolation where the impersonated SA has
+  ;; access to an existing isolated dataset but cannot create new datasets.
+  (let [details    (get conn-spec :details conn-spec)
+        client     (database-details->client details)
+        project-id (get-project-id details)
+        dataset-id (DatasetId/of project-id schema)]
+    (when-not (.getDataset client dataset-id (u/varargs BigQuery$DatasetOption))
+      ;; Dataset doesn't exist, try to create it
+      (let [sql [[(format "CREATE SCHEMA IF NOT EXISTS `%s`;" schema)]]]
+        (driver/execute-raw-queries! driver conn-spec sql)))))
 
 (defmethod driver/schema-exists? :bigquery-cloud-sdk
   [_driver db-id schema]
@@ -1050,3 +1089,414 @@
   [_driver]
   ;; https://cloud.google.com/bigquery/docs/tables
   1024)
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                           Workspace Isolation                                                  |
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; BigQuery workspace isolation uses service account impersonation instead of SQL GRANT statements.
+;;; Each workspace gets its own service account (created automatically) with table-level read permissions.
+;;;
+;;; Required GCP setup for the main service account:
+;;; - Roles: `roles/bigquery.admin`, `roles/iam.serviceAccountAdmin`, `roles/resourcemanager.projectIamAdmin`
+;;; - APIs: `bigquery.googleapis.com`, `iam.googleapis.com`, `cloudresourcemanager.googleapis.com`
+
+(defn- ws-service-account-credentials
+  "Parse ServiceAccountCredentials from database details."
+  ^ServiceAccountCredentials [{:keys [service-account-json]}]
+  (ServiceAccountCredentials/fromStream
+   (ByteArrayInputStream. (.getBytes ^String service-account-json))))
+
+(defn- ws-database-details->client
+  "Create a BigQuery client from database details for workspace isolation."
+  ^BigQuery [details]
+  (let [creds (.createScoped (ws-service-account-credentials details)
+                             (doto (java.util.ArrayList.)
+                               (.add "https://www.googleapis.com/auth/bigquery")))]
+    (-> (BigQueryOptions/newBuilder)
+        ^BigQueryOptions$Builder (.setCredentials creds)
+        ^BigQueryOptions (.build)
+        (.getService))))
+
+(defn- ws-database-details->iam-client
+  "Create an IAM Admin client from database details."
+  ^IAMClient [details]
+  (let [creds (.createScoped (ws-service-account-credentials details)
+                             (doto (java.util.ArrayList.)
+                               (.add "https://www.googleapis.com/auth/cloud-platform")))]
+    (IAMClient/create
+     (-> (IAMSettings/newBuilder)
+         (.setCredentialsProvider (reify com.google.api.gax.core.CredentialsProvider
+                                    (getCredentials [_] creds)))
+         ^IAMSettings (.build)))))
+
+(defn- ws-service-account-id
+  "Generate the service account ID for a workspace (max 30 chars, lowercase, alphanumeric + hyphens)."
+  [workspace]
+  ;; Format: mb-ws-{workspace-id} truncated to 30 chars
+  (let [ws-id (str (:id workspace))
+        sa-id (str "mb-ws-" ws-id)]
+    (-> sa-id
+        (subs 0 (min 30 (count sa-id)))
+        u/lower-case-en)))
+
+(defn- ws-service-account-exists?
+  "Check if a service account exists."
+  [^IAMClient iam-client ^String project-id ^String sa-email]
+  (try
+    (.getServiceAccount iam-client (format "projects/%s/serviceAccounts/%s" project-id sa-email))
+    true
+    (catch com.google.api.gax.rpc.NotFoundException _
+      false)))
+
+(defn- ws-create-service-account!
+  "Create a service account for a workspace if it doesn't exist.
+   Returns the service account email."
+  [^IAMClient iam-client ^String project-id workspace]
+  (let [sa-id        (ws-service-account-id workspace)
+        sa-email     (format "%s@%s.iam.gserviceaccount.com" sa-id project-id)
+        project-name (format "projects/%s" project-id)]
+    ;; Check if already exists first
+    (if (ws-service-account-exists? iam-client project-id sa-email)
+      (log/debugf "Service account already exists: %s" sa-email)
+      (do
+        (log/infof "Creating service account %s in project %s" sa-id project-id)
+        (let [request (-> (CreateServiceAccountRequest/newBuilder)
+                          (.setName project-name)
+                          (.setAccountId sa-id)
+                          (.setServiceAccount
+                           (-> (ServiceAccount/newBuilder)
+                               (.setDisplayName (format "Metabase Workspace %s" (:id workspace)))
+                               (.setDescription "Auto-created by Metabase for workspace isolation")
+                               (.build)))
+                          (.build))]
+          (.createServiceAccount iam-client request)
+          (log/infof "Created service account: %s" sa-email))))
+    sa-email))
+
+(defn- ws-delete-service-account!
+  "Delete a service account for a workspace. Idempotent - does nothing if SA doesn't exist."
+  [^IAMClient iam-client ^String project-id workspace]
+  (let [sa-id    (ws-service-account-id workspace)
+        sa-email (format "%s@%s.iam.gserviceaccount.com" sa-id project-id)
+        sa-name  (format "projects/%s/serviceAccounts/%s" project-id sa-email)]
+    (when (ws-service-account-exists? iam-client project-id sa-email)
+      (log/infof "Deleting service account %s" sa-email)
+      (let [request (-> (DeleteServiceAccountRequest/newBuilder)
+                        (.setName sa-name)
+                        (.build))]
+        (.deleteServiceAccount iam-client request)
+        (log/infof "Deleted service account: %s" sa-email)))))
+
+(defn- ws-has-role-binding?
+  "Check if a policy already has a binding for the given role and member."
+  [^Policy policy ^String role ^String member]
+  (clojure.core/some (fn [^Binding binding]
+                       (and (= (.getRole binding) role)
+                            (clojure.core/some #(= % member) (.getMembersList binding))))
+                     (.getBindingsList policy)))
+
+(defn- ws-grant-impersonation-permission!
+  "Grant the main service account permission to impersonate the workspace service account."
+  [^IAMClient iam-client ^String project-id ^String main-sa-email ^String workspace-sa-email]
+  (let [resource (format "projects/%s/serviceAccounts/%s" project-id workspace-sa-email)
+        role     "roles/iam.serviceAccountTokenCreator"
+        member   (format "serviceAccount:%s" main-sa-email)]
+    (try
+      ;; Get current policy
+      (let [current-policy (.getIamPolicy iam-client resource)]
+        ;; Only add if not already granted
+        (when-not (ws-has-role-binding? current-policy role member)
+          (let [new-binding    (-> (Binding/newBuilder)
+                                   (.setRole role)
+                                   (.addMembers member)
+                                   (.build))
+                updated-policy (-> (Policy/newBuilder current-policy)
+                                   (.addBindings new-binding)
+                                   (.build))
+                request        (-> (SetIamPolicyRequest/newBuilder)
+                                   (.setResource resource)
+                                   (.setPolicy updated-policy)
+                                   (.build))]
+            (.setIamPolicy iam-client request)
+            (log/infof "Granted impersonation permission on %s to %s" workspace-sa-email main-sa-email))))
+      (catch Exception e
+        (log/warn e "Failed to grant impersonation permission (may already exist)")))))
+
+(defn- ws-grant-project-role!
+  "Grant a project-level IAM role to a service account.
+   This is needed for roles like bigquery.jobUser that must be granted at project level."
+  [details ^String project-id ^String service-account-email ^String role]
+  (log/infof "Granting project role %s to %s" role service-account-email)
+  (let [creds          (.createScoped (ws-service-account-credentials details)
+                                      (doto (java.util.ArrayList.)
+                                        (.add "https://www.googleapis.com/auth/cloud-platform")))
+        settings       (-> (ProjectsSettings/newBuilder)
+                           (.setCredentialsProvider (reify com.google.api.gax.core.CredentialsProvider
+                                                      (getCredentials [_] creds)))
+                           ^ProjectsSettings (.build))
+        projects-client (ProjectsClient/create settings)
+        resource       (format "projects/%s" project-id)
+        member         (format "serviceAccount:%s" service-account-email)]
+    (try
+      ;; Get current policy
+      (let [get-request    (-> (GetIamPolicyRequest/newBuilder)
+                               (.setResource resource)
+                               (.build))
+            current-policy (.getIamPolicy projects-client get-request)]
+        ;; Only add if not already granted
+        (when-not (ws-has-role-binding? current-policy role member)
+          (let [new-binding    (-> (Binding/newBuilder)
+                                   (.setRole role)
+                                   (.addMembers member)
+                                   (.build))
+                updated-policy (-> (Policy/newBuilder current-policy)
+                                   (.addBindings new-binding)
+                                   (.build))
+                set-request    (-> (SetIamPolicyRequest/newBuilder)
+                                   (.setResource resource)
+                                   (.setPolicy updated-policy)
+                                   (.build))]
+            (.setIamPolicy projects-client set-request)
+            (log/infof "Granted %s on project %s to %s" role project-id service-account-email))))
+      (finally
+        (.close projects-client)))))
+
+(defn- ws-wait-for-impersonation-ready!
+  "Poll until impersonation is working. GCP IAM changes can take up to 60 seconds to propagate.
+   Tests by actually creating impersonated credentials and making a simple API call."
+  [details ^String target-sa-email & {:keys [max-attempts interval-ms]
+                                      :or   {max-attempts 120
+                                             interval-ms  1000}}]
+  (log/info "Waiting for IAM impersonation to be ready...")
+  (let [base-creds  (.createScoped (ws-service-account-credentials details)
+                                   (doto (java.util.ArrayList.)
+                                     (.add "https://www.googleapis.com/auth/bigquery")))
+        project-id  (get-project-id details)]
+    (loop [attempt 1]
+      (log/debugf "Checking impersonation readiness (attempt %d/%d)" attempt max-attempts)
+      (let [result (try
+                     ;; Try to create impersonated credentials and use them
+                     (let [impersonated (ImpersonatedCredentials/create
+                                         base-creds
+                                         target-sa-email
+                                         nil  ;; delegates
+                                         (doto (java.util.ArrayList.)
+                                           (.add "https://www.googleapis.com/auth/bigquery"))
+                                         3600)
+                           client       (-> (BigQueryOptions/newBuilder)
+                                            ^BigQueryOptions$Builder (.setCredentials impersonated)
+                                            ^BigQueryOptions (.build)
+                                            ^BigQuery (.getService))]
+                       ;; Try a simple operation - list datasets (limited to 1)
+                       (.listDatasets client project-id (into-array BigQuery$DatasetListOption []))
+                       :ready)
+                     (catch Exception e
+                       (if (or (str/includes? (str (ex-message e)) "permission")
+                               (str/includes? (str (ex-message e)) "403")
+                               (str/includes? (str (ex-message e)) "Access Denied"))
+                         :not-ready
+                         (do
+                           (log/debugf "Unexpected error checking impersonation: %s" (ex-message e))
+                           :not-ready))))]
+        (cond
+          (= result :ready)
+          (log/info "IAM impersonation is ready")
+
+          (>= attempt max-attempts)
+          (throw (ex-info "Timeout waiting for IAM impersonation to propagate"
+                          {:target-sa target-sa-email
+                           :attempts  attempt}))
+
+          :else
+          (do
+            (Thread/sleep ^long interval-ms)
+            (recur (inc attempt))))))))
+
+(defn- ws-role-name->acl-role
+  "Convert a BigQuery IAM role name to an Acl$Role."
+  ^Acl$Role [^String role-name]
+  (case role-name
+    "roles/bigquery.dataEditor" Acl$Role/WRITER
+    "roles/bigquery.dataViewer" Acl$Role/READER
+    "roles/bigquery.dataOwner"  Acl$Role/OWNER
+    (throw (ex-info (str "Unknown role: " role-name) {:role role-name}))))
+
+(defn- ws-has-acl-entry?
+  "Check if an ACL list already has an entry for the given entity and role."
+  [acl-list ^Acl$User entity ^Acl$Role role]
+  (clojure.core/some (fn [^Acl acl]
+                       (and (= (.getEntity acl) entity)
+                            (= (.getRole acl) role)))
+                     acl-list))
+
+(defn- ws-grant-dataset-acl!
+  "Grant an ACL role on a dataset to a service account."
+  [^BigQuery client ^DatasetId dataset-id ^String service-account-email ^String role-name]
+  (log/debugf "Granting %s on dataset %s to %s" role-name dataset-id service-account-email)
+  (let [dataset     ^Dataset (.getDataset client dataset-id
+                                          ^"[Lcom.google.cloud.bigquery.BigQuery$DatasetOption;"
+                                          (into-array BigQuery$DatasetOption []))
+        current-acl (into [] (.getAcl dataset))
+        acl-role    (ws-role-name->acl-role role-name)
+        acl-user    (Acl$User. service-account-email)]
+    ;; Only add if not already granted
+    (when-not (ws-has-acl-entry? current-acl acl-user acl-role)
+      (let [new-acl-entry   (Acl/of acl-user acl-role)
+            updated-acl     (conj current-acl new-acl-entry)
+            updated-dataset (-> (.toBuilder dataset)
+                                (.setAcl ^"clojure.lang.PersistentVector" updated-acl)
+                                (.build))]
+        (.update client updated-dataset
+                 ^"[Lcom.google.cloud.bigquery.BigQuery$DatasetOption;"
+                 (into-array BigQuery$DatasetOption []))))))
+
+(defn- ws-has-table-iam-binding?
+  "Check if a table IAM policy already has a binding for the given role and identity."
+  [^com.google.cloud.Policy policy ^Role role ^Identity identity]
+  (let [bindings (.getBindings policy)
+        identity-set (clojure.core/get bindings role)]
+    (and identity-set (.contains ^java.util.Set identity-set identity))))
+
+(defn- ws-grant-table-read-access!
+  "Grant read access on a specific table to a service account using table-level IAM."
+  [^BigQuery client ^TableId table-id ^String service-account-email]
+  (log/debugf "Granting read access on table %s to %s" table-id service-account-email)
+  (let [current-policy (.getIamPolicy client table-id (into-array BigQuery$IAMOption []))
+        role           (Role/of "roles/bigquery.dataViewer")
+        sa-identity    (Identity/serviceAccount service-account-email)]
+    ;; Only add if not already granted
+    (when-not (ws-has-table-iam-binding? current-policy role sa-identity)
+      (let [updated-policy (-> current-policy
+                               (.toBuilder)
+                               (.addIdentity role sa-identity (into-array Identity []))
+                               (.build))]
+        (.setIamPolicy client table-id updated-policy (into-array BigQuery$IAMOption []))))))
+
+(defmethod driver/init-workspace-isolation! :bigquery-cloud-sdk
+  [_driver database workspace]
+  (let [details       (:details database)
+        client        (ws-database-details->client details)
+        iam-client    (ws-database-details->iam-client details)
+        project-id    (get-project-id details)
+        main-sa-email (.getClientEmail (ws-service-account-credentials details))
+        dataset-name  (driver.u/workspace-isolation-namespace-name workspace)]
+
+    (try
+      ;; Create the workspace service account (or get existing)
+      (let [ws-sa-email (ws-create-service-account! iam-client project-id workspace)
+            dataset-id  (DatasetId/of project-id dataset-name)]
+
+        (log/infof "Initializing BigQuery workspace isolation: dataset=%s, service-account=%s"
+                   dataset-name ws-sa-email)
+
+        ;; Grant main SA permission to impersonate workspace SA
+        (ws-grant-impersonation-permission! iam-client project-id main-sa-email ws-sa-email)
+
+        ;; Grant the workspace SA permission to run BigQuery jobs (queries) at project level
+        ;; Note: We intentionally do NOT grant project-level dataEditor as that would give
+        ;; access to all datasets. The workspace SA only gets dataEditor on its isolated dataset.
+        (ws-grant-project-role! details project-id ws-sa-email "roles/bigquery.jobUser")
+
+        ;; Wait for IAM permissions to propagate by polling until impersonation works
+        (ws-wait-for-impersonation-ready! details ws-sa-email)
+
+        ;; Create the isolated dataset if it doesn't exist (using main SA credentials, not impersonated)
+        (when-not (.getDataset client dataset-id
+                               ^"[Lcom.google.cloud.bigquery.BigQuery$DatasetOption;"
+                               (into-array BigQuery$DatasetOption []))
+          (let [dataset-info (-> (DatasetInfo/newBuilder dataset-id)
+                                 (.setDescription (format "Metabase workspace isolation for workspace %s" (:id workspace)))
+                                 (.build))]
+            (.create client dataset-info
+                     ^"[Lcom.google.cloud.bigquery.BigQuery$DatasetOption;"
+                     (into-array BigQuery$DatasetOption []))))
+
+        ;; Grant the workspace service account dataEditor role on the isolated dataset
+        ;; dataEditor allows: create/update/delete tables, insert/update/delete data
+        (ws-grant-dataset-acl! client dataset-id ws-sa-email "roles/bigquery.dataEditor")
+
+        ;; Return workspace connection details for impersonation
+        ;; :user is used by grant-read-access-to-tables! to know which SA to grant access to
+        ;; :impersonate-service-account is used by the connection swap to use impersonated credentials
+        {:schema           dataset-name
+         :database_details {:user                        ws-sa-email
+                            :impersonate-service-account ws-sa-email}})
+      (finally
+        (.close iam-client)))))
+
+(defmethod driver/grant-workspace-read-access! :bigquery-cloud-sdk
+  [_driver database workspace tables]
+  ;; For BigQuery, the workspace contains the service account email in database_details
+  (let [ws-sa-email (-> workspace :database_details :impersonate-service-account)
+        details     (:details database)
+        client      (ws-database-details->client details)
+        project-id  (get-project-id details)]
+
+    (log/debugf "Granting read access to %d tables for %s" (count tables) ws-sa-email)
+
+    ;; Grant dataViewer at table level for each table - proper isolation
+    (doseq [{:keys [schema name]} tables]
+      (let [table-id (TableId/of project-id schema name)]
+        (ws-grant-table-read-access! client table-id ws-sa-email)))))
+
+(def ^:private perm-check-workspace-id 0)
+
+(defmethod driver/check-isolation-permissions :bigquery-cloud-sdk
+  [driver database test-table]
+  ;; BigQuery uses GCP IAM APIs instead of SQL, so we can't use transaction rollback.
+  ;; We run the actual init/grant/destroy operations and clean up immediately.
+  (let [test-workspace {:id   perm-check-workspace-id
+                        :name "_mb_perm_check_"}]
+    (try
+      (let [init-result (try
+                          (driver/init-workspace-isolation! driver database test-workspace)
+                          (catch Exception e
+                            (throw (ex-info (format "Failed to initialize workspace isolation: %s" (ex-message e))
+                                            {:step :init} e))))
+            workspace-with-details (merge test-workspace init-result)]
+        (when test-table
+          (try
+            (driver/grant-workspace-read-access! driver database workspace-with-details [test-table])
+            (catch Exception e
+              (throw (ex-info (format "Failed to grant read access to table %s.%s: %s"
+                                      (:schema test-table) (:name test-table) (ex-message e))
+                              {:step :grant :table test-table} e)))))
+        (try
+          (driver/destroy-workspace-isolation! driver database workspace-with-details)
+          (catch Exception e
+            (throw (ex-info (format "Failed to destroy workspace isolation: %s" (ex-message e))
+                            {:step :destroy} e)))))
+      nil
+      (catch Exception e
+        (ex-message e))
+      (finally
+        (try
+          (driver/destroy-workspace-isolation! driver database test-workspace)
+          (catch Exception _ nil))))))
+
+(defmethod driver/destroy-workspace-isolation! :bigquery-cloud-sdk
+  [_driver database workspace]
+  (let [details      (:details database)
+        client       (ws-database-details->client details)
+        iam-client   (ws-database-details->iam-client details)
+        project-id   (get-project-id details)
+        dataset-name (driver.u/workspace-isolation-namespace-name workspace)
+        dataset-id   (DatasetId/of project-id dataset-name)]
+    (try
+      (log/infof "Destroying BigQuery workspace isolation: dataset=%s" dataset-name)
+
+      ;; Delete the dataset if it exists
+      (when (.getDataset client dataset-id
+                         ^"[Lcom.google.cloud.bigquery.BigQuery$DatasetOption;"
+                         (into-array BigQuery$DatasetOption []))
+        (log/infof "Deleting dataset %s" dataset-name)
+        ;; deleteContents=true removes all tables in the dataset
+        (.delete client dataset-id)
+        (log/infof "Deleted dataset %s" dataset-name))
+
+      ;; Delete the service account (this also removes its IAM bindings)
+      (ws-delete-service-account! iam-client project-id workspace)
+
+      {:success true}
+      (finally
+        (.close iam-client)))))

modules/drivers/clickhouse/src/metabase/driver/clickhouse.clj

@@ -17,10 +17,12 @@
    [metabase.driver.sql-jdbc.execute :as sql-jdbc.execute]
    [metabase.driver.sql.query-processor :as sql.qp]
    [metabase.driver.sql.util :as sql.u]
+   [metabase.driver.util :as driver.u]
    [metabase.util :as u]
    [metabase.util.log :as log])
-  (:import  [com.clickhouse.client.api.query QuerySettings]
-            [java.sql SQLException]))
+  (:import
+   (com.clickhouse.client.api.query QuerySettings)
+   (java.sql Connection SQLException Statement)))
 
 (set! *warn-on-reflection* true)
 
@@ -42,6 +44,7 @@
                               ;; JDBC driver always provides "NO" for the IS_GENERATEDCOLUMN JDBC metadata
                               :describe-is-generated            false
                               :describe-is-nullable             true
+                              :isolation                        true
                               :expression-literals              true
                               :expressions/date                 true
                               :expressions/float                true
@@ -347,3 +350,51 @@
   [_driver _database _table]
   (log/warn "Clickhouse does not support foreign keys. `describe-table-fks` should not have been called!")
   #{})
+
+;;; ------------------------------------------ Workspace Isolation ------------------------------------------
+
+(defmethod driver/init-workspace-isolation! :clickhouse
+  [_driver database workspace]
+  (let [db-name          (driver.u/workspace-isolation-namespace-name workspace)
+        read-user        {:user     (driver.u/workspace-isolation-user-name workspace)
+                          :password (driver.u/random-workspace-password)}
+        escaped-password (sql.u/escape-sql (:password read-user) :ansi)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [sql [(format "CREATE DATABASE IF NOT EXISTS `%s`" db-name)
+                     (format "CREATE USER IF NOT EXISTS `%s` IDENTIFIED BY '%s'"
+                             (:user read-user) escaped-password)
+                     (format "GRANT ALL ON `%s`.* TO `%s`" db-name (:user read-user))]]
+          (.addBatch ^Statement stmt ^String sql))
+        (.executeBatch ^Statement stmt)))
+    {:schema           db-name
+     :database_details read-user}))
+
+(defmethod driver/grant-workspace-read-access! :clickhouse
+  [_driver database workspace tables]
+  (let [read-user-name (-> workspace :database_details :user)
+        sqls           (for [table tables]
+                         (format "GRANT SELECT ON `%s`.`%s` TO `%s`"
+                                 (:schema table)
+                                 (:name table)
+                                 read-user-name))]
+    (when-not read-user-name
+      (throw (ex-info "Workspace isolation is not properly initialized - missing read user name"
+                      {:workspace-id (:id workspace) :step :grant})))
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [sql sqls]
+          (.addBatch ^Statement stmt ^String sql))
+        (.executeBatch ^Statement stmt)))))
+
+(defmethod driver/destroy-workspace-isolation! :clickhouse
+  [_driver database workspace]
+  (let [db-name  (driver.u/workspace-isolation-namespace-name workspace)
+        username (driver.u/workspace-isolation-user-name workspace)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [sql [;; DROP DATABASE cascades to all tables within it
+                     (format "DROP DATABASE IF EXISTS `%s`" db-name)
+                     (format "DROP USER IF EXISTS `%s`" username)]]
+          (.addBatch ^Statement stmt ^String sql))
+        (.executeBatch ^Statement stmt)))))

modules/drivers/mongo/src/metabase/driver/mongo.clj

@@ -440,7 +440,8 @@
                               :index-info                      false
                               :python-transforms               true
                               :transforms/python               true
-                              :database-routing                true}]
+                              :database-routing                true
+                              :isolation                       false}]
   (defmethod driver/database-supports? [:mongo feature] [_driver _feature _db] supported?))
 
 (defmethod driver/database-supports? [:mongo :schemas] [_driver _feat _db] false)
@@ -598,7 +599,7 @@
 
 (defmethod driver/connection-spec :mongo
   [_driver database]
-  (:details database))
+  (driver/maybe-swap-details (:id database) (:details database)))
 
 (defmulti ^:private type->database-type
   "Internal type->database-type multimethod for MongoDB that dispatches on type."

modules/drivers/redshift/src/metabase/driver/redshift.clj

@@ -1,6 +1,7 @@
 (ns metabase.driver.redshift
   "Amazon Redshift Driver."
   (:require
+   [clojure.java.jdbc :as jdbc]
    [clojure.string :as str]
    [honey.sql :as sql]
    [java-time.api :as t]
@@ -15,7 +16,9 @@
    [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
    [metabase.driver.sql-jdbc.sync.describe-database :as sql-jdbc.describe-database]
    [metabase.driver.sql.query-processor :as sql.qp]
+   [metabase.driver.sql.util :as sql.u]
    [metabase.driver.sync :as driver.s]
+   [metabase.driver.util :as driver.u]
    [metabase.util :as u]
    [metabase.util.date-2 :as u.date]
    [metabase.util.honey-sql-2 :as h2x]
@@ -28,6 +31,7 @@
     PreparedStatement
     ResultSet
     ResultSetMetaData
+    Statement
     Types)))
 
 (set! *warn-on-reflection* true)
@@ -51,7 +55,8 @@
                               :test/jvm-timezone-setting        false
                               :transforms/python                true
                               :transforms/table                 true
-                              :uuid-type                        false}]
+                              :uuid-type                        false
+                              :isolation                        true}]
   (defmethod driver/database-supports? [:redshift feature] [_driver _feat _db] supported?))
 
 ;;; +----------------------------------------------------------------------------------------------------------------+
@@ -250,8 +255,7 @@
             :geometry    :type/*    ; spatial data
             :geography   :type/*    ; spatial data
             :intervaly2m :type/*    ; interval literal
-            :intervald2s :type/*}   ; interval literal
-           ))
+            :intervald2s :type/*}))   ; interval literal
 
 (defmethod sql-jdbc.sync/database-type->base-type :redshift
   [driver column-type]
@@ -672,3 +676,80 @@
   ;; https://docs.aws.amazon.com/redshift/latest/mgmt/rsql-query-tool-error-codes.html
   ;; 42P01: undefined_table, 3F000: invalid_schema_name
   (contains? #{"42P01" "3F000"} (sql-jdbc/get-sql-state e)))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                         Workspace Isolation                                                    |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+;; Redshift inherits grant-workspace-read-access! from Postgres.
+;; init-workspace-isolation! is overridden because Redshift doesn't support GRANT user TO user
+;; (PostgreSQL treats users as roles; Redshift requires GRANT ROLE role_name TO user).
+;; Since we don't use DROP OWNED BY in Redshift cleanup, we don't need the role grant anyway.
+;; destroy-workspace-isolation! is overridden because Redshift doesn't support DROP OWNED BY.
+
+(defn- user-exists?
+  "Check if a Redshift user exists."
+  [conn username]
+  (seq (jdbc/query conn ["SELECT 1 FROM pg_user WHERE usename = ?" username])))
+
+(defn- schema-exists?
+  "Check if a schema exists in Redshift."
+  [conn schema-name]
+  (seq (jdbc/query conn ["SELECT 1 FROM pg_namespace WHERE nspname = ?" schema-name])))
+
+(defn- schemas-with-user-grants
+  "Query Redshift to find schemas where the user has been granted privileges."
+  [conn username]
+  (->> (jdbc/query conn
+                   ["SELECT DISTINCT namespace_name FROM svv_relation_privileges
+           WHERE identity_name = ? AND identity_type = 'user'"
+                    username])
+       (keep :namespace_name)))
+
+(defmethod driver/init-workspace-isolation! :redshift
+  [_driver database workspace]
+  (let [schema-name      (driver.u/workspace-isolation-namespace-name workspace)
+        read-user        {:user     (driver.u/workspace-isolation-user-name workspace)
+                          :password (driver.u/random-workspace-password)}
+        escaped-password (sql.u/escape-sql (:password read-user) :ansi)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (let [user-sql (if (user-exists? t-conn (:user read-user))
+                       (format "ALTER USER \"%s\" WITH PASSWORD '%s'" (:user read-user) escaped-password)
+                       (format "CREATE USER \"%s\" WITH PASSWORD '%s'" (:user read-user) escaped-password))]
+        (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+          (doseq [sql [(format "CREATE SCHEMA IF NOT EXISTS \"%s\"" schema-name)
+                       user-sql
+                       (format "GRANT ALL PRIVILEGES ON SCHEMA \"%s\" TO \"%s\"" schema-name (:user read-user))
+                       (format "ALTER DEFAULT PRIVILEGES IN SCHEMA \"%s\" GRANT ALL ON TABLES TO \"%s\"" schema-name (:user read-user))]]
+            (.addBatch ^Statement stmt ^String sql))
+          (.executeBatch ^Statement stmt))))
+    {:schema           schema-name
+     :database_details read-user}))
+
+(defmethod driver/destroy-workspace-isolation! :redshift
+  [_driver database workspace]
+  (let [schema-name (:schema workspace)
+        username    (-> workspace :database_details :user)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (let [user-exists    (user-exists? t-conn username)
+            schema-exists  (schema-exists? t-conn schema-name)
+            granted-schemas (when user-exists
+                              (schemas-with-user-grants t-conn username))]
+        (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+          (when user-exists
+            (doseq [schema granted-schemas]
+              (.addBatch ^Statement stmt
+                         ^String (format "REVOKE ALL PRIVILEGES ON ALL TABLES IN SCHEMA \"%s\" FROM \"%s\""
+                                         schema username))
+              (.addBatch ^Statement stmt
+                         ^String (format "REVOKE ALL PRIVILEGES ON SCHEMA \"%s\" FROM \"%s\""
+                                         schema username)))
+            (when schema-exists
+              (.addBatch ^Statement stmt
+                         ^String (format "ALTER DEFAULT PRIVILEGES IN SCHEMA \"%s\" REVOKE ALL ON TABLES FROM \"%s\""
+                                         schema-name username))))
+          (.addBatch ^Statement stmt
+                     ^String (format "DROP SCHEMA IF EXISTS \"%s\" CASCADE" schema-name))
+          (.addBatch ^Statement stmt
+                     ^String (format "DROP USER IF EXISTS \"%s\"" username))
+          (.executeBatch ^Statement stmt))))))

modules/drivers/redshift/test/metabase/test/data/redshift.clj

@@ -34,6 +34,8 @@
 ;;; need to load this so we can properly override the implementation of `describe-database` below
 (comment metabase.driver.redshift/keep-me)
 
+(def ^:private workspace-isolation-prefix "mb__isolation")
+
 (defmethod driver/database-supports? [:redshift :test/time-type]
   [_driver _feature _database]
   false)
@@ -164,31 +166,71 @@
                               :unknown-error)))]
         (group-by classify schemas)))))
 
+(defn- classify-isolation-schemas
+  "Classifies workspace isolation schemas by age using a single query. Returns a map:
+   {:expired  schemas older than threshold (safe to delete)
+    :recent   schemas created within threshold (might be from parallel test)}"
+  [^java.sql.Connection conn schemas]
+  (if (empty? schemas)
+    {}
+    (let [threshold    (t/minus (t/instant) (t/hours hours-before-expired-threshold))
+          schema-list  (str/join "," (map #(str "'" % "'") schemas))
+          ;; Use pg_class_info joined with pg_namespace to get oldest object creation time per schema
+          sql          (str "SELECT TRIM(n.nspname) as schema_name, MIN(c.relcreationtime) as oldest "
+                            "FROM pg_class_info c "
+                            "JOIN pg_namespace n ON c.relnamespace = n.oid "
+                            "WHERE TRIM(n.nspname) IN (" schema-list ") "
+                            "GROUP BY n.nspname")
+          schema->time (with-open [stmt (.createStatement conn)
+                                   rset (.executeQuery stmt sql)]
+                         (loop [result {}]
+                           (if (.next rset)
+                             (recur (assoc result
+                                           (.getString rset "schema_name")
+                                           (.getTimestamp rset "oldest")))
+                             result)))]
+      (group-by (fn [schema-name]
+                  (if-let [oldest (get schema->time schema-name)]
+                    (if (t/before? (.toInstant oldest) threshold)
+                      :expired
+                      :recent)
+                    ;; Schema not in pg_class_info means no objects - treat as expired
+                    :expired))
+                schemas))))
+
 (defn- delete-old-schemas!
   "Remove unneeded schemas from redshift. Local databases are thrown away after a test run. Shared cloud instances do
   not have this luxury. Test runs can create schemas where models are persisted and nothing cleans these up, leading
-  to redshift clusters hitting the max number of tables allowed."
+  to redshift clusters hitting the max number of tables allowed.
+
+  Also cleans up workspace isolation schemas (mb__isolation_*) and their associated users that may have been
+  left behind by workspace tests. Only deletes isolation schemas older than [[hours-before-expired-threshold]]
+  to avoid interfering with parallel test runs."
   [^java.sql.Connection conn]
-  (let [{old-convention   :old
-         caches-with-info :cache}    (reduce (fn [acc s]
-                                               (cond (sql.tu.unique-prefix/old-dataset-name? s)
-                                                     (update acc :old conj s)
-                                                     (str/starts-with? s "metabase_cache_")
-                                                     (update acc :cache conj s)
-                                                     :else acc))
-                                             {:old [] :cache []}
-                                             (fetch-schemas conn))
+  (let [isolation-pattern (str workspace-isolation-prefix "_")
+        {old-convention   :old
+         caches-with-info :cache
+         isolation        :isolation} (reduce (fn [acc s]
+                                                (cond (sql.tu.unique-prefix/old-dataset-name? s)
+                                                      (update acc :old conj s)
+                                                      (str/starts-with? s "metabase_cache_")
+                                                      (update acc :cache conj s)
+                                                      (str/starts-with? s isolation-pattern)
+                                                      (update acc :isolation conj s)
+                                                      :else acc))
+                                              {:old [] :cache [] :isolation []}
+                                              (fetch-schemas conn))
         {:keys [expired
                 old-style-cache
-                lacking-created-at]} (classify-cache-schemas conn caches-with-info)
-        drop-sql                     (fn [schema-name] (format "DROP SCHEMA IF EXISTS \"%s\" CASCADE;"
-                                                               schema-name))]
-    ;; don't delete unknown-error and recent.
+                lacking-created-at]}  (classify-cache-schemas conn caches-with-info)
+        {expired-isolation :expired}  (classify-isolation-schemas conn isolation)
+        drop-sql                      (fn [schema-name] (format "DROP SCHEMA IF EXISTS \"%s\" CASCADE;" schema-name))]
     (with-open [stmt (.createStatement conn)]
       (doseq [[collection fmt-str] [[old-convention "Dropping old data schema: %s"]
                                     [expired "Dropping expired cache schema: %s"]
                                     [lacking-created-at "Dropping cache without created-at info: %s"]
-                                    [old-style-cache "Dropping old cache schema without `cache_info` table: %s"]]
+                                    [old-style-cache "Dropping old cache schema without `cache_info` table: %s"]
+                                    [expired-isolation "Dropping expired workspace isolation schema: %s"]]
               schema               collection]
         (log/infof fmt-str schema)
         (.execute stmt (drop-sql schema))))))

modules/drivers/snowflake/src/metabase/driver/snowflake.clj

@@ -26,6 +26,7 @@
    [metabase.driver.sql.query-processor.util :as sql.qp.u]
    [metabase.driver.sql.util :as sql.u]
    [metabase.driver.sync :as driver.s]
+   [metabase.driver.util :as driver.u]
    [metabase.util :as u]
    [metabase.util.date-2 :as u.date]
    [metabase.util.honey-sql-2 :as h2x]
@@ -77,7 +78,8 @@
                               :database-routing                       true
                               :metadata/table-existence-check         true
                               :transforms/python                      true
-                              :transforms/table                       true}]
+                              :transforms/table                       true
+                              :isolation                              true}]
   (defmethod driver/database-supports? [:snowflake feature] [_driver _feature _db] supported?))
 
 (defmethod driver/humanize-connection-error-message :snowflake
@@ -958,3 +960,80 @@
 (defmethod sql.qp/->honeysql [:snowflake :ends-with]
   [driver [_ field arg options]]
   (string-filter driver :endswith field arg options))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                         Workspace Isolation                                                    |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(defn- isolation-role-name
+  "Generate role name for workspace isolation."
+  [workspace]
+  (format "MB_ISOLATION_ROLE_%s" (:id workspace)))
+
+(defmethod driver/init-workspace-isolation! :snowflake
+  [_driver database workspace]
+  (let [schema-name      (driver.u/workspace-isolation-namespace-name workspace)
+        db-name          (-> database :details :db)
+        ;; Snowflake identifiers are case-insensitive when unquoted, but we use quoted identifiers
+        ;; in SQL statements. Since warehouse names are stored as uppercase in Snowflake, we need
+        ;; to uppercase here to match (the connection spec does the same in connection-details->spec).
+        warehouse        (some-> (-> database :details :warehouse) u/upper-case-en)
+        role-name        (isolation-role-name workspace)
+        read-user        {:user     (driver.u/workspace-isolation-user-name workspace)
+                          :password (driver.u/random-workspace-password)}
+        escaped-password (sql.u/escape-sql (:password read-user) :ansi)
+        conn-spec        (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    (when-not db-name
+      (throw (ex-info "Snowflake database configuration is missing required 'db' (database name) setting"
+                      {:database-id (:id database) :step :init})))
+    (when-not warehouse
+      (throw (ex-info "Snowflake database configuration is missing required 'warehouse' setting"
+                      {:database-id (:id database) :step :init})))
+    ;; Snowflake RBAC: create schema -> create role -> grant privileges to role -> create user -> grant role to user
+    (doseq [sql [(format "CREATE SCHEMA IF NOT EXISTS \"%s\".\"%s\"" db-name schema-name)
+                 (format "CREATE ROLE IF NOT EXISTS \"%s\"" role-name)
+                 (format "GRANT USAGE ON DATABASE \"%s\" TO ROLE \"%s\"" db-name role-name)
+                 (format "GRANT USAGE ON WAREHOUSE \"%s\" TO ROLE \"%s\"" warehouse role-name)
+                 (format "GRANT USAGE ON SCHEMA \"%s\".\"%s\" TO ROLE \"%s\"" db-name schema-name role-name)
+                 (format "GRANT ALL PRIVILEGES ON SCHEMA \"%s\".\"%s\" TO ROLE \"%s\"" db-name schema-name role-name)
+                 (format "GRANT ALL ON FUTURE TABLES IN SCHEMA \"%s\".\"%s\" TO ROLE \"%s\"" db-name schema-name role-name)
+                 (format "CREATE USER IF NOT EXISTS \"%s\" PASSWORD = '%s' MUST_CHANGE_PASSWORD = FALSE DEFAULT_ROLE = \"%s\""
+                         (:user read-user) escaped-password role-name)
+                 (format "GRANT ROLE \"%s\" TO USER \"%s\"" role-name (:user read-user))]]
+      (jdbc/execute! conn-spec [sql]))
+    {:schema           schema-name
+     :database_details (assoc read-user :role role-name)}))
+
+(defmethod driver/destroy-workspace-isolation! :snowflake
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        db-name     (-> database :details :db)
+        role-name   (isolation-role-name workspace)
+        username    (driver.u/workspace-isolation-user-name workspace)
+        conn-spec   (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    (when-not db-name
+      (throw (ex-info "Snowflake database configuration is missing required 'db' (database name) setting"
+                      {:database-id (:id database) :step :destroy})))
+    ;; Drop in reverse order of creation: schema (CASCADE handles tables) -> user -> role
+    (doseq [sql [(format "DROP SCHEMA IF EXISTS \"%s\".\"%s\" CASCADE" db-name schema-name)
+                 (format "DROP USER IF EXISTS \"%s\"" username)
+                 (format "DROP ROLE IF EXISTS \"%s\"" role-name)]]
+      (jdbc/execute! conn-spec [sql]))))
+
+(defmethod driver/grant-workspace-read-access! :snowflake
+  [_driver database workspace tables]
+  (let [conn-spec (sql-jdbc.conn/db->pooled-connection-spec (:id database))
+        db-name   (-> database :details :db)
+        role-name (-> workspace :database_details :role)
+        schemas   (distinct (map :schema tables))]
+    (when-not db-name
+      (throw (ex-info "Snowflake database configuration is missing required 'db' (database name) setting"
+                      {:database-id (:id database) :step :grant})))
+    (when-not role-name
+      (throw (ex-info "Workspace isolation is not properly initialized - missing role name"
+                      {:workspace-id (:id workspace) :step :grant})))
+    (doseq [schema schemas]
+      (jdbc/execute! conn-spec [(format "GRANT USAGE ON SCHEMA \"%s\".\"%s\" TO ROLE \"%s\"" db-name schema role-name)]))
+    (doseq [table tables]
+      (jdbc/execute! conn-spec [(format "GRANT SELECT ON TABLE \"%s\".\"%s\".\"%s\" TO ROLE \"%s\""
+                                        db-name (:schema table) (:name table) role-name)]))))

modules/drivers/sqlserver/src/metabase/driver/sqlserver.clj

@@ -23,6 +23,7 @@
    [metabase.driver.sql.query-processor.boolean-to-comparison :as sql.qp.boolean-to-comparison]
    [metabase.driver.sql.query-processor.util :as sql.qp.u]
    [metabase.driver.sql.util :as sql.u]
+   [metabase.driver.util :as driver.u]
    [metabase.lib.schema.metadata :as lib.schema.metadata]
    [metabase.util :as u]
    [metabase.util.honey-sql-2 :as h2x]
@@ -61,7 +62,8 @@
                               :jdbc/statements                        false
                               :describe-default-expr                  true
                               :describe-is-nullable                   true
-                              :describe-is-generated                  true}]
+                              :describe-is-generated                  true
+                              :isolation                              true}]
   (defmethod driver/database-supports? [:sqlserver feature] [_driver _feature _db] supported?))
 
 (mu/defmethod driver/database-supports? [:sqlserver :percentile-aggregations]
@@ -1069,3 +1071,69 @@
   [_driver]
   ;; https://learn.microsoft.com/en-us/previous-versions/sql/sql-server-2008-r2/ms191240(v=sql.105)#sysname
   128)
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                         Workspace Isolation                                                    |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+;; SQL Server has separate login (server-level) and user (database-level), but we use the same
+;; name for both to keep things simple. The login is for authentication, the user holds permissions.
+
+(defmethod driver/init-workspace-isolation! :sqlserver
+  [_driver database workspace]
+  (let [schema-name      (driver.u/workspace-isolation-namespace-name workspace)
+        username         (driver.u/workspace-isolation-user-name workspace)
+        password         (driver.u/random-workspace-password)
+        escaped-password (sql.u/escape-sql password :ansi)
+        conn-spec        (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    ;; SQL Server: create login (server level), then user (database level), then schema
+    (doseq [sql [(format (str "IF NOT EXISTS (SELECT name FROM master.sys.server_principals WHERE name = '%s') "
+                              "CREATE LOGIN [%s] WITH PASSWORD = N'%s'")
+                         username username escaped-password)
+                 (format "IF NOT EXISTS (SELECT name FROM sys.database_principals WHERE name = '%s') CREATE USER [%s] FOR LOGIN [%s]"
+                         username username username)
+                 (format "IF NOT EXISTS (SELECT * FROM sys.schemas WHERE name = '%s') EXEC('CREATE SCHEMA [%s]')"
+                         schema-name schema-name)
+                 (format "GRANT CONTROL ON SCHEMA::[%s] TO [%s]" schema-name username)]]
+      (jdbc/execute! conn-spec [sql]))
+    {:schema           schema-name
+     :database_details {:user     username
+                        :password password}}))
+
+(defmethod driver/destroy-workspace-isolation! :sqlserver
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        username    (driver.u/workspace-isolation-user-name workspace)
+        conn-spec   (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    (doseq [sql [(format (str "DECLARE @sql NVARCHAR(MAX) = ''; "
+                              "SELECT @sql += 'DROP TABLE [%s].[' + name + ']; ' "
+                              "FROM sys.tables WHERE schema_id = SCHEMA_ID('%s'); "
+                              "EXEC sp_executesql @sql")
+                         schema-name schema-name)
+                 (format "IF EXISTS (SELECT * FROM sys.schemas WHERE name = '%s') DROP SCHEMA [%s]"
+                         schema-name schema-name)
+                 (format "IF EXISTS (SELECT * FROM sys.database_principals WHERE name = '%s') DROP USER [%s]"
+                         username username)
+                 ;; Kill all sessions using this login before dropping it
+                 (format (str "DECLARE @sql NVARCHAR(MAX) = ''; "
+                              "SELECT @sql += 'KILL ' + CAST(session_id AS VARCHAR(10)) + '; ' "
+                              "FROM sys.dm_exec_sessions WHERE login_name = '%s'; "
+                              "EXEC sp_executesql @sql")
+                         username)
+                 (format "IF EXISTS (SELECT * FROM master.sys.server_principals WHERE name = '%s') DROP LOGIN [%s]"
+                         username username)]]
+      (jdbc/execute! conn-spec [sql]))))
+
+(defmethod driver/grant-workspace-read-access! :sqlserver
+  [_driver database workspace tables]
+  (let [conn-spec (sql-jdbc.conn/db->pooled-connection-spec (:id database))
+        username  (-> workspace :database_details :user)
+        schemas   (distinct (map :schema tables))]
+    (when-not username
+      (throw (ex-info "Workspace isolation is not properly initialized - missing user name"
+                      {:workspace-id (:id workspace) :step :grant})))
+    (doseq [schema schemas]
+      (jdbc/execute! conn-spec [(format "GRANT SELECT ON SCHEMA::[%s] TO [%s]" schema username)]))
+    (doseq [table tables]
+      (jdbc/execute! conn-spec [(format "GRANT SELECT ON [%s].[%s] TO [%s]"
+                                        (:schema table) (:name table) username)]))))

src/metabase/driver.clj

@@ -21,6 +21,7 @@
    [metabase.util :as u]
    [metabase.util.i18n :refer [tru]]
    [metabase.util.json :as json]
+   [metabase.util.log :as log]
    [metabase.util.malli :as mu]
    [metabase.util.malli.registry :as mr]
    [metabase.util.performance :refer [mapv empty?]]
@@ -62,6 +63,91 @@
   [driver & body]
   `(do-with-driver ~driver (fn [] ~@body)))
 
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                         Connection Details Swapping                                            |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(def ^:private ^:dynamic *swapped-connection-details*
+  "A dynamic var that holds a map of database-id -> swapped-details-map for temporarily swapping connection details.
+  When a connection spec is created for a database, if its ID is present in this map, the swap map will be
+  merged into the connection `:details` before they are used to create a connection.
+
+  This provides a mechanism for temporarily using different connection details (e.g., using alternative credentials
+  for workspaces) without mutating the database record.
+
+  The swap map is merged into the database `:details` map. The swap is applied before any connection-specific
+  processing (like hash calculation for connection pooling), so different swaps will result in different
+  connection pools.
+
+  Different drivers may apply this swap at different points in their connection lifecycle, but the semantics
+  are consistent: swapped details are used for the duration of the dynamic scope.
+
+  See [[with-swapped-connection-details]] for usage."
+  nil)
+
+(defn- apply-detail-swaps
+  "Merges the `swap-map` into `details`. Supports nested maps via deep merge."
+  [details swap-map]
+  (reduce-kv
+   (fn [acc k v]
+     (if (and (map? v) (map? (get acc k)))
+       (assoc acc k (apply-detail-swaps (get acc k) v))
+       (assoc acc k v)))
+   details
+   swap-map))
+
+(defn has-connection-swap?
+  "Returns true if there is an active connection detail swap for `database-id`."
+  [database-id]
+  (contains? *swapped-connection-details* database-id))
+
+(defn maybe-swap-details
+  "Returns the database details with any swaps applied from [[*swapped-connection-details*]].
+  If no swap exists for `database-id`, returns `details` unchanged.
+
+  Drivers should call this function when creating connections to apply any active swaps.
+  For JDBC drivers, this is called in [[metabase.driver.sql-jdbc.connection/db->pooled-connection-spec]].
+  For other drivers (e.g., MongoDB), this should be called in their connection creation logic."
+  [database-id details]
+  (if-let [swap-map (get *swapped-connection-details* database-id)]
+    (do
+      (log/debugf "Applying swapped connection details for database %d, swap keys: %s"
+                  database-id (keys swap-map))
+      (apply-detail-swaps details swap-map))
+    details))
+
+(defn do-with-swapped-connection-details
+  "Implementation for [[with-swapped-connection-details]]."
+  [database-id swap-map thunk]
+  (when (contains? *swapped-connection-details* database-id)
+    (throw (ex-info "Nested connection detail swaps are not supported for the same database"
+                    {:database-id database-id})))
+  (log/debugf "Entering swapped connection details scope for database %d, swap keys: %s"
+              database-id (keys swap-map))
+  (binding [*swapped-connection-details* (assoc *swapped-connection-details* database-id swap-map)]
+    (thunk)))
+
+(defmacro with-swapped-connection-details
+  "Temporarily swap the connection details for a specific database within the dynamic scope of `body`.
+
+  The `swap-map` is a map of detail keys to swap values. These will be merged into the database's
+  connection `:details` map. Nested maps are deep-merged.
+
+  Any code that creates a connection for `database-id` within this scope will use the modified details.
+
+  **Important:** Nested swaps for the same database are not supported and will throw an exception.
+  Different databases can have concurrent swaps.
+
+  Example:
+
+    ;; Swap connection to use alternate credentials
+    (driver/with-swapped-connection-details 1 {:user \"workspace-user\" :password \"workspace-pass\"}
+      ;; All connections created in this scope use the swapped credentials
+      (qp/process-query query))"
+  {:style/indent 2}
+  [database-id swap-map & body]
+  `(do-with-swapped-connection-details ~database-id ~swap-map (fn [] ~@body)))
+
 ;;; +----------------------------------------------------------------------------------------------------------------+
 ;;; |                             Driver Registration / Hierarchy / Multimethod Dispatch                             |
 ;;; +----------------------------------------------------------------------------------------------------------------+
@@ -798,7 +884,10 @@
     :describe-is-nullable
 
     ;; Does this driver provide :database-is-generated on (describe-fields) or (describe-table)
-    :describe-is-generated})
+    :describe-is-generated
+
+    ;; Does this driver support isolation, which is used to power workspace
+    :isolation})
 
 (defmulti database-supports?
   "Does this driver and specific instance of a database support a certain `feature`?
@@ -844,7 +933,8 @@
                               :test/dynamic-dataset-loading           true
                               :test/uuids-in-create-table-statements  true
                               :metadata/table-existence-check         false
-                              :metadata/table-writable-check          false}]
+                              :metadata/table-writable-check          false
+                              :isolation                              false}]
   (defmethod database-supports? [::driver feature] [_driver _feature _db] supported?))
 
 ;;; By default a driver supports `:native-parameter-card-reference` if it supports `:native-parameters` AND
@@ -1683,3 +1773,75 @@
       (if (table-known-to-not-exist? driver e)
         false
         (throw e)))))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                           Workspace Isolation                                                  |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(defmulti init-workspace-isolation!
+  "Initialize database isolation for a workspace. Creates an isolated schema/database,
+   user credentials, and grants appropriate permissions for the workspace to operate
+   within its own namespace.
+
+   Returns a map with:
+   - :schema           - The name of the isolated schema/database created
+   - :database_details - Connection details (user, password, etc.) for the isolated user
+
+   Implementations should:
+   - Create an isolated schema or database for the workspace
+   - Create a user with credentials that can only access that schema
+   - Grant appropriate permissions (CREATE, INSERT, SELECT, etc.) on the isolated schema
+
+   This is an enterprise feature. Drivers must also return true for
+   (database-supports? driver :isolation database) to indicate support."
+  {:added "0.59.0" :arglists '([driver database workspace])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmulti destroy-workspace-isolation!
+  "Destroy all database resources created for workspace isolation.
+   This includes dropping schemas/databases, users, roles, and any other
+   resources created by init-workspace-isolation!.
+
+   Should be called when deleting a workspace. Implementations should be
+   idempotent - calling on an already-destroyed workspace should not error."
+  {:added "0.59.0" :arglists '([driver database workspace])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmulti grant-workspace-read-access!
+  "Grant read access on specified tables to a workspace's isolated user.
+   This allows the workspace to read from source tables that it needs as inputs.
+
+   `tables` is a sequence of maps with :schema and :name keys identifying
+   the tables to grant access to."
+  {:added "0.59.0" :arglists '([driver database workspace tables])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)
+
+(defmulti check-isolation-permissions
+  "Check if database connection has sufficient permissions for workspace isolation.
+
+   Rather than directly checking permissions, this method performs the actual isolation
+   operations (init workspace, grant access, destroy resources) in a test workspace
+   because:
+
+   1. Some databases don't provide reliable APIs to check permissions a priori.
+   2. Keeping static permission checks in sync with the actual operations is error-prone.
+   3. A database user might have the necessary workspace permissions even if they
+      lack the introspection permissions to query permission tables.
+
+   Isolation operations run in a transaction that is always rolled back (for databases
+   that support transactional DDL), or are manually cleaned up immediately after testing
+   (for databases where transactions don't work, like BigQuery).
+
+   `test-table` is an optional {:schema ... :name ...} map used to test GRANT SELECT.
+   If nil, the grant test is skipped.
+
+   Returns nil on success, or an error message string on failure.
+
+   Default :sql-jdbc implementation tests CREATE SCHEMA, CREATE USER, GRANT, and DROP.
+   Drivers can override for database-specific syntax."
+  {:added "0.59.0" :arglists '([driver database test-table])}
+  dispatch-on-initialized-driver
+  :hierarchy #'hierarchy)

src/metabase/driver/h2.clj

@@ -1,9 +1,11 @@
 (ns metabase.driver.h2
   (:refer-clojure :exclude [some every?])
   (:require
+   [clojure.java.jdbc :as jdbc]
    [clojure.math.combinatorics :as math.combo]
    [clojure.string :as str]
    [java-time.api :as t]
+   [metabase.config.core :as config]
    [metabase.driver :as driver]
    [metabase.driver-api.core :as driver-api]
    [metabase.driver.common :as driver.common]
@@ -17,6 +19,8 @@
    [metabase.driver.sql-jdbc.sync :as sql-jdbc.sync]
    [metabase.driver.sql.normalize :as sql.normalize]
    [metabase.driver.sql.query-processor :as sql.qp]
+   [metabase.driver.sql.util :as sql.u]
+   [metabase.driver.util :as driver.u]
    [metabase.lib.schema.id :as lib.schema.id]
    [metabase.lib.schema.metadata :as lib.schema.metadata]
    [metabase.util :as u]
@@ -24,9 +28,9 @@
    [metabase.util.i18n :refer [deferred-tru tru]]
    [metabase.util.log :as log]
    [metabase.util.malli :as mu]
-   [metabase.util.performance :refer [every? some]])
+   [metabase.util.performance :refer [every? get-in some]])
   (:import
-   (java.sql Clob ResultSet ResultSetMetaData SQLException)
+   (java.sql Clob Connection ResultSet ResultSetMetaData SQLException Statement)
    (java.time OffsetTime)
    (org.h2.command CommandInterface Parser)
    (org.h2.engine SessionLocal)))
@@ -72,6 +76,7 @@
                               :test/jvm-timezone-setting false
                               :uuid-type                 true
                               :uploads                   true
+                              :isolation                 true
                               :database-routing          true
                               :describe-is-generated     true
                               :describe-is-nullable      true
@@ -140,7 +145,7 @@
   :monday)
 
 ;; TODO - it would be better not to put all the options in the connection string in the first place?
-(defn- connection-string->file+options
+(defn connection-string->file+options
   "Explode a `connection-string` like `file:my-db;OPTION=100;OPTION_2=TRUE` to a pair of file and an options map.
 
     (connection-string->file+options \"file:my-crazy-db;OPTION=100;OPTION_X=TRUE\")
@@ -149,7 +154,7 @@
   {:pre [(string? connection-string)]}
   (let [[file & options] (str/split connection-string #";+")
         options          (into {} (for [option options]
-                                    (str/split option #"=")))]
+                                    (str/split option #"=" 2)))]
     [file options]))
 
 (defn- db-details->user [{:keys [db], :as details}]
@@ -167,8 +172,9 @@
     (when (= (keyword query-type) :native)
       (let [{:keys [details]} (driver-api/database (driver-api/metadata-provider))
             user              (db-details->user details)]
-        (when (or (str/blank? user)
-                  (= user "sa"))        ; "sa" is the default USER
+        (when (and config/is-prod? ;; we elevated permissions in workspace tests
+                   (or (str/blank? user)
+                       (= user "sa")))        ; "sa" is the default USER
           (throw
            (ex-info (tru "Running SQL queries against H2 databases using the default (admin) database user is forbidden.")
                     {:type driver-api/qp.error-type.db})))))))
@@ -517,7 +523,7 @@
 ;; These functions for exploding / imploding the options in the connection strings are here so we can override shady
 ;; options users might try to put in their connection string, like INIT=...
 
-(defn- file+options->connection-string
+(defn file+options->connection-string
   "Implode the results of `connection-string->file+options` back into a connection string."
   [file options]
   (apply str file (for [[k v] options]
@@ -674,3 +680,69 @@
 (defmethod sql/default-schema :h2
   [_]
   "PUBLIC")
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                         Workspace Isolation                                                    |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(defn- replace-credentials
+  "Replace USER and PASSWORD in an H2 connection string."
+  [connection-string new-user new-password]
+  (let [[file options] (connection-string->file+options connection-string)]
+    (file+options->connection-string file (assoc options "USER" new-user "PASSWORD" new-password))))
+
+(defn- get-user-from-connection-string
+  "Extract the USER from an H2 connection string."
+  [connection-string]
+  (let [[_file options] (connection-string->file+options connection-string)]
+    (get options "USER")))
+
+(defmethod driver/init-workspace-isolation! :h2
+  [_driver database workspace]
+  (let [schema-name      (driver.u/workspace-isolation-namespace-name workspace)
+        username         (driver.u/workspace-isolation-user-name workspace)
+        password         (driver.u/random-workspace-password)
+        escaped-password (sql.u/escape-sql password :ansi)
+        ;; H2 embeds credentials in the :db connection string, so we need to build a new one
+        original-db      (get-in database [:details :db])
+        new-db           (replace-credentials original-db username password)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [sql [(format "CREATE SCHEMA IF NOT EXISTS \"%s\"" schema-name)
+                     ;; H2 syntax: CREATE USER userName PASSWORD 'password'
+                     (format "CREATE USER IF NOT EXISTS \"%s\" PASSWORD '%s'" username escaped-password)
+                     ;; Grant access on the isolation schema
+                     (format "GRANT ALL ON SCHEMA \"%s\" TO \"%s\"" schema-name username)]]
+          (.addBatch ^Statement stmt ^String sql))
+        (.executeBatch ^Statement stmt)))
+    {:schema           schema-name
+     :database_details {:db new-db}}))
+
+(defmethod driver/destroy-workspace-isolation! :h2
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        username    (driver.u/workspace-isolation-user-name workspace)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [sql [;; CASCADE drops all objects (tables, etc.) in the schema
+                     (format "DROP SCHEMA IF EXISTS \"%s\" CASCADE" schema-name)
+                     (format "DROP USER IF EXISTS \"%s\"" username)]]
+          (.addBatch ^Statement stmt ^String sql))
+        (.executeBatch ^Statement stmt)))))
+
+(defmethod driver/grant-workspace-read-access! :h2
+  [_driver database workspace tables]
+  (let [username (-> workspace :database_details :db get-user-from-connection-string)
+        schemas  (distinct (map :schema tables))]
+    ;; H2 uses GRANT SELECT ON SCHEMA schemaName TO userName
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [schema schemas]
+          (.addBatch ^Statement stmt
+                     ^String (format "GRANT SELECT ON SCHEMA \"%s\" TO \"%s\"" schema username)))
+        ;; Also grant on individual tables for more fine-grained access
+        (doseq [table tables]
+          (.addBatch ^Statement stmt
+                     ^String (format "GRANT SELECT ON \"%s\".\"%s\" TO \"%s\""
+                                     (:schema table) (:name table) username)))
+        (.executeBatch ^Statement stmt)))))

src/metabase/driver/postgres.clj

@@ -28,6 +28,7 @@
    [metabase.driver.sql.query-processor :as sql.qp]
    [metabase.driver.sql.query-processor.util :as sql.qp.u]
    [metabase.driver.sql.util :as sql.u]
+   [metabase.driver.util :as driver.u]
    [metabase.util :as u]
    [metabase.util.date-2 :as u.date]
    [metabase.util.honey-sql-2 :as h2x]
@@ -42,6 +43,7 @@
     Connection
     ResultSet
     ResultSetMetaData
+    Statement
     Types)
    (java.time LocalDateTime OffsetDateTime OffsetTime)
    (org.apache.commons.codec.binary Hex)
@@ -88,7 +90,8 @@
                               :database-routing         true
                               :transforms/table         true
                               :transforms/python        true
-                              :metadata/table-existence-check true}]
+                              :metadata/table-existence-check true
+                              :isolation                true}]
   (defmethod driver/database-supports? [:postgres feature] [_driver _feature _db] supported?))
 
 (defmethod driver/database-supports? [:postgres :nested-field-columns]
@@ -1304,3 +1307,65 @@
   (let [type  (nippy/thaw-from-in! data-input)
         value (nippy/thaw-from-in! data-input)]
     (doto (PGobject.) (.setType type) (.setValue value))))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                         Workspace Isolation                                                    |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(defn- user-exists?
+  "Check if a PostgreSQL user exists. Uses pg_user which also works in Redshift."
+  [conn username]
+  (seq (jdbc/query conn ["SELECT 1 FROM pg_user WHERE usename = ?" username])))
+
+(defmethod driver/init-workspace-isolation! :postgres
+  [_driver database workspace]
+  (let [schema-name      (driver.u/workspace-isolation-namespace-name workspace)
+        read-user        {:user     (driver.u/workspace-isolation-user-name workspace)
+                          :password (driver.u/random-workspace-password)}
+        escaped-password (sql.u/escape-sql (:password read-user) :ansi)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (let [user-sql (if (user-exists? t-conn (:user read-user))
+                       (format "ALTER USER \"%s\" WITH PASSWORD '%s'" (:user read-user) escaped-password)
+                       (format "CREATE USER \"%s\" WITH PASSWORD '%s'" (:user read-user) escaped-password))]
+        (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+          (doseq [sql [(format "CREATE SCHEMA IF NOT EXISTS \"%s\"" schema-name)
+                       user-sql
+                       (format "GRANT ALL PRIVILEGES ON SCHEMA \"%s\" TO \"%s\"" schema-name (:user read-user))
+                       (format "ALTER DEFAULT PRIVILEGES IN SCHEMA \"%s\" GRANT ALL ON TABLES TO \"%s\"" schema-name (:user read-user))
+                       ;; grant role membership to admin so DROP OWNED BY works during cleanup
+                       (format "GRANT \"%s\" TO CURRENT_USER" (:user read-user))]]
+            (.addBatch ^Statement stmt ^String sql))
+          (.executeBatch ^Statement stmt))))
+    {:schema           schema-name
+     :database_details read-user}))
+
+(defmethod driver/destroy-workspace-isolation! :postgres
+  [_driver database workspace]
+  (let [schema-name (:schema workspace)
+        username    (-> workspace :database_details :user)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [sql (cond-> [(format "DROP SCHEMA IF EXISTS \"%s\" CASCADE" schema-name)]
+                      (user-exists? t-conn username)
+                      (into [(format "DROP OWNED BY \"%s\"" username)
+                             (format "DROP USER IF EXISTS \"%s\"" username)]))]
+          (.addBatch ^Statement stmt ^String sql))
+        (.executeBatch ^Statement stmt)))))
+
+(defmethod driver/grant-workspace-read-access! :postgres
+  [_driver database workspace tables]
+  (let [username      (-> workspace :database_details :user)
+        table-schemas (into #{} (keep :schema) tables)
+        schema-grants (for [s table-schemas
+                            :when (not (str/blank? s))]
+                        (format "GRANT USAGE ON SCHEMA \"%s\" TO \"%s\"" s username))
+        table-grants  (for [{s :schema, t :name} tables]
+                        (if (str/blank? s)
+                          (format "GRANT SELECT ON TABLE \"%s\" TO \"%s\"" t username)
+                          (format "GRANT SELECT ON TABLE \"%s\".\"%s\" TO \"%s\"" s t username)))
+        sqls          (concat schema-grants table-grants)]
+    (jdbc/with-db-transaction [t-conn (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+      (with-open [^Statement stmt (.createStatement ^Connection (:connection t-conn))]
+        (doseq [sql sqls]
+          (.addBatch ^Statement stmt ^String sql))
+        (.executeBatch ^Statement stmt)))))

src/metabase/driver/sql.clj

@@ -183,10 +183,21 @@
   {:table (sql.normalize/normalize-name driver table)
    :schema (some->> schema (sql.normalize/normalize-name driver))})
 
+(defn- parsed-table-refs
+  "Parse a native query and return a sequence of normalized table specs {:table ... :schema ...}."
+  [driver query]
+  (->> query
+       driver-api/raw-native-query
+       macaw/parsed-query
+       macaw/query->components
+       :tables
+       (map :component)
+       (map #(normalize-table-spec driver %))))
+
 (mu/defmethod driver/native-query-deps :sql :- ::driver/native-query-deps
   [driver :- :keyword
    query  :- :metabase.lib.schema/native-only-query]
-  (let [db-tables (driver-api/tables query)
+  (let [db-tables     (driver-api/tables query)
         db-transforms (driver-api/transforms query)]
     (-> query
         driver-api/raw-native-query

src/metabase/driver/sql_jdbc.clj

@@ -331,3 +331,46 @@
      (->> (.getMetaData conn)
           sql-jdbc.describe-database/all-schemas
           (m/find-first #(= % schema))))))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                         Workspace Isolation                                                    |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(def ^:private perm-check-workspace-id 0)
+
+(defmethod driver/check-isolation-permissions :sql-jdbc
+  [driver database test-table]
+  (let [test-workspace {:id   perm-check-workspace-id
+                        :name "_mb_perm_check_"}]
+    (sql-jdbc.execute/do-with-connection-with-options
+     driver
+     database
+     {:write? true}
+     (fn [^Connection conn]
+       (.setAutoCommit conn false)
+       (try
+         (let [init-result (try
+                             (driver/init-workspace-isolation! driver database test-workspace)
+                             (catch Exception e
+                               (throw (ex-info (format "Failed to initialize workspace isolation (CREATE SCHEMA/USER): %s"
+                                                       (ex-message e))
+                                               {:step :init} e))))
+               workspace-with-details (merge test-workspace init-result)]
+           (when test-table
+             (try
+               (driver/grant-workspace-read-access! driver database workspace-with-details [test-table])
+               (catch Exception e
+                 (throw (ex-info (format "Failed to grant read access to table %s.%s: %s"
+                                         (:schema test-table) (:name test-table) (ex-message e))
+                                 {:step :grant :table test-table} e)))))
+           (try
+             (driver/destroy-workspace-isolation! driver database workspace-with-details)
+             (catch Exception e
+               (throw (ex-info (format "Failed to destroy workspace isolation (DROP SCHEMA/USER): %s"
+                                       (ex-message e))
+                               {:step :destroy} e)))))
+         nil
+         (catch Exception e
+           (ex-message e))
+         (finally
+           (.rollback conn)))))))

src/metabase/driver/sql_jdbc/connection.clj

@@ -18,7 +18,9 @@
    ^{:clj-kondo/ignore [:discouraged-namespace]}
    [toucan2.core :as t2])
   (:import
+   (com.google.common.cache Cache CacheBuilder RemovalNotification)
    (com.mchange.v2.c3p0 DataSources)
+   (java.util.concurrent TimeUnit)
    (javax.sql DataSource)
    (org.apache.logging.log4j Level)))
 
@@ -208,11 +210,35 @@
   (driver-api/destroy-connection-pool! pool-spec)
   (ssh/close-tunnel! pool-spec))
 
-(defonce ^:private ^{:doc "A map of our currently open connection pools, keyed by Database `:id`."}
+(def ^:private swapped-pool-ttl-minutes
+  "Time-to-live in minutes for swapped connection pools.
+  Pools are evicted after this duration of inactivity (no access)."
+  15)
+
+(defonce ^:private ^{:doc "A map of our currently open connection pools for canonical (non-swapped) connections,
+  keyed by database-id. Each database has at most one canonical pool."}
   database-id->connection-pool
   (atom {}))
 
-(defonce ^:private ^{:doc "A map of DB details hash values, keyed by Database `:id`."}
+(defonce ^:private ^Cache ^{:doc "A Guava cache of swapped connection pools with TTL-based expiration,
+  keyed by [database-id, details-hash]. Pools are evicted after inactivity. Different swapped
+  details for the same database get separate pools, while identical details share a pool."}
+  swapped-connection-pools
+  (.. (CacheBuilder/newBuilder)
+      (expireAfterAccess swapped-pool-ttl-minutes TimeUnit/MINUTES)
+      (removalListener
+       (reify com.google.common.cache.RemovalListener
+         (^void onRemoval [_ ^RemovalNotification rn]
+           (let [[database-id _details-hash] (.getKey rn)
+                 pool-spec                   (.getValue rn)]
+             (log/debugf "Evicting swapped pool for database %d (cause: %s)"
+                         database-id (.getCause rn))
+             (destroy-pool! database-id pool-spec)
+             nil))))
+      (build)))
+
+(defonce ^:private ^{:doc "A map of DB details hash values for the canonical (non-swapped) details, keyed by Database `:id`.
+  This is used to detect when database details have been updated in the application database."}
   database-id->jdbc-spec-hash
   (atom {}))
 
@@ -223,28 +249,44 @@
   (when (some? database)
     (hash (connection-details->spec driver details))))
 
-(defn- set-pool!
-  "Atomically update the current connection pool for Database `database` with `database-id`. Use this function instead
-  of modifying database-id->connection-pool` directly because it properly closes down old pools in a thread-safe way,
-  ensuring no more than one pool is ever open for a single database. Also modifies the [[database-id->jdbc-spec-hash]]
-  map with the hash value of the given DB's details map."
-  [database-id pool-spec-or-nil database]
-  {:pre [(integer? database-id)]}
-  (let [[old-id->pool] (if pool-spec-or-nil
+(defn- set-canonical-pool!
+  "Atomically update the canonical connection pool for a database.
+
+  Use this function instead of modifying connection pool atoms directly because it properly closes down old pools in a
+  thread-safe way, ensuring no more than one canonical pool is ever open for a specific database.
+
+  This function is only for canonical (non-swapped) pools. Swapped pools are managed by the Guava cache."
+  [database-id details-hash pool-spec-or-nil]
+  {:pre [(integer? database-id) (some? details-hash)]}
+  (let [[old-pool-map] (if pool-spec-or-nil
                          (swap-vals! database-id->connection-pool assoc database-id pool-spec-or-nil)
                          (swap-vals! database-id->connection-pool dissoc database-id))]
     ;; if we replaced a different pool with the new pool that is different from the old one, destroy the old pool
-    (when-let [old-pool-spec (get old-id->pool database-id)]
+    (when-let [old-pool-spec (get old-pool-map database-id)]
       (when-not (identical? old-pool-spec pool-spec-or-nil)
         (destroy-pool! database-id old-pool-spec))))
-  ;; update the db details hash cache with the new hash value
-  (swap! database-id->jdbc-spec-hash assoc database-id (jdbc-spec-hash database))
+  ;; Update canonical hash cache
+  (swap! database-id->jdbc-spec-hash assoc database-id details-hash)
   nil)
 
 (defn invalidate-pool-for-db!
-  "Invalidates the connection pool for the given database by closing it and removing it from the cache."
+  "Invalidates all connection pools for the given database (canonical and swapped) by closing them and removing from cache."
   [database]
-  (set-pool! (u/the-id database) nil nil))
+  (let [db-id (u/the-id database)
+        has-canonical? (contains? @database-id->connection-pool db-id)
+        ;; Find all swapped pool keys for this database (keys are [db-id, details-hash] tuples)
+        swapped-keys (filter (fn [[cached-db-id _details-hash]]
+                               (= cached-db-id db-id))
+                             (keys (.asMap ^Cache swapped-connection-pools)))]
+    (log/debugf "Invalidating connection pools for database %d (canonical: %s, swapped count: %d)"
+                db-id has-canonical? (count swapped-keys))
+    ;; Clear canonical pool
+    (when-let [pool-spec (get @database-id->connection-pool db-id)]
+      (destroy-pool! db-id pool-spec)
+      (swap! database-id->connection-pool dissoc db-id))
+    ;; Clear all swapped pools for this DB (removal listener will call destroy-pool!)
+    (doseq [cache-key swapped-keys]
+      (.invalidate ^Cache swapped-connection-pools cache-key))))
 
 (defn- log-ssh-tunnel-reconnect-msg! [db-id]
   (log/warn (u/format-color :red "ssh tunnel for database %s looks closed; marking pool invalid to reopen it" db-id))
@@ -258,76 +300,168 @@
   (log/warn (u/format-color :yellow "Password of database %s expired; marking pool invalid to reopen it" db-id))
   nil)
 
+(defn- pool-invalidation-reason
+  "Returns a keyword describing why a pool is invalid, or nil if valid.
+  Possible reasons: :password-expired, :tunnel-closed."
+  [pool-spec]
+  (let [{:keys [password-expiry-timestamp tunnel-session]} pool-spec]
+    (cond
+      (and (int? password-expiry-timestamp)
+           (<= password-expiry-timestamp (System/currentTimeMillis)))
+      :password-expired
+
+      (nil? tunnel-session)
+      nil ;; no tunnel, valid
+
+      (ssh/ssh-tunnel-open? pool-spec)
+      nil ;; tunnel open, valid
+
+      :else
+      :tunnel-closed)))
+
+(defn- log-pool-invalidation!
+  "Log a warning about why a pool is being invalidated."
+  [database-id reason]
+  (case reason
+    :password-expired (log-password-expiry! database-id)
+    :tunnel-closed    (log-ssh-tunnel-reconnect-msg! database-id)
+    :hash-changed     (log-jdbc-spec-hash-change-msg! database-id)
+    nil))
+
+(defn- get-pool-if-valid
+  "Returns the pool-spec if it's valid, nil otherwise.
+  Optionally logs the invalidation reason."
+  [pool-spec database-id log-invalidation?]
+  (if-let [reason (pool-invalidation-reason pool-spec)]
+    (do
+      (when log-invalidation?
+        (log-pool-invalidation! database-id reason))
+      nil)
+    pool-spec))
+
+(defn- swap-cache-key
+  "Generate cache key for swapped pools: [database-id, jdbc-spec-hash].
+  Different details (e.g., different user credentials) will produce different keys,
+  ensuring each unique swap gets its own connection pool.
+  Uses jdbc-spec-hash for stable hashing based on the actual connection spec."
+  [{:keys [id] :as db}]
+  [id (jdbc-spec-hash db)])
+
+(defn- get-swapped-pool
+  "Get or create a swapped connection pool from the Guava cache.
+  Validates that existing pools are still valid (password not expired, tunnel open).
+  Pools are keyed by [database-id, details-hash] so different swapped details get separate pools.
+  The `db` should already have swapped details applied."
+  [{:keys [id] :as db}]
+  (let [cache-key (swap-cache-key db)
+        existing  (.getIfPresent ^Cache swapped-connection-pools cache-key)]
+    (cond
+      ;; No existing pool - create a new one
+      (nil? existing)
+      (.get ^Cache swapped-connection-pools
+            cache-key
+            (reify java.util.concurrent.Callable
+              (call [_]
+                (log/debugf "Creating new swapped connection pool for database %d with spec hash %d"
+                            id (second cache-key))
+                (create-pool! db))))
+
+      ;; Existing pool is invalid (password expired, tunnel closed) - invalidate and recreate
+      (nil? (get-pool-if-valid existing id true))
+      (do
+        (log/debugf "Swapped connection pool for database %d is invalid, recreating" id)
+        (.invalidate ^Cache swapped-connection-pools cache-key)
+        (recur db))
+
+      ;; Existing pool is valid - use it
+      :else
+      (do
+        (log/debugf "Using existing swapped connection pool for database %d" id)
+        existing))))
+
+(defn- canonical-pool-hash-changed?
+  "Check if the canonical pool's hash differs from the expected hash.
+  Handles stale DatabaseInstance by re-fetching from app DB."
+  [database-id expected-hash]
+  (let [curr-hash (get @database-id->jdbc-spec-hash database-id)]
+    (when (and (some? curr-hash) (not= curr-hash expected-hash))
+      ;; the hash didn't match, but it's possible that a stale instance of `DatabaseInstance`
+      ;; was passed in (ex: from a long-running sync operation); fetch the latest one from
+      ;; our app DB, and see if it STILL doesn't match
+      (not= curr-hash (-> (t2/select-one [:model/Database :id :engine :details] :id database-id)
+                          jdbc-spec-hash)))))
+
+(defn- get-canonical-pool
+  "Get a canonical pool if it exists and is valid, otherwise return nil."
+  [database-id details-hash log-invalidation?]
+  (let [pool-spec (get @database-id->connection-pool database-id ::not-found)]
+    (cond
+      (= ::not-found pool-spec)
+      nil
+
+      ;; Check if the hash has changed (details were updated in DB)
+      (canonical-pool-hash-changed? database-id details-hash)
+      (do
+        (when log-invalidation?
+          (log-pool-invalidation! database-id :hash-changed))
+        nil)
+
+      ;; Check pool validity (password expiry, tunnel status)
+      :else
+      (get-pool-if-valid pool-spec database-id log-invalidation?))))
+
 (defn db->pooled-connection-spec
   "Return a JDBC connection spec that includes a c3p0 `ComboPooledDataSource`. These connection pools are cached so we
-  don't create multiple ones for the same DB."
+  don't create multiple ones for the same DB.
+
+  When [[metabase.driver/with-swapped-connection-details]] is active for a database, the database details are
+  modified before creating the connection pool. Swapped pools are stored in a separate Guava cache with TTL-based
+  eviction (see [[swapped-pool-ttl-minutes]]), while canonical pools are stored in the main atom cache."
   [db-or-id-or-spec]
   (when-let [db-id (u/id db-or-id-or-spec)]
     (driver-api/check-allowed-access! db-id))
   (cond
     ;; db-or-id-or-spec is a Database instance or an integer ID
     (u/id db-or-id-or-spec)
-    (let [database-id (u/the-id db-or-id-or-spec)
-          ;; we need the Database instance no matter what (in order to compare details hash with cached value)
-          db          (or (when (driver-api/instance-of? :model/Database db-or-id-or-spec)
-                            (driver-api/instance->metadata db-or-id-or-spec :metadata/database))
-                          (when (= (:lib/type db-or-id-or-spec) :metadata/database)
-                            db-or-id-or-spec)
-                          (driver-api/with-metadata-provider database-id
-                            (driver-api/database (driver-api/metadata-provider))))
-          get-fn      (fn [db-id log-invalidation?]
-                        (let [details (get @database-id->connection-pool db-id ::not-found)]
-                          (cond
-                            ;; for the audit db, we pass the datasource for the app-db. This lets us use fewer db
-                            ;; connections with *application-db* and 1 less connection pool. Note: This data-source is
-                            ;; not in [[database-id->connection-pool]].
-                            (or (:is-audit db) (get-in db [:details :is-audit-dev]))
-                            {:datasource (driver-api/data-source)}
+    (let [database-id  (u/the-id db-or-id-or-spec)
+          ;; we need the Database instance no matter what (in order to calculate details hash)
+          db-original  (or (when (driver-api/instance-of? :model/Database db-or-id-or-spec)
+                             (driver-api/instance->metadata db-or-id-or-spec :metadata/database))
+                           (when (= (:lib/type db-or-id-or-spec) :metadata/database)
+                             db-or-id-or-spec)
+                           (driver-api/with-metadata-provider database-id
+                             (driver-api/database (driver-api/metadata-provider))))
+          ;; Apply connection detail swaps if present
+          has-swap?    (driver/has-connection-swap? database-id)
+          db           (update db-original :details #(driver/maybe-swap-details database-id %))
+          ;; Calculate hash from final (possibly swapped) details
+          details-hash (jdbc-spec-hash db)]
+      (cond
+        ;; for the audit db, we pass the datasource for the app-db. This lets us use fewer db
+        ;; connections with *application-db* and 1 less connection pool. Note: This data-source is
+        ;; not in [[database-id->connection-pool]].
+        (or (:is-audit db) (get-in db [:details :is-audit-dev]))
+        {:datasource (driver-api/data-source)}
 
-                            (= ::not-found details)
-                            nil
+        ;; Swapped pool: use Guava cache with TTL
+        has-swap?
+        (get-swapped-pool db)
 
-                            ;; details hash changed from what is cached; invalid
-                            (let [curr-hash (get @database-id->jdbc-spec-hash db-id)
-                                  new-hash  (jdbc-spec-hash db)]
-                              (when (and (some? curr-hash) (not= curr-hash new-hash))
-                                ;; the hash didn't match, but it's possible that a stale instance of `DatabaseInstance`
-                                ;; was passed in (ex: from a long-running sync operation); fetch the latest one from
-                                ;; our app DB, and see if it STILL doesn't match
-                                (not= curr-hash (-> (t2/select-one [:model/Database :id :engine :details] :id database-id)
-                                                    jdbc-spec-hash))))
-                            (when log-invalidation?
-                              (log-jdbc-spec-hash-change-msg! db-id))
-
-                            (let [{:keys [password-expiry-timestamp]} details]
-                              (and (int? password-expiry-timestamp)
-                                   (<= password-expiry-timestamp (System/currentTimeMillis))))
-                            (when log-invalidation?
-                              (log-password-expiry! db-id))
-
-                            (nil? (:tunnel-session details)) ; no tunnel in use; valid
-                            details
-
-                            (ssh/ssh-tunnel-open? details) ; tunnel in use, and open; valid
-                            details
-
-                            :else ; tunnel in use, and not open; invalid
-                            (when log-invalidation?
-                              (log-ssh-tunnel-reconnect-msg! db-id)))))]
-      (or
-       ;; we have an existing pool for this database, so use it
-       (get-fn database-id true)
-       ;; Even tho `set-pool!` will properly shut down old pools if two threads call this method at the same time, we
-       ;; don't want to end up with a bunch of simultaneous threads creating pools only to have them destroyed the
-       ;; very next instant. This will cause their queries to fail. Thus we should do the usual locking here and make
-       ;; sure only one thread will be creating a pool at a given instant.
-       (locking database-id->connection-pool
-         (or
-          ;; check if another thread created the pool while we were waiting to acquire the lock
-          (get-fn database-id false)
-          ;; create a new pool and add it to our cache, then return it
-          (u/prog1 (create-pool! db)
-            (set-pool! database-id <> db))))))
+        ;; Canonical pool: use atom cache
+        :else
+        (or
+         ;; we have an existing valid pool for this database, so use it
+         (get-canonical-pool database-id details-hash true)
+         ;; We don't want to end up with a bunch of simultaneous threads creating pools only to have them destroyed
+         ;; the very next instant. This will cause their queries to fail. Thus we should do the usual locking here
+         ;; and make sure only one thread will be creating a pool at a given instant.
+         (locking database-id->connection-pool
+           (or
+            ;; check if another thread created the pool while we were waiting to acquire the lock
+            (get-canonical-pool database-id details-hash false)
+            ;; create a new pool and add it to our cache, then return it
+            (u/prog1 (create-pool! db)
+              (set-canonical-pool! database-id details-hash <>)))))))
 
     ;; already a `clojure.java.jdbc` spec map
     (map? db-or-id-or-spec)

src/metabase/driver/util.clj

@@ -18,6 +18,7 @@
    [metabase.premium-features.core :as premium-features]
    [metabase.query-processor.error-type :as qp.error-type]
    ^{:clj-kondo/ignore [:deprecated-namespace]} [metabase.query-processor.store :as qp.store]
+   [metabase.system.core :as system]
    [metabase.util :as u]
    [metabase.util.i18n :refer [deferred-tru trs]]
    [metabase.util.log :as log]
@@ -26,7 +27,7 @@
    [metabase.util.performance :as perf :refer [mapv empty?]])
   (:import
    (java.io ByteArrayInputStream)
-   (java.security KeyFactory KeyStore PrivateKey)
+   (java.security KeyFactory KeyStore PrivateKey SecureRandom)
    (java.security.cert Certificate CertificateFactory X509Certificate)
    (java.security.spec PKCS8EncodedKeySpec)
    (javax.net SocketFactory)
@@ -678,3 +679,66 @@
         (auth-provider/fetch-auth auth-provider database-id db-details)
         db-details))
      db-details)))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                        Workspace Isolation Utilities                                           |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(defn- instance-uuid-slug
+  "Create a slug from the site UUID, taking the first character of each section."
+  [site-uuid-string]
+  (->> (str/split site-uuid-string #"-")
+       (map first)
+       (apply str)))
+
+;; WARNING: Changing this prefix requires backwards compatibility handling for existing workspaces.
+;; The prefix is used to identify isolation namespaces in the database, and existing workspaces
+;; will have namespaces created with the current prefix.
+(def ^:private workspace-isolated-prefix "mb__isolation")
+
+(defn workspace-isolation-namespace-name
+  "Generate namespace/database name for workspace isolation following mb__isolation_<slug>_<workspace-id> pattern.
+  Uses 'namespace' as the generic term that maps to 'schema' in Postgres, 'database' in ClickHouse, etc."
+  [workspace]
+  (assert (some? (:id workspace)) "Workspace must have an :id")
+  (let [instance-slug      (instance-uuid-slug (str (system/site-uuid)))
+        clean-workspace-id (str/replace (str (:id workspace)) #"[^a-zA-Z0-9]" "_")]
+    (format "%s_%s_%s" workspace-isolated-prefix instance-slug clean-workspace-id)))
+
+(defn workspace-isolation-user-name
+  "Generate username for workspace isolation."
+  [workspace]
+  (let [instance-slug      (instance-uuid-slug (str (system/site-uuid)))
+        clean-workspace-id (str/replace (str (:id workspace)) #"[^a-zA-Z0-9]" "_")]
+    (format "%s_%s_%s" workspace-isolated-prefix instance-slug clean-workspace-id)))
+
+(def ^:private workspace-password-char-sets
+  "Character sets for password generation. Cycles through these to ensure representation from each."
+  ["ABCDEFGHJKLMNPQRSTUVWXYZ"
+   "abcdefghjkmnpqrstuvwxyz"
+   "123456789"
+   "!#$%&*+-="])
+
+(defn- secure-rand-nth
+  "Like rand-nth but uses SecureRandom for cryptographic security."
+  [^SecureRandom rng coll]
+  (nth coll (.nextInt rng (count coll))))
+
+(defn- secure-shuffle
+  "Like shuffle but uses SecureRandom for cryptographic security."
+  [^SecureRandom rng coll]
+  (let [arr (java.util.ArrayList. ^java.util.Collection coll)]
+    (java.util.Collections/shuffle arr rng)
+    (vec arr)))
+
+(defn random-workspace-password
+  "Generate a random password suitable for most database engines.
+   Uses SecureRandom for cryptographic security. Cycles through character sets
+   (uppercase, lowercase, digits, special) to ensure good distribution."
+  []
+  (let [rng (SecureRandom.)]
+    (->> (cycle workspace-password-char-sets)
+         (take (+ 32 (.nextInt rng 32)))
+         (map #(secure-rand-nth rng %))
+         (secure-shuffle rng)
+         (apply str))))

test/metabase/driver/h2_test.clj

@@ -111,21 +111,22 @@
                t/zone-id
                .normalized)))))
 
-(deftest ^:parallel disallow-admin-accounts-test
+(deftest disallow-admin-accounts-test
   (testing "Check that we're not allowed to run SQL against an H2 database with a non-admin account"
     (mt/with-temp [:model/Database db {:name "Fake-H2-DB", :engine "h2", :details {:db "mem:fake-h2-db"}}]
-      (doseq [[query-type query] {"legacy MBQL query"
-                                  {:database (:id db)
-                                   :type     :native
-                                   :native   {:query "SELECT 1"}}
+      (with-redefs [config/is-prod? true]
+        (doseq [[query-type query] {"legacy MBQL query"
+                                    {:database (:id db)
+                                     :type     :native
+                                     :native   {:query "SELECT 1"}}
 
-                                  "MBQL 5 query"
-                                  (lib/native-query (lib-be/application-database-metadata-provider (:id db)) "SELECT 1")}]
-        (testing query-type
-          (is (thrown-with-msg?
-               clojure.lang.ExceptionInfo
-               #"Running SQL queries against H2 databases using the default \(admin\) database user is forbidden\.$"
-               (qp/process-query query))))))))
+                                    "MBQL 5 query"
+                                    (lib/native-query (lib-be/application-database-metadata-provider (:id db)) "SELECT 1")}]
+          (testing query-type
+            (is (thrown-with-msg?
+                 clojure.lang.ExceptionInfo
+                 #"Running SQL queries against H2 databases using the default \(admin\) database user is forbidden\.$"
+                 (qp/process-query query)))))))))
 
 (deftest ^:parallel add-interval-honeysql-form-test
   (testing "Should convert fractional seconds to milliseconds"

test/metabase/driver/isolation_test.clj

@@ -0,0 +1,305 @@
+(ns ^:mb/driver-tests metabase.driver.isolation-test
+  "Tests for driver-level isolation functionality.
+   These tests verify the driver isolation APIs work correctly without depending
+   on the full workspace infrastructure."
+  (:require
+   [clojure.java.jdbc :as jdbc]
+   [clojure.test :refer :all]
+   [metabase.driver :as driver]
+   [metabase.driver.sql-jdbc.connection :as sql-jdbc.conn]
+   [metabase.driver.util :as driver.u]
+   [metabase.test :as mt]
+   [toucan2.core :as t2]))
+
+(set! *warn-on-reflection* true)
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                       Isolation Resource Checks                                                |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(defmulti isolation-resources-exist?
+  "Check if isolation resources exist for the given workspace.
+   Returns a map with keys like :schema, :user indicating which resources exist.
+   For drivers that use different isolation models (e.g., BigQuery with service accounts),
+   returns nil to indicate resource checking is not applicable."
+  {:arglists '([driver database workspace])}
+  (fn [driver _database _workspace] driver)
+  :hierarchy #'driver/hierarchy)
+
+(defmethod isolation-resources-exist? :default
+  [_driver _database _workspace]
+  ;; Default: nil means we can't check resources for this driver
+  nil)
+
+(defmethod isolation-resources-exist? :postgres
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        username    (driver.u/workspace-isolation-user-name workspace)
+        conn-spec   (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    {:schema (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM information_schema.schemata WHERE schema_name = ?" schema-name])
+                 seq boolean)
+     :user   (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM pg_user WHERE usename = ?" username])
+                 seq boolean)}))
+
+;; Redshift inherits from postgres
+(defmethod isolation-resources-exist? :redshift
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        username    (driver.u/workspace-isolation-user-name workspace)
+        conn-spec   (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    {:schema (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM information_schema.schemata WHERE schema_name = ?" schema-name])
+                 seq boolean)
+     :user   (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM pg_user WHERE usename = ?" username])
+                 seq boolean)}))
+
+(defmethod isolation-resources-exist? :h2
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        username    (driver.u/workspace-isolation-user-name workspace)
+        conn-spec   (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    {:schema (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM INFORMATION_SCHEMA.SCHEMATA WHERE UPPER(SCHEMA_NAME) = UPPER(?)" schema-name])
+                 seq boolean)
+     :user   (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM INFORMATION_SCHEMA.USERS WHERE UPPER(USER_NAME) = UPPER(?)" username])
+                 seq boolean)}))
+
+(defmethod isolation-resources-exist? :snowflake
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        role-name   (driver.u/workspace-isolation-user-name workspace)
+        conn-spec   (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    {:schema (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM INFORMATION_SCHEMA.SCHEMATA WHERE SCHEMA_NAME = ?" schema-name])
+                 seq boolean)
+     :role   (-> (jdbc/query conn-spec
+                             ["SHOW ROLES LIKE ?" role-name])
+                 seq boolean)}))
+
+(defmethod isolation-resources-exist? :sqlserver
+  [_driver database workspace]
+  (let [schema-name (driver.u/workspace-isolation-namespace-name workspace)
+        username    (driver.u/workspace-isolation-user-name workspace)
+        conn-spec   (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    {:schema (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM sys.schemas WHERE name = ?" schema-name])
+                 seq boolean)
+     :user   (-> (jdbc/query conn-spec
+                             ["SELECT 1 FROM sys.database_principals WHERE name = ?" username])
+                 seq boolean)}))
+
+(defmethod isolation-resources-exist? :clickhouse
+  [_driver database workspace]
+  (let [db-name   (driver.u/workspace-isolation-namespace-name workspace)
+        username  (driver.u/workspace-isolation-user-name workspace)
+        conn-spec (sql-jdbc.conn/db->pooled-connection-spec (:id database))]
+    {:database (-> (jdbc/query conn-spec
+                               ["SELECT 1 FROM system.databases WHERE name = ?" db-name])
+                   seq boolean)
+     :user     (-> (jdbc/query conn-spec
+                               ["SELECT 1 FROM system.users WHERE name = ?" username])
+                   seq boolean)}))
+
+;; BigQuery uses service account impersonation - no schema/user resources to check
+(defmethod isolation-resources-exist? :bigquery-cloud-sdk
+  [_driver _database _workspace]
+  ;; BigQuery isolation uses service account impersonation, not schema/user creation
+  ;; Return nil to indicate resource checking is not applicable
+  nil)
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                       Isolation Lifecycle Tests                                                |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(deftest init-and-destroy-workspace-isolation-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (let [workspace {:id (rand-int 100000)}]
+      (let [result    (driver/init-workspace-isolation! driver/*driver* (mt/db) workspace)
+            workspace (merge workspace result)]
+        (try
+          (testing "init-workspace-isolation! creates resources and returns expected structure"
+            (is (contains? result :schema))
+            (is (contains? result :database_details))
+            (when-let [resources (isolation-resources-exist? driver/*driver* (mt/db) workspace)]
+              (testing "isolation resources were created"
+                (is (every? true? (vals resources))
+                    (str "All resources should be removed. Got: " resources)))))
+
+          (testing "destroy-workspace-isolation! removes resources"
+            (driver/destroy-workspace-isolation! driver/*driver* (mt/db) workspace)
+            (when-let [resources (isolation-resources-exist? driver/*driver* (mt/db) workspace)]
+              (testing "isolation resources were removed"
+                (is (every? false? (vals resources))
+                    (str "All resources should be removed. Got: " resources)))))
+
+          (testing "destroy is idempotent - calling twice should not error"
+            ;; destroy-workspace-isolation! is a void operation - just verify it doesn't throw
+            (driver/destroy-workspace-isolation! driver/*driver* (mt/db) workspace))
+
+          (finally
+            (try
+              (driver/destroy-workspace-isolation! driver/*driver* (mt/db) workspace)
+              (catch Exception _))))))))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                       Connection Swapping Tests                                                |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(deftest connection-swapping-integration-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "with-swapped-connection-details affects query execution"
+      (let [query-succeeded? (fn []
+                               (try
+                                 (mt/run-mbql-query venues {:limit 1})
+                                 true
+                                 (catch Exception _e
+                                   false)))]
+        (testing "queries work normally outside swap context"
+          (is (query-succeeded?)))
+
+        (testing "queries fail with invalid swapped credentials"
+          (driver/with-swapped-connection-details (mt/id) {:user "nonexistent_user_12345"
+                                                           :password "wrong_password"}
+            (is (not (query-succeeded?)))))))))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                       Grant Permissions Tests                                                  |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(deftest grant-workspace-read-access-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (let [workspace {:id (rand-int 100000)}
+          query-venues-table (fn []
+                               (try
+                                 (mt/run-mbql-query venues {:limit 1})
+                                 true
+                                 (catch Exception _e
+                                   false)))]
+      (try
+        (let [result           (driver/init-workspace-isolation! driver/*driver* (mt/db) workspace)
+              database_details (:database_details result)
+              venues-table     (t2/select-one :model/Table :id (mt/id :venues))
+              categories-table (t2/select-one :model/Table :id (mt/id :categories))
+              tables           [{:schema (:schema venues-table) :name (:name venues-table)}
+                                {:schema (:schema categories-table) :name (:name categories-table)}]
+              workspace        (merge workspace result)]
+
+          (testing "sanity check: isolated user cannot query tables before grant"
+            (driver/with-swapped-connection-details (mt/id) database_details
+              (is (false? (query-venues-table))
+                  "Query should fail before granting access")))
+
+          (testing "grant-workspace-read-access! succeeds for valid tables"
+            ;; grant-workspace-read-access! is a void operation - just verify it doesn't throw
+            (driver/grant-workspace-read-access! driver/*driver* (mt/db) workspace tables))
+
+          (testing "isolated user can query tables after grant"
+            (driver/with-swapped-connection-details (mt/id) database_details
+              (is (true? (query-venues-table))
+                  "Query should succeed after granting access"))))
+        (finally
+          (driver/destroy-workspace-isolation! driver/*driver* (mt/db) workspace))))))
+
+;;; +----------------------------------------------------------------------------------------------------------------+
+;;; |                                       Check Isolation Permissions Tests                                       |
+;;; +----------------------------------------------------------------------------------------------------------------+
+
+(deftest check-isolation-permissions-success-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "returns nil when connection has all required permissions"
+      (let [database     (mt/db)
+            venues-table (t2/select-one :model/Table :id (mt/id :venues))
+            test-table   {:schema (:schema venues-table)
+                          :name   (:name venues-table)}]
+        (is (nil? (driver/check-isolation-permissions
+                   driver/*driver*
+                   database
+                   test-table)))))))
+
+(deftest check-isolation-permissions-no-artifacts-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "leaves no artifacts after check due to transaction rollback"
+      (let [database       (mt/db)
+            test-workspace {:id   0
+                            :name "_mb_perm_check_"}
+            venues-table   (t2/select-one :model/Table :id (mt/id :venues))
+            test-table     {:schema (:schema venues-table)
+                            :name   (:name venues-table)}]
+        ;; Run the check
+        (driver/check-isolation-permissions driver/*driver* database test-table)
+        ;; Verify no artifacts remain - the test workspace should not have any resources
+        (when-let [resources (isolation-resources-exist? driver/*driver* database test-workspace)]
+          (is (every? false? (vals resources))
+              "No isolation resources should exist after permission check"))))))
+
+(deftest check-isolation-permissions-calls-all-methods-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "all isolation methods are called"
+      (let [init-called?    (atom false)
+            grant-called?   (atom false)
+            destroy-called? (atom false)]
+        (mt/with-dynamic-fn-redefs [driver/init-workspace-isolation!
+                                    (fn [_driver _database _workspace]
+                                      (reset! init-called? true)
+                                      {:schema "test_schema" :database_details {:user "test_user"}})
+                                    driver/grant-workspace-read-access!
+                                    (fn [_driver _database _workspace _tables]
+                                      (reset! grant-called? true)
+                                      nil)
+                                    driver/destroy-workspace-isolation!
+                                    (fn [_driver _database _workspace]
+                                      (reset! destroy-called? true)
+                                      nil)]
+          (let [test-table {:schema "public" :name "test_table"}
+                result     (driver/check-isolation-permissions driver/*driver* (mt/db) test-table)]
+            (is (nil? result) "should return nil on success")
+            (is @init-called? "init-workspace-isolation! should be called")
+            (is @grant-called? "grant-workspace-read-access! should be called when test-table provided")
+            (is @destroy-called? "destroy-workspace-isolation! should be called")))))))
+
+(deftest check-isolation-permissions-init-failure-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "returns error message when init fails"
+      (let [database   (mt/db)
+            test-table {:schema (mt/format-name "public")
+                        :name   (mt/format-name "venues")}]
+        (mt/with-dynamic-fn-redefs [driver/init-workspace-isolation!
+                                    (fn [_driver _database _workspace]
+                                      (throw (ex-info "permission denied" {:step :init})))]
+          (is (some? (driver/check-isolation-permissions driver/*driver* database test-table))
+              "Should return error message when init fails"))))))
+
+(deftest check-isolation-permissions-grant-failure-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "returns error message when grant fails"
+      (let [database   (mt/db)
+            test-table {:schema (mt/format-name "public")
+                        :name   (mt/format-name "venues")}]
+        (mt/with-dynamic-fn-redefs [driver/grant-workspace-read-access!
+                                    (fn [_driver _database _workspace _tables]
+                                      (throw (ex-info "permission denied" {:step :grant})))]
+          (is (some? (driver/check-isolation-permissions driver/*driver* database test-table))
+              "Should return error message when grant fails"))))))
+
+(deftest check-isolation-permissions-destroy-failure-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "returns error message when destroy fails"
+      (let [database   (mt/db)
+            test-table {:schema (mt/format-name "public")
+                        :name   (mt/format-name "venues")}]
+        (mt/with-dynamic-fn-redefs [driver/destroy-workspace-isolation!
+                                    (fn [_driver _database _workspace]
+                                      (throw (ex-info "permission denied" {:step :destroy})))]
+          (is (some? (driver/check-isolation-permissions driver/*driver* database test-table))
+              "Should return error message when destroy fails"))))))
+
+(deftest check-isolation-permissions-nil-table-test
+  (mt/test-drivers (mt/normal-drivers-with-feature :isolation)
+    (testing "returns nil when test-table is nil (skips grant step)"
+      (let [database (mt/db)]
+        (is (nil? (driver/check-isolation-permissions driver/*driver* database nil))
+            "Should succeed when test-table is nil")))))

test/metabase/driver/sql_jdbc/connection_test.clj

@@ -33,6 +33,7 @@
    [next.jdbc :as next.jdbc]
    [toucan2.core :as t2])
   (:import
+   (com.google.common.cache Cache)
    (org.h2.tools Server)))
 
 (set! *warn-on-reflection* true)
@@ -633,3 +634,198 @@
                                                        :auth-provider :aws-iam
                                                        :ssl true})))))
     (log/info "Skipping test: MB_MYSQL_AWS_IAM_TEST not set")))
+
+(defn- count-swapped-pools-for-db
+  "Count the number of swapped connection pools for a given database ID.
+  Since pools are keyed by [db-id, details-hash], we need to iterate through
+  all cache entries and count those matching the db-id."
+  [db-id]
+  (let [cache ^Cache @#'sql-jdbc.conn/swapped-connection-pools]
+    (count (filter (fn [[cached-db-id _details-hash]]
+                     (= cached-db-id db-id))
+                   (keys (.asMap cache))))))
+
+(defn- swap-cache-key
+  "Helper to compute the cache key for a swapped pool, matching the implementation in connection.clj.
+  Takes a db map (with :id, :engine, :details) and returns [db-id, jdbc-spec-hash]."
+  [db]
+  [(:id db) (#'sql-jdbc.conn/jdbc-spec-hash db)])
+
+(deftest with-swapped-connection-details-test
+  (testing "Swap connection details temporarily"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db    (mt/db)
+            db-id (u/the-id db)]
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+        (let [original-spec (sql-jdbc.conn/db->pooled-connection-spec db)]
+          (testing "Swap map is merged into details when creating connection"
+            (driver/with-swapped-connection-details db-id {:test-swap true}
+              (testing "spec is swapped"
+                (is (not= original-spec (sql-jdbc.conn/db->pooled-connection-spec db))))
+              (testing "Pool was created with swap in swapped pools cache"
+                (is (= 1 (count-swapped-pools-for-db db-id)))))))
+
+        (testing "Connection works normally outside swap scope"
+          (sql-jdbc.conn/invalidate-pool-for-db! db)
+          (let [spec (sql-jdbc.conn/db->pooled-connection-spec db)]
+            (is (some? spec))))))))
+
+(deftest different-swap-details-get-separate-pools-test
+  (testing "Different swap details for the same database get separate pools, identical details share pools"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db    (mt/db)
+            db-id (u/the-id db)]
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+        (let [pool-a-1 (atom nil)
+              pool-b   (atom nil)
+              pool-a-2 (atom nil)]
+          (testing "User A swaps with their credentials"
+            (driver/with-swapped-connection-details db-id {:user "user-a" :password "pass-a" :log-level 100}
+              (reset! pool-a-1 (sql-jdbc.conn/db->pooled-connection-spec db))
+              (is (= 1 (count-swapped-pools-for-db db-id)) "First swap creates one pool")))
+          (testing "User B swaps with different credentials"
+            (driver/with-swapped-connection-details db-id {:user "user-b" :password "pass-b" :log-level 99}
+              (reset! pool-b (sql-jdbc.conn/db->pooled-connection-spec db))
+              (is (= 2 (count-swapped-pools-for-db db-id)) "Different swap details create a second pool")
+              (is (not (identical? @pool-a-1 @pool-b)) "Different swap details return different pool instances")))
+          (testing "User A returns - should reuse their original pool (still in cache due to TTL)"
+            (driver/with-swapped-connection-details db-id {:user "user-a" :password "pass-a" :log-level 100}
+              (reset! pool-a-2 (sql-jdbc.conn/db->pooled-connection-spec db))
+              (is (= 2 (count-swapped-pools-for-db db-id)) "Identical swap details reuse existing pool")
+              (is (identical? @pool-a-1 @pool-a-2) "Identical swap details return the same pool instance"))))))))
+
+(deftest with-swapped-connection-details-nested-test
+  (testing "Nested swaps for the same database throw an exception"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db    (mt/db)
+            db-id (u/the-id db)]
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+        (driver/with-swapped-connection-details db-id {:outer-swap true}
+          (sql-jdbc.conn/db->pooled-connection-spec db)
+          (testing "Attempting nested swap for same database throws"
+            (is (thrown-with-msg?
+                 clojure.lang.ExceptionInfo
+                 #"Nested connection detail swaps are not supported"
+                 (driver/with-swapped-connection-details db-id {:inner-swap true}
+                   (sql-jdbc.conn/db->pooled-connection-spec db)))))))))
+
+  (testing "Different databases can have concurrent swaps"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db-1    (mt/db)
+            db-1-id (u/the-id db-1)]
+        ;; We can only test this with one db in most test setups, but the code path works
+        (driver/with-swapped-connection-details db-1-id {:swap-1 true}
+          ;; This would work for a different db-id
+          (is (some? (sql-jdbc.conn/db->pooled-connection-spec db-1))))))))
+
+(deftest invalidate-pool-clears-both-canonical-and-swapped-test
+  (testing "invalidate-pool-for-db! clears both canonical and swapped pools"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db    (mt/db)
+            db-id (u/the-id db)]
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+
+        (sql-jdbc.conn/db->pooled-connection-spec db)
+        (is (contains? @@#'sql-jdbc.conn/database-id->connection-pool db-id))
+
+        (driver/with-swapped-connection-details db-id {:test-swap true}
+          (sql-jdbc.conn/db->pooled-connection-spec db))
+        (is (= 1 (count-swapped-pools-for-db db-id)))
+
+        ;; Now invalidate - should clear both
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+
+        (testing "Canonical pool is cleared"
+          (is (not (contains? @@#'sql-jdbc.conn/database-id->connection-pool db-id))))
+        (testing "Swapped pool is cleared"
+          (is (= 0 (count-swapped-pools-for-db db-id))))))))
+
+(deftest swapped-pool-recreated-when-expired-test
+  (testing "Swapped pools are recreated when password expires"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db           (mt/db)
+            db-id        (u/the-id db)
+            swap-details {:test-swap true}
+            create-count (atom 0)]
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+        (with-redefs [sql-jdbc.conn/create-pool! (let [original @#'sql-jdbc.conn/create-pool!]
+                                                   (fn [db]
+                                                     (swap! create-count inc)
+                                                     (original db)))]
+          (driver/with-swapped-connection-details db-id swap-details
+            ;; First call creates a pool
+            (let [pool-1 (sql-jdbc.conn/db->pooled-connection-spec db)]
+              (is (= 1 @create-count))
+              (is (some? pool-1))
+
+              ;; Simulate password expiration by modifying the cached pool
+              ;; Cache key is [db-id, jdbc-spec-hash-of-swapped-db]
+              (let [cache             ^Cache @#'sql-jdbc.conn/swapped-connection-pools
+                    swapped-db        (update db :details merge swap-details)
+                    cache-key         (swap-cache-key swapped-db)
+                    ;; Use a fixed past timestamp (year 2020) to simulate expired password
+                    expired-timestamp 1577836800000]
+                (.put cache cache-key (assoc pool-1 :password-expiry-timestamp expired-timestamp)))
+
+              ;; Next call should detect invalid pool and recreate
+              (let [pool-2 (sql-jdbc.conn/db->pooled-connection-spec db)]
+                (is (= 2 @create-count) "Pool should have been recreated due to expired password")
+                (is (some? pool-2))
+                (is (not (identical? pool-1 pool-2)) "Should be a different pool instance")))))))))
+
+(deftest swapped-pool-recreated-when-tunnel-closed-test
+  (testing "Swapped pools are recreated when SSH tunnel is closed"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db           (mt/db)
+            db-id        (u/the-id db)
+            swap-details {:test-swap true}
+            create-count (atom 0)]
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+        (with-redefs [sql-jdbc.conn/create-pool! (let [original @#'sql-jdbc.conn/create-pool!]
+                                                   (fn [db]
+                                                     (swap! create-count inc)
+                                                     (original db)))]
+          (driver/with-swapped-connection-details db-id swap-details
+            ;; First call creates a pool
+            (let [pool-1 (sql-jdbc.conn/db->pooled-connection-spec db)]
+              (is (= 1 @create-count))
+              (is (some? pool-1))
+
+              ;; Simulate closed tunnel by modifying the cached pool
+              ;; We add a tunnel-session that reports as closed
+              ;; Cache key is [db-id, jdbc-spec-hash-of-swapped-db]
+              (let [cache      ^Cache @#'sql-jdbc.conn/swapped-connection-pools
+                    swapped-db (update db :details merge swap-details)
+                    cache-key  (swap-cache-key swapped-db)]
+                (.put cache cache-key (assoc pool-1 :tunnel-session :mock-closed-session)))
+
+              ;; Mock ssh-tunnel-open? to return false for our mock session
+              (with-redefs [ssh/ssh-tunnel-open? (fn [pool-spec]
+                                                   (not= :mock-closed-session (:tunnel-session pool-spec)))]
+                ;; Next call should detect invalid pool and recreate
+                (let [pool-2 (sql-jdbc.conn/db->pooled-connection-spec db)]
+                  (is (= 2 @create-count) "Pool should have been recreated due to closed tunnel")
+                  (is (some? pool-2))
+                  (is (not (identical? pool-1 pool-2)) "Should be a different pool instance"))))))))))
+
+(deftest swapped-pool-reused-when-valid-test
+  (testing "Valid swapped pools are reused without recreation"
+    (mt/test-drivers (mt/normal-driver-select {:+parent :sql-jdbc})
+      (let [db           (mt/db)
+            db-id        (u/the-id db)
+            create-count (atom 0)]
+        (sql-jdbc.conn/invalidate-pool-for-db! db)
+        (with-redefs [sql-jdbc.conn/create-pool! (let [original @#'sql-jdbc.conn/create-pool!]
+                                                   (fn [db]
+                                                     (swap! create-count inc)
+                                                     (original db)))]
+          (driver/with-swapped-connection-details db-id {:test-swap true}
+            ;; First call creates a pool
+            (let [pool-1 (sql-jdbc.conn/db->pooled-connection-spec db)]
+              (is (= 1 @create-count))
+              (is (some? pool-1))
+
+              ;; Second call should reuse the same pool
+              (let [pool-2 (sql-jdbc.conn/db->pooled-connection-spec db)]
+                (is (= 1 @create-count) "Pool should be reused, not recreated")
+                (is (identical? pool-1 pool-2) "Should be the same pool instance")))))))))

test/metabase/driver_test.clj

@@ -480,3 +480,47 @@
 (deftest deps-flags-when-supported-driver-is-not-covered-test
   (is (thrown-with-msg? clojure.lang.ExceptionInfo #"Database that supports :dependencies/native does not provide an implementation of driver/native-query-deps"
                         (driver/native-query-deps ::mock-deps-driver nil nil))))
+
+(deftest ^:parallel maybe-swap-details-test
+  (testing "maybe-swap-details merges swap map into details"
+    (driver/with-swapped-connection-details 1 {:user "swap-user" :password "swap-pass"}
+      (is (= {:host "localhost" :user "swap-user" :password "swap-pass"}
+             (driver/maybe-swap-details 1 {:host "localhost" :user "original-user" :password "original-pass"})))))
+
+  (testing "maybe-swap-details returns details unchanged when no swap exists"
+    (driver/with-swapped-connection-details 1 {:user "swap-user"}
+      (is (= {:host "localhost" :user "original-user"}
+             (driver/maybe-swap-details 2 {:host "localhost" :user "original-user"})))))
+
+  (testing "maybe-swap-details supports deep merge for nested maps"
+    (driver/with-swapped-connection-details 1 {:ssl {:key-store-password "new-pass"}}
+      (is (= {:host "localhost" :ssl {:enabled true :key-store-password "new-pass"}}
+             (driver/maybe-swap-details 1 {:host "localhost" :ssl {:enabled true :key-store-password "old-pass"}})))))
+
+  (testing "deep merge adds new keys to nested maps"
+    (driver/with-swapped-connection-details 1 {:ssl {:new-key "new-value"}}
+      (is (= {:host "localhost" :ssl {:enabled true :new-key "new-value"}}
+             (driver/maybe-swap-details 1 {:host "localhost" :ssl {:enabled true}})))))
+
+  (testing "deep merge replaces nested map with non-map value"
+    (driver/with-swapped-connection-details 1 {:ssl "disabled"}
+      (is (= {:host "localhost" :ssl "disabled"}
+             (driver/maybe-swap-details 1 {:host "localhost" :ssl {:enabled true :key-store "path"}})))))
+
+  (testing "deep merge adds nested map where none existed"
+    (driver/with-swapped-connection-details 1 {:ssl {:enabled true}}
+      (is (= {:host "localhost" :ssl {:enabled true}}
+             (driver/maybe-swap-details 1 {:host "localhost"})))))
+
+  (testing "deep merge works with multiple levels of nesting"
+    (driver/with-swapped-connection-details 1 {:advanced {:ssl {:cert {:path "/new/path"}}}}
+      (is (= {:host "localhost" :advanced {:timeout 30 :ssl {:enabled true :cert {:path "/new/path"}}}}
+             (driver/maybe-swap-details 1 {:host "localhost" :advanced {:timeout 30 :ssl {:enabled true :cert {:path "/old/path"}}}})))))
+
+  (testing "nested swaps for the same database throw an exception"
+    (driver/with-swapped-connection-details 1 {:user "outer"}
+      (is (thrown-with-msg?
+           clojure.lang.ExceptionInfo
+           #"Nested connection detail swaps are not supported"
+           (driver/with-swapped-connection-details 1 {:user "inner"}
+             nil))))))

test/metabase/test/data/h2.clj

@@ -34,11 +34,8 @@
   (defmethod sql.tx/field-base-type->sql-type [:h2 base-type] [_ _] database-type))
 
 (defmethod tx/dbdef->connection-details :h2
-  [_driver context dbdef]
-  {:db (str "mem:" (tx/escaped-database-name dbdef) (when (= context :db)
-                                                      ;; Return details with the GUEST user added so SQL queries are
-                                                      ;; allowed.
-                                                      ";USER=GUEST;PASSWORD=guest"))})
+  [_driver _context dbdef]
+  {:db (str "mem:" (tx/escaped-database-name dbdef))})
 
 (defmethod sql.tx/pk-sql-type :h2 [_] "BIGINT AUTO_INCREMENT")