从Oracle向PostgreSQL插入数据时出现22P02错误求助

我正在开发一个API，用于将数据从Oracle迁移至PostgreSQL。流程是从Oracle查询数据存入DataTable，再生成Insert SQL到PostgreSQL执行。但表中包含numeric或timestamp类型字段时，会抛出**"22P02: invalid input syntax for type numeric"**错误。以下是我的代码：

public void Oracle2PG(string fromTable, string toTable)
{
    string pgConn = _configuration.GetValue<string>($"ConnectionStrings:PGConnection");
    string oraConn = _configuration.GetValue<string>($"ConnectionStrings:OracleConnection");

    using (var pgConn = new NpgsqlConnection(sctPgConn))
    using (var oraConn = new OracleConnection(sctOraConn))
    {
        string oraSql = $"select * from {fromTable}";
        var dtOra = GetDatatable(oraSql, "", oraConn);
        string insertSql = GenerateInsertSql(toTable, dtOra);
        pgConn.Execute(insertSql); //occur error "22P02: invalid input syntax for type numeric"
    }
}

private static DataTable GetDatatable(string sql, object para, OracleConnection oraConn)
{
    var res = oraConn.Query(sql, para);
    string json = JsonConvert.SerializeObject(res);
    var dtResult = (DataTable)JsonConvert.DeserializeObject(json, (typeof(DataTable)));

    return dtResult;
}

private static string GenerateInsertSql(string tableName, DataTable dataTable)
{
    var cols = (from DataColumn column in dataTable.Columns select column.ColumnName).ToList();
    string colNames = string.Join(", ", cols);
    string sql = $"Insert into {tableName} ({colNames}) values";
    int cnt = 0;
    foreach (DataRow row in dataTable.Rows)
    {
        cnt += 1;
        sql += $"(";
        var rowValues = row.ItemArray.Select(s => s.ToString()).ToList();
        foreach (string v in rowValues)
        {
            sql += $"'{v}',";
        }
        sql = sql.Substring(0, sql.Length - 1);
        if (cnt == dataTable.Rows.Count)
        {
            sql += ") ";
        }
        else
        {
            sql += "), ";
        }
    }
    return sql;
}

问题根源 #

• 类型处理错误：生成Insert SQL时，把所有字段值都用单引号包裹，但numeric类型不需要单引号，timestamp类型的格式也可能和PostgreSQL要求不匹配，直接导致语法错误。
• 数据类型丢失：通过Json序列化再反序列化生成DataTable的过程中，原本的数值、日期类型会被转成字符串，丢失了原始类型信息，无法准确判断字段类型。
• SQL注入风险：直接拼接表名、字段值到SQL语句中，存在严重的SQL注入漏洞。
• 批量插入效率低：拼接单条大SQL插入大量数据，不仅容易出错，还会拖慢数据库性能。

解决方案 #

1. 用参数化查询替代SQL拼接（最基础的修复） #

参数化查询能自动处理不同数据类型的格式，避免语法错误，同时杜绝SQL注入。修改后的核心代码如下：

public void Oracle2PG(string fromTable, string toTable)
{
    string pgConnStr = _configuration.GetValue<string>($"ConnectionStrings:PGConnection");
    string oraConnStr = _configuration.GetValue<string>($"ConnectionStrings:OracleConnection");

    using (var pgConn = new NpgsqlConnection(pgConnStr))
    using (var oraConn = new OracleConnection(oraConnStr))
    {
        pgConn.Open();
        oraConn.Open();

        string oraSql = $"select * from {fromTable}";
        // 直接获取Oracle数据的原始类型，跳过Json序列化步骤
        var oraData = oraConn.Query<dynamic>(oraSql).ToList();
        if (!oraData.Any()) return;

        // 提取字段名
        var firstRow = oraData.First();
        var cols = ((IDictionary<string, object>)firstRow).Keys.ToList();
        string colNames = string.Join(", ", cols);
        // 生成参数占位符
        string paramPlaceholders = string.Join(", ", cols.Select((_, idx) => $"@p{idx}"));
        string insertSql = $"INSERT INTO {toTable} ({colNames}) VALUES ({paramPlaceholders})";

        using (var cmd = new NpgsqlCommand(insertSql, pgConn))
        {
            foreach (var row in oraData)
            {
                cmd.Parameters.Clear();
                var rowDict = (IDictionary<string, object>)row;
                for (int i = 0; i < cols.Count; i++)
                {
                    var value = rowDict[cols[i]];
                    // 处理Oracle的空值
                    cmd.Parameters.AddWithValue($"@p{i}", value == DBNull.Value ? null : value);
                }
                cmd.ExecuteNonQuery();
            }
        }
    }
}

