diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,14 @@ # Changelog for pgmq-migration +## 0.6.0.0 -- 2026-09-10 + +Native installs now reach tagged PGMQ 1.13.0 through append-only migrations 0004–0006: +upstream 1.12 grouped heads, upstream 1.13 partition identities/premake/metrics, then separate +local four-argument partition re-entry preservation. Existing partitioned queue IDs become +`GENERATED BY DEFAULT`; the metrics composite gains its nullable eighth attribute. +The original three payloads and the 1.11 predecessor import validator remain unchanged. +Upgrading preserves data but does not drain default partitions or schedule maintenance. + ## 0.5.0.0 -- 2026-08-06 ### Bug Fixes diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -4,9 +4,31 @@ compose `pgmqMigrations` into their migration plan; the package intentionally ships no standalone runner. -Fresh databases run the two native migrations normally. Databases installed by +Fresh databases run all six native migrations and reach PGMQ 1.13.0. Databases installed by `pgmq-migration` 0.3 or earlier must import their `public.schema_migrations` predecessor history before the first native `up`, so the PGMQ baseline is recorded rather than replayed. + +The immutable 1.11 baseline and local notification hardening are followed by the pristine +1.12 upgrade, pristine 1.13 upgrade, and a separate four-argument partition re-entry override. +Existing native ledgers run only their pending suffix. Complete that suffix before allowing +application queue-creation traffic; each SQL migration is transactional separately. +The 1.12 checkpoint exists for compatibility testing, not as a public installation mode. + +Partitioned queues still require pg_partman. The 1.13 upgrade preserves existing messages +and changes active partitioned queue IDs to `GENERATED BY DEFAULT`, enabling pg_partman +to move spilled rows with their original IDs. It does not recover spills or schedule maintenance. +`default_partition_length` estimates rows in both queue and archive default partitions; +refresh their statistics with ANALYZE when diagnosing or verifying recovery. + +Run the required partition acceptance with: + +```bash +nix develop .#partman --command cabal test pgmq-migration:pgmq-migration-test --test-show-details=direct +``` + +That shell sets `PGMQ_REQUIRE_PARTMAN=1`; missing or unusable pg_partman fails the tests. +The direct/effect/config fixtures accept `PGMQ_TEST_SCHEMA_VERSION=1.12.0|1.13.0` and +default to the complete native ledger. They package the tagged 1.12 SQL fixture. The existing `pgmqHasqlMigrationSourceConfig` helper requires an exclusive predecessor ledger. When an application deliberately shares that table with its own migrations, use the diff --git a/migrations/0004-upgrade-v1.12.0.sql b/migrations/0004-upgrade-v1.12.0.sql new file mode 100644 --- /dev/null +++ b/migrations/0004-upgrade-v1.12.0.sql @@ -0,0 +1,104 @@ +-- Allow `pgmq.topic_bindings` to be dumped by `pg_dump` when pgmq is installed as an extension +DO +$$ +BEGIN + IF EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pgmq') THEN + PERFORM pg_catalog.pg_extension_config_dump('pgmq.topic_bindings', ''); + END IF; +END +$$; + + +-- read_grouped_head: read the head of N different FIFO groups in a single operation. +-- This supports horizontal scaling by processing groups in parallel while ensuring message ordering is preserved per group. +CREATE OR REPLACE FUNCTION pgmq.read_grouped_head( + queue_name TEXT, + vt INTEGER, + qty INTEGER +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH fifo_groups AS ( + -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility + SELECT + COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key, + MIN(msg_id) AS head_msg_id + FROM pgmq.%1$I + GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') + ), + selected_messages AS ( + -- Take at most 1 message per group + SELECT g.head_msg_id msg_id + FROM fifo_groups g + JOIN pgmq.%1$I q ON q.msg_id = g.head_msg_id + WHERE q.vt <= clock_timestamp() + ORDER BY q.msg_id + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + UPDATE pgmq.%1$I m + SET + vt = clock_timestamp() + %2$L, + read_ct = read_ct + 1, + last_read_at = clock_timestamp() + FROM selected_messages sm + WHERE m.msg_id = sm.msg_id + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers; + $QUERY$, + qtable, make_interval(secs => vt) + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + + +CREATE OR REPLACE FUNCTION pgmq._ensure_pg_partman_installed() +RETURNS void AS $$ +BEGIN + IF NOT pgmq._extension_exists('pg_partman') THEN + RAISE EXCEPTION 'pg_partman is required for partitioned queues'; + END IF; +END; +$$ LANGUAGE plpgsql; + + +DROP FUNCTION IF EXISTS pgmq.enable_notify_insert(queue_name text); + +-- read_grouped_head_with_poll +-- reads the head of N different FIFO groups in a single operation, with polling support +CREATE OR REPLACE FUNCTION pgmq.read_grouped_head_with_poll( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + max_poll_seconds INTEGER DEFAULT 5, + poll_interval_ms INTEGER DEFAULT 100 +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + r pgmq.message_record; + stop_at TIMESTAMPTZ; +BEGIN + stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds); + LOOP + IF clock_timestamp() >= stop_at THEN + RETURN; + END IF; + + FOR r IN + SELECT * FROM pgmq.read_grouped_head(queue_name, vt, qty) + LOOP + RETURN NEXT r; + END LOOP; + IF FOUND THEN + RETURN; + ELSE + PERFORM pg_sleep(poll_interval_ms::numeric / 1000); + END IF; + END LOOP; +END; +$$ LANGUAGE plpgsql; diff --git a/migrations/0005-upgrade-v1.13.0.sql b/migrations/0005-upgrade-v1.13.0.sql new file mode 100644 --- /dev/null +++ b/migrations/0005-upgrade-v1.13.0.sql @@ -0,0 +1,254 @@ +-- Partitioned queues: msg_id becomes GENERATED BY DEFAULT AS IDENTITY. +-- pg_partman's partition_data_* tooling moves rows out of the default +-- partition by inserting them with their existing msg_id, which +-- GENERATED ALWAYS refuses. pgmq's own writers never supply msg_id, so this +-- changes nothing for them. +DO $$ +DECLARE + queue_record RECORD; + qtable TEXT; +BEGIN + FOR queue_record IN SELECT queue_name FROM pgmq.meta WHERE is_partitioned LOOP + qtable := pgmq.format_table_name(queue_record.queue_name, 'q'); + + IF EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'pgmq' + AND table_name = qtable + AND column_name = 'msg_id' + AND identity_generation = 'ALWAYS' + ) THEN + EXECUTE FORMAT('ALTER TABLE pgmq.%I ALTER COLUMN msg_id SET GENERATED BY DEFAULT', qtable); + END IF; + END LOOP; +END; +$$; + +-- create_partitioned gains a premake parameter. The three-argument signature is +-- dropped rather than kept alongside it, so existing calls resolve to one function. +DROP FUNCTION IF EXISTS pgmq.create_partitioned(TEXT, TEXT, TEXT); + +CREATE FUNCTION pgmq.create_partitioned( + queue_name TEXT, + partition_interval TEXT DEFAULT '10000', + retention_interval TEXT DEFAULT '100000', + premake INTEGER DEFAULT 4 +) +RETURNS void AS $$ +DECLARE + partition_col TEXT; + a_partition_col TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + qtable_seq TEXT := qtable || '_msg_id_seq'; + atable TEXT := pgmq.format_table_name(queue_name, 'a'); + fq_qtable TEXT := 'pgmq.' || qtable; + fq_atable TEXT := 'pgmq.' || atable; +BEGIN + PERFORM pgmq.validate_queue_name(queue_name); + PERFORM pgmq.acquire_queue_lock(queue_name); + PERFORM pgmq._ensure_pg_partman_installed(); + IF premake < 1 THEN + RAISE EXCEPTION 'premake must be at least 1, got %', premake; + END IF; + SELECT pgmq._get_partition_col(partition_interval) INTO partition_col; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT GENERATED BY DEFAULT AS IDENTITY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I) + $QUERY$, + qtable, partition_col + ); + + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_premake := %s, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_qtable, + partition_col, + partition_interval, + premake + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I); + $QUERY$, + qtable || '_part_idx', qtable, partition_col + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || qtable + ); + + EXECUTE FORMAT( + $QUERY$ + INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged) + VALUES (%L, true, false) + ON CONFLICT + DO NOTHING; + $QUERY$, + queue_name + ); + + IF partition_col = 'enqueued_at' THEN + a_partition_col := 'archived_at'; + ELSE + a_partition_col := partition_col; + END IF; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT NOT NULL, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I); + $QUERY$, + atable, a_partition_col + ); + + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_premake := %s, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_atable, + a_partition_col, + partition_interval, + premake + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || atable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at); + $QUERY$, + 'archived_at_idx_' || queue_name, atable + ); + +END; +$$ LANGUAGE plpgsql; + +-- metrics_result gains default_partition_length, which metrics() now reports. +ALTER TYPE pgmq.metrics_result ADD ATTRIBUTE default_partition_length bigint; + +-- get metrics for a single queue +CREATE OR REPLACE FUNCTION pgmq.metrics(queue_name TEXT) +RETURNS pgmq.metrics_result AS $$ +DECLARE + result_row pgmq.metrics_result; + query TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + q_default_partition TEXT := qtable || '_default'; + a_default_partition TEXT := pgmq.format_table_name(queue_name, 'a') || '_default'; + default_partition_length BIGINT; +BEGIN + -- Only partitioned queues have default partitions. Messages in them have no + -- partition of their own, which means pg_partman maintenance is failing for + -- this queue; a non-zero value here is the signal to act on. The planner's + -- estimate is used so that a large spill does not slow down every scrape. + IF to_regclass(FORMAT('pgmq.%I', q_default_partition)) IS NOT NULL THEN + SELECT COALESCE(SUM(GREATEST(c.reltuples, 0))::bigint, 0) + INTO default_partition_length + FROM pg_class c + WHERE c.oid IN ( + to_regclass(FORMAT('pgmq.%I', q_default_partition)), + to_regclass(FORMAT('pgmq.%I', a_default_partition)) + ); + END IF; + + query := FORMAT( + $QUERY$ + WITH q_summary AS ( + SELECT + count(*) as queue_length, + count(CASE WHEN vt <= NOW() THEN 1 END) as queue_visible_length, + EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec, + EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec, + NOW() as scrape_time + FROM pgmq.%I + ), + all_metrics AS ( + SELECT CASE + WHEN is_called THEN last_value ELSE 0 + END as total_messages + FROM pgmq.%I + ) + SELECT + %L as queue_name, + q_summary.queue_length, + q_summary.newest_msg_age_sec, + q_summary.oldest_msg_age_sec, + all_metrics.total_messages, + q_summary.scrape_time, + q_summary.queue_visible_length, + %L::bigint as default_partition_length + FROM q_summary, all_metrics + $QUERY$, + qtable, qtable || '_msg_id_seq', queue_name, default_partition_length + ); + EXECUTE query INTO result_row; + RETURN result_row; +END; +$$ LANGUAGE plpgsql; diff --git a/migrations/0006-preserve-partitioned-reentry-v1.13.0.sql b/migrations/0006-preserve-partitioned-reentry-v1.13.0.sql new file mode 100644 --- /dev/null +++ b/migrations/0006-preserve-partitioned-reentry-v1.13.0.sql @@ -0,0 +1,178 @@ +-- Local override of the pristine 1.13 partition function. +-- See docs/adr/pgmq-1.12-1.13-compatibility.md and migration 0003. +CREATE OR REPLACE FUNCTION pgmq.create_partitioned( + queue_name TEXT, + partition_interval TEXT DEFAULT '10000', + retention_interval TEXT DEFAULT '100000', + premake INTEGER DEFAULT 4 +) +RETURNS void AS $$ +DECLARE + partition_col TEXT; + a_partition_col TEXT; + l_already_managed BOOLEAN; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + qtable_seq TEXT := qtable || '_msg_id_seq'; + atable TEXT := pgmq.format_table_name(queue_name, 'a'); + fq_qtable TEXT := 'pgmq.' || qtable; + fq_atable TEXT := 'pgmq.' || atable; +BEGIN + PERFORM pgmq.validate_queue_name(queue_name); + PERFORM pgmq.acquire_queue_lock(queue_name); + PERFORM pgmq._ensure_pg_partman_installed(); + IF premake < 1 THEN + RAISE EXCEPTION 'premake must be at least 1, got %', premake; + END IF; + SELECT pgmq._get_partition_col(partition_interval) INTO partition_col; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT GENERATED BY DEFAULT AS IDENTITY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I) + $QUERY$, + qtable, partition_col + ); + + -- Preserve the parent-registration guard from migration 0003. + EXECUTE FORMAT( + 'SELECT EXISTS (SELECT 1 FROM %I.part_config WHERE parent_table = %L)', + pgmq._get_pg_partman_schema(), fq_qtable + ) INTO l_already_managed; + IF NOT l_already_managed THEN + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_premake := %s, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_qtable, + partition_col, + partition_interval, + premake + ); + END IF; + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I); + $QUERY$, + qtable || '_part_idx', qtable, partition_col + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || qtable + ); + + EXECUTE FORMAT( + $QUERY$ + INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged) + VALUES (%L, true, false) + ON CONFLICT + DO NOTHING; + $QUERY$, + queue_name + ); + + IF partition_col = 'enqueued_at' THEN + a_partition_col := 'archived_at'; + ELSE + a_partition_col := partition_col; + END IF; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT NOT NULL, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I); + $QUERY$, + atable, a_partition_col + ); + + -- Preserve the parent-registration guard from migration 0003. + EXECUTE FORMAT( + 'SELECT EXISTS (SELECT 1 FROM %I.part_config WHERE parent_table = %L)', + pgmq._get_pg_partman_schema(), fq_atable + ) INTO l_already_managed; + IF NOT l_already_managed THEN + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_premake := %s, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_atable, + a_partition_col, + partition_interval, + premake + ); + END IF; + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || atable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at); + $QUERY$, + 'archived_at_idx_' || queue_name, atable + ); + +END; +$$ LANGUAGE plpgsql; diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -1,3 +1,6 @@ 0001-install-v1.11.0.sql 0002-schema-management-comment.sql 0003-notify-crash-safety-and-locking.sql +0004-upgrade-v1.12.0.sql +0005-upgrade-v1.13.0.sql +0006-preserve-partitioned-reentry-v1.13.0.sql diff --git a/pgmq-migration.cabal b/pgmq-migration.cabal --- a/pgmq-migration.cabal +++ b/pgmq-migration.cabal @@ -1,18 +1,18 @@ -cabal-version: 3.4 -name: pgmq-migration -version: 0.5.0.0 -synopsis: PGMQ schema migrations without PostgreSQL extension +cabal-version: 3.4 +name: pgmq-migration +version: 0.6.0.0 +synopsis: PGMQ schema migrations without PostgreSQL extension description: Installs the PGMQ schema into PostgreSQL without requiring the pgmq extension. Exposes a native pg-migrate component and verified predecessor-history imports. -homepage: https://github.com/shinzui/pgmq-hs -license: MIT -license-file: LICENSE -author: Nadeem Bitar -maintainer: Nadeem Bitar -category: Database -build-type: Simple +homepage: https://github.com/shinzui/pgmq-hs +license: MIT +license-file: LICENSE +author: Nadeem Bitar +maintainer: Nadeem Bitar +category: Database +build-type: Simple extra-doc-files: CHANGELOG.md README.md @@ -20,61 +20,71 @@ extra-source-files: migrations/*.sql migrations/manifest + test/fixtures/*.sql vendor/pgmq/pgmq-extension/sql/pgmq--1.10.0--1.10.1.sql vendor/pgmq/pgmq-extension/sql/pgmq--1.10.1--1.11.0.sql + vendor/pgmq/pgmq-extension/sql/pgmq--1.11.0--1.11.1.sql + vendor/pgmq/pgmq-extension/sql/pgmq--1.11.1--1.12.0.sql + vendor/pgmq/pgmq-extension/sql/pgmq--1.12.0--1.13.0.sql vendor/pgmq/pgmq-extension/sql/pgmq.sql common warnings ghc-options: - -Wall -Wcompat -Widentities -Wincomplete-uni-patterns - -Wincomplete-record-updates -Wredundant-constraints - -fhide-source-paths -Wmissing-export-lists -Wpartial-fields + -Wall + -Wcompat + -Widentities + -Wincomplete-uni-patterns + -Wincomplete-record-updates + -Wredundant-constraints + -fhide-source-paths + -Wmissing-export-lists + -Wpartial-fields -Wmissing-deriving-strategies library - import: warnings + import: warnings exposed-modules: Pgmq.Migration Pgmq.Migration.History.HasqlMigration Pgmq.Migration.SchemaContract - other-modules: Pgmq.Migration.Internal.Definition + other-modules: Pgmq.Migration.Internal.Definition default-extensions: ImportQualifiedPost OverloadedStrings build-depends: - , aeson ^>=2.2 - , base >=4.18 && <5 - , bytestring ^>=0.12 - , containers ^>=0.7 - , file-embed ^>=0.0.16 - , hasql ^>=1.10 - , hasql-transaction ^>=1.2 - , pg-migrate ^>=1.1.0.0 - , pg-migrate-embed ^>=1.1.0.0 - , pg-migrate-import-hasql-migration ^>=1.1.0.0 - , text ^>=2.1 + aeson ^>=2.2, + base >=4.18 && <5, + bytestring ^>=0.12, + containers ^>=0.7, + file-embed ^>=0.0.16, + hasql ^>=1.10, + hasql-transaction ^>=1.2, + pg-migrate ^>=1.1.0.0, + pg-migrate-embed ^>=1.1.0.0, + pg-migrate-import-hasql-migration ^>=1.1.0.0, + text ^>=2.1, - hs-source-dirs: src - default-language: GHC2024 + hs-source-dirs: src + default-language: GHC2024 test-suite pgmq-migration-test - import: warnings + import: warnings default-language: GHC2024 - type: exitcode-stdio-1.0 - hs-source-dirs: test - main-is: Main.hs + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: Main.hs build-depends: - , base >=4.18 && <5 - , bytestring - , containers - , directory - , ephemeral-pg >=0.2.1 - , hasql - , pg-migrate - , pg-migrate-import-hasql-migration - , pgmq-migration - , tasty ^>=1.5 - , tasty-hunit ^>=0.10 - , text + base >=4.18 && <5, + bytestring, + containers, + directory, + ephemeral-pg >=0.2.1, + hasql, + pg-migrate, + pg-migrate-import-hasql-migration, + pgmq-migration >=0.6 && <0.7, + tasty ^>=1.5, + tasty-hunit ^>=0.10, + text, diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -3,12 +3,15 @@ module Main (main) where +import Control.Concurrent (forkIO, newEmptyMVar, putMVar, takeMVar) +import Control.Exception (SomeException, try) import Control.Monad (filterM, forM_) import Data.ByteString qualified as ByteString import Data.Foldable (toList) import Data.List.NonEmpty (NonEmpty (..)) import Data.Map.Strict qualified as Map import Data.Text (Text) +import Data.Text qualified as T import Data.Text.Encoding qualified as Text import Database.PostgreSQL.Migrate ( EquivalentHistoryPolicy (AllowEquivalentHistory), @@ -27,6 +30,7 @@ connectionProviderFromSettings, defaultImportOptions, defaultRunOptions, + migrationComponentFromEmbeddedSql, migrationId, migrationPlan, runMigrationPlan, @@ -75,8 +79,9 @@ pgmqHasqlMigrationSourceConfigWithPolicy, ) import System.Directory (doesFileExist) +import System.Environment (lookupEnv) import Test.Tasty (TestTree, defaultMain, testGroup) -import Test.Tasty.HUnit (assertFailure, testCase, (@?=)) +import Test.Tasty.HUnit (assertBool, assertFailure, testCase, (@?=)) main :: IO () main = do @@ -97,12 +102,14 @@ "pgmq-migration" [ testGroup "native definition" - [ testCase "baseline bytes equal vendored pgmq.sql" testNativePayload, + [ testCase "immutable history and upstream payload provenance" (testNativePayload conn), testCase "component pgmq lists the ledger in order and has no dependencies" testNativeComponent ], testGroup "native runner" - [ testCase "fresh install applies once and is idempotent" (testNativeRunner settings conn) + [ testCase "fresh install applies once and is idempotent" (testNativeRunner settings conn), + testCase "1.12 and 1.13 catalog convergence" (testConvergence conn), + testCase "populated partition upgrades and spill recovery" (testPartitions settings conn) ], testGroup "history import" @@ -115,14 +122,201 @@ ] ] -testNativePayload :: IO () -testNativePayload = do - nativePath <- findFile ["pgmq-migration/migrations/0001-install-v1.11.0.sql", "migrations/0001-install-v1.11.0.sql"] - vendorPath <- findFile ["vendor/pgmq/pgmq-extension/sql/pgmq.sql", "../vendor/pgmq/pgmq-extension/sql/pgmq.sql"] - native <- ByteString.readFile nativePath - vendored <- ByteString.readFile vendorPath - native @?= vendored +testNativePayload :: Connection.Connection -> IO () +testNativePayload conn = do + forM_ + [ ("0001-install-v1.11.0.sql", "faa9b8800005f80fbdf6a33d071183d2"), + ("0002-schema-management-comment.sql", "f3010777d47ec6d4f7d92a179bcf09b8"), + ("0003-notify-crash-safety-and-locking.sql", "e0a66c96bd8811607440dbca0cb302ea") + ] + $ \(name, digest) -> do + bytes <- readMigration name + result <- Connection.use conn (Session.statement (Text.decodeUtf8 bytes) (preparable "SELECT md5($1)" (Encoders.param (Encoders.nonNullable Encoders.text)) (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text))))) + either (assertFailure . show) (@?= digest) result + first <- readVendor "pgmq--1.11.0--1.11.1.sql" + second <- readVendor "pgmq--1.11.1--1.12.0.sql" + readMigration "0004-upgrade-v1.12.0.sql" >>= (@?= first <> "\n" <> second) + upstream <- readVendor "pgmq--1.12.0--1.13.0.sql" + readMigration "0005-upgrade-v1.13.0.sql" >>= (@?= upstream) + fixture <- Text.decodeUtf8 <$> readFixture "pgmq-1.12.0.sql" + digest <- Connection.use conn (Session.statement fixture (preparable "SELECT encode(sha256(convert_to($1,'UTF8')),'hex')" (Encoders.param (Encoders.nonNullable Encoders.text)) (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text))))) + either (assertFailure . show) (@?= "be087bfcb0ec5e65abb76610249750f2ec8dc757956125a1b40430ce95fc7f0f") digest +readMigration :: FilePath -> IO ByteString.ByteString +readMigration name = findFile ["pgmq-migration/migrations/" <> name, "migrations/" <> name] >>= ByteString.readFile + +readVendor :: FilePath -> IO ByteString.ByteString +readVendor name = findFile ["vendor/pgmq/pgmq-extension/sql/" <> name, "../vendor/pgmq/pgmq-extension/sql/" <> name] >>= ByteString.readFile + +readFixture :: FilePath -> IO ByteString.ByteString +readFixture name = findFile ["pgmq-migration/test/fixtures/" <> name, "test/fixtures/" <> name] >>= ByteString.readFile + +testConvergence :: Connection.Connection -> IO () +testConvergence conn = forM_ [False, True] $ \latest -> do + resetDb conn + names <- nativeMigrationNames + forM_ (take (if latest then 6 else 4) names) $ \name -> + readMigration (T.unpack name <> ".sql") >>= runSql conn . Text.decodeUtf8 + upgraded <- schemaSnapshot conn + resetDb conn + -- mori://pgmq/pgmq tag v1.12.0, 08ace4087dbf00e51704c5a3d9df2e15fd566127. + -- pgmq-extension/sql/pgmq.sql; artifact URI pending. SHA256: + -- be087bfcb0ec5e65abb76610249750f2ec8dc757956125a1b40430ce95fc7f0f + freshSql <- if latest then readVendor "pgmq.sql" else readFixture "pgmq-1.12.0.sql" + runSql conn (Text.decodeUtf8 freshSql) + fresh <- schemaSnapshot conn + let exceptions = + [ "body:notify_queue_listeners()", + "body:enable_notify_insert(text, integer)", + if latest then "body:create_partitioned(text, text, text, integer)" else "body:create_partitioned(text, text, text)" + ] + normalized snapshot = foldr Map.delete snapshot exceptions + differences a b = + Map.keys (Map.differenceWith (\x y -> if x == y then Nothing else Just x) a b) + <> Map.keys (Map.difference b a) + actual = normalized upgraded + expected = normalized fresh + forM_ exceptions $ \key -> assertBool ("missing local override " <> T.unpack key) (Map.member key upgraded && Map.member key fresh) + differences actual expected @?= [] + let grouped = "body:read_grouped_head(text, integer, integer)" + assertBool "snapshot comparison detects missing grouped function" (grouped `elem` differences (Map.delete grouped actual) expected) + assertBool "snapshot comparison detects altered body" (grouped `elem` differences (Map.insert grouped "changed" actual) expected) + if latest + then do + let attribute = "column:metrics_result:8" + assertBool "eighth metrics attribute exists" (Map.member attribute actual) + assertBool "snapshot comparison detects altered metric" (attribute `elem` differences (Map.insert attribute "wrong type" actual) expected) + else pure () + +schemaSnapshot :: Connection.Connection -> IO (Map.Map Text Text) +schemaSnapshot conn = do + sql <- Text.decodeUtf8 <$> readFixture "schema-snapshot.sql" + result <- + Connection.use + conn + ( Session.statement + () + ( preparable + sql + Encoders.noParams + (Decoders.rowList ((,) <$> Decoders.column (Decoders.nonNullable Decoders.text) <*> Decoders.column (Decoders.nonNullable Decoders.text))) + ) + ) + either (\err -> assertFailure (show err) >> pure Map.empty) (pure . Map.fromList) result + +queryText :: Connection.Connection -> Text -> IO Text +queryText conn sql = do + result <- + Connection.use + conn + ( Session.statement + () + ( Statement.unpreparable + sql + Encoders.noParams + (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.text))) + ) + ) + either (\err -> assertFailure (show err) >> pure "") pure result + +assertSql :: Connection.Connection -> Text -> IO () +assertSql conn predicate = queryText conn ("SELECT (" <> predicate <> ")::text") >>= (@?= "true") + +-- Install a prefix through the real ledger so populated upgrades also exercise +-- checksums, suffix selection and AlreadyApplied, not just raw SQL replay. +installPrefix :: Settings.Settings -> Int -> IO () +installPrefix settings count = do + names <- take count <$> nativeMigrationNames + entries <- + traverse + ( \name -> do + let file = T.unpack name <> ".sql" + bytes <- readMigration file + pure (file, bytes) + ) + names + case entries of + [] -> assertFailure "empty migration prefix" + first : rest -> do + component <- either (assertFailure . show) pure (migrationComponentFromEmbeddedSql "pgmq" mempty (first :| rest)) + plan <- either (assertFailure . show) pure (migrationPlan (component :| [])) + result <- runMigrationPlan defaultRunOptions settings plan + either (assertFailure . show) (const (pure ())) result + +testPartitions :: Settings.Settings -> Connection.Connection -> IO () +testPartitions settings conn = do + required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN" + available <- queryText conn "SELECT EXISTS (SELECT 1 FROM pg_available_extensions WHERE name='pg_partman')::text" + if available /= "true" + then + if required + then assertFailure "PGMQ_REQUIRE_PARTMAN=1 but pg_partman is unavailable" + else putStrLn "SKIPPED: partition upgrade/recovery requires pg_partman" + else do + runSql conn "CREATE SCHEMA IF NOT EXISTS partman; CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman" + version <- queryText conn "SELECT current_setting('server_version') || ' / pg_partman ' || extversion FROM pg_extension WHERE extname='pg_partman'" + putStrLn ("Partition test versions: " <> T.unpack version) + forM_ [3, 4] $ \prefix -> do + resetDb conn + -- Each reset discards registrations as well as fixture-owned tables. + runSql conn "DROP EXTENSION pg_partman CASCADE; CREATE EXTENSION pg_partman SCHEMA partman" + installPrefix settings prefix + runSql conn "SELECT pgmq.create_partitioned('numeric_old','10','100'); SELECT pgmq.create_partitioned('time_old','1 day','30 days'); SELECT pgmq.create('ordinary_old')" + runSql conn "SELECT pgmq.send('numeric_old',jsonb_build_object('n',n)) FROM generate_series(1,20) n; SELECT pgmq.send('time_old','{\"keep\":true}'); SELECT pgmq.send('ordinary_old','{\"keep\":true}'); SELECT pgmq.archive('numeric_old',1::bigint)" + before <- queryText conn preservedDataSql + installPrefix settings 6 + after <- queryText conn preservedDataSql + after @?= before + assertSql conn "(SELECT count(*)=2 FROM information_schema.columns WHERE table_schema='pgmq' AND table_name IN ('q_numeric_old','q_time_old') AND column_name='msg_id' AND identity_generation='BY DEFAULT')" + assertSql conn "(SELECT identity_generation='ALWAYS' FROM information_schema.columns WHERE table_schema='pgmq' AND table_name='q_ordinary_old' AND column_name='msg_id')" + assertSql conn "(SELECT is_identity='NO' FROM information_schema.columns WHERE table_schema='pgmq' AND table_name='a_numeric_old' AND column_name='msg_id')" + installPrefix settings 6 + runSql conn "SELECT pgmq.create_partitioned('defaults','10','100'); SELECT pgmq.create_partitioned('spill','10','100',2); SELECT pgmq.create_partitioned('spill','10','100',2)" + schema <- queryText conn "SELECT pgmq._get_pg_partman_schema()" + let qualified = "\"" <> T.replace "\"" "\"\"" schema <> "\"." + assertSql conn ("(SELECT count(*)=2 AND bool_and(premake=4) FROM " <> qualified <> "part_config WHERE parent_table IN ('pgmq.q_defaults','pgmq.a_defaults'))") + assertSql conn ("(SELECT count(*)=2 AND bool_and(premake=2) FROM " <> qualified <> "part_config WHERE parent_table IN ('pgmq.q_spill','pgmq.a_spill'))") + forM_ ["0", "-1"] $ \invalid -> do + result <- Connection.use conn (Session.script ("SELECT pgmq.create_partitioned('invalid','10','100'," <> invalid <> ")")) + case result of + Left _ -> pure () + Right _ -> assertFailure "invalid premake succeeded" + assertSql conn "to_regclass('pgmq.q_invalid') IS NULL AND NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE queue_name='invalid')" + -- Use independent connections; advisory locking plus guards must let both finish. + let createConcurrent = do + connection <- Connection.acquire settings >>= either (error . show) pure + result <- try (runSql connection "SELECT pgmq.create_partitioned('concurrent','10','100',2)") :: IO (Either SomeException ()) + Connection.release connection + pure result + box <- newEmptyMVar + _ <- forkIO (createConcurrent >>= putMVar box) + right <- createConcurrent + left <- takeMVar box + forM_ [left, right] (either (assertFailure . show) pure) + runSql conn ("UPDATE " <> qualified <> "part_config SET automatic_maintenance='off' WHERE parent_table LIKE 'pgmq.%'") + runSql conn "SELECT pgmq.send('spill',jsonb_build_object('n',n)) FROM generate_series(1,100) n; SELECT pgmq.archive('spill',100::bigint); ANALYZE pgmq.q_spill_default; ANALYZE pgmq.a_spill_default" + assertSql conn "(SELECT count(*)>0 FROM pgmq.q_spill_default) AND (SELECT count(*)=1 FROM pgmq.a_spill_default)" + assertSql conn "(SELECT default_partition_length FROM pgmq.metrics('spill')) = (SELECT count(*) FROM pgmq.q_spill_default) + (SELECT count(*) FROM pgmq.a_spill_default)" + assertSql conn "(SELECT default_partition_length IS NULL FROM pgmq.metrics('ordinary_old'))" + runSql conn "SELECT pgmq.create_unlogged('unlogged');" + assertSql conn "(SELECT default_partition_length IS NULL FROM pgmq.metrics('unlogged'))" + spillBefore <- queryText conn spillDataSql + -- Each CALL is its own simple-query request: the procedure commits internally. + forM_ ["q_spill", "a_spill"] $ \parent -> + runSql conn ("CALL " <> qualified <> "partition_data_proc('pgmq." <> parent <> "', p_wait := 0)") + runSql conn ("SELECT " <> qualified <> "run_maintenance('pgmq.q_spill'); SELECT " <> qualified <> "run_maintenance('pgmq.a_spill')") + runSql conn "ANALYZE pgmq.q_spill_default; ANALYZE pgmq.a_spill_default" + queryText conn spillDataSql >>= (@?= spillBefore) + assertSql conn "(SELECT count(*)=0 FROM pgmq.q_spill_default) AND (SELECT count(*)=0 FROM pgmq.a_spill_default) AND (SELECT default_partition_length=0 FROM pgmq.metrics('spill'))" + resetDb conn + runSql conn "DROP EXTENSION pg_partman CASCADE" + +preservedDataSql :: Text +preservedDataSql = "SELECT jsonb_agg(to_jsonb(m) ORDER BY source,msg_id)::text FROM (SELECT 'numeric' source,msg_id,message FROM pgmq.q_numeric_old UNION ALL SELECT 'archive',msg_id,message FROM pgmq.a_numeric_old UNION ALL SELECT 'time',msg_id,message FROM pgmq.q_time_old UNION ALL SELECT 'ordinary',msg_id,message FROM pgmq.q_ordinary_old) m" + +spillDataSql :: Text +spillDataSql = "SELECT jsonb_agg(to_jsonb(m) ORDER BY source,msg_id)::text FROM (SELECT 'queue' source,msg_id,message FROM pgmq.q_spill UNION ALL SELECT 'archive',msg_id,message FROM pgmq.a_spill) m" + -- | The one place the ledger is spelled out. Every other expectation in this -- suite derives from 'nativeMigrationNames', so appending a migration means -- reviewing this list and nothing else. @@ -138,7 +332,10 @@ migrationNames migrations @?= [ "0001-install-v1.11.0", "0002-schema-management-comment", - "0003-notify-crash-safety-and-locking" + "0003-notify-crash-safety-and-locking", + "0004-upgrade-v1.12.0", + "0005-upgrade-v1.13.0", + "0006-preserve-partitioned-reentry-v1.13.0" ] actual -> assertFailure ("unexpected native PGMQ plan: " <> show actual) @@ -608,6 +805,8 @@ Left err -> assertFailure ("native rerun failed after shared-ledger import: " <> show err) Right report -> (outcome <$> toList (results report)) @?= replicate ledgerLength AlreadyApplied hasCanaryComment connection >>= (@?= True) + assertSql connection "to_regprocedure('pgmq.read_grouped_head(text,integer,integer)') IS NOT NULL AND to_regprocedure('pgmq.read_grouped_head_with_poll(text,integer,integer,integer,integer)') IS NOT NULL AND to_regprocedure('pgmq.create_partitioned(text,text,text)') IS NULL AND to_regprocedure('pgmq.create_partitioned(text,text,text,integer)') IS NOT NULL" + assertSql connection "(SELECT count(*)=8 FROM pg_attribute WHERE attrelid='pgmq.metrics_result'::regclass AND attnum>0 AND NOT attisdropped)" canaryCommentStatement :: Statement.Statement () Bool canaryCommentStatement = diff --git a/test/fixtures/pgmq-1.12.0.sql b/test/fixtures/pgmq-1.12.0.sql new file mode 100644 --- /dev/null +++ b/test/fixtures/pgmq-1.12.0.sql @@ -0,0 +1,2166 @@ +------------------------------------------------------------ +-- Schema, tables, records, privileges, indexes, etc +------------------------------------------------------------ +-- When installed as an extension, we don't need to create the `pgmq` schema +-- because it is automatically created by postgres due to being declared in +-- the extension control file +DO +$$ +BEGIN + IF (SELECT NOT EXISTS( SELECT 1 FROM pg_extension WHERE extname = 'pgmq')) THEN + CREATE SCHEMA IF NOT EXISTS pgmq; + END IF; +END +$$; + +-- Table where queues and metadata about them is stored +CREATE TABLE IF NOT EXISTS pgmq.meta ( + queue_name VARCHAR UNIQUE NOT NULL, + is_partitioned BOOLEAN NOT NULL, + is_unlogged BOOLEAN NOT NULL, + created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL +); + +-- Grant permission to pg_monitor to all tables and sequences +-- These grants are intentionally placed here (after creating `pgmq.meta` but before creating other tables). This +-- allows the `pg_dump` output for a fresh installation to match the output for an installation that followed the +-- upgrade path. +GRANT USAGE ON SCHEMA pgmq TO pg_monitor; +GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pg_monitor; +GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pg_monitor; +ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pg_monitor; +ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pg_monitor; + +-- Table to track notification throttling for queues +CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.notify_insert_throttle ( + queue_name VARCHAR UNIQUE NOT NULL -- Queue name (without 'q_' prefix) + CONSTRAINT notify_insert_throttle_meta_queue_name_fk + REFERENCES pgmq.meta (queue_name) + ON DELETE CASCADE, + throttle_interval_ms INTEGER NOT NULL DEFAULT 0, -- Min milliseconds between notifications (0 = no throttling) + last_notified_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT to_timestamp(0) -- Timestamp of last sent notification +); + +CREATE INDEX IF NOT EXISTS idx_notify_throttle_active + ON pgmq.notify_insert_throttle (queue_name, last_notified_at) + WHERE throttle_interval_ms > 0; + +CREATE TABLE IF NOT EXISTS pgmq.topic_bindings +( + pattern text NOT NULL, -- Wildcard pattern for routing key matching (* = one segment, # = zero or more segments) + queue_name text NOT NULL -- Name of the queue that receives messages when pattern matches + CONSTRAINT topic_bindings_meta_queue_name_fk + REFERENCES pgmq.meta (queue_name) + ON DELETE CASCADE, + bound_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, -- Timestamp when the binding was created + compiled_regex text GENERATED ALWAYS AS ( + -- Pre-compile the pattern to regex for faster matching + -- This avoids runtime compilation on every send_topic call + '^' || + replace( + replace( + regexp_replace(pattern, '([.+?{}()|\[\]\\^$])', '\\\1', 'g'), + '*', '[^.]+' + ), + '#', '.*' + ) || '$' + ) STORED, -- Computed column: stores the compiled regex pattern + CONSTRAINT topic_bindings_unique_pattern_queue UNIQUE (pattern, queue_name) +); + +-- Create covering index for better performance when scanning patterns +-- Includes queue_name and compiled_regex to allow index-only scans (no table access needed) +CREATE INDEX IF NOT EXISTS idx_topic_bindings_covering ON pgmq.topic_bindings (pattern) INCLUDE (queue_name, compiled_regex); + +-- Allow the following `pgmq` tables to be dumped by `pg_dump` when pgmq is installed as an extension +DO +$$ +BEGIN + IF EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pgmq') THEN + PERFORM pg_catalog.pg_extension_config_dump('pgmq.meta', ''); + PERFORM pg_catalog.pg_extension_config_dump('pgmq.notify_insert_throttle', ''); + PERFORM pg_catalog.pg_extension_config_dump('pgmq.topic_bindings', ''); + END IF; +END +$$; + +-- This type has the shape of a message in a queue, and is often returned by +-- pgmq functions that return messages. +-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation +-- expects a specific order of fields. +CREATE TYPE pgmq.message_record AS ( + msg_id BIGINT, + read_ct INTEGER, + enqueued_at TIMESTAMP WITH TIME ZONE, + last_read_at TIMESTAMP WITH TIME ZONE, + vt TIMESTAMP WITH TIME ZONE, + message JSONB, + headers JSONB +); + +-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation +-- expects a specific order of fields. +CREATE TYPE pgmq.queue_record AS ( + queue_name VARCHAR, + is_partitioned BOOLEAN, + is_unlogged BOOLEAN, + created_at TIMESTAMP WITH TIME ZONE +); + +------------------------------------------------------------ +-- Functions +------------------------------------------------------------ + +-- prevents race conditions during queue creation by acquiring a transaction-level advisory lock +-- uses a transaction advisory lock maintain the lock until transaction commit +-- a race condition would still exist if lock was released before commit +CREATE FUNCTION pgmq.acquire_queue_lock(queue_name TEXT) +RETURNS void AS $$ +BEGIN + PERFORM pg_advisory_xact_lock(hashtext('pgmq.queue_' || queue_name)); +END; +$$ LANGUAGE plpgsql; + +-- read_grouped_round_robin +-- reads messages while preserving FIFO within groups and interleaving across groups (layered round-robin) +CREATE FUNCTION pgmq.read_grouped_rr( + queue_name TEXT, + vt INTEGER, + qty INTEGER +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH fifo_groups AS ( + -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility + SELECT + COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key, + MIN(msg_id) AS head_msg_id + FROM pgmq.%1$I + GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') + ), + eligible_groups AS ( + -- Only groups whose head message is currently visible + -- Acquire a transaction-level advisory lock per group to prevent concurrent selection + SELECT + g.fifo_key, + g.head_msg_id, + ROW_NUMBER() OVER (ORDER BY g.head_msg_id) AS group_priority + FROM fifo_groups g + JOIN pgmq.%2$I h ON h.msg_id = g.head_msg_id + WHERE h.vt <= clock_timestamp() + AND pg_try_advisory_xact_lock(pg_catalog.hashtextextended(g.fifo_key, 0)) + ), + available_messages AS ( + -- All currently visible messages starting at the head for each eligible group + SELECT + m.msg_id, + eg.group_priority, + ROW_NUMBER() OVER ( + PARTITION BY eg.fifo_key + ORDER BY m.msg_id + ) AS msg_rank_in_group + FROM pgmq.%3$I m + JOIN eligible_groups eg + ON COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = eg.fifo_key + WHERE m.vt <= clock_timestamp() + AND m.msg_id >= eg.head_msg_id + ), + ordered_messages AS ( + -- Layered round-robin: take rank 1 of all groups by group_priority, then rank 2, etc. + -- Assign selection order before locking + SELECT msg_id, ROW_NUMBER() OVER (ORDER BY msg_rank_in_group, group_priority) as selection_order + FROM available_messages + ), + selected_messages AS ( + -- Lock the messages in the correct order, preserving selection_order + SELECT om.msg_id, om.selection_order + FROM ordered_messages om + JOIN pgmq.%4$I m ON m.msg_id = om.msg_id + WHERE om.selection_order <= $1 + ORDER BY om.selection_order + FOR UPDATE OF m SKIP LOCKED + ), + updated_messages AS ( + UPDATE pgmq.%5$I m + SET + vt = clock_timestamp() + %6$L, + read_ct = read_ct + 1, + last_read_at = clock_timestamp() + FROM selected_messages sm + WHERE m.msg_id = sm.msg_id + AND m.vt <= clock_timestamp() -- final guard to avoid duplicate reads under races + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers, sm.selection_order + ) + SELECT msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers + FROM updated_messages + ORDER BY selection_order; + $QUERY$, + qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt) + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + +-- read_grouped_rr_with_poll +-- reads messages using round-robin layering across groups, with polling support +CREATE FUNCTION pgmq.read_grouped_rr_with_poll( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + max_poll_seconds INTEGER DEFAULT 5, + poll_interval_ms INTEGER DEFAULT 100 +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + r pgmq.message_record; + stop_at TIMESTAMP; +BEGIN + stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds); + LOOP + IF (SELECT clock_timestamp() >= stop_at) THEN + RETURN; + END IF; + + FOR r IN + SELECT * FROM pgmq.read_grouped_rr(queue_name, vt, qty) + LOOP + RETURN NEXT r; + END LOOP; + IF FOUND THEN + RETURN; + ELSE + PERFORM pg_sleep(poll_interval_ms::numeric / 1000); + END IF; + END LOOP; +END; +$$ LANGUAGE plpgsql; + +-- read_grouped_head: read the head of N different FIFO groups in a single operation. +-- This supports horizontal scaling by processing groups in parallel while ensuring message ordering is preserved per group. +CREATE FUNCTION pgmq.read_grouped_head( + queue_name TEXT, + vt INTEGER, + qty INTEGER +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH fifo_groups AS ( + -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility + SELECT + COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key, + MIN(msg_id) AS head_msg_id + FROM pgmq.%1$I + GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') + ), + selected_messages AS ( + -- Take at most 1 message per group + SELECT g.head_msg_id msg_id + FROM fifo_groups g + JOIN pgmq.%1$I q ON q.msg_id = g.head_msg_id + WHERE q.vt <= clock_timestamp() + ORDER BY q.msg_id + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + UPDATE pgmq.%1$I m + SET + vt = clock_timestamp() + %2$L, + read_ct = read_ct + 1, + last_read_at = clock_timestamp() + FROM selected_messages sm + WHERE m.msg_id = sm.msg_id + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers; + $QUERY$, + qtable, make_interval(secs => vt) + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + +-- read_grouped_head_with_poll +-- reads the head of N different FIFO groups in a single operation, with polling support +CREATE FUNCTION pgmq.read_grouped_head_with_poll( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + max_poll_seconds INTEGER DEFAULT 5, + poll_interval_ms INTEGER DEFAULT 100 +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + r pgmq.message_record; + stop_at TIMESTAMPTZ; +BEGIN + stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds); + LOOP + IF clock_timestamp() >= stop_at THEN + RETURN; + END IF; + + FOR r IN + SELECT * FROM pgmq.read_grouped_head(queue_name, vt, qty) + LOOP + RETURN NEXT r; + END LOOP; + IF FOUND THEN + RETURN; + ELSE + PERFORM pg_sleep(poll_interval_ms::numeric / 1000); + END IF; + END LOOP; +END; +$$ LANGUAGE plpgsql; + +-- a helper to format table names and check for invalid characters +CREATE FUNCTION pgmq.format_table_name(queue_name text, prefix text) +RETURNS TEXT AS $$ +BEGIN + IF queue_name ~ '\$|;|--|''' + THEN + RAISE EXCEPTION 'queue name contains invalid characters: $, ;, --, or \'''; + END IF; + RETURN lower(prefix || '_' || queue_name); +END; +$$ LANGUAGE plpgsql; + +-- read +-- reads a number of messages from a queue, setting a visibility timeout on them +CREATE FUNCTION pgmq.read( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + conditional JSONB DEFAULT '{}' +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH cte AS + ( + SELECT msg_id + FROM pgmq.%I + WHERE vt <= clock_timestamp() AND CASE + WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer + ELSE 1 + END = 1 + ORDER BY msg_id ASC + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + UPDATE pgmq.%I m + SET + last_read_at = clock_timestamp(), + vt = clock_timestamp() + %L, + read_ct = read_ct + 1 + FROM cte + WHERE m.msg_id = cte.msg_id + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers; + $QUERY$, + qtable, conditional, qtable, make_interval(secs => vt) + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + +-- read_grouped +-- reads messages with AWS SQS FIFO-style batch retrieval behavior +-- attempts to return as many messages as possible from the same message group +CREATE FUNCTION pgmq.read_grouped( + queue_name TEXT, + vt INTEGER, + qty INTEGER +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH fifo_groups AS ( + -- Find the minimum msg_id for each FIFO group that's ready to be processed + SELECT + COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') as fifo_key, + MIN(msg_id) as min_msg_id + FROM pgmq.%I + WHERE vt <= clock_timestamp() + GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') + ), + locked_groups AS ( + -- Lock the first available message in each FIFO group + SELECT + m.msg_id, + fg.fifo_key + FROM pgmq.%I m + INNER JOIN fifo_groups fg ON + COALESCE(m.headers->>'x-pgmq-group', '_default_fifo_group') = fg.fifo_key + AND m.msg_id = fg.min_msg_id + WHERE m.vt <= clock_timestamp() + ORDER BY m.msg_id ASC + FOR UPDATE SKIP LOCKED + ), + group_priorities AS ( + -- Assign priority to groups based on their oldest message + SELECT + fifo_key, + msg_id as min_msg_id, + ROW_NUMBER() OVER (ORDER BY msg_id) as group_priority + FROM locked_groups + ), + filtered_groups as ( + SELECT * FROM group_priorities gp + WHERE NOT EXISTS ( + -- Ensure no earlier message in this group is currently being processed + SELECT 1 + FROM pgmq.%I m2 + WHERE COALESCE(m2.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key + AND m2.vt > clock_timestamp() + AND m2.msg_id < gp.min_msg_id + ) + ), + available_messages as ( + SELECT gp.fifo_key, t.msg_id,gp.group_priority, + ROW_NUMBER() OVER (PARTITION BY gp.fifo_key ORDER BY t.msg_id) as msg_rank_in_group + FROM filtered_groups gp + CROSS JOIN LATERAL ( + SELECT * + FROM pgmq.%I t + WHERE COALESCE(t.headers->>'x-pgmq-group', '_default_fifo_group') = gp.fifo_key + AND t.vt <= clock_timestamp() + ORDER BY msg_id + LIMIT $1 -- tip to limit query impact, we know we need at most qty in each group + ) t + ORDER BY gp.group_priority + ), + batch_selection AS ( + -- Select messages to fill batch, prioritizing earliest group + SELECT + msg_id, + ROW_NUMBER() OVER (ORDER BY group_priority, msg_rank_in_group) as overall_rank + FROM available_messages + ), + selected_messages AS ( + -- Limit to requested quantity + SELECT msg_id + FROM batch_selection + WHERE overall_rank <= $1 + ORDER BY msg_id + FOR UPDATE SKIP LOCKED + ) + UPDATE pgmq.%I m + SET + vt = clock_timestamp() + %L, + read_ct = read_ct + 1, + last_read_at = clock_timestamp() + FROM selected_messages sm + WHERE m.msg_id = sm.msg_id + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers; + $QUERY$, + qtable, qtable, qtable, qtable, qtable, make_interval(secs => vt) + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + +-- read_grouped_with_poll +-- reads messages with AWS SQS FIFO-style batch retrieval behavior, with polling support +CREATE FUNCTION pgmq.read_grouped_with_poll( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + max_poll_seconds INTEGER DEFAULT 5, + poll_interval_ms INTEGER DEFAULT 100 +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + r pgmq.message_record; + stop_at TIMESTAMP; +BEGIN + stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds); + LOOP + IF (SELECT clock_timestamp() >= stop_at) THEN + RETURN; + END IF; + + FOR r IN + SELECT * FROM pgmq.read_grouped(queue_name, vt, qty) + LOOP + RETURN NEXT r; + END LOOP; + IF FOUND THEN + RETURN; + ELSE + PERFORM pg_sleep(poll_interval_ms::numeric / 1000); + END IF; + END LOOP; +END; +$$ LANGUAGE plpgsql; + +---- read_with_poll +---- reads a number of messages from a queue, setting a visibility timeout on them +CREATE FUNCTION pgmq.read_with_poll( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + max_poll_seconds INTEGER DEFAULT 5, + poll_interval_ms INTEGER DEFAULT 100, + conditional JSONB DEFAULT '{}' +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + r pgmq.message_record; + stop_at TIMESTAMP; + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds); + LOOP + IF (SELECT clock_timestamp() >= stop_at) THEN + RETURN; + END IF; + + sql := FORMAT( + $QUERY$ + WITH cte AS + ( + SELECT msg_id + FROM pgmq.%I + WHERE vt <= clock_timestamp() AND CASE + WHEN %L != '{}'::jsonb THEN (message @> %2$L)::integer + ELSE 1 + END = 1 + ORDER BY msg_id ASC + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + UPDATE pgmq.%I m + SET + last_read_at = clock_timestamp(), + vt = clock_timestamp() + %L, + read_ct = read_ct + 1 + FROM cte + WHERE m.msg_id = cte.msg_id + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers; + $QUERY$, + qtable, conditional, qtable, make_interval(secs => vt) + ); + + FOR r IN + EXECUTE sql USING qty + LOOP + RETURN NEXT r; + END LOOP; + IF FOUND THEN + RETURN; + ELSE + PERFORM pg_sleep(poll_interval_ms::numeric / 1000); + END IF; + END LOOP; +END; +$$ LANGUAGE plpgsql; + +---- archive +---- removes a message from the queue, and sends it to the archive, where its +---- saved permanently. +CREATE FUNCTION pgmq.archive( + queue_name TEXT, + msg_id BIGINT +) +RETURNS BOOLEAN AS $$ +DECLARE + sql TEXT; + result BIGINT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + atable TEXT := pgmq.format_table_name(queue_name, 'a'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH archived AS ( + DELETE FROM pgmq.%I + WHERE msg_id = $1 + RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers + ) + INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers) + SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers + FROM archived + RETURNING msg_id; + $QUERY$, + qtable, atable + ); + EXECUTE sql USING msg_id INTO result; + RETURN NOT (result IS NULL); +END; +$$ LANGUAGE plpgsql; + +---- archive +---- removes an array of message ids from the queue, and sends it to the archive, +---- where these messages will be saved permanently. +CREATE FUNCTION pgmq.archive( + queue_name TEXT, + msg_ids BIGINT[] +) +RETURNS SETOF BIGINT AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + atable TEXT := pgmq.format_table_name(queue_name, 'a'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH archived AS ( + DELETE FROM pgmq.%I + WHERE msg_id = ANY($1) + RETURNING msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers + ) + INSERT INTO pgmq.%I (msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers) + SELECT msg_id, vt, read_ct, enqueued_at, last_read_at, message, headers + FROM archived + RETURNING msg_id; + $QUERY$, + qtable, atable + ); + RETURN QUERY EXECUTE sql USING msg_ids; +END; +$$ LANGUAGE plpgsql; + +---- delete +---- deletes a message id from the queue permanently +CREATE FUNCTION pgmq.delete( + queue_name TEXT, + msg_id BIGINT +) +RETURNS BOOLEAN AS $$ +DECLARE + sql TEXT; + result BIGINT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + DELETE FROM pgmq.%I + WHERE msg_id = $1 + RETURNING msg_id + $QUERY$, + qtable + ); + EXECUTE sql USING msg_id INTO result; + RETURN NOT (result IS NULL); +END; +$$ LANGUAGE plpgsql; + +---- delete +---- deletes an array of message ids from the queue permanently +CREATE FUNCTION pgmq.delete( + queue_name TEXT, + msg_ids BIGINT[] +) +RETURNS SETOF BIGINT AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + DELETE FROM pgmq.%I + WHERE msg_id = ANY($1) + RETURNING msg_id + $QUERY$, + qtable + ); + RETURN QUERY EXECUTE sql USING msg_ids; +END; +$$ LANGUAGE plpgsql; + +-- 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, + msg JSONB +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp()); +$$ LANGUAGE sql; + +-- send: 3 args with headers +CREATE FUNCTION pgmq.send( + queue_name TEXT, + msg JSONB, + headers JSONB +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp()); +$$ LANGUAGE sql; + +-- send: 3 args with integer delay +CREATE FUNCTION pgmq.send( + queue_name TEXT, + msg JSONB, + delay INTEGER +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send(queue_name, msg, NULL, clock_timestamp() + make_interval(secs => delay)); +$$ LANGUAGE sql; + +-- send: 3 args with timestamp +CREATE FUNCTION pgmq.send( + queue_name TEXT, + msg JSONB, + delay TIMESTAMP WITH TIME ZONE +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send(queue_name, msg, NULL, delay); +$$ LANGUAGE sql; + +-- send: 4 args with integer delay +CREATE FUNCTION pgmq.send( + queue_name TEXT, + msg JSONB, + headers JSONB, + delay INTEGER +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send(queue_name, msg, headers, clock_timestamp() + make_interval(secs => delay)); +$$ LANGUAGE sql; + +-- _validate_batch_params: Private function to validate batch parameters +CREATE FUNCTION pgmq._validate_batch_params( + msgs JSONB[], + headers JSONB[] +) RETURNS void AS $$ +BEGIN + -- Validate that msgs is not NULL or empty + IF msgs IS NULL OR array_length(msgs, 1) IS NULL THEN + RAISE EXCEPTION 'msgs cannot be NULL or empty'; + END IF; + + -- Validate that headers array length matches msgs array length if headers is provided + -- Note: array_length returns NULL for empty arrays, so we use COALESCE to treat empty arrays as length 0 + IF headers IS NOT NULL AND COALESCE(array_length(headers, 1), 0) != COALESCE(array_length(msgs, 1), 0) THEN + RAISE EXCEPTION 'headers array length (%) must match msgs array length (%)', + COALESCE(array_length(headers, 1), 0), COALESCE(array_length(msgs, 1), 0); + END IF; +END; +$$ LANGUAGE plpgsql; + +-- _send_batch: Private function that performs the actual batch insert without validation +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; + +-- send_batch: Public function with validation +CREATE FUNCTION pgmq.send_batch( + queue_name TEXT, + msgs JSONB[], + headers JSONB[], + delay TIMESTAMP WITH TIME ZONE +) RETURNS SETOF BIGINT AS $$ +BEGIN + PERFORM pgmq._validate_batch_params(msgs, headers); + RETURN QUERY SELECT * FROM pgmq._send_batch(queue_name, msgs, headers, delay); +END; +$$ LANGUAGE plpgsql; + +-- send batch: 2 args +CREATE FUNCTION pgmq.send_batch( + queue_name TEXT, + msgs JSONB[] +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp()); +$$ LANGUAGE sql; + +-- send batch: 3 args with headers +CREATE FUNCTION pgmq.send_batch( + queue_name TEXT, + msgs JSONB[], + headers JSONB[] +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp()); +$$ LANGUAGE sql; + +-- send batch: 3 args with integer delay +CREATE FUNCTION pgmq.send_batch( + queue_name TEXT, + msgs JSONB[], + delay INTEGER +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, clock_timestamp() + make_interval(secs => delay)); +$$ LANGUAGE sql; + +-- send batch: 3 args with timestamp +CREATE FUNCTION pgmq.send_batch( + queue_name TEXT, + msgs JSONB[], + delay TIMESTAMP WITH TIME ZONE +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send_batch(queue_name, msgs, NULL, delay); +$$ LANGUAGE sql; + +-- send_batch: 4 args with integer delay +CREATE FUNCTION pgmq.send_batch( + queue_name TEXT, + msgs JSONB[], + headers JSONB[], + delay INTEGER +) RETURNS SETOF BIGINT AS $$ + SELECT * FROM pgmq.send_batch(queue_name, msgs, headers, clock_timestamp() + make_interval(secs => delay)); +$$ LANGUAGE sql; + +-- returned by pgmq.metrics() and pgmq.metrics_all +-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation +-- expects a specific order of fields. +CREATE TYPE pgmq.metrics_result AS ( + queue_name text, + queue_length bigint, + newest_msg_age_sec int, + oldest_msg_age_sec int, + total_messages bigint, + scrape_time timestamp with time zone, + queue_visible_length bigint +); + +-- get metrics for a single queue +CREATE FUNCTION pgmq.metrics(queue_name TEXT) +RETURNS pgmq.metrics_result AS $$ +DECLARE + result_row pgmq.metrics_result; + query TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + query := FORMAT( + $QUERY$ + WITH q_summary AS ( + SELECT + count(*) as queue_length, + count(CASE WHEN vt <= NOW() THEN 1 END) as queue_visible_length, + EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec, + EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec, + NOW() as scrape_time + FROM pgmq.%I + ), + all_metrics AS ( + SELECT CASE + WHEN is_called THEN last_value ELSE 0 + END as total_messages + FROM pgmq.%I + ) + SELECT + %L as queue_name, + q_summary.queue_length, + q_summary.newest_msg_age_sec, + q_summary.oldest_msg_age_sec, + all_metrics.total_messages, + q_summary.scrape_time, + q_summary.queue_visible_length + FROM q_summary, all_metrics + $QUERY$, + qtable, qtable || '_msg_id_seq', queue_name + ); + EXECUTE query INTO result_row; + RETURN result_row; +END; +$$ LANGUAGE plpgsql; + +-- get metrics for all queues +CREATE FUNCTION pgmq."metrics_all"() +RETURNS SETOF pgmq.metrics_result AS $$ +DECLARE + row_name RECORD; + result_row pgmq.metrics_result; +BEGIN + FOR row_name IN SELECT queue_name FROM pgmq.meta LOOP + result_row := pgmq.metrics(row_name.queue_name); + RETURN NEXT result_row; + END LOOP; +END; +$$ LANGUAGE plpgsql; + +-- list queues +CREATE FUNCTION pgmq."list_queues"() +RETURNS SETOF pgmq.queue_record AS $$ +BEGIN + RETURN QUERY SELECT * FROM pgmq.meta; +END +$$ LANGUAGE plpgsql; + +-- purge queue, deleting all entries in it. +CREATE OR REPLACE FUNCTION pgmq."purge_queue"(queue_name TEXT) +RETURNS BIGINT AS $$ +DECLARE + deleted_count INTEGER; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + -- Get the row count before truncating + EXECUTE format('SELECT count(*) FROM pgmq.%I', qtable) INTO deleted_count; + + -- Use TRUNCATE for better performance on large tables + EXECUTE format('TRUNCATE TABLE pgmq.%I', qtable); + + -- Return the number of purged rows + RETURN deleted_count; +END +$$ LANGUAGE plpgsql; + +-- unassign archive, so it can be kept when a queue is deleted +CREATE FUNCTION pgmq."detach_archive"(queue_name TEXT) +RETURNS VOID AS $$ +DECLARE + atable TEXT := pgmq.format_table_name(queue_name, 'a'); +BEGIN + RAISE WARNING 'detach_archive(queue_name) is deprecated and is a no-op. It will be removed in PGMQ v2.0. Archive tables are no longer member objects.'; +END +$$ LANGUAGE plpgsql; + +-- pop: implementation +CREATE FUNCTION pgmq.pop(queue_name TEXT, qty INTEGER DEFAULT 1) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + result pgmq.message_record; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH cte AS + ( + SELECT msg_id + FROM pgmq.%I + WHERE vt <= clock_timestamp() + ORDER BY msg_id ASC + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + DELETE from pgmq.%I + WHERE msg_id IN (select msg_id from cte) + RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers; + $QUERY$, + qtable, qtable + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + +-- Sets timestamp vt of a message, returns it +CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt TIMESTAMP WITH TIME ZONE) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + result pgmq.message_record; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + UPDATE pgmq.%I + SET vt = $1 + WHERE msg_id = $2 + RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers; + $QUERY$, + qtable + ); + RETURN QUERY EXECUTE sql USING vt, msg_id; +END; +$$ LANGUAGE plpgsql; + +-- Sets integer vt of a message, returns it +CREATE FUNCTION pgmq.set_vt(queue_name TEXT, msg_id BIGINT, vt INTEGER) +RETURNS SETOF pgmq.message_record AS $$ + SELECT * FROM pgmq.set_vt(queue_name, msg_id, clock_timestamp() + make_interval(secs => vt)); +$$ LANGUAGE sql; + +-- Sets timestamp vt of multiple messages, returns them +CREATE FUNCTION pgmq.set_vt( + queue_name TEXT, + msg_ids BIGINT[], + vt TIMESTAMP WITH TIME ZONE +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + UPDATE pgmq.%I + SET vt = $1 + WHERE msg_id = ANY($2) + RETURNING msg_id, read_ct, enqueued_at, last_read_at, vt, message, headers; + $QUERY$, + qtable + ); + RETURN QUERY EXECUTE sql USING vt, msg_ids; +END; +$$ LANGUAGE plpgsql; + +-- Sets integer vt of multiple messages, returns them +CREATE FUNCTION pgmq.set_vt( + queue_name TEXT, + msg_ids BIGINT[], + vt INTEGER +) +RETURNS SETOF pgmq.message_record AS $$ + SELECT * FROM pgmq.set_vt(queue_name, msg_ids, clock_timestamp() + make_interval(secs => vt)); +$$ LANGUAGE sql; + +CREATE FUNCTION pgmq._get_pg_partman_schema() +RETURNS TEXT AS $$ + SELECT + extnamespace::regnamespace::text + FROM + pg_extension + WHERE + extname = 'pg_partman'; +$$ LANGUAGE SQL; + +CREATE FUNCTION pgmq.drop_queue(queue_name TEXT, partitioned BOOLEAN) +RETURNS BOOLEAN AS $$ +DECLARE + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + fq_qtable TEXT := 'pgmq.' || qtable; + atable TEXT := pgmq.format_table_name(queue_name, 'a'); + fq_atable TEXT := 'pgmq.' || atable; +BEGIN + RAISE WARNING 'drop_queue(queue_name, partitioned) is deprecated and will be removed in PGMQ v2.0. Use drop_queue(queue_name) instead'; + + PERFORM pgmq.drop_queue(queue_name); + + RETURN TRUE; +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq.drop_queue(queue_name TEXT) +RETURNS BOOLEAN AS $$ +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 + PERFORM pgmq.acquire_queue_lock(queue_name); + EXECUTE FORMAT( + $QUERY$ + SELECT is_partitioned FROM pgmq.meta WHERE queue_name = %L + $QUERY$, + queue_name + ) INTO partitioned; + + -- check if the queue exists + IF NOT EXISTS ( + SELECT 1 + FROM information_schema.tables + WHERE table_name = qtable and table_schema = 'pgmq' + ) THEN + RAISE NOTICE 'pgmq queue `%` does not exist', queue_name; + RETURN FALSE; + END IF; + + EXECUTE FORMAT( + $QUERY$ + DROP TABLE IF EXISTS pgmq.%I + $QUERY$, + qtable + ); + + EXECUTE FORMAT( + $QUERY$ + DROP TABLE IF EXISTS pgmq.%I + $QUERY$, + atable + ); + + IF EXISTS ( + SELECT 1 + FROM information_schema.tables + WHERE table_name = 'meta' and table_schema = 'pgmq' + ) THEN + EXECUTE FORMAT( + $QUERY$ + DELETE FROM pgmq.meta WHERE queue_name = %L + $QUERY$, + queue_name + ); + END IF; + + IF partitioned THEN + EXECUTE FORMAT( + $QUERY$ + DELETE FROM %I.part_config where parent_table in (%L, %L) + $QUERY$, + pgmq._get_pg_partman_schema(), fq_qtable, fq_atable + ); + END IF; + + RETURN TRUE; +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq.validate_queue_name(queue_name TEXT) +RETURNS void AS $$ +BEGIN + IF length(queue_name) > 47 THEN + -- complete table identifier must be <= 63 + -- https://www.postgresql.org/docs/17/sql-syntax-lexical.html#SQL-SYNTAX-IDENTIFIERS + -- e.g. template_pgmq_q_my_queue is an identifier for my_queue when partitioned + -- template_pgmq_q_ (16) + (47) = 63 + RAISE EXCEPTION 'queue name is too long, maximum length is 47 characters'; + END IF; +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq._belongs_to_pgmq(table_name TEXT) +RETURNS BOOLEAN AS $$ +DECLARE + sql TEXT; + result BOOLEAN; +BEGIN + SELECT EXISTS ( + SELECT 1 + FROM pg_depend + WHERE refobjid = (SELECT oid FROM pg_extension WHERE extname = 'pgmq') + AND objid = ( + SELECT oid + FROM pg_class + WHERE relname = table_name + ) + ) INTO result; + RETURN result; +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq.create_non_partitioned(queue_name TEXT) +RETURNS void AS $$ +DECLARE + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + qtable_seq TEXT := qtable || '_msg_id_seq'; + atable TEXT := pgmq.format_table_name(queue_name, 'a'); +BEGIN + PERFORM pgmq.validate_queue_name(queue_name); + PERFORM pgmq.acquire_queue_lock(queue_name); + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) + $QUERY$, + qtable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT PRIMARY KEY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ); + $QUERY$, + atable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC); + $QUERY$, + qtable || '_vt_idx', qtable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at); + $QUERY$, + 'archived_at_idx_' || queue_name, atable + ); + + EXECUTE FORMAT( + $QUERY$ + INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged) + VALUES (%L, false, false) + ON CONFLICT + DO NOTHING; + $QUERY$, + queue_name + ); + +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq.create_unlogged(queue_name TEXT) +RETURNS void AS $$ +DECLARE + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + qtable_seq TEXT := qtable || '_msg_id_seq'; + atable TEXT := pgmq.format_table_name(queue_name, 'a'); +BEGIN + PERFORM pgmq.validate_queue_name(queue_name); + PERFORM pgmq.acquire_queue_lock(queue_name); + + EXECUTE FORMAT( + $QUERY$ + CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT PRIMARY KEY GENERATED ALWAYS AS IDENTITY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) + $QUERY$, + qtable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT PRIMARY KEY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ); + $QUERY$, + atable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (vt ASC); + $QUERY$, + qtable || '_vt_idx', qtable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at); + $QUERY$, + 'archived_at_idx_' || queue_name, atable + ); + + EXECUTE FORMAT( + $QUERY$ + INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged) + VALUES (%L, false, true) + ON CONFLICT + DO NOTHING; + $QUERY$, + queue_name + ); +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq._get_partition_col(partition_interval TEXT) +RETURNS TEXT AS $$ +DECLARE + num INTEGER; +BEGIN + BEGIN + num := partition_interval::INTEGER; + RETURN 'msg_id'; + EXCEPTION + WHEN others THEN + RETURN 'enqueued_at'; + END; +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq._extension_exists(extension_name TEXT) + RETURNS BOOLEAN + LANGUAGE SQL +AS $$ +SELECT EXISTS ( + SELECT 1 + FROM pg_extension + WHERE extname = extension_name +) +$$; + +CREATE FUNCTION pgmq._ensure_pg_partman_installed() +RETURNS void AS $$ +BEGIN + IF NOT pgmq._extension_exists('pg_partman') THEN + RAISE EXCEPTION 'pg_partman is required for partitioned queues'; + END IF; +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq._get_pg_partman_major_version() +RETURNS INT +LANGUAGE SQL +AS $$ + SELECT split_part(extversion, '.', 1)::INT + FROM pg_extension + WHERE extname = 'pg_partman' +$$; + +CREATE FUNCTION pgmq.create_partitioned( + queue_name TEXT, + partition_interval TEXT DEFAULT '10000', + retention_interval TEXT DEFAULT '100000' +) +RETURNS void AS $$ +DECLARE + partition_col TEXT; + a_partition_col TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + qtable_seq TEXT := qtable || '_msg_id_seq'; + atable TEXT := pgmq.format_table_name(queue_name, 'a'); + fq_qtable TEXT := 'pgmq.' || qtable; + fq_atable TEXT := 'pgmq.' || atable; +BEGIN + PERFORM pgmq.validate_queue_name(queue_name); + PERFORM pgmq.acquire_queue_lock(queue_name); + PERFORM pgmq._ensure_pg_partman_installed(); + SELECT pgmq._get_partition_col(partition_interval) INTO partition_col; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT GENERATED ALWAYS AS IDENTITY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I) + $QUERY$, + qtable, partition_col + ); + + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_qtable, + partition_col, + partition_interval + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I); + $QUERY$, + qtable || '_part_idx', qtable, partition_col + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || qtable + ); + + EXECUTE FORMAT( + $QUERY$ + INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged) + VALUES (%L, true, false) + ON CONFLICT + DO NOTHING; + $QUERY$, + queue_name + ); + + IF partition_col = 'enqueued_at' THEN + a_partition_col := 'archived_at'; + ELSE + a_partition_col := partition_col; + END IF; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT NOT NULL, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I); + $QUERY$, + atable, a_partition_col + ); + + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_atable, + a_partition_col, + partition_interval + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || atable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at); + $QUERY$, + 'archived_at_idx_' || queue_name, atable + ); + +END; +$$ LANGUAGE plpgsql; + +CREATE FUNCTION pgmq.create(queue_name TEXT) +RETURNS void AS $$ +BEGIN + PERFORM pgmq.create_non_partitioned(queue_name); +END; +$$ LANGUAGE plpgsql; + +-- _create_fifo_index_if_not_exists +-- internal function to create GIN index on headers for better FIFO performance +CREATE OR REPLACE FUNCTION pgmq._create_fifo_index_if_not_exists(queue_name TEXT) +RETURNS void AS $$ +DECLARE + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + index_name TEXT := qtable || '_fifo_idx'; +BEGIN + -- Create GIN index on headers for efficient FIFO key lookups + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I USING GIN (headers); + $QUERY$, + index_name, qtable + ); +END; +$$ LANGUAGE plpgsql; + +-- create_fifo_index +-- creates a GIN index on the headers column to improve FIFO read performance +CREATE FUNCTION pgmq.create_fifo_index(queue_name TEXT) +RETURNS void AS $$ +BEGIN + PERFORM pgmq._create_fifo_index_if_not_exists(queue_name); +END; +$$ LANGUAGE plpgsql; + +-- create_fifo_indexes_all +-- creates FIFO indexes on all existing queues +CREATE FUNCTION pgmq.create_fifo_indexes_all() +RETURNS void AS $$ +DECLARE + queue_record RECORD; +BEGIN + FOR queue_record IN SELECT queue_name FROM pgmq.meta LOOP + PERFORM pgmq.create_fifo_index(queue_record.queue_name); + END LOOP; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION pgmq.convert_archive_partitioned( + table_name TEXT, + partition_interval TEXT DEFAULT '10000', + retention_interval TEXT DEFAULT '100000', + leading_partition INT DEFAULT 10 +) +RETURNS void AS $$ +DECLARE + a_table_name TEXT := pgmq.format_table_name(table_name, 'a'); + a_table_name_old TEXT := pgmq.format_table_name(table_name, 'a') || '_old'; + qualified_a_table_name TEXT := format('pgmq.%I', a_table_name); + partition_col TEXT; + a_partition_col TEXT; +BEGIN + + PERFORM c.relkind + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.relname = a_table_name + AND c.relkind = 'p'; + + IF FOUND THEN + RAISE NOTICE 'Table %s is already partitioned', a_table_name; + RETURN; + END IF; + + PERFORM c.relkind + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE c.relname = a_table_name + AND c.relkind = 'r'; + + IF NOT FOUND THEN + RAISE NOTICE 'Table %s does not exists', a_table_name; + RETURN; + END IF; + + SELECT pgmq._get_partition_col(partition_interval) INTO partition_col; + + -- For archive tables, use archived_at for time-based partitioning + IF partition_col = 'enqueued_at' THEN + a_partition_col := 'archived_at'; + ELSE + a_partition_col := partition_col; + END IF; + + EXECUTE 'ALTER TABLE ' || qualified_a_table_name || ' RENAME TO ' || a_table_name_old; + + -- When partitioning by time (archived_at), we need to exclude constraints and indexes + -- because the existing PRIMARY KEY on msg_id alone is incompatible with partitioning by archived_at. + -- When partitioning by msg_id, we can keep all constraints including PRIMARY KEY. + IF a_partition_col = 'archived_at' THEN + EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including defaults including generated including storage including comments) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col ); + ELSE + EXECUTE format( 'CREATE TABLE pgmq.%I (LIKE pgmq.%I including all) PARTITION BY RANGE (%I)', a_table_name, a_table_name_old, a_partition_col ); + END IF; + + EXECUTE 'ALTER INDEX pgmq.archived_at_idx_' || table_name || ' RENAME TO archived_at_idx_' || table_name || '_old'; + EXECUTE 'CREATE INDEX archived_at_idx_'|| table_name || ' ON ' || qualified_a_table_name ||'(archived_at)'; + + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + qualified_a_table_name, + a_partition_col, + partition_interval + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = false, + infinite_time_partitions = true + WHERE + parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + qualified_a_table_name + ); +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION pgmq.notify_queue_listeners() +RETURNS TRIGGER AS $$ +DECLARE + queue_name_extracted TEXT; -- Queue name extracted from trigger table name + updated_count INTEGER; -- Number of rows updated (0 or 1) +BEGIN + queue_name_extracted := substring(TG_TABLE_NAME from 3); + + UPDATE pgmq.notify_insert_throttle + SET last_notified_at = clock_timestamp() + WHERE queue_name = queue_name_extracted + AND ( + throttle_interval_ms = 0 -- No throttling configured + OR clock_timestamp() - last_notified_at >= + (throttle_interval_ms * INTERVAL '1 millisecond') -- Throttle interval has elapsed + ); + + -- Check how many rows were updated (will be 0 or 1) + GET DIAGNOSTICS updated_count = ROW_COUNT; + + IF updated_count > 0 THEN + PERFORM PG_NOTIFY('pgmq.' || TG_TABLE_NAME || '.' || TG_OP, NULL); + END IF; + +RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION pgmq.enable_notify_insert(queue_name TEXT, throttle_interval_ms INTEGER DEFAULT 250) +RETURNS void AS $$ +DECLARE + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + v_queue_name TEXT := queue_name; + v_throttle_interval_ms INTEGER := throttle_interval_ms; +BEGIN + -- Validate that throttle_interval_ms is non-negative + IF v_throttle_interval_ms < 0 THEN + RAISE EXCEPTION 'throttle_interval_ms must be non-negative'; + END IF; + + -- Validate that the queue table exists + IF NOT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'pgmq' AND table_name = qtable) THEN + RAISE EXCEPTION 'Queue "%" does not exist. Create it first using pgmq.create()', v_queue_name; + END IF; + + PERFORM pgmq.disable_notify_insert(v_queue_name); + + INSERT INTO pgmq.notify_insert_throttle (queue_name, throttle_interval_ms) + VALUES (v_queue_name, v_throttle_interval_ms) + ON CONFLICT ON CONSTRAINT notify_insert_throttle_queue_name_key DO UPDATE + SET throttle_interval_ms = EXCLUDED.throttle_interval_ms, + last_notified_at = to_timestamp(0); + + EXECUTE FORMAT( + $QUERY$ + CREATE CONSTRAINT TRIGGER trigger_notify_queue_insert_listeners + AFTER INSERT ON pgmq.%I + DEFERRABLE FOR EACH ROW + EXECUTE PROCEDURE pgmq.notify_queue_listeners() + $QUERY$, + qtable + ); +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION pgmq.disable_notify_insert(queue_name TEXT) +RETURNS void AS $$ +DECLARE + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + v_queue_name TEXT := queue_name; +BEGIN + EXECUTE FORMAT( + $QUERY$ + DROP TRIGGER IF EXISTS trigger_notify_queue_insert_listeners ON pgmq.%I; + $QUERY$, + qtable + ); + + DELETE FROM pgmq.notify_insert_throttle nit WHERE nit.queue_name = v_queue_name; +END; +$$ LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION pgmq.list_notify_insert_throttles() + RETURNS TABLE + ( + queue_name text, + throttle_interval_ms integer, + last_notified_at TIMESTAMP WITH TIME ZONE + ) + LANGUAGE sql + STABLE +AS +$$ + SELECT queue_name, throttle_interval_ms, last_notified_at + FROM pgmq.notify_insert_throttle + ORDER BY queue_name; +$$; + +CREATE OR REPLACE FUNCTION pgmq.update_notify_insert(queue_name text, throttle_interval_ms integer) + RETURNS void + LANGUAGE plpgsql +AS +$$ +BEGIN + IF throttle_interval_ms < 0 THEN + RAISE EXCEPTION 'throttle_interval_ms must be non-negative, got: %', throttle_interval_ms; + END IF; + + IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = update_notify_insert.queue_name) THEN + RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name; + END IF; + + IF NOT EXISTS (SELECT 1 FROM pgmq.notify_insert_throttle WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name) THEN + RAISE EXCEPTION 'Queue "%" does not have notify_insert enabled. Enable it first using pgmq.enable_notify_insert()', queue_name; + END IF; + + UPDATE pgmq.notify_insert_throttle + SET throttle_interval_ms = update_notify_insert.throttle_interval_ms, + last_notified_at = to_timestamp(0) + WHERE notify_insert_throttle.queue_name = update_notify_insert.queue_name; +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.validate_routing_key(routing_key text) + RETURNS boolean + LANGUAGE plpgsql + IMMUTABLE +AS +$$ +BEGIN + -- Valid routing key examples: + -- "logs.error" + -- "app.user-service.auth" + -- "system_events.db.connection_failed" + -- + -- Invalid routing key examples: + -- "" - empty + -- ".logs.error" - starts with dot + -- "logs.error." - ends with dot + -- "logs..error" - consecutive dots + -- "logs.error!" - invalid character + -- "logs error" - space not allowed + -- "logs.*" - wildcards not allowed in routing keys + + IF routing_key IS NULL OR routing_key = '' THEN + RAISE EXCEPTION 'routing_key cannot be NULL or empty'; + END IF; + + IF length(routing_key) > 255 THEN + RAISE EXCEPTION 'routing_key length cannot exceed 255 characters, got % characters', length(routing_key); + END IF; + + IF routing_key !~ '^[a-zA-Z0-9._-]+$' THEN + RAISE EXCEPTION 'routing_key contains invalid characters. Only alphanumeric, dots, hyphens, and underscores are allowed. Got: %', routing_key; + END IF; + + IF routing_key ~ '^\.' THEN + RAISE EXCEPTION 'routing_key cannot start with a dot. Got: %', routing_key; + END IF; + + IF routing_key ~ '\.$' THEN + RAISE EXCEPTION 'routing_key cannot end with a dot. Got: %', routing_key; + END IF; + + IF routing_key ~ '\.\.' THEN + RAISE EXCEPTION 'routing_key cannot contain consecutive dots. Got: %', routing_key; + END IF; + + RETURN true; +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.validate_topic_pattern(pattern text) + RETURNS boolean + LANGUAGE plpgsql + IMMUTABLE +AS +$$ +BEGIN + -- Valid pattern examples: + -- "logs.*" - matches one segment after logs. (e.g., logs.error, logs.info) + -- "logs.#" - matches one or more segments after logs. (e.g., logs.error, logs.api.error) + -- "*.error" - matches one segment before .error (e.g., app.error, db.error) + -- "#.error" - matches one or more segments before .error (e.g., app.error, x.y.error) + -- "app.*.#" - mixed wildcards (one segment then one or more) + -- "#" - catch-all pattern, matches any routing key + -- + -- Invalid pattern examples: + -- ".logs.*" - starts with dot + -- "logs.*." - ends with dot + -- "logs..error" - consecutive dots + -- "logs.**" - consecutive stars + -- "logs.##" - consecutive hashes + -- "logs.*#" - adjacent wildcards + -- "logs.error!" - invalid character + + IF pattern IS NULL OR pattern = '' THEN + RAISE EXCEPTION 'pattern cannot be NULL or empty'; + END IF; + + IF length(pattern) > 255 THEN + RAISE EXCEPTION 'pattern length cannot exceed 255 characters, got % characters', length(pattern); + END IF; + + IF pattern !~ '^[a-zA-Z0-9._\-*#]+$' THEN + RAISE EXCEPTION 'pattern contains invalid characters. Only alphanumeric, dots, hyphens, underscores, *, and # are allowed. Got: %', pattern; + END IF; + + IF pattern ~ '^\.' THEN + RAISE EXCEPTION 'pattern cannot start with a dot. Got: %', pattern; + END IF; + + IF pattern ~ '\.$' THEN + RAISE EXCEPTION 'pattern cannot end with a dot. Got: %', pattern; + END IF; + + IF pattern ~ '\.\.' THEN + RAISE EXCEPTION 'pattern cannot contain consecutive dots. Got: %', pattern; + END IF; + + IF pattern ~ '\*\*' THEN + RAISE EXCEPTION 'pattern cannot contain consecutive stars (**). Use # for multi-segment matching. Got: %', pattern; + END IF; + + IF pattern ~ '##' THEN + RAISE EXCEPTION 'pattern cannot contain consecutive hashes (##). A single # already matches zero or more segments. Got: %', pattern; + END IF; + + IF pattern ~ '\*#' OR pattern ~ '#\*' THEN + RAISE EXCEPTION 'pattern cannot contain adjacent wildcards (*# or #*). Separate wildcards with dots. Got: %', pattern; + END IF; + + RETURN true; +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.bind_topic(pattern text, queue_name text) + RETURNS void + LANGUAGE plpgsql +AS +$$ +BEGIN + PERFORM pgmq.validate_topic_pattern(pattern); + IF queue_name IS NULL OR queue_name = '' THEN + RAISE EXCEPTION 'queue_name cannot be NULL or empty'; + END IF; + + IF NOT EXISTS (SELECT 1 FROM pgmq.meta WHERE meta.queue_name = bind_topic.queue_name) THEN + RAISE EXCEPTION 'Queue "%" does not exist. Create the queue first using pgmq.create()', queue_name; + END IF; + + INSERT INTO pgmq.topic_bindings (pattern, queue_name) + VALUES (pattern, queue_name) + ON CONFLICT ON CONSTRAINT topic_bindings_unique_pattern_queue DO NOTHING; +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.unbind_topic(pattern text, queue_name text) + RETURNS boolean + LANGUAGE plpgsql +AS +$$ +DECLARE + rows_deleted integer; +BEGIN + IF pattern IS NULL OR pattern = '' THEN + RAISE EXCEPTION 'pattern cannot be NULL or empty'; + END IF; + + IF queue_name IS NULL OR queue_name = '' THEN + RAISE EXCEPTION 'queue_name cannot be NULL or empty'; + END IF; + + DELETE + FROM pgmq.topic_bindings + WHERE topic_bindings.pattern = unbind_topic.pattern + AND topic_bindings.queue_name = unbind_topic.queue_name; + + GET DIAGNOSTICS rows_deleted = ROW_COUNT; + + IF rows_deleted > 0 THEN + RETURN true; + ELSE + RETURN false; + END IF; +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.test_routing(routing_key text) + RETURNS TABLE + ( + pattern text, + queue_name text, + compiled_regex text + ) + LANGUAGE plpgsql + STABLE +AS +$$ +BEGIN + PERFORM pgmq.validate_routing_key(routing_key); + RETURN QUERY + SELECT b.pattern, + b.queue_name, + b.compiled_regex + FROM pgmq.topic_bindings b + WHERE routing_key ~ b.compiled_regex + ORDER BY b.pattern; +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, headers jsonb, delay integer) + RETURNS integer + LANGUAGE plpgsql + VOLATILE +AS +$$ +DECLARE + b RECORD; + matched_count integer := 0; +BEGIN + PERFORM pgmq.validate_routing_key(routing_key); + + IF msg IS NULL THEN + RAISE EXCEPTION 'msg cannot be NULL'; + END IF; + + IF delay < 0 THEN + RAISE EXCEPTION 'delay cannot be negative, got: %', delay; + END IF; + + -- Filter matching patterns in SQL for better performance (uses index) + -- Any failure will rollback the entire transaction + FOR b IN + SELECT DISTINCT tb.queue_name + FROM pgmq.topic_bindings tb + WHERE routing_key ~ tb.compiled_regex + ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name + LOOP + PERFORM pgmq.send(b.queue_name, msg, headers, delay); + matched_count := matched_count + 1; + END LOOP; + + RETURN matched_count; +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb) + RETURNS integer + LANGUAGE plpgsql + VOLATILE +AS +$$ +BEGIN + RETURN pgmq.send_topic(routing_key, msg, NULL, 0); +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.send_topic(routing_key text, msg jsonb, delay integer) + RETURNS integer + LANGUAGE plpgsql + VOLATILE +AS +$$ +BEGIN + RETURN pgmq.send_topic(routing_key, msg, NULL, delay); +END; +$$; + +CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings() + RETURNS TABLE + ( + pattern text, + queue_name text, + bound_at TIMESTAMP WITH TIME ZONE, + compiled_regex text + ) + LANGUAGE sql + STABLE +AS +$$ + SELECT pattern, queue_name, bound_at, compiled_regex + FROM pgmq.topic_bindings + ORDER BY bound_at DESC, pattern, queue_name; +$$; + +CREATE OR REPLACE FUNCTION pgmq.list_topic_bindings(queue_name text) + RETURNS TABLE + ( + pattern text, + queue_name text, + bound_at TIMESTAMP WITH TIME ZONE, + compiled_regex text + ) + LANGUAGE sql + STABLE +AS +$$ + SELECT pattern, tb.queue_name, bound_at, compiled_regex + FROM pgmq.topic_bindings tb + WHERE tb.queue_name = list_topic_bindings.queue_name + ORDER BY bound_at DESC, pattern; +$$; + +-- send_batch_topic: Base implementation with TIMESTAMP WITH TIME ZONE delay +CREATE OR REPLACE FUNCTION pgmq.send_batch_topic( + routing_key text, + msgs jsonb[], + headers jsonb[], + delay TIMESTAMP WITH TIME ZONE +) + RETURNS TABLE(queue_name text, msg_id bigint) + LANGUAGE plpgsql + VOLATILE +AS +$$ +DECLARE + b RECORD; +BEGIN + PERFORM pgmq.validate_routing_key(routing_key); + + -- Validate batch parameters once (not per queue) + PERFORM pgmq._validate_batch_params(msgs, headers); + + -- Filter matching patterns in SQL for better performance (uses index) + -- Any failure will rollback the entire transaction + FOR b IN + SELECT DISTINCT tb.queue_name + FROM pgmq.topic_bindings tb + WHERE routing_key ~ tb.compiled_regex + ORDER BY tb.queue_name -- Deterministic ordering, deduplicated by queue_name + LOOP + -- Use private _send_batch to avoid redundant validation + RETURN QUERY + SELECT b.queue_name, batch_result.msg_id + FROM pgmq._send_batch(b.queue_name, msgs, headers, delay) AS batch_result(msg_id); + END LOOP; + + RETURN; +END; +$$; + +-- send_batch_topic: 2 args (routing_key, msgs) +CREATE OR REPLACE FUNCTION pgmq.send_batch_topic( + routing_key text, + msgs jsonb[] +) + RETURNS TABLE(queue_name text, msg_id bigint) + LANGUAGE sql + VOLATILE +AS +$$ + SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp()); +$$; + +-- send_batch_topic: 3 args with headers +CREATE OR REPLACE FUNCTION pgmq.send_batch_topic( + routing_key text, + msgs jsonb[], + headers jsonb[] +) + RETURNS TABLE(queue_name text, msg_id bigint) + LANGUAGE sql + VOLATILE +AS +$$ + SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp()); +$$; + +-- send_batch_topic: 3 args with integer delay +CREATE OR REPLACE FUNCTION pgmq.send_batch_topic( + routing_key text, + msgs jsonb[], + delay integer +) + RETURNS TABLE(queue_name text, msg_id bigint) + LANGUAGE sql + VOLATILE +AS +$$ + SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, clock_timestamp() + make_interval(secs => delay)); +$$; + +-- send_batch_topic: 3 args with timestamp delay +CREATE OR REPLACE FUNCTION pgmq.send_batch_topic( + routing_key text, + msgs jsonb[], + delay TIMESTAMP WITH TIME ZONE +) + RETURNS TABLE(queue_name text, msg_id bigint) + LANGUAGE sql + VOLATILE +AS +$$ + SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, NULL, delay); +$$; + +-- send_batch_topic: 4 args with integer delay +CREATE OR REPLACE FUNCTION pgmq.send_batch_topic( + routing_key text, + msgs jsonb[], + headers jsonb[], + delay integer +) + RETURNS TABLE(queue_name text, msg_id bigint) + LANGUAGE sql + VOLATILE +AS +$$ + SELECT * FROM pgmq.send_batch_topic(routing_key, msgs, headers, clock_timestamp() + make_interval(secs => delay)); +$$; diff --git a/test/fixtures/schema-snapshot.sql b/test/fixtures/schema-snapshot.sql new file mode 100644 --- /dev/null +++ b/test/fixtures/schema-snapshot.sql @@ -0,0 +1,27 @@ +-- Database-independent catalog keys. Bodies remain separate so only explicit +-- local overrides can be exempted; signatures and defaults are always compared. +SELECT key, value FROM ( + SELECT 'function:' || p.proname || '(' || pg_get_function_identity_arguments(p.oid) || ')' AS key, + json_build_array(pg_get_function_result(p.oid), pg_get_function_arguments(p.oid), + l.lanname, p.provolatile, p.proisstrict, p.prosecdef, p.proparallel, p.proconfig)::text AS value + FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace + JOIN pg_language l ON l.oid=p.prolang WHERE n.nspname='pgmq' + UNION ALL + SELECT 'body:' || p.proname || '(' || oidvectortypes(p.proargtypes) || ')', p.prosrc + FROM pg_proc p JOIN pg_namespace n ON n.oid=p.pronamespace WHERE n.nspname='pgmq' + UNION ALL + SELECT 'column:' || c.relname || ':' || a.attnum, + json_build_array(a.attname, format_type(a.atttypid,a.atttypmod), a.attnotnull, + a.attidentity, a.attgenerated, pg_get_expr(d.adbin,d.adrelid))::text + FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace + JOIN pg_attribute a ON a.attrelid=c.oid + LEFT JOIN pg_attrdef d ON d.adrelid=c.oid AND d.adnum=a.attnum + WHERE n.nspname='pgmq' AND a.attnum>0 AND NOT a.attisdropped + UNION ALL + SELECT 'relation:' || c.relname, json_build_array(c.relkind,c.relpersistence)::text + FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='pgmq' + UNION ALL + SELECT 'constraint:' || c.relname || ':' || con.conname, pg_get_constraintdef(con.oid) + FROM pg_constraint con JOIN pg_class c ON c.oid=con.conrelid + JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='pgmq' +) snapshot ORDER BY key diff --git a/vendor/pgmq/pgmq-extension/sql/pgmq--1.11.0--1.11.1.sql b/vendor/pgmq/pgmq-extension/sql/pgmq--1.11.0--1.11.1.sql new file mode 100644 --- /dev/null +++ b/vendor/pgmq/pgmq-extension/sql/pgmq--1.11.0--1.11.1.sql @@ -0,0 +1,70 @@ +-- Allow `pgmq.topic_bindings` to be dumped by `pg_dump` when pgmq is installed as an extension +DO +$$ +BEGIN + IF EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pgmq') THEN + PERFORM pg_catalog.pg_extension_config_dump('pgmq.topic_bindings', ''); + END IF; +END +$$; + + +-- read_grouped_head: read the head of N different FIFO groups in a single operation. +-- This supports horizontal scaling by processing groups in parallel while ensuring message ordering is preserved per group. +CREATE OR REPLACE FUNCTION pgmq.read_grouped_head( + queue_name TEXT, + vt INTEGER, + qty INTEGER +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH fifo_groups AS ( + -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility + SELECT + COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key, + MIN(msg_id) AS head_msg_id + FROM pgmq.%1$I + GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') + ), + selected_messages AS ( + -- Take at most 1 message per group + SELECT g.head_msg_id msg_id + FROM fifo_groups g + JOIN pgmq.%1$I q ON q.msg_id = g.head_msg_id + WHERE q.vt <= clock_timestamp() + ORDER BY q.msg_id + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + UPDATE pgmq.%1$I m + SET + vt = clock_timestamp() + %2$L, + read_ct = read_ct + 1, + last_read_at = clock_timestamp() + FROM selected_messages sm + WHERE m.msg_id = sm.msg_id + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers; + $QUERY$, + qtable, make_interval(secs => vt) + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + + +CREATE OR REPLACE FUNCTION pgmq._ensure_pg_partman_installed() +RETURNS void AS $$ +BEGIN + IF NOT pgmq._extension_exists('pg_partman') THEN + RAISE EXCEPTION 'pg_partman is required for partitioned queues'; + END IF; +END; +$$ LANGUAGE plpgsql; + + +DROP FUNCTION IF EXISTS pgmq.enable_notify_insert(queue_name text); diff --git a/vendor/pgmq/pgmq-extension/sql/pgmq--1.11.1--1.12.0.sql b/vendor/pgmq/pgmq-extension/sql/pgmq--1.11.1--1.12.0.sql new file mode 100644 --- /dev/null +++ b/vendor/pgmq/pgmq-extension/sql/pgmq--1.11.1--1.12.0.sql @@ -0,0 +1,33 @@ +-- read_grouped_head_with_poll +-- reads the head of N different FIFO groups in a single operation, with polling support +CREATE OR REPLACE FUNCTION pgmq.read_grouped_head_with_poll( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + max_poll_seconds INTEGER DEFAULT 5, + poll_interval_ms INTEGER DEFAULT 100 +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + r pgmq.message_record; + stop_at TIMESTAMPTZ; +BEGIN + stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds); + LOOP + IF clock_timestamp() >= stop_at THEN + RETURN; + END IF; + + FOR r IN + SELECT * FROM pgmq.read_grouped_head(queue_name, vt, qty) + LOOP + RETURN NEXT r; + END LOOP; + IF FOUND THEN + RETURN; + ELSE + PERFORM pg_sleep(poll_interval_ms::numeric / 1000); + END IF; + END LOOP; +END; +$$ LANGUAGE plpgsql; diff --git a/vendor/pgmq/pgmq-extension/sql/pgmq--1.12.0--1.13.0.sql b/vendor/pgmq/pgmq-extension/sql/pgmq--1.12.0--1.13.0.sql new file mode 100644 --- /dev/null +++ b/vendor/pgmq/pgmq-extension/sql/pgmq--1.12.0--1.13.0.sql @@ -0,0 +1,254 @@ +-- Partitioned queues: msg_id becomes GENERATED BY DEFAULT AS IDENTITY. +-- pg_partman's partition_data_* tooling moves rows out of the default +-- partition by inserting them with their existing msg_id, which +-- GENERATED ALWAYS refuses. pgmq's own writers never supply msg_id, so this +-- changes nothing for them. +DO $$ +DECLARE + queue_record RECORD; + qtable TEXT; +BEGIN + FOR queue_record IN SELECT queue_name FROM pgmq.meta WHERE is_partitioned LOOP + qtable := pgmq.format_table_name(queue_record.queue_name, 'q'); + + IF EXISTS ( + SELECT 1 FROM information_schema.columns + WHERE table_schema = 'pgmq' + AND table_name = qtable + AND column_name = 'msg_id' + AND identity_generation = 'ALWAYS' + ) THEN + EXECUTE FORMAT('ALTER TABLE pgmq.%I ALTER COLUMN msg_id SET GENERATED BY DEFAULT', qtable); + END IF; + END LOOP; +END; +$$; + +-- create_partitioned gains a premake parameter. The three-argument signature is +-- dropped rather than kept alongside it, so existing calls resolve to one function. +DROP FUNCTION IF EXISTS pgmq.create_partitioned(TEXT, TEXT, TEXT); + +CREATE FUNCTION pgmq.create_partitioned( + queue_name TEXT, + partition_interval TEXT DEFAULT '10000', + retention_interval TEXT DEFAULT '100000', + premake INTEGER DEFAULT 4 +) +RETURNS void AS $$ +DECLARE + partition_col TEXT; + a_partition_col TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + qtable_seq TEXT := qtable || '_msg_id_seq'; + atable TEXT := pgmq.format_table_name(queue_name, 'a'); + fq_qtable TEXT := 'pgmq.' || qtable; + fq_atable TEXT := 'pgmq.' || atable; +BEGIN + PERFORM pgmq.validate_queue_name(queue_name); + PERFORM pgmq.acquire_queue_lock(queue_name); + PERFORM pgmq._ensure_pg_partman_installed(); + IF premake < 1 THEN + RAISE EXCEPTION 'premake must be at least 1, got %', premake; + END IF; + SELECT pgmq._get_partition_col(partition_interval) INTO partition_col; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT GENERATED BY DEFAULT AS IDENTITY, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I) + $QUERY$, + qtable, partition_col + ); + + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_premake := %s, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_qtable, + partition_col, + partition_interval, + premake + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I); + $QUERY$, + qtable || '_part_idx', qtable, partition_col + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || qtable + ); + + EXECUTE FORMAT( + $QUERY$ + INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged) + VALUES (%L, true, false) + ON CONFLICT + DO NOTHING; + $QUERY$, + queue_name + ); + + IF partition_col = 'enqueued_at' THEN + a_partition_col := 'archived_at'; + ELSE + a_partition_col := partition_col; + END IF; + + EXECUTE FORMAT( + $QUERY$ + CREATE TABLE IF NOT EXISTS pgmq.%I ( + msg_id BIGINT NOT NULL, + read_ct INT DEFAULT 0 NOT NULL, + enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + last_read_at TIMESTAMP WITH TIME ZONE, + archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, + vt TIMESTAMP WITH TIME ZONE NOT NULL, + message JSONB, + headers JSONB + ) PARTITION BY RANGE (%I); + $QUERY$, + atable, a_partition_col + ); + + -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md + -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema. + EXECUTE FORMAT( + $QUERY$ + SELECT %I.create_parent( + p_parent_table := %L, + p_control := %L, + p_interval := %L, + p_premake := %s, + p_type := case + when pgmq._get_pg_partman_major_version() = 5 then 'range' + else 'native' + end + ) + $QUERY$, + pgmq._get_pg_partman_schema(), + fq_atable, + a_partition_col, + partition_interval, + premake + ); + + EXECUTE FORMAT( + $QUERY$ + UPDATE %I.part_config + SET + retention = %L, + retention_keep_table = false, + retention_keep_index = true, + automatic_maintenance = 'on' + WHERE parent_table = %L; + $QUERY$, + pgmq._get_pg_partman_schema(), + retention_interval, + 'pgmq.' || atable + ); + + EXECUTE FORMAT( + $QUERY$ + CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at); + $QUERY$, + 'archived_at_idx_' || queue_name, atable + ); + +END; +$$ LANGUAGE plpgsql; + +-- metrics_result gains default_partition_length, which metrics() now reports. +ALTER TYPE pgmq.metrics_result ADD ATTRIBUTE default_partition_length bigint; + +-- get metrics for a single queue +CREATE OR REPLACE FUNCTION pgmq.metrics(queue_name TEXT) +RETURNS pgmq.metrics_result AS $$ +DECLARE + result_row pgmq.metrics_result; + query TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + q_default_partition TEXT := qtable || '_default'; + a_default_partition TEXT := pgmq.format_table_name(queue_name, 'a') || '_default'; + default_partition_length BIGINT; +BEGIN + -- Only partitioned queues have default partitions. Messages in them have no + -- partition of their own, which means pg_partman maintenance is failing for + -- this queue; a non-zero value here is the signal to act on. The planner's + -- estimate is used so that a large spill does not slow down every scrape. + IF to_regclass(FORMAT('pgmq.%I', q_default_partition)) IS NOT NULL THEN + SELECT COALESCE(SUM(GREATEST(c.reltuples, 0))::bigint, 0) + INTO default_partition_length + FROM pg_class c + WHERE c.oid IN ( + to_regclass(FORMAT('pgmq.%I', q_default_partition)), + to_regclass(FORMAT('pgmq.%I', a_default_partition)) + ); + END IF; + + query := FORMAT( + $QUERY$ + WITH q_summary AS ( + SELECT + count(*) as queue_length, + count(CASE WHEN vt <= NOW() THEN 1 END) as queue_visible_length, + EXTRACT(epoch FROM (NOW() - max(enqueued_at)))::int as newest_msg_age_sec, + EXTRACT(epoch FROM (NOW() - min(enqueued_at)))::int as oldest_msg_age_sec, + NOW() as scrape_time + FROM pgmq.%I + ), + all_metrics AS ( + SELECT CASE + WHEN is_called THEN last_value ELSE 0 + END as total_messages + FROM pgmq.%I + ) + SELECT + %L as queue_name, + q_summary.queue_length, + q_summary.newest_msg_age_sec, + q_summary.oldest_msg_age_sec, + all_metrics.total_messages, + q_summary.scrape_time, + q_summary.queue_visible_length, + %L::bigint as default_partition_length + FROM q_summary, all_metrics + $QUERY$, + qtable, qtable || '_msg_id_seq', queue_name, default_partition_length + ); + EXECUTE query INTO result_row; + RETURN result_row; +END; +$$ LANGUAGE plpgsql; diff --git a/vendor/pgmq/pgmq-extension/sql/pgmq.sql b/vendor/pgmq/pgmq-extension/sql/pgmq.sql --- a/vendor/pgmq/pgmq-extension/sql/pgmq.sql +++ b/vendor/pgmq/pgmq-extension/sql/pgmq.sql @@ -21,6 +21,16 @@ created_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL ); +-- Grant permission to pg_monitor to all tables and sequences +-- These grants are intentionally placed here (after creating `pgmq.meta` but before creating other tables). This +-- allows the `pg_dump` output for a fresh installation to match the output for an installation that followed the +-- upgrade path. +GRANT USAGE ON SCHEMA pgmq TO pg_monitor; +GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pg_monitor; +GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pg_monitor; +ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pg_monitor; +ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pg_monitor; + -- Table to track notification throttling for queues CREATE UNLOGGED TABLE IF NOT EXISTS pgmq.notify_insert_throttle ( queue_name VARCHAR UNIQUE NOT NULL -- Queue name (without 'q_' prefix) @@ -62,7 +72,7 @@ -- Includes queue_name and compiled_regex to allow index-only scans (no table access needed) CREATE INDEX IF NOT EXISTS idx_topic_bindings_covering ON pgmq.topic_bindings (pattern) INCLUDE (queue_name, compiled_regex); --- Allow pgmq.meta to be dumped by `pg_dump` when pgmq is installed as an extension +-- Allow the following `pgmq` tables to be dumped by `pg_dump` when pgmq is installed as an extension DO $$ BEGIN @@ -74,15 +84,10 @@ END $$; --- Grant permission to pg_monitor to all tables and sequences -GRANT USAGE ON SCHEMA pgmq TO pg_monitor; -GRANT SELECT ON ALL TABLES IN SCHEMA pgmq TO pg_monitor; -GRANT SELECT ON ALL SEQUENCES IN SCHEMA pgmq TO pg_monitor; -ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON TABLES TO pg_monitor; -ALTER DEFAULT PRIVILEGES IN SCHEMA pgmq GRANT SELECT ON SEQUENCES TO pg_monitor; - -- This type has the shape of a message in a queue, and is often returned by --- pgmq functions that return messages +-- pgmq functions that return messages. +-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation +-- expects a specific order of fields. CREATE TYPE pgmq.message_record AS ( msg_id BIGINT, read_ct INTEGER, @@ -93,6 +98,8 @@ headers JSONB ); +-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation +-- expects a specific order of fields. CREATE TYPE pgmq.queue_record AS ( queue_name VARCHAR, is_partitioned BOOLEAN, @@ -233,6 +240,87 @@ END; $$ LANGUAGE plpgsql; +-- read_grouped_head: read the head of N different FIFO groups in a single operation. +-- This supports horizontal scaling by processing groups in parallel while ensuring message ordering is preserved per group. +CREATE FUNCTION pgmq.read_grouped_head( + queue_name TEXT, + vt INTEGER, + qty INTEGER +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + sql TEXT; + qtable TEXT := pgmq.format_table_name(queue_name, 'q'); +BEGIN + sql := FORMAT( + $QUERY$ + WITH fifo_groups AS ( + -- Determine the absolute head (oldest) message id per FIFO group, regardless of visibility + SELECT + COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') AS fifo_key, + MIN(msg_id) AS head_msg_id + FROM pgmq.%1$I + GROUP BY COALESCE(headers->>'x-pgmq-group', '_default_fifo_group') + ), + selected_messages AS ( + -- Take at most 1 message per group + SELECT g.head_msg_id msg_id + FROM fifo_groups g + JOIN pgmq.%1$I q ON q.msg_id = g.head_msg_id + WHERE q.vt <= clock_timestamp() + ORDER BY q.msg_id + LIMIT $1 + FOR UPDATE SKIP LOCKED + ) + UPDATE pgmq.%1$I m + SET + vt = clock_timestamp() + %2$L, + read_ct = read_ct + 1, + last_read_at = clock_timestamp() + FROM selected_messages sm + WHERE m.msg_id = sm.msg_id + RETURNING m.msg_id, m.read_ct, m.enqueued_at, m.last_read_at, m.vt, m.message, m.headers; + $QUERY$, + qtable, make_interval(secs => vt) + ); + RETURN QUERY EXECUTE sql USING qty; +END; +$$ LANGUAGE plpgsql; + +-- read_grouped_head_with_poll +-- reads the head of N different FIFO groups in a single operation, with polling support +CREATE FUNCTION pgmq.read_grouped_head_with_poll( + queue_name TEXT, + vt INTEGER, + qty INTEGER, + max_poll_seconds INTEGER DEFAULT 5, + poll_interval_ms INTEGER DEFAULT 100 +) +RETURNS SETOF pgmq.message_record AS $$ +DECLARE + r pgmq.message_record; + stop_at TIMESTAMPTZ; +BEGIN + stop_at := clock_timestamp() + make_interval(secs => max_poll_seconds); + LOOP + IF clock_timestamp() >= stop_at THEN + RETURN; + END IF; + + FOR r IN + SELECT * FROM pgmq.read_grouped_head(queue_name, vt, qty) + LOOP + RETURN NEXT r; + END LOOP; + IF FOUND THEN + RETURN; + ELSE + PERFORM pg_sleep(poll_interval_ms::numeric / 1000); + END IF; + END LOOP; +END; +$$ LANGUAGE plpgsql; + -- a helper to format table names and check for invalid characters CREATE FUNCTION pgmq.format_table_name(queue_name text, prefix text) RETURNS TEXT AS $$ @@ -766,6 +854,8 @@ $$ LANGUAGE sql; -- returned by pgmq.metrics() and pgmq.metrics_all +-- Note: Changing the order of fields in this type is a breaking change -- our Rust Diesel client implementation +-- expects a specific order of fields. CREATE TYPE pgmq.metrics_result AS ( queue_name text, queue_length bigint, @@ -773,7 +863,8 @@ oldest_msg_age_sec int, total_messages bigint, scrape_time timestamp with time zone, - queue_visible_length bigint + queue_visible_length bigint, + default_partition_length bigint ); -- get metrics for a single queue @@ -783,7 +874,24 @@ result_row pgmq.metrics_result; query TEXT; qtable TEXT := pgmq.format_table_name(queue_name, 'q'); + q_default_partition TEXT := qtable || '_default'; + a_default_partition TEXT := pgmq.format_table_name(queue_name, 'a') || '_default'; + default_partition_length BIGINT; BEGIN + -- Only partitioned queues have default partitions. Messages in them have no + -- partition of their own, which means pg_partman maintenance is failing for + -- this queue; a non-zero value here is the signal to act on. The planner's + -- estimate is used so that a large spill does not slow down every scrape. + IF to_regclass(FORMAT('pgmq.%I', q_default_partition)) IS NOT NULL THEN + SELECT COALESCE(SUM(GREATEST(c.reltuples, 0))::bigint, 0) + INTO default_partition_length + FROM pg_class c + WHERE c.oid IN ( + to_regclass(FORMAT('pgmq.%I', q_default_partition)), + to_regclass(FORMAT('pgmq.%I', a_default_partition)) + ); + END IF; + query := FORMAT( $QUERY$ WITH q_summary AS ( @@ -808,10 +916,11 @@ q_summary.oldest_msg_age_sec, all_metrics.total_messages, q_summary.scrape_time, - q_summary.queue_visible_length + q_summary.queue_visible_length, + %L::bigint as default_partition_length FROM q_summary, all_metrics $QUERY$, - qtable, qtable || '_msg_id_seq', queue_name + qtable, qtable || '_msg_id_seq', queue_name, default_partition_length ); EXECUTE query INTO result_row; RETURN result_row; @@ -1267,7 +1376,8 @@ CREATE FUNCTION pgmq.create_partitioned( queue_name TEXT, partition_interval TEXT DEFAULT '10000', - retention_interval TEXT DEFAULT '100000' + retention_interval TEXT DEFAULT '100000', + premake INTEGER DEFAULT 4 ) RETURNS void AS $$ DECLARE @@ -1282,12 +1392,15 @@ PERFORM pgmq.validate_queue_name(queue_name); PERFORM pgmq.acquire_queue_lock(queue_name); PERFORM pgmq._ensure_pg_partman_installed(); + IF premake < 1 THEN + RAISE EXCEPTION 'premake must be at least 1, got %', premake; + END IF; SELECT pgmq._get_partition_col(partition_interval) INTO partition_col; EXECUTE FORMAT( $QUERY$ CREATE TABLE IF NOT EXISTS pgmq.%I ( - msg_id BIGINT GENERATED ALWAYS AS IDENTITY, + msg_id BIGINT GENERATED BY DEFAULT AS IDENTITY, read_ct INT DEFAULT 0 NOT NULL, enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL, last_read_at TIMESTAMP WITH TIME ZONE, @@ -1307,6 +1420,7 @@ p_parent_table := %L, p_control := %L, p_interval := %L, + p_premake := %s, p_type := case when pgmq._get_pg_partman_major_version() = 5 then 'range' else 'native' @@ -1316,7 +1430,8 @@ pgmq._get_pg_partman_schema(), fq_qtable, partition_col, - partition_interval + partition_interval, + premake ); EXECUTE FORMAT( @@ -1381,6 +1496,7 @@ p_parent_table := %L, p_control := %L, p_interval := %L, + p_premake := %s, p_type := case when pgmq._get_pg_partman_major_version() = 5 then 'range' else 'native' @@ -1390,7 +1506,8 @@ pgmq._get_pg_partman_schema(), fq_atable, a_partition_col, - partition_interval + partition_interval, + premake ); EXECUTE FORMAT(