MySQL到PostgreSQL完整迁移指南(含数据迁移/语法差异/性能调优)

mysql (innodb mvcc)：
┌──────────────────────────────────────────────────────┐
│  read view 在事务开始时生成，整个事务看到相同的快照    │
│  问题：长事务会锁定大量历史版本，影响 purge            │
│  读已提交(默认)时，每次语句重新计算快照                │
└──────────────────────────────────────────────────────┘

postgresql：
┌──────────────────────────────────────────────────────┐
│  每个查询独立生成快照（postgresql 17+ 更优化）        │
│  vacuum 异步清理死亡元组，不阻塞读写                  │
│  真正实现读写不互阻塞                                 │
└──────────────────────────────────────────────────────┘

-- mysql: 开启两个会话
-- session 1:
begin;
select balance from accounts where id = 1;  -- 看到 balance = 1000

-- session 2 (另一会话):
update accounts set balance = 2000 where id = 1;
commit;

-- session 1:
select balance from accounts where id = 1;  -- mysql 读已提交：看到 2000（不可重复读）
                                              -- postgresql repeatable read：仍看到 1000

commit;

-- mysql 8.0 json 查询
select json_extract(user_data, '$.name') from users;
select * from users where json_contains(user_data, '"alice"', '$.name');

-- postgresql jsonb 查询（推荐用 jsonb 而非 json，性能差异巨大）
select user_data->>'name' from users;  -- 文本提取
select user_data->'name' from users;   -- json 类型提取
select * from users where user_data->>'name' = 'alice';

-- postgresql jsonb 索引（gin 索引，效果显著）
create index idx_users_data_name on users using gin ((user_data jsonb_path_ops));
create index idx_users_name on users ((user_data->>'name'));

-- mysql 全文索引（innodb，需 fulltext 索引）
alter table articles add fulltext index ft_title_content(title, content);
select * from articles
where match(title, content) against('postgresql 迁移' in natural language mode);

-- postgresql 全文搜索（内置，无需插件）
alter table articles add column search_vector tsvector;
update articles set search_vector = to_tsvector('simple', title || ' ' || content);
create index idx_articles_search on articles using gin(search_vector);

-- 多语言支持（中文需安装 zhparser 扩展）
create extension if not exists zhparser;
create text search configuration chinese_zhparser (parser = zhparser);
alter text search configuration chinese_zhparser add mapping for n,v,a,e,i,l with simple;

select * from articles where search_vector @@ to_tsquery('chinese_zhparser', 'postgresql & 迁移');

-- postgresql 并行查询（默认启用）
show max_parallel_workers_per_gather;  -- 默认 2
show parallel_tuple_cost;               -- 成本阈值

-- 强制使用并行（测试用）
set max_parallel_workers_per_gather = 4;
select * from large_table where status = 'active';  -- 可能触发 parallel seq scan

-- mysql 并行查询（mysql 8.0+ 支持有限并行）
-- 主要用于复制相关的并行读取，写操作无并行
set session cte_max_recursion_depth = 100000;

✅ 数据库对象盘点
  - 表数量、字段类型分布
  - 索引数量及类型（普通/唯一/全文/空间）
  - 存储过程/函数/触发器数量及复杂度
  - 外键约束数量
  - 分区表结构

✅ 数据量评估
  - 单表最大行数
  - 总数据量（gb/tb 级迁移策略不同）
  - lob/blob/text 字段占比
  - json 字段使用情况

✅ sql 复杂度分析
  - 统计高频查询（通过慢查询日志）
  - 识别 mysql 特有语法（handler、get_lock 等）
  - 识别不兼容 sql（需重写）

✅ 应用层适配
  - 连接池（hikaricp/pgbouncer 配置）
  - 驱动版本（mysql connector → postgresql jdbc）
  - orm 框架适配（mybatis/hibernate 参数语法）

场景 1: < 100gb，零停机要求
  → pgloader（主） + 手动补数（增量） + 双写验证
场景 2: 100gb ~ 1tb，有零停机要求
  → aws dms（持续复制） + 应用双写过渡
场景 3: > 1tb，跨云迁移
  → debezium cdc + kafka + 自研消费者

# 安装 pgloader（macos）
brew install pgloader
# docker 方式（推荐，无需本地安装）
docker run --rm -it dimitri/pgloader:latest pgloader --help
# 快速评估 mysql 数据库并生成 postgresql ddl
docker run --rm -it dimitri/pgloader:latest \
  pgloader mysql://root:password@localhost/source_db \
  postgresql://postgres:password@localhost/target_db \
  --dry-run  # 仅评估，不实际迁移

-- mysql:
create table orders (
    id bigint auto_increment primary key,
    customer_id bigint not null,
    created_at datetime default current_timestamp,
    primary key (id)
);

-- postgresql（方式1: 序列，推荐）:
create table orders (
    id bigserial primary key,           -- 等价于 bigint + 自动序列
    customer_id bigint not null,
    created_at timestamptz default now(),
    constraint orders_pkey primary key (id)
);

-- postgresql（方式2: generated always as identity，更标准）:
create table orders (
    id bigint generated always as identity primary key,
    customer_id bigint not null,
    created_at timestamptz default now()
);

-- 如果需要指定自增起始值（承接 mysql 数据）:
create sequence orders_id_seq start with 1000000;
alter table orders alter column id set default nextval('orders_id_seq');

