Skip to main content

How to Insert and Update Many Rows in Batches

In this guide, we load a feed of sensor readings into a SQLite table and then update many of them. By the end, you will know which insert method to use for a given number of rows, how to split a large feed into chunks so that a bad row costs one chunk instead of the whole import, and how to update thousands of rows with one statement or one batch.

We build it one recipe at a time:

  • Choosing between a loop of Repo#insert, Repo#insertBatch and Repo#insertAll, with timings
  • Importing a large feed in chunks, one transaction per chunk
  • What one bad row does to a batch, inside and outside transact
  • Updating many rows: one set-based UPDATE, a loop of Repo#update, and Repo#insertOrUpdateBatch

This guide builds on Build Your First Database Repository on SQLite. We use JdbcTransactor, Repo and sql"..." the way that tutorial introduces them and don't explain them again.

Saving one row at a time with an upsert, and the PRIMARY KEY that upserts need, are covered in How to Upsert Rows and Paginate with Keyset Cursors. We link to that guide instead of repeating it.

The Problem​

A feed of ten thousand sensor readings arrives as a file. Each reading must end up in the database, and a bad row must not leave the table half-loaded without anyone knowing.

With plain JDBC and the default auto-commit mode, the obvious code is one INSERT per reading:

import java.sql.Connection

// One INSERT per reading. With auto-commit on, each executeUpdate is its own transaction.
def loadReadings(conn: Connection, readings: List[(Long, Long, Double)]): Unit = {
  val ps = conn.prepareStatement("INSERT INTO reading (id, sensor_id, celsius, flagged) VALUES (?, ?, ?, 0)")
  readings.foreach { case (id, sensorId, celsius) =>
    ps.setLong(1, id)
    ps.setLong(2, sensorId)
    ps.setDouble(3, celsius)
    ps.executeUpdate()
  }
  ps.close()
}

Every row costs a round trip to the database, and with auto-commit every row also costs a commit. The load gets slower as the feed grows, and if the process dies at row 7,000, the first 7,000 rows stay and the rest are lost. Wrapping the loop in one transaction fixes the commits, but the code is still a loop that the database sees one row at a time. Loading the same feed with batches is a matter of choosing the right call and deciding where the transactions begin and end.

Prerequisites​

Add the sql module and the SQLite JDBC driver to your build.sbt:

// Scala 3.3 to 3.8
libraryDependencies += "dev.zio" %% "zio-blocks-sql" % "0.0.56"

// Scala 3.9 and newer
libraryDependencies += "dev.zio" %% "zio-blocks-next-sql" % "0.0.56"
scalacOptions += "-experimental"

// The SQLite JDBC driver, for every Scala version
libraryDependencies += "org.xerial" % "sqlite-jdbc" % "3.53.4.0"

The sql module is published for Scala 3 only. The code in this guide is the same for both artifact names.

The base imports and a few helpers go in one setup block. Reading has a generated ID, a sensor ID, a temperature and a flag:

import java.nio.file.Files
import scala.util.Try
import zio.blocks.schema.Schema
import zio.blocks.sql.*

case class Reading(id: Long, sensorId: Long, celsius: Double, flagged: Boolean)

object Reading {
  given Schema[Reading] = Schema.derived
}

val repo = Repo.derived[Reading, Long]

// Readings with IDs from `from` to `from + n - 1`, all unflagged
def sample(from: Long, n: Int): List[Reading] =
  (0 until n).toList.map { i =>
    val id = from + i
    Reading(id, id % 4 + 1, 18.0 + (id % 150) / 10.0, flagged = false)
  }

val dbFile     = Files.createTempFile("sensors", ".db")
val transactor = JdbcTransactor.fromUrl(s"jdbc:sqlite:$dbFile", SqlDialect.SQLite)

We create the table ourselves, with the primary key written out. Table#createTable emits no PRIMARY KEY, which is a known bug in 0.0.56, and the upsert and batch methods need one. The upsert guide explains the failure in more detail. Here is the table:

transactor.connect {
  sql"""CREATE TABLE reading (
    id        INTEGER PRIMARY KEY,
    sensor_id INTEGER NOT NULL,
    celsius   REAL NOT NULL,
    flagged   INTEGER NOT NULL
  )""".update
}

Choosing an Insert Method​

The library has three ways to write more than one row. Repo#insertBatch sends one prepared INSERT and all its parameter sets as a JDBC batch, then returns the total row count. Repo#insertAll builds one multi-row INSERT ... VALUES (...), (...) statement and returns the IDs of the rows. A loop of Repo#insert runs one statement per row.

