feat(datatables): Ducklake catalogs on the external instance cluster

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-16ef314b575fde3db031d598f6bbcb462d8903fc
+a7a484b4ce50d2ceff2051cf412bf5c2f563897e

backend/windmill-api-workspaces/src/workspaces.rs

@@ -3768,22 +3768,37 @@ async fn edit_ducklake_config(
     let old_ducklakes: HashMap<String, Ducklake> =
         serde_json::from_value(old_ducklakes).unwrap_or_default();
 
-    // Check that non-superadmins are not abusing Instance databases
-    if !is_superadmin {
-        for (name, dl) in new_config.settings.ducklakes.iter() {
-            if dl.catalog.resource_type == DucklakeCatalogResourceType::Instance {
-                let old_dl = old_ducklakes.get(name);
-                if old_dl.is_none()
-                    || old_dl.unwrap().catalog.resource_type
-                        != DucklakeCatalogResourceType::Instance
-                    || old_dl.unwrap().catalog.resource_path != dl.catalog.resource_path
-                {
-                    return Err(Error::BadRequest(
-                        "Only superadmins can create or modify ducklakes with Instance databases"
-                            .to_string(),
-                    ));
-                }
-            }
+    // Check that non-superadmins are not abusing Instance databases. An unchanged catalog is left
+    // alone either way, so a downgraded instance can still save lakes that already name an
+    // external instance database.
+    for (name, dl) in new_config.settings.ducklakes.iter() {
+        let kind = &dl.catalog.resource_type;
+        if !matches!(
+            kind,
+            DucklakeCatalogResourceType::Instance | DucklakeCatalogResourceType::ExternalInstance
+        ) {
+            continue;
+        }
+        let unchanged = old_ducklakes.get(name).is_some_and(|old| {
+            &old.catalog.resource_type == kind
+                && old.catalog.resource_path == dl.catalog.resource_path
+        });
+        if unchanged {
+            continue;
+        }
+        if *kind == DucklakeCatalogResourceType::ExternalInstance {
+            windmill_common::external_instance_pg::ensure_external_instance_available()?;
+            windmill_common::external_instance_pg::ensure_external_instance_database_registered(
+                &mut tx,
+                &dl.catalog.resource_path,
+            )
+            .await?;
+        }
+        if !is_superadmin {
+            return Err(Error::BadRequest(
+                "Only superadmins can create or modify ducklakes with Instance databases"
+                    .to_string(),
+            ));
         }
     }
 

backend/windmill-api-workspaces/src/workspaces_extra.rs

@@ -1795,7 +1795,17 @@ async fn resolve_fork_catalog_pg(
             "ducklake://{ducklake_name}: malformed registry catalog identity `{catalog}`"
         ))
     })?;
-    let catalog_resource = if resource_type == "instance" {
+    let catalog_resource = if resource_type == "external_instance" {
+        serde_json::to_value(
+            windmill_common::external_instance_pg::external_instance_connection_unchecked(
+                db,
+                resource_path,
+                false,
+            )
+            .await?,
+        )
+        .map_err(|e| Error::internal_err(format!("serializing pg creds: {e}")))?
+    } else if resource_type == "instance" {
         let mut pg_creds = windmill_common::PgDatabase::parse_uri(
             &windmill_common::get_database_url().await?.as_str().await,
         )?;

backend/windmill-api/openapi.yaml

@@ -36063,6 +36063,7 @@ components:
                       - postgresql
                       - mysql
                       - instance
+                      - external_instance
                   resource_path:
                     type: string
                 required:

backend/windmill-common/src/external_instance_pg.rs

@@ -129,7 +129,7 @@ pub async fn external_instance_databases(db: &DB) -> Result<BTreeMap<String, Cus
     Ok(read_external_instance_pg_state(db).await?.databases)
 }
 
-/// The workspaces whose data tables name each database on the external cluster.
+/// The workspaces whose data tables or Ducklake catalogs name each database on the external cluster.
 pub async fn external_instance_database_usages<'c>(
     db: impl sqlx::PgExecutor<'c>,
 ) -> Result<BTreeMap<String, BTreeSet<String>>> {
@@ -142,7 +142,17 @@ pub async fn external_instance_database_usages<'c>(
                  ELSE '{}'::jsonb END
          ) AS dt(k, entry)
          WHERE entry->'database'->>'resource_type' = 'external_instance'
