Repository: GoogleCloudPlatform/pgadapter
Pull request #1406 (2024-02-15 14:28:39)

perf: support virtual threads on Java 21 and higher
Adds support for using virtual threads for both PGAdapter ConnectionHandlers, Spanner Connections, and for the gRPC transport threads. Using virtual threads instead of platform threads can reduce overall resource consumption, especially for PGAdapter instances that handle a very large number of incoming connections.


Patch:
diff --git a/.github/workflows/integration-emulator.yaml b/.github/workflows/integration-emulator.yaml
index ad5c1a6a..c2fec81e 100644
--- a/.github/workflows/integration-emulator.yaml
+++ b/.github/workflows/integration-emulator.yaml
@@ -34,13 +34,17 @@ jobs:
         ports:
           - 9010:9010
           - 9020:9020
+    strategy:
+      fail-fast: false
+      matrix:
+        java: [11, 21]
     steps:
       - uses: actions/checkout@v4
       - name: Setup Java
         uses: actions/setup-java@v4
         with:
           distribution: zulu
-          java-version: 17
+          java-version: ${{matrix.java}}
       - run: java -version
       - name: Setup Go
         uses: actions/setup-go@v5
@@ -65,7 +69,7 @@ jobs:
         run: |
           psql -h /pg -U postgres -c "CREATE DATABASE pgadapter"
       - name: Run integration tests
-        run: mvn verify -B -Dclirr.skip=true -DskipUnits=true -DskipITs=false
+        run: mvn -Dpgadapter.use_virtual_threads=true -Dpgadapter.use_virtual_grpc_transport_threads=true verify -B -Dclirr.skip=true -DskipUnits=true -DskipITs=false
         env:
           POSTGRES_HOST: /pg
           POSTGRES_PORT: 5432
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 37bc0add..f9662c78 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -80,7 +80,7 @@ jobs:
         uses: actions/setup-java@v4
         with:
           distribution: zulu
-          java-version: 17
+          java-version: 21
       - run: java -version
       - name: Setup Go
         uses: actions/setup-go@v5
@@ -106,6 +106,8 @@ jobs:
           psql -h /pg -U postgres -c "CREATE DATABASE pgadapter"
       - name: Run unit tests
         run: mvn test -B -Ptest-all
+      - name: Run unit tests with virtual threads
+        run: mvn test -Dpgadapter.use_virtual_threads=true -Dpgadapter.use_virtual_grpc_transport_threads=true -B -Ptest-all
       - name: Auth
         uses: google-github-actions/auth@v2
         with:
diff --git a/.github/workflows/units.yaml b/.github/workflows/units.yaml
index 2591734a..75b2771e 100644
--- a/.github/workflows/units.yaml
+++ b/.github/workflows/units.yaml
@@ -9,7 +9,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        java: [8, 11, 17]
+        java: [8, 11, 17, 21]
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-java@v4
@@ -32,7 +32,7 @@ jobs:
         with:
           ruby-version: '3.0'
           bundler-cache: true
-      - run: mvn -B test -Ptest-all -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn
+      - run: mvn -Dpgadapter.use_virtual_threads=true -Dpgadapter.use_virtual_grpc_transport_threads=true -B test -Ptest-all -Dorg.slf4j.simpleLogger.log.org.apache.maven.cli.transfer.Slf4jMavenTransferListener=warn
   windows:
     runs-on: windows-latest
     env:
diff --git a/samples/golang/gorm/go.mod b/samples/golang/gorm/go.mod
index 17eb1932..02eb22d1 100644
--- a/samples/golang/gorm/go.mod
+++ b/samples/golang/gorm/go.mod
@@ -7,8 +7,8 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/jackc/pgtype v1.14.2
 	github.com/shopspring/decimal v1.3.1
-	github.com/testcontainers/testcontainers-go v0.27.0
-	github.com/testcontainers/testcontainers-go/modules/postgres v0.27.0
+	github.com/testcontainers/testcontainers-go v0.28.0
+	github.com/testcontainers/testcontainers-go/modules/postgres v0.28.0
 	gorm.io/datatypes v1.2.0
 	gorm.io/driver/postgres v1.5.6
 	gorm.io/gorm v1.25.7
