pgmq-migration-0.1.2.0: vendor/pgmq/pgmq-extension/sql/pgmq--1.10.0--1.10.1.sql
DROP FUNCTION IF EXISTS pgmq.create_non_partitioned(TEXT);
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;
DROP FUNCTION IF EXISTS pgmq.create_unlogged(TEXT);
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;
DROP FUNCTION IF EXISTS pgmq.create_partitioned(TEXT, TEXT, TEXT);
CREATE FUNCTION pgmq.create_partitioned(
queue_name TEXT,
partition_interval TEXT DEFAULT '10000',
retention_interval TEXT DEFAULT '100000'
)
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;