-- mysql 唯一索引
create unique index idx_user_email on users(email);

-- postgresql:
create unique index idx_user_email on users(email);

-- mysql 全文索引（无法直接转换，需重建）
create fulltext index ft_article on articles(title, body);

-- postgresql 全文索引（重建方式）:
alter table articles add column search_vector tsvector 
    generated always as (to_tsvector('simple', coalesce(title,'') || ' ' || coalesce(body,''))) stored;
create index idx_articles_fulltext on articles using gin(search_vector);

-- mysql 前缀索引（postgresql 不支持前缀索引，使用表达式索引替代）
create index idx_user_phone on users(phone(10));  -- mysql: 只索引前10位

-- postgresql: 两种方式
create index idx_user_phone_prefix on users ((phone_text(phone)));  -- 自定义函数
-- 或者直接索引整个字段（postgresql 对短字符串索引效率仍然很高）

-- mysql 多列索引
create index idx_order_status_date on orders(status, created_at);

-- postgresql: 完全相同语法，但建议将选择性高的列放前面
create index idx_order_status_date on orders(status, created_at);

-- postgresql 特有：并发创建索引（生产环境必须用，避免锁表）
create index concurrently idx_order_new on orders(customer_id);

-- mysql 外键（注意：mysql 中外键名称可选，postgresql 必须命名）
create table order_items (
    id bigint auto_increment,
    order_id bigint not null,
    product_id bigint,
    quantity int default 1,
    foreign key (order_id) references orders(id) on delete cascade,
    foreign key (product_id) references products(id) on delete set null
);

-- postgresql（约束必须命名）:
create table order_items (
    id bigserial primary key,
    order_id bigint not null,
    product_id bigint,
    quantity int default 1,
    constraint fk_order_items_order foreign key (order_id) references orders(id) on delete cascade,
    constraint fk_order_items_product foreign key (product_id) references products(id) on delete set null
);

-- mysql check 约束（mysql 8.0 才支持，之前版本会被忽略）
create table users (
    id int auto_increment primary key,
    age int check (age >= 0 and age <= 150),  -- mysql 8.0+ 才生效
    status enum('active','inactive','suspended')
);

-- postgresql（check 约束完整支持）:
create table users (
    id serial primary key,
    age int check (age >= 0 and age <= 150),
    status varchar(20) check (status in ('active','inactive','suspended'))
);

-- mysql 默认值（可使用函数）
created_at datetime default current_timestamp,
updated_at datetime default current_timestamp on update current_timestamp

-- postgresql 默认值
created_at timestamptz default now(),
updated_at timestamptz default now()
-- 动态更新触发器（见触发器迁移章节）

#!/bin/bash
# migrate_mysql_to_pg.sh
# 依赖: pgloader, mysql client, postgresql client
set -e
# ========== 配置区 ==========
mysql_host="192.168.1.100"
mysql_port="3306"
mysql_user="migration"
mysql_pass="your_password"
mysql_db="source_app"
pg_host="192.168.1.101"
pg_port="5432"
pg_user="postgres"
pg_pass="your_password"
pg_db="target_app"
log_dir="./migration_logs"
date_tag=$(date +%y%m%d_%h%m%s)
mkdir -p "$log_dir"
# ========== step 1: 评估阶段（只读分析）==========
echo "[$(date)] step 1: 开始数据库评估..."
pgloader \
  mysql://${mysql_user}:${mysql_pass}@${mysql_host}:${mysql_port}/${mysql_db} \
  postgresql://${pg_user}:${pg_pass}@${pg_host}:${pg_port}/${pg_db} \
  --dry-run \
  --debug \
  2>&1 | tee "${log_dir}/step1_assessment_${date_tag}.log"
# ========== step 2: 完整迁移（加并发加速）==========
echo "[$(date)] step 2: 开始数据迁移..."
pgloader \
  mysql://${mysql_user}:${mysql_pass}@${mysql_host}:${mysql_port}/${mysql_db} \
  postgresql://${pg_user}:${pg_pass}@${pg_host}:${pg_port}/${pg_db} \
  with include drop, create tables, create indexes, foreign keys, data only \
  set postgresql bytea output to 'escape' \
  set mysql bigint unsigned to integer \
  cast type date to timestamp drop default, \
       type enum to text drop default, \
       type blob to bytea drop default \
  2>&1 | tee "${log_dir}/step2_migration_${date_tag}.log"
echo "[$(date)] 迁移完成，日志: ${log_dir}/step2_migration_${date_tag}.log"

