PostgreSQL 详解：高级关系数据库的特性与实践

-- 列出所有数据库
\l

-- 切换数据库
\c demodb

-- 列出当前数据库中的所有表
\dt

-- 查看表结构
\d+ table_name

-- 列出所有索引
\di

-- 查看当前连接信息
\conninfo

-- 查看所有可用的 schema
\dn

-- 查看函数列表
\df

-- 查看视图
\dv

-- 查看执行计划
EXPLAIN ANALYZE SELECT * FROM users WHERE id = 1;

-- 计时开关
\timing on

-- 查询历史
\s

-- 导入 SQL 文件
\i /path/to/file.sql

-- 输出结果到文件
\o /path/to/output.txt

-- 退出
\q

-- 创建包含数组列的表
CREATE TABLE articles_with_tags (
    id SERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    tags TEXT[]           -- 文本数组
);

-- 插入数组数据
INSERT INTO articles_with_tags (title, tags) VALUES
    ('PostgreSQL 入门', ARRAY['数据库', '教程', '新手']),
    ('高级 SQL 技巧', ARRAY['SQL', '高级', '优化']),
    ('JSON 与 NoSQL', ARRAY['JSON', 'NoSQL', 'PostgreSQL']);

-- 或者使用花括号语法
INSERT INTO articles_with_tags (title, tags) VALUES
    ('索引优化实战', '{索引,优化,性能}');

-- 查询
SELECT * FROM articles_with_tags;

-- 数组包含（@>）
SELECT * FROM articles_with_tags
WHERE tags @> ARRAY['SQL'];

-- 数组重叠（&&）
SELECT * FROM articles_with_tags
WHERE tags && ARRAY['数据库', 'NoSQL'];

-- 数组长度
SELECT title, array_length(tags, 1) AS tag_count FROM articles_with_tags;

-- 解包数组为行
SELECT title, unnest(tags) AS tag FROM articles_with_tags;

-- 数组索引访问（PostgreSQL 数组从 1 开始）
SELECT title, tags[1] AS first_tag FROM articles_with_tags;

-- 数组追加
UPDATE articles_with_tags
SET tags = array_append(tags, 'PostgreSQL')
WHERE id = 1;

-- 数组去重
SELECT title, array_distinct(tags) FROM articles_with_tags;

-- 多维数组
CREATE TABLE matrix_data (
    id SERIAL PRIMARY KEY,
    data INTEGER[][]
);

INSERT INTO matrix_data (data) VALUES
    (ARRAY[[1,2,3],[4,5,6],[7,8,9]]);

SELECT data[1:2][1:2] AS sub_matrix FROM matrix_data WHERE id = 1;
-- 结果: {{1,2},{4,5}}

-- 创建测试表
CREATE TABLE products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    attributes JSONB,
    metadata JSON
);

-- 插入 JSONB 数据
INSERT INTO products (name, attributes, metadata) VALUES
    ('笔记本电脑', '{"brand": "Lenovo", "specs": {"cpu": "i7-13700H", "ram": "32GB", "storage": "1TB SSD"}, "price": 8999, "in_stock": true}', '{"created_by": "admin", "notes": "热门商品"}'),
    ('机械键盘', '{"brand": "Filco", "specs": {"layout": "87键", "switch": "Cherry MX 青轴"}, "price": 1499, "in_stock": false}', '{"created_by": "manager", "notes": "缺货中"}'),
    ('显示器', '{"brand": "Dell", "specs": {"size": "27寸", "resolution": "4K", "panel": "IPS"}, "price": 3999, "in_stock": true}', '{"created_by": "admin"}');

-- JSONB 查询操作符

-- 1. -> 返回 JSON 对象字段（保留 JSON 类型）
SELECT attributes -> 'brand' AS brand FROM products;

-- 2. ->> 返回文本
SELECT attributes ->> 'brand' AS brand FROM products;

-- 3. 嵌套路径查询
SELECT
    name,
    attributes -> 'specs' ->> 'cpu' AS cpu,
    attributes -> 'specs' ->> 'ram' AS ram
FROM products
WHERE attributes @> '{"brand": "Lenovo"}'::jsonb;

-- 4. ? 检查存在性
SELECT name FROM products WHERE attributes ? 'warranty';

-- 5. ?| 任意一个存在
SELECT name FROM products WHERE attributes ?| ARRAY['warranty', 'price'];

-- 6. ?& 全部存在
SELECT name FROM products WHERE attributes ?& ARRAY['brand', 'specs', 'price'];

-- 7. @> 包含检查（最常用）
SELECT name FROM products
WHERE attributes @> '{"specs": {"ram": "32GB"}}'::jsonb;

-- 8. <@ 被包含
SELECT name FROM products
WHERE '{"brand": "Lenovo", "price": 8999}'::jsonb <@ attributes;

