RuoYi从MySQL迁移到PostgreSQL数据库的踩坑实录

org.postgresql.util.psqlexception: error: relation "dual" does not exist

# 原来的配置
validationquery: select 1 from dual

# 改成这样就行
validationquery: select 1

# coding=utf-8
import pymysql
import psycopg2
from psycopg2 import extras

# mysql配置
mysql_config = {
    'host': '192.168.1.100',
    'port': 3306,
    'user': 'root',
    'password': 'your_password',
    'database': 'ruoyi',
    'charset': 'utf8mb4'
}

# postgresql配置
pg_config = {
    'host': '192.168.1.101',
    'port': 5432,
    'user': 'postgres',
    'password': 'your_password',
    'database': 'ruoyi'
}

def get_all_tables(mysql_conn):
    """获取mysql中所有表名"""
    cursor = mysql_conn.cursor()
    cursor.execute("show tables")
    tables = [table[0] for table in cursor.fetchall()]
    cursor.close()
    return tables

def get_table_structure(mysql_conn, table_name):
    """获取表的列信息"""
    cursor = mysql_conn.cursor()
    cursor.execute(f"describe {table_name}")
    columns = cursor.fetchall()
    cursor.close()
    return columns

def migrate_table(mysql_conn, pg_conn, table_name):
    """迁移单个表的数据"""
    print(f"正在迁移表: {table_name}")
    
    # 获取mysql数据
    mysql_cursor = mysql_conn.cursor()
    mysql_cursor.execute(f"select * from {table_name}")
    rows = mysql_cursor.fetchall()
    
    if not rows:
        print(f"  表 {table_name} 为空，跳过")
        mysql_cursor.close()
        return
    
    # 获取列名
    columns = [desc[0] for desc in mysql_cursor.description]
    mysql_cursor.close()
    
    # 构建insert语句
    placeholders = ','.join(['%s'] * len(columns))
    columns_str = ','.join(columns)
    insert_sql = f"insert into {table_name} ({columns_str}) values ({placeholders})"
    
    # 批量插入到postgresql
    pg_cursor = pg_conn.cursor()
    try:
        # 使用execute_batch提高性能
        extras.execute_batch(pg_cursor, insert_sql, rows, page_size=1000)
        pg_conn.commit()
        print(f"  成功迁移 {len(rows)} 条记录")
    except exception as e:
        pg_conn.rollback()
        print(f"  迁移失败: {str(e)}")
        # 如果批量插入失败，尝试逐条插入（方便定位问题）
        error_count = 0
        for row in rows:
            try:
                pg_cursor.execute(insert_sql, row)
                pg_conn.commit()
            except exception as e:
                pg_conn.rollback()
                error_count += 1
                if error_count <= 5:  # 只打印前5个错误
                    print(f"    错误记录: {row[:3]}... 错误: {str(e)}")
        
        if error_count > 0:
            print(f"  共有 {error_count} 条记录插入失败")
    finally:
        pg_cursor.close()

def main():
    # 连接数据库
    print("连接mysql...")
    mysql_conn = pymysql.connect(**mysql_config)
    
    print("连接postgresql...")
    pg_conn = psycopg2.connect(**pg_config)
    
    try:
        # 获取所有表
        tables = get_all_tables(mysql_conn)
        print(f"共找到 {len(tables)} 个表\n")
        
        # 逐个迁移
        for i, table in enumerate(tables, 1):
            print(f"[{i}/{len(tables)}] ", end='')
            migrate_table(mysql_conn, pg_conn, table)
        
        print("\n所有表迁移完成！")
        
    finally:
        mysql_conn.close()
        pg_conn.close()
        print("数据库连接已关闭")

if __name__ == '__main__':
    main()

pip install pymysql psycopg2-binary

python migrate_data.py

-- mysql这样写没问题
select `user_id`, `user_name` from sys_user

-- 到pg就报错了
error: syntax error at or near "`"

-- mysql
select ifnull(perms, '') as perms from sys_menu

-- postgresql
select coalesce(perms, '') as perms from sys_menu

-- mysql
update sys_user set update_time = sysdate()

-- postgresql要换成这个
update sys_user set update_time = current_timestamp

-- mysql里这样写没问题，字符类型的status字段可以和数字比较
where status = 0

-- postgresql直接报错
error: operator does not exist: character = integer

where status = '0'

-- mysql的写法
where find_in_set(#{deptid}, ancestors)

-- postgresql的替代方案
where (',' || ancestors || ',') like '%,' || #{deptid} || ',%'

graph lr
    a["ancestors: '100,101,103'"] --> b["加工: ',100,101,103,'"]
    b --> c["匹配: '%,101,%'"]
    c --> d["结果: true"]

where #{deptid} = any(ancestors)

-- mysql
where date_format(create_time, '%y%m%d') >= date_format(#{params.begintime}, '%y%m%d')

-- postgresql得这么写
where to_char(create_time, 'yyyymmdd') >= to_char(to_timestamp(#{params.begintime}, 'yyyy-mm-dd'), 'yyyymmdd')

-- mysql
where table_schema = (select database())

-- postgresql
where table_schema = (select current_database())

-- mysql
truncate table sys_job_log

-- postgresql要加上这个选项
truncate table sys_job_log restart identity

error: duplicate key value violates unique constraint "sys_user_pkey"
详细：key (user_id)=(2) already exists.

do $$
declare
    r record;
    max_id bigint;
begin
    -- 遍历所有带序列的表
    for r in 
        select 
            t.tablename,
            c.column_name,
            pg_get_serial_sequence(t.schemaname || '.' || t.tablename, c.column_name) as sequence_name
        from 
            pg_tables t
            join information_schema.columns c on t.tablename = c.table_name
        where 
            t.schemaname = 'public'
            and pg_get_serial_sequence(t.schemaname || '.' || t.tablename, c.column_name) is not null
    loop
        -- 获取表的最大id
        execute format('select coalesce(max(%i), 0) from %i', r.column_name, r.tablename) into max_id;
        
        if max_id > 0 then
            -- 把序列设置为最大id
            execute format('select setval(%l, %s)', r.sequence_name, max_id);
            raise notice '已修复表 %.%, 序列: %, 当前最大id: %', 
                'public', r.tablename, r.sequence_name, max_id;
        else
            -- 如果表是空的，序列从1开始
            execute format('select setval(%l, 1, false)', r.sequence_name);
            raise notice '表 %.% 为空, 序列 % 已重置为从1开始', 
                'public', r.tablename, r.sequence_name;
        end if;
    end loop;
    
    raise notice '=== 所有序列修复完成 ===';
end $$;

pagehelper:
  helperdialect: postgresql  # 指定数据库方言
  supportmethodsarguments: true
  params: count=countsql

[system.io.file]::writealltext($file.fullname, $content, [system.text.encoding]::utf8)

mvn clean compile -dskiptests

# 1. 替换sysdate()
查找: sysdate\(\)
替换: current_timestamp

# 2. 替换ifnull()
查找: ifnull\(
替换: coalesce(

# 3. 移除反引号
查找: `([a-z_]+)`
替换: $1

# 4. 替换database()
查找: select database\(\)
替换: select current_database()