Operaciones masivas con go-mssqldb

El go-mssqldb controlador soporta operaciones de inserción masiva de alto rendimiento utilizando la mssql.CopyIn función. La inserción masiva omite la ruta normal de fila por fila INSERT y transmite los datos directamente al servidor mediante el protocolo TDS de copia masiva.

Elige copia masiva, TVP o JSON

Utiliza la siguiente guía cuando necesites enviar varias filas o cargas complejas a SQL Server:

Elija...	Cuando mejor encaja	Compensación
Copiar en bloque con mssql.CopyIn	Necesitas la forma más rápida de cargar muchas filas en una tabla objetivo.	Ofrece el mayor rendimiento, pero está orientado a cargas en tablas en lugar de a contratos de procedimientos almacenados o cargas útiles con estructuras mixtas.
Un parámetro con valores en tabla	Necesitas pasar un conjunto de filas fuertemente tipeadas a un procedimiento almacenado o comando parametrizado.	Preserva los límites del esquema y del procedimiento, pero requiere un tipo de tabla definido por el usuario y un orden de campo correspondiente.
JSON con OPENJSON o FOR JSON	Tu aplicación ya intercambia JSON, o la forma de la carga útil es anidada o flexible.	Más portátil para código de app, pero normalmente más lento y menos seguro para escribir que los TVP o copias masivas para insertos estructurados.

Si cargas grandes lotes en una tabla de staging o destino, empieza con copia masiva. Si vas a llamar a procedimientos almacenados con conjuntos de filas estructuradas, empieza con TVPs. Si necesitas documentos anidados o esquemas sueltos, empieza con JSON.

Los ejemplos de este artículo se aplican a la base de datos de ejemplo AdventureWorks2025 . Los ejemplos de copia masiva apuntan a HumanResources.Department y Production.ProductCategory.

Inserto básico a granel

Úsalo mssql.CopyIn para crear una sentencia de copia masiva, y luego úsalo Exec para enviar filas:

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

La llamada final stmt.Exec() , sin argumentos, elimina las filas restantes y completa la operación de copia masiva.

Opciones en bloque

La mssql.BulkOptions estructura configura el comportamiento de copia masiva:

Campo	Tipo	Descripción
CheckConstraints	bool	Comprobar las restricciones durante la inserción masiva.
FireTriggers	bool	Disparar INSERT disparos en la mesa de objetivos.
KeepNulls	bool	Conserva los valores nulos en lugar de insertar valores por defecto.
KilobytesPerBatch	int	Kilobytes por lote. 0 Usa el valor predeterminado del servidor.
RowsPerBatch	int	Filas por lote. 0 Usa el valor predeterminado del servidor.
Order	[]string	ORDER pista para el índice agrupado objetivo (por ejemplo, []string{"Id ASC"}).
Tablock	bool	Consigue un bloqueo a nivel de tabla durante la duración de la copia masiva.

Ejemplo con opciones

Pase BulkOptions para controlar las comprobaciones de restricciones, los disparadores y los bloqueos:

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

Gestión de errores

Si falla alguna fila, falla toda la operación de copia masiva. Compruebe los errores tanto de la llamada de cada fila a Exec como del vaciado final 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
}

Detectar y registrar filas fallidas

Cuando falla una operación de copia masiva, el mensaje de error de SQL Server indica la restricción o el problema de datos, pero no identifica la fila específica. Para identificar filas fallidas, utiliza un enfoque por lotes:

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

Si necesitas omitir filas erróneas y continuar, usa instrucciones individuales INSERT o un patrón de tabla de ensayo: realiza una copia masiva en una tabla de ensayo sin restricciones y, después, usa un MERGE o INSERT...SELECT con control de errores para mover los datos a la tabla de destino.

Transmisión desde archivos CSV

Para archivos CSV grandes, transmite las filas directamente desde el archivo a una copia masiva sin cargar todo el archivo en la memoria:

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

Comparación del rendimiento

La copia masiva es significativamente más rápida que las inserciones individuales para grandes cargas de datos. La siguiente tabla muestra características aproximadas de rendimiento para insertar 100.000 filas:

Método	Velocidad relativa	Viajes de ida y vuelta de la red	Bloqueo
Individual INSERT	Más lento (1x)	100 000	A nivel de fila por inserto.
Agrupado INSERT (1.000 filas por sentencia)	Mediana (5-10 veces)	100	Nivel de fila por lote.
Copia en bloque sin TABLOCK	Rápido (20-50x)	Depende del tamaño del lote	Volumen a nivel de fila.
Copia en bloque con TABLOCK	El más rápido (50-100x)	Depende del tamaño del lote	Bloqueo a nivel de tabla, registro mínimo.

Note

El rendimiento real varía en función de la latencia de la red, la configuración del servidor, los índices de tablas y si hay un registro mínimo disponible. Haz un benchmark con tu carga de trabajo específica usando testing.B. Consulte Optimización del rendimiento.

Ordenamiento de columnas y mapeo de tipos

Las columnas en mssql.CopyIn deben coincidir con el orden y los tipos esperados por la tabla objetivo. El controlador no realiza la coincidencia de nombres de columnas; Utiliza mapeo posicional.

Problemas de tipo común

tipo de Go	Columna de SQL Server	Issue	Solución
string	varchar	Conversión nvarchar implícita.	Usa mssql.VarChar envoltorio.
float64	decimal(18,4)	Pérdida de precisión.	Hacerse pasar por string.
time.Time	datetime2	Conversión de zona horaria.	Usa los horarios UTC.
nil	Cualquier columna anulable	Se requiere KeepNulls: true.	Establezca KeepNulls en BulkOptions.

Ejemplo con tipos explícitos

Especifica los tipos de columnas explícitamente cuando el mapeo de tipos por defecto no coincida con tu 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
    }
}

Consejos de rendimiento

• Uso Tablock para insertos grandes en tablas vacías. Esta opción reduce la contención de bloqueos y permite un registro mínimo.
• Configure RowsPerBatch para controlar la frecuencia con la que el controlador envía datos. Los lotes más grandes reducen los viajes de ida y vuelta pero usan más memoria.
• Aumente packet size en la cadena de conexión (hasta 32767) para reducir la sobrecarga de red.
• Ordena los datos para que coincidan con el índice agrupado de la tabla objetivo y establece la Order opción. Este enfoque evita una ordenación del lado del servidor.
• Elimina índices no agrupados antes de cargas grandes y luego reconstruyelos después. El mantenimiento del índice durante la inserción a granel añade sobrecarga.
• Uso CheckConstraints: false (el predeterminado) para que los datos confiables se salten la comprobación de restricciones durante la copia masiva.

Limitations

• Bulk copy no admite columnas protegidas por Always Encrypted. Para obtener más información, consulta Limitaciones.
• La copia masiva a nivel TDS no está soportada en Azure SQL Database. Para Azure SQL Database, utiliza sentencias agrupadas INSERT o un patrón de tabla de etapas.

Contenido relacionado

• Ajuste del rendimiento
• Consultas y sentencias
• Asignaciones de tipos de datos
• Guía de migración
• Gestión de errores y patrones de reintentos