[PostgreSQL]避开MySQL JSON查询陷阱：PostgreSQL算法应用指南

-- MySQL表结构
CREATE TABLE user_events (
    id BIGINT PRIMARY KEY AUTO_INCREMENT,
    user_id INT NOT NULL,
    event_data JSON NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 典型数据
INSERT INTO user_events (user_id, event_data) VALUES 
(123, '{"type": "purchase", "amount": 299, "product": {"id": 456, "category": "electronics"}}'),
(123, '{"type": "view", "page": "homepage", "duration": 15.5}');

-- 最常见的查询：找买了电子产品的用户
SELECT user_id, event_data FROM user_events 
WHERE JSON_EXTRACT(event_data, '$.type') = 'purchase' 
  AND JSON_EXTRACT(event_data, '$.product.category') = 'electronics';

-- 创建虚拟列并加索引（这才是MySQL的正确姿势）
ALTER TABLE user_events 
ADD COLUMN event_type VARCHAR(20) GENERATED ALWAYS AS 
    (JSON_UNQUOTE(JSON_EXTRACT(event_data, '$.type'))) STORED,
ADD COLUMN product_category VARCHAR(50) GENERATED ALWAYS AS 
    (JSON_UNQUOTE(JSON_EXTRACT(event_data, '$.product.category'))) STORED;

CREATE INDEX idx_event_type ON user_events(event_type);
CREATE INDEX idx_product_category ON user_events(product_category);

-- 改写查询（终于能用上索引了）
SELECT user_id, event_data FROM user_events 
WHERE event_type = 'purchase' 
  AND product_category = 'electronics';

-- MySQL 5.7
UPDATE users SET tags = JSON_MERGE_PATCH(tags, '["vip"]') WHERE id = 1;
-- 结果：["old_tag", "vip"] 完美！

-- 升级到MySQL 8.0后
UPDATE users SET tags = JSON_MERGE_PATCH(tags, '["vip"]') WHERE id = 1;
-- 结果：["old_tag", "vip"] 咦？好像没变？

-- 查文档发现8.0改了实现，需要改成：
UPDATE users SET tags = JSON_ARRAY_APPEND(tags, '$', 'vip') WHERE id = 1;
-- 等等，那5.7又不支持这个函数...卒

# MySQL 8.0（模拟现有生产环境）
docker run -d --name mysql-source \
  -e MYSQL_ROOT_PASSWORD=root123 \
  -p 3306:3306 \
  mysql:8.0.33

# PostgreSQL 15（目标环境）
docker run -d --name pg-target \
  -e POSTGRES_PASSWORD=root123 \
  -p 5432:5432 \
  postgres:15.3-alpine

-- MySQL源表（现有业务）
CREATE DATABASE order_system;
USE order_system;

CREATE TABLE orders (
    order_id BIGINT PRIMARY KEY,
    user_id INT NOT NULL,
    order_data JSON NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_user_id (user_id)
);

-- 模拟500万条订单数据（用存储过程批量生成）
DELIMITER $$

CREATE PROCEDURE generate_orders()
BEGIN
    DECLARE i INT DEFAULT 1;
    WHILE i <= 5000000 DO
        INSERT INTO orders (order_id, user_id, order_data) VALUES (
            i,
            FLOOR(1 + RAND() * 100000),
            JSON_OBJECT(
                'items', JSON_ARRAY(
                    JSON_OBJECT('sku', CONCAT('SKU', FLOOR(RAND()*1000)), 'qty', FLOOR(1 + RAND() * 5), 'price', ROUND(10 + RAND() * 990, 2)),
                    JSON_OBJECT('sku', CONCAT('SKU', FLOOR(RAND()*1000)), 'qty', FLOOR(1 + RAND() * 3), 'price', ROUND(5 + RAND() * 495, 2))
                ),
                'shipping', JSON_OBJECT('city', ELT(1 + FLOOR(RAND() * 4), '北京', '上海', '深圳', '杭州'), 'fee', ROUND(5 + RAND() * 15, 2)),
                'payment', JSON_OBJECT('method', ELT(1 + FLOOR(RAND() * 3), 'alipay', 'wechat', 'card'), 'amount', ROUND(50 + RAND() * 950, 2))
            )
        );
        SET i = i + 1;
    END WHILE;
END$$

DELIMITER ;

-- 开始生成（这个可能要跑几分钟，去泡杯咖啡吧）
CALL generate_orders();

-- 验证数据量
SELECT COUNT(*) FROM orders;
SELECT order_data FROM orders LIMIT 2;

-- 典型查询：查找北京用户的支付宝订单
SELECT order_id, user_id, order_data 
FROM orders 
WHERE JSON_EXTRACT(order_data, '$.shipping.city') = '北京' 
  AND JSON_EXTRACT(order_data, '$.payment.method') = 'alipay'
LIMIT 10;

-- 查看执行计划（重点！）
EXPLAIN ANALYZE 
SELECT COUNT(*) FROM orders 
WHERE JSON_EXTRACT(order_data, '$.payment.amount') > 500;

-- PostgreSQL目标库
CREATE DATABASE order_system;
\c order_system;

-- 订单表（JSONB版本）
CREATE TABLE orders (
    order_id BIGINT PRIMARY KEY,
    user_id INT NOT NULL,
    order_data JSONB NOT NULL,  -- 注意是JSONB不是JSON
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- 创建GIN索引（这才是灵魂！）
CREATE INDEX idx_order_data_gin ON orders USING GIN (order_data);

-- 创建特定路径的GIN索引（更精准的优化）
CREATE INDEX idx_order_city ON orders USING GIN ((order_data -> 'shipping' -> 'city'));
CREATE INDEX idx_payment_method ON orders USING GIN ((order_data -> 'payment' -> 'method'));

-- 还可以创建B-tree索引在常用字段上
CREATE INDEX idx_user_id ON orders(user_id);
CREATE INDEX idx_created_at ON orders(created_at);

-- 表结构说明
COMMENT ON TABLE orders IS '订单主表，使用JSONB存储扩展属性';
COMMENT ON INDEX idx_order_data_gin IS 'GIN索引支持任意JSONB路径查询';

#!/usr/bin/env python3
"""
MySQL到PostgreSQL的JSON数据迁移脚本
支持断点续传、进度显示、错误重试
"""
import mysql.connector
import psycopg2
import json
from datetime import datetime
from tqdm import tqdm  # 进度条库

# 配置
MYSQL_CONFIG = {
    'host': 'localhost',
    'port': 3306,
    'user': 'root',
    'password': 'root123',
    'database': 'order_system',
    'charset': 'utf8mb4'
}

PG_CONFIG = {
    'host': 'localhost',
    'port': 5432,
    'user': 'postgres',
    'password': 'root123',
    'database': 'order_system'
}

def migrate_orders(batch_size=5000, resume_from=0):
    """
    迁移订单数据，带进度条和断点续传
    
    参数:
        batch_size: 每批处理的记录数
        resume_from: 从哪个order_id开始（断点续传用）
    """
    
    # 连接MySQL
    mysql_conn = mysql.connector.connect(**MYSQL_CONFIG)
    mysql_cursor = mysql_conn.cursor(dictionary=True)
    
    # 连接PostgreSQL
    pg_conn = psycopg2.connect(**PG_CONFIG)
    pg_cursor = pg_conn.cursor()
    
    try:
        # 统计总数
        mysql_cursor.execute("SELECT COUNT(*) as total FROM orders WHERE order_id >= %s", (resume_from,))
        total = mysql_cursor.fetchone()['total']
        
        print(f"[{datetime.now()}] 开始迁移数据，总计 {total} 条...")
        
        # 分批查询和插入
        offset = 0
        pbar = tqdm(total=total, desc="迁移进度")
        
        while True:
            # 查询一批MySQL数据
            query = """
                SELECT order_id, user_id, order_data, created_at 
                FROM orders 
                WHERE order_id >= %s
                ORDER BY order_id ASC
                LIMIT %s
            """
            mysql_cursor.execute(query, (resume_from + offset, batch_size))
            rows = mysql_cursor.fetchall()
            
            if not rows:
                break
            
            # 转换并插入PostgreSQL
            for row in rows:
                # MySQL的JSON字段返回的是字符串，需要确认是否为有效JSON
                try:
                    # 验证并格式化JSON（确保是标准JSON）
                    order_data = json.loads(row['order_data'])  # 解析
                    order_data_jsonb = json.dumps(order_data, ensure_ascii=False)  # 重新序列化
                    
                    # 插入PostgreSQL
                    pg_cursor.execute(
                        """
                        INSERT INTO orders (order_id, user_id, order_data, created_at)
                        VALUES (%s, %s, %s::jsonb, %s)
                        ON CONFLICT (order_id) DO NOTHING  -- 避免重复
                        """,
                        (row['order_id'], row['user_id'], order_data_jsonb, row['created_at'])
                    )
                except json.JSONDecodeError as e:
                    print(f"警告: order_id={row['order_id']} JSON解析失败: {e}")
                    continue
            
            # 提交事务
            pg_conn.commit()
            
            # 更新进度
            pbar.update(len(rows))
            offset += batch_size
            
            # 打印状态
            if offset % 50000 == 0:
                print(f"[{datetime.now()}] 已迁移 {offset} 条...")
        
        pbar.close()
        print(f"[{datetime.now()}] 迁移完成！")
        
    except Exception as e:
        print(f"迁移出错: {e}")
        pg_conn.rollback()
        raise
    finally:
        mysql_cursor.close()
        mysql_conn.close()
        pg_cursor.close()
        pg_conn.close()

def verify_migration():
    """验证数据一致性"""
    mysql_conn = mysql.connector.connect(**MYSQL_CONFIG)
    pg_conn = psycopg2.connect(**PG_CONFIG)
    
    try:
        mysql_cursor = mysql_conn.cursor()
        pg_cursor = pg_conn.cursor()
        
        # 统计对比
        mysql_cursor.execute("SELECT COUNT(*) FROM orders")
        mysql_count = mysql_cursor.fetchone()[0]
        
        pg_cursor.execute("SELECT COUNT(*) FROM orders")
        pg_count = pg_cursor.fetchone()[0]
        
        print(f"数据验证:")
        print(f"  MySQL源数据: {mysql_count} 条")
        print(f"  PostgreSQL目标数据: {pg_count} 条")
        print(f"  一致性: {'✓' if mysql_count == pg_count else '✗'}")
        
        # 随机抽样验证
        pg_cursor.execute("""
            SELECT order_id, order_data->>'order_id' as json_order_id
            FROM orders 
            ORDER BY RANDOM()
            LIMIT 5
        """)
        
        print("\n抽样验证JSON数据解析:")
        for row in pg_cursor.fetchall():
            print(f"  order_id={row[0]}, JSON解析order_id={row[1]}")
            
    finally:
        mysql_conn.close()
        pg_conn.close()

if __name__ == '__main__':
    # 执行迁移
    migrate_orders(batch_size=5000, resume_from=0)
    
    # 验证数据
    verify_migration()

-- PostgreSQL版本，体验飞一般的感觉
EXPLAIN (ANALYZE, BUFFERS)
SELECT order_id, user_id, order_data 
FROM orders 
WHERE order_data @> '{"shipping": {"city": "北京"}}'  -- GIN索引神器
  AND order_data -> 'payment' ->> 'method' = 'alipay'
LIMIT 10;

-- 执行计划会显示：
-- Bitmap Heap Scan on orders ...
--   Recheck Cond: (order_data @> '{"shipping": {"city": "北京"}}'::jsonb)
--   Buffers: shared hit=45 read=12
-- 只需要扫描57个buffer！

-- MySQL版本（噩梦）
SELECT * FROM orders 
WHERE JSON_EXTRACT(order_data, '$.payment.amount') > 500
  AND JSON_CONTAINS(
    order_data->'$.items', 
    JSON_OBJECT('sku', 'SKU123')
  );

-- PostgreSQL版本（清晰）
SELECT order_id, user_id, order_data
FROM orders 
WHERE (order_data -> 'payment' ->> 'amount')::DECIMAL > 500
  AND order_data @> '{"items": [{"sku": "SKU123"}]}';

-- 更复杂的：查找items数组中任意商品价格>1000的订单
SELECT order_id, 
       item ->> 'sku' as sku,
       (item ->> 'price')::DECIMAL as price
FROM orders,
     jsonb_array_elements(order_data -> 'items') as item
WHERE (item ->> 'price')::DECIMAL > 1000;

-- 这个用了jsonb_array_elements函数展开数组，然后对展开后的数据过滤
-- 性能依然很棒，因为GIN索引会先过滤候选集

-- 统计各城市的订单数和平均金额
-- MySQL版本（性能爆炸）
SELECT 
    JSON_EXTRACT(order_data, '$.shipping.city') as city,
    COUNT(*) as order_count,
    AVG(JSON_EXTRACT(order_data, '$.payment.amount')) as avg_amount
FROM orders 
GROUP BY JSON_EXTRACT(order_data, '$.shipping.city');

-- PostgreSQL版本（毫秒级响应）
SELECT 
    order_data -> 'shipping' ->> 'city' as city,
    COUNT(*) as order_count,
    AVG((order_data -> 'payment' ->> 'amount')::DECIMAL) as avg_amount
FROM orders 
GROUP BY order_data -> 'shipping' ->> 'city';

-- 更高级的：统计每个城市各支付方式占比
WITH city_payments AS (
    SELECT 
        order_data -> 'shipping' ->> 'city' as city,
        order_data -> 'payment' ->> 'method' as payment_method,
        COUNT(*) as cnt
    FROM orders 
    GROUP BY 1, 2
)
SELECT 
    city,
    payment_method,
    cnt,
    ROUND(cnt * 100.0 / SUM(cnt) OVER (PARTITION BY city), 2) as percentage
FROM city_payments
ORDER BY city, cnt DESC;

-- 这个查询在MySQL里基本写不出来，就算写出来也跑不动

-- 模式1：对整个JSONB字段建GIN索引（最通用）
CREATE INDEX idx_order_data_gin ON orders USING GIN (order_data);
-- 优点：支持任意路径查询
-- 缺点：索引体积大，写入慢

-- 模式2：对特定路径建GIN索引（精准优化）
CREATE INDEX idx_order_city ON orders USING GIN ((order_data -> 'shipping' -> 'city'));
-- 优点：索引小，查询极快
-- 缺点：只支持这个路径

-- 模式3：多列复合GIN索引
CREATE INDEX idx_order_city_payment ON orders USING GIN (
    (order_data -> 'shipping' -> 'city'),
    (order_data -> 'payment' -> 'method')
);
-- 优点：覆盖常见组合查询
-- 缺点：维护成本高

-- 模式4：GIN + B-tree混合（终极方案）
CREATE INDEX idx_order_city_btree ON orders USING BTREE ((order_data ->> 'shipping.city'));
CREATE INDEX idx_order_payment_gin ON orders USING GIN (order_data);
-- 优点：精确匹配用BTREE，模糊查询用GIN
-- 缺点：需要分析查询模式

-- 1. 时间范围用BTREE（范围查询最擅长）
CREATE INDEX idx_created_at ON orders(created_at);

-- 2. 城市用函数式BTREE索引（等值查询）
CREATE INDEX idx_shipping_city ON orders USING BTREE ((order_data ->> 'shipping.city'));

-- 3. 其他JSON路径用GIN（应对不可预知的需求）
CREATE INDEX idx_order_data_gin ON orders USING GIN (order_data);

-- 4. 联合索引应对特定报表
CREATE INDEX idx_user_city_date ON orders 
USING BTREE (user_id, (order_data ->> 'shipping.city'), created_at);

-- 查看索引大小
SELECT 
    indexname,
    pg_size_pretty(pg_relation_size(indexrelid)) as index_size,
    idx_scan as index_scans
FROM pg_stat_user_indexes 
WHERE tablename = 'orders'
ORDER BY pg_relation_size(indexrelid) DESC;

-- 清理未使用的索引（季度维护）
-- 注意：idx_scan=0不代表没用，可能是季度报表才用

-- GIN索引膨胀处理（VACUUM不能解决GIN膨胀）
-- 需要定期重建
REINDEX INDEX idx_order_data_gin;

-- 假设order_data里有个notes字段存用户备注
CREATE INDEX idx_notes_fulltext ON orders USING GIN (
    to_tsvector('english', order_data ->> 'notes') || 
    to_tsvector('simple', order_data -> 'items' ->> 'sku')
);

-- 搜索包含"urgent"且SKU包含"phone"的订单
SELECT order_id, 
       order_data ->> 'notes' as notes,
       item ->> 'sku' as sku
FROM orders,
     jsonb_array_elements(order_data -> 'items') as item
WHERE to_tsvector('english', order_data ->> 'notes') @@ to_tsquery('urgent')
  AND item ->> 'sku' LIKE '%phone%';

-- 创建物化视图：每日城市销售统计
CREATE MATERIALIZED VIEW daily_city_sales AS
SELECT 
    DATE(created_at) as sale_date,
    order_data -> 'shipping' ->> 'city' as city,
    COUNT(*) as order_count,
    SUM((order_data -> 'payment' ->> 'amount')::DECIMAL) as total_amount
FROM orders 
GROUP BY 1, 2;

-- 创建索引
CREATE UNIQUE INDEX ON daily_city_sales(sale_date, city);
CREATE INDEX ON daily_city_sales(sale_date);

-- 刷新策略
-- 手动刷新：REFRESH MATERIALIZED VIEW CONCURRENTLY daily_city_sales;
-- 自动刷新：用pg_cron插件定时任务

-- 查询飞快
SELECT * FROM daily_city_sales 
WHERE sale_date = CURRENT_DATE - 1
ORDER BY total_amount DESC;

-- 按月份分区（PostgreSQL 11+ 支持声明式分区）
CREATE TABLE orders_partitioned (
    order_id BIGINT NOT NULL,
    user_id INT NOT NULL,
    order_data JSONB NOT NULL,
    created_at TIMESTAMPTZ NOT NULL
) PARTITION BY RANGE (created_at);

-- 创建分区
CREATE TABLE orders_y2024m01 PARTITION OF orders_partitioned
    FOR VALUES FROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE orders_y2024m02 PARTITION OF orders_partitioned
    FOR VALUES FROM ('2024-02-01') TO ('2024-03-01');
-- ...按需创建后续分区

-- 索引也建在每个分区上
CREATE INDEX ON orders_y2024m01 USING GIN (order_data);
CREATE INDEX ON orders_y2024m01 (created_at);

-- 查询自动路由到分区
EXPLAIN SELECT * FROM orders_partitioned 
WHERE created_at >= '2024-01-15' 
  AND created_at < '2024-01-20';
-- 会看到只扫描orders_y2024m01分区

# 数据清洗示例：处理MySQL的JSON null问题
def clean_mysql_json(mysql_json_str):
    """MySQL里经常有{"key": null}存成字符串的情况"""
    if mysql_json_str is None:
        return None
    
    # 解析JSON
    try:
        data = json.loads(mysql_json_str)
    except:
        return None
    
    # 递归清理null字符串
    def clean_nulls(obj):
        if isinstance(obj, dict):
            return {k: clean_nulls(v) for k, v in obj.items() if v != "null"}
        elif isinstance(obj, list):
            return [clean_nulls(item) for item in obj if item != "null"]
        else:
            return obj
    
    cleaned = clean_nulls(data)
    return json.dumps(cleaned, ensure_ascii=False)

-- MySQL习惯（隐式提交）
INSERT INTO logs VALUES (1, 'test');  -- 自动提交

-- PostgreSQL（默认事务）
BEGIN;
INSERT INTO logs VALUES (1, 'test');  -- 没提交！
-- 必须COMMIT; 或者设置自动提交

-- 应用层需要调整
# Python psycopg2需要
conn.autocommit = True  # 模拟MySQL行为
# 或者
conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)

-- 解决方案1：调整填充因子
CREATE INDEX idx_order_data_gin ON orders USING GIN (order_data) WITH (fillfactor = 70);
-- 留更多空间减少页分裂

-- 解决方案2：异步写入
-- 应用层先写入到暂存表（无索引），再用后台任务批量转移到主表
CREATE TABLE orders_staging (LIKE orders);
-- 暂存表不加任何索引，写入飞快

-- 解决方案3：分区表降低锁竞争
-- 每个分区独立索引，写入分散

# 安装pgbench（PostgreSQL自带）
pgbench -i -s 100 order_system  # 初始化测试数据

# 自定义脚本测试JSON查询
cat > json_test.sql <<EOF
\\set city random(1, 4)
SELECT count(*) FROM orders 
WHERE order_data @> jsonb_build_object('shipping', jsonb_build_object('city', 
    CASE :city 
        WHEN 1 THEN '北京'
        WHEN 2 THEN '上海' 
        WHEN 3 THEN '深圳'
        ELSE '杭州'
    END
));
EOF

# 执行压测
pgbench -c 50 -j 4 -T 60 -f json_test.sql order_system

-- postgresql.conf关键参数
shared_buffers = 4GB                    # 设为内存的25%
effective_cache_size = 12GB            # 设为内存的75%
work_mem = 64MB                        # 复杂JSON查询需要更大
maintenance_work_mem = 512MB           # 索引构建用
random_page_cost = 1.1                 # SSD设为1.1，机械盘保持4
effective_io_concurrency = 200         # SSD调高
max_worker_processes = 8               # 并行查询
max_parallel_workers_per_gather = 4    # 单个查询并行度
max_parallel_workers = 8               # 总并行工作进程

-- JSONB特定优化
gin_pending_list_limit = 4MB           # GIN索引缓冲，写入密集调大

-- 监控GIN索引扫描效率
SELECT 
    schemaname,
    tablename,
    indexname,
    idx_scan as scans,
    idx_tup_read as tuples_read,
    idx_tup_fetch as tuples_fetched,
    ROUND(idx_tup_fetch::NUMERIC / NULLIF(idx_tup_read, 0), 2) as hit_ratio
FROM pg_stat_user_indexes 
WHERE indexname LIKE 'idx_order%'
ORDER BY scans DESC;

-- 监控JSON函数调用耗时
SELECT 
    userid,
    dbid,
    queryid,
    query,
    mean_exec_time as avg_time_ms,
    calls
FROM pg_stat_statements
WHERE query LIKE '%jsonb%' OR query LIKE '%->>%'
ORDER BY mean_exec_time DESC 
LIMIT 20;

-- 检查键是否存在
WHERE order_data ? 'shipping'

-- 数组包含
WHERE order_data -> 'items' ? 'SKU123'

-- 范围查询
WHERE (order_data -> 'payment' ->> 'amount')::DECIMAL BETWEEN 100 AND 1000

-- 模糊匹配（配合全文检索）
WHERE to_tsvector('simple', order_data ->> 'notes') @@ to_tsquery('fast & delivery')