[PostgreSQL]从MySQL迁移PostgreSQL：算法团队踩过的7个坑

-- MySQL 8.0
CREATE TABLE user_features (
    user_id BIGINT PRIMARY KEY,
    feature_json JSON
);

INSERT INTO user_features VALUES 
(1, '{"embedding": [0.1234567890123456, 0.9876543210987654], "score": 12345678901234567890}');

-- PostgreSQL 15.3（使用pgloader默认迁移）
SELECT user_id, feature_json->>'embedding' as embedding_str,
       (feature_json->>'score')::BIGINT as score
FROM user_features 
WHERE user_id = 1;

-- 结果：
-- user_id | embedding_str                                    | score
-- ---------+--------------------------------------------------+-------
--       1 | [0.1234567890123456, 0.9876543210987654]        | 12345678901234567168  -- 注意最后几位变了！

#!/usr/bin/env python3
# audit_json_precision.py - 审计JSON字段数值精度
import mysql.connector
import json
import sys

def check_precision_loss(table, json_column):
    conn = mysql.connector.connect(
        host='mysql-prod.example.com',
        user='migration_user',
        password='secret',
        database='algorithm_db'
    )
    cursor = conn.cursor(dictionary=True)
    
    cursor.execute(f"SELECT {json_column} FROM {table} LIMIT 10000")
    rows = cursor.fetchall()
    
    precision_issues = []
    for row in rows:
        try:
            data = json.loads(row[json_column])
            # 检查数组中的浮点数
            if 'embedding' in data:
                for val in data['embedding']:
                    if abs(val) < 1e-10:  # 接近0的值
                        continue
                    # 检查双精度表示是否精确
                    if abs(val - float(str(val))) > 1e-15:
                        precision_issues.append({
                            'type': 'float_precision',
                            'original': val,
                            'float_repr': float(str(val))
                        })
        except Exception as e:
            print(f"解析失败: {e}", file=sys.stderr)
    
    return precision_issues

# 在迁移前执行
issues = check_precision_loss('user_features', 'feature_json')
print(f"发现{len(issues)}个精度问题")

-- 方案A：对高精度需求字段使用JSON（文本）而非JSONB
CREATE TABLE user_features (
    user_id BIGINT PRIMARY KEY,
    feature_json JSON,  -- 保留文本形式，避免主动解析
    feature_jsonb JSONB  -- 用于查询加速的冗余字段
);

-- 方案B：迁移脚本中主动转换类型
-- 使用自定义迁移脚本而非pgloader默认配置
COPY (
    SELECT 
        user_id,
        feature_json::JSONB,  -- 默认转换
        -- 对高精度字段特殊处理
        (feature_json->>'embedding')::TEXT as embedding_str,  -- 保留原始字符串
        (feature_json->>'score')::NUMERIC(30,0) as score_numeric  -- 指定精度
    FROM mysql_user_features
) TO '/tmp/features_converted.csv' WITH (FORMAT csv);

-- 方案C：应用层适配（推荐）
-- 在PostgreSQL中存储为TEXT，应用层按需解析
ALTER TABLE user_features 
ALTER COLUMN feature_json TYPE TEXT;

-- 创建透明访问函数
CREATE OR REPLACE FUNCTION get_feature_embedding(p_user_id BIGINT)
RETURNS FLOAT8[] AS $$
DECLARE
    raw_json TEXT;
    embedding_str TEXT;
BEGIN
    SELECT feature_json INTO raw_json FROM user_features WHERE user_id = p_user_id;
    embedding_str := (raw_json::JSONB->>'embedding');
    -- 应用层保证精度
    RETURN STRING_TO_ARRAY(TRANSLATE(embedding_str, '[]', ''), ',')::FLOAT8[];
END;
$$ LANGUAGE plpgsql PARALLEL SAFE;

# Python应用层精度保护
from decimal import Decimal
import json

def safe_json_deserialize(json_text: str) -> dict:
    """
    对PostgreSQL返回的JSON文本进行安全解析
    保持高精度数值为Decimal类型
    """
    return json.loads(
        json_text, 
        parse_float=lambda x: Decimal(x)  # 避免转换为float
    )

# 使用示例
embedding = safe_json_deserialize(user_record['feature_json'])['embedding']
# 计算时使用Decimal保持精度，最终转换为numpy.float32前处理

-- 迁移前执行的全库JSON审计脚本
DO $$
DECLARE
    rec RECORD;
    json_tables CURSOR FOR
        SELECT table_schema, table_name, column_name, data_type
        FROM information_schema.columns
        WHERE data_type IN ('json', 'jsonb');
