fix 1.5.0 migration per issue #362

pgmq-extension/pgmq.control

@@ -1,5 +1,5 @@
 comment = 'A lightweight message queue. Like AWS SQS and RSMQ but on Postgres.'
-default_version = '1.5.0'
+default_version = '1.5.1'
 module_pathname = '$libdir/pgmq'
 schema = 'pgmq'
 relocatable = false

pgmq-extension/sql/pgmq--1.4.5--1.5.0.sql

@@ -561,6 +561,29 @@ DROP FUNCTION pgmq.send_batch(TEXT, JSONB[], INTEGER);
 DROP FUNCTION pgmq.archive(TEXT, BIGINT);
 DROP FUNCTION pgmq.archive(TEXT, BIGINT[]);
 
+-- send: actual implementation
+CREATE FUNCTION pgmq.send(
+    queue_name TEXT,
+    msg JSONB,
+    headers JSONB,
+    delay TIMESTAMP WITH TIME ZONE
+) RETURNS SETOF BIGINT AS $$
+DECLARE
+    sql TEXT;
+    qtable TEXT := pgmq.format_table_name(queue_name, 'q');
+BEGIN
+    sql := FORMAT(
+        $QUERY$
+        INSERT INTO pgmq.%I (vt, message, headers)
+        VALUES ($2, $1, $3)
+        RETURNING msg_id;
+        $QUERY$,
+        qtable
+    );
+    RETURN QUERY EXECUTE sql USING msg, delay, headers;
+END;
+$$ LANGUAGE plpgsql;
+
 -- send: 2 args, no delay or headers
 CREATE FUNCTION pgmq.send(
     queue_name TEXT,
@@ -606,11 +629,11 @@ CREATE FUNCTION pgmq.send(
     SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp() + make_interval(secs => delay));
 $$ LANGUAGE sql;
 
--- send: actual implementation
-CREATE FUNCTION pgmq.send(
+-- send_batch: actual implementation
+CREATE FUNCTION pgmq.send_batch(
     queue_name TEXT,
-    msg JSONB,
-    headers JSONB,
+    msgs JSONB[],
+    headers JSONB[],
     delay TIMESTAMP WITH TIME ZONE
 ) RETURNS SETOF BIGINT AS $$
 DECLARE
@@ -620,12 +643,12 @@ BEGIN
     sql := FORMAT(
         $QUERY$
         INSERT INTO pgmq.%I (vt, message, headers)
-        VALUES ($2, $1, $3)
+        SELECT $2, unnest($1), unnest(coalesce($3, ARRAY[]::jsonb[]))
         RETURNING msg_id;
         $QUERY$,
         qtable
     );
-    RETURN QUERY EXECUTE sql USING msg, delay, headers;
+    RETURN QUERY EXECUTE sql USING msgs, delay, headers;
 END;
 $$ LANGUAGE plpgsql;
 
@@ -674,29 +697,6 @@ CREATE FUNCTION pgmq.send_batch(
     SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp() + make_interval(secs => delay));
 $$ LANGUAGE sql;
 
--- send_batch: actual implementation
-CREATE FUNCTION pgmq.send_batch(
-    queue_name TEXT,
-    msgs JSONB[],
-    headers JSONB[],
-    delay TIMESTAMP WITH TIME ZONE
-) RETURNS SETOF BIGINT AS $$
-DECLARE
-    sql TEXT;
-    qtable TEXT := pgmq.format_table_name(queue_name, 'q');
-BEGIN
-    sql := FORMAT(
-        $QUERY$
-        INSERT INTO pgmq.%I (vt, message, headers)
-        SELECT $2, unnest($1), unnest(coalesce($3, ARRAY[]::jsonb[]))
-        RETURNING msg_id;
-        $QUERY$,
-        qtable
-    );
-    RETURN QUERY EXECUTE sql USING msgs, delay, headers;
-END;
-$$ LANGUAGE plpgsql;
-
 -- archive
 CREATE FUNCTION pgmq.archive(
     queue_name TEXT,

pgmq-extension/sql/pgmq.sql

@@ -1122,4 +1122,4 @@ BEGIN
     qualified_a_table_name
   );
 END;
-$$ LANGUAGE plpgsql;
+$$ LANGUAGE plpgsql;

pgmq-extension/test/expected/base.out

@@ -155,18 +155,36 @@ SELECT msg_id = :msg_id FROM pgmq.read_with_poll('test_default_queue_vt', 10, 1,
  t
 (1 row)
 
--- send a batch of 2 messages
+-- test send_batch
 SELECT pgmq.create('batch_queue');
  create 
 --------
 
 (1 row)
 
+-- send without headers, not explicitly typed
+SELECT ARRAY( SELECT pgmq.send_batch(
+    'batch_queue',
+    ARRAY['{"hello": "world_0"}', '{"hello": "world_1"}']::jsonb[]
+)) = ARRAY[1, 2]::BIGINT[];
+ ?column? 
+----------
+ t
+(1 row)
+
+SELECT msg_id, message, headers from pgmq.read('batch_queue', 10, 2);
+ msg_id |       message        | headers 
+--------+----------------------+---------
+      1 | {"hello": "world_0"} | 
+      2 | {"hello": "world_1"} | 
+(2 rows)
+
+-- send with headers
 SELECT ARRAY( SELECT pgmq.send_batch(
     'batch_queue',
     ARRAY['{"hello": "world_0"}', '{"hello": "world_1"}']::jsonb[],
     ARRAY['{"header": 1}', '{"header": 2}']::jsonb[]
-)) = ARRAY[1, 2]::BIGINT[];
+)) = ARRAY[3, 4]::BIGINT[];
  ?column? 
 ----------
  t
@@ -175,8 +193,8 @@ SELECT ARRAY( SELECT pgmq.send_batch(
 SELECT msg_id, message, headers from pgmq.read('batch_queue', 0, 2);
  msg_id |       message        |    headers    
 --------+----------------------+---------------
-      1 | {"hello": "world_0"} | {"header": 1}
-      2 | {"hello": "world_1"} | {"header": 2}
+      3 | {"hello": "world_0"} | {"header": 1}
+      4 | {"hello": "world_1"} | {"header": 2}
 (2 rows)
 
 -- send a batch of 2 messages with timestamp

pgmq-extension/test/sql/base.sql

@@ -66,13 +66,23 @@ SELECT msg_id = :msg_id FROM pgmq.read('test_default_queue_vt', 2, 1);
 -- read again, now using poll to block until message is ready
 SELECT msg_id = :msg_id FROM pgmq.read_with_poll('test_default_queue_vt', 10, 1, 10);
 
--- send a batch of 2 messages
+-- test send_batch
 SELECT pgmq.create('batch_queue');
+
+-- send without headers, not explicitly typed
+SELECT ARRAY( SELECT pgmq.send_batch(
+    'batch_queue',
+    ARRAY['{"hello": "world_0"}', '{"hello": "world_1"}']::jsonb[]
+)) = ARRAY[1, 2]::BIGINT[];
+
+SELECT msg_id, message, headers from pgmq.read('batch_queue', 10, 2);
+
+-- send with headers
 SELECT ARRAY( SELECT pgmq.send_batch(
     'batch_queue',
     ARRAY['{"hello": "world_0"}', '{"hello": "world_1"}']::jsonb[],
     ARRAY['{"header": 1}', '{"header": 2}']::jsonb[]
-)) = ARRAY[1, 2]::BIGINT[];
+)) = ARRAY[3, 4]::BIGINT[];
 
 SELECT msg_id, message, headers from pgmq.read('batch_queue', 0, 2);