check if queue already exists (#28)

* fix .table bug

* unused import

* add upgrade script

Author: ChuckHend

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "vectorize"
-version = "0.6.0"
+version = "0.6.1"
 edition = "2021"
 publish = false
 

Trunk.toml

@@ -6,7 +6,7 @@ description = "The simplest way to orchestrate vector search on Postgres."
 homepage = "https://github.com/tembo-io/pg_vectorize"
 documentation = "https://github.com/tembo-io/pg_vectorize"
 categories = ["orchestration", "machine_learning"]
-version = "0.6.0"
+version = "0.6.1"
 
 [build]
 postgres_version = "15"

sql/vectorize--0.6.0--0.6.1.sql

@@ -130,9 +130,7 @@ fn table(
     // TODO: first batch update
     // initialize cron
     let _ = init::init_cron(&schedule, &job_name); // handle this error
-    Ok(format!(
-        "{schema}.{table}.{columns:?}.{transformer}.{search_alg}"
-    ))
+    Ok(format!("Successfully created job: {job_name}"))
 }
 
 #[pg_extern]

src/api.rs

@@ -2,7 +2,7 @@ use crate::{query::check_input, types, types::TableMethod, types::Transformer};
 use pgrx::prelude::*;
 use std::collections::HashMap;
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 use lazy_static::lazy_static;
 
 lazy_static! {
@@ -18,12 +18,21 @@ lazy_static! {
 
 pub fn init_pgmq(transformer: &Transformer) -> Result<()> {
     let qname = QUEUE_MAPPING.get(transformer).expect("invalid transformer");
-    let ran: Result<_, spi::Error> = Spi::connect(|mut c| {
-        let _r = c.update(&format!("SELECT pgmq.create('{qname}');"), None, None)?;
-        Ok(())
-    });
-    if let Err(e) = ran {
-        error!("error creating job queue: {}", e);
+    // check if queue already created:
+    let queue_exists: bool = Spi::get_one(&format!(
+        "SELECT EXISTS (SELECT 1 FROM pgmq.meta WHERE queue_name = '{qname}');",
+    ))?
+    .context("error checking if queue exists")?;
+    if queue_exists {
+        return Ok(());
+    } else {
+        let ran: Result<_, spi::Error> = Spi::connect(|mut c| {
+            let _r = c.update(&format!("SELECT pgmq.create('{qname}');"), None, None)?;
+            Ok(())
+        });
+        if let Err(e) = ran {
+            error!("error creating job queue: {}", e);
+        }
     }
     Ok(())
 }

src/init.rs

@@ -2,7 +2,7 @@ pub mod common {
     use log::LevelFilter;
     use sqlx::postgres::{PgConnectOptions, PgPoolOptions};
     use sqlx::ConnectOptions;
-    use sqlx::{FromRow, Pool, Postgres, Row};
+    use sqlx::{Pool, Postgres, Row};
     use url::{ParseError, Url};
 
     pub async fn connect(url: &str) -> Pool<Postgres> {