BEGIN
    FOR rec IN json_tables LOOP
        RAISE NOTICE '审计表: %.%', rec.table_schema, rec.table_name;
        EXECUTE format(
            'SELECT COUNT(*) FROM %I.%I WHERE %I::TEXT ~ ''^[0-9]{19,}$''',
            rec.table_schema, rec.table_name, rec.column_name
        );
    END LOOP;
END $$;

-- MySQL
CREATE TABLE model_predictions (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    model_id INT,
    prediction_result JSON,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO model_predictions (model_id) VALUES (101);  -- id=1
INSERT INTO model_predictions (model_id) VALUES (102);  -- id=2
DELETE FROM model_predictions WHERE id=2;
INSERT INTO model_predictions (model_id) VALUES (103);  -- id=3（MySQL会复用空洞吗？NO！）

-- 使用pgloader默认迁移
CREATE TABLE model_predictions (
    id BIGSERIAL PRIMARY KEY,  -- 实际创建sequence
    model_id INT,
    prediction_result JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 批量插入时
INSERT INTO model_predictions (model_id) 
SELECT generate_series(1, 10000);  -- sequence当前值=10001

-- 数据同步插入
INSERT INTO model_predictions (id, model_id) VALUES (10001, 200);  -- 手动指定ID

-- 业务写入
INSERT INTO model_predictions (model_id) VALUES (201);  -- sequence nextval=10002？NO！还是从1开始！
-- ERROR: duplicate key value violates unique constraint "model_predictions_pkey"

-- CREATE TABLE ... (id SERIAL) 实际执行：
CREATE SEQUENCE model_predictions_id_seq;
CREATE TABLE model_predictions (
    id BIGINT DEFAULT nextval('model_predictions_id_seq') PRIMARY KEY,
    ...
);

-- 迁移完成后，对所有SERIAL字段执行
DO $$
DECLARE
    rec RECORD;
    max_id BIGINT;
BEGIN
    FOR rec IN 
        SELECT 
            t.relname as table_name,
            a.attname as column_name,
            s.relname as seq_name
        FROM pg_class t
        JOIN pg_attribute a ON a.attrelid = t.oid
        JOIN pg_depend d ON d.refobjid = t.oid AND d.refobjsubid = a.attnum
        JOIN pg_class s ON s.oid = d.objid
        WHERE t.relkind = 'r'
          AND a.atttypid = 'int8'::regtype
          AND d.classid = 'pg_class'::regclass
          AND d.refclassid = 'pg_class'::regclass
    LOOP
        -- 获取表中最大ID
        EXECUTE format('SELECT COALESCE(MAX(%I), 0) FROM %I', 
                      rec.column_name, rec.table_name) INTO max_id;
        
        -- 设置Sequence为max(id)+1
        EXECUTE format('SELECT setval(%L, %s)', 
                      rec.seq_name, max_id + 1);
        
        RAISE NOTICE '表 % 的Sequence % 已校准为 %', 
                    rec.table_name, rec.seq_name, max_id + 1;
    END LOOP;
END $$;

#!/usr/bin/env python3
# sync_sequence_after_load.py
import psycopg2
import sys

def sync_sequence(conn, table_name, id_column='id'):
    """同步Sequence当前值到表中max(id)"""
    with conn.cursor() as cur:
        # 获取Sequence名称
        cur.execute("""
            SELECT pg_get_serial_sequence(%s, %s)
        """, (table_name, id_column))
        seq_name = cur.fetchone()[0]
        
        if not seq_name:
            print(f"表 {table_name} 没有SERIAL字段")
            return
        
        # 获取max(id)
        cur.execute(f"SELECT COALESCE(MAX({id_column}), 0) FROM {table_name}")
        max_id = cur.fetchone()[0]
        
        # 校准Sequence
        cur.execute(f"SELECT setval(%s, %s)", (seq_name, max_id + 1))
        
        # 验证
        cur.execute(f"SELECT last_value FROM {seq_name}")
        current_val = cur.fetchone()[0]
        
        print(f"{table_name}: max(id)={max_id}, Sequence已设置为{current_val}")

def main():
    # 连接PostgreSQL
    conn = psycopg2.connect(
        host="pg-prod.example.com",
        database="algorithm_db",
        user="migration_user",
        password="secret"
    )
    
    # 需要校准的表列表
    tables = ['model_predictions', 'user_features', 'experiment_results']
    
    for table in tables:
        sync_sequence(conn, table)
    
    conn.commit()
    conn.close()

if __name__ == '__main__':
    main()

-- 在数据导入期间，临时禁用Sequence
BEGIN;

ALTER TABLE model_predictions ALTER COLUMN id DROP DEFAULT;

-- 执行COPY或批量导入
COPY model_predictions (id, model_id, ...) FROM '/tmp/data.csv';

-- 手动设置Sequence
SELECT setval('model_predictions_id_seq', (SELECT MAX(id) FROM model_predictions));

-- 恢复DEFAULT
ALTER TABLE model_predictions ALTER COLUMN id SET DEFAULT nextval('model_predictions_id_seq');

COMMIT;

# 在业务代码中增加Sequence占位符
def insert_prediction(model_id, result):
    """
    双写兼容逻辑：同时写入MySQL和PostgreSQL
    """
    # MySQL侧：依赖AUTO_INCREMENT
    mysql_cursor.execute(
        "INSERT INTO model_predictions (model_id, result) VALUES (%s, %s)",
        (model_id, result)
    )
    generated_id = mysql_cursor.lastrowid
    
    # PostgreSQL侧：手动指定ID
    postgres_cursor.execute(
        "INSERT INTO model_predictions (id, model_id, result) VALUES (%s, %s, %s)",
        (generated_id, model_id, result)
    )
    
    # 确保Sequence不会冲突
    postgres_cursor.execute(
        "SELECT setval('model_predictions_id_seq', GREATEST(currval('model_predictions_id_seq'), %s))",
        (generated_id,)
    )
    
    mysql_conn.commit()
    postgres_conn.commit()

-- MySQL 8.0 (sql_mode默认宽松)
SELECT 
    user_id,
    city,  -- 不在GROUP BY中，也不在聚合函数中！
    COUNT(*) as order_cnt,
    MAX(amount) as max_amount
FROM orders
GROUP BY user_id;  -- 但MySQL会"随机"返回city值

-- MySQL会返回每个user_id的"某个"city值，通常是分组内的第一条记录

-- PostgreSQL 15.3
SELECT 
    user_id,
    city,  -- ERROR: column "orders.city" must appear in the GROUP BY clause...
    COUNT(*) as order_cnt,
    MAX(amount) as max_amount
FROM orders
GROUP BY user_id;

-- 错误信息：
-- ERROR: column "orders.city" must appear in the GROUP BY clause or be used in an aggregate function
-- LINE 3: city,

-- MySQL原始查询
SELECT user_id, ANY_VALUE(city) as city, COUNT(*)
FROM orders GROUP BY user_id;

-- PostgreSQL等效改写（保留所有city）
SELECT 
    user_id,
    STRING_AGG(DISTINCT city, ',') as cities,  -- 聚合为字符串
    COUNT(*)
FROM orders
GROUP BY user_id;

-- 或使用ARRAY_AGG保留数组形式
SELECT 
    user_id,
    ARRAY_AGG(DISTINCT city) as city_array,
    COUNT(*)
FROM orders
GROUP BY user_id;

-- 原始意图：获取每个user_id的最新city
-- MySQL（依赖隐式排序）
SELECT user_id, city, order_cnt
FROM (
    SELECT user_id, city, COUNT(*) as order_cnt
    FROM orders
    GROUP BY user_id, city
) t
GROUP BY user_id;  -- city不确定

-- PostgreSQL（显式最新值）
WITH user_stats AS (
    SELECT 
        user_id,
        city,
        COUNT(*) as order_cnt,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY MAX(created_at) DESC) as rn
    FROM orders
    GROUP BY user_id, city
)
SELECT user_id, city, order_cnt
FROM user_stats
WHERE rn = 1;  -- 明确选择最新的city

-- 自定义"取第一个值"的聚合函数
CREATE OR REPLACE FUNCTION first_agg(anyelement, anyelement)
RETURNS anyelement
LANGUAGE SQL
IMMUTABLE PARALLEL SAFE
AS $$ SELECT $1 $$;

CREATE AGGREGATE FIRST(anyelement) (
    SFUNC = first_agg,
    STYPE = anyelement,
    INITCOND = ''
);

-- 使用
SELECT 
    user_id,
    FIRST(city ORDER BY created_at) as first_city,  -- 明确排序规则
    COUNT(*)
FROM orders
GROUP BY user_id;

#!/usr/bin/env python3
# mysql_groupby_fixer.py - 自动修复GROUP BY问题
import re
import sqlparse

def fix_group_by(sql):
    """
    自动将宽松的GROUP BY改写为PostgreSQL兼容形式
    """
    parsed = sqlparse.parse(sql)[0]
    
    # 提取SELECT字段
    select_clause = []
    group_by_clause = []
    
    for token in parsed.tokens:
        if token.ttype is sqlparse.tokens.Keyword and token.value.upper() == 'SELECT':
            # 解析后续字段
            for sub_token in token.parent.tokens:
                if sub_token.ttype is sqlparse.tokens.Name:
                    select_clause.append(sub_token.value)
    
    # 自动识别非聚合字段并添加FIRST()聚合
    fixed_columns = []
    for col in select_clause:
        if '(' not in col:  # 简单字段
            fixed_columns.append(f"FIRST({col}) as {col}")
        else:
            fixed_columns.append(col)
    
    # 替换SELECT部分
    fixed_sql = re.sub(
        r'SELECT(.*?)FROM',
        f"SELECT {', '.join(fixed_columns)} FROM",
        sql,
        flags=re.IGNORECASE
    )
    
    return fixed_sql

# 批量处理
if __name__ == '__main__':
    with open('mysql_queries.sql', 'r') as f:
        for line in f:
            fixed = fix_group_by(line.strip())
            print(f"-- 原始: {line.strip()}")
            print(f"{fixed};")
            print()

-- 在PostgreSQL中创建兼容MySQL行为的视图（临时方案）
CREATE VIEW orders_v AS
SELECT 
    user_id,
    (ARRAY_AGG(city ORDER BY created_at DESC))[1] as city,  -- 模拟ANY_VALUE
    COUNT(*) as order_cnt,
    MAX(amount) as max_amount
FROM orders
GROUP BY user_id;

-- 业务代码逐步从orders表迁移到orders_v
-- 最终去除视图，修改业务SQL

# .gitlab-ci.yml
pg_groupby_check:
  stage: test
  script:
    - pip install sqlparse psycopg2
    - python3 tests/check_groupby_strictness.py  # 检查所有SQL文件
    - python3 tests/fix_groupby_issues.py       # 自动修复并提交PR
  only:
    - merge_requests

-- MySQL 8.0，默认utf8mb4_unicode_ci
SELECT * FROM users WHERE nickname LIKE '%用户%';
-- 返回: 用户123, 用户_456, 用户😊789（共150条）

-- 大小写不敏感
SELECT * FROM users WHERE nickname = 'USER';  -- 返回 user, USER, User

-- PostgreSQL 15.3，默认en_US.UTF-8
SELECT * FROM users WHERE nickname LIKE '%用户%';
-- 返回: 用户123, 用户_456（仅120条，emoji记录丢失！）

-- 大小写敏感
SELECT * FROM users WHERE nickname = 'USER';  -- 无结果！

-- 步骤1：创建GIN索引
CREATE EXTENSION pg_trgm;  -- 三元组匹配
CREATE EXTENSION unaccent;  -- 重音符号处理

-- 步骤2：创建文本搜索配置
CREATE TEXT SEARCH CONFIGURATION cn_unaccent (COPY = pg_catalog.simple);
ALTER TEXT SEARCH CONFIGURATION cn_unaccent
    ALTER MAPPING FOR hword, hword_part, word WITH unaccent, simple;

-- 步骤3：创建索引
CREATE INDEX idx_nickname_trgm ON users 
USING GIN (nickname gin_trgm_ops);

-- 步骤4：改写查询
-- MySQL: WHERE nickname LIKE '%用户%'
-- PostgreSQL:
SELECT * FROM users 
WHERE nickname % '用户'  -- 相似度操作符
   OR to_tsvector('cn_unaccent', nickname) @@ to_tsquery('用户');

-- 性能对比：
-- LIKE '%用户%': 800ms，全表扫描
-- %操作符: 15ms，GIN索引
-- @@操作符: 25ms，全文索引

-- 为所有文本字段指定COLLATE
ALTER TABLE users ALTER COLUMN nickname TYPE TEXT COLLATE "zh_CN.utf8";

-- 创建不区分大小写的索引
CREATE INDEX idx_nickname_ci ON users (nickname COLLATE "zh_CN.utf8");

-- 修改数据库默认排序规则
ALTER DATABASE algorithm_db SET LC_COLLATE = 'zh_CN.UTF-8';
ALTER DATABASE algorithm_db SET LC_CTYPE = 'zh_CN.UTF-8';

-- 注意：需重新创建数据库生效

# Python应用层处理大小写和emoji
import re
import hashlib

def normalize_query(query: str) -> str:
    """
    统一查询字符串格式：
    1. 转小写
    2. 去除emoji（可选，根据业务）
    3. 标准化空格
    """
    # 转小写
    query = query.lower()
    
    # 去除emoji（保留表情但不用于模糊匹配）
    emoji_pattern = re.compile(
        "[\U0001F600-\U0001F64F]|[\U0001F300-\U0001F5FF]|...",
        flags=re.UNICODE
    )
    query_clean = emoji_pattern.sub('', query)
    
    return query_clean.strip()

# 查询时
search_term = normalize_query('用户😊123')
# 生成两个查询：一个带emoji，一个不带
sql_with_emoji = "SELECT * FROM users WHERE nickname ILIKE %s"
sql_without = "SELECT * FROM users WHERE nickname ILIKE %s"

-- 创建函数索引实现大小写不敏感
CREATE INDEX idx_nickname_lower ON users (LOWER(nickname));

-- 改写查询
SELECT * FROM users WHERE LOWER(nickname) = LOWER('USER');  -- 使用索引

-- 或使用CITEXT扩展
CREATE EXTENSION citext;  -- 大小写不敏感文本
ALTER TABLE users ALTER COLUMN nickname TYPE CITEXT;

-- 现在可以直接使用，无需LOWER()
SELECT * FROM users WHERE nickname = 'USER';  -- CITEXT索引生效

-- 第1步：识别所有有问题的查询
SELECT query, calls, total_exec_time
FROM pg_stat_statements
WHERE query LIKE '%LIKE%' 
  AND query ILIKE '%nickname%'
ORDER BY total_exec_time DESC;

-- 第2步：创建必要的扩展和索引（在线操作）
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE EXTENSION IF NOT EXISTS citext;

-- 第3步：改写Top10慢查询
-- 使用pg_hint_plan强制索引（需安装扩展）
SELECT /*+ BitmapScan(users idx_nickname_trgm) */ *
FROM users WHERE nickname LIKE '%用户%';

-- MySQL 8.0: 45分钟完成
WITH user_sequence AS (
    SELECT 
        user_id,
        event_type,
        event_time,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time) as rn,
        LAG(event_type, 1) OVER (PARTITION BY user_id ORDER BY event_time) as prev_event
    FROM user_behavior
    WHERE event_time BETWEEN '2024-01-01' AND '2024-01-31'
)
SELECT 
    user_id,
    COUNT(*) as total_events,
    SUM(CASE WHEN event_type = prev_event THEN 1 ELSE 0 END) as repeat_event_cnt
