PostgreSQL跨库复制多关联表查询结果并同步Schema的最佳实践

Hey there, based on your requirement to migrate schema and large datasets from sourceDB to an empty destinationDB (both PostgreSQL 9.6.4 on the same host) via a Java program, here's a step-by-step solution that balances accuracy and efficiency:

1. 先同步数据库Schema #

Since destinationDB has no existing schema, we first need to replicate all table structures, constraints, indexes, and sequences from sourceDB. The easiest and most reliable way is to use PostgreSQL's built-in metadata functions instead of manually building DDL statements (which can miss edge cases like foreign keys or unique constraints).

实现步骤： #

• Connect to sourceDB via JDBC, query all user-defined tables (filter out system tables like pg_catalog or information_schema).
• For each table, use pg_get_tabledef() to fetch the complete CREATE TABLE statement (including constraints and indexes).
• Execute these DDL statements in destinationDB to replicate the schema.

2. 高效迁移大表数据 #

For large associated tables, we need to avoid loading all data into memory at once. Two efficient approaches are available:

方案A：JDBC分批读取+批量插入 #

• Process data in batches (e.g., 1000 rows per batch) to prevent memory overflow.
• Insert batches using PreparedStatement.addBatch() and executeBatch() for better performance than single-row inserts.
• Critical note: Migrate parent tables first (those without foreign key dependencies), then child tables to avoid constraint violations.

方案B：使用PostgreSQL COPY命令（推荐） #

PostgreSQL's COPY command is far faster than JDBC batch inserts for large datasets. We can use the PostgreSQL JDBC extension's CopyManager to stream data directly between databases without writing to intermediate files.

3. 完整Java代码示例 #

依赖准备 #

First, add the PostgreSQL JDBC driver to your project (compatible with 9.6.4):

<!-- Maven dependency -->
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.2.23</version> <!-- Stable version for PostgreSQL 9.6 -->
</dependency>

迁移代码 #

import org.postgresql.copy.CopyManager;
import org.postgresql.core.BaseConnection;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

public class PostgresDbMigration {
    // 源数据库配置
    private static final String SOURCE_DB_URL = "jdbc:postgresql://localhost:5432/sourceDB";
    private static final String SOURCE_DB_USER = "your_username";
    private static final String SOURCE_DB_PWD = "your_password";
    
    // 目标数据库配置
    private static final String DEST_DB_URL = "jdbc:postgresql://localhost:5432/destinationDB";
    private static final String DEST_DB_USER = "your_username";
    private static final String DEST_DB_PWD = "your_password";

    public static void main(String[] args) throws Exception {
        // 加载PostgreSQL驱动
        Class.forName("org.postgresql.Driver");
        
        try (Connection sourceConn = DriverManager.getConnection(SOURCE_DB_URL, SOURCE_DB_USER, SOURCE_DB_PWD);
             Connection destConn = DriverManager.getConnection(DEST_DB_URL, DEST_DB_USER, DEST_DB_PWD)) {
            
            // Step 1: 同步Schema
            syncDatabaseSchema(sourceConn, destConn);
            
            // Step 2: 同步数据
            syncTableData(sourceConn, destConn);
            
            System.out.println("Migration completed successfully!");
        } catch (Exception e) {
            System.err.println("Migration failed: " + e.getMessage());
            throw e;
        }
    }

    private static void syncDatabaseSchema(Connection sourceConn, Connection destConn) throws Exception {
        try (Statement sourceStmt = sourceConn.createStatement()) {
            // 获取所有用户自定义表
            ResultSet tableRs = sourceStmt.executeQuery(
                "SELECT table_schema, table_name FROM information_schema.tables " +
                "WHERE table_type = 'BASE TABLE' AND table_schema NOT IN ('pg_catalog', 'information_schema')"
            );
            
            try (Statement destStmt = destConn.createStatement()) {
                while (tableRs.next()) {
                    String schema = tableRs.getString("table_schema");
                    String tableName = tableRs.getString("table_name");
                    String qualifiedTableName = String.format("%s.%s", schema, tableName);
                    
                    // 获取表的完整DDL语句
                    ResultSet ddlRs = sourceStmt.executeQuery(String.format("SELECT pg_get_tabledef('%s')", qualifiedTableName));
                    if (ddlRs.next()) {
                        String createTableDdl = ddlRs.getString(1);
                        destStmt.execute(createTableDdl);
                        System.out.println("Created table: " + qualifiedTableName);
                    }
                    ddlRs.close();
                }
            }
            tableRs.close();
        }
        // 可选：同步序列、视图等，使用pg_get_sequence_def()、pg_get_viewdef()类似逻辑
    }

    private static void syncTableData(Connection sourceConn, Connection destConn) throws Exception {
        // 注意：这里需要按表的依赖顺序排序（先父表后子表），示例中简化为获取所有表
        try (Statement sourceStmt = sourceConn.createStatement()) {
            ResultSet tableRs = sourceStmt.executeQuery(
                "SELECT table_schema, table_name FROM information_schema.tables " +
                "WHERE table_type = 'BASE TABLE' AND table_schema NOT IN ('pg_catalog', 'information_schema')"
            );
            
            CopyManager copyManager = new CopyManager((BaseConnection) destConn);
            
            while (tableRs.next()) {
                String schema = tableRs.getString("table_schema");
                String tableName = tableRs.getString("table_name");
                String qualifiedTableName = String.format("%s.%s", schema, tableName);
                
                // 查询源表数据
                String selectQuery = String.format("SELECT * FROM %s", qualifiedTableName);
                try (ResultSet dataRs = sourceStmt.executeQuery(selectQuery)) {
                    // 使用COPY命令流式写入目标表
                    String copyCommand = String.format("COPY %s FROM STDIN WITH (FORMAT BINARY)", qualifiedTableName);
                    copyManager.copyIn(copyCommand, dataRs);
                    System.out.println("Copied data for table: " + qualifiedTableName);
                }
            }
            tableRs.close();
        }
    }
}

4. 关键优化与注意事项 #

• 外键约束: To speed up migration, you can temporarily disable foreign key constraints in destinationDB before inserting data, then re-enable them afterward. Use ALTER TABLE table_name DISABLE TRIGGER ALL; and ALTER TABLE table_name ENABLE TRIGGER ALL;.
• Transaction Control: Wrap schema creation and data insertion in transactions to ensure atomicity—if any step fails, roll back the entire process.
• Batch Size: For JDBC batch inserts, adjust the batch size (e.g., 1000-5000 rows) based on your memory capacity.
• Large Tables: For extremely large tables, split the data into chunks using a range query (e.g., WHERE id BETWEEN x AND y) to avoid long-running queries.

内容的提问来源于stack exchange，提问作者Peter