PGMQ read benchmark

read.sql

DO $$
DECLARE
    batch_size INT := 100000;  -- Number of messages to pop per batch
    total_messages INT := 0;   -- Count of total messages popped
    batch_count INT := 0;      -- Count messages in the current batch
    start_time TIMESTAMP;
    end_time TIMESTAMP;
    result RECORD;
    queue_name TEXT := 'my_queue';
    message_ids BIGINT[] := '{}'; -- Array to hold message IDs for bulk delete
BEGIN
    -- Start the clock for benchmarking
    start_time := clock_timestamp();
    
    -- Loop until no more messages are available
    LOOP
        batch_count := 0;  -- Reset batch count for each iteration
        
        -- Read messages in batches
        FOR result IN 
            SELECT * FROM pgmq.read(queue_name => queue_name, vt => 30, qty => batch_size)
        LOOP
            -- Collect message IDs for bulk delete
            message_ids := array_append(message_ids, result.msg_id);
            batch_count := batch_count + 1;
            total_messages := total_messages + 1;
        END LOOP;
        
        -- Bulk delete messages if there are any
        IF array_length(message_ids, 1) > 0 THEN
            -- Call pgmq.delete to delete the collected messages
            PERFORM pgmq.delete(queue_name, message_ids);
            message_ids := '{}';  -- Reset the message_ids array
            RAISE NOTICE 'Processed batch of % messages', batch_count;

            -- Commit after each batch
            COMMIT;
            --RAISE NOTICE 'Committed batch of % messages', batch_count;
        END IF;

        -- Exit if no more messages were processed (queue is empty)
        IF batch_count = 0 THEN
            EXIT;
        END IF;
    END LOOP;
    
    -- End the clock for benchmarking
    end_time := clock_timestamp();
    
    -- Output the total time taken and total messages popped
    RAISE NOTICE 'Total messages read: %', total_messages;
    RAISE NOTICE 'Time taken to read % messages: %', total_messages, end_time - start_time;
END $$;