Repository: GoogleCloudPlatform/pgadapter
Pull request #38 (2022-02-16 12:30:53)

refactor: replace JDBC with Connection API
Replaces the Cloud Spanner JDBC driver with the internal Connection API.
This reduces the number of times that we need to parse the SQL string
and translate between different types of parameters. It also enables us
to more easily make changes in the client library (Connection API) that
we can use for PgAdapter, as the Connection API is an internal API
without a strict contract on the interface.

@aseering I can't add you directly as a reviewer, as you are not listed as a member of the repository.

Vizerai: Copy changes have all been merged at this point.  You can go ahead and resolve the conflicts.
olavloite: > Copy changes have all been merged at this point. You can go ahead and resolve the conflicts.

@Vizerai I've merged in the most recent changes and resolved all conflicts.

Patch:
diff --git a/pom.xml b/pom.xml
index 1027d59e..6b71d308 100644
--- a/pom.xml
+++ b/pom.xml
@@ -49,8 +49,8 @@
     </dependency>
     <dependency>
       <groupId>com.google.cloud</groupId>
-      <artifactId>google-cloud-spanner-jdbc</artifactId>
-      <version>2.6.0</version>
+      <artifactId>google-cloud-spanner</artifactId>
+      <version>6.20.0</version>
     </dependency>
     <dependency>
       <groupId>org.postgresql</groupId>
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java b/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
index 7c7d92d8..57592971 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
@@ -14,6 +14,9 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.ConnectionOptions;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePortalStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
@@ -29,12 +32,11 @@ import java.io.DataOutputStream;
 import java.io.IOException;
 import java.net.Socket;
 import java.security.SecureRandom;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.SQLException;
 import java.text.MessageFormat;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Properties;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