-- 9. 存在路径检查
SELECT name FROM products
WHERE attributes #>> '{specs, cpu}' = 'i7-13700H';

-- 10. JSONB 聚合
SELECT
    jsonb_object_agg(brand, count) AS brand_counts
FROM (
    SELECT attributes ->> 'brand' AS brand, count(*) AS count
    FROM products
    GROUP BY attributes ->> 'brand'
) t;

-- JSONB 修改与更新

-- 设置新字段
UPDATE products
SET attributes = jsonb_set(attributes, '{warranty}', '"2年"')
WHERE name = '笔记本电脑';

-- 删除字段
UPDATE products
SET attributes = attributes - 'price';

-- 合并两个 JSONB
UPDATE products
SET attributes = attributes || '{"discount": 0.1}'::jsonb;

-- JSONB 索引

-- 默认 GIN 索引（支持 @>、?、?|、?&）
CREATE INDEX idx_products_attributes ON products USING GIN (attributes);

-- 针对特定路径的 B-Tree 索引
CREATE INDEX idx_products_brand ON products USING BTREE ((attributes ->> 'brand'));

-- 常用范围类型
-- int4range: 整数范围
-- int8range: 大整数范围
-- numrange: 数值范围
-- daterange: 日期范围
-- tsrange: 时间戳范围
-- tstzrange: 带时区的时间戳范围

CREATE TABLE event_bookings (
    id SERIAL PRIMARY KEY,
    room TEXT NOT NULL,
    booking_period TSTZRANGE,     -- 带时区的预约时间段
    price_range NUMRANGE,          -- 价格区间
    guest_count INT4RANGE,         -- 客人数量范围
    date_range DATERANGE           -- 日期范围
);

-- 插入范围数据
INSERT INTO event_bookings (room, booking_period, price_range, guest_count, date_range) VALUES
    ('会议室A', '[2024-09-15 09:00+08, 2024-09-15 12:00+08)', '[500, 1500]', '[10, 20]', '[2024-09-15, 2024-09-16)'),
    ('宴会厅', '[2024-10-01 18:00+08, 2024-10-01 22:00+08)', '[5000, 20000]', '[50, 200]', '[2024-10-01, 2024-10-02)'),
    ('培训室', '[2024-09-20 14:00+08, 2024-09-21 18:00+08)', '[2000, 5000]', '[20, 50]', '[2024-09-20, 2024-09-22)');

-- 范围操作符

-- @> 包含
SELECT * FROM event_bookings
WHERE booking_period @> '2024-09-15 10:00+08'::timestamptz;

-- && 重叠
SELECT * FROM event_bookings
WHERE booking_period && '[2024-09-14 00:00+08, 2024-09-16 00:00+08)'::tstzrange;

-- << 严格在左（不重叠且在前）
SELECT * FROM event_bookings
WHERE booking_period << '[2024-10-01 00:00+08, 2024-11-01 00:00+08)'::tstzrange;

-- >> 严格在右
SELECT * FROM event_bookings
WHERE booking_period >> '[2024-08-01 00:00+08, 2024-09-01 00:00+08)'::tstzrange;

-- -|- 相邻
SELECT * FROM event_bookings
WHERE booking_period -|- '[2024-09-15 12:00+08, 2024-09-15 18:00+08)'::tstzrange;

-- 范围函数

-- 获取范围下界
SELECT room, lower(booking_period) AS start_time FROM event_bookings;

-- 获取范围上界
SELECT room, upper(booking_period) AS end_time FROM event_bookings;

-- 检查范围是否为空
SELECT room, isempty(price_range) AS no_price_info FROM event_bookings;

-- 范围长度
SELECT room, upper(price_range) - lower(price_range) AS price_span FROM event_bookings;

-- 合并重叠范围
SELECT int4range(1, 5) + int4range(3, 8) AS union_range;     -- [1,8)
SELECT int4range(1, 5) * int4range(3, 8) AS intersect_range; -- [3,5)

-- 使用 GiST 索引加速范围查询
CREATE INDEX idx_booking_period ON event_bookings USING GiST (booking_period);

-- 创建复合类型
CREATE TYPE address AS (
    street TEXT,
    city TEXT,
    state TEXT,
    zip_code TEXT,
    country TEXT DEFAULT '中国'
);

-- 创建包含复合类型的表
CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    home_address address,
    work_address address
);

-- 插入复合类型数据
INSERT INTO customers (name, home_address, work_address) VALUES
    ('张三',
     ROW('长安街1号', '北京', '北京', '100000'),
     ROW('科技园路100号', '北京', '北京', '100080'));

-- 访问复合类型字段
SELECT
    name,
    (home_address).street,
    (home_address).city,
    (work_address).zip_code
FROM customers;

-- 支持索引复合类型字段
CREATE INDEX idx_customers_city ON customers (((home_address).city));