@@ -26,7 +26,7 @@ require (
 	github.com/containerd/log v0.1.0 // indirect
 	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/distribution/reference v0.5.0 // indirect
-	github.com/docker/docker v25.0.1+incompatible // indirect
+	github.com/docker/docker v25.0.2+incompatible // indirect
 	github.com/docker/go-connections v0.5.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
@@ -39,7 +39,8 @@ require (
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
-	github.com/jackc/pgx/v5 v5.4.3 // indirect
+	github.com/jackc/pgx/v5 v5.5.3 // indirect
+	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect
 	github.com/jinzhu/now v1.1.5 // indirect
 	github.com/klauspost/compress v1.17.4 // indirect
@@ -68,6 +69,7 @@ require (
 	golang.org/x/exp v0.0.0-20240119083558-1b970713d09a // indirect
 	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/oauth2 v0.16.0 // indirect
+	golang.org/x/sync v0.6.0 // indirect
 	golang.org/x/sys v0.16.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/tools v0.17.0 // indirect
diff --git a/samples/golang/gorm/go.sum b/samples/golang/gorm/go.sum
index ebf16b11..5b6917f2 100644
--- a/samples/golang/gorm/go.sum
+++ b/samples/golang/gorm/go.sum
@@ -35,8 +35,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/distribution/reference v0.5.0 h1:/FUIFXtfc/x2gpa5/VGfiGLuOIdYa1t65IKK2OFGvA0=
 github.com/distribution/reference v0.5.0/go.mod h1:BbU0aIcezP1/5jX/8MP0YiH4SdvB5Y4f/wlDRiLyi3E=
-github.com/docker/docker v25.0.1+incompatible h1:k5TYd5rIVQRSqcTwCID+cyVA0yRg86+Pcrz1ls0/frA=
-github.com/docker/docker v25.0.1+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
+github.com/docker/docker v25.0.2+incompatible h1:/OaKeauroa10K4Nqavw4zlhcDq/WBcPMc5DbjOGgozY=
+github.com/docker/docker v25.0.2+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/docker/go-connections v0.5.0 h1:USnMq7hx7gwdVZq1L49hLXaFtUdTADjXGp+uj1Br63c=
 github.com/docker/go-connections v0.5.0/go.mod h1:ov60Kzw0kKElRwhNs9UlUHAE/F9Fe6GLaXnqyDdmEXc=
 github.com/docker/go-units v0.5.0 h1:69rxXcBk27SvSaaxTtLh/8llcHD8vYHT7WSdRZ/jvr4=
@@ -119,11 +119,13 @@ github.com/jackc/pgx/v4 v4.0.0-20190421002000-1b8f0016e912/go.mod h1:no/Y67Jkk/9
 github.com/jackc/pgx/v4 v4.0.0-pre1.0.20190824185557-6972a5742186/go.mod h1:X+GQnOEnf1dqHGpw7JmHqHc1NxDoalibchSk9/RWuDc=
 github.com/jackc/pgx/v4 v4.12.1-0.20210724153913-640aa07df17c h1:Dznn52SgVIVst9UyOT9brctYUgxs+CvVfPaC3jKrA50=
 github.com/jackc/pgx/v4 v4.12.1-0.20210724153913-640aa07df17c/go.mod h1:1QD0+tgSXP7iUjYm9C1NxKhny7lq6ee99u/z+IHFcgs=
-github.com/jackc/pgx/v5 v5.4.3 h1:cxFyXhxlvAifxnkKKdlxv8XqUf59tDlYjnV5YYfsJJY=
-github.com/jackc/pgx/v5 v5.4.3/go.mod h1:Ig06C2Vu0t5qXC60W8sqIthScaEnFvojjj9dSljmHRA=
+github.com/jackc/pgx/v5 v5.5.3 h1:Ces6/M3wbDXYpM8JyyPD57ivTtJACFZJd885pdIaV2s=
+github.com/jackc/pgx/v5 v5.5.3/go.mod h1:ez9gk+OAat140fv9ErkZDYFWmXLfV+++K0uAOiwgm1A=
 github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
+github.com/jackc/puddle/v2 v2.2.1 h1:RhxXJtFG022u4ibrCSMSiu5aOq1i77R3OHKNJj77OAk=
+github.com/jackc/puddle/v2 v2.2.1/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
 github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
 github.com/jinzhu/now v1.1.4/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
@@ -214,10 +216,10 @@ github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
-github.com/testcontainers/testcontainers-go v0.27.0 h1:IeIrJN4twonTDuMuBNQdKZ+K97yd7VrmNGu+lDpYcDk=
-github.com/testcontainers/testcontainers-go v0.27.0/go.mod h1:+HgYZcd17GshBUZv9b+jKFJ198heWPQq3KQIp2+N+7U=
-github.com/testcontainers/testcontainers-go/modules/postgres v0.27.0 h1:gbA/HYjBIwOwhE/t4p3kIprfI0qsxCk+YVW7P9XFOus=
-github.com/testcontainers/testcontainers-go/modules/postgres v0.27.0/go.mod h1:VFrFKUUgET2hNXStdtaC7uOIJWviFUrixhKeaVw/4F4=
+github.com/testcontainers/testcontainers-go v0.28.0 h1:1HLm9qm+J5VikzFDYhOd+Zw12NtOl+8drH2E8nTY1r8=
+github.com/testcontainers/testcontainers-go v0.28.0/go.mod h1:COlDpUXbwW3owtpMkEB1zo9gwb1CoKVKlyrVPejF4AU=
+github.com/testcontainers/testcontainers-go/modules/postgres v0.28.0 h1:ff0s4JdYIdNAVSi/SrpN2Pdt1f+IjIw3AKjbHau8Un4=
+github.com/testcontainers/testcontainers-go/modules/postgres v0.28.0/go.mod h1:fXgcYpbyrduNdiz2qRZuYkmvqLnEqsjbQiBNYH1ystI=
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
 github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=
 github.com/tklauser/go-sysconf v0.3.13/go.mod h1:zwleP4Q4OehZHGn4CYZDipCgg9usW5IJePewFCGVEa0=
diff --git a/samples/python/sqlalchemy2-sample/README.md b/samples/python/sqlalchemy2-sample/README.md
index 56233fbc..1ff70a02 100644
--- a/samples/python/sqlalchemy2-sample/README.md
+++ b/samples/python/sqlalchemy2-sample/README.md
@@ -96,7 +96,7 @@ Cloud Spanner supports the following data types in combination with `SQLAlchemy
 | date                                   | Date                    |
 | bytea                                  | LargeBinary             |
 | jsonb                                  | JSONB                   |
-
+| Arrays                                 | Column()                |
 
 ## Limitations
 The following limitations are currently known:
@@ -106,7 +106,6 @@ The following limitations are currently known:
 | Creating and Dropping Tables   | Cloud Spanner does not support the full PostgreSQL DDL dialect. Automated creation of tables using `SQLAlchemy` is therefore not supported.                                                                                                                         |
 | metadata.reflect()             | Cloud Spanner does not support all PostgreSQL `pg_catalog` tables. Using `metadata.reflect()` to get the current objects in the database is therefore not supported.                                                                                                |
 | DDL Transactions               | Cloud Spanner does not support DDL statements in a transaction. Add `?options=-c spanner.ddl_transaction_mode=AutocommitExplicitTransaction` to your connection string to automatically convert DDL transactions to [non-atomic DDL batches](../../../docs/ddl.md). |
-| Generated primary keys         | Manually assign a value to the primary key column in your code. The recommended primary key type is a random UUID. Sequences / SERIAL / IDENTITY columns are currently not supported.                                                                               |
 | INSERT ... ON CONFLICT         | `INSERT ... ON CONFLICT` is not supported.                                                                                                                                                                                                                          |
 | SAVEPOINT                      | Rolling back to a `SAVEPOINT` can fail if the transaction contained at least one query that called a volatile function.                                                                                                                                             |
 | SELECT ... FOR UPDATE          | `SELECT ... FOR UPDATE` is not supported.                                                                                                                                                                                                                           |
@@ -117,21 +116,38 @@ The following limitations are currently known:
 | Other drivers than psycopg 3.x | PGAdapter does not support using SQLAlchemy 2.x with any other drivers than `psycopg 3.x`.                                                                                                                                                                          |
 
 ### Generated Primary Keys
-Generated primary keys are not supported and should be replaced with primary key definitions that
-are manually assigned. See https://cloud.google.com/spanner/docs/schema-design#primary-key-prevent-hotspots
-for more information on choosing a good primary key. This sample uses random UUIDs that are generated
-by the client and stored as strings for primary keys.
+Generated primary keys can be used in combination with a bit-reversed sequence.
 
-```python
-from uuid import uuid4
+The `TicketSale` model in this sample application uses an auto-generated primary key that is
+generated from a bit-reversed sequence:
+1. See [model.py](model.py) for the model definition.
+2. See [create_data_model.sql](create_data_model.sql) for the sequence and table definition.
+
+See https://cloud.google.com/spanner/docs/primary-key-default-value#bit-reversed-sequence for more
+information on bit-reversed sequences in Cloud Spanner.
+
+#### Example Mapping for Generated Primary Key using a Bit-Reversed Sequence
+
+Python model definition:
 
+```python
 class Singer(Base):
-  id = Column(String, primary_key=True)
+  id = Column(Integer, primary_key=True)
   name = Column(String(100))
 
-singer = Singer(
-  id="{}".format(uuid4()),
-  name="Alice")
+singer = Singer(name="Alice")
+session.add(singer)
+session.commit()
+```
+
+Sequence and table definition:
+
+```sql
+create sequence if not exists singers_seq bit_reversed_positive;
+create table if not exists singers (
+  id   bigint not null primary key default nextval('singers_seq'),
+  name varchar not null
+);
 ```
 
 ### ON CONFLICT Clauses
diff --git a/samples/python/sqlalchemy2-sample/create_data_model.sql b/samples/python/sqlalchemy2-sample/create_data_model.sql
index c8e173b3..7ca401e8 100644
--- a/samples/python/sqlalchemy2-sample/create_data_model.sql
+++ b/samples/python/sqlalchemy2-sample/create_data_model.sql
@@ -61,4 +61,25 @@ create table if not exists concerts (
     constraint chk_end_time_after_start_time check (end_time > start_time)
 );
 
+-- Create a bit-reversed sequence that will be used to generate identifiers for the ticket_sales table.
+-- See also https://cloud.google.com/spanner/docs/reference/postgresql/data-definition-language#create_sequence
+-- Note that the 'bit_reversed_positive' keyword is required for Spanner.
+create sequence if not exists ticket_sale_seq
+    bit_reversed_positive
+    skip range 1 1000
+    start counter with 50000
+;
+
+create table if not exists ticket_sales (
+    id bigint not null primary key default nextval('ticket_sale_seq'),
+    version_id       int not null,
+    concert_id       varchar not null,
+    customer_name    varchar not null,
+    price            decimal not null,
+    seats            text[],
+    created_at       timestamptz,
+    updated_at       timestamptz,
+    constraint fk_ticket_sales_concerts foreign key (concert_id) references concerts (id)
+);
+
 run batch;
diff --git a/samples/python/sqlalchemy2-sample/drop_data_model.sql b/samples/python/sqlalchemy2-sample/drop_data_model.sql
index b57ea2ad..bbc6feea 100644
--- a/samples/python/sqlalchemy2-sample/drop_data_model.sql
+++ b/samples/python/sqlalchemy2-sample/drop_data_model.sql
@@ -1,6 +1,8 @@
 -- Executing the schema drop in a batch will improve execution speed.
 start batch ddl;
 
+drop table if exists ticket_sales;
+drop sequence if exists ticket_sale_seq;
 drop table if exists concerts;
 drop table if exists venues;
 drop table if exists tracks;
diff --git a/samples/python/sqlalchemy2-sample/model.py b/samples/python/sqlalchemy2-sample/model.py
index b920cac0..f49c490f 100644
--- a/samples/python/sqlalchemy2-sample/model.py
+++ b/samples/python/sqlalchemy2-sample/model.py
@@ -16,30 +16,26 @@
 from sqlalchemy import Column, Integer, String, Boolean, LargeBinary, Float, \
   Numeric, DateTime, Date, FetchedValue, ForeignKey, ColumnDefault
 from sqlalchemy.orm import registry, relationship
-from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.dialects.postgresql import JSONB, ARRAY
 from datetime import timezone, datetime
 
 mapper_registry = registry()
 Base = mapper_registry.generate_base()
 
 
-"""
- BaseMixin contains properties that are common to all models in this sample. All
- models use a string column that contains a client-side generated UUID as the
- primary key.
- The created_at and updated_at properties are automatically filled with the
- current client system time when a model is created or updated. 
-"""
-
-
 def format_timestamp(timestamp: datetime) -> str:
   return timestamp.astimezone(timezone.utc).isoformat() if timestamp else None
 
 
+"""
+ BaseMixin contains properties that are common to all models in this sample.
+ 
+ The created_at and updated_at properties are automatically filled with the
+ current client system time when a model is created or updated. 
+"""
 class BaseMixin(object):
   __prepare_statements__ = None
 
-  id = Column(String, primary_key=True)
   version_id = Column(Integer, nullable=False)
   created_at = Column(DateTime(timezone=True),
                       # We need to explicitly format the timestamps with a
@@ -53,7 +49,24 @@ class BaseMixin(object):
   __mapper_args__ = {"version_id_col": version_id}
 
 
-class Singer(BaseMixin, Base):
+"""
+ BaseUuidMixin is the base class for all tables that use a client-side
+ generated UUID as the primary key.
+""" 
+class BaseUuidMixin(BaseMixin):
+  id = Column(String, primary_key=True)
+
+
+"""
+ BaseInt64Mixin is the base class for all tables that use an auto-generated
+ INT64 as the primary key. The value is generated by a bit-reversed sequence
+ in the database.
+"""
+class BaseInt64Mixin(BaseMixin):
+  id = Column(Integer, primary_key=True)
+
+
+class Singer(BaseUuidMixin, Base):
   __tablename__ = "singers"
 
   first_name = Column(String(100))
@@ -82,7 +95,7 @@ class Singer(BaseMixin, Base):
     )
 
 
-class Album(BaseMixin, Base):
+class Album(BaseUuidMixin, Base):
   __tablename__ = "albums"
 
   title = Column(String(200))
@@ -112,7 +125,7 @@ class Album(BaseMixin, Base):
     )
 
 
-class Track(BaseMixin, Base):
+class Track(BaseUuidMixin, Base):
   __tablename__ = "tracks"
 
   id = Column(String, ForeignKey("albums.id"), primary_key=True)
@@ -134,7 +147,7 @@ class Track(BaseMixin, Base):
     )
 
 
-class Venue(BaseMixin, Base):
+class Venue(BaseUuidMixin, Base):
   __tablename__ = "venues"
 
   name = Column(String(200))
@@ -152,7 +165,7 @@ class Venue(BaseMixin, Base):
     )
 
 
