PostgreSQL Performance Testing: Query Optimization and Load Testing
PostgreSQL performance problems are rarely obvious until they're in production. A query that runs fine on 10,000 rows starts doing sequential scans on 10 million. An index that covers the happy path misses the query your new feature generates. Connection pool exhaustion looks like application errors, not database slowness.
Performance testing PostgreSQL means testing query plans, not just query results. This guide covers the tools and patterns to do it systematically.
Reading EXPLAIN ANALYZE
Before you can test performance, you need to understand what PostgreSQL is actually doing. EXPLAIN ANALYZE is your primary tool:
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT
o.id,
o.total,
c.name AS customer_name,
COUNT(oi.id) AS item_count
FROM orders o
JOIN customers c ON c.id = o.customer_id
LEFT JOIN order_items oi ON oi.order_id = o.id
WHERE o.created_at >= '2024-01-01'
AND o.status = 'completed'
GROUP BY o.id, o.total, c.name
ORDER BY o.total DESC
LIMIT 100;Key things to look for in the output:
Seq Scan on a large table: Bad. Should be Index Scan or Bitmap Index Scan.
Seq Scan on orders (cost=0.00..45231.00 rows=125000 width=48)
(actual time=0.012..892.345 rows=125000 loops=1)Index Scan: Good. Row count should match estimate closely.
Index Scan using idx_orders_created_status on orders
(cost=0.56..324.12 rows=1024 width=48)
(actual time=0.089..12.341 rows=987 loops=1)Hash Join vs Nested Loop: Hash joins are generally better for large datasets; nested loops are better for small ones. If you see nested loops on large tables, that's a problem.
Buffers: shared hit=X means rows from cache; shared read=X means disk reads.
Automating EXPLAIN Analysis
Don't manually read EXPLAIN output in CI — parse it programmatically:
import psycopg2
import json
import pytest
def get_query_plan(conn, query: str, params=None) -> dict:
"""Execute EXPLAIN ANALYZE and return parsed JSON plan."""
explain_query = f"EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) {query}"
with conn.cursor() as cur:
cur.execute(explain_query, params)
return cur.fetchone()[0][0]
def find_nodes_by_type(plan: dict, node_type: str) -> list:
"""Recursively find all plan nodes of a given type."""
results = []
node = plan.get("Plan", plan)
if node.get("Node Type") == node_type:
results.append(node)
for child in node.get("Plans", []):
results.extend(find_nodes_by_type(child, node_type))
return results
def test_order_search_uses_index(db_conn):
"""Order search must use index scan, not sequential scan."""
plan = get_query_plan(db_conn, """
SELECT id, total, status
FROM orders
WHERE customer_id = %s
AND created_at >= %s
ORDER BY created_at DESC
LIMIT 20
""", params=(1, "2024-01-01"))
seq_scans = find_nodes_by_type(plan, "Seq Scan")
for scan in seq_scans:
if scan.get("Relation Name") == "orders":
row_count = scan.get("Actual Rows", 0)
assert row_count < 1000, \
f"Seq scan on orders table touching {row_count} rows — add index"
def test_customer_order_join_performance(db_conn):
"""Customer-order join must complete under 100ms."""
plan = get_query_plan(db_conn, """
SELECT c.name, COUNT(o.id) AS order_count, SUM(o.total) AS revenue
FROM customers c
LEFT JOIN orders o ON o.customer_id = c.id
WHERE c.region = %s
GROUP BY c.id, c.name
""", params=("US",))
actual_time = plan["Plan"].get("Actual Total Time", 0)
assert actual_time < 100.0, \
f"Join query took {actual_time:.1f}ms, expected under 100ms"pg_stat_statements: Finding Real Slow Queries
pg_stat_statements tracks cumulative query statistics across all executions. It's the best tool for finding what's actually slow in production:
-- Enable the extension (requires superuser, add to postgresql.conf: shared_preload_libraries = 'pg_stat_statements')
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;
-- Top 10 slowest queries by total execution time
SELECT
substring(query, 1, 80) AS query_preview,
calls,
round(total_exec_time::numeric, 2) AS total_ms,
round(mean_exec_time::numeric, 2) AS avg_ms,
round(stddev_exec_time::numeric, 2) AS stddev_ms,
rows
FROM pg_stat_statements
WHERE query NOT LIKE '%pg_stat%'
ORDER BY total_exec_time DESC
LIMIT 10;
-- Queries with high variance (inconsistent performance)
SELECT
substring(query, 1, 80) AS query_preview,
calls,
round(mean_exec_time::numeric, 2) AS avg_ms,
round(stddev_exec_time::numeric, 2) AS stddev_ms,
round((stddev_exec_time / NULLIF(mean_exec_time, 0) * 100)::numeric, 1) AS cv_pct
FROM pg_stat_statements
WHERE calls > 100
AND mean_exec_time > 10
ORDER BY cv_pct DESC
LIMIT 10;Use this in a performance regression test:
def test_no_regressions_in_query_performance(db_conn, baseline_stats):
"""No query should be more than 50% slower than the baseline."""
with db_conn.cursor() as cur:
cur.execute("""
SELECT query, mean_exec_time, calls
FROM pg_stat_statements
WHERE calls > 50
AND mean_exec_time > 5
""")
current_stats = {row[0]: row[1] for row in cur.fetchall()}
regressions = []
for query, current_mean in current_stats.items():
baseline_mean = baseline_stats.get(query)
if baseline_mean and current_mean > baseline_mean * 1.5:
regressions.append({
"query": query[:80],
"baseline_ms": baseline_mean,
"current_ms": current_mean,
"regression_pct": (current_mean / baseline_mean - 1) * 100
})
assert not regressions, (
f"Performance regressions detected:\n" +
"\n".join(
f" {r['query']}: {r['baseline_ms']:.1f}ms → {r['current_ms']:.1f}ms "
f"(+{r['regression_pct']:.0f}%)"
for r in regressions
)
)Index Testing
Indexes are the most impactful performance lever in PostgreSQL. Test that they exist, are used, and don't degrade write performance unacceptably.
Testing Index Existence
def test_required_indexes_exist(db_conn):
"""Critical performance indexes must exist on production tables."""
required_indexes = [
("orders", "customer_id"),
("orders", "created_at"),
("orders", "status"),
("order_items", "order_id"),
("sessions", "user_id"),
("sessions", "expires_at"),
]
with db_conn.cursor() as cur:
for table, column in required_indexes:
cur.execute("""
SELECT COUNT(*)
FROM pg_indexes pi
JOIN pg_attribute pa ON true
JOIN pg_class pc ON pc.relname = pi.tablename
WHERE pi.tablename = %s
AND pi.indexdef LIKE %s
""", (table, f"%({column})%"))
count = cur.fetchone()[0]
assert count > 0, \
f"Missing index on {table}.{column} — add: CREATE INDEX ON {table}({column})"Testing Index Selectivity
A low-selectivity index (e.g., on a boolean column) is often worse than no index:
def test_status_index_selectivity(db_conn):
"""Status index should not be used for low-selectivity values like 'completed'."""
# If 90% of orders are 'completed', a seq scan + filter may be faster than an index scan
with db_conn.cursor() as cur:
cur.execute("SELECT COUNT(*) FROM orders WHERE status = 'completed'")
completed_count = cur.fetchone()[0]
cur.execute("SELECT COUNT(*) FROM orders")
total_count = cur.fetchone()[0]
selectivity = completed_count / total_count if total_count > 0 else 0
if selectivity > 0.5:
# For low-selectivity queries, verify planner chose seq scan (correct behavior)
plan = get_query_plan(db_conn, "SELECT id FROM orders WHERE status = 'completed'")
# Not asserting index usage here — seq scan is correct for 90% selectivity
pytest.skip(f"Status='completed' has {selectivity:.0%} selectivity — index usage not expected")Testing Write Performance with Indexes
Every index slows down INSERT/UPDATE/DELETE. Test that critical write paths stay fast:
import time
def test_order_insert_performance_with_indexes(db_conn, large_order_dataset):
"""Order insert with all indexes should complete under 50ms per insert."""
times = []
with db_conn.cursor() as cur:
for i in range(100):
start = time.monotonic()
cur.execute("""
INSERT INTO orders (customer_id, total, status, created_at)
VALUES (%s, %s, %s, NOW())
RETURNING id
""", (1, 100.0 + i, 'pending'))
db_conn.commit()
times.append(time.monotonic() - start)
p95_ms = sorted(times)[94] * 1000 # 95th percentile
avg_ms = sum(times) / len(times) * 1000
assert p95_ms < 50.0, f"Order insert p95 is {p95_ms:.1f}ms (limit: 50ms)"
assert avg_ms < 20.0, f"Order insert average is {avg_ms:.1f}ms (limit: 20ms)"Connection Pool Load Testing
Connection pool exhaustion is a common production failure mode. Test it explicitly:
import concurrent.futures
import psycopg2.pool
import threading
def test_connection_pool_under_concurrent_load(db_config):
"""100 concurrent requests complete without connection pool exhaustion."""
pool = psycopg2.pool.ThreadedConnectionPool(
minconn=5,
maxconn=20,
**db_config
)
errors = []
latencies = []
def run_query(query_id: int):
try:
conn = pool.getconn()
start = time.monotonic()
with conn.cursor() as cur:
cur.execute("""
SELECT o.id, c.name, SUM(oi.price * oi.quantity) as total
FROM orders o
JOIN customers c ON c.id = o.customer_id
JOIN order_items oi ON oi.order_id = o.id
WHERE o.status = 'completed'
GROUP BY o.id, c.name
LIMIT 10
""")
cur.fetchall()
latencies.append(time.monotonic() - start)
pool.putconn(conn)
except Exception as e:
errors.append(f"Query {query_id}: {e}")
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
futures = [executor.submit(run_query, i) for i in range(100)]
concurrent.futures.wait(futures, timeout=30)
pool.closeall()
assert not errors, f"Connection errors under load:\n" + "\n".join(errors[:5])
p99_ms = sorted(latencies)[98] * 1000
assert p99_ms < 1000, f"p99 query latency under load: {p99_ms:.0f}ms (limit: 1000ms)"Vacuum and Bloat Testing
PostgreSQL's MVCC creates dead tuples that accumulate without vacuuming. Test that autovacuum is keeping up:
-- Check table bloat
SELECT
schemaname,
tablename,
pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS total_size,
pg_size_pretty(pg_relation_size(schemaname||'.'||tablename)) AS table_size,
n_dead_tup,
n_live_tup,
round(n_dead_tup::numeric / NULLIF(n_live_tup + n_dead_tup, 0) * 100, 1) AS dead_pct,
last_autovacuum,
last_autoanalyze
FROM pg_stat_user_tables
WHERE n_live_tup > 10000
ORDER BY dead_pct DESC NULLS LAST
LIMIT 10;def test_table_bloat_within_threshold(db_conn):
"""No large table should have more than 20% dead tuples."""
with db_conn.cursor() as cur:
cur.execute("""
SELECT tablename, n_dead_tup, n_live_tup,
round(n_dead_tup::numeric / NULLIF(n_live_tup + n_dead_tup, 0) * 100, 1) AS dead_pct
FROM pg_stat_user_tables
WHERE n_live_tup > 100000
""")
rows = cur.fetchall()
high_bloat = [
(table, dead_pct)
for table, n_dead, n_live, dead_pct in rows
if dead_pct and dead_pct > 20
]
assert not high_bloat, (
"Tables with excessive bloat (autovacuum may be misconfigured):\n" +
"\n".join(f" {t}: {p}% dead tuples" for t, p in high_bloat)
)Integrating Performance Tests in CI
Performance tests should run on PRs that touch database-related code, and nightly against production-like data:
# .github/workflows/postgres-perf.yml
name: PostgreSQL Performance Tests
on:
schedule:
- cron: '0 2 * * *' # Nightly at 2 AM
pull_request:
paths:
- 'migrations/**'
- 'src/repositories/**'
- 'src/models/**'
jobs:
performance:
runs-on: ubuntu-latest
services:
postgres:
image: postgres:16-alpine
env:
POSTGRES_DB: perftest
POSTGRES_USER: test
POSTGRES_PASSWORD: test
POSTGRES_INITDB_ARGS: "-c shared_preload_libraries=pg_stat_statements"
ports:
- 5432:5432
options: --health-cmd pg_isready --health-interval 10s
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.12'
- run: pip install pytest psycopg2-binary pytest-benchmark
- name: Setup schema and seed data
run: |
psql postgresql://test:test@localhost/perftest -f schema.sql
python scripts/seed_performance_data.py --rows 1000000
- name: Run performance tests
run: pytest tests/performance/ -v --benchmark-autosave
env:
DATABASE_URL: postgresql://test:test@localhost/perftest
- name: Store benchmark results
uses: actions/upload-artifact@v4
with:
name: benchmark-results
path: .benchmarks/Performance Testing Checklist
For every PR that changes queries or schema:
- EXPLAIN ANALYZE on all new/changed queries
- No new Seq Scans on tables > 10,000 rows
- Index tested for new query patterns
- Join performance tested on representative data volume
- Write performance tested if new indexes added
- p95 latency within acceptable threshold
- Connection pool stress test for new concurrent endpoints
PostgreSQL performance testing is not a one-time exercise — it's a practice. Every schema change, every new query, and every new index is a potential regression. The tooling (EXPLAIN ANALYZE, pg_stat_statements, automated plan parsing in CI) makes it possible to catch performance regressions at review time rather than during an incident at 3 AM. Start with the slowest queries in your current pg_stat_statements output and work backward from there.
