Add pgmq.detach_archive function

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "pgmq"
-version = "0.29.0"
+version = "0.30.0"
 edition = "2021"
 authors = ["Tembo.io"]
 description = "Postgres extension for PGMQ"

sql/pgmq--0.29.0--0.30.0.sql

@@ -0,0 +1,9 @@
+-- New function, copied from schema
+-- src/api.rs:26
+-- pgmq::api::detach_archive
+CREATE  FUNCTION pgmq."detach_archive"(
+	"queue_name" TEXT /* alloc::string::String */
+) RETURNS VOID /* core::result::Result<(), pgmq::errors::PgmqExtError> */
+STRICT
+LANGUAGE c /* Rust */
+AS 'MODULE_PATHNAME', 'pgmq_detach_archive_wrapper';

src/api.rs

@@ -8,9 +8,9 @@ use crate::partition;
 use crate::partition::PARTMAN_SCHEMA;
 
 use pgmq_core::{
-    query::{destroy_queue, init_queue},
+    query::{destroy_queue, init_queue, unassign_archive},
     types::{PGMQ_SCHEMA, QUEUE_PREFIX},
-    util::check_input,
+    util::{check_input, CheckedName},
 };
 
 #[pg_extern(name = "drop_queue")]
@@ -22,6 +22,15 @@ fn pgmq_drop_queue(
     Ok(true)
 }
 
+#[pg_extern(name = "detach_archive")]
+fn pgmq_detach_archive(queue_name: String) -> Result<(), PgmqExtError> {
+    let query = unassign_archive(CheckedName::new(&queue_name)?)?;
+    Spi::connect(|mut client| {
+        client.update(query.as_str(), None, None)?;
+        Ok(())
+    })
+}
+
 pub fn delete_queue(queue_name: String, partitioned: bool) -> Result<(), PgmqExtError> {
     // TODO: we should keep track whether queue is partitioned in pgmq_meta
     // then read that to determine we want to delete the part_config entries

tests/integration_tests.rs

@@ -632,6 +632,55 @@ async fn test_transaction_read() {
 
     assert!(read_msg3.is_some());
 }
+// Integration tests are ignored by default
+#[ignore]
+#[tokio::test]
+async fn test_detach_archive() {
+    let conn = init_database().await;
+    let queue_name = "detach_archive_queue";
+    create_queue(&queue_name.to_string(), &conn).await;
+
+    // Without detach, archive is dropped with the extension
+    let _ = sqlx::query(&format!("DROP EXTENSION pgmq CASCADE"))
+        .fetch_one(&conn)
+        .await;
+
+    let table_exists = sqlx::query(&format!(
+        "select from pg_tables where schemaname = 'pgmq' and tablename = 'a_{queue_name}'"
+    ))
+    .fetch_optional(&conn)
+    .await
+    .unwrap();
+
+    assert!(table_exists.is_none());
+
+    // With detach, archive remains
+    let _ = sqlx::query(&format!("CREATE EXTENSION pgmq"))
+        .fetch_one(&conn)
+        .await;
+
+    create_queue(&queue_name.to_string(), &conn).await;
+
+    sqlx::query(&format!(
+        "select from {PGMQ_SCHEMA}.detach_archive('{queue_name}')"
+    ))
+    .fetch_one(&conn)
+    .await
+    .unwrap();
+
+    let _ = sqlx::query(&format!("DROP EXTENSION pgmq CASCADE"))
+        .fetch_one(&conn)
+        .await;
+
+    let table_exists = sqlx::query(&format!(
+        "select from pg_tables where schemaname = 'pgmq' and tablename = 'a_{queue_name}'"
+    ))
+    .fetch_optional(&conn)
+    .await
+    .unwrap();
+
+    assert!(table_exists.is_some());
+}
 
 async fn create_queue(queue_name: &String, conn: &Pool<Postgres>) {
     sqlx::query(&format!("select from {PGMQ_SCHEMA}.create('{queue_name}')"))