Module: Pgbus::TableMaintenance

Defined in:
lib/pgbus/table_maintenance.rb

Overview

Proactive table maintenance to reduce bloat on PGMQ queue tables.

PGMQ’s read operation UPDATEs three columns (vt, read_ct, last_read_at) on every message read. With the default fillfactor of 100, every UPDATE creates a new heap tuple AND a new index entry — the dead tuple and its old index pointer remain until VACUUM. Under sustained load, autovacuum can’t keep up and B-tree indexes bloat.

Setting fillfactor=70 on queue tables reserves 30% of each page for update churn. Because ‘vt` is indexed and changes on every read, these writes are not HOT updates, but leaving headroom on heap pages still reduces page density for a table that is updated heavily between vacuum passes.

More importantly, this module provides targeted VACUUM: instead of relying solely on autovacuum’s global heuristics, the dispatcher periodically checks pg_stat_user_tables for tables with high dead tuple ratios and vacuums them explicitly. This is inspired by pgque’s philosophy of measuring bloat before acting.

Constant Summary collapse

FILLFACTOR = 
70
BLOAT_THRESHOLD = 
0.1
MAINTENANCE_INTERVAL =

6 hours

6 * 3600

Class Method Summary collapse

Class Method Details

.fillfactor_sql_for_all_queues ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/pgbus/table_maintenance.rb', line 33

def fillfactor_sql_for_all_queues
  <<~SQL
    DO $$
    DECLARE
      q RECORD;
    BEGIN
      FOR q IN SELECT queue_name FROM pgmq.meta LOOP
        EXECUTE format('ALTER TABLE pgmq.q_%I SET (fillfactor = #{FILLFACTOR})', q.queue_name);
      END LOOP;
    END $$;
  SQL
end

.fillfactor_sql_for_queue(queue_name) ⇒ Object 



29
30
31
 
# File 'lib/pgbus/table_maintenance.rb', line 29

def fillfactor_sql_for_queue(queue_name)
  "ALTER TABLE pgmq.q_#{queue_name} SET (fillfactor = #{FILLFACTOR});"
end

.reindex_sql(table) ⇒ Object 



78
79
80
81
 
# File 'lib/pgbus/table_maintenance.rb', line 78

def reindex_sql(table)
  schema, relname = table.split(".", 2)
  "REINDEX TABLE CONCURRENTLY \"#{schema}\".\"#{relname}\""
end

.run_maintenance(conn, threshold: BLOAT_THRESHOLD, reindex: true) ⇒ Object 



83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/pgbus/table_maintenance.rb', line 83

def run_maintenance(conn, threshold: BLOAT_THRESHOLD, reindex: true)
  candidates = vacuum_candidates(conn, threshold: threshold)
  return 0 if candidates.empty?

  maintained = 0
  candidates.each do |candidate|
    table = candidate[:table]
    Pgbus.logger.info do
      "[Pgbus::TableMaintenance] Vacuuming #{table} " \
        "(dead_ratio=#{candidate[:dead_ratio]}, dead=#{candidate[:dead_tuples]})"
    end
    conn.exec(vacuum_sql(table))

    if reindex
      Pgbus.logger.info { "[Pgbus::TableMaintenance] Reindexing #{table}" }
      conn.exec(reindex_sql(table))
    end

    maintained += 1
  rescue StandardError => e
    Pgbus.logger.error { "[Pgbus::TableMaintenance] Failed to maintain #{table}: #{e.message}" }
  end

  maintained
end

.vacuum_candidates(conn, threshold: BLOAT_THRESHOLD) ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
 
# File 'lib/pgbus/table_maintenance.rb', line 46

def vacuum_candidates(conn, threshold: BLOAT_THRESHOLD)
  rows = conn.exec(<<~SQL)
    SELECT schemaname, relname, n_dead_tup, n_live_tup
    FROM pg_stat_user_tables
    WHERE schemaname = 'pgmq'
      AND relname LIKE 'q_%'
    ORDER BY n_dead_tup DESC
  SQL

  rows.each_with_object([]) do |row, candidates|
    dead = row["n_dead_tup"].to_i
    live = row["n_live_tup"].to_i
    total = dead + live
    next if total.zero?

    ratio = dead.to_f / total
    next unless ratio > threshold

    candidates << {
      table: "#{row["schemaname"]}.#{row["relname"]}",
      dead_tuples: dead,
      live_tuples: live,
      dead_ratio: ratio.round(4)
    }
  end
end

.vacuum_sql(table) ⇒ Object 



73
74
75
76
 
# File 'lib/pgbus/table_maintenance.rb', line 73

def vacuum_sql(table)
  schema, relname = table.split(".", 2)
  "VACUUM \"#{schema}\".\"#{relname}\""
end