The three give the same rows. We run each on the same five readings, clearing the table between runs:

val rows = sample(1L, 5)
// rows: List[Reading] = List(
//   Reading(id = 1L, sensorId = 2L, celsius = 18.1, flagged = false),
//   Reading(id = 2L, sensorId = 3L, celsius = 18.2, flagged = false),
//   Reading(id = 3L, sensorId = 4L, celsius = 18.3, flagged = false),
//   Reading(id = 4L, sensorId = 1L, celsius = 18.4, flagged = false),
//   Reading(id = 5L, sensorId = 2L, celsius = 18.5, flagged = false)
// )

transactor.transact {
  val counts = rows.map(repo.insert)
  println(s"loop of insert:  ${counts.sum} rows affected, ${repo.count} in table")
  repo.clear()
}
// loop of insert:  5 rows affected, 5 in table
// res2: Int = 5

transactor.transact {
  val affected = repo.insertBatch(rows)
  println(s"insertBatch:     $affected rows affected, ${repo.count} in table")
  repo.clear()
}
// insertBatch:     5 rows affected, 5 in table
// res3: Int = 5

transactor.transact {
  val ids = repo.insertAll(rows)
  println(s"insertAll:       returned ids ${ids.mkString("[", ", ", "]")}, ${repo.count} in table")
}
// insertAll:       returned ids [1, 2, 3, 4, 5], 5 in table

The counts agree, so the choice is about speed and about what you need back. Repo#insertBatch returns a count, not IDs, and it is the one that scales. Repo#insertAll returns the IDs from Repo#getId, in input order, which the batch method doesn't; its SQL grows with the number of rows, so it suits moderate batches. The Repo reference compares the two in detail.

We timed the three on a file-backed SQLite database, one transaction per run, with System.nanoTime. This is one run on one Linux machine with sqlite-jdbc 3.53.4.0, so read the ratios, not the milliseconds:

rows     loop of insert   insertBatch   insertAll
1,000        113.5 ms        30.1 ms      58.3 ms
5,000        183.9 ms        35.4 ms      53.8 ms
10,000       239.2 ms        62.3 ms      98.1 ms

The loop was run in one transaction too, so the gap is the per-statement cost. Running each insert in its own transaction is much worse: 1,000 readings took 19,018.7 ms that way, against 113.5 ms in one transaction. The timing program is Step5TimingExample in the companion examples.

tip

For a few rows, any of the three is fine. For thousands, use Repo#insertBatch. If you need the generated IDs of a few thousand rows, Repo#insertAll with a chunk size (see below) keeps the SQL to a sensible size.

Importing in Chunks​

A feed of a million rows should not sit in one transaction. The transaction holds locks and a large undo log until it commits, and a single bad row rolls back everything before it. Splitting the feed into chunks gives each chunk its own transaction: a failure costs one chunk, and the chunks before it stay written.

The library has no chunking helper, so we split the list ourselves with grouped and call transact once per chunk:

// Inserts rows in chunks of chunkSize, each chunk in its own transaction
def importInChunks(rows: List[Reading], chunkSize: Int): Int = {
  rows.grouped(chunkSize).foldLeft(0) { (written, chunk) =>
    written + transactor.transact(repo.insertBatch(chunk))
  }
}

We run it on a feed of twelve readings in chunks of four. The feed repeats ID 3 in its third chunk, so that chunk fails:

transactor.connect(repo.clear())
// res5: Int = 5
val feed = sample(1L, 8) ++ sample(3L, 1) ++ sample(9L, 3)
// feed: List[Reading] = List(
//   Reading(id = 1L, sensorId = 2L, celsius = 18.1, flagged = false),
//   Reading(id = 2L, sensorId = 3L, celsius = 18.2, flagged = false),
//   Reading(id = 3L, sensorId = 4L, celsius = 18.3, flagged = false),
//   Reading(id = 4L, sensorId = 1L, celsius = 18.4, flagged = false),
//   Reading(id = 5L, sensorId = 2L, celsius = 18.5, flagged = false),
//   Reading(id = 6L, sensorId = 3L, celsius = 18.6, flagged = false),
//   Reading(id = 7L, sensorId = 4L, celsius = 18.7, flagged = false),
//   Reading(id = 8L, sensorId = 1L, celsius = 18.8, flagged = false),
//   Reading(id = 3L, sensorId = 4L, celsius = 18.3, flagged = false),
//   Reading(id = 9L, sensorId = 2L, celsius = 18.9, flagged = false),
//   Reading(id = 10L, sensorId = 3L, celsius = 19.0, flagged = false),
//   Reading(id = 11L, sensorId = 4L, celsius = 19.1, flagged = false)
// )
println(s"feed has ${feed.size} readings, chunks of 4")
// feed has 12 readings, chunks of 4

