Operações em massa com go-mssqldb

O driver go-mssqldb oferece suporte a operações de inserção em massa com alto desempenho usando a função mssql.CopyIn. O bulk insert contorna o caminho normal linha por linha INSERT e transmite dados diretamente para o servidor usando o protocolo de cópia em massa TDS.

Escolha cópia em massa, TVP ou JSON

Use o guia a seguir quando precisar enviar várias linhas ou cargas complexas para o SQL Server:

Se você estiver carregando grandes lotes em uma tabela de preparação ou destino, comece com a cópia em massa. Se você estiver invocando procedimentos armazenados com conjuntos de linhas estruturados, comece usando TVPs. Se precisar de documentos aninhados ou esquemas soltos, comece com JSON.

Os exemplos neste artigo usam o banco de dados de exemplo AdventureWorks2025. Exemplos de cópia em massa têm como destino HumanResources.Department e Production.ProductCategory.

Inserção em massa básica

Use mssql.CopyIn para criar uma instrução de cópia em massa e depois use Exec para enviar linhas:

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()
}

A chamada final stmt.Exec() , sem argumentos, elimina as linhas restantes e completa a operação de cópia em massa.

BulkOptions

A mssql.BulkOptions struct configura o comportamento de cópia em massa:

Exemplo com opções

Passe BulkOptions para controlar as verificações de restrições, os gatilhos e o bloqueio:

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

Tratamento de erros

Se qualquer linha falhar, toda a operação de cópia em massa falha. Verifique os erros de cada chamada Exec de linha e do flush Exec final:

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
}

Detectar e registrar linhas falhadas

Quando uma operação de cópia em massa falha, a mensagem de erro do SQL Server indica a restrição ou problema de dados, mas não identifica a linha específica. Para identificar linhas falhadas, use uma abordagem de agrupamento:

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()
}

Transmitir a partir de arquivos CSV

Para arquivos CSV grandes, faça streaming de linhas diretamente do arquivo para cópias em massa sem carregar o arquivo inteiro na memória:

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()
}

Comparação de desempenho

A cópia em massa é significativamente mais rápida do que inserções individuais para grandes cargas de dados. A tabela a seguir mostra características aproximadas de desempenho para inserir 100.000 linhas:

Ordenação de colunas e mapeamento de tipos

Colunas em mssql.CopyIn devem corresponder à ordem e aos tipos esperados pela tabela alvo. O driver não realiza correspondência de nomes de coluna; ele usa mapeamento posicional.

Questões de tipos comuns

Exemplo com tipos explícitos

Especifique os tipos de colunas explicitamente quando o mapeamento padrão de tipos não corresponder ao seu esquema:

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
    }
}

Dicas de desempenho

• Use Tablock para grandes inserções em tabelas vazias. Esta opção reduz a contenção de bloqueio e permite o registro mínimo.
• Defina RowsPerBatch para controlar com que frequência o driver envia dados. Lotes maiores reduzem viagens de ida e volta, mas consomem mais memória.
• Aumente packet size na cadeia de conexão (até 32767) para reduzir a sobrecarga de rede.
• Ordene os dados para corresponder ao índice agrupado da tabela alvo e defina a Order opção. Essa abordagem evita uma classificação do lado do servidor.
• Elimine índices não agrupados antes de grandes cargas em massa e depois os reconstrua. A manutenção do índice durante a inserção em massa adiciona sobrecarga.
• Uso CheckConstraints: false (o padrão) para que dados confiáveis pulem a verificação de restrições durante a cópia em massa.

Limitações

• Cópia em massa não suporta colunas protegidas pelo Always Encrypted. Para obter mais informações, consulte Limitações.
• Cópia em massa em nível TDS não é suportada no Banco de Dados SQL do Azure. Para o Banco de Dados SQL do Azure, use instruções em lote INSERT ou um padrão de tabela de preparação.

Conteúdo relacionado

• Ajuste de desempenho com go-mssqldb
• Consultas e instruções com go-mssqldb
• Mapeamentos de tipos de dados go-mssqldb
• Migrar para go-mssqldb a partir de outros drivers
• Tratamento de erros e padrões de repetição com go-mssqldb