Операции транзакций и массового копирования

Операции массового копирования можно выполнять как изолированные операции или как часть транзакции с несколькими шагами. Этот последний параметр позволяет выполнять несколько операций массового копирования в той же транзакции, а также выполнять другие операции базы данных (например, вставки, обновления и удаления), сохраняя возможность фиксации или отката всей транзакции.

По умолчанию операция массового копирования выполняется как изолированная операция. Операция массового копирования выполняется без использования транзакций, что исключает возможность отката. Если необходимо откатить все или часть массового копирования при возникновении ошибки, можно использовать SqlBulkCopyуправляемую транзакцию, выполнить операцию массового копирования в рамках существующей транзакции или участвовать в System.TransactionsTransaction.

Выполнение операции пакетного копирования без использования транзакций

В следующем консольном приложении демонстрируется процесс, который происходит, когда операция массового копирования данных без использования транзакций сталкивается с ошибкой.

В примере исходная таблица и целевая таблица включают Identity столбец с именем ProductID. Код сначала подготавливает целевую таблицу, удаляя все строки, а затем вставляя одну строку, идентификатор продукта которой, как известно, существует в исходной таблице. По умолчанию новое значение для столбца Identity создается в целевой таблице для каждой добавленной строки. В этом примере параметр, который устанавливается при открытии подключения, заставляет процесс массовой загрузки использовать Identity значения из исходной таблицы.

Операция массового копирования выполняется с установленным в 10 свойством BatchSize. Когда операция обнаруживает недопустимую строку, создается исключение. В этом первом примере операция массового копирования не выполняется. Все пакеты, скопированные до точки ошибки, фиксируются; Пакет, содержащий повторяющийся ключ, откатывается, и операция массового копирования останавливается перед обработкой других пакетов.

