diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,13 @@
 # Revision history for pgmq-effectful
 
+## 0.6.0.0 -- 2026-09-10
+
+Add grouped-head reads and polling to plain/traced interpreters, with Consumer/receive spans.
+Add explicit premake creation (PGMQ 1.13+) using the existing partition-creation span convention.
+Expose all six grouped reads, their argument records, and explicit premake from `Pgmq.Effectful`.
+The re-exported `QueueMetrics` adds nullable `defaultPartitionLength`; metrics pass through
+unchanged. Users constructing this record must supply the new field.
+
 ## 0.5.0.0 -- 2026-08-06
 
 ### Breaking Changes
diff --git a/pgmq-effectful.cabal b/pgmq-effectful.cabal
--- a/pgmq-effectful.cabal
+++ b/pgmq-effectful.cabal
@@ -1,30 +1,37 @@
-cabal-version:   3.4
-name:            pgmq-effectful
-version:         0.5.0.0
-synopsis:        Effectful effects for PGMQ (PostgreSQL Message Queue)
+cabal-version: 3.4
+name: pgmq-effectful
+version: 0.6.0.0
+synopsis: Effectful effects for PGMQ (PostgreSQL Message Queue)
 description:
   Effectful effects and interpreters for pgmq-hs, a Haskell client
   library for PGMQ (PostgreSQL Message Queue). Includes OpenTelemetry
   instrumentation for tracing pgmq operations.
 
-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
+extra-source-files: test/fixtures/pgmq-1.12.0.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.Effectful
     Pgmq.Effectful.Effect
@@ -47,38 +54,43 @@
     TypeOperators
 
   build-depends:
-    , aeson                                  >=2.0   && <2.3
-    , base                                   >=4.18  && <5
-    , bytestring                             >=0.11  && <0.13
-    , case-insensitive                       >=1.2   && <1.3
-    , effectful-core                         ^>=2.5  || ^>=2.6
-    , hasql                                  ^>=1.10
-    , hasql-pool                             ^>=1.4
-    , hs-opentelemetry-api                   >=1.0   && <2
-    , hs-opentelemetry-semantic-conventions  >=1.40  && <2
-    , http-types                             >=0.12  && <0.13
-    , pgmq-core                              >=0.5   && <0.6
-    , pgmq-hasql                             >=0.5   && <0.6
-    , text                                   >=2.0   && <2.2
-    , unliftio                               >=0.2   && <0.3
-    , unordered-containers                   >=0.2   && <0.3
-    , vector                                 ^>=0.13
+    aeson >=2.0 && <2.3,
+    base >=4.18 && <5,
+    bytestring >=0.11 && <0.13,
+    case-insensitive >=1.2 && <1.3,
+    effectful-core ^>=2.5 || ^>=2.6,
+    hasql ^>=1.10,
+    hasql-pool ^>=1.4,
+    hs-opentelemetry-api >=1.0 && <2,
+    hs-opentelemetry-semantic-conventions >=1.40 && <2,
+    http-types >=0.12 && <0.13,
+    pgmq-core >=0.6 && <0.7,
+    pgmq-hasql >=0.6 && <0.7,
+    text >=2.0 && <2.2,
+    unliftio >=0.2 && <0.3,
+    unordered-containers >=0.2 && <0.3,
+    vector ^>=0.13,
 
-  hs-source-dirs:     src
-  default-language:   GHC2024
+  hs-source-dirs: src
+  default-language: GHC2024
 
 test-suite pgmq-effectful-test
-  import:             warnings
-  default-language:   GHC2024
-  type:               exitcode-stdio-1.0
-  hs-source-dirs:     test
-  main-is:            Main.hs
-  ghc-options:        -threaded -rtsopts -with-rtsopts=-N
+  import: warnings
+  default-language: GHC2024
+  type: exitcode-stdio-1.0
+  hs-source-dirs: test
+  main-is: Main.hs
+  ghc-options:
+    -threaded
+    -rtsopts
+    -with-rtsopts=-N
+
   other-modules:
     ClassificationSpec
     EphemeralDb
     PlainInterpreterSpec
     TracedInterpreterSpec
