chore(pgmq): update

ansible/files/postgresql_extension_custom_scripts/pgmq/after-create.sql

@@ -18,25 +18,29 @@ begin
     this update is backwards compatible with version 1.4.4 but should be removed once we're on
     physical backups everywhere
 */
--- Detach and delete the official function
-if extversion = '1.4.4' then
-  alter extension pgmq drop function pgmq.drop_queue;
-  drop function pgmq.drop_queue;
-else -- 1.5.1+
-  alter extension pgmq drop function pgmq.drop_queue(TEXT);
-  drop function pgmq.drop_queue(TEXT);
-end if;
+  -- detach and drop both historical drop_queue signatures (1.4.4 only ever
+  -- has (text, boolean); 1.5.0+ has both (text) and (text, boolean))
+  begin
+    alter extension pgmq drop function pgmq.drop_queue(text);
+  exception when others then null;
+  end;
+  begin
+    alter extension pgmq drop function pgmq.drop_queue(text, boolean);
+  exception when others then null;
+  end;
+
+  drop function if exists pgmq.drop_queue(text);
+  drop function if exists pgmq.drop_queue(text, boolean);
 
 -- Create and reattach the patched function
-CREATE FUNCTION pgmq.drop_queue(queue_name TEXT)
+CREATE FUNCTION pgmq.drop_queue(queue_name TEXT, partitioned BOOLEAN DEFAULT FALSE)
 RETURNS BOOLEAN AS $func$
 DECLARE
     qtable TEXT := pgmq.format_table_name(queue_name, 'q');
     qtable_seq TEXT := qtable || '_msg_id_seq';
     fq_qtable TEXT := 'pgmq.' || qtable;
     atable TEXT := pgmq.format_table_name(queue_name, 'a');
     fq_atable TEXT := 'pgmq.' || atable;
