Крупные операции с помощью go-mssqldb

Драйвер go-mssqldb поддерживает высокопроизводительные операции массовой вставки с помощью функции mssql.CopyIn. Bulk Insert обходит обычный путь по INSERT строкам и направляет данные непосредственно на сервер с помощью протокола массового копирования TDS.

Выбирайте массовое копиирование, TVP или JSON

Используйте следующую инструкцию, если вам нужно отправить несколько строк или сложные наборы данных в SQL Server:

Выбрать...	Когда это подходит лучше всего	Компромиссы
Массовое копирование с mssql.CopyIn	Вам нужен самый быстрый способ загрузить множество строк в одну целевой таблицу.	Наилучшая пропускная способность, но ориентировано это скорее на загрузку таблиц, а не на контракты хранимых процедур или полезные нагрузки смешанной структуры.
Параметр с табличным значением	Нужно передать сильно типизированный набор строк в сохранённую процедуру или параметризованную команду.	Сохраняет границы схемы и процедур, но требует пользовательского типа таблицы и соответствующего порядка полей.
JSON с OPENJSON или FOR JSON	Ваше приложение уже обменивается данными в формате JSON, или полезная нагрузка имеет вложенную либо гибкую структуру.	Более портативный для кода приложения, но обычно медленнее и менее безопасно для шрифта, чем TVP или массовые копии для структурированных вставок.

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

Примеры в этой статье выполняются в образце базы данных AdventureWorks2025. Примеры массового копирования предназначены для HumanResources.Department и Production.ProductCategory.

Базовая массовая вставка

Используйте mssql.CopyIn для создания инструкции bulk copy, а затем используйте Exec для отправки строк:

import (
    "database/sql"
    "log"

    "github.com/microsoft/go-mssqldb"
)

func bulkInsert(db *sql.DB) error {
    txn, err := db.Begin()
    if err != nil {
        return err
    }
    defer txn.Rollback()

    stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department", mssql.BulkOptions{},
        "Name", "GroupName"))
    if err != nil {
        return err
    }

    // Add rows
    _, err = stmt.Exec("Data Science", "Research and Development")
    if err != nil {
        return err
    }
    _, err = stmt.Exec("Cloud Ops", "Information Technology")
    if err != nil {
        return err
    }
    _, err = stmt.Exec("Developer Relations", "Sales and Marketing")
    if err != nil {
        return err
    }

    // Flush and finalize the bulk copy
    result, err := stmt.Exec()
    if err != nil {
        return err
    }

    if err = stmt.Close(); err != nil {
        return err
    }

    rowsAffected, _ := result.RowsAffected()
    log.Printf("Bulk inserted %d rows\n", rowsAffected)

    return txn.Commit()
}

Последний stmt.Exec() вызов без аргументов очищает все оставшиеся строки и завершает операцию массового копирования.

BulkOptions

Структура mssql.BulkOptions настраивает поведение массового копирования:

Поле	Type	Description
CheckConstraints	bool	Проверять ограничения при пакетной вставке.
FireTriggers	bool	Запускайте INSERT триггеры на столе цели.
KeepNulls	bool	Сохраняйте значения null вместо подстановки значений по умолчанию.
KilobytesPerBatch	int	Килобайты на партию. 0 использует значение сервера по умолчанию.
RowsPerBatch	int	Ряды на партию. 0 использует значение сервера по умолчанию.
Order	[]string	ORDER подсказка для целевого кластерного индекса (например, []string{"Id ASC"}).
Tablock	bool	Установить блокировку на уровне таблицы на время выполнения операции массового копирования.

Пример с опциями

Передайте BulkOptions, чтобы управлять проверками ограничений, триггерами и блокировкой:

stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department",
    mssql.BulkOptions{
        CheckConstraints: true,
        FireTriggers:     true,
        Tablock:          true,
        RowsPerBatch:     1000,
    },
    "Name", "GroupName"))

Обработка ошибок

Если хотя бы одна строка завершается с ошибкой, вся операция массового копирования завершается с ошибкой. Проверяйте ошибки как при вызове Exec для каждой строки, так и при окончательном сбросе Exec:

for _, emp := range employees {
    _, err = stmt.Exec(emp.Name, emp.GroupName)
    if err != nil {
        txn.Rollback()
        return err
    }
}

// Final flush
_, err = stmt.Exec()
if err != nil {
    txn.Rollback()
    return err
}

Обнаружение и запись неудачных строк

Когда операция массового копирования не удаётся, сообщение об ошибке от SQL Server указывает на ограничение или проблему с данными, но не указывает конкретную строку. Для выявления неисправных строк используйте пакетный подход:

func bulkInsertWithRowTracking(db *sql.DB, departments []Department) error {
    txn, err := db.Begin()
    if err != nil {
        return err
    }

    stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department",
        mssql.BulkOptions{RowsPerBatch: 500}, "Name", "GroupName"))
    if err != nil {
        txn.Rollback()
        return err
    }

    for i, dept := range departments {
        _, err = stmt.Exec(dept.Name, dept.GroupName)
        if err != nil {
            txn.Rollback()
            log.Printf("Bulk copy failed at row %d (Name=%q): %v", i, dept.Name, err)
            return fmt.Errorf("bulk copy failed at row %d: %w", i, err)
        }
    }

    _, err = stmt.Exec()
    if err != nil {
        txn.Rollback()
        return fmt.Errorf("bulk copy flush failed: %w", err)
    }

    if err = stmt.Close(); err != nil {
        txn.Rollback()
        return err
    }

    return txn.Commit()
}

