make the advisory lock prefix the oid

Author: cevian

CHANGELOG.md

@@ -4,7 +4,7 @@
 
 ## 0.0.2
 
-Add ability to info start and end date from filters
+Add ability to infer start and end date from filters
 
 
 ## 0.0.1

nbs/01_pgvectorizer.ipynb

@@ -72,7 +72,7 @@
     "        self.trigger_name_fn = client.QueryBuilder._quote_ident(trigger_name_fn) \n",
     "\n",
     "\n",
-    "    def register(self):\n",
+    "    def register(self):        \n",
     "        with psycopg2.connect(self.service_url) as conn:\n",
     "            with conn.cursor() as cursor:\n",
     "                cursor.execute(f\"\"\"\n",
@@ -81,9 +81,7 @@
     "                table_exists = cursor.fetchone()[0]\n",
     "                if table_exists:\n",
     "                    return\n",
-    "        \n",
-    "        with psycopg2.connect(self.service_url) as conn:\n",
-    "            with conn.cursor() as cursor:\n",
+    "                \n",
     "                cursor.execute(f\"\"\"\n",
     "                    CREATE TABLE {self.schema_name}.{self.work_queue_table_name} (\n",
     "                        id int\n",
@@ -110,21 +108,26 @@
     "                    INSERT INTO {self.schema_name}.{self.work_queue_table_name} SELECT {self.id_column_name} FROM {self.schema_name}.{self.table_name};\n",
     "                \"\"\")\n",
     "\n",
-    "    def process(self, embed_and_write_cb, batch_size:int=10, advisory_prefix=47859, autoregister=True):\n",
+    "    def process(self, embed_and_write_cb, batch_size:int=10, autoregister=True):\n",
     "        if autoregister:\n",
     "            self.register()\n",
     "            \n",
     "        with psycopg2.connect(self.service_url) as conn:\n",
     "            with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:\n",
     "                cursor.execute(f\"\"\"\n",
+    "                    SELECT to_regclass('{self.schema_name}.{self.work_queue_table_name}')::oid; \n",
+    "                \"\"\")\n",
+    "                table_oid = cursor.fetchone()[0]\n",
+    "            \n",
+    "                cursor.execute(f\"\"\"\n",
     "                    WITH selected_rows AS (\n",
     "                        SELECT id\n",
     "                        FROM {self.schema_name}.{self.work_queue_table_name}\n",
     "                        LIMIT {int(batch_size)}\n",
     "                        FOR UPDATE SKIP LOCKED\n",
     "                    ), \n",
     "                    locked_items AS (\n",
-    "                        SELECT id, pg_try_advisory_xact_lock({int(advisory_prefix)}, id) AS locked\n",
+    "                        SELECT id, pg_try_advisory_xact_lock({int(table_oid)}, id) AS locked\n",
     "                        FROM (SELECT DISTINCT id FROM selected_rows ORDER BY id) as ids\n",
     "                    ),\n",
     "                    deleted_rows AS (\n",

timescale_vector/pgvectorizer.py

@@ -45,7 +45,7 @@ def __init__(self,
         self.trigger_name_fn = client.QueryBuilder._quote_ident(trigger_name_fn) 
 
 
-    def register(self):
+    def register(self):        
         with psycopg2.connect(self.service_url) as conn:
             with conn.cursor() as cursor:
                 cursor.execute(f"""
@@ -54,9 +54,7 @@ def register(self):
                 table_exists = cursor.fetchone()[0]
                 if table_exists:
                     return
-        
-        with psycopg2.connect(self.service_url) as conn:
-            with conn.cursor() as cursor:
+                
                 cursor.execute(f"""
                     CREATE TABLE {self.schema_name}.{self.work_queue_table_name} (
                         id int
@@ -83,12 +81,17 @@ def register(self):
                     INSERT INTO {self.schema_name}.{self.work_queue_table_name} SELECT {self.id_column_name} FROM {self.schema_name}.{self.table_name};
                 """)
 
-    def process(self, embed_and_write_cb, batch_size:int=10, advisory_prefix=47859, autoregister=True):
+    def process(self, embed_and_write_cb, batch_size:int=10, autoregister=True):
         if autoregister:
             self.register()
 
         with psycopg2.connect(self.service_url) as conn:
             with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
+                cursor.execute(f"""
+                    SELECT to_regclass('{self.schema_name}.{self.work_queue_table_name}')::oid; 
+                """)
+                table_oid = cursor.fetchone()[0]
+            
                 cursor.execute(f"""
                     WITH selected_rows AS (
                         SELECT id
@@ -97,7 +100,7 @@ def process(self, embed_and_write_cb, batch_size:int=10, advisory_prefix=47859,
                         FOR UPDATE SKIP LOCKED
                     ), 
                     locked_items AS (
-                        SELECT id, pg_try_advisory_xact_lock({int(advisory_prefix)}, id) AS locked
+                        SELECT id, pg_try_advisory_xact_lock({int(table_oid)}, id) AS locked
                         FROM (SELECT DISTINCT id FROM selected_rows ORDER BY id) as ids
                     ),
                     deleted_rows AS (