Repository: GoogleCloudPlatform/pgadapter
Pull request #621 (2023-01-27 15:29:19)

build: measure p50 in ycsb benchmarks
Add p50 as a measurement to the YCSB benchmarks.

codecov[bot]: # [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/621?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> Merging [#621](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/621?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (fb2b4b9) into [postgresql-dialect](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/commit/5e9f95a720f1648236b39167b227cc70bd40e323?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (5e9f95a) will **not change** coverage.
> The diff coverage is `n/a`.

```diff
@@                  Coverage Diff                  @@
##             postgresql-dialect     #621   +/-   ##
=====================================================
  Coverage                 88.84%   88.84%           
  Complexity                 2149     2149           
=====================================================
  Files                       124      124           
  Lines                      7125     7125           
  Branches                   1021     1021           
=====================================================
  Hits                       6330     6330           
  Misses                      558      558           
  Partials                    237      237           
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `88.84% <ø> (ø)` | |

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.


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


Patch:
diff --git a/README.md b/README.md
index ef3e3029..bbade18a 100644
--- a/README.md
+++ b/README.md
@@ -11,8 +11,8 @@ PGAdapter can be used with the following drivers and clients:
 4. `psycopg2`: Version 2.9.3 and higher (but not `psycopg3`) are supported. See [psycopg2](docs/psycopg2.md) for more details.
 5. `node-postgres`: Version 8.8.0 and higher are supported. See [node-postgres support](docs/node-postgres.md) for more details.
 
-## Frameworks
-PGAdapter can be used with the following frameworks:
+## Frameworks and Tools
+PGAdapter can be used with the following frameworks and tools:
 1. `Liquibase`: Version 4.12.0 and higher are supported. See [Liquibase support](docs/liquibase.md)
    for more details. See also [this directory](samples/java/liquibase) for a sample application using `Liquibase`.
 2. `gorm`: Version 1.23.8 and higher are supported. See [gorm support](docs/gorm.md) for more details.
@@ -20,6 +20,8 @@ PGAdapter can be used with the following frameworks:
 3. `SQLAlchemy`: Version 1.4.45 has _experimental support_. See [SQLAlchemy support](docs/sqlalchemy.md)
    for more details. See also [this directory](samples/python/sqlalchemy-sample) for a sample
    application using `SQLAlchemy`.
+4. `pgbench` can be used with PGAdapter, but with some limitations. See [pgbench.md](docs/pgbench.md)
+   for more details.
 
 ## FAQ
 See [Frequently Asked Questions](docs/faq.md) for answers to frequently asked questions.
diff --git a/benchmarks/ycsb/run.sh b/benchmarks/ycsb/run.sh
index dfeb48e0..94768749 100644
--- a/benchmarks/ycsb/run.sh
+++ b/benchmarks/ycsb/run.sh
@@ -31,16 +31,19 @@ psql -h localhost -c "CREATE TABLE IF NOT EXISTS usertable (
                   read_min        float,
                   read_max        float,
                   read_avg        float,
+                  read_p50        float,
                   read_p95        float,
                   read_p99        float,
                   update_min      float,
                   update_max      float,
                   update_avg      float,
+                  update_p50      float,
                   update_p95      float,
                   update_p99      float,
                   insert_min      float,
                   insert_max      float,
                   insert_avg      float,
+                  insert_p50      float,
                   insert_p95      float,
                   insert_p99      float,
                   primary key (executed_at, deployment, workload, threads, batch_size, operation_count)
@@ -98,6 +101,7 @@ do
         fi
         ./bin/ycsb run jdbc -P workloads/workload$WORKLOAD \
           -threads $THREADS \
+          -p hdrhistogram.percentiles=50,95,99 \
           -p operationcount=$OPERATION_COUNT \
           -p recordcount=100000 \
           -p db.batchsize=$BATCH_SIZE \
@@ -112,30 +116,33 @@ do
         READ_AVG=$(grep '\[READ\], AverageLatency(us), ' ycsb.log | sed 's/^.*, //' | sed "s/NaN/'NaN'/" || echo null)
         READ_MIN=$(grep '\[READ\], MinLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         READ_MAX=$(grep '\[READ\], MaxLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
+        READ_P50=$(grep '\[READ\], 50thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         READ_P95=$(grep '\[READ\], 95thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         READ_P99=$(grep '\[READ\], 99thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
 
         UPDATE_AVG=$(grep '\[UPDATE\], AverageLatency(us), ' ycsb.log | sed 's/^.*, //' | sed "s/NaN/'NaN'/" || echo null)
         UPDATE_MIN=$(grep '\[UPDATE\], MinLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         UPDATE_MAX=$(grep '\[UPDATE\], MaxLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
+        UPDATE_P50=$(grep '\[UPDATE\], 50thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         UPDATE_P95=$(grep '\[UPDATE\], 95thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         UPDATE_P99=$(grep '\[UPDATE\], 99thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
 
         INSERT_AVG=$(grep '\[INSERT\], AverageLatency(us), ' ycsb.log | sed 's/^.*, //' | sed "s/NaN/'NaN'/" || echo null)
         INSERT_MIN=$(grep '\[INSERT\], MinLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         INSERT_MAX=$(grep '\[INSERT\], MaxLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
+        INSERT_P50=$(grep '\[INSERT\], 50thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         INSERT_P95=$(grep '\[INSERT\], 95thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
         INSERT_P99=$(grep '\[INSERT\], 99thPercentileLatency(us), ' ycsb.log | sed 's/^.*, //' || echo null)
 
         psql -h localhost \
           -c "insert into run (executed_at, deployment, workload, threads, batch_size, operation_count, run_time, throughput,
-                               read_min, read_max, read_avg, read_p95, read_p99,
-                               update_min, update_max, update_avg, update_p95, update_p99,
-                               insert_min, insert_max, insert_avg, insert_p95, insert_p99) values
+                               read_min, read_max, read_avg, read_p50, read_p95, read_p99,
+                               update_min, update_max, update_avg, update_p50, update_p95, update_p99,
+                               insert_min, insert_max, insert_avg, insert_p50, insert_p95, insert_p99) values
                                ('$EXECUTED_AT', '$DEPLOYMENT', '$WORKLOAD', $THREADS, $BATCH_SIZE, $OPERATION_COUNT, $OVERALL_RUNTIME, $OVERALL_THROUGHPUT,
-                                $READ_MIN, $READ_MAX, $READ_AVG, $READ_P95, $READ_P99,
-                                $UPDATE_MIN, $UPDATE_MAX, $UPDATE_AVG, $UPDATE_P95, $UPDATE_P99,
-                                $INSERT_MIN, $INSERT_MAX, $INSERT_AVG, $INSERT_P95, $INSERT_P99)"
+                                $READ_MIN, $READ_MAX, $READ_AVG, $READ_P50, $READ_P95, $READ_P99,
+                                $UPDATE_MIN, $UPDATE_MAX, $UPDATE_AVG, $UPDATE_P50, $UPDATE_P95, $UPDATE_P99,
+                                $INSERT_MIN, $INSERT_MAX, $INSERT_AVG, $INSERT_P50, $INSERT_P95, $INSERT_P99)"
       done
     done
   done
diff --git a/docs/pgbench.md b/docs/pgbench.md
new file mode 100644
index 00000000..eb393d90
--- /dev/null
+++ b/docs/pgbench.md
@@ -0,0 +1,100 @@
+# PGAdapter - pgbench
+
+[pgbench](https://www.postgresql.org/docs/current/pgbench.html) can be used with PGAdapter, but with
+some limitations.
+
+Follow these steps to initialize and run benchmarks with `pgbench` with PGAdapter:
+
+## Create Data Model
+The default data model that is generated by `pgbench` does not include primary keys for the tables.
+Cloud Spanner requires all tables to have primary keys. Execute the following command to manually
+create the data model for `pgbench`:
+
+```shell
+psql -h /tmp -p 5432 -d my-database \
+        -c "START BATCH DDL;
+            CREATE TABLE pgbench_accounts (
+                aid integer primary key  NOT NULL,            
+                bid integer  NULL,                
+                abalance integer  NULL,           
+                filler varchar(84)  NULL
+            );
+            CREATE TABLE pgbench_branches (
+                bid integer primary key  NOT NULL,            
+                bbalance integer  NULL,           
+                filler varchar(88)  NULL
+            );
+            CREATE TABLE pgbench_history (       
+                tid integer  NOT NULL DEFAULT -1,                      
+                bid integer  NOT NULL DEFAULT -1,                      
+                aid integer  NOT NULL DEFAULT -1,                      
+                delta integer  NULL,                    
+                mtime timestamptz  NULL,
+                filler varchar(22)  NULL,
+                primary key (tid, bid, aid)
+            );
+            CREATE TABLE pgbench_tellers (
+                tid integer  primary key NOT NULL,           
+                bid integer  NULL,               
+                tbalance integer  NULL,          
+                filler varchar(84)  NULL
+            );
+            RUN BATCH;"
+```
+
+## Initialize Data
+`pgbench` deletes and inserts data into PostgreSQL using a combination of `truncate`, `insert` and
+`copy` statements. These statements all run in a single transaction. The amount of data that is
+modified during this transaction will exceed the transaction mutation limits of Cloud Spanner. This
+can be worked around by adding the following options to the `pgbench` initialization command:
+
+```shell
+pgbench "host=/tmp port=5432 dbname=my-database \
+        options='-c spanner.force_autocommit=on -c spanner.autocommit_dml_mode=\'partitioned_non_atomic\''" \
+        -i -Ig \
+        --scale=100
+```
+
+These additional options do the following:
+1. `spanner.force_autocommit=true`: This instructs PGAdapter to ignore any transaction statements and
+   execute all statements in autocommit mode. This prevents the initialization from being executed as
+   a single, large transaction.
+2. `spanner.autocommit_dml_mode='partitioned_non_atomic'`: This instructs PGAdapter to use Partitioned
+   DML for (large) update statements. This ensures that a single statement succeeds even if it would
+   exceed the transaction limits of Cloud Spanner, including large `copy` operations.
+3. `-i` activates initialization mode of `pgbench`.
+4. `-Ig` instructs `pgbench` to generate test data client side. PGAdapter does not support generating
+   test data server side.
+
+## Running Benchmarks
+You can run different benchmarks after finishing the steps above.
+
+### Default Benchmark
+Run a default benchmark to verify that everything works as expected.
+
+```shell
+pgbench "host=/tmp port=5432 dbname=my-database"
+```
+
+### Number of Clients
+Increase the number of clients and threads to increase the number of parallel transactions.
+
+```shell
+pgbench "host=/tmp port=5432 dbname=my-database" \
+        --client=100 --jobs=100 \
+        --progress=10
+```
+
+## Dropping Tables
+Execute the following command to remove the `pgbench` tables from your database if you no longer
+need them.
+
+```shell
+psql -h /tmp -p 5432 -d my-database \
+        -c "START BATCH DDL;
+            DROP TABLE pgbench_history;
+            DROP TABLE pgbench_tellers;
+            DROP TABLE pgbench_branches;
+            DROP TABLE pgbench_accounts;
+            RUN BATCH;"
+```
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 176a2f48..0213ac5a 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
@@ -120,7 +120,7 @@ public class ConnectionHandler extends Thread {
   private int invalidMessagesCount;
   private Connection spannerConnection;
   private DatabaseId databaseId;
-  private WellKnownClient wellKnownClient;
+  private WellKnownClient wellKnownClient = WellKnownClient.UNSPECIFIED;
   private boolean hasDeterminedClientUsingQuery;
   private ExtendedQueryProtocolHandler extendedQueryProtocolHandler;
   private CopyStatement activeCopyStatement;
@@ -476,12 +476,12 @@ public class ConnectionHandler extends Thread {
     DataOutputStream output = getConnectionMetadata().getOutputStream();
     if (this.status == ConnectionStatus.TERMINATED
         || this.status == ConnectionStatus.UNAUTHENTICATED) {
-      new ErrorResponse(output, exception).send();
+      new ErrorResponse(this, exception).send();
       new TerminateResponse(output).send();
     } else if (this.status == ConnectionStatus.COPY_IN) {
-      new ErrorResponse(output, exception).send();
+      new ErrorResponse(this, exception).send();
     } else {
-      new ErrorResponse(output, exception).send();
+      new ErrorResponse(this, exception).send();
       new ReadyResponse(output, ReadyResponse.Status.IDLE).send();
     }
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
index 6abac46f..f3b595cc 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
@@ -88,7 +88,7 @@ public class OptionsMetadata {
   }
 
   private static final Logger logger = Logger.getLogger(OptionsMetadata.class.getName());
-  private static final String DEFAULT_SERVER_VERSION = "14.1";
+  public static final String DEFAULT_SERVER_VERSION = "14.1";
   private static final String DEFAULT_USER_AGENT = "pg-adapter";
 
   private static final String OPTION_SERVER_PORT = "s";
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java b/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java
index 4692e590..ca37be80 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java
@@ -374,6 +374,11 @@ public class SessionState {
     this.transactionSettings = null;
   }
 
+  /** Returns the PostgreSQL version. */
+  public String getServerVersion() {
+    return getStringSetting(null, "server_version", OptionsMetadata.DEFAULT_SERVER_VERSION);
+  }
+
   /**
    * Returns whether transaction statements should be ignored and all statements should be executed
    * in autocommit mode.
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 a5cdf21d..e9171ca5 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
@@ -59,12 +59,14 @@ import com.google.cloud.spanner.pgadapter.statements.DdlExecutor.NotExecuted;
 import com.google.cloud.spanner.pgadapter.statements.SessionStatementParser.SessionStatement;
 import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
 import com.google.cloud.spanner.pgadapter.statements.local.LocalStatement;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
 import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 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.base.Suppliers;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableMap.Builder;
@@ -87,6 +89,7 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Future;
 import java.util.function.Function;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
@@ -208,9 +211,10 @@ public class BackendConnection {
         //  block always ends with a ROLLBACK, PGAdapter should skip the entire execution of that
         //  block.
         SessionStatement sessionStatement = getSessionManagementStatement(parsedStatement);
-        if (!localStatements.isEmpty() && localStatements.containsKey(statement.getSql())) {
+        if (!localStatements.get().isEmpty()
+            && localStatements.get().containsKey(statement.getSql())) {
           result.set(
-              Objects.requireNonNull(localStatements.get(statement.getSql()))
+              Objects.requireNonNull(localStatements.get().get(statement.getSql()))
                   .execute(BackendConnection.this));
         } else if (sessionStatement != null) {
           result.set(sessionStatement.execute(sessionState));
@@ -253,7 +257,7 @@ public class BackendConnection {
           // Potentially replace pg_catalog table references with common table expressions.
           updatedStatement =
               sessionState.isReplacePgCatalogTables()
-                  ? pgCatalog.replacePgCatalogTables(statement)
+                  ? pgCatalog.get().replacePgCatalogTables(statement)
                   : statement;
           updatedStatement = bindStatement(updatedStatement);
           result.set(analyzeOrExecute(updatedStatement));
@@ -567,8 +571,8 @@ public class BackendConnection {
   private static final Statement ROLLBACK = Statement.of("ROLLBACK");
 
   private final SessionState sessionState;
-  private final PgCatalog pgCatalog;
-  private final ImmutableMap<String, LocalStatement> localStatements;
+  private final Supplier<PgCatalog> pgCatalog;
+  private final Supplier<ImmutableMap<String, LocalStatement>> localStatements;
   private ConnectionState connectionState = ConnectionState.IDLE;
   private TransactionMode transactionMode = TransactionMode.IMPLICIT;
   private final String currentSchema = "public";
@@ -581,24 +585,31 @@ public class BackendConnection {
   BackendConnection(
       DatabaseId databaseId,
       Connection spannerConnection,
+      Supplier<WellKnownClient> wellKnownClient,
       OptionsMetadata optionsMetadata,
-      ImmutableList<LocalStatement> localStatements) {
+      Supplier<ImmutableList<LocalStatement>> localStatements) {
     this.sessionState = new SessionState(optionsMetadata);
-    this.pgCatalog = new PgCatalog(this.sessionState);
+    this.pgCatalog =
+        Suppliers.memoize(
+            () -> new PgCatalog(BackendConnection.this.sessionState, wellKnownClient.get()));
     this.spannerConnection = spannerConnection;
     this.databaseId = databaseId;
     this.ddlExecutor = new DdlExecutor(databaseId, this);
-    if (localStatements.isEmpty()) {
-      this.localStatements = EMPTY_LOCAL_STATEMENTS;
-    } else {
-      Builder<String, LocalStatement> builder = ImmutableMap.builder();
-      for (LocalStatement localStatement : localStatements) {
-        for (String sql : localStatement.getSql()) {
-          builder.put(new SimpleImmutableEntry<>(sql, localStatement));
-        }
-      }
-      this.localStatements = builder.build();
-    }
+    this.localStatements =
+        Suppliers.memoize(
+            () -> {
+              if (localStatements.get().isEmpty()) {
+                return EMPTY_LOCAL_STATEMENTS;
+              } else {
+                Builder<String, LocalStatement> builder = ImmutableMap.builder();
+                for (LocalStatement localStatement : localStatements.get()) {
+                  for (String sql : localStatement.getSql()) {
+                    builder.put(new SimpleImmutableEntry<>(sql, localStatement));
+                  }
+                }
+                return builder.build();
+              }
+            });
   }
 
   /** Returns the current connection state. */
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 fe950b4e..e615f107 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
@@ -563,7 +563,7 @@ public class CopyStatement extends IntermediatePortalStatement {
     }
     ParsedCopyStatement.Builder builder = new ParsedCopyStatement.Builder();
     if (parser.eatToken("(")) {
-      builder.query = parser.parseExpressionUntilKeyword(ImmutableList.of(), true, true);
+      builder.query = parser.parseExpressionUntilKeyword(ImmutableList.of(), true, true, false);
       if (!parser.eatToken(")")) {
         throw PGExceptionFactory.newPGException(
             "missing closing parentheses after query", SQLState.SyntaxError);
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatement.java
index 876e8862..a003a2d9 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatement.java
@@ -57,7 +57,7 @@ public class CopyToStatement extends IntermediatePortalStatement {
       new byte[] {'P', 'G', 'C', 'O', 'P', 'Y', '\n', -1, '\r', '\n', '\0'};
 
   private final ParsedCopyStatement parsedCopyStatement;
-  private final CSVFormat csvFormat;
+  private CSVFormat csvFormat;
   private final AtomicBoolean hasReturnedData = new AtomicBoolean(false);
 
   public CopyToStatement(
@@ -272,6 +272,10 @@ public class CopyToStatement extends IntermediatePortalStatement {
       }
     }
     String row = csvFormat.format((Object[]) data);
+    // Only include the header with the first row.
+    if (!csvFormat.getSkipHeaderRecord()) {
+      csvFormat = csvFormat.builder().setSkipHeaderRecord(true).build();
+    }
     return new CopyDataResponse(this.outputStream, row, csvFormat.getRecordSeparator().charAt(0));
   }
 
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 ae61125d..9d3c5bf7 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
@@ -40,8 +40,9 @@ public class ExtendedQueryProtocolHandler {
         new BackendConnection(
             connectionHandler.getDatabaseId(),
             connectionHandler.getSpannerConnection(),
+            connectionHandler::getWellKnownClient,
             connectionHandler.getServer().getOptions(),
-            connectionHandler.getWellKnownClient().getLocalStatements(connectionHandler));
+            () -> connectionHandler.getWellKnownClient().getLocalStatements(connectionHandler));
   }
 
   /** Constructor only intended for testing. */
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/PgCatalog.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/PgCatalog.java
index a1d4aac4..0fa0dcbf 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/PgCatalog.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/PgCatalog.java
@@ -20,6 +20,8 @@ import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.Value;
 import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
@@ -27,11 +29,13 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
+import java.util.function.Supplier;
 import java.util.regex.Pattern;
+import javax.annotation.Nonnull;
 
 @InternalApi
 public class PgCatalog {
-  private static final ImmutableMap<TableOrIndexName, TableOrIndexName> TABLE_REPLACEMENTS =
+  private static final ImmutableMap<TableOrIndexName, TableOrIndexName> DEFAULT_TABLE_REPLACEMENTS =
       ImmutableMap.<TableOrIndexName, TableOrIndexName>builder()
           .put(
               new TableOrIndexName("pg_catalog", "pg_namespace"),
@@ -63,6 +67,12 @@ public class PgCatalog {
           Pattern.compile("pg_table_is_visible\\(.+\\)"), "true",
           Pattern.compile("ANY\\(current_schemas\\(true\\)\\)"), "'public'");
 
+  private final ImmutableSet<String> checkPrefixes;
+
+  private final ImmutableMap<TableOrIndexName, TableOrIndexName> tableReplacements;
+
+  private final ImmutableMap<Pattern, Supplier<String>> functionReplacements;
+
   private final Map<TableOrIndexName, PgCatalogTable> pgCatalogTables =
       ImmutableMap.of(
           new TableOrIndexName(null, "pg_namespace"), new PgNamespace(),
@@ -73,14 +83,31 @@ public class PgCatalog {
           new TableOrIndexName(null, "pg_settings"), new PgSettings());
   private final SessionState sessionState;
 
-  public PgCatalog(SessionState sessionState) {
-    this.sessionState = sessionState;
+  public PgCatalog(@Nonnull SessionState sessionState, @Nonnull WellKnownClient wellKnownClient) {
+    this.sessionState = Preconditions.checkNotNull(sessionState);
+    this.checkPrefixes = wellKnownClient.getPgCatalogCheckPrefixes();
+    ImmutableMap.Builder<TableOrIndexName, TableOrIndexName> builder =
+        ImmutableMap.<TableOrIndexName, TableOrIndexName>builder()
+            .putAll(DEFAULT_TABLE_REPLACEMENTS);
+    wellKnownClient
+        .getTableReplacements()
+        .forEach((k, v) -> builder.put(TableOrIndexName.of(k), TableOrIndexName.of(v)));
+    this.tableReplacements = builder.build();
+    this.functionReplacements =
+        ImmutableMap.<Pattern, Supplier<String>>builder()
+            .put(Pattern.compile("pg_catalog.pg_table_is_visible\\(.+\\)"), () -> "true")
+            .put(Pattern.compile("pg_table_is_visible\\(.+\\)"), () -> "true")
+            .put(Pattern.compile("ANY\\(current_schemas\\(true\\)\\)"), () -> "'public'")
+            .put(
+                Pattern.compile("version\\(\\)"), () -> "'" + sessionState.getServerVersion() + "'")
+            .putAll(wellKnownClient.getFunctionReplacements())
+            .build();
   }
 
   /** Replace supported pg_catalog tables with Common Table Expressions. */
   public Statement replacePgCatalogTables(Statement statement) {
     Tuple<Set<TableOrIndexName>, Statement> replacedTablesStatement =
-        new TableParser(statement).detectAndReplaceTables(TABLE_REPLACEMENTS);
+        new TableParser(statement).detectAndReplaceTables(DEFAULT_TABLE_REPLACEMENTS);
     if (replacedTablesStatement.x().isEmpty()) {
       return replacedTablesStatement.y();
     }
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 ccbd8eb9..8918b341 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
@@ -354,6 +354,15 @@ public class SimpleParser {
       ImmutableList<String> keywords,
       boolean sameParensLevelAsStart,
       boolean stopAtEndOfExpression) {
+    return parseExpressionUntilKeyword(
+        keywords, sameParensLevelAsStart, stopAtEndOfExpression, true);
+  }
+
+  String parseExpressionUntilKeyword(
+      ImmutableList<String> keywords,
+      boolean sameParensLevelAsStart,
+      boolean stopAtEndOfExpression,
+      boolean stopAtComma) {
     skipWhitespaces();
     int start = pos;
     boolean valid;
@@ -366,7 +375,7 @@ public class SimpleParser {
         if (stopAtEndOfExpression && parens < 0) {
           break;
         }
-      } else if (stopAtEndOfExpression && parens == 0 && sql.charAt(pos) == ',') {
+      } else if (stopAtEndOfExpression && parens == 0 && stopAtComma && sql.charAt(pos) == ',') {
         break;
       }
       if ((!sameParensLevelAsStart || parens == 0)
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java
index 44db4631..3532d495 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java
@@ -17,6 +17,8 @@ package com.google.cloud.spanner.pgadapter.utils;
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
 import com.google.cloud.spanner.pgadapter.statements.local.DjangoGetTableNamesStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.ListDatabasesStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.LocalStatement;
@@ -24,10 +26,17 @@ import com.google.cloud.spanner.pgadapter.statements.local.SelectCurrentCatalogS
 import com.google.cloud.spanner.pgadapter.statements.local.SelectCurrentDatabaseStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.SelectCurrentSchemaStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.SelectVersionStatement;
+import com.google.cloud.spanner.pgadapter.wireoutput.NoticeResponse;
+import com.google.cloud.spanner.pgadapter.wireoutput.NoticeResponse.NoticeSeverity;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
+import java.util.function.Supplier;
+import java.util.regex.Pattern;
 import javax.annotation.Nonnull;
 import org.postgresql.core.Oid;
 
@@ -45,6 +54,10 @@ public class ClientAutoDetector {
           SelectCurrentCatalogStatement.INSTANCE,
           SelectVersionStatement.INSTANCE,
           DjangoGetTableNamesStatement.INSTANCE);
+  private static final ImmutableSet<String> DEFAULT_CHECK_PG_CATALOG_PREFIXES =
+      ImmutableSet.of("pg_");
+  public static final String PGBENCH_USAGE_HINT =
+      "See https://github.com/GoogleCloudPlatform/pgadapter/blob/-/docs/pgbench.md for how to use pgbench with PGAdapter";
 
   public enum WellKnownClient {
     PSQL {
@@ -67,6 +80,48 @@ public class ClientAutoDetector {
         return ImmutableList.of(new ListDatabasesStatement(connectionHandler));
       }
     },
+    PGBENCH {
+      final ImmutableList<String> errorHints = ImmutableList.of(PGBENCH_USAGE_HINT);
+      volatile long lastHintTimestampMillis = 0L;
+
+      @Override
+      public void reset() {
+        lastHintTimestampMillis = 0L;
+      }
+
+      @Override
+      boolean isClient(List<String> orderedParameterKeys, Map<String, String> parameters) {
+        // PGBENCH makes it easy for us, as it sends its own name in the application_name parameter.
+        return parameters.containsKey("application_name")
+            && parameters.get("application_name").equals("pgbench");
+      }
+
+      @Override
+      public ImmutableList<NoticeResponse> createStartupNoticeResponses(
+          ConnectionHandler connection) {
+        synchronized (PGBENCH) {
+          // Only send the hint at most once every 30 seconds, to prevent benchmark runs that open
+          // multiple connections from showing the hint every time.
+          if (Duration.ofMillis(System.currentTimeMillis() - lastHintTimestampMillis).getSeconds()
+              > 30L) {
+            lastHintTimestampMillis = System.currentTimeMillis();
+            return ImmutableList.of(
+                new NoticeResponse(
+                    connection.getConnectionMetadata().getOutputStream(),
+                    SQLState.Success,
+                    NoticeSeverity.INFO,
+                    "Detected connection from pgbench",
+                    PGBENCH_USAGE_HINT + "\n"));
+          }
+        }
+        return super.createStartupNoticeResponses(connection);
+      }
+
+      @Override
+      public ImmutableList<String> getErrorHints(PGException exception) {
+        return errorHints;
+      }
+    },
     JDBC {
       @Override
       boolean isClient(List<String> orderedParameterKeys, Map<String, String> parameters) {
@@ -153,6 +208,10 @@ public class ClientAutoDetector {
 
     abstract boolean isClient(List<String> orderedParameterKeys, Map<String, String> parameters);
 
+    /** Resets any cached or temporary settings for the client. */
+    @VisibleForTesting
+    public void reset() {}
+
     boolean isClient(List<Statement> statements) {
       return false;
     }
@@ -164,6 +223,29 @@ public class ClientAutoDetector {
       return EMPTY_LOCAL_STATEMENTS;
     }
 
+    public ImmutableSet<String> getPgCatalogCheckPrefixes() {
+      return DEFAULT_CHECK_PG_CATALOG_PREFIXES;
+    }
+
+    public ImmutableMap<String, String> getTableReplacements() {
+      return ImmutableMap.of();
+    }
+
+    public ImmutableMap<Pattern, Supplier<String>> getFunctionReplacements() {
+      return ImmutableMap.of();
+    }
+
+    /** Creates specific notice messages for a client after startup. */
+    public ImmutableList<NoticeResponse> createStartupNoticeResponses(
+        ConnectionHandler connection) {
+      return ImmutableList.of();
+    }
+
+    /** Returns the client-specific hint(s) that should be included with the given exception. */
+    public ImmutableList<String> getErrorHints(PGException exception) {
+      return ImmutableList.of();
+    }
+
     public ImmutableMap<String, String> getDefaultParameters() {
       return ImmutableMap.of();
     }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponse.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponse.java
index e37964fd..35502d2d 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponse.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponse.java
@@ -15,9 +15,11 @@
 package com.google.cloud.spanner.pgadapter.wireoutput;
 
 import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
 import com.google.common.base.Strings;
-import java.io.DataOutputStream;
+import com.google.common.collect.ImmutableList;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.text.MessageFormat;
@@ -25,6 +27,7 @@ import java.text.MessageFormat;
 /** Sends error information back to client. */
 @InternalApi
 public class ErrorResponse extends WireOutput {
+  private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
 
   private static final int HEADER_LENGTH = 4;
   private static final int FIELD_IDENTIFIER_LENGTH = 1;
@@ -36,58 +39,83 @@ public class ErrorResponse extends WireOutput {
   private static final byte HINT_FLAG = 'H';
   private static final byte NULL_TERMINATOR = 0;
 
-  private final byte[] severity;
-  private final byte[] errorMessage;
-  private final byte[] errorState;
-  private final byte[] hints;
-
-  public ErrorResponse(DataOutputStream output, PGException pgException) {
-    super(output, calculateLength(pgException));
-    this.errorMessage = pgException.getMessage().getBytes(StandardCharsets.UTF_8);
-    this.errorState = pgException.getSQLState().getBytes();
-    this.severity = pgException.getSeverity().name().getBytes(StandardCharsets.UTF_8);
-    this.hints =
-        Strings.isNullOrEmpty(pgException.getHints())
-            ? null
-            : pgException.getHints().getBytes(StandardCharsets.UTF_8);
+  private final PGException pgException;
+  private final WellKnownClient client;
+
+  public ErrorResponse(ConnectionHandler connection, PGException pgException) {
+    super(
+        connection.getConnectionMetadata().getOutputStream(),
+        calculateLength(pgException, connection.getWellKnownClient()));
+    this.pgException = pgException;
+    this.client = connection.getWellKnownClient();
   }
 
-  static int calculateLength(PGException pgException) {
+  static int calculateLength(PGException pgException, WellKnownClient client) {
     int length =
         HEADER_LENGTH
             + FIELD_IDENTIFIER_LENGTH
-            + pgException.getSeverity().name().getBytes(StandardCharsets.UTF_8).length
+            + convertSeverityToWireProtocol(pgException).length
             + NULL_TERMINATOR_LENGTH
             + FIELD_IDENTIFIER_LENGTH
-            + pgException.getSQLState().getBytes().length
+            + convertSQLStateToWireProtocol(pgException).length
             + NULL_TERMINATOR_LENGTH
             + FIELD_IDENTIFIER_LENGTH
-            + pgException.getMessage().getBytes(StandardCharsets.UTF_8).length
+            + convertMessageToWireProtocol(pgException).length
             + NULL_TERMINATOR_LENGTH
             + NULL_TERMINATOR_LENGTH;
-    if (!Strings.isNullOrEmpty(pgException.getHints())) {
-      length +=
-          FIELD_IDENTIFIER_LENGTH
-              + pgException.getHints().getBytes(StandardCharsets.UTF_8).length
-              + NULL_TERMINATOR_LENGTH;
+    byte[] hints = convertHintsToWireProtocol(pgException, client);
+    if (hints.length > 0) {
+      length += FIELD_IDENTIFIER_LENGTH + hints.length + NULL_TERMINATOR_LENGTH;
     }
     return length;
   }
 
+  static byte[] convertSeverityToWireProtocol(PGException pgException) {
+    return pgException.getSeverity().name().getBytes(StandardCharsets.UTF_8);
+  }
+
+  static byte[] convertSQLStateToWireProtocol(PGException pgException) {
+    return pgException.getSQLState().getBytes();
+  }
+
+  static byte[] convertMessageToWireProtocol(PGException pgException) {
+    return pgException.getMessage().getBytes(StandardCharsets.UTF_8);
+  }
+
+  static byte[] convertHintsToWireProtocol(PGException pgException, WellKnownClient client) {
+    if (Strings.isNullOrEmpty(pgException.getHints())
+        && client.getErrorHints(pgException).isEmpty()) {
+      return EMPTY_BYTE_ARRAY;
+    }
+    String hints = "";
+    if (!Strings.isNullOrEmpty(pgException.getHints())) {
+      hints += pgException.getHints();
+    }
+    ImmutableList<String> clientHints = client.getErrorHints(pgException);
+    if (!clientHints.isEmpty()) {
+      if (hints.length() > 0) {
+        hints += "\n";
+      }
+      hints += String.join("\n", clientHints);
+    }
+    return hints.getBytes(StandardCharsets.UTF_8);
+  }
+
   @Override
   protected void sendPayload() throws IOException {
     this.outputStream.writeByte(SEVERITY_FLAG);
-    this.outputStream.write(severity);
+    this.outputStream.write(convertSeverityToWireProtocol(pgException));
     this.outputStream.writeByte(NULL_TERMINATOR);
     this.outputStream.writeByte(CODE_FLAG);
-    this.outputStream.write(this.errorState);
+    this.outputStream.write(convertSQLStateToWireProtocol(pgException));
     this.outputStream.writeByte(NULL_TERMINATOR);
     this.outputStream.writeByte(MESSAGE_FLAG);
-    this.outputStream.write(this.errorMessage);
+    this.outputStream.write(convertMessageToWireProtocol(pgException));
     this.outputStream.writeByte(NULL_TERMINATOR);
-    if (this.hints != null) {
+    byte[] hints = convertHintsToWireProtocol(pgException, client);
+    if (hints.length > 0) {
       this.outputStream.writeByte(HINT_FLAG);
-      this.outputStream.write(this.hints);
+      this.outputStream.write(hints);
       this.outputStream.writeByte(NULL_TERMINATOR);
     }
     this.outputStream.writeByte(NULL_TERMINATOR);
@@ -109,8 +137,10 @@ public class ErrorResponse extends WireOutput {
         .format(
             new Object[] {
               this.length,
-              new String(this.errorMessage, UTF8),
-              this.hints == null ? "" : new String(this.hints, UTF8)
+              new String(convertMessageToWireProtocol(pgException), UTF8),
+              convertHintsToWireProtocol(pgException, client).length == 0
+                  ? ""
+                  : new String(convertHintsToWireProtocol(pgException, client), UTF8)
             });
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/NoticeResponse.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/NoticeResponse.java
new file mode 100644
index 00000000..bcf3d748
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireoutput/NoticeResponse.java
@@ -0,0 +1,129 @@
+// Copyright 2023 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.wireoutput;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.common.base.Preconditions;
+import java.io.DataOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.text.MessageFormat;
+
+/**
+ * Notices can be sent as asynchronous messages and can include warnings, informational messages,
+ * debug information, etc.
+ */
+@InternalApi
+public class NoticeResponse extends WireOutput {
+  public enum NoticeSeverity {
+    WARNING,
+    NOTICE,
+    DEBUG,
+    INFO,
+    LOG,
+    TIP,
+  }
+
+  private static final int HEADER_LENGTH = 4;
+  private static final int FIELD_IDENTIFIER_LENGTH = 1;
+  private static final int NULL_TERMINATOR_LENGTH = 1;
+
+  private static final byte MESSAGE_FLAG = 'M';
+  private static final byte CODE_FLAG = 'C';
+  private static final byte SEVERITY_FLAG = 'S';
+  private static final byte HINT_FLAG = 'H';
+  private static final byte NULL_TERMINATOR = 0;
+
+  private SQLState sqlState;
+  private final NoticeSeverity severity;
+  private final String message;
+  private final String hint;
+
+  public NoticeResponse(
+      DataOutputStream output,
+      SQLState sqlState,
+      NoticeSeverity severity,
+      String message,
+      String hint) {
+    super(
+        output,
+        HEADER_LENGTH
+            + FIELD_IDENTIFIER_LENGTH
+            + Preconditions.checkNotNull(severity).name().getBytes(StandardCharsets.UTF_8).length
+            + NULL_TERMINATOR_LENGTH
+            + FIELD_IDENTIFIER_LENGTH
+            + sqlState.getBytes().length
+            + NULL_TERMINATOR_LENGTH
+            + FIELD_IDENTIFIER_LENGTH
+            + Preconditions.checkNotNull(message).getBytes(StandardCharsets.UTF_8).length
+            + NULL_TERMINATOR_LENGTH
+            + (hint == null
+                ? 0
+                : (FIELD_IDENTIFIER_LENGTH
+                    + hint.getBytes(StandardCharsets.UTF_8).length
+                    + NULL_TERMINATOR_LENGTH))
+            + NULL_TERMINATOR_LENGTH);
+    this.sqlState = sqlState;
+    this.severity = severity;
+    this.message = message;
+    this.hint = hint;
+  }
+
+  @Override
+  protected void sendPayload() throws Exception {
+    this.outputStream.writeByte(SEVERITY_FLAG);
+    this.outputStream.write(severity.name().getBytes(StandardCharsets.UTF_8));
+    this.outputStream.writeByte(NULL_TERMINATOR);
+    this.outputStream.writeByte(CODE_FLAG);
+    this.outputStream.write(sqlState.getBytes());
+    this.outputStream.writeByte(NULL_TERMINATOR);
+    this.outputStream.writeByte(MESSAGE_FLAG);
+    this.outputStream.write(message.getBytes(StandardCharsets.UTF_8));
+    this.outputStream.writeByte(NULL_TERMINATOR);
+    if (this.hint != null) {
+      this.outputStream.writeByte(HINT_FLAG);
+      this.outputStream.write(hint.getBytes(StandardCharsets.UTF_8));
+      this.outputStream.writeByte(NULL_TERMINATOR);
+    }
+    this.outputStream.writeByte(NULL_TERMINATOR);
+  }
+
+  @Override
+  public byte getIdentifier() {
+    return 'N';
+  }
+
+  @Override
+  protected String getMessageName() {
+    return "Notice";
+  }
+
+  public String getMessage() {
+    return this.message;
+  }
+
+  public String getHint() {
+    return this.hint;
+  }
+
+  @Override
+  protected String getPayloadString() {
+    return new MessageFormat("Length: {0}, Severity: {1}, Notice Message: {2}, Hint: {3}")
+        .format(
+            new Object[] {
+              this.length, this.severity, this.message, this.hint == null ? "" : this.hint
+            });
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BootstrapMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BootstrapMessage.java
index 8cc11a6c..e4ac643e 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BootstrapMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BootstrapMessage.java
@@ -19,6 +19,7 @@ import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.session.SessionState;
 import com.google.cloud.spanner.pgadapter.wireoutput.AuthenticationOkResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.KeyDataResponse;
+import com.google.cloud.spanner.pgadapter.wireoutput.NoticeResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ParameterStatusResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.ReadyResponse.Status;
@@ -107,7 +108,11 @@ public abstract class BootstrapMessage extends WireMessage {
    * @throws Exception
    */
   public static void sendStartupMessage(
-      DataOutputStream output, int connectionId, int secret, SessionState sessionState)
+      DataOutputStream output,
+      int connectionId,
+      int secret,
+      SessionState sessionState,
+      Iterable<NoticeResponse> startupNotices)
       throws Exception {
     new AuthenticationOkResponse(output).send(false);
     new KeyDataResponse(output, connectionId, secret).send(false);
@@ -166,6 +171,9 @@ public abstract class BootstrapMessage extends WireMessage {
             "TimeZone".getBytes(StandardCharsets.UTF_8),
             ZoneId.systemDefault().getId().getBytes(StandardCharsets.UTF_8))
         .send(false);
+    for (NoticeResponse noticeResponse : startupNotices) {
+      noticeResponse.send(false);
+    }
     new ReadyResponse(output, Status.IDLE).send();
   }
 }
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 e97f269a..67ea96f8 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
@@ -179,7 +179,7 @@ public abstract class ControlMessage extends WireMessage {
         connection.increaseInvalidMessageCount();
         if (connection.getInvalidMessageCount() > MAX_INVALID_MESSAGE_COUNT) {
           new ErrorResponse(
-                  connection.getConnectionMetadata().getOutputStream(),
+                  connection,
                   PGException.newBuilder(
                           String.format(
                               "Received %d invalid/unexpected messages. Last received message: '%c'",
@@ -233,7 +233,7 @@ public abstract class ControlMessage extends WireMessage {
    * @throws Exception if there is some issue in the sending of the error messages.
    */
   protected void handleError(Exception exception) throws Exception {
-    new ErrorResponse(this.outputStream, PGExceptionFactory.toPGException(exception)).send(false);
+    new ErrorResponse(this.connection, PGExceptionFactory.toPGException(exception)).send(false);
   }
 
   /**
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/PasswordMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/PasswordMessage.java
index a289b44a..45d29d62 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/PasswordMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/PasswordMessage.java
@@ -70,7 +70,7 @@ public class PasswordMessage extends ControlMessage {
   protected void sendPayload() throws Exception {
     if (!useAuthentication()) {
       new ErrorResponse(
-              this.outputStream,
+              this.connection,
               PGException.newBuilder("Received PasswordMessage while authentication is disabled.")
                   .setSQLState(SQLState.ProtocolViolation)
                   .setSeverity(Severity.ERROR)
@@ -83,7 +83,7 @@ public class PasswordMessage extends ControlMessage {
     Credentials credentials = checkCredentials(this.username, this.password);
     if (credentials == null) {
       new ErrorResponse(
-              this.outputStream,
+              this.connection,
               PGException.newBuilder("Invalid credentials received.")
                   .setHints(
                       "PGAdapter expects credentials to be one of the following:\n"
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
index c2c7f001..17d9f666 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
@@ -96,7 +96,8 @@ public class StartupMessage extends BootstrapMessage {
         connection.getConnectionMetadata().getOutputStream(),
         connection.getConnectionId(),
         connection.getSecret(),
-        connection.getExtendedQueryProtocolHandler().getBackendConnection().getSessionState());
+        connection.getExtendedQueryProtocolHandler().getBackendConnection().getSessionState(),
+        connection.getWellKnownClient().createStartupNoticeResponses(connection));
     connection.setStatus(ConnectionStatus.AUTHENTICATED);
   }
 


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java
index d151f407..6c8bb88e 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/CopyOutMockServerTest.java
@@ -291,7 +291,13 @@ public class CopyOutMockServerTest extends AbstractMockServerTest {
   @Test
   public void testCopyOutCsvWithHeader() throws SQLException, IOException {
     mockSpanner.putStatementResult(
-        StatementResult.query(Statement.of("select * from all_types"), ALL_TYPES_RESULTSET));
+        StatementResult.query(
+            Statement.of("select * from all_types"),
+            ALL_TYPES_RESULTSET
+                .toBuilder()
+                .addRows(ALL_TYPES_RESULTSET.getRows(0))
+                .addRows(ALL_TYPES_NULLS_RESULTSET.getRows(0))
+                .build()));
 
     try (Connection connection = DriverManager.getConnection(createUrl())) {
       connection.createStatement().execute("set time zone 'UTC'");
@@ -302,7 +308,9 @@ public class CopyOutMockServerTest extends AbstractMockServerTest {
 
       assertEquals(
           "col_bigint-col_bool-col_bytea-col_float8-col_int-col_numeric-col_timestamptz-col_date-col_varchar-col_jsonb\n"
-              + "1-t-\\x74657374-3.14-100-6.626-\"2022-02-16 13:18:02.123456+00\"-\"2022-03-29\"-test-\"{~\"key~\": ~\"value~\"}\"\n",
+              + "1-t-\\x74657374-3.14-100-6.626-\"2022-02-16 13:18:02.123456+00\"-\"2022-03-29\"-test-\"{~\"key~\": ~\"value~\"}\"\n"
+              + "1-t-\\x74657374-3.14-100-6.626-\"2022-02-16 13:18:02.123456+00\"-\"2022-03-29\"-test-\"{~\"key~\": ~\"value~\"}\"\n"
+              + "\\N-\\N-\\N-\\N-\\N-\\N-\\N-\\N-\\N-\\N\n",
           writer.toString());
     }
   }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPgbenchMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPgbenchMockServerTest.java
new file mode 100644
index 00000000..9e477b2d
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPgbenchMockServerTest.java
@@ -0,0 +1,114 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+
+import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
+import io.grpc.Status;
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.SQLWarning;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.postgresql.PGProperty;
+import org.postgresql.util.PSQLException;
+import org.postgresql.util.PSQLWarning;
+
+@RunWith(JUnit4.class)
+public class EmulatedPgbenchMockServerTest extends AbstractMockServerTest {
+
+  @BeforeClass
+  public static void loadPgJdbcDriver() throws Exception {
+    // Make sure the PG JDBC driver is loaded.
+    Class.forName("org.postgresql.Driver");
+  }
+
+  /**
+   * Creates a JDBC connection string that instructs the PG JDBC driver to use the default simple
+   * mode. It also adds 'pgbench' as the application name, which will make PGAdapter automatically
+   * recognize the connection as a pgbench connection.
+   */
+  private String createUrl() {
+    return String.format(
+        "jdbc:postgresql://localhost:%d/db?preferQueryMode=simple&%s=pgbench&%s=090000",
+        pgServer.getLocalPort(),
+        PGProperty.APPLICATION_NAME.getName(),
+        PGProperty.ASSUME_MIN_SERVER_VERSION.getName());
+  }
+
+  @Test
+  public void testClientDetectionAndHint() throws Exception {
+    WellKnownClient.PGBENCH.reset();
+    // Verify that we get the notice response that indicates that the client was automatically
+    // detected.
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      SQLWarning warning = connection.getWarnings();
+      assertNotNull(warning);
+      PSQLWarning psqlWarning = (PSQLWarning) warning;
+      assertNotNull(psqlWarning.getServerErrorMessage());
+      assertNotNull(psqlWarning.getServerErrorMessage().getSQLState());
+      assertArrayEquals(
+          SQLState.Success.getBytes(),
+          psqlWarning.getServerErrorMessage().getSQLState().getBytes(StandardCharsets.UTF_8));
+      assertEquals(
+          "Detected connection from pgbench", psqlWarning.getServerErrorMessage().getMessage());
+      assertEquals(
+          ClientAutoDetector.PGBENCH_USAGE_HINT + "\n",
+          psqlWarning.getServerErrorMessage().getHint());
+
+      assertNull(warning.getNextWarning());
+    }
+
+    // Verify that creating a second connection directly afterwards with pgbench does not repeat the
+    // hint.
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      SQLWarning warning = connection.getWarnings();
+      assertNull(warning);
+    }
+  }
+
+  @Test
+  public void testErrorHint() throws SQLException {
+    // Verify that any error message includes the pgbench usage hint.
+    String sql = "select * from foo where bar=1";
+    mockSpanner.putStatementResult(
+        StatementResult.exception(
+            Statement.of(sql),
+            Status.INVALID_ARGUMENT.withDescription("test error").asRuntimeException()));
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      PSQLException exception =
+          assertThrows(PSQLException.class, () -> connection.createStatement().execute(sql));
+      assertNotNull(exception.getServerErrorMessage());
+      assertEquals(
+          String.format("test error - Statement: '%s'", sql),
+          exception.getServerErrorMessage().getMessage());
+      assertEquals(
+          ClientAutoDetector.PGBENCH_USAGE_HINT, exception.getServerErrorMessage().getHint());
+    }
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
index beef711e..541f76f2 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
@@ -761,7 +761,11 @@ public class ITPsqlTest implements IntegrationTest {
           // Iran observed DST in 1978. Not all databases agree on this.
           ZoneId.of("Asia/Tehran"),
           // Rankin_Inlet did not observer DST in 1970-1979, but not all databases agree.
-          ZoneId.of("America/Rankin_Inlet"));
+          ZoneId.of("America/Rankin_Inlet"),
+          // Pangnirtung did not observer DST in 1970-1979, but not all databases agree.
+          ZoneId.of("America/Pangnirtung"),
+          // Niue switched from -11:30 to -11 in 1978. Not all JDKs know that.
+          ZoneId.of("Pacific/Niue"));
 
   private LocalDate generateRandomLocalDate() {
     return LocalDate.ofEpochDay(random.nextInt(365 * 100));
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java
index dac991ac..d67eb144 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java
@@ -32,6 +32,7 @@ import com.google.cloud.spanner.pgadapter.error.PGException;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode;
 import com.google.cloud.spanner.pgadapter.statements.PgCatalog;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
 import com.google.common.collect.ImmutableMap;
 import java.util.List;
 import java.util.Map;
@@ -463,7 +464,7 @@ public class SessionStateTest {
   @Test
   public void testAddSessionState() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
-    PgCatalog pgCatalog = new PgCatalog(state);
+    PgCatalog pgCatalog = new PgCatalog(state, WellKnownClient.UNSPECIFIED);
     Statement statement = Statement.of("select * from pg_settings");
 
     Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
@@ -476,7 +477,7 @@ public class SessionStateTest {
   @Test
   public void testAddSessionStateWithParameters() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
-    PgCatalog pgCatalog = new PgCatalog(state);
+    PgCatalog pgCatalog = new PgCatalog(state, WellKnownClient.UNSPECIFIED);
     Statement statement =
         Statement.newBuilder("select * from pg_settings where name=$1")
             .bind("p1")
@@ -497,7 +498,7 @@ public class SessionStateTest {
   @Test
   public void testAddSessionStateWithoutPgSettings() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
-    PgCatalog pgCatalog = new PgCatalog(state);
+    PgCatalog pgCatalog = new PgCatalog(state, WellKnownClient.UNSPECIFIED);
     Statement statement = Statement.of("select * from some_table");
 
     Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
@@ -508,7 +509,7 @@ public class SessionStateTest {
   @Test
   public void testAddSessionStateWithComments() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
-    PgCatalog pgCatalog = new PgCatalog(state);
+    PgCatalog pgCatalog = new PgCatalog(state, WellKnownClient.UNSPECIFIED);
     Statement statement = Statement.of("/* This comment is preserved */ select * from pg_settings");
 
     Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
@@ -521,7 +522,7 @@ public class SessionStateTest {
   @Test
   public void testAddSessionStateWithExistingCTE() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
-    PgCatalog pgCatalog = new PgCatalog(state);
+    PgCatalog pgCatalog = new PgCatalog(state, WellKnownClient.UNSPECIFIED);
     Statement statement =
         Statement.of(
             "with my_cte as (select col1, col2 from foo) select * from pg_settings inner join my_cte on my_cte.col1=pg_settings.name");
@@ -538,7 +539,7 @@ public class SessionStateTest {
   @Test
   public void testAddSessionStateWithCommentsAndExistingCTE() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
-    PgCatalog pgCatalog = new PgCatalog(state);
+    PgCatalog pgCatalog = new PgCatalog(state, WellKnownClient.UNSPECIFIED);
     Statement statement =
         Statement.of(
             "/* This comment is preserved */ with foo as (select * from bar)\nselect * from pg_settings");
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 f149bcf6..55ca7b76 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
@@ -52,6 +52,7 @@ import com.google.cloud.spanner.pgadapter.statements.BackendConnection.QueryResu
 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.ClientAutoDetector.WellKnownClient;
 import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.common.collect.ImmutableList;
@@ -134,8 +135,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
 
     backendConnection.execute(
         PARSER.parse(Statement.of("CREATE TABLE \"Foo\" (id bigint primary key)")),
@@ -168,8 +170,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     Future<StatementResult> result =
         backendConnection.executeCopy(parsedStatement, statement, receiver, writer, executor);
     backendConnection.flush();
@@ -205,8 +208,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     onlyDmlStatements.execute(parsedUpdateStatement, updateStatement, Function.identity());
     onlyDmlStatements.execute(parsedUpdateStatement, updateStatement, Function.identity());
     assertTrue(onlyDmlStatements.hasDmlOrCopyStatementsAfter(0));
@@ -216,8 +220,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     onlyCopyStatements.executeCopy(parsedCopyStatement, copyStatement, receiver, writer, executor);
     onlyCopyStatements.executeCopy(parsedCopyStatement, copyStatement, receiver, writer, executor);
     assertTrue(onlyCopyStatements.hasDmlOrCopyStatementsAfter(0));
@@ -227,8 +232,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     dmlAndCopyStatements.execute(parsedUpdateStatement, updateStatement, Function.identity());
     dmlAndCopyStatements.executeCopy(
         parsedCopyStatement, copyStatement, receiver, writer, executor);
@@ -239,8 +245,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     onlySelectStatements.execute(parsedSelectStatement, selectStatement, Function.identity());
     onlySelectStatements.execute(parsedSelectStatement, selectStatement, Function.identity());
     assertFalse(onlySelectStatements.hasDmlOrCopyStatementsAfter(0));
@@ -250,8 +257,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     onlyClientSideStatements.execute(
         parsedClientSideStatement, clientSideStatement, Function.identity());
     onlyClientSideStatements.execute(
@@ -263,8 +271,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     onlyUnknownStatements.execute(parsedUnknownStatement, unknownStatement, Function.identity());
     onlyUnknownStatements.execute(parsedUnknownStatement, unknownStatement, Function.identity());
     assertFalse(onlyUnknownStatements.hasDmlOrCopyStatementsAfter(0));
@@ -274,8 +283,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     dmlAndSelectStatements.execute(parsedUpdateStatement, updateStatement, Function.identity());
     dmlAndSelectStatements.execute(parsedSelectStatement, selectStatement, Function.identity());
     assertTrue(dmlAndSelectStatements.hasDmlOrCopyStatementsAfter(0));
@@ -285,8 +295,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     copyAndSelectStatements.executeCopy(
         parsedCopyStatement, copyStatement, receiver, writer, executor);
     copyAndSelectStatements.execute(parsedSelectStatement, selectStatement, Function.identity());
@@ -297,8 +308,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            ImmutableList.of());
+            ImmutableList::of);
     copyAndUnknownStatements.executeCopy(
         parsedCopyStatement, copyStatement, receiver, writer, executor);
     copyAndUnknownStatements.execute(parsedUnknownStatement, unknownStatement, Function.identity());
@@ -323,7 +335,11 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
-            DatabaseId.of("p", "i", "d"), connection, mock(OptionsMetadata.class), localStatements);
+            DatabaseId.of("p", "i", "d"),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            mock(OptionsMetadata.class),
+            () -> localStatements);
     Future<StatementResult> resultFuture =
         backendConnection.execute(
             parsedListDatabasesStatement,
@@ -355,7 +371,11 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
-            DatabaseId.of("p", "i", "d"), connection, mock(OptionsMetadata.class), localStatements);
+            DatabaseId.of("p", "i", "d"),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            mock(OptionsMetadata.class),
+            () -> localStatements);
     Future<StatementResult> resultFuture =
         backendConnection.execute(parsedStatement, statement, Function.identity());
     backendConnection.flush();
@@ -388,8 +408,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             connection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            EMPTY_LOCAL_STATEMENTS);
+            () -> EMPTY_LOCAL_STATEMENTS);
     Future<StatementResult> resultFuture =
         backendConnection.execute(parsedStatement, statement, Function.identity());
     backendConnection.flush();
@@ -412,8 +433,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             connection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            EMPTY_LOCAL_STATEMENTS);
+            () -> EMPTY_LOCAL_STATEMENTS);
     Future<StatementResult> resultFuture =
         backendConnection.execute(parsedStatement, statement, Function.identity());
     backendConnection.flush();
@@ -449,8 +471,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             connection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            EMPTY_LOCAL_STATEMENTS);
+            () -> EMPTY_LOCAL_STATEMENTS);
     Future<StatementResult> resultFuture1 =
         backendConnection.execute(parsedStatement1, statement1, Function.identity());
     backendConnection.execute(parsedStatement2, statement2, Function.identity());
@@ -474,7 +497,11 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
-            DatabaseId.of("p", "i", "d"), connection, options, EMPTY_LOCAL_STATEMENTS);
+            DatabaseId.of("p", "i", "d"),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     backendConnection.execute(parsedStatement, statement, Function.identity());
     backendConnection.flush();
@@ -517,7 +544,11 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
-            DatabaseId.of("p", "i", "d"), connection, options, EMPTY_LOCAL_STATEMENTS);
+            DatabaseId.of("p", "i", "d"),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     backendConnection.execute(parsedStatement, statement, Function.identity());
     backendConnection.flush();
@@ -537,8 +568,9 @@ public class BackendConnectionTest {
         new BackendConnection(
             DatabaseId.of("p", "i", "d"),
             connection,
+            () -> WellKnownClient.UNSPECIFIED,
             mock(OptionsMetadata.class),
-            EMPTY_LOCAL_STATEMENTS);
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     backendConnection.execute(parsedStatement, statement, Function.identity());
     backendConnection.flush();
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatementTest.java
index 3c32d859..f118d104 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/CopyToStatementTest.java
@@ -15,6 +15,7 @@
 package com.google.cloud.spanner.pgadapter.statements;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 import static org.mockito.Mockito.when;
 
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
@@ -91,4 +92,14 @@ public class CopyToStatementTest {
         new CopyToStatement(connectionHandler, options, "", parsedCopyStatement);
     assertEquals("\n", statement.getCsvFormat().getRecordSeparator());
   }
+
+  @Test
+  public void testCopyQuery() {
+    String query =
+        "select executed_at, workload, threads, batch_size, operation_count, round(read_avg/1000) as read_avg, round(read_p95/1000) as read_p95 from run where true and workload='a' and threads=20 and deployment='java_uds' order by executed_at desc limit 100";
+    ParsedCopyStatement parsedCopyStatement =
+        CopyStatement.parse(String.format("copy (%s) to stdout\n", query));
+    assertNotNull(parsedCopyStatement);
+    assertEquals(parsedCopyStatement.query, query);
+  }
 }
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 571d33dc..9e0a67fd 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
@@ -53,6 +53,7 @@ import com.google.cloud.spanner.pgadapter.error.SQLState;
 import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.session.SessionState;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage;
@@ -195,7 +196,11 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
-            connectionHandler.getDatabaseId(), connection, options, EMPTY_LOCAL_STATEMENTS);
+            connectionHandler.getDatabaseId(),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     assertFalse(intermediateStatement.isExecuted());
     assertEquals("UPDATE", intermediateStatement.getCommand());
@@ -281,7 +286,11 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
-            connectionHandler.getDatabaseId(), connection, options, EMPTY_LOCAL_STATEMENTS);
+            connectionHandler.getDatabaseId(),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.executeAsync(backendConnection);
     backendConnection.flush();
@@ -316,7 +325,11 @@ public class StatementTest {
             .build();
     BackendConnection backendConnection =
         new BackendConnection(
-            connectionHandler.getDatabaseId(), connection, options, EMPTY_LOCAL_STATEMENTS);
+            connectionHandler.getDatabaseId(),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     IntermediatePreparedStatement intermediateStatement =
         new IntermediatePreparedStatement(
@@ -419,7 +432,11 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
-            connectionHandler.getDatabaseId(), connection, options, EMPTY_LOCAL_STATEMENTS);
+            connectionHandler.getDatabaseId(),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     when(connection.execute(Statement.of(sqlStatement)))
         .thenThrow(
@@ -481,7 +498,11 @@ public class StatementTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
-            DatabaseId.of("p", "i", "d"), connection, options, ImmutableList.of());
+            DatabaseId.of("p", "i", "d"),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            ImmutableList::of);
     statement.executeAsync(backendConnection);
 
     ExecutorService executor = Executors.newSingleThreadExecutor();
@@ -549,7 +570,11 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
-            connectionHandler.getDatabaseId(), connection, options, EMPTY_LOCAL_STATEMENTS);
+            connectionHandler.getDatabaseId(),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            () -> EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.executeAsync(backendConnection);
 
@@ -570,7 +595,11 @@ public class StatementTest {
     setupQueryInformationSchemaResults();
     BackendConnection backendConnection =
         new BackendConnection(
-            DatabaseId.of("p", "i", "d"), connection, options, ImmutableList.of());
+            DatabaseId.of("p", "i", "d"),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            ImmutableList::of);
 
     byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/batch-size-test.txt"));
 
@@ -619,7 +648,11 @@ public class StatementTest {
     setupQueryInformationSchemaResults();
     BackendConnection backendConnection =
         new BackendConnection(
-            DatabaseId.of("p", "i", "d"), connection, options, ImmutableList.of());
+            DatabaseId.of("p", "i", "d"),
+            connection,
+            () -> WellKnownClient.UNSPECIFIED,
+            options,
+            ImmutableList::of);
 
     byte[] payload = "1\t'one'\n2".getBytes();
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java
index 8b6fd843..790258ed 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java
@@ -24,9 +24,12 @@ import static org.mockito.Mockito.when;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.ProxyServer;
+import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.cloud.spanner.pgadapter.statements.local.ListDatabasesStatement;
 import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
+import com.google.cloud.spanner.pgadapter.wireoutput.NoticeResponse;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import java.util.ArrayList;
@@ -37,6 +40,39 @@ import org.junit.runners.JUnit4;
 
 @RunWith(JUnit4.class)
 public class ClientAutoDetectorTest {
+
+  @Test
+  public void testUnspecified() {
+    WellKnownClient.UNSPECIFIED.reset();
+
+    assertEquals(
+        WellKnownClient.UNSPECIFIED,
+        ClientAutoDetector.detectClient(ImmutableList.of(), ImmutableMap.of()));
+    assertEquals(
+        WellKnownClient.UNSPECIFIED,
+        ClientAutoDetector.detectClient(
+            ImmutableList.of("some-param"), ImmutableMap.of("some-param", "some-value")));
+
+    ConnectionHandler connection = mock(ConnectionHandler.class);
+    ProxyServer server = mock(ProxyServer.class);
+    OptionsMetadata options = mock(OptionsMetadata.class);
+    when(server.getOptions()).thenReturn(options);
+    when(connection.getServer()).thenReturn(server);
+
+    when(options.useDefaultLocalStatements()).thenReturn(true);
+    assertEquals(
+        DEFAULT_LOCAL_STATEMENTS, WellKnownClient.UNSPECIFIED.getLocalStatements(connection));
+
+    when(options.useDefaultLocalStatements()).thenReturn(false);
+    assertEquals(ImmutableList.of(), WellKnownClient.UNSPECIFIED.getLocalStatements(connection));
+
+    assertEquals(
+        ImmutableList.of(), WellKnownClient.UNSPECIFIED.createStartupNoticeResponses(connection));
+    assertEquals(
+        ImmutableList.of(), WellKnownClient.UNSPECIFIED.getErrorHints(mock(PGException.class)));
+    assertEquals(ImmutableMap.of(), WellKnownClient.UNSPECIFIED.getDefaultParameters());
+  }
+
   @Test
   public void testJdbc() {
     // The JDBC driver will **always** include these startup parameters in exactly this order.
@@ -291,4 +327,52 @@ public class ClientAutoDetectorTest {
                         + "SELECT ns.nspname, t.oid, t.typname, t.typtype, t.typnotnull, t.elemtypoid\n"),
                 Statement.of("SELECT version();"))));
   }
+
+  @Test
+  public void testPgbench() {
+    // pgbench always includes itself as the application_name.
+    assertEquals(
+        WellKnownClient.PGBENCH,
+        ClientAutoDetector.detectClient(
+            ImmutableList.of("application_name"), ImmutableMap.of("application_name", "pgbench")));
+    assertEquals(
+        WellKnownClient.PGBENCH,
+        ClientAutoDetector.detectClient(
+            ImmutableList.of("application_name", "some-param"),
+            ImmutableMap.of(
+                "application_name", "pgbench",
+                "some-param", "some-value")));
+    assertEquals(
+        WellKnownClient.PGBENCH,
+        ClientAutoDetector.detectClient(
+            ImmutableList.of("some-param1", "application_name", "some-param2"),
+            ImmutableMap.of(
+                "some-param1", "some-value1",
+                "application_name", "pgbench",
+                "some-param2", "some-value2")));
+    assertNotEquals(
+        WellKnownClient.PGBENCH,
+        ClientAutoDetector.detectClient(ImmutableList.of(), ImmutableMap.of()));
+    assertNotEquals(
+        WellKnownClient.PGBENCH,
+        ClientAutoDetector.detectClient(
+            ImmutableList.of("some-param1"), ImmutableMap.of("some-param1", "some-value1")));
+    assertNotEquals(
+        WellKnownClient.PGBENCH,
+        ClientAutoDetector.detectClient(
+            ImmutableList.of("application_name"), ImmutableMap.of("application_name", "JDBC")));
+    assertNotEquals(
+        WellKnownClient.PGBENCH,
+        ClientAutoDetector.detectClient(
+            ImmutableList.of("application_name"), ImmutableMap.of("application_name", "PGBENCH")));
+
+    WellKnownClient.PGBENCH.reset();
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(mock(ConnectionMetadata.class));
+    ImmutableList<NoticeResponse> startupNotices =
+        WellKnownClient.PGBENCH.createStartupNoticeResponses(connectionHandler);
+    assertEquals(1, startupNotices.size());
+    assertEquals("Detected connection from pgbench", startupNotices.get(0).getMessage());
+    assertEquals(ClientAutoDetector.PGBENCH_USAGE_HINT + "\n", startupNotices.get(0).getHint());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponseTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponseTest.java
index 443c1ee6..9dc39244 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponseTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/wireoutput/ErrorResponseTest.java
@@ -16,10 +16,16 @@ package com.google.cloud.spanner.pgadapter.wireoutput;
 
 import static com.google.cloud.spanner.pgadapter.wireoutput.ErrorResponse.calculateLength;
 import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.error.PGException;
 import com.google.cloud.spanner.pgadapter.error.SQLState;
 import com.google.cloud.spanner.pgadapter.error.Severity;
+import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector;
+import com.google.cloud.spanner.pgadapter.utils.ClientAutoDetector.WellKnownClient;
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import org.junit.Test;
@@ -44,7 +50,8 @@ public class ErrorResponseTest {
             PGException.newBuilder("test message")
                 .setSeverity(Severity.ERROR)
                 .setSQLState(SQLState.RaiseException)
-                .build()));
+                .build(),
+            WellKnownClient.UNSPECIFIED));
     assertEquals(
         4
             + "test message".length()
@@ -61,16 +68,59 @@ public class ErrorResponseTest {
                 .setSeverity(Severity.ERROR)
                 .setSQLState(SQLState.RaiseException)
                 .setHints("test hint")
-                .build()));
+                .build(),
+            WellKnownClient.UNSPECIFIED));
+
+    assertEquals(
+        4
+            + "test message".length()
+            + /* Field header + null terminator */ 2
+            + "ERROR".length()
+            + 2
+            + "P0001".length()
+            + 2
+            + ClientAutoDetector.PGBENCH_USAGE_HINT.length()
+            + 2
+            + 1,
+        calculateLength(
+            PGException.newBuilder("test message")
+                .setSeverity(Severity.ERROR)
+                .setSQLState(SQLState.RaiseException)
+                .build(),
+            WellKnownClient.PGBENCH));
+    assertEquals(
+        4
+            + "test message".length()
+            + /* Field header + null terminator */ 2
+            + "ERROR".length()
+            + 2
+            + "P0001".length()
+            + 2
+            + "test hint\n".length()
+            + ClientAutoDetector.PGBENCH_USAGE_HINT.length()
+            + 2
+            + 1,
+        calculateLength(
+            PGException.newBuilder("test message")
+                .setSeverity(Severity.ERROR)
+                .setSQLState(SQLState.RaiseException)
+                .setHints("test hint")
+                .build(),
+            WellKnownClient.PGBENCH));
   }
 
   @Test
   public void testSendPayload() throws Exception {
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     DataOutputStream output = new DataOutputStream(out);
+    ConnectionMetadata metadata = mock(ConnectionMetadata.class);
+    when(metadata.getOutputStream()).thenReturn(output);
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    when(connectionHandler.getWellKnownClient()).thenReturn(WellKnownClient.UNSPECIFIED);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(metadata);
     ErrorResponse response =
         new ErrorResponse(
-            output,
+            connectionHandler,
             PGException.newBuilder("test message")
                 .setSeverity(Severity.ERROR)
                 .setSQLState(SQLState.RaiseException)
@@ -86,9 +136,14 @@ public class ErrorResponseTest {
   public void testSendPayloadWithHint() throws Exception {
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     DataOutputStream output = new DataOutputStream(out);
+    ConnectionMetadata metadata = mock(ConnectionMetadata.class);
+    when(metadata.getOutputStream()).thenReturn(output);
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    when(connectionHandler.getWellKnownClient()).thenReturn(WellKnownClient.UNSPECIFIED);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(metadata);
     ErrorResponse response =
         new ErrorResponse(
-            output,
+            connectionHandler,
             PGException.newBuilder("test message")
                 .setSeverity(Severity.ERROR)
                 .setSQLState(SQLState.RaiseException)
@@ -102,4 +157,63 @@ public class ErrorResponseTest {
         "Length: 51, Error Message: test message, Hints: test hint\n" + "line 2",
         response.getPayloadString());
   }
+
+  @Test
+  public void testSendPayloadWithClientHint() throws Exception {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    DataOutputStream output = new DataOutputStream(out);
+    ConnectionMetadata metadata = mock(ConnectionMetadata.class);
+    when(metadata.getOutputStream()).thenReturn(output);
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    when(connectionHandler.getWellKnownClient()).thenReturn(WellKnownClient.PGBENCH);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(metadata);
+    ErrorResponse response =
+        new ErrorResponse(
+            connectionHandler,
+            PGException.newBuilder("test message")
+                .setSeverity(Severity.ERROR)
+                .setSQLState(SQLState.RaiseException)
+                .build());
+
+    response.sendPayload();
+
+    assertEquals(
+        String.format(
+            "SERROR\0CP0001\0Mtest message\0H%s\0\0", ClientAutoDetector.PGBENCH_USAGE_HINT),
+        out.toString());
+    assertEquals(
+        "Length: 148, Error Message: test message, Hints: " + ClientAutoDetector.PGBENCH_USAGE_HINT,
+        response.getPayloadString());
+  }
+
+  @Test
+  public void testSendPayloadWithErrorAndClientHint() throws Exception {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    DataOutputStream output = new DataOutputStream(out);
+    ConnectionMetadata metadata = mock(ConnectionMetadata.class);
+    when(metadata.getOutputStream()).thenReturn(output);
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    when(connectionHandler.getWellKnownClient()).thenReturn(WellKnownClient.PGBENCH);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(metadata);
+    ErrorResponse response =
+        new ErrorResponse(
+            connectionHandler,
+            PGException.newBuilder("test message")
+                .setSeverity(Severity.ERROR)
+                .setSQLState(SQLState.RaiseException)
+                .setHints("test hint\nline 2")
+                .build());
+
+    response.sendPayload();
+
+    assertEquals(
+        String.format(
+            "SERROR\0CP0001\0Mtest message\0Htest hint\nline 2\n%s\0\0",
+            ClientAutoDetector.PGBENCH_USAGE_HINT),
+        out.toString());
+    assertEquals(
+        "Length: 165, Error Message: test message, Hints: test hint\nline 2\n"
+            + ClientAutoDetector.PGBENCH_USAGE_HINT,
+        response.getPayloadString());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/wireoutput/NoticeResponseTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/wireoutput/NoticeResponseTest.java
new file mode 100644
index 00000000..3a4b628b
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/wireoutput/NoticeResponseTest.java
@@ -0,0 +1,69 @@
+// Copyright 2023 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.wireoutput;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.cloud.spanner.pgadapter.wireoutput.NoticeResponse.NoticeSeverity;
+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 NoticeResponseTest {
+
+  @Test
+  public void testBasics() {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    DataOutputStream output = new DataOutputStream(out);
+    NoticeResponse response =
+        new NoticeResponse(output, SQLState.Success, NoticeSeverity.NOTICE, "test notice", null);
+
+    assertEquals('N', response.getIdentifier());
+    assertEquals("Notice", response.getMessageName());
+  }
+
+  @Test
+  public void testSendPayload() throws Exception {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    DataOutputStream output = new DataOutputStream(out);
+    NoticeResponse response =
+        new NoticeResponse(output, SQLState.Success, NoticeSeverity.NOTICE, "test notice", null);
+    response.sendPayload();
+
+    assertEquals("SNOTICE\0C00000\0Mtest notice\0\0", out.toString());
+    assertEquals(
+        "Length: 33, Severity: NOTICE, Notice Message: test notice, Hint: ",
+        response.getPayloadString());
+  }
+
+  @Test
+  public void testSendPayloadWithHint() throws Exception {
+    ByteArrayOutputStream out = new ByteArrayOutputStream();
+    DataOutputStream output = new DataOutputStream(out);
+    NoticeResponse response =
+        new NoticeResponse(
+            output, SQLState.Success, NoticeSeverity.NOTICE, "test notice", "some hint");
+    response.sendPayload();
+
+    assertEquals("SNOTICE\0C00000\0Mtest notice\0Hsome hint\0\0", out.toString());
+    assertEquals(
+        "Length: 44, Severity: NOTICE, Notice Message: test notice, Hint: some hint",
+        response.getPayloadString());
+  }
+}
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 43cc6e71..8e6f4df6 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
@@ -986,6 +986,7 @@ public class ProtocolTest {
         PGExceptionFactory.newPGException("test error", SQLState.SyntaxError);
     when(intermediatePortalStatement.hasException()).thenReturn(true);
     when(intermediatePortalStatement.getException()).thenReturn(testException);
+    when(connectionHandler.getWellKnownClient()).thenReturn(WellKnownClient.UNSPECIFIED);
     when(connectionHandler.getPortal(anyString())).thenReturn(intermediatePortalStatement);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
@@ -1890,6 +1891,7 @@ public class ProtocolTest {
 
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+    when(connectionHandler.getWellKnownClient()).thenReturn(WellKnownClient.UNSPECIFIED);
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionHandler.getStatus()).thenReturn(ConnectionStatus.AUTHENTICATED);
     doCallRealMethod().when(connectionHandler).increaseInvalidMessageCount();