#!/usr/bin/env python3
"""
mysql -> postgresql 迁移脚本（支持断点续传和增量迁移）
依赖: pip install mysql-connector-python psycopg2-binary
"""
import os
import sys
import time
import logging
from datetime import datetime
from typing import optional, generator, any
import mysql.connector
import psycopg2
from psycopg2 import sql
logging.basicconfig(
    level=logging.info,
    format='%(asctime)s [%(levelname)s] %(message)s',
    handlers=[
        logging.filehandler(f'migration_{datetime.now():%y%m%d_%h%m%s}.log'),
        logging.streamhandler()
    ]
)
logger = logging.getlogger(__name__)
class mysqltopostgresmigrator:
    """mysql → postgresql 迁移器，支持断点续传"""
    def __init__(
        self,
        mysql_conf: dict,
        pg_conf: dict,
        checkpoint_file: str = ".migration_checkpoint.json",
        batch_size: int = 5000,
    ):
        self.mysql_conf = mysql_conf
        self.pg_conf = pg_conf
        self.checkpoint_file = checkpoint_file
        self.batch_size = batch_size
        self.mysql_conn = none
        self.pg_conn = none
        self.checkpoints = self._load_checkpoints()
    def _load_checkpoints(self) -> dict:
        """加载断点文件"""
        if os.path.exists(self.checkpoint_file):
            import json
            with open(self.checkpoint_file) as f:
                return json.load(f)
        return {}
    def _save_checkpoint(self, table: str, max_id: any):
        """保存断点"""
        self.checkpoints[table] = max_id
        import json
        with open(self.checkpoint_file, 'w') as f:
            json.dump(self.checkpoints, f)
        logger.info(f"断点已保存: {table} -> {max_id}")
    def connect(self):
        """建立数据库连接"""
        self.mysql_conn = mysql.connector.connect(**self.mysql_conf)
        self.pg_conn = psycopg2.connect(**self.pg_conf)
        self.pg_conn.autocommit = false
        logger.info("数据库连接已建立")
    def close(self):
        """关闭连接"""
        if self.mysql_conn:
            self.mysql_conn.close()
        if self.pg_conn:
            self.pg_conn.commit()
            self.pg_conn.close()
        logger.info("数据库连接已关闭")
    def get_tables(self) -> list[dict]:
        """获取所有需要迁移的表"""
        cursor = self.mysql_conn.cursor(dictionary=true)
        cursor.execute("""
            select table_name, table_rows, data_length, index_length
            from information_schema.tables
            where table_schema = %s and table_type = 'base table'
            order by data_length desc
        """, (self.mysql_conf['database'],))
        tables = cursor.fetchall()
        cursor.close()
        return tables
    def get_columns(self, table: str) -> list[dict]:
        """获取表字段信息"""
        cursor = self.mysql_conn.cursor(dictionary=true)
        cursor.execute("""
            select column_name, data_type, is_nullable, column_key,
                   column_default, character_maximum_length, numeric_precision,
                   numeric_scale, extra
            from information_schema.columns
            where table_schema = %s and table_name = %s
            order by ordinal_position
        """, (self.mysql_conf['database'], table))
        columns = cursor.fetchall()
        cursor.close()
        return columns
    def get_primary_key(self, table: str) -> optional[str]:
        """获取主键字段名"""
        cursor = self.mysql_conn.cursor(dictionary=true)
        cursor.execute("""
            select column_name
            from information_schema.key_column_usage
            where table_schema = %s and table_name = %s
              and constraint_name = 'primary'
        """, (self.mysql_conf['database'], table))
        result = cursor.fetchone()
        cursor.close()
        return result['column_name'] if result else none
    def _convert_value(self, value: any, col_type: str) -> any:
        """数据类型转换"""
        if value is none:
            return none
        if 'blob' in col_type.lower() or 'binary' in col_type.lower():
            # mysql blob -> postgresql bytea
            return psycopg2.binary(value)
        return value
    def migrate_table(self, table: str):
        """迁移单个表"""
        start_id = self.checkpoints.get(table, 0)
        pk = self.get_primary_key(table)
        if not pk:
            logger.warning(f"表 {table} 无主键，使用全表扫描模式")
            pk = self.get_columns(table)[0]['column_name']
            # 无主键时不支持断点续传
            start_id = none
        columns = self.get_columns(table)
        col_names = [c['column_name'] for c in columns]
        col_placeholders = ','.join(['%s'] * len(col_names))
        col_names_str = ', '.join(col_names)
        # 构建 insert 语句
        insert_sql = sql.sql("insert into {table} ({cols}) values ({placeholders}) on conflict do nothing").format(
            table=sql.identifier(table),
            cols=sql.sql(', ').join(map(sql.identifier, col_names)),
            placeholders=sql.sql(', ').join(['%s'] * len(col_names))
        )
        mysql_cursor = self.mysql_conn.cursor(dictionary=true)
        pg_cursor = self.pg_conn.cursor()
        total_rows = 0
        max_id = start_id
        # 分批查询 mysql
        if start_id:
            mysql_query = f"select {col_names_str} from {table} where {pk} > %s order by {pk} limit %s"
            params = (start_id, self.batch_size)
        else:
            mysql_query = f"select {col_names_str} from {table} limit %s"
            params = (self.batch_size,)
        logger.info(f"开始迁移表: {table}，起始 {pk}={start_id or 0}")
        while true:
            mysql_cursor.execute(mysql_query, params)
            rows = mysql_cursor.fetchall()
            if not rows:
                break
            # 数据类型转换
            converted_rows = []
            for row in rows:
                converted = tuple(
                    self._convert_value(row[c], c['data_type'])
                    for c in zip(col_names, columns)
                )
                converted_rows.append(converted)
            # 批量插入 postgresql
            try:
                pg_cursor.executemany(insert_sql.as_string(self.pg_conn), converted_rows)
                self.pg_conn.commit()
            except exception as e:
                logger.error(f"插入失败: {e}，尝试单条插入...")
                for row in converted_rows:
                    try:
                        pg_cursor.execute(insert_sql.as_string(self.pg_conn), row)
                    except exception as row_err:
                        logger.error(f"单条插入失败: {row_err}")
                self.pg_conn.commit()
            total_rows += len(rows)
            max_id = rows[-1][pk]
            logger.info(f"  {table}: 已迁移 {total_rows} 行，最新 {pk}={max_id}")
            self._save_checkpoint(table, max_id)
            if len(rows) < self.batch_size:
                break
            if start_id:
                params = (max_id, self.batch_size)
        mysql_cursor.close()
        pg_cursor.close()
        logger.info(f"表 {table} 迁移完成，共 {total_rows} 行")
        return total_rows
    def migrate_all(self):
        """迁移所有表"""
        self.connect()
        tables = self.get_tables()
        for table_info in tables:
            table = table_info['table_name']
            try:
                # 跳过系统表
                if table.startswith('_') or table in ('migration_log',):
                    continue
                self.migrate_table(table)
            except exception as e:
                logger.error(f"表 {table} 迁移失败: {e}")
                continue
        self.close()
        logger.info("全部迁移完成！")
