connector-rdbms-jdbc/com.river.connector.rdbms.jdbc/batchUpdate

batchUpdate

@ExperimentalCoroutinesApi

fun <T> Jdbc.batchUpdate(sql: String, upstream: Flow<T>, concurrency: Int = 1, groupStrategy: GroupStrategy = TimeWindow(100, 250.milliseconds), prepare: suspend PreparedStatement.(T) -> Unit = {}): Flow<Int>

Executes a batch update for the given SQL statement using the provided upstream Flow as input. The batch update is performed in chunks, as specified by the groupStrategy parameter, and can be executed concurrently using the specified concurrency level.

Return

A Flow of the number of rows affected by the update for each batch.

Example usage:

val data = flowOf(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
val updatedRows = Jdbc.batchUpdate("UPDATE my_table SET value = ? WHERE id = ?", data) { ps, value ->
    ps.setInt(1, value)
    ps.setInt(2, value)
}
updatedRows.collect { println("Updated $it rows.") }

Parameters

sql

the SQL statement to be executed

upstream

the Flow of input elements to be used in the batch update

concurrency

the level of concurrency to be used in the batch update (default: 1)

groupStrategy

the chunking strategy to be used for processing the input elements (default: TimeWindow(100, 250.milliseconds))

prepare

a function that prepares the PreparedStatement using the current input element (default: {})