FROM user_sequence
GROUP BY user_id;

-- PostgreSQL 15.3: 2小时32分钟
-- 相同SQL，资源使用率低（CPU 30%，I/O 40%）

-- MySQL EXPLAIN
/*
-> Window function: row_number() OVER (PARTITION BY user_id ORDER BY event_time)
   -> Index scan on user_behavior (user_id, event_time)  -- 覆盖索引
     Cost: 125000 rows, 45min
*/

-- PostgreSQL EXPLAIN
/*
WindowAgg  (cost=100000.00..500000.00 rows=125000 width=48)
  -> Sort  (cost=100000.00..102500.00 rows=1000000 width=40)
        Sort Key: user_id, event_time
        -> Seq Scan on user_behavior  (cost=0.00..25000.00 rows=1000000 width=40)
              Filter: (event_time BETWEEN '2024-01-01' AND '2024-01-31')
*/

-- 关键问题：
-- 1. 未使用索引（event_time条件在Filter而非Index Cond）
-- 2. 全表扫描后排序（Sort），内存不足时写磁盘
-- 3. 串行执行窗口函数

-- 步骤1：创建覆盖索引（Index Only Scan）
CREATE INDEX idx_behavior_window ON user_behavior 
(user_id, event_time) INCLUDE (event_type);

