Operazioni in blocco con go-mssqldb

Il driver go-mssqldb supporta operazioni di inserimento in blocco ad alte prestazioni utilizzando la funzione mssql.CopyIn. L'inserimento in massa bypassa il normale percorso riga per riga INSERT e trasmette i dati direttamente al server utilizzando il protocollo TDS di copia in massa.

Scegli copia di massa, TVP o JSON

Usa la seguente guida quando devi inviare più righe o payload complessi a SQL Server:

Se carichi batch di grandi dimensioni in una tabella di staging o di destinazione, inizia con la copia bulk. Se stai chiamando procedure archiviate con set di righe strutturati, inizia dai TVP. Se hai bisogno di documenti annidati o schemi sparsi, inizia con JSON.

Gli esempi in questo articolo vengono confrontati con il database di esempio AdventureWorks2025 . Esempi di copia in blocco target HumanResources.Department e Production.ProductCategory.

Inserimento in blocco di base

Usare mssql.CopyIn per creare un'istruzione copia in massa, e poi usare Exec per inviare righe:

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

L'ultima stmt.Exec() chiamata senza argomentazioni elimina tutte le righe rimanenti e completa l'operazione di copia in massa.

BulkOptions

La mssql.BulkOptions struct configura il comportamento delle copie di massa:

Esempio con opzioni

Passare BulkOptions per controllare i controlli dei vincoli, i trigger e il blocco:

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

Gestione degli errori

Se una riga fallisce, l'intera operazione di copia in blocco fallisce. Controlla gli errori sia dalla chiamata Exec di ogni riga che dal flush finale 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
}

Rilevare e registrare le righe fallite

Quando un'operazione di copia in massa fallisce, il messaggio di errore di SQL Server indica il vincolo o il problema dei dati ma non identifica la riga specifica. Per identificare le righe non riuscite, utilizza un approccio per batch:

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

Streaming da file CSV

Per file CSV di grandi dimensioni, trasmetti le righe direttamente dal file in copia in massa senza caricare l'intero file in 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()
}

Confronto delle prestazioni

La copia di massa è significativamente più veloce rispetto agli inserti individuali per carichi dati elevati. La tabella seguente mostra le caratteristiche di prestazione approssimative per l'inserimento di 100.000 righe:

Ordinamento delle colonne e mappatura dei tipi

Le colonne in mssql.CopyIn devono corrispondere all'ordine e ai tipi attesi dalla tabella target. Il driver non esegue il matching dei nomi delle colonne; Utilizza la mappatura posizionale.

Problemi di tipo comune

Esempio con tipi espliciti

Specifica esplicitamente i tipi di colonna quando la mappatura predefinita dei tipi non corrisponde al tuo schema:

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

Suggerimenti per le prestazioni

• Usa Tablock per inserimenti di grandi dimensioni in tabelle vuote. Questa opzione riduce la contenzione dei lock e consente una registrazione minima.
• Set RowsPerBatch per controllare con quale frequenza il driver invia dati. I lotti più grandi riducono i viaggi di andata e ritorno ma usano più memoria.
• Aumentare packet size nella stringa di connessione (fino a 32767) per ridurre il sovraccarico di rete.
• Ordina i dati per corrispondere all'indice clusterizzato della tabella target e imposta l'opzione Order . Questo approccio evita un ordinamento sul server.
• Elimina gli indici non clusterizzati prima di carichi grandi in massa, poi ricostruiscili dopo. La manutenzione dell'indice durante l'inserimento in massa aggiunge costi aggiuntivi.
• Utilizzo CheckConstraints: false (il valore predefinito) per permettere ai dati affidabili di saltare il controllo dei vincoli durante la copia in massa.

Limitations

• Bulk copy non supporta colonne protette da Always Encrypted. Per altre informazioni, vedere Limitazioni.
• La copia di massa a livello TDS non è supportata su database SQL di Azure. Con database SQL di Azure, usa invece istruzioni in batch INSERT o un modello con tabella di staging.

Contenuti correlati

• Ottimizzazione delle prestazioni
• Interrogazioni e istruzioni
• Mapping dei tipi di dati
• Guida alla migrazione
• Gestione degli errori e schemi di tentativi