+    UmbrellaExportsSpec
 
   default-extensions:
     DataKinds
@@ -89,24 +101,26 @@
     TypeOperators
 
   build-depends:
-    , aeson                                  >=2.0   && <2.3
-    , base                                   >=4.18  && <5
-    , effectful-core                         ^>=2.5  || ^>=2.6
-    , ephemeral-pg                           >=0.2.1
-    , hasql                                  ^>=1.10
-    , hasql-pool                             ^>=1.4
-    , hs-opentelemetry-api                   >=1.0   && <2
-    , hs-opentelemetry-exporter-in-memory    >=1.0   && <2
-    , hs-opentelemetry-propagator-w3c        >=1.0   && <2
-    , hs-opentelemetry-sdk                   >=1.0   && <2
-    , hs-opentelemetry-semantic-conventions  >=1.40  && <2
-    , pg-migrate
-    , pgmq-core                              >=0.5   && <0.6
-    , pgmq-effectful
-    , pgmq-migration                         >=0.5   && <0.6
-    , random                                 ^>=1.2
-    , tasty                                  ^>=1.5
-    , tasty-hunit                            ^>=0.10
-    , text                                   >=2.0   && <2.2
-    , unordered-containers                   >=0.2   && <0.3
-    , vector                                 ^>=0.13
+    aeson >=2.0 && <2.3,
+    base >=4.18 && <5,
+    directory,
+    effectful-core ^>=2.5 || ^>=2.6,
+    ephemeral-pg >=0.2.1,
+    hasql ^>=1.10,
+    hasql-pool ^>=1.4,
+    hs-opentelemetry-api >=1.0 && <2,
+    hs-opentelemetry-exporter-in-memory >=1.0 && <2,
+    hs-opentelemetry-propagator-w3c >=1.0 && <2,
+    hs-opentelemetry-sdk >=1.0 && <2,
+    hs-opentelemetry-semantic-conventions >=1.40 && <2,
+    pg-migrate,
+    pgmq-core >=0.6 && <0.7,
+    pgmq-effectful >=0.6 && <0.7,
+    pgmq-hasql >=0.6 && <0.7,
+    pgmq-migration >=0.6 && <0.7,
+    random ^>=1.2,
+    tasty ^>=1.5,
+    tasty-hunit ^>=0.10,
+    text >=2.0 && <2.2,
+    unordered-containers >=0.2 && <0.3,
+    vector ^>=0.13,
diff --git a/src/Pgmq/Effectful.hs b/src/Pgmq/Effectful.hs
--- a/src/Pgmq/Effectful.hs
+++ b/src/Pgmq/Effectful.hs
@@ -34,6 +34,7 @@
     createQueue,
     dropQueue,
     createPartitionedQueue,
+    createPartitionedQueueWithPremake,
     createUnloggedQueue,
     detachArchive,
 
@@ -63,6 +64,20 @@
     readWithPoll,
     pop,
 
+    -- * FIFO / Grouped Reads
+
+    -- | Head reads (PGMQ 1.12+) lease at most one absolute head per group.
+    -- An invisible head blocks its group; expiry makes that same head eligible.
+    -- Polling occupies a database connection. Leases do not guarantee exactly-once processing.
+    readGrouped,
+    readGroupedWithPoll,
+    readGroupedRoundRobin,
+    readGroupedRoundRobinWithPoll,
+    readGroupedHead,
+    readGroupedHeadWithPoll,
+    ReadGrouped (..),
+    ReadGroupedWithPoll (..),
+
     -- * Topic Routing (pgmq 1.11.0+)
 
     -- ** Topic Management
@@ -164,6 +179,7 @@
     bindTopic,
     changeVisibilityTimeout,
     createPartitionedQueue,
+    createPartitionedQueueWithPremake,
     createQueue,
     createUnloggedQueue,
     deleteAllMessagesFromQueue,
@@ -178,6 +194,12 @@
     listTopicBindingsForQueue,
     pop,
     queueMetrics,
+    readGrouped,
+    readGroupedHead,
+    readGroupedHeadWithPoll,
+    readGroupedRoundRobin,
+    readGroupedRoundRobinWithPoll,
+    readGroupedWithPoll,
     readMessage,
     readWithPoll,
     sendMessage,