-- 创建域（Domain）——基于已有类型的约束
CREATE DOMAIN email_address AS TEXT
    CHECK (VALUE ~ '^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$');

CREATE TABLE contacts (
    id SERIAL PRIMARY KEY,
    email email_address
);

-- 合法
INSERT INTO contacts (email) VALUES ('user@example.com');
-- 非法，会报错
-- INSERT INTO contacts (email) VALUES ('not-an-email');

-- 准备示例数据
CREATE TABLE sales_data (
    id SERIAL PRIMARY KEY,
    product TEXT NOT NULL,
    category TEXT NOT NULL,
    sales_amount NUMERIC(10,2),
    sale_date DATE
);

INSERT INTO sales_data (product, category, sales_amount, sale_date) VALUES
    ('笔记本电脑', '电子产品', 15000, '2024-01-15'),
    ('手机', '电子产品', 12000, '2024-01-16'),
    ('T恤', '服装', 3000,  '2024-01-15'),
    ('牛仔裤', '服装', 4500,  '2024-01-17'),
    ('笔记本电脑', '电子产品', 18000, '2024-02-15'),
    ('手机', '电子产品', 11000, '2024-02-18'),
    ('T恤', '服装', 3500,  '2024-02-20'),
    ('牛仔裤', '服装', 5000,  '2024-02-22'),
    ('笔记本电脑', '电子产品', 20000, '2024-03-15'),
    ('手机', '电子产品', 13000, '2024-03-16'),
    ('T恤', '服装', 2800,  '2024-03-18'),
    ('牛仔裤', '服装', 4800,  '2024-03-20');

-- ROW_NUMBER：为每一行分配唯一编号
SELECT
    product,
    category,
    sales_amount,
    sale_date,
    ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales_amount DESC) AS rank_by_category
FROM sales_data;

-- RANK：排名（相同值并列，跳过后续排名）
SELECT
    product,
    category,
    sales_amount,
    RANK() OVER (PARTITION BY category ORDER BY sales_amount DESC) AS rank
FROM sales_data;

-- DENSE_RANK：密集排名（相同值并列，不跳过排名）
SELECT
    product,
    category,
    sales_amount,
    DENSE_RANK() OVER (PARTITION BY category ORDER BY sales_amount DESC) AS dense_rank
FROM sales_data;

-- LAG：访问前一行的值
SELECT
    product,
    category,
    sales_amount,
    sale_date,
    LAG(sales_amount, 1, 0) OVER (PARTITION BY product ORDER BY sale_date) AS prev_month_sales,
    sales_amount - LAG(sales_amount, 1, 0) OVER (PARTITION BY product ORDER BY sale_date) AS month_growth
FROM sales_data;

-- LEAD：访问后一行的值
SELECT
    product,
    category,
    sales_amount,
    sale_date,
    LEAD(sales_amount, 1) OVER (PARTITION BY product ORDER BY sale_date) AS next_month_sales
FROM sales_data;

-- 移动平均（3个月滑动窗口）
SELECT
    product,
    sale_date,
    sales_amount,
    AVG(sales_amount) OVER (
        PARTITION BY product
        ORDER BY sale_date
        ROWS BETWEEN 2 PRECEDING AND CURRENT ROW
    ) AS moving_avg_3m
FROM sales_data;

-- FIRST_VALUE / LAST_VALUE
SELECT
    product,
    category,
    sales_amount,
    FIRST_VALUE(sales_amount) OVER (
        PARTITION BY category ORDER BY sale_date
    ) AS first_sale_amount,
    LAST_VALUE(sales_amount) OVER (
        PARTITION BY category ORDER BY sale_date
        RANGE BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
    ) AS last_sale_amount
FROM sales_data;

-- NTILE：分桶（例如分为4个四分位）
SELECT
    product,
    sales_amount,
    NTILE(4) OVER (ORDER BY sales_amount) AS quartile
FROM sales_data;

-- 聚合函数+窗口
SELECT
    product,
    category,
    sales_amount,
    SUM(sales_amount) OVER (PARTITION BY category) AS category_total,
    ROUND(sales_amount * 100.0 / SUM(sales_amount) OVER (PARTITION BY category), 2) AS pct_of_category
FROM sales_data;

-- CUME_DIST：累积分布
SELECT
    sales_amount,
    ROUND(CUME_DIST() OVER (ORDER BY sales_amount)::numeric, 4) AS cum_dist
FROM (SELECT DISTINCT sales_amount FROM sales_data) t
ORDER BY sales_amount;

-- 递归 CTE：树形结构查询

CREATE TABLE employees (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    manager_id INTEGER REFERENCES employees(id)
);

INSERT INTO employees (id, name, manager_id) VALUES
    (1, 'CEO', NULL),
    (2, 'CTO', 1),
    (3, 'CFO', 1),
    (4, '技术总监', 2),
    (5, '财务主任', 3),
    (6, '高级工程师', 4),
    (7, '初级工程师', 6);

