Repository: GoogleCloudPlatform/pgadapter
Pull request #168 (2022-06-02 18:17:38)

feat: add support for extended query protocol batching
Adds support for extended query protocol batches and does a major internal refactor of PGAdapter:
1. The simple and extended query protocol now use the same code paths. This significantly reduces the amount of code duplication, and fixes the handling of implicit/explicit transactions in the extended query protocol.
2. The extended query protocol now buffers any messages and corresponding database statements in memory until it receives a flush or sync message. This allows database statements to be batched together when sent to Cloud Spanner if the frontend client sends multiple extended query protocol messages without a flush/sync. This feature is used by several native PG drivers to implement batching (JDBC and pgx).
3. Handling of implicit/explicit transactions and aborted transactions has been moved to a new `BackendConnection` class. This separates wire-protocol handling from the backend transaction and connection management.
4. The special `MatcherStatement` class that was only used for psql has been removed.
5. The simple query protocol now uses the extended query protocol internally by generating parse-bind-describe-execute messages.

codecov[bot]: # [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> Merging [#168](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (cdaec89) into [postgresql-dialect](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/commit/30ee4b945a269329082d1a89c42ab908984f478f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (30ee4b9) will **decrease** coverage by `3.45%`.
> The diff coverage is `80.75%`.

```diff
@@                   Coverage Diff                    @@
##             postgresql-dialect     #168      +/-   ##
========================================================
- Coverage                 77.64%   74.18%   -3.46%     
- Complexity                  958     1011      +53     
========================================================
  Files                        85       89       +4     
  Lines                      3283     3703     +420     
  Branches                    384      436      +52     
========================================================
+ Hits                       2549     2747     +198     
- Misses                      578      759     +181     
- Partials                    156      197      +41     
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `74.18% <80.75%> (-3.46%)` | :arrow_down: |

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

| [Impacted Files](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | Coverage Δ | |
|---|---|---|
| [...panner/pgadapter/wireprotocol/DescribeMessage.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3dpcmVwcm90b2NvbC9EZXNjcmliZU1lc3NhZ2UuamF2YQ==) | `71.42% <53.57%> (-9.01%)` | :arrow_down: |
| [...dapter/statements/IntermediatePortalStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvSW50ZXJtZWRpYXRlUG9ydGFsU3RhdGVtZW50LmphdmE=) | `89.18% <62.50%> (-7.37%)` | :arrow_down: |
| [...er/pgadapter/statements/IntermediateStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvSW50ZXJtZWRpYXRlU3RhdGVtZW50LmphdmE=) | `55.32% <66.66%> (-38.82%)` | :arrow_down: |
| [...pter/statements/IntermediatePreparedStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvSW50ZXJtZWRpYXRlUHJlcGFyZWRTdGF0ZW1lbnQuamF2YQ==) | `76.44% <75.00%> (-7.95%)` | :arrow_down: |
| [...panner/pgadapter/statements/BackendConnection.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvQmFja2VuZENvbm5lY3Rpb24uamF2YQ==) | `77.19% <77.19%> (ø)` | |
| [...ner/pgadapter/statements/SimpleQueryStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvU2ltcGxlUXVlcnlTdGF0ZW1lbnQuamF2YQ==) | `81.94% <81.94%> (ø)` | |
| [...apter/statements/ExtendedQueryProtocolHandler.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvRXh0ZW5kZWRRdWVyeVByb3RvY29sSGFuZGxlci5qYXZh) | `88.00% <88.00%> (ø)` | |
| [...d/spanner/pgadapter/wireprotocol/QueryMessage.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3dpcmVwcm90b2NvbC9RdWVyeU1lc3NhZ2UuamF2YQ==) | `89.74% <96.42%> (-2.71%)` | :arrow_down: |
| [...gle/cloud/spanner/pgadapter/ConnectionHandler.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL0Nvbm5lY3Rpb25IYW5kbGVyLmphdmE=) | `72.64% <100.00%> (-0.22%)` | :arrow_down: |
| [...ter/wireprotocol/AbstractQueryProtocolMessage.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3dpcmVwcm90b2NvbC9BYnN0cmFjdFF1ZXJ5UHJvdG9jb2xNZXNzYWdlLmphdmE=) | `100.00% <100.00%> (ø)` | |
| ... and [14 more](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform). Last update [30ee4b9...cdaec89](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/168?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform).


Patch:
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index ecc46a94..21b14949 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -21,7 +21,7 @@ jobs:
   integration-test:
     needs: [check-env]
     if: needs.check-env.outputs.has-key == 'true'
-    timeout-minutes: 30
+    timeout-minutes: 60
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v3
diff --git a/clirr-ignored-differences.xml b/clirr-ignored-differences.xml
index e9511a75..a43f9602 100644
--- a/clirr-ignored-differences.xml
+++ b/clirr-ignored-differences.xml
@@ -7,6 +7,12 @@
     <className>com/google/cloud/spanner/pgadapter/ProxyServer</className>
     <method>void run()</method>
   </difference>
+  <!-- Moved transaction status to BackendConnection. -->
+  <difference>
+    <differenceType>6001</differenceType>
+    <className>com/google/cloud/spanner/pgadapter/ConnectionHandler$ConnectionStatus</className>
+    <field>IDLE</field>
+  </difference>
 
   <!-- Add ignores for all sub packages, as these are considered internal. -->
   <difference>
diff --git a/pom.xml b/pom.xml
index 32452b9f..7a40a8f5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -304,6 +304,7 @@
         <artifactId>maven-surefire-plugin</artifactId>
         <version>3.0.0-M5</version>
         <configuration>
+          <forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
           <argLine>${surefire.jacoco.args}</argLine>
           <excludedGroups>${excludedTests}</excludedGroups>
           <reportNameSuffix>sponge_log</reportNameSuffix>
@@ -412,6 +413,7 @@
           </execution>
         </executions>
         <configuration>
+          <forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
           <argLine>${failsafe.jacoco.args}</argLine>
           <skipITs>${skipITs}</skipITs>
         </configuration>
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 f52584d6..cea50164 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
@@ -23,13 +23,13 @@ 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.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.ExtendedQueryProtocolHandler;
 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.wireoutput.ErrorResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse.Severity;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
-import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
 import com.google.cloud.spanner.pgadapter.wireoutput.TerminateResponse;
 import com.google.cloud.spanner.pgadapter.wireprotocol.BootstrapMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
@@ -83,6 +83,7 @@ public class ConnectionHandler extends Thread {
   private ConnectionMetadata connectionMetadata;
   private WireMessage message;
   private Connection spannerConnection;
+  private ExtendedQueryProtocolHandler extendedQueryProtocolHandler;
 
   ConnectionHandler(ProxyServer server, Socket socket) {
     super("ConnectionHandler-" + CONNECTION_HANDLER_ID_GENERATOR.incrementAndGet());
@@ -145,6 +146,7 @@ public class ConnectionHandler extends Thread {
       throw e;
     }
     this.spannerConnection = spannerConnection;
+    this.extendedQueryProtocolHandler = new ExtendedQueryProtocolHandler(this);
   }
 
   private String appendPropertiesToUrl(String url, Properties info) {
@@ -293,7 +295,7 @@ public class ConnectionHandler extends Thread {
     } else if (this.status == ConnectionStatus.COPY_IN) {
       new ErrorResponse(output, e, ErrorResponse.State.InternalError).send();
     } else {
-      this.status = ConnectionStatus.IDLE;
+      this.status = ConnectionStatus.AUTHENTICATED;
       new ErrorResponse(output, e, ErrorResponse.State.InternalError).send();
       new ReadyResponse(output, ReadyResponse.Status.IDLE).send();
     }
@@ -301,10 +303,8 @@ public class ConnectionHandler extends Thread {
 
   /** Closes portals and statements if the result of an execute was the end of a transaction. */
   public void cleanUp(IntermediateStatement statement) throws Exception {
-    for (int index = 0; index < statement.getStatementCount(); index++) {
-      if (!statement.isHasMoreData(index) && statement.isBound()) {
-        statement.close(index);
-      }
+    if (!statement.isHasMoreData() && statement.isBound()) {
+      statement.close();
     }
     // TODO when we have transaction data from jdbcConnection, close all portals if done
   }
@@ -451,6 +451,10 @@ public class ConnectionHandler extends Thread {
     return connectionMetadata;
   }
 
+  public ExtendedQueryProtocolHandler getExtendedQueryProtocolHandler() {
+    return extendedQueryProtocolHandler;
+  }
+
   public synchronized IntermediateStatement getActiveStatement() {
     return activeStatementsMap.get(this.connectionId);
   }
@@ -465,22 +469,10 @@ public class ConnectionHandler extends Thread {
 
   /** Status of a {@link ConnectionHandler} */
   public enum ConnectionStatus {
-    UNAUTHENTICATED(Status.IDLE),
-    IDLE(Status.IDLE),
-    TRANSACTION(Status.TRANSACTION),
-    COPY_IN(Status.IDLE),
-    TERMINATED(Status.IDLE),
-    TRANSACTION_ABORTED(Status.FAILED);
-
-    private final ReadyResponse.Status readyResponseStatus;
-
-    ConnectionStatus(ReadyResponse.Status readyResponseStatus) {
-      this.readyResponseStatus = readyResponseStatus;
-    }
-
-    public ReadyResponse.Status getReadyResponseStatus() {
-      return this.readyResponseStatus;
-    }
+    UNAUTHENTICATED,
+    AUTHENTICATED,
+    COPY_IN,
+    TERMINATED,
   }
 
   /**
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 cab50725..1aa88e01 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
@@ -49,6 +49,8 @@ class TimestampParser extends Parser<Timestamp> {
 
   private static final DateTimeFormatter TIMESTAMP_FORMATTER =
       new DateTimeFormatterBuilder()
+          .parseLenient()
+          .parseCaseInsensitive()
           .appendPattern("yyyy-MM-dd HH:mm:ss")
           .appendFraction(ChronoField.MICRO_OF_SECOND, 0, 9, true)
           .appendOffset("+HH:mm", "Z")
@@ -66,7 +68,7 @@ class TimestampParser extends Parser<Timestamp> {
     if (item != null) {
       switch (formatCode) {
         case TEXT:
-          String stringValue = new String(item, StandardCharsets.UTF_8);
+          String stringValue = toPGString(new String(item, StandardCharsets.UTF_8));
           TemporalAccessor temporalAccessor = TIMESTAMP_FORMATTER.parse(stringValue);
           this.item =
               Timestamp.ofTimeSecondsAndNanos(
@@ -94,12 +96,12 @@ class TimestampParser extends Parser<Timestamp> {
    */
   static boolean isTimestamp(String value) {
     Preconditions.checkNotNull(value);
-    return TIMESTAMP_PATTERN.matcher(value).matches();
+    return TIMESTAMP_PATTERN.matcher(toPGString(value)).matches();
   }
 
   @Override
   protected String stringParse() {
-    return this.item == null ? null : toPGString();
+    return this.item == null ? null : toPGString(this.item.toString());
   }
 
   @Override
@@ -126,11 +128,8 @@ class TimestampParser extends Parser<Timestamp> {
    *
    * @return a {@link String} that can be interpreted by PostgreSQL.
    */
-  private String toPGString() {
-    return this.item
-        .toString()
-        .replace(TIMESTAMP_SEPARATOR, EMPTY_SPACE)
-        .replace(ZERO_TIMEZONE, PG_ZERO_TIMEZONE);
+  private static String toPGString(String value) {
+    return value.replace(TIMESTAMP_SEPARATOR, EMPTY_SPACE).replace(ZERO_TIMEZONE, PG_ZERO_TIMEZONE);
   }
 
   @Override
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
new file mode 100644
index 00000000..5a03035b
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
@@ -0,0 +1,570 @@
+// 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.statements;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.SpannerBatchUpdateException;
+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.ParsedStatement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
+import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode;
+import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
+import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.SettableFuture;
+import java.util.LinkedList;
+import java.util.Objects;
+import java.util.concurrent.Future;
+
+/**
+ * This class emulates a backend PostgreSQL connection. Statements are buffered in memory until a
+ * flush/sync is received. This makes it possible to batch multiple statements together before
+ * sending these to Cloud Spanner. This class also keeps track of the transaction status of the
+ * connection.
+ */
+@InternalApi
+public class BackendConnection {
+  public static final String TRANSACTION_ABORTED_ERROR =
+      "current transaction is aborted, commands ignored until end of transaction block";
+
+  /**
+   * Connection state indicates whether the backend connection is idle, in a transaction or in an
+   * aborted transaction.
+   */
+  public enum ConnectionState {
+    /** Connection is idle, no transaction is active. */
+    IDLE(Status.IDLE),
+    /** An implicit or explicit transaction is active. */
+    TRANSACTION(Status.TRANSACTION),
+    /** The current transaction is aborted. All statements are refused until rollback. */
+    ABORTED(Status.FAILED);
+
+    private final ReadyResponse.Status readyResponseStatus;
+
+    ConnectionState(ReadyResponse.Status readyResponseStatus) {
+      this.readyResponseStatus = readyResponseStatus;
+    }
+
+    public ReadyResponse.Status getReadyResponseStatus() {
+      return this.readyResponseStatus;
+    }
+  }
+
+  /**
+   * {@link TransactionMode} indicates whether the current transaction on a backend connection is
+   * implicit or explicit. Implicit transactions are automatically committed/rolled back when a
+   * batch of statements finishes execution (i.e. when we receive a Sync message).
+   */
+  enum TransactionMode {
+    IMPLICIT,
+    EXPLICIT,
+  }
+
+  /**
+   * Buffered statements are kept in memory until a flush or sync message is received. This makes it
+   * possible to batch multiple statements together when sending them to Cloud Spanner.
+   */
+  abstract class BufferedStatement<T> {
+    final ParsedStatement parsedStatement;
+    final Statement statement;
+    final SettableFuture<T> result;
+
+    BufferedStatement(ParsedStatement parsedStatement, Statement statement) {
+      this.parsedStatement = parsedStatement;
+      this.statement = statement;
+      this.result = SettableFuture.create();
+    }
+
+    abstract void execute();
+
+    void checkConnectionState() {
+      // Only COMMIT or ROLLBACK is allowed if we are in an ABORTED transaction.
+      if (connectionState == ConnectionState.ABORTED
+          && !(isCommit(parsedStatement) || isRollback(parsedStatement))) {
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT, TRANSACTION_ABORTED_ERROR);
+      }
+    }
+  }
+
+  private final class Execute extends BufferedStatement<StatementResult> {
+    Execute(ParsedStatement parsedStatement, Statement statement) {
+      super(parsedStatement, statement);
+    }
+
+    @Override
+    void execute() {
+      try {
+        checkConnectionState();
+        if (connectionState == ConnectionState.ABORTED
+            && !spannerConnection.isInTransaction()
+            && (isRollback(parsedStatement) || isCommit(parsedStatement))) {
+          result.set(ROLLBACK_RESULT);
+        } else if (isBegin(parsedStatement) && spannerConnection.isInTransaction()) {
+          // Ignore the statement as it is a no-op to execute BEGIN when we are already in a
+          // transaction. TODO: Return a warning.
+          result.set(NO_RESULT);
+        } else if ((isCommit(parsedStatement) || isRollback(parsedStatement))
+            && !spannerConnection.isInTransaction()) {
+          // Ignore the statement as it is a no-op to execute COMMIT/ROLLBACK when we are not in a
+          // transaction. TODO: Return a warning.
+          result.set(NO_RESULT);
+        } else if (statement.getSql().isEmpty()) {
+          result.set(NO_RESULT);
+        } else {
+          result.set(spannerConnection.execute(statement));
+        }
+      } catch (Exception exception) {
+        result.setException(exception);
+        throw exception;
+      }
+    }
+  }
+
+  private static final StatementResult NO_RESULT = new NoResult();
+  private static final StatementResult ROLLBACK_RESULT = new NoResult("ROLLBACK");
+  private static final Statement ROLLBACK = Statement.of("ROLLBACK");
+
+  private ConnectionState connectionState = ConnectionState.IDLE;
+  private TransactionMode transactionMode = TransactionMode.IMPLICIT;
+  private final LinkedList<BufferedStatement<?>> bufferedStatements = new LinkedList<>();
+  private final Connection spannerConnection;
+  private final DdlTransactionMode ddlTransactionMode;
+
+  /**
+   * Creates a PG backend connection that uses the given Spanner {@link Connection} and {@link
+   * DdlTransactionMode}.
+   */
+  BackendConnection(Connection spannerConnection, DdlTransactionMode ddlTransactionMode) {
+    this.spannerConnection = spannerConnection;
+    this.ddlTransactionMode = ddlTransactionMode;
+  }
+
+  /** Returns the current connection state. */
+  public ConnectionState getConnectionState() {
+    return this.connectionState;
+  }
+
+  /**
+   * Buffers the given statement for execution on the backend connection when the next flush/sync
+   * message is received. The returned future will contain the result of the statement when
+   * execution has finished.
+   */
+  public Future<StatementResult> execute(ParsedStatement parsedStatement, Statement statement) {
+    Execute execute = new Execute(parsedStatement, statement);
+    bufferedStatements.add(execute);
+    return execute.result;
+  }
+
+  /** Flushes the buffered statements to Spanner. */
+  void flush() {
+    flush(false);
+  }
+
+  /**
+   * Flushes the buffered statements to Spanner and commits/rollbacks the implicit transaction (if
+   * any).
+   */
+  void sync() {
+    try {
+      flush(true);
+    } finally {
+      endImplicitTransaction();
+    }
+  }
+
+  /**
+   * Flushes all buffered statements to Cloud Spanner and commits/rolls back the transaction at the
+   * end if isSync=true.
+   */
+  private void flush(boolean isSync) {
+    int index = 0;
+    try {
+      while (index < bufferedStatements.size()) {
+        BufferedStatement<?> bufferedStatement = bufferedStatements.get(index);
+        maybeBeginImplicitTransaction(index, isSync);
+        // Prepare the connection for executing a DDL statement. This could include committing the
+        // current transaction, depending on the settings for execute DDL in transactions.
+        if (bufferedStatement.parsedStatement.isDdl()) {
+          prepareExecuteDdl(bufferedStatement);
+        }
+        boolean canUseBatch = false;
+        if (index < (getStatementCount() - 1)) {
+          StatementType statementType = getStatementType(index);
+          StatementType nextStatementType = getStatementType(index + 1);
+          canUseBatch = canBeBatchedTogether(statementType, nextStatementType);
+        }
+
+        if (canUseBatch) {
+          index += executeStatementsInBatch(index);
+        } else {
+          bufferedStatement.execute();
+
+          if (isBegin(index)) {
+            transactionMode = TransactionMode.EXPLICIT;
+            connectionState = ConnectionState.TRANSACTION;
+          } else if (isCommit(index) || isRollback(index)) {
+            transactionMode = TransactionMode.IMPLICIT;
+            connectionState = ConnectionState.IDLE;
+          }
+          index++;
+        }
+      }
+    } catch (Exception exception) {
+      connectionState = ConnectionState.ABORTED;
+      if (spannerConnection.isInTransaction()) {
+        spannerConnection.setStatementTag(null);
+        spannerConnection.execute(ROLLBACK);
+      }
+    } finally {
+      bufferedStatements.clear();
+    }
+  }
+
+  /** Starts an implicit transaction if that is necessary. */
+  private void maybeBeginImplicitTransaction(int index, boolean isSync) {
+    if (connectionState != ConnectionState.IDLE) {
+      return;
+    }
+
+    // Only start an implicit transaction if we have more than one statement left. Otherwise, just
+    // let the Spanner connection execute the statement in auto-commit mode.
+    if (isSync && index == bufferedStatements.size() - 1) {
+      return;
+    }
+    // Don't start an implicit transaction if this is already a transaction statement.
+    if (isTransactionStatement(index)) {
+      return;
+    }
+    // No need to start a transaction for DDL or client side statements.
+    if (bufferedStatements.get(index).parsedStatement.getType() == StatementType.DDL
+        || bufferedStatements.get(index).parsedStatement.getType() == StatementType.CLIENT_SIDE) {
+      return;
+    }
+    // If there are only DML statements left, those can be executed as an auto-commit dml batch.
+    if (isSync && hasOnlyDmlStatementsAfter(index)) {
+      return;
+    }
+
+    // We need to start an implicit transaction.
+    // Check if a read-only transaction suffices.
+    spannerConnection.beginTransaction();
+    if (isSync && !hasDmlStatementsAfter(index)) {
+      spannerConnection.setTransactionMode(
+          com.google.cloud.spanner.connection.TransactionMode.READ_ONLY_TRANSACTION);
+    }
+    transactionMode = TransactionMode.IMPLICIT;
+    connectionState = ConnectionState.TRANSACTION;
+  }
+
+  /** Ends the current implicit transaction (if any). */
+  private void endImplicitTransaction() {
+    // Only touch the transaction if it is an implicit transaction.
+    if (transactionMode != TransactionMode.IMPLICIT) {
+      return;
+    }
+
+    try {
+      if (spannerConnection.isInTransaction()) {
+        spannerConnection.setStatementTag(null);
+        if (connectionState == ConnectionState.ABORTED) {
+          spannerConnection.rollback();
+        } else {
+          spannerConnection.commit();
+        }
+      }
+    } finally {
+      connectionState = ConnectionState.IDLE;
+    }
+  }
+
+  /**
+   * Prepares the connection for executing a DDL statement. This can include committing the current
+   * transaction.
+   *
+   * <p>Executing the DDL statement may or may not be allowed depending on the state of the
+   * transaction and the selected {@link
+   * com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode}. The method
+   * will throw a {@link SpannerException} if executing a DDL statement at this point is not
+   * allowed.
+   */
+  private void prepareExecuteDdl(BufferedStatement<?> bufferedStatement) {
+    try {
+      // Single statements are simpler to check, so we do that in a separate check.
+      if (bufferedStatements.size() == 1) {
+        switch (ddlTransactionMode) {
+          case Single:
+          case Batch:
+          case AutocommitImplicitTransaction:
+            // Single DDL statements outside explicit transactions are always allowed. For a single
+            // statement, there can also not be an implicit transaction that needs to be committed.
+            if (transactionMode == TransactionMode.EXPLICIT) {
+              throw SpannerExceptionFactory.newSpannerException(
+                  ErrorCode.FAILED_PRECONDITION,
+                  "DDL statements are only allowed outside explicit transactions.");
+            }
+            // Fall-through to commit the transaction if necessary.
+          case AutocommitExplicitTransaction:
+            // DDL statements are allowed even in explicit transactions. Commit any transaction that
+            // might be active.
+            if (spannerConnection.isInTransaction()) {
+              spannerConnection.commit();
+              transactionMode = TransactionMode.IMPLICIT;
+            }
+        }
+        return;
+      }
+
+      // We are in a batch of statements.
+      switch (ddlTransactionMode) {
+        case Single:
+          throw SpannerExceptionFactory.newSpannerException(
+              ErrorCode.FAILED_PRECONDITION,
+              "DDL statements are only allowed outside batches and transactions.");
+        case Batch:
+          if (spannerConnection.isInTransaction()
+              || bufferedStatements.stream()
+                  .anyMatch(statement -> !statement.parsedStatement.isDdl())) {
+            throw SpannerExceptionFactory.newSpannerException(
+                ErrorCode.FAILED_PRECONDITION,
+                "DDL statements are not allowed in mixed batches or transactions.");
+          }
+          break;
+        case AutocommitImplicitTransaction:
+          if (spannerConnection.isInTransaction() && transactionMode != TransactionMode.IMPLICIT) {
+            throw SpannerExceptionFactory.newSpannerException(
+                ErrorCode.FAILED_PRECONDITION,
+                "DDL statements are only allowed outside explicit transactions.");
+          }
+          // Fallthrough to commit the transaction if necessary.
+        case AutocommitExplicitTransaction:
+          // Commit any transaction that might be active and allow executing the statement.
+          // Switch the execution state to implicit transaction.
+          if (spannerConnection.isInTransaction()) {
+            spannerConnection.commit();
+            transactionMode = TransactionMode.IMPLICIT;
+          }
+      }
+    } catch (SpannerException exception) {
+      bufferedStatement.result.setException(exception);
+      throw exception;
+    }
+  }
+
+  private boolean isBegin(int index) {
+    return isBegin(bufferedStatements.get(index).parsedStatement);
+  }
+
+  private boolean isBegin(ParsedStatement parsedStatement) {
+    return parsedStatement.getType() == StatementType.CLIENT_SIDE
+        && parsedStatement.getClientSideStatementType() == ClientSideStatementType.BEGIN;
+  }
+
+  private boolean isCommit(int index) {
+    return isCommit(bufferedStatements.get(index).parsedStatement);
+  }
+
+  private boolean isCommit(ParsedStatement parsedStatement) {
+    return parsedStatement.getType() == StatementType.CLIENT_SIDE
+        && parsedStatement.getClientSideStatementType() == ClientSideStatementType.COMMIT;
+  }
+
+  private boolean isRollback(int index) {
+    return isRollback(bufferedStatements.get(index).parsedStatement);
+  }
+
+  private boolean isRollback(ParsedStatement parsedStatement) {
+    return parsedStatement.getType() == StatementType.CLIENT_SIDE
+        && parsedStatement.getClientSideStatementType() == ClientSideStatementType.ROLLBACK;
+  }
+
+  private boolean isTransactionStatement(int index) {
+    return isBegin(index) || isCommit(index) || isRollback(index);
+  }
+
+  private boolean hasDmlStatementsAfter(int index) {
+    return bufferedStatements.subList(index, bufferedStatements.size()).stream()
+        .anyMatch(statement -> statement.parsedStatement.getType() == StatementType.UPDATE);
+  }
+
+  private boolean hasOnlyDmlStatementsAfter(int index) {
+    return bufferedStatements.subList(index, bufferedStatements.size()).stream()
+        .allMatch(statement -> statement.parsedStatement.getType() == StatementType.UPDATE);
+  }
+
+  private int getStatementCount() {
+    return bufferedStatements.size();
+  }
+
+  private StatementType getStatementType(int index) {
+    return bufferedStatements.get(index).parsedStatement.getType();
+  }
+
+  private boolean canBeBatchedTogether(StatementType statementType1, StatementType statementType2) {
+    if (Objects.equals(statementType1, StatementType.QUERY)
+        || Objects.equals(statementType1, StatementType.CLIENT_SIDE)) {
+      return false;
+    }
+    return Objects.equals(statementType1, statementType2);
+  }
+
+  /**
+   * Executes the statements from fromIndex in a DML/DDL batch. The batch will consist of all
+   * statements from fromIndex till the first statement that is of a different type than the
+   * statement at fromIndex. That is; If the first statement is a DML statement, the batch will
+   * contain all statements that follow until it encounters a statement that is not a DML statement.
+   * The same also applies to DDL statements. Query statements and other statements can not be
+   * batched.
+   *
+   * @param fromIndex The index of the statements array where the batch should start
+   * @return The number of statements included in the batch.
+   */
+  private int executeStatementsInBatch(int fromIndex) {
+    Preconditions.checkArgument(fromIndex < getStatementCount() - 1);
+    Preconditions.checkArgument(
+        canBeBatchedTogether(getStatementType(fromIndex), getStatementType(fromIndex + 1)));
+    StatementType batchType = getStatementType(fromIndex);
+    switch (batchType) {
+      case UPDATE:
+        spannerConnection.startBatchDml();
+        break;
+      case DDL:
+        spannerConnection.startBatchDdl();
+        break;
+      default:
+        throw SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT, "Statement type is not supported for batching");
+    }
+    int index = fromIndex;
+    while (index < getStatementCount()) {
+      StatementType statementType = getStatementType(index);
+      if (canBeBatchedTogether(batchType, statementType)) {
+        spannerConnection.execute(bufferedStatements.get(index).statement);
+        index++;
+      } else {
+        // End the batch here, as the statement type on this index can not be batched together with
+        // the other statements in the batch.
+        break;
+      }
+    }
+    try {
+      long[] counts = spannerConnection.runBatch();
+      updateBatchResultCount(fromIndex, counts);
+    } catch (SpannerBatchUpdateException batchUpdateException) {
+      long[] counts = batchUpdateException.getUpdateCounts();
+      updateBatchResultCount(fromIndex, counts);
+      Execute failedExecute = (Execute) bufferedStatements.get(fromIndex + counts.length);
+      failedExecute.result.setException(batchUpdateException);
+      throw batchUpdateException;
+    }
+    return index - fromIndex;
+  }
+
+  /** Updates the results of the buffered statements after finishing executing a batch. */
+  private void updateBatchResultCount(int fromIndex, long[] updateCounts) {
+    for (int index = fromIndex; index < fromIndex + updateCounts.length; index++) {
+      Execute execute = (Execute) bufferedStatements.get(index);
+      if (execute.parsedStatement.getType() == StatementType.DDL) {
+        execute.result.set(new NoResult());
+      } else {
+        execute.result.set(new UpdateCount(updateCounts[index - fromIndex]));
+      }
+    }
+  }
+
+  /**
+   * {@link StatementResult} implementation for statements that do not return anything (e.g. DDL).
+   */
+  @InternalApi
+  public static final class NoResult implements StatementResult {
+    private final String commandTag;
+
+    private NoResult() {
+      this.commandTag = null;
+    }
+
+    private NoResult(String commandTag) {
+      this.commandTag = commandTag;
+    }
+
+    @Override
+    public ResultType getResultType() {
+      return ResultType.NO_RESULT;
+    }
+
+    public boolean hasCommandTag() {
+      return this.commandTag != null;
+    }
+
+    public String getCommandTag() {
+      return this.commandTag;
+    }
+
+    @Override
+    public ClientSideStatementType getClientSideStatementType() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public ResultSet getResultSet() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Long getUpdateCount() {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  /**
+   * Implementation of {@link StatementResult} for statements that return an update count (e.g.
+   * DML).
+   */
+  @InternalApi
+  public static final class UpdateCount implements StatementResult {
+    private final Long updateCount;
+
+    public UpdateCount(Long updateCount) {
+      this.updateCount = updateCount;
+    }
+
+    @Override
+    public ResultType getResultType() {
+      return ResultType.UPDATE_COUNT;
+    }
+
+    @Override
+    public ClientSideStatementType getClientSideStatementType() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public ResultSet getResultSet() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Long getUpdateCount() {
+      return updateCount;
+    }
+  }
+}
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 288502c6..dfd0a5cd 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
@@ -71,7 +71,7 @@ public class CopyStatement extends IntermediateStatement {
 
   public Exception getException() {
     // Do not clear exceptions on a CopyStatement.
-    return this.exceptions[0];
+    return this.exception;
   }
 
   @Override
@@ -83,7 +83,7 @@ public class CopyStatement extends IntermediateStatement {
     super.close();
   }
 
-  public Long getUpdateCount() {
+  public long getUpdateCount() {
     try {
       return updateCount.get();
     } catch (ExecutionException e) {
@@ -94,7 +94,7 @@ public class CopyStatement extends IntermediateStatement {
   }
 
   @Override
-  public StatementType getStatementType(int index) {
+  public StatementType getStatementType() {
     return StatementType.UPDATE;
   }
 
@@ -289,19 +289,14 @@ public class CopyStatement extends IntermediateStatement {
     return 0;
   }
 
-  public void handleExecutionException(SpannerException e) {
-    this.handleExecutionException(0, e);
-  }
-
   @Override
-  public void handleExecutionException(int index, SpannerException exception) {
+  public void handleExecutionException(SpannerException exception) {
     executor.shutdownNow();
-    super.handleExecutionException(index, exception);
+    super.handleExecutionException(exception);
   }
 
-  @Override
   public void execute() {
-    this.executedCount++;
+    this.executed = true;
     try {
       parseCopyStatement();
       queryInformationSchema();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
new file mode 100644
index 00000000..7d6afc25
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
@@ -0,0 +1,98 @@
+// 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.statements;
+
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.wireprotocol.AbstractQueryProtocolMessage;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * Handles the message flow for the extended query protocol. Wire-protocol messages are buffered in
+ * memory until a flush/sync is received.
+ */
+public class ExtendedQueryProtocolHandler {
+  private final LinkedList<AbstractQueryProtocolMessage> messages = new LinkedList<>();
+  private final BackendConnection backendConnection;
+
+  /** Creates an {@link ExtendedQueryProtocolHandler} for the given connection. */
+  public ExtendedQueryProtocolHandler(ConnectionHandler connectionHandler) {
+    this.backendConnection =
+        new BackendConnection(
+            connectionHandler.getSpannerConnection(),
+            connectionHandler.getServer().getOptions().getDdlTransactionMode());
+  }
+
+  /** Constructor only intended for testing. */
+  @VisibleForTesting
+  public ExtendedQueryProtocolHandler(BackendConnection backendConnection) {
+    this.backendConnection = backendConnection;
+  }
+
+  /** Returns the backend PG connection for this query handler. */
+  public BackendConnection getBackendConnection() {
+    return backendConnection;
+  }
+
+  /** Returns a copy of the currently buffered messages in this handler. */
+  @VisibleForTesting
+  List<AbstractQueryProtocolMessage> getMessages() {
+    return new ArrayList<>(messages);
+  }
+
+  /**
+   * Buffer an extended query protocol message for execution when the next flush/sync message is
+   * received.
+   */
+  public void buffer(AbstractQueryProtocolMessage message) {
+    messages.add(message);
+  }
+
+  /**
+   * Flushes the current queue of messages. Any pending database statements are first executed,
+   * before sending the wire-protocol responses to the frontend. A flush does not commit the
+   * implicit transaction (if any).
+   */
+  public void flush() throws Exception {
+    backendConnection.flush();
+    flushMessages();
+  }
+
+  /**
+   * Flushes the current queue of messages and commits the implicit transaction (if any). Any
+   * pending database statements are first executed, before sending the wire-protocol responses to
+   * the frontend.
+   */
+  public void sync() throws Exception {
+    backendConnection.sync();
+    flushMessages();
+  }
+
+  /** Flushes the wire-protocol messages to the frontend. */
+  private void flushMessages() throws Exception {
+    try {
+      for (AbstractQueryProtocolMessage message : messages) {
+        message.flush();
+        if (message.isReturnedErrorResponse()) {
+          break;
+        }
+      }
+    } finally {
+      messages.clear();
+    }
+  }
+}
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 e8b1dc6e..0e758886 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
@@ -15,16 +15,16 @@
 package com.google.cloud.spanner.pgadapter.statements;
 
 import com.google.api.core.InternalApi;
-import com.google.cloud.spanner.ResultSet;
-import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.metadata.DescribeMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.DescribePortalMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.common.util.concurrent.Futures;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Future;
 import java.util.logging.Logger;
 
 /**
@@ -33,7 +33,6 @@ import java.util.logging.Logger;
  */
 @InternalApi
 public class IntermediatePortalStatement extends IntermediatePreparedStatement {
-
   private static final Logger logger =
       Logger.getLogger(IntermediatePreparedStatement.class.getName());
   protected List<Short> parameterFormatCodes;
@@ -84,19 +83,17 @@ public class IntermediatePortalStatement extends IntermediatePreparedStatement {
   }
 
   @Override
-  public DescribeMetadata describe() {
-    try {
-      // Pre-emptively execute the statement, even though it is only asked to be described. This is
-      // a lot more efficient than taking two round trips to the server, and getting a
-      // DescribePortal message without a following Execute message is extremely rare, as that would
-      // only happen if the client is ill-behaved, or if the client crashes between the
-      // DescribePortal and Execute.
-      ResultSet statementResult = connection.executeQuery(this.statement);
-      setStatementResult(0, statementResult);
-      return new DescribePortalMetadata(statementResult);
-    } catch (SpannerException exception) {
-      handleExecutionExceptionAndTransactionStatus(0, exception);
-      throw exception;
-    }
+  public Future<DescribePortalMetadata> describeAsync(BackendConnection backendConnection) {
+    // Pre-emptively execute the statement, even though it is only asked to be described. This is
+    // a lot more efficient than taking two round trips to the server, and getting a
+    // DescribePortal message without a following Execute message is extremely rare, as that would
+    // only happen if the client is ill-behaved, or if the client crashes between the
+    // DescribePortal and Execute.
+    Future<StatementResult> statementResultFuture =
+        backendConnection.execute(this.parsedStatement, this.statement);
+    setFutureStatementResult(statementResultFuture);
+    return Futures.lazyTransform(
+        statementResultFuture,
+        statementResult -> new DescribePortalMetadata(statementResult.getResultSet()));
   }
 }
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 11a58923..15ffe825 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
@@ -15,19 +15,14 @@
 package com.google.cloud.spanner.pgadapter.statements;
 
 import com.google.api.core.InternalApi;
-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.Type.StructField;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
-import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
 import com.google.cloud.spanner.connection.Connection;
-import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeStatementMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
@@ -41,7 +36,6 @@ import java.util.Arrays;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Set;
-import java.util.logging.Logger;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 import org.postgresql.core.Oid;
@@ -51,9 +45,6 @@ import org.postgresql.core.Oid;
  */
 @InternalApi
 public class IntermediatePreparedStatement extends IntermediateStatement {
-  private static final Logger logger =
-      Logger.getLogger(IntermediatePreparedStatement.class.getName());
-
   private final String name;
   protected int[] parameterDataTypes;
   protected Statement statement;
@@ -93,52 +84,12 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
   }
 
   @Override
-  public void execute() {
-    // TODO(230579451): Refactor to use ClientSideStatement information.
-    if (connectionHandler.getStatus() == ConnectionStatus.TRANSACTION_ABORTED) {
-      handleTransactionAborted();
-      return;
-    }
-
+  public void executeAsync(BackendConnection backendConnection) {
     // If the portal has already been described, the statement has already been executed, and we
     // don't need to do that once more.
-    if (getStatementResult(0) == null) {
-      this.executedCount++;
-      try {
-        if (!connection.isInTransaction()
-            // TODO(230579451): Refactor to use ClientSideStatement information.
-            && this.parsedStatement.getType().equals(StatementType.CLIENT_SIDE)
-            && (this.commands.get(0).equals("ROLLBACK") || this.commands.get(0).equals("COMMIT"))) {
-          // TODO(230579929): Return warning that no transaction if connection status == IDLE.
-          connectionHandler.setStatus(ConnectionStatus.IDLE);
-        } else {
-          StatementResult result = connection.execute(this.statement);
-          this.updateResultCount(0, result);
-          connectionHandler.setStatus(
-              connection.isInTransaction() ? ConnectionStatus.TRANSACTION : ConnectionStatus.IDLE);
-        }
-      } catch (SpannerException exception) {
-        handleExecutionExceptionAndTransactionStatus(0, exception);
-      }
-    }
-  }
-
-  private void handleTransactionAborted() {
-    // TODO(230579451): Refactor to use ClientSideStatement information.
-    String command = getCommand(0);
-    if ("COMMIT".equals(command) || "ROLLBACK".equals(command)) {
-      connectionHandler.setStatus(ConnectionStatus.IDLE);
-      // COMMIT rollbacks aborted transaction
-      commandTags.set(0, "ROLLBACK");
-      if (connection.isInTransaction()) {
-        connection.rollback();
-      }
-    } else {
-      handleExecutionException(
-          executedCount,
-          SpannerExceptionFactory.newSpannerException(
-              ErrorCode.INVALID_ARGUMENT, TRANSACTION_ABORTED_ERROR));
-      executedCount++;
+    if (futureStatementResult == null && getStatementResult() == null) {
+      this.executed = true;
+      setFutureStatementResult(backendConnection.execute(parsedStatement, statement));
     }
   }
 
@@ -175,7 +126,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
   }
 
   @Override
-  public DescribeMetadata describe() {
+  public DescribeMetadata<?> describe() {
     Set<String> parameters =
         ImmutableSortedSet.<String>orderedBy(Comparator.comparing(o -> o.substring(1)))
             .addAll(PARSER.getQueryParameters(this.parsedStatement.getSqlWithoutComments()))
@@ -194,8 +145,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
       } else if (parameters.size() != this.parameterDataTypes.length
           || Arrays.stream(this.parameterDataTypes).anyMatch(p -> p == 0)) {
         // Note: We are only asking the backend to parse the types if there is at least one
-        // parameter
-        // with unspecified type. Otherwise, we will rely on the types given in PARSE.
+        // parameter with unspecified type. Otherwise, we will rely on the types given in PARSE.
 
         // Transform the statement into a select statement that selects the parameters, and then
         // extract the types from the result set metadata.
@@ -203,8 +153,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
         if (selectParamsStatement == null) {
           // The transformation failed. Just rely on the types given in the PARSE message. If the
           // transformation failed because the statement was malformed, the backend will catch that
-          // at
-          // a later stage.
+          // at a later stage.
           describeFailed = true;
           ensureParameterLength(parameters.size());
         } else {
@@ -226,11 +175,9 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
           && (describeFailed || !Strings.isNullOrEmpty(this.name))) {
         // Let the backend analyze the statement if it is a named prepared statement or if the query
         // that was used to determine the parameter types failed, so we can return a reasonable
-        // error
-        // message if the statement is invalid. If it is the unnamed statement or getting the param
-        // types succeeded, we will let the following EXECUTE message handle that, instead of
-        // sending
-        // the statement twice to the backend.
+        // error message if the statement is invalid. If it is the unnamed statement or getting the
+        // param types succeeded, we will let the following EXECUTE message handle that, instead of
+        // sending the statement twice to the backend.
         connection.analyzeUpdate(
             Statement.of(this.parsedStatement.getSqlWithoutComments()), QueryAnalyzeMode.PLAN);
       }
@@ -284,7 +231,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
    * statement.
    */
   private Statement transformDmlToSelectParams(Set<String> parameters) {
-    switch (getCommand(0)) {
+    switch (getCommand()) {
       case "INSERT":
         return transformInsertToSelectParams(
             this.connection, this.parsedStatement.getSqlWithoutComments(), parameters);
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
deleted file mode 100644
index a49d3817..00000000
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/MatcherStatement.java
+++ /dev/null
@@ -1,66 +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.statements;
-
-import com.google.api.core.InternalApi;
-import com.google.cloud.spanner.Statement;
-import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.commands.Command;
-import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
-
-/**
- * Meant to be utilized when running as a proxy for any interactive terminal tool either PSQL or
- * custom versions. Not meant for production runs. Generally here, all statements will take some
- * penalty by running matchers to determine whether they belong to a specific meta-command. If
- * matches, translates the command into something Spanner can handle.
- */
-@InternalApi
-public class MatcherStatement extends IntermediateStatement {
-
-  public MatcherStatement(
-      OptionsMetadata options,
-      ParsedStatement parsedStatement,
-      ConnectionHandler connectionHandler) {
-    super(options, translateSQL(parsedStatement, connectionHandler), connectionHandler);
-  }
-
-  @Override
-  public void execute() {
-    super.execute();
-  }
-
-  /**
-   * Translate a Postgres Specific command into something Spanner can handle. Currently, this is
-   * only concerned with PSQL specific meta-commands.
-   *
-   * @param parsedStatement The SQL statement to be translated.
-   * @return The translated SQL statement if it matches any {@link Command} statement. Otherwise
-   *     gives out the original Statement.
-   */
-  private static ParsedStatement translateSQL(
-      ParsedStatement parsedStatement, ConnectionHandler connectionHandler) {
-    for (Command currentCommand :
-        Command.getCommands(
-            parsedStatement.getSqlWithoutComments(),
-            connectionHandler.getSpannerConnection(),
-            connectionHandler.getServer().getOptions().getCommandMetadataJSON())) {
-      if (currentCommand.is()) {
-        return PARSER.parse(Statement.of(currentCommand.translate()));
-      }
-    }
-    return parsedStatement;
-  }
-}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleQueryStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleQueryStatement.java
new file mode 100644
index 00000000..ba183623
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleQueryStatement.java
@@ -0,0 +1,184 @@
+// 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.statements;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.PostgreSQLStatementParser;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.commands.Command;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.wireprotocol.BindMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.ManuallyCreatedToken;
+import com.google.cloud.spanner.pgadapter.wireprotocol.DescribeMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ExecuteMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.SyncMessage;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import java.util.List;
+
+/**
+ * Class that represents a simple query protocol statement. This statement can contain multiple
+ * semi-colon separated SQL statements. The simple query protocol internally uses the extended query
+ * protocol to execute the statement(s) in the SQL string, but does not return all the messages that
+ * would have been returned by the extended protocol.
+ */
+@InternalApi
+public class SimpleQueryStatement {
+  private static final PostgreSQLStatementParser PARSER =
+      (PostgreSQLStatementParser) AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
+
+  private final ConnectionHandler connectionHandler;
+  private final OptionsMetadata options;
+  private final ImmutableList<ParsedStatement> statements;
+
+  private static final char STATEMENT_DELIMITER = ';';
+  private static final char SINGLE_QUOTE = '\'';
+  private static final char DOUBLE_QUOTE = '"';
+
+  public SimpleQueryStatement(
+      OptionsMetadata options,
+      ParsedStatement parsedStatement,
+      ConnectionHandler connectionHandler) {
+    this.connectionHandler = connectionHandler;
+    this.options = options;
+    this.statements = parseStatements(parsedStatement);
+  }
+
+  public void execute() throws Exception {
+    // Do a Parse-Describe-Bind-Execute round-trip for each statement in the query string.
+    // Finish with a Sync to close any implicit transaction and to return the results.
+    for (ParsedStatement statement : this.statements) {
+      if (options.requiresMatcher()) {
+        statement = translatePotentialMetadataCommand(statement, connectionHandler);
+      }
+      statement = replaceKnownUnsupportedQueries(this.options, statement);
+      new ParseMessage(connectionHandler, statement).send();
+      new BindMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN).send();
+      new DescribeMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN).send();
+      new ExecuteMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN).send();
+    }
+    new SyncMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN).send();
+  }
+
+  /** Replaces any known unsupported query (e.g. JDBC metadata queries). */
+  static ParsedStatement replaceKnownUnsupportedQueries(
+      OptionsMetadata options, ParsedStatement parsedStatement) {
+    if (options.isReplaceJdbcMetadataQueries()
+        && JdbcMetadataStatementHelper.isPotentialJdbcMetadataStatement(
+            parsedStatement.getSqlWithoutComments())) {
+      return PARSER.parse(
+          Statement.of(
+              JdbcMetadataStatementHelper.replaceJdbcMetadataStatement(
+                  parsedStatement.getSqlWithoutComments())));
+    }
+    return parsedStatement;
+  }
+
+  /**
+   * Translate a Postgres Specific command into something Spanner can handle. Currently, this is
+   * only concerned with PSQL specific meta-commands.
+   *
+   * @param parsedStatement The SQL statement to be translated.
+   * @return The translated SQL statement if it matches any {@link Command} statement. Otherwise,
+   *     returns the original Statement.
+   */
+  @VisibleForTesting
+  static ParsedStatement translatePotentialMetadataCommand(
+      ParsedStatement parsedStatement, ConnectionHandler connectionHandler) {
+    for (Command currentCommand :
+        Command.getCommands(
+            parsedStatement.getSqlWithoutComments(),
+            connectionHandler.getSpannerConnection(),
+            connectionHandler.getServer().getOptions().getCommandMetadataJSON())) {
+      if (currentCommand.is()) {
+        return PARSER.parse(Statement.of(currentCommand.translate()));
+      }
+    }
+    return parsedStatement;
+  }
+
+  static ImmutableList<String> splitStatements(String sql) {
+    // First check trivial cases with only one statement.
+    int firstIndexOfDelimiter = sql.indexOf(STATEMENT_DELIMITER);
+    if (firstIndexOfDelimiter == -1) {
+      return ImmutableList.of(sql);
+    }
+    if (firstIndexOfDelimiter == sql.length() - 1) {
+      return ImmutableList.of(sql.substring(0, sql.length() - 1));
+    }
+
+    ImmutableList.Builder<String> builder = ImmutableList.builder();
+    // TODO: Fix this parsing, as it does not take all types of quotes into consideration.
+    boolean insideQuotes = false;
+    char quoteChar = 0;
+    int index = 0;
+    for (int i = 0; i < sql.length(); ++i) {
+      if (insideQuotes) {
+        if (sql.charAt(i) == quoteChar) {
+          if (sql.length() > (i + 1) && sql.charAt(i + 1) == quoteChar) {
+            // This is an escaped quote. Skip one ahead.
+            i++;
+          } else {
+            insideQuotes = false;
+          }
+        }
+      } else {
+        if (sql.charAt(i) == SINGLE_QUOTE || sql.charAt(i) == DOUBLE_QUOTE) {
+          quoteChar = sql.charAt(i);
+          insideQuotes = true;
+        } else {
+          // skip semicolon inside quotes
+          if (sql.charAt(i) == STATEMENT_DELIMITER) {
+            String stmt = sql.substring(index, i).trim();
+            // Statements with only ';' character are empty and dropped.
+            if (stmt.length() > 0) {
+              builder.add(stmt);
+            }
+            index = i + 1;
+          }
+        }
+      }
+    }
+
+    if (index < sql.length()) {
+      builder.add(sql.substring(index).trim());
+    }
+    return builder.build();
+  }
+
+  protected static ImmutableList<ParsedStatement> parseStatements(ParsedStatement stmt) {
+    String sql = stmt.getSqlWithoutComments();
+    Preconditions.checkNotNull(sql);
+    ImmutableList.Builder<ParsedStatement> builder = ImmutableList.builder();
+    for (String statement : splitStatements(sql)) {
+      builder.add(PARSER.parse(Statement.of(statement)));
+    }
+    return builder.build();
+  }
+
+  public List<ParsedStatement> getStatements() {
+    return this.statements;
+  }
+
+  public String getStatement(int index) {
+    return this.statements.get(index).getSqlWithoutComments();
+  }
+}
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 5f831acc..01e943db 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
@@ -39,14 +39,12 @@ public class DataRowResponse extends WireOutput {
 
   public DataRowResponse(
       DataOutputStream output,
-      int resultIndex,
       IntermediateStatement statement,
       OptionsMetadata options,
-      QueryMode mode)
-      throws Exception {
+      QueryMode mode) {
     super(output, 0);
     this.mode = mode;
-    this.resultSet = statement.getStatementResult(resultIndex);
+    this.resultSet = statement.getStatementResult().getResultSet();
     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.
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/AbstractQueryProtocolMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/AbstractQueryProtocolMessage.java
new file mode 100644
index 00000000..3f2a9d1d
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/AbstractQueryProtocolMessage.java
@@ -0,0 +1,66 @@
+// 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.wireprotocol;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
+import com.google.cloud.spanner.pgadapter.statements.ExtendedQueryProtocolHandler;
+import java.io.IOException;
+
+/**
+ * Abstract base class for extended query protocol messages. These can be buffered in memory locally
+ * until a flush or sync message is received. Any corresponding database statement will only be
+ * executed once a flush/sync is received.
+ */
+@InternalApi
+public abstract class AbstractQueryProtocolMessage extends ControlMessage {
+  private final ExtendedQueryProtocolHandler handler;
+  protected final QueryMode queryMode;
+  private boolean returnedErrorResponse;
+
+  AbstractQueryProtocolMessage(ConnectionHandler connection) throws IOException {
+    super(connection);
+    this.handler = connection.getExtendedQueryProtocolHandler();
+    this.queryMode = QueryMode.EXTENDED;
+  }
+
+  AbstractQueryProtocolMessage(
+      ConnectionHandler connection, int length, ManuallyCreatedToken manuallyCreatedToken) {
+    super(connection, length, manuallyCreatedToken);
+    this.handler = connection.getExtendedQueryProtocolHandler();
+    this.queryMode = QueryMode.SIMPLE;
+  }
+
+  @Override
+  protected final void sendPayload() throws Exception {
+    this.buffer(handler.getBackendConnection());
+    handler.buffer(this);
+  }
+
+  abstract void buffer(BackendConnection backendConnection) throws Exception;
+
+  public abstract void flush() throws Exception;
+
+  public boolean isReturnedErrorResponse() {
+    return returnedErrorResponse;
+  }
+
+  protected void handleError(Exception exception) throws Exception {
+    super.handleError(exception);
+    this.returnedErrorResponse = true;
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
index 2c07e78e..b3d952b1 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
@@ -16,8 +16,10 @@ package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
 import com.google.cloud.spanner.pgadapter.wireoutput.BindCompleteResponse;
+import com.google.common.collect.ImmutableList;
 import java.text.MessageFormat;
 import java.util.Arrays;
 import java.util.List;
@@ -28,16 +30,17 @@ import java.util.List;
  * unless it fails.
  */
 @InternalApi
-public class BindMessage extends ControlMessage {
+public class BindMessage extends AbstractQueryProtocolMessage {
   protected static final char IDENTIFIER = 'B';
 
-  private String portalName;
-  private String statementName;
-  private List<Short> formatCodes;
-  private List<Short> resultFormatCodes;
-  private byte[][] parameters;
-  private IntermediatePreparedStatement statement;
+  private final String portalName;
+  private final String statementName;
+  private final List<Short> formatCodes;
+  private final List<Short> resultFormatCodes;
+  private final byte[][] parameters;
+  private final IntermediatePreparedStatement statement;
 
+  /** Constructor for Bind messages that are received from the front-end. */
   public BindMessage(ConnectionHandler connection) throws Exception {
     super(connection);
     this.portalName = this.readString();
@@ -48,18 +51,32 @@ public class BindMessage extends ControlMessage {
     this.statement = connection.getStatement(this.statementName);
   }
 
-  /**
-   * Given the prepared statement, bind it and save it locally.
-   *
-   * @throws Exception If the binding fails.
-   */
+  /** Constructor for Bind messages that are constructed to execute a Query message. */
+  public BindMessage(ConnectionHandler connection, ManuallyCreatedToken manuallyCreatedToken) {
+    super(connection, 4, manuallyCreatedToken);
+    this.portalName = "";
+    this.statementName = "";
+    this.formatCodes = ImmutableList.of();
+    this.resultFormatCodes = ImmutableList.of();
+    this.parameters = new byte[0][];
+    this.statement = connection.getStatement("");
+  }
+
+  /** Given the prepared statement, bind it and save it locally. */
   @Override
-  protected void sendPayload() throws Exception {
+  void buffer(BackendConnection backendConnection) {
     this.connection.registerPortal(
         this.portalName,
         this.statement.bind(
             this.portalName, this.parameters, this.formatCodes, this.resultFormatCodes));
-    new BindCompleteResponse(this.outputStream).send();
+  }
+
+  @Override
+  public void flush() throws Exception {
+    // The simple query protocol does not expect a BindComplete response.
+    if (isExtendedProtocol()) {
+      new BindCompleteResponse(this.outputStream).send();
+    }
   }
 
   @Override
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 8000f145..26f64cad 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
@@ -23,9 +23,12 @@ import com.google.cloud.spanner.pgadapter.metadata.SendResultSetState;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.cloud.spanner.pgadapter.wireoutput.CommandCompleteResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.DataRowResponse;
+import com.google.cloud.spanner.pgadapter.wireoutput.EmptyQueryResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse.State;
 import com.google.cloud.spanner.pgadapter.wireoutput.PortalSuspendedResponse;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -37,9 +40,29 @@ import java.util.List;
  */
 @InternalApi
 public abstract class ControlMessage extends WireMessage {
+  /**
+   * Token that is used to mark {@link ControlMessage}s that are manually created to execute a
+   * {@link QueryMessage}.
+   */
+  public enum ManuallyCreatedToken {
+    MANUALLY_CREATED_TOKEN
+  }
+
+  private final ManuallyCreatedToken manuallyCreatedToken;
 
   public ControlMessage(ConnectionHandler connection) throws IOException {
     super(connection, connection.getConnectionMetadata().getInputStream().readInt());
+    this.manuallyCreatedToken = null;
+  }
+
+  /** Constructor for manually created Control messages. */
+  protected ControlMessage(ConnectionHandler connection, int length, ManuallyCreatedToken token) {
+    super(connection, length);
+    this.manuallyCreatedToken = token;
+  }
+
+  protected boolean isExtendedProtocol() {
+    return manuallyCreatedToken == null;
   }
 
   /**
@@ -133,16 +156,11 @@ public abstract class ControlMessage extends WireMessage {
   /**
    * Takes an Exception Object and relates its results to the user within the client.
    *
-   * @param e The exception to be related.
+   * @param exception The exception to be related.
    * @throws Exception if there is some issue in the sending of the error messages.
    */
-  protected void handleError(Exception e) throws Exception {
-    new ErrorResponse(this.outputStream, e, State.RaiseException).send(false);
-    if (connection.getSpannerConnection().isInTransaction()) {
-      connection.getSpannerConnection().rollbackAsync();
-      connection.setStatus(ConnectionStatus.TRANSACTION_ABORTED);
-    }
-    this.outputStream.flush();
+  protected void handleError(Exception exception) throws Exception {
+    new ErrorResponse(this.outputStream, exception, State.RaiseException).send(false);
   }
 
   /**
@@ -154,37 +172,40 @@ public abstract class ControlMessage extends WireMessage {
    *
    * <p>NOTE: This method does not flush the output stream.
    */
-  public boolean sendSpannerResult(
-      int resultIndex, IntermediateStatement statement, QueryMode mode, long maxRows)
+  public void sendSpannerResult(IntermediateStatement statement, QueryMode mode, long maxRows)
       throws Exception {
-    String command = statement.getCommandTag(resultIndex);
-    switch (statement.getStatementType(resultIndex)) {
+    String command = statement.getCommandTag();
+    if (Strings.isNullOrEmpty(command)) {
+      new EmptyQueryResponse(this.outputStream).send(false);
+      return;
+    }
+
+    switch (statement.getStatementType()) {
       case DDL:
       case CLIENT_SIDE:
         new CommandCompleteResponse(this.outputStream, command).send(false);
-        return false;
+        break;
       case QUERY:
-        SendResultSetState state = sendResultSet(resultIndex, statement, mode, maxRows);
-        statement.setHasMoreData(resultIndex, state.hasMoreRows());
+        SendResultSetState state = sendResultSet(statement, mode, maxRows);
+        statement.setHasMoreData(state.hasMoreRows());
         if (state.hasMoreRows()) {
           new PortalSuspendedResponse(this.outputStream).send(false);
         } else {
-          statement.close(resultIndex);
+          statement.close();
           new CommandCompleteResponse(this.outputStream, "SELECT " + state.getNumberOfRowsSent())
               .send(false);
         }
-        return state.hasMoreRows();
+        break;
       case UPDATE:
         // For an INSERT command, the tag is INSERT oid rows, where rows is the number of rows
         // inserted. oid used to be the object ID of the inserted row if rows was 1 and the target
         // table had OIDs, but OIDs system columns are not supported anymore; therefore oid is
         // always 0.
-        command += ("INSERT".equals(command) ? " 0 " : " ") + statement.getUpdateCount(resultIndex);
+        command += ("INSERT".equals(command) ? " 0 " : " ") + statement.getUpdateCount();
         new CommandCompleteResponse(this.outputStream, command).send(false);
-        return false;
+        break;
       default:
-        throw new IllegalStateException(
-            "Unknown statement type: " + statement.getStatement(resultIndex));
+        throw new IllegalStateException("Unknown statement type: " + statement.getStatement());
     }
   }
 
@@ -199,18 +220,15 @@ public abstract class ControlMessage extends WireMessage {
    * @throws com.google.cloud.spanner.SpannerException if traversing the {@link ResultSet} fails.
    */
   public SendResultSetState sendResultSet(
-      int resultIndex, IntermediateStatement describedResult, QueryMode mode, long maxRows)
-      throws Exception {
+      IntermediateStatement describedResult, QueryMode mode, long maxRows) throws Exception {
+    Preconditions.checkArgument(
+        describedResult.containsResultSet(), "The statement result must be a result set");
     long rows = 0;
-    boolean hasData = describedResult.isHasMoreData(resultIndex);
-    ResultSet resultSet = describedResult.getStatementResult(resultIndex);
+    ResultSet resultSet = describedResult.getStatementResult().getResultSet();
+    boolean hasData = describedResult.isHasMoreData();
     while (hasData) {
       new DataRowResponse(
-              this.outputStream,
-              resultIndex,
-              describedResult,
-              this.connection.getServer().getOptions(),
-              mode)
+              this.outputStream, describedResult, this.connection.getServer().getOptions(), mode)
           .send(false);
       rows++;
       hasData = resultSet.next();
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 d9973fc8..08717de5 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
@@ -52,7 +52,7 @@ public class CopyDataMessage extends ControlMessage {
     }
     // If backend error occurred during copy-in mode, drop any subsequent CopyData messages.
     MutationWriter mutationWriter = this.statement.getMutationWriter();
-    if (!statement.hasException(0)) {
+    if (!statement.hasException()) {
       try {
         mutationWriter.addCopyData(this.payload);
       } catch (SpannerException exception) {
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDoneMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDoneMessage.java
index 0dd73d85..c2580a8d 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDoneMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyDoneMessage.java
@@ -18,6 +18,7 @@ import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.UpdateCount;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
@@ -45,17 +46,18 @@ public class CopyDoneMessage extends ControlMessage {
     if (this.statement != null) {
       MutationWriter mutationWriter = this.statement.getMutationWriter();
       statement.close();
-      if (!statement.hasException(0)) {
+      if (!statement.hasException()) {
         try {
           long rowCount = this.statement.getUpdateCount();
-          statement.setUpdateCount(0, rowCount); // Set the row count of number of rows copied.
-          this.sendSpannerResult(0, this.statement, QueryMode.SIMPLE, 0L);
+          statement.setStatementResult(
+              new UpdateCount(rowCount)); // Set the row count of number of rows copied.
+          this.sendSpannerResult(this.statement, QueryMode.SIMPLE, 0L);
           this.outputStream.flush();
         } catch (Exception e) {
           // Spanner returned an error when trying to commit the batch of mutations.
           mutationWriter.writeErrorFile(e);
           mutationWriter.closeErrorFile();
-          this.connection.setStatus(ConnectionStatus.IDLE);
+          this.connection.setStatus(ConnectionStatus.AUTHENTICATED);
           this.connection.removeActiveStatement(this.statement);
           throw e;
         }
@@ -63,7 +65,7 @@ public class CopyDoneMessage extends ControlMessage {
         mutationWriter.closeErrorFile();
       }
     }
-    this.connection.setStatus(ConnectionStatus.IDLE);
+    this.connection.setStatus(ConnectionStatus.AUTHENTICATED);
     this.connection.removeActiveStatement(this.statement);
     new ReadyResponse(this.outputStream, ReadyResponse.Status.IDLE).send();
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyFailMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyFailMessage.java
index 280b2b33..68f7236d 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyFailMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/CopyFailMessage.java
@@ -52,12 +52,12 @@ public class CopyFailMessage extends ControlMessage {
       mutationWriter.rollback();
       mutationWriter.closeErrorFile();
       statement.close();
-      if (!statement.hasException(0)) {
+      if (!statement.hasException()) {
         new ErrorResponse(this.outputStream, new Exception(this.errorMessage), State.IOError)
             .send();
       }
     }
-    this.connection.setStatus(ConnectionStatus.IDLE);
+    this.connection.setStatus(ConnectionStatus.AUTHENTICATED);
     this.connection.removeActiveStatement(this.statement);
     new ReadyResponse(this.outputStream, ReadyResponse.Status.IDLE).send();
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/DescribeMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/DescribeMessage.java
index a0eac6e9..d15a0fb2 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/DescribeMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/DescribeMessage.java
@@ -16,25 +16,30 @@ package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.metadata.DescribePortalMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeStatementMetadata;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.cloud.spanner.pgadapter.wireoutput.NoDataResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ParameterDescriptionResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.RowDescriptionResponse;
+import com.google.common.annotations.VisibleForTesting;
 import java.text.MessageFormat;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 /** Calls describe on a portal or prepared statement. */
 @InternalApi
-public class DescribeMessage extends ControlMessage {
+public class DescribeMessage extends AbstractQueryProtocolMessage {
 
   protected static final char IDENTIFIER = 'D';
 
   private final PreparedType type;
   private final String name;
   private final IntermediateStatement statement;
+  private Future<DescribePortalMetadata> describePortalMetadata;
 
   public DescribeMessage(ConnectionHandler connection) throws Exception {
     super(connection);
@@ -47,8 +52,25 @@ public class DescribeMessage extends ControlMessage {
     }
   }
 
+  /** Constructor for manually created Describe messages from the simple query protocol. */
+  public DescribeMessage(ConnectionHandler connection, ManuallyCreatedToken manuallyCreatedToken) {
+    super(connection, 4, manuallyCreatedToken);
+    this.type = PreparedType.Portal;
+    this.name = "";
+    this.statement = this.connection.getPortal(this.name);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  void buffer(BackendConnection backendConnection) {
+    if (this.type == PreparedType.Portal && this.statement.containsResultSet()) {
+      describePortalMetadata =
+          (Future<DescribePortalMetadata>) this.statement.describeAsync(backendConnection);
+    }
+  }
+
   @Override
-  protected void sendPayload() throws Exception {
+  public void flush() throws Exception {
     try {
       if (this.type == PreparedType.Portal) {
         this.handleDescribePortal();
@@ -94,33 +116,47 @@ public class DescribeMessage extends ControlMessage {
    *
    * @throws Exception if sending the message back to the client causes an error.
    */
-  public void handleDescribePortal() throws Exception {
-    if (this.statement.hasException(0)) {
-      this.handleError(this.statement.getException(0));
+  @VisibleForTesting
+  void handleDescribePortal() throws Exception {
+    if (this.statement.hasException()) {
+      throw this.statement.getException();
     } else {
-      switch (this.statement.getStatementType(0)) {
-        case UPDATE:
-        case DDL:
-        case CLIENT_SIDE:
+      if (this.statement.containsResultSet()) {
+        try {
+          new RowDescriptionResponse(
+                  this.outputStream,
+                  this.statement,
+                  getPortalMetadata().getMetadata(),
+                  this.connection.getServer().getOptions(),
+                  this.queryMode)
+              .send();
+        } catch (SpannerException exception) {
+          this.handleError(exception);
+        }
+      } else {
+        // The simple query protocol does not expect a NoData response in case of a non-query
+        // statement.
+        if (isExtendedProtocol()) {
           new NoDataResponse(this.outputStream).send();
-          break;
-        case QUERY:
-          try {
-            new RowDescriptionResponse(
-                    this.outputStream,
-                    this.statement,
-                    ((DescribePortalMetadata) this.statement.describe()).getMetadata(),
-                    this.connection.getServer().getOptions(),
-                    QueryMode.EXTENDED)
-                .send();
-            break;
-          } catch (SpannerException exception) {
-            this.handleError(exception);
-          }
+        }
       }
     }
   }
 
+  @VisibleForTesting
+  DescribePortalMetadata getPortalMetadata() {
+    if (!this.describePortalMetadata.isDone()) {
+      throw new IllegalStateException("Trying to get Portal Metadata before it has been described");
+    }
+    try {
+      return this.describePortalMetadata.get();
+    } catch (ExecutionException executionException) {
+      throw SpannerExceptionFactory.asSpannerException(executionException.getCause());
+    } catch (InterruptedException interruptedException) {
+      throw SpannerExceptionFactory.propagateInterrupt(interruptedException);
+    }
+  }
+
   /**
    * Called when a describe message of type 'S' is received.
    *
@@ -136,7 +172,7 @@ public class DescribeMessage extends ControlMessage {
                 this.statement,
                 metadata.getResultSet(),
                 this.connection.getServer().getOptions(),
-                QueryMode.EXTENDED)
+                this.queryMode)
             .send();
       } else {
         new NoDataResponse(this.outputStream).send();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java
index bdcbb304..7fb148da 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java
@@ -16,18 +16,18 @@ package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import java.text.MessageFormat;
 
 /** Executes a portal. */
 @InternalApi
-public class ExecuteMessage extends ControlMessage {
+public class ExecuteMessage extends AbstractQueryProtocolMessage {
   protected static final char IDENTIFIER = 'E';
 
-  private String name;
-  private int maxRows;
-  private IntermediateStatement statement;
+  private final String name;
+  private final int maxRows;
+  private final IntermediateStatement statement;
 
   public ExecuteMessage(ConnectionHandler connection) throws Exception {
     super(connection);
@@ -36,9 +36,21 @@ public class ExecuteMessage extends ControlMessage {
     this.statement = this.connection.getPortal(this.name);
   }
 
+  /** Constructor for execute messages that are generated by the simple query protocol. */
+  public ExecuteMessage(ConnectionHandler connection, ManuallyCreatedToken manuallyCreatedToken) {
+    super(connection, 8, manuallyCreatedToken);
+    this.name = "";
+    this.maxRows = 0;
+    this.statement = this.connection.getPortal(this.name);
+  }
+
+  @Override
+  void buffer(BackendConnection backendConnection) {
+    this.statement.executeAsync(backendConnection);
+  }
+
   @Override
-  protected void sendPayload() throws Exception {
-    this.statement.execute();
+  public void flush() throws Exception {
     this.handleExecute();
   }
 
@@ -77,11 +89,11 @@ public class ExecuteMessage extends ControlMessage {
    * @throws Exception if sending the message back to the client causes an error.
    */
   private void handleExecute() throws Exception {
-    if (this.statement.hasException(0)) {
-      this.handleError(this.statement.getException(0));
+    if (this.statement.hasException()) {
+      this.handleError(this.statement.getException());
     } else {
       try {
-        this.sendSpannerResult(0, this.statement, QueryMode.EXTENDED, this.maxRows);
+        this.sendSpannerResult(this.statement, this.queryMode, this.maxRows);
         this.outputStream.flush();
       } catch (Exception exception) {
         handleError(exception);
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 13c6c1a9..8aeab56b 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
@@ -35,6 +35,7 @@ public class FlushMessage extends ControlMessage {
 
   @Override
   protected void sendPayload() throws Exception {
+    connection.getExtendedQueryProtocolHandler().flush();
     boolean inTransaction = connection.getSpannerConnection().isInTransaction();
     new ReadyResponse(
             this.outputStream, inTransaction ? Status.TRANSACTION : ReadyResponse.Status.IDLE)
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 fb94c1f3..589671dc 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
@@ -20,6 +20,7 @@ import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
 import com.google.cloud.spanner.pgadapter.wireoutput.ParseCompleteResponse;
 import com.google.common.base.Strings;
@@ -27,7 +28,7 @@ import java.text.MessageFormat;
 
 /** Creates a prepared statement. */
 @InternalApi
-public class ParseMessage extends ControlMessage {
+public class ParseMessage extends AbstractQueryProtocolMessage {
   private static final AbstractStatementParser PARSER =
       AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
   protected static final char IDENTIFIER = 'P';
@@ -51,13 +52,36 @@ public class ParseMessage extends ControlMessage {
     this.statement.setParameterDataTypes(this.parameterDataTypes);
   }
 
+  /**
+   * Constructor for manually created Parse messages that originate from the simple query protocol.
+   */
+  public ParseMessage(ConnectionHandler connection, ParsedStatement parsedStatement) {
+    super(
+        connection,
+        5 + parsedStatement.getSqlWithoutComments().length(),
+        ManuallyCreatedToken.MANUALLY_CREATED_TOKEN);
+    this.name = "";
+    this.parameterDataTypes = new int[0];
+    this.statement =
+        new IntermediatePreparedStatement(
+            connection, connection.getServer().getOptions(), name, parsedStatement);
+    this.statement.setParameterDataTypes(this.parameterDataTypes);
+  }
+
   @Override
-  protected void sendPayload() throws Exception {
+  void buffer(BackendConnection backendConnection) {
     if (!Strings.isNullOrEmpty(this.name) && this.connection.hasStatement(this.name)) {
       throw new IllegalStateException("Must close statement before reusing name.");
     }
     this.connection.registerStatement(this.name, this.statement);
-    new ParseCompleteResponse(this.outputStream).send();
+  }
+
+  @Override
+  public void flush() throws Exception {
+    // The simple query protocol does not need the ParseComplete response.
+    if (isExtendedProtocol()) {
+      new ParseCompleteResponse(this.outputStream).send();
+    }
   }
 
   @Override
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 dadbdc4f..d2e05197 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
@@ -21,17 +21,11 @@ import com.google.cloud.spanner.connection.AbstractStatementParser;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
-import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
-import com.google.cloud.spanner.pgadapter.statements.MatcherStatement;
+import com.google.cloud.spanner.pgadapter.statements.SimpleQueryStatement;
 import com.google.cloud.spanner.pgadapter.utils.StatementParser;
 import com.google.cloud.spanner.pgadapter.wireoutput.CopyInResponse;
-import com.google.cloud.spanner.pgadapter.wireoutput.EmptyQueryResponse;
-import com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse;
-import com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse.State;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
-import com.google.cloud.spanner.pgadapter.wireoutput.RowDescriptionResponse;
 import java.text.MessageFormat;
 
 /** Executes a simple statement. */
@@ -42,34 +36,34 @@ public class QueryMessage extends ControlMessage {
   protected static final char IDENTIFIER = 'Q';
   public static final String COPY = "COPY";
 
+  private final ParsedStatement statement;
+  private final CopyStatement copyStatement;
+  private final SimpleQueryStatement simpleQueryStatement;
   private final boolean isCopy;
-  private final IntermediateStatement statement;
 
   public QueryMessage(ConnectionHandler connection) throws Exception {
     super(connection);
-    ParsedStatement parsedStatement = PARSER.parse(Statement.of(this.readAll()));
-    this.isCopy = StatementParser.isCommand(COPY, parsedStatement.getSqlWithoutComments());
+    this.statement = PARSER.parse(Statement.of(this.readAll()));
+    this.isCopy = StatementParser.isCommand(COPY, statement.getSqlWithoutComments());
     if (isCopy) {
-      this.statement =
-          new CopyStatement(connection, connection.getServer().getOptions(), parsedStatement);
-    } else if (!connection.getServer().getOptions().requiresMatcher()) {
-      this.statement =
-          new IntermediateStatement(
-              connection.getServer().getOptions(), parsedStatement, this.connection);
+      this.copyStatement =
+          new CopyStatement(connection, connection.getServer().getOptions(), statement);
+      this.simpleQueryStatement = null;
+      this.connection.addActiveStatement(this.copyStatement);
     } else {
-      this.statement =
-          new MatcherStatement(
-              connection.getServer().getOptions(), parsedStatement, this.connection);
+      this.simpleQueryStatement =
+          new SimpleQueryStatement(connection.getServer().getOptions(), statement, this.connection);
+      this.copyStatement = null;
     }
-    this.connection.addActiveStatement(this.statement);
   }
 
   @Override
   protected void sendPayload() throws Exception {
-    this.statement.execute();
-    this.handleQuery();
-    if (!this.statement.getCommand(0).equalsIgnoreCase(COPY)) {
-      this.connection.removeActiveStatement(this.statement);
+    if (this.isCopy) {
+      this.copyStatement.execute();
+      handleCopy();
+    } else {
+      this.simpleQueryStatement.execute();
     }
   }
 
@@ -81,7 +75,7 @@ public class QueryMessage extends ControlMessage {
   @Override
   protected String getPayloadString() {
     return new MessageFormat("Length: {0}, SQL: {1}")
-        .format(new Object[] {this.length, this.statement.getSql()});
+        .format(new Object[] {this.length, this.statement.getSqlWithoutComments()});
   }
 
   @Override
@@ -89,58 +83,32 @@ public class QueryMessage extends ControlMessage {
     return String.valueOf(IDENTIFIER);
   }
 
-  public IntermediateStatement getStatement() {
+  public ParsedStatement getStatement() {
     return this.statement;
   }
 
-  /**
-   * Simple Query handler, which examined the state of the statement and processes accordingly (if
-   * error, handle error, otherwise sends the result and if contains result set, send row
-   * description)
-   *
-   * @throws Exception if handling the query fails
-   */
-  public void handleQuery() throws Exception {
-    // Skip unexecuted statements, as no response needs be returned
-    for (int index = 0; index < statement.getExecutedCount(); index++) {
-      if (this.statement.hasException(index)) {
-        new ErrorResponse(
-                this.outputStream, this.statement.getException(index), State.InternalError)
-            .send();
-      } else if (this.statement.getCommand(index).equalsIgnoreCase(COPY)) {
-        CopyStatement copyStatement = (CopyStatement) this.statement;
-        new CopyInResponse(
-                this.outputStream,
-                copyStatement.getTableColumns().size(),
-                copyStatement.getFormatCode())
-            .send();
-        this.connection.setStatus(ConnectionStatus.COPY_IN);
+  public SimpleQueryStatement getSimpleQueryStatement() {
+    return this.simpleQueryStatement;
+  }
 
-        // Return early as we do not respond with CommandComplete after a COPY command.
-        return;
-      } else if ("".equals(this.statement.getCommandTag(index))) {
-        new EmptyQueryResponse(outputStream).send();
-      } else {
-        if (this.statement.containsResultSet(index)) {
-          new RowDescriptionResponse(
-                  this.outputStream,
-                  this.statement,
-                  this.statement.getStatementResult(index),
-                  this.connection.getServer().getOptions(),
-                  QueryMode.SIMPLE)
-              .send(false);
-        }
-        this.sendSpannerResult(index, this.statement, QueryMode.SIMPLE, 0L);
-      }
-    }
-    if (connection.getSpannerConnection().isInTransaction()
-        && connection.getStatus() == ConnectionStatus.TRANSACTION_ABORTED) {
-      // Actively rollback the aborted transaction but still block clients
-      // Clear any statement tags, as these are not allowed for rollbacks.
-      connection.getSpannerConnection().setStatementTag(null);
-      connection.getSpannerConnection().rollback();
+  private void handleCopy() throws Exception {
+    if (this.copyStatement.hasException()) {
+      handleError(this.copyStatement.getException());
+      new ReadyResponse(
+              this.outputStream,
+              connection
+                  .getExtendedQueryProtocolHandler()
+                  .getBackendConnection()
+                  .getConnectionState()
+                  .getReadyResponseStatus())
+          .send();
+    } else {
+      new CopyInResponse(
+              this.outputStream,
+              copyStatement.getTableColumns().size(),
+              copyStatement.getFormatCode())
+          .send();
+      this.connection.setStatus(ConnectionStatus.COPY_IN);
     }
-    new ReadyResponse(this.outputStream, connection.getStatus().getReadyResponseStatus()).send();
-    this.connection.cleanUp(this.statement);
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
index 12388ea3..1923c702 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
@@ -50,7 +50,7 @@ public class StartupMessage extends BootstrapMessage {
           this.connection.getConnectionId(),
           this.connection.getSecret(),
           this.connection.getServer().getOptions());
-      this.connection.setStatus(ConnectionStatus.IDLE);
+      this.connection.setStatus(ConnectionStatus.AUTHENTICATED);
     } else {
       new MD5AuthenticationRequest(this.outputStream, 0).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 b59833af..51f22f1f 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
@@ -32,9 +32,21 @@ public class SyncMessage extends ControlMessage {
     super(connection);
   }
 
+  public SyncMessage(ConnectionHandler connection, ManuallyCreatedToken manuallyCreatedToken) {
+    super(connection, 4, manuallyCreatedToken);
+  }
+
   @Override
   protected void sendPayload() throws Exception {
-    new ReadyResponse(this.outputStream, connection.getStatus().getReadyResponseStatus()).send();
+    connection.getExtendedQueryProtocolHandler().sync();
+    new ReadyResponse(
+            this.outputStream,
+            connection
+                .getExtendedQueryProtocolHandler()
+                .getBackendConnection()
+                .getConnectionState()
+                .getReadyResponseStatus())
+        .send();
   }
 
   @Override


Test patch:
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 8f588f0b..5b9fcdd9 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
@@ -16,30 +16,22 @@ package com.google.cloud.spanner.pgadapter.statements;
 
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.Dialect;
-import com.google.cloud.spanner.ErrorCode;
-import com.google.cloud.spanner.ResultSet;
-import com.google.cloud.spanner.SpannerBatchUpdateException;
 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.AbstractStatementParser.ParsedStatement;
 import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
 import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.PostgreSQLStatementParser;
 import com.google.cloud.spanner.connection.StatementResult;
-import com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType;
-import com.google.cloud.spanner.connection.TransactionMode;
+import com.google.cloud.spanner.connection.StatementResult.ResultType;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.metadata.DescribeMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.NoResult;
 import com.google.cloud.spanner.pgadapter.utils.StatementParser;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.ImmutableList;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 /**
  * Data type to store simple SQL statement with designated metadata. Allows manipulation of
@@ -48,29 +40,20 @@ import java.util.Objects;
  */
 @InternalApi
 public class IntermediateStatement {
-  public static final String TRANSACTION_ABORTED_ERROR =
-      "current transaction is aborted, commands ignored until end of transaction block";
   protected static final PostgreSQLStatementParser PARSER =
       (PostgreSQLStatementParser) AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
 
   protected final OptionsMetadata options;
-  private final ImmutableList<StatementType> statementTypes;
-  protected ResultSet[] statementResults;
-  protected final boolean[] hasMoreData;
-  protected final SpannerException[] exceptions;
+  protected StatementResult statementResult;
+  protected boolean hasMoreData;
+  protected Future<StatementResult> futureStatementResult;
+  protected SpannerException exception;
   protected final ParsedStatement parsedStatement;
-  protected final ImmutableList<String> commands;
-  protected final List<String> commandTags;
-  protected int executedCount = 0;
+  protected final String command;
+  protected String commandTag;
+  protected boolean executed;
   protected final Connection connection;
-  protected long[] updateCounts;
-  protected final ImmutableList<ParsedStatement> statements;
   protected final ConnectionHandler connectionHandler;
-  private ExecutionStatus executionStatus;
-
-  private static final char STATEMENT_DELIMITER = ';';
-  private static final char SINGLE_QUOTE = '\'';
-  private static final char DOUBLE_QUOTE = '"';
 
   public IntermediateStatement(
       OptionsMetadata options,
@@ -85,114 +68,11 @@ public class IntermediateStatement {
       ParsedStatement parsedStatement) {
     this.connectionHandler = connectionHandler;
     this.options = options;
-    this.parsedStatement = replaceKnownUnsupportedQueries(parsedStatement);
+    this.parsedStatement =
+        SimpleQueryStatement.replaceKnownUnsupportedQueries(this.options, parsedStatement);
     this.connection = connectionHandler.getSpannerConnection();
-    this.statements = parseStatements(parsedStatement);
-    this.commands = StatementParser.parseCommands(this.statements);
-    this.commandTags = new ArrayList<>(this.commands);
-    this.statementTypes = determineStatementTypes(this.statements);
-    this.hasMoreData = new boolean[this.statements.size()];
-    this.exceptions = new SpannerException[this.statements.size()];
-  }
-
-  protected ParsedStatement replaceKnownUnsupportedQueries(ParsedStatement parsedStatement) {
-    if (this.options.isReplaceJdbcMetadataQueries()
-        && JdbcMetadataStatementHelper.isPotentialJdbcMetadataStatement(
-            parsedStatement.getSqlWithoutComments())) {
-      return PARSER.parse(
-          Statement.of(
-              JdbcMetadataStatementHelper.replaceJdbcMetadataStatement(
-                  parsedStatement.getSqlWithoutComments())));
-    }
-    return parsedStatement;
-  }
-
-  /** @return The number of SQL statements in this {@link IntermediateStatement} */
-  public int getStatementCount() {
-    return statements.size();
-  }
-
-  /**
-   * Determines the statement types based on the given statement strings. The statement string must
-   * already been stripped of any comments that might precede the actual statement string.
-   *
-   * @param statements The parsed statements to determine the type for
-   * @return The list of {@link StatementType} that the given statement strings will produce
-   */
-  protected static ImmutableList<StatementType> determineStatementTypes(
-      List<ParsedStatement> statements) {
-    ImmutableList.Builder<StatementType> builder = ImmutableList.builder();
-    for (ParsedStatement stmt : statements) {
-      if (stmt.isUpdate()) {
-        builder.add(StatementType.UPDATE);
-      } else if (stmt.isQuery()) {
-        builder.add(StatementType.QUERY);
-      } else if (stmt.isDdl()) {
-        builder.add(StatementType.DDL);
-      } else {
-        builder.add(StatementType.CLIENT_SIDE);
-      }
-    }
-    return builder.build();
-  }
-
-  static ImmutableList<String> splitStatements(String sql) {
-    // First check trivial cases with only one statement.
-    int firstIndexOfDelimiter = sql.indexOf(STATEMENT_DELIMITER);
-    if (firstIndexOfDelimiter == -1) {
-      return ImmutableList.of(sql);
-    }
-    if (firstIndexOfDelimiter == sql.length() - 1) {
-      return ImmutableList.of(sql.substring(0, sql.length() - 1));
-    }
-
-    ImmutableList.Builder<String> builder = ImmutableList.builder();
-    // TODO: Fix this parsing, as it does not take all types of quotes into consideration.
-    boolean insideQuotes = false;
-    char quoteChar = 0;
-    int index = 0;
-    for (int i = 0; i < sql.length(); ++i) {
-      if (insideQuotes) {
-        if (sql.charAt(i) == quoteChar) {
-          if (sql.length() > (i + 1) && sql.charAt(i + 1) == quoteChar) {
-            // This is an escaped quote. Skip one ahead.
-            i++;
-          } else {
-            insideQuotes = false;
-          }
-        }
-      } else {
-        if (sql.charAt(i) == SINGLE_QUOTE || sql.charAt(i) == DOUBLE_QUOTE) {
-          quoteChar = sql.charAt(i);
-          insideQuotes = true;
-        } else {
-          // skip semicolon inside quotes
-          if (sql.charAt(i) == STATEMENT_DELIMITER) {
-            String stmt = sql.substring(index, i).trim();
-            // Statements with only ';' character are empty and dropped.
-            if (stmt.length() > 0) {
-              builder.add(stmt);
-            }
-            index = i + 1;
-          }
-        }
-      }
-    }
-
-    if (index < sql.length()) {
-      builder.add(sql.substring(index).trim());
-    }
-    return builder.build();
-  }
-
-  protected static ImmutableList<ParsedStatement> parseStatements(ParsedStatement stmt) {
-    String sql = stmt.getSqlWithoutComments();
-    Preconditions.checkNotNull(sql);
-    ImmutableList.Builder<ParsedStatement> builder = ImmutableList.builder();
-    for (String statement : splitStatements(sql)) {
-      builder.add(PARSER.parse(Statement.of(statement)));
-    }
-    return builder.build();
+    this.command = StatementParser.parseCommand(this.parsedStatement.getSqlWithoutComments());
+    this.commandTag = this.command;
   }
 
   /**
@@ -211,443 +91,141 @@ public class IntermediateStatement {
    * @throws Exception if closing fails server-side.
    */
   public void close() throws Exception {
-    for (int index = 0; index < this.getStatementCount(); index++) {
-      close(index);
-    }
-  }
-
-  public void close(int index) throws Exception {
-    if (this.statementResults != null && this.statementResults[index] != null) {
-      this.statementResults[index].close();
-      this.statementResults[index] = null;
+    if (statementResult != null && statementResult.getResultType() == ResultType.RESULT_SET) {
+      statementResult.getResultSet().close();
+      statementResult = null;
     }
   }
 
   /** @return True if this is a select statement, false otherwise. */
-  public boolean containsResultSet(int index) {
-    return this.statementTypes.get(index) == StatementType.QUERY;
+  public boolean containsResultSet() {
+    return this.parsedStatement.isQuery();
   }
 
   /** @return True if this statement was executed, False otherwise. */
   public boolean isExecuted() {
-    return executedCount > 0;
-  }
-
-  public int getExecutedCount() {
-    return this.executedCount;
+    return executed;
   }
 
   /**
    * @return The number of items that were modified by this execution for DML. 0 for DDL and -1 for
    *     QUERY.
    */
-  public long getUpdateCount(int index) {
-    if (this.updateCounts == null) {
-      return 0;
-    }
-    return this.updateCounts[index];
-  }
-
-  public void setUpdateCount(int index, long updateCount) {
-    if (this.updateCounts == null) {
-      this.updateCounts = new long[getStatementCount()];
+  public long getUpdateCount() {
+    initFutureResult();
+    switch (this.parsedStatement.getType()) {
+      case QUERY:
+        return -1L;
+      case UPDATE:
+        return this.statementResult.getUpdateCount();
+      case CLIENT_SIDE:
+      case DDL:
+      case UNKNOWN:
+      default:
+        return 0L;
     }
-    this.updateCounts[index] = updateCount;
-    this.hasMoreData[index] = false;
   }
 
   /** @return True if at some point in execution, and exception was thrown. */
-  public boolean hasException(int index) {
-    return this.exceptions[index] != null;
+  public boolean hasException() {
+    initFutureResult();
+    return this.exception != null;
   }
 
   /** @return True if only a subset of the available data has been returned. */
-  public boolean isHasMoreData(int index) {
-    return this.hasMoreData[index];
+  public boolean isHasMoreData() {
+    return this.hasMoreData;
   }
 
-  public void setHasMoreData(int index, boolean hasMoreData) {
-    this.hasMoreData[index] = hasMoreData;
+  public void setHasMoreData(boolean hasMoreData) {
+    this.hasMoreData = hasMoreData;
   }
 
   public Connection getConnection() {
     return this.connection;
   }
 
-  public List<ParsedStatement> getStatements() {
-    return this.statements;
-  }
-
-  public String getStatement(int index) {
-    return this.statements.get(index).getSqlWithoutComments();
-  }
-
-  public ResultSet getStatementResult(int index) {
-    if (this.statementResults == null) {
-      return null;
-    }
-    return this.statementResults[index];
-  }
-
-  protected void setStatementResult(int index, ResultSet resultSet) {
-    if (this.statementResults == null) {
-      this.statementResults = new ResultSet[getStatementCount()];
-    }
-    this.statementResults[index] = resultSet;
-    this.hasMoreData[index] = resultSet.next();
-    if (this.updateCounts == null) {
-      this.updateCounts = new long[getStatementCount()];
-    }
-    this.updateCounts[index] = -1;
-  }
-
-  public StatementType getStatementType(int index) {
-    return this.statementTypes.get(index);
-  }
-
-  public String getSql() {
+  public String getStatement() {
     return this.parsedStatement.getSqlWithoutComments();
   }
 
-  public Exception getException(int index) {
-    Exception e = this.exceptions[index];
-    this.exceptions[index] = null;
-    return e;
-  }
-
-  /**
-   * 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.
-   */
-  protected void updateResultCount(int index, StatementResult result) {
-    switch (result.getResultType()) {
-      case RESULT_SET:
-        setStatementResult(index, result.getResultSet());
-        break;
-      case UPDATE_COUNT:
-        setUpdateCount(index, result.getUpdateCount());
-        break;
-      case NO_RESULT:
-        this.hasMoreData[index] = false;
-        break;
-      default:
-        throw SpannerExceptionFactory.newSpannerException(
-            ErrorCode.INTERNAL, "Unknown or unsupported result type: " + result.getResultType());
-    }
-  }
-
-  protected void updateBatchResultCount(int fromIndex, long[] updateCounts) {
-    for (int index = fromIndex; index < fromIndex + updateCounts.length; index++) {
-      setUpdateCount(index, updateCounts[index - fromIndex]);
-    }
-  }
-
-  /**
-   * Clean up and save metadata when an exception occurs.
-   *
-   * @param exception The exception to store.
-   */
-  protected void handleExecutionException(int index, SpannerException exception) {
-    this.exceptions[index] = exception;
-    this.hasMoreData[index] = false;
-  }
-
-  protected void handleExecutionExceptionAndTransactionStatus(
-      int index, SpannerException exception) {
-    if (executionStatus == ExecutionStatus.EXPLICIT_TRANSACTION) {
-      connectionHandler.setStatus(ConnectionStatus.TRANSACTION_ABORTED);
-    }
-    handleExecutionException(index, exception);
-  }
-
-  /** Execute the SQL statement, storing metadata. */
-  public void execute() {
-    if (connection.isInTransaction()) {
-      executionStatus = ExecutionStatus.EXPLICIT_TRANSACTION;
-    } else {
-      executionStatus = ExecutionStatus.IMPLICIT_TRANSACTION;
-    }
-    while (executedCount < getStatementCount()) {
-      maybeStartImplicitTransaction(executedCount);
-
-      if (connectionHandler.getStatus() == ConnectionStatus.TRANSACTION_ABORTED
-          && !isRollback(executedCount)
-          && !isCommit(executedCount)) {
-        handleExecutionException(
-            executedCount,
-            SpannerExceptionFactory.newSpannerException(
-                ErrorCode.INVALID_ARGUMENT, TRANSACTION_ABORTED_ERROR));
-        executedCount++;
-        continue;
-      }
-      if (getStatementType(executedCount) == StatementType.DDL) {
-        try {
-          prepareExecuteDdl(executedCount);
-        } catch (SpannerException e) {
-          handleExecutionExceptionAndTransactionStatus(executedCount, e);
-          executedCount++;
-          break;
-        }
-      }
-      boolean canUseBatch = false;
-      if (executedCount < (getStatementCount() - 1)) {
-        StatementType statementType = getStatementType(executedCount);
-        StatementType nextStatementType = getStatementType(executedCount + 1);
-        canUseBatch = canBeBatchedTogether(statementType, nextStatementType);
-      }
-      if (canUseBatch) {
-        executedCount += executeStatementsInBatch(executedCount);
-      } else {
-        executeSingleStatement(executedCount);
-        executedCount++;
-      }
-      // Return directly after any exception.
-      if (hasException(executedCount - 1)) {
-        break;
+  private void initFutureResult() {
+    if (this.futureStatementResult != null) {
+      if (!this.futureStatementResult.isDone()) {
+        throw new IllegalStateException("Statement result cannot be retrieved before flush/sync");
       }
-    }
-    if (executionStatus == ExecutionStatus.IMPLICIT_TRANSACTION && connection.isInTransaction()) {
-      if (hasException(executedCount - 1)) {
-        connection.rollback();
-      } else {
-        connection.commit();
+      try {
+        setStatementResult(this.futureStatementResult.get());
+      } catch (ExecutionException executionException) {
+        this.exception = SpannerExceptionFactory.asSpannerException(executionException.getCause());
+      } catch (InterruptedException interruptedException) {
+        this.exception = SpannerExceptionFactory.propagateInterrupt(interruptedException);
+      } finally {
+        this.futureStatementResult = null;
       }
     }
   }
 
-  /**
-   * Prepares the connection for executing a DDL statement. This can include committing the current
-   * transaction.
-   *
-   * <p>Executing the DDL statement may or may not be allowed depending on the state of the
-   * transaction and the selected {@link
-   * com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode}. The method
-   * will throw a {@link SpannerException} if executing a DDL statement at this point is not
-   * allowed.
-   */
-  protected void prepareExecuteDdl(int index) {
-    // Single statements are simpler to check, so we do that in a separate check.
-    if (statements.size() == 1) {
-      switch (options.getDdlTransactionMode()) {
-        case Single:
-        case Batch:
-        case AutocommitImplicitTransaction:
-          // Single DDL statements outside explicit transactions are always allowed. For a single
-          // statement, there can also not be an implicit transaction that needs to be committed.
-          if (connection.isInTransaction()) {
-            throw SpannerExceptionFactory.newSpannerException(
-                ErrorCode.FAILED_PRECONDITION,
-                "DDL statements are only allowed outside explicit transactions.");
-          }
-          break;
-        case AutocommitExplicitTransaction:
-          // DDL statements are allowed even in explicit transactions. Commit any transaction that
-          // might be active.
-          if (connection.isInTransaction()) {
-            connection.commit();
-          }
-      }
-      return;
-    }
-
-    // We are in a batch of statements.
-    switch (options.getDdlTransactionMode()) {
-      case Single:
-        throw SpannerExceptionFactory.newSpannerException(
-            ErrorCode.FAILED_PRECONDITION,
-            "DDL statements are only allowed outside batches and transactions.");
-      case Batch:
-        if (connection.isInTransaction()
-            || statements.stream().anyMatch(statement -> !statement.isDdl())) {
-          throw SpannerExceptionFactory.newSpannerException(
-              ErrorCode.FAILED_PRECONDITION,
-              "DDL statements are not allowed in mixed batches or transactions.");
-        }
-        break;
-      case AutocommitImplicitTransaction:
-        if (connection.isInTransaction()
-            && executionStatus != ExecutionStatus.IMPLICIT_TRANSACTION) {
-          throw SpannerExceptionFactory.newSpannerException(
-              ErrorCode.FAILED_PRECONDITION,
-              "DDL statements are only allowed outside explicit transactions.");
-        }
-        if (connection.isInTransaction()) {
-          // Commit the implicit transaction before we continue.
-          connection.commit();
-        }
-        break;
-      case AutocommitExplicitTransaction:
-        // Commit any transaction that might be active and allow executing the statement.
-        // Switch the execution state to implicit transaction.
-        if (connection.isInTransaction()) {
-          connection.commit();
-          executionStatus = ExecutionStatus.IMPLICIT_TRANSACTION;
-        }
-    }
+  public StatementResult getStatementResult() {
+    initFutureResult();
+    return this.statementResult;
   }
 
-  private void maybeStartImplicitTransaction(int index) {
-    if (connection.isInTransaction()) {
-      return;
-    }
-    if (executionStatus == ExecutionStatus.EXPLICIT_TRANSACTION) {
-      return;
-    }
-    // No need to start a transaction for the last statement.
-    if (index == statements.size() - 1) {
-      return;
-    }
-    // No need to start a transaction for DDL or client side statements.
-    if (statementTypes.get(index) == StatementType.DDL
-        || statementTypes.get(index) == StatementType.CLIENT_SIDE) {
-      return;
-    }
-    // If there are only DML statements left, those can be executed as an auto-commit dml batch.
-    if (hasOnlyDmlStatementsAfter(index)) {
-      return;
-    }
-    // We need to start an implicit transaction.
-    // Check if a read-only transaction suffices.
-    connection.beginTransaction();
-    if (!hasDmlStatementsAfter(index)) {
-      connection.setTransactionMode(TransactionMode.READ_ONLY_TRANSACTION);
+  public void setStatementResult(StatementResult statementResult) {
+    this.statementResult = statementResult;
+    if (statementResult.getResultType() == ResultType.RESULT_SET) {
+      this.hasMoreData = statementResult.getResultSet().next();
+    } else if (statementResult instanceof NoResult
+        && ((NoResult) statementResult).hasCommandTag()) {
+      this.commandTag = ((NoResult) statementResult).getCommandTag();
     }
   }
 
-  private boolean hasDmlStatementsAfter(int index) {
-    return statementTypes.subList(index, statementTypes.size()).stream()
-        .anyMatch(type -> type == StatementType.UPDATE);
+  protected void setFutureStatementResult(Future<StatementResult> result) {
+    this.futureStatementResult = result;
   }
 
-  private boolean hasOnlyDmlStatementsAfter(int index) {
-    return statementTypes.subList(index, statementTypes.size()).stream()
-        .allMatch(type -> type == StatementType.UPDATE);
+  public StatementType getStatementType() {
+    return this.parsedStatement.getType();
   }
 
-  private boolean isBegin(int index) {
-    return getStatements().get(index).getType() == StatementType.CLIENT_SIDE
-        && getStatements().get(index).getClientSideStatementType() == ClientSideStatementType.BEGIN;
-  }
-
-  private boolean isCommit(int index) {
-    return getStatements().get(index).getType() == StatementType.CLIENT_SIDE
-        && getStatements().get(index).getClientSideStatementType()
-            == ClientSideStatementType.COMMIT;
-  }
-
-  private boolean isRollback(int index) {
-    return getStatements().get(index).getType() == StatementType.CLIENT_SIDE
-        && getStatements().get(index).getClientSideStatementType()
-            == ClientSideStatementType.ROLLBACK;
+  public String getSql() {
+    return this.parsedStatement.getSqlWithoutComments();
   }
 
-  private void executeSingleStatement(int index) {
-    // Before executing the statement, handle specific statements that change the transaction status
-    String command = getCommand(index);
-    String statement = getStatement(index);
-    if (isBegin(index)) {
-      // Executing a BEGIN statement when a transaction is already active will set the execution
-      // mode to EXPLICIT_TRANSACTION. The current transaction is not committed.
-      executionStatus = ExecutionStatus.EXPLICIT_TRANSACTION;
-      if (!connection.isInTransaction()) {
-        connection.execute(Statement.of(statement));
-      }
-      return;
-    }
-    if (isCommit(index) || isRollback(index)) {
-      if (connectionHandler.getStatus() == ConnectionStatus.TRANSACTION_ABORTED) {
-        connectionHandler.setStatus(ConnectionStatus.IDLE);
-        // COMMIT rollbacks aborted transaction
-        statement = "ROLLBACK";
-        commandTags.set(index, "ROLLBACK");
-      }
-      // Executing ROLLBACK or COMMIT when there is no active transaction is a no-op, but will
-      // change the transaction mode to implicit.
-      executionStatus = ExecutionStatus.IMPLICIT_TRANSACTION;
-      if (!connection.isInTransaction()) {
-        return;
-      }
-    }
-    // Ignore empty statements.
-    if (!"".equals(statement)) {
-      try {
-        StatementResult result = this.connection.execute(Statement.of(statement));
-        updateResultCount(index, result);
-      } catch (SpannerException e) {
-        handleExecutionExceptionAndTransactionStatus(index, e);
-      }
-    }
+  public Exception getException() {
+    Exception e = this.exception;
+    this.exception = null;
+    return e;
   }
 
   /**
-   * Executes the statements from fromIndex in a DML/DDL batch. The batch will consist of all
-   * statements from fromIndex till the first statement that is of a different type than the
-   * statement at fromIndex. That is; If the first statement is a DML statement, the batch will
-   * contain all statements that follow until it encounters a statement that is not a DML statement.
-   * The same also applies to DDL statements. Query statements and other statements can not be
-   * batched.
+   * Clean up and save metadata when an exception occurs.
    *
-   * @param fromIndex The index of the statements array where the batch should start
-   * @return The number of statements included in the batch.
+   * @param exception The exception to store.
    */
-  private int executeStatementsInBatch(int fromIndex) {
-    Preconditions.checkArgument(fromIndex < getStatementCount() - 1);
-    Preconditions.checkArgument(
-        canBeBatchedTogether(getStatementType(fromIndex), getStatementType(fromIndex + 1)));
-    StatementType batchType = getStatementType(fromIndex);
-    switch (batchType) {
-      case UPDATE:
-        connection.startBatchDml();
-        break;
-      case DDL:
-        connection.startBatchDdl();
-        break;
-      default:
-        throw SpannerExceptionFactory.newSpannerException(
-            ErrorCode.INVALID_ARGUMENT, "Statement type is not supported for batching");
-    }
-    int index = fromIndex;
-    while (index < getStatementCount()) {
-      StatementType statementType = getStatementType(index);
-      if (canBeBatchedTogether(batchType, statementType)) {
-        connection.execute(Statement.of(getStatement(index)));
-        index++;
-      } else {
-        // End the batch here, as the statement type on this index can not be batched together with
-        // the other statements in the batch.
-        break;
-      }
-    }
-    try {
-      long[] counts = connection.runBatch();
-      updateBatchResultCount(fromIndex, counts);
-    } catch (SpannerBatchUpdateException e) {
-      long[] counts = e.getUpdateCounts();
-      updateBatchResultCount(fromIndex, counts);
-      handleExecutionExceptionAndTransactionStatus(fromIndex + counts.length, e);
-      return counts.length + 1;
-    }
-    return index - fromIndex;
+  protected void handleExecutionException(SpannerException exception) {
+    this.exception = exception;
+    this.hasMoreData = false;
   }
 
-  private boolean canBeBatchedTogether(StatementType statementType1, StatementType statementType2) {
-    if (Objects.equals(statementType1, StatementType.QUERY)
-        || Objects.equals(statementType1, StatementType.CLIENT_SIDE)) {
-      return false;
-    }
-    return Objects.equals(statementType1, statementType2);
+  public void executeAsync(BackendConnection backendConnection) {
+    throw new UnsupportedOperationException();
   }
 
   /**
    * Moreso meant for inherited classes, allows one to call describe on a statement. Since raw
    * statements cannot be described, throw an error.
    */
-  public DescribeMetadata describe() {
+  public DescribeMetadata<?> describe() {
     throw new IllegalStateException(
         "Cannot describe a simple statement " + "(only prepared statements and portals)");
   }
 
+  public Future<? extends DescribeMetadata<?>> describeAsync(BackendConnection backendConnection) {
+    throw new UnsupportedOperationException();
+  }
+
   /**
    * Moreso intended for inherited classes (prepared statements et al) which allow the setting of
    * result format codes. Here we dafault to string.
@@ -657,22 +235,12 @@ public class IntermediateStatement {
   }
 
   /** @return the extracted command (first word) from the SQL statement. */
-  public String getCommand(int index) {
-    return this.commands.get(index);
+  public String getCommand() {
+    return this.command;
   }
 
   /** @return the extracted command (first word) from the really executed SQL statement. */
-  public String getCommandTag(int index) {
-    return this.commandTags.get(index);
-  }
-
-  /* Used for testing purposes */
-  public boolean isBatchedQuery() {
-    return (statements.size() > 1);
-  }
-
-  enum ExecutionStatus {
-    EXPLICIT_TRANSACTION,
-    IMPLICIT_TRANSACTION
+  public String getCommandTag() {
+    return this.commandTag;
   }
 }
diff --git a/src/test/golang/pgadapter_pgx_tests/pgx.go b/src/test/golang/pgadapter_pgx_tests/pgx.go
index accd1acd..3e7fb807 100644
--- a/src/test/golang/pgadapter_pgx_tests/pgx.go
+++ b/src/test/golang/pgadapter_pgx_tests/pgx.go
@@ -22,6 +22,7 @@ import (
 	"github.com/jackc/pgtype"
 	"github.com/jackc/pgx/v4"
 	"reflect"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -334,6 +335,112 @@ func TestPrepareSelectStatement(connString string) *C.char {
 	return nil
 }
 
+//export TestInsertBatch
+func TestInsertBatch(connString string) *C.char {
+	ctx := context.Background()
+	conn, err := pgx.Connect(ctx, connString)
+	if err != nil {
+		return C.CString(err.Error())
+	}
+	defer conn.Close(ctx)
+
+	batch := &pgx.Batch{}
+	batchSize := 10
+	if err := insertBatch(batch, connString, batchSize); err != nil {
+		return C.CString(err.Error())
+	}
+	res := conn.SendBatch(ctx, batch)
+	for i := 0; i < batchSize; i++ {
+		tag, err := res.Exec()
+		if err != nil {
+			return C.CString(fmt.Sprintf("failed to execute insert statement %d: %v", i, err))
+		}
+		if !tag.Insert() {
+			return C.CString(fmt.Sprintf("statement %d was not recognized as an insert", i))
+		}
+		if g, w := tag.RowsAffected(), int64(1); g != w {
+			return C.CString(fmt.Sprintf("rows affected mismatch for statement %d:\n Got: %v\nWant: %v", i, g, w))
+		}
+	}
+	if err := res.Close(); err != nil {
+		return C.CString(err.Error())
+	}
+
+	return nil
+}
+
+//export TestMixedBatch
+func TestMixedBatch(connString string) *C.char {
+	ctx := context.Background()
+	conn, err := pgx.Connect(ctx, connString)
+	if err != nil {
+		return C.CString(err.Error())
+	}
+	defer conn.Close(ctx)
+
+	batch := &pgx.Batch{}
+	batchSize := 5
+	if err := insertBatch(batch, connString, batchSize); err != nil {
+		return C.CString(err.Error())
+	}
+	batch.Queue("select count(*) from all_types where col_bool=$1", true)
+	batch.Queue("update all_types set col_bool=false where col_bool=$1", true)
+
+	res := conn.SendBatch(ctx, batch)
+	for i := 0; i < batchSize; i++ {
+		tag, err := res.Exec()
+		if err != nil {
+			return C.CString(fmt.Sprintf("failed to execute insert statement %d: %v", i, err))
+		}
+		if !tag.Insert() {
+			return C.CString(fmt.Sprintf("statement %d was not recognized as an insert", i))
+		}
+		if g, w := tag.RowsAffected(), int64(1); g != w {
+			return C.CString(fmt.Sprintf("rows affected mismatch for statement %d:\n Got: %v\nWant: %v", i, g, w))
+		}
+	}
+	var count int64
+	if err := res.QueryRow().Scan(&count); err != nil {
+		return C.CString(fmt.Sprintf("failed to get row count: %v", err.Error()))
+	}
+	tag, err := res.Exec()
+	if err != nil {
+		return C.CString(fmt.Sprintf("failed to execute update: %v", err.Error()))
+	}
+	if !tag.Update() {
+		return C.CString("update statement was not recognized as an update")
+	}
+	if g, w := tag.RowsAffected(), count; g != w {
+		return C.CString(fmt.Sprintf("rows affected mismatch for update statement:\n Got: %v\nWant: %v", g, w))
+	}
+	if err := res.Close(); err != nil {
+		return C.CString(err.Error())
+	}
+
+	return nil
+}
+
+func insertBatch(batch *pgx.Batch, connString string, batchSize int) error {
+	sql := "INSERT INTO all_types (col_bigint, col_bool, col_bytea, col_float8, col_int, col_numeric, col_timestamptz, col_date, col_varchar) values ($1, $2, $3, $4, $5, $6, $7, $8, $9)"
+	numeric := pgtype.Numeric{}
+	for i := 0; i < batchSize; i++ {
+		_ = numeric.Set(strconv.Itoa(i) + ".123")
+		var timestamptz interface{}
+		var date interface{}
+		// TODO: Remove this when the backend supports Zulu timestamp/date literals.
+		if strings.Contains(connString, "prefer_simple_protocol=true") {
+			date = fmt.Sprintf("2022-04-%02d", i+1)
+			timestamptz = fmt.Sprintf("2022-03-24 %02d:39:10.123456000+00", i)
+		} else {
+			date = &pgtype.Date{}
+			_ = date.(*pgtype.Date).Set(fmt.Sprintf("2022-04-%02d", i+1))
+			timestamptz, _ = time.Parse(time.RFC3339Nano, fmt.Sprintf("2022-03-24T%02d:39:10.123456000Z", i))
+		}
+		batch.Queue(sql, 100+i, i%2 == 0, []byte(strconv.Itoa(i)+"test_bytes"), 3.14+float64(i), i, numeric, timestamptz, date, "test_string"+strconv.Itoa(i))
+	}
+	return nil
+}
+
 //export TestWrongDialect
 func TestWrongDialect(connString string) *C.char {
 	ctx := context.Background()
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 ac11304c..639a281a 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java
@@ -65,10 +65,9 @@ public final class ControlMessageTest {
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
-    when(intermediateStatement.getStatementType(0)).thenReturn(StatementType.UPDATE);
-    when(intermediateStatement.getCommandTag(0)).thenReturn("INSERT");
-    when(intermediateStatement.getUpdateCount(0)).thenReturn(1L);
-    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(intermediateStatement.getStatementType()).thenReturn(StatementType.UPDATE);
+    when(intermediateStatement.getCommandTag()).thenReturn("INSERT");
+    when(intermediateStatement.getUpdateCount()).thenReturn(1L);
 
     JSONParser parser = new JSONParser();
     JSONObject commandMetadata = (JSONObject) parser.parse(EMPTY_COMMAND_JSON);
@@ -86,7 +85,7 @@ public final class ControlMessageTest {
     when(connectionHandler.getServer()).thenReturn(server);
 
     ControlMessage controlMessage = ControlMessage.create(connectionHandler);
-    controlMessage.sendSpannerResult(0, intermediateStatement, QueryMode.SIMPLE, 0L);
+    controlMessage.sendSpannerResult(intermediateStatement, QueryMode.SIMPLE, 0L);
 
     DataInputStream outputReader =
         new DataInputStream(new ByteArrayInputStream(buffer.toByteArray()));
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitExplicitTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitExplicitTest.java
index bd030db5..cb045079 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitExplicitTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitExplicitTest.java
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
+import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
 import com.google.spanner.v1.CommitRequest;
 import com.google.spanner.v1.ExecuteBatchDmlRequest;
 import com.google.spanner.v1.ExecuteSqlRequest;
@@ -31,12 +32,13 @@ import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.Collections;
 import java.util.List;
+import java.util.stream.Collectors;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.junit.runners.Parameterized;
 
-@RunWith(JUnit4.class)
+@RunWith(Parameterized.class)
 public class DdlTransactionModeAutocommitExplicitTest
     extends DdlTransactionModeAutocommitImplicitTest {
   @BeforeClass
@@ -162,4 +164,24 @@ public class DdlTransactionModeAutocommitExplicitTest
     // The transaction fails and is automatically rolled back.
     assertEquals(1, mockSpanner.countRequestsOfType(RollbackRequest.class));
   }
+
+  @Test
+  public void testSingleDdlStatementInExplicitTransaction() throws SQLException {
+    String sql = "CREATE TABLE foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        statement.execute("begin");
+        assertFalse(statement.execute(sql));
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitImplicitTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitImplicitTest.java
index ef2f127d..e34aaf5c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitImplicitTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeAutocommitImplicitTest.java
@@ -14,13 +14,13 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.TRANSACTION_ABORTED_ERROR;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
-import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
 import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.CommitRequest;
@@ -38,9 +38,9 @@ import java.util.stream.Collectors;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.junit.runners.Parameterized;
 
-@RunWith(JUnit4.class)
+@RunWith(Parameterized.class)
 public class DdlTransactionModeAutocommitImplicitTest extends DdlTransactionModeBatchTest {
   @BeforeClass
   public static void startMockSpannerAndPgAdapterServers() throws Exception {
@@ -162,8 +162,7 @@ public class DdlTransactionModeAutocommitImplicitTest extends DdlTransactionMode
             assertThrows(
                 SQLException.class, () -> statement.execute("show transaction isolation level"));
         assertTrue(
-            exception.getMessage(),
-            exception.getMessage().contains(IntermediateStatement.TRANSACTION_ABORTED_ERROR));
+            exception.getMessage(), exception.getMessage().contains(TRANSACTION_ABORTED_ERROR));
       }
     }
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeBatchTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeBatchTest.java
index 2cc92e23..a75b4d64 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeBatchTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeBatchTest.java
@@ -14,12 +14,12 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.TRANSACTION_ABORTED_ERROR;
 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.pgadapter.statements.IntermediateStatement;
 import com.google.spanner.v1.BeginTransactionRequest;
 import com.google.spanner.v1.CommitRequest;
 import com.google.spanner.v1.RollbackRequest;
@@ -31,9 +31,9 @@ import java.util.Collections;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.junit.runners.Parameterized;
 
-@RunWith(JUnit4.class)
+@RunWith(Parameterized.class)
 public class DdlTransactionModeBatchTest extends DdlTransactionModeNoneTest {
 
   @BeforeClass
@@ -159,8 +159,7 @@ public class DdlTransactionModeBatchTest extends DdlTransactionModeNoneTest {
             assertThrows(
                 SQLException.class, () -> statement.execute("show transaction isolation level"));
         assertTrue(
-            exception.getMessage(),
-            exception.getMessage().contains(IntermediateStatement.TRANSACTION_ABORTED_ERROR));
+            exception.getMessage(), exception.getMessage().contains(TRANSACTION_ABORTED_ERROR));
       }
     }
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeNoneTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeNoneTest.java
index ceb9a592..9152f6cf 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeNoneTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DdlTransactionModeNoneTest.java
@@ -24,16 +24,31 @@ import java.sql.Connection;
 import java.sql.DriverManager;
 import java.sql.SQLException;
 import java.sql.Statement;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.stream.Collectors;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameter;
+import org.junit.runners.Parameterized.Parameters;
 
-@RunWith(JUnit4.class)
+@RunWith(Parameterized.class)
 public class DdlTransactionModeNoneTest extends AbstractMockServerTest {
+  @Parameter public String preferQueryMode;
+
+  @Parameters(name = "preferQueryMode = {0}")
+  public static List<Object[]> data() {
+    String[] queryModes = {"extended", "simple"};
+    List<Object[]> parameters = new ArrayList<>();
+    for (String queryMode : queryModes) {
+      parameters.add(new Object[] {queryMode});
+    }
+    return parameters;
+  }
+
   @BeforeClass
   public static void loadPgJdbcDriver() throws Exception {
     // Make sure the PG JDBC driver is loaded.
@@ -45,14 +60,10 @@ public class DdlTransactionModeNoneTest extends AbstractMockServerTest {
     doStartMockSpannerAndPgAdapterServers("d", Collections.singleton("-ddl=Single"));
   }
 
-  /**
-   * 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.
-   */
   protected String createUrl() {
     return String.format(
-        "jdbc:postgresql://localhost:%d/?preferQueryMode=simple", pgServer.getLocalPort());
+        "jdbc:postgresql://localhost:%d/?preferQueryMode=%s",
+        pgServer.getLocalPort(), preferQueryMode);
   }
 
   @Test
@@ -81,6 +92,31 @@ public class DdlTransactionModeNoneTest extends AbstractMockServerTest {
         updateDatabaseDdlRequests.get(0).getStatements(0));
   }
 
+  @Test
+  public void testSingleDdlStatementInExplicitTransaction() throws SQLException {
+    String sql = "CREATE TABLE foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        statement.execute("begin");
+        SQLException exception = assertThrows(SQLException.class, () -> statement.execute(sql));
+        assertTrue(
+            exception.getMessage(),
+            exception
+                .getMessage()
+                .contains("DDL statements are only allowed outside explicit transactions."));
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(0, updateDatabaseDdlRequests.size());
+  }
+
   @Test
   public void testDdlBatch() throws SQLException {
     String sql =
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ErrorHandlingTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ErrorHandlingTest.java
index 21e616dc..0bde76bb 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ErrorHandlingTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ErrorHandlingTest.java
@@ -14,7 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter;
 
-import static com.google.cloud.spanner.pgadapter.statements.IntermediateStatement.TRANSACTION_ABORTED_ERROR;
+import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.TRANSACTION_ABORTED_ERROR;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
@@ -118,6 +118,9 @@ public class ErrorHandlingTest extends AbstractMockServerTest {
 
       // Committing the transaction will actually execute a rollback.
       connection.commit();
+
+      // The connection should now be usable.
+      assertTrue(connection.createStatement().execute("show transaction isolation level"));
     }
     // Check that we only received a rollback and no commit.
     assertTrue(mockSpanner.countRequestsOfType(RollbackRequest.class) > 0);
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 fe32ed80..9c3b6ce3 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
@@ -529,7 +529,7 @@ public class ITJdbcTest implements IntegrationTest {
   }
 
   @Test
-  public void testTwoDmlStatementsInSimpleMode() throws SQLException {
+  public void testTwoDmlStatements() throws SQLException {
     try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
       try (java.sql.Statement statement = connection.createStatement()) {
         // Statement#execute(String) returns false if the result is an update count or no result.
@@ -569,7 +569,7 @@ public class ITJdbcTest implements IntegrationTest {
   }
 
   @Test
-  public void testDmlAndQueryInBatchInSimpleMode() throws SQLException {
+  public void testDmlAndQueryInBatch() throws SQLException {
     try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
       try (java.sql.Statement statement = connection.createStatement()) {
         assertFalse(
@@ -594,6 +594,43 @@ public class ITJdbcTest implements IntegrationTest {
     }
   }
 
+  @Test
+  public void testDml() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      try (java.sql.PreparedStatement statement =
+          connection.prepareStatement("INSERT INTO numbers VALUES (?, ?)")) {
+        statement.setLong(1, 2L);
+        statement.setString(2, "Two");
+        statement.addBatch();
+
+        statement.setLong(1, 3L);
+        statement.setString(2, "Three");
+        statement.addBatch();
+
+        statement.setLong(1, 4L);
+        statement.setString(2, "Four");
+        statement.addBatch();
+
+        int[] updateCounts = statement.executeBatch();
+        assertArrayEquals(new int[] {1, 1, 1}, updateCounts);
+
+        // Read back the data to verify.
+        try (ResultSet resultSet =
+            connection.createStatement().executeQuery("SELECT name FROM numbers ORDER BY num")) {
+          assertTrue(resultSet.next());
+          assertEquals("One", resultSet.getString("name"));
+          assertTrue(resultSet.next());
+          assertEquals("Two", resultSet.getString("name"));
+          assertTrue(resultSet.next());
+          assertEquals("Three", resultSet.getString("name"));
+          assertTrue(resultSet.next());
+          assertEquals("Four", resultSet.getString("name"));
+          assertFalse(resultSet.next());
+        }
+      }
+    }
+  }
+
   @Test
   public void testFetchSize() throws SQLException {
     writeExtraTestRows();
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITParameterizedQueryTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITParameterizedQueryTest.java
deleted file mode 100644
index a199560e..00000000
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITParameterizedQueryTest.java
+++ /dev/null
@@ -1,320 +0,0 @@
-// Copyright 2021 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.assertArrayEquals;
-
-import com.google.cloud.spanner.Database;
-import com.google.common.collect.ImmutableList;
-import com.google.common.primitives.Bytes;
-import java.io.DataInputStream;
-import java.io.DataOutputStream;
-import java.net.Socket;
-import java.nio.ByteBuffer;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.junit.experimental.categories.Category;
-import org.junit.runner.RunWith;
-import org.junit.runners.JUnit4;
-import org.postgresql.core.Oid;
-
-@Category(IntegrationTest.class)
-@RunWith(JUnit4.class)
-public final class ITParameterizedQueryTest implements IntegrationTest {
-  private static final PgAdapterTestEnv testEnv = new PgAdapterTestEnv();
-  private static Database database;
-
-  private Socket clientSocket;
-  private DataOutputStream out;
-  private DataInputStream in;
-
-  @BeforeClass
-  public static void setup() {
-    testEnv.setUp();
-    database = testEnv.createDatabase(getDdlStatements());
-    testEnv.startPGAdapterServerWithDefaultDatabase(database.getId(), Collections.emptyList());
-  }
-
-  @AfterClass
-  public static void teardown() {
-    testEnv.stopPGAdapterServer();
-    testEnv.cleanUp();
-  }
-
-  @Before
-  public void insertTestData() {
-    List<String> values =
-        new ArrayList<>(Arrays.asList("(1, 1, '1')", "(2, 20, 'Joe')", "(3, 23, 'Jack')"));
-    String dml = "INSERT INTO users (id, age, name) VALUES " + String.join(", ", values);
-    testEnv.updateTables(
-        database.getId().getDatabase(), ImmutableList.of("delete from users", dml));
-  }
-
-  private static Iterable<String> getDdlStatements() {
-    return Collections.singletonList(
-        "CREATE TABLE users (\n"
-            + "  id   bigint PRIMARY KEY,\n"
-            + "  age  bigint,\n"
-            + "  name   text\n"
-            + ");");
-  }
-
-  @After
-  public void closeSocket() throws Exception {
-    clientSocket.close();
-    out.close();
-    in.close();
-  }
-
-  // '\0' for name will refer to an unnamed statement.
-  private byte[] constructParseMessage(
-      short parameters, String query, String name, ArrayList<Integer> oidTypes) {
-    // Unnamed prepared statement = "\0"
-    ByteBuffer metadata = ByteBuffer.allocate(5);
-    metadata.put(0, (byte) 'P');
-    // Message size: 4 additional bytes per parameter + 4 for size + 2 for number of parameters
-    metadata.putInt(1, name.length() + query.length() + parameters * 4 + 6);
-
-    ByteBuffer typeBuffer = ByteBuffer.allocate(parameters * 4 + 2);
-    typeBuffer.putShort(parameters);
-    for (Integer oidType : oidTypes) {
-      typeBuffer.putInt(oidType);
-    }
-    return Bytes.concat(metadata.array(), name.getBytes(), query.getBytes(), typeBuffer.array());
-  }
-
-  // '\0' for statementName or portalName will refer to an unnamed statement.
-  private byte[] constructBindMessage(
-      String statementName, String portalName, ArrayList<PgAdapterTestEnv.Parameter> values) {
-    // 5 bytes: 1 byte for message type + 4 for size
-    ByteBuffer metadata = ByteBuffer.allocate(5);
-    metadata.put(0, (byte) 'B');
-    int valueSize = 0;
-    for (PgAdapterTestEnv.Parameter value : values) {
-      // 4 bytes for size field + parameter value size
-      valueSize += value.getSize() + 4;
-    }
-    // Message size:  4 bytes for size +
-    //                total size of parameters (includes parameter sizes) +
-    //                2 bytes per parameter for parameter length +
-    //                2 bytes for number of parameters +
-    //                2 bytes for number of parameter format codes +
-    //                2 bytes indicating number of result column format feilds (0 == use default on
-    // all)
-    metadata.putInt(
-        1, portalName.length() + statementName.length() + valueSize + values.size() * 2 + 10);
-
-    // 2 bytes for number of format codes + 2 bytes per format code + 2 bytes for number of
-    // parameters
-    ByteBuffer parameterData = ByteBuffer.allocate(values.size() * 2 + 4);
-    parameterData.putShort((short) values.size());
-    for (int i = 0; i < values.size(); ++i) {
-      // 0 = TEXT, 1 = BINARY
-      parameterData.putShort((short) 0);
-    }
-    // Number of parameters
-    parameterData.putShort((short) values.size());
-
-    // 4 bytes per parameter for size + size of paramter itself
-    byte[] parameterValues = {};
-    for (PgAdapterTestEnv.Parameter value : values) {
-      byte[] sizeArray = ByteBuffer.allocate(4).putInt(value.getSize()).array();
-      parameterValues = Bytes.concat(parameterValues, sizeArray, value.getValue());
-    }
-    // 2 bytes for number of result column format codes (0 == use default for all)
-    ByteBuffer trailingMetadata = ByteBuffer.allocate(2);
-    trailingMetadata.putShort((short) 0);
-
-    return Bytes.concat(
-        metadata.array(),
-        portalName.getBytes(),
-        statementName.getBytes(),
-        parameterData.array(),
-        parameterValues,
-        trailingMetadata.array());
-  }
-
-  // Constructs a Describe message that will describe either a portal (portal == true) or a prepared
-  // statement (portal == false)
-  private byte[] constructDescribeMessage(boolean portal, String name) {
-    // 6 bytes: 1 byte for message type + 4 for size + 1 for Portal/Statement type
-    ByteBuffer metadata = ByteBuffer.allocate(6);
-    metadata.put(0, (byte) 'D');
-    metadata.putInt(1, name.length() + 5);
-    if (portal) {
-      // Portal
-      metadata.put(5, (byte) 'P');
-    } else {
-      // Statement
-      metadata.put(5, (byte) 'S');
-    }
-    return Bytes.concat(metadata.array(), name.getBytes());
-  }
-
-  private byte[] constructExecuteMessage(String portalName) {
-    // 5 bytes: 1 byte for message type + 4 for size
-    ByteBuffer metadata = ByteBuffer.allocate(5);
-    metadata.put(0, (byte) 'E');
-    // 8 bytes: 4 bytes for size (includes self) + 4 for maximum number of return rows (0 == no
-    // limit)
-    metadata.putInt(1, portalName.length() + 8);
-    ByteBuffer limit = ByteBuffer.allocate(4);
-    limit.putInt(0, 0);
-    return Bytes.concat(metadata.array(), portalName.getBytes(), limit.array());
-  }
-
-  // If portal is true this will close the corresponding portal, if false this will close the
-  // corresponding prepared statement.
-  private byte[] constructCloseMessage(boolean portal, String name) {
-    // 6 bytes: 1 byte for message type + 4 for size + 1 for Portal/Statement type
-    ByteBuffer metadata = ByteBuffer.allocate(6);
-    metadata.put(0, (byte) 'C');
-    metadata.putInt(1, name.length() + 5);
-    if (portal) {
-      // Portal
-      metadata.put(5, (byte) 'P');
-    } else {
-      // Statement
-      metadata.put(5, (byte) 'S');
-    }
-    return Bytes.concat(metadata.array(), name.getBytes());
-  }
-
-  private byte[] constructSyncMessage() {
-    // 5 bytes: 1 byte for message type + 4 for size
-    ByteBuffer metadata = ByteBuffer.allocate(5);
-    metadata.put(0, (byte) 'S');
-    metadata.putInt(1, 4);
-    return metadata.array();
-  }
-
-  @Test
-  public void unnamedParameterTest() throws Exception {
-    testEnv.waitForServer();
-
-    clientSocket = new Socket(testEnv.getPGAdapterHost(), testEnv.getPGAdapterPort());
-    out = new DataOutputStream(clientSocket.getOutputStream());
-    in = new DataInputStream(clientSocket.getInputStream());
-    testEnv.initializeConnection(out);
-    testEnv.consumeStartUpMessages(in);
-
-    // Send Parse message.
-    ArrayList<Integer> oidList = new ArrayList<>();
-    oidList.add(Oid.INT8);
-    byte[] parseMessage =
-        constructParseMessage((short) 1, "SELECT * FROM users WHERE id = $1\0", "\0", oidList);
-    out.write(parseMessage, 0, parseMessage.length);
-    // Check for parse complete message.
-    testEnv.consumePGMessage('1', in);
-
-    // Send Bind message.
-    ArrayList<PgAdapterTestEnv.Parameter> paramList = new ArrayList<>();
-    String param1 = "1";
-    paramList.add(new PgAdapterTestEnv.Parameter(Oid.INT8, param1));
-    byte[] bindMessage = constructBindMessage("\0", "\0", paramList);
-    out.write(bindMessage, 0, bindMessage.length);
-    // Check for bind complete message.
-    testEnv.consumePGMessage('2', in);
-
-    // Send Execute message.
-    byte[] exectueMessage = constructExecuteMessage("\0");
-    out.write(exectueMessage, 0, exectueMessage.length);
-    PgAdapterTestEnv.PGMessage[] dataRows = {testEnv.consumePGMessage('D', in)};
-    PgAdapterTestEnv.PGMessage commandComplete = testEnv.consumePGMessage('C', in);
-
-    // Send Close message.
-    byte[] closeMessage = constructCloseMessage(true, "\0");
-    out.write(closeMessage, 0, closeMessage.length);
-    testEnv.consumePGMessage('3', in);
-
-    // Send Sync message.
-    byte[] syncMessage = constructSyncMessage();
-    out.write(syncMessage, 0, syncMessage.length);
-    PgAdapterTestEnv.PGMessage readyForQuery = testEnv.consumePGMessage('Z', in);
-
-    // Verify results
-    byte[] dataRow0 = {0, 3, 0, 0, 0, 1, 49, 0, 0, 0, 1, 49, 0, 0, 0, 1, 49};
-    byte[] commandCompleteData = {83, 69, 76, 69, 67, 84, 32, 49, 0};
-    byte[] readyForQueryData = {73};
-
-    assertArrayEquals(dataRow0, dataRows[0].getPayload());
-    assertArrayEquals(commandCompleteData, commandComplete.getPayload());
-    assertArrayEquals(readyForQueryData, readyForQuery.getPayload());
-  }
-
-  @Test
-  public void parameterTest() throws Exception {
-    testEnv.waitForServer();
-    String statementName = "test-statement\0";
-    String portalName = "test-portal\0";
-
-    clientSocket = new Socket(testEnv.getPGAdapterHost(), testEnv.getPGAdapterPort());
-    out = new DataOutputStream(clientSocket.getOutputStream());
-    in = new DataInputStream(clientSocket.getInputStream());
-    testEnv.initializeConnection(out);
-    testEnv.consumeStartUpMessages(in);
-
-    // Send Parse message.
-    ArrayList<Integer> oidList = new ArrayList<>();
-    oidList.add(Oid.INT8);
-    byte[] parseMessage =
-        constructParseMessage(
-            (short) 1, "SELECT * FROM users WHERE id = $1\0", statementName, oidList);
-    out.write(parseMessage, 0, parseMessage.length);
-    // Check for parse complete message.
-    testEnv.consumePGMessage('1', in);
-
-    // Send Bind message.
-    ArrayList<PgAdapterTestEnv.Parameter> paramList = new ArrayList<>();
-    String param1 = "1";
-    paramList.add(new PgAdapterTestEnv.Parameter(Oid.INT8, param1));
-    byte[] bindMessage = constructBindMessage(statementName, portalName, paramList);
-    out.write(bindMessage, 0, bindMessage.length);
-    // Check for bind complete message.
-    testEnv.consumePGMessage('2', in);
-
-    // Send Execute message.
-    byte[] exectueMessage = constructExecuteMessage(portalName);
-    out.write(exectueMessage, 0, exectueMessage.length);
-    PgAdapterTestEnv.PGMessage[] dataRows = {testEnv.consumePGMessage('D', in)};
-    PgAdapterTestEnv.PGMessage commandComplete = testEnv.consumePGMessage('C', in);
-
-    // Send Close message.
-    byte[] closeMessage = constructCloseMessage(true, portalName);
-    out.write(closeMessage, 0, closeMessage.length);
-    testEnv.consumePGMessage('3', in);
-
-    // Send Sync message.
-    byte[] syncMessage = constructSyncMessage();
-    out.write(syncMessage, 0, syncMessage.length);
-    PgAdapterTestEnv.PGMessage readyForQuery = testEnv.consumePGMessage('Z', in);
-
-    // Verify results
-    byte[] dataRow0 = {0, 3, 0, 0, 0, 1, 49, 0, 0, 0, 1, 49, 0, 0, 0, 1, 49};
-    byte[] commandCompleteData = {83, 69, 76, 69, 67, 84, 32, 49, 0};
-    byte[] readyForQueryData = {73};
-
-    assertArrayEquals(dataRow0, dataRows[0].getPayload());
-    assertArrayEquals(commandCompleteData, commandComplete.getPayload());
-    assertArrayEquals(readyForQueryData, readyForQuery.getPayload());
-  }
-}
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 ae32c5bb..f720ea73 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
@@ -33,6 +33,7 @@ import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.PreparedTy
 import com.google.cloud.spanner.pgadapter.wireprotocol.DescribeMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ExecuteMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
+import com.google.common.collect.ImmutableList;
 import com.google.protobuf.Value;
 import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
 import com.google.spanner.v1.ExecuteBatchDmlRequest;
@@ -260,6 +261,56 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     }
   }
 
+  @Test
+  public void testQueryWithNonExistingTable() throws SQLException {
+    String sql = "select * from non_existing_table where id=$1";
+    mockSpanner.putStatementResult(
+        StatementResult.exception(
+            Statement.of(sql),
+            Status.NOT_FOUND
+                .withDescription("Table non_existing_table not found")
+                .asRuntimeException()));
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+        SQLException exception = assertThrows(SQLException.class, preparedStatement::executeQuery);
+        assertTrue(
+            exception.getMessage(),
+            exception.getMessage().contains("Table non_existing_table not found"));
+      }
+    }
+
+    // PGAdapter tries to execute the query directly when describing the portal, so we receive one
+    // ExecuteSqlRequest in normal execute mode.
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(sql, requests.get(0).getSql());
+    assertEquals(QueryMode.NORMAL, requests.get(0).getQueryMode());
+  }
+
+  @Test
+  public void testDmlWithNonExistingTable() throws SQLException {
+    String sql = "update non_existing_table set value=$2 where id=$1";
+    mockSpanner.putStatementResult(
+        StatementResult.exception(
+            Statement.of(sql),
+            Status.NOT_FOUND
+                .withDescription("Table non_existing_table not found")
+                .asRuntimeException()));
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+        SQLException exception = assertThrows(SQLException.class, preparedStatement::executeUpdate);
+        assertTrue(
+            exception.getMessage(),
+            exception.getMessage().contains("Table non_existing_table not found"));
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(sql, requests.get(0).getSql());
+    assertEquals(QueryMode.NORMAL, requests.get(0).getQueryMode());
+  }
+
   @Test
   public void testNullValues() throws SQLException {
     mockSpanner.putStatementResult(
@@ -443,12 +494,14 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     }
 
     // The DML statements are split by the JDBC driver and sent as separate statements to PgAdapter.
-    // PgAdapter therefore also simply executes these as separate DML statements.
-    assertTrue(mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).isEmpty());
-    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-    assertEquals(2, requests.size());
-    assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
-    assertEquals(UPDATE_STATEMENT.getSql(), requests.get(1).getSql());
+    // The Sync message is however sent after the second DML statement, which means that PGAdapter
+    // is able to batch these together into one ExecuteBatchDml statement.
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteBatchDmlRequest.class));
+    ExecuteBatchDmlRequest request =
+        mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).get(0);
+    assertEquals(2, request.getStatementsCount());
+    assertEquals(INSERT_STATEMENT.getSql(), request.getStatements(0).getSql());
+    assertEquals(UPDATE_STATEMENT.getSql(), request.getStatements(1).getSql());
   }
 
   @Test
@@ -465,15 +518,15 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
       }
     }
 
-    // The PostgreSQL JDBC driver will send the DML statements as separated statements to PG when
-    // executing a batch using simple mode. This means that Spanner will receive two separate DML
-    // requests.
-    assertTrue(mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).isEmpty());
-    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
-
-    assertEquals(2, requests.size());
-    assertEquals(INSERT_STATEMENT.getSql(), requests.get(0).getSql());
-    assertEquals(UPDATE_STATEMENT.getSql(), requests.get(1).getSql());
+    // The PostgreSQL JDBC driver will send the DML statements as separated statements to PG, but it
+    // will only send a Sync after the second statement. This means that PGAdapter is able to batch
+    // these together in one ExecuteBatchDml request.
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteBatchDmlRequest.class));
+    ExecuteBatchDmlRequest request =
+        mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).get(0);
+    assertEquals(2, request.getStatementsCount());
+    assertEquals(INSERT_STATEMENT.getSql(), request.getStatements(0).getSql());
+    assertEquals(UPDATE_STATEMENT.getSql(), request.getStatements(1).getSql());
   }
 
   @Test
@@ -530,6 +583,37 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     assertEquals(sql, updateDatabaseDdlRequests.get(0).getStatements(0));
   }
 
+  @Test
+  public void testDdlBatch() throws SQLException {
+    ImmutableList<String> statements =
+        ImmutableList.of(
+            "CREATE TABLE foo (id bigint primary key)",
+            "CREATE TABLE bar (id bigint primary key, value text)",
+            "CREATE INDEX idx_foo ON bar (text)");
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        for (String sql : statements) {
+          statement.addBatch(sql);
+        }
+        int[] updateCounts = statement.executeBatch();
+        assertArrayEquals(new int[] {0, 0, 0}, updateCounts);
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(3, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    for (int i = 0; i < statements.size(); i++) {
+      assertEquals(statements.get(i), updateDatabaseDdlRequests.get(0).getStatements(i));
+    }
+  }
+
   @Test
   public void testPreparedStatement() throws SQLException {
     mockSpanner.putStatementResult(
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
index bf3c0929..e8242b57 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcSimpleModeMockServerTest.java
@@ -14,7 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter;
 
-import static com.google.cloud.spanner.pgadapter.statements.IntermediateStatement.TRANSACTION_ABORTED_ERROR;
+import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.TRANSACTION_ABORTED_ERROR;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -120,6 +120,55 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
     assertTrue(request.getTransaction().getSingleUse().hasReadOnly());
   }
 
+  @Test
+  public void testDml() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result is an update count or no result.
+        assertFalse(statement.execute(INSERT_STATEMENT.getSql()));
+        assertEquals(1, statement.getUpdateCount());
+        // There are no more results. This is indicated by getMoreResults returning false AND
+        // getUpdateCount returning -1.
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    ExecuteSqlRequest request = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0);
+    assertEquals(QueryMode.NORMAL, request.getQueryMode());
+    assertEquals(INSERT_STATEMENT.getSql(), request.getSql());
+    assertTrue(request.getTransaction().hasBegin());
+    assertTrue(request.getTransaction().getBegin().hasReadWrite());
+
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertEquals(1, commitRequests.size());
+  }
+
+  @Test
+  public void testBegin() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute("BEGIN"));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+  }
+
+  @Test
+  public void testCommit() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute("COMMIT"));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, statement.getUpdateCount());
+      }
+    }
+  }
+
   @Test
   public void testEmptyStatement() throws SQLException {
     String sql = "";
@@ -132,6 +181,68 @@ public class JdbcSimpleModeMockServerTest extends AbstractMockServerTest {
     assertEquals(0, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
   }
 
+  @Test
+  public void testInvalidDml() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        SQLException exception =
+            assertThrows(SQLException.class, () -> statement.execute(INVALID_DML.getSql()));
+        assertThat(
+            exception.getMessage(), containsString("INVALID_ARGUMENT: Statement is invalid."));
+
+        // Verify that the transaction was rolled back and that the connection is usable.
+        assertTrue(statement.execute("show transaction isolation level"));
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(INVALID_DML.getSql(), requests.get(0).getSql());
+    assertTrue(requests.get(0).getTransaction().hasBegin());
+    // There is no Rollback request, because the first statement failed and did not return a
+    // transaction.
+    assertEquals(0, mockSpanner.countRequestsOfType(RollbackRequest.class));
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+  }
+
+  @Test
+  public void testInvalidQuery() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        SQLException exception =
+            assertThrows(SQLException.class, () -> statement.execute(INVALID_SELECT.getSql()));
+        assertThat(
+            exception.getMessage(), containsString("INVALID_ARGUMENT: Statement is invalid."));
+
+        // Verify that the transaction was rolled back and that the connection is usable.
+        assertTrue(statement.execute("show transaction isolation level"));
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(INVALID_SELECT.getSql(), requests.get(0).getSql());
+    assertTrue(requests.get(0).getTransaction().hasSingleUse());
+    assertTrue(requests.get(0).getTransaction().getSingleUse().hasReadOnly());
+  }
+
+  @Test
+  public void testInvalidDdl() throws SQLException {
+    addDdlExceptionToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        SQLException exception =
+            assertThrows(SQLException.class, () -> statement.execute(INVALID_DDL.getSql()));
+        assertThat(
+            exception.getMessage(), containsString("INVALID_ARGUMENT: Statement is invalid."));
+
+        // Verify that the transaction was rolled back and that the connection is usable.
+        assertTrue(statement.execute("show transaction isolation level"));
+      }
+    }
+  }
+
   @Test
   public void testEmptyStatementFollowedByNonEmptyStatement() throws SQLException {
     try (Connection connection = DriverManager.getConnection(createUrl())) {
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/golang/ITPgxTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/golang/ITPgxTest.java
index b40b8695..7be53f86 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/golang/ITPgxTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/golang/ITPgxTest.java
@@ -14,14 +14,21 @@
 
 package com.google.cloud.spanner.pgadapter.golang;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 
 import com.google.cloud.ByteArray;
 import com.google.cloud.Date;
 import com.google.cloud.Timestamp;
 import com.google.cloud.spanner.Database;
+import com.google.cloud.spanner.DatabaseClient;
 import com.google.cloud.spanner.Key;
+import com.google.cloud.spanner.KeySet;
 import com.google.cloud.spanner.Mutation;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.pgadapter.IntegrationTest;
 import com.google.cloud.spanner.pgadapter.PgAdapterTestEnv;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
@@ -179,4 +186,62 @@ public class ITPgxTest implements IntegrationTest {
 
     assertNull(pgxTest.TestInsertNullsAllDataTypes(createConnString()));
   }
+
+  @Test
+  public void testInsertBatch() {
+    // Make sure the table is empty before we execute the test.
+    String databaseId = database.getId().getDatabase();
+    testEnv.write(
+        databaseId, Collections.singletonList(Mutation.delete("all_types", KeySet.all())));
+    DatabaseClient client = testEnv.getSpanner().getDatabaseClient(database.getId());
+    try (ResultSet resultSet =
+        client.singleUse().executeQuery(Statement.of("SELECT COUNT(*) FROM all_types"))) {
+      assertTrue(resultSet.next());
+      assertEquals(0L, resultSet.getLong(0));
+      assertFalse(resultSet.next());
+    }
+
+    assertNull(pgxTest.TestInsertBatch(createConnString()));
+
+    final long batchSize = 10L;
+    try (ResultSet resultSet =
+        client.singleUse().executeQuery(Statement.of("SELECT COUNT(*) FROM all_types"))) {
+      assertTrue(resultSet.next());
+      assertEquals(batchSize, resultSet.getLong(0));
+      assertFalse(resultSet.next());
+    }
+  }
+
+  @Test
+  public void testMixedBatch() {
+    // Make sure the table is empty before we execute the test.
+    String databaseId = database.getId().getDatabase();
+    testEnv.write(
+        databaseId, Collections.singletonList(Mutation.delete("all_types", KeySet.all())));
+    DatabaseClient client = testEnv.getSpanner().getDatabaseClient(database.getId());
+    try (ResultSet resultSet =
+        client.singleUse().executeQuery(Statement.of("SELECT COUNT(*) FROM all_types"))) {
+      assertTrue(resultSet.next());
+      assertEquals(0L, resultSet.getLong(0));
+      assertFalse(resultSet.next());
+    }
+
+    assertNull(pgxTest.TestMixedBatch(createConnString()));
+
+    final long batchSize = 5L;
+    try (ResultSet resultSet =
+        client.singleUse().executeQuery(Statement.of("SELECT COUNT(*) FROM all_types"))) {
+      assertTrue(resultSet.next());
+      assertEquals(batchSize, resultSet.getLong(0));
+      assertFalse(resultSet.next());
+    }
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(Statement.of("SELECT COUNT(*) FROM all_types WHERE col_bool=true"))) {
+      assertTrue(resultSet.next());
+      assertEquals(0L, resultSet.getLong(0));
+      assertFalse(resultSet.next());
+    }
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxMockServerTest.java
index bf133fe4..59375e6c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxMockServerTest.java
@@ -28,8 +28,12 @@ import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.pgadapter.AbstractMockServerTest;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.common.collect.ImmutableList;
+import com.google.protobuf.AbstractMessage;
+import com.google.protobuf.ByteString;
 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.ResultSet;
@@ -40,6 +44,7 @@ import com.google.spanner.v1.Type;
 import com.google.spanner.v1.TypeCode;
 import java.io.IOException;
 import java.util.List;
+import java.util.stream.Collectors;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -292,6 +297,358 @@ public class PgxMockServerTest extends AbstractMockServerTest {
     assertEquals(QueryMode.NORMAL, executeRequest.getQueryMode());
   }
 
+  @Test
+  public void testInsertBatch() {
+    String sql =
+        "INSERT INTO all_types "
+            + "(col_bigint, col_bool, col_bytea, col_float8, col_int, col_numeric, col_timestamptz, col_date, col_varchar) "
+            + "values ($1, $2, $3, $4, $5, $6, $7, $8, $9)";
+    String describeSql =
+        "select $1, $2, $3, $4, $5, $6, $7, $8, $9 from (select col_bigint=$1, col_bool=$2, col_bytea=$3, col_float8=$4, col_int=$5, col_numeric=$6, col_timestamptz=$7, col_date=$8, col_varchar=$9 from all_types) p";
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(describeSql),
+            ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(
+                            TypeCode.INT64,
+                            TypeCode.BOOL,
+                            TypeCode.BYTES,
+                            TypeCode.FLOAT64,
+                            TypeCode.INT64,
+                            TypeCode.NUMERIC,
+                            TypeCode.TIMESTAMP,
+                            TypeCode.DATE,
+                            TypeCode.STRING)))
+                .build()));
+    mockSpanner.putStatementResult(StatementResult.update(Statement.of(sql), 0L));
+    int batchSize = 10;
+    for (int i = 0; i < batchSize; i++) {
+      mockSpanner.putStatementResult(
+          StatementResult.update(
+              Statement.newBuilder(sql)
+                  .bind("p1")
+                  .to(100L + i)
+                  .bind("p2")
+                  .to(i % 2 == 0)
+                  .bind("p3")
+                  .to(ByteArray.copyFrom(i + "test_bytes"))
+                  .bind("p4")
+                  .to(3.14d + i)
+                  .bind("p5")
+                  .to(i)
+                  .bind("p6")
+                  .to(com.google.cloud.spanner.Value.pgNumeric(i + ".123"))
+                  .bind("p7")
+                  .to(
+                      Timestamp.parseTimestamp(
+                          String.format("2022-03-24T%02d:39:10.123456000Z", i)))
+                  .bind("p8")
+                  .to(Date.parseDate(String.format("2022-04-%02d", i + 1)))
+                  .bind("p9")
+                  .to("test_string" + i)
+                  .build(),
+              1L));
+    }
+
+    String res = pgxTest.TestInsertBatch(createConnString());
+
+    assertNull(res);
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    // pgx by default always uses prepared statements. That means that each request is sent 2 times
+    // to the backend to be described, and then `batchSize` times to be executed.
+    // 1. DescribeStatement (parameters)
+    // 2. DescribeStatement (verify validity / PARSE) -- This step could be skipped.
+    // 3. Execute 10 times.
+    assertEquals(2, requests.size());
+    ExecuteSqlRequest describeParamsRequest = requests.get(0);
+    assertEquals(describeSql, describeParamsRequest.getSql());
+    assertEquals(QueryMode.PLAN, describeParamsRequest.getQueryMode());
+    // The 'describe' query for the parameters will be executed as a single use transaction.
+    assertTrue(describeParamsRequest.getTransaction().hasSingleUse());
+
+    // The analyzeUpdate that is executed to verify the validity of the DML statement is executed as
+    // a separate transaction.
+    ExecuteSqlRequest describeRequest = requests.get(1);
+    assertEquals(sql, describeRequest.getSql());
+    assertEquals(QueryMode.PLAN, describeRequest.getQueryMode());
+    assertTrue(describeRequest.getTransaction().hasBegin());
+
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteBatchDmlRequest.class));
+    ExecuteBatchDmlRequest batchDmlRequest =
+        mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).get(0);
+    assertEquals(batchSize, batchDmlRequest.getStatementsCount());
+    assertTrue(batchDmlRequest.getTransaction().hasBegin());
+
+    // There are two commit requests, as the 'Describe statement' message is executed as a separate
+    // transaction.
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertEquals(2, commitRequests.size());
+  }
+
+  @Test
+  public void testMixedBatch() {
+    String insertSql =
+        "INSERT INTO all_types "
+            + "(col_bigint, col_bool, col_bytea, col_float8, col_int, col_numeric, col_timestamptz, col_date, col_varchar) "
+            + "values ($1, $2, $3, $4, $5, $6, $7, $8, $9)";
+    String describeInsertSql =
+        "select $1, $2, $3, $4, $5, $6, $7, $8, $9 from (select col_bigint=$1, col_bool=$2, col_bytea=$3, col_float8=$4, col_int=$5, col_numeric=$6, col_timestamptz=$7, col_date=$8, col_varchar=$9 from all_types) p";
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(describeInsertSql),
+            ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(
+                            TypeCode.INT64,
+                            TypeCode.BOOL,
+                            TypeCode.BYTES,
+                            TypeCode.FLOAT64,
+                            TypeCode.INT64,
+                            TypeCode.NUMERIC,
+                            TypeCode.TIMESTAMP,
+                            TypeCode.DATE,
+                            TypeCode.STRING)))
+                .build()));
+    mockSpanner.putStatementResult(StatementResult.update(Statement.of(insertSql), 0L));
+
+    String selectSql = "select count(*) from all_types where col_bool=$1";
+    ResultSet resultSet =
+        ResultSet.newBuilder()
+            .setMetadata(
+                ResultSetMetadata.newBuilder()
+                    .setRowType(
+                        StructType.newBuilder()
+                            .addFields(
+                                Field.newBuilder()
+                                    .setName("c")
+                                    .setType(Type.newBuilder().setCode(TypeCode.INT64).build())
+                                    .build())
+                            .build())
+                    .build())
+            .addRows(
+                ListValue.newBuilder()
+                    .addValues(Value.newBuilder().setStringValue("3").build())
+                    .build())
+            .build();
+    mockSpanner.putStatementResult(StatementResult.query(Statement.of(selectSql), resultSet));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(selectSql).bind("p1").to(true).build(), resultSet));
+
+    String describeParamsSelectSql =
+        "select $1 from (select count(*) from all_types where col_bool=$1) p";
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(describeParamsSelectSql),
+            ResultSet.newBuilder()
+                .setMetadata(createMetadata(ImmutableList.of(TypeCode.BOOL)))
+                .build()));
+
+    String updateSql = "update all_types set col_bool=false where col_bool=$1";
+    mockSpanner.putStatementResult(StatementResult.update(Statement.of(updateSql), 0L));
+    mockSpanner.putStatementResult(
+        StatementResult.update(Statement.newBuilder(updateSql).bind("p1").to(true).build(), 3L));
+    String describeUpdateSql =
+        "select $1 from (select col_bool=false from all_types where col_bool=$1) p";
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(describeUpdateSql),
+            ResultSet.newBuilder()
+                .setMetadata(createMetadata(ImmutableList.of(TypeCode.BOOL)))
+                .build()));
+
+    int batchSize = 5;
+    for (int i = 0; i < batchSize; i++) {
+      mockSpanner.putStatementResult(
+          StatementResult.update(
+              Statement.newBuilder(insertSql)
+                  .bind("p1")
+                  .to(100L + i)
+                  .bind("p2")
+                  .to(i % 2 == 0)
+                  .bind("p3")
+                  .to(ByteArray.copyFrom(i + "test_bytes"))
+                  .bind("p4")
+                  .to(3.14d + i)
+                  .bind("p5")
+                  .to(i)
+                  .bind("p6")
+                  .to(com.google.cloud.spanner.Value.pgNumeric(i + ".123"))
+                  .bind("p7")
+                  .to(
+                      Timestamp.parseTimestamp(
+                          String.format("2022-03-24T%02d:39:10.123456000Z", i)))
+                  .bind("p8")
+                  .to(Date.parseDate(String.format("2022-04-%02d", i + 1)))
+                  .bind("p9")
+                  .to("test_string" + i)
+                  .build(),
+              1L));
+    }
+
+    String res = pgxTest.TestMixedBatch(createConnString());
+
+    assertNull(res);
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    // pgx by default always uses prepared statements. That means that we get the following list of
+    // statements:
+    // 1. Describe parameters of insert statement in PLAN mode.
+    // 2. Parse insert statement in PLAN mode.
+    // 3. Describe columns of select statement in PLAN mode.
+    // 4. Describe parameters of select statement in PLAN mode.
+    // 5. Describe parameters of update statement in PLAN mode.
+    // 6. Parse update statement in PLAN mode.
+    // 7. Execute select statement.
+    // 8. Execute update statement.
+    assertEquals(8, requests.size());
+
+    // NOTE: pgx will first create prepared statements for sql strings that it does not yet know.
+    // All those describe statement messages will be executed in separate (single-use) transactions.
+    // The order in which the describe statements are executed is random.
+
+    ExecuteSqlRequest describeInsertParamsRequest =
+        requests.stream()
+            .filter(request -> request.getSql().equals(describeInsertSql))
+            .findFirst()
+            .orElse(ExecuteSqlRequest.getDefaultInstance());
+    assertEquals(describeInsertSql, describeInsertParamsRequest.getSql());
+    assertEquals(QueryMode.PLAN, describeInsertParamsRequest.getQueryMode());
+    // The 'describe' query for the parameters will be executed as a single use transaction.
+    assertTrue(describeInsertParamsRequest.getTransaction().hasSingleUse());
+
+    ExecuteSqlRequest parseInsertRequest =
+        requests.stream()
+            .filter(
+                request ->
+                    request.getSql().equals(insertSql) && request.getQueryMode() == QueryMode.PLAN)
+            .findFirst()
+            .orElse(ExecuteSqlRequest.getDefaultInstance());
+    assertEquals(insertSql, parseInsertRequest.getSql());
+    assertEquals(QueryMode.PLAN, parseInsertRequest.getQueryMode());
+    assertTrue(parseInsertRequest.getTransaction().hasBegin());
+
+    ExecuteSqlRequest describeSelectColumnsRequest =
+        requests.stream()
+            .filter(
+                request ->
+                    request.getSql().equals(selectSql) && request.getQueryMode() == QueryMode.PLAN)
+            .findFirst()
+            .orElse(ExecuteSqlRequest.getDefaultInstance());
+    assertEquals(selectSql, describeSelectColumnsRequest.getSql());
+    assertEquals(QueryMode.PLAN, describeSelectColumnsRequest.getQueryMode());
+    assertTrue(describeSelectColumnsRequest.getTransaction().hasSingleUse());
+
+    ExecuteSqlRequest describeSelectParamsRequest =
+        requests.stream()
+            .filter(request -> request.getSql().equals(describeParamsSelectSql))
+            .findFirst()
+            .orElse(ExecuteSqlRequest.getDefaultInstance());
+    assertEquals(describeParamsSelectSql, describeSelectParamsRequest.getSql());
+    assertEquals(QueryMode.PLAN, describeSelectParamsRequest.getQueryMode());
+    assertTrue(describeSelectParamsRequest.getTransaction().hasSingleUse());
+
+    ExecuteSqlRequest describeUpdateParamsRequest =
+        requests.stream()
+            .filter(request -> request.getSql().equals(describeUpdateSql))
+            .findFirst()
+            .orElse(ExecuteSqlRequest.getDefaultInstance());
+    assertEquals(describeUpdateSql, describeUpdateParamsRequest.getSql());
+    assertEquals(QueryMode.PLAN, describeUpdateParamsRequest.getQueryMode());
+    assertTrue(describeUpdateParamsRequest.getTransaction().hasSingleUse());
+
+    ExecuteSqlRequest parseUpdateRequest =
+        requests.stream()
+            .filter(
+                request ->
+                    request.getSql().equals(updateSql) && request.getQueryMode() == QueryMode.PLAN)
+            .findFirst()
+            .orElse(ExecuteSqlRequest.getDefaultInstance());
+    assertEquals(updateSql, parseUpdateRequest.getSql());
+    assertEquals(QueryMode.PLAN, parseUpdateRequest.getQueryMode());
+    assertTrue(parseUpdateRequest.getTransaction().hasBegin());
+
+    // From here we start with the actual statement execution.
+    ExecuteSqlRequest executeSelectRequest = requests.get(6);
+    assertEquals(selectSql, executeSelectRequest.getSql());
+    assertEquals(QueryMode.NORMAL, executeSelectRequest.getQueryMode());
+    // The SELECT statement should use the transaction that was started by the BatchDml request.
+    assertTrue(executeSelectRequest.getTransaction().hasId());
+
+    ExecuteSqlRequest executeUpdateRequest = requests.get(7);
+    assertEquals(updateSql, executeUpdateRequest.getSql());
+    assertEquals(QueryMode.NORMAL, executeUpdateRequest.getQueryMode());
+    assertTrue(executeUpdateRequest.getTransaction().hasId());
+
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteBatchDmlRequest.class));
+    ExecuteBatchDmlRequest batchDmlRequest =
+        mockSpanner.getRequestsOfType(ExecuteBatchDmlRequest.class).get(0);
+    assertEquals(batchSize, batchDmlRequest.getStatementsCount());
+    assertTrue(batchDmlRequest.getTransaction().hasBegin());
+
+    // There are three commit requests:
+    // 1. Describe insert statement.
+    // 2. Describe update statement.
+    // 3. Commit the actual transaction.
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertEquals(3, commitRequests.size());
+
+    // Verify that the Batch DML request was sent first, and that the following ExecuteSql requests
+    // used the same transaction ids.
+    List<AbstractMessage> allRequests =
+        mockSpanner.getRequests().stream()
+            .filter(
+                request ->
+                    request instanceof ExecuteSqlRequest
+                        || request instanceof ExecuteBatchDmlRequest
+                        || request instanceof CommitRequest)
+            .collect(Collectors.toList());
+    // 12 == 3 Commit + 1 Batch DML + 8 ExecuteSql.
+    assertEquals(12, allRequests.size());
+
+    // We don't know the exact order of the DESCRIBE requests.
+    // The order of EXECUTE requests is known and fixed.
+
+    // The (theoretical) order of DESCRIBE requests is:
+    // 1. Describe parameters of insert statement in PLAN mode.
+    // 2. Parse insert statement in PLAN mode.
+    // 3. Commit.
+    // 4. Describe columns of select statement in PLAN mode.
+    // 5. Describe parameters of select statement in PLAN mode.
+    // 6. Describe parameters of update statement in PLAN mode.
+    // 7. Parse update statement in PLAN mode.
+    // 8. Commit.
+
+    // The fixed order of EXECUTE requests is:
+    // 9. Execute insert batch (ExecuteBatchDml).
+    // 10. Execute select statement.
+    // 11. Execute update statement.
+    // 12. Commit transaction.
+    assertEquals(
+        2,
+        allRequests.subList(0, 8).stream()
+            .filter(request -> request instanceof CommitRequest)
+            .count());
+    assertEquals(
+        6,
+        allRequests.subList(0, 8).stream()
+            .filter(
+                request ->
+                    request instanceof ExecuteSqlRequest
+                        && ((ExecuteSqlRequest) request).getQueryMode() == QueryMode.PLAN)
+            .count());
+    assertEquals(ExecuteBatchDmlRequest.class, allRequests.get(8).getClass());
+    assertEquals(ExecuteSqlRequest.class, allRequests.get(9).getClass());
+    assertEquals(ExecuteSqlRequest.class, allRequests.get(10).getClass());
+    assertEquals(CommitRequest.class, allRequests.get(11).getClass());
+
+    ByteString transactionId = ((CommitRequest) allRequests.get(11)).getTransactionId();
+    assertEquals(transactionId, ((ExecuteSqlRequest) allRequests.get(9)).getTransaction().getId());
+    assertEquals(transactionId, ((ExecuteSqlRequest) allRequests.get(10)).getTransaction().getId());
+  }
+
   @Test
   public void testInsertNullsAllDataTypes() {
     String sql =
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxTest.java
index aef52f26..cfb8345d 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/golang/PgxTest.java
@@ -39,6 +39,10 @@ public interface PgxTest extends Library {
 
   String TestPrepareSelectStatement(GoString connString);
 
+  String TestInsertBatch(GoString connString);
+
+  String TestMixedBatch(GoString connString);
+
   String TestWrongDialect(GoString connString);
 
   static PgxTest compile() throws IOException, InterruptedException {
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandlerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandlerTest.java
new file mode 100644
index 00000000..6e662bb3
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandlerTest.java
@@ -0,0 +1,112 @@
+// 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.statements;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+import com.google.cloud.spanner.pgadapter.wireprotocol.BindMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.DescribeMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ExecuteMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
+import com.google.common.collect.ImmutableList;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnit;
+import org.mockito.junit.MockitoRule;
+
+@RunWith(JUnit4.class)
+public class ExtendedQueryProtocolHandlerTest {
+  @Rule public MockitoRule rule = MockitoJUnit.rule();
+
+  @Mock private BackendConnection backendConnection;
+
+  @Test
+  public void testBuffer() {
+    ParseMessage parseMessage = mock(ParseMessage.class);
+    BindMessage bindMessage = mock(BindMessage.class);
+    DescribeMessage describeMessage = mock(DescribeMessage.class);
+    ExecuteMessage executeMessage = mock(ExecuteMessage.class);
+
+    ExtendedQueryProtocolHandler handler = new ExtendedQueryProtocolHandler(backendConnection);
+    handler.buffer(parseMessage);
+    handler.buffer(bindMessage);
+    handler.buffer(describeMessage);
+    handler.buffer(executeMessage);
+
+    assertEquals(
+        ImmutableList.of(parseMessage, bindMessage, describeMessage, executeMessage),
+        handler.getMessages());
+  }
+
+  @Test
+  public void testFlush() throws Exception {
+    ParseMessage parseMessage = mock(ParseMessage.class);
+    BindMessage bindMessage = mock(BindMessage.class);
+    DescribeMessage describeMessage = mock(DescribeMessage.class);
+    ExecuteMessage executeMessage = mock(ExecuteMessage.class);
+
+    ExtendedQueryProtocolHandler handler = new ExtendedQueryProtocolHandler(backendConnection);
+    handler.buffer(parseMessage);
+    handler.buffer(bindMessage);
+    handler.buffer(describeMessage);
+    handler.buffer(executeMessage);
+    assertEquals(
+        ImmutableList.of(parseMessage, bindMessage, describeMessage, executeMessage),
+        handler.getMessages());
+
+    handler.flush();
+
+    assertEquals(0, handler.getMessages().size());
+    verify(backendConnection).flush();
+    verify(backendConnection, never()).sync();
+    verify(parseMessage).flush();
+    verify(bindMessage).flush();
+    verify(describeMessage).flush();
+    verify(executeMessage).flush();
+  }
+
+  @Test
+  public void testSync() throws Exception {
+    ParseMessage parseMessage = mock(ParseMessage.class);
+    BindMessage bindMessage = mock(BindMessage.class);
+    DescribeMessage describeMessage = mock(DescribeMessage.class);
+    ExecuteMessage executeMessage = mock(ExecuteMessage.class);
+
+    ExtendedQueryProtocolHandler handler = new ExtendedQueryProtocolHandler(backendConnection);
+    handler.buffer(parseMessage);
+    handler.buffer(bindMessage);
+    handler.buffer(describeMessage);
+    handler.buffer(executeMessage);
+    assertEquals(
+        ImmutableList.of(parseMessage, bindMessage, describeMessage, executeMessage),
+        handler.getMessages());
+
+    handler.sync();
+
+    assertEquals(0, handler.getMessages().size());
+    verify(backendConnection, never()).flush();
+    verify(backendConnection).sync();
+    verify(parseMessage).flush();
+    verify(bindMessage).flush();
+    verify(describeMessage).flush();
+    verify(executeMessage).flush();
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatementTest.java
index f329d294..e13f70cf 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatementTest.java
@@ -17,9 +17,10 @@ package com.google.cloud.spanner.pgadapter.statements;
 import static com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement.transformDeleteToSelectParams;
 import static com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement.transformInsertToSelectParams;
 import static com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement.transformUpdateToSelectParams;
-import static com.google.cloud.spanner.pgadapter.statements.IntermediateStatement.splitStatements;
+import static com.google.cloud.spanner.pgadapter.statements.SimpleQueryStatement.splitStatements;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
@@ -104,11 +105,11 @@ public class IntermediateStatementTest {
     when(result.getResultType()).thenReturn(ResultType.RESULT_SET);
     when(result.getResultSet()).thenReturn(resultSet);
 
-    statement.updateResultCount(0, result);
+    statement.setStatementResult(result);
 
-    assertTrue(statement.hasMoreData[0]);
-    assertEquals(-1, statement.getUpdateCount(0));
-    assertSame(resultSet, statement.getStatementResult(0));
+    assertTrue(statement.hasMoreData);
+    assertEquals(-1, statement.getUpdateCount());
+    assertSame(resultSet, statement.getStatementResult().getResultSet());
   }
 
   @Test
@@ -121,11 +122,12 @@ public class IntermediateStatementTest {
     when(result.getResultType()).thenReturn(ResultType.UPDATE_COUNT);
     when(result.getUpdateCount()).thenReturn(100L);
 
-    statement.updateResultCount(0, result);
+    statement.setStatementResult(result);
 
-    assertFalse(statement.hasMoreData[0]);
-    assertEquals(100L, statement.getUpdateCount(0));
-    assertNull(statement.getStatementResult(0));
+    assertFalse(statement.hasMoreData);
+    assertEquals(100L, statement.getUpdateCount());
+    assertNotNull(statement.getStatementResult());
+    assertEquals(ResultType.UPDATE_COUNT, statement.getStatementResult().getResultType());
   }
 
   @Test
@@ -139,11 +141,12 @@ public class IntermediateStatementTest {
     StatementResult result = mock(StatementResult.class);
     when(result.getResultType()).thenReturn(ResultType.NO_RESULT);
 
-    statement.updateResultCount(0, result);
+    statement.setStatementResult(result);
 
-    assertFalse(statement.hasMoreData[0]);
-    assertEquals(0, statement.getUpdateCount(0));
-    assertNull(statement.getStatementResult(0));
+    assertFalse(statement.hasMoreData);
+    assertEquals(0, statement.getUpdateCount());
+    assertNotNull(statement.getStatementResult());
+    assertEquals(ResultType.NO_RESULT, statement.getStatementResult().getResultType());
   }
 
   @Test
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/PSQLTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/PSQLTest.java
similarity index 87%
rename from src/test/java/com/google/cloud/spanner/pgadapter/PSQLTest.java
rename to src/test/java/com/google/cloud/spanner/pgadapter/statements/PSQLTest.java
index f35a8fa7..b46ac2ce 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/PSQLTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/PSQLTest.java
@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.google.cloud.spanner.pgadapter;
+package com.google.cloud.spanner.pgadapter.statements;
 
 import static org.junit.Assert.assertEquals;
 
@@ -21,9 +21,10 @@ import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
 import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.ProxyServer;
 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 org.json.simple.JSONObject;
 import org.json.simple.parser.JSONParser;
 import org.junit.Before;
@@ -64,6 +65,11 @@ public class PSQLTest {
     Mockito.when(options.getCommandMetadataJSON()).thenReturn(defaultCommands);
   }
 
+  private String translate(String sql) {
+    return SimpleQueryStatement.translatePotentialMetadataCommand(parse(sql), connectionHandler)
+        .getSqlWithoutComments();
+  }
+
   @Test
   public void testDescribeTranslates() {
     // PSQL equivalent: \d
@@ -94,10 +100,7 @@ public class PSQLTest {
             + "WHERE"
             + " t.table_schema = 'public'";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -124,10 +127,7 @@ public class PSQLTest {
             + " AND"
             + " LOWER(t.table_name) = LOWER('users')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -154,10 +154,7 @@ public class PSQLTest {
             + " AND"
             + " LOWER(t.table_name) = LOWER('bobby''; DROP TABLE USERS; SELECT''')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -180,10 +177,7 @@ public class PSQLTest {
             + " '' as str1,"
             + " '' as str2";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -217,10 +211,7 @@ public class PSQLTest {
             + " t.table_schema='public'"
             + " AND t.table_name = '-1'";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -255,10 +246,7 @@ public class PSQLTest {
             + " t.table_schema='public'"
             + " AND t.table_name = 'bobby''; DROP TABLE USERS; SELECT'''";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -270,10 +258,7 @@ public class PSQLTest {
             + " 'p' ORDER BY inhseqno;";
     String expected = "SELECT 1 LIMIT 0";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -285,10 +270,7 @@ public class PSQLTest {
             + " c.relname;";
     String expected = "SELECT 1 LIMIT 0";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -303,10 +285,7 @@ public class PSQLTest {
             + "ORDER BY 1;";
     String expected = "SELECT '' AS Name";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -322,10 +301,7 @@ public class PSQLTest {
             + "ORDER BY 1;";
     String expected = "SELECT '' AS Name";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -343,10 +319,7 @@ public class PSQLTest {
 
     // TODO: Add Connection#getDatabase() to Connection API and test here what happens if that
     // method throws an exception.
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -370,10 +343,7 @@ public class PSQLTest {
             + "ORDER BY 1,2;";
     String expected = "SELECT * FROM information_schema.tables";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -397,10 +367,7 @@ public class PSQLTest {
     String expected =
         "SELECT * FROM information_schema.tables WHERE LOWER(table_name) = LOWER('users')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -425,10 +392,7 @@ public class PSQLTest {
         "SELECT * FROM information_schema.tables WHERE LOWER(table_name) ="
             + " LOWER('bobby''; DROP TABLE USERS; SELECT''')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -456,10 +420,7 @@ public class PSQLTest {
     String expected =
         "SELECT table_catalog, table_schema, table_name, index_name, index_type, parent_table_name, is_unique, is_null_filtered, index_state, spanner_is_managed FROM information_schema.indexes";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -487,10 +448,7 @@ public class PSQLTest {
         "SELECT table_catalog, table_schema, table_name, index_name, index_type, parent_table_name, is_unique, is_null_filtered, index_state, spanner_is_managed FROM information_schema.indexes WHERE LOWER(index_name) ="
             + " LOWER('index')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -518,10 +476,7 @@ public class PSQLTest {
         "SELECT table_catalog, table_schema, table_name, index_name, index_type, parent_table_name, is_unique, is_null_filtered, index_state, spanner_is_managed FROM information_schema.indexes WHERE LOWER(index_name) ="
             + " LOWER('bobby''; DROP TABLE USERS; SELECT''')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -535,10 +490,7 @@ public class PSQLTest {
             + "ORDER BY 1;";
     String expected = "SELECT * FROM information_schema.schemata";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -553,10 +505,7 @@ public class PSQLTest {
     String expected =
         "SELECT * FROM information_schema.schemata WHERE LOWER(schema_name) = LOWER('schema')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -572,10 +521,7 @@ public class PSQLTest {
         "SELECT * FROM information_schema.schemata WHERE LOWER(schema_name) = LOWER('bobby''; DROP"
             + " TABLE USERS; SELECT''')";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -610,10 +556,7 @@ public class PSQLTest {
             + " table_schema = 'public' and STARTS_WITH(LOWER(table_name),"
             + " LOWER('user')) LIMIT 1000";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -647,10 +590,7 @@ public class PSQLTest {
             + " table_schema = 'public' and STARTS_WITH(LOWER(table_name),"
             + " LOWER('user')) LIMIT 1000";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -668,10 +608,7 @@ public class PSQLTest {
         "SELECT column_name AS quote_ident FROM information_schema.columns WHERE"
             + " table_name = 'user' AND STARTS_WITH(LOWER(COLUMN_NAME), LOWER('age')) LIMIT 1000";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -703,10 +640,7 @@ public class PSQLTest {
         "SELECT table_name AS quote_ident FROM information_schema.tables WHERE "
             + "table_schema = 'public' AND STARTS_WITH(LOWER(table_name), LOWER('user')) LIMIT 1000";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -738,10 +672,7 @@ public class PSQLTest {
         "SELECT table_name AS quote_ident FROM INFORMATION_SCHEMA.TABLES WHERE"
             + " STARTS_WITH(LOWER(table_name), LOWER('user')) LIMIT 1000";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -773,10 +704,7 @@ public class PSQLTest {
         "SELECT index_name AS quote_ident FROM INFORMATION_SCHEMA.INDEXES WHERE"
             + " STARTS_WITH(LOWER(index_name), LOWER('index')) LIMIT 1000";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -790,10 +718,7 @@ public class PSQLTest {
         "SELECT schema_name AS quote_ident FROM INFORMATION_SCHEMA.SCHEMATA WHERE"
             + " STARTS_WITH(LOWER(schema_name), LOWER('schema')) LIMIT 1000";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-
-    assertEquals(expected, matcherStatement.getSql());
+    assertEquals(expected, translate(sql));
   }
 
   @Test
@@ -825,12 +750,8 @@ public class PSQLTest {
     String secondSQL = "SELECT name FROM USERS WHERE age = 30;";
     String expectedSecondResult = "RESULT 2: selector=name, arg2=30, arg1=age";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(firstSQL), connectionHandler);
-    assertEquals(expectedFirstResult, matcherStatement.getSql());
-
-    matcherStatement = new MatcherStatement(options, parse(secondSQL), connectionHandler);
-    assertEquals(expectedSecondResult, matcherStatement.getSql());
+    assertEquals(expectedFirstResult, translate(firstSQL));
+    assertEquals(expectedSecondResult, translate(secondSQL));
   }
 
   @Test
@@ -855,8 +776,6 @@ public class PSQLTest {
     String sql = "SELECT * FROM USERS;";
     String expectedResult = "TABLE: USERS, EXPRESSION: *";
 
-    MatcherStatement matcherStatement =
-        new MatcherStatement(options, parse(sql), connectionHandler);
-    assertEquals(expectedResult, matcherStatement.getSql());
+    assertEquals(expectedResult, translate(sql));
   }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
similarity index 63%
rename from src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
rename to src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
index 6629e39f..506f254b 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/StatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
@@ -12,10 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.google.cloud.spanner.pgadapter;
+package com.google.cloud.spanner.pgadapter.statements;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
@@ -36,13 +37,11 @@ import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStateme
 import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
 import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.ProxyServer;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode;
-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.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage;
@@ -80,6 +79,8 @@ public class StatementTest {
   @Mock private Connection connection;
   @Mock private ConnectionHandler connectionHandler;
   @Mock private ConnectionMetadata connectionMetadata;
+  @Mock private ExtendedQueryProtocolHandler extendedQueryProtocolHandler;
+  @Mock private BackendConnection backendConnection;
   @Mock private ProxyServer server;
   @Mock private OptionsMetadata options;
   @Mock private StatementResult statementResult;
@@ -95,59 +96,45 @@ public class StatementTest {
 
   @Test
   public void testBasicSelectStatement() throws Exception {
-    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);
-
+    String sql = "SELECT * FROM users";
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(options, parse("SELECT * FROM users"), connectionHandler);
+    IntermediatePortalStatement intermediateStatement =
+        new IntermediatePortalStatement(connectionHandler, options, "", parse(sql));
 
     assertFalse(intermediateStatement.isExecuted());
-    assertEquals("SELECT", intermediateStatement.getCommand(0));
+    assertEquals("SELECT", intermediateStatement.getCommand());
 
-    intermediateStatement.execute();
+    intermediateStatement.executeAsync(backendConnection);
 
-    verify(connection).execute(Statement.of("SELECT * FROM users"));
-    assertTrue(intermediateStatement.containsResultSet(0));
+    verify(backendConnection).execute(parse(sql), Statement.of(sql));
+    assertTrue(intermediateStatement.containsResultSet());
     assertTrue(intermediateStatement.isExecuted());
-    assertEquals(StatementType.QUERY, intermediateStatement.getStatementType(0));
-    assertEquals(resultSet, intermediateStatement.getStatementResult(0));
-    assertTrue(intermediateStatement.isHasMoreData(0));
-    assertFalse(intermediateStatement.hasException(0));
+    assertEquals(StatementType.QUERY, intermediateStatement.getStatementType());
     assertEquals(0, intermediateStatement.getResultFormatCode(0));
 
     intermediateStatement.close();
-
-    verify(resultSet).close();
   }
 
   @Test
   public void testBasicUpdateStatement() throws Exception {
-    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);
+    String sql = "UPDATE users SET name = someName WHERE id = 10";
 
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(
-            options, parse("UPDATE users SET name = someName WHERE id = 10"), connectionHandler);
+    IntermediatePortalStatement intermediateStatement =
+        new IntermediatePortalStatement(connectionHandler, options, "", parse(sql));
 
     assertFalse(intermediateStatement.isExecuted());
-    assertEquals("UPDATE", intermediateStatement.getCommand(0));
+    assertEquals("UPDATE", intermediateStatement.getCommand());
 
-    intermediateStatement.execute();
+    intermediateStatement.executeAsync(backendConnection);
 
-    verify(connection).execute(Statement.of("UPDATE users SET name = someName WHERE id = 10"));
-    assertFalse(intermediateStatement.containsResultSet(0));
-    assertEquals(1L, intermediateStatement.getUpdateCount(0));
+    verify(backendConnection).execute(parse(sql), Statement.of(sql));
+    assertFalse(intermediateStatement.containsResultSet());
     assertTrue(intermediateStatement.isExecuted());
-    assertEquals(StatementType.UPDATE, intermediateStatement.getStatementType(0));
-    assertNull(intermediateStatement.getStatementResult(0));
-    assertFalse(intermediateStatement.isHasMoreData(0));
-    assertFalse(intermediateStatement.hasException(0));
+    assertEquals(StatementType.UPDATE, intermediateStatement.getStatementType());
+    assertNull(intermediateStatement.getStatementResult());
+    assertFalse(intermediateStatement.isHasMoreData());
+    assertFalse(intermediateStatement.hasException());
     assertEquals(0, intermediateStatement.getResultFormatCode(0));
 
     intermediateStatement.close();
@@ -157,30 +144,28 @@ public class StatementTest {
 
   @Test
   public void testBasicZeroUpdateCountResultStatement() throws Exception {
-    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);
+    String sql = "UPDATE users SET name = someName WHERE id = -1";
 
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(
-            options, parse("UPDATE users SET name = someName WHERE id = -1"), connectionHandler);
+    when(connection.execute(Statement.of(sql))).thenReturn(statementResult);
+    IntermediatePortalStatement intermediateStatement =
+        new IntermediatePortalStatement(connectionHandler, options, "", parse(sql));
+    BackendConnection backendConnection =
+        new BackendConnection(connection, DdlTransactionMode.Batch);
 
     assertFalse(intermediateStatement.isExecuted());
-    assertEquals("UPDATE", intermediateStatement.getCommand(0));
+    assertEquals("UPDATE", intermediateStatement.getCommand());
 
-    intermediateStatement.execute();
+    intermediateStatement.executeAsync(backendConnection);
+    backendConnection.flush();
 
-    Mockito.verify(connection)
-        .execute(Statement.of("UPDATE users SET name = someName WHERE id = -1"));
-    assertFalse(intermediateStatement.containsResultSet(0));
-    assertEquals(0L, intermediateStatement.getUpdateCount(0));
+    assertFalse(intermediateStatement.containsResultSet());
+    assertEquals(0L, intermediateStatement.getUpdateCount());
     assertTrue(intermediateStatement.isExecuted());
-    assertEquals(StatementType.UPDATE, intermediateStatement.getStatementType(0));
-    assertNull(intermediateStatement.getStatementResult(0));
-    assertFalse(intermediateStatement.isHasMoreData(0));
-    assertFalse(intermediateStatement.hasException(0));
+    assertEquals(StatementType.UPDATE, intermediateStatement.getStatementType());
+    assertNotNull(intermediateStatement.getStatementResult());
+    assertFalse(intermediateStatement.isHasMoreData());
+    assertFalse(intermediateStatement.hasException());
     assertEquals(0, intermediateStatement.getResultFormatCode(0));
 
     intermediateStatement.close();
@@ -190,31 +175,25 @@ public class StatementTest {
 
   @Test
   public void testBasicNoResultStatement() throws Exception {
-    when(options.getDdlTransactionMode()).thenReturn(DdlTransactionMode.Batch);
-    when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.NO_RESULT);
-    when(connection.execute(Statement.of("CREATE TABLE users (name varchar(100) primary key)")))
-        .thenReturn(statementResult);
+    String sql = "CREATE TABLE users (name varchar(100) primary key)";
 
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(
-            options,
-            parse("CREATE TABLE users (name varchar(100) primary key)"),
-            connectionHandler);
+    IntermediatePortalStatement intermediateStatement =
+        new IntermediatePortalStatement(connectionHandler, options, "", parse(sql));
 
     assertFalse(intermediateStatement.isExecuted());
-    assertEquals("CREATE", intermediateStatement.getCommand(0));
+    assertEquals("CREATE", intermediateStatement.getCommand());
 
-    intermediateStatement.execute();
+    intermediateStatement.executeAsync(backendConnection);
 
-    verify(connection).execute(Statement.of("CREATE TABLE users (name varchar(100) primary key)"));
-    assertFalse(intermediateStatement.containsResultSet(0));
-    assertEquals(0, intermediateStatement.getUpdateCount(0));
+    verify(backendConnection).execute(parse(sql), Statement.of(sql));
+    assertFalse(intermediateStatement.containsResultSet());
+    assertEquals(0, intermediateStatement.getUpdateCount());
     assertTrue(intermediateStatement.isExecuted());
-    assertEquals(StatementType.DDL, intermediateStatement.getStatementType(0));
-    assertNull(intermediateStatement.getStatementResult(0));
-    assertFalse(intermediateStatement.isHasMoreData(0));
-    assertFalse(intermediateStatement.hasException(0));
+    assertEquals(StatementType.DDL, intermediateStatement.getStatementType());
+    assertNull(intermediateStatement.getStatementResult());
+    assertFalse(intermediateStatement.isHasMoreData());
+    assertFalse(intermediateStatement.hasException());
     assertEquals(0, intermediateStatement.getResultFormatCode(0));
 
     intermediateStatement.close();
@@ -232,19 +211,22 @@ public class StatementTest {
 
   @Test
   public void testBasicStatementExceptionGetsSetOnExceptedExecution() {
+    String sql = "SELECT * FROM users";
     SpannerException thrownException =
         SpannerExceptionFactory.newSpannerException(ErrorCode.INVALID_ARGUMENT, "test error");
 
-    when(connection.execute(Statement.of("SELECT * FROM users"))).thenThrow(thrownException);
-
+    when(connection.execute(Statement.of(sql))).thenThrow(thrownException);
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(options, parse("SELECT * FROM users"), connectionHandler);
+    IntermediatePortalStatement intermediateStatement =
+        new IntermediatePortalStatement(connectionHandler, options, "", parse(sql));
+    BackendConnection backendConnection =
+        new BackendConnection(connection, DdlTransactionMode.Batch);
 
-    intermediateStatement.execute();
+    intermediateStatement.executeAsync(backendConnection);
+    backendConnection.flush();
 
-    assertTrue(intermediateStatement.hasException(0));
-    assertEquals(thrownException, intermediateStatement.getException(0));
+    assertTrue(intermediateStatement.hasException());
+    assertEquals(thrownException, intermediateStatement.getException());
   }
 
   @Test
@@ -262,9 +244,8 @@ public class StatementTest {
             .bind("p3")
             .to(30)
             .build();
-    when(statementResult.getResultType()).thenReturn(StatementResult.ResultType.UPDATE_COUNT);
-    when(statementResult.getUpdateCount()).thenReturn(0L);
-    when(connection.execute(statement)).thenReturn(statementResult);
+    BackendConnection backendConnection =
+        new BackendConnection(connection, DdlTransactionMode.Batch);
 
     IntermediatePreparedStatement intermediateStatement =
         new IntermediatePreparedStatement(connectionHandler, options, "", parse(sqlStatement));
@@ -276,11 +257,13 @@ public class StatementTest {
     IntermediatePortalStatement intermediatePortalStatement =
         intermediateStatement.bind(
             "", parameters, Arrays.asList((short) 0, (short) 0, (short) 0), new ArrayList<>());
-    intermediateStatement.execute();
+    intermediateStatement.executeAsync(backendConnection);
+    backendConnection.flush();
+
     verify(connection).execute(statement);
 
     assertEquals(sqlStatement, intermediatePortalStatement.getSql());
-    assertEquals("SELECT", intermediatePortalStatement.getCommand(0));
+    assertEquals("SELECT", intermediatePortalStatement.getCommand());
     assertFalse(intermediatePortalStatement.isExecuted());
     assertTrue(intermediateStatement.isBound());
   }
@@ -321,14 +304,16 @@ public class StatementTest {
   public void testPortalStatement() {
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     String sqlStatement = "SELECT * FROM users WHERE age > $1 AND age < $2 AND name = $3";
-    when(connection.executeQuery(Statement.of(sqlStatement))).thenReturn(resultSet);
 
     IntermediatePortalStatement intermediateStatement =
         new IntermediatePortalStatement(connectionHandler, options, "", parse(sqlStatement));
+    BackendConnection backendConnection =
+        new BackendConnection(connection, DdlTransactionMode.Batch);
 
-    intermediateStatement.describe();
+    intermediateStatement.describeAsync(backendConnection);
+    backendConnection.flush();
 
-    verify(connection).executeQuery(Statement.of(sqlStatement));
+    verify(connection).execute(Statement.of(sqlStatement));
 
     assertEquals(0, intermediateStatement.getParameterFormatCode(0));
     assertEquals(0, intermediateStatement.getParameterFormatCode(1));
@@ -365,73 +350,19 @@ public class StatementTest {
 
     IntermediatePortalStatement intermediateStatement =
         new IntermediatePortalStatement(connectionHandler, options, "", parse(sqlStatement));
+    BackendConnection backendConnection =
+        new BackendConnection(connection, DdlTransactionMode.Batch);
 
-    when(connection.executeQuery(Statement.of(sqlStatement)))
+    when(connection.execute(Statement.of(sqlStatement)))
         .thenThrow(
             SpannerExceptionFactory.newSpannerException(ErrorCode.INVALID_ARGUMENT, "test error"));
 
-    SpannerException exception =
-        assertThrows(SpannerException.class, intermediateStatement::describe);
-    assertEquals(ErrorCode.INVALID_ARGUMENT, exception.getErrorCode());
-  }
+    intermediateStatement.describeAsync(backendConnection);
+    backendConnection.flush();
 
-  @Test
-  public void testBatchStatements() {
-    String sql =
-        "INSERT INTO users (id) VALUES (1); INSERT INTO users (id) VALUES (2);INSERT INTO users (id) VALUES (3);";
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(options, parse(sql), connectionHandler);
-
-    assertTrue(intermediateStatement.isBatchedQuery());
-    assertEquals(3, intermediateStatement.getStatements().size(), 3);
-    assertEquals("INSERT INTO users (id) VALUES (1)", intermediateStatement.getStatement(0));
-    assertEquals("INSERT INTO users (id) VALUES (2)", intermediateStatement.getStatement(1));
-    assertEquals("INSERT INTO users (id) VALUES (3)", intermediateStatement.getStatement(2));
-  }
-
-  @Test
-  public void testAdditionalBatchStatements() {
-    String sql =
-        "BEGIN TRANSACTION; INSERT INTO users (id) VALUES (1); INSERT INTO users (id) VALUES (2); INSERT INTO users (id) VALUES (3); COMMIT;";
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(options, parse(sql), connectionHandler);
-
-    assertTrue(intermediateStatement.isBatchedQuery());
-
-    assertEquals(5, intermediateStatement.getStatements().size(), 5);
-    assertEquals("BEGIN TRANSACTION", intermediateStatement.getStatement(0));
-    assertEquals("INSERT INTO users (id) VALUES (1)", intermediateStatement.getStatement(1));
-    assertEquals("INSERT INTO users (id) VALUES (2)", intermediateStatement.getStatement(2));
-    assertEquals("INSERT INTO users (id) VALUES (3)", intermediateStatement.getStatement(3));
-    assertEquals("COMMIT", intermediateStatement.getStatement(4));
-  }
-
-  @Test
-  public void testBatchStatementsWithEmptyStatements() {
-    String sql = "INSERT INTO users (id) VALUES (1); ;;; INSERT INTO users (id) VALUES (2);";
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(options, parse(sql), connectionHandler);
-
-    assertTrue(intermediateStatement.isBatchedQuery());
-    assertEquals(2, intermediateStatement.getStatements().size());
-    assertEquals("INSERT INTO users (id) VALUES (1)", intermediateStatement.getStatement(0));
-    assertEquals("INSERT INTO users (id) VALUES (2)", intermediateStatement.getStatement(1));
-  }
-
-  @Test
-  public void testBatchStatementsWithQuotes() {
-    String sql =
-        "INSERT INTO users (name) VALUES (';;test;;'); INSERT INTO users (name1, name2) VALUES ('''''', ';'';');";
-    IntermediateStatement intermediateStatement =
-        new IntermediateStatement(options, parse(sql), connectionHandler);
-
-    assertTrue(intermediateStatement.isBatchedQuery());
-    assertEquals(2, intermediateStatement.getStatements().size());
-    assertEquals(
-        "INSERT INTO users (name) VALUES (';;test;;')", intermediateStatement.getStatement(0));
-    assertEquals(
-        "INSERT INTO users (name1, name2) VALUES ('''''', ';'';')",
-        intermediateStatement.getStatement(1));
+    assertTrue(intermediateStatement.hasException());
+    SpannerException exception = (SpannerException) intermediateStatement.getException();
+    assertEquals(ErrorCode.INVALID_ARGUMENT, exception.getErrorCode());
   }
 
   @Test
@@ -442,25 +373,22 @@ public class StatementTest {
     byte[] value = Bytes.concat(messageMetadata, payload.getBytes());
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
 
-    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
-    when(options.requiresMatcher()).thenReturn(false);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
-    IntermediateStatement intermediateStatement = ((QueryMessage) message).getStatement();
+    SimpleQueryStatement simpleQueryStatement = ((QueryMessage) message).getSimpleQueryStatement();
 
-    assertTrue(intermediateStatement.isBatchedQuery());
-    assertEquals(2, intermediateStatement.getStatements().size());
+    assertEquals(2, simpleQueryStatement.getStatements().size());
     assertEquals(
-        "INSERT INTO users (name) VALUES (';;test;;')", intermediateStatement.getStatement(0));
+        "INSERT INTO users (name) VALUES (';;test;;')", simpleQueryStatement.getStatement(0));
     assertEquals(
         "INSERT INTO users (name1, name2) VALUES ('''''', ';'';')",
-        intermediateStatement.getStatement(1));
+        simpleQueryStatement.getStatement(1));
   }
 
   @Test
@@ -487,6 +415,47 @@ public class StatementTest {
     verify(resultSet, never()).close();
   }
 
+  @Test
+  public void testIntermediateStatementExecuteAsyncIsUnsupported() {
+    String sql = "select * from foo";
+
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    IntermediateStatement intermediateStatement =
+        new IntermediateStatement(options, parse(sql), connectionHandler);
+
+    assertThrows(
+        UnsupportedOperationException.class,
+        () -> intermediateStatement.executeAsync(backendConnection));
+  }
+
+  @Test
+  public void testIntermediateStatementDescribeAsyncIsUnsupported() {
+    String sql = "select * from foo";
+
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    IntermediateStatement intermediateStatement =
+        new IntermediateStatement(options, parse(sql), connectionHandler);
+
+    assertThrows(
+        UnsupportedOperationException.class,
+        () -> intermediateStatement.describeAsync(backendConnection));
+  }
+
+  @Test
+  public void testGetStatementResultBeforeFlushFails() {
+    String sql = "select * from foo";
+
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    IntermediatePortalStatement intermediateStatement =
+        new IntermediatePortalStatement(connectionHandler, options, "", parse(sql));
+    BackendConnection backendConnection =
+        new BackendConnection(connection, DdlTransactionMode.Batch);
+
+    intermediateStatement.executeAsync(backendConnection);
+
+    assertThrows(IllegalStateException.class, intermediateStatement::getStatementResult);
+  }
+
   private void setupQueryInformationSchemaResults() {
     ResultSet spannerType = mock(ResultSet.class);
     when(spannerType.getString("column_name")).thenReturn("key", "value");
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ProtocolTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
similarity index 89%
rename from src/test/java/com/google/cloud/spanner/pgadapter/ProtocolTest.java
rename to src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
index 5ae8b5a7..cf2fec67 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ProtocolTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
@@ -12,19 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package com.google.cloud.spanner.pgadapter;
+package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
@@ -39,40 +38,26 @@ import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
 import com.google.cloud.spanner.connection.Connection;
-import com.google.cloud.spanner.connection.StatementResult;
-import com.google.cloud.spanner.connection.StatementResult.ResultType;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
+import com.google.cloud.spanner.pgadapter.ProxyServer;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.ConnectionState;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
+import com.google.cloud.spanner.pgadapter.statements.ExtendedQueryProtocolHandler;
 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.MatcherStatement;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
-import com.google.cloud.spanner.pgadapter.wireprotocol.BindMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.BootstrapMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.CancelMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.CloseMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.ManuallyCreatedToken;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.PreparedType;
-import com.google.cloud.spanner.pgadapter.wireprotocol.CopyDataMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.CopyDoneMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.CopyFailMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.DescribeMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.ExecuteMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.FlushMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.FunctionCallMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.SSLMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.StartupMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.SyncMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.TerminateMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
 import com.google.common.primitives.Bytes;
+import com.google.common.util.concurrent.SettableFuture;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
@@ -89,7 +74,6 @@ 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.Rule;
@@ -110,6 +94,8 @@ public class ProtocolTest {
   @Rule public MockitoRule rule = MockitoJUnit.rule();
   @Mock private ConnectionHandler connectionHandler;
   @Mock private Connection connection;
+  @Mock private ExtendedQueryProtocolHandler extendedQueryProtocolHandler;
+  @Mock private BackendConnection backendConnection;
   @Mock private ProxyServer server;
   @Mock private OptionsMetadata options;
   @Mock private IntermediatePreparedStatement intermediatePreparedStatement;
@@ -117,7 +103,6 @@ public class ProtocolTest {
   @Mock private ConnectionMetadata connectionMetadata;
   @Mock private DataOutputStream outputStream;
   @Mock private ResultSet resultSet;
-  @Mock private StatementResult statementResult;
 
   private byte[] intToBytes(int value) {
     byte[] parameters = new byte[4];
@@ -164,31 +149,31 @@ public class ProtocolTest {
     byte[] value = Bytes.concat(messageMetadata, payload.getBytes());
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
+    DataOutputStream outputStream = new DataOutputStream(new ByteArrayOutputStream());
 
     String expectedSQL = "SELECT * FROM users";
 
-    when(connection.execute(Statement.of(expectedSQL))).thenReturn(statementResult);
-    when(statementResult.getResultType()).thenReturn(ResultType.RESULT_SET);
-    when(statementResult.getResultSet()).thenReturn(resultSet);
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
     when(options.requiresMatcher()).thenReturn(false);
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
+    when(extendedQueryProtocolHandler.getBackendConnection()).thenReturn(backendConnection);
+    when(backendConnection.getConnectionState()).thenReturn(ConnectionState.IDLE);
+    when(connectionHandler.getStatement("")).thenReturn(intermediatePortalStatement);
+    when(connectionHandler.getPortal("")).thenReturn(intermediatePortalStatement);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
-    assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSql());
+    assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSqlWithoutComments());
 
     QueryMessage messageSpy = (QueryMessage) spy(message);
 
-    doNothing().when(messageSpy).handleQuery();
-
     messageSpy.send();
-    // Execute
-    verify(connection).execute(Statement.of(expectedSQL));
   }
 
   @Test
@@ -202,30 +187,16 @@ public class ProtocolTest {
 
     String expectedSQL = "SELECT * FROM users";
 
-    when(connection.execute(Statement.of(expectedSQL))).thenReturn(statementResult);
-    when(statementResult.getResultType()).thenReturn(ResultType.RESULT_SET);
-    when(statementResult.getResultSet()).thenReturn(resultSet);
     when(connectionHandler.getServer()).thenReturn(server);
     when(server.getOptions()).thenReturn(options);
-    when(options.requiresMatcher()).thenReturn(true);
-    when(options.getCommandMetadataJSON())
-        .thenReturn((JSONObject) parser.parse("{\"commands\": []}"));
-    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
-    assertEquals(MatcherStatement.class, ((QueryMessage) message).getStatement().getClass());
-    assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSql());
-
-    QueryMessage messageSpy = (QueryMessage) spy(message);
-
-    doNothing().when(messageSpy).handleQuery();
-
-    messageSpy.send();
-    verify(connection).execute(Statement.of(expectedSQL));
+    assertNotNull(((QueryMessage) message).getSimpleQueryStatement());
+    assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSqlWithoutComments());
   }
 
   @Test
@@ -283,6 +254,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
@@ -294,6 +267,7 @@ public class ProtocolTest {
 
     when(connectionHandler.hasStatement(anyString())).thenReturn(false);
     message.send();
+    ((ParseMessage) message).flush();
     verify(connectionHandler)
         .registerStatement(expectedMessageName, ((ParseMessage) message).getStatement());
 
@@ -344,6 +318,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
@@ -355,6 +331,7 @@ public class ProtocolTest {
 
     when(connectionHandler.hasStatement(anyString())).thenReturn(false);
     message.send();
+    ((ParseMessage) message).flush();
     verify(connectionHandler)
         .registerStatement(expectedMessageName, ((ParseMessage) message).getStatement());
 
@@ -406,6 +383,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
@@ -417,6 +396,7 @@ public class ProtocolTest {
 
     when(connectionHandler.hasStatement(anyString())).thenReturn(false);
     message.send();
+    ((ParseMessage) message).flush();
     verify(connectionHandler)
         .registerStatement(expectedMessageName, ((ParseMessage) message).getStatement());
 
@@ -453,6 +433,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(ParseMessage.class, message.getClass());
@@ -464,6 +446,7 @@ public class ProtocolTest {
 
     when(connectionHandler.hasStatement(anyString())).thenReturn(false);
     message.send();
+    ((ParseMessage) message).flush();
     verify(connectionHandler)
         .registerStatement(expectedMessageName, ((ParseMessage) message).getStatement());
 
@@ -508,6 +491,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
 
@@ -552,6 +537,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
 
@@ -597,6 +584,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
 
@@ -654,6 +643,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(BindMessage.class, message.getClass());
@@ -671,6 +662,7 @@ public class ProtocolTest {
         .thenReturn(intermediatePortalStatement);
 
     message.send();
+    ((BindMessage) message).flush();
     verify(connectionHandler).registerPortal(expectedPortalName, intermediatePortalStatement);
 
     // BindCompleteResponse
@@ -838,6 +830,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(DescribeMessage.class, message.getClass());
@@ -849,6 +843,7 @@ public class ProtocolTest {
     doNothing().when(messageSpy).handleDescribePortal();
 
     messageSpy.send();
+    messageSpy.flush();
 
     verify(messageSpy).handleDescribePortal();
   }
@@ -872,6 +867,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(DescribeMessage.class, message.getClass());
@@ -883,6 +880,7 @@ public class ProtocolTest {
     doNothing().when(messageSpy).handleDescribeStatement();
 
     messageSpy.send();
+    messageSpy.flush();
 
     verify(messageSpy).handleDescribeStatement();
   }
@@ -907,6 +905,9 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
+    when(extendedQueryProtocolHandler.getBackendConnection()).thenReturn(backendConnection);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(ExecuteMessage.class, message.getClass());
@@ -916,16 +917,59 @@ public class ProtocolTest {
     verify(connectionHandler).getPortal("some portal");
     ExecuteMessage messageSpy = (ExecuteMessage) spy(message);
 
-    doReturn(false)
+    doNothing()
         .when(messageSpy)
-        .sendSpannerResult(
-            anyInt(), any(IntermediatePortalStatement.class), any(QueryMode.class), anyLong());
+        .sendSpannerResult(any(IntermediatePortalStatement.class), any(QueryMode.class), anyLong());
 
     messageSpy.send();
+    messageSpy.flush();
 
-    verify(intermediatePortalStatement).execute();
+    verify(intermediatePortalStatement).executeAsync(backendConnection);
     verify(messageSpy)
-        .sendSpannerResult(0, intermediatePortalStatement, QueryMode.EXTENDED, totalRows);
+        .sendSpannerResult(intermediatePortalStatement, QueryMode.EXTENDED, totalRows);
+    verify(connectionHandler).cleanUp(intermediatePortalStatement);
+  }
+
+  @Test
+  public void testExecuteMessageWithException() throws Exception {
+    byte[] messageMetadata = {'E'};
+    String statementName = "some portal\0";
+    int totalRows = 99999;
+
+    byte[] length = intToBytes(4 + statementName.length() + 4);
+
+    byte[] value =
+        Bytes.concat(messageMetadata, length, statementName.getBytes(), intToBytes(totalRows));
+
+    String expectedStatementName = "some portal";
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    Exception testException = new Exception("test error");
+    when(intermediatePortalStatement.hasException()).thenReturn(true);
+    when(intermediatePortalStatement.getException()).thenReturn(testException);
+    when(connectionHandler.getPortal(anyString())).thenReturn(intermediatePortalStatement);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
+    when(extendedQueryProtocolHandler.getBackendConnection()).thenReturn(backendConnection);
+
+    WireMessage message = ControlMessage.create(connectionHandler);
+    assertEquals(ExecuteMessage.class, message.getClass());
+    assertEquals(expectedStatementName, ((ExecuteMessage) message).getName());
+    assertEquals(totalRows, ((ExecuteMessage) message).getMaxRows());
+
+    verify(connectionHandler).getPortal("some portal");
+    ExecuteMessage messageSpy = (ExecuteMessage) spy(message);
+
+    messageSpy.send();
+    messageSpy.flush();
+
+    verify(intermediatePortalStatement).executeAsync(backendConnection);
+    verify(messageSpy).handleError(testException);
     verify(connectionHandler).cleanUp(intermediatePortalStatement);
   }
 
@@ -1016,10 +1060,14 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.IDLE);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
+    when(extendedQueryProtocolHandler.getBackendConnection()).thenReturn(backendConnection);
+    when(backendConnection.getConnectionState()).thenReturn(ConnectionState.IDLE);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(message.getClass(), SyncMessage.class);
@@ -1045,10 +1093,14 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.TRANSACTION);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
+    when(extendedQueryProtocolHandler.getBackendConnection()).thenReturn(backendConnection);
+    when(backendConnection.getConnectionState()).thenReturn(ConnectionState.TRANSACTION);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(SyncMessage.class, message.getClass());
@@ -1078,6 +1130,8 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
 
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(message.getClass(), FlushMessage.class);
@@ -1104,6 +1158,8 @@ public class ProtocolTest {
     DataOutputStream outputStream = new DataOutputStream(result);
 
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
     when(connection.isInTransaction()).thenReturn(true);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -1134,20 +1190,28 @@ public class ProtocolTest {
     String expectedSQL = "INSERT INTO users (name) VALUES ('test')";
 
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
-    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.TRANSACTION);
-    when(statementResult.getResultType()).thenReturn(ResultType.UPDATE_COUNT);
-    when(statementResult.getUpdateCount()).thenReturn(1L);
-    when(connection.execute(Statement.of(expectedSQL))).thenReturn(statementResult);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getServer()).thenReturn(server);
+    when(connectionHandler.getStatement("")).thenReturn(intermediatePortalStatement);
+    when(connectionHandler.getPortal("")).thenReturn(intermediatePortalStatement);
+    when(intermediatePortalStatement.getCommandTag()).thenReturn("INSERT");
+    when(intermediatePortalStatement.getStatementType()).thenReturn(StatementType.UPDATE);
+    when(intermediatePortalStatement.getUpdateCount()).thenReturn(1L);
+    when(backendConnection.getConnectionState()).thenReturn(ConnectionState.TRANSACTION);
     OptionsMetadata options = mock(OptionsMetadata.class);
     when(server.getOptions()).thenReturn(options);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
+    ExtendedQueryProtocolHandler extendedQueryProtocolHandler =
+        new ExtendedQueryProtocolHandler(backendConnection);
+    when(connectionHandler.getExtendedQueryProtocolHandler())
+        .thenReturn(extendedQueryProtocolHandler);
+
     WireMessage message = ControlMessage.create(connectionHandler);
     assertEquals(QueryMessage.class, message.getClass());
-    assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSql());
+    assertEquals(expectedSQL, ((QueryMessage) message).getStatement().getSqlWithoutComments());
 
     message.send();
 
@@ -1313,13 +1377,12 @@ public class ProtocolTest {
 
     assertEquals(CopyDoneMessage.class, message.getClass());
     CopyDoneMessage messageSpy = (CopyDoneMessage) spy(message);
-    doReturn(false)
+    doNothing()
         .when(messageSpy)
-        .sendSpannerResult(
-            anyInt(), any(IntermediateStatement.class), any(QueryMode.class), anyLong());
+        .sendSpannerResult(any(IntermediateStatement.class), any(QueryMode.class), anyLong());
 
     messageSpy.send();
-    verify(messageSpy).sendSpannerResult(0, copyStatement, QueryMode.SIMPLE, 0L);
+    verify(messageSpy).sendSpannerResult(copyStatement, QueryMode.SIMPLE, 0L);
   }
 
   @Test
@@ -1399,8 +1462,8 @@ public class ProtocolTest {
 
     assertEquals("TEXT", copyStatement.getFormatType());
     assertEquals('\t', copyStatement.getDelimiterChar());
-    assertFalse(copyStatement.hasException(0));
-    assertEquals(12L, copyStatement.getUpdateCount().longValue());
+    assertFalse(copyStatement.hasException());
+    assertEquals(12L, copyStatement.getUpdateCount());
     assertEquals(12L, mw.getRowCount());
 
     copyStatement.close();
@@ -1759,6 +1822,21 @@ public class ProtocolTest {
     assertThrows(IOException.class, message::send);
   }
 
+  @Test
+  public void testGetPortalMetadataBeforeFlushFails() {
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getPortal(anyString())).thenReturn(intermediatePortalStatement);
+    when(intermediatePortalStatement.containsResultSet()).thenReturn(true);
+    when(intermediatePortalStatement.describeAsync(backendConnection))
+        .thenReturn(SettableFuture.create());
+
+    DescribeMessage describeMessage =
+        new DescribeMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN);
+    describeMessage.buffer(backendConnection);
+
+    assertThrows(IllegalStateException.class, describeMessage::getPortalMetadata);
+  }
+
   private void setupQueryInformationSchemaResults() {
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     ResultSet spannerType = mock(ResultSet.class);