-- 步骤2：强制并行执行
SET max_parallel_workers_per_gather = 8;
SET work_mem = '512MB';  -- 为排序分配更多内存

-- 步骤3：改写查询以支持并行窗口
WITH user_sequence AS (
    SELECT 
        user_id,
        event_type,
        event_time,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time) as rn,
        LAG(event_type, 1) OVER (PARTITION BY user_id ORDER BY event_time) as prev_event
    FROM user_behavior
    WHERE event_time BETWEEN '2024-01-01' AND '2024-01-31'
)
SELECT 
    user_id,
    COUNT(*) as total_events,
    SUM(CASE WHEN event_type = prev_event THEN 1 ELSE 0 END) as repeat_event_cnt
FROM user_sequence
GROUP BY user_id;

-- 优化后执行计划：
-- WindowAgg  (cost=125000.00..200000.00 rows=125000 width=48)
--   -> Gather Merge  (cost=100000.00..125000.00 rows=125000 width=48)
--         Workers Planned: 8
--         ->  Sort  (cost=99000.00..100000.00 rows=15625 width=48)
--               Sort Key: user_id, event_time
--               ->  Parallel Index Only Scan on user_behavior

-- 当数据量超过内存时，使用临时表物化
CREATE TEMP TABLE user_sequence_temp AS
SELECT 
    user_id,
    event_type,
    event_time,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time) as rn,
    LAG(event_type, 1) OVER (PARTITION BY user_id ORDER BY event_time) as prev_event
