Repository: GoogleCloudPlatform/pgadapter
Pull request #565 (2022-12-28 10:47:22)

feat: support more timestamp values for query parameters
PostgreSQL allows query parameters to be given in a wide range of input formats, and the server will automatically convert this to the correct server value before using these in the query. This adds support for lenient parsing of timestamp values. These will be interpreted in the current timezone of the connection. This means that if for example a timestamp value is given as just a date value, it will be interpreted as midnight at the given date in the timezone of the current connection.

codecov[bot]: # [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> Merging [#565](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (2189bc7) into [postgresql-dialect](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/commit/71cb0a4fe6a7ff268c0b209d1cb22ffb23920d4a?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (71cb0a4) will **increase** coverage by `0.00%`.
> The diff coverage is `93.02%`.

```diff
@@                  Coverage Diff                  @@
##             postgresql-dialect     #565   +/-   ##
=====================================================
  Coverage                 88.41%   88.41%           
- Complexity                 2069     2072    +3     
=====================================================
  Files                       122      122           
  Lines                      6791     6822   +31     
  Branches                    956      960    +4     
=====================================================
+ Hits                       6004     6032   +28     
- Misses                      558      559    +1     
- Partials                    229      231    +2     
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `88.41% <93.02%> (+<0.01%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | Coverage Δ | |
|---|---|---|
| [...oud/spanner/pgadapter/parsers/TimestampParser.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3BhcnNlcnMvVGltZXN0YW1wUGFyc2VyLmphdmE=) | `90.72% <89.65%> (-1.06%)` | :arrow_down: |
| [.../cloud/spanner/pgadapter/session/CopySettings.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3Nlc3Npb24vQ29weVNldHRpbmdzLmphdmE=) | `96.61% <100.00%> (+0.05%)` | :arrow_up: |
| [...le/cloud/spanner/pgadapter/utils/CopyInParser.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3V0aWxzL0NvcHlJblBhcnNlci5qYXZh) | `50.00% <100.00%> (ø)` | |
| [...e/cloud/spanner/pgadapter/utils/CsvCopyParser.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3V0aWxzL0NzdkNvcHlQYXJzZXIuamF2YQ==) | `100.00% <100.00%> (ø)` | |
| [.../cloud/spanner/pgadapter/utils/MutationWriter.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/565/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3V0aWxzL011dGF0aW9uV3JpdGVyLmphdmE=) | `92.16% <100.00%> (+0.07%)` | :arrow_up: |

:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform)


Patch:
diff --git a/pom.xml b/pom.xml
index 320feb1e..ccf5422f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -21,7 +21,7 @@
     <skipTests>false</skipTests>
     <skipUnits>${skipTests}</skipUnits>
     <skipITs>true</skipITs>
-    <it.forkCount>8</it.forkCount>
+    <it.forkCount>16</it.forkCount>
     <clirr.skip>true</clirr.skip>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParser.java
index 0db60f15..54682366 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParser.java
@@ -26,8 +26,11 @@ import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.session.SessionState;
 import java.nio.charset.StandardCharsets;
 import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
+import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeFormatterBuilder;
 import java.time.temporal.ChronoField;