-- 递归 CTE：查找所有下属
WITH RECURSIVE org_tree AS (
    -- 基础案例：根节点
    SELECT id, name, manager_id, 0 AS level, ARRAY[id] AS path
    FROM employees
    WHERE manager_id IS NULL

    UNION ALL

    -- 递归步骤：连接下属
    SELECT
        e.id,
        e.name,
        e.manager_id,
        ot.level + 1,
        ot.path || e.id
    FROM employees e
    INNER JOIN org_tree ot ON e.manager_id = ot.id
)
SELECT
    level,
    REPEAT('  ', level) || name AS org_chart,
    name,
    path
FROM org_tree
ORDER BY path;

-- 递归 CTE：查询所有上级（反向）
WITH RECURSIVE report_chain AS (
    SELECT id, name, manager_id, ARRAY[name] AS chain
    FROM employees
    WHERE name = '初级工程师'

    UNION ALL

    SELECT
        e.id,
        e.name,
        e.manager_id,
        rc.chain || e.name
    FROM employees e
    INNER JOIN report_chain rc ON e.id = rc.manager_id
)
SELECT chain FROM report_chain
WHERE manager_id IS NULL;

-- 递归 CTE：生成序列
WITH RECURSIVE number_series AS (
    SELECT 1 AS n
    UNION ALL
    SELECT n + 1 FROM number_series WHERE n < 10
)
SELECT * FROM number_series;

-- 递归 CTE：生成日历
WITH RECURSIVE calendar AS (
    SELECT '2024-01-01'::DATE AS date
    UNION ALL
    SELECT date + 1 FROM calendar WHERE date < '2024-01-31'
)
SELECT
    date,
    EXTRACT(DOW FROM date) AS day_of_week,
    TO_CHAR(date, 'Day') AS day_name
FROM calendar;

-- 场景：为每个分类找出销售额最高的前2个产品

-- 不使用 LATERAL（需要窗口函数+子查询）
WITH ranked AS (
    SELECT *,
        ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales_amount DESC) AS rn
    FROM sales_data
)
SELECT * FROM ranked WHERE rn <= 2;

-- 使用 LATERAL
SELECT DISTINCT s.category, top_products.*
FROM (SELECT DISTINCT category FROM sales_data) cat
CROSS JOIN LATERAL (
    SELECT product, sales_amount, sale_date
    FROM sales_data s
    WHERE s.category = cat.category
    ORDER BY s.sales_amount DESC
    LIMIT 2
) top_products;

-- LATERAL 结合 LEFT JOIN
SELECT * FROM categories c
LEFT JOIN LATERAL (
    SELECT * FROM products p
    WHERE p.category_id = c.id
    ORDER BY p.price DESC
    LIMIT 5
) top5 ON true;

-- 实战场景：最近交易
CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name TEXT
);

CREATE TABLE transactions (
    id SERIAL PRIMARY KEY,
    customer_id INTEGER REFERENCES customers(id),
    amount NUMERIC(10,2),
    created_at TIMESTAMPTZ DEFAULT now()
);

INSERT INTO customers (name) VALUES ('Alice'), ('Bob'), ('Charlie');
INSERT INTO transactions (customer_id, amount, created_at) VALUES
    (1, 100, '2024-01-01'),
    (1, 200, '2024-02-15'),
    (1, 150, '2024-03-10'),
    (2, 500, '2024-01-20'),
    (2, 300, '2024-03-01'),
    (3, 50, '2024-02-28');

-- 每个客户的最近3笔交易
SELECT
    c.name,
    recent.*
FROM customers c
LEFT JOIN LATERAL (
    SELECT amount, created_at
    FROM transactions t
    WHERE t.customer_id = c.id
    ORDER BY t.created_at DESC
    LIMIT 3
) recent ON true;

-- LATERAL 在子查询中
SELECT * FROM products
WHERE EXISTS (
    SELECT 1 FROM LATERAL (
        SELECT (attributes ->> 'price')::NUMERIC AS p
    ) price_info
    WHERE p > 100
);

-- 创建全文搜索表
CREATE TABLE documents (
    id SERIAL PRIMARY KEY,
    title TEXT NOT NULL,
    body TEXT NOT NULL,
    search_vector TSVECTOR       -- 预计算的 tsvector
);

-- 插入测试数据
INSERT INTO documents (title, body) VALUES
    ('PostgreSQL 基本概念', 'PostgreSQL 是一个强大的开源关系数据库系统，支持 ACID 事务和复杂查询'),
    ('全文搜索入门', '全文搜索允许用户快速在大量文本中找到相关文档，PostgreSQL 内置了强大的全文搜索功能'),
    ('索引优化技巧', '合理的索引设计可以大幅提升查询性能，PostgreSQL 支持多种索引类型包括 B-Tree 和 GIN'),
    ('高级 SQL 查询', '窗口函数、CTE 和 LATERAL JOIN 是 PostgreSQL 的高级查询特性');