FROM user_behavior
WHERE event_time BETWEEN '2024-01-01' AND '2024-01-31';

-- 在临时表上创建索引加速后续聚合
CREATE INDEX idx_temp_user ON user_sequence_temp (user_id);

-- 最终聚合
SELECT 
    user_id,
    COUNT(*) as total_events,
    SUM(CASE WHEN event_type = prev_event THEN 1 ELSE 0 END) as repeat_event_cnt
FROM user_sequence_temp
GROUP BY user_id;

-- 使用ON COMMIT DROP自动清理
CREATE TEMP TABLE user_sequence_temp (...) ON COMMIT DROP;

-- 创建并行安全函数
CREATE OR REPLACE FUNCTION calc_user_features_parallel()
RETURNS TABLE (user_id BIGINT, repeat_cnt INT) AS $$
    import pandas as pd
    import numpy as np
    from sklearn.feature_extraction import DictVectorizer
    
    # 从PostgreSQL读取数据（并行执行）
    query = """
        SELECT user_id, event_type, event_time::TEXT
        FROM user_behavior
        WHERE event_time BETWEEN '2024-01-01' AND '2024-01-31'
    """
    
    # 使用pandas原生窗口函数
    df = plpy.execute(query)
    df['event_time'] = pd.to_datetime(df['event_time'])
    df = df.sort_values(['user_id', 'event_time'])
    
    # 向量化计算
    df['prev_event'] = df.groupby('user_id')['event_type'].shift(1)
    df['is_repeat'] = (df['event_type'] == df['prev_event']).astype(int)
    
    result = df.groupby('user_id').agg({
        'is_repeat': 'sum',
        'event_type': 'count'
    }).reset_index()
    
    return result.itertuples(index=False)