Try(importInChunks(feed, 4)) match {
  case scala.util.Success(n) => println(s"imported $n readings")
  case scala.util.Failure(e) => println(s"import stopped: ${e.getMessage}")
}
// import stopped: [SQLITE_CONSTRAINT_PRIMARYKEY] A PRIMARY KEY constraint failed (UNIQUE constraint failed: reading.id)

transactor.connect {
  println(s"readings in table: ${repo.count}")
  println(s"ids in table: ${repo.all.map(_.id).mkString(", ")}")
}
// readings in table: 8
// ids in table: 1, 2, 3, 4, 5, 6, 7, 8

The first two chunks, IDs 1 to 8, are committed. The third chunk, which holds the repeated ID 3, rolls back completely, so none of IDs 9 to 11 are written either. The exception stops the import, so the rest of the feed is never tried. If you want to keep going past a bad chunk, catch the exception inside the foldLeft function, log the chunk, and continue. Pick the chunk size from how much you can afford to redo: a few thousand rows is a common range, and the time each chunk takes is the measure to check.

warning

The chunk boundaries are not a resume point by themselves. If the process dies between chunks, the table holds the chunks that committed. To resume, record the last committed ID and skip the rows before it, or upsert the feed so that running it again is safe; the upsert guide shows how.

What One Bad Row Does​

Repo#insertBatch doesn't open a transaction. It runs the batch on whatever connection it is given, so what survives a failure depends on where the call sits. We feed it five readings, the third of which repeats an ID that already exists, first outside any transaction and then inside transact:

val batch = sample(1L, 5) // IDs 1 to 5; ID 3 is about to clash
// batch: List[Reading] = List(
//   Reading(id = 1L, sensorId = 2L, celsius = 18.1, flagged = false),
//   Reading(id = 2L, sensorId = 3L, celsius = 18.2, flagged = false),
//   Reading(id = 3L, sensorId = 4L, celsius = 18.3, flagged = false),
//   Reading(id = 4L, sensorId = 1L, celsius = 18.4, flagged = false),
//   Reading(id = 5L, sensorId = 2L, celsius = 18.5, flagged = false)
// )

transactor.connect {
  repo.clear()
  repo.insert(sample(3L, 1).head)
}
// res9: Int = 1

val outside = Try(transactor.connect(repo.insertBatch(batch)))
// outside: Try[Int] = Failure(
//   org.sqlite.SQLiteException: [SQLITE_CONSTRAINT_PRIMARYKEY] A PRIMARY KEY constraint failed (UNIQUE constraint failed: reading.id)
// )
println(s"without transact: ${outside.failed.map(_.getMessage).getOrElse("no error")}")
// without transact: [SQLITE_CONSTRAINT_PRIMARYKEY] A PRIMARY KEY constraint failed (UNIQUE constraint failed: reading.id)
transactor.connect(println(s"  readings in table: ${repo.count}"))
//   readings in table: 3

transactor.connect {
  repo.clear()
  repo.insert(sample(3L, 1).head)
}
// res12: Int = 1

val inside = Try(transactor.transact(repo.insertBatch(batch)))
// inside: Try[Int] = Failure(
//   org.sqlite.SQLiteException: [SQLITE_CONSTRAINT_PRIMARYKEY] A PRIMARY KEY constraint failed (UNIQUE constraint failed: reading.id)
// )
println(s"with transact:    ${inside.failed.map(_.getMessage).getOrElse("no error")}")
// with transact:    [SQLITE_CONSTRAINT_PRIMARYKEY] A PRIMARY KEY constraint failed (UNIQUE constraint failed: reading.id)
transactor.connect(println(s"  readings in table: ${repo.count}"))
//   readings in table: 1

Outside a transaction, the rows before the failing one stay: IDs 1 and 2 were written, plus the existing ID 3. Inside transact, the failure rolls the whole batch back, and only the existing row remains. This is SQLite's behaviour with this driver. The source doesn't promise that a driver keeps the rows before a failed row, so the safe rule is to always wrap a batch in transact, as the chunked import above does. transact commits when the block returns and rolls back when it throws, and the exception is rethrown.

Updating Many Rows​

