pgmq-migration-0.6.0.0: migrations/0006-preserve-partitioned-reentry-v1.13.0.sql

-- Local override of the pristine 1.13 partition function.
-- See docs/adr/pgmq-1.12-1.13-compatibility.md and migration 0003.
CREATE OR REPLACE FUNCTION pgmq.create_partitioned(
  queue_name TEXT,
  partition_interval TEXT DEFAULT '10000',
  retention_interval TEXT DEFAULT '100000',
  premake INTEGER DEFAULT 4
)
RETURNS void AS $$
DECLARE
  partition_col TEXT;
  a_partition_col TEXT;
  l_already_managed BOOLEAN;
  qtable TEXT := pgmq.format_table_name(queue_name, 'q');
  qtable_seq TEXT := qtable || '_msg_id_seq';
  atable TEXT := pgmq.format_table_name(queue_name, 'a');
  fq_qtable TEXT := 'pgmq.' || qtable;
  fq_atable TEXT := 'pgmq.' || atable;
BEGIN
  PERFORM pgmq.validate_queue_name(queue_name);
  PERFORM pgmq.acquire_queue_lock(queue_name);
  PERFORM pgmq._ensure_pg_partman_installed();
  IF premake < 1 THEN
    RAISE EXCEPTION 'premake must be at least 1, got %', premake;
  END IF;
  SELECT pgmq._get_partition_col(partition_interval) INTO partition_col;

  EXECUTE FORMAT(
    $QUERY$
    CREATE TABLE IF NOT EXISTS pgmq.%I (
        msg_id BIGINT GENERATED BY DEFAULT AS IDENTITY,
        read_ct INT DEFAULT 0 NOT NULL,
        enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
        last_read_at TIMESTAMP WITH TIME ZONE,
        vt TIMESTAMP WITH TIME ZONE NOT NULL,
        message JSONB,
        headers JSONB
    ) PARTITION BY RANGE (%I)
    $QUERY$,
    qtable, partition_col
  );

  -- Preserve the parent-registration guard from migration 0003.
  EXECUTE FORMAT(
    'SELECT EXISTS (SELECT 1 FROM %I.part_config WHERE parent_table = %L)',
    pgmq._get_pg_partman_schema(), fq_qtable
  ) INTO l_already_managed;
  IF NOT l_already_managed THEN
    -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md
    -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.
    EXECUTE FORMAT(
      $QUERY$
      SELECT %I.create_parent(
        p_parent_table := %L,
        p_control := %L,
        p_interval := %L,
        p_premake := %s,
        p_type := case
          when pgmq._get_pg_partman_major_version() = 5 then 'range'
          else 'native'
        end
      )
      $QUERY$,
      pgmq._get_pg_partman_schema(),
      fq_qtable,
      partition_col,
      partition_interval,
      premake
    );
  END IF;

  EXECUTE FORMAT(
    $QUERY$
    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (%I);
    $QUERY$,
    qtable || '_part_idx', qtable, partition_col
  );

  EXECUTE FORMAT(
    $QUERY$
    UPDATE %I.part_config
    SET
        retention = %L,
        retention_keep_table = false,
        retention_keep_index = true,
        automatic_maintenance = 'on'
    WHERE parent_table = %L;
    $QUERY$,
    pgmq._get_pg_partman_schema(),
    retention_interval,
    'pgmq.' || qtable
  );

  EXECUTE FORMAT(
    $QUERY$
    INSERT INTO pgmq.meta (queue_name, is_partitioned, is_unlogged)
    VALUES (%L, true, false)
    ON CONFLICT
    DO NOTHING;
    $QUERY$,
    queue_name
  );

  IF partition_col = 'enqueued_at' THEN
    a_partition_col := 'archived_at';
  ELSE
    a_partition_col := partition_col;
  END IF;

  EXECUTE FORMAT(
    $QUERY$
    CREATE TABLE IF NOT EXISTS pgmq.%I (
      msg_id BIGINT NOT NULL,
      read_ct INT DEFAULT 0 NOT NULL,
      enqueued_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
      last_read_at TIMESTAMP WITH TIME ZONE,
      archived_at TIMESTAMP WITH TIME ZONE DEFAULT now() NOT NULL,
      vt TIMESTAMP WITH TIME ZONE NOT NULL,
      message JSONB,
      headers JSONB
    ) PARTITION BY RANGE (%I);
    $QUERY$,
    atable, a_partition_col
  );

  -- Preserve the parent-registration guard from migration 0003.
  EXECUTE FORMAT(
    'SELECT EXISTS (SELECT 1 FROM %I.part_config WHERE parent_table = %L)',
    pgmq._get_pg_partman_schema(), fq_atable
  ) INTO l_already_managed;
  IF NOT l_already_managed THEN
    -- https://github.com/pgpartman/pg_partman/blob/master/doc/pg_partman.md
    -- p_parent_table - the existing parent table. MUST be schema qualified, even if in public schema.
    EXECUTE FORMAT(
      $QUERY$
      SELECT %I.create_parent(
        p_parent_table := %L,
        p_control := %L,
        p_interval := %L,
        p_premake := %s,
        p_type := case
          when pgmq._get_pg_partman_major_version() = 5 then 'range'
          else 'native'
        end
      )
      $QUERY$,
      pgmq._get_pg_partman_schema(),
      fq_atable,
      a_partition_col,
      partition_interval,
      premake
    );
  END IF;

  EXECUTE FORMAT(
    $QUERY$
    UPDATE %I.part_config
    SET
        retention = %L,
        retention_keep_table = false,
        retention_keep_index = true,
        automatic_maintenance = 'on'
    WHERE parent_table = %L;
    $QUERY$,
    pgmq._get_pg_partman_schema(),
    retention_interval,
    'pgmq.' || atable
  );

  EXECUTE FORMAT(
    $QUERY$
    CREATE INDEX IF NOT EXISTS %I ON pgmq.%I (archived_at);
    $QUERY$,
    'archived_at_idx_' || queue_name, atable
  );

END;
$$ LANGUAGE plpgsql;