@@ -233,6 +255,8 @@
     MessageQuery (..),
     PopMessage (..),
     QueueMetrics (..),
+    ReadGrouped (..),
+    ReadGroupedWithPoll (..),
     ReadMessage (..),
     ReadWithPollMessage (..),
     SendMessage (..),
diff --git a/src/Pgmq/Effectful/Effect.hs b/src/Pgmq/Effectful/Effect.hs
--- a/src/Pgmq/Effectful/Effect.hs
+++ b/src/Pgmq/Effectful/Effect.hs
@@ -6,6 +6,7 @@
     createQueue,
     dropQueue,
     createPartitionedQueue,
+    createPartitionedQueueWithPremake,
     createUnloggedQueue,
     detachArchive,
 
@@ -47,6 +48,10 @@
     readGrouped,
     readGroupedWithPoll,
 
+    -- ** Grouped heads (pgmq 1.12.0+)
+    readGroupedHead,
+    readGroupedHeadWithPoll,
+
     -- ** Round-Robin FIFO Read (pgmq 1.9.0+)
     readGroupedRoundRobin,
     readGroupedRoundRobinWithPoll,
@@ -141,6 +146,7 @@
   CreateQueue :: QueueName -> Pgmq m ()
   DropQueue :: QueueName -> Pgmq m Bool
   CreatePartitionedQueue :: CreatePartitionedQueue -> Pgmq m ()
+  CreatePartitionedQueueWithPremake :: CreatePartitionedQueue -> Int32 -> Pgmq m ()
   CreateUnloggedQueue :: QueueName -> Pgmq m ()
   DetachArchive :: QueueName -> Pgmq m ()
   EnableNotifyInsert :: EnableNotifyInsert -> Pgmq m ()
@@ -169,6 +175,9 @@
   BatchSetVisibilityTimeoutAt :: BatchVisibilityTimeoutAtQuery -> Pgmq m (Vector Message)
   ReadWithPoll :: ReadWithPollMessage -> Pgmq m (Vector Message)
   Pop :: PopMessage -> Pgmq m (Vector Message)
+  -- Grouped heads (pgmq 1.12.0+)
+  ReadGroupedHead :: ReadGrouped -> Pgmq m (Vector Message)
+  ReadGroupedHeadWithPoll :: ReadGroupedWithPoll -> Pgmq m (Vector Message)
   -- FIFO Read (pgmq 1.8.0+)
   ReadGrouped :: ReadGrouped -> Pgmq m (Vector Message)
   ReadGroupedWithPoll :: ReadGroupedWithPoll -> Pgmq m (Vector Message)
@@ -213,6 +222,11 @@
 createPartitionedQueue :: (Pgmq :> es) => CreatePartitionedQueue -> Eff es ()
 createPartitionedQueue = send . CreatePartitionedQueue
 
+-- | Create with an explicit pg_partman premake count (PGMQ 1.13+).
+-- Counts below one are server errors; both queue and archive use this count.
+createPartitionedQueueWithPremake :: (Pgmq :> es) => CreatePartitionedQueue -> Int32 -> Eff es ()
+createPartitionedQueueWithPremake q n = send (CreatePartitionedQueueWithPremake q n)
+
 createUnloggedQueue :: (Pgmq :> es) => QueueName -> Eff es ()
 createUnloggedQueue = send . CreateUnloggedQueue
 
@@ -319,6 +333,16 @@
 readGroupedRoundRobinWithPoll :: (Pgmq :> es) => ReadGroupedWithPoll -> Eff es (Vector Message)
 readGroupedRoundRobinWithPoll = send . ReadGroupedRoundRobinWithPoll
 
+-- | Lease at most one absolute head per group (PGMQ 1.12+). An invisible
+-- head blocks its group. Expiry permits redelivery; this is not exactly-once.
+readGroupedHead :: (Pgmq :> es) => ReadGrouped -> Eff es (Vector Message)
+readGroupedHead = send . ReadGroupedHead
+
+-- | Poll for group heads (PGMQ 1.12+), occupying a database connection while
+-- waiting. The same head lease and redelivery semantics as 'readGroupedHead' apply.
+readGroupedHeadWithPoll :: (Pgmq :> es) => ReadGroupedWithPoll -> Eff es (Vector Message)
+readGroupedHeadWithPoll = send . ReadGroupedHeadWithPoll
+
 -- Topic Management (pgmq 1.11.0+)
 
 -- | Bind a topic pattern to a queue (pgmq 1.11.0+)
