selectStream performance (#189)

* made selectStream more efficient by batching multiple rows is a single zio and using chunks

* drop scala 2.13 api use

Author: oridag

core/src/main/scala/zio/jdbc/Query.scala

@@ -54,19 +54,27 @@ final case class Query[+A](sql: SqlFragment, decode: ZResultSet => A) {
   /**
    * Performs a SQL select query, returning a stream of results.
    */
-  def selectStream: ZStream[ZConnection, Throwable, A] =
+  def selectStream(chunkSize: => Int = ZStream.DefaultChunkSize): ZStream[ZConnection, Throwable, A] =
     ZStream.unwrapScoped {
       for {
         zrs   <- executeQuery(sql)
-        stream = ZStream.repeatZIOOption {
-                   ZIO
-                     .suspend(if (zrs.next()) ZIO.attempt(Some(decode(zrs))) else ZIO.none)
-                     .mapError(Option(_))
-                     .flatMap {
-                       case None    => ZIO.fail(None)
-                       case Some(v) => ZIO.succeed(v)
+        stream = ZStream.paginateChunkZIO(())(_ =>
+                   ZIO.attemptBlocking {
+                     val builder = ChunkBuilder.make[A](chunkSize)
+                     var hasNext = false
+                     var i       = 0
+                     while (
+                       i < chunkSize && {
+                         hasNext = zrs.next()
+                         hasNext
+                       }
+                     ) {
+                       builder.addOne(decode(zrs))
+                       i += 1
                      }
-                 }
+                     (builder.result(), if (hasNext) Some(()) else None)
+                   }
+                 )
       } yield stream
     }
 

core/src/test/scala/zio/jdbc/ZConnectionPoolSpec.scala

@@ -329,11 +329,11 @@ object ZConnectionPoolSpec extends ZIOSpecDefault {
               } +
               test("select stream") {
                 for {
-                  _     <- createUsers *> insertSherlock *> insertWatson
+                  _     <- createUsersNoId *> insertFive
                   value <- transaction {
-                             sql"select name, age from users".query[User].selectStream.runCollect
+                             sql"select name, age from users_no_id".query[UserNoId].selectStream(2).chunks.runCollect
                            }
-                } yield assertTrue(value == Chunk(sherlockHolmes, johnWatson))
+                } yield assertTrue(value == Chunk(Chunk(user1, user2), Chunk(user3, user4), Chunk(user5)))
               } +
               test("delete") {
                 for {