From 7d7d4e872e4e8f48a94555078e9e6aacbc5847f8 Mon Sep 17 00:00:00 2001 From: asmhatre Date: Thu, 29 Oct 2026 04:16:34 +0000 Subject: [PATCH] r2dbc: Spring Data R2DBC CRUD, DatabaseClient, reactive transactions and Testcontainers Postgres (post 2) Co-Authored-By: Claude Sonnet 5.5 Claude-Session: https://claude.ai/code/session_017ic4b4VQVEJWetTT1dYzyJ --- README.md | 1 + r2dbc/README.md | 25 ++ r2dbc/output/01-schema-init.txt | 1 + r2dbc/output/02-crud.txt | 6 + r2dbc/output/03-lazy-pipeline.txt | 2 + r2dbc/output/04-queries.txt | 5 + r2dbc/output/05-no-relations.txt | 6 + r2dbc/output/06-transactions.txt | 6 + r2dbc/output/07-throughput-indicative.txt | 5 + .../08-missing-testcontainers-r2dbc.txt | 4 + r2dbc/pom.xml | 97 ++++++++ r2dbc/scripts/repro-missing-artifact.sh | 20 ++ r2dbc/scripts/run-all.sh | 9 + .../main/java/com/ankurm/r2dbc/Account.java | 10 + .../java/com/ankurm/r2dbc/AccountService.java | 71 ++++++ .../main/java/com/ankurm/r2dbc/Customer.java | 15 ++ .../ankurm/r2dbc/CustomerOrderService.java | 85 +++++++ .../com/ankurm/r2dbc/CustomerRepository.java | 23 ++ .../com/ankurm/r2dbc/CustomerWithOrders.java | 7 + .../java/com/ankurm/r2dbc/PurchaseOrder.java | 12 + .../ankurm/r2dbc/PurchaseOrderRepository.java | 8 + .../ankurm/r2dbc/R2dbcDemoApplication.java | 9 + .../src/main/resources/application.properties | 5 + r2dbc/src/main/resources/schema.sql | 22 ++ .../java/com/ankurm/r2dbc/R2dbcDemoTest.java | 218 ++++++++++++++++++ .../r2dbc/ThroughputComparisonTest.java | 112 +++++++++ .../java/com/ankurm/r2dbc/Transcript.java | 29 +++ 27 files changed, 813 insertions(+) create mode 100644 r2dbc/README.md create mode 100644 r2dbc/output/01-schema-init.txt create mode 100644 r2dbc/output/02-crud.txt create mode 100644 r2dbc/output/03-lazy-pipeline.txt create mode 100644 r2dbc/output/04-queries.txt create mode 100644 r2dbc/output/05-no-relations.txt create mode 100644 r2dbc/output/06-transactions.txt create mode 100644 r2dbc/output/07-throughput-indicative.txt create mode 100644 r2dbc/output/08-missing-testcontainers-r2dbc.txt create mode 100644 r2dbc/pom.xml create mode 100755 r2dbc/scripts/repro-missing-artifact.sh create mode 100755 r2dbc/scripts/run-all.sh create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/Account.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/AccountService.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/Customer.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/CustomerOrderService.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/CustomerRepository.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/CustomerWithOrders.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrder.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrderRepository.java create mode 100644 r2dbc/src/main/java/com/ankurm/r2dbc/R2dbcDemoApplication.java create mode 100644 r2dbc/src/main/resources/application.properties create mode 100644 r2dbc/src/main/resources/schema.sql create mode 100644 r2dbc/src/test/java/com/ankurm/r2dbc/R2dbcDemoTest.java create mode 100644 r2dbc/src/test/java/com/ankurm/r2dbc/ThroughputComparisonTest.java create mode 100644 r2dbc/src/test/java/com/ankurm/r2dbc/Transcript.java diff --git a/README.md b/README.md index d79f992..4e84938 100644 --- a/README.md +++ b/README.md @@ -8,6 +8,7 @@ Companion code for a series of posts on [ankurm.com](https://ankurm.com) coverin | [`mongodb/`](./mongodb) | Spring Boot 4 with MongoDB: Spring Data MongoDB, Aggregations and Transactions | MongoDB 9.0.2 single-node replica set, Spring Data MongoDB | | [`elasticsearch/`](./elasticsearch) | Full-Text Search with Spring Data Elasticsearch on Spring Boot 4 | Elasticsearch 9.4.8, Spring Data Elasticsearch 6.1.1 | | [`distributed-cache/`](./distributed-cache) | Distributed Caching in Spring Boot: Hazelcast vs Infinispan vs Redis | Hazelcast 5.5.0 (embedded), Infinispan 16.1.4 (embedded), Redis 8.8.3 (Testcontainers) | +| [`r2dbc/`](./r2dbc) | Reactive Data Access with Spring Data R2DBC: CRUD, Transactions and When to Choose It | Spring Data R2DBC 4.1.2, r2dbc-postgresql 1.1.3.RELEASE, PostgreSQL 17 (Testcontainers 2.0.5) | Each module has its own `pom.xml`, a `docker-compose.yml` for its backing services, and a `docs/output/` directory holding the real, re-runnable transcripts that every number quoted in its post traces back to. There are no narrative doc chapters here — the deeper material for each post lives in accordions inside the post itself. diff --git a/r2dbc/README.md b/r2dbc/README.md new file mode 100644 index 0000000..70228f6 --- /dev/null +++ b/r2dbc/README.md @@ -0,0 +1,25 @@ +# r2dbc + +Companion code for the ankurm.com post +[Reactive Data Access with Spring Data R2DBC: CRUD, Transactions and When to Choose It](https://ankurm.com/spring-data-r2dbc-crud-transactions/). + +Spring Boot 4.1.2, JDK 25, Spring Data R2DBC, r2dbc-postgresql 1.1.3.RELEASE, PostgreSQL 17 (Testcontainers 2.0.5). + +## What is in here + +- `Customer`, `PurchaseOrder`, `Account`: plain records, one row each. No relationship mapping. +- `CustomerRepository`: `ReactiveCrudRepository`, derived queries, `@Query`, `@Modifying`. +- `CustomerOrderService`: three ways to load customers with orders (1+N, two queries, one JOIN with `DatabaseClient`). +- `AccountService`: `@Transactional` and `TransactionalOperator`, with rollback on error. +- `schema.sql`: run by Spring Boot through the R2DBC connection at startup. +- Tests use `StepVerifier` and write the transcripts in `output/`. + +## Run + +```bash +cd r2dbc +./scripts/run-all.sh # needs Docker; regenerates output/*.txt +``` + +`output/07-throughput-indicative.txt` is an indicative JDBC vs R2DBC run on a shared 2-vCPU sandbox. +Treat it as a harness, not a result: run it on your own hardware. diff --git a/r2dbc/output/01-schema-init.txt b/r2dbc/output/01-schema-init.txt new file mode 100644 index 0000000..58af37c --- /dev/null +++ b/r2dbc/output/01-schema-init.txt @@ -0,0 +1 @@ +tables created by schema.sql: [account, customer, purchase_order] diff --git a/r2dbc/output/02-crud.txt b/r2dbc/output/02-crud.txt new file mode 100644 index 0000000..5fe7ccc --- /dev/null +++ b/r2dbc/output/02-crud.txt @@ -0,0 +1,6 @@ +save -> id=1 name=Asha Rao +findById -> Customer[id=1, name=Asha Rao, email=asha@example.com, tier=GOLD] +update -> name=Asha Rao-Iyer (same id=1, SQL UPDATE) +count -> 1 +deleteById-> count now 0 +findById(9999) -> empty Mono (completes with no value, no exception) diff --git a/r2dbc/output/03-lazy-pipeline.txt b/r2dbc/output/03-lazy-pipeline.txt new file mode 100644 index 0000000..998d9bf --- /dev/null +++ b/r2dbc/output/03-lazy-pipeline.txt @@ -0,0 +1,2 @@ +built save(...) pipeline, never subscribed -> rows in table: 0 +after .block() on the same pipeline -> rows in table: 1 diff --git a/r2dbc/output/04-queries.txt b/r2dbc/output/04-queries.txt new file mode 100644 index 0000000..3ffd04c --- /dev/null +++ b/r2dbc/output/04-queries.txt @@ -0,0 +1,5 @@ +findByTier(GOLD) -> [Asha Rao, Meera Joshi] +@Query searchByName(%RAO%) -> [Asha Rao] +@Modifying changeTier -> 1 row updated +GOLD customers now -> 3 +DatabaseClient totalSpent(ravi) -> 200.00 diff --git a/r2dbc/output/05-no-relations.txt b/r2dbc/output/05-no-relations.txt new file mode 100644 index 0000000..34e0516 --- /dev/null +++ b/r2dbc/output/05-no-relations.txt @@ -0,0 +1,6 @@ +findAll() on Customer returns rows only. There is no orders field to be lazy or eager: + Customer[id=1, name=A, email=a@example.com, tier=STANDARD] +1+N queries (4 SQL statements for 3 customers) -> [A:2, B:1, C:0] +two queries (2 SQL statements) -> [A:2, B:1, C:0] +one LEFT JOIN, grouped by hand (1 statement) -> [A:2, B:1, C:0] +all three strategies return identical aggregates: true diff --git a/r2dbc/output/06-transactions.txt b/r2dbc/output/06-transactions.txt new file mode 100644 index 0000000..a044b3a --- /dev/null +++ b/r2dbc/output/06-transactions.txt @@ -0,0 +1,6 @@ +NO transaction, failure after debit -> alice=60.00 bob=50.00 (money vanished) +@Transactional, failure after debit -> alice=100.00 bob=50.00 (rolled back) +TransactionalOperator, failure -> alice=100.00 bob=50.00 (rolled back) +TransactionalOperator, success -> alice=60.00 bob=90.00 (committed) +@Transactional, CHECK(balance>=0) violated on debit -> DataIntegrityViolationException; alice=100.00 bob=50.00 +transfer to a non-existent account -> no error at all; alice=90.00 (debited, credit hit 0 rows) diff --git a/r2dbc/output/07-throughput-indicative.txt b/r2dbc/output/07-throughput-indicative.txt new file mode 100644 index 0000000..66d6bb0 --- /dev/null +++ b/r2dbc/output/07-throughput-indicative.txt @@ -0,0 +1,5 @@ +INDICATIVE, shared 2-vCPU sandbox, database in a container on the same host. +4000 x 'SELECT ... WHERE id = ?' per round, 100 in flight, pool size 20, 5 rounds after 1 warm-up each. +queries/second, sorted: + R2DBC DatabaseClient : [233, 288, 361, 427, 435] median=361 + JDBC + virtual threads : [1240, 1281, 1348, 1558, 3084] median=1348 diff --git a/r2dbc/output/08-missing-testcontainers-r2dbc.txt b/r2dbc/output/08-missing-testcontainers-r2dbc.txt new file mode 100644 index 0000000..1d4b717 --- /dev/null +++ b/r2dbc/output/08-missing-testcontainers-r2dbc.txt @@ -0,0 +1,4 @@ +mvn test with testcontainers-r2dbc removed from the pom: +Tests run: 1, Failures: 0, Errors: 1, Skipped: 0, Time elapsed: 57.47 s <<< FAILURE! -- in com.ankurm.r2dbc.R2dbcDemoTest +Failed to instantiate [io.r2dbc.pool.ConnectionPool]: Factory method 'connectionFactory' threw exception with message: org/testcontainers/r2dbc/R2DBCDatabaseContainer +Caused by: java.lang.ClassNotFoundException: org.testcontainers.r2dbc.R2DBCDatabaseContainer diff --git a/r2dbc/pom.xml b/r2dbc/pom.xml new file mode 100644 index 0000000..5625b70 --- /dev/null +++ b/r2dbc/pom.xml @@ -0,0 +1,97 @@ + + + 4.0.0 + + + org.springframework.boot + spring-boot-starter-parent + 4.1.2 + + + com.ankurm + r2dbc-demo + 1.0.0 + jar + + + 25 + 25 + + + + + org.springframework.boot + spring-boot-starter-data-r2dbc + + + org.postgresql + r2dbc-postgresql + runtime + + + + + org.springframework.boot + spring-boot-starter-jdbc + test + + + org.postgresql + postgresql + test + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.boot + spring-boot-testcontainers + test + + + org.testcontainers + testcontainers-postgresql + test + + + org.testcontainers + testcontainers-r2dbc + test + + + org.testcontainers + testcontainers-junit-jupiter + test + + + io.projectreactor + reactor-test + test + + + + + + + org.testcontainers + testcontainers-bom + 2.0.5 + pom + import + + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/r2dbc/scripts/repro-missing-artifact.sh b/r2dbc/scripts/repro-missing-artifact.sh new file mode 100755 index 0000000..7cca27b --- /dev/null +++ b/r2dbc/scripts/repro-missing-artifact.sh @@ -0,0 +1,20 @@ +#!/usr/bin/env bash +# Shows the real failure when testcontainers-r2dbc is NOT on the test classpath. +# Builds a temporary pom without that dependency, runs one test, keeps the key lines. +set -uo pipefail +cd "$(dirname "$0")/.." +python3 - <<'PY' +import re +s = open('pom.xml').read() +s = re.sub(r'\s*\s*org.testcontainers\s*testcontainers-r2dbc\s*test\s*', '', s) +open('pom-no-tc-r2dbc.xml', 'w').write(s) +PY +mvn -B -q -T1 -f pom-no-tc-r2dbc.xml test -Dtest='R2dbcDemoTest#schemaCreatedByBootAtStartup' > /tmp/no-tc-r2dbc.log 2>&1 +{ + echo "mvn test with testcontainers-r2dbc removed from the pom:" + grep -m1 -o 'Tests run:.*' /tmp/no-tc-r2dbc.log + grep -m1 -o "Failed to instantiate \[io.r2dbc.pool.ConnectionPool\]: Factory method 'connectionFactory' threw exception with message: [^ ]*" /tmp/no-tc-r2dbc.log + grep -m1 'Caused by: java.lang.ClassNotFoundException' /tmp/no-tc-r2dbc.log +} > output/08-missing-testcontainers-r2dbc.txt +rm -f pom-no-tc-r2dbc.xml +cat output/08-missing-testcontainers-r2dbc.txt diff --git a/r2dbc/scripts/run-all.sh b/r2dbc/scripts/run-all.sh new file mode 100755 index 0000000..7cf1533 --- /dev/null +++ b/r2dbc/scripts/run-all.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +# Regenerates every file in output/ by running the test suite against a real +# PostgreSQL 17 container (Testcontainers). Needs a running Docker daemon. +set -euo pipefail +cd "$(dirname "$0")/.." +rm -f output/*.txt +mvn -B -q test +./scripts/repro-missing-artifact.sh >/dev/null +echo "wrote:"; ls output diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/Account.java b/r2dbc/src/main/java/com/ankurm/r2dbc/Account.java new file mode 100644 index 0000000..279c0f1 --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/Account.java @@ -0,0 +1,10 @@ +package com.ankurm.r2dbc; + +import org.springframework.data.annotation.Id; +import org.springframework.data.relational.core.mapping.Table; + +import java.math.BigDecimal; + +@Table("account") +public record Account(@Id Long id, String owner, BigDecimal balance) { +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/AccountService.java b/r2dbc/src/main/java/com/ankurm/r2dbc/AccountService.java new file mode 100644 index 0000000..6971ff8 --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/AccountService.java @@ -0,0 +1,71 @@ +package com.ankurm.r2dbc; + +import org.springframework.r2dbc.core.DatabaseClient; +import org.springframework.stereotype.Service; +import org.springframework.transaction.ReactiveTransactionManager; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.reactive.TransactionalOperator; +import reactor.core.publisher.Mono; + +import java.math.BigDecimal; + +@Service +public class AccountService { + + private final DatabaseClient db; + private final TransactionalOperator tx; + + public AccountService(DatabaseClient db, ReactiveTransactionManager tm) { + this.db = db; + this.tx = TransactionalOperator.create(tm); + } + + private Mono adjust(long id, BigDecimal delta) { + return db.sql("UPDATE account SET balance = balance + :d WHERE id = :id") + .bind("d", delta).bind("id", id) + .fetch().rowsUpdated(); + } + + /** Declarative: the returned Mono is wrapped in a transaction by the proxy. */ + @Transactional + public Mono transfer(long from, long to, BigDecimal amount) { + return adjust(from, amount.negate()) + .then(adjust(to, amount)) + .then(); + } + + /** Same work, no transaction. The debit commits on its own before the failure. */ + public Mono transferNoTx(long from, long to, BigDecimal amount) { + return adjust(from, amount.negate()) + .then(Mono.error(new IllegalStateException("simulated failure after debit"))) + .then(adjust(to, amount)) + .then(); + } + + /** Programmatic: TransactionalOperator wraps exactly the pipeline you hand it. */ + public Mono transferWithOperator(long from, long to, BigDecimal amount, boolean failAfterDebit) { + Mono work = adjust(from, amount.negate()) + .then(failAfterDebit + ? Mono.error(new IllegalStateException("simulated failure after debit")) + : Mono.just(0L)) + .then(adjust(to, amount)) + .then(); + return tx.transactional(work); + } + + /** Declarative, failing: the exception rolls the debit back. */ + @Transactional + public Mono transferThenFail(long from, long to, BigDecimal amount) { + return adjust(from, amount.negate()) + .then(Mono.error(new IllegalStateException("simulated failure after debit"))) + .then(adjust(to, amount)) + .then(); + } + + public Mono balance(long id) { + return db.sql("SELECT balance FROM account WHERE id = :id") + .bind("id", id) + .map(row -> row.get("balance", BigDecimal.class)) + .one(); + } +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/Customer.java b/r2dbc/src/main/java/com/ankurm/r2dbc/Customer.java new file mode 100644 index 0000000..0f2fe85 --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/Customer.java @@ -0,0 +1,15 @@ +package com.ankurm.r2dbc; + +import org.springframework.data.annotation.Id; +import org.springframework.data.relational.core.mapping.Table; + +/** + * A plain row. Note what is NOT here: no {@code List} field, no + * @OneToMany. Spring Data R2DBC maps one row to one object and nothing else. + */ +@Table("customer") +public record Customer(@Id Long id, String name, String email, String tier) { + public static Customer ofNew(String name, String email, String tier) { + return new Customer(null, name, email, tier); + } +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerOrderService.java b/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerOrderService.java new file mode 100644 index 0000000..ff4df48 --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerOrderService.java @@ -0,0 +1,85 @@ +package com.ankurm.r2dbc; + +import org.springframework.r2dbc.core.DatabaseClient; +import org.springframework.stereotype.Service; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +import java.math.BigDecimal; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * Three ways to get "customers with their orders" when there is no + * relationship mapping. Each makes the cost of that choice visible. + */ +@Service +public class CustomerOrderService { + + private final CustomerRepository customers; + private final PurchaseOrderRepository orders; + private final DatabaseClient db; + + public CustomerOrderService(CustomerRepository customers, PurchaseOrderRepository orders, DatabaseClient db) { + this.customers = customers; + this.orders = orders; + this.db = db; + } + + /** Way 1: 1 + N queries. Correct, readable, and one round trip per customer. */ + public Flux loadNPlusOne() { + return customers.findAll() + .concatMap(c -> orders.findByCustomerId(c.id()).collectList() + .map(list -> new CustomerWithOrders(c, list))); + } + + /** Way 2: two queries total, joined in memory. */ + public Flux loadTwoQueries() { + return customers.findAll().collectList().flatMapMany(cs -> + orders.findAll().collectList().flatMapMany(os -> { + Map> byCustomer = new LinkedHashMap<>(); + os.forEach(o -> byCustomer.computeIfAbsent(o.customerId(), k -> new ArrayList<>()).add(o)); + return Flux.fromIterable(cs).map(c -> + new CustomerWithOrders(c, byCustomer.getOrDefault(c.id(), List.of()))); + })); + } + + /** Way 3: one JOIN with DatabaseClient, grouped by hand. */ + public Flux loadWithJoin() { + return db.sql(""" + SELECT c.id AS cid, c.name, c.email, c.tier, + o.id AS oid, o.total + FROM customer c LEFT JOIN purchase_order o ON o.customer_id = c.id + ORDER BY c.id, o.id + """) + .map((row, meta) -> new Object[]{ + new Customer(row.get("cid", Long.class), row.get("name", String.class), + row.get("email", String.class), row.get("tier", String.class)), + row.get("oid", Long.class), + row.get("total", BigDecimal.class)}) + .all() + .bufferUntilChanged(r -> ((Customer) r[0]).id()) + .map(group -> { + Customer c = (Customer) group.get(0)[0]; + List os = group.stream() + .filter(r -> r[1] != null) + .map(r -> new PurchaseOrder((Long) r[1], c.id(), (BigDecimal) r[2])) + .toList(); + return new CustomerWithOrders(c, os); + }); + } + + /** DatabaseClient with a named parameter and an aggregate - SQL stays SQL. */ + public Mono totalSpent(String email) { + return db.sql(""" + SELECT COALESCE(SUM(o.total), 0) AS spent + FROM customer c JOIN purchase_order o ON o.customer_id = c.id + WHERE c.email = :email + """) + .bind("email", email) + .map(row -> row.get("spent", BigDecimal.class)) + .one(); + } +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerRepository.java b/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerRepository.java new file mode 100644 index 0000000..1354af5 --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerRepository.java @@ -0,0 +1,23 @@ +package com.ankurm.r2dbc; + +import org.springframework.data.r2dbc.repository.Query; +import org.springframework.data.repository.reactive.ReactiveCrudRepository; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +public interface CustomerRepository extends ReactiveCrudRepository { + + /** Derived query: Spring Data builds the SQL from the method name. */ + Flux findByTier(String tier); + + Mono findByEmail(String email); + + /** Hand-written SQL: R2DBC repositories take plain SQL, not JPQL. */ + @Query("SELECT * FROM customer WHERE lower(name) LIKE lower(:pattern) ORDER BY name") + Flux searchByName(String pattern); + + /** Modifying query: must return {@code Mono} (rows updated), {@code Mono} or {@code Mono}. */ + @org.springframework.data.r2dbc.repository.Modifying + @Query("UPDATE customer SET tier = :tier WHERE id = :id") + Mono changeTier(Long id, String tier); +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerWithOrders.java b/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerWithOrders.java new file mode 100644 index 0000000..913f39c --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/CustomerWithOrders.java @@ -0,0 +1,7 @@ +package com.ankurm.r2dbc; + +import java.util.List; + +/** The aggregate you assemble yourself, because the mapping layer will not. */ +public record CustomerWithOrders(Customer customer, List orders) { +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrder.java b/r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrder.java new file mode 100644 index 0000000..24894b6 --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrder.java @@ -0,0 +1,12 @@ +package com.ankurm.r2dbc; + +import org.springframework.data.annotation.Id; +import org.springframework.data.relational.core.mapping.Column; +import org.springframework.data.relational.core.mapping.Table; + +import java.math.BigDecimal; + +/** The child row holds only the foreign key value, never a Customer object. */ +@Table("purchase_order") +public record PurchaseOrder(@Id Long id, @Column("customer_id") Long customerId, BigDecimal total) { +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrderRepository.java b/r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrderRepository.java new file mode 100644 index 0000000..a24e8bd --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/PurchaseOrderRepository.java @@ -0,0 +1,8 @@ +package com.ankurm.r2dbc; + +import org.springframework.data.repository.reactive.ReactiveCrudRepository; +import reactor.core.publisher.Flux; + +public interface PurchaseOrderRepository extends ReactiveCrudRepository { + Flux findByCustomerId(Long customerId); +} diff --git a/r2dbc/src/main/java/com/ankurm/r2dbc/R2dbcDemoApplication.java b/r2dbc/src/main/java/com/ankurm/r2dbc/R2dbcDemoApplication.java new file mode 100644 index 0000000..65b4e0e --- /dev/null +++ b/r2dbc/src/main/java/com/ankurm/r2dbc/R2dbcDemoApplication.java @@ -0,0 +1,9 @@ +package com.ankurm.r2dbc; + +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; + +@SpringBootApplication +public class R2dbcDemoApplication { + public static void main(String[] args) { SpringApplication.run(R2dbcDemoApplication.class, args); } +} diff --git a/r2dbc/src/main/resources/application.properties b/r2dbc/src/main/resources/application.properties new file mode 100644 index 0000000..b7ac9c1 --- /dev/null +++ b/r2dbc/src/main/resources/application.properties @@ -0,0 +1,5 @@ +# Boot runs schema.sql through the R2DBC ConnectionFactory (not JDBC) at startup. +spring.sql.init.mode=always +spring.r2dbc.pool.initial-size=5 +spring.r2dbc.pool.max-size=20 +logging.level.org.springframework.r2dbc=INFO diff --git a/r2dbc/src/main/resources/schema.sql b/r2dbc/src/main/resources/schema.sql new file mode 100644 index 0000000..5a1f2c4 --- /dev/null +++ b/r2dbc/src/main/resources/schema.sql @@ -0,0 +1,22 @@ +DROP TABLE IF EXISTS purchase_order; +DROP TABLE IF EXISTS account; +DROP TABLE IF EXISTS customer; + +CREATE TABLE customer ( + id BIGSERIAL PRIMARY KEY, + name TEXT NOT NULL, + email TEXT NOT NULL UNIQUE, + tier TEXT NOT NULL DEFAULT 'STANDARD' +); + +CREATE TABLE purchase_order ( + id BIGSERIAL PRIMARY KEY, + customer_id BIGINT NOT NULL REFERENCES customer(id), + total NUMERIC(10,2) NOT NULL +); + +CREATE TABLE account ( + id BIGSERIAL PRIMARY KEY, + owner TEXT NOT NULL, + balance NUMERIC(12,2) NOT NULL CHECK (balance >= 0) +); diff --git a/r2dbc/src/test/java/com/ankurm/r2dbc/R2dbcDemoTest.java b/r2dbc/src/test/java/com/ankurm/r2dbc/R2dbcDemoTest.java new file mode 100644 index 0000000..3def7a8 --- /dev/null +++ b/r2dbc/src/test/java/com/ankurm/r2dbc/R2dbcDemoTest.java @@ -0,0 +1,218 @@ +package com.ankurm.r2dbc; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.r2dbc.core.DatabaseClient; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.postgresql.PostgreSQLContainer; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; + +import java.math.BigDecimal; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Every transcript in output/ is written by one of these tests, against a real + * PostgreSQL 17 started by Testcontainers. The schema comes from + * src/main/resources/schema.sql, run by Spring Boot at startup. + */ +@Testcontainers +@SpringBootTest +class R2dbcDemoTest { + + @Container + @ServiceConnection + static PostgreSQLContainer postgres = new PostgreSQLContainer("postgres:17-alpine"); + + @Autowired CustomerRepository customers; + @Autowired PurchaseOrderRepository orders; + @Autowired CustomerOrderService service; + @Autowired AccountService accounts; + @Autowired DatabaseClient db; + + @BeforeEach + void clean() { + db.sql("TRUNCATE purchase_order, customer, account RESTART IDENTITY CASCADE").then().block(); + } + + @Test + void schemaCreatedByBootAtStartup() { + Transcript t = new Transcript("01-schema-init.txt"); + List tables = db.sql(""" + SELECT table_name FROM information_schema.tables + WHERE table_schema = 'public' ORDER BY table_name""") + .map(r -> r.get("table_name", String.class)).all().collectList().block(); + t.line("tables created by schema.sql: %s", tables); + assertEquals(List.of("account", "customer", "purchase_order"), tables); + t.write(); + } + + @Test + void crudWithStepVerifier() { + Transcript t = new Transcript("02-crud.txt"); + + StepVerifier.create(customers.save(Customer.ofNew("Asha Rao", "asha@example.com", "GOLD"))) + .assertNext(c -> { assertNotNull(c.id()); t.line("save -> id=%d name=%s", c.id(), c.name()); }) + .verifyComplete(); + + Customer asha = customers.findByEmail("asha@example.com").block(); + StepVerifier.create(customers.findById(asha.id())) + .assertNext(c -> t.line("findById -> %s", c)) + .verifyComplete(); + + StepVerifier.create(customers.save(new Customer(asha.id(), "Asha Rao-Iyer", asha.email(), "GOLD"))) + .assertNext(c -> t.line("update -> name=%s (same id=%d, SQL UPDATE)", c.name(), c.id())) + .verifyComplete(); + + StepVerifier.create(customers.count()).expectNext(1L).verifyComplete(); + t.line("count -> 1"); + + StepVerifier.create(customers.deleteById(asha.id())).verifyComplete(); + StepVerifier.create(customers.count()).expectNext(0L).verifyComplete(); + t.line("deleteById-> count now 0"); + + StepVerifier.create(customers.findById(9999L)).verifyComplete(); + t.line("findById(9999) -> empty Mono (completes with no value, no exception)"); + t.write(); + } + + @Test + void nothingHappensUntilYouSubscribe() { + Transcript t = new Transcript("03-lazy-pipeline.txt"); + Mono unsubscribed = customers.save(Customer.ofNew("Ghost", "ghost@example.com", "STANDARD")); + long before = customers.count().block(); + t.line("built save(...) pipeline, never subscribed -> rows in table: %d", before); + assertEquals(0L, before); + unsubscribed.block(); + long after = customers.count().block(); + t.line("after .block() on the same pipeline -> rows in table: %d", after); + assertEquals(1L, after); + t.write(); + } + + @Test + void derivedAndCustomQueries() { + Transcript t = new Transcript("04-queries.txt"); + customers.saveAll(List.of( + Customer.ofNew("Asha Rao", "asha@example.com", "GOLD"), + Customer.ofNew("Ravi Kulkarni", "ravi@example.com", "STANDARD"), + Customer.ofNew("Meera Joshi", "meera@example.com", "GOLD"))).blockLast(); + + StepVerifier.create(customers.findByTier("GOLD").map(Customer::name).collectList()) + .assertNext(n -> { t.line("findByTier(GOLD) -> %s", n); assertEquals(2, n.size()); }) + .verifyComplete(); + + StepVerifier.create(customers.searchByName("%RAO%").map(Customer::name)) + .expectNext("Asha Rao") + .verifyComplete(); + t.line("@Query searchByName(%%RAO%%) -> [Asha Rao]"); + + Customer ravi = customers.findByEmail("ravi@example.com").block(); + StepVerifier.create(customers.changeTier(ravi.id(), "GOLD")).expectNext(1L).verifyComplete(); + t.line("@Modifying changeTier -> 1 row updated"); + StepVerifier.create(customers.findByTier("GOLD").count()).expectNext(3L).verifyComplete(); + t.line("GOLD customers now -> 3"); + + orders.saveAll(List.of( + new PurchaseOrder(null, ravi.id(), new BigDecimal("120.50")), + new PurchaseOrder(null, ravi.id(), new BigDecimal("79.50")))).blockLast(); + StepVerifier.create(service.totalSpent("ravi@example.com")) + .assertNext(s -> { t.line("DatabaseClient totalSpent(ravi) -> %s", s); assertEquals(0, s.compareTo(new BigDecimal("200.00"))); }) + .verifyComplete(); + t.write(); + } + + @Test + void noRelationsMeansYouAssembleAggregatesYourself() { + Transcript t = new Transcript("05-no-relations.txt"); + List saved = customers.saveAll(List.of( + Customer.ofNew("A", "a@example.com", "STANDARD"), + Customer.ofNew("B", "b@example.com", "STANDARD"), + Customer.ofNew("C", "c@example.com", "STANDARD"))).collectList().block(); + orders.saveAll(List.of( + new PurchaseOrder(null, saved.get(0).id(), new BigDecimal("10.00")), + new PurchaseOrder(null, saved.get(0).id(), new BigDecimal("20.00")), + new PurchaseOrder(null, saved.get(1).id(), new BigDecimal("30.00")))).blockLast(); + + t.line("findAll() on Customer returns rows only. There is no orders field to be lazy or eager:"); + t.line(" %s", customers.findById(saved.get(0).id()).block()); + + List nPlusOne = shape(service.loadNPlusOne()); + List twoQ = shape(service.loadTwoQueries()); + List join = shape(service.loadWithJoin()); + t.line("1+N queries (4 SQL statements for 3 customers) -> %s", nPlusOne); + t.line("two queries (2 SQL statements) -> %s", twoQ); + t.line("one LEFT JOIN, grouped by hand (1 statement) -> %s", join); + assertEquals(nPlusOne, twoQ); + assertEquals(nPlusOne, join); + t.line("all three strategies return identical aggregates: true"); + t.write(); + } + + private List shape(Flux f) { + return f.map(cw -> cw.customer().name() + ":" + cw.orders().size()).collectList().block(); + } + + @Test + void transactionalRollsBackOnError() { + Transcript t = new Transcript("06-transactions.txt"); + seedAccounts(); + + // 1. no transaction: the debit survives the failure + StepVerifier.create(accounts.transferNoTx(1, 2, new BigDecimal("40.00"))) + .expectErrorMessage("simulated failure after debit").verify(); + BigDecimal a1 = accounts.balance(1).block(), b1 = accounts.balance(2).block(); + t.line("NO transaction, failure after debit -> alice=%s bob=%s (money vanished)", a1, b1); + assertEquals(0, a1.compareTo(new BigDecimal("60.00"))); + assertEquals(0, b1.compareTo(new BigDecimal("50.00"))); + + // 2. @Transactional: rolled back + seedAccounts(); + StepVerifier.create(accounts.transferThenFail(1, 2, new BigDecimal("40.00"))) + .expectErrorMessage("simulated failure after debit").verify(); + BigDecimal a2 = accounts.balance(1).block(), b2 = accounts.balance(2).block(); + t.line("@Transactional, failure after debit -> alice=%s bob=%s (rolled back)", a2, b2); + assertEquals(0, a2.compareTo(new BigDecimal("100.00"))); + assertEquals(0, b2.compareTo(new BigDecimal("50.00"))); + + // 3. TransactionalOperator, failure + StepVerifier.create(accounts.transferWithOperator(1, 2, new BigDecimal("40.00"), true)) + .expectErrorMessage("simulated failure after debit").verify(); + t.line("TransactionalOperator, failure -> alice=%s bob=%s (rolled back)", + accounts.balance(1).block(), accounts.balance(2).block()); + assertEquals(0, accounts.balance(1).block().compareTo(new BigDecimal("100.00"))); + + // 4. success commits + StepVerifier.create(accounts.transferWithOperator(1, 2, new BigDecimal("40.00"), false)).verifyComplete(); + t.line("TransactionalOperator, success -> alice=%s bob=%s (committed)", + accounts.balance(1).block(), accounts.balance(2).block()); + assertEquals(0, accounts.balance(1).block().compareTo(new BigDecimal("60.00"))); + assertEquals(0, accounts.balance(2).block().compareTo(new BigDecimal("90.00"))); + + // 5. database-raised error (CHECK balance >= 0) on @Transactional + seedAccounts(); + StepVerifier.create(accounts.transfer(1, 2, new BigDecimal("500.00"))) + .expectError(DataIntegrityViolationException.class).verify(); + t.line("@Transactional, CHECK(balance>=0) violated on debit -> DataIntegrityViolationException; alice=%s bob=%s", + accounts.balance(1).block(), accounts.balance(2).block()); + + // 6. the credit side fails (unknown target is NOT an error: 0 rows updated) + StepVerifier.create(accounts.transfer(1, 999, new BigDecimal("10.00"))).verifyComplete(); + t.line("transfer to a non-existent account -> no error at all; alice=%s (debited, credit hit 0 rows)", + accounts.balance(1).block()); + t.write(); + } + + private void seedAccounts() { + db.sql("TRUNCATE account RESTART IDENTITY").then().block(); + db.sql("INSERT INTO account(owner, balance) VALUES ('alice', 100.00), ('bob', 50.00)").then().block(); + } +} diff --git a/r2dbc/src/test/java/com/ankurm/r2dbc/ThroughputComparisonTest.java b/r2dbc/src/test/java/com/ankurm/r2dbc/ThroughputComparisonTest.java new file mode 100644 index 0000000..4ed4a17 --- /dev/null +++ b/r2dbc/src/test/java/com/ankurm/r2dbc/ThroughputComparisonTest.java @@ -0,0 +1,112 @@ +package com.ankurm.r2dbc; + +import com.zaxxer.hikari.HikariConfig; +import com.zaxxer.hikari.HikariDataSource; +import org.junit.jupiter.api.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.boot.testcontainers.service.connection.ServiceConnection; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.r2dbc.core.DatabaseClient; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; +import org.testcontainers.postgresql.PostgreSQLContainer; +import reactor.core.publisher.Flux; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * INDICATIVE ONLY. Same PostgreSQL container, same query, same pool size (20), + * same logical concurrency (100): R2DBC through DatabaseClient versus plain JDBC + * (HikariCP + JdbcTemplate) on virtual threads. Runs on a shared 2-vCPU sandbox + * together with the database itself, so absolute numbers mean little; the point + * is the shape: both are limited by the same 20 connections. + */ +@Testcontainers +@SpringBootTest +class ThroughputComparisonTest { + + @Container + @ServiceConnection + static PostgreSQLContainer postgres = new PostgreSQLContainer("postgres:17-alpine"); + + @Autowired DatabaseClient db; + + static final int ROWS = 1000, QUERIES = 4_000, CONCURRENCY = 100, POOL = 20, ROUNDS = 5; + + @Test + void compare() throws Exception { + db.sql("INSERT INTO customer(name,email) SELECT 'n'||g, 'n'||g||'@x.io' FROM generate_series(1,%d) g".formatted(ROWS)) + .then().block(); + + HikariConfig cfg = new HikariConfig(); + cfg.setJdbcUrl(postgres.getJdbcUrl()); + cfg.setUsername(postgres.getUsername()); + cfg.setPassword(postgres.getPassword()); + cfg.setMaximumPoolSize(POOL); + cfg.setMinimumIdle(POOL); + + List r2 = new ArrayList<>(), jd = new ArrayList<>(); + try (HikariDataSource ds = new HikariDataSource(cfg)) { + JdbcTemplate jdbc = new JdbcTemplate(ds); + runR2dbc(); // warm-up, discarded + runJdbc(jdbc); // warm-up, discarded + for (int i = 0; i < ROUNDS; i++) { + r2.add(runR2dbc()); + jd.add(runJdbc(jdbc)); + } + } + Collections.sort(r2); + Collections.sort(jd); + + Transcript t = new Transcript("07-throughput-indicative.txt"); + t.line("INDICATIVE, shared 2-vCPU sandbox, database in a container on the same host."); + t.line("%d x 'SELECT ... WHERE id = ?' per round, %d in flight, pool size %d, %d rounds after 1 warm-up each.", + QUERIES, CONCURRENCY, POOL, ROUNDS); + t.line("queries/second, sorted:"); + t.line(" R2DBC DatabaseClient : %s median=%.0f", fmt(r2), r2.get(ROUNDS / 2)); + t.line(" JDBC + virtual threads : %s median=%.0f", fmt(jd), jd.get(ROUNDS / 2)); + t.write(); + } + + private static String fmt(List xs) { + return xs.stream().map(d -> String.format("%.0f", d)).toList().toString(); + } + + private double runR2dbc() { + long t0 = System.nanoTime(); + Long n = Flux.range(0, QUERIES) + .flatMap(i -> db.sql("SELECT name FROM customer WHERE id = :id") + .bind("id", 1L + (i % ROWS)) + .map(r -> r.get("name", String.class)).one(), CONCURRENCY) + .count().block(); + assertEquals(QUERIES, n); + return QUERIES / ((System.nanoTime() - t0) / 1e9); + } + + private double runJdbc(JdbcTemplate jdbc) throws Exception { + long t0 = System.nanoTime(); + java.util.concurrent.Semaphore inFlight = new java.util.concurrent.Semaphore(CONCURRENCY); + List> fs = new ArrayList<>(QUERIES); + try (ExecutorService ex = Executors.newVirtualThreadPerTaskExecutor()) { + for (int i = 0; i < QUERIES; i++) { + long id = 1L + (i % ROWS); + inFlight.acquire(); + fs.add(ex.submit(() -> { + try { + return jdbc.queryForObject("SELECT name FROM customer WHERE id = ?", String.class, id); + } finally { inFlight.release(); } + })); + } + for (Future f : fs) f.get(); + } + return QUERIES / ((System.nanoTime() - t0) / 1e9); + } +} diff --git a/r2dbc/src/test/java/com/ankurm/r2dbc/Transcript.java b/r2dbc/src/test/java/com/ankurm/r2dbc/Transcript.java new file mode 100644 index 0000000..5e0eb7a --- /dev/null +++ b/r2dbc/src/test/java/com/ankurm/r2dbc/Transcript.java @@ -0,0 +1,29 @@ +package com.ankurm.r2dbc; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; + +/** Writes output/NN-*.txt while a test asserts the same numbers. */ +final class Transcript { + private final List lines = new ArrayList<>(); + private final Path file; + + Transcript(String name) { this.file = Path.of("output/" + name); } + + Transcript line(String fmt, Object... args) { + String s = fmt.formatted(args); + lines.add(s); + System.out.println(s); + return this; + } + + void write() { + try { + Files.createDirectories(file.getParent()); + Files.writeString(file, String.join("\n", lines) + "\n"); + } catch (IOException e) { throw new RuntimeException(e); } + } +}