@@ -401,6 +425,8 @@
 listFifoIndexQueueNames :: (Pgmq :> es) => Eff es [Text]
 listFifoIndexQueueNames = send ListFifoIndexQueueNames
 
+-- | Partition estimates are Nothing on PGMQ 1.12 and for ordinary queues;
+-- Just values on PGMQ 1.13 are planner estimates, not exact live counts.
 queueMetrics :: (Pgmq :> es) => QueueName -> Eff es QueueMetrics
 queueMetrics = send . QueueMetrics
 
diff --git a/src/Pgmq/Effectful/Interpreter.hs b/src/Pgmq/Effectful/Interpreter.hs
--- a/src/Pgmq/Effectful/Interpreter.hs
+++ b/src/Pgmq/Effectful/Interpreter.hs
@@ -118,6 +118,7 @@
   CreateQueue q -> runSession pool $ Sessions.createQueue q
   DropQueue q -> runSession pool $ Sessions.dropQueue q
   CreatePartitionedQueue q -> runSession pool $ Sessions.createPartitionedQueue q
+  CreatePartitionedQueueWithPremake q n -> runSession pool $ Sessions.createPartitionedQueueWithPremake q n
   CreateUnloggedQueue q -> runSession pool $ Sessions.createUnloggedQueue q
   DetachArchive _q -> pure ()
   EnableNotifyInsert config -> runSession pool $ Sessions.enableNotifyInsert config
@@ -147,6 +148,8 @@
   ReadWithPoll query -> runSession pool $ Sessions.readWithPoll query
   Pop query -> runSession pool $ Sessions.pop query
   -- FIFO Read (pgmq 1.8.0+)
+  ReadGroupedHead query -> runSession pool $ Sessions.readGroupedHead query
+  ReadGroupedHeadWithPoll query -> runSession pool $ Sessions.readGroupedHeadWithPoll query
   ReadGrouped query -> runSession pool $ Sessions.readGrouped query
   ReadGroupedWithPoll query -> runSession pool $ Sessions.readGroupedWithPoll query
   -- Round-robin FIFO Read (pgmq 1.9.0+)
diff --git a/src/Pgmq/Effectful/Interpreter/Traced.hs b/src/Pgmq/Effectful/Interpreter/Traced.hs
--- a/src/Pgmq/Effectful/Interpreter/Traced.hs
+++ b/src/Pgmq/Effectful/Interpreter/Traced.hs
@@ -186,6 +186,9 @@
   CreatePartitionedQueue pq@(Types.CreatePartitionedQueue qn _ _) ->
     withTracedOp config pool (queueOp "pgmq.create_partitioned" OTel.Internal qn) $
       Sessions.createPartitionedQueue pq
+  CreatePartitionedQueueWithPremake pq@(Types.CreatePartitionedQueue qn _ _) n ->
+    withTracedOp config pool (queueOp "pgmq.create_partitioned" OTel.Internal qn) $
+      Sessions.createPartitionedQueueWithPremake pq n
   CreateUnloggedQueue q ->
     withTracedOp config pool (queueOp "pgmq.create_unlogged" OTel.Internal q) $
       Sessions.createUnloggedQueue q
@@ -239,6 +242,12 @@
     withTracedOp config pool (receiveOp "pgmq.pop" qn) $
       Sessions.pop query
   -- FIFO Read (Consumer spans)
+  ReadGroupedHead query@(Types.ReadGrouped qn _ _) ->
+    withTracedOp config pool (receiveOp "pgmq.read_grouped_head" qn) $
+      Sessions.readGroupedHead query
+  ReadGroupedHeadWithPoll query@(Types.ReadGroupedWithPoll qn _ _ _ _) ->
+    withTracedOp config pool (receiveOp "pgmq.read_grouped_head_with_poll" qn) $
+      Sessions.readGroupedHeadWithPoll query
   ReadGrouped query@(Types.ReadGrouped qn _ _) ->
     withTracedOp config pool (receiveOp "pgmq.read_grouped" qn) $
       Sessions.readGrouped query
