Skip to main content

Composing operations

Operations compose as values, so multiple database actions run in a single transaction without manual connection handling.

Combining independent operations​

.combineWith() combines two operations that don't depend on each other. Both run in the same transaction, and a function combines their results:

data class User(val id: Int, val name: String)
data class Order(val id: Int, val userId: Int, val product: String)
data class Dashboard(val userCount: Long, val recentOrders: List<Order>)
data class Stats(val userCount: Long, val orderCount: Long, val revenue: Long)

val orderCodec: RowCodec<Order> =
    RowCodec.builder<Order>()
        .field(PgTypes.int4, Order::id)
        .field(PgTypes.int4, Order::userId)
        .field(PgTypes.text, Order::product)
        .build(::Order)

lateinit var tx: Transactor

// Combine two independent queries — both run in one transaction
val countUsers: OperationRead<Long> =
    sql { "SELECT count(*) FROM users" }
        .query(RowCodec.of(PgTypes.int8).exactlyOne())
val recentOrders: OperationRead<List<Order>> =
    sql { "SELECT * FROM orders ORDER BY id DESC LIMIT 10" }
        .query(orderCodec.all())

fun dashboard(): Dashboard =
    countUsers
        .combineWith(recentOrders, ::Dashboard)
        .transact(tx)

// Three-way: all run in one transaction, results combined
val countOrders: OperationRead<Long> =
    sql { "SELECT count(*) FROM orders" }
        .query(RowCodec.of(PgTypes.int8).exactlyOne())
val totalRevenue: OperationRead<Long> =
    sql { "SELECT coalesce(sum(amount), 0) FROM orders" }
        .query(RowCodec.of(PgTypes.int8).exactlyOne())

fun stats(): Stats =
    countUsers
        .combineWith(countOrders, totalRevenue, ::Stats)
        .transact(tx)

Running multiple writes​

When you have several write operations and only care about completion (not individual results), use Operation.allOf():

// Run multiple writes in one transaction, discard individual results
val insertUser: Operation<Int> =
    sql { "INSERT INTO users(name) VALUES(${PgTypes.text("Alice")})" }.update()
val insertAudit: Operation<Int> =
    sql { "INSERT INTO audit_log(action) VALUES(${PgTypes.text("user_created")})" }.update()
val updateStats: Operation<Int> =
    sql { "UPDATE stats SET user_count = user_count + 1" }.update()

fun createUserWithAudit() {
    Operation.allOf(insertUser, insertAudit, updateStats).transact(tx)
}

Sequencing a list​

When you have a dynamic list of operations, OperationRead.sequence() runs them all and collects the results:

// Execute a list of operations and collect all results
val names = listOf("Alice", "Bob", "Charlie")

fun insertAll(): List<Int> {
    val inserts: List<OperationRead<Int>> = names.map { name ->
        sql { "INSERT INTO users(name) VALUES(${PgTypes.text(name)}) RETURNING id" }
            .query(RowCodec.of(PgTypes.int4).exactlyOne())
    }

    return OperationRead.sequence(inserts).transact(tx)
}

Data flow between operations​

Use .then() to feed one operation's result into a continuation function that returns the next operation. The first operation runs, and its result becomes the input to the function:

// Reusable queries as methods
fun insertUser(name: String): OperationRead<Int> =
    Fragment.of("INSERT INTO users(name) VALUES(")
        .value(PgTypes.text, name)
        .append(") RETURNING id")
        .query(RowCodec.of(PgTypes.int4).exactlyOne())

fun ordersByUser(userId: Int): OperationRead<List<Order>> =
    Fragment.of("SELECT id, user_id, product FROM orders WHERE user_id = ")
        .value(PgTypes.int4, userId)
        .query(orderCodec.all())

// Chain: insert user, then use returned id to fetch their orders.
fun insertAndFetchOrders(): List<Order> =
    insertUser("Alice").then { id -> ordersByUser(id) }.transact(tx)

When the first operation returns a record, you can destructure inside the continuation:

// Insert and return the new user (id + name).
fun insertUser(name: String): OperationRead<NewUser> =
    Fragment.of("INSERT INTO users(name) VALUES(")
        .value(PgTypes.text, name)
        .append(") RETURNING id, name")
        .query(newUserCodec.exactlyOne())

// Log the creation, taking the new user as input.
fun logCreation(user: NewUser): Operation<Int> =
    Fragment.of("INSERT INTO audit_log(user_id, username) VALUES(")
        .value(PgTypes.int4, user.id)
        .append(", ")
        .value(PgTypes.text, user.name)
        .append(")")
        .update()