if __name__ == '__main__':
    mysql_conf = {
        'host': '192.168.1.100',
        'port': 3306,
        'user': 'migration',
        'password': 'your_mysql_password',
        'database': 'source_app',
        'charset': 'utf8mb4',
    }
    pg_conf = {
        'host': '192.168.1.101',
        'port': 5432,
        'user': 'postgres',
        'password': 'your_pg_password',
        'database': 'target_app',
    }
    migrator = mysqltopostgresmigrator(
        mysql_conf=mysql_conf,
        pg_conf=pg_conf,
        checkpoint_file='.migration_checkpoint.json',
        batch_size=5000,
    )
    migrator.migrate_all()

#!/usr/bin/env python3
"""数据校验脚本：对比 mysql 和 postgresql 的行数和 checksum"""
import hashlib
import mysql.connector
import psycopg2
mysql_conf = {'host': '192.168.1.100', 'port': 3306, 
              'user': 'root', 'password': 'xxx', 'database': 'source_app'}
pg_conf = {'host': '192.168.1.101', 'port': 5432,
           'user': 'postgres', 'password': 'xxx', 'database': 'target_app'}
def table_checksum(cursor, table: str, pk: str) -> tuple[int, str]:
    """计算表的行数和 md5 checksum"""
    cursor.execute(f"select count(*) from {table}")
    count = cursor.fetchone()[0]
    cursor.execute(f"select {pk} from {table} order by {pk}")
    ids = [str(r[0]) for r in cursor.fetchall()]
    checksum = hashlib.md5('|'.join(ids).encode()).hexdigest()
    return count, checksum
mysql_conn = mysql.connector.connect(**mysql_conf)
pg_conn = psycopg2.connect(**pg_conf)
mysql_cursor = mysql_conn.cursor()
pg_cursor = pg_conn.cursor()
# 获取所有表
mysql_cursor.execute("""
    select table_name from information_schema.tables
    where table_schema = %s and table_type = 'base table'
""", (mysql_conf['database'],))
tables = [r[0] for r in mysql_cursor.fetchall()]
errors = []
for table in tables:
    # 获取主键
    pg_cursor.execute("""
        select a.attname from pg_index i
        join pg_attribute a on a.attrelid = i.indrelid and a.attnum = any(i.indkey)
        where i.indrelid = %s::regclass and i.indisprimary
    """, (table,))
    pk_result = pg_cursor.fetchone()
    if not pk_result:
        print(f"⚠ {table}: 无主键，跳过 checksum 校验")
        continue
    pk = pk_result[0]
    mysql_count, mysql_cs = table_checksum(mysql_cursor, table, pk)
    pg_count, pg_cs = table_checksum(pg_cursor, table, pk)
    status = "✅" if mysql_count == pg_count and mysql_cs == pg_cs else "❌"
    if status == "❌":
        errors.append(table)
    print(f"{status} {table}: mysql({mysql_count}, {mysql_cs[:8]}) vs pg({pg_count}, {pg_cs[:8]})")
mysql_cursor.close()
pg_cursor.close()
mysql_conn.close()
pg_conn.close()
if errors:
    print(f"\n❌ 校验失败表: {errors}")
else:
    print("\n✅ 所有表校验通过！")

-- ============= mysql 存储过程 =============
delimiter //

create procedure get_user_orders(
    in p_user_id bigint,
    in p_status varchar(20),
    out p_total_amount decimal(12,2)
)
begin
    declare v_count int default 0;

    select count(*) into v_count from orders
    where user_id = p_user_id and status = p_status;

    if v_count = 0 then
        select 'no orders found' as message;
    else
        select o.*, p.product_name
        from orders o
        join order_items oi on o.id = oi.order_id
        join products p on oi.product_id = p.id
        where o.user_id = p_user_id and o.status = p_status;

        select sum(total_amount) into p_total_amount
        from orders
        where user_id = p_user_id and status = p_status;
    end if;
end //

delimiter ;

-- ============= postgresql 等价实现 =============

-- 函数版本（推荐，postgresql 函数比过程更常用）
create or replace function get_user_orders(
    p_user_id bigint,
    p_status varchar(20)
) returns table (
    message text,
    order_id bigint,
    product_name text,
    total_amount numeric(12,2)
) as $$
declare
    v_count int;