diff --git a/test/EphemeralDb.hs b/test/EphemeralDb.hs
--- a/test/EphemeralDb.hs
+++ b/test/EphemeralDb.hs
@@ -13,7 +13,9 @@
   )
 where
 
+import Control.Monad (filterM, when)
 import Data.List.NonEmpty (NonEmpty (..))
+import Data.Text.IO qualified as TextIO
 import Database.PostgreSQL.Migrate
   ( defaultRunOptions,
     migrationPlan,
@@ -26,7 +28,10 @@
   )
 import Hasql.Pool qualified as Pool
 import Hasql.Pool.Config qualified as PoolConfig
+import Hasql.Session qualified as Session
 import Pgmq.Migration qualified as Migration
+import System.Directory (doesFileExist)
+import System.Environment (lookupEnv)
 
 -- | Run an action with a temporary PostgreSQL database that has the
 -- pgmq schema installed.
@@ -39,12 +44,31 @@
             PoolConfig.staticConnectionSettings connSettings
           ]
   pool <- Pool.acquire poolConfig
-  component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations
-  plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))
-  installResult <- runMigrationPlan defaultRunOptions connSettings plan
-  case installResult of
-    Left migrationErr -> error $ "Migration failed: " <> show migrationErr
-    Right _ -> action pool
+  version <- lookupEnv "PGMQ_TEST_SCHEMA_VERSION"
+  case version of
+    Just "1.12.0" -> do
+      paths <- filterM doesFileExist ["test/fixtures/pgmq-1.12.0.sql", "pgmq-effectful/test/fixtures/pgmq-1.12.0.sql"]
+      path <- case paths of
+        candidate : _ -> pure candidate
+        [] -> error "Missing packaged PGMQ 1.12.0 test fixture"
+      sql <- TextIO.readFile path
+      Pool.use pool (Session.script sql) >>= either (error . show) pure
+    Nothing -> installNative connSettings
+    Just "1.13.0" -> installNative connSettings
+    Just invalid -> error ("Invalid PGMQ_TEST_SCHEMA_VERSION: " <> invalid)
+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"
+  -- Required runs fail for absent or unusable pg_partman, not just missing metadata.
+  partman <- Pool.use pool (Session.script "CREATE SCHEMA IF NOT EXISTS partman; CREATE EXTENSION IF NOT EXISTS pg_partman SCHEMA partman")
+  case partman of
+    Left err -> when required (error ("PGMQ_REQUIRE_PARTMAN=1: " <> show err))
+    Right () -> pure ()
+  action pool
+  where
+    installNative connSettings = do
+      component <- either (error . ("Invalid PGMQ migration component: " <>) . show) pure Migration.pgmqMigrations
+      plan <- either (error . ("Invalid PGMQ migration plan: " <>) . show) pure (migrationPlan (component :| []))
+      installResult <- runMigrationPlan defaultRunOptions connSettings plan
+      either (error . ("Migration failed: " <>) . show) (const (pure ())) installResult
 
 -- | Alias for 'withPgmqDb' kept for parallelism with pgmq-hasql.
 withPgmqPool :: (Pool.Pool -> IO a) -> IO (Either StartError a)
diff --git a/test/Main.hs b/test/Main.hs
--- a/test/Main.hs
+++ b/test/Main.hs
@@ -7,6 +7,7 @@
 import PlainInterpreterSpec qualified
 import Test.Tasty (defaultMain, testGroup)
 import TracedInterpreterSpec qualified
+import UmbrellaExportsSpec ()
 
 main :: IO ()
 main = do
diff --git a/test/PlainInterpreterSpec.hs b/test/PlainInterpreterSpec.hs
--- a/test/PlainInterpreterSpec.hs
+++ b/test/PlainInterpreterSpec.hs
@@ -17,12 +17,14 @@
   )
 import Test.Tasty (TestTree, testGroup)
 import Test.Tasty.HUnit (assertFailure, testCase)