-           AND entry->'database'->>'resource_path' IS NOT NULL",
+           AND entry->'database'->>'resource_path' IS NOT NULL
+         UNION ALL
+         SELECT ws.workspace_id, entry->'catalog'->>'resource_path'
+         FROM workspace_settings ws
+         CROSS JOIN LATERAL jsonb_each(
+             CASE WHEN jsonb_typeof(ws.ducklake->'ducklakes') = 'object'
+                 THEN ws.ducklake->'ducklakes'
+                 ELSE '{}'::jsonb END
+         ) AS dl(k, entry)
+         WHERE entry->'catalog'->>'resource_type' = 'external_instance'
+           AND entry->'catalog'->>'resource_path' IS NOT NULL",
     )
     .fetch_all(db)
     .await?;
@@ -173,7 +183,7 @@ pub async fn ensure_external_instance_pg_removable(db: &DB) -> Result<()> {
         .join(", ");
     Err(Error::BadRequest(format!(
         "The external instance cluster still holds databases in use ({names}). Drop them and \
-         repoint the data tables using them before removing {EXTERNAL_INSTANCE_PG_SETTING}."
+         repoint the data tables and Ducklake catalogs using them before removing {EXTERNAL_INSTANCE_PG_SETTING}."
     )))
 }
 

backend/windmill-common/src/workspaces.rs

@@ -2096,6 +2096,10 @@ pub enum DucklakeCatalogResourceType {
     Postgresql,
     Mysql,
     Instance,
+    /// On the external instance cluster ([`crate::external_instance_pg`]). Enterprise Edition.
+    #[serde(rename = "external_instance")]
+    #[strum(serialize = "external_instance")]
+    ExternalInstance,
 }
 
 #[derive(Deserialize, Serialize)]
@@ -2623,7 +2627,16 @@ async fn ducklake_conn_data(
     let ducklake = serde_json::from_value::<Ducklake>(ducklake)?;
 
     let catalog_resource =
-        if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance {
+        if ducklake.catalog.resource_type == DucklakeCatalogResourceType::ExternalInstance {
+            let pg_creds = crate::external_instance_pg::external_instance_connection_unchecked(
+                db,
+                &ducklake.catalog.resource_path,
+                false,
+            )
+            .await?;
+            serde_json::to_value(&pg_creds)
+                .map_err(|e| Error::internal_err(format!("Error serializing pg creds: {}", e)))?
+        } else if ducklake.catalog.resource_type == DucklakeCatalogResourceType::Instance {
             let mut pg_creds = PgDatabase::parse_uri(&get_database_url().await?.as_str().await)?;
             pg_creds.dbname = ducklake.catalog.resource_path.clone();
             pg_creds.user = Some("custom_instance_user".to_string());

backend/windmill-worker/src/duckdb_executor.rs

@@ -2391,7 +2391,9 @@ async fn transform_attach_ducklake(
         format!(", {}", user_extra_args)
     };
     let db_type = match ducklake.catalog.resource_type {
-        DucklakeCatalogResourceType::Instance => "postgres",
+        DucklakeCatalogResourceType::Instance | DucklakeCatalogResourceType::ExternalInstance => {
+            "postgres"
+        }
         _ => ducklake.catalog.resource_type.as_ref(),
     };
 
@@ -2507,7 +2509,8 @@ fn fork_defer_statements(
             hidden_passwords.lock().unwrap().push(pwd.to_string());
         }
         let db_type = match a.catalog.resource_type {
-            DucklakeCatalogResourceType::Instance => "postgres",
+            DucklakeCatalogResourceType::Instance
+            | DucklakeCatalogResourceType::ExternalInstance => "postgres",
             _ => a.catalog.resource_type.as_ref(),
         };
         stmts.push(get_attach_db_install_str(db_type)?.to_string());