Revert "#57 Fail fast if query has zero or multiple statements"

This reverts commit d3b7913 - reverting for now, as this change appears to be breaking the batching.
thoughtworks · May 25, 2022 · 901b71b · 901b71b
1 parent 7ae43b3
commit 901b71b
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 60 deletions.
diff --git a/src/main/kotlin/recce/server/dataset/DatasetRecService.kt b/src/main/kotlin/recce/server/dataset/DatasetRecService.kt
@@ -56,9 +56,8 @@ open class DatasetRecService(
         batchSaver: (List<HashedRow>, RecRun) -> Flux<LazyDatasetMeta>
     ): Mono<DatasetMeta> =
         def.runQuery()
-            .doFirst { logger.info { "${def.datasourceDescriptor} query completed; streaming to Recce DB" } }
-            .single()
-            .flatMapMany { it.map(hashingStrategy::hash) }
+            .doOnNext { logger.info { "${def.datasourceDescriptor} query completed; streaming to Recce DB" } }
+            .flatMap { result -> result.map(hashingStrategy::hash) }
             .buffer(recConfig.defaults.batchSize)
             .zipWith(run.repeat())
             .flatMap({ (rows, run) -> batchSaver(rows, run) }, recConfig.defaults.batchConcurrency)
@@ -108,4 +107,4 @@ open class DatasetRecService(
     )
 }
 
-class DataLoadException(message: String, cause: Throwable? = null) : Exception(message, cause)
+class DataLoadException(message: String, cause: Throwable) : Exception(message, cause)
diff --git a/src/test/kotlin/recce/server/dataset/DatasetRecServiceTest.kt b/src/test/kotlin/recce/server/dataset/DatasetRecServiceTest.kt
@@ -275,60 +275,4 @@ internal class DatasetRecServiceTest {
         assertThat(DatasetRecService(RecConfiguration(datasetConfig), mock(), mock(), mock(), mock()).availableDataSets)
             .hasSameElementsAs(listOf(first, second))
     }
-
-    @Test
-    fun `should throw error if there is more than one result from query`() {
-        val multipleResults = mock<DataLoadDefinition> {
-            on { runQuery() } doReturn Flux.just(singleRowResult, singleRowResult)
-        }
-
-        val service = DatasetRecService(
-            RecConfiguration(mapOf(testDataset to DatasetConfiguration(multipleResults, multipleResults))),
-            mock(),
-            mock(),
-            runService,
-            mock()
-        )
-
-        service.runFor(testDataset)
-            .test()
-            .assertNext { assertThat(it.status).isEqualTo(RunStatus.Failed) }
-            .verifyComplete()
-
-        val errorCaptor = argumentCaptor<Throwable>()
-        verify(runService).failed(eq(recRun), errorCaptor.capture())
-
-        assertThat(errorCaptor.firstValue)
-            .isInstanceOf(DataLoadException::class.java)
-            .hasMessageContaining("Source emitted more than one item")
-            .hasCauseExactlyInstanceOf(IndexOutOfBoundsException::class.java)
-    }
-
-    @Test
-    fun `should throw error if there are no results from query`() {
-        val zeroResult = mock<DataLoadDefinition> {
-            on { runQuery() } doReturn Flux.empty()
-        }
-
-        val service = DatasetRecService(
-            RecConfiguration(mapOf(testDataset to DatasetConfiguration(zeroResult, zeroResult))),
-            mock(),
-            mock(),
-            runService,
-            mock()
-        )
-
-        service.runFor(testDataset)
-            .test()
-            .assertNext { assertThat(it.status).isEqualTo(RunStatus.Failed) }
-            .verifyComplete()
-
-        val errorCaptor = argumentCaptor<Throwable>()
-        verify(runService).failed(eq(recRun), errorCaptor.capture())
-
-        assertThat(errorCaptor.firstValue)
-            .isInstanceOf(DataLoadException::class.java)
-            .hasMessageContaining("Source was empty")
-            .hasCauseExactlyInstanceOf(NoSuchElementException::class.java)
-    }
 }