-- 手动创建 tsvector
SELECT to_tsvector('english', 'PostgreSQL is a powerful database system');

-- 解析文本为词素
SELECT to_tsvector('english', 'running runs ran')::TEXT;
-- 结果: 'ran':4 'run':1,3

-- tsquery 查询
SELECT to_tsvector('english', 'PostgreSQL is a powerful database system') @@
       to_tsquery('english', 'powerful & database');

-- 更新 search_vector 列
UPDATE documents
SET search_vector = to_tsvector('simple', title || ' ' || body);

-- 或者使用自动更新触发器
CREATE FUNCTION documents_search_vector_update()
RETURNS TRIGGER AS $$
BEGIN
    NEW.search_vector := to_tsvector('simple', NEW.title || ' ' || NEW.body);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trg_documents_search_vector
    BEFORE INSERT OR UPDATE ON documents
    FOR EACH ROW
    EXECUTE FUNCTION documents_search_vector_update();

-- 现在插入时会自动计算 tsvector
INSERT INTO documents (title, body) VALUES
    ('PostgreSQL 与 NoSQL', 'PostgreSQL 的 JSONB 支持使其兼具关系数据库和 NoSQL 数据库的优势');

-- 基本全文搜索
SELECT id, title,
       ts_rank(search_vector, query) AS rank
FROM documents,
     to_tsquery('simple', 'PostgreSQL & 数据库') AS query
WHERE search_vector @@ query
ORDER BY rank DESC;

-- ts_rank 与 ts_rank_cd（覆盖密度排名）
SELECT id, title,
       ts_rank(search_vector, to_tsquery('simple', 'PostgreSQL')) AS rank_standard,
       ts_rank_cd(search_vector, to_tsquery('simple', 'PostgreSQL')) AS rank_coverage
FROM documents
WHERE search_vector @@ to_tsquery('simple', 'PostgreSQL');

-- 高亮搜索结果
SELECT id, title,
       ts_headline('simple', body, to_tsquery('simple', 'PostgreSQL & 数据库'),
                   'StartSel=<mark>, StopSel=</mark>, MaxWords=30, MinWords=15') AS snippet
FROM documents
WHERE search_vector @@ to_tsquery('simple', 'PostgreSQL & 数据库');

-- 短语搜索（<-> 操作符）
SELECT * FROM documents
WHERE search_vector @@ to_tsquery('simple', 'PostgreSQL <-> 查询');

-- 距离搜索（<N> 表示两个词之间的最大距离）
SELECT * FROM documents
WHERE search_vector @@ to_tsquery('simple', 'PostgreSQL <2> 数据库');

-- 权重分配（A=标题，B=正文）
ALTER TABLE documents ADD COLUMN search_vector_weighted TSVECTOR;

UPDATE documents SET search_vector_weighted =
    setweight(to_tsvector('simple', title), 'A') ||
    setweight(to_tsvector('simple', body), 'B');

-- 带权重的搜索排名
SELECT id, title,
       ts_rank(search_vector_weighted, query) AS rank
FROM documents,
     to_tsquery('simple', 'PostgreSQL') AS query
WHERE search_vector_weighted @@ query
ORDER BY rank DESC;

-- 准备工作
CREATE TABLE mvcc_demo2 (
    id SERIAL PRIMARY KEY,
    value TEXT NOT NULL,
    version INTEGER DEFAULT 1
);

-- 查看当前事务 ID
SELECT txid_current();

-- 事务隔离级别
SHOW transaction_isolation;  -- 默认: read committed

-- 行可见性系统列详解

-- 插入新行
INSERT INTO mvcc_demo2 (value) VALUES ('Initial');

-- 查看元组信息
SELECT
    ctid,                    -- 物理位置（页面:行偏移）
    xmin,                    -- 创建该行版本的事务 ID
    xmax,                    -- 删除/更新的事务 ID，0 表示可见
    cmin,                    -- 创建命令的序号（事务内）
    cmax,                    -- 删除命令的序号
    *
FROM mvcc_demo2;

-- 更新行（实际上创建了一个新版本）
BEGIN;
UPDATE mvcc_demo2 SET value = 'Updated', version = 2 WHERE id = 1;
-- 此时可以看到两个元组
SELECT ctid, xmin, xmax, * FROM mvcc_demo2;
COMMIT;

-- 事务隔离演示

-- 会话 A
BEGIN;
UPDATE mvcc_demo2 SET value = 'SessionA', version = 3 WHERE id = 1;
-- 不提交

-- 会话 B（READ COMMITTED 下）
BEGIN;
SELECT * FROM mvcc_demo2 WHERE id = 1;
-- 看到 'Updated'（旧版本，因为 A 未提交）