$$ LANGUAGE plpython3u PARALLEL SAFE;

-- 调用（自动并行）
SELECT * FROM calc_user_features_parallel();

-- 开启JIT编译，优化窗口函数表达式
ALTER SYSTEM SET jit = on;
ALTER SYSTEM SET jit_above_cost = 50000;  -- 降低阈值
SELECT pg_reload_conf();

-- 重新执行查询，观察JIT效果
EXPLAIN (ANALYZE, TIMING)
SELECT ...;  -- 在输出中应看到"JIT:"段落

-- 创建性能对比视图
CREATE VIEW v_window_perf AS
SELECT 
    query,
    calls,
    total_exec_time,
    mean_exec_time,
    stddev_exec_time,
    rows
FROM pg_stat_statements
WHERE query LIKE '%WINDOW%' OR query LIKE '%OVER%'
ORDER BY total_exec_time DESC;

-- 基线监控
SELECT 
    now(),
    (SELECT total_exec_time FROM v_window_perf WHERE query LIKE '%user_behavior%') as window_time,
    (SELECT total_exec_time FROM v_window_perf WHERE query LIKE '%model_predictions%') as model_time;

# 数据目录空间分布
$ du -sh /data/pgsql/data/*
1.2T    base        # 实际数据
50G     pg_wal      # WAL日志
1.3T    pg_xact     # 事务目录（异常大！）
20G     pg_log

# pg_xact目录下大量活跃事务文件
$ ls -lh pg_xact/ | head
-rw------- 1 postgres postgres 256M Jan 15 10:23 0000
-rw------- 1 postgres postgres 256M Jan 15 10:24 0001
...
-rw------- 1 postgres postgres 256M Jan 15 10:45 0512  # 512个文件！

-- 步骤1：终止长事务
SELECT pg_terminate_backend(pid) 
FROM pg_stat_activity 
WHERE state = 'active' 
  AND query_start < NOW() - INTERVAL '2 hours'
  AND usename = 'analyst_user';

-- 步骤2：手动VACUUM（不锁表，但慢）
VACUUM (VERBOSE, ANALYZE) user_behavior;

-- 步骤3：VACUUM FULL（锁表，彻底回收空间）
VACUUM FULL user_behavior;  -- 需在维护窗口执行

-- 步骤4：清理pg_xact
CHECKPOINT;  -- 强制刷盘
SELECT pg_clean_xlog();  -- 需要superuser

-- 修改postgresql.conf
ALTER SYSTEM SET autovacuum = on;
ALTER SYSTEM SET autovacuum_max_workers = 8;  -- 增加worker数
ALTER SYSTEM SET autovacuum_naptime = '30s';  -- 缩短检查间隔

-- 大表专用策略
ALTER SYSTEM SET autovacuum_vacuum_scale_factor = 0.05;  -- 5%变更就触发
ALTER SYSTEM SET autovacuum_vacuum_threshold = 1000;      -- 至少1000条变更

-- 针对特定表（如频繁更新的特征表）
ALTER TABLE user_behavior SET (
    autovacuum_vacuum_scale_factor = 0.01,  -- 1%就触发
    autovacuum_vacuum_threshold = 500,
    fillfactor = 85  -- 预留15%空间给UPDATE，减少页分裂
);

-- 应用配置
SELECT pg_reload_conf();

-- 监控Autovacuum
CREATE VIEW v_autovacuum_stats AS
SELECT 
    schemaname,
    relname,
    last_vacuum,
    last_autovacuum,
    n_tup_ins,
    n_tup_upd,
    n_tup_del,
    n_live_tup,
    n_dead_tup,
    ROUND(100.0 * n_dead_tup / NULLIF(n_live_tup + n_dead_tup, 0), 2) as dead_ratio
FROM pg_stat_user_tables
WHERE n_dead_tup > 1000
ORDER BY dead_ratio DESC;

-- 将大表按时间分区，旧分区只读，无需VACUUM
CREATE TABLE user_behavior_range (
    user_id BIGINT,
    event_time TIMESTAMPTZ,
    ...
) PARTITION BY RANGE (event_time);

-- 每月一个分区，旧分区设置为只读
ALTER TABLE user_behavior_range_202312 SET (
    autovacuum_enabled = false,  -- 关闭旧分区VACUUM
    fillfactor = 100
);

-- 使用pg_partman自动管理分区
CREATE EXTENSION pg_partman;
SELECT partman.create_parent('public.user_behavior_range', 'event_time', 'native', 'monthly');

-- 创建膨胀率监控函数
CREATE OR REPLACE FUNCTION check_table_bloat()
RETURNS TABLE (
    table_name TEXT,
    dead_tup_count BIGINT,
    dead_ratio FLOAT,
    action TEXT
) AS $$
    SELECT 
        relname,
        n_dead_tup,
        ROUND(100.0 * n_dead_tup / NULLIF(n_live_tup + n_dead_tup, 0), 2),
        CASE 
            WHEN n_dead_tup > 100000 AND 
                 100.0 * n_dead_tup / NULLIF(n_live_tup + n_dead_tup, 0) > 20 
            THEN 'URGENT: VACUUM FULL required'
            WHEN n_dead_tup > 10000 THEN 'WARNING: Run VACUUM'
            ELSE 'OK'
        END
    FROM pg_stat_user_tables
    WHERE n_dead_tup > 1000;
$$ LANGUAGE sql;

-- 定时任务（cron）
*/30 * * * * psql -d algorithm_db -c "SELECT * FROM check_table_bloat()" | mail -s "PG Bloat Alert" dba@company.com