begin
    select count(*) into v_count
    from orders
    where user_id = p_user_id and status = p_status;

    if v_count = 0 then
        return query select 'no orders found'::text, null::bigint, null::text, null::numeric(12,2);
    else
        return query
        select null::text, o.id, p.product_name, o.total_amount
        from orders o
        join order_items oi on o.id = oi.order_id
        join products p on oi.product_id = p.id
        where o.user_id = p_user_id and o.status = p_status;
    end if;
end;
$$ language plpgsql;

-- 调用
select * from get_user_orders(12345, 'completed');

-- 过程版本（postgresql 11+ 支持 create procedure）
create or replace procedure get_user_orders_proc(
    p_user_id bigint,
    p_status varchar(20),
    out p_total_amount numeric(12,2)
) as $$
declare
    v_count int;
begin
    select count(*) into v_count
    from orders
    where user_id = p_user_id and status = p_status;

    if v_count = 0 then
        raise notice 'no orders found for user %', p_user_id;
    else
        select sum(total_amount) into p_total_amount
        from orders
        where user_id = p_user_id and status = p_status;
    end if;
end;
$$ language plpgsql;

-- 调用（postgresql call 语法）
do $$
declare
    v_total numeric(12,2);
begin
    call get_user_orders_proc(12345, 'completed', v_total);
    raise notice 'total amount: %', v_total;
end $$;

-- ============= mysql 触发器 =============
create trigger tr_order_after_insert
after insert on orders
for each row
begin
    insert into order_audit (action, order_id, user_id, amount, created_at)
    values ('insert', new.id, new.user_id, new.total_amount, now());

    update user_stats set order_count = order_count + 1, total_spent = total_spent + new.total_amount
    where user_id = new.user_id;
end;

-- ============= postgresql 等价触发器 =============
create or replace function fn_order_audit_insert()
returns trigger as $$
begin
    insert into order_audit (action, order_id, user_id, amount, created_at)
    values ('insert', new.id, new.user_id, new.total_amount, now());

    update user_stats
    set order_count = order_count + 1,
        total_spent = total_spent + new.total_amount
    where user_id = new.user_id;

    return new;
end;
$$ language plpgsql;

create trigger tr_order_after_insert
after insert on orders
for each row
execute function fn_order_audit_insert();

-- 更新时间的自动更新触发器（mysql on update current_timestamp）
create or replace function fn_update_timestamp()
returns trigger as $$
begin
    new.updated_at = now();
    return new;
end;
$$ language plpgsql;

create trigger tr_users_update_timestamp
before update on users
for each row
execute function fn_update_timestamp();

-- mysql get_lock（分布式锁）
-- mysql:
-- get_lock('my_lock', 10);  -- 获取锁，超时10秒
-- do release_lock('my_lock');

-- postgresql: 使用 advisory lock（更强大的机制）
do $$
begin
    -- 尝试获取 advisory lock（key 为 bigint）
    if pg_try_advisory_lock(12345) then
        raise notice 'lock acquired!';
        -- 执行业务逻辑
        perform pg_advisory_unlock(12345);
    else
        raise notice 'lock not available, waiting...';
        -- 等待锁
        perform pg_advisory_lock(12345);
        raise notice 'lock acquired after waiting!';
        perform pg_advisory_unlock(12345);
    end if;
end $$;

-- postgresql advisory lock 批量模式
select pg_try_advisory_lock(1), pg_try_advisory_lock(2), pg_try_advisory_lock(3);
-- ... 业务逻辑 ...
select pg_advisory_unlock(1), pg_advisory_unlock(2), pg_advisory_unlock(3);

postgresql 索引类型
├── b-tree（默认，等价 mysql innodb b+tree）
│   ├── 适用: =, <, <=, >, >=, between, is null, like 前缀匹配
│   └── 场景: 主键、唯一索引、大多数普通查询
├── hash
│   ├── 适用: = 精确匹配（仅等值查询）
│   ├── 优点: 比 b-tree 体积小，查找快
│   └── 场景: uuid、临时表、不需要范围查询的字段
├── gin（generalized inverted index）
│   ├── 适用: 数组、jsonb、全文搜索
│   ├── 优点: 一条索引覆盖多值字段
│   └── 场景: 标签查询、商品属性搜索
├── gist（generalized search tree）
│   ├── 适用: 几何类型、范围类型、全文搜索、postgis
│   └── 场景: 地理位置范围查询
├── brin（block range index）
│   ├── 适用: 物理顺序与逻辑顺序一致的大表（时序数据）
│   ├── 优点: 索引极小，适合 tb 级表
│   └── 场景: 日志表、传感器数据、订单表（按时间分区）
└── sp-gist（space-partitioned gist）
    ├── 适用: 非平衡数据结构（树、网格）
    └── 场景: 字典树、四叉树、k-d 树

              ┌─────────────────────────┐
              │  什么类型的查询？        │
              └───────────┬─────────────┘
                          │
           ┌──────────────┼──────────────┐
           ▼              ▼              ▼
      等值比较        范围/排序       多值/特殊
           │              │              │
     ┌─────┴─────┐    ┌────┴────┐    ┌────┴─────┐
     │ like 'x%'?│    │between等?│    │ jsonb/数组│
     └─────┬─────┘    └────┬────┘    └────┬─────┘
           │              │              │
           ▼              ▼              ▼
      b-tree(默认)    b-tree(默认)    gin
                      或 brin
                      
     ┌─────────────────────────────────────────┐
     │ 时序数据 / 按时间分区的表 → 优先 brin     │
     │ 数据量 > 100gb 且按时间顺序插入 → brin   │
     │ 地理位置数据 → gist + postgis            │
     │ jsonb 多条件查询 → gin 索引              │
     │ 只需要 = 比较 → hash 索引（体积更小）     │
     └─────────────────────────────────────────┘