-- 会话 A 提交后
-- 会话 B 再次查询（READ COMMITTED 会重新获取快照）
SELECT * FROM mvcc_demo2 WHERE id = 1;
-- 看到 'SessionA'（A 已提交）

-- 清理死元组
VACUUM VERBOSE mvcc_demo2;

-- 查看和设置事务隔离级别

-- 读未提交（在 PostgreSQL 中行为等同于读已提交）
SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;

-- 读已提交（默认）
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;

-- 可重复读
SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;

-- 可序列化
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;

-- SSI 演示：检测序列化冲突
-- 会话 A
BEGIN ISOLATION LEVEL SERIALIZABLE;
UPDATE accounts SET balance = balance - 100 WHERE id = 1;

-- 会话 B
BEGIN ISOLATION LEVEL SERIALIZABLE;
UPDATE accounts SET balance = balance - 50 WHERE id = 1;

-- 其中一个会话提交时会收到:
-- ERROR:  could not serialize access due to read/write dependencies among transactions
-- DETAIL:  Reason code: Canceled on identification as a pivot, during commit attempt.
-- HINT:  The transaction might succeed if retried.

-- 查看序列化冲突统计
SELECT * FROM pg_stat_database WHERE datname = current_database();
-- serialization_failures 列显示序列化冲突次数

-- PostgreSQL 锁类型

-- 1. 表级锁
LOCK TABLE orders IN ACCESS SHARE MODE;        -- SELECT 获取
LOCK TABLE orders IN ROW SHARE MODE;           -- SELECT ... FOR UPDATE
LOCK TABLE orders IN EXCLUSIVE MODE;           -- DDL 操作
LOCK TABLE orders IN ACCESS EXCLUSIVE MODE;    -- DROP TABLE, VACUUM FULL

-- 2. 行级锁
BEGIN;
SELECT * FROM orders WHERE id = 1 FOR UPDATE;        -- 行级排他锁
SELECT * FROM orders WHERE id = 2 FOR NO KEY UPDATE; -- 较弱排他锁
SELECT * FROM orders WHERE id = 3 FOR SHARE;         -- 行级共享锁
SELECT * FROM orders WHERE id = 4 FOR KEY SHARE;     -- 最弱行级锁
COMMIT;

-- 3. 锁监控

-- 当前锁信息
SELECT
    l.locktype,
    l.relation::regclass AS relation,
    l.mode,
    l.granted,
    a.pid,
    a.state,
    a.query,
    a.wait_event_type,
    a.wait_event
FROM pg_locks l
JOIN pg_stat_activity a ON l.pid = a.pid
WHERE a.pid <> pg_backend_pid()
ORDER BY a.query_start;

-- 锁等待统计
SELECT
    relation::regclass AS table_name,
    mode,
    granted,
    pid,
    wait_event
FROM pg_locks
WHERE NOT granted;

-- 阻塞链查询
WITH blocking_info AS (
    SELECT
        a.pid AS blocked_pid,
        a.query AS blocked_query,
        l.pid AS blocking_pid,
        l.mode
    FROM pg_stat_activity a
    JOIN pg_locks l ON l.pid != a.pid
    WHERE a.wait_event_type = 'Lock'
)
SELECT * FROM blocking_info;

-- 死锁日志位置
SHOW log_directory;
SHOW log_filename;

-- 设置死锁超时
SHOW deadlock_timeout;  -- 默认 1s

-- 死锁演示

-- 准备数据
CREATE TABLE account_a (id INT PRIMARY KEY, balance NUMERIC(10,2));
CREATE TABLE account_b (id INT PRIMARY KEY, balance NUMERIC(10,2));
INSERT INTO account_a VALUES (1, 1000);
INSERT INTO account_b VALUES (1, 1000);

-- 会话 A
BEGIN;
UPDATE account_a SET balance = balance - 100 WHERE id = 1;
-- 然后等待会话 B

-- 会话 B
BEGIN;
UPDATE account_b SET balance = balance - 100 WHERE id = 1;
-- 然后等待会话 A

-- 会话 A 尝试获取 B 的锁
UPDATE account_b SET balance = balance + 100 WHERE id = 1;
-- 阻塞...

-- 会话 B 尝试获取 A 的锁
UPDATE account_a SET balance = balance + 100 WHERE id = 1;
-- 产生死锁，其中一个会话会被终止:
-- ERROR:  deadlock detected
-- DETAIL:  Process X waits for ShareLock on transaction Y; blocked by process Z.
-- Process Z waits for ShareLock on transaction Y; blocked by process X.
-- HINT:  See server log for query details.

-- 预防死锁：使用一致的锁顺序
-- 例如，总是先锁 A 再锁 B

-- 安装
CREATE EXTENSION IF NOT EXISTS postgis;
CREATE EXTENSION IF NOT EXISTS postgis_topology;

-- 验证安装
SELECT postgis_version();
SELECT postgis_full_version();

