Repository: GoogleCloudPlatform/pgadapter
Pull request #234 (2022-06-27 11:41:26)

docs: document pgx connection tips and performance considerations 
Adds some simple connection tips for using the Golang pgx driver.

codecov[bot]: # [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/234?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> :exclamation: No coverage uploaded for pull request base (`postgresql-dialect@823a72f`). [Click here to learn what that means](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#section-missing-base-commit).
> The diff coverage is `n/a`.

```diff
@@                  Coverage Diff                  @@
##             postgresql-dialect     #234   +/-   ##
=====================================================
  Coverage                      ?   80.66%           
  Complexity                    ?     1066           
=====================================================
  Files                         ?       92           
  Lines                         ?     3647           
  Branches                      ?      431           
=====================================================
  Hits                          ?     2942           
  Misses                        ?      538           
  Partials                      ?      167           
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `80.66% <0.00%> (?)` | |

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.


------

[Continue to review full report at Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/234?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/234?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform). Last update [823a72f...2fc6062](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/234?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/.ci/evaluate-with-psql.sh b/.ci/evaluate-with-psql.sh
index 5a14e708..b0238cbe 100644
--- a/.ci/evaluate-with-psql.sh
+++ b/.ci/evaluate-with-psql.sh
@@ -131,6 +131,17 @@ echo "------Test \"COPY FROM STDIN\"------"
 17	17	17
 \.
 EOF
-echo "SELECT * FROM users;" | /usr/lib/postgresql/"${PSQL_VERSION}"/bin/psql -h localhost -p 4242 -d "${GOOGLE_CLOUD_DATABASE_WITH_VERSION}" > .ci/e2e-result/copy-from-stdin.txt
+echo "SELECT * FROM users order by id;" | /usr/lib/postgresql/"${PSQL_VERSION}"/bin/psql -h localhost -p 4242 -d "${GOOGLE_CLOUD_DATABASE_WITH_VERSION}" > .ci/e2e-result/copy-from-stdin.txt
 diff -i -w -s .ci/e2e-result/copy-from-stdin.txt .ci/e2e-expected/copy-from-stdin.txt
-RETURN_CODE=$((${RETURN_CODE}||$?))
\ No newline at end of file
+RETURN_CODE=$((${RETURN_CODE}||$?))
+
+echo "------Test \"COPY FROM STDIN in batch\"------"
+/usr/lib/postgresql/"${PSQL_VERSION}"/bin/psql -h localhost -p 4242 -d "${GOOGLE_CLOUD_DATABASE_WITH_VERSION}" -c "select 0; COPY users FROM STDIN; COPY users FROM STDIN; select 1;" <<EOF
+18	18	18
+\.
+19	19	19
+\.
+EOF
+echo "SELECT * FROM users order by id;" | /usr/lib/postgresql/"${PSQL_VERSION}"/bin/psql -h localhost -p 4242 -d "${GOOGLE_CLOUD_DATABASE_WITH_VERSION}" > .ci/e2e-result/copy-from-stdin-batch.txt
+diff -i -w -s .ci/e2e-result/copy-from-stdin-batch.txt .ci/e2e-expected/copy-from-stdin-batch.txt
+RETURN_CODE=$((${RETURN_CODE}||$?))
diff --git a/README.md b/README.md
index 254dc014..f3fa1062 100644
--- a/README.md
+++ b/README.md
@@ -293,7 +293,7 @@ Other `psql` meta-commands are __not__ supported.
 Spanner database. `COPY` operations are atomic by default, but the standard transaction limits of
 Cloud Spanner apply to these transactions. That means that at most 20,000 mutations can be included
 in one `COPY` operation. `COPY` can also be executed in non-atomic mode by executing the statement
-`SET AUTOCOMMIT_DML_MODE='PARTITIONED_NON_ATOMIC'` before executing the copy operation.
+`SET SPANNER.AUTOCOMMIT_DML_MODE='PARTITIONED_NON_ATOMIC'` before executing the copy operation.
 
 Although only `STDIN` is supported, export files can still be imported using `COPY` by piping files
 into `psql`. See the examples below.
@@ -315,7 +315,7 @@ create table numbers (number bigint not null primary key, name varchar);
 ```
 
 ```shell
-cat numbers.txt | psql -h localhost -d test-db -c "set autocommit_dml_mode='partitioned_non_atomic'" -c "copy numbers from stdin;"
+cat numbers.txt | psql -h localhost -d test-db -c "set spanner.autocommit_dml_mode='partitioned_non_atomic'; copy numbers from stdin;"
 ```
 
 The above operation will automatically split the data over multiple transactions if the file
@@ -348,8 +348,7 @@ Larger datasets require that the Cloud Spanner database is set to `PARTITIONED_N
 psql -h localhost -p 5432 -d my-local-db \
   -c "copy (select i, to_char(i, 'fm000') from generate_series(1, 1000000) s(i)) to stdout" \
   | psql -h localhost -p 5433 -d my-spanner-db \
-  -c "set autocommit_dml_mode='partitioned_non_atomic'" \
-  -c "copy numbers from stdin;"
+  -c "set spanner.autocommit_dml_mode='partitioned_non_atomic'; copy numbers from stdin;"
 ```
 
 ## Limitations
diff --git a/docs/jdbc.md b/docs/jdbc.md
index 8f8b7647..b57accfe 100644
--- a/docs/jdbc.md
+++ b/docs/jdbc.md
@@ -9,18 +9,54 @@ First start PGAdapter:
 
 ```shell
 wget https://storage.googleapis.com/pgadapter-jar-releases/pgadapter.tar.gz && tar -xzvf pgadapter.tar.gz
-java -jar pgadapter.jar -p my-project -i my-instance -d my-database
+java -jar pgadapter.jar -p my-project -i my-instance
 ```
 
-Connect to PGAdapter like this:
+Connect to PGAdapter using TCP like this:
 
 ```java
 // Make sure the PG JDBC driver is loaded.
 Class.forName("org.postgresql.Driver");
 
 // Replace localhost and 5432 with the host and port number where PGAdapter is running.
-// prepareThreshold=0 disables the use of server-side prepared statements in JDBC.
-try (Connection connection = DriverManager.getConnection("jdbc:postgresql://localhost:5432/?prepareThreshold=0")) {
+try (Connection connection = DriverManager.getConnection("jdbc:postgresql://localhost:5432/my-database")) {
+  try (ResultSet resultSet = connection.createStatement().executeQuery("select 'Hello world!' as hello")) {
+    while (resultSet.next()) {
+      System.out.printf("Greeting from Cloud Spanner PostgreSQL: %s\n", resultSet.getString(1));
+    }
+  }
+}
+```
+
+
+### Unix Domain Sockets
+Connect to PGAdapter using Unix Domain Sockets like this:
+
+Add this to your dependencies:
+```xml
+<dependency>
+  <groupId>com.kohlschutter.junixsocket</groupId>
+  <artifactId>junixsocket-core</artifactId>
+  <version>2.5.0</version>
+  <type>pom</type>
+</dependency>
+<dependency>
+  <groupId>com.kohlschutter.junixsocket</groupId>
+  <artifactId>junixsocket-common</artifactId>
+  <version>2.5.0</version>
+</dependency>
+```
+
+```java
+// Make sure the PG JDBC driver is loaded.
+Class.forName("org.postgresql.Driver");
+
+// '/tmp' is the default domain socket directory for PGAdapter. This can be changed using the -dir
+// command line argument. 5432 is the default port number used by PGAdapter. Change this in the
+// connection string if PGAdapter is running on a custom port.
+try (Connection connection = DriverManager.getConnection("jdbc:postgresql://localhost/my-database"
+    + "?socketFactory=org.newsclub.net.unix.AFUNIXSocketFactory$FactoryArg" 
+    + "&socketFactoryArg=/tmp/.s.PGSQL.5432")) {
   try (ResultSet resultSet = connection.createStatement().executeQuery("select 'Hello world!' as hello")) {
     while (resultSet.next()) {
       System.out.printf("Greeting from Cloud Spanner PostgreSQL: %s\n", resultSet.getString(1));
@@ -33,3 +69,39 @@ try (Connection connection = DriverManager.getConnection("jdbc:postgresql://loca
 
 This example uses the pre-built jar to run PGAdapter as a standalone process.
 See [README](../README.md) for more possibilities on how to run PGAdapter.
+
+## Performance Considerations
+
+The following will give you the best possible performance when using the JDBC driver with PGAdapter.
+
+### Unix Domain Sockets
+Use Unix Domain Socket connections for the lowest possible latency when PGAdapter and the client
+application are running on the same host. See https://jdbc.postgresql.org/documentation/head/connect.html
+for more information on connection options for the JDBC driver.
+
+### JDBC Batching
+Use the JDBC batching API when executing multiple DDL or DML statements. PGAdapter will combine
+DML and DDL statements that are executed in a JDBC batch into a single request on Cloud Spanner.
+This can significantly reduce the overhead of executing multiple DML or DDL statements.
+
+Example for DML statements:
+
+```java
+try (java.sql.Statement statement = connection.createStatement()) {
+  statement.addBatch("insert into my_table (key, value) values ('k1', 'value1')");
+  statement.addBatch("insert into my_table (key, value) values ('k2', 'value2')");
+  // This will execute the above statements as a single batch DML request on Cloud Spanner.
+  int[] updateCounts = statement.executeBatch();
+}
+```
+
+Example for DDL statements:
+
+```java
+try (java.sql.Statement statement = connection.createStatement()) {
+  statement.addBatch("create table singers (singerid varchar primary key, name varchar)");
+  statement.addBatch("create index idx_singers_name on singers (name)");
+  // This will execute the above statements as a single DDL update request on Cloud Spanner.
+  int[] updateCounts = statement.executeBatch();
+}
+```
diff --git a/docs/pgx.md b/docs/pgx.md
index 3616c26e..c8d78a7a 100644
--- a/docs/pgx.md
+++ b/docs/pgx.md
@@ -15,18 +15,18 @@ docker run \
   -v ${GOOGLE_APPLICATION_CREDENTIALS}:${GOOGLE_APPLICATION_CREDENTIALS}:ro \
   -e GOOGLE_APPLICATION_CREDENTIALS \
   gcr.io/cloud-spanner-pg-adapter/pgadapter \
-  -p my-project -i my-instance -d my-database \
+  -p my-project -i my-instance \
   -x
 ```
 
-Then connect to PGAdapter like this:
+Then connect to PGAdapter using TCP like this:
 
 ```go
 // pwd:uid is not used by PGAdapter, but it is required in the connection string.
 // Replace localhost and 5432 with the host and port number where PGAdapter is running.
 // sslmode=disable instructs pgx to try plain text mode directly. Otherwise, pgx will try two times
 // with SSL enabled before trying plain text.
-connString := "postgres://uid:pwd@localhost:5432/?sslmode=disable"
+connString := "postgres://uid:pwd@localhost:5432/my-database?sslmode=disable"
 ctx := context.Background()
 conn, err := pgx.Connect(ctx, connString)
 if err != nil {
@@ -42,7 +42,64 @@ if err != nil {
 fmt.Printf("Greeting from Cloud Spanner PostgreSQL: %v\n", greeting)
 ```
 
+You can also connect to PGAdapter using Unix Domain Sockets if PGAdapter is running on the same host
+as the client application:
+
+```go
+// '/tmp' is the default domain socket directory for PGAdapter. This can be changed using the -dir
+// command line argument. 5432 is the default port number used by PGAdapter. Change this in the
+// connection string if PGAdapter is running on a custom port.
+connString := "host=/tmp port=5432 database=my-database"
+ctx := context.Background()
+conn, err := pgx.Connect(ctx, connString)
+if err != nil {
+    return err
+}
+defer conn.Close(ctx)
+
+var greeting string
+err = conn.QueryRow(ctx, "select 'Hello world!' as hello").Scan(&greeting)
+if err != nil {
+    return err
+}
+fmt.Printf("Greeting from Cloud Spanner PostgreSQL: %v\n", greeting)
+```
+
+
 ## Running PGAdapter
 
 This example uses the pre-built Docker image to run PGAdapter.
 See [README](../README.md) for more possibilities on how to run PGAdapter.
+
+
+## Performance Considerations
+
+The following will give you the best possible performance when using pgx with PGAdapter.
+
+### Unix Domain Sockets
+Use Unix Domain Socket connections for the lowest possible latency when PGAdapter and the client
+application are running on the same host. See https://pkg.go.dev/github.com/jackc/pgx#ConnConfig
+for more information on connection options for pgx.
+
+### Batching
+Use the pgx batching API when executing multiple DDL or DML statements. PGAdapter will combine
+DML and DDL statements that are executed in a batch into a single request on Cloud Spanner.
+This can significantly reduce the overhead of executing multiple DML or DDL statements.
+
+Example for DML statements:
+
+```go
+batch := &pgx.Batch{}
+batch.Queue("insert into my_table (key, value) values ($1, $2)", "k1", "value1")
+batch.Queue("insert into my_table (key, value) values ($1, $2)", "k2", "value2")
+res := conn.SendBatch(context.Background(), batch)
+```
+
+Example for DDL statements:
+
+```go
+batch := &pgx.Batch{}
+batch.Queue("create table singers (singerid varchar primary key, name varchar)")
+batch.Queue("create index idx_singers_name on singers (name)")
+res := conn.SendBatch(context.Background(), batch)
+```
diff --git a/pom.xml b/pom.xml
index 5bd693c3..745bce86 100644
--- a/pom.xml
+++ b/pom.xml
@@ -29,7 +29,7 @@
       com.google.cloud.spanner.pgadapter.python.PythonTest
     </excludedTests>
 
-    <spanner.version>6.25.6</spanner.version>
+    <spanner.version>6.25.7</spanner.version>
     <junixsocket.version>2.5.0</junixsocket.version>
   </properties>
 
@@ -120,14 +120,14 @@
     <dependency>
       <groupId>com.google.api</groupId>
       <artifactId>gax-grpc</artifactId>
-      <version>2.18.1</version>
+      <version>2.18.2</version>
       <classifier>testlib</classifier>
       <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>net.java.dev.jna</groupId>
       <artifactId>jna</artifactId>
-      <version>5.11.0</version>
+      <version>5.12.1</version>
       <scope>test</scope>
     </dependency>
     <dependency>
@@ -302,6 +302,7 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-surefire-plugin</artifactId>
         <configuration>
+          <forkedProcessExitTimeoutInSeconds>3</forkedProcessExitTimeoutInSeconds>
           <forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
           <argLine>${surefire.jacoco.args}</argLine>
           <excludedGroups>${excludedTests}</excludedGroups>
@@ -410,6 +411,8 @@
         </executions>
         <configuration>
           <forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
+          <forkCount>8</forkCount>
+          <reuseForks>true</reuseForks>
           <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 abd0eb43..4d6e9ef0 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
@@ -98,6 +98,7 @@ public class ConnectionHandler extends Thread {
   private static final AtomicInteger incrementingConnectionId = new AtomicInteger(0);
   private ConnectionMetadata connectionMetadata;
   private WireMessage message;
+  private int invalidMessagesCount;
   private Connection spannerConnection;
   private DatabaseId databaseId;
   private WellKnownClient wellKnownClient;
@@ -251,15 +252,7 @@ public class ConnectionHandler extends Thread {
           }
         }
         while (this.status != ConnectionStatus.TERMINATED) {
-          try {
-            message.nextHandler();
-            message.send();
-          } catch (IllegalArgumentException | IllegalStateException | EOFException fatalException) {
-            this.handleError(output, fatalException);
-            this.status = ConnectionStatus.TERMINATED;
-          } catch (Exception e) {
-            this.handleError(output, e);
-          }
+          handleMessages();
         }
       } catch (Exception e) {
         this.handleError(output, e);
@@ -293,6 +286,27 @@ public class ConnectionHandler extends Thread {
     }
   }
 