-- 测试数据: users 表 1000万行，每行有 user_data jsonb 字段
-- 查询: where user_data->>'city' = 'beijing'

-- 无索引
explain (analyze, buffers) 
select * from users where user_data->>'city' = 'beijing';

-- 添加 gin 索引后
create index idx_users_city on users using gin ((user_data jsonb_path_ops));

-- orders 表 1亿行，按 created_at 时间顺序插入
-- 数据量: 约 80gb

-- b-tree 索引（常规做法）
create index idx_orders_created_btree on orders(created_at);
-- 索引大小: ~3.2gb

-- brin 索引（利用物理顺序）
create index idx_orders_created_brin on orders using brin(created_at);
-- 索引大小: ~128kb（仅为 b-tree 的 0.004%）

-- 查询性能对比: select * from orders where created_at between '2026-01-01' and '2026-01-31'

-- 1. 查看慢查询的真正执行计划
explain (analyze, buffers, format json)
select o.*, u.username from orders o
join users u on o.user_id = u.id
where o.status = 'pending' and o.created_at > now() - interval '7 days';

-- 2. postgresql 特有: 部分索引（mysql 不支持）
-- 只索引活跃订单，大幅减少索引体积
create index idx_orders_pending on orders(created_at)
where status = 'pending';

-- 3. postgresql 特有: 表达式索引
create index idx_users_email_lower on users (lower(email));

-- 4. 覆盖索引（index only scan）
create index idx_users_covering on users (user_id) include (username, email);
-- mysql: 在索引列后直接加 include
-- postgresql: 用 include（pg 11+）或表达式模拟

-- 5. 监控索引使用情况
select
    schemaname,
    tablename,
    indexname,
    idx_scan,
    idx_tup_read,
    idx_tup_fetch,
    pg_size_pretty(pg_relation_size(indexrelid)) as index_size
from pg_stat_user_indexes
order by idx_scan asc;

-- 找出从未被使用的索引（可删除节省空间）
select
    schemaname || '.' || tablename as table_name,
    indexname,
    pg_size_pretty(pg_relation_size(indexrelid)) as index_size
from pg_stat_user_indexes
where idx_scan = 0
  and indexrelid not in (select conindid from pg_constraint)
order by pg_relation_size(indexrelid) desc;

-- 6. 重建臃肿索引（postgresql 索引会膨胀）
reindex index concurrently idx_users_email;
vacuum analyze;

# postgresql.conf 关键参数调优（8核32gb机器示例）
# === 内存参数 ===
shared_buffers = 8gb              # 推荐 25% 物理内存
effective_cache_size = 24gb        # 推荐 75% 物理内存（留给 os 缓存）
work_mem = 256mb                   # 每个排序/哈希操作内存上限
maintenance_work_mem = 2gb         # vacuum/reindex/建索引时可用内存
# === 并行查询 ===
max_worker_processes = 8
max_parallel_workers_per_gather = 4
max_parallel_workers = 8
max_parallel_maintenance_workers = 4
parallel_tuple_cost = 0.1
parallel_setup_cost = 1000
# === 写入性能 ===
wal_buffers = 64mb
checkpoint_completion_target = 0.9
max_wal_size = 4gb
min_wal_size = 1gb
# === 连接数（配合 pgbouncer 使用）===
max_connections = 200              # 直接连 pg 的应用连接上限
# 生产环境推荐: max_connections 设小，用 pgbouncer 分 2000 个虚拟连接
# === autovacuum（重要！mysql 无此概念）===
autovacuum_max_workers = 4
autovacuum_naptime = 30s
autovacuum_vacuum_threshold = 50
autovacuum_analyze_threshold = 50

# pgbouncer pgbouncer.ini 配置（连接池）
[databases]
target_app = host=127.0.0.1 port=5432 dbname=target_app
[pgbouncer]
listen_addr = 127.0.0.1
listen_port = 6432
auth_type = md5
auth_file = /etc/pgbouncer/userlist.txt
pool_mode = transaction          # 事务级连接池（推荐），session 模式更保守
max_client_conn = 5000
default_pool_size = 50          # 每个 db/user 的连接数
min_pool_size = 10
reserve_pool_size = 5
reserve_pool_timeout = 3
server_idle_timeout = 600

-- ⚠️ mysql 的 utf8mb4 与 postgresql 的 utf8 行为差异

-- mysql 8.0（默认排序规则 utf8mb4_0900_ai_ci）
create table users (
    name varchar(50) character set utf8mb4 collate utf8mb4_0900_ai_ci
);