-- 创建地理空间表
CREATE TABLE poi_locations (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    location GEOGRAPHY(Point, 4326),    -- WGS84 经纬度
    geom GEOMETRY(Point, 3857)           -- Web Mercator 投影
);

-- 插入地理数据
INSERT INTO poi_locations (name, location, geom) VALUES
    ('天安门广场', ST_GeogFromText('SRID=4326;POINT(116.3972 39.9087)'),
                  ST_Transform(ST_GeomFromText('POINT(116.3972 39.9087)', 4326), 3857)),
    ('故宫博物院', ST_GeogFromText('SRID=4326;POINT(116.3970 39.9163)'),
                  ST_Transform(ST_GeomFromText('POINT(116.3970 39.9163)', 4326), 3857)),
    ('上海东方明珠', ST_GeogFromText('SRID=4326;POINT(121.4997 31.2397)'),
                    ST_Transform(ST_GeomFromText('POINT(121.4997 31.2397)', 4326), 3857));

-- 创建 GiST 空间索引
CREATE INDEX idx_poi_geog ON poi_locations USING GiST (location);
CREATE INDEX idx_poi_geom ON poi_locations USING GiST (geom);

-- 距离查询
SELECT
    name,
    ST_Distance(location, ST_GeogFromText('SRID=4326;POINT(116.4074 39.9042)')) AS distance_m
FROM poi_locations
ORDER BY location <-> ST_GeogFromText('SRID=4326;POINT(116.4074 39.9042)')
LIMIT 5;

-- 半径搜索（5公里内）
SELECT name,
       ST_Distance(location, ST_GeogFromText('SRID=4326;POINT(116.4074 39.9042)')) AS distance_m
FROM poi_locations
WHERE ST_DWithin(location, ST_GeogFromText('SRID=4326;POINT(116.4074 39.9042)'), 5000)
ORDER BY distance_m;

-- 几何计算
SELECT
    ST_Area(ST_GeomFromText('POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))')) AS area_sq_units,
    ST_Length(ST_GeomFromText('LINESTRING(0 0, 1 1, 2 0)')) AS length_units;

-- 空间连接
SELECT a.name AS point_a, b.name AS point_b,
       ST_Distance(a.location, b.location) AS distance
FROM poi_locations a
CROSS JOIN poi_locations b
WHERE a.id < b.id
ORDER BY distance;

-- 安装
CREATE EXTENSION IF NOT EXISTS pg_stat_statements;

-- 配置（在 postgresql.conf 中）
-- shared_preload_libraries = 'pg_stat_statements'
-- pg_stat_statements.max = 10000
-- pg_stat_statements.track = all

-- 重启后查看配置
SHOW shared_preload_libraries;

-- 查询性能统计
SELECT
    queryid,
    LEFT(query, 80) AS query_preview,
    calls,
    total_exec_time / calls AS avg_time_ms,
    min_exec_time,
    max_exec_time,
    rows,
    shared_blks_hit,
    shared_blks_read,
    shared_blks_dirtied,
    shared_blks_written,
    temp_blks_read,
    temp_blks_written,
    blk_read_time,
    blk_write_time
FROM pg_stat_statements
WHERE query NOT LIKE '%pg_stat%'
ORDER BY total_exec_time DESC
LIMIT 20;

-- 查看最频繁的查询
SELECT
    LEFT(query, 100) AS query,
    calls,
    ROUND(total_exec_time::NUMERIC, 2) AS total_ms,
    ROUND((total_exec_time / calls)::NUMERIC, 2) AS avg_ms
FROM pg_stat_statements
ORDER BY calls DESC
LIMIT 10;

-- 查看 I/O 最多的查询
SELECT
    LEFT(query, 100) AS query,
    calls,
    shared_blks_read,
    shared_blks_hit,
    ROUND(shared_blks_hit * 100.0 / NULLIF(shared_blks_hit + shared_blks_read, 0), 2) AS hit_ratio
FROM pg_stat_statements
WHERE shared_blks_read > 0
ORDER BY shared_blks_read DESC
LIMIT 10;

-- 重置统计信息
SELECT pg_stat_statements_reset();

-- 安装
CREATE EXTENSION IF NOT EXISTS pg_partman;
CREATE SCHEMA partman;

-- 创建分区表
CREATE TABLE measurement (
    log_time TIMESTAMPTZ NOT NULL,
    device_id INTEGER,
    value NUMERIC(10,2)
) PARTITION BY RANGE (log_time);

-- 使用 pg_partman 自动创建和管理分区
SELECT partman.create_parent(
    p_parent_table := 'public.measurement',
    p_control := 'log_time',
    p_type := 'native',
    p_interval := '1 day',       -- 每天一个分区
    p_premake := 30               -- 预创建未来 30 天
);

-- 查看分区信息
SELECT * FROM partman.part_config
WHERE parent_table = 'public.measurement';