-- MySQL 8.0 + GIS扩展
SELECT 
    merchant_id,
    ST_Distance_Sphere(
        POINT(user_lon, user_lat), 
        POINT(merchant_lon, merchant_lat)
    ) / 1000 as distance_km
FROM users, merchants
WHERE ST_Contains(
    ST_GeomFromText('POLYGON((116.39 39.9, 116.40 39.9, ...))'), 
    POINT(user_lon, user_lat)
);

-- 使用PostGIS扩展
SELECT 
    merchant_id,
    ST_Distance(
        ST_MakePoint(user_lon, user_lat)::geography, 
        ST_MakePoint(merchant_lon, merchant_lat)::geography
    ) / 1000 as distance_km
FROM users, merchants
WHERE ST_Contains(
    ST_GeomFromText('POLYGON((116.39 39.9, 116.40 39.9, ...))', 4326), 
    ST_MakePoint(user_lon, user_lat)::geometry
);

-- 结果：距离偏差20%，商圈判定漏掉30%用户

-- 步骤1：升级数据类型
ALTER TABLE users 
ALTER COLUMN location TYPE geography(POINT, 4326) 
USING ST_MakePoint(lon, lat)::geography;

ALTER TABLE merchants 
ALTER COLUMN location TYPE geography(POINT, 4326) 
USING ST_MakePoint(lon, lat)::geography;

-- 步骤2：创建geography索引
CREATE INDEX idx_users_location_geo ON users USING GIST (location);
CREATE INDEX idx_merchants_location_geo ON merchants USING GIST (location);

-- 步骤3：改写查询为球面距离（精确）
SELECT 
    merchant_id,
    ST_Distance(
        u.location, 
        m.location
    ) / 1000 as distance_km  -- 结果单位已经是米