-- postgresql（默认排序规则继承数据库/集群级别设置）
-- 推荐创建数据库时指定:
create database target_app
    with encoding = 'utf8'
    lc_collate = 'en_us.utf8'      -- 影响 order by 和字符串比较
    lc_ctype = 'en_us.utf8'       -- 影响字符分类（大写/小写/数字）
    template = template0;         -- 避免模板数据库污染

-- 中文环境推荐:
create database target_app
    with encoding = 'utf8'
    lc_collate = 'c'
    lc_ctype = 'c'
    template = template0;

-- 关键差异：postgresql 区分大小写！
select * from users where username = 'alice';  -- 大小写敏感！
select * from users where username ilike 'alice';  -- 不区分大小写

-- 解决方案：使用不区分大小写的排序规则或创建表达式索引
create index idx_users_username_ci on users ((lower(username)));
create extension if not exists citext;  -- citext 类型，自动不区分大小写

-- 修改字段为 citext 类型
alter table users alter column username type citext using username::citext;

-- mysql: 表名/列名在 windows/macos 不区分大小写，linux 区分
-- postgresql: 总是区分大小写（除非用双引号包裹）

-- 问题：mysql 表名 user_orders 在 pg 中必须完全匹配
select * from user_orders;  -- ❌ 报错: relation "user_orders" does not exist
select * from user_orders;  -- ✅ 正确

-- 解决方案：统一使用小写下划线命名
-- 建议在迁移脚本中加入: table_name.lower().replace('-', '_')

-- 函数名、索引名同理

-- mysql 分页（传统方式）
select * from orders order by id limit 1000000, 20;
-- mysql 在 id 上有索引时，优化器可以跳过前100万行

-- postgresql offset 性能陷阱（深分页极慢）
select * from orders order by id limit 20 offset 1000000;
-- postgresql 必须扫描 1000020 行才能返回结果！

-- 解决方案 1：游标分页（推荐，性能恒定）
select * from orders
where id > :last_seen_id
order by id
limit 20;

-- 解决方案 2：keyset 分页（适合前端无限滚动）
-- 应用层传入上一页最后一条的 id/cursor

-- 解决方案 3：如果必须用 offset，加子查询（利用索引覆盖）
select * from (
    select id from orders order by id limit 20 offset 1000000
) as page_ids
join orders o using (id);

-- 解决方案 4：使用 offset + 已知边界（分批处理）
do $$
declare
    offset_val bigint := 0;
    batch_size int := 10000;
begin
    for i in 1..100 loop
        -- 每次处理 10000 行
        perform do_something_with_batch(offset_val, batch_size);
        offset_val := offset_val + batch_size;
        raise notice 'processed offset %', offset_val;
    end loop;
end $$;

-- mysql: null = null 结果为 true（违反 sql 标准）
-- postgresql: null = null 结果为 null（标准行为）

-- 这会导致大量 where 条件失效！
select * from users where deleted_at = null;  -- ❌ postgresql 中永远返回空

-- 必须写成:
select * from users where deleted_at is null;

-- 批量替换技巧（mysql → postgresql 迁移后）
update users set deleted_at = null where deleted_at = '0000-00-00 00:00:00';  -- 清掉 mysql 的 0 时间戳
update users set deleted_at = null where deleted_at = '1970-01-01';         -- 清掉 unix 纪元

-- 重要：mysql 的 group by 不要求 select 字段在 group by 中
-- postgresql 严格模式要求所有 select 字段要么在 group by 中，要么是聚合函数

-- mysql（可能报错）:
-- select status, count(*), created_at from orders group by status;
-- postgresql（严格）:
select status, count(*), min(created_at) from orders group by status;
-- 或关闭严格模式（不推荐）:
set sql_mode = '';

-- mysql: 默认 autocommit=1（每条语句自动提交）
set autocommit = 0;  -- 手动开启事务

-- postgresql: 默认 autocommit=off（在事务块内才提交）
-- 单条语句也在事务中，但无需 begin

begin;  -- postgresql 中 begin 是可选的（但显式写是好习惯）
select * from orders where id = 1;
update orders set status = 'shipped' where id = 1;
commit;

-- ⚠️ postgresql 没有 rollback to savepoint 以外的回滚点语法（部分支持）
savepoint my_savepoint;
-- ... some work ...
rollback to savepoint my_savepoint;  -- ✅ 支持
release savepoint my_savepoint;

-- mysql timestamp 在不同版本行为不一致（取决于 time_zone 设置）
-- postgresql timestamptz 明确存储 utc，转换由客户端负责

-- 推荐：统一使用 timestamptz
create table events (
    id bigserial primary key,
    event_time timestamptz not null default now(),
    created_at timestamptz not null default now()
);

-- 统一时区显示
set time zone 'asia/shanghai';

-- 日期计算差异
-- mysql:
-- select date_sub(now(), interval 7 day);

-- postgresql:
select now() - interval '7 days';
select current_timestamp at time zone 'asia/shanghai';

-- ⚠️ postgresql 大多数 ddl 操作会锁表！
create index idx_orders_user_id on orders(user_id);  -- 阻塞写操作

-- ✅ 解决方案：concurrently（不锁表，但速度慢 2-3 倍）
create index concurrently idx_orders_user_id on orders(user_id);
drop index concurrently idx_orders_user_id;

