PostgreSQL 使用全面指南：从入门到企业级应用

# Ubuntu / Debian
sudo apt update && sudo apt install postgresql postgresql-contrib

# CentOS / RHEL
sudo dnf install postgresql-server postgresql-contrib
sudo postgresql-setup --initdb

# macOS
brew install postgresql@16 && brew services start postgresql@16

sudo systemctl enable --now postgresql
sudo -u postgres psql      # 用 postgres 系统用户进入

CREATE USER app WITH PASSWORD 'StrongPass!2026';

CREATE DATABASE sales_db OWNER app ENCODING 'UTF8';

-- 用户级默认权限（可选，免得到处 GRANT）
ALTER USER app SET search_path TO app_schema, public;

# 类型   数据库   用户    地址            认证方法
local    all      all                     peer
host     all      all     127.0.0.1/32    scram-sha-256
host     all      all     ::1/128         scram-sha-256
host     sales_db app     10.0.0.0/8      scram-sha-256

SELECT pg_reload_conf();      -- 或 sudo systemctl reload postgresql

# postgresql.conf
listen_addresses = '*'

-- 传统写法
CREATE TABLE t_serial (id SERIAL PRIMARY KEY, name TEXT);

-- SQL 标准写法（PG 10+，推荐）
CREATE TABLE t_identity (
    id   INT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name TEXT
);