The library has Repo#update, which updates one row by its ID, and Repo#insertOrUpdateBatch, which upserts many rows in one JDBC batch. It has no updateBatch, and no Repo or Frag method runs a custom UPDATE as a JDBC batch: the sql"..." interpolator returns a Frag, and a Frag runs one statement. Batching your own statement means working one level down, with the prepareStatement, addBatch and executeBatch methods of DbCon#connection, which is what Repo#insertBatch uses; this guide doesn't cover that route. So for "update many rows", you pick one of three approaches.

The three examples below share a table of twelve readings, so we reset it first:

def reset(rows: List[Reading]): Unit = transactor.connect {
  repo.clear()
  repo.insertBatch(rows)
  ()
}

reset(sample(1L, 12))

One Set-Based Statement​

When the change is the same for every row that matches a condition, write one UPDATE with a WHERE clause. The database does the whole job in one statement:

transactor.transact {
  val threshold = 18.5
  val flagged   = sql"UPDATE reading SET flagged = 1 WHERE celsius > $threshold".update
  println(s"set-based UPDATE flagged $flagged readings")
}
// set-based UPDATE flagged 7 readings

This is the fastest option when it fits, because the rows never leave the database. It doesn't fit a change that depends on each row's own values, which is where the next two come in: use Repo#update in a loop inside transact for a few rows with different values, and Repo#insertOrUpdateBatch inside transact for many full entities, accepting that it also inserts missing rows.

A Loop of Repo#update in One Transaction​

When each row gets a different value, Repo#update writes all non-ID columns of an entity by its ID, and returns 0 when no row has that ID. Calling it in a loop inside one transact keeps the work atomic:

val recalibrated = transactor.connect(repo.all.take(4)).map(r => r.copy(celsius = r.celsius + 0.5))
// recalibrated: List[Reading] = List(
//   Reading(id = 1L, sensorId = 2L, celsius = 18.6, flagged = false),
//   Reading(id = 2L, sensorId = 3L, celsius = 18.7, flagged = false),
//   Reading(id = 3L, sensorId = 4L, celsius = 18.8, flagged = false),
//   Reading(id = 4L, sensorId = 1L, celsius = 18.9, flagged = false)
// )

transactor.transact {
  val counts = recalibrated.map(repo.update)
  println(s"loop of update:          ${counts.sum} rows affected")
}
// loop of update:          4 rows affected

Each Repo#update is a separate statement, so this costs one round trip per row. It is still one transaction, so a failure rolls back the whole loop.

Repo#insertOrUpdateBatch as a Batch Update​

When each row is a full entity and you have every column's new value, Repo#insertOrUpdateBatch sends all of them as one JDBC batch, and it updates the rows that exist. It also inserts the rows that don't, so use it when a missing row is acceptable. We add one row with a new ID to the batch to show the insert side:

val recalibratedAgain = recalibrated.map(r => r.copy(celsius = r.celsius + 0.5)) :+
  Reading(99L, 1L, 21.0, flagged = false)
// recalibratedAgain: List[Reading] = List(
//   Reading(id = 1L, sensorId = 2L, celsius = 19.1, flagged = false),
//   Reading(id = 2L, sensorId = 3L, celsius = 19.2, flagged = false),
//   Reading(id = 3L, sensorId = 4L, celsius = 19.3, flagged = false),
//   Reading(id = 4L, sensorId = 1L, celsius = 19.4, flagged = false),
//   Reading(id = 99L, sensorId = 1L, celsius = 21.0, flagged = false)
// )

transactor.transact {
  val affected = repo.insertOrUpdateBatch(recalibratedAgain)
  println(s"insertOrUpdateBatch:     $affected rows affected")
}
// insertOrUpdateBatch:     5 rows affected

transactor.connect {
  println(s"readings in table: ${repo.count}")
  repo.all.take(5).foreach(println)
}
// readings in table: 13
// Reading(1,2,19.1,false)
// Reading(2,3,19.2,false)
// Reading(3,4,19.3,false)
// Reading(4,1,19.4,false)
// Reading(5,2,18.5,false)

Four of the five rows existed, so they were updated; ID 99 didn't, so it was inserted. The count is 13 (12 loaded, plus 99), and the statement's count of 5 can't tell the two apart, which is the same caveat the upsert guide gives for single rows.

Running the Examples​

All runnable code from this guide is in sql-examples/src/main/scala/batchwritesmanyrows/. Run any object from the sql-examples directory with sbt "runMain batchwritesmanyrows.<ObjectName>", for example sbt "runMain batchwritesmanyrows.Step4UpdateManyExample". Step5TimingExample prints numbers that depend on your machine.

Going Further​