Inserting many jobs at once

Batch inserts are executed with Client.InsertMany and Client.InsertManyTx. Both take a slice of InsertManyParams structs, which like a call to a normal non-batch insert function, take job args and optional InsertOpts.

results, err := riverClient.InsertMany(ctx, []river.InsertManyParams{
    {Args: BatchInsertArgs{}},
    {Args: BatchInsertArgs{}},
    {Args: BatchInsertArgs{}},
    {Args: BatchInsertArgs{}, InsertOpts: &river.InsertOpts{Priority: 3}},
    {Args: BatchInsertArgs{}, InsertOpts: &river.InsertOpts{Priority: 4}},
})
if err != nil {
    panic(err)
}

fmt.Printf("Inserted %d jobs\n", len(results))

See the BatchInsert example for complete code.

InsertManyTx takes a transaction, and like InsertTx, all the normal transactional enqueuing benefits apply, like that jobs aren’t worked until the transaction commits, and are removed if it rolls back.

results, err := riverClient.InsertManyTx(ctx, tx, []river.InsertManyParams{
    ...
}

Pass an array of args to insert_many. Wrap an entry in InsertManyParams when it needs its own insertion options:

results = client.insert_many([
  BatchInsertArgs.new,
  River::InsertManyParams.new(BatchInsertArgs.new, priority: 3),
  River::InsertManyParams.new(BatchInsertArgs.new, priority: 4)
])

puts "Inserted #{results.length} jobs"

Wrap this call and related application writes in a transaction on the same database connection to commit them together.

Pass a collection of args to insert_many and await the batch result. Here, all three jobs use their default insertion options:

let results = client.insert_many([
    BatchInsertArgs {},
    BatchInsertArgs {},
    BatchInsertArgs {},
]).await?;

println!("Inserted {} jobs", results.len());

Use InsertManyItem for per-job options, or insert_batch for different argument types in one batch. Both support .tx(&mut tx) for insertion in an application transaction.

Each insertMany entry pairs a job definition with its arguments and optional insertion options. This lets one batch contain jobs with different priorities:

const results = await client.insertMany([
  { job: batchInsert, args: {} },
  { job: batchInsert, args: {}, options: { priority: 3 } },
  { job: batchInsert, args: {}, options: { priority: 4 } },
]);

console.log(`Inserted ${results.length} jobs`);

Entries can have different job definitions and argument types. Pass { tx } as the second argument to use an application transaction; every entry is inserted atomically.

Normal job insertions are quite fast so it’s usually not necessary to resort to batch job insertion, but it may be desirable in situations where multiple jobs are being inserted at once because batch insertion requires fewer round trips.

A third option exists for cases where thousands of jobs are being inserted at once: InsertManyFast / InsertManyFastTx. Under the hood these methods use Postgres COPY FROM (only in the riverpgxv5 driver), has a few further performance benefits like reduced logging overhead. This method has some limitations, however, such as its inability to return inserted data or to detect and handle unique conflicts cleanly without rolling back the entire transaction.

Ruby uses insert_many for bulk insertion. There is no separate public COPY insertion API.

Rust uses insert_many or insert_batch for bulk insertion. There is no separate public COPY insertion API.

TypeScript uses insertMany for bulk insertion. There is no separate public COPY insertion API.