CREATE TABLE employees (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    name        VARCHAR(100) NOT NULL,
    email       VARCHAR(100) NOT NULL UNIQUE,
    department  VARCHAR(50),
    salary      NUMERIC(10,2) CHECK (salary >= 0),
    skills      TEXT[],
    profile     JSONB DEFAULT '{}'::jsonb,
    hire_date   DATE DEFAULT CURRENT_DATE,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- 批量插入
INSERT INTO employees (name, email, department, salary) VALUES
('张伟', 'zhang@company.com', '研发部', 15000),
('李娜', 'li@company.com', '市场部', 12000);

-- RETURNING：插入的同时拿回生成的值（PG 特色，省一次查询）
INSERT INTO employees (name, email, salary)
VALUES ('王芳', 'wang@company.com', 13000)
RETURNING id, created_at;

-- UPSERT：冲突时更新
INSERT INTO employees (id, name, salary) VALUES (1, '张伟', 16000)
ON CONFLICT (id) DO UPDATE SET salary = EXCLUDED.salary;

-- 冲突时不做任何事
INSERT INTO employees (id, name, salary) VALUES (1, '张伟', 16000)
ON CONFLICT (id) DO NOTHING;

UPDATE employees SET salary = salary * 1.1
WHERE department = '研发部'
RETURNING id, name, salary;      -- 直接看到改了什么

DELETE FROM employees WHERE id = 5 RETURNING *;

SELECT
    name, department, salary,
    AVG(salary) OVER (PARTITION BY department)              AS dept_avg,
    RANK()      OVER (PARTITION BY department ORDER BY salary DESC) AS dept_rank,
    salary - AVG(salary) OVER (PARTITION BY department)     AS diff_from_avg
FROM employees;

WITH dept_stats AS (
    SELECT department, COUNT(*) AS cnt, AVG(salary) AS avg_salary
    FROM employees
    GROUP BY department
)
SELECT * FROM dept_stats WHERE cnt > 3 ORDER BY avg_salary DESC;

WITH RECURSIVE org AS (
    SELECT id, name, manager_id, 1 AS level
    FROM employees WHERE manager_id IS NULL
    UNION ALL
    SELECT e.id, e.name, e.manager_id, o.level + 1
    FROM employees e JOIN org o ON e.manager_id = o.id
)
SELECT * FROM org ORDER BY level, id;

SELECT DISTINCT ON (department) department, name, salary
FROM employees
ORDER BY department, salary DESC;

-- 每个部门取工资最高的 2 个人
SELECT d.name AS dept, e.name, e.salary
FROM departments d
CROSS JOIN LATERAL (
    SELECT name, salary FROM employees
    WHERE department = d.name
    ORDER BY salary DESC LIMIT 2
) e;

-- 生成日期序列（做报表补全空缺日期特别好用）
SELECT d::date AS day
FROM generate_series('2026-01-01'::date, '2026-01-10'::date, '1 day') d;

-- 生成 1-10
SELECT generate_series(1, 10);

-- FILTER：对聚合加条件（标准 SQL）
SELECT
    department,
    COUNT(*)                              AS total,
    COUNT(*) FILTER (WHERE salary > 13000) AS high_paid,
    string_agg(name, ', ' ORDER BY salary DESC) AS members,
    array_agg(name)                       AS names_array
FROM employees
GROUP BY department;

-- 查询包含某元素的数组
SELECT * FROM employees WHERE 'Python' = ANY(skills);

-- 展开数组为多行
SELECT name, unnest(skills) AS skill FROM employees;

-- 匹配数组
SELECT * FROM employees WHERE skills @> ARRAY['Python', 'SQL'];

-- 数组长度
SELECT name, array_length(skills, 1) FROM employees;

CREATE TABLE products (
    id         SERIAL PRIMARY KEY,
    name       VARCHAR(100),
    attributes JSONB
);

INSERT INTO products (name, attributes) VALUES
('Laptop', '{"color": "silver", "memory": "16GB", "ports": ["USB-C", "HDMI"]}'),
('Mouse',  '{"color": "black", "wireless": true}');

SELECT name,
       attributes ->> 'color'  AS color,        -- 文本
       attributes -> 'ports'   AS ports_jsonb,  -- JSONB
       attributes #>> '{ports,0}' AS first_port -- 路径取值
FROM products;

-- 包含关系（GIN 索引可以加速）
SELECT * FROM products WHERE attributes @> '{"memory": "16GB"}';

-- 键存在
SELECT * FROM products WHERE attributes ? 'wireless';

-- 嵌套条件
SELECT * FROM products WHERE attributes ->> 'color' = 'silver';

-- 通用包含查询索引
CREATE INDEX idx_products_attrs ON products USING GIN (attributes);

-- 针对特定路径的索引（更小更快）
CREATE INDEX idx_products_color ON products ((attributes ->> 'color'));

UPDATE products
SET attributes = jsonb_set(attributes, '{memory}', '"32GB"')
WHERE name = 'Laptop';

CREATE TABLE documents (
    id    SERIAL PRIMARY KEY,
    title TEXT,
    body  TEXT
);

-- 建表达式索引（GIN 是文本搜索的标准选择）
CREATE INDEX idx_documents_fts ON documents
USING GIN (to_tsvector('english', body));

-- 搜索 + 高亮
SELECT title,
       ts_headline('english', body, to_tsquery('english', 'database & performance')) AS snippet
FROM documents
WHERE to_tsvector('english', body) @@ to_tsquery('english', 'database & performance');

-- 排序按相关度
SELECT title, ts_rank(to_tsvector('english', body), q) AS rank
FROM documents, to_tsquery('english', 'database') q
WHERE to_tsvector('english', body) @@ q
ORDER BY rank DESC;

CREATE EXTENSION pg_trgm;
CREATE INDEX idx_title_trgm ON documents USING GIN (title gin_trgm_ops);
SELECT * FROM documents WHERE title ILIKE '%数据库%';

-- 表达式索引：查询里常用 lower(email) 就索引它
CREATE INDEX idx_users_email_lower ON users (lower(email));
-- 查询要写成同样的表达式才能命中：
SELECT * FROM users WHERE lower(email) = lower('A@B.com');

-- 部分索引：只索引满足条件的行（索引更小）
CREATE INDEX idx_orders_pending ON orders (created_at) WHERE status = 'pending';

-- 覆盖索引：把查询需要的列放进索引，避免回表（PG 11+）
CREATE INDEX idx_orders_cover ON orders (customer_id) INCLUDE (amount, status);

-- ❌ 对索引列做函数运算
SELECT * FROM orders WHERE date_trunc('day', created_at) = '2026-01-01';
-- ✅ 改写成范围
SELECT * FROM orders WHERE created_at >= '2026-01-01' AND created_at < '2026-01-02';

-- ❌ 隐式类型转换（id 是 bigint，传了字符串）
SELECT * FROM orders WHERE id = '123';
-- ✅ 类型对上
SELECT * FROM orders WHERE id = 123;

EXPLAIN (ANALYZE, BUFFERS) SELECT * FROM orders WHERE customer_id = 42;

-- 声明式分区（PG 10+）
CREATE TABLE sales (
    id       BIGINT GENERATED ALWAYS AS IDENTITY,
    sale_date DATE NOT NULL,
    amount   NUMERIC(12,2)
) PARTITION BY RANGE (sale_date);

CREATE TABLE sales_2026_q1 PARTITION OF sales
    FOR VALUES FROM ('2026-01-01') TO ('2026-04-01');

CREATE TABLE sales_2026_q2 PARTITION OF sales
    FOR VALUES FROM ('2026-04-01') TO ('2026-07-01');

SHOW default_transaction_isolation;                       -- 默认 read committed
SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;

-- 行级锁：锁定读到的行
BEGIN;
SELECT * FROM accounts WHERE id = 1 FOR UPDATE;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;
COMMIT;

-- FOR UPDATE NOWAIT：拿不到锁立即报错，不等待
SELECT * FROM accounts WHERE id = 1 FOR UPDATE NOWAIT;

-- FOR UPDATE SKIP LOCKED：跳过已被锁的行（做任务队列的经典技巧）
SELECT * FROM jobs WHERE status = 'pending' LIMIT 1 FOR UPDATE SKIP LOCKED;

CREATE OR REPLACE FUNCTION calculate_tax(amount NUMERIC, rate NUMERIC DEFAULT 0.1)
RETURNS NUMERIC AS $$
BEGIN
    RETURN ROUND(amount * rate, 2);
END;
$$ LANGUAGE plpgsql IMMUTABLE;

SELECT calculate_tax(1000);        -- 100.00
SELECT calculate_tax(1000, 0.06);  -- 60.00

CREATE OR REPLACE FUNCTION get_employees(dept VARCHAR)
RETURNS TABLE (id BIGINT, name TEXT, salary NUMERIC) AS $$
BEGIN
    RETURN QUERY
    SELECT e.id, e.name, e.salary
    FROM employees e
    WHERE e.department = dept;
END;
$$ LANGUAGE plpgsql STABLE;

SELECT * FROM get_employees('研发部');

CREATE OR REPLACE PROCEDURE raise_salaries(pct NUMERIC)
LANGUAGE plpgsql AS $$
BEGIN
    UPDATE employees SET salary = salary * (1 + pct / 100);
    COMMIT;      -- 过程内可以控制事务
END;
$$;

CALL raise_salaries(10);

CREATE TABLE audit_log (
    id          BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    table_name  TEXT,
    action      TEXT,
    old_data    JSONB,
    new_data    JSONB,
    change_time TIMESTAMPTZ DEFAULT now()
);

CREATE OR REPLACE FUNCTION log_employee_changes()
RETURNS TRIGGER AS $$
BEGIN
    IF (TG_OP = 'DELETE') THEN
        INSERT INTO audit_log (table_name, action, old_data)
        VALUES (TG_TABLE_NAME, 'DELETE', to_jsonb(OLD));
    ELSIF (TG_OP = 'UPDATE') THEN
        INSERT INTO audit_log (table_name, action, old_data, new_data)
        VALUES (TG_TABLE_NAME, 'UPDATE', to_jsonb(OLD), to_jsonb(NEW));
    ELSIF (TG_OP = 'INSERT') THEN
        INSERT INTO audit_log (table_name, action, new_data)
        VALUES (TG_TABLE_NAME, 'INSERT', to_jsonb(NEW));
    END IF;
    RETURN COALESCE(NEW, OLD);
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER employees_audit
AFTER INSERT OR UPDATE OR DELETE ON employees
FOR EACH ROW EXECUTE FUNCTION log_employee_changes();

CREATE EXTENSION IF NOT EXISTS pg_stat_statements;   -- 慢查询统计（强烈推荐）
CREATE EXTENSION IF NOT EXISTS pg_trgm;              -- 三元组模糊匹配
CREATE EXTENSION IF NOT EXISTS "uuid-ossp";          -- UUID 生成
CREATE EXTENSION IF NOT EXISTS pgcrypto;             -- 加密函数
CREATE EXTENSION IF NOT EXISTS postgis;              -- 地理空间
CREATE EXTENSION IF NOT EXISTS vector;               -- pgvector 向量检索

SELECT * FROM pg_extension;    -- 查看已装扩展

-- 装完需要重启或改 shared_preload_libraries
CREATE EXTENSION pg_stat_statements;

SELECT queryid, calls, round(total_exec_time::numeric, 1) AS total_ms,
       round(mean_exec_time::numeric, 2) AS mean_ms, left(query, 60) AS q
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 10;

-- 正在跑的查询
SELECT pid, usename, state, wait_event_type, wait_event,
       now() - query_start AS duration, left(query, 60) AS q
FROM pg_stat_activity
WHERE state != 'idle'
ORDER BY duration DESC;

-- 温和取消（等价 Ctrl+C）
SELECT pg_cancel_backend(12345);
-- 强制断开
SELECT pg_terminate_backend(12345);

-- 表的扫描情况
SELECT relname, seq_scan, idx_scan, n_live_tup, n_dead_tup
FROM pg_stat_user_tables
ORDER BY seq_scan DESC;

-- 索引使用率（idx_scan=0 的索引可能多余）
SELECT relname, indexrelname, idx_scan
FROM pg_stat_user_indexes
ORDER BY idx_scan LIMIT 10;

SELECT pg_size_pretty(pg_total_relation_size('employees'));      -- 单表
SELECT pg_size_pretty(pg_database_size(current_database()));      -- 当前库

-- 表大小排行
SELECT relname, pg_size_pretty(pg_total_relation_size(relid)) AS size
FROM pg_statio_user_tables
ORDER BY pg_total_relation_size(relid) DESC LIMIT 10;

VACUUM employees;                 -- 回收死元组，不阻塞读写（可并发）
VACUUM ANALYZE employees;         -- 回收 + 更新统计信息
VACUUM FULL employees;            -- ⚠️ 整理并归还空间给 OS，会独占锁表！
ANALYZE employees;                -- 只更新统计信息
REINDEX TABLE employees;          -- 重建索引
REINDEX TABLE CONCURRENTLY employees;  -- 并发重建（不阻塞写，但更慢）

-- 看 autovacuum 是否跟上
SELECT relname, n_dead_tup, last_autovacuum, last_vacuum
FROM pg_stat_user_tables
WHERE n_dead_tup > 10000
ORDER BY n_dead_tup DESC;

# 纯文本 SQL
pg_dump -U app -d sales_db -f sales_db.sql

# 自定义格式（推荐：压缩 + 支持并行/选择性恢复）
pg_dump -U app -d sales_db -Fc -f sales_db.dump

# 目录格式 + 并行（大库提速）
pg_dump -U app -d sales_db -Fd -j 4 -f /backup/sales_db_dir

# 所有数据库（含角色等全局对象）
pg_dumpall -U postgres -f all.sql

# 只导结构 / 只导数据 / 只导某表
pg_dump -U app -d sales_db --schema-only -f schema.sql
pg_dump -U app -d sales_db --data-only -f data.sql
pg_dump -U app -d sales_db -t employees -f employees.sql

# 纯文本
psql -U app -d sales_db -f sales_db.sql

# 自定义 / 目录格式
pg_restore -U app -d sales_db sales_db.dump
pg_restore -U app -d sales_db -j 4 /backup/sales_db_dir

# 只恢复某张表
pg_restore -U app -d sales_db -t employees sales_db.dump

# postgresql.conf
wal_level = replica
archive_mode = on
archive_command = 'cp %p /var/lib/postgresql/wal_archive/%f'

# 基础备份
pg_basebackup -D /backup/base -U replicator -P -Fp -Xs -R

# 1. 恢复基础备份到数据目录
tar -xf /backup/base.tar -C /var/lib/postgresql/data

# 2. 创建 recovery.signal 标记文件（PG 12+ 用这个，不再是 recovery.conf！）
touch /var/lib/postgresql/data/recovery.signal

# 3. 在 postgresql.conf 里配置恢复参数

restore_command = 'cp /var/lib/postgresql/wal_archive/%f %p'
recovery_target_time = '2026-09-22 12:00:00'

# 4. 启动，PG 会自动回放到目标时间点
sudo systemctl start postgresql

# 主库 postgresql.conf
wal_level = replica
max_wal_senders = 10
hot_standby = on

CREATE USER replicator WITH REPLICATION PASSWORD 'ReplPass!2026';

# 备库：用 pg_basebackup 基础备份 + -R 自动写连接配置
pg_basebackup -h 10.0.0.1 -U replicator -D /var/lib/postgresql/data -P -Xs -R

# postgresql.auto.conf
primary_conninfo = 'host=10.0.0.1 port=5432 user=replicator password=ReplPass!2026'

sudo systemctl start postgresql   # 备库自动进入流复制

# 主库查看复制状态
psql -c "SELECT client_addr, state, sync_state, replay_lag FROM pg_stat_replication;"

; /etc/pgbouncer/pgbouncer.ini
[databases]
sales_db = host=127.0.0.1 port=5432 dbname=sales_db

[pgbouncer]
listen_port = 6432
pool_mode = transaction      ; transaction 模式并发能力最强
max_client_conn = 1000
default_pool_size = 20

import psycopg2
from psycopg2 import sql

conn = None
try:
    conn = psycopg2.connect(
        dbname='sales_db',
        user='app',
        password='StrongPass!2026',
        host='localhost',
        port=5432,
    )

    with conn.cursor() as cursor:
        # 查询：%s 是占位符（不要自己拼字符串）
        cursor.execute(
            "SELECT id, name, salary FROM employees WHERE department = %s",
            ('研发部',),
        )
        for emp_id, name, salary in cursor.fetchall():
            print(f"{emp_id}\t{name}\t{salary}")

        # 插入并用 RETURNING 拿回自增 id
        cursor.execute(
            """INSERT INTO employees (name, email, department, salary)
               VALUES (%s, %s, %s, %s) RETURNING id""",
            ('王芳', 'wang@company.com', '市场部', 13000),
        )
        new_id = cursor.fetchone()[0]
        print(f"新员工 ID: {new_id}")

    conn.commit()

except Exception as e:
    print(f"数据库错误: {e}")
    if conn:
        conn.rollback()

finally:
    if conn:
        conn.close()