如何将Oracle查询结果转为动态结构插入PostgreSQL？

核心思路 #

无需转换为动态结构体，通过获取表元数据+动态构造SQL+动态参数绑定即可实现。利用Oracle的列信息生成适配PostgreSQL的插入逻辑，直接以[]interface{}切片存储每行数据，配合pgx的批量插入能力完成迁移。

步骤1：获取Oracle表的列信息 #

先查询目标表的列名，用于后续构造查询和插入语句：

import (
    "fmt"
    "strings"
    "github.com/sijms/go-ora/v2"
    "github.com/jackc/pgx/v5"
    "context"
)

func getOracleColumns(conn *go-ora.Connection, tableName string) ([]string, error) {
    query := fmt.Sprintf(`
        SELECT column_name 
        FROM all_tab_columns 
        WHERE table_name = UPPER('%s') 
        ORDER BY column_id
    `, tableName)
    rows, err := conn.Query(query)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var columns []string
    for rows.Next() {
        var colName string
        if err := rows.Scan(&colName); err != nil {
            return nil, err
        }
        columns = append(columns, colName)
    }
    return columns, rows.Err()
}

步骤2：从Oracle读取全量动态数据 #

根据列名构造查询语句，逐行读取数据并存储为[][]interface{}：

func fetchOracleData(conn *go-ora.Connection, tableName string, columns []string) ([][]interface{}, error) {
    colStr := strings.Join(columns, ", ")
    query := fmt.Sprintf("SELECT %s FROM %s", colStr, tableName)
    rows, err := conn.Query(query)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var data [][]interface{}
    for rows.Next() {
        rowVals := make([]interface{}, len(columns))
        rowPtrs := make([]interface{}, len(columns))
        for i := range rowVals {
            rowPtrs[i] = &rowVals[i]
        }
        if err := rows.Scan(rowPtrs...); err != nil {
            return nil, err
        }

        // 处理Oracle特殊类型（如CLOB）
        for i, val := range rowVals {
            if clob, ok := val.(go-ora.CLOB); ok {
                strVal, err := clob.ReadAll()
                if err != nil {
                    return nil, err
                }
                rowVals[i] = string(strVal)
            }
        }
        data = append(data, rowVals)
    }
    return data, rows.Err()
}

步骤3：动态插入PostgreSQL #

推荐使用pgx的CopyFrom实现高性能批量插入，也可构造动态INSERT语句处理小数据量：

方式1：高性能批量插入（CopyFrom） #

func insertIntoPG(ctx context.Context, pgConn *pgx.Conn, tableName string, columns []string, data [][]interface{}) error {
    colIdentifiers := make([]pgx.Identifier, len(columns))
    for i, col := range columns {
        colIdentifiers[i] = pgx.Identifier{col}
    }

    rows := pgx.CopyFromRows(data)
    _, err := pgConn.CopyFrom(ctx, colIdentifiers, tableName, rows)
    return err
}

方式2：动态INSERT语句（小数据量） #

func insertWithBatch(ctx context.Context, pgConn *pgx.Conn, tableName string, columns []string, data [][]interface{}) error {
    placeholders := make([]string, len(columns))
    for i := range placeholders {
        placeholders[i] = fmt.Sprintf("$%d", i+1)
    }
    insertQuery := fmt.Sprintf(
        "INSERT INTO %s (%s) VALUES (%s)",
        pgx.Identifier{tableName}.Sanitize(),
        strings.Join(columns, ", "),
        strings.Join(placeholders, ", "),
    )

    batch := &pgx.Batch{}
    for _, row := range data {
        batch.Queue(insertQuery, row...)
    }
    results := pgConn.SendBatch(ctx, batch)
    defer results.Close()

    for i := 0; i < len(data); i++ {
        _, err := results.Exec()
        if err != nil {
            return fmt.Errorf("行%d插入失败: %w", i, err)
        }
    }
    return nil
}

完整调用流程 #

func main() {
    // 连接Oracle
    oracleDSN := "user/password@host:port/service_name"
    oracleConn, err := go-ora.NewConnection(oracleDSN)
    if err != nil {
        panic(err)
    }
    defer oracleConn.Close()
    if err := oracleConn.Open(); err != nil {
        panic(err)
    }

    // 连接PostgreSQL
    pgDSN := "postgres://user:password@host:port/dbname?sslmode=disable"
    ctx := context.Background()
    pgConn, err := pgx.Connect(ctx, pgDSN)
    if err != nil {
        panic(err)
    }
    defer pgConn.Close(ctx)

    // 迁移目标表
    tableName := "your_target_table"

    // 执行迁移
    columns, err := getOracleColumns(oracleConn, tableName)
    if err != nil {
        panic(err)
    }
    data, err := fetchOracleData(oracleConn, tableName, columns)
    if err != nil {
        panic(err)
    }
    if err := insertIntoPG(ctx, pgConn, tableName, columns, data); err != nil {
        panic(err)
    }

    fmt.Println("数据迁移完成")
}

关键注意事项 #

• 表结构兼容：确保PostgreSQL目标表已创建，列名、数据类型与Oracle匹配（可扩展逻辑自动创建PG表）
• 大表分页：数据量过大时，用Oracle的ROWNUM或FETCH NEXT分页查询，避免内存溢出
• 事务保障：迁移过程中开启PostgreSQL事务，确保数据一致性
• 类型适配：手动处理Oracle特殊类型（如CLOB、BLOB），转换为PostgreSQL兼容格式

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