var connectionString = GetConnectionString();
// Open a sourceConnection to the AdventureWorks database.
using (SqlConnection sourceConnection =
           new(connectionString))
{
    sourceConnection.Open();

    //  Delete all from the destination table.
    SqlCommand commandDelete = new()
    {
        Connection = sourceConnection,
        CommandText =
        "DELETE FROM dbo.BulkCopyDemoMatchingColumns"
    };
    commandDelete.ExecuteNonQuery();

    //  Add a single row that will result in duplicate key
    //  when all rows from source are bulk copied.
    //  Note that this technique will only be successful in
    //  illustrating the point if a row with ProductID = 446
    //  exists in the AdventureWorks Production.Products table.
    //  If you have made changes to the data in this table, change
    //  the SQL statement in the code to add a ProductID that
    //  does exist in your version of the Production.Products
    //  table. Choose any ProductID in the middle of the table
    //  (not first or last row) to best illustrate the result.
    SqlCommand commandInsert = new()
    {
        Connection = sourceConnection,
        CommandText =
        "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns ON;" +
        "INSERT INTO " + "dbo.BulkCopyDemoMatchingColumns " +
        "([ProductID], [Name] ,[ProductNumber]) " +
        "VALUES(446, 'Lock Nut 23','LN-3416');" +
        "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns OFF"
    };
    commandInsert.ExecuteNonQuery();

    // Perform an initial count on the destination table.
    SqlCommand commandRowCount = new(
        "SELECT COUNT(*) FROM dbo.BulkCopyDemoMatchingColumns;",
        sourceConnection);
    long countStart = Convert.ToInt32(
        commandRowCount.ExecuteScalar());
    Console.WriteLine($"Starting row count = {countStart}");

    //  Get data from the source table as a SqlDataReader.
    SqlCommand commandSourceData = new(
        "SELECT ProductID, Name, ProductNumber " +
        "FROM Production.Product;", sourceConnection);
    SqlDataReader reader = commandSourceData.ExecuteReader();

    // Set up the bulk copy object using the KeepIdentity option.
    using (SqlBulkCopy bulkCopy = new(
               connectionString, SqlBulkCopyOptions.KeepIdentity))
    {
        bulkCopy.BatchSize = 10;
        bulkCopy.DestinationTableName =
            "dbo.BulkCopyDemoMatchingColumns";

        // Write from the source to the destination.
        // This should fail with a duplicate key error
        // after some of the batches have been copied.
        try
        {
            bulkCopy.WriteToServer(reader);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
        finally
        {
            reader.Close();
        }
    }

    // Perform a final count on the destination
    // table to see how many rows were added.
    long countEnd = Convert.ToInt32(
        commandRowCount.ExecuteScalar());
    Console.WriteLine($"Ending row count = {countEnd}");
    Console.WriteLine($"{countEnd - countStart} rows were added.");
    Console.WriteLine("Press Enter to finish.");
    Console.ReadLine();
}

Imports System.Data.SqlClient

Module Module1
    Sub Main()
        Dim connectionString As String = GetConnectionString()

        ' Open a sourceConnection to the AdventureWorks database.
        Using sourceConnection As SqlConnection = _
           New SqlConnection(connectionString)
            sourceConnection.Open()

            ' Delete all from the destination table.
            Dim commandDelete As New SqlCommand
            commandDelete.Connection = sourceConnection
            commandDelete.CommandText = _
               "DELETE FROM dbo.BulkCopyDemoMatchingColumns"
            commandDelete.ExecuteNonQuery()

            ' Add a single row that will result in duplicate key
            ' when all rows from source are bulk copied.
            ' Note that this technique will only be successful in
            ' illustrating the point if a row with ProductID = 446
            ' exists in the AdventureWorks Production.Products table.
            ' If you have made changes to the data in this table, change
            ' the SQL statement in the code to add a ProductID that
            ' does exist in your version of the Production.Products
            ' table. Choose any ProductID in the middle of the table
            ' (not first or last row) to best illustrate the result.
            Dim commandInsert As New SqlCommand
            commandInsert.Connection = sourceConnection
            commandInsert.CommandText = _
               "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns ON;" & _
               "INSERT INTO dbo.BulkCopyDemoMatchingColumns " & _
               "([ProductID], [Name] ,[ProductNumber]) " & _
               "VALUES(446, 'Lock Nut 23','LN-3416');" & _
               "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns OFF"
            commandInsert.ExecuteNonQuery()

            ' Perform an initial count on the destination table.
            Dim commandRowCount As New SqlCommand( _
               "SELECT COUNT(*) FROM dbo.BulkCopyDemoMatchingColumns;", _
                sourceConnection)
            Dim countStart As Long = _
               System.Convert.ToInt32(commandRowCount.ExecuteScalar())
            Console.WriteLine("Starting row count = {0}", countStart)

            ' Get data from the source table as a SqlDataReader.
            Dim commandSourceData As SqlCommand = New SqlCommand( _
               "SELECT ProductID, Name, ProductNumber " & _
               "FROM Production.Product;", sourceConnection)
            Dim reader As SqlDataReader = _
             commandSourceData.ExecuteReader()

            ' Set up the bulk copy object using the KeepIdentity option.
            Using bulkCopy As SqlBulkCopy = New SqlBulkCopy(connectionString, _
              SqlBulkCopyOptions.KeepIdentity)
                bulkCopy.BatchSize = 10
                bulkCopy.DestinationTableName = "dbo.BulkCopyDemoMatchingColumns"

                ' Write from the source to the destination.
                ' This should fail with a duplicate key error
                ' after some of the batches have already been copied.
                Try
                    bulkCopy.WriteToServer(reader)

                Catch ex As Exception
                    Console.WriteLine(ex.Message)

                Finally
                    reader.Close()
                End Try
            End Using

            ' Perform a final count on the destination table
            ' to see how many rows were added.
            Dim countEnd As Long = _
                System.Convert.ToInt32(commandRowCount.ExecuteScalar())
            Console.WriteLine("Ending row count = {0}", countEnd)
            Console.WriteLine("{0} rows were added.", countEnd - countStart)
            Console.WriteLine("Press Enter to finish.")
            Console.ReadLine()
        End Using
    End Sub

    Private Function GetConnectionString() As String
        Throw New NotImplementedException()
    End Function
End Module

Выполнение выделенной операции массового копирования в транзакции

По умолчанию операция массового копирования является собственной транзакцией. Если требуется выполнить выделенную операцию массового копирования, создайте новый экземпляр SqlBulkCopy со строкой подключения или используйте существующий SqlConnection объект без активной транзакции. В каждом сценарии операция массового копирования создает транзакцию, а затем ее фиксирует или откатывает.

Можно явно указать UseInternalTransaction параметр в конструкторе SqlBulkCopy классов, чтобы явно вызвать выполнение операции массового копирования в собственной транзакции, что приводит к выполнению каждой пакетной операции массового копирования в рамках отдельной транзакции.

Следующее консольное приложение аналогично предыдущему примеру. В этом примере операция массового копирования управляет собственными транзакциями. Все пакеты, скопированные до точки ошибки, фиксируются; Пакет, содержащий повторяющийся ключ, откатывается, и операция массового копирования останавливается перед обработкой других пакетов.

var connectionString = GetConnectionString();
// Open a sourceConnection to the AdventureWorks database.
using (SqlConnection sourceConnection =
           new(connectionString))
{
    sourceConnection.Open();

    //  Delete all from the destination table.
    SqlCommand commandDelete = new()
    {
        Connection = sourceConnection,
        CommandText =
        "DELETE FROM dbo.BulkCopyDemoMatchingColumns"
    };
    commandDelete.ExecuteNonQuery();

    //  Add a single row that will result in duplicate key
    //  when all rows from source are bulk copied.
    //  Note that this technique will only be successful in
    //  illustrating the point if a row with ProductID = 446
    //  exists in the AdventureWorks Production.Products table.
    //  If you have made changes to the data in this table, change
    //  the SQL statement in the code to add a ProductID that
    //  does exist in your version of the Production.Products
    //  table. Choose any ProductID in the middle of the table
    //  (not first or last row) to best illustrate the result.
    SqlCommand commandInsert = new()
    {
        Connection = sourceConnection,
        CommandText =
        "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns ON;" +
        "INSERT INTO " + "dbo.BulkCopyDemoMatchingColumns " +
        "([ProductID], [Name] ,[ProductNumber]) " +
        "VALUES(446, 'Lock Nut 23','LN-3416');" +
        "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns OFF"
    };
    commandInsert.ExecuteNonQuery();

    // Perform an initial count on the destination table.
    SqlCommand commandRowCount = new(
        "SELECT COUNT(*) FROM dbo.BulkCopyDemoMatchingColumns;",
        sourceConnection);
    long countStart = Convert.ToInt32(
        commandRowCount.ExecuteScalar());
    Console.WriteLine($"Starting row count = {countStart}");

    //  Get data from the source table as a SqlDataReader.
    SqlCommand commandSourceData = new(
        "SELECT ProductID, Name, ProductNumber " +
        "FROM Production.Product;", sourceConnection);
    SqlDataReader reader = commandSourceData.ExecuteReader();

    // Set up the bulk copy object.
    // Note that when specifying the UseInternalTransaction
    // option, you cannot also specify an external transaction.
    // Therefore, you must use the SqlBulkCopy construct that
    // requires a string for the connection, rather than an
    // existing SqlConnection object.
    using (SqlBulkCopy bulkCopy = new(
               connectionString, SqlBulkCopyOptions.KeepIdentity |
               SqlBulkCopyOptions.UseInternalTransaction))
    {
        bulkCopy.BatchSize = 10;
        bulkCopy.DestinationTableName =
            "dbo.BulkCopyDemoMatchingColumns";

        // Write from the source to the destination.
        // This should fail with a duplicate key error
        // after some of the batches have been copied.
        try
        {
            bulkCopy.WriteToServer(reader);
        }
        catch (Exception ex)
        {
            Console.WriteLine(ex.Message);
        }
        finally
        {
            reader.Close();
        }
    }

    // Perform a final count on the destination
    // table to see how many rows were added.
    long countEnd = Convert.ToInt32(
        commandRowCount.ExecuteScalar());
    Console.WriteLine($"Ending row count = {countEnd}");
    Console.WriteLine($"{countEnd - countStart} rows were added.");
    Console.WriteLine("Press Enter to finish.");
    Console.ReadLine();
}

Imports System.Data.SqlClient

Module Module1
    Sub Main()
        Dim connectionString As String = GetConnectionString()

        ' Open a sourceConnection to the AdventureWorks database.
        Using sourceConnection As SqlConnection = _
           New SqlConnection(connectionString)
            sourceConnection.Open()

            ' Delete all from the destination table.
            Dim commandDelete As New SqlCommand
            commandDelete.Connection = sourceConnection
            commandDelete.CommandText = _
               "DELETE FROM dbo.BulkCopyDemoMatchingColumns"
            commandDelete.ExecuteNonQuery()

            ' Add a single row that will result in duplicate key
            ' when all rows from source are bulk copied.
            ' Note that this technique will only be successful in
            ' illustrating the point if a row with ProductID = 446
            ' exists in the AdventureWorks Production.Products table.
            ' If you have made changes to the data in this table, change
            ' the SQL statement in the code to add a ProductID that
            ' does exist in your version of the Production.Products
            ' table. Choose any ProductID in the middle of the table
            ' (not first or last row) to best illustrate the result.
            Dim commandInsert As New SqlCommand
            commandInsert.Connection = sourceConnection
            commandInsert.CommandText = _
               "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns ON;" & _
               "INSERT INTO dbo.BulkCopyDemoMatchingColumns " & _
               "([ProductID], [Name] ,[ProductNumber]) " & _
               "VALUES(446, 'Lock Nut 23','LN-3416');" & _
               "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns OFF"
            commandInsert.ExecuteNonQuery()

            ' Perform an initial count on the destination table.
            Dim commandRowCount As New SqlCommand( _
               "SELECT COUNT(*) FROM dbo.BulkCopyDemoMatchingColumns;", _
                sourceConnection)
            Dim countStart As Long = _
               System.Convert.ToInt32(commandRowCount.ExecuteScalar())
            Console.WriteLine("Starting row count = {0}", countStart)

            ' Get data from the source table as a SqlDataReader.
            Dim commandSourceData As SqlCommand = New SqlCommand( _
               "SELECT ProductID, Name, ProductNumber " & _
               "FROM Production.Product;", sourceConnection)
            Dim reader As SqlDataReader = _
             commandSourceData.ExecuteReader()

            ' Set up the bulk copy object.
            ' Note that when specifying the UseInternalTransaction option,
            ' you cannot also specify an external transaction. Therefore,
            ' you must use the SqlBulkCopy construct that requires a string
            ' for the connection, rather than an existing SqlConnection object.
            Using bulkCopy As SqlBulkCopy = New SqlBulkCopy(connectionString, _
             SqlBulkCopyOptions.UseInternalTransaction Or _
             SqlBulkCopyOptions.KeepIdentity)
                bulkCopy.BatchSize = 10
                bulkCopy.DestinationTableName = "dbo.BulkCopyDemoMatchingColumns"

                ' Write from the source to the destination.
                ' This should fail with a duplicate key error
                ' after some of the batches have already been copied.
                Try
                    bulkCopy.WriteToServer(reader)

                Catch ex As Exception
                    Console.WriteLine(ex.Message)

                Finally
                    reader.Close()
                End Try
            End Using

            ' Perform a final count on the destination table
            ' to see how many rows were added.
            Dim countEnd As Long = _
                System.Convert.ToInt32(commandRowCount.ExecuteScalar())
            Console.WriteLine("Ending row count = {0}", countEnd)
            Console.WriteLine("{0} rows were added.", countEnd - countStart)
            Console.WriteLine("Press Enter to finish.")
            Console.ReadLine()
        End Using
    End Sub

    Private Function GetConnectionString() As String
        Throw New NotImplementedException()
    End Function
End Module

Использование существующих транзакций

Существующий SqlTransaction объект можно указать в качестве параметра в конструкторе SqlBulkCopy . В этой ситуации операция массового копирования выполняется в существующей транзакции, и при этом состояние транзакции не изменяется (т. е. транзакция не фиксируется и не прерывается). Это позволяет приложению включать операцию массового копирования в транзакцию с другими операциями базы данных. Однако если вы не укажете объект SqlTransaction и передадите нулевую ссылку, при наличии активной транзакции в соединении возникает исключение.

Если необходимо откатить всю операцию массового копирования, так как возникает ошибка, или если массовая копия должна выполняться как часть более крупного процесса, который можно откатить, можно предоставить SqlTransaction объект конструктору SqlBulkCopy .

Следующее консольное приложение аналогично первому примеру (без транзакций), за исключением одного исключения: в этом примере операция массового копирования включается в большую внешнюю транзакцию. При возникновении ошибки нарушения первичного ключа вся транзакция откатывается, и ни одна строка не добавляется в целевую таблицу.

var connectionString = GetConnectionString();
// Open a sourceConnection to the AdventureWorks database.
using (SqlConnection sourceConnection =
           new(connectionString))
{
    sourceConnection.Open();

    //  Delete all from the destination table.
    SqlCommand commandDelete = new()
    {
        Connection = sourceConnection,
        CommandText =
        "DELETE FROM dbo.BulkCopyDemoMatchingColumns"
    };
    commandDelete.ExecuteNonQuery();

    //  Add a single row that will result in duplicate key
    //  when all rows from source are bulk copied.
    //  Note that this technique will only be successful in
    //  illustrating the point if a row with ProductID = 446
    //  exists in the AdventureWorks Production.Products table.
    //  If you have made changes to the data in this table, change
    //  the SQL statement in the code to add a ProductID that
    //  does exist in your version of the Production.Products
    //  table. Choose any ProductID in the middle of the table
    //  (not first or last row) to best illustrate the result.
    SqlCommand commandInsert = new()
    {
        Connection = sourceConnection,
        CommandText =
        "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns ON;" +
        "INSERT INTO " + "dbo.BulkCopyDemoMatchingColumns " +
        "([ProductID], [Name] ,[ProductNumber]) " +
        "VALUES(446, 'Lock Nut 23','LN-3416');" +
        "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns OFF"
    };
    commandInsert.ExecuteNonQuery();

    // Perform an initial count on the destination table.
    SqlCommand commandRowCount = new(
        "SELECT COUNT(*) FROM dbo.BulkCopyDemoMatchingColumns;",
        sourceConnection);
    long countStart = Convert.ToInt32(
        commandRowCount.ExecuteScalar());
    Console.WriteLine($"Starting row count = {countStart}");

    //  Get data from the source table as a SqlDataReader.
    SqlCommand commandSourceData = new(
        "SELECT ProductID, Name, ProductNumber " +
        "FROM Production.Product;", sourceConnection);
    SqlDataReader reader = commandSourceData.ExecuteReader();

    //Set up the bulk copy object inside the transaction.
    using (SqlConnection destinationConnection =
               new(connectionString))
    {
        destinationConnection.Open();

        using (SqlTransaction transaction =
                   destinationConnection.BeginTransaction())
        {
            using (SqlBulkCopy bulkCopy = new(
                       destinationConnection, SqlBulkCopyOptions.KeepIdentity,
                       transaction))
            {
                bulkCopy.BatchSize = 10;
                bulkCopy.DestinationTableName =
                    "dbo.BulkCopyDemoMatchingColumns";

                // Write from the source to the destination.
                // This should fail with a duplicate key error.
                try
                {
                    bulkCopy.WriteToServer(reader);
                    transaction.Commit();
                }
                catch (Exception ex)
                {
                    Console.WriteLine(ex.Message);
                    transaction.Rollback();
                }
                finally
                {
                    reader.Close();
                }
            }
        }
    }

    // Perform a final count on the destination
    // table to see how many rows were added.
    long countEnd = Convert.ToInt32(
        commandRowCount.ExecuteScalar());
    Console.WriteLine($"Ending row count = {countEnd}");
    Console.WriteLine($"{countEnd - countStart} rows were added.");
    Console.WriteLine("Press Enter to finish.");
    Console.ReadLine();
}

Imports System.Data.SqlClient

Module Module1
    Sub Main()
        Dim connectionString As String = GetConnectionString()

        ' Open a sourceConnection to the AdventureWorks database.
        Using sourceConnection As SqlConnection = _
           New SqlConnection(connectionString)
            sourceConnection.Open()

            ' Delete all from the destination table.
            Dim commandDelete As New SqlCommand
            commandDelete.Connection = sourceConnection
            commandDelete.CommandText = _
               "DELETE FROM dbo.BulkCopyDemoMatchingColumns"
            commandDelete.ExecuteNonQuery()

            ' Add a single row that will result in duplicate key
            ' when all rows from source are bulk copied.
            ' Note that this technique will only be successful in
            ' illustrating the point if a row with ProductID = 446
            ' exists in the AdventureWorks Production.Products table.
            ' If you have made changes to the data in this table, change
            ' the SQL statement in the code to add a ProductID that
            ' does exist in your version of the Production.Products
            ' table. Choose any ProductID in the middle of the table
            ' (not first or last row) to best illustrate the result.
            Dim commandInsert As New SqlCommand
            commandInsert.Connection = sourceConnection
            commandInsert.CommandText = _
               "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns ON;" & _
               "INSERT INTO dbo.BulkCopyDemoMatchingColumns " & _
               "([ProductID], [Name] ,[ProductNumber]) " & _
               "VALUES(446, 'Lock Nut 23','LN-3416');" & _
               "SET IDENTITY_INSERT dbo.BulkCopyDemoMatchingColumns OFF"
            commandInsert.ExecuteNonQuery()

            ' Perform an initial count on the destination table.
            Dim commandRowCount As New SqlCommand( _
               "SELECT COUNT(*) FROM dbo.BulkCopyDemoMatchingColumns;", _
                sourceConnection)
            Dim countStart As Long = _
               System.Convert.ToInt32(commandRowCount.ExecuteScalar())
            Console.WriteLine("Starting row count = {0}", countStart)

            ' Get data from the source table as a SqlDataReader.
            Dim commandSourceData As SqlCommand = New SqlCommand( _
               "SELECT ProductID, Name, ProductNumber " & _
               "FROM Production.Product;", sourceConnection)
            Dim reader As SqlDataReader = _
             commandSourceData.ExecuteReader()

            ' Set up the bulk copy object inside the transaction.
            Using destinationConnection As SqlConnection = _
               New SqlConnection(connectionString)
                destinationConnection.Open()

                Using transaction As SqlTransaction = _
                  destinationConnection.BeginTransaction()

                    Using bulkCopy As SqlBulkCopy = New _
                      SqlBulkCopy(destinationConnection, _
                         SqlBulkCopyOptions.KeepIdentity, transaction)
                        bulkCopy.BatchSize = 10
                        bulkCopy.DestinationTableName = _
                         "dbo.BulkCopyDemoMatchingColumns"

                        ' Write from the source to the destination.
                        ' This should fail with a duplicate key error.
                        Try
                            bulkCopy.WriteToServer(reader)
                            transaction.Commit()

                        Catch ex As Exception
                            Console.WriteLine(ex.Message)
                            transaction.Rollback()

                        Finally
                            reader.Close()
                        End Try
                    End Using
                End Using
            End Using

            ' Perform a final count on the destination table
            ' to see how many rows were added.
            Dim countEnd As Long = _
                System.Convert.ToInt32(commandRowCount.ExecuteScalar())
            Console.WriteLine("Ending row count = {0}", countEnd)
            Console.WriteLine("{0} rows were added.", countEnd - countStart)
            Console.WriteLine("Press Enter to finish.")
            Console.ReadLine()
        End Using
    End Sub

    Private Function GetConnectionString() As String
        Throw New NotImplementedException()
    End Function
End Module

См. также

• Операции массового копирования в SQL Server
• Общие сведения об ADO.NET