Improve R2DBC result handling (#1076)

Fixes #1019

.github/workflows/gradle.yml

@@ -46,7 +46,13 @@ jobs:
       - name: Build with Gradle
         run: ./gradlew dependencyUpdates check --no-daemon --parallel --continue
         env:
-           TESTCONTAINERS_RYUK_DISABLED: true        
+           TESTCONTAINERS_RYUK_DISABLED: true
+      - name: Publish Test Report
+        if: always()
+        uses: mikepenz/action-junit-report@v2
+        with:
+           check_name: Java CI / Test Report (${{ matrix.java }})
+           report_paths: '**/build/test-results/test/TEST-*.xml'
       - name: Publish to Sonatype Snapshots
         if: success() && github.event_name == 'push' && matrix.java == '8'
         env:

data-jdbc/src/test/groovy/io/micronaut/data/jdbc/DatabaseTestPropertyProvider.groovy

@@ -4,6 +4,7 @@ import io.micronaut.data.model.query.builder.sql.Dialect
 import io.micronaut.data.runtime.config.SchemaGenerate
 import io.micronaut.test.support.TestPropertyProvider
 import org.testcontainers.containers.*
+import org.testcontainers.utility.DockerImageName
 
 trait DatabaseTestPropertyProvider implements TestPropertyProvider {
 
@@ -42,7 +43,9 @@ trait DatabaseTestPropertyProvider implements TestPropertyProvider {
             case "sqlserver":
                 return new MSSQLServerContainer<>()
             case "oracle":
-                return new OracleContainer("registry.gitlab.com/micronaut-projects/micronaut-graal-tests/oracle-database:18.4.0-xe")
+                return new OracleContainer(DockerImageName.parse("gvenzl/oracle-xe:18"))
+                        .withEnv("ORACLE_PASSWORD", "password")
+                        .withPassword("password")
             case "mariadb":
                 return new MariaDBContainer<>("mariadb:10.5")
             case "mysql":

data-model/src/main/java/io/micronaut/data/exceptions/NonUniqueResultException.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2017-2020 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.data.exceptions;
+
+/**
+ * The exception represents the error state when one result has been requested by data layer returned multiple results.
+ *
+ * @author Denis Stepanov
+ * @since 2.4.7
+ */
+public class NonUniqueResultException extends DataAccessException {
+
+    public NonUniqueResultException() {
+        super("Query did not return a unique result");
+    }
+
+    public NonUniqueResultException(String message) {
+        super(message);
+    }
+}

data-r2dbc/build.gradle

@@ -77,12 +77,7 @@ dependencies {
 
 test {
     systemProperty "oracle.jdbc.timezoneAsRegion", "false"
-    exclude "**/r2dbc/mysql/**"
-    exclude "**/r2dbc/postgres/**"
-    exclude "**/r2dbc/sqlserver/**"
     exclude "**/r2dbc/oraclexe/**"
-//    exclude "**/r2dbc/mariadb/**"
-//    exclude "**/r2dbc/h2/**"
 }
 
 micronautBuild {

data-r2dbc/src/test/groovy/io/micronaut/data/r2dbc/DatabaseTestPropertyProvider.groovy

@@ -4,6 +4,7 @@ import io.micronaut.data.model.query.builder.sql.Dialect
 import io.micronaut.data.runtime.config.SchemaGenerate
 import io.micronaut.test.support.TestPropertyProvider
 import org.testcontainers.containers.*
+import org.testcontainers.utility.DockerImageName
 
 trait DatabaseTestPropertyProvider implements TestPropertyProvider {
 
@@ -36,16 +37,16 @@ trait DatabaseTestPropertyProvider implements TestPropertyProvider {
     String getR2dbUrlSuffix(String driverName, JdbcDatabaseContainer container) {
         switch (driverName) {
             case "postgresql":
-                return "localhost:${container.getFirstMappedPort()}/${container.getDatabaseName()}"
+                return "${container.getHost()}:${container.getFirstMappedPort()}/${container.getDatabaseName()}"
             case "h2":
                 return "/testdb"
             case "sqlserver":
-                return "localhost:${container.getFirstMappedPort()}"
+                return "${container.getHost()}:${container.getFirstMappedPort()}"
             case "oracle":
-                return "localhost:${container.getFirstMappedPort()}/xe"
+                return "${container.getHost()}:${container.getFirstMappedPort()}/xe"
             case "mariadb":
             case "mysql":
-                return "${container.getUsername()}:${container.getPassword()}@localhost:${container.getFirstMappedPort()}/${container.getDatabaseName()}"
+                return "${container.getUsername()}:${container.getPassword()}@${container.getHost()}:${container.getFirstMappedPort()}/${container.getDatabaseName()}"
         }
     }
 
@@ -58,11 +59,13 @@ trait DatabaseTestPropertyProvider implements TestPropertyProvider {
             case "sqlserver":
                 return new MSSQLServerContainer<>()
             case "oracle":
-                return new OracleContainer("registry.gitlab.com/micronaut-projects/micronaut-graal-tests/oracle-database:18.4.0-xe")
+                return new OracleContainer(DockerImageName.parse("gvenzl/oracle-xe:18"))
+                        .withEnv("ORACLE_PASSWORD", "password")
+                        .withPassword("password")
             case "mariadb":
                 return new MariaDBContainer<>("mariadb:10.5")
             case "mysql":
-                return new MySQLContainer<>("mysql:8.0.17")
+                return new MySQLContainer<>(DockerImageName.parse("mysql/mysql-server:8.0").asCompatibleSubstituteFor("mysql"))
         }
     }
 

data-r2dbc/src/test/groovy/io/micronaut/data/r2dbc/PlainR2dbcSpec.groovy

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2017-2020 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.data.r2dbc
+
+import io.micronaut.context.ApplicationContext
+import io.micronaut.data.tck.entities.Author
+import io.micronaut.data.tck.repositories.AuthorRepository
+import io.micronaut.transaction.reactive.ReactiveTransactionOperations
+import io.r2dbc.spi.Connection
+import io.r2dbc.spi.ConnectionFactory
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+import spock.lang.AutoCleanup
+import spock.lang.Shared
+import spock.lang.Specification
+
+abstract class PlainR2dbcSpec extends Specification {
+
+    @AutoCleanup
+    @Shared
+    ApplicationContext context = ApplicationContext.run(properties)
+
+    ConnectionFactory connectionFactory = context.getBean(ConnectionFactory)
+
+    ReactiveTransactionOperations<Connection> reactiveTransactionOperations = context.getBean(ReactiveTransactionOperations)
+
+    protected abstract AuthorRepository getAuthorRepository()
+
+    protected String getInsertQuery() {
+        'INSERT INTO author (name) VALUES ($1)'
+    }
+
+    def "save one"() {
+        when:
+            def author = new Author(name: "Denis")
+            def result = Flux.usingWhen(connectionFactory.create(), connection -> {
+                return Flux.usingWhen(Mono.from(connection.beginTransaction()).then(Mono.just(connection)),
+                        (b) -> {
+                            return Flux.from(connection.createStatement(getInsertQuery())
+                                    .bind(0, author.getName())
+                                    .execute()).flatMap { r -> Mono.from(r.getRowsUpdated()) };
+                        },
+                        (b) -> connection.commitTransaction(),
+                        (b, throwable) -> connection.rollbackTransaction(),
+                        (b) -> connection.commitTransaction())
+
+            }, { it -> it.close() })
+                    .collectList()
+                    .block()
+        then:
+            result.size() == 1
+            result[0] == 1
+            authorRepository.findByNameContains("Denis").size() == 1
+    }
+
+    def "save one - convert flux to mono"() {
+        when:
+            def author = new Author(name: "Zed")
+            def result = Mono.from(Flux.usingWhen(connectionFactory.create(), connection -> {
+                return Flux.usingWhen(Mono.from(connection.beginTransaction()).then(Mono.just(connection)),
+                        (b) -> {
+                            return Flux.from(connection.createStatement(getInsertQuery())
+                                    .bind(0, author.getName())
+                                    .execute()).flatMap { r -> Mono.from(r.getRowsUpdated()) };
+                        },
+                        (b) -> connection.commitTransaction(),
+                        (b, throwable) -> connection.rollbackTransaction(),
+                        (b) -> connection.commitTransaction())
+
+            }, { it -> it.close() })
+            ).block()
+        then:
+            result == 1
+            if (isFailsRandomlyWhenConvertingFluxToMono()) {
+                true
+            } else {
+                authorRepository.findByNameContains("Zed").size() == (isFailsWhenConvertingFluxToMono() ? 0 : 1)
+            }
+    }
+
+    def "save one - reactiveTransactionOperations"() {
+        when:
+            def author = new Author(name: "John")
+            def result = Flux.from(reactiveTransactionOperations.withTransaction { status ->
+                return Flux.from(status.connection.createStatement(getInsertQuery())
+                        .bind(0, author.getName())
+                        .execute()).flatMap { r -> Mono.from(r.getRowsUpdated()) };
+            })
+                    .collectList()
+                    .block()
+        then:
+            result.size() == 1
+            result[0] == 1
+            authorRepository.findByNameContains("John").size() == 1
+    }
+
+    def "save one - reactiveTransactionOperations - convert flux to mono"() {
+        when:
+            def author = new Author(name: "Josh")
+            def result = Mono.from(reactiveTransactionOperations.withTransaction { status ->
+                return Flux.from(status.connection.createStatement(getInsertQuery())
+                        .bind(0, author.getName())
+                        .execute()).flatMap { r -> Mono.from(r.getRowsUpdated()) };
+            }).block()
+        then:
+            result == 1
+        if (isFailsRandomlyWhenConvertingFluxToMono()) {
+            true
+        } else {
+            authorRepository.findByNameContains("Josh").size() == (isFailsWhenConvertingFluxToMono() ? 0 : 1)
+        }
+    }
+
+    boolean isFailsWhenConvertingFluxToMono() {
+        // Override if the driver is not correctly handling `cancel` when Flux is converted to Mono
+        return false
+    }
+
+    boolean isFailsRandomlyWhenConvertingFluxToMono() {
+        return false
+    }
+
+    def "save one - without `getRowsUpdated` call nothing is saved"() {
+        when:
+            def author = new Author(name: "Fred")
+            Flux.usingWhen(connectionFactory.create(), connection -> {
+                return Flux.usingWhen(Mono.from(connection.beginTransaction()).then(Mono.just(connection)),
+                        (b) -> {
+                            return Flux.from(connection.createStatement(getInsertQuery())
+                                    .bind(0, author.getName())
+                                    .execute()).flatMap { r -> Mono.just(1) };
+                        },
+                        (b) -> connection.commitTransaction(),
+                        (b, throwable) -> connection.rollbackTransaction(),
+                        (b) -> connection.commitTransaction())
+
+            }, { it -> it.close() })
+                    .collectList()
+                    .block()
+        then:
+            authorRepository.findByNameContains("Fred").size() == (isFailsToInsertWithoutGetRowsUpdatedCall() ? 0 : 1)
+    }
+
+    boolean isFailsToInsertWithoutGetRowsUpdatedCall() {
+        // Override if the driver is not inserting a record without `getRowsUpdated` call
+        return false
+    }
+
+}

data-r2dbc/src/test/groovy/io/micronaut/data/r2dbc/h2/H2PlainR2dbcSpec.groovy

@@ -0,0 +1,29 @@
+/*
+ * Copyright 2017-2020 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.data.r2dbc.h2
+
+import groovy.transform.Memoized
+import io.micronaut.data.r2dbc.PlainR2dbcSpec
+
+class H2PlainR2dbcSpec extends PlainR2dbcSpec implements H2TestPropertyProvider {
+
+    @Memoized
+    @Override
+    H2AuthorRepository getAuthorRepository() {
+        return context.getBean(H2AuthorRepository)
+    }
+
+}

data-r2dbc/src/test/groovy/io/micronaut/data/r2dbc/mariadb/MariaDbPlainR2dbcSpec.groovy

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2017-2020 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.data.r2dbc.mariadb
+
+import groovy.transform.Memoized
+import io.micronaut.data.r2dbc.PlainR2dbcSpec
+import io.micronaut.data.r2dbc.mysql.MySqlAuthorRepository
+import io.micronaut.data.tck.repositories.AuthorRepository
+
+class MariaDbPlainR2dbcSpec extends PlainR2dbcSpec implements MariaDbTestPropertyProvider {
+
+    @Memoized
+    @Override
+    AuthorRepository getAuthorRepository() {
+        return context.getBean(MySqlAuthorRepository)
+    }
+
+    @Override
+    String getInsertQuery() {
+        return 'INSERT INTO author (name) VALUES (?)'
+    }
+}

data-r2dbc/src/test/groovy/io/micronaut/data/r2dbc/mariadb/MariaDbReactiveRepositoryPoolSpec.groovy

@@ -0,0 +1,24 @@
+/*
+ * Copyright 2017-2020 original authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.micronaut.data.r2dbc.mariadb
+
+class MariaDbReactiveRepositoryPoolSpec extends MariaDbReactiveRepositorySpec {
+
+    @Override
+    boolean usePool() {
+        return true
+    }
+}