@@ -67,15 +69,20 @@ public class ConnectionHandler extends Thread {
   private static AtomicInteger incrementingConnectionId = new AtomicInteger(0);
   private ConnectionMetadata connectionMetadata;
   private WireMessage message;
-  private Connection jdbcConnection;
+  private Connection spannerConnection;
 
-  ConnectionHandler(ProxyServer server, Socket socket) throws SQLException {
+  ConnectionHandler(ProxyServer server, Socket socket) {
     super("ConnectionHandler-" + CONNECTION_HANDLER_ID_GENERATOR.incrementAndGet());
     this.server = server;
     this.socket = socket;
     this.secret = new SecureRandom().nextInt();
-    this.jdbcConnection =
-        DriverManager.getConnection(server.getOptions().getConnectionURL(), server.getProperties());
+    String uri = server.getOptions().getConnectionURL();
+    if (uri.startsWith("jdbc:")) {
+      uri = uri.substring("jdbc:".length());
+    }
+    uri = appendPropertiesToUrl(uri, server.getProperties());
+    ConnectionOptions connectionOptions = ConnectionOptions.newBuilder().setUri(uri).build();
+    this.spannerConnection = connectionOptions.getConnection();
     setDaemon(true);
     logger.log(
         Level.INFO,
@@ -83,6 +90,19 @@ public class ConnectionHandler extends Thread {
         new Object[] {getName(), socket.getInetAddress().getHostAddress()});
   }
 
+  private String appendPropertiesToUrl(String url, Properties info) {
+    if (info == null || info.isEmpty()) {
+      return url;
+    }
+    StringBuilder result = new StringBuilder(url);
+    for (Entry<Object, Object> entry : info.entrySet()) {
+      if (entry.getValue() != null && !"".equals(entry.getValue())) {
+        result.append(";").append(entry.getKey()).append("=").append(entry.getValue());
+      }
+    }
+    return result.toString();
+  }
+
   /**
    * Simple runner starts a loop which keeps taking inbound messages, processing them, sending them
    * to Spanner, getting a result, processing that result, and replying to the client (in that
@@ -130,13 +150,16 @@ public class ConnectionHandler extends Thread {
     } finally {
       logger.log(Level.INFO, "Closing connection handler with ID {0}", getName());
       try {
-        if (this.jdbcConnection != null) {
-          this.jdbcConnection.close();
+        if (this.spannerConnection != null) {
+          this.spannerConnection.close();
         }
         this.socket.close();
-      } catch (SQLException | IOException e) {
+      } catch (SpannerException | IOException e) {
         logger.log(
-            Level.WARNING, "Exception while closing connection handler with ID {0}", getName());
+            Level.WARNING,
+            e,
+            () ->
+                String.format("Exception while closing connection handler with ID %s", getName()));
       }
       this.server.deregister(this);
       logger.log(Level.INFO, "Connection handler with ID {0} closed", getName());
@@ -146,7 +169,7 @@ public class ConnectionHandler extends Thread {
   /** Called when a Terminate message is received. This closes this {@link ConnectionHandler}. */
   public void handleTerminate() throws Exception {
     closeAllPortals();
-    this.jdbcConnection.close();
+    this.spannerConnection.close();
     this.status = ConnectionStatus.TERMINATED;
   }
 
@@ -261,7 +284,7 @@ public class ConnectionHandler extends Thread {
       IntermediateStatement statement = activeStatementsMap.remove(connectionId);
       // We can mostly ignore the exception since cancel does not expect any result (positive or
       // otherwise)
-      statement.getStatement().cancel();
+      statement.getConnection().cancel();
     }
   }
 
@@ -291,8 +314,8 @@ public class ConnectionHandler extends Thread {
     return this.server;
   }
 
-  public Connection getJdbcConnection() {
-    return this.jdbcConnection;
+  public Connection getSpannerConnection() {
+    return this.spannerConnection;
   }
 
   public int getConnectionId() {
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
index b662bf78..cd01d9aa 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
@@ -14,6 +14,10 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.TextFormat;
@@ -23,7 +27,6 @@ import java.io.IOException;
 import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketException;
-import java.sql.SQLException;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -154,7 +157,7 @@ public class ProxyServer extends Thread {
           Level.INFO,
           "Socket exception on port {0}: {1}. This is normal when the server is stopped.",
           new Object[] {getLocalPort(), e});
-    } catch (SQLException e) {
+    } catch (SpannerException e) {
       logger.log(
           Level.SEVERE,
           "Something went wrong in establishing a Spanner connection: {0}",
@@ -169,12 +172,20 @@ public class ProxyServer extends Thread {
    * Creates and runs the {@link ConnectionHandler}, saving an instance of it locally.
    *
    * @param socket The socket the {@link ConnectionHandler} will read from.
-   * @throws SQLException if the {@link ConnectionHandler} is unable to Connect.
+   * @throws SpannerException if the {@link ConnectionHandler} is unable to connect to Cloud Spanner
+   *     or if the dialect of the database is not PostgreSQL.
    */
-  void createConnectionHandler(Socket socket) throws SQLException {
+  void createConnectionHandler(Socket socket) {
     ConnectionHandler handler = new ConnectionHandler(this, socket);
-    if (!handler.getJdbcConnection().isValid(0)) {
-      throw new SQLException("Invalid JDBC connection. Make sure credentials are valid.");
+    // Note: Calling getDialect() will cause a SpannerException if the connection itself is
+    // invalid, for example as a result of the credentials being wrong.
+    if (handler.getSpannerConnection().getDialect() != Dialect.POSTGRESQL) {
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.INVALID_ARGUMENT,
+          String.format(
+              "The database uses dialect %s. Currently PGAdapter only supports connections to PostgreSQL dialect databases. "
+                  + "These can be created using https://cloud.google.com/spanner/docs/quickstart-console#postgresql",
+              handler.getSpannerConnection().getDialect()));
     }
     register(handler);
     handler.start();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/commands/Command.java b/src/main/java/com/google/cloud/spanner/pgadapter/commands/Command.java
index decf6cda..a7e0b39a 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/commands/Command.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/commands/Command.java
@@ -14,8 +14,8 @@
 
 package com.google.cloud.spanner.pgadapter.commands;
 
+import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.pgadapter.metadata.DynamicCommandMetadata;
-import java.sql.Connection;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/commands/ListCommand.java b/src/main/java/com/google/cloud/spanner/pgadapter/commands/ListCommand.java
index 5640ae87..0fe2874c 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/commands/ListCommand.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/commands/ListCommand.java
@@ -14,8 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter.commands;
 
-import java.sql.Connection;
-import java.sql.SQLException;
+import com.google.cloud.spanner.connection.Connection;
 import java.util.regex.Pattern;
 
 /**
@@ -50,10 +49,7 @@ public class ListCommand extends Command {
 
   @Override
   public String translate() {
-    try {
-      return String.format(OUTPUT_QUERY, this.connection.getCatalog());
-    } catch (SQLException e) {
-      return String.format(OUTPUT_QUERY, "UNKNOWN");
-    }
+    // TODO: Pass in the database name, or add it to the Connection API surface and get it here.
+    return String.format(OUTPUT_QUERY, "");
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/metadata/DescribePortalMetadata.java b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/DescribePortalMetadata.java
index df5732f5..d9b292e9 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/metadata/DescribePortalMetadata.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/DescribePortalMetadata.java
@@ -14,12 +14,12 @@
 
 package com.google.cloud.spanner.pgadapter.metadata;
 
-import java.sql.ResultSetMetaData;
+import com.google.cloud.spanner.ResultSet;
 
 /** Simple POJO to hold describe metadata, specific to portal describes. */
-public class DescribePortalMetadata extends DescribeMetadata<ResultSetMetaData> {
+public class DescribePortalMetadata extends DescribeMetadata<ResultSet> {
 
-  public DescribePortalMetadata(ResultSetMetaData metadata) {
+  public DescribePortalMetadata(ResultSet metadata) {
     this.metadata = metadata;
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/ArrayParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/ArrayParser.java
index bc337919..5d9851c0 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/ArrayParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/ArrayParser.java
@@ -14,12 +14,16 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.Type;
+import com.google.cloud.spanner.Type.Code;
+import com.google.cloud.spanner.Value;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
-import java.sql.ResultSet;
-import java.sql.SQLException;
 import java.sql.Types;
-import java.util.Arrays;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.stream.Collectors;
@@ -28,29 +32,66 @@ import java.util.stream.Collectors;
  * Translate wire protocol to array. Since arrays house any other specified types (including
  * potentially arrays), we use all parser types to parse each item within.
  */
-public class ArrayParser extends Parser<List> {
+public class ArrayParser extends Parser<List<?>> {
+  private static final String STRING_TOGGLE = "\"";
+  private static final String ARRAY_DELIMITER = ",";
+  private static final String PG_ARRAY_OPEN = "{", PG_ARRAY_CLOSE = "}";
+  private static final String SPANNER_ARRAY_OPEN = "[", SPANNER_ARRAY_CLOSE = "]";
 
-  private static String STRING_TOGGLE = "\"";
-  private static String ARRAY_DELIMITER = ",";
-  private static String PG_ARRAY_OPEN = "{", PG_ARRAY_CLOSE = "}";
-  private static String SPANNER_ARRAY_OPEN = "[", SPANNER_ARRAY_CLOSE = "]";
+  private final Type arrayElementType;
+  private final boolean isStringEquivalent;
 
-  int arrayType;
-  boolean isStringEquivalent;
-
-  public ArrayParser(ResultSet item, int position) throws SQLException {
+  public ArrayParser(ResultSet item, int position) {
     if (item != null) {
-      this.item = Arrays.asList((Object[]) item.getArray(position).getArray());
-      this.arrayType = item.getArray(position).getBaseType();
-      if (this.arrayType == Types.ARRAY) {
+      this.arrayElementType = item.getColumnType(position).getArrayElementType();
+      if (this.arrayElementType.getCode() == Code.ARRAY) {
         throw new IllegalArgumentException(
             "Spanner does not support embedded Arrays."
                 + " If you are seeing this, something went wrong!");
       }
-      this.isStringEquivalent = stringEquivalence(this.arrayType);
+      this.item = toList(item.getValue(position), this.arrayElementType.getCode());
+      this.isStringEquivalent = stringEquivalence(this.arrayElementType.getCode());
+    } else {
+      arrayElementType = null;
+      isStringEquivalent = false;
+    }
+  }
+
+  private List<?> toList(Value value, Code arrayElementType) {
+    switch (arrayElementType) {
+      case BOOL:
+        return value.getBoolArray();
+      case DATE:
+        return value.getDateArray();
+      case JSON:
+        return value.getJsonArray();
+      case BYTES:
+        return value.getBytesArray();
+      case INT64:
+        return value.getInt64Array();
+      case STRING:
+        return value.getStringArray();
+      case PG_NUMERIC:
+        return value.getNumericArray();
+      case TIMESTAMP:
+        return value.getTimestampArray();
+      case FLOAT64:
+        return value.getFloat64Array();
+      case NUMERIC: // Only PG_NUMERIC is supported
+      case ARRAY:
+      case STRUCT:
+      default:
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT,
+            String.format("Array of %s is not supported", arrayElementType));
     }
   }
 
+  @Override
+  public List<?> getItem() {
+    return this.item;
+  }
+
   @Override
   public int getSqlType() {
     return Types.ARRAY;
@@ -59,14 +100,14 @@ public class ArrayParser extends Parser<List> {
   /**
    * Whether a type is represented as string.
    *
-   * @param arrayType The type to check.
+   * @param arrayElementType The type to check.
    * @return True if the type uses strings, false otherwise.
    */
-  private boolean stringEquivalence(int arrayType) {
-    return arrayType == Types.BINARY
-        || arrayType == Types.DATE
-        || arrayType == Types.NVARCHAR
-        || arrayType == Types.TIMESTAMP;
+  private boolean stringEquivalence(Code arrayElementType) {
+    return arrayElementType == Code.BYTES
+        || arrayElementType == Code.DATE
+        || arrayElementType == Code.STRING
+        || arrayElementType == Code.TIMESTAMP;
   }
 
   /**
@@ -84,9 +125,13 @@ public class ArrayParser extends Parser<List> {
 
   @Override
   protected String stringParse() {
+    if (this.item == null) {
+      return null;
+    }
     List<String> results = new LinkedList<>();
     for (Object currentItem : this.item) {
-      results.add(stringify(Parser.create(currentItem, this.arrayType).stringParse()));
+      results.add(
+          stringify(Parser.create(currentItem, this.arrayElementType.getCode()).stringParse()));
     }
     return results.stream()
         .collect(Collectors.joining(ARRAY_DELIMITER, PG_ARRAY_OPEN, PG_ARRAY_CLOSE));
@@ -94,9 +139,13 @@ public class ArrayParser extends Parser<List> {
 
   @Override
   protected String spannerParse() {
+    if (this.item == null) {
+      return null;
+    }
     List<String> results = new LinkedList<>();
     for (Object currentItem : this.item) {
-      results.add(stringify(Parser.create(currentItem, this.arrayType).spannerParse()));
+      results.add(
+          stringify(Parser.create(currentItem, this.arrayElementType.getCode()).spannerParse()));
     }
     return results.stream()
         .collect(Collectors.joining(ARRAY_DELIMITER, SPANNER_ARRAY_OPEN, SPANNER_ARRAY_CLOSE));
@@ -104,18 +153,21 @@ public class ArrayParser extends Parser<List> {
 
   @Override
   protected byte[] binaryParse() {
+    if (this.item == null) {
+      return null;
+    }
     ByteArrayOutputStream arrayStream = new ByteArrayOutputStream();
     try {
       arrayStream.write(IntegerParser.binaryParse(1)); // dimension
       arrayStream.write(IntegerParser.binaryParse(1)); // Set null flag
-      arrayStream.write(IntegerParser.binaryParse(this.arrayType)); // Set type
+      arrayStream.write(IntegerParser.binaryParse(Parser.toOid(this.arrayElementType))); // Set type
       arrayStream.write(IntegerParser.binaryParse(this.item.size())); // Set array length
       arrayStream.write(IntegerParser.binaryParse(0)); // Lower bound (?)
       for (Object currentItem : this.item) {
         if (currentItem == null) {
           arrayStream.write(IntegerParser.binaryParse(-1));
         } else {
-          byte[] data = Parser.create(currentItem, this.arrayType).binaryParse();
+          byte[] data = Parser.create(currentItem, this.arrayElementType.getCode()).binaryParse();
           arrayStream.write(IntegerParser.binaryParse(data.length));
           arrayStream.write(data);
         }
@@ -125,4 +177,9 @@ public class ArrayParser extends Parser<List> {
       throw new RuntimeException(e);
     }
   }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    throw SpannerExceptionFactory.newSpannerException(
+        ErrorCode.UNIMPLEMENTED, "Array parameters not yet supported");
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BinaryParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BinaryParser.java
index 49d3ad0c..878639f5 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BinaryParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BinaryParser.java
@@ -14,8 +14,10 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.ByteArray;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import java.nio.charset.StandardCharsets;
-import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.sql.Types;
 import org.postgresql.util.PGbytea;
@@ -24,14 +26,14 @@ import org.postgresql.util.PGbytea;
  * Parse specified type to binary (generally this is the simplest parse class, as items are
  * generally represented in binary for wire format).
  */
-public class BinaryParser extends Parser<byte[]> {
+public class BinaryParser extends Parser<ByteArray> {
 
-  public BinaryParser(ResultSet item, int position) throws SQLException {
+  public BinaryParser(ResultSet item, int position) {
     this.item = item.getBytes(position);
   }
 
   public BinaryParser(Object item) {
-    this.item = (byte[]) item;
+    this.item = (ByteArray) item;
   }
 
   public BinaryParser(byte[] item, FormatCode formatCode) {
@@ -39,14 +41,14 @@ public class BinaryParser extends Parser<byte[]> {
       switch (formatCode) {
         case TEXT:
           try {
-            this.item = PGbytea.toBytes(item);
+            this.item = ByteArray.copyFrom(PGbytea.toBytes(item));
             break;
           } catch (SQLException e) {
             throw new IllegalArgumentException(
                 "Invalid binary value: " + new String(item, StandardCharsets.UTF_8), e);
           }
         case BINARY:
-          this.item = item;
+          this.item = ByteArray.copyFrom(item);
           break;
         default:
           throw new IllegalArgumentException("Unsupported format: " + formatCode);
@@ -61,16 +63,20 @@ public class BinaryParser extends Parser<byte[]> {
 
   @Override
   protected String stringParse() {
-    return PGbytea.toPGString(this.item);
+    return this.item == null ? null : PGbytea.toPGString(this.item.toByteArray());
   }
 
   @Override
   protected byte[] spannerBinaryParse() {
-    return this.item;
+    return this.item == null ? null : this.item.toByteArray();
   }
 
   @Override
   protected byte[] binaryParse() {
-    return this.item;
+    return this.item == null ? null : this.item.toByteArray();
+  }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BooleanParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BooleanParser.java
index fbeae34d..00ab14dd 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BooleanParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/BooleanParser.java
@@ -14,9 +14,9 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import com.google.common.collect.ImmutableSet;
-import java.sql.ResultSet;
-import java.sql.SQLException;
 import java.sql.Types;
 import java.util.Locale;
 import java.util.Set;
@@ -35,7 +35,7 @@ public class BooleanParser extends Parser<Boolean> {
   private static final Set<String> FALSE_VALUES =
       ImmutableSet.of("f", "fa", "fal", "fals", "false", "n", "no", "of", "off");
 
-  public BooleanParser(ResultSet item, int position) throws SQLException {
+  public BooleanParser(ResultSet item, int position) {
     this.item = item.getBoolean(position);
   }
 
@@ -72,18 +72,28 @@ public class BooleanParser extends Parser<Boolean> {
 
   @Override
   protected String stringParse() {
+    if (this.item == null) {
+      return null;
+    }
     return this.item ? TRUE_VALUE : FALSE_VALUE;
   }
 
   @Override
   protected String spannerParse() {
-    return Boolean.toString(this.item);
+    return this.item == null ? null : Boolean.toString(this.item);
   }
 
   @Override
   protected byte[] binaryParse() {
+    if (this.item == null) {
+      return null;
+    }
     byte[] result = new byte[1];
     ByteConverter.bool(result, 0, this.item);
     return result;
   }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DateParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DateParser.java
index bbde61ab..bd3433e4 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DateParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DateParser.java
@@ -14,9 +14,10 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.Date;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import com.google.common.base.Preconditions;
-import java.sql.ResultSet;
-import java.sql.SQLException;
 import java.sql.Types;
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
@@ -24,14 +25,14 @@ import java.time.LocalDate;
 import org.postgresql.util.ByteConverter;
 
 /** Translate wire protocol dates to desired formats. */
-public class DateParser extends Parser<java.sql.Date> {
+public class DateParser extends Parser<Date> {
 
-  public DateParser(ResultSet item, int position) throws SQLException {
+  public DateParser(ResultSet item, int position) {
     this.item = item.getDate(position);
   }
 
   public DateParser(Object item) {
-    this.item = (java.sql.Date) item;
+    this.item = (Date) item;
   }
 
   public DateParser(byte[] item, FormatCode formatCode) {
@@ -39,12 +40,15 @@ public class DateParser extends Parser<java.sql.Date> {
       switch (formatCode) {
         case TEXT:
           String stringValue = new String(item, UTF8);
-          this.item = java.sql.Date.valueOf(stringValue);
+          this.item = Date.parseDate(stringValue);
           break;
         case BINARY:
           long days = ByteConverter.int4(item, 0) + PG_EPOCH_DAYS;
           this.validateRange(days);
-          this.item = java.sql.Date.valueOf(LocalDate.ofEpochDay(days));
+          LocalDate localDate = LocalDate.ofEpochDay(days);
+          this.item =
+              Date.fromYearMonthDay(
+                  localDate.getYear(), localDate.getMonthValue(), localDate.getDayOfMonth());
           break;
         default:
           throw new IllegalArgumentException("Unsupported format: " + formatCode);
@@ -86,12 +90,17 @@ public class DateParser extends Parser<java.sql.Date> {
 
   @Override
   protected String stringParse() {
-    return item.toString();
+    return this.item == null ? null : item.toString();
   }
 
   @Override
   protected byte[] binaryParse() {
-    long days = this.item.toLocalDate().toEpochDay() - PG_EPOCH_DAYS;
+    if (this.item == null) {
+      return null;
+    }
+    LocalDate localDate =
+        LocalDate.of(this.item.getYear(), this.item.getMonth(), this.item.getDayOfMonth());
+    long days = localDate.toEpochDay() - PG_EPOCH_DAYS;
     int daysAsInt = validateRange(days);
     return IntegerParser.binaryParse(daysAsInt);
   }
@@ -108,4 +117,8 @@ public class DateParser extends Parser<java.sql.Date> {
     }
     return (int) days;
   }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DoubleParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DoubleParser.java
index ee24125c..9621a1cc 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DoubleParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/DoubleParser.java
@@ -14,15 +14,15 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
-import java.sql.ResultSet;
-import java.sql.SQLException;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import java.sql.Types;
 import org.postgresql.util.ByteConverter;
 
 /** Translate from wire protocol to double. */
 public class DoubleParser extends Parser<Double> {
 
-  public DoubleParser(ResultSet item, int position) throws SQLException {
+  public DoubleParser(ResultSet item, int position) {
     this.item = item.getDouble(position);
   }
 
@@ -52,13 +52,20 @@ public class DoubleParser extends Parser<Double> {
 
   @Override
   protected String stringParse() {
-    return Double.toString(this.item);
+    return this.item == null ? null : Double.toString(this.item);
   }
 
   @Override
   protected byte[] binaryParse() {
+    if (this.item == null) {
+      return null;
+    }
     byte[] result = new byte[8];
     ByteConverter.float8(result, 0, this.item);
     return result;
   }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/IntegerParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/IntegerParser.java
index d521dc26..f7c05f57 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/IntegerParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/IntegerParser.java
@@ -14,16 +14,16 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
-import java.sql.ResultSet;
-import java.sql.SQLException;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import java.sql.Types;
 import org.postgresql.util.ByteConverter;
 
 /** Translate from wire protocol to int. */
 public class IntegerParser extends Parser<Integer> {
 
-  public IntegerParser(ResultSet item, int position) throws SQLException {
-    this.item = item.getInt(position);
+  public IntegerParser(ResultSet item, int position) {
+    this.item = Math.toIntExact(item.getLong(position));
   }
 
   public IntegerParser(Object item) {
@@ -52,12 +52,12 @@ public class IntegerParser extends Parser<Integer> {
 
   @Override
   protected String stringParse() {
-    return Integer.toString(this.item);
+    return this.item == null ? null : Integer.toString(this.item);
   }
 
   @Override
   protected byte[] binaryParse() {
-    return binaryParse(this.item);
+    return this.item == null ? null : binaryParse(this.item);
   }
 
   public static byte[] binaryParse(int value) {
@@ -65,4 +65,9 @@ public class IntegerParser extends Parser<Integer> {
     ByteConverter.int4(result, 0, value);
     return result;
   }
+
+  @Override
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item == null ? null : this.item.longValue());
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/LongParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/LongParser.java
index 57ff2c84..f2337d36 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/LongParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/LongParser.java
@@ -14,15 +14,15 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
-import java.sql.ResultSet;
-import java.sql.SQLException;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import java.sql.Types;
 import org.postgresql.util.ByteConverter;
 
 /** Translate from wire protocol to long. */
 public class LongParser extends Parser<Long> {
 
-  public LongParser(ResultSet item, int position) throws SQLException {
+  public LongParser(ResultSet item, int position) {
     this.item = item.getLong(position);
   }
 
@@ -52,13 +52,20 @@ public class LongParser extends Parser<Long> {
 
   @Override
   protected String stringParse() {
-    return Long.toString(this.item);
+    return this.item == null ? null : Long.toString(this.item);
   }
 
   @Override
   protected byte[] binaryParse() {
+    if (this.item == null) {
+      return null;
+    }
     byte[] result = new byte[8];
     ByteConverter.int8(result, 0, this.item);
     return result;
   }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/NumericParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/NumericParser.java
index ada0cfed..a05114f2 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/NumericParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/NumericParser.java
@@ -14,18 +14,24 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.Value;
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
-import java.sql.ResultSet;
-import java.sql.SQLException;
 import java.sql.Types;
 import org.postgresql.util.ByteConverter;
 
 /** Translate from wire protocol to {@link Number}. */
 public class NumericParser extends Parser<Number> {
-  public NumericParser(ResultSet item, int position) throws SQLException {
+  public NumericParser(ResultSet item, int position) {
     // This should be either a BigDecimal value or a Double.NaN.
-    this.item = (Number) item.getObject(position);
+    Value value = item.getValue(position);
+    if (Value.NAN.equalsIgnoreCase(value.getString())) {
+      this.item = Double.NaN;
+    } else {
+      this.item = value.getNumeric();
+    }
   }
 
   public NumericParser(Object item) {
@@ -59,14 +65,24 @@ public class NumericParser extends Parser<Number> {
 
   @Override
   protected String stringParse() {
+    if (this.item == null) {
+      return null;
+    }
     return Double.isNaN(this.item.doubleValue()) ? "NaN" : ((BigDecimal) this.item).toPlainString();
   }
 
   @Override
   protected byte[] binaryParse() {
+    if (this.item == null) {
+      return null;
+    }
     if (Double.isNaN(this.item.doubleValue())) {
       return "NaN".getBytes(StandardCharsets.UTF_8);
     }
     return ByteConverter.numeric((BigDecimal) this.item);
   }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(Value.pgNumeric(stringParse()));
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/Parser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/Parser.java
index f9af12c0..ac63e293 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/Parser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/Parser.java
@@ -14,12 +14,15 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.Type;
+import com.google.cloud.spanner.Type.Code;
 import com.google.cloud.spanner.pgadapter.ProxyServer.DataFormat;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Types;
 import org.postgresql.core.Oid;
 
 /**
@@ -78,7 +81,7 @@ public abstract class Parser<T> {
    * @param formatCode The format of the data to be parsed
    * @return The parser object for the designated data type.
    */
-  public static Parser create(byte[] item, int oidType, FormatCode formatCode) {
+  public static Parser<?> create(byte[] item, int oidType, FormatCode formatCode) {
     switch (oidType) {
       case Oid.BOOL:
       case Oid.BIT:
@@ -119,37 +122,34 @@ public abstract class Parser<T> {
    * Factory method for parsing given a result set, for the specified column.
    *
    * @param result The result set object containing all data retrieved.
-   * @param oidType The type of the data to be parsed.
+   * @param type The type of the data to be parsed.
    * @param columnarPosition Column from the result to be parsed.
    * @return The parser object for the designated data type.
-   * @throws SQLException if the result data type does not correspond with the desired type.
    */
-  public static Parser create(ResultSet result, int oidType, int columnarPosition)
-      throws SQLException {
-    switch (oidType) {
-      case Types.BOOLEAN:
+  public static Parser<?> create(ResultSet result, Type type, int columnarPosition) {
+    switch (type.getCode()) {
+      case BOOL:
         return new BooleanParser(result, columnarPosition);
-      case Types.BINARY:
+      case BYTES:
         return new BinaryParser(result, columnarPosition);
-      case Types.DATE:
+      case DATE:
         return new DateParser(result, columnarPosition);
-      case Types.DOUBLE:
+      case FLOAT64:
         return new DoubleParser(result, columnarPosition);
-      case Types.BIGINT:
+      case INT64:
         return new LongParser(result, columnarPosition);
-      case Types.INTEGER:
-        return new IntegerParser(result, columnarPosition);
-      case Types.NUMERIC:
+      case PG_NUMERIC:
         return new NumericParser(result, columnarPosition);
-      case Types.NVARCHAR:
+      case STRING:
         return new StringParser(result, columnarPosition);
-      case Types.TIMESTAMP:
+      case TIMESTAMP:
         return new TimestampParser(result, columnarPosition);
-      case Types.ARRAY:
+      case ARRAY:
         return new ArrayParser(result, columnarPosition);
-      case Types.STRUCT:
+      case NUMERIC:
+      case STRUCT:
       default:
-        throw new IllegalArgumentException("Illegal or unknown element oidType: " + oidType);
+        throw new IllegalArgumentException("Illegal or unknown element type: " + type);
     }
   }
 
@@ -157,33 +157,93 @@ public abstract class Parser<T> {
    * Factory method for parsing generic data given a specified type.
    *
    * @param result The generic object to parse.
-   * @param oidType The type of the object to be parsed.
+   * @param typeCode The type of the object to be parsed.
    * @return The parser object for the designated data type.
    */
-  protected static Parser create(Object result, int oidType) {
-    switch (oidType) {
-      case Types.BOOLEAN:
+  protected static Parser<?> create(Object result, Code typeCode) {
+    switch (typeCode) {
+      case BOOL:
         return new BooleanParser(result);
-      case Types.BINARY:
+      case BYTES:
         return new BinaryParser(result);
-      case Types.DATE:
+      case DATE:
         return new DateParser(result);
-      case Types.DOUBLE:
+      case FLOAT64:
         return new DoubleParser(result);
-      case Types.BIGINT:
+      case INT64:
         return new LongParser(result);
-      case Types.INTEGER:
-        return new IntegerParser(result);
-      case Types.NUMERIC:
+      case PG_NUMERIC:
         return new NumericParser(result);
-      case Types.NVARCHAR:
+      case STRING:
         return new StringParser(result);
-      case Types.TIMESTAMP:
+      case TIMESTAMP:
         return new TimestampParser(result);
-      case Types.ARRAY:
-      case Types.STRUCT:
+      case NUMERIC:
+      case ARRAY:
+      case STRUCT:
       default:
-        throw new IllegalArgumentException("Illegal or unknown element type: " + oidType);
+        throw new IllegalArgumentException("Illegal or unknown element type: " + typeCode);
+    }
+  }
+
+  /**
+   * Translates the given Cloud Spanner {@link Type} to a PostgreSQL OID constant.
+   *
+   * @param type the type to translate
+   * @return The OID constant value for the type
+   */
+  public static int toOid(Type type) {
+    switch (type.getCode()) {
+      case BOOL:
+        return Oid.BOOL;
+      case INT64:
+        return Oid.INT8;
+      case PG_NUMERIC:
+        return Oid.NUMERIC;
+      case FLOAT64:
+        return Oid.FLOAT8;
+      case STRING:
+        return Oid.VARCHAR;
+      case JSON:
+        return Oid.JSONB;
+      case BYTES:
+        return Oid.BYTEA;
+      case TIMESTAMP:
+        return Oid.TIMESTAMP;
+      case DATE:
+        return Oid.DATE;
+      case ARRAY:
+        switch (type.getArrayElementType().getCode()) {
+          case BOOL:
+            return Oid.BOOL_ARRAY;
+          case INT64:
+            return Oid.INT8_ARRAY;
+          case PG_NUMERIC:
+            return Oid.NUMERIC_ARRAY;
+          case FLOAT64:
+            return Oid.FLOAT8_ARRAY;
+          case STRING:
+            return Oid.VARCHAR_ARRAY;
+          case JSON:
+            return Oid.JSONB_ARRAY;
+          case BYTES:
+            return Oid.BYTEA_ARRAY;
+          case TIMESTAMP:
+            return Oid.TIMESTAMP_ARRAY;
+          case DATE:
+            return Oid.DATE_ARRAY;
+          case NUMERIC:
+          case ARRAY:
+          case STRUCT:
+          default:
+            throw SpannerExceptionFactory.newSpannerException(
+                ErrorCode.INVALID_ARGUMENT, "Unsupported or unknown array type: " + type);
+        }
+      case NUMERIC:
+      case STRUCT:
+      default:
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT, "Unsupported or unknown type: " + type);
     }
   }
 
@@ -241,4 +301,6 @@ public abstract class Parser<T> {
 
   /** Used to parse data type onto binary. Override this to change binary representation. */
   protected abstract byte[] binaryParse();
+
+  public abstract void bind(Statement.Builder statementBuilder, String name);
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/StringParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/StringParser.java
index b410199c..428013b4 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/StringParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/StringParser.java
@@ -14,15 +14,15 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import java.nio.charset.StandardCharsets;
-import java.sql.ResultSet;
-import java.sql.SQLException;
 import java.sql.Types;
 
 /** Translate from wire protocol to string. */
 public class StringParser extends Parser<String> {
 
-  public StringParser(ResultSet item, int position) throws SQLException {
+  public StringParser(ResultSet item, int position) {
     this.item = item.getString(position);
   }
 
@@ -48,6 +48,10 @@ public class StringParser extends Parser<String> {
 
   @Override
   protected byte[] binaryParse() {
-    return this.item.getBytes(StandardCharsets.UTF_8);
+    return this.item == null ? null : this.item.getBytes(StandardCharsets.UTF_8);
+  }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
   }
 }
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 e842d6d2..6574ff95 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
@@ -14,13 +14,12 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import com.google.common.base.Preconditions;
 import java.nio.charset.StandardCharsets;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Timestamp;
 import java.sql.Types;
-import java.time.Instant;
 import java.time.format.DateTimeFormatter;
 import java.time.format.DateTimeFormatterBuilder;
 import java.time.temporal.ChronoField;
@@ -35,8 +34,9 @@ public class TimestampParser extends Parser<Timestamp> {
   public static final int MICROSECONDS_IN_SECOND = 1000000;
   public static final long NANOSECONDS_IN_MICROSECONDS = 1000L;
   private static final char TIMESTAMP_SEPARATOR = 'T';
-  private static final char ZERO_TIMEZONE = 'Z';
   private static final char EMPTY_SPACE = ' ';
+  private static final String ZERO_TIMEZONE = "Z";
+  private static final String PG_ZERO_TIMEZONE = "+00";
 
   /** Regular expression for parsing timestamps. */
   private static final String TIMESTAMP_REGEX =
@@ -56,7 +56,7 @@ public class TimestampParser extends Parser<Timestamp> {
           .appendOffset("+HH:mm", "Z")
           .toFormatter();
 
-  public TimestampParser(ResultSet item, int position) throws SQLException {
+  public TimestampParser(ResultSet item, int position) {
     this.item = item.getTimestamp(position);
   }
 
@@ -70,16 +70,17 @@ public class TimestampParser extends Parser<Timestamp> {
         case TEXT:
           String stringValue = new String(item, StandardCharsets.UTF_8);
           TemporalAccessor temporalAccessor = TIMESTAMP_FORMATTER.parse(stringValue);
-          this.item = Timestamp.from(Instant.from(temporalAccessor));
+          this.item =
+              Timestamp.ofTimeSecondsAndNanos(
+                  temporalAccessor.getLong(ChronoField.INSTANT_SECONDS),
+                  temporalAccessor.get(ChronoField.NANO_OF_SECOND));
           break;
         case BINARY:
           long pgMicros = ByteConverter.int8(item, 0);
           com.google.cloud.Timestamp ts = com.google.cloud.Timestamp.ofTimeMicroseconds(pgMicros);
           long javaSeconds = ts.getSeconds() + PG_EPOCH_SECONDS;
           int javaNanos = ts.getNanos();
-          this.item =
-              com.google.cloud.Timestamp.ofTimeSecondsAndNanos(javaSeconds, javaNanos)
-                  .toSqlTimestamp();
+          this.item = Timestamp.ofTimeSecondsAndNanos(javaSeconds, javaNanos);
           break;
         default:
           throw new IllegalArgumentException("Unsupported format: " + formatCode);
@@ -105,18 +106,21 @@ public class TimestampParser extends Parser<Timestamp> {
 
   @Override
   protected String stringParse() {
-    return toPGString();
+    return this.item == null ? null : toPGString();
   }
 
   @Override
   protected String spannerParse() {
-    return item.toString();
+    return this.item == null ? null : item.toString();
   }
 
   @Override
   protected byte[] binaryParse() {
+    if (this.item == null) {
+      return null;
+    }
     long microseconds =
-        ((this.item.getTime() / MILLISECONDS_IN_SECOND - PG_EPOCH_SECONDS) * MICROSECONDS_IN_SECOND)
+        ((this.item.getSeconds() - PG_EPOCH_SECONDS) * MICROSECONDS_IN_SECOND)
             + (this.item.getNanos() / NANOSECONDS_IN_MICROSECONDS);
     byte[] result = new byte[8];
     ByteConverter.int8(result, 0, microseconds);
@@ -133,6 +137,10 @@ public class TimestampParser extends Parser<Timestamp> {
     return this.item
         .toString()
         .replace(TIMESTAMP_SEPARATOR, EMPTY_SPACE)
-        .replace(ZERO_TIMEZONE, EMPTY_SPACE);
+        .replace(ZERO_TIMEZONE, PG_ZERO_TIMEZONE);
+  }
+
+  public void bind(Statement.Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/UnspecifiedParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/UnspecifiedParser.java
index 3d0d88fb..1109f45a 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/parsers/UnspecifiedParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/parsers/UnspecifiedParser.java
@@ -14,6 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter.parsers;
 
+import com.google.cloud.spanner.Statement.Builder;
 import com.google.cloud.spanner.Value;
 import java.nio.charset.StandardCharsets;
 import java.sql.Types;
@@ -25,7 +26,7 @@ import java.sql.Types;
 public class UnspecifiedParser extends Parser<Value> {
 
   public UnspecifiedParser(byte[] item, FormatCode formatCode) {
-    this.item = Value.string(item == null ? null : new String(item, UTF8));
+    this.item = item == null ? null : Value.string(new String(item, UTF8));
   }
 
   @Override
@@ -35,11 +36,18 @@ public class UnspecifiedParser extends Parser<Value> {
 
   @Override
   protected String stringParse() {
-    return this.item.isNull() ? null : this.item.getString();
+    return this.item == null || this.item.isNull() ? null : this.item.getString();
   }
 
   @Override
   protected byte[] binaryParse() {
-    return this.item.isNull() ? null : this.item.getString().getBytes(StandardCharsets.UTF_8);
+    return this.item == null || this.item.isNull()
+        ? null
+        : this.item.getString().getBytes(StandardCharsets.UTF_8);
+  }
+
+  @Override
+  public void bind(Builder statementBuilder, String name) {
+    statementBuilder.bind(name).to(this.item);
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java
index 532f7a4f..d397e9a3 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyStatement.java
@@ -17,16 +17,18 @@ package com.google.cloud.spanner.pgadapter.statements;
 import static com.google.cloud.spanner.pgadapter.parsers.copy.Copy.parse;
 import static com.google.cloud.spanner.pgadapter.parsers.copy.CopyTreeParser.CopyOptions.Format;
 
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.pgadapter.parsers.copy.CopyTreeParser;
 import com.google.cloud.spanner.pgadapter.parsers.copy.TokenMgrError;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.utils.StatementParser;
 import com.google.common.base.Strings;
 import com.google.spanner.v1.TypeCode;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -45,12 +47,11 @@ public class CopyStatement extends IntermediateStatement {
   private Map<String, TypeCode> tableColumns;
   private MutationWriter mutationWriter;
 
-  public CopyStatement(String sql, Connection connection) throws SQLException {
+  public CopyStatement(String sql, Connection connection) {
     super(sql);
     this.sql = sql;
     this.command = StatementParser.parseCommand(sql);
     this.connection = connection;
-    this.statement = connection.createStatement();
   }
 
   /** @return Mapping of table column names to column type. */
@@ -131,19 +132,21 @@ public class CopyStatement extends IntermediateStatement {
     return (options.getFormat() == CopyTreeParser.CopyOptions.Format.BINARY) ? 1 : 0;
   }
 
-  private void verifyCopyColumns() throws SQLException {
+  private void verifyCopyColumns() {
     if (options.getColumnNames().size() == 0) {
       // Use all columns if none were specified.
       return;
     }
     if (options.getColumnNames().size() > this.tableColumns.size()) {
-      throw new SQLException("Number of copy columns provided exceed table column count");
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.INVALID_ARGUMENT, "Number of copy columns provided exceed table column count");
     }
     LinkedHashMap<String, TypeCode> tempTableColumns = new LinkedHashMap<>();
     // Verify that every copy column given is a valid table column name
     for (String copyColumn : options.getColumnNames()) {
       if (!this.tableColumns.containsKey(copyColumn)) {
-        throw new SQLException(
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT,
             "Column \"" + copyColumn + "\" of relation \"" + getTableName() + "\" does not exist");
       }
       // Update table column ordering with the copy column ordering
@@ -182,29 +185,30 @@ public class CopyStatement extends IntermediateStatement {
     }
   }
 
-  private void queryInformationSchema() throws SQLException {
+  private void queryInformationSchema() {
     Map<String, TypeCode> tableColumns = new LinkedHashMap<>();
-    // The mutation API requires GoogleSQL type names instead of PostgreSQL type names for the table
-    // column types. Issue the information_schema table query as a GoogleSQL query so that it will
-    // return GoogleSQL type names.
-    PreparedStatement statement =
-        this.connection.prepareStatement(
-            "SELECT "
-                + COLUMN_NAME
-                + ", "
-                + DATA_TYPE
-                + " FROM information_schema.columns WHERE table_name = ?");
-    statement.setString(1, getTableName());
-    ResultSet result = statement.executeQuery();
-
-    while (result.next()) {
-      String columnName = result.getString(COLUMN_NAME);
-      TypeCode type = parsePostgreSQLDataType(result.getString(DATA_TYPE));
-      tableColumns.put(columnName, type);
+    Statement statement =
+        Statement.newBuilder(
+                "SELECT "
+                    + COLUMN_NAME
+                    + ", "
+                    + DATA_TYPE
+                    + " FROM information_schema.columns WHERE table_name = $1")
+            .bind("p1")
+            .to(getTableName())
+            .build();
+    try (ResultSet result = connection.executeQuery(statement)) {
+      while (result.next()) {
+        String columnName = result.getString(COLUMN_NAME);
+        TypeCode type = parsePostgreSQLDataType(result.getString(DATA_TYPE));
+        tableColumns.put(columnName, type);
+      }
     }
 
     if (tableColumns.isEmpty()) {
-      throw new SQLException("Table " + getTableName() + " is not found in information_schema");
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.INVALID_ARGUMENT,
+          "Table " + getTableName() + " is not found in information_schema");
     }
 
     this.tableColumns = tableColumns;
@@ -215,7 +219,7 @@ public class CopyStatement extends IntermediateStatement {
   }
 
   @Override
-  public void handleExecutionException(SQLException e) {
+  public void handleExecutionException(SpannerException e) {
     super.handleExecutionException(e);
   }
 
@@ -229,9 +233,10 @@ public class CopyStatement extends IntermediateStatement {
       mutationWriter =
           new MutationWriter(
               options.getTableName(), getTableColumns(), getParserFormat(), hasHeader());
-      updateResultCount(); // Gets update count, set hasMoreData false and statement result
+      updateCount = (long) mutationWriter.getRowCount();
     } catch (Exception e) {
-      SQLException se = new SQLException(e.toString());
+      SpannerException se =
+          SpannerExceptionFactory.newSpannerException(ErrorCode.UNKNOWN, e.getMessage(), e);
       handleExecutionException(se);
     }
   }
@@ -240,7 +245,8 @@ public class CopyStatement extends IntermediateStatement {
     try {
       parse(sql, this.options);
     } catch (Exception | TokenMgrError e) {
-      throw new SQLException("Invalid COPY statement syntax: " + e.toString());
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.INVALID_ARGUMENT, "Invalid COPY statement syntax: " + e);
     }
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java
index b82ba060..c1ff7be5 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java
@@ -14,13 +14,13 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
+import com.google.cloud.spanner.ReadContext.QueryAnalyzeMode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.DescribePortalMetadata;
-import com.google.common.collect.SetMultimap;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.logging.Level;
@@ -37,21 +37,21 @@ public class IntermediatePortalStatement extends IntermediatePreparedStatement {
   protected List<Short> parameterFormatCodes;
   protected List<Short> resultFormatCodes;
 
-  public IntermediatePortalStatement(
-      PreparedStatement statement,
-      String sql,
-      int parameterCount,
-      SetMultimap<Integer, Integer> parameterIndexToPositions,
-      Connection connection) {
-    super(statement, sql, parameterCount, parameterIndexToPositions, connection);
+  public IntermediatePortalStatement(String sql, Connection connection) {
+    super(sql, connection);
+    this.statement = Statement.of(sql);
     this.parameterFormatCodes = new ArrayList<>();
     this.resultFormatCodes = new ArrayList<>();
   }
 
+  void setBoundStatement(Statement statement) {
+    this.statement = statement;
+  }
+
   public short getParameterFormatCode(int index) {
     if (this.parameterFormatCodes.size() == 0) {
       return 0;
-    } else if (this.parameterFormatCodes.size() == 1) {
+    } else if (index >= this.parameterFormatCodes.size()) {
       return this.parameterFormatCodes.get(0);
     } else {
       return this.parameterFormatCodes.get(index);
@@ -79,10 +79,14 @@ public class IntermediatePortalStatement extends IntermediatePreparedStatement {
 
   @Override
   public DescribeMetadata describe() throws Exception {
-    try {
-      ResultSetMetaData metaData = ((PreparedStatement) this.statement).getMetaData();
-      return new DescribePortalMetadata(metaData);
-    } catch (SQLException e) {
+    // TODO: Consider replacing this with an execute call, so we don't take two round-trips to the
+    // backend just to first describe and then execute a query.
+    try (ResultSet resultSet = connection.analyzeQuery(this.statement, QueryAnalyzeMode.PLAN)) {
+      // TODO: Remove ResultSet.next() call once this is supported in the client library.
+      // See https://github.com/googleapis/java-spanner/pull/1691
+      resultSet.next();
+      return new DescribePortalMetadata(resultSet);
+    } catch (SpannerException e) {
       /* Generally this error will occur when a non-SELECT portal statement is described in Spanner,
         however, it could occur when a statement is incorrectly formatted. Though we could catch
         this early if we could parse the type of statement, it is a significant burden on the
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java
index 5086f8af..f60c3c41 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePreparedStatement.java
@@ -14,19 +14,18 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeMetadata;
-import com.google.cloud.spanner.pgadapter.metadata.SQLMetadata;
 import com.google.cloud.spanner.pgadapter.parsers.Parser;
 import com.google.cloud.spanner.pgadapter.parsers.Parser.FormatCode;
-import com.google.cloud.spanner.pgadapter.utils.Converter;
 import com.google.cloud.spanner.pgadapter.utils.StatementParser;
-import com.google.common.collect.SetMultimap;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.SQLException;
 import java.util.List;
+import org.postgresql.core.Oid;
 
 /**
  * Intermediate representation for prepared statements (i.e.: statements before they become portals)
@@ -34,34 +33,13 @@ import java.util.List;
 public class IntermediatePreparedStatement extends IntermediateStatement {
 
   private static final Charset UTF8 = StandardCharsets.UTF_8;
-  protected final int parameterCount;
-  protected final SetMultimap<Integer, Integer> parameterIndexToPositions;
   protected List<Integer> parameterDataTypes;
+  protected Statement statement;
 
-  public IntermediatePreparedStatement(String sql, Connection connection) throws SQLException {
-    super(sql);
-    SQLMetadata parsedSQL = Converter.toJDBCParams(sql);
-    this.parameterCount = parsedSQL.getParameterCount();
-    this.sql = parsedSQL.getSqlString();
-    this.parameterIndexToPositions = parsedSQL.getParameterIndexToPositions();
-    this.command = StatementParser.parseCommand(sql);
-    this.connection = connection;
-    this.statement = this.connection.prepareStatement(this.sql);
-    this.parameterDataTypes = null;
-  }
-
-  protected IntermediatePreparedStatement(
-      PreparedStatement statement,
-      String sql,
-      int totalParameters,
-      SetMultimap<Integer, Integer> parameterIndexToPositions,
-      Connection connection) {
+  public IntermediatePreparedStatement(String sql, Connection connection) {
     super(sql);
     this.sql = sql;
     this.command = StatementParser.parseCommand(sql);
-    this.parameterCount = totalParameters;
-    this.parameterIndexToPositions = parameterIndexToPositions;
-    this.statement = statement;
     this.connection = connection;
     this.parameterDataTypes = null;
   }
@@ -78,7 +56,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
     if (this.parameterDataTypes.size() > index) {
       return this.parameterDataTypes.get(index);
     } else {
-      throw new IllegalArgumentException("PgAdapter does not support untyped parameters.");
+      return Oid.UNSPECIFIED;
     }
   }
 
@@ -86,26 +64,17 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
     return this.parameterDataTypes;
   }
 
-  public void setParameterDataTypes(List<Integer> parameterDataTypes)
-      throws IllegalArgumentException {
+  public void setParameterDataTypes(List<Integer> parameterDataTypes) {
     this.parameterDataTypes = parameterDataTypes;
-    if (this.parameterDataTypes.size() != this.parameterCount) {
-      throw new IllegalArgumentException(
-          "Number of supplied parameter data types must match number of parameters. PgAdapter does not support untyped parameters.");
-    }
-  }
-
-  public int getParameterCount() {
-    return this.parameterCount;
   }
 
   @Override
   public void execute() {
     this.executed = true;
     try {
-      ((PreparedStatement) this.statement).execute();
-      this.updateResultCount();
-    } catch (SQLException e) {
+      StatementResult result = connection.execute(this.statement);
+      this.updateResultCount(result);
+    } catch (SpannerException e) {
       handleExecutionException(e);
     }
   }
@@ -118,36 +87,22 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
    * @param parameterFormatCodes A list of the format of each parameter.
    * @param resultFormatCodes A list of the desired format of each result.
    * @return An Intermediate Portal Statement (or rather a bound version of this statement)
-   * @throws SQLException If the binding fails.
    */
   public IntermediatePortalStatement bind(
-      byte[][] parameters, List<Short> parameterFormatCodes, List<Short> resultFormatCodes)
-      throws SQLException {
-    IntermediatePortalStatement portal =
-        new IntermediatePortalStatement(
-            this.connection.prepareStatement(this.sql),
-            this.sql,
-            this.parameterCount,
-            this.parameterIndexToPositions,
-            this.connection);
+      byte[][] parameters, List<Short> parameterFormatCodes, List<Short> resultFormatCodes) {
+    IntermediatePortalStatement portal = new IntermediatePortalStatement(this.sql, this.connection);
     portal.setParameterFormatCodes(parameterFormatCodes);
     portal.setResultFormatCodes(resultFormatCodes);
-    if (parameters.length != parameterCount) {
-      throw new IllegalArgumentException(
-          "Wrong number of parameters for prepared statement, expected: "
-              + parameterCount
-              + " but got: "
-              + parameters.length);
-    }
+    Statement.Builder builder = Statement.newBuilder(sql);
     for (int index = 0; index < parameters.length; index++) {
       short formatCode = portal.getParameterFormatCode(index);
       int type = this.parseType(parameters, index);
-      for (Integer position : parameterIndexToPositions.get(index)) {
-        Parser<?> parser = Parser.create(parameters[index], type, FormatCode.of(formatCode));
-        ((PreparedStatement) portal.statement)
-            .setObject(position, parser.getItem(), parser.getSqlType());
-      }
+      Parser<?> parser = Parser.create(parameters[index], type, FormatCode.of(formatCode));
+      parser.bind(builder, "p" + (index + 1));
     }
+    this.statement = builder.build();
+    portal.setBoundStatement(statement);
+
     return portal;
   }
 
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/MatcherStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/MatcherStatement.java
index e46402ee..dcf2ba44 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/MatcherStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/MatcherStatement.java
@@ -17,7 +17,6 @@ package com.google.cloud.spanner.pgadapter.statements;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.commands.Command;
 import com.google.cloud.spanner.pgadapter.utils.StatementParser;
-import java.sql.SQLException;
 import org.json.simple.JSONObject;
 
 /**
@@ -30,10 +29,9 @@ public class MatcherStatement extends IntermediateStatement {
 
   private JSONObject commandMetadataJSON;
 
-  public MatcherStatement(String sql, ConnectionHandler connectionHandler) throws SQLException {
+  public MatcherStatement(String sql, ConnectionHandler connectionHandler) {
     super(sql);
-    this.connection = connectionHandler.getJdbcConnection();
-    this.statement = this.connection.createStatement();
+    this.connection = connectionHandler.getSpannerConnection();
     this.commandMetadataJSON = connectionHandler.getServer().getOptions().getCommandMetadataJSON();
     this.sql = translateSQL(sql);
     this.statements = parseStatements(sql);
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/Converter.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/Converter.java
index 72211492..acd05613 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/Converter.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/Converter.java
@@ -14,120 +14,26 @@
 
 package com.google.cloud.spanner.pgadapter.utils;
 
+import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.pgadapter.ProxyServer.DataFormat;
-import com.google.cloud.spanner.pgadapter.metadata.SQLMetadata;
 import com.google.cloud.spanner.pgadapter.parsers.Parser;
 import com.google.common.base.Preconditions;
-import com.google.common.collect.HashMultimap;
-import com.google.common.collect.SetMultimap;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
 
 /** Utility class for converting between generic PostgreSQL conversions. */
 public class Converter {
-
-  /**
-   * PostgreSQL parameters occur as $\\d+, whereas JDBC expects a question mark. For now, doing a
-   * simple iteration to correct. Taking into account both quoted strings and escape sequences.
-   *
-   * @param sql The PostgreSQL String.
-   * @return A {@link SQLMetadata} object containing both the corrected SQL String as well as the
-   *     number of parameters iterated.
-   */
-  public static SQLMetadata toJDBCParams(String sql) {
-    Preconditions.checkNotNull(sql);
-    StringBuilder result = new StringBuilder();
-    // Multimap from the 0-based parameter index to the associated set of JDBC parameter positions
-    // (1-based).
-    SetMultimap<Integer, Integer> parameterIndexToPositions = HashMultimap.create();
-    final char SINGLE_QUOTE = '\'';
-    final char DOUBLE_QUOTE = '"';
-    final char DOLLAR = '$';
-    final char BACK_SLASH = '\\';
-    String currentTag = null;
-    boolean isInQuoted = false;
-    char startQuote = 0;
-    boolean lastCharWasEscapeChar = false;
-    int parameterOrder = 0;
-    int totalParameterCount = 0;
-    int index = 0;
-    while (index < sql.length()) {
-      char c = sql.charAt(index);
-      if (isInQuoted) {
-        if (c == startQuote) {
-          if (c == DOLLAR) {
-            // Check if this is the end of the current dollar quoted string.
-            String tag = StatementParser.parseDollarQuotedString(sql, index + 1);
-            if (tag != null && tag.equals(currentTag)) {
-              index += tag.length() + 1;
-              result.append(c);
-              result.append(tag);
-              isInQuoted = false;
-              startQuote = 0;
-            }
-          } else if (lastCharWasEscapeChar) {
-            lastCharWasEscapeChar = false;
-          } else {
-            isInQuoted = false;
-            startQuote = 0;
-          }
-        } else if (c == BACK_SLASH) {
-          lastCharWasEscapeChar = true;
-        } else {
-          lastCharWasEscapeChar = false;
-        }
-      } else {
-        if (c == SINGLE_QUOTE || c == DOUBLE_QUOTE) {
-          isInQuoted = true;
-          startQuote = c;
-        } else if (c == DOLLAR && sql.length() > index + 1) {
-          if (Character.isDigit(sql.charAt(index + 1))) {
-            // Consume the parameter index.
-            int beginIndex = index + 1;
-            result.append('?');
-            while (++index < sql.length() && Character.isDigit(sql.charAt(index))) {}
-            int parameterIndex = Integer.valueOf(sql.substring(beginIndex, index)) - 1;
-            if (parameterIndex < 0) {
-              throw new IllegalArgumentException("Parameter index should be >= 1");
-            }
-            parameterOrder++;
-            parameterIndexToPositions.put(parameterIndex, parameterOrder);
-            totalParameterCount = Integer.max(totalParameterCount, parameterIndex + 1);
-            continue;
-          }
-          currentTag = StatementParser.parseDollarQuotedString(sql, index + 1);
-          if (currentTag != null) {
-            isInQuoted = true;
-            startQuote = DOLLAR;
-            index += currentTag.length() + 1;
-            result.append(c);
-            result.append(currentTag);
-          }
-        }
-      }
-      result.append(c);
-      index++;
-    }
-    return new SQLMetadata(result.toString(), totalParameterCount, parameterIndexToPositions);
-  }
-
   /**
    * Return the data of the specified column of the {@link ResultSet} as a byte array. The column
    * may not contain a null value.
    *
    * @param result The {@link ResultSet} to read the data from.
-   * @param metadata The {@link ResultSetMetaData} object holding the metadata for the result.
    * @param columnarIndex The columnar index.
    * @param format The {@link DataFormat} format to use to encode the data.
    * @return a byte array containing the data in the specified format.
    */
-  public byte[] parseData(
-      ResultSet result, ResultSetMetaData metadata, int columnarIndex, DataFormat format)
-      throws SQLException {
+  public byte[] parseData(ResultSet result, int columnarIndex, DataFormat format) {
     Preconditions.checkArgument(
-        result.getObject(columnarIndex) != null, "Column may not contain a null value");
-    Parser parser = Parser.create(result, metadata.getColumnType(columnarIndex), columnarIndex);
+        !result.isNull(columnarIndex), "Column may not contain a null value");
+    Parser<?> parser = Parser.create(result, result.getColumnType(columnarIndex), columnarIndex);
     return parser.parse(format);
   }
 }
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 9531cb18..f3ceaebb 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
@@ -14,9 +14,11 @@
 
 package com.google.cloud.spanner.pgadapter.utils;
 
+import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.Mutation;
 import com.google.cloud.spanner.Mutation.WriteBuilder;
-import com.google.cloud.spanner.jdbc.CloudSpannerJdbcConnection;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.spanner.v1.TypeCode;
 import java.io.ByteArrayOutputStream;
@@ -24,8 +26,6 @@ import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.sql.Connection;
-import java.sql.SQLException;
 import java.time.format.DateTimeParseException;
 import java.util.ArrayList;
 import java.util.List;
@@ -76,7 +76,8 @@ public class MutationWriter {
     this.payload.write(payload, 0, payload.length);
     if (!commitSizeIsWithinLimit()) {
       handleError(connectionHandler);
-      throw new SQLException(
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.FAILED_PRECONDITION,
           "Commit size: " + this.payload.size() + " has exceeded the limit: " + COMMIT_LIMIT);
     }
   }
@@ -88,7 +89,8 @@ public class MutationWriter {
       // Check that the number of columns in a record matches the number of columns in the table
       if (record.size() != this.tableColumns.keySet().size()) {
         handleError(connectionHandler);
-        throw new SQLException(
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT,
             "Invalid COPY data: Row length mismatched. Expected "
                 + this.tableColumns.keySet().size()
                 + " columns, but only found "
@@ -123,7 +125,8 @@ public class MutationWriter {
           }
         } catch (NumberFormatException | DateTimeParseException e) {
           handleError(connectionHandler);
-          throw new SQLException(
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.INVALID_ARGUMENT,
               "Invalid input syntax for type "
                   + columnType.toString()
                   + ":"
@@ -132,7 +135,8 @@ public class MutationWriter {
                   + "\"");
         } catch (IllegalArgumentException e) {
           handleError(connectionHandler);
-          throw new SQLException("Invalid input syntax for column \"" + columnName + "\"");
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.INVALID_ARGUMENT, "Invalid input syntax for column \"" + columnName + "\"");
         } catch (Exception e) {
           handleError(connectionHandler);
           throw e;
@@ -144,7 +148,8 @@ public class MutationWriter {
     }
     if (!mutationCountIsWithinLimit()) {
       handleError(connectionHandler);
-      throw new SQLException(
+      throw SpannerExceptionFactory.newSpannerException(
+          ErrorCode.FAILED_PRECONDITION,
           "Mutation count: " + mutationCount + " has exceeded the limit: " + MUTATION_LIMIT);
     }
   }
@@ -191,14 +196,12 @@ public class MutationWriter {
    *
    * @return count of the number of rows updated.
    */
-  public int writeToSpanner(ConnectionHandler connectionHandler) throws SQLException {
+  public int writeToSpanner(ConnectionHandler connectionHandler) {
     if (this.mutations.isEmpty()) {
       return 0;
     }
-    Connection connection = connectionHandler.getJdbcConnection();
-    CloudSpannerJdbcConnection spannerConnection =
-        connection.unwrap(CloudSpannerJdbcConnection.class);
-    spannerConnection.write(this.mutations); // Write mutation to spanner
+    Connection connection = connectionHandler.getSpannerConnection();
+    connection.write(this.mutations); // Write mutation to spanner
     // Reset mutations, mutation counter, and batch size count for a new batch
     this.mutations = new ArrayList<>();
     this.mutationCount = 0;
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/DataRowResponse.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/DataRowResponse.java
index b4b6aeb1..9575ea04 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/DataRowResponse.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/DataRowResponse.java
@@ -14,14 +14,13 @@
 
 package com.google.cloud.spanner.pgadapter.wireoutput;
 
+import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.ProxyServer.DataFormat;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.cloud.spanner.pgadapter.utils.Converter;
 import java.io.DataOutputStream;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -35,7 +34,6 @@ public class DataRowResponse extends WireOutput {
   private final IntermediateStatement statement;
   private final QueryMode mode;
   private final ResultSet resultSet;
-  private final ResultSetMetaData metadata;
   private final List<byte[]> columns;
 
   public DataRowResponse(
@@ -48,20 +46,18 @@ public class DataRowResponse extends WireOutput {
     this.statement = statement;
     this.mode = mode;
     this.resultSet = statement.getStatementResult();
-    this.metadata = resultSet.getMetaData();
-    this.columns = new ArrayList<>(this.metadata.getColumnCount());
+    this.columns = new ArrayList<>(this.resultSet.getColumnCount());
     this.length = HEADER_LENGTH + COLUMN_NUMBER_LENGTH;
     // TODO profile this with immense rows/column count to see if it's an aread of optimization.
-    for (int column_index = 1; /* column indices start at 1 */
-        column_index <= this.metadata.getColumnCount();
+    for (int column_index = 0; /* column indices start at 0 */
+        column_index < this.resultSet.getColumnCount();
         column_index++) {
       length += COLUMN_SIZE_LENGTH;
-      if (resultSet.getObject(column_index) == null) {
+      if (resultSet.isNull(column_index)) {
         columns.add(null);
       } else {
         DataFormat format = DataFormat.getDataFormat(column_index, statement, this.mode, options);
-        byte[] column =
-            new Converter().parseData(this.resultSet, this.metadata, column_index, format);
+        byte[] column = new Converter().parseData(this.resultSet, column_index, format);
         length += column.length;
         columns.add(column);
       }
@@ -70,7 +66,7 @@ public class DataRowResponse extends WireOutput {
 
   @Override
   protected void sendPayload() throws Exception {
-    this.outputStream.writeShort(this.metadata.getColumnCount());
+    this.outputStream.writeShort(this.resultSet.getColumnCount());
     for (byte[] column_value : columns) {
       if (column_value == null) {
         this.outputStream.writeInt(-1);
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/RowDescriptionResponse.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/RowDescriptionResponse.java
index dc343476..c386a9c1 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/RowDescriptionResponse.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/RowDescriptionResponse.java
@@ -14,17 +14,14 @@
 
 package com.google.cloud.spanner.pgadapter.wireoutput;
 
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Type;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.ProxyServer.DataFormat;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.parsers.Parser;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import java.io.DataOutputStream;
-import java.math.BigDecimal;
-import java.sql.Date;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Timestamp;
-import java.sql.Types;
 import java.text.MessageFormat;
 import org.postgresql.core.Oid;
 
@@ -44,7 +41,7 @@ public class RowDescriptionResponse extends WireOutput {
   private static final byte DEFAULT_FLAG = 0;
 
   private final IntermediateStatement statement;
-  private final ResultSetMetaData metadata;
+  private final ResultSet resultSet;
   private final QueryMode mode;
   private final OptionsMetadata options;
   private final int columnCount;
@@ -52,25 +49,25 @@ public class RowDescriptionResponse extends WireOutput {
   public RowDescriptionResponse(
       DataOutputStream output,
       IntermediateStatement statement,
-      ResultSetMetaData metadata,
+      ResultSet resultSet,
       OptionsMetadata options,
       QueryMode mode)
       throws Exception {
-    super(output, calculateLength(metadata));
+    super(output, calculateLength(resultSet));
     this.statement = statement;
-    this.metadata = metadata;
+    this.resultSet = resultSet;
     this.options = options;
     this.mode = mode;
-    this.columnCount = metadata.getColumnCount();
+    this.columnCount = resultSet.getColumnCount();
   }
 
-  private static int calculateLength(ResultSetMetaData metadata) throws Exception {
+  private static int calculateLength(ResultSet resultSet) throws Exception {
     int length = HEADER_LENGTH + FIELD_NUMBER_LENGTH;
-    for (int column_index = 1; /* columns start at 1 */
-        column_index <= metadata.getColumnCount();
+    for (int column_index = 0; /* columns start at 0 */
+        column_index < resultSet.getColumnCount();
         column_index++) {
       length +=
-          metadata.getColumnName(column_index).length()
+          resultSet.getType().getStructFields().get(column_index).getName().length()
               + NULL_TERMINATOR_LENGTH
               + TABLE_OID_LENGTH
               + COLUMN_INDEX_LENGTH
@@ -86,10 +83,11 @@ public class RowDescriptionResponse extends WireOutput {
   protected void sendPayload() throws Exception {
     this.outputStream.writeShort(this.columnCount);
     DataFormat defaultFormat = DataFormat.getDataFormat(0, this.statement, this.mode, this.options);
-    for (int column_index = 1; /* columns start at 1 */
-        column_index <= this.columnCount;
+    for (int column_index = 0; /* columns start at 0 */
+        column_index < this.columnCount;
         column_index++) {
-      this.outputStream.write(this.metadata.getColumnName(column_index).getBytes(UTF8));
+      this.outputStream.write(
+          this.resultSet.getType().getStructFields().get(column_index).getName().getBytes(UTF8));
       // If it can be identified as a column of a table, the object ID of the table.
       this.outputStream.writeByte(DEFAULT_FLAG);
       // TODO: pass through Postgres types
@@ -107,7 +105,7 @@ public class RowDescriptionResponse extends WireOutput {
               ? defaultFormat.getCode()
               : this.statement.getResultFormatCode(column_index) == 0
                   ? defaultFormat.getCode()
-                  : this.statement.getResultFormatCode(column_index - 1);
+                  : this.statement.getResultFormatCode(column_index);
       this.outputStream.writeShort(format);
     }
   }
@@ -131,67 +129,9 @@ public class RowDescriptionResponse extends WireOutput {
             });
   }
 
-  int getOidType(int column_index) throws SQLException {
-    int type = this.metadata.getColumnType(column_index);
-    switch (type) {
-      case Types.SMALLINT:
-        return Oid.INT2;
-      case Types.INTEGER:
-        return Oid.INT4;
-      case Types.BIGINT:
-        return Oid.INT8;
-      case Types.NUMERIC:
-        return Oid.NUMERIC;
-      case Types.REAL:
-        return Oid.FLOAT4;
-      case Types.DOUBLE:
-        return Oid.FLOAT8;
-      case Types.CHAR:
-        return Oid.CHAR;
-      case Types.VARCHAR:
-        return Oid.VARCHAR;
-      case Types.BINARY:
-        return Oid.BYTEA;
-      case Types.BIT:
-        return Oid.BOOL;
-      case Types.DATE:
-        return Oid.DATE;
-      case Types.TIME:
-        return Oid.TIME;
-      case Types.TIMESTAMP:
-        return Oid.TIMESTAMP;
-      case Types.ARRAY:
-        // TODO: Rewrite to use Cloud Spanner ResultSetMetaData when refactored to use the
-        // Connection API instead of the JDBC driver, instead of checking the class name.
-        String typeName = metadata.getColumnClassName(column_index);
-        if (Boolean[].class.getName().equals(typeName)) {
-          return Oid.BOOL_ARRAY;
-        }
-        if (Long[].class.getName().equals(typeName)) {
-          return Oid.INT8_ARRAY;
-        }
-        if (Double[].class.getName().equals(typeName)) {
-          return Oid.FLOAT8_ARRAY;
-        }
-        if (BigDecimal[].class.getName().equals(typeName)) {
-          return Oid.NUMERIC_ARRAY;
-        }
-        if (String[].class.getName().equals(typeName)) {
-          return Oid.VARCHAR_ARRAY;
-        }
-        if (byte[][].class.getName().equals(typeName)) {
-          return Oid.BYTEA_ARRAY;
-        }
-        if (Date[].class.getName().equals(typeName)) {
-          return Oid.DATE_ARRAY;
-        }
-        if (Timestamp[].class.getName().equals(typeName)) {
-          return Oid.TIMESTAMP_ARRAY;
-        }
-        return Oid.UNSPECIFIED;
-      default:
-        return Oid.UNSPECIFIED;
-    }
+  int getOidType(int column_index) {
+    Type type = this.resultSet.getColumnType(column_index);
+    return Parser.toOid(type);
   }
 
   int getOidTypeSize(int oid_type) {
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
index 1da089f1..41c8ff3e 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
@@ -14,6 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
+import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
@@ -27,7 +28,6 @@ import com.google.cloud.spanner.pgadapter.wireoutput.PortalSuspendedResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
 import java.io.DataInputStream;
 import java.io.IOException;
-import java.sql.ResultSet;
 import java.util.ArrayList;
 import java.util.List;
 
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDataMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDataMessage.java
index 0606fefa..356b8b66 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDataMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDataMessage.java
@@ -14,11 +14,11 @@
 
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
+import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import java.io.IOException;
-import java.sql.SQLException;
 import java.text.MessageFormat;
 
 /**
@@ -51,7 +51,7 @@ public class CopyDataMessage extends ControlMessage {
     if (!statement.hasException()) {
       try {
         mw.addCopyData(this.connection, this.payload);
-      } catch (SQLException e) {
+      } catch (SpannerException e) {
         mw.writeCopyDataToErrorFile();
         statement.handleExecutionException(e);
         throw e;
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FlushMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FlushMessage.java
index 594b879b..93553b26 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FlushMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/FlushMessage.java
@@ -14,7 +14,6 @@
 
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
-import com.google.cloud.spanner.jdbc.CloudSpannerJdbcConnection;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
@@ -34,8 +33,7 @@ public class FlushMessage extends ControlMessage {
 
   @Override
   protected void sendPayload() throws Exception {
-    boolean inTransaction =
-        connection.getJdbcConnection().unwrap(CloudSpannerJdbcConnection.class).isInTransaction();
+    boolean inTransaction = connection.getSpannerConnection().isInTransaction();
     new ReadyResponse(
             this.outputStream, inTransaction ? Status.TRANSACTION : ReadyResponse.Status.IDLE)
         .send();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java
index b95a6d86..bd13853c 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java
@@ -42,7 +42,8 @@ public class ParseMessage extends ControlMessage {
       int type = this.inputStream.readInt();
       this.parameterDataTypes.add(type);
     }
-    this.statement = new IntermediatePreparedStatement(queryString, connection.getJdbcConnection());
+    this.statement =
+        new IntermediatePreparedStatement(queryString, connection.getSpannerConnection());
     this.statement.setParameterDataTypes(this.parameterDataTypes);
   }
 
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
index fc8fbb48..f5ca637f 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
@@ -14,7 +14,6 @@
 
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
-import com.google.cloud.spanner.jdbc.CloudSpannerJdbcConnection;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
@@ -40,10 +39,10 @@ public class QueryMessage extends ControlMessage {
     super(connection);
     String query = StatementParser.removeCommentsAndTrim(this.readAll());
     String command = StatementParser.parseCommand(query);
-    if (command.equalsIgnoreCase(COPY)) {
-      this.statement = new CopyStatement(query, this.connection.getJdbcConnection());
+    if (COPY.equalsIgnoreCase(command)) {
+      this.statement = new CopyStatement(query, this.connection.getSpannerConnection());
     } else if (!connection.getServer().getOptions().requiresMatcher()) {
-      this.statement = new IntermediateStatement(query, this.connection.getJdbcConnection());
+      this.statement = new IntermediateStatement(query, this.connection.getSpannerConnection());
     } else {
       this.statement = new MatcherStatement(query, this.connection);
     }
@@ -105,14 +104,13 @@ public class QueryMessage extends ControlMessage {
         new RowDescriptionResponse(
                 this.outputStream,
                 this.statement,
-                this.statement.getStatementResult().getMetaData(),
+                this.statement.getStatementResult(),
                 this.connection.getServer().getOptions(),
                 QueryMode.SIMPLE)
             .send();
       }
       this.sendSpannerResult(this.statement, QueryMode.SIMPLE, 0L);
-      boolean inTransaction =
-          connection.getJdbcConnection().unwrap(CloudSpannerJdbcConnection.class).isInTransaction();
+      boolean inTransaction = connection.getSpannerConnection().isInTransaction();
       new ReadyResponse(
               this.outputStream, inTransaction ? Status.TRANSACTION : ReadyResponse.Status.IDLE)
           .send();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/SyncMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/SyncMessage.java
index 3cc4392e..22e9c207 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/SyncMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/SyncMessage.java
@@ -14,7 +14,6 @@
 
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
-import com.google.cloud.spanner.jdbc.CloudSpannerJdbcConnection;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
@@ -34,8 +33,7 @@ public class SyncMessage extends ControlMessage {
 
   @Override
   protected void sendPayload() throws Exception {
-    boolean inTransaction =
-        connection.getJdbcConnection().unwrap(CloudSpannerJdbcConnection.class).isInTransaction();
+    boolean inTransaction = connection.getSpannerConnection().isInTransaction();
     new ReadyResponse(
             this.outputStream, inTransaction ? Status.TRANSACTION : ReadyResponse.Status.IDLE)
         .send();


Test patch:
diff --git a/.ci/e2e-expected/ddl-dml-batching.txt b/.ci/e2e-expected/ddl-dml-batching.txt
index 5724a086..f1244266 100644
--- a/.ci/e2e-expected/ddl-dml-batching.txt
+++ b/.ci/e2e-expected/ddl-dml-batching.txt
@@ -1,1 +1,1 @@
-ERROR:  INVALID_ARGUMENT: Mixing DML and DDL statements in a batch is not allowed. "CREATE TABLE testtable (id integer PRIMARY KEY, data integer); INSERT INTO testtable (id, data) VALUES (1, 2); INSERT INTO testtable (id, data) VALUES (3, 4);"
+ERROR: FAILED_PRECONDITION: Executing updates is not allowed for DDL batches. "CREATE TABLE testtable (id integer PRIMARY KEY, data integer); INSERT INTO testtable (id, data) VALUES (1, 2); INSERT INTO testtable (id, data) VALUES (3, 4);"
diff --git a/.ci/e2e-expected/invalid-commit-batching.txt b/.ci/e2e-expected/invalid-commit-batching.txt
index ae898e43..78eddb5e 100644
--- a/.ci/e2e-expected/invalid-commit-batching.txt
+++ b/.ci/e2e-expected/invalid-commit-batching.txt
@@ -1,1 +1,1 @@
-ERROR:  INVALID_ARGUMENT: The statement is not suitable for batching. Only DML and DDL statements are allowed for batching. "BEGIN TRANSACTION; INSERT INTO users (id, age, name) VALUES (99, 99, 'test'); COMMIT;"
+ERROR: FAILED_PRECONDITION: This connection has an active batch and cannot begin a transaction "BEGIN TRANSACTION; INSERT INTO users (id, age, name) VALUES (99, 99, 'test'); COMMIT;"
diff --git a/.ci/e2e-expected/select-ddl-batching.txt b/.ci/e2e-expected/select-ddl-batching.txt
index b6aca940..fdfb07ae 100644
--- a/.ci/e2e-expected/select-ddl-batching.txt
+++ b/.ci/e2e-expected/select-ddl-batching.txt
@@ -1,1 +1,1 @@
-ERROR:  INVALID_ARGUMENT: The statement is not suitable for batching. Only DML and DDL statements are allowed for batching. "CREATE TABLE test (id integer PRIMARY KEY); SELECT * FROM test; SELECT * FROM test;"
+ERROR: FAILED_PRECONDITION: Executing queries is not allowed for DDL batches. "CREATE TABLE test (id integer PRIMARY KEY); SELECT * FROM test; SELECT * FROM test;"
diff --git a/.ci/e2e-expected/select-dml-batching.txt b/.ci/e2e-expected/select-dml-batching.txt
index 0382970f..87a9f5d9 100644
--- a/.ci/e2e-expected/select-dml-batching.txt
+++ b/.ci/e2e-expected/select-dml-batching.txt
@@ -1,1 +1,1 @@
-ERROR:  INVALID_ARGUMENT: The statement is not suitable for batching. Only DML and DDL statements are allowed for batching. "SELECT * FROM users; INSERT INTO users (id, age, name) VALUES (99, 99, 'person'); SELECT * FROM users;"
+ERROR: INVALID_ARGUMENT: Statement type is not supported for batching "SELECT * FROM users; INSERT INTO users (id, age, name) VALUES (99, 99, 'person'); SELECT * FROM users;"
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
index 8b860754..eeed9a99 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
@@ -15,14 +15,17 @@
 package com.google.cloud.spanner.pgadapter.statements;
 
 import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeMetadata;
 import com.google.cloud.spanner.pgadapter.utils.StatementParser;
 import com.google.common.base.Preconditions;
-import java.sql.Connection;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -35,27 +38,25 @@ public class IntermediateStatement {
   private static final AbstractStatementParser PARSER =
       AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
 
-  protected Statement statement;
   private final ResultType resultType;
   protected ResultSet statementResult;
   protected boolean hasMoreData;
-  protected Exception exception;
+  protected SpannerException exception;
   protected String sql;
   protected String command;
   protected boolean executed;
   protected Connection connection;
-  protected Integer updateCount;
+  protected Long updateCount;
   protected List<String> statements;
 
   private static final char STATEMENT_DELIMITER = ';';
   private static final char SINGLE_QUOTE = '\'';
 
-  public IntermediateStatement(String sql, Connection connection) throws SQLException {
+  public IntermediateStatement(String sql, Connection connection) {
     this.sql = sql;
     this.statements = parseStatements(sql);
     this.command = StatementParser.parseCommand(sql);
     this.connection = connection;
-    this.statement = connection.createStatement();
     // Note: This determines the result type based on the first statement in the SQL statement. That
     // means that it assumes that if this is a batch of statements, all the statements in the batch
     // will have the same type of result (that is; they are all DML statements, all DDL statements,
@@ -148,7 +149,7 @@ public class IntermediateStatement {
   }
 
   /** @return The number of items that were modified by this execution. */
-  public Integer getUpdateCount() {
+  public Long getUpdateCount() {
     return this.updateCount;
   }
 
@@ -170,8 +171,8 @@ public class IntermediateStatement {
     this.hasMoreData = hasMoreData;
   }
 
-  public Statement getStatement() {
-    return this.statement;
+  public Connection getConnection() {
+    return this.connection;
   }
 
   public List<String> getStatements() {
@@ -200,22 +201,20 @@ public class IntermediateStatement {
    * Processes the results from an execute/executeBatch execution, extracting metadata from that
    * execution (including results and update counts). An array of updateCounts is needed in the case
    * of updateBatchResultCount.
-   *
-   * @throws SQLException If an issue occurred in extracting result metadata.
    */
-  protected void updateResultCount() throws SQLException {
-    if (this.containsResultSet()) {
-      this.statementResult = this.statement.getResultSet();
+  protected void updateResultCount(StatementResult result) {
+    if (result.getResultType() == StatementResult.ResultType.RESULT_SET) {
+      this.statementResult = result.getResultSet();
       this.hasMoreData = this.statementResult.next();
     } else {
-      this.updateCount = this.statement.getUpdateCount();
+      this.updateCount = result.getUpdateCount();
       this.hasMoreData = false;
       this.statementResult = null;
     }
   }
 
-  protected void updateBatchResultCount(int[] updateCounts) throws SQLException {
-    this.updateCount = 0;
+  protected void updateBatchResultCount(long[] updateCounts) {
+    this.updateCount = 0L;
     for (int i = 0; i < updateCounts.length; ++i) {
       this.updateCount += updateCounts[i];
     }
@@ -228,7 +227,7 @@ public class IntermediateStatement {
    *
    * @param e The exception to store.
    */
-  protected void handleExecutionException(SQLException e) {
+  protected void handleExecutionException(SpannerException e) {
     this.exception = e;
     this.hasMoreData = false;
     this.statementResult = null;
@@ -237,21 +236,33 @@ public class IntermediateStatement {
   /** Execute the SQL statement, storing metadata. */
   public void execute() {
     this.executed = true;
-    int[] updateCounts = null;
     try {
       if (statements.size() > 1) {
+        // TODO: Clean this up a little once the statement parsing in the client library has been
+        // released: https://github.com/googleapis/java-spanner/pull/1690
+        // Also, the restriction that mixed batches are not allowed will be removed in a future PR.
+        if (resultType == ResultType.UPDATE_COUNT) {
+          connection.startBatchDml();
+        } else if (resultType == ResultType.NO_RESULT) {
+          connection.startBatchDdl();
+        } else {
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.INVALID_ARGUMENT, "Statement type is not supported for batching");
+        }
         for (String stmt : statements) {
-          this.statement.addBatch(stmt);
+          connection.execute(Statement.of(stmt));
         }
-        updateCounts = this.statement.executeBatch();
-        this.updateBatchResultCount(updateCounts);
+        long[] updateCounts = connection.runBatch();
+        updateBatchResultCount(updateCounts);
       } else {
-        this.statement.execute(this.sql);
-        this.updateResultCount();
+        StatementResult result = connection.execute(Statement.of(this.sql));
+        updateResultCount(result);
       }
-    } catch (SQLException e) {
+    } catch (SpannerException e) {
       if (statements.size() > 1) {
-        SQLException exception = new SQLException(e.getMessage() + " \"" + this.sql + "\"", e);
+        SpannerException exception =
+            SpannerExceptionFactory.newSpannerException(
+                e.getErrorCode(), e.getMessage() + " \"" + this.sql + "\"", e);
         handleExecutionException(exception);
       } else {
         handleExecutionException(e);
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
index 966cbe90..ae27a06c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
@@ -21,6 +21,7 @@ import com.google.api.gax.longrunning.OperationTimedPollAlgorithm;
 import com.google.api.gax.retrying.RetrySettings;
 import com.google.api.gax.rpc.UnaryCallSettings;
 import com.google.cloud.NoCredentials;
+import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.MockDatabaseAdminServiceImpl;
 import com.google.cloud.spanner.MockOperationsServiceImpl;
@@ -34,6 +35,7 @@ import com.google.cloud.spanner.connection.SpannerPool;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.ListValue;
+import com.google.protobuf.NullValue;
 import com.google.protobuf.Value;
 import com.google.spanner.admin.database.v1.CreateDatabaseRequest;
 import com.google.spanner.admin.database.v1.InstanceName;
@@ -41,11 +43,14 @@ import com.google.spanner.v1.ResultSetMetadata;
 import com.google.spanner.v1.StructType;
 import com.google.spanner.v1.StructType.Field;
 import com.google.spanner.v1.Type;
+import com.google.spanner.v1.TypeAnnotationCode;
 import com.google.spanner.v1.TypeCode;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Server;
 import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
 import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
 import java.util.logging.Logger;
 import org.junit.AfterClass;
 import org.junit.Before;
@@ -109,6 +114,79 @@ abstract class AbstractMockServerTest {
   protected static final Statement INSERT_STATEMENT = Statement.of("INSERT INTO FOO VALUES (1)");
   protected static final int INSERT_COUNT = 1;
 
+  protected static final ResultSetMetadata ALL_TYPES_METADATA =
+      ResultSetMetadata.newBuilder()
+          .setRowType(
+              StructType.newBuilder()
+                  .addFields(
+                      Field.newBuilder()
+                          .setName("col_bigint")
+                          .setType(Type.newBuilder().setCode(TypeCode.INT64).build()))
+                  .addFields(
+                      Field.newBuilder()
+                          .setName("col_bool")
+                          .setType(Type.newBuilder().setCode(TypeCode.BOOL).build()))
+                  .addFields(
+                      Field.newBuilder()
+                          .setName("col_bytea")
+                          .setType(Type.newBuilder().setCode(TypeCode.BYTES).build()))
+                  .addFields(
+                      Field.newBuilder()
+                          .setName("col_float8")
+                          .setType(Type.newBuilder().setCode(TypeCode.FLOAT64).build()))
+                  .addFields(
+                      Field.newBuilder()
+                          .setName("col_numeric")
+                          .setType(
+                              Type.newBuilder()
+                                  .setCode(TypeCode.NUMERIC)
+                                  .setTypeAnnotation(TypeAnnotationCode.PG_NUMERIC)
+                                  .build()))
+                  .addFields(
+                      Field.newBuilder()
+                          .setName("col_timestamptz")
+                          .setType(Type.newBuilder().setCode(TypeCode.TIMESTAMP).build()))
+                  .addFields(
+                      Field.newBuilder()
+                          .setName("col_varchar")
+                          .setType(Type.newBuilder().setCode(TypeCode.STRING).build()))
+                  .build())
+          .build();
+  protected static final com.google.spanner.v1.ResultSet ALL_TYPES_RESULTSET =
+      com.google.spanner.v1.ResultSet.newBuilder()
+          .setMetadata(ALL_TYPES_METADATA)
+          .addRows(
+              ListValue.newBuilder()
+                  .addValues(Value.newBuilder().setStringValue("1").build())
+                  .addValues(Value.newBuilder().setBoolValue(true).build())
+                  .addValues(
+                      Value.newBuilder()
+                          .setStringValue(
+                              Base64.getEncoder()
+                                  .encodeToString("test".getBytes(StandardCharsets.UTF_8)))
+                          .build())
+                  .addValues(Value.newBuilder().setNumberValue(3.14d).build())
+                  .addValues(Value.newBuilder().setStringValue("6.626").build())
+                  .addValues(
+                      Value.newBuilder().setStringValue("2022-02-16T13:18:02.123456789Z").build())
+                  .addValues(Value.newBuilder().setStringValue("test").build())
+                  .build())
+          .build();
+  protected static final com.google.spanner.v1.ResultSet ALL_TYPES_NULLS_RESULTSET =
+      com.google.spanner.v1.ResultSet.newBuilder()
+          .setMetadata(ALL_TYPES_METADATA)
+          .addRows(
+              ListValue.newBuilder()
+                  .addValues(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+                  .addValues(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+                  .addValues(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+                  .addValues(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+                  .addValues(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+                  .addValues(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+                  .addValues(Value.newBuilder().setNullValue(NullValue.NULL_VALUE).build())
+                  .build())
+          .build();
+
   protected static MockSpannerServiceImpl mockSpanner;
   protected static MockOperationsServiceImpl mockOperationsService;
   protected static MockDatabaseAdminServiceImpl mockDatabaseAdmin;
@@ -123,6 +201,8 @@ abstract class AbstractMockServerTest {
     mockSpanner.putStatementResult(StatementResult.query(SELECT2, SELECT2_RESULTSET));
     mockSpanner.putStatementResult(StatementResult.update(UPDATE_STATEMENT, UPDATE_COUNT));
     mockSpanner.putStatementResult(StatementResult.update(INSERT_STATEMENT, INSERT_COUNT));
+    mockSpanner.putStatementResult(
+        MockSpannerServiceImpl.StatementResult.detectDialectResult(Dialect.POSTGRESQL));
 
     mockOperationsService = new MockOperationsServiceImpl();
     mockDatabaseAdmin = new MockDatabaseAdminServiceImpl(mockOperationsService);
@@ -218,6 +298,11 @@ abstract class AbstractMockServerTest {
   @AfterClass
   public static void stopMockSpannerAndPgAdapterServers() throws Exception {
     pgServer.stopServer();
+    while (pgServer.isAlive()) {
+      // TODO: Remove once the pgServer.stopServer() is blocking until it has actually stopped (or
+      // there is some other way to block until it has stopped).
+      Thread.sleep(1L);
+    }
     try {
       SpannerPool.closeSpannerPool();
     } catch (SpannerException e) {
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ArraysMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ArraysMockServerTest.java
index fe3d6d10..b00db613 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ArraysMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ArraysMockServerTest.java
@@ -28,6 +28,7 @@ import com.google.spanner.v1.ResultSetMetadata;
 import com.google.spanner.v1.StructType;
 import com.google.spanner.v1.StructType.Field;
 import com.google.spanner.v1.Type;
+import com.google.spanner.v1.TypeAnnotationCode;
 import com.google.spanner.v1.TypeCode;
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
@@ -48,6 +49,15 @@ import org.junit.runners.JUnit4;
 public class ArraysMockServerTest extends AbstractMockServerTest {
   private static com.google.spanner.v1.ResultSet createResultSet(
       String columnName, TypeCode arrayElementType, Iterable<Value> values) {
+    return createResultSet(
+        columnName, arrayElementType, TypeAnnotationCode.TYPE_ANNOTATION_CODE_UNSPECIFIED, values);
+  }
+
+  private static com.google.spanner.v1.ResultSet createResultSet(
+      String columnName,
+      TypeCode arrayElementType,
+      TypeAnnotationCode arrayElementAnnotationCode,
+      Iterable<Value> values) {
     com.google.spanner.v1.ResultSet.Builder builder = com.google.spanner.v1.ResultSet.newBuilder();
     builder.setMetadata(
         ResultSetMetadata.newBuilder()
@@ -60,7 +70,10 @@ public class ArraysMockServerTest extends AbstractMockServerTest {
                                 Type.newBuilder()
                                     .setCode(TypeCode.ARRAY)
                                     .setArrayElementType(
-                                        Type.newBuilder().setCode(arrayElementType).build())
+                                        Type.newBuilder()
+                                            .setCode(arrayElementType)
+                                            .setTypeAnnotation(arrayElementAnnotationCode)
+                                            .build())
                                     .build())
                             .build())
                     .build())
@@ -161,6 +174,7 @@ public class ArraysMockServerTest extends AbstractMockServerTest {
             createResultSet(
                 "NUMERIC_ARRAY",
                 TypeCode.NUMERIC,
+                TypeAnnotationCode.PG_NUMERIC,
                 ImmutableList.of(
                     Values.of("3.14"), Values.of("1000.0"), Values.of("-0.123456789")))));
 
@@ -278,14 +292,15 @@ public class ArraysMockServerTest extends AbstractMockServerTest {
       try (ResultSet resultSet = connection.createStatement().executeQuery(sql)) {
         assertTrue(resultSet.next());
         Array array = resultSet.getArray("TIMESTAMP_ARRAY");
-        assertArrayEquals(
+        Timestamp[] expected =
             new Timestamp[] {
               com.google.cloud.Timestamp.parseTimestamp("2022-02-14T11:47:10.123456700Z")
                   .toSqlTimestamp(),
               com.google.cloud.Timestamp.parseTimestamp("2000-02-29T00:00:01.00000100Z")
                   .toSqlTimestamp()
-            },
-            (Timestamp[]) array.getArray());
+            };
+        Timestamp[] actual = (Timestamp[]) array.getArray();
+        assertArrayEquals(expected, actual);
         assertFalse(resultSet.next());
       }
     }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java
index f2e627e8..ab0b572a 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java
@@ -14,6 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
@@ -27,7 +28,6 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.sql.Connection;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.junit.Assert;
@@ -70,8 +70,8 @@ public final class ControlMessageTest {
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(intermediateStatement.getResultType()).thenReturn(ResultType.UPDATE_COUNT);
     Mockito.when(intermediateStatement.getCommand()).thenReturn("INSERT");
-    Mockito.when(intermediateStatement.getUpdateCount()).thenReturn(1);
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(intermediateStatement.getUpdateCount()).thenReturn(1L);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
 
     JSONParser parser = new JSONParser();
     JSONObject commandMetadata = (JSONObject) parser.parse(EMPTY_COMMAND_JSON);
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ConverterTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ConverterTest.java
deleted file mode 100644
index 0ca289ec..00000000
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ConverterTest.java
+++ /dev/null
@@ -1,100 +0,0 @@
-// Copyright 2020 Google LLC
-//
-// 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 com.google.cloud.spanner.pgadapter;
-
-import com.google.cloud.spanner.pgadapter.metadata.SQLMetadata;
-import com.google.cloud.spanner.pgadapter.utils.Converter;
-import com.google.common.collect.ImmutableSetMultimap;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class ConverterTest {
-
-  @Test
-  public void testToJDBCParams() {
-    String sqlStatement =
-        "SELECT * FROM users WHERE name = $1 AND age > $2 AND age < $3"; // Simple case
-    String expectedResult = "SELECT * FROM users WHERE name = ? AND age > ? AND age < ?";
-    int parameterCount = 3;
-    SQLMetadata result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-    Assert.assertEquals(result.getParameterCount(), parameterCount);
-    Assert.assertEquals(
-        result.getParameterIndexToPositions(), ImmutableSetMultimap.of(0, 1, 1, 2, 2, 3));
-
-    sqlStatement = "SELECT * FROM users WHERE name = $1$ AND AGE = $1$";
-    expectedResult = "SELECT * FROM users WHERE name = ?$ AND AGE = ?$";
-    parameterCount = 1;
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-    Assert.assertEquals(result.getParameterCount(), parameterCount);
-    Assert.assertEquals(result.getParameterIndexToPositions(), ImmutableSetMultimap.of(0, 1, 0, 2));
-
-    sqlStatement =
-        "SELECT * FROM users WHERE data = \"this data should not be changed $123 $$\" AND name = $1"; // Text should not be modified
-    expectedResult =
-        "SELECT * FROM users WHERE data = \"this data should not be changed $123 $$\" AND name = ?";
-    parameterCount = 1;
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-    Assert.assertEquals(result.getParameterCount(), parameterCount);
-    Assert.assertEquals(result.getParameterIndexToPositions(), ImmutableSetMultimap.of(0, 1));
-
-    sqlStatement =
-        "SELECT * FROM users WHERE data = \'this data should not be changed $123 $$\' AND name = $1"; // idem
-    expectedResult =
-        "SELECT * FROM users WHERE data = \'this data should not be changed $123 $$\' AND name = ?";
-    parameterCount = 1;
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-    Assert.assertEquals(result.getParameterCount(), parameterCount);
-    Assert.assertEquals(result.getParameterIndexToPositions(), ImmutableSetMultimap.of(0, 1));
-
-    // Out of order and multiple instances of the same index.
-    sqlStatement = "SELECT * FROM users WHERE (name = $2 OR surname = $2) AND age = $1";
-    expectedResult = "SELECT * FROM users WHERE (name = ? OR surname = ?) AND age = ?";
-    parameterCount = 2;
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-    Assert.assertEquals(result.getParameterCount(), parameterCount);
-    Assert.assertEquals(
-        result.getParameterIndexToPositions(), ImmutableSetMultimap.of(0, 3, 1, 1, 1, 2));
-
-    // Some parameter indices missing, this expects 123 parameters and would use only the last one.
-    sqlStatement = "SELECT $123";
-    expectedResult = "SELECT ?";
-    parameterCount = 123;
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-    Assert.assertEquals(result.getParameterCount(), parameterCount);
-    Assert.assertEquals(result.getParameterIndexToPositions(), ImmutableSetMultimap.of(122, 1));
-
-    // When parameters are mixed with dollar-quoted string
-    sqlStatement = "$1$$?it$?s$$$2";
-    expectedResult = "?$$?it$?s$$?";
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-
-    sqlStatement = "$1$tag$?it$$?s$tag$$2";
-    expectedResult = "?$tag$?it$$?s$tag$?";
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-
-    sqlStatement = "$1$$?it\\'?s \n ?it\\'?s$$$2";
-    expectedResult = "?$$?it\\'?s \n ?it\\'?s$$?";
-    result = Converter.toJDBCParams(sqlStatement);
-    Assert.assertEquals(result.getSqlString(), expectedResult);
-  }
-}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
index 6b649e36..dc435007 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
@@ -54,9 +54,9 @@ public class ITJdbcTest implements IntegrationTest {
   private static ProxyServer server;
   private static Database database;
 
-  @Parameter public String preferredQueryMode;
+  @Parameter public String preferQueryMode;
 
-  @Parameters(name = "preferredQueryMode = {0}")
+  @Parameters(name = "preferQueryMode = {0}")
   public static Object[] data() {
     return new Object[] {"extended", "simple"};
   }
@@ -148,8 +148,8 @@ public class ITJdbcTest implements IntegrationTest {
 
   private String getConnectionUrl() {
     return String.format(
-        "jdbc:postgresql://localhost:%d/?preferredQueryMode=%s",
-        server.getLocalPort(), preferredQueryMode);
+        "jdbc:postgresql://localhost:%d/?preferQueryMode=%s",
+        server.getLocalPort(), preferQueryMode);
   }
 
   @Test
@@ -166,24 +166,29 @@ public class ITJdbcTest implements IntegrationTest {
 
   @Test
   public void testSelectWithParameters() throws SQLException {
+    boolean isSimpleMode = "simple".equalsIgnoreCase(preferQueryMode);
+    String sql =
+        "select col_bigint, col_bool, col_bytea, col_float8, col_int, col_numeric, col_timestamptz, col_varchar "
+            + "from all_types "
+            + "where col_bigint=? "
+            + "and col_bool=? "
+            // The PG JDBC driver does not support bytea parameters in simple mode.
+            + (isSimpleMode ? "" : "and col_bytea=? ")
+            + "and col_float8=? "
+            + "and col_int=? "
+            + "and col_numeric=? "
+            + "and col_timestamptz=? "
+            + "and col_varchar=?";
+
     try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
-      try (PreparedStatement statement =
-          connection.prepareStatement(
-              "select col_bigint, col_bool, col_bytea, col_float8, col_int, col_numeric, col_timestamptz, col_varchar "
-                  + "from all_types "
-                  + "where col_bigint=? "
-                  + "and col_bool=? "
-                  + "and col_bytea=? "
-                  + "and col_float8=? "
-                  + "and col_int=? "
-                  + "and col_numeric=? "
-                  + "and col_timestamptz=? "
-                  + "and col_varchar=?")) {
+      try (PreparedStatement statement = connection.prepareStatement(sql)) {
 
         int index = 0;
         statement.setLong(++index, 1);
         statement.setBoolean(++index, true);
-        statement.setBytes(++index, "test".getBytes(StandardCharsets.UTF_8));
+        if (!isSimpleMode) {
+          statement.setBytes(++index, "test".getBytes(StandardCharsets.UTF_8));
+        }
         statement.setDouble(++index, 3.14d);
         statement.setInt(++index, 1);
         statement.setBigDecimal(++index, new BigDecimal("3.14"));
@@ -196,7 +201,9 @@ public class ITJdbcTest implements IntegrationTest {
 
           assertEquals(1, resultSet.getLong(1));
           assertTrue(resultSet.getBoolean(2));
-          assertArrayEquals("test".getBytes(StandardCharsets.UTF_8), resultSet.getBytes(3));
+          if (!isSimpleMode) {
+            assertArrayEquals("test".getBytes(StandardCharsets.UTF_8), resultSet.getBytes(3));
+          }
           assertEquals(3.14d, resultSet.getDouble(4), 0.0d);
           assertEquals(1, resultSet.getInt(5));
           assertEquals(new BigDecimal("3.14"), resultSet.getBigDecimal(6));
@@ -213,6 +220,7 @@ public class ITJdbcTest implements IntegrationTest {
 
   @Test
   public void testInsertWithParameters() throws SQLException {
+    boolean isSimpleMode = "simple".equalsIgnoreCase(preferQueryMode);
     try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
       try (PreparedStatement statement =
           connection.prepareStatement(
@@ -222,7 +230,12 @@ public class ITJdbcTest implements IntegrationTest {
         int index = 0;
         statement.setLong(++index, 2);
         statement.setBoolean(++index, true);
-        statement.setBytes(++index, "bytes_test".getBytes(StandardCharsets.UTF_8));
+        // The PG JDBC driver does not support bytea parameters in simple mode.
+        if (isSimpleMode) {
+          statement.setNull(++index, Types.BINARY);
+        } else {
+          statement.setBytes(++index, "bytes_test".getBytes(StandardCharsets.UTF_8));
+        }
         statement.setDouble(++index, 10.1);
         statement.setInt(++index, 100);
         statement.setBigDecimal(++index, new BigDecimal("6.626"));
@@ -239,7 +252,9 @@ public class ITJdbcTest implements IntegrationTest {
 
         assertEquals(2, resultSet.getLong(1));
         assertTrue(resultSet.getBoolean(2));
-        assertArrayEquals("bytes_test".getBytes(StandardCharsets.UTF_8), resultSet.getBytes(3));
+        if (!isSimpleMode) {
+          assertArrayEquals("bytes_test".getBytes(StandardCharsets.UTF_8), resultSet.getBytes(3));
+        }
         assertEquals(10.1d, resultSet.getDouble(4), 0.0d);
         assertEquals(100, resultSet.getInt(5));
         assertEquals(new BigDecimal("6.626"), resultSet.getBigDecimal(6));
@@ -255,21 +270,26 @@ public class ITJdbcTest implements IntegrationTest {
 
   @Test
   public void testUpdateWithParameters() throws SQLException {
+    boolean isSimpleMode = "simple".equalsIgnoreCase(preferQueryMode);
+    String sql =
+        "update all_types set "
+            + "col_bool=?, "
+            // The PG JDBC driver does not support bytea parameters in simple mode.
+            + (isSimpleMode ? "" : "col_bytea=?, ")
+            + "col_float8=?, "
+            + "col_int=?, "
+            + "col_numeric=?, "
+            + "col_timestamptz=?, "
+            + "col_varchar=? "
+            + "where col_bigint=?";
+
     try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
-      try (PreparedStatement statement =
-          connection.prepareStatement(
-              "update all_types set "
-                  + "col_bool=?, "
-                  + "col_bytea=?, "
-                  + "col_float8=?, "
-                  + "col_int=?, "
-                  + "col_numeric=?, "
-                  + "col_timestamptz=?, "
-                  + "col_varchar=? "
-                  + "where col_bigint=?")) {
+      try (PreparedStatement statement = connection.prepareStatement(sql)) {
         int index = 0;
         statement.setBoolean(++index, false);
-        statement.setBytes(++index, "updated".getBytes(StandardCharsets.UTF_8));
+        if (!isSimpleMode) {
+          statement.setBytes(++index, "updated".getBytes(StandardCharsets.UTF_8));
+        }
         statement.setDouble(++index, 3.14d * 2d);
         statement.setInt(++index, 2);
         statement.setBigDecimal(++index, new BigDecimal("10.0"));
@@ -290,7 +310,9 @@ public class ITJdbcTest implements IntegrationTest {
 
         assertEquals(1, resultSet.getLong(1));
         assertFalse(resultSet.getBoolean(2));
-        assertArrayEquals("updated".getBytes(StandardCharsets.UTF_8), resultSet.getBytes(3));
+        if (!isSimpleMode) {
+          assertArrayEquals("updated".getBytes(StandardCharsets.UTF_8), resultSet.getBytes(3));
+        }
         assertEquals(3.14d * 2d, resultSet.getDouble(4), 0.0d);
         assertEquals(2, resultSet.getInt(5));
         assertEquals(new BigDecimal("10.0"), resultSet.getBigDecimal(6));
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITQueryTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITQueryTest.java
index 0301b1b9..82d3a3d3 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITQueryTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITQueryTest.java
@@ -189,7 +189,7 @@ public final class ITQueryTest implements IntegrationTest {
     // Check exact message results.
     byte[] rowDescriptionData = {
       0, 4, 105, 100, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 20, 0, 8, 0, 0, 0, 0, 0, 0, 110, 97, 109, 101,
-      0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, -1, -1, 0, 0, 0, 0, 0, 0, 97, 103, 101, 0, 0, 0, 0, 0, 0, 0,
+      0, 0, 0, 0, 0, 0, 0, 0, 0, 4, 19, -1, -1, 0, 0, 0, 0, 0, 0, 97, 103, 101, 0, 0, 0, 0, 0, 0, 0,
       0, 0, 0, 20, 0, 8, 0, 0, 0, 0, 0, 0, 100, 97, 116, 97, 0, 0, 0, 0, 0, 0, 0, 0, 0, 6, -92, -1,
       -1, 0, 0, 0, 0, 0, 0
     };
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
index e82913b5..fd7f0a8d 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
@@ -16,30 +16,46 @@ package com.google.cloud.spanner.pgadapter;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
+import com.google.cloud.ByteArray;
+import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
+import com.google.cloud.spanner.Statement;
 import com.google.protobuf.ListValue;
 import com.google.protobuf.Value;
 import com.google.spanner.v1.CommitRequest;
 import com.google.spanner.v1.ExecuteBatchDmlRequest;
 import com.google.spanner.v1.ExecuteSqlRequest;
+import com.google.spanner.v1.ExecuteSqlRequest.QueryMode;
 import com.google.spanner.v1.Mutation;
 import com.google.spanner.v1.Mutation.OperationCase;
 import com.google.spanner.v1.ResultSetMetadata;
 import com.google.spanner.v1.StructType;
 import com.google.spanner.v1.StructType.Field;
 import com.google.spanner.v1.Type;
+import com.google.spanner.v1.TypeAnnotationCode;
 import com.google.spanner.v1.TypeCode;
 import java.io.File;
 import java.io.IOException;
 import java.io.StringReader;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
 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.Instant;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.Base64;
 import java.util.List;
+import java.util.Map;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -66,13 +82,218 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
 
   @Test
   public void testQuery() throws SQLException {
+    String sql = "SELECT 1";
+
     try (Connection connection = DriverManager.getConnection(createUrl())) {
-      try (ResultSet resultSet = connection.createStatement().executeQuery("SELECT 1")) {
+      try (ResultSet resultSet = connection.createStatement().executeQuery(sql)) {
         assertTrue(resultSet.next());
         assertEquals(1L, resultSet.getLong(1));
         assertFalse(resultSet.next());
       }
     }
+
+    // The statement is sent twice to the mock server:
+    // 1. The first time it is sent with the PLAN mode enabled.
+    // 2. The second time it is sent in normal execute mode.
+    // TODO: Consider skipping the PLAN step and always execute the query already when we receive a
+    // DESCRIBE portal message.
+    assertEquals(2, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    ExecuteSqlRequest planRequest = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0);
+    assertEquals(QueryMode.PLAN, planRequest.getQueryMode());
+    ExecuteSqlRequest executeRequest =
+        mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(1);
+    assertEquals(QueryMode.NORMAL, executeRequest.getQueryMode());
+
+    for (ExecuteSqlRequest request : mockSpanner.getRequestsOfType(ExecuteSqlRequest.class)) {
+      assertEquals(sql, request.getSql());
+      assertTrue(request.getTransaction().hasSingleUse());
+      assertTrue(request.getTransaction().getSingleUse().hasReadOnly());
+    }
+  }
+
+  @Test
+  public void testQueryWithParameters() throws SQLException {
+    String jdbcSql =
+        "select col_bigint, col_bool, col_bytea, col_float8, col_numeric, col_timestamptz, col_varchar "
+            + "from all_types "
+            + "where col_bigint=? "
+            + "and col_bool=? "
+            + "and col_bytea=? "
+            + "and col_float8=? "
+            + "and col_numeric=? "
+            + "and col_timestamptz=? "
+            + "and col_varchar=?";
+    String pgSql =
+        "select col_bigint, col_bool, col_bytea, col_float8, col_numeric, col_timestamptz, col_varchar "
+            + "from all_types "
+            + "where col_bigint=$1 "
+            + "and col_bool=$2 "
+            + "and col_bytea=$3 "
+            + "and col_float8=$4 "
+            + "and col_numeric=$5 "
+            + "and col_timestamptz=$6 "
+            + "and col_varchar=$7";
+    mockSpanner.putStatementResult(StatementResult.query(Statement.of(pgSql), ALL_TYPES_RESULTSET));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(pgSql)
+                .bind("p1")
+                .to(1L)
+                .bind("p2")
+                .to(true)
+                .bind("p3")
+                .to(ByteArray.copyFrom("test"))
+                .bind("p4")
+                .to(3.14d)
+                .bind("p5")
+                .to(com.google.cloud.spanner.Value.pgNumeric("6.626"))
+                .bind("p6")
+                .to(Timestamp.parseTimestamp("2022-02-16T13:18:02.123457000Z"))
+                .bind("p7")
+                .to("test")
+                .build(),
+            ALL_TYPES_RESULTSET));
+
+    OffsetDateTime zonedDateTime =
+        LocalDateTime.of(2022, 2, 16, 13, 18, 2, 123456789).atOffset(ZoneOffset.UTC);
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (PreparedStatement preparedStatement = connection.prepareStatement(jdbcSql)) {
+        preparedStatement.setLong(1, 1L);
+        preparedStatement.setBoolean(2, true);
+        preparedStatement.setBytes(3, "test".getBytes(StandardCharsets.UTF_8));
+        preparedStatement.setDouble(4, 3.14d);
+        preparedStatement.setBigDecimal(5, new BigDecimal("6.626"));
+        preparedStatement.setTimestamp(6, java.sql.Timestamp.from(Instant.from(zonedDateTime)));
+        preparedStatement.setString(7, "test");
+        try (ResultSet resultSet = preparedStatement.executeQuery()) {
+          assertTrue(resultSet.next());
+          assertEquals(1L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(2, requests.size());
+
+    ExecuteSqlRequest planRequest = requests.get(0);
+    assertEquals(QueryMode.PLAN, planRequest.getQueryMode());
+    assertEquals(pgSql, planRequest.getSql());
+
+    ExecuteSqlRequest executeRequest = requests.get(1);
+    assertEquals(QueryMode.NORMAL, executeRequest.getQueryMode());
+    assertEquals(pgSql, executeRequest.getSql());
+
+    for (ExecuteSqlRequest request : requests) {
+      Map<String, Value> params = request.getParams().getFieldsMap();
+      Map<String, Type> types = request.getParamTypesMap();
+
+      assertEquals(TypeCode.INT64, types.get("p1").getCode());
+      assertEquals("1", params.get("p1").getStringValue());
+      assertEquals(TypeCode.BOOL, types.get("p2").getCode());
+      assertTrue(params.get("p2").getBoolValue());
+      assertEquals(TypeCode.BYTES, types.get("p3").getCode());
+      assertEquals(
+          Base64.getEncoder().encodeToString("test".getBytes(StandardCharsets.UTF_8)),
+          params.get("p3").getStringValue());
+      assertEquals(TypeCode.FLOAT64, types.get("p4").getCode());
+      assertEquals(3.14d, params.get("p4").getNumberValue(), 0.0d);
+      assertEquals(TypeCode.NUMERIC, types.get("p5").getCode());
+      assertEquals(TypeAnnotationCode.PG_NUMERIC, types.get("p5").getTypeAnnotation());
+      assertEquals("6.626", params.get("p5").getStringValue());
+      assertEquals(TypeCode.TIMESTAMP, types.get("p6").getCode());
+      assertEquals("2022-02-16T13:18:02.123457000Z", params.get("p6").getStringValue());
+      assertEquals(TypeCode.STRING, types.get("p7").getCode());
+      assertEquals("test", params.get("p7").getStringValue());
+    }
+  }
+
+  @Test
+  public void testNullValues() throws SQLException {
+    mockSpanner.putStatementResult(
+        StatementResult.update(
+            Statement.newBuilder(
+                    "insert into all_types "
+                        + "(col_bigint, col_bool, col_bytea, col_float8, col_int, col_numeric, col_timestamptz, col_varchar) "
+                        + "values ($1, $2, $3, $4, $5, $6, $7, $8)")
+                .bind("p1")
+                .to(2L)
+                .bind("p2")
+                .to((Boolean) null)
+                .bind("p3")
+                .to((ByteArray) null)
+                .bind("p4")
+                .to((Double) null)
+                .bind("p5")
+                .to((Long) null)
+                .bind("p6")
+                .to(com.google.cloud.spanner.Value.pgNumeric(null))
+                .bind("p7")
+                .to((com.google.cloud.spanner.Value) null)
+                .bind("p8")
+                .to((String) null)
+                .build(),
+            1L));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from all_types where col_bigint is null"),
+            ALL_TYPES_NULLS_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      // TODO: Add col_timestamptz to the statement when PgAdapter allows untyped null values.
+      // The PG JDBC driver sends date and timestamp parameters with type code Oid.UNSPECIFIED.
+      // TODO: Add col_numeric to the statement when the mock serverr supports numeric parameters.
+      try (PreparedStatement statement =
+          connection.prepareStatement(
+              "insert into all_types "
+                  + "(col_bigint, col_bool, col_bytea, col_float8, col_int, col_numeric, col_timestamptz, col_varchar) "
+                  + "values (?, ?, ?, ?, ?, ?, ?, ?)")) {
+        int index = 0;
+        statement.setLong(++index, 2);
+        statement.setNull(++index, Types.BOOLEAN);
+        statement.setNull(++index, Types.BINARY);
+        statement.setNull(++index, Types.DOUBLE);
+        statement.setNull(++index, Types.INTEGER);
+        // TODO: Enable the next line when the mock server supports numeric parameter values.
+        statement.setNull(++index, Types.NUMERIC);
+        // TODO: Enable the next line when PgAdapter allows untyped null values.
+        // This is currently blocked on both the Spangres backend allowing untyped null values in
+        // SQL statements, as well as the Java client library and/or JDBC driver allowing untyped
+        // null values.
+        // See also https://github.com/googleapis/java-spanner/pull/1680
+        statement.setNull(++index, Types.TIMESTAMP_WITH_TIMEZONE);
+        statement.setNull(++index, Types.VARCHAR);
+
+        assertEquals(1, statement.executeUpdate());
+      }
+
+      try (ResultSet resultSet =
+          connection
+              .createStatement()
+              .executeQuery("select * from all_types where col_bigint is null")) {
+        assertTrue(resultSet.next());
+
+        int index = 0;
+        // Note: JDBC returns the zero-value for primitive types if the value is NULL, and you have
+        // to call wasNull() to determine whether the value was NULL or zero.
+        assertEquals(0L, resultSet.getLong(++index));
+        assertTrue(resultSet.wasNull());
+        assertFalse(resultSet.getBoolean(++index));
+        assertTrue(resultSet.wasNull());
+        assertNull(resultSet.getBytes(++index));
+        assertTrue(resultSet.wasNull());
+        assertEquals(0d, resultSet.getDouble(++index), 0.0d);
+        assertTrue(resultSet.wasNull());
+        assertNull(resultSet.getBigDecimal(++index));
+        assertTrue(resultSet.wasNull());
+        assertNull(resultSet.getTimestamp(++index));
+        assertTrue(resultSet.wasNull());
+        assertNull(resultSet.getString(++index));
+        assertTrue(resultSet.wasNull());
+
+        assertFalse(resultSet.next());
+      }
+    }
   }
 
   @Test
@@ -157,7 +378,7 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     mockSpanner.putStatementResult(
         StatementResult.query(
             com.google.cloud.spanner.Statement.newBuilder(
-                    "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = @p1")
+                    "SELECT column_name, data_type FROM information_schema.columns WHERE table_name = $1")
                 .bind("p1")
                 .to("users")
                 .build(),
@@ -198,11 +419,9 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     // PgAdapter therefore also simply executes these as separate DML statements.
     assertTrue(mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).isEmpty());
     List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-    // We get three requests: One keep-alive and the two DML statements.
-    assertEquals(3, requests.size());
-    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
-    assertEquals(INSERT_STATEMENT.getSql(), requests.get(1).getSql());
-    assertEquals(UPDATE_STATEMENT.getSql(), requests.get(2).getSql());
+    assertEquals(2, requests.size());
+    assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
+    assertEquals(UPDATE_STATEMENT.getSql(), requests.get(1).getSql());
   }
 
   @Test
@@ -225,12 +444,9 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     assertTrue(mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).isEmpty());
     List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
 
-    // The server will also receive a 'SELECT 1' statement from PgAdapter, as PgAdapter calls
-    // connection.isAlive() before using it.
-    assertEquals(3, requests.size());
-    assertEquals(SELECT1.getSql(), requests.get(0).getSql());
-    assertEquals(INSERT_STATEMENT.getSql(), requests.get(1).getSql());
-    assertEquals(UPDATE_STATEMENT.getSql(), requests.get(2).getSql());
+    assertEquals(2, requests.size());
+    assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
+    assertEquals(UPDATE_STATEMENT.getSql(), requests.get(1).getSql());
   }
 
   @Test
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
new file mode 100644
index 00000000..14015de6
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
@@ -0,0 +1,216 @@
+// Copyright 2022 Google LLC
+//
+// 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 com.google.cloud.spanner.pgadapter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
+import com.google.spanner.v1.ExecuteSqlRequest;
+import com.google.spanner.v1.ExecuteSqlRequest.QueryMode;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
+import java.time.ZoneOffset;
+import java.util.TimeZone;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.postgresql.jdbc.TimestampUtils;
+
+/**
+ * Tests the native PG JDBC driver in simple query mode. This is similar to the protocol that is
+ * used by psql, and for example allows batches to be given as semicolon-separated strings.
+ */
+@RunWith(JUnit4.class)
+public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
+  @BeforeClass
+  public static void loadPgJdbcDriver() throws Exception {
+    // Make sure the PG JDBC driver is loaded.
+    Class.forName("org.postgresql.Driver");
+  }
+
+  /**
+   * Creates a JDBC connection string that instructs the PG JDBC driver to use the default simple
+   * mode for queries and DML statements. This makes the JDBC driver behave in (much) the same way
+   * as psql.
+   */
+  private String createUrl() {
+    return String.format(
+        "jdbc:postgresql://localhost:%d/?preferQueryMode=simple", pgServer.getLocalPort());
+  }
+
+  @Test
+  public void testQuery() throws SQLException {
+    String sql = "SELECT 1";
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (ResultSet resultSet = connection.createStatement().executeQuery(sql)) {
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+    }
+
+    // The statement is sent only once to the mock server in simple query mode.
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    ExecuteSqlRequest request = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0);
+    assertEquals(QueryMode.NORMAL, request.getQueryMode());
+    assertEquals(sql, request.getSql());
+    assertTrue(request.getTransaction().hasSingleUse());
+    assertTrue(request.getTransaction().getSingleUse().hasReadOnly());
+  }
+
+  @Test
+  public void testQueryWithParameters() throws SQLException {
+    // Query parameters are not supported by the PG wire protocol in the simple query mode. The JDBC
+    // driver will therefore convert parameters to literals before sending them to PostgreSQL.
+    // The bytea data type is not supported for that (by the PG JDBC driver).
+    // Also, the JDBC driver always uses the default timezone of the JVM when setting a timestamp.
+    // This is a requirement in the JDBC API (and one that causes about a trillion confusions per
+    // year). So we need to extract that from the env in order to determine what the timestamp
+    // string will be.
+    OffsetDateTime zonedDateTime =
+        LocalDateTime.of(2022, 2, 16, 13, 18, 2, 123456789).atOffset(ZoneOffset.UTC);
+    String timestampString =
+        new TimestampUtils(false, TimeZone::getDefault)
+            .timeToString(java.sql.Timestamp.from(Instant.from(zonedDateTime)), true);
+
+    String pgSql =
+        "select col_bigint, col_bool, col_bytea, col_float8, col_numeric, col_timestamptz, col_varchar "
+            + "from all_types "
+            + "where col_bigint=1 "
+            + "and col_bool='TRUE' "
+            + "and col_float8=3.14 "
+            + "and col_numeric=6.626 "
+            + String.format("and col_timestamptz='%s' ", timestampString)
+            + "and col_varchar='test'";
+    String jdbcSql =
+        "select col_bigint, col_bool, col_bytea, col_float8, col_numeric, col_timestamptz, col_varchar "
+            + "from all_types "
+            + "where col_bigint=? "
+            + "and col_bool=? "
+            + "and col_float8=? "
+            + "and col_numeric=? "
+            + "and col_timestamptz=? "
+            + "and col_varchar=?";
+    mockSpanner.putStatementResult(
+        StatementResult.query(com.google.cloud.spanner.Statement.of(pgSql), ALL_TYPES_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (PreparedStatement preparedStatement = connection.prepareStatement(jdbcSql)) {
+        int index = 0;
+        preparedStatement.setLong(++index, 1L);
+        preparedStatement.setBoolean(++index, true);
+        preparedStatement.setDouble(++index, 3.14d);
+        preparedStatement.setBigDecimal(++index, new BigDecimal("6.626"));
+        preparedStatement.setTimestamp(
+            ++index, java.sql.Timestamp.from(Instant.from(zonedDateTime)));
+        preparedStatement.setString(++index, "test");
+        try (ResultSet resultSet = preparedStatement.executeQuery()) {
+          assertTrue(resultSet.next());
+          assertEquals(1L, resultSet.getLong(1));
+          assertFalse(resultSet.next());
+        }
+      }
+    }
+
+    // The statement is sent only once to the mock server in simple query mode.
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    ExecuteSqlRequest request = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0);
+    assertEquals(QueryMode.NORMAL, request.getQueryMode());
+    assertEquals(pgSql, request.getSql());
+    assertTrue(request.getTransaction().hasSingleUse());
+    assertTrue(request.getTransaction().getSingleUse().hasReadOnly());
+  }
+
+  @Test
+  public void testMixedBatch() throws SQLException {
+    // TODO: Change this test case once we support mixed batches.
+    String sql =
+        "CREATE TABLE testtable (id integer PRIMARY KEY, data integer); INSERT INTO testtable (id, data) VALUES (1, 2); INSERT INTO testtable (id, data) VALUES (3, 4);";
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertEquals(
+            String.format(
+                "ERROR: FAILED_PRECONDITION: Executing updates is not allowed for DDL batches. \"%s\"",
+                sql),
+            exception.getMessage());
+      }
+    }
+  }
+
+  @Test
+  public void testSelectInDdlBatch() throws SQLException {
+    // TODO: Change this test case once we support mixed batches.
+    String sql =
+        "CREATE TABLE test (id integer PRIMARY KEY); SELECT * FROM test; SELECT * FROM test;";
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertEquals(
+            String.format(
+                "ERROR: FAILED_PRECONDITION: Executing queries is not allowed for DDL batches. \"%s\"",
+                sql),
+            exception.getMessage());
+      }
+    }
+  }
+
+  @Test
+  public void testTransactionStatementsInBatch() throws SQLException {
+    // TODO: Change this test case once we support mixed batches.
+    String sql =
+        "BEGIN TRANSACTION; INSERT INTO users (id, age, name) VALUES (99, 99, 'test'); COMMIT;";
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertEquals(
+            String.format(
+                "ERROR: FAILED_PRECONDITION: This connection has an active batch and cannot begin a transaction \"%s\"",
+                sql),
+            exception.getMessage());
+      }
+    }
+  }
+
+  @Test
+  public void testSelectAtStartOfBatch() throws SQLException {
+    // TODO: Change this test case once we support mixed batches.
+    String sql =
+        "SELECT * FROM users; INSERT INTO users (id, age, name) VALUES (99, 99, 'person'); SELECT * FROM users;";
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertEquals(
+            String.format(
+                "ERROR: INVALID_ARGUMENT: Statement type is not supported for batching \"%s\"",
+                sql),
+            exception.getMessage());
+      }
+    }
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/PSQLTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/PSQLTest.java
index 28153443..9ea7170b 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/PSQLTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/PSQLTest.java
@@ -14,12 +14,10 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.pgadapter.metadata.CommandMetadataParser;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.statements.MatcherStatement;
-import java.sql.Connection;
-import java.sql.SQLException;
-import java.sql.Statement;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.junit.Assert;
@@ -46,20 +44,17 @@ public class PSQLTest {
 
   @Mock private OptionsMetadata options;
 
-  @Mock private Statement statement;
-
   @Before
   public void setup() throws Exception {
     final JSONObject defaultCommands = new CommandMetadataParser().defaultCommands();
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getServer()).thenReturn(server);
     Mockito.when(server.getOptions()).thenReturn(options);
     Mockito.when(options.getCommandMetadataJSON()).thenReturn(defaultCommands);
-    Mockito.when(connection.createStatement()).thenReturn(statement);
   }
 
   @Test
-  public void testDescribeTranslates() throws SQLException {
+  public void testDescribeTranslates() {
     // PSQL equivalent: \d
     String sql =
         "SELECT n.nspname as \"Schema\",\n"
@@ -94,7 +89,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableMatchTranslates() throws SQLException {
+  public void testDescribeTableMatchTranslates() {
     // PSQL equivalent: \d <table> (1)
     String sql =
         "SELECT c.oid,\n"
@@ -123,7 +118,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableMatchHandlesBobbyTables() throws SQLException {
+  public void testDescribeTableMatchHandlesBobbyTables() {
     // PSQL equivalent: \d <table> (1)
     String sql =
         "SELECT c.oid,\n"
@@ -152,7 +147,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableCatalogTranslates() throws SQLException {
+  public void testDescribeTableCatalogTranslates() {
     // PSQL equivalent: \d <table> (2)
     String sql =
         "SELECT relchecks, relkind, relhasindex, relhasrules, reltriggers <> 0, false, false,"
@@ -177,7 +172,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableMetadataTranslates() throws SQLException {
+  public void testDescribeTableMetadataTranslates() {
     // PSQL equivalent: \d <table> (3)
     String sql =
         "SELECT a.attname,\n"
@@ -213,7 +208,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableMetadataHandlesBobbyTables() throws SQLException {
+  public void testDescribeTableMetadataHandlesBobbyTables() {
     // PSQL equivalent: \d <table> (3)
     String sql =
         "SELECT a.attname,\n"
@@ -250,7 +245,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableAttributesTranslates() throws SQLException {
+  public void testDescribeTableAttributesTranslates() {
     // PSQL equivalent: \d <table> (4)
     String sql =
         "SELECT c.oid::pg_catalog.regclass FROM pg_catalog.pg_class c, pg_catalog.pg_inherits i"
@@ -264,7 +259,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeMoreTableAttributesTranslates() throws SQLException {
+  public void testDescribeMoreTableAttributesTranslates() {
     // PSQL equivalent: \d <table> (5)
     String sql =
         "SELECT c.oid::pg_catalog.regclass FROM pg_catalog.pg_class c, pg_catalog.pg_inherits i"
@@ -278,7 +273,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testListCommandTranslates() throws SQLException {
+  public void testListCommandTranslates() {
     // PSQL equivalent: \l
     String sql =
         "SELECT d.datname as \"Name\",\n"
@@ -287,9 +282,7 @@ public class PSQLTest {
             + "       pg_catalog.array_to_string(d.datacl, '\\n') AS \"Access privileges\"\n"
             + "FROM pg_catalog.pg_database d\n"
             + "ORDER BY 1;";
-    String result = "SELECT 'users' AS Name;";
-
-    Mockito.when(connection.getCatalog()).thenReturn("users");
+    String result = "SELECT '' AS Name;";
 
     MatcherStatement matcherStatement = new MatcherStatement(sql, connectionHandler);
 
@@ -297,7 +290,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testListDatabaseCommandTranslates() throws SQLException {
+  public void testListDatabaseCommandTranslates() {
     // PSQL equivalent: \l <table>
     String sql =
         "SELECT d.datname as \"Name\",\n"
@@ -307,9 +300,7 @@ public class PSQLTest {
             + "FROM pg_catalog.pg_database d\n"
             + "WHERE d.datname OPERATOR(pg_catalog.~) '^(users)$'\n"
             + "ORDER BY 1;";
-    String result = "SELECT 'users' AS Name;";
-
-    Mockito.when(connection.getCatalog()).thenReturn("users");
+    String result = "SELECT '' AS Name;";
 
     MatcherStatement matcherStatement = new MatcherStatement(sql, connectionHandler);
 
@@ -317,7 +308,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testListDatabaseCommandFailsButPrintsUnknown() throws SQLException {
+  public void testListDatabaseCommandFailsButPrintsUnknown() {
     // PSQL equivalent: \l <table>
     String sql =
         "SELECT d.datname as \"Name\",\n"
@@ -327,17 +318,17 @@ public class PSQLTest {
             + "FROM pg_catalog.pg_database d\n"
             + "WHERE d.datname OPERATOR(pg_catalog.~) '^(users)$'\n"
             + "ORDER BY 1;";
-    String result = "SELECT 'UNKNOWN' AS Name;";
-
-    Mockito.when(connection.getCatalog()).thenThrow(SQLException.class);
+    String result = "SELECT '' AS Name;";
 
+    // TODO: Add Connection#getDatabase() to Connection API and test here what happens if that
+    // method throws an exception.
     MatcherStatement matcherStatement = new MatcherStatement(sql, connectionHandler);
 
     Assert.assertEquals(matcherStatement.getSql(), result);
   }
 
   @Test
-  public void testDescribeAllTableMetadataTranslates() throws SQLException {
+  public void testDescribeAllTableMetadataTranslates() {
     // PSQL equivalent: \dt
     String sql =
         "SELECT n.nspname as \"Schema\",\n"
@@ -363,7 +354,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeSelectedTableMetadataTranslates() throws SQLException {
+  public void testDescribeSelectedTableMetadataTranslates() {
     // PSQL equivalent: \dt <table>
     String sql =
         "SELECT n.nspname as \"Schema\",\n"
@@ -389,7 +380,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeSelectedTableMetadataHandlesBobbyTables() throws SQLException {
+  public void testDescribeSelectedTableMetadataHandlesBobbyTables() {
     // PSQL equivalent: \dt <table>
     String sql =
         "SELECT n.nspname as \"Schema\",\n"
@@ -416,7 +407,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeAllIndexMetadataTranslates() throws SQLException {
+  public void testDescribeAllIndexMetadataTranslates() {
     // PSQL equivalent: \di
     String sql =
         "SELECT n.nspname as \"Schema\",\n"
@@ -446,7 +437,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeSelectedIndexMetadataTranslates() throws SQLException {
+  public void testDescribeSelectedIndexMetadataTranslates() {
     // PSQL equivalent: \di <index>
     String sql =
         "SELECT n.nspname as \"Schema\",\n"
@@ -476,7 +467,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeSelectedIndexMetadataHandlesBobbyTables() throws SQLException {
+  public void testDescribeSelectedIndexMetadataHandlesBobbyTables() {
     // PSQL equivalent: \di <index>
     String sql =
         "SELECT n.nspname as \"Schema\",\n"
@@ -506,7 +497,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeAllSchemaMetadataTranslates() throws SQLException {
+  public void testDescribeAllSchemaMetadataTranslates() {
     // PSQL equivalent: \dn
     String sql =
         "SELECT n.nspname AS \"Name\",\n"
@@ -522,7 +513,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeSelectedSchemaMetadataTranslates() throws SQLException {
+  public void testDescribeSelectedSchemaMetadataTranslates() {
     // PSQL equivalent: \dn <schema>
     String sql =
         "SELECT n.nspname AS \"Name\",\n"
@@ -539,7 +530,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeSelectedSchemaMetadataHandlesBobbyTables() throws SQLException {
+  public void testDescribeSelectedSchemaMetadataHandlesBobbyTables() {
     // PSQL equivalent: \dn <schema>
     String sql =
         "SELECT n.nspname AS \"Name\",\n"
@@ -557,7 +548,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testTableSelectAutocomplete() throws SQLException {
+  public void testTableSelectAutocomplete() {
     // PSQL equivalent: SELECT <table>
     String sql =
         "SELECT pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c WHERE c.relkind IN"
@@ -594,7 +585,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testTableInsertAutocomplete() throws SQLException {
+  public void testTableInsertAutocomplete() {
     // PSQL equivalent: INSERT INTO <table>
     String sql =
         "SELECT pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c WHERE c.relkind IN"
@@ -630,7 +621,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testTableAttributesAutocomplete() throws SQLException {
+  public void testTableAttributesAutocomplete() {
     // PSQL equivalent: INSERT INTO table_name (<attribute>)
     // PSQL equivalent: SELECT * FROM table_name WHERE <attribute>
     String sql =
@@ -650,7 +641,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableAutocomplete() throws SQLException {
+  public void testDescribeTableAutocomplete() {
     // PSQL equivalent: \\d <table>
     String sql =
         "SELECT pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c WHERE"
@@ -684,7 +675,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeTableMetadataAutocomplete() throws SQLException {
+  public void testDescribeTableMetadataAutocomplete() {
     // PSQL equivalent: \\dt <table>
     String sql =
         "SELECT pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c WHERE c.relkind IN"
@@ -718,7 +709,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeIndexMetadataAutocomplete() throws SQLException {
+  public void testDescribeIndexMetadataAutocomplete() {
     // PSQL equivalent: \\di <index>
     String sql =
         "SELECT pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c WHERE c.relkind IN"
@@ -752,7 +743,7 @@ public class PSQLTest {
   }
 
   @Test
-  public void testDescribeSchemaMetadataAutocomplete() throws SQLException {
+  public void testDescribeSchemaMetadataAutocomplete() {
     // PSQL equivalent: \\dn <schema>
     String sql =
         "SELECT pg_catalog.quote_ident(nspname) FROM pg_catalog.pg_namespace  WHERE"
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ParserTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ParserTest.java
index ff3cd950..4f541ade 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ParserTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ParserTest.java
@@ -23,7 +23,14 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
-
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.ByteArray;
+import com.google.cloud.Date;
+import com.google.cloud.Timestamp;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Type;
+import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.pgadapter.ProxyServer.DataFormat;
 import com.google.cloud.spanner.pgadapter.parsers.ArrayParser;
 import com.google.cloud.spanner.pgadapter.parsers.BinaryParser;
@@ -45,12 +52,6 @@ import com.google.cloud.spanner.pgadapter.parsers.copy.TokenMgrError;
 import com.google.cloud.spanner.pgadapter.utils.StatementParser;
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
-import java.sql.Array;
-import java.sql.Date;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Timestamp;
-import java.sql.Types;
 import java.util.Arrays;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -208,7 +209,7 @@ public class ParserTest {
 
   @Test
   public void testDateParsing() {
-    Date value = new Date(904910400000L); // Google founding date :)
+    Date value = Date.fromYearMonthDay(1998, 9, 4); // Google founding date :)
 
     byte[] byteResult = {-1, -1, -2, 28};
     byte[] stringResult = {'1', '9', '9', '8', '-', '0', '9', '-', '0', '4'};
@@ -258,7 +259,7 @@ public class ParserTest {
 
   @Test
   public void testTimestampParsingBytePart() {
-    Timestamp value = new Timestamp(904910400000L);
+    Timestamp value = Timestamp.ofTimeMicroseconds(904910400000000L);
 
     byte[] byteResult = {-1, -1, -38, 1, -93, -70, 48, 0};
 
@@ -298,7 +299,7 @@ public class ParserTest {
 
   @Test
   public void testBinaryParsing() {
-    byte[] value = {(byte) 0b01010101, (byte) 0b10101010};
+    ByteArray value = ByteArray.copyFrom(new byte[] {(byte) 0b01010101, (byte) 0b10101010});
     byte[] byteResult = {(byte) 0b01010101, (byte) 0b10101010};
     byte[] stringResult = {'U', '\\', '2', '5', '2'};
 
@@ -310,11 +311,18 @@ public class ParserTest {
   }
 
   @Test
-  public void testStringArrayParsing() throws SQLException {
+  public void testStringArrayParsing() {
     String[] value = {"abc", "def", "jhi"};
+    // The binary format of a PG array should contain the OID of the element type and not the array
+    // OID. For VARCHAR that means 1043 and not 1015
+    // See
+    // https://github.com/pgjdbc/pgjdbc/blob/60a81034fd003551fc863033f491b2d0ed1dfa80/pgjdbc/src/main/java/org/postgresql/jdbc/ArrayDecoding.java#L505
+    // We can test this more thoroughly when
+    // https://github.com/GoogleCloudPlatform/pgadapter/pull/36
+    // has been merged.
     byte[] byteResult = {
-      0, 0, 0, 1, 0, 0, 0, 1, -1, -1, -1, -9, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 3, 97, 98, 99, 0, 0,
-      0, 3, 100, 101, 102, 0, 0, 0, 3, 106, 104, 105
+      0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 4, 19, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 3, 97, 98, 99, 0, 0, 0,
+      3, 100, 101, 102, 0, 0, 0, 3, 106, 104, 105
     };
     byte[] stringResult = {
       '{', '"', 'a', 'b', 'c', '"', ',', '"', 'd', 'e', 'f', '"', ',', '"', 'j', 'h', 'i', '"', '}'
@@ -323,56 +331,53 @@ public class ParserTest {
       '[', '"', 'a', 'b', 'c', '"', ',', '"', 'd', 'e', 'f', '"', ',', '"', 'j', 'h', 'i', '"', ']'
     };
 
-    Array sqlArray = Mockito.mock(Array.class);
     ResultSet resultSet = Mockito.mock(ResultSet.class);
+    when(resultSet.getColumnType(0)).thenReturn(Type.array(Type.string()));
+    when(resultSet.getValue(0)).thenReturn(Value.stringArray(Arrays.asList(value)));
 
-    Mockito.when(sqlArray.getBaseType()).thenReturn(Types.NVARCHAR);
-    Mockito.when(sqlArray.getArray()).thenReturn(value);
-    Mockito.when(resultSet.getArray(0)).thenReturn(sqlArray);
-
-    Parser parser = new ArrayParser(resultSet, 0);
+    ArrayParser parser = new ArrayParser(resultSet, 0);
 
     validate(parser, byteResult, stringResult, spannerResult);
   }
 
   @Test
-  public void testLongArrayParsing() throws SQLException {
+  public void testLongArrayParsing() {
     Long[] value = {1L, 2L, 3L};
+    // The binary format of a PG array should contain the OID of the element type and not the array
+    // OID. For INT8 that means 20 and not 1016
+    // See
+    // https://github.com/pgjdbc/pgjdbc/blob/60a81034fd003551fc863033f491b2d0ed1dfa80/pgjdbc/src/main/java/org/postgresql/jdbc/ArrayDecoding.java#L505
+    // We can test this more thoroughly when
+    // https://github.com/GoogleCloudPlatform/pgadapter/pull/36
+    // has been merged.
     byte[] byteResult = {
-      0, 0, 0, 1, 0, 0, 0, 1, -1, -1, -1, -5, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0,
-      0, 1, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 3
+      0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0, 20, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0,
+      1, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 2, 0, 0, 0, 8, 0, 0, 0, 0, 0, 0, 0, 3
     };
     byte[] stringResult = {'{', '1', ',', '2', ',', '3', '}'};
     byte[] spannerResult = {'[', '1', ',', '2', ',', '3', ']'};
 
-    Array sqlArray = Mockito.mock(Array.class);
     ResultSet resultSet = Mockito.mock(ResultSet.class);
+    when(resultSet.getColumnType(0)).thenReturn(Type.array(Type.int64()));
+    when(resultSet.getValue(0)).thenReturn(Value.int64Array(Arrays.asList(value)));
 
-    Mockito.when(sqlArray.getBaseType()).thenReturn(Types.BIGINT);
-    Mockito.when(sqlArray.getArray()).thenReturn(value);
-    Mockito.when(resultSet.getArray(0)).thenReturn(sqlArray);
-
-    Parser parser = new ArrayParser(resultSet, 0);
+    ArrayParser parser = new ArrayParser(resultSet, 0);
 
     validate(parser, byteResult, stringResult, spannerResult);
   }
 
   @Test(expected = IllegalArgumentException.class)
-  public void testArrayArrayParsingFails() throws SQLException {
+  public void testArrayArrayParsingFails() {
     Long[] value = {1L, 2L, 3L};
 
-    Array sqlArray = Mockito.mock(Array.class);
     ResultSet resultSet = Mockito.mock(ResultSet.class);
-
-    Mockito.when(sqlArray.getBaseType()).thenReturn(Types.ARRAY);
-    Mockito.when(sqlArray.getArray()).thenReturn(value);
-    Mockito.when(resultSet.getArray(0)).thenReturn(sqlArray);
+    when(resultSet.getColumnType(0)).thenReturn(Type.array(Type.array(Type.int64())));
 
     new ArrayParser(resultSet, 0);
   }
 
   @Test
-  public void testNumericParsing() throws SQLException {
+  public void testNumericParsing() {
     BigDecimal value = new BigDecimal("1234567890.1234567890");
 
     byte[] byteResult = ByteConverter.numeric(new BigDecimal("1234567890.1234567890"));
@@ -390,7 +395,7 @@ public class ParserTest {
   }
 
   @Test
-  public void testNumericParsingNaN() throws SQLException {
+  public void testNumericParsingNaN() {
     Number value = Double.NaN;
 
     byte[] stringResult = {'N', 'a', 'N'};
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ProtocolTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ProtocolTest.java
index c1d3a37f..80722acc 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ProtocolTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ProtocolTest.java
@@ -22,7 +22,12 @@ import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
-import com.google.cloud.spanner.jdbc.CloudSpannerJdbcConnection;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
@@ -66,19 +71,15 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Paths;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
@@ -97,8 +98,6 @@ public class ProtocolTest {
   @Rule public MockitoRule rule = MockitoJUnit.rule();
   @Mock private ConnectionHandler connectionHandler;
   @Mock private Connection connection;
-  @Mock private Statement statement;
-  @Mock private PreparedStatement preparedStatement;
   @Mock private ProxyServer server;
   @Mock private OptionsMetadata options;
   @Mock private IntermediateStatement intermediateStatement;
@@ -109,6 +108,7 @@ public class ProtocolTest {
   @Mock private ConnectionMetadata connectionMetadata;
   @Mock private DataOutputStream outputStream;
   @Mock private ResultSet resultSet;
+  @Mock private StatementResult statementResult;
   @Mock private MutationWriter mutationWriter;
 
   private byte[] intToBytes(int value) {
@@ -137,6 +137,13 @@ public class ProtocolTest {
     return result;
   }
 
+  @AfterClass
+  public static void cleanup() {
+    // TODO: Make error log file configurable and turn off writing to a file during tests.
+    File outputFile = new File("output.txt");
+    outputFile.delete();
+  }
+
   @Test
   public void testQueryMessage() throws Exception {
     byte[] messageMetadata = {'Q', 0, 0, 0, 24};
@@ -147,12 +154,12 @@ public class ProtocolTest {
 
     String expectedSQL = "SELECT * FROM users";
 
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(statement.getResultSet()).thenReturn(mock(ResultSet.class));
+    Mockito.when(connection.execute(Statement.of(expectedSQL))).thenReturn(statementResult);
+    when(statementResult.getResultSet()).thenReturn(resultSet);
     Mockito.when(connectionHandler.getServer()).thenReturn(server);
     Mockito.when(server.getOptions()).thenReturn(options);
     Mockito.when(options.requiresMatcher()).thenReturn(false);
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -167,7 +174,7 @@ public class ProtocolTest {
 
     messageSpy.send();
     // Execute
-    Mockito.verify(statement, Mockito.times(1)).execute(expectedSQL);
+    Mockito.verify(connection, Mockito.times(1)).execute(Statement.of(expectedSQL));
   }
 
   @Test
@@ -181,14 +188,14 @@ public class ProtocolTest {
 
     String expectedSQL = "SELECT * FROM users";
 
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(statement.getResultSet()).thenReturn(mock(ResultSet.class));
+    when(connection.execute(Statement.of(expectedSQL))).thenReturn(statementResult);
+    Mockito.when(statementResult.getResultSet()).thenReturn(resultSet);
     Mockito.when(connectionHandler.getServer()).thenReturn(server);
     Mockito.when(server.getOptions()).thenReturn(options);
     Mockito.when(options.requiresMatcher()).thenReturn(true);
     Mockito.when(options.getCommandMetadataJSON())
         .thenReturn((JSONObject) parser.parse("{\"commands\": []}"));
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -203,7 +210,7 @@ public class ProtocolTest {
     Mockito.doNothing().when(messageSpy).handleQuery();
 
     messageSpy.send();
-    Mockito.verify(statement, Mockito.times(1)).execute(expectedSQL);
+    Mockito.verify(connection, Mockito.times(1)).execute(Statement.of(expectedSQL));
   }
 
   @Test(expected = IOException.class)
@@ -251,14 +258,14 @@ public class ProtocolTest {
             parameters);
 
     List<Integer> expectedParameterDataTypes = Arrays.asList(1002);
-    String expectedSQL = "/*GSQL*/SELECT * FROM users WHERE name = ?";
+    String expectedSQL = "/*GSQL*/SELECT * FROM users WHERE name = $1";
     String expectedMessageName = "some statement";
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -310,14 +317,14 @@ public class ProtocolTest {
             parameters);
 
     List<Integer> expectedParameterDataTypes = Arrays.asList(1002);
-    String expectedSQL = "SELECT * FROM users WHERE name = ?";
+    String expectedSQL = "SELECT * FROM users WHERE name = $1";
     String expectedMessageName = "some statement";
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -370,14 +377,14 @@ public class ProtocolTest {
             parameters);
 
     List<Integer> expectedParameterDataTypes = Arrays.asList(0);
-    String expectedSQL = "SELECT * FROM users WHERE name = ?";
+    String expectedSQL = "SELECT * FROM users WHERE name = $1";
     String expectedMessageName = "some statement";
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -401,8 +408,8 @@ public class ProtocolTest {
     Assert.assertEquals(outputResult.readInt(), 4);
   }
 
-  @Test(expected = IllegalArgumentException.class)
-  public void testParseMessageExceptsWithNonMatchingParameterTypeCount() throws Exception {
+  @Test
+  public void testParseMessageWithNonMatchingParameterTypeCount() throws Exception {
     byte[] messageMetadata = {'P'};
     String statementName = "some statement\0";
     String payload =
@@ -414,13 +421,15 @@ public class ProtocolTest {
         Bytes.concat(
             messageMetadata, length, statementName.getBytes(), payload.getBytes(), intToBytes(0));
 
-    List<Integer> expectedParameterDataTypes = Arrays.asList(0);
-    String expectedSQL = "SELECT * FROM users WHERE name = ?";
+    List<Integer> expectedParameterDataTypes = Collections.emptyList();
+    String expectedSQL = "SELECT * FROM users WHERE name = $1";
     String expectedMessageName = "some statement";
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -435,6 +444,13 @@ public class ProtocolTest {
 
     Mockito.when(connectionHandler.hasStatement(anyString())).thenReturn(false);
     message.send();
+    Mockito.verify(connectionHandler, Mockito.times(1))
+        .registerStatement(expectedMessageName, ((ParseMessage) message).getStatement());
+
+    // ParseCompleteResponse
+    DataInputStream outputResult = inputStreamFromOutputStream(result);
+    Assert.assertEquals(outputResult.readByte(), '1');
+    Assert.assertEquals(outputResult.readInt(), 4);
   }
 
   @Test(expected = IllegalStateException.class)
@@ -466,7 +482,7 @@ public class ProtocolTest {
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -508,7 +524,7 @@ public class ProtocolTest {
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -553,7 +569,7 @@ public class ProtocolTest {
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -984,9 +1000,7 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(connection.unwrap(CloudSpannerJdbcConnection.class))
-        .thenReturn(mock(CloudSpannerJdbcConnection.class));
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -1015,10 +1029,8 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    CloudSpannerJdbcConnection cloudSpannerConnection = mock(CloudSpannerJdbcConnection.class);
-    when(cloudSpannerConnection.isInTransaction()).thenReturn(true);
-    when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    when(connection.unwrap(CloudSpannerJdbcConnection.class)).thenReturn(cloudSpannerConnection);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.isInTransaction()).thenReturn(true);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -1047,9 +1059,7 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(connection.unwrap(CloudSpannerJdbcConnection.class))
-        .thenReturn(mock(CloudSpannerJdbcConnection.class));
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     Mockito.when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     Mockito.when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     Mockito.when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -1078,10 +1088,8 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    CloudSpannerJdbcConnection cloudSpannerConnection = mock(CloudSpannerJdbcConnection.class);
-    when(cloudSpannerConnection.isInTransaction()).thenReturn(true);
-    when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    when(connection.unwrap(CloudSpannerJdbcConnection.class)).thenReturn(cloudSpannerConnection);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.isInTransaction()).thenReturn(true);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -1110,14 +1118,11 @@ public class ProtocolTest {
 
     String expectedSQL = "INSERT INTO users (name) VALUES ('test')";
 
-    CloudSpannerJdbcConnection cloudSpannerConnection = mock(CloudSpannerJdbcConnection.class);
-    when(cloudSpannerConnection.isInTransaction()).thenReturn(true);
-    when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    when(connection.unwrap(CloudSpannerJdbcConnection.class)).thenReturn(cloudSpannerConnection);
-    when(connection.createStatement()).thenReturn(statement);
+    when(connection.isInTransaction()).thenReturn(true);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     ResultSet resultSet = mock(ResultSet.class);
-    when(statement.getResultSet()).thenReturn(resultSet);
-    when(resultSet.getMetaData()).thenReturn(mock(ResultSetMetaData.class));
+    when(statementResult.getResultSet()).thenReturn(resultSet);
+    when(connection.execute(Statement.of(expectedSQL))).thenReturn(statementResult);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getServer()).thenReturn(server);
     OptionsMetadata options = mock(OptionsMetadata.class);
@@ -1217,12 +1222,10 @@ public class ProtocolTest {
 
   @Test
   public void testMultipleCopyDataMessages() throws Exception {
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.execute(any(Statement.class))).thenReturn(statementResult);
+    when(statementResult.getUpdateCount()).thenReturn(1L);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
 
     byte[] messageMetadata = {'d'};
     byte[] payload1 = "1\t'one'\n2\t".getBytes();
@@ -1243,7 +1246,7 @@ public class ProtocolTest {
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
 
     CopyStatement copyStatement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
     copyStatement.execute();
@@ -1347,11 +1350,9 @@ public class ProtocolTest {
 
   @Test
   public void testCopyFromFilePipe() throws Exception {
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
+    when(connection.execute(any(Statement.class))).thenReturn(statementResult);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    Mockito.when(statementResult.getUpdateCount()).thenReturn(1L);
 
     byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/small-file-test.txt"));
 
@@ -1359,7 +1360,7 @@ public class ProtocolTest {
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
 
     CopyStatement copyStatement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
     copyStatement.execute();
@@ -1380,11 +1381,9 @@ public class ProtocolTest {
 
   @Test
   public void testCopyBatchSizeLimit() throws Exception {
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.execute(any(Statement.class))).thenReturn(statementResult);
+    when(statementResult.getUpdateCount()).thenReturn(1L);
 
     byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/batch-size-test.txt"));
 
@@ -1392,7 +1391,7 @@ public class ProtocolTest {
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
 
     CopyStatement copyStatement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
 
@@ -1418,11 +1417,9 @@ public class ProtocolTest {
 
   @Test
   public void testCopyDataRowLengthMismatchLimit() throws Exception {
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.execute(any(Statement.class))).thenReturn(statementResult);
+    Mockito.when(statementResult.getUpdateCount()).thenReturn(1L);
 
     byte[] payload = "1\t'one'\n2".getBytes();
 
@@ -1430,7 +1427,7 @@ public class ProtocolTest {
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
 
     CopyStatement copyStatement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
 
@@ -1442,27 +1439,28 @@ public class ProtocolTest {
     MutationWriter mwSpy = Mockito.spy(mw);
     Mockito.when(mwSpy.writeToSpanner(Mockito.any(ConnectionHandler.class))).thenReturn(1);
 
-    Exception thrown =
+    SpannerException thrown =
         Assert.assertThrows(
-            SQLException.class,
+            SpannerException.class,
             () -> {
               mwSpy.addCopyData(connectionHandler, payload);
               mwSpy.buildMutationList(connectionHandler);
-              ;
             });
+    Assert.assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
     Assert.assertEquals(
-        thrown.getMessage(),
-        "Invalid COPY data: Row length mismatched. Expected 2 columns, but only found 1");
+        "INVALID_ARGUMENT: Invalid COPY data: Row length mismatched. Expected 2 columns, but only found 1",
+        thrown.getMessage());
     copyStatement.close();
+
+    File outputFile = new File("output.txt");
+    outputFile.delete();
   }
 
   @Test
   public void testCopyResumeErrorOutputFile() throws Exception {
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.execute(any(Statement.class))).thenReturn(statementResult);
+    Mockito.when(statementResult.getUpdateCount()).thenReturn(1L);
 
     byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/test-copy-output.txt"));
 
@@ -1470,29 +1468,33 @@ public class ProtocolTest {
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
 
     CopyStatement copyStatement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
     Assert.assertFalse(copyStatement.isExecuted());
     copyStatement.execute();
     Assert.assertTrue(copyStatement.isExecuted());
 
+    File outputFile = new File("output.txt");
+    outputFile.delete();
+
     MutationWriter mw = copyStatement.getMutationWriter();
     MutationWriter mwSpy = Mockito.spy(mw);
     Mockito.when(mwSpy.writeToSpanner(connectionHandler)).thenReturn(4);
 
-    Exception thrown =
+    SpannerException thrown =
         Assert.assertThrows(
-            SQLException.class,
+            SpannerException.class,
             () -> {
               mwSpy.addCopyData(connectionHandler, payload);
               mwSpy.buildMutationList(connectionHandler);
               mwSpy.writeToSpanner(connectionHandler);
             });
-    Assert.assertEquals(thrown.getMessage(), "Invalid input syntax for type INT64:\"'5'\"");
+    Assert.assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
+    Assert.assertEquals(
+        "INVALID_ARGUMENT: Invalid input syntax for type INT64:\"'5'\"", thrown.getMessage());
 
     Mockito.verify(mwSpy, Mockito.times(1)).writeCopyDataToErrorFile();
-    File outputFile = new File("output.txt");
     Assert.assertTrue(outputFile.exists());
     Assert.assertTrue(outputFile.isFile());
 
@@ -1502,11 +1504,9 @@ public class ProtocolTest {
 
   @Test
   public void testCopyResumeErrorStartOutputFile() throws Exception {
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.execute(any(Statement.class))).thenReturn(statementResult);
+    when(statementResult.getUpdateCount()).thenReturn(1L);
 
     byte[] payload =
         Files.readAllBytes(Paths.get("./src/test/resources/test-copy-start-output.txt"));
@@ -1515,7 +1515,12 @@ public class ProtocolTest {
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
+
+    // Pre-emptively try to delete the output file if it is lingering from a previous test run.
+    // TODO: Make the output file for COPY configurable, so we can use a temp file for this.
+    File outputFile = new File("output.txt");
+    outputFile.delete();
 
     CopyStatement copyStatement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
     Assert.assertFalse(copyStatement.isExecuted());
@@ -1526,18 +1531,19 @@ public class ProtocolTest {
     MutationWriter mwSpy = Mockito.spy(mw);
     Mockito.when(mwSpy.writeToSpanner(connectionHandler)).thenReturn(1);
 
-    Exception thrown =
+    SpannerException thrown =
         Assert.assertThrows(
-            SQLException.class,
+            SpannerException.class,
             () -> {
               mwSpy.addCopyData(connectionHandler, payload);
               mwSpy.buildMutationList(connectionHandler);
               mwSpy.writeToSpanner(connectionHandler);
             });
-    Assert.assertEquals(thrown.getMessage(), "Invalid input syntax for type INT64:\"'1'\"");
+    Assert.assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
+    Assert.assertEquals(
+        "INVALID_ARGUMENT: Invalid input syntax for type INT64:\"'1'\"", thrown.getMessage());
 
     Mockito.verify(mwSpy, Mockito.times(1)).writeCopyDataToErrorFile();
-    File outputFile = new File("output.txt");
     Assert.assertTrue(outputFile.exists());
     Assert.assertTrue(outputFile.isFile());
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/RowDescriptionTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/RowDescriptionTest.java
index d60eed8e..125932dc 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/RowDescriptionTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/RowDescriptionTest.java
@@ -16,7 +16,11 @@ package com.google.cloud.spanner.pgadapter.wireoutput;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.is;
+import static org.mockito.Mockito.when;
 
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Type;
+import com.google.cloud.spanner.Type.StructField;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.TextFormat;
@@ -27,8 +31,6 @@ import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
-import java.sql.ResultSetMetaData;
-import java.sql.Types;
 import org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.junit.Assert;
@@ -54,27 +56,29 @@ public final class RowDescriptionTest {
   private ByteArrayOutputStream buffer = new ByteArrayOutputStream();
   private DataOutputStream output = new DataOutputStream(buffer);
   @Mock private IntermediateStatement statement;
-  @Mock private ResultSetMetaData metadata;
+  @Mock private ResultSet metadata;
+  private Type rowType;
 
   @Before
-  public void setUp() {}
+  public void setUp() {
+    rowType =
+        Type.struct(
+            StructField.of("default-column-name", Type.int64()),
+            StructField.of("default-column-name", Type.int64()));
+    when(metadata.getType()).thenReturn(rowType);
+  }
 
   @Test
   public void OidTest() throws Exception {
-    Mockito.when(metadata.getColumnCount()).thenReturn(0);
-    Mockito.when(metadata.getColumnType(1)).thenReturn(Types.SMALLINT);
-    Mockito.when(metadata.getColumnType(2)).thenReturn(Types.INTEGER);
-    Mockito.when(metadata.getColumnType(3)).thenReturn(Types.BIGINT);
-    Mockito.when(metadata.getColumnType(4)).thenReturn(Types.NUMERIC);
-    Mockito.when(metadata.getColumnType(5)).thenReturn(Types.REAL);
-    Mockito.when(metadata.getColumnType(6)).thenReturn(Types.DOUBLE);
-    Mockito.when(metadata.getColumnType(7)).thenReturn(Types.CHAR);
-    Mockito.when(metadata.getColumnType(8)).thenReturn(Types.VARCHAR);
-    Mockito.when(metadata.getColumnType(9)).thenReturn(Types.BINARY);
-    Mockito.when(metadata.getColumnType(10)).thenReturn(Types.BIT);
-    Mockito.when(metadata.getColumnType(11)).thenReturn(Types.DATE);
-    Mockito.when(metadata.getColumnType(12)).thenReturn(Types.TIME);
-    Mockito.when(metadata.getColumnType(13)).thenReturn(Types.TIMESTAMP);
+    when(metadata.getColumnCount()).thenReturn(0);
+    when(metadata.getColumnType(0)).thenReturn(Type.int64());
+    when(metadata.getColumnType(1)).thenReturn(Type.pgNumeric());
+    when(metadata.getColumnType(2)).thenReturn(Type.float64());
+    when(metadata.getColumnType(3)).thenReturn(Type.string());
+    when(metadata.getColumnType(4)).thenReturn(Type.bytes());
+    when(metadata.getColumnType(5)).thenReturn(Type.bool());
+    when(metadata.getColumnType(6)).thenReturn(Type.date());
+    when(metadata.getColumnType(7)).thenReturn(Type.timestamp());
 
     JSONParser parser = new JSONParser();
     JSONObject commandMetadata = (JSONObject) parser.parse(EMPTY_COMMAND_JSON);
@@ -91,44 +95,29 @@ public final class RowDescriptionTest {
     RowDescriptionResponse response =
         new RowDescriptionResponse(output, statement, metadata, options, mode);
 
-    // Types.SMALLINT
-    Assert.assertEquals(response.getOidType(1), Oid.INT2);
-    Assert.assertEquals(response.getOidTypeSize(Oid.INT2), 2);
-    // Types.INTEGER
-    Assert.assertEquals(response.getOidType(2), Oid.INT4);
-    Assert.assertEquals(response.getOidTypeSize(Oid.INT4), 4);
     // Types.BIGINT
-    Assert.assertEquals(response.getOidType(3), Oid.INT8);
+    Assert.assertEquals(response.getOidType(0), Oid.INT8);
     Assert.assertEquals(response.getOidTypeSize(Oid.INT8), 8);
     // Types.NUMERIC
-    Assert.assertEquals(response.getOidType(4), Oid.NUMERIC);
+    Assert.assertEquals(response.getOidType(1), Oid.NUMERIC);
     Assert.assertEquals(response.getOidTypeSize(Oid.NUMERIC), -1);
-    // Types.REAL
-    Assert.assertEquals(response.getOidType(5), Oid.FLOAT4);
-    Assert.assertEquals(response.getOidTypeSize(Oid.FLOAT4), 4);
     // Types.DOUBLE
-    Assert.assertEquals(response.getOidType(6), Oid.FLOAT8);
+    Assert.assertEquals(response.getOidType(2), Oid.FLOAT8);
     Assert.assertEquals(response.getOidTypeSize(Oid.FLOAT8), 8);
-    // Types.CHAR
-    Assert.assertEquals(response.getOidType(7), Oid.CHAR);
-    Assert.assertEquals(response.getOidTypeSize(Oid.CHAR), 1);
     // Types.VARCHAR
-    Assert.assertEquals(response.getOidType(8), Oid.VARCHAR);
+    Assert.assertEquals(response.getOidType(3), Oid.VARCHAR);
     Assert.assertEquals(response.getOidTypeSize(Oid.VARCHAR), -1);
     // Types.BINARY
-    Assert.assertEquals(response.getOidType(9), Oid.BYTEA);
+    Assert.assertEquals(response.getOidType(4), Oid.BYTEA);
     Assert.assertEquals(response.getOidTypeSize(Oid.BYTEA), -1);
     // Types.BIT
-    Assert.assertEquals(response.getOidType(10), Oid.BOOL);
+    Assert.assertEquals(response.getOidType(5), Oid.BOOL);
     Assert.assertEquals(response.getOidTypeSize(Oid.BOOL), 1);
     // Types.DATE
-    Assert.assertEquals(response.getOidType(11), Oid.DATE);
+    Assert.assertEquals(response.getOidType(6), Oid.DATE);
     Assert.assertEquals(response.getOidTypeSize(Oid.DATE), 8);
-    // Types.TIME
-    Assert.assertEquals(response.getOidType(12), Oid.TIME);
-    Assert.assertEquals(response.getOidTypeSize(Oid.TIME), 8);
     // Types.TIMESTAMP
-    Assert.assertEquals(response.getOidType(13), Oid.TIMESTAMP);
+    Assert.assertEquals(response.getOidType(7), Oid.TIMESTAMP);
     Assert.assertEquals(response.getOidTypeSize(Oid.TIMESTAMP), 12);
   }
 
@@ -136,9 +125,11 @@ public final class RowDescriptionTest {
   public void SendPayloadNullStatementTest() throws Exception {
     int COLUMN_COUNT = 1;
     String COLUMN_NAME = "default-column-name";
-    Mockito.when(metadata.getColumnCount()).thenReturn(COLUMN_COUNT);
-    Mockito.when(metadata.getColumnName(Mockito.anyInt())).thenReturn(COLUMN_NAME);
-    Mockito.when(metadata.getColumnType(Mockito.anyInt())).thenReturn(Types.SMALLINT);
+    Type rowType = Type.struct(StructField.of(COLUMN_NAME, Type.string()));
+    when(metadata.getColumnCount()).thenReturn(COLUMN_COUNT);
+    when(metadata.getType()).thenReturn(rowType);
+    when(metadata.getColumnType(Mockito.anyInt())).thenReturn(Type.int64());
+
     JSONParser parser = new JSONParser();
     JSONObject commandMetadata = (JSONObject) parser.parse(EMPTY_COMMAND_JSON);
     OptionsMetadata options =
@@ -171,9 +162,9 @@ public final class RowDescriptionTest {
     // column index
     Assert.assertThat(outputReader.readShort(), is(equalTo((short) DEFAULT_FLAG)));
     // type oid
-    Assert.assertEquals(outputReader.readInt(), Oid.INT2);
+    Assert.assertEquals(outputReader.readInt(), Oid.INT8);
     // type size
-    Assert.assertThat(outputReader.readShort(), is(equalTo((short) 2)));
+    Assert.assertThat(outputReader.readShort(), is(equalTo((short) 8)));
     // type modifier
     Assert.assertThat(outputReader.readInt(), is(equalTo((int) DEFAULT_FLAG)));
     // format code
@@ -184,10 +175,11 @@ public final class RowDescriptionTest {
   public void SendPayloadStatementWithBinaryOptionTest() throws Exception {
     int COLUMN_COUNT = 1;
     String COLUMN_NAME = "default-column-name";
-    Mockito.when(metadata.getColumnCount()).thenReturn(COLUMN_COUNT);
-    Mockito.when(metadata.getColumnName(Mockito.anyInt())).thenReturn(COLUMN_NAME);
-    Mockito.when(metadata.getColumnType(Mockito.anyInt())).thenReturn(Types.SMALLINT);
-    Mockito.when(statement.getResultFormatCode(Mockito.anyInt())).thenReturn((short) 0);
+    Type rowType = Type.struct(StructField.of(COLUMN_NAME, Type.string()));
+    when(metadata.getColumnCount()).thenReturn(COLUMN_COUNT);
+    when(metadata.getType()).thenReturn(rowType);
+    when(metadata.getColumnType(Mockito.anyInt())).thenReturn(Type.int64());
+    when(statement.getResultFormatCode(Mockito.anyInt())).thenReturn((short) 0);
     JSONParser parser = new JSONParser();
     JSONObject commandMetadata = (JSONObject) parser.parse(EMPTY_COMMAND_JSON);
     OptionsMetadata options =
@@ -220,9 +212,9 @@ public final class RowDescriptionTest {
     // column index
     Assert.assertThat(outputReader.readShort(), is(equalTo((short) DEFAULT_FLAG)));
     // type oid
-    Assert.assertEquals(outputReader.readInt(), Oid.INT2);
+    Assert.assertEquals(outputReader.readInt(), Oid.INT8);
     // type size
-    Assert.assertThat(outputReader.readShort(), is(equalTo((short) 2)));
+    Assert.assertThat(outputReader.readShort(), is(equalTo((short) 8)));
     // type modifier
     Assert.assertThat(outputReader.readInt(), is(equalTo((int) DEFAULT_FLAG)));
     // format code
@@ -233,10 +225,13 @@ public final class RowDescriptionTest {
   public void SendPayloadStatementTest() throws Exception {
     int COLUMN_COUNT = 2;
     String COLUMN_NAME = "default-column-name";
-    Mockito.when(metadata.getColumnCount()).thenReturn(COLUMN_COUNT);
-    Mockito.when(metadata.getColumnName(Mockito.anyInt())).thenReturn(COLUMN_NAME);
-    Mockito.when(metadata.getColumnType(Mockito.anyInt())).thenReturn(Types.SMALLINT);
-    Mockito.when(statement.getResultFormatCode(Mockito.anyInt()))
+    Type rowType =
+        Type.struct(
+            StructField.of(COLUMN_NAME, Type.string()), StructField.of(COLUMN_NAME, Type.string()));
+    when(metadata.getColumnCount()).thenReturn(COLUMN_COUNT);
+    when(metadata.getType()).thenReturn(rowType);
+    when(metadata.getColumnType(Mockito.anyInt())).thenReturn(Type.int64());
+    when(statement.getResultFormatCode(Mockito.anyInt()))
         .thenReturn((short) 0)
         .thenReturn((short) 0)
         .thenReturn((short) 1);
@@ -272,9 +267,9 @@ public final class RowDescriptionTest {
       // column index
       Assert.assertThat(outputReader.readShort(), is(equalTo((short) DEFAULT_FLAG)));
       // type oid
-      Assert.assertEquals(outputReader.readInt(), Oid.INT2);
+      Assert.assertEquals(outputReader.readInt(), Oid.INT8);
       // type size
-      Assert.assertThat(outputReader.readShort(), is(equalTo((short) 2)));
+      Assert.assertThat(outputReader.readShort(), is(equalTo((short) 8)));
       // type modifier
       Assert.assertThat(outputReader.readInt(), is(equalTo((int) DEFAULT_FLAG)));
       // format code
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
index 77d6a77a..e19fb26d 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
@@ -17,19 +17,24 @@ package com.google.cloud.spanner.pgadapter;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.when;
 
-import com.google.cloud.spanner.jdbc.JdbcConstants;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ReadContext.QueryAnalyzeMode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
-import com.google.cloud.spanner.pgadapter.metadata.SQLMetadata;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePortalStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement.ResultType;
-import com.google.cloud.spanner.pgadapter.utils.Converter;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage;
@@ -38,21 +43,16 @@ import com.google.common.primitives.Bytes;
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.sql.Types;
+import java.io.File;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
-import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnit;
@@ -67,10 +67,9 @@ public class StatementTest {
   @Mock private Connection connection;
   @Mock private ConnectionHandler connectionHandler;
   @Mock private ConnectionMetadata connectionMetadata;
-  @Mock private Statement statement;
-  @Mock private PreparedStatement preparedStatement;
   @Mock private ProxyServer server;
   @Mock private OptionsMetadata options;
+  @Mock private StatementResult statementResult;
   @Mock private ResultSet resultSet;
   @Mock private DataOutputStream outputStream;
 
@@ -80,11 +79,18 @@ public class StatementTest {
     return parameters;
   }
 
+  @AfterClass
+  public static void cleanup() {
+    // TODO: Make error log file configurable and turn off writing to a file during tests.
+    File outputFile = new File("output.txt");
+    outputFile.delete();
+  }
+
   @Test
   public void testBasicSelectStatement() throws Exception {
-    when(connection.createStatement()).thenReturn(statement);
-    when(statement.getUpdateCount()).thenReturn(JdbcConstants.STATEMENT_RESULT_SET);
-    when(statement.getResultSet()).thenReturn(resultSet);
+    when(connection.execute(Statement.of("SELECT * FROM users"))).thenReturn(statementResult);
+    when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.RESULT_SET);
+    when(statementResult.getResultSet()).thenReturn(resultSet);
     when(resultSet.next()).thenReturn(true);
 
     IntermediateStatement intermediateStatement =
@@ -95,7 +101,7 @@ public class StatementTest {
 
     intermediateStatement.execute();
 
-    Mockito.verify(statement, Mockito.times(1)).execute("SELECT * FROM users");
+    Mockito.verify(connection, Mockito.times(1)).execute(Statement.of("SELECT * FROM users"));
     assertTrue(intermediateStatement.containsResultSet());
     assertTrue(intermediateStatement.isExecuted());
     assertEquals(intermediateStatement.getResultType(), ResultType.RESULT_SET);
@@ -111,8 +117,10 @@ public class StatementTest {
 
   @Test
   public void testBasicUpdateStatement() throws Exception {
-    when(connection.createStatement()).thenReturn(statement);
-    when(statement.getUpdateCount()).thenReturn(1);
+    when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.UPDATE_COUNT);
+    when(statementResult.getUpdateCount()).thenReturn(1L);
+    when(connection.execute(Statement.of("UPDATE users SET name = someName WHERE id = 10")))
+        .thenReturn(statementResult);
 
     IntermediateStatement intermediateStatement =
         new IntermediateStatement("UPDATE users SET name = someName WHERE id = 10", connection);
@@ -122,10 +130,10 @@ public class StatementTest {
 
     intermediateStatement.execute();
 
-    Mockito.verify(statement, Mockito.times(1))
-        .execute("UPDATE users SET name = someName WHERE id = 10");
+    Mockito.verify(connection, Mockito.times(1))
+        .execute(Statement.of("UPDATE users SET name = someName WHERE id = 10"));
     assertFalse(intermediateStatement.containsResultSet());
-    assertEquals((int) intermediateStatement.getUpdateCount(), 1);
+    assertEquals(intermediateStatement.getUpdateCount().longValue(), 1L);
     assertTrue(intermediateStatement.isExecuted());
     assertEquals(intermediateStatement.getResultType(), ResultType.UPDATE_COUNT);
     Assert.assertNull(intermediateStatement.getStatementResult());
@@ -140,8 +148,10 @@ public class StatementTest {
 
   @Test
   public void testBasicZeroUpdateCountResultStatement() throws Exception {
-    when(connection.createStatement()).thenReturn(statement);
-    when(statement.getUpdateCount()).thenReturn(0);
+    when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.UPDATE_COUNT);
+    when(statementResult.getUpdateCount()).thenReturn(0L);
+    when(connection.execute(Statement.of("UPDATE users SET name = someName WHERE id = -1")))
+        .thenReturn(statementResult);
 
     IntermediateStatement intermediateStatement =
         new IntermediateStatement("UPDATE users SET name = someName WHERE id = -1", connection);
@@ -151,10 +161,10 @@ public class StatementTest {
 
     intermediateStatement.execute();
 
-    Mockito.verify(statement, Mockito.times(1))
-        .execute("UPDATE users SET name = someName WHERE id = -1");
+    Mockito.verify(connection, Mockito.times(1))
+        .execute(Statement.of("UPDATE users SET name = someName WHERE id = -1"));
     assertFalse(intermediateStatement.containsResultSet());
-    assertEquals((int) intermediateStatement.getUpdateCount(), 0);
+    assertEquals(intermediateStatement.getUpdateCount().longValue(), 0L);
     assertTrue(intermediateStatement.isExecuted());
     assertEquals(intermediateStatement.getResultType(), ResultType.UPDATE_COUNT);
     Assert.assertNull(intermediateStatement.getStatementResult());
@@ -169,8 +179,10 @@ public class StatementTest {
 
   @Test
   public void testBasicNoResultStatement() throws Exception {
-    when(connection.createStatement()).thenReturn(statement);
-    when(statement.getUpdateCount()).thenReturn(0);
+    when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.NO_RESULT);
+    when(statementResult.getUpdateCount()).thenReturn(0L);
+    when(connection.execute(Statement.of("CREATE TABLE users (name varchar(100) primary key)")))
+        .thenReturn(statementResult);
 
     IntermediateStatement intermediateStatement =
         new IntermediateStatement("CREATE TABLE users (name varchar(100) primary key)", connection);
@@ -180,10 +192,10 @@ public class StatementTest {
 
     intermediateStatement.execute();
 
-    Mockito.verify(statement, Mockito.times(1))
-        .execute("CREATE TABLE users (name varchar(100) primary key)");
+    Mockito.verify(connection, Mockito.times(1))
+        .execute(Statement.of("CREATE TABLE users (name varchar(100) primary key)"));
     assertFalse(intermediateStatement.containsResultSet());
-    assertEquals((int) intermediateStatement.getUpdateCount(), 0);
+    assertEquals(intermediateStatement.getUpdateCount().longValue(), 0L);
     assertTrue(intermediateStatement.isExecuted());
     assertEquals(intermediateStatement.getResultType(), ResultType.NO_RESULT);
     Assert.assertNull(intermediateStatement.getStatementResult());
@@ -198,8 +210,6 @@ public class StatementTest {
 
   @Test(expected = IllegalStateException.class)
   public void testDescribeBasicStatementThrowsException() throws Exception {
-    when(connection.createStatement()).thenReturn(statement);
-
     IntermediateStatement intermediateStatement =
         new IntermediateStatement("SELECT * FROM users", connection);
 
@@ -208,10 +218,10 @@ public class StatementTest {
 
   @Test
   public void testBasicStatementExceptionGetsSetOnExceptedExecution() throws Exception {
-    SQLException thrownException = new SQLException();
+    SpannerException thrownException =
+        SpannerExceptionFactory.newSpannerException(ErrorCode.INVALID_ARGUMENT, "test error");
 
-    when(connection.createStatement()).thenReturn(statement);
-    when(statement.execute(ArgumentMatchers.anyString())).thenThrow(thrownException);
+    when(connection.execute(Statement.of("SELECT * FROM users"))).thenThrow(thrownException);
 
     IntermediateStatement intermediateStatement =
         new IntermediateStatement("SELECT * FROM users", connection);
@@ -227,35 +237,34 @@ public class StatementTest {
     String sqlStatement = "SELECT * FROM users WHERE age > $2 AND age < $3 AND name = $1";
     List<Integer> parameterDataTypes = Arrays.asList(Oid.VARCHAR, Oid.INT8, Oid.INT4);
 
-    when(connection.prepareStatement(ArgumentMatchers.anyString())).thenReturn(preparedStatement);
-    when(preparedStatement.getResultSet()).thenReturn(mock(ResultSet.class));
+    Statement statement =
+        Statement.newBuilder(sqlStatement)
+            .bind("p1")
+            .to("userName")
+            .bind("p2")
+            .to(20L)
+            .bind("p3")
+            .to(30)
+            .build();
+    when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.UPDATE_COUNT);
+    when(statementResult.getUpdateCount()).thenReturn(0L);
+    when(connection.execute(statement)).thenReturn(statementResult);
 
     IntermediatePreparedStatement intermediateStatement =
         new IntermediatePreparedStatement(sqlStatement, connection);
     intermediateStatement.setParameterDataTypes(parameterDataTypes);
 
-    String expectedSQL = "SELECT * FROM users WHERE age > ? AND age < ? AND name = ?";
-
-    assertEquals(intermediateStatement.getSql(), expectedSQL);
-    assertEquals(intermediateStatement.getParameterCount(), 3);
-
-    intermediateStatement.execute();
-
-    Mockito.verify(preparedStatement, Mockito.times(1)).execute();
-    Mockito.verify(connection, Mockito.times(1)).prepareStatement(expectedSQL);
+    assertEquals(intermediateStatement.getSql(), sqlStatement);
 
     byte[][] parameters = {"userName".getBytes(), "20".getBytes(), "30".getBytes()};
     IntermediatePortalStatement intermediatePortalStatement =
         intermediateStatement.bind(
             parameters, Arrays.asList((short) 0, (short) 0, (short) 0), new ArrayList<>());
+    intermediateStatement.execute();
+    Mockito.verify(connection, Mockito.times(1)).execute(statement);
 
-    Mockito.verify(preparedStatement, Mockito.times(1)).setObject(1, 20L, Types.BIGINT);
-    Mockito.verify(preparedStatement, Mockito.times(1)).setObject(2, 30, Types.INTEGER);
-    Mockito.verify(preparedStatement, Mockito.times(1)).setObject(3, "userName", Types.VARCHAR);
-
-    assertEquals(intermediatePortalStatement.getSql(), expectedSQL);
+    assertEquals(intermediatePortalStatement.getSql(), sqlStatement);
     assertEquals(intermediatePortalStatement.getCommand(), "SELECT");
-    assertEquals(intermediatePortalStatement.getStatement(), preparedStatement);
     assertFalse(intermediatePortalStatement.isExecuted());
     assertTrue(intermediateStatement.isBound());
   }
@@ -265,8 +274,6 @@ public class StatementTest {
     String sqlStatement = "SELECT * FROM users WHERE metadata = $1";
     List<Integer> parameterDataTypes = Arrays.asList(Oid.JSON);
 
-    when(connection.prepareStatement(ArgumentMatchers.anyString())).thenReturn(preparedStatement);
-
     IntermediatePreparedStatement intermediateStatement =
         new IntermediatePreparedStatement(sqlStatement, connection);
     intermediateStatement.setParameterDataTypes(parameterDataTypes);
@@ -280,8 +287,6 @@ public class StatementTest {
   public void testPreparedStatementDescribeThrowsException() throws Exception {
     String sqlStatement = "SELECT * FROM users WHERE name = $1 AND age > $2 AND age < $3";
 
-    when(connection.prepareStatement(ArgumentMatchers.anyString())).thenReturn(preparedStatement);
-
     IntermediatePreparedStatement intermediateStatement =
         new IntermediatePreparedStatement(sqlStatement, connection);
 
@@ -291,21 +296,16 @@ public class StatementTest {
   @Test
   public void testPortalStatement() throws Exception {
     String sqlStatement = "SELECT * FROM users WHERE age > $1 AND age < $2 AND name = $3";
-    List<Integer> parameterDataTypes = Arrays.asList(Oid.INT8, Oid.INT4);
-    byte[][] parameters = {"20".getBytes(), "30".getBytes(), "userName".getBytes()};
-    SQLMetadata metadata = Converter.toJDBCParams(sqlStatement);
+    when(connection.analyzeQuery(Statement.of(sqlStatement), QueryAnalyzeMode.PLAN))
+        .thenReturn(resultSet);
 
     IntermediatePortalStatement intermediateStatement =
-        new IntermediatePortalStatement(
-            preparedStatement,
-            metadata.getSqlString(),
-            metadata.getParameterCount(),
-            metadata.getParameterIndexToPositions(),
-            connection);
+        new IntermediatePortalStatement(sqlStatement, connection);
 
     intermediateStatement.describe();
 
-    Mockito.verify(preparedStatement, Mockito.times(1)).getMetaData();
+    Mockito.verify(connection, Mockito.times(1))
+        .analyzeQuery(Statement.of(sqlStatement), QueryAnalyzeMode.PLAN);
 
     assertEquals(intermediateStatement.getParameterFormatCode(0), 0);
     assertEquals(intermediateStatement.getParameterFormatCode(1), 0);
@@ -336,17 +336,13 @@ public class StatementTest {
   @Test(expected = IllegalStateException.class)
   public void testPortalStatementDescribePropagatesFailure() throws Exception {
     String sqlStatement = "SELECT * FROM users WHERE age > $1 AND age < $2 AND name = $3";
-    SQLMetadata metadata = Converter.toJDBCParams(sqlStatement);
 
     IntermediatePortalStatement intermediateStatement =
-        new IntermediatePortalStatement(
-            preparedStatement,
-            metadata.getSqlString(),
-            metadata.getParameterCount(),
-            metadata.getParameterIndexToPositions(),
-            connection);
+        new IntermediatePortalStatement(sqlStatement, connection);
 
-    when(preparedStatement.getMetaData()).thenThrow(new SQLException());
+    when(connection.analyzeQuery(Statement.of(sqlStatement), QueryAnalyzeMode.PLAN))
+        .thenThrow(
+            SpannerExceptionFactory.newSpannerException(ErrorCode.INVALID_ARGUMENT, "test error"));
 
     intermediateStatement.describe();
   }
@@ -414,8 +410,7 @@ public class StatementTest {
     byte[] value = Bytes.concat(messageMetadata, payload.getBytes());
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
-    when(connection.createStatement()).thenReturn(statement);
-    when(connectionHandler.getJdbcConnection()).thenReturn(connection);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
@@ -436,17 +431,16 @@ public class StatementTest {
 
   @Test
   public void testCopyBuildMutation() throws Exception {
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    Mockito.when(statementResult.getResultType())
+        .thenReturn(StatementResult.ResultType.UPDATE_COUNT);
+    Mockito.when(statementResult.getUpdateCount()).thenReturn(1L);
 
     ResultSet spannerType = Mockito.mock(ResultSet.class);
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
 
     CopyStatement statement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
     statement.execute();
@@ -468,17 +462,16 @@ public class StatementTest {
 
   @Test
   public void testCopyInvalidBuildMutation() throws Exception {
-    Mockito.when(connection.createStatement()).thenReturn(statement);
-    Mockito.when(connection.prepareStatement(ArgumentMatchers.anyString()))
-        .thenReturn(preparedStatement);
-    Mockito.when(connectionHandler.getJdbcConnection()).thenReturn(connection);
-    Mockito.when(statement.getUpdateCount()).thenReturn(1);
+    Mockito.when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    Mockito.when(statementResult.getResultType())
+        .thenReturn(StatementResult.ResultType.UPDATE_COUNT);
+    Mockito.when(statementResult.getUpdateCount()).thenReturn(1L);
 
     ResultSet spannerType = Mockito.mock(ResultSet.class);
     Mockito.when(spannerType.getString("column_name")).thenReturn("key", "value");
     Mockito.when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     Mockito.when(spannerType.next()).thenReturn(true, true, false);
-    Mockito.when(preparedStatement.executeQuery()).thenReturn(spannerType);
+    Mockito.when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
 
     CopyStatement statement = new CopyStatement("COPY keyvalue FROM STDIN;", connection);
     statement.execute();
@@ -486,16 +479,17 @@ public class StatementTest {
     byte[] payload = "2 3\n".getBytes();
     MutationWriter mw = statement.getMutationWriter();
 
-    Exception thrown =
+    SpannerException thrown =
         Assert.assertThrows(
-            SQLException.class,
+            SpannerException.class,
             () -> {
               mw.addCopyData(connectionHandler, payload);
               mw.buildMutationList(connectionHandler);
             });
+    Assert.assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
     Assert.assertEquals(
-        thrown.getMessage(),
-        "Invalid COPY data: Row length mismatched. Expected 2 columns, but only found 1");
+        "INVALID_ARGUMENT: Invalid COPY data: Row length mismatched. Expected 2 columns, but only found 1",
+        thrown.getMessage());
 
     statement.close();
     Mockito.verify(resultSet, Mockito.times(0)).close();