2. 优化DataTable的获取方式（保留原始类型） #

如果必须使用DataTable，不要用Json序列化，直接用DataAdapter填充，保留原始数据类型：

private static DataTable GetDatatable(string sql, OracleConnection oraConn)
{
    var dt = new DataTable();
    using (var cmd = new OracleCommand(sql, oraConn))
    using (var adapter = new OracleDataAdapter(cmd))
    {
        adapter.Fill(dt);
    }
    return dt;
}

然后生成参数化SQL时，根据DataColumn的类型处理值：

private static void BulkInsert(DataTable dt, string toTable, NpgsqlConnection pgConn)
{
    string colNames = string.Join(", ", dt.Columns.Cast<DataColumn>().Select(c => c.ColumnName));
    string paramPlaceholders = string.Join(", ", dt.Columns.Cast<DataColumn>().Select((_, idx) => $"@p{idx}"));
    string insertSql = $"INSERT INTO {toTable} ({colNames}) VALUES ({paramPlaceholders})";

    using (var cmd = new NpgsqlCommand(insertSql, pgConn))
    {
        foreach (DataRow row in dt.Rows)
        {
            cmd.Parameters.Clear();
            for (int i = 0; i < dt.Columns.Count; i++)
            {
                var colType = dt.Columns[i].DataType;
                var value = row[i];
                cmd.Parameters.AddWithValue($"@p{i}", value == DBNull.Value ? DBNull.Value : Convert.ChangeType(value, colType));
            }
            cmd.ExecuteNonQuery();
        }
    }
}

3. 使用Npgsql批量复制（大数据量最优解） #

如果迁移的数据量很大，推荐用Npgsql的NpgsqlBinaryImporter，效率远高于普通插入，还能精准处理类型映射：

public void Oracle2PGBulk(string fromTable, string toTable)
{
    string pgConnStr = _configuration.GetValue<string>($"ConnectionStrings:PGConnection");
    string oraConnStr = _configuration.GetValue<string>($"ConnectionStrings:OracleConnection");

    using (var pgConn = new NpgsqlConnection(pgConnStr))
    using (var oraConn = new OracleConnection(oraConnStr))
    {
        pgConn.Open();
        oraConn.Open();

        string oraSql = $"select * from {fromTable}";
        var dt = GetDatatable(oraSql, oraConn);
        if (dt.Rows.Count == 0) return;

        // 开启二进制批量导入
        using (var writer = pgConn.BeginBinaryImport($"COPY {toTable} FROM STDIN BINARY"))
        {
            foreach (DataRow row in dt.Rows)
            {
                writer.StartRow();
                foreach (DataColumn col in dt.Columns)
                {
                    var value = row[col];
                    if (value == DBNull.Value)
                    {
                        writer.WriteNull();
                        continue;
                    }
                    // 根据Oracle类型映射到PostgreSQL对应类型
                    switch (col.DataType.Name)
                    {
                        case "Decimal":
                            writer.Write((decimal)value, NpgsqlDbType.Numeric);
                            break;
                        case "DateTime":
                            writer.Write((DateTime)value, NpgsqlDbType.TimestampTz);
                            break;
                        case "Int32":
                            writer.Write((int)value, NpgsqlDbType.Integer);
                            break;
                        // 其他类型按需添加，比如字符串、布尔值等
                        default:
                            writer.Write(value.ToString());
                            break;
                    }
                }
            }
            writer.Complete();
        }
    }
}

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