+  /**
+   * Reads and handles wire-protocol messages. This method is normally only called from this {@link
+   * ConnectionHandler}, but certain sub-protocols such as the COPY protocol also need to process
+   * messages in line.
+   */
+  public void handleMessages() throws Exception {
+    try {
+      message.nextHandler();
+      message.send();
+    } catch (IllegalArgumentException | IllegalStateException | EOFException fatalException) {
+      this.handleError(getConnectionMetadata().getOutputStream(), fatalException);
+      if (this.status == ConnectionStatus.COPY_IN) {
+        this.status = ConnectionStatus.COPY_FAILED;
+      } else {
+        terminate();
+      }
+    } catch (Exception e) {
+      this.handleError(getConnectionMetadata().getOutputStream(), e);
+    }
+  }
+
   /** Called when a Terminate message is received. This closes this {@link ConnectionHandler}. */
   public void handleTerminate() {
     closeAllPortals();
@@ -307,42 +321,41 @@ public class ConnectionHandler extends Thread {
    * shutting down while the connection is still active.
    */
   void terminate() {
-    if (this.status != ConnectionStatus.TERMINATED) {
-      handleTerminate();
-      try {
-        if (!socket.isClosed()) {
-          socket.close();
-        }
-      } catch (IOException exception) {
-        logger.log(
-            Level.WARNING,
-            exception,
-            () ->
-                String.format(
-                    "Failed to close connection handler with ID %s: %s", getName(), exception));
+    handleTerminate();
+    try {
+      if (!socket.isClosed()) {
+        socket.close();
       }
+    } catch (IOException exception) {
+      logger.log(
+          Level.WARNING,
+          exception,
+          () ->
+              String.format(
+                  "Failed to close connection handler with ID %s: %s", getName(), exception));
     }
   }
 
   /**
    * 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 IOException if there is some issue in the sending of the error messages.
    */
-  private void handleError(DataOutputStream output, Exception e) throws Exception {
+  private void handleError(DataOutputStream output, Exception exception) throws Exception {
     logger.log(
         Level.WARNING,
-        e,
-        () -> String.format("Exception on connection handler with ID %s: %s", getName(), e));
+        exception,
+        () ->
+            String.format("Exception on connection handler with ID %s: %s", getName(), exception));
     if (this.status == ConnectionStatus.TERMINATED) {
-      new ErrorResponse(output, e, ErrorResponse.State.InternalError, Severity.FATAL).send();
+      new ErrorResponse(output, exception, ErrorResponse.State.InternalError, Severity.FATAL)
+          .send();
       new TerminateResponse(output).send();
     } else if (this.status == ConnectionStatus.COPY_IN) {
-      new ErrorResponse(output, e, ErrorResponse.State.InternalError).send();
+      new ErrorResponse(output, exception, ErrorResponse.State.InternalError).send();
     } else {
-      this.status = ConnectionStatus.AUTHENTICATED;
-      new ErrorResponse(output, e, ErrorResponse.State.InternalError).send();
+      new ErrorResponse(output, exception, ErrorResponse.State.InternalError).send();
       new ReadyResponse(output, ReadyResponse.Status.IDLE).send();
     }
   }
@@ -497,6 +510,31 @@ public class ConnectionHandler extends Thread {
     this.message = this.server.recordMessage(message);
   }
 
+  /**
+   * Returns the number of invalid messages that this connection has received in a row. This can for
+   * example happen if a client has initiated a COPY operation and the copy operation fails on the
+   * server. The server will then respond with an error response, but if the client fails to read
+   * that message and continues to send copy data messages, the server could get flooded. This value
+   * is used to detect such a situation and breaks the connection if too many invalid messages in a
+   * row are received.
+   */
+  public int getInvalidMessageCount() {
+    return this.invalidMessagesCount;
+  }
+
+  /** Increases the number of invalid messages that was received in a row by 1. */
+  public void increaseInvalidMessageCount() {
+    this.invalidMessagesCount++;
+  }
+
+  /**
+   * Clears the number of invalid messages that was received. This is called whenever a valid
+   * message is encountered.
+   */
+  public void clearInvalidMessageCount() {
+    this.invalidMessagesCount = 0;
+  }
+
   public ConnectionMetadata getConnectionMetadata() {
     return connectionMetadata;
   }
@@ -530,6 +568,8 @@ public class ConnectionHandler extends Thread {
     UNAUTHENTICATED,
     AUTHENTICATED,
     COPY_IN,
+    COPY_DONE,
+    COPY_FAILED,
     TERMINATED,
   }
 
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
index d233d994..b752a0d5 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
@@ -14,7 +14,10 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
+import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.COPY;
+
 import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.DatabaseId;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.SpannerBatchUpdateException;
@@ -27,16 +30,30 @@ 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.statements.DdlExecutor.NotExecuted;
 import com.google.cloud.spanner.pgadapter.statements.local.LocalStatement;
+import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
+import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
+import com.google.cloud.spanner.pgadapter.utils.StatementParser;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 
@@ -144,6 +161,8 @@ public class BackendConnection {
           result.set(NO_RESULT);
         } else if (statement.getSql().isEmpty()) {
           result.set(NO_RESULT);
+        } else if (parsedStatement.isDdl()) {
+          result.set(ddlExecutor.execute(parsedStatement, statement));
         } else {
           result.set(spannerConnection.execute(statement));
         }
@@ -154,6 +173,57 @@ public class BackendConnection {
     }
   }
 
+  /**
+   * This statement represents a COPY table FROM STDIN statement. This has no one-on-one mapping
+   * with a Cloud Spanner SQL statement and is therefore executed using a custom {@link
+   * MutationWriter}. As the COPY implementation uses mutations instead of DML, it has slightly
+   * different transaction semantics than in real PostgreSQL. A COPY operation will by default be
+   * atomic, but can be configured to behave non-atomically for large batches. Also, if a COPY
+   * operation is executed in a transaction (both implicit and explicit), it will commit the
+   * transaction when the COPY operation is done. This is required to flush the mutations to the
+   * database.
+   */
+  private final class Copy extends BufferedStatement<StatementResult> {
+    private final CopyDataReceiver copyDataReceiver;
+    private final MutationWriter mutationWriter;
+    private final ListeningExecutorService executor;
+
+    Copy(
+        ParsedStatement parsedStatement,
+        Statement statement,
+        CopyDataReceiver copyDataReceiver,
+        MutationWriter mutationWriter,
+        ExecutorService executor) {
+      super(parsedStatement, statement);
+      this.copyDataReceiver = copyDataReceiver;
+      this.mutationWriter = mutationWriter;
+      this.executor = MoreExecutors.listeningDecorator(executor);
+    }
+
+    @Override
+    void execute() {
+      try {
+        checkConnectionState();
+        // Execute the MutationWriter and the CopyDataReceiver both asynchronously and wait for both
+        // to finish before continuing with the next statement. This ensures that all statements are
+        // applied in sequential order.
+        ListenableFuture<StatementResult> statementResultFuture = executor.submit(mutationWriter);
+        ListenableFuture<Void> copyDataReceiverFuture = executor.submit(copyDataReceiver);
+        this.result.setFuture(statementResultFuture);
+        Futures.allAsList(copyDataReceiverFuture, statementResultFuture).get();
+      } catch (ExecutionException executionException) {
+        result.setException(executionException.getCause());
+        throw SpannerExceptionFactory.asSpannerException(executionException.getCause());
+      } catch (InterruptedException interruptedException) {
+        result.setException(SpannerExceptionFactory.propagateInterrupt(interruptedException));
+        throw SpannerExceptionFactory.propagateInterrupt(interruptedException);
+      } catch (Exception exception) {
+        result.setException(exception);
+        throw exception;
+      }
+    }
+  }
+
   private static final ImmutableMap<String, LocalStatement> EMPTY_LOCAL_STATEMENTS =
       ImmutableMap.of();
   private static final StatementResult NO_RESULT = new NoResult();
@@ -163,8 +233,10 @@ public class BackendConnection {
   private final ImmutableMap<String, LocalStatement> localStatements;
   private ConnectionState connectionState = ConnectionState.IDLE;
   private TransactionMode transactionMode = TransactionMode.IMPLICIT;
+  private final String currentSchema = "public";
   private final LinkedList<BufferedStatement<?>> bufferedStatements = new LinkedList<>();
   private final Connection spannerConnection;
+  private final DdlExecutor ddlExecutor;
   private final DdlTransactionMode ddlTransactionMode;
 
   /**
@@ -172,10 +244,12 @@ public class BackendConnection {
    * DdlTransactionMode}.
    */
   BackendConnection(
+      DatabaseId databaseId,
       Connection spannerConnection,
       DdlTransactionMode ddlTransactionMode,
       ImmutableList<LocalStatement> localStatements) {
     this.spannerConnection = spannerConnection;
+    this.ddlExecutor = new DdlExecutor(databaseId, this);
     this.ddlTransactionMode = ddlTransactionMode;
     //noinspection UnstableApiUsage
     this.localStatements =
@@ -205,6 +279,22 @@ public class BackendConnection {
     return execute.result;
   }
 
+  /**
+   * Buffers the given COPY operation for execution on the backend connection when the next
+   * flush/sync message is received. The returned future will contain the result of the COPY
+   * operation when execution has finished.
+   */
+  public Future<StatementResult> executeCopy(
+      ParsedStatement parsedStatement,
+      Statement statement,
+      CopyDataReceiver copyDataReceiver,
+      MutationWriter mutationWriter,
+      ExecutorService executor) {
+    Copy copy = new Copy(parsedStatement, statement, copyDataReceiver, mutationWriter, executor);
+    bufferedStatements.add(copy);
+    return copy.result;
+  }
+
   /** Flushes the buffered statements to Spanner. */
   void flush() {
     flush(false);
@@ -222,6 +312,14 @@ public class BackendConnection {
     }
   }
 
+  Connection getSpannerConnection() {
+    return this.spannerConnection;
+  }
+
+  String getCurrentSchema() {
+    return this.currentSchema;
+  }
+
   /**
    * Flushes all buffered statements to Cloud Spanner and commits/rolls back the transaction at the
    * end if isSync=true.
@@ -298,7 +396,7 @@ public class BackendConnection {
     // We need to start an implicit transaction.
     // Check if a read-only transaction suffices.
     spannerConnection.beginTransaction();
-    if (isSync && !hasDmlStatementsAfter(index)) {
+    if (isSync && !hasDmlOrCopyStatementsAfter(index)) {
       spannerConnection.setTransactionMode(
           com.google.cloud.spanner.connection.TransactionMode.READ_ONLY_TRANSACTION);
     }
@@ -431,14 +529,20 @@ public class BackendConnection {
     return isBegin(index) || isCommit(index) || isRollback(index);
   }
 
-  private boolean hasDmlStatementsAfter(int index) {
+  private boolean hasOnlyDmlStatementsAfter(int index) {
     return bufferedStatements.subList(index, bufferedStatements.size()).stream()
-        .anyMatch(statement -> statement.parsedStatement.getType() == StatementType.UPDATE);
+        .allMatch(statement -> statement.parsedStatement.getType() == StatementType.UPDATE);
   }
 
-  private boolean hasOnlyDmlStatementsAfter(int index) {
+  @VisibleForTesting
+  boolean hasDmlOrCopyStatementsAfter(int index) {
     return bufferedStatements.subList(index, bufferedStatements.size()).stream()
-        .allMatch(statement -> statement.parsedStatement.getType() == StatementType.UPDATE);
+        .anyMatch(
+            statement ->
+                statement.parsedStatement.getType() == StatementType.UPDATE
+                    || statement.parsedStatement.getType() == StatementType.UNKNOWN
+                        && StatementParser.isCommand(
+                            COPY, statement.parsedStatement.getSqlWithoutComments()));
   }
 
   private int getStatementCount() {
@@ -450,11 +554,11 @@ public class BackendConnection {
   }
 
   private boolean canBeBatchedTogether(StatementType statementType1, StatementType statementType2) {
-    if (Objects.equals(statementType1, StatementType.QUERY)
-        || Objects.equals(statementType1, StatementType.CLIENT_SIDE)) {
-      return false;
+    if (Objects.equals(statementType1, StatementType.DDL)
+        || Objects.equals(statementType2, StatementType.UPDATE)) {
+      return Objects.equals(statementType1, statementType2);
     }
-    return Objects.equals(statementType1, statementType2);
+    return false;
   }
 
   /**
@@ -484,11 +588,22 @@ public class BackendConnection {
         throw SpannerExceptionFactory.newSpannerException(
             ErrorCode.INVALID_ARGUMENT, "Statement type is not supported for batching");
     }
+    List<StatementResult> statementResults = new ArrayList<>(getStatementCount());
     int index = fromIndex;
     while (index < getStatementCount()) {
       StatementType statementType = getStatementType(index);
       if (canBeBatchedTogether(batchType, statementType)) {
-        spannerConnection.execute(bufferedStatements.get(index).statement);
+        // Send DDL statements to the DdlExecutor instead of executing them directly on the
+        // connection, so we can support certain DDL constructs that are currently not supported by
+        // the backend, such as IF [NOT] EXISTS.
+        if (batchType == StatementType.DDL) {
+          statementResults.add(
+              ddlExecutor.execute(
+                  bufferedStatements.get(index).parsedStatement,
+                  bufferedStatements.get(index).statement));
+        } else {
+          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
@@ -498,9 +613,17 @@ public class BackendConnection {
     }
     try {
       long[] counts = spannerConnection.runBatch();
+      if (batchType == StatementType.DDL) {
+        counts = extractDdlUpdateCounts(statementResults, counts);
+      }
       updateBatchResultCount(fromIndex, counts);
     } catch (SpannerBatchUpdateException batchUpdateException) {
-      long[] counts = batchUpdateException.getUpdateCounts();
+      long[] counts;
+      if (batchType == StatementType.DDL) {
+        counts = extractDdlUpdateCounts(statementResults, batchUpdateException.getUpdateCounts());
+      } else {
+        counts = batchUpdateException.getUpdateCounts();
+      }
       updateBatchResultCount(fromIndex, counts);
       Execute failedExecute = (Execute) bufferedStatements.get(fromIndex + counts.length);
       failedExecute.result.setException(batchUpdateException);
@@ -509,6 +632,32 @@ public class BackendConnection {
     return index - fromIndex;
   }
 
+  /**
+   * Extracts the update count for a list of DDL statements. It could be that the DdlExecutor
+   * decided to skip some DDL statements. This is indicated by the executor returning a {@link
+   * NotExecuted} result for that statement. The result that we return to the client should still
+   * indicate that the statement was 'executed'.
+   */
+  static long[] extractDdlUpdateCounts(
+      List<StatementResult> statementResults, long[] returnedUpdateCounts) {
+    int index = 0;
+    int successfullyExecutedCount = 0;
+    while (index < returnedUpdateCounts.length
+        && successfullyExecutedCount < statementResults.size()) {
+      if (!(statementResults.get(successfullyExecutedCount) instanceof NotExecuted)) {
+        index++;
+      }
+      successfullyExecutedCount++;
+    }
+    while (successfullyExecutedCount < statementResults.size()
+        && statementResults.get(successfullyExecutedCount) instanceof NotExecuted) {
+      successfullyExecutedCount++;
+    }
+    long[] updateCounts = new long[successfullyExecutedCount];
+    Arrays.fill(updateCounts, 1L);
+    return updateCounts;
+  }
+
   /** 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++) {
@@ -528,7 +677,7 @@ public class BackendConnection {
   public static final class NoResult implements StatementResult {
     private final String commandTag;
 
-    private NoResult() {
+    NoResult() {
       this.commandTag = null;
     }
 
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 f2a2375f..c211427e 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
@@ -30,6 +30,7 @@ import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.parsers.copy.CopyTreeParser;
 import com.google.cloud.spanner.pgadapter.parsers.copy.TokenMgrError;
+import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter.CopyTransactionMode;
 import com.google.common.base.Strings;
@@ -38,17 +39,19 @@ import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.apache.commons.csv.CSVFormat;
 
+/**
+ * {@link CopyStatement} models a `COPY table FROM STDIN` statement. The same class is used both as
+ * an {@link IntermediatePreparedStatement} and {@link IntermediatePortalStatement}, as COPY does
+ * not support any statement parameters, which means that there is no difference between the two.
+ */
 @InternalApi
-public class CopyStatement extends IntermediateStatement {
-
+public class CopyStatement extends IntermediatePortalStatement {
   private static final String COLUMN_NAME = "column_name";
   private static final String DATA_TYPE = "data_type";
 
@@ -59,22 +62,43 @@ public class CopyStatement extends IntermediateStatement {
   private Map<String, TypeCode> tableColumns;
   private int indexedColumnsCount;
   private MutationWriter mutationWriter;
-  private Future<Long> updateCount;
-  private final ExecutorService executor = Executors.newSingleThreadExecutor();
+  // We need two threads to execute a CopyStatement:
+  // 1. A thread to receive the CopyData messages from the client.
+  // 2. A thread to write the received data to Cloud Spanner.
+  // The two are kept separate to allow each to execute at the maximum speed that it can.
+  // The link between them is a piped output/input stream that ensures that backpressure is applied
+  // to the client if the client is sending data at a higher speed than that the writer can handle.
+  private final ExecutorService executor = Executors.newFixedThreadPool(2);
 
   public CopyStatement(
       ConnectionHandler connectionHandler,
       OptionsMetadata options,
+      String name,
       ParsedStatement parsedStatement,
       Statement originalStatement) {
-    super(connectionHandler, options, parsedStatement, originalStatement);
+    super(connectionHandler, options, name, parsedStatement, originalStatement);
   }
 
-  public Exception getException() {
+  @Override
+  public boolean hasException() {
+    return this.exception != null;
+  }
+
+  @Override
+  public SpannerException getException() {
     // Do not clear exceptions on a CopyStatement.
     return this.exception;
   }
 
+  @Override
+  public long getUpdateCount() {
+    // COPY statements continue to execute while the server continues to receive a stream of
+    // CopyData messages AFTER we have received a flush/sync message. We therefore need to block
+    // while waiting for the update count to be available once the server has received a CopyDone
+    // or CopyFailed message.
+    return getUpdateCount(ResultNotReadyBehavior.BLOCK);
+  }
+
   @Override
   public void close() throws Exception {
     if (this.mutationWriter != null) {
@@ -84,16 +108,6 @@ public class CopyStatement extends IntermediateStatement {
     super.close();
   }
 
-  public long getUpdateCount() {
-    try {
-      return updateCount.get();
-    } catch (ExecutionException e) {
-      throw SpannerExceptionFactory.asSpannerException(e.getCause());
-    } catch (InterruptedException e) {
-      throw SpannerExceptionFactory.propagateInterrupt(e);
-    }
-  }
-
   @Override
   public StatementType getStatementType() {
     return StatementType.UPDATE;
@@ -243,7 +257,7 @@ public class CopyStatement extends IntermediateStatement {
             .bind("p1")
             .to(getTableName())
             .build();
-    try (ResultSet result = connection.executeQuery(statement)) {
+    try (ResultSet result = connection.getDatabaseClient().singleUse().executeQuery(statement)) {
       while (result.next()) {
         String columnName = result.getString(COLUMN_NAME);
         TypeCode type = parsePostgreSQLDataType(result.getString(DATA_TYPE));
@@ -282,7 +296,7 @@ public class CopyStatement extends IntermediateStatement {
       paramIndex++;
     }
     Statement statement = builder.build();
-    try (ResultSet resultSet = connection.executeQuery(statement)) {
+    try (ResultSet resultSet = connection.getDatabaseClient().singleUse().executeQuery(statement)) {
       if (resultSet.next()) {
         return (int) resultSet.getLong(0);
       }
@@ -290,13 +304,24 @@ public class CopyStatement extends IntermediateStatement {
     return 0;
   }
 
+  @Override
+  public IntermediatePortalStatement bind(
+      String name,
+      byte[][] parameters,
+      List<Short> parameterFormatCodes,
+      List<Short> resultFormatCodes) {
+    // COPY does not support binding any parameters, so we just return the same statement.
+    return this;
+  }
+
   @Override
   public void handleExecutionException(SpannerException exception) {
     executor.shutdownNow();
     super.handleExecutionException(exception);
   }
 
-  public void execute() {
+  @Override
+  public void executeAsync(BackendConnection backendConnection) {
     this.executed = true;
     try {
       parseCopyStatement();
@@ -311,7 +336,13 @@ public class CopyStatement extends IntermediateStatement {
               indexedColumnsCount,
               getParserFormat(),
               hasHeader());
-      updateCount = executor.submit(mutationWriter);
+      setFutureStatementResult(
+          backendConnection.executeCopy(
+              parsedStatement,
+              statement,
+              new CopyDataReceiver(this, this.connectionHandler),
+              mutationWriter,
+              executor));
     } catch (Exception e) {
       SpannerException spannerException = SpannerExceptionFactory.asSpannerException(e);
       handleExecutionException(spannerException);
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java
new file mode 100644
index 00000000..58cb8220
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java
@@ -0,0 +1,340 @@
+// 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.AbstractLazyInitializer;
+import com.google.cloud.spanner.DatabaseAdminClient;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.DatabaseId;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Spanner;
+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.Connection;
+import com.google.cloud.spanner.connection.ConnectionOptionsHelper;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+
+/**
+ * {@link DdlExecutor} inspects DDL statements before executing these to support commonly used DDL
+ * features that are not (yet) supported by the backend.
+ */
+class DdlExecutor {
+  /**
+   * This class checks whether the backend supports 'if [not] exists' style statements. This is used
+   * to determine whether PGAdapter should do the existence check itself, or whether it should send
+   * the DDL statement unmodified to the backend. The check is only executed once for each
+   * connection, and only if at least one 'if [not] exists' statement is executed.
+   */
+  static class BackendSupportsIfExists extends AbstractLazyInitializer<Boolean> {
+    private final DatabaseId databaseId;
+    private final Connection connection;
+
+    BackendSupportsIfExists(DatabaseId databaseId, Connection connection) {
+      this.databaseId = databaseId;
+      this.connection = connection;
+    }
+
+    @Override
+    protected Boolean initialize() {
+      Spanner spanner = ConnectionOptionsHelper.getSpanner(this.connection);
+      DatabaseAdminClient adminClient = spanner.getDatabaseAdminClient();
+      try {
+        // NOTE: This update will *NEVER* succeed. Either, the first statement fails to parse
+        // because the backend does not support 'if [not] exists'-style statements, or the second
+        // statement fails because it misses a data type for the `id` column. The error that is
+        // returned indicates whether the backend supports `if [not] exists`, as Spanner will parse
+        // the statements in order, and return an error for the first statement that fails.
+        adminClient
+            .updateDatabaseDdl(
+                this.databaseId.getInstanceId().getInstance(),
+                this.databaseId.getDatabase(),
+                ImmutableList.of(
+                    "create table if not exists test (id bigint primary key)",
+                    "create table invalid (id primary key)"),
+                null)
+            .get();
+        // This statement should not be reachable, as at least one of the statements should fail.
+        return Boolean.TRUE;
+      } catch (ExecutionException exception) {
+        SpannerException spannerException =
+            SpannerExceptionFactory.asSpannerException(exception.getCause());
+        if (spannerException.getErrorCode() == ErrorCode.INVALID_ARGUMENT
+            && spannerException
+                .getMessage()
+                .contains("<IF NOT EXISTS> clause is not supported in <CREATE TABLE> statement")) {
+          return Boolean.FALSE;
+        }
+        return Boolean.TRUE;
+      } catch (InterruptedException interruptedException) {
+        throw SpannerExceptionFactory.propagateInterrupt(interruptedException);
+      }
+    }
+  }
+
+  /**
+   * {@link NotExecuted} is used to indicate that a statement was skipped because the 'if [not]
+   * exists' check indicated that the statement should be skipped.
+   */
+  static final class NotExecuted implements StatementResult {
+    @Override
+    public ResultType getResultType() {
+      return ResultType.NO_RESULT;
+    }
+
+    @Override
+    public ClientSideStatementType getClientSideStatementType() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public ResultSet getResultSet() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Long getUpdateCount() {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  private static final NotExecuted NOT_EXECUTED = new NotExecuted();
+  private final AbstractLazyInitializer<Boolean> backendSupportsIfExists;
+  private final BackendConnection backendConnection;
+  private final Connection connection;
+
+  /**
+   * Constructor only intended for testing. This will create an executor that always assumes that
+   * the backend does not support 'if [not] exists'.
+   */
+  @VisibleForTesting
+  DdlExecutor(BackendConnection backendConnection) {
+    this(
+        new AbstractLazyInitializer<Boolean>() {
+          @Override
+          protected Boolean initialize() {
+            return Boolean.FALSE;
+          }
+        },
+        backendConnection);
+  }
+
+  /** Constructor for a {@link DdlExecutor} for the given database and connection. */
+  DdlExecutor(DatabaseId databaseId, BackendConnection backendConnection) {
+    this(
+        new BackendSupportsIfExists(databaseId, backendConnection.getSpannerConnection()),
+        backendConnection);
+  }
+
+  private DdlExecutor(
+      AbstractLazyInitializer<Boolean> backendSupportsIfExists,
+      BackendConnection backendConnection) {
+    this.backendConnection = backendConnection;
+    this.connection = backendConnection.getSpannerConnection();
+    this.backendSupportsIfExists = backendSupportsIfExists;
+  }
+
+  /**
+   * Removes any quotes around the given identifier, and folds the identifier to lower case if it
+   * was not quoted.
+   */
+  static String unquoteIdentifier(String identifier) {
+    if (identifier.length() < 2) {
+      return identifier;
+    }
+    if (identifier.startsWith("\"")) {
+      return identifier.substring(1, identifier.length() - 1);
+    }
+    return identifier.toLowerCase();
+  }
+
+  /**
+   * Executes the given DDL statement. This can include translating the statement into a check
+   * whether the to create/drop object exists, and skipping the actual execution of the DDL
+   * statement depending on the result of the `if [not] exists` check.
+   */
+  StatementResult execute(ParsedStatement parsedStatement, Statement statement) {
+    Statement translated = translate(parsedStatement, statement);
+    if (translated != null) {
+      return connection.execute(translated);
+    }
+    return NOT_EXECUTED;
+  }
+
+  /**
+   * Translates a DDL statement that can contain an `if [not] exists` statement into one that is
+   * supported by the backend.
+   */
+  Statement translate(ParsedStatement parsedStatement, Statement statement) {
+    SimpleParser parser = new SimpleParser(parsedStatement.getSqlWithoutComments());
+    if (parser.eat("create")) {
+      statement = translateCreate(parser, statement);
+    } else if (parser.eat("drop")) {
+      statement = translateDrop(parser, statement);
+    }
+
+    return statement;
+  }
+
+  Statement translateCreate(SimpleParser parser, Statement statement) {
+    if (parser.eat("table")) {
+      return translateCreateTable(parser, statement);
+    }
+    boolean unique = parser.eat("unique");
+    if (parser.eat("index")) {
+      return translateCreateIndex(parser, statement, unique);
+    }
+    return statement;
+  }
+
+  Statement translateDrop(SimpleParser parser, Statement statement) {
+    if (parser.eat("table")) {
+      return translateDropTable(parser, statement);
+    }
+    if (parser.eat("index")) {
+      return translateDropIndex(parser, statement);
+    }
+    return statement;
+  }
+
+  Statement translateCreateTable(SimpleParser parser, Statement statement) {
+    return translateCreateTableOrIndex(parser, statement, "table", this::tableExists);
+  }
+
+  Statement translateCreateIndex(SimpleParser parser, Statement statement, boolean unique) {
+    return translateCreateTableOrIndex(
+        parser, statement, unique ? "unique index" : "index", this::indexExists);
+  }
+
+  Statement translateDropTable(SimpleParser parser, Statement statement) {
+    return translateDropTableOrIndex(parser, statement, "table", this::tableExists);
+  }
+
+  Statement translateDropIndex(SimpleParser parser, Statement statement) {
+    return translateDropTableOrIndex(parser, statement, "index", this::indexExists);
+  }
+
+  private Statement translateCreateTableOrIndex(
+      SimpleParser parser,
+      Statement statement,
+      String type,
+      Function<TableOrIndexName, Boolean> existsFunction) {
+    if (!parser.eat("if", "not", "exists")) {
+      return statement;
+    }
+    return maybeExecuteStatement(
+        parser, statement, "create", type, name -> !existsFunction.apply(name));
+  }
+
+  private Statement translateDropTableOrIndex(
+      SimpleParser parser,
+      Statement statement,
+      String type,
+      Function<TableOrIndexName, Boolean> existsFunction) {
+    if (!parser.eat("if", "exists")) {
+      return statement;
+    }
+    return maybeExecuteStatement(parser, statement, "drop", type, existsFunction);
+  }
+
+  /** Executes the given DDL statement if the shouldExecuteFunction indicates so. */
+  Statement maybeExecuteStatement(
+      SimpleParser parser,
+      Statement statement,
+      String command,
+      String type,
+      Function<TableOrIndexName, Boolean> shouldExecuteFunction) {
+    if (backendSupportsIfExists()) {
+      return statement;
+    }
+    int identifierPosition = parser.getPos();
+    if (identifierPosition >= parser.getSql().length()) {
+      return statement;
+    }
+    TableOrIndexName name = parser.readTableOrIndexName();
+    if (name == null) {
+      return statement;
+    }
+    // Check if the table exists or not.
+    if (!shouldExecuteFunction.apply(name)) {
+      // Return null to indicate that the statement should be skipped.
+      return null;
+    }
+
+    // Return the DDL statement without the 'if [not] exists' clause.
+    return Statement.of(command + " " + type + parser.getSql().substring(identifierPosition));
+  }
+
+  boolean tableExists(TableOrIndexName tableName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                    .bind("p1")
+                    .to(
+                        unquoteIdentifier(
+                            tableName.schema == null
+                                ? backendConnection.getCurrentSchema()
+                                : tableName.schema))
+                    .bind("p2")
+                    .to(unquoteIdentifier(tableName.name))
+                    .build())) {
+      return resultSet.next();
+    }
+  }
+
+  boolean indexExists(TableOrIndexName indexName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                    .bind("p1")
+                    .to(
+                        unquoteIdentifier(
+                            indexName.schema == null
+                                ? backendConnection.getCurrentSchema()
+                                : indexName.schema))
+                    .bind("p2")
+                    .to(unquoteIdentifier(indexName.name))
+                    .build())) {
+      return resultSet.next();
+    }
+  }
+
+  /**
+   * Returns true if the backend supports `if [not] exists`-style statements. This will cause the
+   * {@link DdlExecutor} to send these statements unmodified to the backend.
+   */
+  boolean backendSupportsIfExists() {
+    try {
+      return backendSupportsIfExists.get();
+    } catch (Exception exception) {
+      throw SpannerExceptionFactory.asSpannerException(exception);
+    }
+  }
+}
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
index 64b69afa..861d02bf 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
@@ -41,6 +41,7 @@ public class ExtendedQueryProtocolHandler {
             : connectionHandler.getWellKnownClient().getLocalStatements(connectionHandler);
     this.backendConnection =
         new BackendConnection(
+            connectionHandler.getDatabaseId(),
             connectionHandler.getSpannerConnection(),
             connectionHandler.getServer().getOptions().getDdlTransactionMode(),
             localStatements);
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 8b66a621..71fd1042 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
@@ -28,6 +28,7 @@ import com.google.cloud.spanner.pgadapter.metadata.DescribeStatementMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.parsers.Parser;
 import com.google.cloud.spanner.pgadapter.parsers.Parser.FormatCode;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableSortedSet;
@@ -280,7 +281,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
       return null;
     }
     parser.eat("into");
-    String table = parser.readIdentifier();
+    TableOrIndexName table = parser.readTableOrIndexName();
     if (table == null) {
       return null;
     }
@@ -346,10 +347,10 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
    * without a column list. The query that is used does not use the INFORMATION_SCHEMA, but queries
    * the table directly, so it can use the same transaction as the actual insert statement.
    */
-  static List<String> getAllColumns(Connection connection, String table) {
+  static List<String> getAllColumns(Connection connection, TableOrIndexName table) {
     try (ResultSet resultSet =
         connection.analyzeQuery(
-            Statement.of("SELECT * FROM \"" + table + "\" LIMIT 1"), QueryAnalyzeMode.PLAN)) {
+            Statement.of("SELECT * FROM " + table + " LIMIT 1"), QueryAnalyzeMode.PLAN)) {
       return resultSet.getType().getStructFields().stream()
           .map(StructField::getName)
           .collect(Collectors.toList());
@@ -377,7 +378,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
       return null;
     }
     parser.eat("only");
-    String table = parser.readIdentifier();
+    TableOrIndexName table = parser.readTableOrIndexName();
     if (table == null) {
       return null;
     }
@@ -427,7 +428,7 @@ public class IntermediatePreparedStatement extends IntermediateStatement {
       return null;
     }
     parser.eat("from");
-    String table = parser.readIdentifier();
+    TableOrIndexName table = parser.readTableOrIndexName();
     if (table == null) {
       return null;
     }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
index b1428126..0afb8fc8 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
@@ -14,14 +14,52 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
-import com.google.common.collect.ImmutableSet;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Set;
+import java.util.Objects;
 import javax.annotation.Nullable;
 
+/** A very simple parser that can interpret SQL statements to find specific parts in the string. */
 class SimpleParser {
-  private static final Set<Character> OPERATORS = ImmutableSet.of('+', '-', '*', '/', '!');
+  /** Name of table or index. */
+  static class TableOrIndexName {
+    /** Schema is an optional schema name prefix. */
+    final String schema;
+    /** Name is the actual object name. */
+    final String name;
+
+    TableOrIndexName(String name) {
+      this.schema = null;
+      this.name = name;
+    }
+
+    TableOrIndexName(String schema, String name) {
+      this.schema = schema;
+      this.name = name;
+    }
+
+    @Override
+    public String toString() {
+      if (schema == null) {
+        return name;
+      }
+      return schema + "." + name;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (!(o instanceof TableOrIndexName)) {
+        return false;
+      }
+      TableOrIndexName other = (TableOrIndexName) o;
+      return Objects.equals(this.schema, other.schema) && Objects.equals(this.name, other.name);
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(this.schema, this.name);
+    }
+  }
 
   private final String sql;
   private int pos;
@@ -103,7 +141,22 @@ class SimpleParser {
     return sql.substring(start, pos).trim();
   }
 
-  String readIdentifier() {
+  TableOrIndexName readTableOrIndexName() {
+    String nameOrSchema = readTableOrIndexNamePart();
+    if (nameOrSchema == null) {
+      return null;
+    }
+    if (eat(".")) {
+      String name = readTableOrIndexNamePart();
+      if (name == null) {
+        name = "";
+      }
+      return new TableOrIndexName(nameOrSchema, name);
+    }
+    return new TableOrIndexName(nameOrSchema);
+  }
+
+  String readTableOrIndexNamePart() {
     skipWhitespaces();
     boolean quoted = sql.charAt(pos) == '"';
     int start = pos;
@@ -112,11 +165,18 @@ class SimpleParser {
     }
     while (pos < sql.length()) {
       if (quoted) {
-        if (sql.charAt(pos) == '"' && sql.charAt(pos - 1) != '\\') {
-          return sql.substring(start, ++pos);
+        if (sql.charAt(pos) == '"') {
+          if (pos < (sql.length() - 1) && sql.charAt(pos + 1) == '"') {
+            pos++;
+          } else {
+            return sql.substring(start, ++pos);
+          }
         }
       } else {
-        if (Character.isWhitespace(sql.charAt(pos))) {
+        if (Character.isWhitespace(sql.charAt(pos))
+            || sql.charAt(pos) == '.'
+            || sql.charAt(pos) == ','
+            || sql.charAt(pos) == '"') {
           return sql.substring(start, pos);
         }
       }
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
index 3ea5e098..d823b095 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleQueryStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleQueryStatement.java
@@ -15,6 +15,7 @@
 package com.google.cloud.spanner.pgadapter.statements;
 
 import static com.google.cloud.spanner.pgadapter.utils.StatementParser.splitStatements;
+import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.COPY;
 
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.Dialect;
@@ -26,10 +27,12 @@ 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.utils.ClientAutoDetector.WellKnownClient;
+import com.google.cloud.spanner.pgadapter.utils.StatementParser;
 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.FlushMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.SyncMessage;
 import com.google.common.annotations.VisibleForTesting;
@@ -76,10 +79,20 @@ public class SimpleQueryStatement {
         // The original statement was replaced.
         originalStatement = Statement.of(parsedStatement.getSqlWithoutComments());
       }
+      // We need to flush the entire pipeline if we encounter a COPY statement, as COPY statements
+      // require additional messages to be sent back and forth, and this ensures that we get
+      // everything in the correct order.
+      boolean isCopy = StatementParser.isCommand(COPY, parsedStatement.getSqlWithoutComments());
+      if (isCopy) {
+        new FlushMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN).send();
+      }
       new ParseMessage(connectionHandler, parsedStatement, originalStatement).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();
+      if (isCopy) {
+        new FlushMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN).send();
+      }
     }
     new SyncMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN).send();
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/CopyDataReceiver.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/CopyDataReceiver.java
new file mode 100644
index 00000000..a349b2e5
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/CopyDataReceiver.java
@@ -0,0 +1,92 @@
+// 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.utils;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
+import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
+import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement.ResultNotReadyBehavior;
+import com.google.cloud.spanner.pgadapter.wireoutput.CopyInResponse;
+import com.google.common.annotations.VisibleForTesting;
+import java.util.concurrent.Callable;
+
+/**
+ * This message receiver is activated when the COPY sub-protocol is activated by a COPY statement.
+ * This handler will run and receive all incoming messages until the state of the connection changes
+ * from COPY_IN to either COPY_DONE or COPY_FAILED.
+ */
+@InternalApi
+public class CopyDataReceiver implements Callable<Void> {
+  private final CopyStatement copyStatement;
+  private final ConnectionHandler connectionHandler;
+
+  public CopyDataReceiver(CopyStatement copyStatement, ConnectionHandler connectionHandler) {
+    this.copyStatement = copyStatement;
+    this.connectionHandler = connectionHandler;
+  }
+
+  @Override
+  public Void call() throws Exception {
+    handleCopy();
+    return null;
+  }
+
+  /**
+   * Sends a {@link CopyInResponse} to the client and then waits for copy data, done and fail
+   * messages. The incoming data messages are fed into the {@link MutationWriter} that is associated
+   * with this {@link CopyStatement}. The method blocks until it sees a {@link
+   * com.google.cloud.spanner.pgadapter.wireprotocol.CopyDoneMessage} or {@link
+   * com.google.cloud.spanner.pgadapter.wireprotocol.CopyFailMessage}, or until the {@link
+   * MutationWriter} changes the status of the connection to a non-COPY_IN status, for example as a
+   * result of an error while copying the data to Cloud Spanner.
+   */
+  @VisibleForTesting
+  void handleCopy() throws Exception {
+    if (copyStatement.hasException()) {
+      throw copyStatement.getException();
+    } else {
+      this.connectionHandler.addActiveStatement(copyStatement);
+      new CopyInResponse(
+              this.connectionHandler.getConnectionMetadata().getOutputStream(),
+              copyStatement.getTableColumns().size(),
+              copyStatement.getFormatCode())
+          .send();
+      ConnectionStatus initialConnectionStatus = this.connectionHandler.getStatus();
+      try {
+        this.connectionHandler.setStatus(ConnectionStatus.COPY_IN);
+        // Loop here until COPY_IN mode has finished.
+        while (this.connectionHandler.getStatus() == ConnectionStatus.COPY_IN) {
+          this.connectionHandler.handleMessages();
+        }
+        if (copyStatement.hasException(ResultNotReadyBehavior.BLOCK)
+            || this.connectionHandler.getStatus() == ConnectionStatus.COPY_FAILED) {
+          if (copyStatement.hasException(ResultNotReadyBehavior.BLOCK)) {
+            throw copyStatement.getException();
+          } else {
+            throw SpannerExceptionFactory.newSpannerException(
+                ErrorCode.INTERNAL, "Copy failed with unknown reason");
+          }
+        }
+      } finally {
+        this.connectionHandler.removeActiveStatement(copyStatement);
+        this.copyStatement.close();
+        this.connectionHandler.setStatus(initialConnectionStatus);
+      }
+    }
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
index b4e6b878..986990e9 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
@@ -28,6 +28,8 @@ import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.UpdateCount;
 import com.google.common.collect.ImmutableList;
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
@@ -50,6 +52,7 @@ import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.Callable;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
@@ -66,7 +69,7 @@ import org.apache.commons.csv.CSVRecord;
 import org.postgresql.jdbc.TimestampUtils;
 
 @InternalApi
-public class MutationWriter implements Callable<Long>, Closeable {
+public class MutationWriter implements Callable<StatementResult>, Closeable {
   public enum CopyTransactionMode {
     /**
      * 'Normal' auto-commit mode. The entire COPY operation is atomic. If the number of mutations
@@ -129,7 +132,9 @@ public class MutationWriter implements Callable<Long>, Closeable {
   private final CSVParser parser;
   private PrintWriter errorFileWriter;
   private final PipedOutputStream payload = new PipedOutputStream();
+  private final AtomicBoolean commit = new AtomicBoolean(false);
   private final AtomicBoolean rollback = new AtomicBoolean(false);
+  private final CountDownLatch closedLatch = new CountDownLatch(1);
   private final ListeningExecutorService executorService =
       MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
 
@@ -199,14 +204,23 @@ public class MutationWriter implements Callable<Long>, Closeable {
     try {
       this.payload.write(payload);
     } catch (IOException e) {
-      SpannerException spannerException =
-          SpannerExceptionFactory.newSpannerException(
-              ErrorCode.INTERNAL, "Could not write copy data to buffer", e);
-      logger.log(Level.SEVERE, spannerException.getMessage(), spannerException);
-      throw spannerException;
+      // Ignore the exception if the executor has already been shutdown. That means that an error
+      // occurred that ended the COPY operation while we were writing data to the buffer.
+      if (!executorService.isShutdown()) {
+        SpannerException spannerException =
+            SpannerExceptionFactory.newSpannerException(
+                ErrorCode.INTERNAL, "Could not write copy data to buffer", e);
+        logger.log(Level.SEVERE, spannerException.getMessage(), spannerException);
+        throw spannerException;
+      }
     }
   }
 
+  /** Indicate that this mutation writer should commit. */
+  public void commit() {
+    this.commit.set(true);
+  }
+
   /**
    * Indicate that this mutation writer should be rolled back. This will not rollback any changes
    * that have already been committed if the mutation writer is running in {@link
@@ -219,10 +233,11 @@ public class MutationWriter implements Callable<Long>, Closeable {
   @Override
   public void close() throws IOException {
     this.payload.close();
+    this.closedLatch.countDown();
   }
 
   @Override
-  public Long call() throws Exception {
+  public StatementResult call() throws Exception {
     // This LinkedBlockingDeque holds a reference to all transactions that are currently active. The
     // max capacity of this deque is what ensures that we never have more than maxParallelism
     // transactions running at the same time. We could also achieve that by using a thread pool with
@@ -304,9 +319,15 @@ public class MutationWriter implements Callable<Long>, Closeable {
       // Write any remaining mutations in the buffer.
       if (!rollback.get() && !mutations.isEmpty()) {
         if (transactionMode == CopyTransactionMode.Explicit) {
-          connection.bufferedWrite(mutations);
+          connection.write(mutations);
         } else {
-          allCommitFutures.add(writeToSpannerAsync(activeCommitFutures, mutations));
+          // Wait until we have received a CopyDone message before writing the remaining data to
+          // Spanner. If we are in a non-atomic transaction, there might already be data that have
+          // been written to Spanner.
+          closedLatch.await();
+          if (commit.get()) {
+            allCommitFutures.add(writeToSpannerAsync(activeCommitFutures, mutations));
+          }
         }
       }
       // Wait for all commits to finish. We do this even if something went wrong, as it ensures two
@@ -339,7 +360,7 @@ public class MutationWriter implements Callable<Long>, Closeable {
       this.payload.close();
       this.parser.close();
     }
-    return rowCount;
+    return new UpdateCount(rowCount);
   }
 
   private long addMutationAndMaybeFlushTransaction(
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/CopyInResponse.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/CopyInResponse.java
index e3b9b646..588ea9c5 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/CopyInResponse.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/CopyInResponse.java
@@ -52,7 +52,6 @@ public class CopyInResponse extends WireOutput {
     this.outputStream.writeByte(this.formatCode);
     this.outputStream.writeShort(this.numColumns);
     this.outputStream.write(this.columnFormat);
-    this.outputStream.flush();
   }
 
   @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 26f64cad..cd06719f 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
@@ -25,6 +25,7 @@ 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.Severity;
 import com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse.State;
 import com.google.cloud.spanner.pgadapter.wireoutput.PortalSuspendedResponse;
 import com.google.common.base.Preconditions;
@@ -40,6 +41,9 @@ import java.util.List;
  */
 @InternalApi
 public abstract class ControlMessage extends WireMessage {
+  /** Maximum number of invalid messages in a row allowed before we terminate the connection. */
+  static final int MAX_INVALID_MESSAGE_COUNT = 50;
+
   /**
    * Token that is used to mark {@link ControlMessage}s that are manually created to execute a
    * {@link QueryMessage}.
@@ -73,50 +77,85 @@ public abstract class ControlMessage extends WireMessage {
    * @throws Exception If construction or reading fails.
    */
   public static ControlMessage create(ConnectionHandler connection) throws Exception {
+    boolean validMessage = true;
     char nextMsg = (char) connection.getConnectionMetadata().getInputStream().readUnsignedByte();
-    if (connection.getStatus() == ConnectionStatus.COPY_IN) {
-      switch (nextMsg) {
-        case CopyDoneMessage.IDENTIFIER:
-          return new CopyDoneMessage(connection);
-        case CopyDataMessage.IDENTIFIER:
-          return new CopyDataMessage(connection);
-        case CopyFailMessage.IDENTIFIER:
-          return new CopyFailMessage(connection);
-        default:
-          // Drop the connection if we receive an invalid message to prevent further CopyData
-          // messages from coming in. There is no error handling in the COPY protocol, and some
-          // clients will blindly continue to send data and never check any possible responses from
-          // the server during a (large) copy operation, so this is the safest option.
-          connection.setStatus(ConnectionStatus.TERMINATED);
-          throw new IllegalStateException(
-              String.format(
-                  "Expected CopyData ('d'), CopyDone ('c') or CopyFail ('f') messages, got: '%c'",
-                  nextMsg));
+    try {
+      if (connection.getStatus() == ConnectionStatus.COPY_IN) {
+        switch (nextMsg) {
+          case CopyDoneMessage.IDENTIFIER:
+            return new CopyDoneMessage(connection);
+          case CopyDataMessage.IDENTIFIER:
+            return new CopyDataMessage(connection);
+          case CopyFailMessage.IDENTIFIER:
+            return new CopyFailMessage(connection);
+          case SyncMessage.IDENTIFIER:
+          case FlushMessage.IDENTIFIER:
+            // Skip sync/flush in COPY_IN. This is consistent with real PostgreSQL which also does
+            // this to accommodate clients that do not check what type of statement they sent in an
+            // ExecuteMessage, and instead always blindly send a flush/sync after each execute.
+            return SkipMessage.createForValidStream(connection);
+          default:
+            // Skip other unexpected messages and throw an exception to fail the copy operation.
+            validMessage = false;
+            SkipMessage.createForInvalidStream(connection);
+            throw new IllegalStateException(
+                String.format(
+                    "Expected CopyData ('d'), CopyDone ('c') or CopyFail ('f') messages, got: '%c'",
+                    nextMsg));
+        }
+      } else {
+        switch (nextMsg) {
+          case QueryMessage.IDENTIFIER:
+            return new QueryMessage(connection);
+          case ParseMessage.IDENTIFIER:
+            return new ParseMessage(connection);
+          case BindMessage.IDENTIFIER:
+            return new BindMessage(connection);
+          case DescribeMessage.IDENTIFIER:
+            return new DescribeMessage(connection);
+          case ExecuteMessage.IDENTIFIER:
+            return new ExecuteMessage(connection);
+          case CloseMessage.IDENTIFIER:
+            return new CloseMessage(connection);
+          case SyncMessage.IDENTIFIER:
+            return new SyncMessage(connection);
+          case TerminateMessage.IDENTIFIER:
+            return new TerminateMessage(connection);
+          case FunctionCallMessage.IDENTIFIER:
+            return new FunctionCallMessage(connection);
+          case FlushMessage.IDENTIFIER:
+            return new FlushMessage(connection);
+          case CopyDoneMessage.IDENTIFIER:
+          case CopyDataMessage.IDENTIFIER:
+          case CopyFailMessage.IDENTIFIER:
+            // Silently skip COPY messages in non-COPY mode. This is consistent with the PG wire
+            // protocol. If we continue to receive COPY messages while in non-COPY mode, we'll
+            // terminate the connection to prevent the server from being flooded with invalid
+            // messages.
+            validMessage = false;
+            // Note: The stream itself is still valid as we received a message that we recognized.
+            return SkipMessage.createForValidStream(connection);
+          default:
+            throw new IllegalStateException(String.format("Unknown message: %c", nextMsg));
+        }
       }
-    } else {
-      switch (nextMsg) {
-        case QueryMessage.IDENTIFIER:
-          return new QueryMessage(connection);
-        case ParseMessage.IDENTIFIER:
-          return new ParseMessage(connection);
-        case BindMessage.IDENTIFIER:
-          return new BindMessage(connection);
-        case DescribeMessage.IDENTIFIER:
-          return new DescribeMessage(connection);
-        case ExecuteMessage.IDENTIFIER:
-          return new ExecuteMessage(connection);
-        case CloseMessage.IDENTIFIER:
-          return new CloseMessage(connection);
-        case SyncMessage.IDENTIFIER:
-          return new SyncMessage(connection);
-        case TerminateMessage.IDENTIFIER:
-          return new TerminateMessage(connection);
-        case FunctionCallMessage.IDENTIFIER:
-          return new FunctionCallMessage(connection);
-        case FlushMessage.IDENTIFIER:
-          return new FlushMessage(connection);
-        default:
-          throw new IllegalStateException(String.format("Unknown message: %c", nextMsg));
+    } finally {
+      if (validMessage) {
+        connection.clearInvalidMessageCount();
+      } else {
+        connection.increaseInvalidMessageCount();
+        if (connection.getInvalidMessageCount() > MAX_INVALID_MESSAGE_COUNT) {
+          new ErrorResponse(
+                  connection.getConnectionMetadata().getOutputStream(),
+                  new IllegalStateException(
+                      String.format(
+                          "Received %d invalid/unexpected messages. Last received message: '%c'",
+                          connection.getInvalidMessageCount(), nextMsg)),
+                  State.InternalError,
+                  Severity.FATAL)
+              .send();
+          connection.setStatus(ConnectionStatus.TERMINATED);
+        }
       }
     }
   }
@@ -204,6 +243,7 @@ public abstract class ControlMessage extends WireMessage {
         command += ("INSERT".equals(command) ? " 0 " : " ") + statement.getUpdateCount();
         new CommandCompleteResponse(this.outputStream, command).send(false);
         break;
+      case UNKNOWN:
       default:
         throw new IllegalStateException("Unknown statement type: " + statement.getStatement());
     }
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 cd1d7c24..25613a83 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
@@ -17,7 +17,6 @@ package com.google.cloud.spanner.pgadapter.wireprotocol;
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import java.text.MessageFormat;
@@ -48,7 +47,10 @@ public class CopyDataMessage extends ControlMessage {
   @Override
   protected void sendPayload() throws Exception {
     if (statement == null) {
-      throw new IllegalStateException("This connection is no (longer) in COPY IN mode.");
+      // Do not handle this message if there is no CopyStatement available anymore. This means that
+      // the copy operation failed and stopped while the client was still sending data to the
+      // server.
+      return;
     }
     // If backend error occurred during copy-in mode, drop any subsequent CopyData messages.
     MutationWriter mutationWriter = this.statement.getMutationWriter();
@@ -59,12 +61,6 @@ public class CopyDataMessage extends ControlMessage {
         statement.handleExecutionException(exception);
         throw exception;
       }
-    } else {
-      // If we get another CopyData message after we have already failed, then we drop the
-      // connection to ensure that the client is not overflowing us with messages. Some clients do
-      // not check for error messages during a COPY operation.
-      connection.setStatus(ConnectionStatus.TERMINATED);
-      throw statement.getException();
     }
   }
 
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 4c013730..7c2d4e1a 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
@@ -17,11 +17,7 @@ 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.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;
 import java.text.MessageFormat;
 
 /**
@@ -44,26 +40,13 @@ public class CopyDoneMessage extends ControlMessage {
   protected void sendPayload() throws Exception {
     // If backend error occurred during copy-in mode, drop any subsequent CopyDone messages.
     if (this.statement != null) {
-      MutationWriter mutationWriter = this.statement.getMutationWriter();
+      statement.getMutationWriter().commit();
       statement.close();
-      if (!statement.hasException()) {
-        try {
-          long rowCount = this.statement.getUpdateCount();
-          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.
-          this.connection.setStatus(ConnectionStatus.AUTHENTICATED);
-          this.connection.removeActiveStatement(this.statement);
-          throw e;
-        }
-      }
     }
-    this.connection.setStatus(ConnectionStatus.AUTHENTICATED);
-    this.connection.removeActiveStatement(this.statement);
-    new ReadyResponse(this.outputStream, ReadyResponse.Status.IDLE).send();
+    // Clear the COPY_IN status to indicate that we finished successfully. This will cause the
+    // inline handling of incoming (copy) messages to stop and the server to resume normal
+    // operation.
+    this.connection.setStatus(ConnectionStatus.COPY_DONE);
   }
 
   @Override
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 cdb9b9c6..0c55b5d4 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
@@ -15,13 +15,13 @@
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
+import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
-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 java.text.MessageFormat;
 
 /**
@@ -32,33 +32,35 @@ import java.text.MessageFormat;
  */
 @InternalApi
 public class CopyFailMessage extends ControlMessage {
-
   protected static final char IDENTIFIER = 'f';
-  private CopyStatement statement;
 
-  private String errorMessage;
+  private final CopyStatement statement;
+  private final String errorMessage;
 
   public CopyFailMessage(ConnectionHandler connection) throws Exception {
     super(connection);
     this.errorMessage = this.readAll();
-    this.statement = (CopyStatement) connection.getActiveStatement();
+    IntermediateStatement activeStatement = connection.getActiveStatement();
+    if (activeStatement instanceof CopyStatement) {
+      this.statement = (CopyStatement) activeStatement;
+    } else {
+      this.statement = null;
+    }
   }
 
   @Override
   protected void sendPayload() throws Exception {
-    // If backend error occurred during copy-in mode, drop any subsequent CopyFail messages.
     if (this.statement != null) {
       MutationWriter mutationWriter = this.statement.getMutationWriter();
       mutationWriter.rollback();
       statement.close();
-      if (!statement.hasException()) {
-        new ErrorResponse(this.outputStream, new Exception(this.errorMessage), State.IOError)
-            .send();
-      }
+      this.statement.handleExecutionException(
+          SpannerExceptionFactory.newSpannerException(ErrorCode.CANCELLED, this.errorMessage));
     }
-    this.connection.setStatus(ConnectionStatus.AUTHENTICATED);
-    this.connection.removeActiveStatement(this.statement);
-    new ReadyResponse(this.outputStream, ReadyResponse.Status.IDLE).send();
+    // Clear the COPY_IN status to indicate that we finished unsuccessfully. This will cause the
+    // inline handling of incoming (copy) messages to stop and the server to return an error message
+    // and to resume normal operation.
+    this.connection.setStatus(ConnectionStatus.COPY_FAILED);
   }
 
   @Override
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 a3b4834d..880cbf02 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
@@ -21,7 +21,7 @@ import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 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.statements.IntermediatePreparedStatement;
 import com.google.cloud.spanner.pgadapter.wireoutput.NoDataResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ParameterDescriptionResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.RowDescriptionResponse;
@@ -38,7 +38,7 @@ public class DescribeMessage extends AbstractQueryProtocolMessage {
 
   private final PreparedType type;
   private final String name;
-  private final IntermediateStatement statement;
+  private final IntermediatePreparedStatement statement;
   private Future<DescribePortalMetadata> describePortalMetadata;
 
   public DescribeMessage(ConnectionHandler connection) throws Exception {
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 7fb148da..3fb6f623 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
@@ -17,7 +17,7 @@ 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.IntermediateStatement;
+import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
 import java.text.MessageFormat;
 
 /** Executes a portal. */
@@ -27,7 +27,7 @@ public class ExecuteMessage extends AbstractQueryProtocolMessage {
 
   private final String name;
   private final int maxRows;
-  private final IntermediateStatement statement;
+  private final IntermediatePreparedStatement statement;
 
   public ExecuteMessage(ConnectionHandler connection) throws Exception {
     super(connection);
@@ -94,7 +94,6 @@ public class ExecuteMessage extends AbstractQueryProtocolMessage {
     } else {
       try {
         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 8aeab56b..a40a5bc4 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
@@ -16,13 +16,11 @@ 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.wireoutput.ReadyResponse;
-import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
 import java.text.MessageFormat;
 
 /**
- * Handles a flush command from the user. JDBC does not require this step, so server-side this is a
- * noop.
+ * Handles a flush command from the user. This flushes any pending changes to Cloud Spanner and then
+ * sends any pending messages to the client.
  */
 @InternalApi
 public class FlushMessage extends ControlMessage {
@@ -33,13 +31,13 @@ public class FlushMessage extends ControlMessage {
     super(connection);
   }
 
+  public FlushMessage(ConnectionHandler connection, ManuallyCreatedToken manuallyCreatedToken) {
+    super(connection, 4, manuallyCreatedToken);
+  }
+
   @Override
   protected void sendPayload() throws Exception {
     connection.getExtendedQueryProtocolHandler().flush();
-    boolean inTransaction = connection.getSpannerConnection().isInTransaction();
-    new ReadyResponse(
-            this.outputStream, inTransaction ? Status.TRANSACTION : ReadyResponse.Status.IDLE)
-        .send();
   }
 
   @Override
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 635ee581..4f279c0d 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
@@ -14,6 +14,8 @@
 
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
+import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.COPY;
+
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.Statement;
@@ -21,7 +23,9 @@ 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.CopyStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
+import com.google.cloud.spanner.pgadapter.utils.StatementParser;
 import com.google.cloud.spanner.pgadapter.wireoutput.ParseCompleteResponse;
 import com.google.common.base.Strings;
 import java.text.MessageFormat;
@@ -48,13 +52,7 @@ public class ParseMessage extends AbstractQueryProtocolMessage {
       parameterDataTypes[i] = this.inputStream.readInt();
     }
     this.statement =
-        new IntermediatePreparedStatement(
-            connection,
-            connection.getServer().getOptions(),
-            name,
-            parsedStatement,
-            originalStatement);
-    this.statement.setParameterDataTypes(this.parameterDataTypes);
+        createStatement(connection, name, parsedStatement, originalStatement, parameterDataTypes);
   }
 
   /**
@@ -69,13 +67,33 @@ public class ParseMessage extends AbstractQueryProtocolMessage {
     this.name = "";
     this.parameterDataTypes = new int[0];
     this.statement =
-        new IntermediatePreparedStatement(
-            connection,
-            connection.getServer().getOptions(),
-            name,
-            parsedStatement,
-            originalStatement);
-    this.statement.setParameterDataTypes(this.parameterDataTypes);
+        createStatement(connection, name, parsedStatement, originalStatement, parameterDataTypes);
+  }
+
+  static IntermediatePreparedStatement createStatement(
+      ConnectionHandler connectionHandler,
+      String name,
+      ParsedStatement parsedStatement,
+      Statement originalStatement,
+      int[] parameterDataTypes) {
+    if (StatementParser.isCommand(COPY, parsedStatement.getSqlWithoutComments())) {
+      return new CopyStatement(
+          connectionHandler,
+          connectionHandler.getServer().getOptions(),
+          name,
+          parsedStatement,
+          originalStatement);
+    } else {
+      IntermediatePreparedStatement statement =
+          new IntermediatePreparedStatement(
+              connectionHandler,
+              connectionHandler.getServer().getOptions(),
+              name,
+              parsedStatement,
+              originalStatement);
+      statement.setParameterDataTypes(parameterDataTypes);
+      return statement;
+    }
   }
 
   @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 91280cc1..7ee21079 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
@@ -15,62 +15,31 @@
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 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.pgadapter.ConnectionHandler;
-import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
-import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
 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.ReadyResponse;
 import java.text.MessageFormat;
 
 /** Executes a simple statement. */
 @InternalApi
 public class QueryMessage extends ControlMessage {
-  private static final AbstractStatementParser PARSER =
-      AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
   protected static final char IDENTIFIER = 'Q';
   public static final String COPY = "COPY";
 
   private final Statement originalStatement;
-  private final CopyStatement copyStatement;
   private final SimpleQueryStatement simpleQueryStatement;
-  private final boolean isCopy;
 
   public QueryMessage(ConnectionHandler connection) throws Exception {
     super(connection);
     this.originalStatement = Statement.of(this.readAll());
-    ParsedStatement parsedStatement = PARSER.parse(originalStatement);
-    this.isCopy = StatementParser.isCommand(COPY, parsedStatement.getSqlWithoutComments());
-    if (isCopy) {
-      this.copyStatement =
-          new CopyStatement(
-              connection,
-              connection.getServer().getOptions(),
-              parsedStatement,
-              this.originalStatement);
-      this.simpleQueryStatement = null;
-      this.connection.addActiveStatement(this.copyStatement);
-    } else {
-      this.simpleQueryStatement =
-          new SimpleQueryStatement(
-              connection.getServer().getOptions(), this.originalStatement, this.connection);
-      this.copyStatement = null;
-    }
+    this.simpleQueryStatement =
+        new SimpleQueryStatement(
+            connection.getServer().getOptions(), this.originalStatement, this.connection);
   }
 
   @Override
   protected void sendPayload() throws Exception {
-    if (this.isCopy) {
-      this.copyStatement.execute();
-      handleCopy();
-    } else {
-      this.simpleQueryStatement.execute();
-    }
+    this.simpleQueryStatement.execute();
   }
 
   @Override
@@ -96,25 +65,4 @@ public class QueryMessage extends ControlMessage {
   public SimpleQueryStatement getSimpleQueryStatement() {
     return this.simpleQueryStatement;
   }
-
-  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);
-    }
-  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/SkipMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/SkipMessage.java
new file mode 100644
index 00000000..7449bccc
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/SkipMessage.java
@@ -0,0 +1,84 @@
+// 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 java.io.DataInputStream;
+import java.io.IOException;
+import java.text.MessageFormat;
+
+/**
+ * {@link SkipMessage} is a pseudo wire-protocol message that is used to read and skip messages that
+ * we receive that we do not want at that time. This applies to COPY messages during normal
+ * operation, and sync/flush messages during COPY operation.
+ */
+@InternalApi
+public class SkipMessage extends ControlMessage {
+
+  public static SkipMessage createForValidStream(ConnectionHandler connectionHandler)
+      throws IOException {
+    return new SkipMessage(connectionHandler, false);
+  }
+
+  public static SkipMessage createForInvalidStream(ConnectionHandler connectionHandler)
+      throws IOException {
+    return new SkipMessage(connectionHandler, true);
+  }
+
+  private SkipMessage(ConnectionHandler connectionHandler, boolean streamIsPotentiallyInvalid)
+      throws IOException {
+    super(connectionHandler);
+    int skipLength = this.length - 4;
+    int skipped = 0;
+    // Read and skip bytes until we have reached the total message length.
+    DataInputStream input = connection.getConnectionMetadata().getInputStream();
+    // If the stream is potentially invalid we'll use a more pessimistic skipping algorithm than if
+    // we just received a message that we want to skip. We do this because the Unix domain socket
+    // implementation might block on skip(n) if the stream has no bytes more to read.
+    if (streamIsPotentiallyInvalid) {
+      skipInvalid(input, skipLength);
+    } else {
+      while (skipped < skipLength) {
+        skipped += input.skip(skipLength - skipped);
+      }
+    }
+  }
+
+  private void skipInvalid(DataInputStream input, int numBytes) throws IOException {
+    int skipped = 0;
+    while (skipped < numBytes && input.read() != -1) {
+      skipped++;
+    }
+  }
+
+  @Override
+  protected void sendPayload() throws Exception {}
+
+  @Override
+  protected String getMessageName() {
+    return "Skip";
+  }
+
+  @Override
+  protected String getPayloadString() {
+    return new MessageFormat("Length: {0}").format(new Object[] {this.length});
+  }
+
+  @Override
+  protected String getIdentifier() {
+    return "";
+  }
+}


Test patch:
diff --git a/.ci/e2e-expected/copy-from-stdin-batch.txt b/.ci/e2e-expected/copy-from-stdin-batch.txt
new file mode 100644
index 00000000..0da509b7
--- /dev/null
+++ b/.ci/e2e-expected/copy-from-stdin-batch.txt
@@ -0,0 +1,22 @@
+  id | age | name
+-----+-----+--------
+   1 |   1 | John
+   2 |  20 | Joe
+   3 |  23 | Jack
+   7 |   7 | 7
+   8 |   8 | 8
+   9 |   9 | 9
+  10 |  10 | 10
+  11 |  11 | 11
+  12 |  12 | 12
+  13 |  13 | 13
+  14 |  14 | 14
+  15 |  15 | 15
+  16 |  16 | 16
+  17 |  17 | 17
+  18 |  18 | 18
+  19 |  19 | 19
+  99 |  99 | person
+ 100 | 100 | test
+(18 rows)
+
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 6f4deda2..4b477774 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
@@ -41,6 +41,19 @@ import java.util.concurrent.Future;
  */
 @InternalApi
 public class IntermediateStatement {
+  /**
+   * Indicates whether an attempt to get the result of a statement should block or fail if the
+   * result is not yet available. Normal SQL commands that can be executed directly on Cloud Spanner
+   * should always have their results available when a sync/flush message is received. COPY
+   * statements do not have that, as they require additional messages after a flush/sync has been
+   * received. Attempts to get the result of a COPY statement should therefore block until it is
+   * available, which is after a CopyDone or CopyFail message has been received.
+   */
+  public enum ResultNotReadyBehavior {
+    FAIL,
+    BLOCK;
+  }
+
   protected static final PostgreSQLStatementParser PARSER =
       (PostgreSQLStatementParser) AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
 
@@ -122,11 +135,24 @@ public class IntermediateStatement {
 
   /**
    * @return The number of items that were modified by this execution for DML. 0 for DDL and -1 for
-   *     QUERY.
+   *     QUERY. Fails if the result is not yet available.
    */
   public long getUpdateCount() {
-    initFutureResult();
-    switch (this.parsedStatement.getType()) {
+    return getUpdateCount(ResultNotReadyBehavior.FAIL);
+  }
+
+  /**
+   * @return The number of items that were modified by this execution for DML. 0 for DDL and -1 for
+   *     QUERY. Will block or fail depending on the given {@link ResultNotReadyBehavior} if the
+   *     result is not yet available.
+   */
+  public long getUpdateCount(ResultNotReadyBehavior resultNotReadyBehavior) {
+    initFutureResult(resultNotReadyBehavior);
+    if (hasException()) {
+      throw getException();
+    }
+    // Note: getStatementType() returns UPDATE for COPY statements.
+    switch (getStatementType()) {
       case QUERY:
         return -1L;
       case UPDATE:
@@ -139,9 +165,20 @@ public class IntermediateStatement {
     }
   }
 
-  /** @return True if at some point in execution, and exception was thrown. */
+  /**
+   * @return True if at some point in execution an exception was thrown. Fails if execution has not
+   *     yet finished.
+   */
   public boolean hasException() {
-    initFutureResult();
+    return hasException(ResultNotReadyBehavior.FAIL);
+  }
+
+  /**
+   * @return True if at some point in execution an exception was thrown. Fails or blocks depending
+   *     on the given {@link ResultNotReadyBehavior} if execution has not yet finished.
+   */
+  public boolean hasException(ResultNotReadyBehavior resultNotReadyBehavior) {
+    initFutureResult(resultNotReadyBehavior);
     return this.exception != null;
   }
 
@@ -162,25 +199,30 @@ public class IntermediateStatement {
     return this.parsedStatement.getSqlWithoutComments();
   }
 
-  private void initFutureResult() {
+  private void initFutureResult(ResultNotReadyBehavior resultNotReadyBehavior) {
     if (this.futureStatementResult != null) {
-      if (!this.futureStatementResult.isDone()) {
+      if (resultNotReadyBehavior == ResultNotReadyBehavior.FAIL
+          && !this.futureStatementResult.isDone()) {
         throw new IllegalStateException("Statement result cannot be retrieved before flush/sync");
       }
       try {
         setStatementResult(this.futureStatementResult.get());
       } catch (ExecutionException executionException) {
-        this.exception = SpannerExceptionFactory.asSpannerException(executionException.getCause());
+        setException(SpannerExceptionFactory.asSpannerException(executionException.getCause()));
       } catch (InterruptedException interruptedException) {
-        this.exception = SpannerExceptionFactory.propagateInterrupt(interruptedException);
+        setException(SpannerExceptionFactory.propagateInterrupt(interruptedException));
       } finally {
         this.futureStatementResult = null;
       }
     }
   }
 
+  /**
+   * Returns the result of this statement as a {@link StatementResult}. Fails if the result is not
+   * yet available.
+   */
   public StatementResult getStatementResult() {
-    initFutureResult();
+    initFutureResult(ResultNotReadyBehavior.FAIL);
     return this.statementResult;
   }
 
@@ -206,10 +248,18 @@ public class IntermediateStatement {
     return this.parsedStatement.getSqlWithoutComments();
   }
 
-  public Exception getException() {
-    Exception e = this.exception;
-    this.exception = null;
-    return e;
+  /** Returns any execution exception registered for this statement. */
+  public SpannerException getException() {
+    return this.exception;
+  }
+
+  void setException(SpannerException exception) {
+    // Do not override any exception that has already been registered. COPY statements can receive
+    // multiple errors as they execute asynchronously while receiving a stream of data from the
+    // client. We always return the first exception that we encounter.
+    if (this.exception == null) {
+      this.exception = exception;
+    }
   }
 
   /**
@@ -218,7 +268,7 @@ public class IntermediateStatement {
    * @param exception The exception to store.
    */
   protected void handleExecutionException(SpannerException exception) {
-    this.exception = exception;
+    setException(exception);
     this.hasMoreData = false;
   }
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
index 1bf3e92c..99ca458e 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
@@ -106,6 +106,8 @@ public abstract class AbstractMockServerTest {
                           .build())
                   .build())
           .build();
+  protected static final com.google.spanner.v1.ResultSet EMPTY_RESULTSET =
+      com.google.spanner.v1.ResultSet.newBuilder().setMetadata(SELECT1_METADATA).build();
   protected static final com.google.spanner.v1.ResultSet SELECT1_RESULTSET =
       com.google.spanner.v1.ResultSet.newBuilder()
           .addRows(
@@ -405,11 +407,15 @@ public abstract class AbstractMockServerTest {
     }
   }
 
+  protected void closeSpannerPool() {
+    closeSpannerPool(false);
+  }
+
   /**
    * Closes all open Spanner instances in the pool. Use this to force the recreation of a Spanner
    * instance for a test case. This method will ignore any errors and retry if closing fails.
    */
-  protected void closeSpannerPool() {
+  protected void closeSpannerPool(boolean ignoreException) {
     SpannerException exception = null;
     for (int attempt = 0; attempt < 1000; attempt++) {
       try {
@@ -424,7 +430,9 @@ public abstract class AbstractMockServerTest {
         exception = e;
       }
     }
-    throw exception;
+    if (!ignoreException) {
+      throw exception;
+    }
   }
 
   @Before
@@ -450,4 +458,11 @@ public abstract class AbstractMockServerTest {
     mockDatabaseAdmin.addException(
         Status.INVALID_ARGUMENT.withDescription("Statement is invalid.").asRuntimeException());
   }
+
+  protected static void addIfNotExistsDdlException() {
+    mockDatabaseAdmin.addException(
+        Status.INVALID_ARGUMENT
+            .withDescription("<IF NOT EXISTS> clause is not supported in <CREATE TABLE> statement.")
+            .asRuntimeException());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
index 0e5e65e0..e05ca7a1 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
@@ -14,11 +14,20 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
+import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
+import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.io.EOFException;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.Socket;
@@ -74,4 +83,60 @@ public class ConnectionHandlerTest {
     connection.terminate();
     verify(socket).close();
   }
+
+  @Test
+  public void testHandleMessages_NonFatalException() throws Exception {
+    ProxyServer server = mock(ProxyServer.class);
+    Socket socket = mock(Socket.class);
+    InetAddress address = mock(InetAddress.class);
+    when(socket.getInetAddress()).thenReturn(address);
+    DataOutputStream dataOutputStream = new DataOutputStream(new ByteArrayOutputStream());
+    ConnectionMetadata connectionMetadata = mock(ConnectionMetadata.class);
+    when(connectionMetadata.getOutputStream()).thenReturn(dataOutputStream);
+    WireMessage message = mock(WireMessage.class);
+    when(server.recordMessage(message)).thenReturn(message);
+    doThrow(
+            SpannerExceptionFactory.newSpannerException(
+                ErrorCode.FAILED_PRECONDITION, "non-fatal test exception"))
+        .when(message)
+        .send();
+
+    ConnectionHandler connection =
+        new ConnectionHandler(server, socket) {
+          public ConnectionMetadata getConnectionMetadata() {
+            return connectionMetadata;
+          }
+        };
+
+    connection.setMessageState(message);
+    connection.handleMessages();
+
+    assertEquals(ConnectionStatus.UNAUTHENTICATED, connection.getStatus());
+  }
+
+  @Test
+  public void testHandleMessages_FatalException() throws Exception {
+    ProxyServer server = mock(ProxyServer.class);
+    Socket socket = mock(Socket.class);
+    InetAddress address = mock(InetAddress.class);
+    when(socket.getInetAddress()).thenReturn(address);
+    DataOutputStream dataOutputStream = new DataOutputStream(new ByteArrayOutputStream());
+    ConnectionMetadata connectionMetadata = mock(ConnectionMetadata.class);
+    when(connectionMetadata.getOutputStream()).thenReturn(dataOutputStream);
+    WireMessage message = mock(WireMessage.class);
+    when(server.recordMessage(message)).thenReturn(message);
+    doThrow(new EOFException("fatal test exception")).when(message).send();
+
+    ConnectionHandler connection =
+        new ConnectionHandler(server, socket) {
+          public ConnectionMetadata getConnectionMetadata() {
+            return connectionMetadata;
+          }
+        };
+
+    connection.setMessageState(message);
+    connection.handleMessages();
+
+    assertEquals(ConnectionStatus.TERMINATED, connection.getStatus());
+  }
 }
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 639a281a..a66bb9b1 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ControlMessageTest.java
@@ -15,6 +15,8 @@
 package com.google.cloud.spanner.pgadapter;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
@@ -25,6 +27,8 @@ import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.TextFormat;
 import com.google.cloud.spanner.pgadapter.statements.IntermediateStatement;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.ManuallyCreatedToken;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ExecuteMessage;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataInputStream;
@@ -99,4 +103,21 @@ public final class ControlMessageTest {
     assertEquals(numOfBytes, outputReader.read(bytes, 0, numOfBytes));
     assertEquals(resultMessage, new String(bytes, UTF8));
   }
+
+  @Test
+  public void testUnknownStatementTypeThrowsError() {
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    ExecuteMessage executeMessage =
+        new ExecuteMessage(connectionHandler, ManuallyCreatedToken.MANUALLY_CREATED_TOKEN);
+    IntermediateStatement intermediateStatement = mock(IntermediateStatement.class);
+    when(intermediateStatement.getCommandTag()).thenReturn("PARSE");
+    when(intermediateStatement.getStatementType()).thenReturn(StatementType.UNKNOWN);
+    when(intermediateStatement.getStatement()).thenReturn("parse foo from bar");
+
+    IllegalStateException exception =
+        assertThrows(
+            IllegalStateException.class,
+            () -> executeMessage.sendSpannerResult(intermediateStatement, QueryMode.SIMPLE, 0L));
+    assertEquals("Unknown statement type: parse foo from bar", exception.getMessage());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/CopyInMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/CopyInMockServerTest.java
index 482a65a0..7362789e 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/CopyInMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/CopyInMockServerTest.java
@@ -18,6 +18,7 @@ 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 static org.junit.Assume.assumeTrue;
 
 import com.google.cloud.spanner.MockSpannerServiceImpl.SimulatedExecutionTime;
 import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
@@ -33,8 +34,11 @@ import com.google.spanner.v1.StructType.Field;
 import com.google.spanner.v1.Type;
 import com.google.spanner.v1.TypeCode;
 import io.grpc.Status;
+import java.io.BufferedReader;
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
 import java.io.StringReader;
 import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
@@ -42,8 +46,11 @@ import java.sql.DriverManager;
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.List;
+import java.util.stream.Collectors;
 import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.Timeout;
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameter;
@@ -54,11 +61,12 @@ import org.postgresql.copy.CopyManager;
 import org.postgresql.core.BaseConnection;
 import org.postgresql.core.PGStream;
 import org.postgresql.core.QueryExecutorBase;
-import org.postgresql.core.v3.CopyOperationImpl;
 import org.postgresql.core.v3.QueryExecutorImpl;
+import org.postgresql.jdbc.PgConnection;
 
 @RunWith(Parameterized.class)
 public class CopyInMockServerTest extends AbstractMockServerTest {
+  @Rule public Timeout globalTimeout = Timeout.seconds(10);
 
   @Parameter public boolean useDomainSocket;
 
@@ -79,14 +87,20 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
    * mode for queries and DML statements.
    */
   private String createUrl() {
+    return createUrl("extended");
+  }
+
+  private String createUrl(String queryMode) {
     if (useDomainSocket) {
       return String.format(
           "jdbc:postgresql://localhost/?"
               + "socketFactory=org.newsclub.net.unix.AFUNIXSocketFactory$FactoryArg"
-              + "&socketFactoryArg=/tmp/.s.PGSQL.%d",
-          pgServer.getLocalPort());
+              + "&socketFactoryArg=/tmp/.s.PGSQL.%d"
+              + "&preferQueryMode=%s",
+          pgServer.getLocalPort(), queryMode);
     }
-    return String.format("jdbc:postgresql://localhost:%d/", pgServer.getLocalPort());
+    return String.format(
+        "jdbc:postgresql://localhost:%d/?preferQueryMode=%s", pgServer.getLocalPort(), queryMode);
   }
 
   @Test
@@ -115,6 +129,63 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
     assertEquals(3, mutation.getInsert().getValuesCount());
   }
 
+  @Test
+  public void testCopyInWithExplicitTransaction() throws SQLException, IOException {
+    setupCopyInformationSchemaResults();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+
+      CopyManager copyManager = new CopyManager(connection.unwrap(BaseConnection.class));
+      copyManager.copyIn("COPY users FROM STDIN;", new StringReader("5\t5\t5\n6\t6\t6\n7\t7\t7\n"));
+
+      // Verify that we can use the connection for normal queries.
+      try (ResultSet resultSet = connection.createStatement().executeQuery("SELECT 1")) {
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+
+      // Verify that the transaction has not yet been committed.
+      assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+
+      connection.commit();
+    }
+
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertEquals(1, commitRequests.size());
+    CommitRequest commitRequest = commitRequests.get(0);
+    assertEquals(1, commitRequest.getMutationsCount());
+
+    Mutation mutation = commitRequest.getMutations(0);
+    assertEquals(OperationCase.INSERT, mutation.getOperationCase());
+    assertEquals(3, mutation.getInsert().getValuesCount());
+  }
+
+  @Test
+  public void testCopyInWithExplicitTransaction_Rollback() throws SQLException, IOException {
+    setupCopyInformationSchemaResults();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+
+      CopyManager copyManager = new CopyManager(connection.unwrap(BaseConnection.class));
+      copyManager.copyIn("COPY users FROM STDIN;", new StringReader("5\t5\t5\n6\t6\t6\n7\t7\t7\n"));
+
+      // Verify that we can use the connection for normal queries.
+      try (ResultSet resultSet = connection.createStatement().executeQuery("SELECT 1")) {
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+
+      connection.rollback();
+    }
+
+    // Verify that the COPY operation was not committed.
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+  }
+
   @Test
   public void testCopyIn_Small() throws SQLException, IOException {
     setupCopyInformationSchemaResults();
@@ -131,7 +202,7 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
   }
 
   @Test
-  public void testCopyIn_Large_FailsWhenAtomic() throws SQLException {
+  public void testCopyIn_Large_FailsWhenAtomic() throws Exception {
     setupCopyInformationSchemaResults();
 
     try (Connection connection = DriverManager.getConnection(createUrl())) {
@@ -144,12 +215,14 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
                   copyManager.copyIn(
                       "copy all_types from stdin;",
                       new FileInputStream("./src/test/resources/all_types_data.txt")));
-      // The JDBC driver CopyManager takes the COPY protocol quite literally, and as the COPY
-      // protocol does not include any error handling, the JDBC driver will just send all data to
-      // the server and ignore any error messages the server might send during the copy operation.
-      // PGAdapter therefore drops the connection if it continues to receive CopyData messages after
-      // it sent back an error message.
-      assertTrue(exception.getMessage().contains("Database connection failed"));
+      assertTrue(
+          exception.getMessage(),
+          exception
+                  .getMessage()
+                  .contains("FAILED_PRECONDITION: Record count: 2001 has exceeded the limit: 2000.")
+              || exception
+                  .getMessage()
+                  .contains("Database connection failed when canceling copy operation"));
     }
   }
 
@@ -187,6 +260,7 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
                   copyManager.copyIn(
                       "COPY users FROM STDIN;", new StringReader("5\t5\t5\n6\t6\t6\n7\t7\t7\n")));
       assertTrue(
+          exception.getMessage(),
           exception.getMessage().contains("io.grpc.StatusRuntimeException: INVALID_ARGUMENT"));
     }
 
@@ -222,7 +296,7 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
   }
 
   @Test
-  public void testCopyInWithInvalidRow() throws SQLException {
+  public void testCopyInWithInvalidRow() throws SQLException, IOException {
     setupCopyInformationSchemaResults();
 
     try (Connection connection = DriverManager.getConnection(createUrl())) {
@@ -255,7 +329,9 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
               () ->
                   copyManager.copyIn(
                       "COPY users FROM STDIN;", new StringReader("5\t5\t5\n6\t6\t6\n7\t7\t7\n")));
-      assertTrue(exception.getMessage().contains("Commit size: 20 has exceeded the limit: 10"));
+      assertTrue(
+          exception.getMessage(),
+          exception.getMessage().contains("Commit size: 20 has exceeded the limit: 10"));
     } finally {
       System.getProperties().remove("copy_in_commit_limit");
     }
@@ -351,20 +427,27 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
 
     byte[] payload = "5\t5\t5\n".getBytes(StandardCharsets.UTF_8);
     try (Connection connection = DriverManager.getConnection(createUrl())) {
-      CopyManager copyManager = new CopyManager(connection.unwrap(BaseConnection.class));
-      // Start a copy operation and then try to execute a query during the copy.
-      CopyIn copyOperation = copyManager.copyIn("COPY users FROM STDIN;");
-      copyOperation.writeToCopy(payload, 0, payload.length);
-      copyOperation.flushCopy();
-
-      // Use reflection to get hold of the underlying stream, so we can send an invalid message.
-      java.lang.reflect.Field queryExecutorField =
-          CopyOperationImpl.class.getDeclaredField("queryExecutor");
-      queryExecutorField.setAccessible(true);
-      QueryExecutorImpl queryExecutor = (QueryExecutorImpl) queryExecutorField.get(copyOperation);
+      PgConnection pgConnection = connection.unwrap(PgConnection.class);
+      QueryExecutorImpl queryExecutor = (QueryExecutorImpl) pgConnection.getQueryExecutor();
+      // Use reflection to get hold of the underlying stream, so we can send any message that we
+      // want.
       java.lang.reflect.Field pgStreamField = QueryExecutorBase.class.getDeclaredField("pgStream");
       pgStreamField.setAccessible(true);
       PGStream stream = (PGStream) pgStreamField.get(queryExecutor);
+
+      String sql = "copy users from stdin;";
+      int length = sql.length() + 5;
+      stream.sendChar('Q');
+      stream.sendInteger4(length);
+      stream.send(sql.getBytes(StandardCharsets.UTF_8));
+      stream.sendChar(0);
+      stream.flush();
+
+      // Send a CopyData message and then a 'Q', which is not allowed.
+      stream.sendChar('d');
+      stream.sendInteger4(payload.length + 4);
+      stream.send(payload);
+
       stream.sendChar('Q');
       // Length = 4 + 8 + 1 = 13
       // (msg length 4 bytes, 8 bytes for SELECT 1, 1 byte for \0)
@@ -373,18 +456,261 @@ public class CopyInMockServerTest extends AbstractMockServerTest {
       stream.sendChar(0);
       stream.flush();
 
-      // PGAdapter drops the connection if an invalid message is received during a COPY. This is a
-      // safety measure as there is no other error handling in the COPY protocol, and the server
-      // could otherwise have been completely flushed with garbage if it continued to receive
-      // messages after receiving an invalid message.
-      SQLException exception = assertThrows(SQLException.class, copyOperation::endCopy);
-      assertEquals("Database connection failed when ending copy", exception.getMessage());
+      boolean receivedErrorMessage = false;
+      StringBuilder errorMessage = new StringBuilder();
+      while (!receivedErrorMessage) {
+        int command = stream.receiveChar();
+        if (command == 'E') {
+          receivedErrorMessage = true;
+          // Read and ignore the length.
+          stream.receiveInteger4();
+          while (stream.receiveChar() > 0) {
+            errorMessage.append(stream.receiveString()).append('\n');
+          }
+        } else {
+          // Just skip everything in the message.
+          length = stream.receiveInteger4();
+          stream.skip(length - 4);
+        }
+      }
+      assertEquals(
+          "ERROR\n"
+              + "XX000\n"
+              + "Expected CopyData ('d'), CopyDone ('c') or CopyFail ('f') messages, got: 'Q'\n",
+          errorMessage.toString());
+
+      stream.sendChar('c');
+      stream.sendInteger4(4);
+      stream.sendChar('x');
+      stream.sendInteger4(4);
+      stream.flush();
     }
 
     List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
     assertTrue(commitRequests.isEmpty());
   }
 
+  private static boolean isPsqlAvailable() {
+    ProcessBuilder builder = new ProcessBuilder();
+    String[] psqlCommand = new String[] {"psql", "--version"};
+    builder.command(psqlCommand);
+    try {
+      Process process = builder.start();
+      int res = process.waitFor();
+
+      return res == 0;
+    } catch (Exception ignored) {
+      return false;
+    }
+  }
+
+  @Test
+  public void testCopyInBatchPsql() throws Exception {
+    assumeTrue("This test requires psql to be installed", isPsqlAvailable());
+    setupCopyInformationSchemaResults();
+
+    String host = useDomainSocket ? "/tmp" : "localhost";
+    ProcessBuilder builder = new ProcessBuilder();
+    String[] psqlCommand =
+        new String[] {"psql", "-h", host, "-p", String.valueOf(pgServer.getLocalPort())};
+    builder.command(psqlCommand);
+    Process process = builder.start();
+    String errors;
+    String output;
+
+    try (OutputStreamWriter writer = new OutputStreamWriter(process.getOutputStream());
+        BufferedReader reader =
+            new BufferedReader(new InputStreamReader(process.getInputStream()));
+        BufferedReader errorReader =
+            new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+      writer.write(
+          "SELECT 1\\;copy users from stdin\\;copy users from stdin\\;SELECT 2;\n"
+              + "1\t2\t3\n"
+              + "\\.\n"
+              + "4\t5\t6\n"
+              + "\\.\n"
+              + "\n"
+              + "\\q\n");
+      writer.flush();
+      errors = errorReader.lines().collect(Collectors.joining("\n"));
+      output = reader.lines().collect(Collectors.joining("\n"));
+    }
+
+    assertEquals("", errors);
+    assertEquals(" C \n---\n 2\n(1 row)\n", output);
+    int res = process.waitFor();
+    assertEquals(0, res);
+
+    // The batch is executed as one implicit transaction.
+    List<CommitRequest> commitRequests = mockSpanner.getRequestsOfType(CommitRequest.class);
+    assertEquals(1, commitRequests.size());
+    CommitRequest commitRequest = commitRequests.get(0);
+    assertEquals(1, commitRequest.getMutationsCount());
+    // We buffer 2 mutations, but the proto builder in the Java client library combines these two
+    // mutations into 1 mutation with 2 rows.
+    assertEquals(2, commitRequest.getMutations(0).getInsert().getValuesCount());
+    assertEquals(OperationCase.INSERT, commitRequest.getMutations(0).getOperationCase());
+    assertEquals(3, commitRequest.getMutations(0).getInsert().getColumnsCount());
+    assertEquals(
+        "1", commitRequest.getMutations(0).getInsert().getValues(0).getValues(0).getStringValue());
+    assertEquals(
+        "2", commitRequest.getMutations(0).getInsert().getValues(0).getValues(1).getStringValue());
+    assertEquals(
+        "3", commitRequest.getMutations(0).getInsert().getValues(0).getValues(2).getStringValue());
+    assertEquals(OperationCase.INSERT, commitRequest.getMutations(0).getOperationCase());
+    assertEquals(
+        "4", commitRequest.getMutations(0).getInsert().getValues(1).getValues(0).getStringValue());
+    assertEquals(
+        "5", commitRequest.getMutations(0).getInsert().getValues(1).getValues(1).getStringValue());
+    assertEquals(
+        "6", commitRequest.getMutations(0).getInsert().getValues(1).getValues(2).getStringValue());
+  }
+
+  @Test
+  public void testCopyInBatch() throws Exception {
+    setupCopyInformationSchemaResults();
+
+    byte[] row1 = "5\t6\t7\n".getBytes(StandardCharsets.UTF_8);
+    byte[] row2 = "6\t7\t8\n".getBytes(StandardCharsets.UTF_8);
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      PgConnection pgConnection = connection.unwrap(PgConnection.class);
+      QueryExecutorImpl queryExecutor = (QueryExecutorImpl) pgConnection.getQueryExecutor();
+      // Use reflection to get hold of the underlying stream, so we can send any message that we
+      // want.
+      java.lang.reflect.Field pgStreamField = QueryExecutorBase.class.getDeclaredField("pgStream");
+      pgStreamField.setAccessible(true);
+      PGStream stream = (PGStream) pgStreamField.get(queryExecutor);
+
+      String sql = "SELECT 1;copy users from stdin;copy users from stdin;SELECT 2;";
+      int length = sql.length() + 5;
+      stream.sendChar('Q');
+      stream.sendInteger4(length);
+      stream.send(sql.getBytes(StandardCharsets.UTF_8));
+      stream.sendChar(0);
+      stream.flush();
+
+      // Send CopyData + CopyDone for the first copy operation.
+      stream.sendChar('d');
+      stream.sendInteger4(row1.length + 4);
+      stream.send(row1);
+      stream.sendChar('c');
+      stream.sendInteger4(4);
+      stream.flush();
+
+      // Send CopyData + CopyDone for the second copy operation.
+      stream.sendChar('d');
+      stream.sendInteger4(row2.length + 4);
+      stream.send(row2);
+      stream.sendChar('c');
+      stream.sendInteger4(4);
+      stream.flush();
+
+      boolean receivedReadyForQuery = false;
+      while (!receivedReadyForQuery) {
+        int command = stream.receiveChar();
+        if (command == 'Z') {
+          receivedReadyForQuery = true;
+        } else {
+          // Just skip everything in the message.
+          length = stream.receiveInteger4();
+          stream.skip(length - 4);
+        }
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(CommitRequest.class));
+    CommitRequest commitRequest = mockSpanner.getRequestsOfType(CommitRequest.class).get(0);
+    assertEquals(1, commitRequest.getMutationsCount());
+    assertEquals(2, commitRequest.getMutations(0).getInsert().getValuesCount());
+    assertEquals(
+        "5", commitRequest.getMutations(0).getInsert().getValues(0).getValues(0).getStringValue());
+    assertEquals(
+        "6", commitRequest.getMutations(0).getInsert().getValues(0).getValues(1).getStringValue());
+    assertEquals(
+        "7", commitRequest.getMutations(0).getInsert().getValues(0).getValues(2).getStringValue());
+    assertEquals(
+        "6", commitRequest.getMutations(0).getInsert().getValues(1).getValues(0).getStringValue());
+    assertEquals(
+        "7", commitRequest.getMutations(0).getInsert().getValues(1).getValues(1).getStringValue());
+    assertEquals(
+        "8", commitRequest.getMutations(0).getInsert().getValues(1).getValues(2).getStringValue());
+  }
+
+  @Test
+  public void testCopyInBatchWithCopyFail() throws Exception {
+    setupCopyInformationSchemaResults();
+
+    byte[] row1 = "5\t6\t7\n".getBytes(StandardCharsets.UTF_8);
+    byte[] row2 = "8\t9\t10\n".getBytes(StandardCharsets.UTF_8);
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      PgConnection pgConnection = connection.unwrap(PgConnection.class);
+      QueryExecutorImpl queryExecutor = (QueryExecutorImpl) pgConnection.getQueryExecutor();
+      // Use reflection to get hold of the underlying stream, so we can send any message that we
+      // want.
+      java.lang.reflect.Field pgStreamField = QueryExecutorBase.class.getDeclaredField("pgStream");
+      pgStreamField.setAccessible(true);
+      PGStream stream = (PGStream) pgStreamField.get(queryExecutor);
+
+      String sql = "SELECT 1;copy users from stdin;copy users from stdin;SELECT 2;";
+      int length = sql.getBytes(StandardCharsets.UTF_8).length + 5;
+      stream.sendChar('Q');
+      stream.sendInteger4(length);
+      stream.send(sql.getBytes(StandardCharsets.UTF_8));
+      stream.sendChar(0);
+      stream.flush();
+
+      // Send CopyData + CopyDone for the first copy operation.
+      stream.sendChar('d');
+      stream.sendInteger4(row1.length + 4);
+      stream.send(row1);
+      stream.sendChar('c');
+      stream.sendInteger4(4);
+      stream.flush();
+
+      // Send CopyData + CopyFail for the second copy operation.
+      stream.sendChar('d');
+      stream.sendInteger4(row2.length + 4);
+      stream.send(row2);
+      String error = "Changed my mind";
+      stream.sendChar('f');
+      stream.sendInteger4(4 + error.getBytes(StandardCharsets.UTF_8).length + 1);
+      stream.send(error.getBytes(StandardCharsets.UTF_8));
+      stream.sendChar(0);
+      stream.flush();
+
+      boolean receivedReadyForQuery = false;
+      boolean receivedErrorMessage = false;
+      StringBuilder errorMessage = new StringBuilder();
+      while (!receivedReadyForQuery) {
+        int command = stream.receiveChar();
+        if (command == 'Z') {
+          receivedReadyForQuery = true;
+        } else if (command == 'E') {
+          receivedErrorMessage = true;
+          // Read and ignore the length.
+          stream.receiveInteger4();
+          while (stream.receiveChar() > 0) {
+            errorMessage.append(stream.receiveString()).append('\n');
+          }
+        } else {
+          // Just skip everything in the message.
+          length = stream.receiveInteger4();
+          stream.skip(length - 4);
+        }
+      }
+      assertTrue(receivedErrorMessage);
+      assertEquals(
+          "ERROR\n"
+              + "P0001\n"
+              + "CANCELLED: Changed my mind\n"
+              + "ERROR\n"
+              + "P0001\n"
+              + "INVALID_ARGUMENT: current transaction is aborted, commands ignored until end of transaction block\n",
+          errorMessage.toString());
+    }
+    assertEquals(0, mockSpanner.countRequestsOfType(CommitRequest.class));
+  }
+
   private void setupCopyInformationSchemaResults() {
     setupCopyInformationSchemaResults(true);
   }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
index 328b517e..a16a528c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
@@ -22,7 +22,6 @@ import static org.junit.Assert.assertTrue;
 import com.google.cloud.spanner.MockSpannerServiceImpl.SimulatedExecutionTime;
 import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
 import com.google.cloud.spanner.Statement;
-import com.google.cloud.spanner.connection.SpannerPool;
 import com.google.common.collect.ImmutableList;
 import com.google.rpc.ResourceInfo;
 import com.google.spanner.admin.database.v1.Database;
@@ -48,6 +47,7 @@ import java.sql.SQLException;
 import java.util.List;
 import org.junit.After;
 import org.junit.BeforeClass;
+import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
@@ -55,6 +55,8 @@ import org.postgresql.PGProperty;
 
 @RunWith(JUnit4.class)
 public class EmulatedPsqlMockServerTest extends AbstractMockServerTest {
+  // @Rule public Timeout globalTimeout = Timeout.seconds(10);
+
   private static final String INSERT1 = "insert into foo values (1)";
   private static final String INSERT2 = "insert into foo values (2)";
 
@@ -130,10 +132,12 @@ public class EmulatedPsqlMockServerTest extends AbstractMockServerTest {
     assertEquals(DatabaseName.parse(databaseName), gotDatabaseName);
   }
 
+  @Ignore(
+      "Skipped because of a bug in the gRPC server implementation that causes random NullPointerExceptions")
   @Test
   public void testConnectToNonExistingDatabase() {
     try {
-      mockSpanner.setExecuteStreamingSqlExecutionTime(
+      mockSpanner.setBatchCreateSessionsExecutionTime(
           SimulatedExecutionTime.stickyDatabaseNotFoundException("non-existing-db"));
       // The Connection API calls listInstanceConfigs(..) once first when the connection is a
       // localhost connection. It does so to verify that the connection is valid and to quickly
@@ -177,7 +181,7 @@ public class EmulatedPsqlMockServerTest extends AbstractMockServerTest {
           exception.getMessage(),
           exception.getMessage().contains("\tprojects/p/instances/i/databases/google-sql-db\n"));
     } finally {
-      SpannerPool.closeSpannerPool();
+      closeSpannerPool(true);
     }
   }
 
@@ -215,7 +219,7 @@ public class EmulatedPsqlMockServerTest extends AbstractMockServerTest {
       assertTrue(
           exception.getMessage(), exception.getMessage().contains("\tprojects/p/instances/i\n"));
     } finally {
-      SpannerPool.closeSpannerPool();
+      closeSpannerPool(true);
     }
   }
 
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 9c3b6ce3..6e6f1b92 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
@@ -38,6 +38,7 @@ import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.sql.Types;
 import java.time.LocalDate;
 import java.util.ArrayList;
@@ -134,6 +135,14 @@ public class ITJdbcTest implements IntegrationTest {
   }
 
   private String getConnectionUrl() {
+    if (useDomainSocket) {
+      return String.format(
+          "jdbc:postgresql://localhost/%s?"
+              + "preferQueryMode=%s"
+              + "&socketFactory=org.newsclub.net.unix.AFUNIXSocketFactory$FactoryArg"
+              + "&socketFactoryArg=/tmp/.s.PGSQL.%d",
+          database.getId().getDatabase(), preferQueryMode, testEnv.getPGAdapterPort());
+    }
     return String.format(
         "jdbc:postgresql://%s/%s?preferQueryMode=%s",
         testEnv.getPGAdapterHostAndPort(), database.getId().getDatabase(), preferQueryMode);
@@ -151,6 +160,31 @@ public class ITJdbcTest implements IntegrationTest {
     }
   }
 
+  @Test
+  public void testCreateTableIfNotExists() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        // This table already exists, so it should be a no-op.
+        assertFalse(
+            statement.execute("create table if not exists all_types (id bigint primary key)"));
+        assertFalse(statement.getMoreResults());
+      }
+    }
+  }
+
+  @Test
+  public void testCreateTableIfNotExists_withSchemaPrefix() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        // This table already exists, so it should be a no-op.
+        assertFalse(
+            statement.execute(
+                "create table if not exists public.all_types (id bigint primary key)"));
+        assertFalse(statement.getMoreResults());
+      }
+    }
+  }
+
   @Test
   public void testSelectWithParameters() throws SQLException {
     boolean isSimpleMode = "simple".equalsIgnoreCase(preferQueryMode);
@@ -476,12 +510,14 @@ public class ITJdbcTest implements IntegrationTest {
                   copyManager.copyIn(
                       "copy all_types from stdin;",
                       new FileInputStream("./src/test/resources/all_types_data.txt")));
-      // The JDBC driver CopyManager takes the COPY protocol quite literally, and as the COPY
-      // protocol does not include any error handling, the JDBC driver will just send all data to
-      // the server and ignore any error messages the server might send during the copy operation.
-      // PGAdapter therefore drops the connection if it continues to receive CopyData messages after
-      // it sent back an error message.
-      assertTrue(exception.getMessage().contains("Database connection failed"));
+      assertTrue(
+          exception.getMessage(),
+          exception
+                  .getMessage()
+                  .contains("FAILED_PRECONDITION: Record count: 2001 has exceeded the limit: 2000.")
+              || exception
+                  .getMessage()
+                  .contains("Database connection failed when canceling copy operation"));
     }
 
     // Verify that the table is still empty.
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/InvalidMessagesTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/InvalidMessagesTest.java
index 636b68d8..3b16ada0 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/InvalidMessagesTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/InvalidMessagesTest.java
@@ -115,11 +115,14 @@ public class InvalidMessagesTest extends AbstractMockServerTest {
         outputStream.flush();
 
         // Read until the end of the stream. The stream should be closed by the backend.
-        int bytesRead = 0;
-        while (inputStream.read() > -1 && bytesRead < 1 << 16) {
-          bytesRead++;
+        try {
+          int bytesRead = 0;
+          while (inputStream.read() > -1 && bytesRead < 1 << 16) {
+            bytesRead++;
+          }
+          assertEquals(-1, inputStream.read());
+        } catch (IOException ignore) {
         }
-        assertEquals(-1, inputStream.read());
       }
     }
   }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcBatchMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcBatchMockServerTest.java
index d8cf1b45..a3b583a4 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcBatchMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcBatchMockServerTest.java
@@ -57,7 +57,7 @@ public class JdbcBatchMockServerTest extends AbstractMockServerTest {
 
   @Parameter public String preferQueryMode;
 
-  @Parameters(name = "preferQueryMode = {0}, useDomainSocket = {1}")
+  @Parameters(name = "preferQueryMode = {0}")
   public static List<Object[]> data() {
     String[] queryModes = {"extended", "simple"};
     List<Object[]> parameters = new ArrayList<>();
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 9783160f..3e84c0c8 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
@@ -47,6 +47,7 @@ import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
 import java.sql.Connection;
 import java.sql.DriverManager;
+import java.sql.ParameterMetaData;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
@@ -65,7 +66,6 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 import org.postgresql.PGStatement;
-import org.postgresql.core.Oid;
 import org.postgresql.jdbc.PgStatement;
 
 @RunWith(JUnit4.class)
@@ -92,12 +92,7 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
    * mode for queries and DML statements.
    */
   private String createUrl() {
-    // Force binary transfer of date parameters to prevent PostgreSQL from sending them with type
-    // Oid.UNSPECIFIED. This setting is however at this moment ignored and depends on this PR:
-    // https://github.com/pgjdbc/pgjdbc/pull/2476
-    return String.format(
-        "jdbc:postgresql://localhost:%d/?binaryTransferEnable=%d",
-        pgServer.getLocalPort(), Oid.DATE);
+    return String.format("jdbc:postgresql://localhost:%d/", pgServer.getLocalPort());
   }
 
   @Test
@@ -483,6 +478,55 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     assertEquals(QueryMode.PLAN, requests.get(1).getQueryMode());
   }
 
+  @Test
+  public void testDescribeDmlWithSchemaPrefix() throws SQLException {
+    String sql = "update public.my_table set value=? where id=?";
+    String describeSql = "select $1, $2 from (select value=$1 from public.my_table where id=$2) p";
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(describeSql),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(createMetadata(ImmutableList.of(TypeCode.STRING, TypeCode.INT64)))
+                .build()));
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+        ParameterMetaData metadata = preparedStatement.getParameterMetaData();
+        assertEquals(Types.VARCHAR, metadata.getParameterType(1));
+        assertEquals(Types.BIGINT, metadata.getParameterType(2));
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(describeSql, requests.get(0).getSql());
+    assertEquals(QueryMode.PLAN, requests.get(0).getQueryMode());
+  }
+
+  @Test
+  public void testDescribeDmlWithQuotedSchemaPrefix() throws SQLException {
+    String sql = "update \"public\".\"my_table\" set value=? where id=?";
+    String describeSql =
+        "select $1, $2 from (select value=$1 from \"public\".\"my_table\" where id=$2) p";
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(describeSql),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(createMetadata(ImmutableList.of(TypeCode.STRING, TypeCode.INT64)))
+                .build()));
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (PreparedStatement preparedStatement = connection.prepareStatement(sql)) {
+        ParameterMetaData metadata = preparedStatement.getParameterMetaData();
+        assertEquals(Types.VARCHAR, metadata.getParameterType(1));
+        assertEquals(Types.BIGINT, metadata.getParameterType(2));
+      }
+    }
+
+    List<ExecuteSqlRequest> requests = mockSpanner.getRequestsOfType(ExecuteSqlRequest.class);
+    assertEquals(1, requests.size());
+    assertEquals(describeSql, requests.get(0).getSql());
+    assertEquals(QueryMode.PLAN, requests.get(0).getQueryMode());
+  }
+
   @Test
   public void testTwoDmlStatements() throws SQLException {
     try (Connection connection = DriverManager.getConnection(createUrl())) {
@@ -633,6 +677,369 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     }
   }
 
+  @Test
+  public void testCreateTableIfNotExists_tableDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "create table foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testCreateTableIfNotExists_tableExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    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 testCreateIndexIfNotExists_indexDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE INDEX IF NOT EXISTS foo on bar (value)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "create index foo on bar (value)", updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testCreateIndexIfNotExists_indexExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE INDEX IF NOT EXISTS foo on bar (value)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    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 testDropTableIfExists_tableDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP TABLE IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    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 testDropTableIfExists_tableExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP TABLE IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals("drop table foo", updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testDropIndexIfExists_indexDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP INDEX IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    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 testDropIndexIfExists_indexExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP INDEX IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals("drop index foo", updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testDdlBatchWithIfNotExists() throws SQLException {
+    addIfNotExistsDdlException();
+    ImmutableList<String> statements =
+        ImmutableList.of(
+            "CREATE TABLE IF NOT EXISTS \"Foo\" (id bigint primary key)",
+            "CREATE TABLE IF NOT EXISTS bar (id bigint primary key, value text)",
+            "CREATE INDEX IF NOT EXISTS idx_foo ON bar (text)");
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("Foo")
+                .build(),
+            SELECT1_RESULTSET));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("bar")
+                .build(),
+            SELECT1_RESULTSET));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("idx_foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    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(0, updateDatabaseDdlRequests.size());
+  }
+
+  @Test
+  public void testCreateTableIfNotExists_withBackendSupport() throws SQLException {
+    // Add a generic error that is returned for DDL statements. This will cause PGAdapter to think
+    // that the backend supports `IF [NOT] EXISTS`, as it does not receive a specific error
+    // regarding an `IF NOT EXISTS` statement.
+    addDdlExceptionToSpannerAdmin();
+    String sql = "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)";
+    // Add a response for the DDL statement that is sent to Spanner.
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
   @Test
   public void testPreparedStatement() throws SQLException {
     mockSpanner.putStatementResult(
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java b/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java
index 034c4937..d065ff8e 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/PgAdapterTestEnv.java
@@ -44,7 +44,7 @@ import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Random;
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 import java.util.logging.Level;
@@ -267,10 +267,7 @@ public class PgAdapterTestEnv {
     if (databaseId == null) {
       databaseId = System.getProperty(TEST_DATABASE_PROPERTY, DEFAULT_DATABASE_ID);
     }
-    String id =
-        String.format(
-            "%s_%d_%d",
-            databaseId, System.currentTimeMillis(), new Random().nextInt(Short.MAX_VALUE));
+    String id = String.format("%s_%s", databaseId, UUID.randomUUID().toString().replace('-', '_'));
     // Make sure the database id is not longer than the max allowed 30 characters.
     if (id.length() > 30) {
       id = id.substring(0, 30);
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/python/ITPsycopg2Test.java b/src/test/java/com/google/cloud/spanner/pgadapter/python/ITPsycopg2Test.java
new file mode 100644
index 00000000..513d1ee7
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/python/ITPsycopg2Test.java
@@ -0,0 +1,93 @@
+// 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.python;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.cloud.spanner.Database;
+import com.google.cloud.spanner.pgadapter.IntegrationTest;
+import com.google.cloud.spanner.pgadapter.PgAdapterTestEnv;
+import java.io.IOException;
+import java.util.Collections;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+@Category(IntegrationTest.class)
+public class ITPsycopg2Test extends PythonTestSetup {
+
+  private static final PgAdapterTestEnv testEnv = new PgAdapterTestEnv();
+  private static Database database;
+
+  private static Iterable<String> getDdlStatements() {
+    return Collections.singletonList(
+        "create table all_types ("
+            + "col_bigint bigint not null primary key, "
+            + "col_bool bool, "
+            + "col_bytea bytea, "
+            + "col_float8 float8, "
+            + "col_int int, "
+            + "col_numeric numeric, "
+            + "col_timestamptz timestamptz, "
+            + "col_date date, "
+            + "col_varchar varchar(100))");
+  }
+
+  @BeforeClass
+  public static void setup() throws Exception {
+    testEnv.setUp();
+    database = testEnv.createDatabase(getDdlStatements());
+    testEnv.startPGAdapterServerWithDefaultDatabase(database.getId(), Collections.emptyList());
+  }
+
+  @AfterClass
+  public static void teardown() {
+    testEnv.stopPGAdapterServer();
+    testEnv.cleanUp();
+  }
+
+  @Test
+  public void testBasicInsert() throws IOException, InterruptedException {
+    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("
+            + "2, "
+            + "TRUE, "
+            + "'some string'::bytea, "
+            + "9.7, "
+            + "10, "
+            + "300.321, "
+            + "'2018-03-11 02:00:00'::timestamptz, "
+            + "'2022-10-02', "
+            + "'some string')";
+
+    String actualOutput = executeWithoutParameters(testEnv.getPGAdapterPort(), sql, "update");
+    String expectedOutput = "1\n";
+    assertEquals(expectedOutput, actualOutput);
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
index d4e7fe94..f1636edd 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
@@ -14,33 +14,255 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
+import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.extractDdlUpdateCounts;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.cloud.spanner.DatabaseId;
+import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.ResultSet;
 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.connection.StatementResult;
 import com.google.cloud.spanner.connection.StatementResult.ResultType;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.NoResult;
 import com.google.cloud.spanner.pgadapter.statements.BackendConnection.QueryResult;
+import com.google.cloud.spanner.pgadapter.statements.DdlExecutor.NotExecuted;
 import com.google.cloud.spanner.pgadapter.statements.local.ListDatabasesStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.LocalStatement;
+import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
+import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.common.collect.ImmutableList;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
 
 @RunWith(JUnit4.class)
 public class BackendConnectionTest {
+  private final AbstractStatementParser PARSER =
+      AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
+  private static final NotExecuted NOT_EXECUTED = new NotExecuted();
+  private static final NoResult NO_RESULT = new NoResult();
+
+  @Test
+  public void testExtractDdlUpdateCounts() {
+    assertArrayEquals(new long[] {}, extractDdlUpdateCounts(ImmutableList.of(), new long[] {}));
+    assertArrayEquals(
+        new long[] {1L}, extractDdlUpdateCounts(ImmutableList.of(NO_RESULT), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(ImmutableList.of(NO_RESULT, NOT_EXECUTED), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(ImmutableList.of(NOT_EXECUTED, NO_RESULT), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NO_RESULT, NO_RESULT), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NO_RESULT, NOT_EXECUTED), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NOT_EXECUTED, NO_RESULT, NOT_EXECUTED),
+            new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NOT_EXECUTED, NO_RESULT, NOT_EXECUTED, NO_RESULT),
+            new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NO_RESULT, NO_RESULT, NOT_EXECUTED), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L, 1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(
+                NOT_EXECUTED,
+                NO_RESULT,
+                NO_RESULT,
+                NOT_EXECUTED,
+                NOT_EXECUTED,
+                NO_RESULT,
+                NO_RESULT),
+            new long[] {1L, 1L, 1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NOT_EXECUTED, NOT_EXECUTED), new long[] {}));
+  }
+
+  @Test
+  public void testCopyPropagatesNonSpannerException() {
+    String sql = "COPY users FROM STDIN";
+    Statement statement = Statement.of(sql);
+    ParsedStatement parsedStatement = PARSER.parse(statement);
+    Connection spannerConnection = mock(Connection.class);
+    CopyDataReceiver receiver = mock(CopyDataReceiver.class);
+    MutationWriter writer = mock(MutationWriter.class);
+    ExecutorService executor = mock(ExecutorService.class);
+    RejectedExecutionException rejectedExecutionException = new RejectedExecutionException();
+    doThrow(rejectedExecutionException).when(executor).execute(any(Runnable.class));
+
+    BackendConnection backendConnection =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    Future<StatementResult> result =
+        backendConnection.executeCopy(parsedStatement, statement, receiver, writer, executor);
+    backendConnection.flush();
+
+    assertTrue(result.isDone());
+    ExecutionException executionException = assertThrows(ExecutionException.class, result::get);
+    assertSame(rejectedExecutionException, executionException.getCause());
+  }
+
+  @Test
+  public void testHasDmlOrCopyStatementsAfter() {
+    CopyDataReceiver receiver = mock(CopyDataReceiver.class);
+    MutationWriter writer = mock(MutationWriter.class);
+    ExecutorService executor = mock(ExecutorService.class);
+    Connection spannerConnection = mock(Connection.class);
+    String updateSql = "insert into foo values (1)";
+    Statement updateStatement = Statement.of(updateSql);
+    ParsedStatement parsedUpdateStatement = PARSER.parse(updateStatement);
+    String selectSql = "select * from foo";
+    Statement selectStatement = Statement.of(selectSql);
+    ParsedStatement parsedSelectStatement = PARSER.parse(selectStatement);
+    String copySql = "copy foo from stdin";
+    Statement copyStatement = Statement.of(copySql);
+    ParsedStatement parsedCopyStatement = PARSER.parse(copyStatement);
+    String clientSideSql = "set spanner.statement_tag='foo'";
+    Statement clientSideStatement = Statement.of(clientSideSql);
+    ParsedStatement parsedClientSideStatement = PARSER.parse(clientSideStatement);
+    String unknownSql = "merge foo";
+    Statement unknownStatement = Statement.of(unknownSql);
+    ParsedStatement parsedUnknownStatement = PARSER.parse(unknownStatement);
+
+    BackendConnection onlyDmlStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    onlyDmlStatements.execute(parsedUpdateStatement, updateStatement);
+    onlyDmlStatements.execute(parsedUpdateStatement, updateStatement);
+    assertTrue(onlyDmlStatements.hasDmlOrCopyStatementsAfter(0));
+    assertTrue(onlyDmlStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection onlyCopyStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    onlyCopyStatements.executeCopy(parsedCopyStatement, copyStatement, receiver, writer, executor);
+    onlyCopyStatements.executeCopy(parsedCopyStatement, copyStatement, receiver, writer, executor);
+    assertTrue(onlyCopyStatements.hasDmlOrCopyStatementsAfter(0));
+    assertTrue(onlyCopyStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection dmlAndCopyStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    dmlAndCopyStatements.execute(parsedUpdateStatement, updateStatement);
+    dmlAndCopyStatements.executeCopy(
+        parsedCopyStatement, copyStatement, receiver, writer, executor);
+    assertTrue(dmlAndCopyStatements.hasDmlOrCopyStatementsAfter(0));
+    assertTrue(dmlAndCopyStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection onlySelectStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    onlySelectStatements.execute(parsedSelectStatement, selectStatement);
+    onlySelectStatements.execute(parsedSelectStatement, selectStatement);
+    assertFalse(onlySelectStatements.hasDmlOrCopyStatementsAfter(0));
+    assertFalse(onlySelectStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection onlyClientSideStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    onlyClientSideStatements.execute(parsedClientSideStatement, clientSideStatement);
+    onlyClientSideStatements.execute(parsedClientSideStatement, clientSideStatement);
+    assertFalse(onlyClientSideStatements.hasDmlOrCopyStatementsAfter(0));
+    assertFalse(onlyClientSideStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection onlyUnknownStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    onlyUnknownStatements.execute(parsedUnknownStatement, unknownStatement);
+    onlyUnknownStatements.execute(parsedUnknownStatement, unknownStatement);
+    assertFalse(onlyUnknownStatements.hasDmlOrCopyStatementsAfter(0));
+    assertFalse(onlyUnknownStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection dmlAndSelectStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    dmlAndSelectStatements.execute(parsedUpdateStatement, updateStatement);
+    dmlAndSelectStatements.execute(parsedSelectStatement, selectStatement);
+    assertTrue(dmlAndSelectStatements.hasDmlOrCopyStatementsAfter(0));
+    assertFalse(dmlAndSelectStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection copyAndSelectStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    copyAndSelectStatements.executeCopy(
+        parsedCopyStatement, copyStatement, receiver, writer, executor);
+    copyAndSelectStatements.execute(parsedSelectStatement, selectStatement);
+    assertTrue(copyAndSelectStatements.hasDmlOrCopyStatementsAfter(0));
+    assertFalse(copyAndSelectStatements.hasDmlOrCopyStatementsAfter(1));
+
+    BackendConnection copyAndUnknownStatements =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
+    copyAndUnknownStatements.executeCopy(
+        parsedCopyStatement, copyStatement, receiver, writer, executor);
+    copyAndUnknownStatements.execute(parsedUnknownStatement, unknownStatement);
+    assertTrue(copyAndUnknownStatements.hasDmlOrCopyStatementsAfter(0));
+    assertFalse(copyAndUnknownStatements.hasDmlOrCopyStatementsAfter(1));
+  }
 
   @Test
   public void testExecuteLocalStatement() throws ExecutionException, InterruptedException {
@@ -56,7 +278,8 @@ public class BackendConnectionTest {
         .thenReturn(ListDatabasesStatement.LIST_DATABASES_SQL);
 
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, localStatements);
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, localStatements);
     Future<StatementResult> resultFuture =
         backendConnection.execute(
             parsedListDatabasesStatement, Statement.of(ListDatabasesStatement.LIST_DATABASES_SQL));
@@ -84,7 +307,8 @@ public class BackendConnectionTest {
     when(connection.execute(statement)).thenReturn(statementResult);
 
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, localStatements);
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, localStatements);
     Future<StatementResult> resultFuture = backendConnection.execute(parsedStatement, statement);
     backendConnection.flush();
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java
new file mode 100644
index 00000000..72840d1a
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java
@@ -0,0 +1,392 @@
+// 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 com.google.cloud.spanner.pgadapter.statements.DdlExecutor.unquoteIdentifier;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class DdlExecutorTest {
+  private static final AbstractStatementParser PARSER =
+      AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
+
+  @Test
+  public void testUnquoteIdentifier() {
+    assertEquals("foo", unquoteIdentifier("foo"));
+    assertEquals("foo", unquoteIdentifier("FOO"));
+    assertEquals("foo", unquoteIdentifier("Foo"));
+
+    assertEquals("foo", unquoteIdentifier("\"foo\""));
+    assertEquals("FOO", unquoteIdentifier("\"FOO\""));
+    assertEquals("Foo", unquoteIdentifier("\"Foo\""));
+
+    assertEquals("", unquoteIdentifier(""));
+    assertEquals("a", unquoteIdentifier("a"));
+    assertEquals("", unquoteIdentifier("\"\""));
+  }
+
+  private Statement translate(String sql, DdlExecutor executor) {
+    Statement statement = Statement.of(sql);
+    return executor.translate(PARSER.parse(statement), statement);
+  }
+
+  @Test
+  public void testTranslateCreateTableNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create table foo (id int)"),
+        translate("create table foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"Foo\" (id int)"),
+        translate("create table \"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.foo (id int)"),
+        translate("create table s.foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.\"Foo\" (id int)"),
+        translate("create table s.\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"s\".\"Foo\" (id int)"),
+        translate("create table \"s\".\"Foo\" (id int)", executor));
+    assertEquals(Statement.of("create table"), translate("create table", executor));
+    assertEquals(Statement.of("create table "), translate("create table ", executor));
+
+    assertEquals(
+        Statement.of("create table foo (id int)"),
+        translate("create table if not exists foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"Foo\" (id int)"),
+        translate("create table if not exists \"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.foo (id int)"),
+        translate("create table if not exists s.foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.\"Foo\" (id int)"),
+        translate("create table if not exists s.\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"s\".\"Foo\" (id int)"),
+        translate("create table if not exists \"s\".\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table if not exists"),
+        translate("create table if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateTableExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create table foo (id int)"),
+        translate("create table foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"Foo\" (id int)"),
+        translate("create table \"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.foo (id int)"),
+        translate("create table s.foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"S\".\"Foo\" (id int)"),
+        translate("create table \"S\".\"Foo\" (id int)", executor));
+    assertEquals(Statement.of("create table"), translate("create table", executor));
+
+    assertNull(translate("create table if not exists foo (id int)", executor));
+    assertNull(translate("create table if not exists \"Foo\" (id int)", executor));
+    assertNull(translate("create table if not exists s.foo (id int)", executor));
+    assertNull(translate("create table if not exists \"S\".\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table if not exists"),
+        translate("create table if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateIndexNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create index foo on bar(id)"),
+        translate("create index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"Foo\" on bar(id)"),
+        translate("create index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index s.foo on bar(id)"),
+        translate("create index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"s\".\"Foo\" on bar(id)"),
+        translate("create index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create index"), translate("create index", executor));
+    assertEquals(Statement.of("create index "), translate("create index ", executor));
+
+    assertEquals(
+        Statement.of("create index foo on bar(id)"),
+        translate("create index if not exists foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"Foo\" on bar(id)"),
+        translate("create index if not exists \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index s.foo on bar(id)"),
+        translate("create index if not exists s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"s\".\"Foo\" on bar(id)"),
+        translate("create index if not exists \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index if not exists"),
+        translate("create index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateIndexExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create index foo on bar(id)"),
+        translate("create index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"Foo\" on bar(id)"),
+        translate("create index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index s.foo on bar(id)"),
+        translate("create index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"s\".\"Foo\" on bar(id)"),
+        translate("create index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create index"), translate("create index", executor));
+
+    assertNull(translate("create index if not exists foo on bar(id int)", executor));
+    assertNull(translate("create index if not exists \"Foo\" on bar(id int)", executor));
+    assertNull(translate("create index if not exists s.foo on bar(id int)", executor));
+    assertNull(translate("create index if not exists \"s\".\"Foo\" on bar(id int)", executor));
+    assertEquals(
+        Statement.of("create index if not exists"),
+        translate("create index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateUniqueIndexNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create unique index foo on bar(id)"),
+        translate("create unique index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"Foo\" on bar(id)"),
+        translate("create unique index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index s.foo on bar(id)"),
+        translate("create unique index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"s\".\"Foo\" on bar(id)"),
+        translate("create unique index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create unique index"), translate("create unique index", executor));
+    assertEquals(Statement.of("create unique index "), translate("create unique index ", executor));
+
+    assertEquals(
+        Statement.of("create unique index foo on bar(id)"),
+        translate("create unique index if not exists foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"Foo\" on bar(id)"),
+        translate("create unique index if not exists \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index s.foo on bar(id)"),
+        translate("create unique index if not exists s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"s\".\"Foo\" on bar(id)"),
+        translate("create unique index if not exists \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index if not exists"),
+        translate("create unique index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateUniqueIndexExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create unique index foo on bar(id)"),
+        translate("create unique index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"Foo\" on bar(id)"),
+        translate("create unique index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index s.foo on bar(id)"),
+        translate("create unique index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"s\".\"Foo\" on bar(id)"),
+        translate("create unique index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create unique index"), translate("create unique index", executor));
+
+    assertNull(translate("create unique index if not exists foo on bar(id int)", executor));
+    assertNull(translate("create unique index if not exists \"Foo\" on bar(id int)", executor));
+    assertNull(translate("create unique index if not exists s.foo on bar(id int)", executor));
+    assertNull(
+        translate("create unique index if not exists \"s\".\"Foo\" on bar(id int)", executor));
+    assertEquals(
+        Statement.of("create unique index if not exists"),
+        translate("create unique index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropTableNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(Statement.of("drop table foo"), translate("drop table foo", executor));
+    assertEquals(Statement.of("drop table \"Foo\""), translate("drop table \"Foo\"", executor));
+    assertEquals(Statement.of("drop table s.foo"), translate("drop table s.foo", executor));
+    assertEquals(
+        Statement.of("drop table \"s\".\"Foo\""), translate("drop table \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table"), translate("drop table", executor));
+    assertEquals(Statement.of("drop table "), translate("drop table ", executor));
+
+    assertNull(translate("drop table if exists foo", executor));
+    assertNull(translate("drop table if exists \"Foo\"", executor));
+    assertNull(translate("drop table if exists sfoo", executor));
+    assertNull(translate("drop table if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table if exists"), translate("drop table if exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropTableExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(Statement.of("drop table foo"), translate("drop table foo", executor));
+    assertEquals(Statement.of("drop table \"Foo\""), translate("drop table \"Foo\"", executor));
+    assertEquals(Statement.of("drop table s.foo"), translate("drop table s.foo", executor));
+    assertEquals(
+        Statement.of("drop table \"s\".\"Foo\""), translate("drop table \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table"), translate("drop table", executor));
+
+    assertEquals(Statement.of("drop table foo"), translate("drop table if exists foo", executor));
+    assertEquals(
+        Statement.of("drop table \"Foo\""), translate("drop table if exists \"Foo\"", executor));
+    assertEquals(
+        Statement.of("drop table s.foo"), translate("drop table if exists s.foo", executor));
+    assertEquals(
+        Statement.of("drop table \"s\".\"Foo\""),
+        translate("drop table if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table if exists"), translate("drop table if exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropIndexNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(Statement.of("drop index foo"), translate("drop index foo", executor));
+    assertEquals(Statement.of("drop index \"Foo\""), translate("drop index \"Foo\"", executor));
+    assertEquals(Statement.of("drop index s.foo"), translate("drop index s.foo", executor));
+    assertEquals(
+        Statement.of("drop index \"s\".\"Foo\""), translate("drop index \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index"), translate("drop index", executor));
+    assertEquals(Statement.of("drop index "), translate("drop index ", executor));
+
+    assertNull(translate("drop index if exists foo", executor));
+    assertNull(translate("drop index if exists \"Foo\"", executor));
+    assertNull(translate("drop index if exists s.foo", executor));
+    assertNull(translate("drop index if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index if exists"), translate("drop index if exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropIndexExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(Statement.of("drop index foo"), translate("drop index foo", executor));
+    assertEquals(Statement.of("drop index \"Foo\""), translate("drop index \"Foo\"", executor));
+    assertEquals(Statement.of("drop index s.foo"), translate("drop index s.foo", executor));
+    assertEquals(
+        Statement.of("drop index \"s\".\"Foo\""), translate("drop index \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index"), translate("drop index", executor));
+
+    assertEquals(Statement.of("drop index foo"), translate("drop index if exists foo", executor));
+    assertEquals(
+        Statement.of("drop index \"Foo\""), translate("drop index if exists \"Foo\"", executor));
+    assertEquals(
+        Statement.of("drop index s.foo"), translate("drop index if exists s.foo", executor));
+    assertEquals(
+        Statement.of("drop index \"s\".\"Foo\""),
+        translate("drop index if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index if exists"), translate("drop index if exists", executor));
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java
index 41cc20fb..15508665 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java
@@ -19,6 +19,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
 import java.util.Arrays;
 import java.util.Collections;
 import org.junit.Test;
@@ -46,16 +47,55 @@ public class SimpleParserTest {
   }
 
   @Test
-  public void testReadIdentifier() {
-    assertEquals("foo", new SimpleParser("foo bar").readIdentifier());
-    assertEquals("foo", new SimpleParser("foo").readIdentifier());
-    assertEquals("\"foo\"", new SimpleParser("\"foo\" bar").readIdentifier());
-    assertEquals("\"foo\"", new SimpleParser("\"foo\"").readIdentifier());
-    assertEquals("foo", new SimpleParser(" foo bar").readIdentifier());
-    assertEquals("foo", new SimpleParser("\tfoo").readIdentifier());
-    assertEquals("\"foo\"", new SimpleParser("\n\"foo\" bar").readIdentifier());
-    assertEquals("\"foo\"", new SimpleParser("    \"foo\"").readIdentifier());
-    assertNull(new SimpleParser("\"foo").readIdentifier());
+  public void testReadTableOrIndexNamePart() {
+    assertEquals("foo", new SimpleParser("foo bar").readTableOrIndexNamePart());
+    assertEquals("foo", new SimpleParser("foo").readTableOrIndexNamePart());
+    assertEquals("\"foo\"", new SimpleParser("\"foo\" bar").readTableOrIndexNamePart());
+    assertEquals("\"foo\"", new SimpleParser("\"foo\"").readTableOrIndexNamePart());
+    assertEquals("foo", new SimpleParser(" foo bar").readTableOrIndexNamePart());
+    assertEquals("foo", new SimpleParser("\tfoo").readTableOrIndexNamePart());
+    assertEquals("\"foo\"", new SimpleParser("\n\"foo\" bar").readTableOrIndexNamePart());
+    assertEquals("\"foo\"", new SimpleParser("    \"foo\"").readTableOrIndexNamePart());
+    assertEquals("\"foo\"\"bar\"", new SimpleParser("\"foo\"\"bar\"").readTableOrIndexNamePart());
+    assertEquals("foo", new SimpleParser("foo\"bar\"").readTableOrIndexNamePart());
+    assertEquals("foo", new SimpleParser("foo.bar").readTableOrIndexNamePart());
+    assertEquals("foo", new SimpleParser("foo").readTableOrIndexNamePart());
+    assertEquals("\"foo\"", new SimpleParser("\"foo\".bar").readTableOrIndexNamePart());
+    assertEquals("\"foo\"", new SimpleParser("\"foo\"").readTableOrIndexNamePart());
+    assertNull(new SimpleParser("\"foo").readTableOrIndexNamePart());
+  }
+
+  @Test
+  public void testReadTableOrIndexName() {
+    assertEquals(new TableOrIndexName("foo"), new SimpleParser("foo bar").readTableOrIndexName());
+    assertEquals(new TableOrIndexName("foo"), new SimpleParser("foo").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\""), new SimpleParser("\"foo\" bar").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\""), new SimpleParser("\"foo\"").readTableOrIndexName());
+    assertEquals(new TableOrIndexName("foo"), new SimpleParser(" foo bar").readTableOrIndexName());
+    assertEquals(new TableOrIndexName("foo"), new SimpleParser("\tfoo").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\""), new SimpleParser("\n\"foo\" bar").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\""), new SimpleParser("    \"foo\"").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\"\"bar\""),
+        new SimpleParser("\"foo\"\"bar\"").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("foo"), new SimpleParser("foo\"bar\"").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("foo", "bar"), new SimpleParser("foo.bar").readTableOrIndexName());
+    assertEquals(new TableOrIndexName("foo"), new SimpleParser("foo").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\"", "bar"),
+        new SimpleParser("\"foo\".bar").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\"", "\"bar\""),
+        new SimpleParser("\"foo\".\"bar\"").readTableOrIndexName());
+    assertEquals(
+        new TableOrIndexName("\"foo\""), new SimpleParser("\"foo\"").readTableOrIndexName());
+    assertNull(new SimpleParser("\"foo").readTableOrIndexName());
   }
 
   @Test
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
index 0344746d..85755ed1 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
@@ -26,8 +26,11 @@ import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.DatabaseId;
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ReadContext;
 import com.google.cloud.spanner.ReadContext.QueryAnalyzeMode;
 import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.SpannerException;
@@ -47,11 +50,15 @@ import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
+import com.google.common.collect.ImmutableList;
 import com.google.common.primitives.Bytes;
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -90,9 +97,15 @@ public class StatementTest {
 
   @AfterClass
   public static void cleanup() {
+    deleteLogFile();
+  }
+
+  private static void deleteLogFile() {
     // TODO: Make error log file configurable and turn off writing to a file during tests.
-    File outputFile = new File("output.txt");
-    outputFile.delete();
+    try {
+      Files.deleteIfExists(new File("output.txt").toPath());
+    } catch (IOException ignore) {
+    }
   }
 
   @Test
@@ -155,7 +168,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sql), Statement.of(sql));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     assertFalse(intermediateStatement.isExecuted());
     assertEquals("UPDATE", intermediateStatement.getCommand());
@@ -227,7 +244,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sql), Statement.of(sql));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.executeAsync(backendConnection);
     backendConnection.flush();
@@ -252,7 +273,11 @@ public class StatementTest {
             .to(30)
             .build();
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     IntermediatePreparedStatement intermediateStatement =
         new IntermediatePreparedStatement(
@@ -319,7 +344,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sqlStatement), Statement.of(sqlStatement));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.describeAsync(backendConnection);
     backendConnection.flush();
@@ -363,7 +392,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sqlStatement), Statement.of(sqlStatement));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     when(connection.execute(Statement.of(sqlStatement)))
         .thenThrow(
@@ -411,8 +444,13 @@ public class StatementTest {
     String sql = "COPY keyvalue FROM STDIN;";
     CopyStatement statement =
         new CopyStatement(
-            connectionHandler, mock(OptionsMetadata.class), parse(sql), Statement.of(sql));
-    statement.execute();
+            connectionHandler, mock(OptionsMetadata.class), "", parse(sql), Statement.of(sql));
+
+    BackendConnection backendConnection =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, ImmutableList.of());
+    statement.executeAsync(backendConnection);
+    backendConnection.flush();
 
     byte[] payload = "2 3\n".getBytes();
     MutationWriter mutationWriter = statement.getMutationWriter();
@@ -463,28 +501,108 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sql), Statement.of(sql));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.executeAsync(backendConnection);
 
     assertThrows(IllegalStateException.class, intermediateStatement::getStatementResult);
   }
 
+  @Test
+  public void testCopyBatchSizeLimit() throws Exception {
+    setupQueryInformationSchemaResults();
+    BackendConnection backendConnection =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, ImmutableList.of());
+
+    byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/batch-size-test.txt"));
+
+    String sql = "COPY keyvalue FROM STDIN;";
+    CopyStatement copyStatement =
+        new CopyStatement(
+            connectionHandler, mock(OptionsMetadata.class), "", parse(sql), Statement.of(sql));
+
+    assertFalse(copyStatement.isExecuted());
+    copyStatement.executeAsync(backendConnection);
+    assertTrue(copyStatement.isExecuted());
+
+    backendConnection.flush();
+    MutationWriter mw = copyStatement.getMutationWriter();
+    //    // Inject a mock DatabaseClient for now.
+    //    // TODO: Fix this once we can use the Connection API.
+    //    Field databaseClientField = MutationWriter.class.getDeclaredField("databaseClient");
+    //    databaseClientField.setAccessible(true);
+    //    databaseClientField.set(mw, mock(DatabaseClient.class));
+    mw.addCopyData(payload);
+    mw.close();
+
+    assertEquals("TEXT", copyStatement.getFormatType());
+    assertEquals('\t', copyStatement.getDelimiterChar());
+    assertFalse(copyStatement.hasException());
+    assertEquals(12L, copyStatement.getUpdateCount());
+    assertEquals(12L, mw.getRowCount());
+
+    copyStatement.close();
+  }
+
+  @Test
+  public void testCopyDataRowLengthMismatchLimit() throws Exception {
+    setupQueryInformationSchemaResults();
+    BackendConnection backendConnection =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, ImmutableList.of());
+
+    byte[] payload = "1\t'one'\n2".getBytes();
+
+    String sql = "COPY keyvalue FROM STDIN;";
+    CopyStatement copyStatement =
+        new CopyStatement(
+            connectionHandler, mock(OptionsMetadata.class), "", parse(sql), Statement.of(sql));
+
+    assertFalse(copyStatement.isExecuted());
+    copyStatement.executeAsync(backendConnection);
+    assertTrue(copyStatement.isExecuted());
+
+    backendConnection.flush();
+    MutationWriter mw = copyStatement.getMutationWriter();
+    mw.addCopyData(payload);
+    mw.close();
+
+    SpannerException thrown = assertThrows(SpannerException.class, copyStatement::getUpdateCount);
+    assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
+    assertEquals(
+        "INVALID_ARGUMENT: Invalid COPY data: Row length mismatched. Expected 2 columns, but only found 1",
+        thrown.getMessage());
+
+    copyStatement.close();
+
+    deleteLogFile();
+  }
+
   private void setupQueryInformationSchemaResults() {
+    DatabaseClient databaseClient = mock(DatabaseClient.class);
+    ReadContext singleUseReadContext = mock(ReadContext.class);
+    when(databaseClient.singleUse()).thenReturn(singleUseReadContext);
+    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.getDatabaseClient()).thenReturn(databaseClient);
     ResultSet spannerType = mock(ResultSet.class);
     when(spannerType.getString("column_name")).thenReturn("key", "value");
     when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     when(spannerType.next()).thenReturn(true, true, false);
-    when(connection.executeQuery(
+    when(singleUseReadContext.executeQuery(
             ArgumentMatchers.argThat(
                 statement ->
                     statement != null && statement.getSql().startsWith("SELECT column_name"))))
         .thenReturn(spannerType);
 
-    ResultSet countResult = Mockito.mock(ResultSet.class);
+    ResultSet countResult = mock(ResultSet.class);
     when(countResult.getLong(0)).thenReturn(2L);
     when(countResult.next()).thenReturn(true, false);
-    when(connection.executeQuery(
+    when(singleUseReadContext.executeQuery(
             ArgumentMatchers.argThat(
                 statement ->
                     statement != null && statement.getSql().startsWith("SELECT COUNT(*)"))))
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/utils/CopyDataReceiverTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/utils/CopyDataReceiverTest.java
new file mode 100644
index 00000000..a0658916
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/utils/CopyDataReceiverTest.java
@@ -0,0 +1,72 @@
+// 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.utils;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler.ConnectionStatus;
+import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
+import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class CopyDataReceiverTest {
+
+  @Test
+  public void testCopyStatementWithException() {
+    SpannerException exception =
+        SpannerExceptionFactory.newSpannerException(
+            ErrorCode.INVALID_ARGUMENT, "Invalid copy statement");
+    CopyStatement statement = mock(CopyStatement.class);
+    when(statement.hasException()).thenReturn(true);
+    when(statement.getException()).thenReturn(exception);
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+
+    CopyDataReceiver receiver = new CopyDataReceiver(statement, connectionHandler);
+    SpannerException spannerException = assertThrows(SpannerException.class, receiver::handleCopy);
+
+    assertSame(exception, spannerException);
+  }
+
+  @Test
+  public void testCopyStatementWithCopyFailed() {
+    DataOutputStream outputStream = new DataOutputStream(new ByteArrayOutputStream());
+    CopyStatement statement = mock(CopyStatement.class);
+    ConnectionMetadata connectionMetadata = mock(ConnectionMetadata.class);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    when(connectionHandler.getStatus())
+        .thenReturn(ConnectionStatus.AUTHENTICATED, ConnectionStatus.COPY_FAILED);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+
+    CopyDataReceiver receiver = new CopyDataReceiver(statement, connectionHandler);
+    SpannerException spannerException = assertThrows(SpannerException.class, receiver::handleCopy);
+
+    assertEquals(ErrorCode.INTERNAL, spannerException.getErrorCode());
+    assertEquals("INTERNAL: Copy failed with unknown reason", spannerException.getMessage());
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java
index 0842a7bf..4edfe88d 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/utils/MutationWriterTest.java
@@ -34,6 +34,7 @@ import com.google.cloud.spanner.Mutation;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter.CopyTransactionMode;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
@@ -128,11 +129,12 @@ public class MutationWriterTest {
             false);
 
     mutationWriter.addCopyData("1\t\"One\"\n2\t\"Two\"\n".getBytes(StandardCharsets.UTF_8));
+    mutationWriter.commit();
     mutationWriter.close();
 
-    long updateCount = mutationWriter.call();
+    StatementResult updateCount = mutationWriter.call();
 
-    assertEquals(2L, updateCount);
+    assertEquals(2L, updateCount.getUpdateCount().longValue());
     List<Mutation> expectedMutations =
         ImmutableList.of(
             Mutation.newInsertBuilder("numbers").set("number").to(1L).set("name").to("One").build(),
@@ -208,11 +210,12 @@ public class MutationWriterTest {
       mutationWriter.addCopyData(
           "1\t\"One\"\n2\t\"Two\"\n3\t\"Three\"\n4\t\"Four\"\n5\t\"Five\"\n"
               .getBytes(StandardCharsets.UTF_8));
+      mutationWriter.commit();
       mutationWriter.close();
 
-      long updateCount = mutationWriter.call();
+      StatementResult updateCount = mutationWriter.call();
 
-      assertEquals(5L, updateCount);
+      assertEquals(5L, updateCount.getUpdateCount().longValue());
       verify(databaseClient, times(3)).write(anyIterable());
     } finally {
       System.getProperties().remove("copy_in_mutation_limit");
@@ -285,11 +288,12 @@ public class MutationWriterTest {
       mutationWriter.addCopyData(
           "1\t\"One\"\n2\t\"Two\"\n3\t\"Three\"\n4\t\"Four\"\n5\t\"Five\"\n"
               .getBytes(StandardCharsets.UTF_8));
+      mutationWriter.commit();
       mutationWriter.close();
 
-      long updateCount = mutationWriter.call();
+      StatementResult updateCount = mutationWriter.call();
 
-      assertEquals(5L, updateCount);
+      assertEquals(5L, updateCount.getUpdateCount().longValue());
       // We expect two batches, because:
       // 1. The commit limit is 80 bytes. That is for safety halved down to 40 bytes.
       // 2. Each record is 20 bytes.
@@ -329,6 +333,7 @@ public class MutationWriterTest {
     mutationWriter.addCopyData(
         "1\t\"One\"\n2\t\"Two\"\n3\t\"Three\"\n4\t\"Four\"\n5\t\"Five\"\n"
             .getBytes(StandardCharsets.UTF_8));
+    mutationWriter.commit();
     mutationWriter.close();
 
     ExecutorService executor = Executors.newFixedThreadPool(2);
@@ -344,9 +349,9 @@ public class MutationWriterTest {
           mutationWriter.close();
           return null;
         });
-    Future<Long> updateCount = executor.submit(mutationWriter);
+    Future<StatementResult> updateCount = executor.submit(mutationWriter);
 
-    assertEquals(5L, updateCount.get().longValue());
+    assertEquals(5L, updateCount.get().getUpdateCount().longValue());
     List<Mutation> expectedMutations =
         ImmutableList.of(
             Mutation.newInsertBuilder("numbers").set("number").to(1L).set("name").to("One").build(),
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
index 6bea11f1..842e7b85 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
@@ -14,15 +14,15 @@
 
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
+import static com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.MAX_INVALID_MESSAGE_COUNT;
 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.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doCallRealMethod;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
@@ -33,8 +33,10 @@ import static org.mockito.Mockito.when;
 import com.google.cloud.spanner.DatabaseClient;
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ReadContext;
 import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
@@ -52,7 +54,6 @@ 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.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.ManuallyCreatedToken;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.PreparedType;
@@ -935,7 +936,8 @@ public class ProtocolTest {
     ByteArrayOutputStream result = new ByteArrayOutputStream();
     DataOutputStream outputStream = new DataOutputStream(result);
 
-    Exception testException = new Exception("test error");
+    SpannerException testException =
+        SpannerExceptionFactory.newSpannerException(ErrorCode.INVALID_ARGUMENT, "test error");
     when(intermediatePortalStatement.hasException()).thenReturn(true);
     when(intermediatePortalStatement.getException()).thenReturn(testException);
     when(connectionHandler.getPortal(anyString())).thenReturn(intermediatePortalStatement);
@@ -1112,10 +1114,8 @@ public class ProtocolTest {
     byte[] value = Bytes.concat(messageMetadata, length);
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
-    ByteArrayOutputStream result = new ByteArrayOutputStream();
-    DataOutputStream outputStream = new DataOutputStream(result);
+    DataOutputStream outputStream = mock(DataOutputStream.class);
 
-    when(connectionHandler.getSpannerConnection()).thenReturn(connection);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -1127,11 +1127,8 @@ public class ProtocolTest {
 
     message.send();
 
-    // ReadyResponse
-    DataInputStream outputResult = inputStreamFromOutputStream(result);
-    assertEquals('Z', outputResult.readByte());
-    assertEquals(5, outputResult.readInt());
-    assertEquals('I', outputResult.readByte());
+    verify(extendedQueryProtocolHandler).flush();
+    verify(extendedQueryProtocolHandler, never()).sync();
   }
 
   @Test
@@ -1143,13 +1140,10 @@ public class ProtocolTest {
     byte[] value = Bytes.concat(messageMetadata, length);
 
     DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
-    ByteArrayOutputStream result = new ByteArrayOutputStream();
-    DataOutputStream outputStream = new DataOutputStream(result);
+    DataOutputStream outputStream = mock(DataOutputStream.class);
 
-    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);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
@@ -1159,11 +1153,8 @@ public class ProtocolTest {
 
     message.send();
 
-    // ReadyResponse
-    DataInputStream outputResult = inputStreamFromOutputStream(result);
-    assertEquals('Z', outputResult.readByte());
-    assertEquals(5, outputResult.readInt());
-    assertEquals('T', outputResult.readByte());
+    verify(extendedQueryProtocolHandler).flush();
+    verify(extendedQueryProtocolHandler, never()).sync();
   }
 
   @Test
@@ -1283,6 +1274,26 @@ public class ProtocolTest {
     verify(mw).addCopyData(payload);
   }
 
+  @Test
+  public void testCopyDataMessageWithNoCopyStatement() throws Exception {
+    byte[] messageMetadata = {'d'};
+    byte[] payload = "This is the payload".getBytes();
+    byte[] length = intToBytes(4 + payload.length);
+    byte[] value = Bytes.concat(messageMetadata, length, payload);
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
+
+    when(connectionHandler.getActiveStatement()).thenReturn(null);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+
+    WireMessage message = ControlMessage.create(connectionHandler);
+    // This should be a no-op.
+    message.sendPayload();
+  }
+
   @Test
   public void testMultipleCopyDataMessages() throws Exception {
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
@@ -1303,16 +1314,10 @@ public class ProtocolTest {
     DataInputStream inputStream2 = new DataInputStream(new ByteArrayInputStream(value2));
     DataInputStream inputStream3 = new DataInputStream(new ByteArrayInputStream(value3));
 
-    ResultSet spannerType = mock(ResultSet.class);
-    when(spannerType.getString("column_name")).thenReturn("key", "value");
-    when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
-    when(spannerType.next()).thenReturn(true, true, false);
-    when(connection.executeQuery(any(Statement.class))).thenReturn(spannerType);
-
     String sql = "COPY keyvalue FROM STDIN;";
     CopyStatement copyStatement =
-        new CopyStatement(connectionHandler, options, parse(sql), Statement.of(sql));
-    copyStatement.execute();
+        new CopyStatement(connectionHandler, options, "", parse(sql), Statement.of(sql));
+    copyStatement.executeAsync(mock(BackendConnection.class));
 
     when(connectionHandler.getActiveStatement()).thenReturn(copyStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
@@ -1355,24 +1360,22 @@ public class ProtocolTest {
     DataOutputStream outputStream = new DataOutputStream(result);
 
     CopyStatement copyStatement = mock(CopyStatement.class);
+    MutationWriter mutationWriter = mock(MutationWriter.class);
+    when(copyStatement.getMutationWriter()).thenReturn(mutationWriter);
     when(connectionHandler.getActiveStatement()).thenReturn(copyStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
 
-    MutationWriter mb = mock(MutationWriter.class);
-    when(copyStatement.getMutationWriter()).thenReturn(mb);
     WireMessage message = ControlMessage.create(connectionHandler);
 
     assertEquals(CopyDoneMessage.class, message.getClass());
     CopyDoneMessage messageSpy = (CopyDoneMessage) spy(message);
-    doNothing()
-        .when(messageSpy)
-        .sendSpannerResult(any(IntermediateStatement.class), any(QueryMode.class), anyLong());
 
     messageSpy.send();
-    verify(messageSpy).sendSpannerResult(copyStatement, QueryMode.SIMPLE, 0L);
+    verify(messageSpy).sendPayload();
+    verify(mutationWriter).commit();
   }
 
   @Test
@@ -1415,8 +1418,8 @@ public class ProtocolTest {
     String sql = "COPY keyvalue FROM STDIN;";
     CopyStatement copyStatement =
         new CopyStatement(
-            connectionHandler, mock(OptionsMetadata.class), parse(sql), Statement.of(sql));
-    copyStatement.execute();
+            connectionHandler, mock(OptionsMetadata.class), "", parse(sql), Statement.of(sql));
+    copyStatement.executeAsync(mock(BackendConnection.class));
 
     MutationWriter mw = copyStatement.getMutationWriter();
     mw.addCopyData(payload);
@@ -1428,113 +1431,6 @@ public class ProtocolTest {
     verify(resultSet, never()).close();
   }
 
-  @Test
-  public void testCopyBatchSizeLimit() throws Exception {
-    setupQueryInformationSchemaResults();
-    when(connection.getDatabaseClient()).thenReturn(mock(DatabaseClient.class));
-
-    byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/batch-size-test.txt"));
-
-    String sql = "COPY keyvalue FROM STDIN;";
-    CopyStatement copyStatement =
-        new CopyStatement(
-            connectionHandler, mock(OptionsMetadata.class), parse(sql), Statement.of(sql));
-
-    assertFalse(copyStatement.isExecuted());
-    copyStatement.execute();
-    assertTrue(copyStatement.isExecuted());
-
-    MutationWriter mw = copyStatement.getMutationWriter();
-    mw.addCopyData(payload);
-    mw.close();
-
-    assertEquals("TEXT", copyStatement.getFormatType());
-    assertEquals('\t', copyStatement.getDelimiterChar());
-    assertFalse(copyStatement.hasException());
-    assertEquals(12L, copyStatement.getUpdateCount());
-    assertEquals(12L, mw.getRowCount());
-
-    copyStatement.close();
-  }
-
-  @Test
-  public void testCopyDataRowLengthMismatchLimit() throws Exception {
-    setupQueryInformationSchemaResults();
-
-    byte[] payload = "1\t'one'\n2".getBytes();
-
-    String sql = "COPY keyvalue FROM STDIN;";
-    CopyStatement copyStatement =
-        new CopyStatement(
-            connectionHandler, mock(OptionsMetadata.class), parse(sql), Statement.of(sql));
-
-    assertFalse(copyStatement.isExecuted());
-    copyStatement.execute();
-    assertTrue(copyStatement.isExecuted());
-
-    MutationWriter mw = copyStatement.getMutationWriter();
-    mw.addCopyData(payload);
-    mw.close();
-
-    SpannerException thrown = assertThrows(SpannerException.class, copyStatement::getUpdateCount);
-    assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
-    assertEquals(
-        "INVALID_ARGUMENT: Invalid COPY data: Row length mismatched. Expected 2 columns, but only found 1",
-        thrown.getMessage());
-
-    copyStatement.close();
-  }
-
-  @Test
-  public void testCopyResumeErrorOutputFile() throws Exception {
-    setupQueryInformationSchemaResults();
-
-    byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/test-copy-output.txt"));
-
-    String sql = "COPY keyvalue FROM STDIN;";
-    CopyStatement copyStatement =
-        new CopyStatement(
-            connectionHandler, mock(OptionsMetadata.class), parse(sql), Statement.of(sql));
-    assertFalse(copyStatement.isExecuted());
-    copyStatement.execute();
-    assertTrue(copyStatement.isExecuted());
-
-    MutationWriter mw = copyStatement.getMutationWriter();
-    mw.addCopyData(payload);
-
-    SpannerException thrown = assertThrows(SpannerException.class, copyStatement::getUpdateCount);
-    assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
-    assertEquals(
-        "INVALID_ARGUMENT: Invalid input syntax for type INT64:\"'5'\"", thrown.getMessage());
-
-    copyStatement.close();
-  }
-
-  @Test
-  public void testCopyResumeErrorStartOutputFile() throws Exception {
-    setupQueryInformationSchemaResults();
-
-    byte[] payload =
-        Files.readAllBytes(Paths.get("./src/test/resources/test-copy-start-output.txt"));
-
-    String sql = "COPY keyvalue FROM STDIN;";
-    CopyStatement copyStatement =
-        new CopyStatement(connectionHandler, options, parse(sql), Statement.of(sql));
-    assertFalse(copyStatement.isExecuted());
-    copyStatement.execute();
-    assertTrue(copyStatement.isExecuted());
-
-    MutationWriter mw = copyStatement.getMutationWriter();
-    mw.addCopyData(payload);
-
-    SpannerException thrown = assertThrows(SpannerException.class, copyStatement::getUpdateCount);
-    assertEquals(ErrorCode.INVALID_ARGUMENT, thrown.getErrorCode());
-    assertEquals(
-        "INVALID_ARGUMENT: Invalid input syntax for type INT64:\"'1'\"", thrown.getMessage());
-
-    copyStatement.close();
-  }
-
   @Test
   public void testFunctionCallMessageThrowsException() throws Exception {
     byte[] messageMetadata = {'F'};
@@ -1801,13 +1697,184 @@ public class ProtocolTest {
     assertThrows(IllegalStateException.class, describeMessage::getPortalMetadata);
   }
 
+  @Test
+  public void testSkipMessage() throws Exception {
+    byte[] messageMetadata = {0, 0, 0, 45};
+    String payload = "INSERT INTO users (name) VALUES ('test')\0";
+    byte[] value = Bytes.concat(messageMetadata, payload.getBytes());
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+
+    SkipMessage message = SkipMessage.createForValidStream(connectionHandler);
+    message.send();
+
+    // Verify that nothing was written to the output.
+    assertEquals(0, result.size());
+    assertEquals("Skip", message.getMessageName());
+    assertEquals("", message.getIdentifier());
+    assertEquals("Length: 45", message.getPayloadString());
+  }
+
+  @Test
+  public void testFlushSkippedInCopyMode() throws Exception {
+    byte[] messageMetadata = {FlushMessage.IDENTIFIER, 0, 0, 0, 4};
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(messageMetadata));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
+
+    ControlMessage message = ControlMessage.create(connectionHandler);
+
+    assertEquals(SkipMessage.class, message.getClass());
+    // Verify that nothing was written to the output.
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  public void testSyncSkippedInCopyMode() throws Exception {
+    byte[] messageMetadata = {SyncMessage.IDENTIFIER, 0, 0, 0, 4};
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(messageMetadata));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.COPY_IN);
+
+    ControlMessage message = ControlMessage.create(connectionHandler);
+
+    assertEquals(SkipMessage.class, message.getClass());
+    // Verify that nothing was written to the output.
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  public void testCopyDataSkippedInNormalMode() throws Exception {
+    byte[] messageMetadata = {CopyDataMessage.IDENTIFIER, 0, 0, 0, 4};
+    String payload = "1\t'One'\n2\t'Two'\n";
+    byte[] value = Bytes.concat(messageMetadata, payload.getBytes());
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
+
+    ControlMessage message = ControlMessage.create(connectionHandler);
+
+    assertEquals(SkipMessage.class, message.getClass());
+    // Verify that nothing was written to the output.
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  public void testCopyDoneSkippedInNormalMode() throws Exception {
+    byte[] messageMetadata = {CopyDoneMessage.IDENTIFIER, 0, 0, 0, 4};
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(messageMetadata));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
+
+    ControlMessage message = ControlMessage.create(connectionHandler);
+
+    assertEquals(SkipMessage.class, message.getClass());
+    // Verify that nothing was written to the output.
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  public void testCopyFailSkippedInNormalMode() throws Exception {
+    byte[] messageMetadata = {CopyFailMessage.IDENTIFIER, 0, 0, 0, 4};
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(messageMetadata));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
+
+    ControlMessage message = ControlMessage.create(connectionHandler);
+
+    assertEquals(SkipMessage.class, message.getClass());
+    // Verify that nothing was written to the output.
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  public void testRepeatedCopyDataInNormalMode_TerminatesConnectionAndReturnsError()
+      throws Exception {
+    String payload = "1\t'One'\n2\t'Two'\n";
+    byte[] messageMetadata = {
+      CopyDataMessage.IDENTIFIER,
+      0,
+      0,
+      0,
+      (byte) (4 + payload.getBytes(StandardCharsets.UTF_8).length)
+    };
+    byte[] value = new byte[0];
+    for (int i = 0; i <= MAX_INVALID_MESSAGE_COUNT; i++) {
+      value = Bytes.concat(value, messageMetadata, payload.getBytes());
+    }
+
+    DataInputStream inputStream = new DataInputStream(new ByteArrayInputStream(value));
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+    DataOutputStream outputStream = new DataOutputStream(result);
+
+    when(connectionMetadata.getInputStream()).thenReturn(inputStream);
+    when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
+    when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
+    doCallRealMethod().when(connectionHandler).increaseInvalidMessageCount();
+    when(connectionHandler.getInvalidMessageCount()).thenCallRealMethod();
+
+    for (int i = 0; i < MAX_INVALID_MESSAGE_COUNT; i++) {
+      ControlMessage message = ControlMessage.create(connectionHandler);
+      assertEquals(SkipMessage.class, message.getClass());
+      // Verify that nothing was written to the output.
+      assertEquals(0, result.size());
+      verify(connectionHandler, never()).setStatus(ConnectionStatus.TERMINATED);
+    }
+
+    ControlMessage.create(connectionHandler);
+    verify(connectionHandler).setStatus(ConnectionStatus.TERMINATED);
+    byte[] resultBytes = result.toByteArray();
+    assertEquals('E', resultBytes[0]);
+  }
+
   private void setupQueryInformationSchemaResults() {
+    DatabaseClient databaseClient = mock(DatabaseClient.class);
+    ReadContext singleUseReadContext = mock(ReadContext.class);
+    when(databaseClient.singleUse()).thenReturn(singleUseReadContext);
     when(connectionHandler.getSpannerConnection()).thenReturn(connection);
+    when(connection.getDatabaseClient()).thenReturn(databaseClient);
     ResultSet spannerType = mock(ResultSet.class);
     when(spannerType.getString("column_name")).thenReturn("key", "value");
     when(spannerType.getString("data_type")).thenReturn("bigint", "character varying");
     when(spannerType.next()).thenReturn(true, true, false);
-    when(connection.executeQuery(
+    when(singleUseReadContext.executeQuery(
             ArgumentMatchers.argThat(
                 statement ->
                     statement != null && statement.getSql().startsWith("SELECT column_name"))))
@@ -1816,7 +1883,7 @@ public class ProtocolTest {
     ResultSet countResult = mock(ResultSet.class);
     when(countResult.getLong(0)).thenReturn(2L);
     when(countResult.next()).thenReturn(true, false);
-    when(connection.executeQuery(
+    when(singleUseReadContext.executeQuery(
             ArgumentMatchers.argThat(
                 statement ->
                     statement != null && statement.getSql().startsWith("SELECT COUNT(*)"))))