-class Concert(BaseMixin, Base):
+class Concert(BaseUuidMixin, Base):
   __tablename__ = "concerts"
 
   name = Column(String(200))
@@ -162,6 +175,7 @@ class Concert(BaseMixin, Base):
   singer = relationship("Singer")
   start_time = Column(DateTime(timezone=True))
   end_time = Column(DateTime(timezone=True))
+  ticket_sales = relationship("TicketSale", back_populates="concert")
 
   def __repr__(self):
     return (
@@ -176,3 +190,33 @@ class Concert(BaseMixin, Base):
       f"updated_at={format_timestamp(self.updated_at)!r}"
       f")"
     )
+
+
+class TicketSale(BaseInt64Mixin, Base):
+  __tablename__ = "ticket_sales"
+
+  concert_id = Column(String, ForeignKey("concerts.id"))
+  concert = relationship("Concert")
+  customer_name = Column(String)
+  price = Column(Numeric)
+  # "seats" has type text[] in the database. By not specifying any type here,
+  # we ensure that SQLAlchemy just sends an untyped string value to the
+  # database. If we were to configure the type to be ARRAY(String) here, then
+  # we would get an error 'Array casting / coercion is not supported', as
+  # SQLAlchemy would try to cast the value to a VARCHAR[].
+  # TODO: Use this mapping when array casting/coercion is supported.
+  #       seats = Column(ARRAY(String))
+  seats = Column()
+
+  def __repr__(self):
+    return (
+      f"ticket_sales("
+      f"id={self.id!r},"
+      f"concert={self.concert!r},"
+      f"customer_name={self.customer_name!r},"
+      f"price={self.price!r},"
+      f"seats={self.seats!r},"
+      f"created_at={format_timestamp(self.created_at)!r},"
+      f"updated_at={format_timestamp(self.updated_at)!r}"
+      f")"
+    )
diff --git a/samples/python/sqlalchemy2-sample/sample.py b/samples/python/sqlalchemy2-sample/sample.py
index 249f3e0b..1685eb40 100644
--- a/samples/python/sqlalchemy2-sample/sample.py
+++ b/samples/python/sqlalchemy2-sample/sample.py
@@ -12,11 +12,12 @@
  See the License for the specific language governing permissions and
  limitations under the License.
 """
+from random import random
 
 from connect import create_test_engine
 from sqlalchemy.orm import Session, joinedload
 from sqlalchemy import text, func, or_, Integer
-from model import Singer, Album, Track, Venue, Concert
+from model import Singer, Album, Track, Venue, Concert, TicketSale
 from util_random_names import random_first_name, random_last_name, \
   random_album_title, random_release_date, random_marketing_budget, \
   random_cover_picture
@@ -143,7 +144,15 @@ def create_venue_and_concert_in_transaction():
       start_time=datetime.fromisoformat("2023-02-01T20:00:00-05:00"),
       end_time=datetime.fromisoformat("2023-02-02T02:00:00-05:00"),
     )
-    session.add_all([venue, concert])
+    # Create a TicketSale. TicketSale uses an auto-generated primary
+    # key, so we do not need to assign it a primary key in code.
+    ticket_sale = TicketSale(
+        concert=concert,
+        customer_name=random_first_name() + " " + random_last_name(),
+        price=random() * 1000,
+        seats=["A19", "A20", "A21"],
+    )
+    session.add_all([venue, concert, ticket_sale])
     session.commit()
     print()
     print("Created Venue and Concert")
@@ -169,6 +178,9 @@ def print_concerts():
                     concert.start_time.astimezone(timezone.utc).isoformat(),
                     concert.singer.full_name,
                     concert.venue.name))
+      for ticket_sale in concert.ticket_sales:
+        print("  Ticket sold to {} for seats {}"
+              .format(ticket_sale.customer_name, ticket_sale.seats))
 
 
 # Prints all albums with a release date before 1980-01-01.
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 35ff8101..beaa5030 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
@@ -59,6 +59,7 @@ import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.SSLMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
@@ -85,7 +86,6 @@ import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import javax.annotation.Nullable;
@@ -97,13 +97,13 @@ import javax.net.ssl.SSLSocketFactory;
  * representation {@link IntermediateStatement} that servers as a middle layer between Postgres and
  * Spanner.
  *
- * <p>Each {@link ConnectionHandler} is also a {@link Thread}. Although a TCP connection does not
- * necessarily need to have its own thread, this makes the implementation more straightforward.
+ * <p>Each {@link ConnectionHandler} is also a {@link Runnable} so it can be handed to a single
+ * handler thread. Although a TCP connection does not necessarily need to have its own thread, this
+ * makes the implementation more straightforward. The handler thread may be a virtual thread.
  */
 @InternalApi
-public class ConnectionHandler extends Thread {
+public class ConnectionHandler implements Runnable {
   private static final Logger logger = Logger.getLogger(ConnectionHandler.class.getName());
-  private static final AtomicLong CONNECTION_HANDLER_ID_GENERATOR = new AtomicLong(0L);
   private static final String CHANNEL_PROVIDER_PROPERTY = "CHANNEL_PROVIDER";
 
   private final ProxyServer server;
@@ -119,6 +119,7 @@ public class ConnectionHandler extends Thread {
   private static final Map<Integer, ConnectionHandler> CONNECTION_HANDLERS =
       new ConcurrentHashMap<>();
   private volatile ConnectionStatus status = ConnectionStatus.UNAUTHENTICATED;
+  private Thread thread;
   private final int connectionId;
   private final int secret;
   // Separate the following from the threat ID generator, since PG connection IDs are maximum
@@ -152,20 +153,37 @@ public class ConnectionHandler extends Thread {
   /** Constructor only for testing. */
   @VisibleForTesting
   ConnectionHandler(ProxyServer server, Socket socket, Connection spannerConnection) {
-    super("ConnectionHandler-" + CONNECTION_HANDLER_ID_GENERATOR.incrementAndGet());
     this.server = server;
     this.socket = socket;
     this.secret = new SecureRandom().nextInt();
     this.connectionId = incrementingConnectionId.incrementAndGet();
     CONNECTION_HANDLERS.put(this.connectionId, this);
-    setDaemon(true);
+    this.spannerConnection = spannerConnection;
+  }
+
+  void start() {
+    Preconditions.checkState(thread != null, "Cannot start a ConnectionHandler without a thread");
+    thread.start();
+  }
+
+  String getName() {
+    Preconditions.checkState(
+        thread != null, "Cannot get the name of a ConnectionHandler without a thread");
+    return thread.getName();
+  }
+
+  Thread getThread() {
+    return this.thread;
+  }
+
+  void setThread(Thread thread) {
+    this.thread = thread;
     logger.log(
         Level.INFO,
         () ->
             String.format(
                 "Connection handler with ID %s created for client %s",
                 getName(), socket.getInetAddress().getHostAddress()));
-    this.spannerConnection = spannerConnection;
   }
 
   void createSSLSocket() throws IOException {
@@ -273,13 +291,14 @@ public class ConnectionHandler extends Thread {
     return uri;
   }
 
-  private static String appendPropertiesToUrl(String url, Properties info) {
+  @VisibleForTesting
+  static String appendPropertiesToUrl(String url, Properties info) {
     if (info == null || info.isEmpty()) {
       return url;
     }
     StringBuilder result = new StringBuilder(url);
     for (Entry<Object, Object> entry : info.entrySet()) {
-      if (entry.getValue() != null && !"".equals(entry.getValue())) {
+      if (!Strings.isNullOrEmpty((String) entry.getValue())) {
         result.append(";").append(entry.getKey()).append("=").append(entry.getValue());
       }
     }
@@ -299,8 +318,8 @@ public class ConnectionHandler extends Thread {
             "Run",
             () ->
                 String.format(
-                    "Connection handler with ID %s starting for client %s",
-                    getName(), socket.getInetAddress().getHostAddress())));
+                    "Connection handler with ID %s starting for client %s with thread %s",
+                    getName(), socket.getInetAddress().getHostAddress(), thread.toString())));
     if (runConnection(false) == RunConnectionState.RESTART_WITH_SSL) {
       logger.log(
           Level.INFO,
@@ -630,7 +649,7 @@ public class ConnectionHandler extends Thread {
     // otherwise)
     try {
       connectionToCancel.getSpannerConnection().cancel();
-      connectionToCancel.interrupt();
+      connectionToCancel.getThread().interrupt();
       return true;
     } catch (Throwable ignore) {
     }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
index 44d94cd7..ed638ff4 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java
@@ -19,6 +19,7 @@ import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.ThreadFactoryUtil;
 import com.google.cloud.spanner.connection.SpannerPool;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler.QueryMode;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
@@ -41,6 +42,7 @@ import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
@@ -81,6 +83,8 @@ public class ProxyServer extends AbstractApiService {
   private final ConcurrentLinkedQueue<WireMessage> debugMessages = new ConcurrentLinkedQueue<>();
   private final AtomicInteger debugMessageCount = new AtomicInteger();
 
+  private final ThreadFactory threadFactory;
+
   /**
    * Instantiates the ProxyServer from CLI-gathered metadata.
    *
@@ -116,6 +120,9 @@ public class ProxyServer extends AbstractApiService {
     this.localPort = optionsMetadata.getProxyPort();
     this.properties = properties;
     this.debugMode = optionsMetadata.isDebugMode();
+    this.threadFactory =
+        ThreadFactoryUtil.createVirtualOrPlatformDaemonThreadFactory(
+            "ConnectionHandler", optionsMetadata.isUseVirtualThreads());
     addConnectionProperties();
   }
 
@@ -329,6 +336,8 @@ public class ProxyServer extends AbstractApiService {
     socket.setTcpNoDelay(true);
     ConnectionHandler handler = new ConnectionHandler(this, socket);
     register(handler);
+    Thread thread = threadFactory.newThread(handler);
+    handler.setThread(thread);
     handler.start();
   }
 
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 e5730088..5e29b028 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
@@ -23,6 +23,7 @@ import com.google.cloud.spanner.InstanceId;
 import com.google.cloud.spanner.SessionPoolOptions;
 import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.SpannerOptions;
+import com.google.cloud.spanner.connection.ConnectionOptions;
 import com.google.cloud.spanner.pgadapter.ProxyServer;
 import com.google.cloud.spanner.pgadapter.Server;
 import com.google.common.annotations.VisibleForTesting;
@@ -50,6 +51,12 @@ import org.json.simple.JSONObject;
 
 /** Metadata extractor for CLI. */
 public class OptionsMetadata {
+  /** These system properties can be used to turn on the use of virtual threads. */
+  public static String USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME = "pgadapter.use_virtual_threads";
+
+  public static String USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME =
+      "pgadapter.use_virtual_grpc_transport_threads";
+
   static Duration DEFAULT_STARTUP_TIMEOUT = Duration.ofSeconds(30L);
 
   /**
@@ -82,6 +89,8 @@ public class OptionsMetadata {
     private boolean enableOpenTelemetry;
     private Double openTelemetryTraceRatio;
     private boolean skipLocalhostCheck;
+    private boolean useVirtualThreads;
+    private boolean useVirtualGrpcTransportThreads;
     private SslMode sslMode;
     private int port;
     private String unixDomainSocketDirectory;
@@ -255,6 +264,35 @@ public class OptionsMetadata {
       return this;
     }
 
+    /**
+     * Creates a virtual thread for each connection instead of a platform thread. Enabling this
+     * option only has effect if PGAdapter is running on Java 21 or higher. Enabling virtual threads
+     * can reduce memory consumption for PGAdapter instances that have a large number of incoming
+     * connections, and reduces the total number of platform threads that are created.
+     *
+     * <p>See <a href="https://docs.oracle.com/en/java/javase/21/core/virtual-threads.html">Virtual
+     * threads</a> for more information on virtual threads.
+     */
+    public Builder useVirtualThreads() {
+      this.useVirtualThreads = true;
+      return this;
+    }
+
+    /**
+     * Uses virtual threads instead of platform threads for the gRPC executor pool. Enabling this
+     * option only has effect if PGAdapter is running on Java 21 or higher. Enabling virtual threads
+     * for gRPC can reduce memory consumption for PGAdapter instances that execute a large number of
+     * parallel queries or transactions, and reduces the total number of platform threads that are
+     * created.
+     *
+     * <p>See <a href="https://docs.oracle.com/en/java/javase/21/core/virtual-threads.html">Virtual
+     * threads</a> for more information on virtual threads.
+     */
+    public Builder useVirtualGrpcTransportThreads() {
+      this.useVirtualGrpcTransportThreads = true;
+      return this;
+    }
+
     /**
      * PGAdapter by default does not support SSL connections. Call this method to allow or require
      * SSL connections.
@@ -391,7 +429,12 @@ public class OptionsMetadata {
       if (endpoint != null) {
         addOption(args, OPTION_SPANNER_ENDPOINT, endpoint);
       }
-      if (usePlainText || numChannels != null || databaseRole != null || autoConfigEmulator) {
+      if (usePlainText
+          || numChannels != null
+          || databaseRole != null
+          || autoConfigEmulator
+          || useVirtualThreads
+          || useVirtualGrpcTransportThreads) {
         StringBuilder jdbcOptionBuilder = new StringBuilder();
         if (usePlainText) {
           jdbcOptionBuilder.append("usePlainText=true;");
@@ -405,6 +448,16 @@ public class OptionsMetadata {
         if (autoConfigEmulator) {
           jdbcOptionBuilder.append("autoConfigEmulator=true;");
         }
+        if (useVirtualThreads) {
+          jdbcOptionBuilder
+              .append(ConnectionOptions.USE_VIRTUAL_THREADS_PROPERTY_NAME)
+              .append("=true;");
+        }
+        if (useVirtualGrpcTransportThreads) {
+          jdbcOptionBuilder
+              .append(ConnectionOptions.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_PROPERTY_NAME)
+              .append("=true;");
+        }
         addOption(args, OPTION_JDBC_PROPERTIES, jdbcOptionBuilder.toString());
       }
       if (debugMode) {
@@ -730,6 +783,18 @@ public class OptionsMetadata {
         }
       }
     }
+    // Set virtual thread properties from system properties if:
+    // 1. The virtual thread property has not been set in the command line arguments.
+    // 2. The corresponding system property has been set.
+    if (properties.get(ConnectionOptions.USE_VIRTUAL_THREADS_PROPERTY_NAME) == null
+        && Boolean.parseBoolean(System.getProperty(USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME))) {
+      properties.put(ConnectionOptions.USE_VIRTUAL_THREADS_PROPERTY_NAME, "true");
+    }
+    if (properties.get(ConnectionOptions.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_PROPERTY_NAME) == null
+        && Boolean.parseBoolean(
+            System.getProperty(USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME))) {
+      properties.put(ConnectionOptions.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_PROPERTY_NAME, "true");
+    }
     return properties;
   }
 
@@ -1266,6 +1331,16 @@ public class OptionsMetadata {
     return this.debugMode;
   }
 
+  public boolean isUseVirtualThreads() {
+    return Boolean.parseBoolean(
+        getPropertyMap().get(ConnectionOptions.USE_VIRTUAL_THREADS_PROPERTY_NAME));
+  }
+
+  public boolean isUseGrpcTransportVirtualThreads() {
+    return Boolean.parseBoolean(
+        getPropertyMap().get(ConnectionOptions.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_PROPERTY_NAME));
+  }
+
   public Duration getStartupTimeout() {
     return this.startupTimeout;
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
index 5ce27fdf..f2c32e5f 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/MutationWriter.java
@@ -14,6 +14,8 @@
 
 package com.google.cloud.spanner.pgadapter.utils;
 
+import static com.google.cloud.spanner.ThreadFactoryUtil.createVirtualOrPlatformDaemonThreadFactory;
+
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiFutures;
 import com.google.api.core.InternalApi;
@@ -64,6 +66,7 @@ import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -97,6 +100,9 @@ public class MutationWriter implements Callable<StatementResult>, Closeable {
 
   private static final Logger logger = Logger.getLogger(MutationWriter.class.getName());
 
+  private static final ThreadFactory THREAD_FACTORY =
+      createVirtualOrPlatformDaemonThreadFactory("copy-worker", true);
+
   private final CopyTransactionMode transactionMode;
   private long rowCount;
   private final Connection connection;
@@ -117,7 +123,7 @@ public class MutationWriter implements Callable<StatementResult>, Closeable {
   private final AtomicBoolean rollback = new AtomicBoolean(false);
   private final CountDownLatch closedLatch = new CountDownLatch(1);
   private final ListeningExecutorService executorService =
-      MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
+      MoreExecutors.listeningDecorator(Executors.newCachedThreadPool(THREAD_FACTORY));
 
   private final Object lock = new Object();
 


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
index 83d20e8f..36b89b8b 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
@@ -14,21 +14,33 @@
 
 package com.google.cloud.spanner.pgadapter;
 
+import static com.google.cloud.spanner.pgadapter.ConnectionHandler.appendPropertiesToUrl;
 import static com.google.cloud.spanner.pgadapter.ConnectionHandler.buildConnectionURL;
+import static com.google.cloud.spanner.pgadapter.ConnectionHandler.listDatabasesOrInstances;
+import static com.google.cloud.spanner.pgadapter.EmulatedPsqlMockServerTest.newStatusResourceNotFoundException;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.api.core.SettableApiFuture;
+import com.google.api.gax.paging.Pages;
 import com.google.cloud.NoCredentials;
+import com.google.cloud.spanner.DatabaseAdminClient;
 import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.Spanner;
 import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerException.ResourceNotFoundException;
 import com.google.cloud.spanner.SpannerExceptionFactory;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
@@ -48,6 +60,9 @@ import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
 import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.spanner.v1.DatabaseName;
+import io.grpc.StatusRuntimeException;
+import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import java.io.EOFException;
@@ -56,8 +71,10 @@ import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.InetAddress;
 import java.net.Socket;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -70,8 +87,7 @@ public class ConnectionHandlerTest {
   public void testRegisterStatement() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     IntermediatePreparedStatement statement = mock(IntermediatePreparedStatement.class);
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
@@ -88,8 +104,7 @@ public class ConnectionHandlerTest {
   public void testCloseUnknownStatement() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
 
@@ -100,8 +115,7 @@ public class ConnectionHandlerTest {
   public void testCloseAll() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     IntermediatePreparedStatement statement1 = mock(IntermediatePreparedStatement.class);
     IntermediatePreparedStatement statement2 = mock(IntermediatePreparedStatement.class);
 
@@ -119,8 +133,7 @@ public class ConnectionHandlerTest {
   public void testTerminateClosesSocket() throws IOException {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
 
@@ -133,8 +146,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
     when(socket.isClosed()).thenReturn(false, true);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
 
@@ -150,12 +162,12 @@ public class ConnectionHandlerTest {
   public void testTerminateHandlesCloseError() throws IOException {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     // IOException should be handled internally in terminate().
     doThrow(new IOException("test exception")).when(socket).close();
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
+    connection.setThread(mock(Thread.class));
 
     connection.terminate();
     verify(socket).close();
@@ -165,8 +177,7 @@ public class ConnectionHandlerTest {
   public void testTerminateClosesAllPortals() throws Exception {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     IntermediatePortalStatement portal1 = mock(IntermediatePortalStatement.class);
     IntermediatePortalStatement portal2 = mock(IntermediatePortalStatement.class);
 
@@ -184,8 +195,7 @@ public class ConnectionHandlerTest {
   public void testTerminateIgnoresPortalCloseError() throws Exception {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     IntermediatePortalStatement portal1 = mock(IntermediatePortalStatement.class);
     IntermediatePortalStatement portal2 = mock(IntermediatePortalStatement.class);
     doThrow(new Exception("test")).when(portal2).close();
@@ -204,8 +214,7 @@ public class ConnectionHandlerTest {
   public void testGetPortal() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     IntermediatePortalStatement portal1 = mock(IntermediatePortalStatement.class);
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
@@ -218,8 +227,7 @@ public class ConnectionHandlerTest {
   public void testGetUnknownPortal() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
     PGException exception =
@@ -231,8 +239,7 @@ public class ConnectionHandlerTest {
   public void testClosePortal() throws Exception {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     IntermediatePortalStatement portal1 = mock(IntermediatePortalStatement.class);
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
@@ -246,8 +253,7 @@ public class ConnectionHandlerTest {
   public void testCloseUnknownPortal() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
     PGException exception =
@@ -259,8 +265,7 @@ public class ConnectionHandlerTest {
   public void testHandleMessages_NonFatalException() throws Exception {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     DataOutputStream dataOutputStream = new DataOutputStream(new ByteArrayOutputStream());
     ConnectionMetadata connectionMetadata = mock(ConnectionMetadata.class);
     when(connectionMetadata.getOutputStream()).thenReturn(dataOutputStream);
@@ -278,6 +283,7 @@ public class ConnectionHandlerTest {
             return connectionMetadata;
           }
         };
+    connection.setThread(mock(Thread.class));
 
     connection.setMessageState(message);
     connection.handleMessages();
@@ -289,8 +295,7 @@ public class ConnectionHandlerTest {
   public void testHandleMessages_FatalException() throws Exception {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     DataOutputStream dataOutputStream = new DataOutputStream(new ByteArrayOutputStream());
     ConnectionMetadata connectionMetadata = mock(ConnectionMetadata.class);
     when(connectionMetadata.getOutputStream()).thenReturn(dataOutputStream);
@@ -304,6 +309,7 @@ public class ConnectionHandlerTest {
             return connectionMetadata;
           }
         };
+    connection.setThread(mock(Thread.class));
 
     connection.setMessageState(message);
     connection.handleMessages();
@@ -315,15 +321,14 @@ public class ConnectionHandlerTest {
   public void testCancelActiveStatement() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
-    when(address.getHostAddress()).thenReturn("address1");
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     Connection spannerConnection = mock(Connection.class);
 
     ConnectionHandler connectionHandler =
         new ConnectionHandler(server, socket, mock(Connection.class));
     ConnectionHandler connectionHandlerToCancel =
         new ConnectionHandler(server, socket, spannerConnection);
+    connectionHandlerToCancel.setThread(mock(Thread.class));
 
     // Cancelling yourself is not allowed.
     assertFalse(
@@ -354,8 +359,7 @@ public class ConnectionHandlerTest {
   public void testRestartConnectionWithSsl_CreatesSslSocket() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     AtomicBoolean calledCreateSSLSocket = new AtomicBoolean();
     ConnectionHandler connection =
         new ConnectionHandler(server, socket) {
@@ -364,6 +368,7 @@ public class ConnectionHandlerTest {
             calledCreateSSLSocket.set(true);
           }
         };
+    connection.setThread(mock(Thread.class));
     connection.restartConnectionWithSsl();
 
     assertTrue(calledCreateSSLSocket.get());
@@ -373,8 +378,7 @@ public class ConnectionHandlerTest {
   public void testRestartConnectionWithSsl_SslSocketCreationFailureIsConvertedToPGException() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connection =
         new ConnectionHandler(server, socket) {
           @Override
@@ -391,8 +395,7 @@ public class ConnectionHandlerTest {
       testRestartConnectionWithSsl_SslSocketCreationFailureIsConvertedToPGExceptionWithMessage() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connection =
         new ConnectionHandler(server, socket) {
           @Override
@@ -408,8 +411,7 @@ public class ConnectionHandlerTest {
   public void testRestartConnectionWithSsl_SendsPGException() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
 
     ConnectionHandler connection =
         new ConnectionHandler(server, socket) {
@@ -432,8 +434,7 @@ public class ConnectionHandlerTest {
   public void testRestartConnectionWithSsl_IgnoresExceptionErrorHandling() {
     ProxyServer server = mock(ProxyServer.class);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
 
     ConnectionHandler connection =
         new ConnectionHandler(server, socket) {
@@ -463,8 +464,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler =
         new ConnectionHandler(server, socket) {
           @Override
@@ -473,11 +473,7 @@ public class ConnectionHandlerTest {
           }
         };
 
-    when(address.isLoopbackAddress()).thenReturn(true);
     assertTrue(connectionHandler.checkValidConnection(false));
-
-    when(address.isLoopbackAddress()).thenReturn(false);
-    assertFalse(connectionHandler.checkValidConnection(false));
   }
 
   @Test
@@ -486,8 +482,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler =
         new ConnectionHandler(server, socket) {
           @Override
@@ -496,11 +491,7 @@ public class ConnectionHandlerTest {
           }
         };
 
-    when(address.isAnyLocalAddress()).thenReturn(true);
     assertTrue(connectionHandler.checkValidConnection(false));
-
-    when(address.isAnyLocalAddress()).thenReturn(false);
-    assertFalse(connectionHandler.checkValidConnection(false));
   }
 
   @Test
@@ -509,8 +500,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler =
         new ConnectionHandler(server, socket) {
           @Override
@@ -518,15 +508,11 @@ public class ConnectionHandlerTest {
             return new ConnectionMetadata(mock(InputStream.class), mock(OutputStream.class));
           }
         };
+    connectionHandler.setThread(mock(Thread.class));
 
     when(options.getSslMode()).thenReturn(SslMode.Enable);
-    when(address.isLoopbackAddress()).thenReturn(true);
     assertTrue(connectionHandler.checkValidConnection(false));
 
-    assertTrue(connectionHandler.checkValidConnection(true));
-    when(address.isLoopbackAddress()).thenReturn(false);
-    assertTrue(connectionHandler.checkValidConnection(true));
-
     when(options.getSslMode()).thenReturn(SslMode.Require);
     assertFalse(connectionHandler.checkValidConnection(false));
     assertTrue(connectionHandler.checkValidConnection(true));
@@ -539,8 +525,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler = new ConnectionHandler(server, socket);
 
     assertTrue(connectionHandler.checkValidConnection(false));
@@ -552,8 +537,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler = new ConnectionHandler(server, socket);
 
     when(options.shouldAutoDetectClient()).thenReturn(true);
@@ -570,8 +554,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler = new ConnectionHandler(server, socket);
 
     when(options.shouldAutoDetectClient()).thenReturn(true);
@@ -592,8 +575,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler = new ConnectionHandler(server, socket);
 
     when(options.shouldAutoDetectClient()).thenReturn(false);
@@ -614,8 +596,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler = new ConnectionHandler(server, socket);
 
     when(options.shouldAutoDetectClient()).thenReturn(true);
@@ -640,8 +621,7 @@ public class ConnectionHandlerTest {
     ProxyServer server = mock(ProxyServer.class);
     when(server.getOptions()).thenReturn(options);
     Socket socket = mock(Socket.class);
-    InetAddress address = mock(InetAddress.class);
-    when(socket.getInetAddress()).thenReturn(address);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
     ConnectionHandler connectionHandler = new ConnectionHandler(server, socket);
 
     when(options.shouldAutoDetectClient()).thenReturn(true);
@@ -670,6 +650,12 @@ public class ConnectionHandlerTest {
 
   @Test
   public void testBuildConnectionUrl() {
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME) == null);
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME)
+            == null);
+
     OptionsMetadata options =
         OptionsMetadata.newBuilder().setCredentials(NoCredentials.getInstance()).build();
     // Check that the dialect is included in the connection URL. This is required to support the
@@ -748,6 +734,155 @@ public class ConnectionHandlerTest {
         System.setProperty("ENABLE_CHANNEL_PROVIDER", currentEnableChannelProvider);
       }
     }
+
+    // Verify that the virtual threads properties are carried over to the connection URL.
+    assertEquals(
+        "cloudspanner:/projects/my-project/instances/my-instance/databases/my-database;userAgent=pg-adapter;useVirtualThreads=true;dialect=postgresql",
+        buildConnectionURL(
+            "projects/my-project/instances/my-instance/databases/my-database",
+            options,
+            buildProperties(ImmutableMap.of("useVirtualThreads", "true"))));
+    assertEquals(
+        "cloudspanner:/projects/my-project/instances/my-instance/databases/my-database;userAgent=pg-adapter;useVirtualGrpcTransportThreads=true;dialect=postgresql",
+        buildConnectionURL(
+            "projects/my-project/instances/my-instance/databases/my-database",
+            options,
+            buildProperties(ImmutableMap.of("useVirtualGrpcTransportThreads", "true"))));
+
+    runWithSystemProperty(
+        OptionsMetadata.USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME,
+        "true",
+        () -> {
+          OptionsMetadata optionsWithSystemProperty =
+              OptionsMetadata.newBuilder().setCredentials(NoCredentials.getInstance()).build();
+          assertEquals(
+              "cloudspanner:/projects/my-project/instances/my-instance/databases/my-database;userAgent=pg-adapter;useVirtualThreads=true;dialect=postgresql",
+              buildConnectionURL(
+                  "projects/my-project/instances/my-instance/databases/my-database",
+                  optionsWithSystemProperty,
+                  buildProperties(optionsWithSystemProperty.getPropertyMap())));
+        });
+    runWithSystemProperty(
+        OptionsMetadata.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME,
+        "true",
+        () -> {
+          OptionsMetadata optionsWithSystemProperty =
+              OptionsMetadata.newBuilder().setCredentials(NoCredentials.getInstance()).build();
+          assertEquals(
+              "cloudspanner:/projects/my-project/instances/my-instance/databases/my-database;userAgent=pg-adapter;useVirtualGrpcTransportThreads=true;dialect=postgresql",
+              buildConnectionURL(
+                  "projects/my-project/instances/my-instance/databases/my-database",
+                  optionsWithSystemProperty,
+                  buildProperties(optionsWithSystemProperty.getPropertyMap())));
+        });
+  }
+
+  @Test
+  public void testGetName() {
+    ProxyServer server = mock(ProxyServer.class);
+    Socket socket = mock(Socket.class);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
+    Connection connection = mock(Connection.class);
+    ConnectionHandler connectionHandler = new ConnectionHandler(server, socket, connection);
+
+    assertThrows(IllegalStateException.class, connectionHandler::getName);
+
+    Thread thread = new Thread(() -> {}, "test-thread");
+    connectionHandler.setThread(thread);
+    assertEquals("test-thread", connectionHandler.getName());
+  }
+
+  @Test
+  public void testStart() throws Exception {
+    ProxyServer server = mock(ProxyServer.class);
+    Socket socket = mock(Socket.class);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getLoopbackAddress());
+    Connection connection = mock(Connection.class);
+    ConnectionHandler connectionHandler = new ConnectionHandler(server, socket, connection);
+
+    assertThrows(IllegalStateException.class, connectionHandler::start);
+
+    SettableApiFuture<Boolean> started = SettableApiFuture.create();
+    Thread thread = new Thread(() -> started.set(true), "test-thread");
+    connectionHandler.setThread(thread);
+
+    connectionHandler.start();
+    assertTrue(started.get(10L, TimeUnit.SECONDS));
+  }
+
+  @Test
+  public void testAppendPropertiesToUrl() {
+    assertEquals(
+        "cloudspanner:/project/p/instances/i/databases/d",
+        appendPropertiesToUrl("cloudspanner:/project/p/instances/i/databases/d", new Properties()));
+    assertEquals(
+        "cloudspanner:/project/p/instances/i/databases/d",
+        appendPropertiesToUrl("cloudspanner:/project/p/instances/i/databases/d", null));
+    assertEquals(
+        "cloudspanner:/project/p/instances/i/databases/d",
+        appendPropertiesToUrl(
+            "cloudspanner:/project/p/instances/i/databases/d",
+            buildProperties(ImmutableMap.of("key", ""))));
+    assertEquals(
+        "cloudspanner:/project/p/instances/i/databases/d;key=value",
+        appendPropertiesToUrl(
+            "cloudspanner:/project/p/instances/i/databases/d",
+            buildProperties(ImmutableMap.of("key", "value"))));
+  }
+
+  @Test
+  public void testCheckValidConnection_External() throws Exception {
+    OptionsMetadata options = mock(OptionsMetadata.class);
+    ProxyServer server = mock(ProxyServer.class);
+    when(server.getOptions()).thenReturn(options);
+    Socket socket = mock(Socket.class);
+    when(socket.getInetAddress()).thenReturn(InetAddress.getByName("google.com"));
+    Connection connection = mock(Connection.class);
+
+    ByteArrayOutputStream bos = new ByteArrayOutputStream();
+    ConnectionMetadata connectionMetadata =
+        new ConnectionMetadata(new ByteArrayInputStream(new byte[0]), bos);
+
+    ConnectionHandler connectionHandler =
+        new ConnectionHandler(server, socket, connection) {
+          @Override
+          public ConnectionMetadata getConnectionMetadata() {
+            return connectionMetadata;
+          }
+        };
+    connectionHandler.setThread(new Thread(() -> {}, "test-thread"));
+
+    assertFalse(connectionHandler.checkValidConnection(false));
+    byte[] bytes = bos.toByteArray();
+    assertEquals(74, bytes.length);
+    assertEquals('E', bytes[0]); // Error
+    // The total length of the Error message is 68.
+    assertArrayEquals(new byte[] {0, 0, 0, 68}, Arrays.copyOfRange(bytes, 1, 5));
+    // The next message is a Terminate ('X') message.
+    assertEquals('X', bytes[69]);
+  }
+
+  @Test
+  public void testListDatabasesOrInstances() {
+    Spanner spanner = mock(Spanner.class);
+    DatabaseAdminClient adminClient = mock(DatabaseAdminClient.class);
+    when(spanner.getDatabaseAdminClient()).thenReturn(adminClient);
+    when(adminClient.listDatabases(eq("my-instance"), any())).thenReturn(Pages.empty());
+    StatusRuntimeException exception =
+        newStatusResourceNotFoundException(
+            "test-database",
+            "type.googleapis.com/google.spanner.admin.database.v1.Database",
+            "projects/my-project/instances/my-instance/databases/test-database");
+    ResourceNotFoundException resourceNotFoundException =
+        (ResourceNotFoundException) SpannerExceptionFactory.asSpannerException(exception);
+    assertEquals(
+        "Database projects/my-project/instances/my-instance/databases/test-database not found.\n"
+            + "\n"
+            + "These PostgreSQL databases are available on instance projects/my-project/instances/my-instance:\n",
+        listDatabasesOrInstances(
+            resourceNotFoundException,
+            DatabaseName.of("my-project", "my-instance", "test-database"),
+            spanner));
   }
 
   static Properties buildProperties(Map<String, String> map) {
@@ -755,4 +890,22 @@ public class ConnectionHandlerTest {
     properties.putAll(map);
     return properties;
   }
+
+  void runWithSystemProperty(String property, String value, Runnable runnable) {
+    String currentValue = System.getProperty(property);
+    try {
+      if (value == null) {
+        System.clearProperty(property);
+      } else {
+        System.setProperty(property, value);
+      }
+      runnable.run();
+    } finally {
+      if (currentValue == null) {
+        System.clearProperty(property);
+      } else {
+        System.setProperty(property, currentValue);
+      }
+    }
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
index eb3606a9..5ec2f0c1 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/EmulatedPsqlMockServerTest.java
@@ -99,11 +99,16 @@ public class EmulatedPsqlMockServerTest extends AbstractMockServerTest {
    * recognize the connection as a psql connection.
    */
   private String createUrl(String database) {
+    return createUrl(database, "psql");
+  }
+
+  private String createUrl(String database, String applicationName) {
     return String.format(
-        "jdbc:postgresql://localhost:%d/%s?preferQueryMode=simple&%s=psql&%s=090000",
+        "jdbc:postgresql://localhost:%d/%s?preferQueryMode=simple&%s=%s&%s=090000",
         pgServer.getLocalPort(),
         database,
         PGProperty.APPLICATION_NAME.getName(),
+        applicationName,
         PGProperty.ASSUME_MIN_SERVER_VERSION.getName());
   }
 
@@ -210,37 +215,61 @@ public class EmulatedPsqlMockServerTest extends AbstractMockServerTest {
 
   @Test
   public void testConnectToNonExistingInstance() {
+    for (String appName : new String[] {"psql", "other-app"}) {
+      try {
+        mockSpanner.setExecuteStreamingSqlExecutionTime(
+            SimulatedExecutionTime.ofStickyException(
+                newStatusResourceNotFoundException(
+                    "i",
+                    "type.googleapis.com/google.spanner.admin.instance.v1.Instance",
+                    "projects/p/instances/i")));
+        // The Connection API calls listInstanceConfigs(..) once first when the connection is a
+        // localhost connection. It does so to verify that the connection is valid and to quickly
+        // return an error if someone is for example trying to connect to the emulator while the
+        // emulator is not running. This does not happen when you connect to a remote host. We
+        // therefore need to add a response for the listInstanceConfigs as well.
+        mockInstanceAdmin.addResponse(ListInstanceConfigsResponse.getDefaultInstance());
+        mockInstanceAdmin.addResponse(
+            ListInstancesResponse.newBuilder()
+                .addInstances(
+                    Instance.newBuilder()
+                        .setConfig("projects/p/instanceConfigs/ic")
+                        .setName("projects/p/instances/i")
+                        .build())
+                .build());
+
+        SQLException exception =
+            assertThrows(
+                SQLException.class,
+                () -> DriverManager.getConnection(createUrl("non-existing-db", appName)));
+        assertTrue(exception.getMessage(), exception.getMessage().contains("NOT_FOUND"));
+
+        boolean isPsql = "psql".equals(appName);
+        assertEquals(
+            exception.getMessage(),
+            isPsql,
+            exception.getMessage().contains("These instances are available in project p:"));
+        assertEquals(
+            exception.getMessage(),
+            isPsql,
+            exception.getMessage().contains("\tprojects/p/instances/i\n"));
+      } finally {
+        closeSpannerPool(true);
+      }
+    }
+  }
+
+  @Test
+  public void testConnectFailed() {
     try {
       mockSpanner.setExecuteStreamingSqlExecutionTime(
           SimulatedExecutionTime.ofStickyException(
-              newStatusResourceNotFoundException(
-                  "i",
-                  "type.googleapis.com/google.spanner.admin.instance.v1.Instance",
-                  "projects/p/instances/i")));
-      // The Connection API calls listInstanceConfigs(..) once first when the connection is a
-      // localhost connection. It does so to verify that the connection is valid and to quickly
-      // return an error if someone is for example trying to connect to the emulator while the
-      // emulator is not running. This does not happen when you connect to a remote host. We
-      // therefore need to add a response for the listInstanceConfigs as well.
-      mockInstanceAdmin.addResponse(ListInstanceConfigsResponse.getDefaultInstance());
-      mockInstanceAdmin.addResponse(
-          ListInstancesResponse.newBuilder()
-              .addInstances(
-                  Instance.newBuilder()
-                      .setConfig("projects/p/instanceConfigs/ic")
-                      .setName("projects/p/instances/i")
-                      .build())
-              .build());
-
+              Status.INVALID_ARGUMENT.withDescription("test error").asRuntimeException()));
       SQLException exception =
           assertThrows(
               SQLException.class, () -> DriverManager.getConnection(createUrl("non-existing-db")));
-      assertTrue(exception.getMessage(), exception.getMessage().contains("NOT_FOUND"));
-      assertTrue(
-          exception.getMessage(),
-          exception.getMessage().contains("These instances are available in project p:"));
-      assertTrue(
-          exception.getMessage(), exception.getMessage().contains("\tprojects/p/instances/i\n"));
+      assertTrue(exception.getMessage(), exception.getMessage().contains("test error"));
+
     } finally {
       closeSpannerPool(true);
     }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java
index 5bd82644..fc7c7a27 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java
@@ -22,6 +22,7 @@ import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeTrue;
 
 import com.google.auth.oauth2.AccessToken;
 import com.google.auth.oauth2.OAuth2Credentials;
@@ -123,6 +124,12 @@ public class OptionsMetadataTest {
 
   @Test
   public void testBuildConnectionUrlWithFullPath() {
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME) == null);
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME)
+            == null);
+
     assertEquals(
         "cloudspanner:/projects/test-project/instances/test-instance/databases/test-database;userAgent=pg-adapter;credentials=credentials.json",
         new OptionsMetadata(new String[] {"-c", "credentials.json"})
@@ -157,6 +164,12 @@ public class OptionsMetadataTest {
 
   @Test
   public void testBuildConnectionUrlWithDefaultProjectId() {
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME) == null);
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME)
+            == null);
+
     OptionsMetadata useDefaultProjectIdOptions =
         new OptionsMetadata(new String[] {"-i", "test-instance", "-c", "credentials.json"}) {
           @Override
@@ -182,6 +195,12 @@ public class OptionsMetadataTest {
 
   @Test
   public void testBuildConnectionUrlWithDefaultCredentials() {
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME) == null);
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME)
+            == null);
+
     OptionsMetadata useDefaultCredentials =
         new OptionsMetadata(new String[] {"-p", "test-project", "-i", "test-instance"}) {
           @Override
@@ -299,6 +318,12 @@ public class OptionsMetadataTest {
 
   @Test
   public void testBuilder() {
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_THREADS_SYSTEM_PROPERTY_NAME) == null);
+    assumeTrue(
+        System.getProperty(OptionsMetadata.USE_VIRTUAL_GRPC_TRANSPORT_THREADS_SYSTEM_PROPERTY_NAME)
+            == null);
+
     assertFalse(
         OptionsMetadata.newBuilder()
             .setProject("my-project")
@@ -398,6 +423,41 @@ public class OptionsMetadataTest {
             .build()
             .getPropertyMap()
             .get("usePlainText"));
+    assertNull(
+        OptionsMetadata.newBuilder()
+            .setCredentials(NoCredentials.getInstance())
+            .build()
+            .getPropertyMap()
+            .get("useVirtualThreads"));
+    assertEquals(
+        "true",
+        OptionsMetadata.newBuilder()
+            .useVirtualThreads()
+            .setCredentials(NoCredentials.getInstance())
+            .build()
+            .getPropertyMap()
+            .get("useVirtualThreads"));
+    assertNull(
+        OptionsMetadata.newBuilder()
+            .setCredentials(NoCredentials.getInstance())
+            .build()
+            .getPropertyMap()
+            .get("useVirtualGrpcTransportThreads"));
+    assertEquals(
+        "true",
+        OptionsMetadata.newBuilder()
+            .useVirtualGrpcTransportThreads()
+            .setCredentials(NoCredentials.getInstance())
+            .build()
+            .getPropertyMap()
+            .get("useVirtualGrpcTransportThreads"));
+    assertFalse(OptionsMetadata.newBuilder().build().isUseGrpcTransportVirtualThreads());
+    assertTrue(
+        OptionsMetadata.newBuilder()
+            .useVirtualGrpcTransportThreads()
+            .build()
+            .isUseGrpcTransportVirtualThreads());
+
     assertEquals(
         DdlTransactionMode.Batch,
         OptionsMetadata.newBuilder()
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/python/sqlalchemy2/SqlAlchemy2SampleTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/python/sqlalchemy2/SqlAlchemy2SampleTest.java
index 647a6a06..fbec6888 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/python/sqlalchemy2/SqlAlchemy2SampleTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/python/sqlalchemy2/SqlAlchemy2SampleTest.java
@@ -343,11 +343,31 @@ public class SqlAlchemy2SampleTest extends AbstractMockServerTest {
                         .addAllValues(singerValues)
                         .build())
                 .build()));
+    String ticketSalesSql =
+        "SELECT ticket_sales.concert_id AS ticket_sales_concert_id, ticket_sales.customer_name AS ticket_sales_customer_name, ticket_sales.price AS ticket_sales_price, ticket_sales.seats AS ticket_sales_seats, ticket_sales.id AS ticket_sales_id, ticket_sales.version_id AS ticket_sales_version_id, ticket_sales.created_at AS ticket_sales_created_at, ticket_sales.updated_at AS ticket_sales_updated_at \n"
+            + "FROM ticket_sales \n"
+            + "WHERE $1::VARCHAR = ticket_sales.concert_id";
+    mockSpanner.putPartialStatementResult(
+        StatementResult.query(
+            Statement.of(ticketSalesSql),
+            ResultSet.newBuilder()
+                .setMetadata(createTicketSalesMetadata("ticket_sales_"))
+                .addRows(
+                    createTicketSaleRow(
+                        1L,
+                        "c1",
+                        "Alice",
+                        "99.99",
+                        Timestamp.parseTimestamp("2022-12-02T17:30:00Z"),
+                        Timestamp.parseTimestamp("2022-12-02T17:30:00Z")))
+                .build()));
+
     String output =
         execute(SAMPLE_DIR, "test_print_concerts.py", "localhost", pgServer.getLocalPort());
     assertEquals(
         "\n"
-            + "Concert 'Avenue Park Open' starting at 2023-02-02T01:00:00+00:00 with Pete Allison will be held at Avenue Park\n",
+            + "Concert 'Avenue Park Open' starting at 2023-02-02T01:00:00+00:00 with Pete Allison will be held at Avenue Park\n"
+            + "  Ticket sold to Alice for seats ['A10']\n",
         output);
   }
 
@@ -420,6 +440,97 @@ public class SqlAlchemy2SampleTest extends AbstractMockServerTest {
             Statement.of(
                 "INSERT INTO concerts (name, venue_id, singer_id, start_time, end_time, id, version_id, created_at, updated_at) "),
             1L));
+    String insertTicketSaleSql =
+        "INSERT INTO ticket_sales (concert_id, customer_name, price, seats, version_id, created_at, updated_at) "
+            + "VALUES ($1::VARCHAR, $2::VARCHAR, $3, $4, $5::INTEGER, $6::TIMESTAMP WITH TIME ZONE, $7::TIMESTAMP WITH TIME ZONE) "
+            + "RETURNING ticket_sales.id";
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(insertTicketSaleSql),
+            ResultSet.newBuilder()
+                .setMetadata(
+                    ResultSetMetadata.newBuilder()
+                        .setRowType(
+                            StructType.newBuilder()
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("id")
+                                        .setType(Type.newBuilder().setCode(TypeCode.INT64).build())
+                                        .build())
+                                .build())
+                        .setUndeclaredParameters(
+                            StructType.newBuilder()
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p1")
+                                        .setType(Type.newBuilder().setCode(TypeCode.STRING).build())
+                                        .build())
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p2")
+                                        .setType(Type.newBuilder().setCode(TypeCode.STRING).build())
+                                        .build())
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p3")
+                                        .setType(
+                                            Type.newBuilder().setCode(TypeCode.NUMERIC).build())
+                                        .build())
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p4")
+                                        .setType(
+                                            Type.newBuilder()
+                                                .setCode(TypeCode.ARRAY)
+                                                .setArrayElementType(
+                                                    Type.newBuilder()
+                                                        .setCode(TypeCode.STRING)
+                                                        .build())
+                                                .build())
+                                        .build())
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p5")
+                                        .setType(Type.newBuilder().setCode(TypeCode.INT64).build())
+                                        .build())
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p6")
+                                        .setType(
+                                            Type.newBuilder().setCode(TypeCode.TIMESTAMP).build())
+                                        .build())
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("p7")
+                                        .setType(
+                                            Type.newBuilder().setCode(TypeCode.TIMESTAMP).build())
+                                        .build())
+                                .build())
+                        .build())
+                .setStats(ResultSetStats.newBuilder().build())
+                .build()));
+    mockSpanner.putPartialStatementResult(
+        StatementResult.query(
+            Statement.of(
+                "INSERT INTO ticket_sales (concert_id, customer_name, price, seats, version_id, created_at, updated_at) "),
+            ResultSet.newBuilder()
+                .setMetadata(
+                    ResultSetMetadata.newBuilder()
+                        .setRowType(
+                            StructType.newBuilder()
+                                .addFields(
+                                    Field.newBuilder()
+                                        .setName("id")
+                                        .setType(Type.newBuilder().setCode(TypeCode.INT64).build())
+                                        .build())
+                                .build())
+                        .build())
+                .setStats(ResultSetStats.newBuilder().setRowCountExact(1L).build())
+                .addRows(
+                    ListValue.newBuilder()
+                        .addValues(Value.newBuilder().setStringValue("1").build())
+                        .build())
+                .build()));
 
     String output =
         execute(
@@ -1087,6 +1198,7 @@ public class SqlAlchemy2SampleTest extends AbstractMockServerTest {
         output);
   }
 
+  @Ignore("Requires support for array casting/coercion")
   @Test
   public void testCreateDataModel() throws Exception {
     String checkTableExistsSql =
@@ -1228,6 +1340,27 @@ public class SqlAlchemy2SampleTest extends AbstractMockServerTest {
             ResultSet.newBuilder()
                 .setMetadata(createMetadata(ImmutableList.of(TypeCode.STRING)))
                 .build()));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(checkTableExistsSql)
+                .bind("p1")
+                .to("ticket_sales")
+                .bind("p2")
+                .to("r")
+                .bind("p3")
+                .to("p")
+                .bind("p4")
+                .to("f")
+                .bind("p5")
+                .to("v")
+                .bind("p6")
+                .to("m")
+                .bind("p7")
+                .to("pg_catalog")
+                .build(),
+            ResultSet.newBuilder()
+                .setMetadata(createMetadata(ImmutableList.of(TypeCode.STRING)))
+                .build()));
     addDdlResponseToSpannerAdmin();
 
     String actualOutput =
@@ -1241,7 +1374,7 @@ public class SqlAlchemy2SampleTest extends AbstractMockServerTest {
             .map(req -> (UpdateDatabaseDdlRequest) req)
             .collect(Collectors.toList());
     assertEquals(1, requests.size());
-    assertEquals(5, requests.get(0).getStatementsCount());
+    assertEquals(6, requests.get(0).getStatementsCount());
     assertEquals(
         "CREATE TABLE singers (\n"
             + "\tfirst_name VARCHAR(100), \n"
@@ -1312,6 +1445,20 @@ public class SqlAlchemy2SampleTest extends AbstractMockServerTest {
             + "\tFOREIGN KEY(id) REFERENCES albums (id)\n"
             + ")",
         requests.get(0).getStatements(4));
+    assertEquals(
+        "CREATE TABLE ticket_sales (\n"
+            + "\tconcert_id VARCHAR, \n"
+            + "\tcustomer_name VARCHAR, \n"
+            + "\tprice NUMERIC, \n"
+            + "\tseats VARCHAR[], \n"
+            + "\tid SERIAL NOT NULL, \n"
+            + "\tversion_id INTEGER NOT NULL, \n"
+            + "\tcreated_at TIMESTAMP WITH TIME ZONE, \n"
+            + "\tupdated_at TIMESTAMP WITH TIME ZONE, \n"
+            + "\tPRIMARY KEY (id), \n"
+            + "\tFOREIGN KEY(concert_id) REFERENCES concerts (id)\n"
+            + ")",
+        requests.get(0).getStatements(5));
   }
 
   @Ignore("Uses too much pg_catalog tables")
@@ -1573,6 +1720,88 @@ public class SqlAlchemy2SampleTest extends AbstractMockServerTest {
         .build();
   }
 
+  static ResultSetMetadata createTicketSalesMetadata(String prefix) {
+    return ResultSetMetadata.newBuilder()
+        .setRowType(
+            StructType.newBuilder()
+                .addFields(
+                    Field.newBuilder()
+                        .setType(Type.newBuilder().setCode(TypeCode.STRING).build())
+                        .setName(prefix + "concert_id")
+                        .build())
+                .addFields(
+                    Field.newBuilder()
+                        .setType(Type.newBuilder().setCode(TypeCode.STRING).build())
+                        .setName(prefix + "customer_name")
+                        .build())
+                .addFields(
+                    Field.newBuilder()
+                        .setType(
+                            Type.newBuilder()
+                                .setCode(TypeCode.NUMERIC)
+                                .setTypeAnnotation(TypeAnnotationCode.PG_NUMERIC)
+                                .build())
+                        .setName(prefix + "price")
+                        .build())
+                .addFields(
+                    Field.newBuilder()
+                        .setType(
+                            Type.newBuilder()
+                                .setCode(TypeCode.ARRAY)
+                                .setArrayElementType(
+                                    Type.newBuilder().setCode(TypeCode.STRING).build())
+                                .build())
+                        .setName(prefix + "seats")
+                        .build())
+                .addFields(
+                    Field.newBuilder()
+                        .setType(Type.newBuilder().setCode(TypeCode.INT64).build())
+                        .setName(prefix + "id")
+                        .build())
+                .addFields(
+                    Field.newBuilder()
+                        .setType(Type.newBuilder().setCode(TypeCode.INT64).build())
+                        .setName(prefix + "version_id")
+                        .build())
+                .addFields(
+                    Field.newBuilder()
+                        .setType(Type.newBuilder().setCode(TypeCode.TIMESTAMP).build())
+                        .setName(prefix + "created_at")
+                        .build())
+                .addFields(
+                    Field.newBuilder()
+                        .setType(Type.newBuilder().setCode(TypeCode.TIMESTAMP).build())
+                        .setName(prefix + "updated_at")
+                        .build())
+                .build())
+        .build();
+  }
+
+  static ListValue createTicketSaleRow(
+      long id,
+      String concertId,
+      String customerName,
+      String price,
+      Timestamp createdAt,
+      Timestamp updatedAt) {
+    return ListValue.newBuilder()
+        .addValues(Value.newBuilder().setStringValue(concertId).build())
+        .addValues(Value.newBuilder().setStringValue(customerName).build())
+        .addValues(Value.newBuilder().setStringValue(price).build())
+        .addValues(
+            Value.newBuilder()
+                .setListValue(
+                    ListValue.newBuilder()
+                        .addValues(Value.newBuilder().setStringValue("A10").build())
+                        .build())
+                .build())
+        .addValues(Value.newBuilder().setStringValue(String.valueOf(id)).build())
+        .addValues(Value.newBuilder().setStringValue("1").build())
+        .addValues(Value.newBuilder().setStringValue(createdAt.toString()).build())
+        .addValues(Value.newBuilder().setStringValue(updatedAt.toString()).build())
+        .build();
+  }
+
   static ResultSetMetadata createConcertsMetadata(String prefix) {
     return ResultSetMetadata.newBuilder()
         .setRowType(