-- 手动维护分区
-- 创建新分区
CALL partman.create_partitions('public.measurement', 7);

-- 删除旧分区（保留最近 90 天）
CALL partman.partition_data_proc(
    'public.measurement',
    p_interval := '1 day',
    p_batch := 1000,
    p_keep_table := 'f'
);

-- 设置自动维护
UPDATE partman.part_config
SET infinite_time_partitions = true,
    retention = '90 days',
    retention_keep_table = false
WHERE parent_table = 'public.measurement';

-- 创建模板表用于新分区的结构
CREATE TABLE measurement_template (LIKE measurement INCLUDING ALL);

-- 更新配置使用模板
UPDATE partman.part_config
SET template_table = 'public.measurement_template'
WHERE parent_table = 'public.measurement';

-- 监控复制状态（主库）
SELECT
    client_addr,
    state,                    -- streaming / catchup / startup
    sync_state,              -- sync / async / potential
    write_lag,
    flush_lag,
    replay_lag,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), write_lsn)) AS write_delay
FROM pg_stat_replication;

-- 监控复制状态（从库）
SELECT
    pg_is_in_recovery(),
    pg_last_wal_receive_lsn(),
    pg_last_wal_replay_lsn(),
    pg_last_xact_replay_timestamp(),
    GREATEST(
        0,
        EXTRACT(EPOCH FROM (now() - pg_last_xact_replay_timestamp()))
    ) AS replay_lag_seconds
FROM pg_stat_wal_receiver;

-- 同步复制配置
-- 在 postgresql.conf 中设置
-- synchronous_standby_names = 'FIRST 2 (*)'

-- 查看同步状态
SELECT application_name, sync_priority, sync_state FROM pg_stat_replication;

-- 级联复制（从库还可以有从库）
-- 分布式从库设置
-- primary_conninfo = 'host=中间层从库_ip port=5432 user=replicator password=replica_secret'

# patroni.yml 配置示例
scope: pg_cluster
namespace: /db/
name: pg-node1

restapi:
  listen: 0.0.0.0:8008
  connect_address: 192.168.1.10:8008

etcd:
  host: 192.168.1.100:2379

bootstrap:
  dcs:
    ttl: 30
    loop_wait: 10
    retry_timeout: 10
    maximum_lag_on_failover: 1048576
    postgresql:
      use_pg_rewind: true
      use_slots: true
      parameters:
        wal_level: replica
        hot_standby: "on"
        max_wal_senders: 10
        max_replication_slots: 10
        wal_log_hints: "on"

  initdb:
  - encoding: UTF8
  - data-checksums

  pg_hba:
  - host replication replicator 0.0.0.0/0 md5
  - host all all 0.0.0.0/0 md5

postgresql:
  listen: 0.0.0.0:5432
  connect_address: 192.168.1.10:5432
  data_dir: /data/postgresql/16/main
  bin_dir: /usr/lib/postgresql/16/bin
  authentication:
    replication:
      username: replicator
      password: replica_secret
    superuser:
      username: postgres
      password: super_secret
    rewind:
      username: rewind_user
      password: rewind_secret

tags:
  nofailover: false
  noloadbalance: false
  clonefrom: false
  replicatefrom: null

# pgpool.conf 关键配置

# 连接池设置
max_pool = 4                    # 每个子进程的连接池大小
num_init_children = 32          # 初始子进程数
child_life_time = 300           # 子进程最大生存时间
child_max_connections = 0       # 子进程处理最大连接数（0=不限）
connection_life_time = 0        # 连接缓存时间（0=不限）
client_idle_limit = 0           # 客户端空闲断开时间（0=不限）

# 后端数据库节点配置
backend_hostname0 = '192.168.1.10'
backend_port0 = 5432
backend_weight0 = 1             # 读权重
backend_data_directory0 = '/data/postgresql/16/main'
backend_flag0 = 'ALLOW_TO_FAILOVER'

backend_hostname1 = '192.168.1.20'
backend_port1 = 5432
backend_weight1 = 2             # 从库分配更多读流量
backend_data_directory1 = '/data/postgresql/16/main'
backend_flag1 = 'ALLOW_TO_FAILOVER'

# 复制模式
replication_mode = off
load_balance_mode = on          # 启用负载均衡
master_slave_mode = on          # 主从模式
master_slave_sub_mode = 'stream'  # 流复制模式

# 健康检查
health_check_period = 10        # 健康检查周期（秒）
health_check_timeout = 20       # 超时
health_check_user = 'pgpool'
health_check_password = 'pgpool_pass'
health_check_max_retries = 3
health_check_retry_delay = 1

# 自动故障转移
failover_command = '/etc/pgpool/failover.sh %H %P %R'
follow_master_command = '/etc/pgpool/follow_master.sh %H %P %R'

# 连接池
connection_cache = on