Tip

Если нужно пропустить ошибочные строки и продолжить, используйте отдельные операторы INSERT или шаблон с промежуточной таблицей: выполните массовое копирование в промежуточную таблицу без ограничений, затем используйте MERGE или INSERT...SELECT с обработкой ошибок, чтобы перенести данные в целевую таблицу.

Поток из CSV-файлов

Для больших CSV-файлов передавайте строки напрямую из файла в bulk copy, не загружая весь файл в память:

import (
    "encoding/csv"
    "io"
    "os"
)

func bulkInsertFromCSV(db *sql.DB, filePath string) error {
    f, err := os.Open(filePath)
    if err != nil {
        return err
    }
    defer f.Close()

    reader := csv.NewReader(f)

    // Skip the header row.
    _, err = reader.Read()
    if err != nil {
        return err
    }

    txn, err := db.Begin()
    if err != nil {
        return err
    }

    stmt, err := txn.Prepare(mssql.CopyIn("HumanResources.Department",
        mssql.BulkOptions{Tablock: true, RowsPerBatch: 5000},
        "Name", "GroupName"))
    if err != nil {
        txn.Rollback()
        return err
    }

    var rowCount int
    for {
        record, err := reader.Read()
        if err == io.EOF {
            break
        }
        if err != nil {
            txn.Rollback()
            return fmt.Errorf("CSV read error at row %d: %w", rowCount+1, err)
        }

        _, err = stmt.Exec(record[0], record[1])
        if err != nil {
            txn.Rollback()
            return fmt.Errorf("row %d: %w", rowCount+1, err)
        }
        rowCount++
    }

    // Flush remaining rows.
    result, err := stmt.Exec()
    if err != nil {
        txn.Rollback()
        return err
    }
    if err = stmt.Close(); err != nil {
        txn.Rollback()
        return err
    }

    affected, _ := result.RowsAffected()
    log.Printf("Bulk inserted %d rows from CSV", affected)

    return txn.Commit()
}

Сравнение производительности

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

Метод	Относительная скорость	Маршруты по сети туда и обратно	Блокировка
Индивидуальный INSERT	Самый медленный (1 раз)	100,000	Уровень строки для каждой вставки.
Пакетная обработка INSERT (1 000 строк в одном выражении)	Средний (5–10 раз)	100	На уровне строки для каждого пакета.
Массовое копирование без TABLOCK	Быстро (20–50 раз)	Зависит от размера партии	Пакетная обработка на уровне строк.
Массовое копирование с TABLOCK	Самый быстрый (50-100 раз)	Зависит от размера партии	Блокировка на уровне таблицы, минимальное журналирование.

Замечание

Фактическая производительность зависит от сетевой задержки, конфигурации сервера, индексов таблиц и возможности минимального логирования. Проводите бенчмарк с вашей конкретной нагрузкой с помощью testing.B. См. раздел "Настройка производительности".

Упорядочение столбцов и отображение типов

Столбцы в mssql.CopyIn должны соответствовать порядку и типам данных, требуемым целевой таблицей. Драйвер не выполняет сопоставление имён столбцов; Он использует позиционное отображение.

Распространённые типографические проблемы

тип Go	Столбец SQL Server	Issue	Solution
string	varchar	Неявное nvarchar преобразование.	Используйте mssql.VarChar обёртку.
float64	decimal(18,4)	Потеря точности.	Пройти как string.
time.Time	datetime2	Преобразование часовых поясов.	Используйте время UTC.
nil	Любой нулируемый столбец	Требует использования KeepNulls: true.	Установите KeepNulls в BulkOptions.

Пример с явными типами

Указывайте типы столбцов явно, если стандартное отображение типов не совпадает с вашей схемой:

stmt, err := txn.Prepare(mssql.CopyIn("Production.ProductCategory",
    mssql.BulkOptions{KeepNulls: true},
    "Name"))
if err != nil {
    return err
}

for _, p := range categories {
    _, err = stmt.Exec(p.Name)
    if err != nil {
        return err
    }
}

Советы по производительности

• Используйте Tablock для массовой вставки в пустые таблицы. Этот параметр снижает конкуренцию за блокировки и позволяет использовать минимальное журналирование.
• Установите RowsPerBatch, чтобы управлять тем, как часто драйвер отправляет данные. Большие партии сокращают круговые поездки, но требуют больше памяти.
• Увеличьте packet size в строке подключения (до 32767), чтобы снизить сетевые накладные расходы.
• Упорядочивайте данные так, чтобы они совпадали с кластерным индексом целевой таблицы, и задайте эту Order опцию. Этот подход избегает сортировки на стороне сервера.
• Убрать некластерные индексы перед крупными массовыми загрузками, а потом восстановить их после. Обслуживание индекса во время массовой вставки создаёт дополнительные накладные расходы.
• Используйте CheckConstraints: false (по умолчанию) для доверенных данных, чтобы пропустить проверку ограничений при массовом копировании.

Ограничения

• Массовое копирование не поддерживает колонки, защищённые Always Encrypted. Дополнительные сведения см. в статье Ограничения.
• Массовое копирование уровня TDS не поддерживается в База данных SQL Azure. Для База данных SQL Azure вместо этого используйте операторы INSERT, выполняемые пакетами, или подход с промежуточной таблицей.

Связанные материалы

• Настройка производительности
• Запросы и инструкции
• Сопоставления типов данных
• Руководство по миграции
• Обработка ошибок и повторные попытки