Don't discard the timezone information when reading timezone with tim…

build.sbt

@@ -84,11 +84,13 @@ lazy val integration = project
     publish / skip := true,
     Test / fork    := true,
     libraryDependencies ++= Seq(
-      "org.testcontainers" % "postgresql"   % "1.19.1"   % Test,
-      "org.postgresql"     % "postgresql"   % "42.6.0"   % Test,
-      "dev.zio"           %% "zio-test"     % ZioVersion % Test,
-      "dev.zio"           %% "zio-test-sbt" % ZioVersion % Test,
-      "org.slf4j"          % "slf4j-api"    % "2.0.9"    % Test,
-      "org.slf4j"          % "slf4j-simple" % "2.0.9"    % Test
+      "org.testcontainers" % "postgresql"            % "1.19.1"         % Test,
+      "org.postgresql"     % "postgresql"            % "42.6.0"         % Test,
+      "org.duckdb"         % "duckdb_jdbc"           % "0.10.1"         % Test,
+      "dev.zio"           %% "zio-schema-derivation" % ZioSchemaVersion % Test,
+      "dev.zio"           %% "zio-test"              % ZioVersion       % Test,
+      "dev.zio"           %% "zio-test-sbt"          % ZioVersion       % Test,
+      "org.slf4j"          % "slf4j-api"             % "2.0.9"          % Test,
+      "org.slf4j"          % "slf4j-simple"          % "2.0.9"          % Test
     )
   )

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

@@ -18,6 +18,7 @@ package zio.jdbc
 import zio._
 
 import java.io._
+import java.time.{ OffsetDateTime, OffsetTime }
 import java.sql.{ Array => _, _ }
 import scala.collection.immutable.ListMap
 
