feat(postgres): support editing table owners

crates/dbx-core/src/db/postgres.rs

@@ -7190,6 +7190,14 @@ pub async fn list_owners(pool: &Pool, schema: &str) -> Result<Vec<OwnerInfo>, St
         .collect())
 }
 
+pub async fn get_table_owner(pool: &Pool, schema: &str, table: &str) -> Result<Option<String>, String> {
+    let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
+    let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = [&schema, &table];
+    let rows = postgres_query_cached(&client, POSTGRES_TABLE_OWNER_SQL, &params).await.map_err(pg_error_to_string)?;
+
+    Ok(rows.first().map(|row| pg_row_try_string(row, 0)).filter(|owner| !owner.is_empty()))
+}
+
 pub async fn get_table_access(pool: &Pool, schema: &str, table: &str) -> Result<PostgresTableAccessInfo, String> {
     let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
     let params: [&(dyn tokio_postgres::types::ToSql + Sync); 2] = [&schema, &table];

crates/dbx-core/src/schema.rs

@@ -7049,6 +7049,25 @@ pub async fn list_owners_core(
     .await
 }
 
+pub async fn get_table_owner_core(
+    state: &AppState,
+    connection_id: &str,
+    database: &str,
+    schema: &str,
+    table: &str,
+) -> Result<Option<String>, String> {
+    retry_metadata_connection(state, connection_id, Some(database), || async {
+        let pool_key = state.get_or_create_metadata_pool_for_session(connection_id, Some(database), None).await?;
+        let pool = clone_metadata_pool(state, &pool_key).await.ok_or("Pool not found")?;
+
+        match &pool {
+            PoolKind::Postgres(p) => db::postgres::get_table_owner(p, schema, table).await,
+            _ => Ok(None),
+        }
+    })
+    .await
+}
+
 /// Whether to widen or normalize a single-table DDL fetch for its caller.
 ///
 /// Database export and table transfer render one relation at a time because

crates/dbx-core/src/table_structure_sql.rs

@@ -6,6 +6,7 @@ mod create_table;
 mod dialect;
 mod foreign_keys;
 mod indexes;
+mod owner;
 mod sqlite_rebuild;
 mod triggers;
 mod types;
@@ -17,6 +18,7 @@ mod tests;
 
 pub use column_alter::build_single_column_alter_sql;
 pub use create_table::build_create_table_sql;
+pub use owner::build_table_owner_change_sql;
 pub use sqlite_rebuild::{apply_sqlite_table_structure_change, preview_sqlite_table_structure_change};
 pub use types::*;
 

crates/dbx-core/src/table_structure_sql/owner.rs

@@ -0,0 +1,79 @@
+use super::dialect::StructureDialect;
+use super::types::{TableOwnerChangeSqlOptions, TableStructureSqlResult};
+use super::util::{qualified_table, quote_ident};
+use crate::models::connection::DatabaseType;
+
+pub fn build_table_owner_change_sql(options: TableOwnerChangeSqlOptions) -> TableStructureSqlResult {
+    let owner = &options.owner;
+    if owner == &options.original_owner {
+        return TableStructureSqlResult { statements: Vec::new(), warnings: Vec::new() };
+    }
+    if owner.is_empty() {
+        return TableStructureSqlResult {
+            statements: Vec::new(),
+            warnings: vec!["Table owner cannot be empty.".to_string()],
+        };
+    }
+    if options.database_type != Some(DatabaseType::Postgres) {
+        return TableStructureSqlResult {
+            statements: Vec::new(),
+            warnings: vec!["Changing the table owner is currently supported only for PostgreSQL.".to_string()],
+        };
+    }
+
+    let table = qualified_table(StructureDialect::Postgres, options.schema.as_deref(), &options.table_name);
+    let owner = quote_ident(StructureDialect::Postgres, owner);
+    TableStructureSqlResult { statements: vec![format!("ALTER TABLE {table} OWNER TO {owner};")], warnings: Vec::new() }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn options(owner: &str, original_owner: &str) -> TableOwnerChangeSqlOptions {
+        TableOwnerChangeSqlOptions {
+            database_type: Some(DatabaseType::Postgres),
+            schema: Some("app schema".to_string()),
+            table_name: "order\"items".to_string(),
+            owner: owner.to_string(),
+            original_owner: original_owner.to_string(),
+        }
+    }
+
+    #[test]
+    fn builds_quoted_postgres_owner_change() {
+        let result = build_table_owner_change_sql(options("reporting\"role", "app_user"));
+        assert!(result.warnings.is_empty());
+        assert_eq!(
+            result.statements,
+            vec!["ALTER TABLE \"app schema\".\"order\"\"items\" OWNER TO \"reporting\"\"role\";".to_string()]
+        );
+    }
+
+    #[test]
+    fn preserves_whitespace_in_postgres_role_names() {
+        let changed = build_table_owner_change_sql(options(" app_user ", "app_user"));
+        assert!(changed.warnings.is_empty());
+        assert_eq!(
+            changed.statements,
+            vec!["ALTER TABLE \"app schema\".\"order\"\"items\" OWNER TO \" app_user \";".to_string()]
+        );
+
+        let unchanged = build_table_owner_change_sql(options(" app_user ", " app_user "));
+        assert!(unchanged.statements.is_empty());
+        assert!(unchanged.warnings.is_empty());
+    }
+
+    #[test]
+    fn rejects_empty_or_unsupported_owner_changes() {
+        let empty = build_table_owner_change_sql(options("", "app_user"));
+        assert!(empty.statements.is_empty());
+        assert_eq!(empty.warnings, vec!["Table owner cannot be empty."]);
+
+        let mut unsupported = options("reporting", "app_user");
+        unsupported.database_type = Some(DatabaseType::Mysql);
+        let unsupported = build_table_owner_change_sql(unsupported);
+        assert!(unsupported.statements.is_empty());
+        assert_eq!(unsupported.warnings, vec!["Changing the table owner is currently supported only for PostgreSQL."]);
+    }
+}

crates/dbx-core/src/table_structure_sql/types.rs

@@ -231,6 +231,20 @@ pub struct TableStructureSqlResult {
     pub warnings: Vec<String>,
 }
 
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct TableOwnerChangeSqlOptions {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub database_type: Option<DatabaseType>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub schema: Option<String>,
+    pub table_name: String,
+    #[serde(default)]
+    pub owner: String,
+    #[serde(default)]
+    pub original_owner: String,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct SqliteTableStructurePreview {