具有不同排序规则的 SqlBulkCopy

Question

我需要将数据从一个数据库迁移到另一个数据库。我选择使用 SqlBulkCopy，但有一个问题，因为源数据库的排序规则与目标数据库不同，所以，我遇到了一个例外：

System.InvalidOperationException: The locale id '1049' of the source column 'Id' and the locale id '1033' of the destination column 'Id' do not match.
   at System.Data.SqlClient.SqlBulkCopy.AnalyzeTargetAndCreateUpdateBulkCommand(BulkCopySimpleResultSet internalResults)
   at System.Data.SqlClient.SqlBulkCopy.WriteToServerInternal()
   at System.Data.SqlClient.SqlBulkCopy.WriteRowSourceToServer(Int32 columnCount)
   at System.Data.SqlClient.SqlBulkCopy.WriteToServer(IDataReader reader)
   at MigrateToNormalized.DirectMapCommand.Migrate(SqlConnection source, SqlConnection destination, SqlTransaction transaction) in D:\Projects\APS\DTE\MigrateTo
Normalized\MigrateToNormalized\MigrateToNormalized\DirectMapCommand.cs:line 53
   at MigrateToNormalized.Program.Main(String[] args) in D:\Projects\APS\DTE\MigrateToNormalized\MigrateToNormalized\MigrateToNormalized\Program.cs:line 32

谁能告诉我，如何在 SQL 查询中不直接使用 COLLATE 语句的情况下解决此问题？是否有一些简单的方法可以更改源数据库中所有列的排序规则？

Answer 1

没错，当我们使用SqlBulkCopy时，有时会出现错误，这是使用SqlBulkCopy时映射列的最佳方法。

我以前的代码：

SqlConnectionStringBuilder cb = new SqlConnectionStringBuilder("Data Source=ServerName;User Id=userid;Password=****;Initial Catalog=Deepak; Pooling=true; Max pool size=200; Min pool size=0");
SqlConnection con = new SqlConnection(cb.ConnectionString);
SqlCommand cmd = new SqlCommand("select Name,Class,Section,RollNo from Student", con);
con.Open();
SqlDataReader rdr = cmd.ExecuteReader();
SqlBulkCopy sbc = new SqlBulkCopy("Data Source=DestinationServer;User Id=destinationserveruserid;Password=******;Initial Catalog=DeepakTransfer; Pooling=true; Max pool size=200; Min pool size=0");
sbc.DestinationTableName = "StudentTrans";   
sbc.WriteToServer(rdr); 
sbc.Close();
rdr.Close();
con.Close();

该代码给我的错误为：

源列“RollNo”的区域设置 ID“0”与目标列“Section”的区域设置 ID“1033”不匹配。

现在，在列映射之后，我的代码已成功运行。

我修改后的代码是：

SqlConnectionStringBuilder cb = new SqlConnectionStringBuilder("Data Source=ServerName;User Id=userid;Password=****;Initial Catalog=Deepak;");
SqlConnection con = new SqlConnection(cb.ConnectionString);
SqlCommand cmd = new SqlCommand("select Name,Class,Section,RollNo from Student", con);
con.Open();
SqlDataReader rdr = cmd.ExecuteReader();
SqlBulkCopy sbc = new SqlBulkCopy("Data Source=DestinationServer;User Id=destinationserveruserid;Password=******;Initial Catalog=DeepakTransfer;");
sbc.DestinationTableName = "StudentTrans";
sbc.ColumnMappings.Add("Name", "Name");
sbc.ColumnMappings.Add("Class", "Class");
sbc.ColumnMappings.Add("Section", "Section");
sbc.ColumnMappings.Add("RollNo", "RollNo");
sbc.WriteToServer(rdr);
sbc.Close();
rdr.Close();
con.Close();

此代码运行成功。

Answer 2

您可以选择具有不同排序规则的列：

SELECT Foo COLLATE SQL_Latin1_General_CP1_CI_AS AS Bar FROM Baz

这会将列 Foo 的排序规则转换为新的排序规则。在上面的示例中，列 Foo 被转换为排序规则 SQL_Latin1_General_CP1_CI_AS 并在查询中命名为 Bar。

然后，您需要为批量复制命令的新列添加列映射：

using (var bulkCopy = new SqlBulkCopy(connection))
{
    bulkCopy.DestinationTableName = "FooBars";
    bulkCopy.ColumnMappings.Add("Bar", "FooBar"); 
    bulkCopy.WriteToServer(reader);
}