// Chain: insertUser → returned NewUser → logCreation.
fun insertAndLog(): Int =
    insertUser("Alice").then { user -> logCreation(user) }.transact(tx)

Conditional execution​

OperationRead.ifEmpty() implements the find-or-create pattern: run the first operation, and if it returns empty (empty Optional, null, or None), run the fallback instead:

// Find-or-create pattern
fun findUser(email: String): OperationRead<User?> =
    Fragment.of("SELECT id, name, email FROM users WHERE email = ")
        .value(PgTypes.text, email)
        .query(userCodec.maxOne())

fun createUser(name: String, email: String): OperationRead<User> =
    Fragment.of("INSERT INTO users(name, email) VALUES(")
        .value(PgTypes.text, name)
        .append(", ")
        .value(PgTypes.text, email)
        .append(") RETURNING *")
        .query(userCodec.exactlyOne())

fun findOrCreate(): User =
    Operation.ifEmpty(findUser(email), createUser(name, email)).transact(tx)

Read-only composition​

When you compose OperationRead values, the result is always OperationRead. Mix in a single write Operation, and the result becomes Operation. The type system tracks this automatically:

// Combining read-only operations yields OperationRead
val bothReads: OperationRead<Pair<List<Int>, Long>> =
    findIds.combine(countUsers)

// Mixing in a write operation yields Operation (not OperationRead)
val writeAndRead: Operation<Pair<Int, List<Int>>> =
    insertUser.combine(findIds)

// transactRead works for read-only compositions
val readResult: Pair<List<Int>, Long> = bothReads.transactRead(tx)

// transact required when writes are involved
val writeResult: Pair<Int, List<Int>> = writeAndRead.transact(tx)

This means transactRead works for any tree of pure reads, and the compiler enforces it. See Read-only transactions for more.

Performance: why composition matters​

How you compose operations determines whether the execution engine can optimize them.

combine() is parallel, then() is sequential​

When you write a.combine(b), you're telling the execution engine that a and b are independent: neither needs the other's result. When you write a.then(continuation), the continuation depends on a's result.

The distinction affects execution:

CombinatorDependencyExecution
combine()IndependentParallelizable — backend decides strategy
combineWith()IndependentSame as combine
sequence()IndependentDecomposes into combine tree
allOf()IndependentDecomposes into combine tree
then()DependentSequential (must be)
ifEmpty()ConditionalSequential (check, then maybe fallback)
map()TransformNo I/O (pure function)

The OperationRunner delegates Combine nodes to a pluggable CombineStrategy. JDBC uses SEQUENTIAL (one query at a time on the same connection). Other backends can implement PARALLEL to execute both halves concurrently.

Since combine() nests, a.combine(b).combine(c).combine(d) creates a tree of Combine nodes, and the parallelization is recursive. All leaf operations are submitted concurrently when the strategy supports it.

When to use which​

Use combine() / combineWith() when queries are independent:

// Dashboard: load user + orders + preferences in ~1 RTT
var page = tx.execute(
    findUser.on(userId)
        .combineWith(getOrders.on(userId), getPrefs.on(userId),
            (user, orders, prefs) -> new DashboardPage(user, orders, prefs))
);

Use sequence() for dynamic lists of independent operations:

// Fan-out: load N items in ~1 RTT
var items = tx.execute(
    OperationRead.sequence(ids.stream()
        .map(id -> findItem.on(id))
        .toList())
);

Use then() when the next query depends on the previous result:

// Chain: insert, then read back with generated ID
var created = tx.execute(
    insertUser.on(newUser).then(findUserById)
);

Use ifEmpty() for find-or-create patterns:

// Conditional: find existing, or create if missing
var user = tx.execute(
    OperationRead.ifEmpty(findUser.on(email), createUser.on(newUser))
);
Applicative vs monadic

This design is the applicative functor pattern from functional programming. combine() is the applicative product: it declares that two computations are independent, so the runtime can execute them in parallel. then() is the monadic bind: it declares a dependency, forcing sequential execution.

Many database libraries only offer monadic composition (each query depends on the previous connection state). By also offering applicative composition, foundations-jdbc lets the pipeline optimizer batch independent queries into a single network round-trip.

Analyzing composed operations​

QueryAnalyzer walks the entire operation tree and analyzes every SQL statement in one call. See Query analysis for details.

fun analyzeComposedOperation() {
    val transaction: Operation<*> =
        insertUser("Alice").productL(allUsers)

    // Analyze every SQL statement in the tree — one call
    val results: List<QueryAnalysis> =
        QueryAnalyzer.analyze(transaction, conn)

    for (analysis in results) {
        if (!analysis.succeeded()) {
            System.err.println(analysis.report())
        }
    }
}