@@ -58,7 +61,7 @@ public class TimestampParser extends Parser<Timestamp> {
           .appendOffset(OptionsMetadata.isJava8() ? "+HH:mm" : "+HH:mm:ss", "+00")
           .toFormatter();
 
-  private static final DateTimeFormatter TIMESTAMP_INPUT_FORMATTER =
+  private static final DateTimeFormatter TIMESTAMPTZ_INPUT_FORMATTER =
       new DateTimeFormatterBuilder()
           .parseLenient()
           .parseCaseInsensitive()
@@ -67,6 +70,13 @@ public class TimestampParser extends Parser<Timestamp> {
           // Java 8 does not support seconds in timezone offset.
           .appendOffset(OptionsMetadata.isJava8() ? "+HH:mm" : "+HH:mm:ss", "+00:00:00")
           .toFormatter();
+  private static final DateTimeFormatter TIMESTAMP_INPUT_FORMATTER =
+      new DateTimeFormatterBuilder()
+          .parseLenient()
+          .parseCaseInsensitive()
+          .appendPattern("yyyy-MM-dd[[ ]['T']HH:mm[:ss][XXX]]")
+          .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+          .toFormatter();
 
   private final SessionState sessionState;
 
@@ -85,7 +95,8 @@ public class TimestampParser extends Parser<Timestamp> {
     if (item != null) {
       switch (formatCode) {
         case TEXT:
-          this.item = toTimestamp(new String(item, StandardCharsets.UTF_8));
+          this.item =
+              toTimestamp(new String(item, StandardCharsets.UTF_8), sessionState.getTimezone());
           break;
         case BINARY:
           this.item = toTimestamp(item);
@@ -109,16 +120,41 @@ public class TimestampParser extends Parser<Timestamp> {
     return Timestamp.ofTimeSecondsAndNanos(javaSeconds, javaNanos);
   }
 
-  /** Converts the given string value to a {@link Timestamp}. */
-  public static Timestamp toTimestamp(String value) {
+  /**
+   * Converts the given string value to a {@link Timestamp}. The given timezone is used to determine
+   * the actual point in time if the timestamp string itself does not contain a timezone offset.
+   */
+  public static Timestamp toTimestamp(String value, ZoneId timezone) {
     try {
       String stringValue = toPGString(value);
-      TemporalAccessor temporalAccessor = TIMESTAMP_INPUT_FORMATTER.parse(stringValue);
+      TemporalAccessor temporalAccessor = TIMESTAMPTZ_INPUT_FORMATTER.parse(stringValue);
       return Timestamp.ofTimeSecondsAndNanos(
           temporalAccessor.getLong(ChronoField.INSTANT_SECONDS),
           temporalAccessor.get(ChronoField.NANO_OF_SECOND));
-    } catch (Exception exception) {
-      throw PGExceptionFactory.newPGException("Invalid timestamp value: " + value);
+    } catch (Exception ignore) {
+      try {
+        TemporalAccessor temporalAccessor =
+            TIMESTAMP_INPUT_FORMATTER.parseBest(
+                value, ZonedDateTime::from, LocalDateTime::from, LocalDate::from);
+        ZonedDateTime zonedDateTime = null;
+        if (temporalAccessor instanceof ZonedDateTime) {
+          zonedDateTime = (ZonedDateTime) temporalAccessor;
+        } else if (temporalAccessor instanceof LocalDateTime) {
+          LocalDateTime localDateTime = (LocalDateTime) temporalAccessor;
+          zonedDateTime = localDateTime.atZone(timezone);
+        } else if (temporalAccessor instanceof LocalDate) {
+          LocalDate localDate = (LocalDate) temporalAccessor;
+          zonedDateTime = localDate.atStartOfDay().atZone(timezone);
+        }
+        if (zonedDateTime != null) {
+          return Timestamp.ofTimeSecondsAndNanos(
+              zonedDateTime.getLong(ChronoField.INSTANT_SECONDS),
+              zonedDateTime.get(ChronoField.NANO_OF_SECOND));
+        }
+        throw PGExceptionFactory.newPGException("Invalid timestamp value: " + value);
+      } catch (Exception exception) {
+        throw PGExceptionFactory.newPGException("Invalid timestamp value: " + value);
+      }
     }
   }
 
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/session/CopySettings.java b/src/main/java/com/google/cloud/spanner/pgadapter/session/CopySettings.java
index 3c9de5ce..1bb29f7f 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/session/CopySettings.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/session/CopySettings.java
@@ -78,6 +78,11 @@ public class CopySettings {
     this.sessionState = sessionState;
   }
 
+  /** Returns the {@link SessionState} associated with this {@link CopySettings}. */
+  public SessionState getSessionState() {
+    return this.sessionState;
+  }
+
   /** Returns the maximum number of parallel transactions for a single COPY operation. */
   public int getMaxParallelism() {
     return sessionState.getIntegerSetting("spanner", "copy_max_parallelism", 128);
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/CopyInParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/CopyInParser.java
index 32d7a6eb..813fc09a 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/CopyInParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/CopyInParser.java
@@ -17,6 +17,7 @@ package com.google.cloud.spanner.pgadapter.utils;
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement.Format;
 import java.io.IOException;
 import java.io.PipedInputStream;
@@ -32,12 +33,16 @@ public interface CopyInParser {
    * for non-binary formats.
    */
   static CopyInParser create(
-      Format format, @Nullable CSVFormat csvFormat, PipedInputStream inputStream, boolean hasHeader)
+      SessionState sessionState,
+      Format format,
+      @Nullable CSVFormat csvFormat,
+      PipedInputStream inputStream,
+      boolean hasHeader)
       throws IOException {
     switch (format) {
       case TEXT:
       case CSV:
-        return new CsvCopyParser(csvFormat, inputStream, hasHeader);
+        return new CsvCopyParser(sessionState, csvFormat, inputStream, hasHeader);
       case BINARY:
         return new BinaryCopyParser(inputStream);
       default:
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParser.java
index df3fdf0a..9265faf6 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParser.java
@@ -24,6 +24,7 @@ import com.google.cloud.spanner.Type;
 import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.pgadapter.parsers.BooleanParser;
 import com.google.cloud.spanner.pgadapter.parsers.TimestampParser;
+import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.common.collect.Iterators;
 import java.io.IOException;
 import java.io.InputStreamReader;
@@ -43,12 +44,18 @@ import org.apache.commons.csv.CSVRecord;
 class CsvCopyParser implements CopyInParser {
   private static final Logger logger = Logger.getLogger(CsvCopyParser.class.getName());
 
+  private final SessionState sessionState;
   private final CSVFormat format;
   private final boolean hasHeader;
   private final CSVParser parser;
 
-  CsvCopyParser(CSVFormat csvFormat, PipedInputStream inputStream, boolean hasHeader)
+  CsvCopyParser(
+      SessionState sessionState,
+      CSVFormat csvFormat,
+      PipedInputStream inputStream,
+      boolean hasHeader)
       throws IOException {
+    this.sessionState = sessionState;
     this.format = csvFormat;
     this.hasHeader = hasHeader;
     this.parser = createParser(inputStream);
@@ -57,7 +64,7 @@ class CsvCopyParser implements CopyInParser {
   @Override
   public Iterator<CopyRecord> iterator() {
     return Iterators.transform(
-        parser.iterator(), record -> new CsvCopyRecord(record, this.hasHeader));
+        parser.iterator(), record -> new CsvCopyRecord(this.sessionState, record, this.hasHeader));
   }
 
   @Override
@@ -75,10 +82,12 @@ class CsvCopyParser implements CopyInParser {
   }
 
   static class CsvCopyRecord implements CopyRecord {
+    private final SessionState sessionState;
     private final CSVRecord record;
     private final boolean hasHeader;
 
-    CsvCopyRecord(CSVRecord record, boolean hasHeader) {
+    CsvCopyRecord(SessionState sessionState, CSVRecord record, boolean hasHeader) {
+      this.sessionState = sessionState;
       this.record = record;
       this.hasHeader = hasHeader;
     }
@@ -96,16 +105,17 @@ class CsvCopyParser implements CopyInParser {
     @Override
     public Value getValue(Type type, String columnName) throws SpannerException {
       String recordValue = record.get(columnName);
-      return getSpannerValue(type, recordValue);
+      return getSpannerValue(sessionState, type, recordValue);
     }
 
     @Override
     public Value getValue(Type type, int columnIndex) throws SpannerException {
       String recordValue = record.get(columnIndex);
-      return getSpannerValue(type, recordValue);
+      return getSpannerValue(sessionState, type, recordValue);
     }
 
-    static Value getSpannerValue(Type type, String recordValue) throws SpannerException {
+    static Value getSpannerValue(SessionState sessionState, Type type, String recordValue)
+        throws SpannerException {
       try {
         switch (type.getCode()) {
           case STRING:
@@ -134,7 +144,9 @@ class CsvCopyParser implements CopyInParser {
             return Value.date(recordValue == null ? null : Date.parseDate(recordValue));
           case TIMESTAMP:
             Timestamp timestamp =
-                recordValue == null ? null : TimestampParser.toTimestamp(recordValue);
+                recordValue == null
+                    ? null
+                    : TimestampParser.toTimestamp(recordValue, sessionState.getTimezone());
             return Value.timestamp(timestamp);
           default:
             SpannerException spannerException =
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
index 2dc290fe..55eeda09 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
@@ -210,7 +210,9 @@ public class MutationWriter implements Callable<StatementResult>, Closeable {
   public StatementResult call() throws Exception {
     PipedInputStream inputStream = new PipedInputStream(payload, copySettings.getPipeBufferSize());
     pipeCreatedLatch.countDown();
-    final CopyInParser parser = CopyInParser.create(copyFormat, csvFormat, inputStream, hasHeader);
+    final CopyInParser parser =
+        CopyInParser.create(
+            copySettings.getSessionState(), copyFormat, csvFormat, inputStream, hasHeader);
     // This LinkedBlockingDeque holds a reference to all transactions that are currently active. The
     // max capacity of this deque is what ensures that we never have more than maxParallelism
     // transactions running at the same time. We could also achieve that by using a thread pool with


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java
index 15373904..fe7da284 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeTrue;
+import static org.mockito.Mockito.mock;
 
 import com.google.cloud.ByteArray;
 import com.google.cloud.Date;
@@ -33,6 +34,7 @@ import com.google.cloud.spanner.Type;
 import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.connection.RandomResultSetGenerator;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement.Format;
 import com.google.cloud.spanner.pgadapter.utils.CopyInParser;
 import com.google.cloud.spanner.pgadapter.utils.CopyRecord;
@@ -455,7 +457,9 @@ public class CopyOutMockServerTest extends AbstractMockServerTest {
     }
     PipedOutputStream pipedOutputStream = new PipedOutputStream();
     PipedInputStream inputStream = new PipedInputStream(pipedOutputStream, 1 << 16);
-    CopyInParser copyParser = CopyInParser.create(Format.BINARY, null, inputStream, false);
+    SessionState sessionState = mock(SessionState.class);
+    CopyInParser copyParser =
+        CopyInParser.create(sessionState, Format.BINARY, null, inputStream, false);
     int b;
     while ((b = process.getInputStream().read()) != -1) {
       pipedOutputStream.write(b);
@@ -507,7 +511,9 @@ public class CopyOutMockServerTest extends AbstractMockServerTest {
     }
     PipedOutputStream pipedOutputStream = new PipedOutputStream();
     PipedInputStream inputStream = new PipedInputStream(pipedOutputStream, 1 << 16);
-    CopyInParser copyParser = CopyInParser.create(Format.BINARY, null, inputStream, false);
+    SessionState sessionState = mock(SessionState.class);
+    CopyInParser copyParser =
+        CopyInParser.create(sessionState, Format.BINARY, null, inputStream, false);
     int b;
     while ((b = process.getInputStream().read()) != -1) {
       pipedOutputStream.write(b);
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
index ea21dd67..70660518 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
+import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.MockSpannerServiceImpl.SimulatedExecutionTime;
 import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
 import com.google.cloud.spanner.Statement;
@@ -404,6 +405,61 @@ public class EmulatedPsqlMockServerTest extends AbstractMockServerTest {
     }
   }
 
+  @Test
+  public void testDateForTimestamptzParameter() throws SQLException {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from my_table where ts=$1"),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(
+                    createAllTypesResultSetMetadata("")
+                        .toBuilder()
+                        .setUndeclaredParameters(
+                            StructType.newBuilder()
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p1")
+                                        .setType(
+                                            Type.newBuilder().setCode(TypeCode.TIMESTAMP).build())
+                                        .build())
+                                .build()))
+                .build()));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder("select * from my_table where ts=$1")
+                .bind("p1")
+                .to(Timestamp.parseTimestamp("2022-12-27T23:00:00Z"))
+                .build(),
+            createAllTypesResultSet("")));
+
+    try (Connection connection = DriverManager.getConnection(createUrl("my-db"))) {
+      connection.createStatement().execute("set time zone 'Europe/Amsterdam'");
+      connection
+          .createStatement()
+          .execute("prepare my_prepared_statement as select * from my_table where ts=$1");
+      // '2022-12-28' is interpreted in timezone 'Europe/Amsterdam', which means
+      // '2022-12-28T23:00:00Z'.
+      try (ResultSet resultSet =
+          connection
+              .createStatement()
+              .executeQuery("execute my_prepared_statement ('2022-12-28')")) {
+        assertTrue(resultSet.next());
+        assertFalse(resultSet.next());
+      }
+    }
+    assertEquals(2, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    ExecuteSqlRequest executeRequest =
+        mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(1);
+    assertEquals(QueryMode.NORMAL, executeRequest.getQueryMode());
+    assertEquals(1, executeRequest.getParamTypesCount());
+    assertEquals(
+        Type.newBuilder().setCode(TypeCode.TIMESTAMP).build(),
+        executeRequest.getParamTypesMap().get("p1"));
+    assertEquals(
+        "2022-12-27T23:00:00Z",
+        executeRequest.getParams().getFieldsMap().get("p1").getStringValue());
+  }
+
   static StatusRuntimeException newStatusResourceNotFoundException(
       String shortName, String resourceType, String resourceName) {
     ResourceInfo resourceInfo =
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
index 59bddc2f..a5d5f1fe 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
@@ -28,17 +28,25 @@ import com.google.cloud.spanner.KeySet;
 import com.google.cloud.spanner.Mutation;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.OutputStreamWriter;
+import java.io.Writer;
 import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Types;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Random;
@@ -70,6 +78,8 @@ public class ITPsqlTest implements IntegrationTest {
   public static final String POSTGRES_DATABASE = System.getenv("POSTGRES_DATABASE");
   public static final String POSTGRES_PASSWORD = System.getenv("POSTGRES_PASSWORD");
 
+  private final Random random = new Random();
+
   @BeforeClass
   public static void setup() throws Exception {
     assumeTrue("This test requires psql to be installed", isPsqlAvailable());
@@ -395,7 +405,7 @@ public class ITPsqlTest implements IntegrationTest {
         DriverManager.getConnection(createJdbcUrlForLocalPg(), POSTGRES_USER, POSTGRES_PASSWORD)) {
       try (PreparedStatement preparedStatement =
           connection.prepareStatement("insert into all_types (col_bigint) values (?)")) {
-        preparedStatement.setLong(1, new Random().nextLong());
+        preparedStatement.setLong(1, random.nextLong());
         assertEquals(1, preparedStatement.executeUpdate());
       }
     }
@@ -513,6 +523,135 @@ public class ITPsqlTest implements IntegrationTest {
     }
   }
 
+  @Test
+  public void testTimestamptzParsing() throws Exception {
+    final int numTests = 10;
+    for (int i = 0; i < numTests; i++) {
+      String timezone;
+      try (Connection pgConnection =
+          DriverManager.getConnection(
+              createJdbcUrlForLocalPg(), POSTGRES_USER, POSTGRES_PASSWORD)) {
+        int numTimezones;
+        try (ResultSet resultSet =
+            pgConnection
+                .createStatement()
+                .executeQuery(
+                    "select count(*) from pg_timezone_names where not name like '%posix%'")) {
+          assertTrue(resultSet.next());
+          numTimezones = resultSet.getInt(1);
+        }
+
+        while (true) {
+          try (ResultSet resultSet =
+              pgConnection
+                  .createStatement()
+                  .executeQuery(
+                      String.format(
+                          "select name from pg_timezone_names where not name like '%%posix%%' offset %d limit 1",
+                          random.nextInt(numTimezones)))) {
+            assertTrue(resultSet.next());
+            timezone = resultSet.getString(1);
+            if (!PROBLEMATIC_ZONE_IDS.contains(ZoneId.of(timezone))) {
+              break;
+            }
+          }
+        }
+      }
+      for (String timestamp :
+          new String[] {
+            generateRandomLocalDate().toString(),
+            generateRandomLocalDateTime().toString().replace('T', ' '),
+            generateRandomOffsetDateTime().toString().replace('T', ' ')
+          }) {
+        ProcessBuilder realPgBuilder = new ProcessBuilder();
+        realPgBuilder.command(
+            "bash",
+            "-c",
+            "psql"
+                + " -h "
+                + POSTGRES_HOST
+                + " -p "
+                + POSTGRES_PORT
+                + " -U "
+                + POSTGRES_USER
+                + " -d "
+                + POSTGRES_DATABASE);
+        setPgPassword(realPgBuilder);
+
+        ProcessBuilder pgAdapterBuilder = new ProcessBuilder();
+        pgAdapterBuilder.command(
+            "bash",
+            "-c",
+            "psql"
+                + " -h "
+                + (POSTGRES_HOST.startsWith("/") ? "/tmp" : testEnv.getPGAdapterHost())
+                + " -p "
+                + testEnv.getPGAdapterPort()
+                + " -d "
+                + database.getId().getDatabase());
+
+        String realPgOutput = "";
+        String pgAdapterOutput = "";
+        for (ProcessBuilder builder : new ProcessBuilder[] {realPgBuilder, pgAdapterBuilder}) {
+          Process process = builder.start();
+          try (Writer writer = new OutputStreamWriter(process.getOutputStream())) {
+            writer.write(String.format("set time zone '%s';\n", timezone));
+            writer.write("prepare test (timestamptz) as select $1;\n");
+            writer.write(String.format("execute test ('%s');\n", timestamp));
+            writer.write("\\q\n");
+          }
+
+          StringBuilder error = new StringBuilder();
+          try (Scanner scanner = new Scanner(process.getErrorStream())) {
+            while (scanner.hasNextLine()) {
+              error.append(scanner.nextLine()).append('\n');
+            }
+          }
+          StringBuilder output = new StringBuilder();
+          try (Scanner scanner = new Scanner(process.getInputStream())) {
+            while (scanner.hasNextLine()) {
+              output.append(scanner.nextLine()).append('\n');
+            }
+          }
+          int res = process.waitFor();
+          assertEquals(error.toString(), 0, res);
+          if (builder == realPgBuilder) {
+            realPgOutput = output.toString();
+          } else {
+            pgAdapterOutput = output.toString();
+          }
+        }
+        assertEquals(
+            String.format("Timestamp: %s, Timezone: %s", timestamp, timezone),
+            realPgOutput,
+            pgAdapterOutput);
+      }
+    }
+  }
+
+  private static final ImmutableSet<ZoneId> PROBLEMATIC_ZONE_IDS =
+      ImmutableSet.of(
+          // Mexico abolished DST in 2022, but not all databases contain this information.
+          ZoneId.of("America/Chihuahua"));
+
+  private LocalDate generateRandomLocalDate() {
+    return LocalDate.ofEpochDay(random.nextInt(365 * 100));
+  }
+
+  private LocalDateTime generateRandomLocalDateTime() {
+    return LocalDateTime.of(generateRandomLocalDate(), generateRandomLocalTime());
+  }
+
+  private LocalTime generateRandomLocalTime() {
+    return LocalTime.ofSecondOfDay(random.nextInt(24 * 60 * 60));
+  }
+
+  private OffsetDateTime generateRandomOffsetDateTime() {
+    return OffsetDateTime.of(
+        generateRandomLocalDateTime(),
+        ZoneOffset.ofHoursMinutes(random.nextInt(12), random.nextInt(60)));
+  }
+
   private void compareTableContents() throws SQLException {
     try (Connection pgConnection =
             DriverManager.getConnection(
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/nodejs/NodeJSTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/nodejs/NodeJSTest.java
index 6d14cf2c..8bf1c97c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/nodejs/NodeJSTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/nodejs/NodeJSTest.java
@@ -51,7 +51,7 @@ public interface NodeJSTest {
     Process process = builder.start();
     InputStream inputStream = process.getInputStream();
     InputStream errorStream = process.getErrorStream();
-    boolean finished = process.waitFor(30L, TimeUnit.SECONDS);
+    boolean finished = process.waitFor(120L, TimeUnit.SECONDS);
 
     String output = readAll(inputStream);
     String errors = readAll(errorStream);
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParserTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParserTest.java
index 0f3492a9..229afcbd 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParserTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/parsers/TimestampParserTest.java
@@ -110,4 +110,29 @@ public class TimestampParserTest {
             new TimestampParser(
                 "foo".getBytes(StandardCharsets.UTF_8), FormatCode.TEXT, sessionState));
   }
+
+  @Test
+  public void testTextToTimestamp() {
+    assertEquals(
+        Timestamp.parseTimestamp("2022-10-09T19:09:18Z"),
+        TimestampParser.toTimestamp("2022-10-09 10:09:18", ZoneId.of("-09:00")));
+    assertEquals(
+        Timestamp.parseTimestamp("2022-12-28T09:00Z"),
+        TimestampParser.toTimestamp("2022-12-28 10:00", ZoneId.of("CET")));
+    assertEquals(
+        Timestamp.parseTimestamp("2022-08-28T08:00Z"),
+        TimestampParser.toTimestamp("2022-08-28 10:00", ZoneId.of("CET")));
+    assertEquals(
+        Timestamp.parseTimestamp("2022-08-28T08:00Z"),
+        TimestampParser.toTimestamp("2022-08-28 10:00", ZoneId.of("Europe/Amsterdam")));
+    assertEquals(
+        Timestamp.parseTimestamp("2022-12-27T14:00:00Z"),
+        TimestampParser.toTimestamp("2022-12-28", ZoneId.of("+10:00")));
+    assertEquals(
+        Timestamp.parseTimestamp("2022-12-28T08:00Z"),
+        TimestampParser.toTimestamp("2022-12-28 10:00+02:00", ZoneId.of("CET")));
+    assertEquals(
+        Timestamp.parseTimestamp("2022-12-28T07:30Z"),
+        TimestampParser.toTimestamp("2022-12-28 10:00+02:30", ZoneId.of("CET")));
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/utils/CopyInParserTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/utils/CopyInParserTest.java
index 4a18ff97..aa0d48ba 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/utils/CopyInParserTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/utils/CopyInParserTest.java
@@ -15,7 +15,9 @@
 package com.google.cloud.spanner.pgadapter.utils;
 
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
 
+import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement.Format;
 import java.io.IOException;
 import java.io.PipedInputStream;
@@ -32,6 +34,7 @@ public class CopyInParserTest {
   public void testCreateText() throws IOException {
     CopyInParser parser =
         CopyInParser.create(
+            mock(SessionState.class),
             Format.TEXT,
             CSVFormat.POSTGRESQL_TEXT,
             new PipedInputStream(new PipedOutputStream(), 256),
@@ -43,6 +46,7 @@ public class CopyInParserTest {
   public void testCreateCsv() throws IOException {
     CopyInParser parser =
         CopyInParser.create(
+            mock(SessionState.class),
             Format.CSV,
             CSVFormat.POSTGRESQL_CSV,
             new PipedInputStream(new PipedOutputStream(), 256),
@@ -54,7 +58,11 @@ public class CopyInParserTest {
   public void testCreateBinary() throws IOException {
     CopyInParser parser =
         CopyInParser.create(
-            Format.BINARY, null, new PipedInputStream(new PipedOutputStream(), 256), false);
+            mock(SessionState.class),
+            Format.BINARY,
+            null,
+            new PipedInputStream(new PipedOutputStream(), 256),
+            false);
     assertTrue(parser instanceof BinaryCopyParser);
   }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParserTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParserTest.java
index e836c5e2..15614b9f 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParserTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/utils/CsvCopyParserTest.java
@@ -17,6 +17,7 @@ package com.google.cloud.spanner.pgadapter.utils;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
+import static org.mockito.Mockito.mock;
 
 import com.google.cloud.ByteArray;
 import com.google.cloud.Date;
@@ -25,6 +26,7 @@ import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.Type;
 import com.google.cloud.spanner.Type.StructField;
 import com.google.cloud.spanner.Value;
+import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.cloud.spanner.pgadapter.utils.CsvCopyParser.CsvCopyRecord;
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -44,7 +46,10 @@ public class CsvCopyParserTest {
 
     CsvCopyParser parser =
         new CsvCopyParser(
-            CSVFormat.POSTGRESQL_TEXT, new PipedInputStream(new PipedOutputStream(), 256), false);
+            mock(SessionState.class),
+            CSVFormat.POSTGRESQL_TEXT,
+            new PipedInputStream(new PipedOutputStream(), 256),
+            false);
     assertNotNull(parser.iterator());
     parser.close();
   }
@@ -70,164 +75,222 @@ public class CsvCopyParserTest {
             })
         .start();
     CsvCopyParser parser =
-        new CsvCopyParser(CSVFormat.POSTGRESQL_TEXT, new PipedInputStream(outputStream, 256), true);
+        new CsvCopyParser(
+            mock(SessionState.class),
+            CSVFormat.POSTGRESQL_TEXT,
+            new PipedInputStream(outputStream, 256),
+            true);
     assertNotNull(parser.iterator());
     parser.close();
   }
 
   @Test
   public void testGetSpannerValueBool() {
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "t"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "tr"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "tru"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "true"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "1"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "on"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "y"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "ye"));
-    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(Type.bool(), "yes"));
-
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "f"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "fa"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "fal"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "fals"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "false"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "0"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "off"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "of"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "n"));
-    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(Type.bool(), "no"));
-
-    assertEquals(Value.bool(null), CsvCopyRecord.getSpannerValue(Type.bool(), null));
+    SessionState sessionState = mock(SessionState.class);
+
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "t"));
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "tr"));
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "tru"));
+    assertEquals(
+        Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "true"));
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "1"));
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "on"));
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "y"));
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "ye"));
+    assertEquals(Value.bool(true), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "yes"));
+
+    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "f"));
+    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "fa"));
+    assertEquals(
+        Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "fal"));
+    assertEquals(
+        Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "fals"));
+    assertEquals(
+        Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "false"));
+    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "0"));
+    assertEquals(
+        Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "off"));
+    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "of"));
+    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "n"));
+    assertEquals(Value.bool(false), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "no"));
+
+    assertEquals(Value.bool(null), CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), null));
   }
 
   @Test
   public void testGetSpannerValueBytes() {
+    SessionState sessionState = mock(SessionState.class);
     assertEquals(
         Value.bytes(ByteArray.copyFrom("test")),
-        CsvCopyRecord.getSpannerValue(Type.bytes(), "\\x74657374"));
+        CsvCopyRecord.getSpannerValue(sessionState, Type.bytes(), "\\x74657374"));
     assertEquals(
-        Value.bytes(ByteArray.copyFrom("")), CsvCopyRecord.getSpannerValue(Type.bytes(), "\\x"));
-    assertEquals(Value.bytes(null), CsvCopyRecord.getSpannerValue(Type.bytes(), null));
+        Value.bytes(ByteArray.copyFrom("")),
+        CsvCopyRecord.getSpannerValue(sessionState, Type.bytes(), "\\x"));
+    assertEquals(
+        Value.bytes(null), CsvCopyRecord.getSpannerValue(sessionState, Type.bytes(), null));
   }
 
   @Test
   public void testGetSpannerValueInt64() {
-    assertEquals(Value.int64(-1L), CsvCopyRecord.getSpannerValue(Type.int64(), "-1"));
-    assertEquals(Value.int64(1L), CsvCopyRecord.getSpannerValue(Type.int64(), "1"));
-    assertEquals(Value.int64(0L), CsvCopyRecord.getSpannerValue(Type.int64(), "0"));
+    SessionState sessionState = mock(SessionState.class);
+    assertEquals(Value.int64(-1L), CsvCopyRecord.getSpannerValue(sessionState, Type.int64(), "-1"));
+    assertEquals(Value.int64(1L), CsvCopyRecord.getSpannerValue(sessionState, Type.int64(), "1"));
+    assertEquals(Value.int64(0L), CsvCopyRecord.getSpannerValue(sessionState, Type.int64(), "0"));
     assertEquals(
         Value.int64(Long.MAX_VALUE),
-        CsvCopyRecord.getSpannerValue(Type.int64(), "9223372036854775807"));
+        CsvCopyRecord.getSpannerValue(sessionState, Type.int64(), "9223372036854775807"));
     assertEquals(
         Value.int64(Long.MIN_VALUE),
-        CsvCopyRecord.getSpannerValue(Type.int64(), "-9223372036854775808"));
-    assertEquals(Value.int64(null), CsvCopyRecord.getSpannerValue(Type.int64(), null));
+        CsvCopyRecord.getSpannerValue(sessionState, Type.int64(), "-9223372036854775808"));
+    assertEquals(
+        Value.int64(null), CsvCopyRecord.getSpannerValue(sessionState, Type.int64(), null));
   }
 
   @Test
   public void testGetSpannerValueFloat64() {
-    assertEquals(Value.float64(-1.0D), CsvCopyRecord.getSpannerValue(Type.float64(), "-1.0"));
-    assertEquals(Value.float64(0.0D), CsvCopyRecord.getSpannerValue(Type.float64(), "0.0"));
-    assertEquals(Value.float64(1.0D), CsvCopyRecord.getSpannerValue(Type.float64(), "1.0"));
-    assertEquals(Value.float64(null), CsvCopyRecord.getSpannerValue(Type.float64(), null));
+    SessionState sessionState = mock(SessionState.class);
+    assertEquals(
+        Value.float64(-1.0D), CsvCopyRecord.getSpannerValue(sessionState, Type.float64(), "-1.0"));
+    assertEquals(
+        Value.float64(0.0D), CsvCopyRecord.getSpannerValue(sessionState, Type.float64(), "0.0"));
+    assertEquals(
+        Value.float64(1.0D), CsvCopyRecord.getSpannerValue(sessionState, Type.float64(), "1.0"));
+    assertEquals(
+        Value.float64(null), CsvCopyRecord.getSpannerValue(sessionState, Type.float64(), null));
   }
 
   @Test
   public void testGetSpannerValueNumeric() {
-    assertEquals(Value.pgNumeric("-1.0"), CsvCopyRecord.getSpannerValue(Type.pgNumeric(), "-1.0"));
-    assertEquals(Value.pgNumeric("0.0"), CsvCopyRecord.getSpannerValue(Type.pgNumeric(), "0.0"));
-    assertEquals(Value.pgNumeric("1.0"), CsvCopyRecord.getSpannerValue(Type.pgNumeric(), "1.0"));
-    assertEquals(Value.pgNumeric(null), CsvCopyRecord.getSpannerValue(Type.pgNumeric(), null));
+    SessionState sessionState = mock(SessionState.class);
+    assertEquals(
+        Value.pgNumeric("-1.0"),
+        CsvCopyRecord.getSpannerValue(sessionState, Type.pgNumeric(), "-1.0"));
+    assertEquals(
+        Value.pgNumeric("0.0"),
+        CsvCopyRecord.getSpannerValue(sessionState, Type.pgNumeric(), "0.0"));
+    assertEquals(
+        Value.pgNumeric("1.0"),
+        CsvCopyRecord.getSpannerValue(sessionState, Type.pgNumeric(), "1.0"));
+    assertEquals(
+        Value.pgNumeric(null), CsvCopyRecord.getSpannerValue(sessionState, Type.pgNumeric(), null));
   }
 
   @Test
   public void testGetSpannerValueString() {
-    assertEquals(Value.string("test"), CsvCopyRecord.getSpannerValue(Type.string(), "test"));
-    assertEquals(Value.string(""), CsvCopyRecord.getSpannerValue(Type.string(), ""));
-    assertEquals(Value.string(null), CsvCopyRecord.getSpannerValue(Type.string(), null));
+    SessionState sessionState = mock(SessionState.class);
+    assertEquals(
+        Value.string("test"), CsvCopyRecord.getSpannerValue(sessionState, Type.string(), "test"));
+    assertEquals(Value.string(""), CsvCopyRecord.getSpannerValue(sessionState, Type.string(), ""));
+    assertEquals(
+        Value.string(null), CsvCopyRecord.getSpannerValue(sessionState, Type.string(), null));
   }
 
   @Test
   public void testGetSpannerValueDate() {
+    SessionState sessionState = mock(SessionState.class);
     assertEquals(
         Value.date(Date.parseDate("2022-08-17")),
-        CsvCopyRecord.getSpannerValue(Type.date(), "2022-08-17"));
-    assertEquals(Value.date(null), CsvCopyRecord.getSpannerValue(Type.date(), null));
+        CsvCopyRecord.getSpannerValue(sessionState, Type.date(), "2022-08-17"));
+    assertEquals(Value.date(null), CsvCopyRecord.getSpannerValue(sessionState, Type.date(), null));
   }
 
   @Test
   public void testGetSpannerValueTimestamp() {
+    SessionState sessionState = mock(SessionState.class);
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2093-08-02T14:53:40.481913Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2093-08-02T14:53:40.481913+00"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2093-08-02T14:53:40.481913+00"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17T10:11:12.123456789Z"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17T10:11:12.123456789Z"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17 10:11:12.123456789Z"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17 10:11:12.123456789Z"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17 10:11:12.123456789+00"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17 10:11:12.123456789+00"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17 10:11:12.123456789+00:00"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17 10:11:12.123456789+00:00"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17T10:11:12.123456789+00"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17T10:11:12.123456789+00"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17T10:11:12.123456789+00:00"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17T10:11:12.123456789+00:00"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17 12:11:12.123456789+02"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17 12:11:12.123456789+02"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17T12:11:12.123456789+02"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17T12:11:12.123456789+02"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17 08:11:12.123456789-02"));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17 08:11:12.123456789-02"));
     assertEquals(
         Value.timestamp(Timestamp.parseTimestamp("2022-08-17T10:11:12.123456789Z")),
-        CsvCopyRecord.getSpannerValue(Type.timestamp(), "2022-08-17T08:11:12.123456789-02"));
-    assertEquals(Value.date(null), CsvCopyRecord.getSpannerValue(Type.date(), null));
+        CsvCopyRecord.getSpannerValue(
+            sessionState, Type.timestamp(), "2022-08-17T08:11:12.123456789-02"));
+    assertEquals(Value.date(null), CsvCopyRecord.getSpannerValue(sessionState, Type.date(), null));
   }
 
   @Test
   public void testGetSpannerValue_InvalidBytesValue() {
+    SessionState sessionState = mock(SessionState.class);
     assertThrows(
-        SpannerException.class, () -> CsvCopyRecord.getSpannerValue(Type.bytes(), "value"));
+        SpannerException.class,
+        () -> CsvCopyRecord.getSpannerValue(sessionState, Type.bytes(), "value"));
   }
 
   @Test
   public void testGetSpannerValue_InvalidNumberValue() {
+    SessionState sessionState = mock(SessionState.class);
     assertThrows(
-        SpannerException.class, () -> CsvCopyRecord.getSpannerValue(Type.int64(), "value"));
+        SpannerException.class,
+        () -> CsvCopyRecord.getSpannerValue(sessionState, Type.int64(), "value"));
   }
 
   @Test
   public void testGetSpannerValue_InvalidBoolValue() {
-    assertThrows(SpannerException.class, () -> CsvCopyRecord.getSpannerValue(Type.bool(), "value"));
+    SessionState sessionState = mock(SessionState.class);
+    assertThrows(
+        SpannerException.class,
+        () -> CsvCopyRecord.getSpannerValue(sessionState, Type.bool(), "value"));
   }
 
   @Test
   public void testGetSpannerValue_InvalidDateValue() {
-    assertThrows(SpannerException.class, () -> CsvCopyRecord.getSpannerValue(Type.date(), "value"));
+    SessionState sessionState = mock(SessionState.class);
+    assertThrows(
+        SpannerException.class,
+        () -> CsvCopyRecord.getSpannerValue(sessionState, Type.date(), "value"));
   }
 
   @Test
   public void testGetSpannerValue_InvalidTimestampValue() {
+    SessionState sessionState = mock(SessionState.class);
     assertThrows(
-        SpannerException.class, () -> CsvCopyRecord.getSpannerValue(Type.timestamp(), "value"));
+        SpannerException.class,
+        () -> CsvCopyRecord.getSpannerValue(sessionState, Type.timestamp(), "value"));
   }
 
   @Test
   public void testGetSpannerValue_UnsupportedType() {
+    SessionState sessionState = mock(SessionState.class);
     assertThrows(
         SpannerException.class,
         () ->
             CsvCopyRecord.getSpannerValue(
-                Type.struct(StructField.of("f1", Type.string())), "value"));
+                sessionState, Type.struct(StructField.of("f1", Type.string())), "value"));
   }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java
index b6881a67..3f8aca1d 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java
@@ -602,7 +602,8 @@ public class MutationWriterTest {
               new StringReader("col\n\\N\n"), CSVFormat.POSTGRESQL_TEXT.withFirstRecordAsHeader());
       CSVRecord record = parser.getRecords().get(0);
 
-      Mutation mutation = mutationWriter.buildMutation(new CsvCopyRecord(record, true));
+      Mutation mutation =
+          mutationWriter.buildMutation(new CsvCopyRecord(sessionState, record, true));
 
       assertEquals(String.format("Type: %s", type), 1, mutation.asMap().size());
     }