#57 Fail if query has multiple statements

For some reason using `.single().flatMapMany()` seems to cause the Flux to get stuck sometimes, perhaps related to batching. This seems to work, although it's not really failing "fast" since it still has to execute the queries and it may have partially inserted some data for the original statement. Nonetheless, it does work sufficiently for now.

src/main/kotlin/recce/server/dataset/DataLoadDefinition.kt

@@ -7,6 +7,8 @@ import io.micronaut.data.r2dbc.operations.R2dbcOperations
 import io.micronaut.inject.qualifiers.Qualifiers
 import io.r2dbc.spi.Result
 import reactor.core.publisher.Flux
+import reactor.kotlin.core.util.function.component1
+import reactor.kotlin.core.util.function.component2
 import recce.server.PostConstructable
 import java.nio.file.Path
 import java.util.*
@@ -38,6 +40,8 @@ class DataLoadDefinition
         { it.createStatement(queryStatement).execute() },
         { it.close() }
     )
+        .index()
+        .map { (i, r) -> if (i > 0) throw IllegalArgumentException("More than one query found.") else r }
 
     val datasourceDescriptor: String
         get() = "$role(ref=$datasourceRef)"

src/test/kotlin/recce/server/dataset/DataLoadDefinitionTest.kt

@@ -14,6 +14,7 @@ import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.mockito.Answers
 import org.mockito.kotlin.*
+import reactor.core.publisher.Flux
 import reactor.core.publisher.Mono
 import reactor.kotlin.test.test
 import java.util.*
@@ -169,4 +170,31 @@ internal class DataLoadDefinitionTest {
             verify(mockConnection).close()
         }
     }
+
+    @Test
+    fun `should fail on query with more than one statement`() {
+        val result = mock<Result>()
+
+        val statement: Statement = mock {
+            on { execute() } doReturn Flux.just(result, result)
+        }
+
+        definitionQuery.dbOperations = mock(defaultAnswer = Answers.RETURNS_DEEP_STUBS) {
+            on { connectionFactory().create() } doReturn Mono.just(mockConnection)
+        }
+
+        definitionQuery.queryStatement = testQuery
+
+        whenever(mockConnection.createStatement(eq(testQuery))).thenReturn(statement)
+
+        definitionQuery.runQuery()
+            .test()
+            .expectNext(result)
+            .consumeErrorWith {
+                assertThat(it)
+                    .hasMessageContaining("More than one query found")
+                    .isExactlyInstanceOf(IllegalArgumentException::class.java)
+            }
+            .verify()
+    }
 }

src/test/kotlin/recce/server/dataset/DatasetRecServiceTest.kt

@@ -2,7 +2,8 @@ package recce.server.dataset
 
 import io.r2dbc.spi.Row
 import io.r2dbc.spi.RowMetadata
-import org.assertj.core.api.Assertions.*
+import org.assertj.core.api.Assertions.assertThat
+import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.jupiter.api.Test
 import org.mockito.ArgumentMatchers.anyList
 import org.mockito.kotlin.*