+import TracedInterpreterSpec (plainFeatureTests)
 
 tests :: Pool.Pool -> TestTree
 tests pool =
   testGroup
     "Plain interpreter error propagation"
-    [ testCase "statement error surfaces PgmqSessionError" $ do
+    [ plainFeatureTests,
+      testCase "statement error surfaces PgmqSessionError" $ do
         bogus <- case parseQueueName "queue_that_does_not_exist_xyz" of
           Right q -> pure q
           Left err -> assertFailure ("could not build test queue name: " <> show err) >> fail ""
diff --git a/test/TracedInterpreterSpec.hs b/test/TracedInterpreterSpec.hs
--- a/test/TracedInterpreterSpec.hs
+++ b/test/TracedInterpreterSpec.hs
@@ -1,18 +1,25 @@
 {-# LANGUAGE OverloadedStrings #-}
 
-module TracedInterpreterSpec (tests) where
+module TracedInterpreterSpec (tests, plainFeatureTests) where
 
 import Control.Concurrent.MVar (MVar, newMVar, putMVar, takeMVar)
 import Control.Exception (bracket)
-import Control.Monad (unless)
+import Control.Monad (unless, when)
+import Data.Aeson (object, (.=))
 import Data.HashMap.Strict qualified as HM
 import Data.IORef (IORef, readIORef)
+import Data.Int (Int32)
 import Data.Text (Text)
 import Data.Text qualified as T
 import Data.Vector qualified as V
-import Effectful (runEff)
-import Effectful.Error.Static (runError)
+import Effectful (Eff, IOE, runEff)
+import Effectful.Error.Static (Error, runError)
+import EphemeralDb (withPgmqDb)
+import Hasql.Decoders qualified as D
+import Hasql.Encoders qualified as E
 import Hasql.Pool qualified as Pool
+import Hasql.Session qualified as Session
+import Hasql.Statement (preparable)
 import OpenTelemetry.Attributes qualified as Attrs
 import OpenTelemetry.Context qualified as Ctxt
 import OpenTelemetry.Context.ThreadLocal qualified as CtxtLocal
@@ -38,7 +45,11 @@
     sendMessage,
     sendMessageTraced,
   )
+import Pgmq.Effectful.Effect qualified as Eff
+import Pgmq.Effectful.Interpreter (runPgmq)
 import Pgmq.Effectful.Telemetry (extractTraceContext, jsonToTraceHeaders)
+import Pgmq.Hasql.Sessions qualified as Sessions
+import Pgmq.Hasql.Statements.Types qualified as Types
 import Pgmq.Types qualified as Pgmq
 import System.Environment (lookupEnv, setEnv, unsetEnv)
 import System.IO.Unsafe (unsafePerformIO)
@@ -50,7 +61,8 @@
 tests pool =
   testGroup
     "Traced interpreter"
-    [ testGroup
+    [ featureTests True,
+      testGroup
         "Error propagation"
         [ testCase "statement error surfaces PgmqSessionError via Error channel" $ do
             (tracer, _provider, _spansRef) <- setupTracer
@@ -412,3 +424,123 @@
 semconvEnvLock :: MVar ()
 semconvEnvLock = unsafePerformIO (newMVar ())
 {-# NOINLINE semconvEnvLock #-}
+
+-- Each case owns its database: metrics_all must not race unrelated queue drops.
+plainFeatureTests :: TestTree
+plainFeatureTests = featureTests False
+
+featureTests :: Bool -> TestTree
+featureTests traced =
+  testGroup (if traced then "GroupedHead and partition compatibility traced" else "GroupedHead and partition compatibility plain") $
+    [ testCase (if polling then "polling heads" else "heads") $
+        withSemconvOptIn "messaging/dup,database/dup" $
+          isolated $ \pool -> do
+            (tracer, _, spansRef) <- setupTracer
+            queue <- mkUniqueQueue "head"
+            let run :: Eff '[Eff.Pgmq, Error PgmqRuntimeError, IOE] a -> IO a
+                run action = assertRight =<< runEff (runError @PgmqRuntimeError ((if traced then runPgmqTraced pool tracer else runPgmq pool) action))
+            session pool (Sessions.createQueue queue)
+            ids <- traverse (\group -> session pool (Sessions.sendMessageWithHeaders (Types.SendMessageWithHeaders queue (MessageBody "head") (Pgmq.MessageHeaders (object ["x-pgmq-group" .= (group :: Text)])) Nothing))) ["a", "a", "b", "b"]
+            let readHeads = if polling then Eff.readGroupedHeadWithPoll (Types.ReadGroupedWithPoll queue 30 10 1 10) else Eff.readGroupedHead (Types.ReadGrouped queue 30 10)
+            messages <- run readHeads
+            assertEqual "one absolute head per group" [msg | (i, msg) <- zip [0 :: Int ..] ids, even i] (map Pgmq.messageId (V.toList messages))
+            when traced $ do
+              spans <- readIORef spansRef
+              observedSpan <- singleSpan spans "grouped head"
+              spanName observedSpan >>= assertEqual "receive span name" ("receive " <> queueNameToText queue)
+              assertSpanKindConsumer observedSpan
+              assertAttrText observedSpan "messaging.destination.name" (queueNameToText queue)
+              let label = if polling then "pgmq.read_grouped_head_with_poll" else "pgmq.read_grouped_head"
+              assertAttrText observedSpan "db.operation" label
+              assertAttrText observedSpan "db.operation.name" label
+            -- Leased heads block their groups even though later messages are visible.
+            blocked <- run (Eff.readGroupedHead (Types.ReadGrouped queue 30 10))
+            assertEqual "heads block tails" 0 (V.length blocked)
+    | polling <- [False, True]
+    ]
+      ++ [ testCase "nullable metrics and explicit premake" $
+             withSemconvOptIn "messaging/dup,database/dup" $
+               isolated $ \pool -> do
+                 (tracer, _, spansRef) <- setupTracer
+                 let run :: Eff '[Eff.Pgmq, Error PgmqRuntimeError, IOE] a -> IO a
+                     run action = assertRight =<< runEff (runError @PgmqRuntimeError ((if traced then runPgmqTraced pool tracer else runPgmq pool) action))
+                 queue <- mkUniqueQueue "ordinary"
+                 run (Eff.createQueue queue)
+                 ordinary <- run (Eff.queueMetrics queue)
+                 assertEqual "ordinary metric" Nothing (Types.defaultPartitionLength ordinary)
+                 ordinaryAll <- run Eff.allQueueMetrics
+                 assertEqual "all ordinary metrics" [Nothing] (map Types.defaultPartitionLength ordinaryAll)
+                 withPartman pool $ do
+                   stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"
+                   partitioned <- mkUniqueQueue "partitioned"
+                   let request = Types.CreatePartitionedQueue partitioned "10" "100"
+                   if stock
+                     then run (Eff.createPartitionedQueue request)
+                     else run (Eff.createPartitionedQueueWithPremake request 2)
+                   counts <-
+                     session pool $
+                       Session.statement (queueNameToText partitioned) $
+                         preparable
+                           "select premake from partman.part_config where parent_table in ('pgmq.q_' || $1, 'pgmq.a_' || $1) order by parent_table"
+                           (E.param (E.nonNullable E.text))
+                           (D.rowList (D.column (D.nonNullable D.int4)))
+                   assertEqual "both parents" (replicate 2 (if stock then 4 else 2 :: Int32)) counts
+                   let q = queueNameToText partitioned
+                   session pool (Session.script ("ANALYZE pgmq.q_" <> q <> "_default; ANALYZE pgmq.a_" <> q <> "_default"))
+                   metric <- run (Eff.queueMetrics partitioned)
+                   let expected = if stock then Nothing else Just 0
+                   assertEqual "partition estimate" expected (Types.defaultPartitionLength metric)
+                   metrics <- run Eff.allQueueMetrics
+                   assertEqual "all partition estimates" [expected] [Types.defaultPartitionLength m | m <- metrics, metricsName m == q]
+                   when traced $ do
+                     spans <- readIORef spansRef
+                     creates <- spansWithFirstWord "pgmq.create_partitioned" spans
+                     observedSpan <- singleSpan creates "partition creation"
+                     spanName observedSpan >>= assertEqual "partition span name" ("pgmq.create_partitioned " <> q)
+                     case OTel.spanKind observedSpan of
+                       OTel.Internal -> pure ()
+                       other -> assertFailure ("expected Internal: " <> show other)
+                     assertAttrText observedSpan "db.operation" "pgmq.create_partitioned"
+                     assertAttrText observedSpan "db.operation.name" "pgmq.create_partitioned",
+           testCase "explicit premake errors reach runtime error channel" $
+             withSemconvOptIn "" $ do
+               -- Fresh pools avoid cached failed prepares on the unsupported 1.12 signature.
+               mapM_
+                 ( \n -> isolated $ \pool -> withPartman pool $ do
+                     (tracer, _, _) <- setupTracer
+                     queue <- mkUniqueQueue "invalid_premake"
+                     stock <- (== Just "1.12.0") <$> lookupEnv "PGMQ_TEST_SCHEMA_VERSION"
+                     result <-
+                       runEff
+                         ( runError @PgmqRuntimeError
+                             ( (if traced then runPgmqTraced pool tracer else runPgmq pool)
+                                 (Eff.createPartitionedQueueWithPremake (Types.CreatePartitionedQueue queue "10" "100") n)
+                             )
+                         )
+                     case result of
+                       Left (_, PgmqSessionError err) -> assertBool "server rejection is retained" (T.isInfixOf (if stock then "42883" else "premake must be at least 1") (T.pack (show err)))
+                       other -> assertFailure ("expected session error, got " <> show other)
+                 )
+                 [0, -1]
+         ]
+
+session :: Pool.Pool -> Session.Session a -> IO a
+session pool action = assertRight =<< Pool.use pool action
+
+isolated :: (Pool.Pool -> IO ()) -> IO ()
+isolated action = assertRight =<< withPgmqDb (\pool -> bracket (pure pool) Pool.release action)
+
+withPartman :: Pool.Pool -> IO () -> IO ()
+withPartman pool action = do
+  available <-
+    session pool $
+      Session.statement () $
+        preparable
+          "select exists (select from pg_extension where extname = 'pg_partman')"
+          E.noParams
+          (D.singleRow (D.column (D.nonNullable D.bool)))
+  required <- (== Just "1") <$> lookupEnv "PGMQ_REQUIRE_PARTMAN"
+  if available then action else if required then assertFailure "pg_partman required" else putStrLn "SKIPPED: pg_partman is unavailable"
+
+metricsName :: Types.QueueMetrics -> Text
+metricsName Types.QueueMetrics {Types.queueName = q} = q
diff --git a/test/UmbrellaExportsSpec.hs b/test/UmbrellaExportsSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/UmbrellaExportsSpec.hs
@@ -0,0 +1,27 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeOperators #-}
+
+-- Compile witnesses: this module intentionally uses only the public umbrella.
+module UmbrellaExportsSpec (grouped, polling, creation, arguments, metric) where
+
+import Data.Int (Int32, Int64)
+import Data.Vector (Vector)
+import Effectful (Eff, (:>))
+import Pgmq.Effectful
+
+grouped :: (Pgmq :> es) => ReadGrouped -> [Eff es (Vector Message)]
+grouped args = [readGrouped args, readGroupedRoundRobin args, readGroupedHead args]
+
+polling :: (Pgmq :> es) => ReadGroupedWithPoll -> [Eff es (Vector Message)]
+polling args = [readGroupedWithPoll args, readGroupedRoundRobinWithPoll args, readGroupedHeadWithPoll args]
+
+creation :: (Pgmq :> es) => CreatePartitionedQueue -> Int32 -> [Eff es ()]
+creation args n = [createPartitionedQueue args, createPartitionedQueueWithPremake args n]
+
+arguments :: QueueName -> (ReadGrouped, ReadGroupedWithPoll, CreatePartitionedQueue)
+arguments q = (ReadGrouped q 30 10, ReadGroupedWithPoll q 30 10 5 100, CreatePartitionedQueue q "100" "1000")
+
+metric :: QueueMetrics -> Maybe Int64
+metric = defaultPartitionLength
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) + <a max length queue name> (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));
+$$;