FROM users u, merchants m
WHERE ST_DWithin(
    u.location,
    m.location,
    5000  -- 5公里范围，利用geography索引
);

-- 步骤4：验证准确性
SELECT 
    ST_Distance(
        ST_MakePoint(116.40, 39.90)::geography,
        ST_MakePoint(116.41, 39.91)::geography
    ) as distance;  -- 应返回约1480米（球面距离）

-- 对需要平面计算的场景（如网格统计）
-- 1. 先转换为geometry
ALTER TABLE users ADD COLUMN location_geom geometry(POINT, 4326);
UPDATE users SET location_geom = location::geometry;

-- 2. 转换到UTM投影（北京区域：UTM Zone 50N）
ALTER TABLE users ADD COLUMN location_utm geometry(POINT, 32650);
UPDATE users SET location_utm = ST_Transform(location_geom, 32650);

-- 3. 使用平面距离（更快，但需定期重新投影）
CREATE INDEX idx_users_location_utm ON users USING GIST (location_utm);

-- 4. 查询（10公里内）
SELECT 
    merchant_id,
    ST_Distance(u.location_utm, m.location_utm) / 1000 as distance_km
FROM users u, merchants m
WHERE ST_DWithin(u.location_utm, m.location_utm, 10000);

-- 创建MySQL兼容函数
CREATE OR REPLACE FUNCTION st_distance_sphere(lat1 float, lon1 float, lat2 float, lon2 float)
RETURNS float AS $$
    SELECT ST_Distance(
        ST_MakePoint(lon1, lat1)::geography,
        ST_MakePoint(lon2, lat2)::geography
    );
$$ LANGUAGE SQL IMMUTABLE PARALLEL SAFE;

-- 改写业务SQL（最小改动）
SELECT 
    merchant_id,
    st_distance_sphere(user_lat, user_lon, merchant_lat, merchant_lon) / 1000 as distance_km
FROM users, merchants;

-- 创建函数索引（推荐）
CREATE INDEX idx_distance_sphere ON merchants 
USING GIST (ST_MakePoint(merchant_lon, merchant_lat)::geography);

-- 检查Polygon有效性
SELECT 
    ST_IsValidReason(geofence) as validity,
    ST_Area(geofence) as area
FROM business_districts;

-- 修复顶点顺序（强制逆时针）
UPDATE business_districts
SET geofence = ST_ForcePolygonCCW(geofence)
WHERE NOT ST_IsValid(geofence);

-- 确保闭合
UPDATE business_districts
SET geofence = ST_AddPoint(geofence, ST_StartPoint(geofence))
WHERE NOT ST_Equals(ST_StartPoint(geofence), ST_EndPoint(geofence));

-- 验证包含关系
SELECT 
    user_id,
    district_name
FROM users u
JOIN business_districts d ON ST_Contains(d.geofence, u.location::geometry)
WHERE ST_IsValid(d.geofence);

#!/usr/bin/env python3
# validate_gis_migration.py
import psycopg2
import math

def haversine_distance(lon1, lat1, lon2, lat2):
    """手动计算球面距离（基准）"""
    R = 6371.0  # 地球半径（公里）
    dlon = math.radians(lon2 - lon1)
    dlat = math.radians(lat2 - lat1)
    a = math.sin(dlat/2)**2 + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon/2)**2
    c = 2 * math.atan2(math.sqrt(a), math.sqrt(1-a))
    return R * c

def test_postgis_accuracy(conn):
    """验证PostGIS距离计算准确性"""
    cur = conn.cursor()
    
    test_cases = [
        (116.40, 39.90, 116.41, 39.91),  # 北京附近
        (121.47, 31.23, 121.48, 31.24),  # 上海附近
    ]
    
    for lon1, lat1, lon2, lat2 in test_cases:
        # 手动计算
        manual = haversine_distance(lon1, lat1, lon2, lat2)
        
        # PostGIS计算
        cur.execute("""
            SELECT ST_Distance(
                ST_MakePoint(%s, %s)::geography,
                ST_MakePoint(%s, %s)::geography
            ) / 1000
        """, (lon1, lat1, lon2, lat2))
        postgis = cur.fetchone()[0]
        
        # 误差容忍度：<0.1%
        error = abs(manual - postgis) / manual * 100
        status = "PASS" if error < 0.1 else "FAIL"
        
        print(f"手动: {manual:.6f}km, PostGIS: {postgis:.6f}km, 误差: {error:.4f}% [{status}]")

if __name__ == '__main__':
    conn = psycopg2.connect("host=pg-prod dbname=algorithm_db user=gis_user")
    test_postgis_accuracy(conn)
    conn.close()