-- ✅ 迁移阶段推荐：一次性重建所有索引
-- 1. 先创建所有索引为 concurrently（不加唯一约束）
-- 2. 迁移数据
-- 3. 最后加唯一约束和外键

-- ⚠️ alter table 也可能锁表，pg 15+ 支持更多并发操作
alter table orders add column tracking_no varchar(50);  -- pg 11+ 不锁表
alter table orders alter column status type varchar(20);  -- 可能锁表
-- 替代方案（大表）:
alter table orders alter column status type varchar(20) using status::varchar(20);
-- pg 15+ 允许 using 表达式避免全表重写（如果兼容的话）

阶段 1: 并行写入期（1-2周）
┌─────────────┐     ┌─────────────┐
│  mysql ←────│─────│  应用层     │────→  postgresql
│  (主库)     │     │  双写逻辑   │     (新库, 只写)
└─────────────┘     └─────────────┘
     ↑                      ↑
     │                      │
  读取                  增量校验

阶段 2: 读流量灰度（逐步切 5% → 20% → 50% → 100%）
┌─────────────┐     ┌─────────────┐
│  mysql      │◄────│  灰度网关   │────→ postgresql
│  (95%读)    │     │  (权重分流) │     (5%-100%读)
└─────────────┘     └─────────────┘

阶段 3: 写流量切流（确认无误后）
┌─────────────┐     ┌─────────────┐
│  mysql      │◄────│  切流开关   │────→ postgresql
│  (备用回滚) │     │  (一键切换) │     (100%写)
└─────────────┘     └─────────────┘

阶段 4: 观察期（1周）+ 数据回填校验

#!/usr/bin/env python3
"""灰度切流控制器（伪代码）"""

from enum import enum
import random
import hashlib
import time

class dbmode(enum):
    mysql = "mysql"
    pg = "postgres"
    dual = "dual_write"

class trafficrouter:
    """基于用户 id / 请求 id 的确定性灰度路由"""

    def __init__(self, pg_read_weight: float = 0.0):
        """
        pg_read_weight: postgresql 读流量占比 (0.0 ~ 1.0)
        """
        self.pg_read_weight = pg_read_weight

    def should_use_pg(self, user_id: str, operation: str) -> bool:
        """决定是否使用 postgresql（确定性路由，同一 user_id 结果一致）"""
        if operation == "write":
            # 写入期：mysql 写，postgresql 同步写
            return true  # 双写模式下两边都写

        # 读取：根据灰度权重
        key = f"{user_id}:{operation}"
        hash_val = int(hashlib.md5(key.encode()).hexdigest(), 16)
        return (hash_val % 100) < (self.pg_read_weight * 100)

    def route_query(self, user_id: str, sql: str) -> dbmode:
        """根据路由决定查询走哪个数据库"""
        if self.should_use_pg(user_id, "read"):
            return dbmode.pg
        return dbmode.mysql

    def route_write(self, sql: str) -> list[dbmode]:
        """写入路由（双写期两边都写）"""
        return [dbmode.mysql, dbmode.pg]


# 使用示例
router = trafficrouter(pg_read_weight=0.2)  # 20% 读流量切 pg

def execute_read(user_id: str, sql: str):
    target = router.route_query(user_id, sql)
    if target == dbmode.pg:
        return pg_execute(sql)
    return mysql_execute(sql)

def execute_write(sql: str):
    results = []
    for target in router.route_write(sql):
        if target == dbmode.mysql:
            results.append(("mysql", mysql_execute(sql)))
        else:
            results.append(("pg", pg_execute(sql)))

    # 校验双写结果一致性（可选）
    compare_results(results)
    return results[0]  # 返回主库结果

# 灰度升级脚本
def upgrade_pg_read_traffic(target_weight: float, step: float = 0.1, interval: int = 3600):
    """渐进式提升 pg 读流量权重"""
    current = 0.0
    while current < target_weight:
        current = min(current + step, target_weight)
        router.pg_read_weight = current
        print(f"[{time.strftime('%h:%m:%s')}] pg 读流量: {current*100:.0f}%")
        time.sleep(interval)

-- 1. 实时数据对比视图（mysql 侧运行）
create or replace view v_data_consistency as
select 
    'orders' as table_name,
    (select count(*) from orders) as mysql_count,
    (select count(*) from orders_pg) as pg_count,
    (select count(*) from orders o
     left join orders_pg p on o.id = p.id
     where p.id is null) as mysql_only_count,
    (select count(*) from orders_pg p
     left join orders o on p.id = o.id
     where o.id is null) as pg_only_count
union all
select 
    'users',
    (select count(*) from users),
    (select count(*) from users_pg),
    (select count(*) from users u left join users_pg p on u.id = p.id where p.id is null),
    (select count(*) from users_pg p left join users u on p.id = u.id where u.id is null);

select * from v_data_consistency;

□ 数据量 < 50gb，sql 复杂度低 → pgloader 即可完成
□ 数据量 50gb~1tb，零停机 → aws dms 或 pgloader + 双写
□ 数据量 > 1tb → debezium cdc 方案
□ 有大量存储过程/触发器 → 预留 40% 额外迁移时间
□ mysql 特有语法多 → 先评估重写工作量
□ json 查询密集 → postgresql jsonb 收益巨大
□ 地理位置需求 → postgis 绝对值得迁移
□ 时序数据 → timescaledb 扩展可大幅简化分区管理