Added hierarchy of exceptions

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

@@ -28,9 +28,10 @@ import scala.collection.immutable.ListMap
 trait JdbcDecoder[+A] { self =>
   def unsafeDecode(columIndex: Int, rs: ResultSet): (Int, A)
 
-  final def decode(columnIndex: Int, rs: ResultSet): Either[Throwable, (Int, A)] =
-    try Right(unsafeDecode(columnIndex, rs))
-    catch { case e: JdbcDecoderError => Left(e) }
+  final def decode(columnIndex: Int, rs: ResultSet): IO[JdbcDecoderError, (Int, A)] =
+    ZIO.attempt(unsafeDecode(columnIndex, rs)).refineOrDie { case e =>
+      JdbcDecoderError(e.getMessage(), e, rs.getMetaData(), rs.getRow())
+    }
 
   final def map[B](f: A => B): JdbcDecoder[B] =
     new JdbcDecoder[B] {
@@ -137,9 +138,9 @@ object JdbcDecoder extends JdbcDecoderLowPriorityImplicits {
   implicit def optionDecoder[A](implicit decoder: JdbcDecoder[A]): JdbcDecoder[Option[A]] =
     JdbcDecoder(rs =>
       int =>
-        decoder.decode(int, rs) match {
-          case Left(_)      => None
-          case Right(value) => Option(value._2)
+        try Some(decoder.unsafeDecode(int, rs)._2)
+        catch {
+          case _: Throwable => None
         }
     )
 

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

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2022 John A. De Goes and the ZIO Contributors
+ *
+ * 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
+ *
+ *     http://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 zio.jdbc
+
+import java.io.IOException
+import java.sql.{ ResultSetMetaData, SQLException, SQLTimeoutException }
+
+/**
+ * Trait to encapsule all the exceptions employed by ZIO JDBC
+ */
+sealed trait JdbcException extends Exception
+
+/**
+ * Exceptions subtraits to specify the type of error
+ */
+sealed trait ConnectionException extends JdbcException
+sealed trait QueryException      extends JdbcException
+sealed trait CodecException      extends JdbcException with QueryException
+
+sealed trait FatalException       extends JdbcException
+sealed trait RecoverableException extends JdbcException
+
+// ZTimeoutException groups all the errors produced by SQLTimeoutException.
+sealed trait ZTimeoutException extends JdbcException
+
+/**
+ * ConnectionException. Related to the connection operations with a database
+ */
+final case class DriverNotFound(cause: Throwable, driver: String)
+    extends Exception(s"Could not found driver: $driver", cause)
+    with ConnectionException
+    with FatalException
+final case class DBError(cause: Throwable) extends Exception(cause) with ConnectionException with FatalException
+final case class FailedMakingRestorable(cause: Throwable)
+    extends Exception(cause)
+    with ConnectionException
+    with FatalException
+final case class ConnectionTimeout(cause: Throwable)
+    extends Exception(cause)
+    with ConnectionException
+    with ZTimeoutException
+    with RecoverableException
+
+/**
+ * CodecExceptions. Related to the decoding and encoding of the data in a transaction
+ */
+final case class DecodeException(cause: Throwable) extends Exception(cause) with CodecException with FatalException
+final case class JdbcDecoderError(
+  message: String,
+  cause: Throwable,
+  metadata: ResultSetMetaData,
+  row: Int,
+  column: Option[Int] = None
+) extends IOException(message, cause)
+    with CodecException
+    with FatalException
+final case class JdbcEncoderError(message: String, cause: Throwable)
+    extends IOException(message, cause)
+    with CodecException
+    with FatalException
+
+/**
+ * FailedQueries. Related to the failure of actions executed directly on a database
+ */
+final case class ZSQLException(cause: SQLException)
+    extends Exception(cause)
+    with QueryException
+    with RecoverableException
+final case class ZSQLTimeoutException(cause: SQLTimeoutException)
+    extends Exception(cause)
+    with QueryException
+    with ZTimeoutException
+    with RecoverableException

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

@@ -18,74 +18,74 @@ package zio.jdbc
 import zio._
 import zio.stream._
 
-final case class Query[+A](sql: SqlFragment, decode: ZResultSet => A) {
+import java.sql.{ SQLException, SQLTimeoutException }
+
+final case class Query[+A](decode: ZResultSet => IO[CodecException, A], sql: SqlFragment) {
 
   def as[B](implicit decoder: JdbcDecoder[B]): Query[B] =
-    Query(sql, zrs => decoder.unsafeDecode(1, zrs.resultSet)._2)
+    Query(zrs => decoder.decode(1, zrs.resultSet).map(_._2), sql)
 
   def map[B](f: A => B): Query[B] =
-    Query(sql, zrs => f(decode(zrs)))
+    Query(zrs => decode(zrs).map(f), sql)
 
   /**
    * Performs a SQL select query, returning all results in a chunk.
    */
-  def selectAll: ZIO[ZConnection, Throwable, Chunk[A]] =
+  def selectAll: ZIO[ZConnection, QueryException, Chunk[A]] =
     ZIO.scoped(for {
       zrs   <- executeQuery(sql)
-      chunk <- ZIO.attempt {
-                 val builder = ChunkBuilder.make[A]()
-                 while (zrs.next())
-                   builder += decode(zrs)
-                 builder.result()
+      chunk <- ZIO.iterate(ChunkBuilder.make[A]())(_ => zrs.next()) { builder =>
+                 for {
+                   decoded <- decode(zrs)
+                 } yield builder += decoded
                }
-    } yield chunk)
+    } yield chunk.result())
 
   /**
    * Performs a SQL select query, returning the first result, if any.
    */
-  def selectOne: ZIO[ZConnection, Throwable, Option[A]] =
+  def selectOne: ZIO[ZConnection, QueryException, Option[A]] =
     ZIO.scoped(for {
       zrs    <- executeQuery(sql)
-      option <- ZIO.attempt {
-                  if (zrs.next()) Some(decode(zrs)) else None
-                }
+      option <-
+        if (zrs.next()) decode(zrs).map(Some(_))
+        else ZIO.none
     } yield option)
 
   /**
    * Performs a SQL select query, returning a stream of results.
    */
-  def selectStream(chunkSize: => Int = ZStream.DefaultChunkSize): ZStream[ZConnection, Throwable, A] =
+  def selectStream(chunkSize: => Int = ZStream.DefaultChunkSize): ZStream[ZConnection, QueryException, A] =
     ZStream.unwrapScoped {
       for {
         zrs   <- executeQuery(sql)
         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
+                   ZIO
+                     .iterate((ChunkBuilder.make[A](chunkSize), 0)) { case (_, i) =>
+                       i < chunkSize && zrs.next()
+                     } { case (builder, i) =>
+                       for {
+                         decoded <- decode(zrs)
+                       } yield (builder += decoded, i + 1)
+                     }
+                     .map { case (builder, i) =>
+                       (builder.result(), if (i >= chunkSize) Some(()) else None)
                      }
-                     (builder.result(), if (hasNext) Some(()) else None)
-                   }
                  )
       } yield stream
     }
 
   def withDecode[B](f: ZResultSet => B): Query[B] =
     Query(sql, f)
 
-  private[jdbc] def executeQuery(sql: SqlFragment): ZIO[Scope with ZConnection, Throwable, ZResultSet] = for {
+  private[jdbc] def executeQuery(sql: SqlFragment): ZIO[Scope with ZConnection, QueryException, ZResultSet] = for {
     connection <- ZIO.service[ZConnection]
     zrs        <- connection.executeSqlWith(sql, false) { ps =>
                     ZIO.acquireRelease {
-                      ZIO.attempt(ZResultSet(ps.executeQuery()))
+                      ZIO.attempt(ZResultSet(ps.executeQuery())).refineOrDie {
+                        case e: SQLTimeoutException => ZSQLTimeoutException(e)
+                        case e: SQLException        => ZSQLException(e)
+                      }
                     }(_.close)
                   }
   } yield zrs
@@ -94,7 +94,15 @@ final case class Query[+A](sql: SqlFragment, decode: ZResultSet => A) {
 
 object Query {
 
+  def apply[A](sql: SqlFragment, decode: ZResultSet => A): Query[A] = {
+    def decodeZIO(zrs: ZResultSet): IO[DecodeException, A] =
+      ZIO.attempt(decode(zrs)).refineOrDie { case e: Throwable =>
+        DecodeException(e)
+      }
+    new Query[A](zrs => decodeZIO(zrs), sql)
+  }
+
   def fromSqlFragment[A](sql: SqlFragment)(implicit decoder: JdbcDecoder[A]): Query[A] =
-    Query[A](sql, zrs => decoder.unsafeDecode(1, zrs.resultSet)._2)
+    Query[A](sql, (zrs: ZResultSet) => decoder.unsafeDecode(1, zrs.resultSet)._2)
 
 }

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

@@ -18,7 +18,7 @@ package zio.jdbc
 import zio._
 import zio.jdbc.SqlFragment.Segment
 
-import java.sql.{ PreparedStatement, Types }
+import java.sql.{ PreparedStatement, SQLException, SQLTimeoutException, Types }
 import java.time.{ OffsetDateTime, ZoneOffset }
 import scala.language.implicitConversions
 
@@ -172,18 +172,22 @@ sealed trait SqlFragment { self =>
   /**
    * Executes a SQL statement, such as one that creates a table.
    */
-  def execute: ZIO[ZConnection, Throwable, Unit] =
-    ZIO.scoped(for {
-      connection <- ZIO.service[ZConnection]
-      _          <- connection.executeSqlWith(self, false) { ps =>
-                      ZIO.attempt(ps.executeUpdate())
-                    }
-    } yield ())
+  def execute: ZIO[ZConnection, QueryException, Unit] =
+    ZIO
+      .scoped(for {
+        connection <- ZIO.service[ZConnection]
+        _          <- connection.executeSqlWith(self, false) { ps =>
+                        ZIO.attempt(ps.executeUpdate()).refineOrDie {
+                          case e: SQLTimeoutException => ZSQLTimeoutException(e)
+                          case e: SQLException        => ZSQLException(e)
+                        }
+                      }
+      } yield ())
 
   /**
    * Executes a SQL delete query.
    */
-  def delete: ZIO[ZConnection, Throwable, Long] =
+  def delete: ZIO[ZConnection, QueryException, Long] =
     ZIO.scoped(executeLargeUpdate(self))
 
   /**
@@ -212,33 +216,36 @@ sealed trait SqlFragment { self =>
    * parsed and returned as `Chunk[Long]`. If keys are non-numeric, a
    * `Chunk.empty` is returned.
    */
-  def insertWithKeys: ZIO[ZConnection, Throwable, UpdateResult[Long]] =
+  def insertWithKeys: ZIO[ZConnection, QueryException, UpdateResult[Long]] =
     ZIO.scoped(executeWithReturning(self, JdbcDecoder[Long]))
 
   /**
    * Executes a SQL update query with a RETURNING clause, materialized
    * as values of type `A`.
    */
-  def updateReturning[A: JdbcDecoder]: ZIO[ZConnection, Throwable, UpdateResult[A]] =
+  def updateReturning[A: JdbcDecoder]: ZIO[ZConnection, QueryException, UpdateResult[A]] =
     ZIO.scoped(executeWithReturning(self, JdbcDecoder[A]))
 
   /**
    * Performs a SQL update query, returning a count of rows updated.
    */
-  def update: ZIO[ZConnection, Throwable, Long] =
+  def update: ZIO[ZConnection, QueryException, Long] =
     ZIO.scoped(executeLargeUpdate(self))
 
-  private def executeLargeUpdate(sql: SqlFragment): ZIO[Scope with ZConnection, Throwable, Long] = for {
+  private def executeLargeUpdate(sql: SqlFragment): ZIO[Scope with ZConnection, QueryException, Long] = for {
     connection <- ZIO.service[ZConnection]
     count      <- connection.executeSqlWith(sql, false) { ps =>
-                    ZIO.attempt(ps.executeLargeUpdate())
+                    ZIO.attempt(ps.executeLargeUpdate()).refineOrDie {
+                      case e: SQLTimeoutException => ZSQLTimeoutException(e)
+                      case e: SQLException        => ZSQLException(e)
+                    }
                   }
   } yield count
 
   private def executeWithReturning[A](
     sql: SqlFragment,
     decoder: JdbcDecoder[A]
-  ): ZIO[Scope with ZConnection, Throwable, UpdateResult[A]] =
+  ): ZIO[Scope with ZConnection, QueryException, UpdateResult[A]] =
     for {
       updateRes       <- executeUpdate(sql, true)
       (count, maybeRs) = updateRes
@@ -250,24 +257,31 @@ sealed trait SqlFragment { self =>
                                while (rs.next())
                                  builder += decoder.unsafeDecode(1, rs.resultSet)._2
                                builder.result()
+                             }.refineOrDie {
+                               case e: SQLTimeoutException => ZSQLTimeoutException(e)
+                               case e: SQLException        => ZSQLException(e)
                              }
                          }
     } yield UpdateResult(count, keys)
 
   private[jdbc] def executeUpdate(
     sql: SqlFragment,
     returnAutoGeneratedKeys: Boolean
-  ): ZIO[Scope with ZConnection, Throwable, (Long, Option[ZResultSet])] =
+  ): ZIO[Scope with ZConnection, QueryException, (Long, Option[ZResultSet])] =
     for {
       connection <- ZIO.service[ZConnection]
       result     <- connection.executeSqlWith(sql, returnAutoGeneratedKeys) { ps =>
-                      ZIO.acquireRelease(ZIO.attempt {
-                        val rowsUpdated = ps.executeLargeUpdate()
-                        val updatedKeys = if (returnAutoGeneratedKeys) Some(ps.getGeneratedKeys) else None
-                        (rowsUpdated, updatedKeys.map(ZResultSet(_)))
-
-                      })(_._2.map(_.close).getOrElse(ZIO.unit))
-
+                      ZIO
+                        .acquireRelease(ZIO.attempt {
+                          val rowsUpdated = ps.executeLargeUpdate()
+                          val updatedKeys = if (returnAutoGeneratedKeys) Some(ps.getGeneratedKeys) else None
+                          (rowsUpdated, updatedKeys.map(ZResultSet(_)))
+
+                        })(_._2.map(_.close).getOrElse(ZIO.unit))
+                        .refineOrDie {
+                          case e: SQLTimeoutException => ZSQLTimeoutException(e)
+                          case e: SQLException        => ZSQLException(e)
+                        }
                     }
     } yield result