Bulk Inserts using pgx in Go

[robd003]

Hey guys,

I'm trying to do bulk inserts from Go and wondering what the best way to go about it is. Since cursors are now supported does that mean you can wrap many prepared statements in a transaction and then execute the transaction?

I was worried about using COPY FROM since the schema to this table is evolving and I'm worried about putting items in the wrong column. (If I use INSERT INTO (field1, field2, field3) then everything will match up even if the schema changes)

Example of what I'm thinking of doing:

import (
    "context"
    "github.com/jackc/pgx/v5"
)

func bulkInsert(conn *pgx.Conn, rows [][]interface{}) error {
    // Start a transaction
    tx, err := conn.Begin(context.Background())
    if err != nil {
        return err
    }
    defer tx.Rollback(context.Background()) // this is useless, correct?

    // Prepare the insert statement with placeholders for each column
    stmt, err := tx.Prepare(context.Background(), "INSERT INTO mytable (field1, field2, field3) VALUES ($1, $2, $3)")
    if err != nil {
        return err
    }
    defer stmt.Close()

    // Iterate over the rows and add each one to the prepared statement
    for _, row := range rows {
        _, err = stmt.Exec(context.Background(), row[0], row[1], row[2])
        if err != nil {
            return err
        }
    }

    // Commit the transaction
    err = tx.Commit(context.Background())
    if err != nil {
        return err
    }

    return nil
}

[matriv]

I'm trying to do bulk inserts from Go and wondering what the best way to go about it is. Since cursors are now supported does that mean you can wrap many prepared statements in a transaction and then execute the transaction?

Cursors, are for select statements, they don't have any connection with the prepared statements.
CrateDB doesn't support transactions, so what you want to do is batch insert using a prepared statement like your example.

I was worried about using COPY FROM since the schema to this table is evolving and I'm worried about putting items in the wrong column. (If I use INSERT INTO (field1, field2, field3) then everything will match up even if the schema changes)

It's a good idea to provide the exact columns and their order for the insert which you can also do when using COPY FROM, see here.

Just FYI, there is also cr8 tool that allows you to do inserts independently of your application.

[matriv]

You can checkout the cr8 tool for this as well.: https://github.com/mfussenegger/cr8#run-spec
and how it's used in our benchmarks: https://github.com/crate/crate-benchmarks

How big is your cluster? You might want to consider (apart from testing and adjusting the batch size), to use multiple threads to exploit the whole performance potential of the cluster.

You can start with multiple threads (like 10 for example) and play around with increasing them to reach a saturation point, where you could have a rough number of inserts/sec you can do.

[robd003]

Does it make sense to break up the batches and insert them on different hosts in the cluster?

Here's my current test code for connecting to a single host:

import (
	"context"
	"fmt"
	"log"
        "time"

	"github.com/jackc/pgx/v5"
)

// MyTableRow represents a row in test.mytable
type MyTableRow struct {
	Field1 int
	Field2 string
	Field3 string
}

func BulkInsertMyTableRows(ctx context.Context, db *pgxpool.Pool, rows []MyTableRow) error {
	const batchSize = 1000

	// Prepare the statement.
	sql := "INSERT INTO mytable (field1, field2, field3) VALUES ($1, $2, $3)"
	_, err := db.Prepare(ctx, "bulk_insert_mytable", sql)
	if err != nil {
		return fmt.Errorf("failed to prepare statement: %w", err)
	}

	// Perform bulk inserts in batches.
	for i := 0; i < len(rows); i += batchSize {
		batch := &pgx.Batch{}
		batchEnd := i + batchSize
		if batchEnd > len(rows) {
			batchEnd = len(rows)
		}

		for _, row := range rows[i:batchEnd] {
			batch.Queue("bulk_insert_mytable", row.Field1, row.Field2, row.Field3)
		}

		results := db.SendBatch(ctx, batch)
		err = results.Close()
		if err != nil {
			return fmt.Errorf("failed to execute batch: %w", err)
		}
	}

	return nil
}

func main() {
        // @matriv should I connect to all hosts in the pool and round-robin each batch of 1,000 inserts to a different host?
	connString := "postgres://crate-1:5432/test?sslmode=disable"
	db, err := pgx.Connect(context.Background(), connString)
	if err != nil {
		log.Fatalf("Unable to connect to database: %v\n", err)
	}
	defer db.Close()

	// Generate sample data.
	rows := make([]MyTableRow, 50000)
	for i := range rows {
		rows[i] = MyTableRow{
			Field1: i,
			Field2: fmt.Sprintf("Field2-%d", i),
			Field3: fmt.Sprintf("Field3-%d", i),
		}
	}

	// Perform the bulk insert.
        start := time.Now()
	err = BulkInsertMyTableRows(context.Background(), db, rows)
	if err != nil {
		log.Fatalf("Failed to bulk insert rows: %v\n", err)
	}
        duration := time.Since(start)
	fmt.Printf("Successfully inserted rows in %s\n", duration)
}

[matriv]

Yep, it makes sense, you can test both scenarios, first to hit one node and then to use round robin.

[proddata]

Not to big clusters can be easily saturated with a single thread and async calls.
Small nodes with ~2 vCPUs can handle 10-20 000 records per second (of course depending on record size)

Maybe also interesting:
https://community.crate.io/t/connecting-to-cratedb-from-go/642/1
https://github.com/proddata/nodeIngestBench

[matriv]

@robd003 Did you have any progress on this? Do you need more help?