@@ -732,14 +733,12 @@ trait JdbcDecoderLowPriorityImplicits {
               valueOrNone(timestamp, DynamicValue.Primitive(timestamp.toLocalDateTime, StandardType.LocalDateTimeType))
 
             case SqlTypes.TIMESTAMP_WITH_TIMEZONE =>
-              // TODO: Timezone
-              val timestamp = resultSet.getTimestamp(columnIndex)
-              valueOrNone(timestamp, DynamicValue.Primitive(timestamp.toInstant(), StandardType.InstantType))
+              val timestamp = resultSet.getObject(columnIndex, classOf[OffsetDateTime])
+              valueOrNone(timestamp, DynamicValue.Primitive(timestamp, StandardType.OffsetDateTimeType))
 
             case SqlTypes.TIME_WITH_TIMEZONE =>
-              // TODO: Timezone
-              val time = resultSet.getTime(columnIndex)
-              valueOrNone(time, DynamicValue.Primitive(time.toLocalTime(), StandardType.LocalTimeType))
+              val time = resultSet.getObject(columnIndex, classOf[OffsetTime])
+              valueOrNone(time, DynamicValue.Primitive(time, StandardType.OffsetTimeType))
 
             case SqlTypes.TINYINT =>
               val short = resultSet.getShort(columnIndex)

integration/src/test/scala/zio/jdbc/DuckDbSpec.scala

@@ -0,0 +1,78 @@
+package zio.jdbc
+
+import org.duckdb.DuckDBConnection
+import org.testcontainers.jdbc.ConnectionWrapper
+import zio._
+import zio.schema.{ DeriveSchema, Schema }
+import zio.test.TestAspect.{ repeats, sequential, shrinks, withLiveClock }
+import zio.test.{ Gen, Spec, TestEnvironment, ZIOSpec, assertTrue, check }
+
+import java.sql.{ Array => _, _ }
+import java.time.{ OffsetDateTime, ZoneOffset }
+import java.time.temporal.ChronoUnit
+import java.util.Properties
+
+object DuckDbSpec extends ZIOSpec[ZConnection] {
+
+  override def bootstrap: ZLayer[Any, Any, ZConnection] =
+    ZLayer.scoped(
+      for {
+        connection  <- ZIO.acquireRelease(
+                         ZIO.succeedBlocking {
+                           val props  = new Properties()
+                           val duckDb = DriverManager
+                             .getConnection(s"jdbc:duckdb:", props)
+                             .asInstanceOf[DuckDBConnection]
+                           WorkaroundDuckDBConnection(duckDb)
+                         }
+                       )(c => ZIO.succeed(c.close()))
+        zConnection <- ZConnection.make(connection)
+      } yield zConnection
+    )
+
+  case class OffsetDateTimeRow(value: OffsetDateTime)
+  object OffsetDateTimeRow {
+    implicit val schema: Schema[OffsetDateTimeRow]           = DeriveSchema.gen[OffsetDateTimeRow]
+    implicit val jdbcDecoder: JdbcDecoder[OffsetDateTimeRow] = JdbcDecoder.fromSchema
+  }
+
+  override def spec: Spec[ZConnection with TestEnvironment with Scope, Any] = suite("DuckDB")(
+    test("should be able to decode case classes with OffsetDateTime fields and handle timezones correctly") {
+      check(
+        Gen.offsetDateTime(
+          OffsetDateTime.of(1970, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC),
+          OffsetDateTime.of(2100, 1, 1, 0, 0, 0, 0, ZoneOffset.UTC)
+        )
+      ) { offsetDateTime =>
+        for {
+          _       <- sql"""CREATE TABLE offset_datetime (value TIMESTAMP WITH TIME ZONE)""".execute
+          _       <- sql"""INSERT INTO offset_datetime VALUES ($offsetDateTime)""".execute
+          d       <- sql"""SELECT value FROM offset_datetime""".query[OffsetDateTimeRow].selectOne
+          _       <- sql"DROP TABLE offset_datetime".execute
+          expected =
+            offsetDateTime
+              .truncatedTo(ChronoUnit.MICROS)
+              .withOffsetSameInstant(ZoneOffset.UTC)
+        } yield assertTrue(
+          d.isDefined,
+          d.get.value == expected
+        )
+      }
+    }
+  ) @@ sequential @@ shrinks(0) @@ repeats(100) @@ withLiveClock
+
+  val noop = new Runnable() {
+    override def run(): Unit = ()
+  }
+
+  /**
+   * The DuckDBConnection hasn't implemented some operations that are used by zio-jdbc. This class works around those.
+   */
+  case class WorkaroundDuckDBConnection(duckDb: DuckDBConnection) extends ConnectionWrapper(duckDb, noop) {
+    // Work around this feature not being implemented by duckdb jdbc
+    override def prepareStatement(sql: String, autoGeneratedKeys: Int): PreparedStatement = prepareStatement(sql)
+    override def getClientInfo(name: String): String                                      = null
+    override def getClientInfo: Properties                                                = new Properties()
+  }
+
+}

integration/src/test/scala/zio/jdbc/JavaTimeSupportSpec.scala

@@ -59,9 +59,11 @@ object JavaTimeSupportSpec extends PgSpec {
   val genPGLocalDate: Gen[Any, LocalDate]           = Gen.localDate(MIN_LOCAL_DATE, MAX_LOCAL_DATE)
   val genPGLocalTime: Gen[Any, LocalTime]           = Gen.localTime(LocalTime.MIN, MAX_TIME)
   val genPGLocalDateTime: Gen[Any, LocalDateTime]   = Gen.localDateTime(MIN_LOCAL_DATETIME, MAX_LOCAL_DATETIME)
-  // We need to set `UTC` as PG will move the date to UTC and so can generate a date that is not in the range of `MIN_TIMESTAMP` and `MAX_TIMESTAMP`
+  // We need to limit the offset range because postgres doesn't support offsets > +/-17:00
   val genPGOffsetDateTime: Gen[Any, OffsetDateTime] =
-    Gen.offsetDateTime(MIN_OFFSET_DATETIME, MAX_OFFSET_DATETIME).map(_.withOffsetSameInstant(ZoneOffset.UTC))
+    Gen
+      .offsetDateTime(MIN_OFFSET_DATETIME, MAX_OFFSET_DATETIME)
+      .filter(_.getOffset.getTotalSeconds.abs < 17 * 60)
   val genPGInstant: Gen[Any, Instant]               = Gen.instant(MIN_TIMESTAMP, MAX_TIMESTAMP)
 
   /**