-    partitioned BOOLEAN;
 BEGIN
     EXECUTE FORMAT(
         $QUERY$
@@ -140,11 +144,7 @@ BEGIN
 END;
 $func$ LANGUAGE plpgsql;
 
-if extversion = '1.4.4' then
-  alter extension pgmq add function pgmq.drop_queue;
-else -- 1.5.1+
-  alter extension pgmq add function pgmq.drop_queue(TEXT);
-end if;
+  alter extension pgmq add function pgmq.drop_queue(text, boolean);
 
 
   update pg_extension set extowner = 'postgres'::regrole where extname = 'pgmq';

nix/ext/pgmq/default.nix

@@ -36,7 +36,7 @@ let
       inherit pname version;
       buildInputs = [ postgresql ];
       src = fetchFromGitHub {
-        owner = "tembo-io";
+        owner = "pgmq";
         repo = pname;
         rev = "v${version}";
         inherit hash;
@@ -69,19 +69,6 @@ let
             echo "default_version = '${version}'"
             cat $out/share/postgresql/extension/${pname}--${version}.control
           } > $out/share/postgresql/extension/${pname}.control
-          cat >> sql/pgmq--1.5.0--1.5.1.sql <<EOF
-
-        CREATE FUNCTION pgmq._extension_exists(extension_name TEXT)
-        RETURNS BOOLEAN
-        LANGUAGE SQL
-        AS \$\$
-        SELECT EXISTS (
-            SELECT 1
-            FROM pg_extension
-            WHERE extname = extension_name
-        )
-        \$\$;
-        EOF
           cp sql/*.sql $out/share/postgresql/extension
         fi
 
@@ -90,7 +77,7 @@ let
 
       meta = with lib; {
         description = "A lightweight message queue. Like AWS SQS and RSMQ but on Postgres.";
-        homepage = "https://github.com/tembo-io/pgmq";
+        homepage = "https://github.com/pgmq/pgmq";
         inherit (postgresql.meta) platforms;
         license = licenses.postgresql;
       };

nix/ext/tests/pgmq-drop-queue-overload.nix

@@ -0,0 +1,85 @@
+{ self, pkgs }:
+let
+  pname = "pgmq";
+  inherit (pkgs) lib;
+  system = pkgs.pkgsLinux.stdenv.hostPlatform.system;
+  testLib = import ./lib.nix { inherit self pkgs; };
+
+  installedExtension = self.legacyPackages.${system}."psql_15".exts."${pname}";
+  versions = installedExtension.versions;
+in
+pkgs.testers.runNixOSTest {
+  name = "pgmq-drop-queue-overload";
+  nodes.server =
+    { ... }:
+    {
+      imports = [
+        (testLib.makeSupabaseTestConfig {
+          majorVersion = "15";
+        })
+      ];
+    };
+  testScript =
+    { ... }:
+    let
+      versionList = lib.concatStringsSep ", " (map (v: ''"${v}"'') versions);
+    in
+    # python
+    ''
+      versions = [${versionList}]
+
+      def sql(query):
+          return server.succeed(
+              "psql -U supabase_admin -d postgres -t -A -F',' -c \"" + query.replace('"', '\\"') + "\""
+          ).strip()
+
+      def drop_queue_overloads():
+          # owned flag first: pg_get_function_identity_arguments() can itself
+          # contain a comma ("queue_name text, partitioned boolean"), so put the
+          # single-char flag first and split on the first comma only.
+          out = sql(
+              "select (d.objid is not null), pg_get_function_identity_arguments(p.oid) "
+              "from pg_proc p "
+              "left join pg_depend d on d.objid = p.oid and d.deptype = 'e' "
+              "  and d.refobjid = (select oid from pg_extension where extname = 'pgmq') "
+              "where p.pronamespace = 'pgmq'::regnamespace and p.proname = 'drop_queue' "
+              "order by 2;"
+          )
+          return [line.split(",", 1) for line in out.splitlines() if line]
+
+      def check_callers(qname):
+          sql(f"select pgmq.create('{qname}_a'); select pgmq.drop_queue('{qname}_a');")
+          sql(f"select pgmq.create('{qname}_b'); select pgmq.drop_queue('{qname}_b', false);")
+          # WRONG flag on purpose (queue isn't partitioned) - must still
+          # succeed, safely ignored in favour of pgmq.meta
+          sql(f"select pgmq.create('{qname}_c'); select pgmq.drop_queue('{qname}_c', true);")
+          sql(
+              f"select pgmq.create('{qname}_d'); "
+              f"select pgmq.drop_queue(queue_name => '{qname}_d', partitioned => true);"
+          )
+
+      start_all()
+      server.wait_for_unit("supabase-db-init.service")
+
+      for version in versions:
+          with subtest(f"install pgmq {version}"):
+              server.succeed("psql -U supabase_admin -d postgres -c 'DROP EXTENSION IF EXISTS pgmq;'")
+              server.succeed(
+                  f"psql -U supabase_admin -d postgres -c \"CREATE EXTENSION pgmq WITH VERSION '{version}' CASCADE;\""
+              )
+
+              overloads = drop_queue_overloads()
+              print(f"[{version}] drop_queue overloads: {overloads}")
+              assert len(overloads) == 1, (
+                  f"[{version}] expected exactly one drop_queue overload, got: {overloads}"
+              )
+              assert overloads[0][0] == "t", (
+                  f"[{version}] drop_queue is not extension-owned: {overloads}"
+              )
+              assert overloads[0][1] == "queue_name text, partitioned boolean", (
+                  f"[{version}] expected merged text,boolean signature, got: {overloads[0][1]}"
+              )
+
+              check_callers(f"q_{version.replace('.', '_')}")
+    '';
+}

nix/ext/versions.json

@@ -409,6 +409,13 @@
           "17"
         ],
         "hash": "sha256-IU+i6ONPwtgsFKdzya6E+222ualR66gkbb0lDr+7Rb8="
+      },
+      "1.13.0": {
+        "postgresql": [
+          "15",
+          "17"
+        ],
+        "hash": "sha256-0/L/ic6WZ64Uc6JDNnKGVFVpQa2a+3OGIeyQwPFTJ5U="
       }
     },
   "pgroonga": {

nix/tests/expected/pgmq.out

@@ -156,49 +156,84 @@ where
   n.nspname = 'pgmq'
 order by
   p.proname;
- schema_name |         function_name         |  owner   
--------------+-------------------------------+----------
- pgmq        | _belongs_to_pgmq              | postgres
- pgmq        | _ensure_pg_partman_installed  | postgres
- pgmq        | _extension_exists             | postgres
- pgmq        | _get_partition_col            | postgres
- pgmq        | _get_pg_partman_major_version | postgres
- pgmq        | _get_pg_partman_schema        | postgres
- pgmq        | archive                       | postgres
- pgmq        | archive                       | postgres
- pgmq        | convert_archive_partitioned   | postgres
- pgmq        | create                        | postgres
- pgmq        | create_non_partitioned        | postgres
- pgmq        | create_partitioned            | postgres
- pgmq        | create_unlogged               | postgres
- pgmq        | delete                        | postgres
- pgmq        | delete                        | postgres
- pgmq        | detach_archive                | postgres
- pgmq        | drop_queue                    | postgres
- pgmq        | drop_queue                    | postgres
- pgmq        | format_table_name             | postgres
- pgmq        | list_queues                   | postgres
- pgmq        | metrics                       | postgres
- pgmq        | metrics_all                   | postgres
- pgmq        | pop                           | postgres
- pgmq        | purge_queue                   | postgres
- pgmq        | read                          | postgres
- pgmq        | read_with_poll                | postgres
- pgmq        | send                          | postgres
- pgmq        | send                          | postgres
- pgmq        | send                          | postgres
- pgmq        | send                          | postgres
- pgmq        | send                          | postgres
- pgmq        | send                          | postgres
- pgmq        | send_batch                    | postgres
- pgmq        | send_batch                    | postgres
- pgmq        | send_batch                    | postgres
- pgmq        | send_batch                    | postgres
- pgmq        | send_batch                    | postgres
- pgmq        | send_batch                    | postgres
- pgmq        | set_vt                        | postgres
- pgmq        | validate_queue_name           | postgres
-(40 rows)
+ schema_name |          function_name           |  owner   
+-------------+----------------------------------+----------
+ pgmq        | _belongs_to_pgmq                 | postgres
+ pgmq        | _create_fifo_index_if_not_exists | postgres
+ pgmq        | _ensure_pg_partman_installed     | postgres
+ pgmq        | _extension_exists                | postgres
+ pgmq        | _get_partition_col               | postgres
+ pgmq        | _get_pg_partman_major_version    | postgres
+ pgmq        | _get_pg_partman_schema           | postgres
+ pgmq        | _send_batch                      | postgres
+ pgmq        | _validate_batch_params           | postgres
+ pgmq        | acquire_queue_lock               | postgres
+ pgmq        | archive                          | postgres
+ pgmq        | archive                          | postgres
+ pgmq        | bind_topic                       | postgres
+ pgmq        | convert_archive_partitioned      | postgres
+ pgmq        | create                           | postgres
+ pgmq        | create_fifo_index                | postgres
+ pgmq        | create_fifo_indexes_all          | postgres
+ pgmq        | create_non_partitioned           | postgres
+ pgmq        | create_partitioned               | postgres
+ pgmq        | create_unlogged                  | postgres
+ pgmq        | delete                           | postgres
+ pgmq        | delete                           | postgres
+ pgmq        | detach_archive                   | postgres
+ pgmq        | disable_notify_insert            | postgres
+ pgmq        | drop_queue                       | postgres
+ pgmq        | enable_notify_insert             | postgres
+ pgmq        | format_table_name                | postgres
+ pgmq        | list_notify_insert_throttles     | postgres
+ pgmq        | list_queues                      | postgres
+ pgmq        | list_topic_bindings              | postgres
+ pgmq        | list_topic_bindings              | postgres
+ pgmq        | metrics                          | postgres
+ pgmq        | metrics_all                      | postgres
+ pgmq        | notify_queue_listeners           | postgres
+ pgmq        | pop                              | postgres
+ pgmq        | purge_queue                      | postgres
+ pgmq        | read                             | postgres
+ pgmq        | read_grouped                     | postgres
+ pgmq        | read_grouped_head                | postgres
+ pgmq        | read_grouped_head_with_poll      | postgres
+ pgmq        | read_grouped_rr                  | postgres
+ pgmq        | read_grouped_rr_with_poll        | postgres
+ pgmq        | read_grouped_with_poll           | postgres
+ pgmq        | read_with_poll                   | postgres
+ pgmq        | send                             | postgres
+ pgmq        | send                             | postgres
+ pgmq        | send                             | postgres
+ pgmq        | send                             | postgres
+ pgmq        | send                             | postgres
+ pgmq        | send                             | postgres
+ pgmq        | send_batch                       | postgres
+ pgmq        | send_batch                       | postgres
+ pgmq        | send_batch                       | postgres
+ pgmq        | send_batch                       | postgres
+ pgmq        | send_batch                       | postgres
+ pgmq        | send_batch                       | postgres
+ pgmq        | send_batch_topic                 | postgres
+ pgmq        | send_batch_topic                 | postgres
+ pgmq        | send_batch_topic                 | postgres
+ pgmq        | send_batch_topic                 | postgres
+ pgmq        | send_batch_topic                 | postgres
+ pgmq        | send_batch_topic                 | postgres
+ pgmq        | send_topic                       | postgres
+ pgmq        | send_topic                       | postgres
+ pgmq        | send_topic                       | postgres
+ pgmq        | set_vt                           | postgres
+ pgmq        | set_vt                           | postgres
+ pgmq        | set_vt                           | postgres
+ pgmq        | set_vt                           | postgres
+ pgmq        | test_routing                     | postgres
+ pgmq        | unbind_topic                     | postgres
+ pgmq        | update_notify_insert             | postgres
+ pgmq        | validate_queue_name              | postgres
+ pgmq        | validate_routing_key             | postgres
+ pgmq        | validate_topic_pattern           | postgres
+(75 rows)
 
 -- assert search_path is preserved after after-create script is run
 show search_path;