Answer 3

简单添加列映射对我来说不起作用。我已经通过 SqlBulkCopy 和 DataTable 实现了插入 - 这个工作正常。

private void BulkCopyTable(string sourceConnection, string targetConnection, Table sTable, Table tTable)
{
    using (SqlConnection sourceConn = new SqlConnection(sourceConnection))
    {
        if (cbFixStructure.Checked)
            CheckAndRecreateTarget(targetConnection, sTable, tTable);

        string selectSql = "SELECT * FROM " + sTable.Schema + ".[" + sTable.Name + "]";

        string selectCntSql = "SELECT COUNT(*) FROM " + sTable.Schema + ".[" + sTable.Name + "] WITH(NOLOCK)";
        using (SqlCommand selectCmd = new SqlCommand(selectSql, sourceConn))
        {
            selectCmd.CommandTimeout = 60 * 100 * 1000;
            sourceConn.Open();
            Int64 totalCount = 0;
            using (SqlCommand cntCommand = new SqlCommand(selectCntSql, sourceConn))
            {
                cntCommand.CommandTimeout = 60 * 100 * 1000;
                totalCount = Convert.ToInt64(cntCommand.ExecuteScalar());
            }

            DataTable dtBuffer = new DataTable();
            var columns = sTable.Columns.Cast<Column>().Where(p => p.Computed == false).ToList();
            foreach (var clm in columns)
            {
                var sdt = clm.DataType.SqlDataType;
                if (sdt == SqlDataType.UserDefinedDataType)
                {
                    var lst = Enum.GetValues(typeof(SqlDataType)).Cast<SqlDataType>();
                    sdt = lst.Where(p => p.ToString().ToLower() == TargetDataBase.UserDefinedDataTypes[clm.DataType.Name].SystemType.ToString()).First();
                }

                dtBuffer.Columns.Add(new DataColumn(clm.Name, GetClrType(sdt)));
            }
            using (SqlDataReader reader = selectCmd.ExecuteReader())
            {
                using (SqlBulkCopy blkCopy = new SqlBulkCopy(targetConnection, SqlBulkCopyOptions.KeepIdentity))
                {
                    blkCopy.BulkCopyTimeout = 60 * 100 * 1000;
                    blkCopy.DestinationTableName = sTable.Schema + ".[" + sTable.Name + "]";

                    foreach (var colmn in columns)
                    {
                        blkCopy.ColumnMappings.Add(colmn.Name, colmn.Name);
                    }

                    int bufferCountLengthMax = 500;
                    int rowCnt = 0;
                    int globalCounter = 0;
                    while (reader.Read())
                    {
                        var dataRow = dtBuffer.NewRow();
                        foreach (var clm in columns)
                        {
                            dataRow[clm.Name] = reader[clm.Name];
                        }
                        dtBuffer.Rows.Add(dataRow);
                        rowCnt++;
                        globalCounter++;
                        if (rowCnt >= bufferCountLengthMax)
                        {
                            dtBuffer.AcceptChanges();
                            blkCopy.WriteToServer(dtBuffer);
                            rowCnt = 0;
                            dtBuffer.Rows.Clear();
                            GC.Collect();
                            DoLogText(String.Format("Table \"{0}\" copied rows {1} out of {2}", sTable.Schema + ".[" + sTable.Name + "]", globalCounter, totalCount));
                        }
                    }
                    if (rowCnt > 0)
                    {
                        dtBuffer.AcceptChanges();
                        blkCopy.WriteToServer(dtBuffer);
                        rowCnt = 0;
                        dtBuffer.Rows.Clear();
                        GC.Collect();
                        DoLogText(String.Format("Table \"{0}\" copied rows {1} out of {2}", sTable.Schema + ".[" + sTable.Name + "]", globalCounter, totalCount));
                    }
                }
            }
        }
    }
    DoLogText(String.Format("Table \"{0}\" done", sTable.Name));
}

Answer 4

您可以更改用于 sqlbulkcopy 的表中列的排序规则。

例如

CREATE TABLE T3
(
    C1 int PRIMARY KEY,
    C2 varchar(50) NULL,
    C3 int NULL,
    C4 int 
) ;
GO

ALTER TABLE T3 ALTER COLUMN C2 varchar(50) COLLATE Latin1_General_BIN