Repository: GoogleCloudPlatform/pgadapter
Pull request #797 (2023-04-30 19:40:33)

feat: support cursors and foreign data wrapper
Add support for DECLARE, FETCH/MOVE and CLOSE cursors. This also enables the use of PGAdapter as a `FOREIGN SERVER` in a real PostgreSQL database, which again makes it possible to use features like user-defined functions, stored procedures or built-in functions that are not supported by Cloud Spanner for data in Cloud Spanner (albeit with limitations).

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

```diff
@@                   Coverage Diff                    @@
##             postgresql-dialect     #797      +/-   ##
========================================================
- Coverage                 89.61%   88.85%   -0.76%     
- Complexity                 2366     2413      +47     
========================================================
  Files                       131      136       +5     
  Lines                      7837     8069     +232     
  Branches                   1122     1164      +42     
========================================================
+ Hits                       7023     7170     +147     
- Misses                      566      642      +76     
- Partials                    248      257       +9     
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `88.85% <64.95%> (-0.76%)` | :arrow_down: |

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

| [Impacted Files](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | Coverage Δ | |
|---|---|---|
| [...gle/cloud/spanner/pgadapter/ConnectionHandler.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL0Nvbm5lY3Rpb25IYW5kbGVyLmphdmE=) | `88.41% <ø> (ø)` | |
| [...spanner/pgadapter/statements/ExecuteStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvRXhlY3V0ZVN0YXRlbWVudC5qYXZh) | `91.52% <ø> (ø)` | |
| [...d/spanner/pgadapter/wireprotocol/QueryMessage.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3dpcmVwcm90b2NvbC9RdWVyeU1lc3NhZ2UuamF2YQ==) | `100.00% <ø> (ø)` | |
| [...d/spanner/pgadapter/statements/FetchStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvRmV0Y2hTdGF0ZW1lbnQuamF2YQ==) | `28.57% <28.57%> (ø)` | |
| [...d/spanner/pgadapter/statements/CloseStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvQ2xvc2VTdGF0ZW1lbnQuamF2YQ==) | `44.11% <44.11%> (ø)` | |
| [...ud/spanner/pgadapter/statements/MoveStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvTW92ZVN0YXRlbWVudC5qYXZh) | `50.00% <50.00%> (ø)` | |
| [...d/spanner/pgadapter/wireprotocol/ParseMessage.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3dpcmVwcm90b2NvbC9QYXJzZU1lc3NhZ2UuamF2YQ==) | `86.07% <50.00%> (-7.08%)` | :arrow_down: |
| [...spanner/pgadapter/statements/DeclareStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvRGVjbGFyZVN0YXRlbWVudC5qYXZh) | `61.53% <61.53%> (ø)` | |
| [...apter/statements/AbstractFetchOrMoveStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvQWJzdHJhY3RGZXRjaE9yTW92ZVN0YXRlbWVudC5qYXZh) | `73.33% <73.33%> (ø)` | |
| [...oud/spanner/pgadapter/statements/SimpleParser.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvU2ltcGxlUGFyc2VyLmphdmE=) | `92.23% <89.28%> (-0.17%)` | :arrow_down: |
| ... and [1 more](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/797?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | |

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


Patch:
diff --git a/.github/workflows/integration.yaml b/.github/workflows/integration.yaml
index 8762611e..1bdcba13 100644
--- a/.github/workflows/integration.yaml
+++ b/.github/workflows/integration.yaml
@@ -52,6 +52,7 @@ jobs:
           - 5432:5432
         volumes:
           - /pg:/var/run/postgresql
+          - /tmp:/tmp
         options: >-
           --health-cmd pg_isready
           --health-interval 10s
diff --git a/README.md b/README.md
index 0f08c258..b764e28a 100644
--- a/README.md
+++ b/README.md
@@ -16,6 +16,7 @@ PGAdapter can be used with the following drivers and clients:
 1. `node-postgres`: Version 8.8.0 and higher are supported. See [node-postgres support](docs/node-postgres.md) for more details.
 1. `npgsql`: Version 6.0.x and higher have experimental support. See [npgsql support](docs/npgsql.md) for more details.
 1. `hibernate`: Version 5.3.20. Final and higher are supported. See [hibernate support](samples/java/hibernate/README.md) for more details.
+1. `postgres_fdw`: The PostgreSQL foreign data wrapper has experimental support. See [Foreign Data Wrapper sample](samples/foreign-data-wrapper) for more details.
 
 ## Frameworks and Tools
 PGAdapter can be used with the following frameworks and tools:
diff --git a/docs/faq.md b/docs/faq.md
index a4eba8e1..9b360cdf 100644
--- a/docs/faq.md
+++ b/docs/faq.md
@@ -133,6 +133,27 @@ PGAdapter supports `COPY table_name FROM STDIN [BINARY]` and `COPY table_name TO
 These commands can be used to copy data between different Cloud Spanner databases, or between a
 Cloud Spanner database and a PostgreSQL database. See [copy support](copy.md) for more information.
 
+## Can I use user-defined functions or stored procedures?
+PGAdapter and Cloud Spanner do not support user-defined functions or stored procedures. You can
+however add PGAdapter as a [foreign server](https://www.postgresql.org/docs/current/sql-createserver.html)
+to a normal PostgreSQL database and create user-defined functions and stored procedures in that
+database. Those functions can be used with data from Cloud Spanner, as all tables from Cloud Spanner
+can be [imported as foreign tables](https://www.postgresql.org/docs/current/sql-importforeignschema.html)
+into the PostgreSQL database.
+
+The local PostgreSQL database should be considered the same as your application. That means that it:
+1. Should only define code / logic. It should not store any actual data.
+2. Should be disposable. Dropping and re-creating it should always be possible.
+3. Should let Cloud Spanner do as much as possible of the data processing. It should only fetch the
+   rows from Cloud Spanner that it actually needs.
+
+__This feature can also be used as a workaround for built-in PostgreSQL functions that are not supported
+by Cloud Spanner.__
+
+See the [foreign data wrapper sample](../samples/foreign-data-wrapper) for a complete sample on how
+to set this up.
+
+
 ## How can I set a statement timeout?
 You can set the statement timeout that a connection should use by executing a
 `set statement_timeout=<timeout>` statement. The timeout is specified in milliseconds.
diff --git a/samples/foreign-data-wrapper/README.md b/samples/foreign-data-wrapper/README.md
new file mode 100644
index 00000000..0baef4c6
--- /dev/null
+++ b/samples/foreign-data-wrapper/README.md
@@ -0,0 +1,217 @@
+# Foreign Data Wrapper
+
+PGAdapter can be used as a foreign server in a real PostgreSQL database. Tables in Cloud Spanner can
+be added to a local PostgreSQL database as foreign tables. This enables the use of some PostgreSQL
+features that are not available in Cloud Spanner, such as user-defined functions, stored procedures,
+or missing system functions.
+
+Note that PostgreSQL in this case should be treated as a 'client'. That is; executing a user-defined
+function on data in a Cloud Spanner database requires PostgreSQL to fetch the data from Cloud Spanner
+and execute the function locally. You should therefore use a WHERE clause that will limit the data
+that is fetched as much as possible. The WHERE clause will be pushed down to Cloud Spanner, as long
+as the WHERE clause uses standard comparison operators and built-in functions.
+
+## Use Cases
+
+Adding PGAdapter as a foreign server in a real PostgreSQL database can be useful if you need access
+to PostgreSQL features that are not available in Cloud Spanner, such as:
+* Built-in functions and operators that are not available in Cloud Spanner.
+* User-defined functions.
+* Stored procedures.
+* PL/pgSQL procedural language.
+
+Note that many of the above features are commonly used in PostgreSQL to reduce the amount of data
+that needs to be transferred from the database server to a client for processing. This is not the
+case when using a foreign data wrapper. This feature should therefore be used with care.
+
+## Example
+
+This example shows how you can use Docker Compose to start a PGAdapter and a PostgreSQL container
+and link the two. Note that both containers are completely disposable and do not store any permanent
+data.
+
+1. Set up some environment variables. These will be used by all following steps.
+
+```shell
+export GOOGLE_APPLICATION_CREDENTIALS=/path/to/credentials.json
+export GOOGLE_CLOUD_PROJECT=my-project
+export SPANNER_INSTANCE=my-instance
+export SPANNER_DATABASE=my-database
+```
+
+2. (Optional) Create the Cloud Spanner PostgreSQL-dialect database if it does not already exist:
+
+```shell
+gcloud spanner databases create ${SPANNER_DATABASE} --instance=${SPANNER_INSTANCE} --database-dialect=POSTGRESQL
+```
+
+3. Start PGAdapter and PostgreSQL using Docker Compose. This will start both PGAdapter and PostgreSQL
+   in a Docker container and connect both to the same Docker network.
+
+```shell
+docker compose up -d
+```
+
+4. Create a table in the Cloud Spanner database and insert a couple of test rows.
+
+```shell
+docker run -it --rm \
+  --network foreign-data-wrapper_default \
+  postgres psql -h pgadapter \
+    -c "create table if not exists concerts (
+          id bigint not null primary key,
+          name varchar,
+          start_time timestamptz,
+          end_time timestamptz);" \
+    -c "insert into concerts values (1, 'Open Air', '2020-03-01 19:30+01'::timestamptz, '2020-03-02 01:00+01'::timestamptz)" \
+    -c "insert into concerts values (2, 'Closed Air', '2020-04-01 13:30+01'::timestamptz, '2020-04-01 18:00+01'::timestamptz)"
+```
+
+5. Add PGAdapter as a foreign server to the PostgreSQL instance and add the Cloud Spanner table as a
+   foreign table in the PostgreSQL database:
+
+```shell
+docker run -it --rm \
+  --network foreign-data-wrapper_default \
+  --env PGPASSWORD=mysecret \
+  postgres psql -h postgres -U postgres \
+  -c "-- Create the foreign data wrapper extension.
+      CREATE EXTENSION IF NOT EXISTS postgres_fdw;" \
+  -c "-- Create a server. It is highly recommended to create a 'read-only' server if you are only going to
+      -- query the foreign table. This will ensure that Cloud Spanner uses a read-only transaction for any
+      -- query that is executed on the foreign table.
+      CREATE SERVER IF NOT EXISTS pgadapter_read_only FOREIGN DATA WRAPPER postgres_fdw 
+          OPTIONS (
+              -- This is the host name of the PGAdapter container.
+              host 'pgadapter', port '5432',
+              -- Change to match your database id.
+              dbname 'my-database',
+              -- This instructs PGAdapter to use read-only transactions.
+              options '-c spanner.readonly=true');" \
+  -c "-- We must create a user mapping for the local user. The user is ignored by PGAdapter, unless PGAdapter
+      -- has been started with the command line argument that requires clients to authenticate.
+      CREATE USER MAPPING IF NOT EXISTS FOR postgres SERVER pgadapter_read_only
+          OPTIONS (user 'postgres', password_required 'false');" \
+  -c "-- Create a schema that we will use for the Cloud Spanner tables.
+      -- This makes it easier to just drop and re-create and re-import the schema if
+      -- the schema in Cloud Spanner has changed.
+      CREATE SCHEMA IF NOT EXISTS spanner;" \
+  -c "-- Switch to the spanner schema.
+      SET search_path TO spanner;" \
+  -c "-- Import the schema from Cloud Spanner.
+      IMPORT FOREIGN SCHEMA public from server pgadapter_read_only into spanner;" \
+  -c "-- Create a user-defined function in PostgreSQL.
+      CREATE FUNCTION timestamp_diff_minutes(earlier timestamptz, later timestamptz) RETURNS integer AS \$\$
+          SELECT (extract(epoch from later) - extract(epoch from earlier)) / 60;
+      \$\$ LANGUAGE SQL;" \
+  -c "-- Select data from Cloud Spanner and apply the user-defined function in data that is selected.
+      SELECT name, timestamp_diff_minutes(start_time, end_time) as duration_in_minutes
+      FROM concerts
+      WHERE name='Open Air'"
+```
+
+The following command will start an interactive psql session on the PostgreSQL database. This
+database has the `concerts` foreign table defined that actually resides in Cloud Spanner. You can
+create user-defined functions or stored procedures and use these with the data from Cloud Spanner.
+
+You can also use standard PostgreSQL functions that are not supported by Cloud Spanner. Normally,
+the PostgreSQL foreign data wrapper extension will push down the execution of built-in functions to
+the foreign server. You can prevent this by creating a user-defined function that is just a wrapper
+around the built-in function:
+
+```shell
+docker run -it --rm \
+  --network foreign-data-wrapper_default \
+  --env PGPASSWORD=mysecret \
+  postgres psql -h postgres -U postgres
+
+set search_path to spanner;
+
+create function to_char(ts timestamptz, format text) returns text as $$
+  select pg_catalog.to_char(ts, format);
+$$ language sql;
+
+select to_char(start_time, 'HH12:MI:SS') from concerts;
+```
+
+### Re-importing the schema
+Changes to the schema in Cloud Spanner are not directly visible in the imported schema in PostgreSQL.
+You can re-import the schema in PostgreSQL by dropping, re-creating and re-importing the schema in
+PostgreSQL:
+
+```sql
+drop schema spanner cascade;
+create schema spanner;
+import foreign schema public from server pgadapter_read_only into spanner;
+```
+
+## Limitations
+
+### Parameterized Queries for Foreign Tables
+You can use parameterized queries with Cloud Spanner foreign tables. The PostgreSQL Foreign Data
+Wrapper extension will however replace any parameter values with literals before sending the query
+to Cloud Spanner. That means that Cloud Spanner will see the query as an un-parameterized query,
+which again means that Cloud Spanner will need to re-compile the query each time the statement is
+executed with different parameter values.
+
+Example:
+
+```sql
+PREPARE my_statement AS SELECT * FROM f_concerts WHERE name=$1;
+EXECUTE my_statement ('Open Air');
+
+-- postgresql_fdw will internally translate the above prepared statement into the following query
+-- string that is sent to Cloud Spanner:
+SELECT * FROM f_concerts WHERE name='Open Air';
+```
+
+### Transactions
+The PostgreSQL Foreign Data Wrapper extension fetches data from Cloud Spanner using a CURSOR.
+Cursors can only be used in transactions. This means that postgres_fdw will always start a
+transaction before executing a query or any other statement on a foreign table. It is recommended to
+use a read-only transaction for statements that only read. The best way to achieve this is by
+defining a FOREIGN SERVER that uses a read-only connection to Cloud Spanner. You can specify this
+by adding the option `-c spanner.read_only=true` to the connection options:
+
+```sql
+CREATE SERVER IF NOT EXISTS pgadapter_read_only FOREIGN DATA WRAPPER postgres_fdw 
+    OPTIONS (
+        -- This assumes that PGAdapter runs on a host named 'pgadapter'.
+        host 'pgadapter', port '5432',
+        -- Change to match your database id.
+        dbname 'my-database',
+        -- This instructs PGAdapter to use read-only transactions.
+        options '-c spanner.readonly=true');
+```
+
+### Writing Data
+The PostgreSQL Foreign Data Wrapper extension also supports writing data to a foreign table. You
+should only use this if your insert/update/delete statement requires a feature that is not available
+in Cloud Spanner, as writing through a foreign table adds latency to each write operation in two
+ways:
+1. Each statement first has to go through the local PostgreSQL server to PGAdapter and then to Cloud Spanner.
+2. The PostgreSQL Foreign Data Wrapper extension replaces any query parameters with literals. This means
+   that Cloud Spanner needs to re-compile the statement each time it is executed with new values.
+
+#### Example
+
+```sql
+CREATE SERVER IF NOT EXISTS pgadapter_read_write FOREIGN DATA WRAPPER postgres_fdw 
+    OPTIONS (
+       -- This assumes that PGAdapter runs on a host named 'pgadapter'.
+        host 'pgadapter', port '5432',
+        -- Change to match your database id.
+        dbname 'my-database');
+
+CREATE USER MAPPING IF NOT EXISTS FOR postgres SERVER pgadapter_read_write
+   OPTIONS (user 'postgres', password_required 'false');
+
+CREATE FOREIGN TABLE IF NOT EXISTS concerts_writeable (id bigint, name varchar, start_time timestamptz, end_time timestamptz)
+    server pgadapter_read_write
+    options (table_name 'concerts');
+
+PREPARE insert_concert AS
+    INSERT INTO concerts_writeable (id, name, start_time, end_time)
+    VALUES ($1, $2, $3, $4);
+EXECUTE insert_concert (3, 'Clean Air', '2023-06-01 20:00:00+01'::timestamptz, '2023-06-02 02:30:00+01'::timestamptz);
+```
diff --git a/samples/foreign-data-wrapper/docker-compose.yml b/samples/foreign-data-wrapper/docker-compose.yml
new file mode 100644
index 00000000..4b150c83
--- /dev/null
+++ b/samples/foreign-data-wrapper/docker-compose.yml
@@ -0,0 +1,19 @@
+version: "3.9"
+services:
+  pgadapter:
+    image: "gcr.io/cloud-spanner-pg-adapter/pgadapter"
+    pull_policy: always
+    container_name: pgadapter
+    volumes:
+      - "${GOOGLE_APPLICATION_CREDENTIALS}:/credentials.json:ro"
+    command:
+      - "-p ${GOOGLE_CLOUD_PROJECT}"
+      - "-i ${SPANNER_INSTANCE}"
+      - "-d ${SPANNER_DATABASE}"
+      - "-c /credentials.json"
+      - "-x"
+  postgres:
+    image: "postgres"
+    container_name: postgres
+    environment:
+      POSTGRES_PASSWORD: mysecret
diff --git a/samples/java/hibernate/src/main/resources/sample-schema-sql b/samples/java/hibernate/src/main/resources/sample-schema-sql
index c6277777..d39ce909 100644
--- a/samples/java/hibernate/src/main/resources/sample-schema-sql
+++ b/samples/java/hibernate/src/main/resources/sample-schema-sql
@@ -5,7 +5,9 @@ create table if not exists singers (
     id         varchar not null primary key,
     first_name varchar,
     last_name  varchar not null,
-    full_name  varchar generated always as (coalesce(concat(first_name, ' '::varchar, last_name), last_name)) stored,
+    full_name  varchar generated always as (CASE WHEN first_name IS NULL THEN last_name
+                                                 WHEN last_name  IS NULL THEN first_name
+                                                 ELSE first_name || ' ' || last_name END) stored,
     active     boolean,
     created_at timestamptz,
     updated_at timestamptz
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 5d62b171..655d3e72 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java
@@ -448,6 +448,7 @@ public class ConnectionHandler extends Thread {
   public void handleTerminate() {
     synchronized (this) {
       closeAllPortals();
+      closeAllStatements();
       if (this.spannerConnection != null) {
         this.spannerConnection.close();
       }
@@ -510,7 +511,7 @@ public class ConnectionHandler extends Thread {
   }
 
   /** Closes all named and unnamed portals on this connection. */
-  private void closeAllPortals() {
+  public void closeAllPortals() {
     for (IntermediatePortalStatement statement : portalsMap.values()) {
       try {
         statement.close();
@@ -520,12 +521,12 @@ public class ConnectionHandler extends Thread {
       }
     }
     this.portalsMap.clear();
-    this.statementsMap.clear();
   }
 
   public IntermediatePortalStatement getPortal(String portalName) {
     if (!hasPortal(portalName)) {
-      throw new IllegalStateException("Unregistered portal: " + portalName);
+      throw PGExceptionFactory.newPGException(
+          "unrecognized portal name: " + portalName, SQLState.InvalidCursorName);
     }
     return this.portalsMap.get(portalName);
   }
@@ -536,9 +537,10 @@ public class ConnectionHandler extends Thread {
 
   public void closePortal(String portalName) throws Exception {
     if (!hasPortal(portalName)) {
-      throw new IllegalStateException("Unregistered portal: " + portalName);
+      throw PGExceptionFactory.newPGException(
+          "unrecognized portal name: " + portalName, SQLState.InvalidCursorName);
     }
-    IntermediatePortalStatement portal = this.portalsMap.get(portalName);
+    IntermediatePortalStatement portal = this.portalsMap.remove(portalName);
     portal.close();
   }
 
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/error/SQLState.java b/src/main/java/com/google/cloud/spanner/pgadapter/error/SQLState.java
index 21b20e01..86f6a91d 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/error/SQLState.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/error/SQLState.java
@@ -149,6 +149,9 @@ public enum SQLState {
   InFailedSqlTransaction("25P02"),
   IdleInTransactionSessionTimeout("25P03"),
 
+  // Class 34 - Cursor
+  InvalidCursorName("34000"),
+
   // Class 3B — Savepoint Exception
   SavepointException("3B000"),
   InvalidSavepointSpecification("3B001"),
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/AbstractFetchOrMoveStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/AbstractFetchOrMoveStatement.java
new file mode 100644
index 00000000..cb1a4005
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/AbstractFetchOrMoveStatement.java
@@ -0,0 +1,272 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import static com.google.cloud.spanner.pgadapter.statements.SimpleParser.unquoteOrFoldIdentifier;
+
+import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.error.PGExceptionFactory;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.common.base.Preconditions;
+import com.google.common.util.concurrent.Futures;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.Future;
+
+abstract class AbstractFetchOrMoveStatement extends IntermediatePortalStatement {
+  enum Direction {
+    FORWARD_ALL {
+      @Override
+      public int getDefaultCount() {
+        return 0;
+      }
+    },
+    BACKWARD_ALL {
+      @Override
+      boolean isSupported() {
+        return false;
+      }
+    },
+    NEXT,
+    PRIOR {
+      @Override
+      boolean isSupported() {
+        return false;
+      }
+    },
+    FIRST {
+      @Override
+      boolean isSupported() {
+        return false;
+      }
+    },
+    LAST {
+      @Override
+      boolean isSupported() {
+        return false;
+      }
+    },
+    ABSOLUTE(true) {
+      @Override
+      boolean isSupported() {
+        return false;
+      }
+    },
+    RELATIVE(true) {
+      @Override
+      boolean isSupported() {
+        return false;
+      }
+    },
+    ALL {
+      @Override
+      public int getDefaultCount() {
+        return 0;
+      }
+    },
+    FORWARD(true),
+    BACKWARD(true) {
+      @Override
+      boolean isSupported() {
+        return false;
+      }
+    };
+
+    final boolean supportsCount;
+
+    Direction() {
+      this(false);
+    }
+
+    Direction(boolean supportsCount) {
+      this.supportsCount = supportsCount;
+    }
+
+    boolean isSupported() {
+      return true;
+    }
+
+    public int getDefaultCount() {
+      return 1;
+    }
+
+    @Override
+    public String toString() {
+      return name().toLowerCase().replace('_', ' ');
+    }
+  }
+
+  abstract static class ParsedFetchOrMoveStatement {
+    final String name;
+    final Direction direction;
+    final Integer count;
+
+    ParsedFetchOrMoveStatement(String name, Direction direction, Integer count) {
+      this.name = name;
+      this.direction = direction;
+      this.count = count;
+    }
+  }
+
+  protected final ParsedFetchOrMoveStatement fetchOrMoveStatement;
+
+  public AbstractFetchOrMoveStatement(
+      String name,
+      IntermediatePreparedStatement preparedStatement,
+      byte[][] parameters,
+      List<Short> parameterFormatCodes,
+      List<Short> resultFormatCodes,
+      ParsedFetchOrMoveStatement fetchOrMoveStatement) {
+    super(name, preparedStatement, parameters, parameterFormatCodes, resultFormatCodes);
+    this.fetchOrMoveStatement = fetchOrMoveStatement;
+  }
+
+  @Override
+  public void executeAsync(BackendConnection backendConnection) {
+    if (!this.executed) {
+      try {
+        checkSupported();
+        int count;
+        if (fetchOrMoveStatement.count == null) {
+          count =
+              fetchOrMoveStatement.direction == null
+                  ? 1
+                  : fetchOrMoveStatement.direction.getDefaultCount();
+        } else {
+          count = fetchOrMoveStatement.count;
+        }
+        execute(backendConnection, count);
+        // Set a null result to indicate that this statement should not return any result.
+        setFutureStatementResult(Futures.immediateFuture(null));
+      } catch (Exception exception) {
+        setFutureStatementResult(Futures.immediateFailedFuture(exception));
+      }
+    }
+  }
+
+  protected void checkSupported() throws PGException {
+    if (fetchOrMoveStatement.count != null && fetchOrMoveStatement.count <= 0) {
+      throw PGExceptionFactory.newPGException(
+          "only positive count is supported", SQLState.FeatureNotSupported);
+    }
+    if (fetchOrMoveStatement.direction != null && !fetchOrMoveStatement.direction.isSupported()) {
+      throw PGExceptionFactory.newPGException(
+          fetchOrMoveStatement.direction + " is not supported", SQLState.FeatureNotSupported);
+    }
+  }
+
+  protected abstract void execute(BackendConnection backendConnection, int count) throws Exception;
+
+  @Override
+  public StatementType getStatementType() {
+    return StatementType.CLIENT_SIDE;
+  }
+
+  @Override
+  public Future<StatementResult> describeAsync(BackendConnection backendConnection) {
+    // Return null to indicate that the FETCH or MOVE statement does not return any
+    // RowDescriptionResponse.
+    return Futures.immediateFuture(null);
+  }
+
+  @Override
+  public IntermediatePortalStatement createPortal(
+      String name,
+      byte[][] parameters,
+      List<Short> parameterFormatCodes,
+      List<Short> resultFormatCodes) {
+    // FETCH and MOVE do not support binding any parameters, so we just return the same statement.
+    return this;
+  }
+
+  static <T extends ParsedFetchOrMoveStatement> T parse(String sql, String type, Class<T> clazz) {
+    Preconditions.checkNotNull(sql);
+    Preconditions.checkNotNull(type);
+
+    // {MOVE | FETCH} [ direction ] [ FROM | IN ] cursor_name
+    SimpleParser parser = new SimpleParser(sql);
+    if (!parser.eatKeyword(type)) {
+      throw PGExceptionFactory.newPGException(
+          "not a valid " + type.toUpperCase() + " statement: " + sql, SQLState.SyntaxError);
+    }
+    Direction direction;
+    if (parser.eatKeyword("forward")) {
+      if (parser.eatKeyword("all")) {
+        direction = Direction.FORWARD_ALL;
+      } else {
+        direction = Direction.FORWARD;
+      }
+    } else if (parser.eatKeyword("backward")) {
+      if (parser.eatKeyword("all")) {
+        direction = Direction.BACKWARD_ALL;
+      } else {
+        direction = Direction.BACKWARD;
+      }
+    } else {
+      direction =
+          Arrays.stream(Direction.values())
+              // This ensures that the stream will return the first valid direction keyword that it
+              // finds, or null if no valid direction keyword is found. If no valid direction
+              // keyword is found, then the position of the parser will also not be moved.
+              .filter(dir -> parser.eatKeyword(dir.name()))
+              .findFirst()
+              .orElse(null);
+    }
+    Long count = null;
+    if (parser.peekNumericLiteral()) {
+      count = parser.readIntegerLiteral();
+      if (count == null) {
+        throw PGExceptionFactory.newPGException("syntax error: " + sql, SQLState.SyntaxError);
+      }
+      if (count > Integer.MAX_VALUE || count < Integer.MIN_VALUE) {
+        throw PGExceptionFactory.newPGException(
+            "count out of range: " + count, SQLState.NumericValueOutOfRange);
+      }
+    }
+    if (count != null && direction != null && !direction.supportsCount) {
+      throw PGExceptionFactory.newPGException(
+          "unexpected <count> argument: " + sql, SQLState.SyntaxError);
+    } else if (count == null && direction == Direction.ABSOLUTE) {
+      throw PGExceptionFactory.newPGException(
+          "missing or invalid <count> argument for ABSOLUTE: " + sql, SQLState.SyntaxError);
+    } else if (count == null && direction == Direction.RELATIVE) {
+      throw PGExceptionFactory.newPGException(
+          "missing or invalid <count> argument for RELATIVE: " + sql, SQLState.SyntaxError);
+    }
+    // Skip 'from' or 'in'.
+    boolean ignore = parser.eatKeyword("from") || parser.eatKeyword("in");
+    TableOrIndexName name = parser.readTableOrIndexName();
+    if (name == null || name.schema != null) {
+      throw PGExceptionFactory.newPGException("invalid cursor name: " + sql, SQLState.SyntaxError);
+    }
+    if (parser.hasMoreTokens()) {
+      throw PGExceptionFactory.newPGException(
+          "unexpected tokens after cursor name: " + sql, SQLState.SyntaxError);
+    }
+
+    Integer integerCount = count == null ? null : count.intValue();
+    try {
+      return clazz
+          .getDeclaredConstructor(String.class, Direction.class, Integer.class)
+          .newInstance(unquoteOrFoldIdentifier(name.name), direction, integerCount);
+    } catch (Exception exception) {
+      throw PGExceptionFactory.newPGException(
+          "internal error: " + exception.getMessage(), SQLState.InternalError);
+    }
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
index cb153959..617a1cc1 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
@@ -15,9 +15,11 @@
 package com.google.cloud.spanner.pgadapter.statements;
 
 import static com.google.cloud.spanner.pgadapter.error.PGExceptionFactory.toPGException;
+import static com.google.cloud.spanner.pgadapter.statements.IntermediateStatement.PARSER;
 
 import com.google.api.core.InternalApi;
 import com.google.cloud.ByteArray;
+import com.google.cloud.Tuple;
 import com.google.cloud.spanner.BatchClient;
 import com.google.cloud.spanner.BatchReadOnlyTransaction;
 import com.google.cloud.spanner.BatchTransactionId;
@@ -141,6 +143,23 @@ public class BackendConnection {
     return pgException;
   }
 
+  boolean shouldReplaceStatement(Statement statement) {
+    if (!localStatements.get().isEmpty() && localStatements.get().containsKey(statement.getSql())) {
+      LocalStatement localStatement = localStatements.get().get(statement.getSql());
+      if (localStatement != null) {
+        return localStatement.hasReplacementStatement();
+      }
+    }
+    return false;
+  }
+
+  Tuple<Statement, ParsedStatement> replaceStatement(Statement statement) {
+    LocalStatement localStatement = localStatements.get().get(statement.getSql());
+    Statement replacement =
+        Objects.requireNonNull(localStatement).getReplacementStatement(statement);
+    return Tuple.of(replacement, Objects.requireNonNull(PARSER.parse(replacement)));
+  }
+
   /**
    * Buffered statements are kept in memory until a flush or sync message is received. This makes it
    * possible to batch multiple statements together when sending them to Cloud Spanner.
@@ -151,6 +170,11 @@ public class BackendConnection {
     final SettableFuture<T> result;
 
     BufferedStatement(ParsedStatement parsedStatement, Statement statement) {
+      if (shouldReplaceStatement(statement)) {
+        Tuple<Statement, ParsedStatement> replacement = replaceStatement(statement);
+        statement = replacement.x();
+        parsedStatement = replacement.y();
+      }
       this.parsedStatement = parsedStatement;
       this.statement = statement;
       this.result = SettableFuture.create();
@@ -216,10 +240,13 @@ public class BackendConnection {
         //  block.
         SessionStatement sessionStatement = getSessionManagementStatement(parsedStatement);
         if (!localStatements.get().isEmpty()
-            && localStatements.get().containsKey(statement.getSql())) {
-          result.set(
-              Objects.requireNonNull(localStatements.get().get(statement.getSql()))
-                  .execute(BackendConnection.this));
+            && localStatements.get().containsKey(statement.getSql())
+            && localStatements.get().get(statement.getSql()) != null
+            && !Objects.requireNonNull(localStatements.get().get(statement.getSql()))
+                .hasReplacementStatement()) {
+          LocalStatement localStatement =
+              Objects.requireNonNull(localStatements.get().get(statement.getSql()));
+          result.set(localStatement.execute(BackendConnection.this));
         } else if (sessionStatement != null) {
           result.set(sessionStatement.execute(sessionState));
         } else if (connectionState == ConnectionState.ABORTED
@@ -684,6 +711,7 @@ public class BackendConnection {
   private static final StatementResult ROLLBACK_RESULT = new NoResult("ROLLBACK");
   private static final Statement ROLLBACK = Statement.of("ROLLBACK");
 
+  private final Runnable closeAllPortals;
   private final SessionState sessionState;
   private final Supplier<PgCatalog> pgCatalog;
   private final Supplier<ImmutableMap<String, LocalStatement>> localStatements;
@@ -697,11 +725,13 @@ public class BackendConnection {
 
   /** Creates a PG backend connection that uses the given Spanner {@link Connection} and options. */
   BackendConnection(
+      Runnable closeAllPortals,
       DatabaseId databaseId,
       Connection spannerConnection,
       Supplier<WellKnownClient> wellKnownClient,
       OptionsMetadata optionsMetadata,
       Supplier<ImmutableList<LocalStatement>> localStatements) {
+    this.closeAllPortals = closeAllPortals;
     this.sessionState = new SessionState(optionsMetadata);
     this.pgCatalog =
         Suppliers.memoize(
@@ -936,6 +966,7 @@ public class BackendConnection {
             } else {
               sessionState.rollback();
             }
+            closeAllPortals.run();
             transactionMode = TransactionMode.IMPLICIT;
             connectionState = ConnectionState.IDLE;
           }
@@ -944,6 +975,7 @@ public class BackendConnection {
       }
     } catch (Exception exception) {
       connectionState = ConnectionState.ABORTED;
+      closeAllPortals.run();
       sessionState.rollback();
       if (spannerConnection.isInTransaction()) {
         if (spannerConnection.isDmlBatchActive()) {
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/CloseStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CloseStatement.java
new file mode 100644
index 00000000..19c0b8dd
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/CloseStatement.java
@@ -0,0 +1,127 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import static com.google.cloud.spanner.pgadapter.statements.SimpleParser.unquoteOrFoldIdentifier;
+
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.error.PGExceptionFactory;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.NoResult;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Futures;
+import java.util.List;
+import java.util.concurrent.Future;
+
+public class CloseStatement extends IntermediatePortalStatement {
+  static final class ParsedCloseStatement {
+    final String name;
+
+    private ParsedCloseStatement(String name) {
+      this.name = name;
+    }
+  }
+
+  private final ParsedCloseStatement closeStatement;
+
+  public CloseStatement(
+      ConnectionHandler connectionHandler,
+      OptionsMetadata options,
+      String name,
+      ParsedStatement parsedStatement,
+      Statement originalStatement) {
+    super(
+        name,
+        new IntermediatePreparedStatement(
+            connectionHandler,
+            options,
+            name,
+            NO_PARAMETER_TYPES,
+            parsedStatement,
+            originalStatement),
+        NO_PARAMS,
+        ImmutableList.of(),
+        ImmutableList.of());
+    this.closeStatement = parse(originalStatement.getSql());
+  }
+
+  @Override
+  public String getCommandTag() {
+    return "CLOSE CURSOR";
+  }
+
+  @Override
+  public StatementType getStatementType() {
+    return StatementType.CLIENT_SIDE;
+  }
+
+  @Override
+  public void executeAsync(BackendConnection backendConnection) {
+    this.executed = true;
+    try {
+      if (closeStatement.name == null) {
+        connectionHandler.closeAllPortals();
+      } else {
+        connectionHandler.closePortal(closeStatement.name);
+      }
+    } catch (Exception exception) {
+      setFutureStatementResult(Futures.immediateFailedFuture(exception));
+      return;
+    }
+    setFutureStatementResult(Futures.immediateFuture(new NoResult()));
+  }
+
+  @Override
+  public Future<StatementResult> describeAsync(BackendConnection backendConnection) {
+    // Return null to indicate that this CLOSE statement does not return any
+    // RowDescriptionResponse.
+    return Futures.immediateFuture(null);
+  }
+
+  @Override
+  public IntermediatePortalStatement createPortal(
+      String name,
+      byte[][] parameters,
+      List<Short> parameterFormatCodes,
+      List<Short> resultFormatCodes) {
+    // CLOSE does not support binding any parameters, so we just return the same statement.
+    return this;
+  }
+
+  static ParsedCloseStatement parse(String sql) {
+    Preconditions.checkNotNull(sql);
+
+    SimpleParser parser = new SimpleParser(sql);
+    if (!parser.eatKeyword("close")) {
+      throw PGExceptionFactory.newPGException("not a valid CLOSE statement: " + sql);
+    }
+    String statementName = null;
+    if (!parser.eatKeyword("all")) {
+      TableOrIndexName name = parser.readTableOrIndexName();
+      if (name == null || name.schema != null) {
+        throw PGExceptionFactory.newPGException("invalid cursor name");
+      }
+      statementName = unquoteOrFoldIdentifier(name.name);
+    }
+    parser.throwIfHasMoreTokens();
+    return new ParsedCloseStatement(statementName);
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/DeclareStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/DeclareStatement.java
new file mode 100644
index 00000000..32d4a1f3
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/DeclareStatement.java
@@ -0,0 +1,260 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import static com.google.cloud.spanner.pgadapter.statements.SimpleParser.unquoteOrFoldIdentifier;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.error.PGExceptionFactory;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.ConnectionState;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.NoResult;
+import com.google.cloud.spanner.pgadapter.statements.DeclareStatement.ParsedDeclareStatement.Builder;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.cloud.spanner.pgadapter.wireprotocol.BindMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.ManuallyCreatedToken;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.util.concurrent.Futures;
+import java.util.List;
+import java.util.concurrent.Future;
+
+@InternalApi
+public class DeclareStatement extends IntermediatePortalStatement {
+  enum Sensitivity {
+    ASENSITIVE,
+    INSENSITIVE,
+  }
+
+  enum Scroll {
+    SCROLL,
+    NO_SCROLL,
+  }
+
+  enum Holdability {
+    HOLD,
+    NO_HOLD,
+  }
+
+  static final class ParsedDeclareStatement {
+    static final class Builder {
+      String name;
+      boolean binary;
+      Sensitivity sensitivity;
+      Scroll scroll;
+      Holdability holdability;
+      String sql;
+    }
+
+    final String name;
+    final boolean binary;
+    final Sensitivity sensitivity;
+    final Scroll scroll;
+    final Holdability holdability;
+    final Statement originalPreparedStatement;
+    final ParsedStatement parsedPreparedStatement;
+
+    private ParsedDeclareStatement(Builder builder) {
+      this.name = builder.name;
+      this.binary = builder.binary;
+      this.sensitivity = builder.sensitivity;
+      this.scroll = builder.scroll;
+      this.holdability = builder.holdability;
+      this.originalPreparedStatement = Statement.of(builder.sql);
+      this.parsedPreparedStatement =
+          AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(originalPreparedStatement);
+    }
+  }
+
+  private final ParsedDeclareStatement declareStatement;
+
+  public DeclareStatement(
+      ConnectionHandler connectionHandler,
+      OptionsMetadata options,
+      String name,
+      ParsedStatement parsedStatement,
+      Statement originalStatement) {
+    super(
+        name,
+        new IntermediatePreparedStatement(
+            connectionHandler,
+            options,
+            name,
+            NO_PARAMETER_TYPES,
+            parsedStatement,
+            originalStatement),
+        NO_PARAMS,
+        ImmutableList.of(),
+        ImmutableList.of());
+    this.declareStatement = parse(originalStatement.getSql());
+  }
+
+  @Override
+  public String getCommandTag() {
+    return "DECLARE CURSOR";
+  }
+
+  @Override
+  public StatementType getStatementType() {
+    return StatementType.CLIENT_SIDE;
+  }
+
+  @Override
+  public void executeAsync(BackendConnection backendConnection) {
+    if (!this.executed) {
+      this.executed = true;
+      try {
+        if (!declareStatement.parsedPreparedStatement.isQuery()) {
+          throw PGExceptionFactory.newPGException(
+              "cursors can only be defined for queries", SQLState.FeatureNotSupported);
+        }
+        if (declareStatement.scroll == Scroll.SCROLL) {
+          throw PGExceptionFactory.newPGException(
+              "scrollable cursors are not supported", SQLState.FeatureNotSupported);
+        }
+        if (declareStatement.holdability == Holdability.HOLD) {
+          throw PGExceptionFactory.newPGException(
+              "holdable cursors are not supported", SQLState.FeatureNotSupported);
+        }
+        if (backendConnection.getConnectionState() == ConnectionState.IDLE) {
+          throw PGExceptionFactory.newPGException(
+              "DECLARE CURSOR can only be used in transaction blocks",
+              SQLState.NoActiveSqlTransaction);
+        }
+        if (backendConnection.getConnectionState() == ConnectionState.ABORTED) {
+          throw PGExceptionFactory.newTransactionAbortedException();
+        }
+        new ParseMessage(
+                connectionHandler,
+                "",
+                new int[0],
+                declareStatement.parsedPreparedStatement,
+                declareStatement.originalPreparedStatement)
+            .send();
+        new BindMessage(
+                connectionHandler,
+                "",
+                declareStatement.name,
+                new byte[0][],
+                ManuallyCreatedToken.MANUALLY_CREATED_TOKEN)
+            .send();
+      } catch (Exception exception) {
+        setFutureStatementResult(Futures.immediateFailedFuture(exception));
+        return;
+      }
+      setFutureStatementResult(Futures.immediateFuture(new NoResult()));
+    }
+  }
+
+  @Override
+  public Future<StatementResult> describeAsync(BackendConnection backendConnection) {
+    // Return null to indicate that this DECLARE statement does not return any
+    // RowDescriptionResponse.
+    return Futures.immediateFuture(null);
+  }
+
+  @Override
+  public IntermediatePortalStatement createPortal(
+      String name,
+      byte[][] parameters,
+      List<Short> parameterFormatCodes,
+      List<Short> resultFormatCodes) {
+    // DECLARE does not support binding any parameters, so we just return the same statement.
+    return this;
+  }
+
+  static ParsedDeclareStatement parse(String sql) {
+    Preconditions.checkNotNull(sql);
+
+    // DECLARE name [ BINARY ] [ ASENSITIVE | INSENSITIVE ] [ [ NO ] SCROLL ]
+    //    CURSOR [ { WITH | WITHOUT } HOLD ] FOR query
+    // The keywords ASENSITIVE, BINARY, INSENSITIVE, and SCROLL can appear in any order.
+    SimpleParser parser = new SimpleParser(sql);
+    if (!parser.eatKeyword("declare")) {
+      throw PGExceptionFactory.newPGException(
+          "not a valid DECLARE statement: " + sql, SQLState.SyntaxError);
+    }
+    Builder builder = new Builder();
+    TableOrIndexName name = parser.readTableOrIndexName();
+    if (name == null || name.schema != null) {
+      throw PGExceptionFactory.newPGException("invalid cursor name: " + sql, SQLState.SyntaxError);
+    }
+    builder.name = unquoteOrFoldIdentifier(name.name);
+    while (parser.hasMoreTokens()
+        && !parser.peekKeyword("cursor")
+        && !parser.peekKeyword("for")
+        && !parser.peekKeyword("with")
+        && !parser.peekKeyword("without")) {
+      if (parser.eatKeyword("binary")) {
+        builder.binary = true;
+      } else if (parser.eatKeyword("asensitive")) {
+        if (builder.sensitivity == Sensitivity.INSENSITIVE) {
+          throw PGExceptionFactory.newPGException(
+              "cannot specify both INSENSITIVE and ASENSITIVE", SQLState.SyntaxError);
+        }
+        builder.sensitivity = Sensitivity.ASENSITIVE;
+      } else if (parser.eatKeyword("insensitive")) {
+        if (builder.sensitivity == Sensitivity.ASENSITIVE) {
+          throw PGExceptionFactory.newPGException(
+              "cannot specify both INSENSITIVE and ASENSITIVE", SQLState.SyntaxError);
+        }
+        builder.sensitivity = Sensitivity.INSENSITIVE;
+      } else if (parser.eatKeyword("scroll")) {
+        if (builder.scroll == Scroll.NO_SCROLL) {
+          throw PGExceptionFactory.newPGException(
+              "cannot specify both SCROLL and NO SCROLL", SQLState.SyntaxError);
+        }
+        builder.scroll = Scroll.SCROLL;
+      } else if (parser.eatKeyword("no", "scroll")) {
+        if (builder.scroll == Scroll.SCROLL) {
+          throw PGExceptionFactory.newPGException(
+              "cannot specify both SCROLL and NO SCROLL", SQLState.SyntaxError);
+        }
+        builder.scroll = Scroll.NO_SCROLL;
+      } else {
+        throw PGExceptionFactory.newPGException("syntax error: " + sql, SQLState.SyntaxError);
+      }
+    }
+    if (!parser.eatKeyword("cursor")) {
+      throw PGExceptionFactory.newPGException(
+          "missing expected CURSOR keyword: " + sql, SQLState.SyntaxError);
+    }
+    if (parser.eatKeyword("with", "hold")) {
+      builder.holdability = Holdability.HOLD;
+    } else if (parser.eatKeyword("without", "hold")) {
+      builder.holdability = Holdability.NO_HOLD;
+    }
+    if (!parser.eatKeyword("for")) {
+      throw PGExceptionFactory.newPGException(
+          "missing expected FOR keyword: " + sql, SQLState.SyntaxError);
+    }
+    if (!parser.hasMoreTokens()) {
+      throw PGExceptionFactory.newPGException(
+          "missing query for cursor: " + sql, SQLState.SyntaxError);
+    }
+    builder.sql = parser.getSql().substring(parser.getPos()).trim();
+
+    return new ParsedDeclareStatement(builder);
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
index aea67a98..f6722d15 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
@@ -39,6 +39,7 @@ public class ExtendedQueryProtocolHandler {
     this.connectionHandler = Preconditions.checkNotNull(connectionHandler);
     this.backendConnection =
         new BackendConnection(
+            connectionHandler::closeAllPortals,
             connectionHandler.getDatabaseId(),
             connectionHandler.getSpannerConnection(),
             connectionHandler::getWellKnownClient,
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/FetchStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/FetchStatement.java
new file mode 100644
index 00000000..21bfec3e
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/FetchStatement.java
@@ -0,0 +1,78 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.ManuallyCreatedToken;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.PreparedType;
+import com.google.cloud.spanner.pgadapter.wireprotocol.DescribeMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ExecuteMessage;
+import com.google.common.collect.ImmutableList;
+
+public class FetchStatement extends AbstractFetchOrMoveStatement {
+
+  public FetchStatement(
+      ConnectionHandler connectionHandler,
+      OptionsMetadata options,
+      String name,
+      ParsedStatement parsedStatement,
+      Statement originalStatement) {
+    super(
+        name,
+        new IntermediatePreparedStatement(
+            connectionHandler,
+            options,
+            name,
+            NO_PARAMETER_TYPES,
+            parsedStatement,
+            originalStatement),
+        NO_PARAMS,
+        ImmutableList.of(),
+        ImmutableList.of(),
+        parse(originalStatement.getSql(), "fetch", ParsedFetchStatement.class));
+  }
+
+  @Override
+  public String getCommandTag() {
+    return "FETCH";
+  }
+
+  @Override
+  protected void execute(BackendConnection backendConnection, int count) throws Exception {
+    new DescribeMessage(
+            connectionHandler,
+            PreparedType.Portal,
+            fetchOrMoveStatement.name,
+            ManuallyCreatedToken.MANUALLY_CREATED_TOKEN)
+        .send();
+    new ExecuteMessage(
+            connectionHandler,
+            fetchOrMoveStatement.name,
+            count,
+            "FETCH",
+            false,
+            ManuallyCreatedToken.MANUALLY_CREATED_TOKEN)
+        .send();
+  }
+
+  static class ParsedFetchStatement extends ParsedFetchOrMoveStatement {
+    ParsedFetchStatement(String name, Direction direction, Integer count) {
+      super(name, direction, count);
+    }
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java
index 0c3a3b16..b5492389 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediatePortalStatement.java
@@ -21,6 +21,7 @@ import com.google.cloud.spanner.connection.StatementResult.ResultType;
 import com.google.cloud.spanner.pgadapter.parsers.Parser;
 import com.google.cloud.spanner.pgadapter.parsers.Parser.FormatCode;
 import com.google.cloud.spanner.pgadapter.statements.BackendConnection.NoResult;
+import com.google.common.util.concurrent.Futures;
 import java.util.List;
 import java.util.concurrent.Future;
 
@@ -133,10 +134,13 @@ public class IntermediatePortalStatement extends IntermediatePreparedStatement {
     // DescribePortal message without a following Execute message is extremely rare, as that would
     // only happen if the client is ill-behaved, or if the client crashes between the
     // DescribePortal and Execute.
-    Future<StatementResult> statementResultFuture =
-        backendConnection.execute(this.parsedStatement, this.statement, this::bind);
-    setFutureStatementResult(statementResultFuture);
-    this.executed = true;
-    return statementResultFuture;
+    if (futureStatementResult == null && getStatementResult() == null) {
+      Future<StatementResult> statementResultFuture =
+          backendConnection.execute(this.parsedStatement, this.statement, this::bind);
+      setFutureStatementResult(statementResultFuture);
+      this.executed = true;
+      return statementResultFuture;
+    }
+    return Futures.immediateFuture(getStatementResult());
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/MoveStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/MoveStatement.java
new file mode 100644
index 00000000..28d5908c
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/MoveStatement.java
@@ -0,0 +1,75 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ControlMessage.ManuallyCreatedToken;
+import com.google.cloud.spanner.pgadapter.wireprotocol.ExecuteMessage;
+import com.google.common.collect.ImmutableList;
+
+/**
+ * MOVE is the same as FETCH, except it just skips the results instead of actually sending the rows
+ * back to the client.
+ */
+public class MoveStatement extends AbstractFetchOrMoveStatement {
+  static final String MOVE_COMMAND_TAG = "MOVE";
+
+  public MoveStatement(
+      ConnectionHandler connectionHandler,
+      OptionsMetadata options,
+      String name,
+      ParsedStatement parsedStatement,
+      Statement originalStatement) {
+    super(
+        name,
+        new IntermediatePreparedStatement(
+            connectionHandler,
+            options,
+            name,
+            NO_PARAMETER_TYPES,
+            parsedStatement,
+            originalStatement),
+        NO_PARAMS,
+        ImmutableList.of(),
+        ImmutableList.of(),
+        parse(originalStatement.getSql(), "move", ParsedMoveStatement.class));
+  }
+
+  @Override
+  public String getCommandTag() {
+    return MOVE_COMMAND_TAG;
+  }
+
+  @Override
+  protected void execute(BackendConnection backendConnection, int maxRows) throws Exception {
+    new ExecuteMessage(
+            connectionHandler,
+            fetchOrMoveStatement.name,
+            maxRows,
+            MOVE_COMMAND_TAG,
+            false,
+            ManuallyCreatedToken.MANUALLY_CREATED_TOKEN)
+        .send();
+  }
+
+  static class ParsedMoveStatement extends ParsedFetchOrMoveStatement {
+    ParsedMoveStatement(String name, Direction direction, Integer count) {
+      super(name, direction, count);
+    }
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
index f86d4e59..7efb65e0 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
@@ -836,6 +836,53 @@ public class SimpleParser {
     }
   }
 
+  Long readIntegerLiteral() {
+    skipWhitespaces();
+    int startPos = pos;
+    if (skipNumericLiteral()) {
+      try {
+        return Long.valueOf(sql.substring(startPos, pos));
+      } catch (NumberFormatException ignore) {
+        return null;
+      }
+    }
+    return null;
+  }
+
+  boolean peekNumericLiteral() {
+    int originalPos = pos;
+    if (skipNumericLiteral()) {
+      pos = originalPos;
+      return true;
+    }
+    return false;
+  }
+
+  boolean skipNumericLiteral() {
+    skipWhitespaces();
+    int startPos = pos;
+    if (eatToken("+") || eatToken("-")) {
+      startPos++;
+    }
+    if (eatKeyword("inf")) {
+      return true;
+    }
+    boolean foundDecimalDot = false;
+    while (pos < sql.length()) {
+      if (!foundDecimalDot && sql.charAt(pos) == '.') {
+        foundDecimalDot = true;
+        pos++;
+      } else if (Character.isDigit(sql.charAt(pos))) {
+        pos++;
+      } else if (!isValidEndOfKeyword(pos)) {
+        return false;
+      } else {
+        break;
+      }
+    }
+    return pos > startPos;
+  }
+
   QuotedString readSingleQuotedString() {
     return readQuotedString(SINGLE_QUOTE, false);
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/AbortTransaction.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/AbortTransaction.java
new file mode 100644
index 00000000..867ec963
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/AbortTransaction.java
@@ -0,0 +1,53 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements.local;
+
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
+
+// TODO: Remove this once support for ABORT TRANSACTION has been added to the Connection API.
+/** Simple replacement for ABORT TRANSACTION. */
+public class AbortTransaction implements LocalStatement {
+  public static final AbortTransaction INSTANCE = new AbortTransaction();
+  private static final Statement REPLACEMENT_STATEMENT = Statement.of("ROLLBACK");
+
+  // This is the very specific command string that postgres_fdw uses.
+  // https://github.com/postgres/postgres/blob/503b0556d96f2c8df6ed91c5a8cf11b23f37ce6d/contrib/postgres_fdw/connection.c#L98
+  private static final String FDW_ABORT_COMMAND = "ABORT TRANSACTION";
+
+  private AbortTransaction() {}
+
+  @Override
+  public String[] getSql() {
+    return new String[] {FDW_ABORT_COMMAND};
+  }
+
+  @Override
+  public boolean hasReplacementStatement() {
+    return true;
+  }
+
+  @Override
+  public Statement getReplacementStatement(Statement statement) {
+    return REPLACEMENT_STATEMENT;
+  }
+
+  @Override
+  public StatementResult execute(BackendConnection backendConnection) {
+    // This should not be possible.
+    throw new UnsupportedOperationException();
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/LocalStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/LocalStatement.java
index 7e853b4e..e0439d17 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/LocalStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/LocalStatement.java
@@ -14,6 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter.statements.local;
 
+import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
 
@@ -32,6 +33,19 @@ public interface LocalStatement {
    */
   String[] getSql();
 
+  /** Returns true if the statement has a static replacement SQL string. */
+  default boolean hasReplacementStatement() {
+    return false;
+  }
+
+  /**
+   * Returns the static replacement statement for this local statement, or null if {@link
+   * #hasReplacementStatement()} returns false.
+   */
+  default Statement getReplacementStatement(Statement statement) {
+    return null;
+  }
+
   /** Executes the local statement and returns the result. */
   StatementResult execute(BackendConnection backendConnection);
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/StartTransactionIsolationLevelRepeatableRead.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/StartTransactionIsolationLevelRepeatableRead.java
new file mode 100644
index 00000000..c3409285
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/local/StartTransactionIsolationLevelRepeatableRead.java
@@ -0,0 +1,54 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements.local;
+
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
+
+/** Simple replacement for statements that start a repeatable-read transaction. */
+public class StartTransactionIsolationLevelRepeatableRead implements LocalStatement {
+  public static final StartTransactionIsolationLevelRepeatableRead INSTANCE =
+      new StartTransactionIsolationLevelRepeatableRead();
+  private static final Statement REPLACEMENT_STATEMENT = Statement.of("BEGIN");
+
+  // This is the very specific command string that postgres_fdw uses.
+  // https://github.com/postgres/postgres/blob/503b0556d96f2c8df6ed91c5a8cf11b23f37ce6d/contrib/postgres_fdw/connection.c#L742
+  private static final String FDW_START_COMMAND =
+      "START TRANSACTION ISOLATION LEVEL REPEATABLE READ";
+
+  private StartTransactionIsolationLevelRepeatableRead() {}
+
+  @Override
+  public String[] getSql() {
+    return new String[] {FDW_START_COMMAND};
+  }
+
+  @Override
+  public boolean hasReplacementStatement() {
+    return true;
+  }
+
+  @Override
+  public Statement getReplacementStatement(Statement statement) {
+    return REPLACEMENT_STATEMENT;
+  }
+
+  @Override
+  public StatementResult execute(BackendConnection backendConnection) {
+    // This should not be possible.
+    throw new UnsupportedOperationException();
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java b/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java
index 05a797b8..498fb799 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetector.java
@@ -21,6 +21,7 @@ import com.google.cloud.spanner.pgadapter.error.PGException;
 import com.google.cloud.spanner.pgadapter.error.SQLState;
 import com.google.cloud.spanner.pgadapter.session.PGSetting;
 import com.google.cloud.spanner.pgadapter.statements.PgCatalog.PgCatalogTable;
+import com.google.cloud.spanner.pgadapter.statements.local.AbortTransaction;
 import com.google.cloud.spanner.pgadapter.statements.local.DjangoGetTableNamesStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.ListDatabasesStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.LocalStatement;
@@ -28,6 +29,7 @@ import com.google.cloud.spanner.pgadapter.statements.local.SelectCurrentCatalogS
 import com.google.cloud.spanner.pgadapter.statements.local.SelectCurrentDatabaseStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.SelectCurrentSchemaStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.SelectVersionStatement;
+import com.google.cloud.spanner.pgadapter.statements.local.StartTransactionIsolationLevelRepeatableRead;
 import com.google.cloud.spanner.pgadapter.wireoutput.NoticeResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.NoticeResponse.NoticeSeverity;
 import com.google.cloud.spanner.pgadapter.wireprotocol.ParseMessage;
@@ -38,6 +40,7 @@ import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import java.time.Duration;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.regex.Pattern;
@@ -85,6 +88,43 @@ public class ClientAutoDetector {
         return ImmutableList.of(new ListDatabasesStatement(connectionHandler));
       }
     },
+    PG_FDW {
+      final ImmutableList<QueryPartReplacer> functionReplacements =
+          ImmutableList.of(
+              RegexQueryPartReplacer.replace(
+                  Pattern.compile("format_type\\s*\\(\\s*atttypid\\s*,\\s*atttypmod\\s*\\)"),
+                  "spanner_type as format_type"),
+              RegexQueryPartReplacer.replace(
+                  Pattern.compile("pg_get_expr\\s*\\(\\s*adbin\\s*,\\s*adrelid\\s*\\)"),
+                  "adbin as pg_get_expr"));
+
+      @Override
+      boolean isClient(List<String> orderedParameterKeys, Map<String, String> parameters) {
+        // postgres_fdw by default sends its own name.
+        return parameters.containsKey("application_name")
+            && parameters
+                .get("application_name")
+                .toLowerCase(Locale.ENGLISH)
+                .contains("postgres_fdw");
+      }
+
+      @Override
+      public ImmutableList<LocalStatement> getLocalStatements(ConnectionHandler connectionHandler) {
+        if (connectionHandler.getServer().getOptions().useDefaultLocalStatements()) {
+          return ImmutableList.<LocalStatement>builder()
+              .addAll(DEFAULT_LOCAL_STATEMENTS)
+              .add(StartTransactionIsolationLevelRepeatableRead.INSTANCE)
+              .add(AbortTransaction.INSTANCE)
+              .build();
+        }
+        return ImmutableList.of(StartTransactionIsolationLevelRepeatableRead.INSTANCE);
+      }
+
+      @Override
+      public ImmutableList<QueryPartReplacer> getQueryPartReplacements() {
+        return functionReplacements;
+      }
+    },
     PGBENCH {
       final ImmutableList<String> errorHints = ImmutableList.of(PGBENCH_USAGE_HINT);
       volatile long lastHintTimestampMillis = 0L;
@@ -201,7 +241,7 @@ public class ClientAutoDetector {
       }
 
       @Override
-      public ImmutableMap<String, String> getDefaultParameters() {
+      public ImmutableMap<String, String> getDefaultParameters(Map<String, String> parameters) {
         return ImmutableMap.of("spanner.emulate_pg_class_tables", "true");
       }
 
@@ -264,7 +304,7 @@ public class ClientAutoDetector {
       }
 
       @Override
-      public ImmutableMap<String, String> getDefaultParameters() {
+      public ImmutableMap<String, String> getDefaultParameters(Map<String, String> parameters) {
         return ImmutableMap.of(
             "spanner.guess_types", String.format("%d,%d", Oid.TIMESTAMPTZ, Oid.DATE));
       }
@@ -442,7 +482,7 @@ public class ClientAutoDetector {
       return ImmutableList.of();
     }
 
-    public ImmutableMap<String, String> getDefaultParameters() {
+    public ImmutableMap<String, String> getDefaultParameters(Map<String, String> parameters) {
       return ImmutableMap.of();
     }
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
index 9f852614..a572de46 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BindMessage.java
@@ -59,17 +59,18 @@ public class BindMessage extends AbstractQueryProtocolMessage {
 
   /** Constructor for Bind messages that are constructed to execute a Query message. */
   public BindMessage(ConnectionHandler connection, ManuallyCreatedToken manuallyCreatedToken) {
-    this(connection, "", new byte[0][], manuallyCreatedToken);
+    this(connection, "", "", new byte[0][], manuallyCreatedToken);
   }
 
   /** Constructor for Bind messages that are created by EXECUTE statements. */
   public BindMessage(
       ConnectionHandler connection,
       String statementName,
+      String portalName,
       byte[][] parameters,
       ManuallyCreatedToken manuallyCreatedToken) {
     super(connection, 4, manuallyCreatedToken);
-    this.portalName = "";
+    this.portalName = portalName;
     this.statementName = statementName;
     this.formatCodes = ImmutableList.of();
     this.resultFormatCodes = ImmutableList.of();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
index 67ea96f8..41d93153 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ControlMessage.java
@@ -271,10 +271,12 @@ public abstract class ControlMessage extends WireMessage {
         if (statement.getStatementResult().getResultType() == ResultType.RESULT_SET) {
           SendResultSetState state = sendResultSet(statement, mode, maxRows);
           statement.setHasMoreData(state.hasMoreRows());
-          if (state.hasMoreRows()) {
+          if (state.hasMoreRows() && mode == QueryMode.EXTENDED) {
             new PortalSuspendedResponse(this.outputStream).send(false);
           } else {
-            statement.close();
+            if (!state.hasMoreRows() && mode == QueryMode.EXTENDED) {
+              statement.close();
+            }
             new CommandCompleteResponse(this.outputStream, state.getCommandAndNumRows())
                 .send(false);
           }
@@ -501,13 +503,15 @@ public abstract class ControlMessage extends WireMessage {
         long rows = 0L;
         while (hasData) {
           WireOutput wireOutput = describedResult.createDataRowResponse(converter);
-          if (!converter.isIncludeBinaryCopyHeaderInFirstRow()) {
-            binaryCopyHeaderSentLatch.await();
+          if (wireOutput != null) {
+            if (!converter.isIncludeBinaryCopyHeaderInFirstRow()) {
+              binaryCopyHeaderSentLatch.await();
+            }
+            synchronized (describedResult) {
+              wireOutput.send(false);
+            }
+            binaryCopyHeaderSentLatch.countDown();
           }
-          synchronized (describedResult) {
-            wireOutput.send(false);
-          }
-          binaryCopyHeaderSentLatch.countDown();
           if (Thread.interrupted()) {
             throw PGExceptionFactory.newQueryCancelledException();
           }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java
index 934e30bf..6545dd2d 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ExecuteMessage.java
@@ -29,20 +29,36 @@ public class ExecuteMessage extends AbstractQueryProtocolMessage {
   private final String name;
   private final int maxRows;
   private final IntermediatePreparedStatement statement;
+  private final boolean cleanupAfterExecute;
 
   public ExecuteMessage(ConnectionHandler connection) throws Exception {
     super(connection);
     this.name = this.readAll();
     this.maxRows = this.inputStream.readInt();
     this.statement = this.connection.getPortal(this.name);
+    this.cleanupAfterExecute = true;
   }
 
   /** Constructor for execute messages that are generated by the simple query protocol. */
   public ExecuteMessage(ConnectionHandler connection, ManuallyCreatedToken manuallyCreatedToken) {
+    this(connection, "", 0, null, true, manuallyCreatedToken);
+  }
+
+  public ExecuteMessage(
+      ConnectionHandler connection,
+      String name,
+      int maxRows,
+      String commandTag,
+      boolean cleanupAfterExecute,
+      ManuallyCreatedToken manuallyCreatedToken) {
     super(connection, 8, manuallyCreatedToken);
-    this.name = "";
-    this.maxRows = 0;
+    this.name = name;
+    this.maxRows = maxRows;
     this.statement = this.connection.getPortal(this.name);
+    if (commandTag != null) {
+      this.statement.setCommandTag(commandTag);
+    }
+    this.cleanupAfterExecute = cleanupAfterExecute;
   }
 
   @Override
@@ -111,6 +127,8 @@ public class ExecuteMessage extends AbstractQueryProtocolMessage {
         }
       }
     }
-    this.connection.cleanUp(this.statement);
+    if (cleanupAfterExecute) {
+      this.connection.cleanUp(this.statement);
+    }
   }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java
index bf08103e..514f109f 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/ParseMessage.java
@@ -15,9 +15,13 @@
 package com.google.cloud.spanner.pgadapter.wireprotocol;
 
 import static com.google.cloud.spanner.pgadapter.statements.SimpleParser.isCommand;
+import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.CLOSE;
 import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.COPY;
 import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.DEALLOCATE;
+import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.DECLARE;
 import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.EXECUTE;
+import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.FETCH;
+import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.MOVE;
 import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.PREPARE;
 import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.RELEASE;
 import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.ROLLBACK;
@@ -33,11 +37,15 @@ import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStateme
 import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
 import com.google.cloud.spanner.pgadapter.statements.BackendConnection;
+import com.google.cloud.spanner.pgadapter.statements.CloseStatement;
 import com.google.cloud.spanner.pgadapter.statements.CopyStatement;
 import com.google.cloud.spanner.pgadapter.statements.DeallocateStatement;
+import com.google.cloud.spanner.pgadapter.statements.DeclareStatement;
 import com.google.cloud.spanner.pgadapter.statements.ExecuteStatement;
+import com.google.cloud.spanner.pgadapter.statements.FetchStatement;
 import com.google.cloud.spanner.pgadapter.statements.IntermediatePreparedStatement;
 import com.google.cloud.spanner.pgadapter.statements.InvalidStatement;
+import com.google.cloud.spanner.pgadapter.statements.MoveStatement;
 import com.google.cloud.spanner.pgadapter.statements.PrepareStatement;
 import com.google.cloud.spanner.pgadapter.statements.ReleaseStatement;
 import com.google.cloud.spanner.pgadapter.statements.RollbackToStatement;
@@ -134,6 +142,34 @@ public class ParseMessage extends AbstractQueryProtocolMessage {
             name,
             parsedStatement,
             originalStatement);
+      } else if (isCommand(DECLARE, originalStatement.getSql())) {
+        return new DeclareStatement(
+            connectionHandler,
+            connectionHandler.getServer().getOptions(),
+            name,
+            parsedStatement,
+            originalStatement);
+      } else if (isCommand(FETCH, originalStatement.getSql())) {
+        return new FetchStatement(
+            connectionHandler,
+            connectionHandler.getServer().getOptions(),
+            name,
+            parsedStatement,
+            originalStatement);
+      } else if (isCommand(MOVE, originalStatement.getSql())) {
+        return new MoveStatement(
+            connectionHandler,
+            connectionHandler.getServer().getOptions(),
+            name,
+            parsedStatement,
+            originalStatement);
+      } else if (isCommand(CLOSE, originalStatement.getSql())) {
+        return new CloseStatement(
+            connectionHandler,
+            connectionHandler.getServer().getOptions(),
+            name,
+            parsedStatement,
+            originalStatement);
       } else if (isCommand(VACUUM, originalStatement.getSql())) {
         return new VacuumStatement(
             connectionHandler,
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
index 47d805d8..fc6f7048 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/QueryMessage.java
@@ -33,6 +33,10 @@ public class QueryMessage extends ControlMessage {
   public static final String SAVEPOINT = "SAVEPOINT";
   public static final String RELEASE = "RELEASE";
   public static final String ROLLBACK = "ROLLBACK";
+  public static final String DECLARE = "DECLARE";
+  public static final String FETCH = "FETCH";
+  public static final String MOVE = "MOVE";
+  public static final String CLOSE = "CLOSE";
   private final Statement originalStatement;
   private final SimpleQueryStatement simpleQueryStatement;
 
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
index 327cf7a3..2384709a 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/StartupMessage.java
@@ -76,7 +76,7 @@ public class StartupMessage extends BootstrapMessage {
     connection.connectToSpanner(database, credentials);
     for (Entry<String, String> parameter :
         Iterables.concat(
-            connection.getWellKnownClient().getDefaultParameters().entrySet(),
+            connection.getWellKnownClient().getDefaultParameters(parameters).entrySet(),
             parameters.entrySet())) {
       connection
           .getExtendedQueryProtocolHandler()


Test patch:
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExecuteStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExecuteStatement.java
index f8fa71d9..e3dcdf01 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExecuteStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExecuteStatement.java
@@ -91,6 +91,7 @@ public class ExecuteStatement extends IntermediatePortalStatement {
       new BindMessage(
               connectionHandler,
               executeStatement.name,
+              "",
               executeStatement.parameters,
               ManuallyCreatedToken.MANUALLY_CREATED_TOKEN)
           .send();
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
index 8be705cc..ee438e6b 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/IntermediateStatement.java
@@ -14,6 +14,7 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
+import static com.google.cloud.spanner.pgadapter.statements.MoveStatement.MOVE_COMMAND_TAG;
 import static com.google.cloud.spanner.pgadapter.statements.SimpleParser.parseCommand;
 
 import com.google.api.core.InternalApi;
@@ -37,6 +38,7 @@ import com.google.cloud.spanner.pgadapter.utils.Converter;
 import com.google.cloud.spanner.pgadapter.wireoutput.DataRowResponse;
 import com.google.cloud.spanner.pgadapter.wireoutput.WireOutput;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import java.io.DataOutputStream;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
@@ -322,6 +324,10 @@ public class IntermediateStatement {
     return this.command;
   }
 
+  public void setCommandTag(String commandTag) {
+    this.commandTag = Preconditions.checkNotNull(commandTag);
+  }
+
   /** @return the extracted command (first word) from the really executed SQL statement. */
   public String getCommandTag() {
     return this.commandTag;
@@ -334,7 +340,10 @@ public class IntermediateStatement {
   }
 
   public WireOutput createDataRowResponse(Converter converter) {
-    return new DataRowResponse(this.outputStream, converter);
+    // MOVE should not return any of the data.
+    return MOVE_COMMAND_TAG.equals(commandTag)
+        ? null
+        : new DataRowResponse(this.outputStream, converter);
   }
 
   public WireOutput[] createResultSuffix() {
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 b44c9679..b802df38 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ConnectionHandlerTest.java
@@ -214,7 +214,9 @@ public class ConnectionHandlerTest {
     when(socket.getInetAddress()).thenReturn(address);
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
-    assertThrows(IllegalStateException.class, () -> connection.getPortal("unknown portal"));
+    PGException exception =
+        assertThrows(PGException.class, () -> connection.getPortal("unknown portal"));
+    assertEquals(SQLState.InvalidCursorName, exception.getSQLState());
   }
 
   @Test
@@ -240,7 +242,9 @@ public class ConnectionHandlerTest {
     when(socket.getInetAddress()).thenReturn(address);
 
     ConnectionHandler connection = new ConnectionHandler(server, socket);
-    assertThrows(IllegalStateException.class, () -> connection.closePortal("unknown portal"));
+    PGException exception =
+        assertThrows(PGException.class, () -> connection.closePortal("unknown portal"));
+    assertEquals(SQLState.InvalidCursorName, exception.getSQLState());
   }
 
   @Test
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/CursorMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/CursorMockServerTest.java
new file mode 100644
index 00000000..c9cb003e
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/CursorMockServerTest.java
@@ -0,0 +1,519 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
+import com.google.cloud.spanner.RandomResultSetGenerator;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.common.collect.ImmutableList;
+import com.google.spanner.v1.BeginTransactionRequest;
+import com.google.spanner.v1.ExecuteSqlRequest;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.util.Random;
+import org.junit.After;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.postgresql.PGProperty;
+import org.postgresql.util.PSQLException;
+
+@RunWith(JUnit4.class)
+public class CursorMockServerTest extends AbstractMockServerTest {
+
+  @BeforeClass
+  public static void loadPgJdbcDriver() throws Exception {
+    // Make sure the PG JDBC driver is loaded.
+    Class.forName("org.postgresql.Driver");
+  }
+
+  @BeforeClass
+  public static void startMockSpannerAndPgAdapterServers() throws Exception {
+    // Start PGAdapter without a default database.
+    doStartMockSpannerAndPgAdapterServers(null, ImmutableList.of());
+  }
+
+  @After
+  public void removeExecutionTimes() {
+    mockSpanner.removeAllExecutionTimes();
+  }
+
+  private String createUrl() {
+    return createUrl("postgres_fdw");
+  }
+
+  private String createUrl(String applicationName) {
+    return String.format(
+        "jdbc:postgresql://localhost:%d/d?preferQueryMode=simple&%s=%s&%s=090000",
+        pgServer.getLocalPort(),
+        PGProperty.APPLICATION_NAME.getName(),
+        applicationName,
+        PGProperty.ASSUME_MIN_SERVER_VERSION.getName());
+  }
+
+  @Test
+  public void testReadWriteTransaction() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+
+      try (ResultSet resultSet = connection.createStatement().executeQuery(SELECT1.getSql())) {
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+    }
+
+    assertEquals(0, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    assertTrue(mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0).hasTransaction());
+    assertTrue(
+        mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0).getTransaction().hasBegin());
+    assertTrue(
+        mockSpanner
+            .getRequestsOfType(ExecuteSqlRequest.class)
+            .get(0)
+            .getTransaction()
+            .getBegin()
+            .hasReadWrite());
+  }
+
+  @Test
+  public void testReadOnlyTransaction() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+      connection.setReadOnly(true);
+
+      try (ResultSet resultSet = connection.createStatement().executeQuery(SELECT1.getSql())) {
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+    }
+
+    assertEquals(1, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
+    assertTrue(
+        mockSpanner
+            .getRequestsOfType(BeginTransactionRequest.class)
+            .get(0)
+            .getOptions()
+            .hasReadOnly());
+  }
+
+  @Test
+  public void testTransactionReadWrite() throws SQLException {
+    try (Connection connection =
+        DriverManager.getConnection(createUrl("postgres_fdw readonly=false"))) {
+      connection.setAutoCommit(false);
+      try (ResultSet resultSet = connection.createStatement().executeQuery(SELECT1.getSql())) {
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+    }
+
+    assertEquals(0, mockSpanner.countRequestsOfType(BeginTransactionRequest.class));
+    assertEquals(1, mockSpanner.countRequestsOfType(ExecuteSqlRequest.class));
+    assertTrue(
+        mockSpanner.getRequestsOfType(ExecuteSqlRequest.class).get(0).getTransaction().hasBegin());
+    assertTrue(
+        mockSpanner
+            .getRequestsOfType(ExecuteSqlRequest.class)
+            .get(0)
+            .getTransaction()
+            .getBegin()
+            .hasReadWrite());
+  }
+
+  @Test
+  public void testSelectRandom() throws SQLException {
+    int numRows = new Random().nextInt(2000) + 1;
+    int totalRows = 0;
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from random"),
+            new RandomResultSetGenerator(numRows).generate()));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      connection.createStatement().execute("declare c1 cursor for select * from random");
+      while (true) {
+        boolean foundRows = false;
+        try (ResultSet resultSet = connection.createStatement().executeQuery("fetch 100 c1")) {
+          while (resultSet.next()) {
+            foundRows = true;
+            totalRows++;
+          }
+        }
+        if (!foundRows) {
+          break;
+        }
+      }
+      connection.createStatement().execute("close c1");
+    }
+    assertEquals(numRows, totalRows);
+  }
+
+  @Test
+  public void testMove() throws SQLException {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from random"),
+            new RandomResultSetGenerator(new Random().nextInt(10) + 2).generate()));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      connection.createStatement().execute("declare c1 cursor for select * from random");
+      // execute returns false if no rows were returned.
+      assertFalse(connection.createStatement().execute("move 3 c1"));
+      connection.createStatement().execute("close c1");
+    }
+  }
+
+  @Test
+  public void testSelectWithParameters() throws SQLException {
+    int numRows = new Random().nextInt(150) + 25;
+    int totalRows = 0;
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from random where some_col='some-value'"),
+            new RandomResultSetGenerator(numRows).generate()));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      try (PreparedStatement preparedStatement =
+          connection.prepareStatement(
+              "declare c1 cursor for select * from random where some_col=?")) {
+        preparedStatement.setString(1, "some-value");
+        preparedStatement.execute();
+      }
+      while (true) {
+        boolean foundRows = false;
+        try (ResultSet resultSet = connection.createStatement().executeQuery("fetch 100 from c1")) {
+          while (resultSet.next()) {
+            foundRows = true;
+            totalRows++;
+          }
+        }
+        if (!foundRows) {
+          break;
+        }
+      }
+      connection.createStatement().execute("close c1");
+    }
+    assertEquals(numRows, totalRows);
+  }
+
+  @Test
+  public void testRandomFetchAndMove() throws SQLException {
+    Random random = new Random();
+    int numRows = random.nextInt(750) + 50;
+    int totalRows = 0;
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from random"),
+            new RandomResultSetGenerator(numRows).generate()));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      connection.createStatement().execute("declare c1 cursor for select * from random");
+      while (true) {
+        boolean fetch = random.nextBoolean();
+        int fetchSize = random.nextInt(200) + 1;
+        String sql = (fetch ? "fetch " : "move ") + fetchSize + " c1";
+        boolean foundRows = false;
+        if (fetch) {
+          try (ResultSet resultSet = connection.createStatement().executeQuery(sql)) {
+            while (resultSet.next()) {
+              foundRows = true;
+              totalRows++;
+            }
+          }
+        } else {
+          int count = connection.createStatement().executeUpdate(sql);
+          foundRows = count > 0;
+          totalRows += count;
+        }
+        if (!foundRows) {
+          break;
+        }
+      }
+      connection.createStatement().execute("close c1");
+    }
+    assertEquals(numRows, totalRows);
+  }
+
+  @Test
+  public void testUnsupportedFetchAndMove() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      for (String operation : new String[] {"move", "fetch"}) {
+        for (String dir :
+            new String[] {
+              "backward",
+              "backward 1",
+              "backward all",
+              "prior",
+              "first",
+              "last",
+              "absolute 1",
+              "relative 1",
+              "0",
+              "forward 0",
+              "-1",
+              "forward -1"
+            }) {
+          connection.createStatement().execute("begin transaction");
+          connection.createStatement().execute("declare c1 cursor for " + SELECT1.getSql());
+          PSQLException exception =
+              assertThrows(
+                  PSQLException.class,
+                  () -> connection.createStatement().execute(operation + " " + dir + " from c1"));
+          assertEquals(SQLState.FeatureNotSupported.toString(), exception.getSQLState());
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testFetchAll() throws SQLException {
+    for (String command : new String[] {"all", "forward all"}) {
+      int numRows = new Random().nextInt(10) + 5;
+      mockSpanner.putStatementResult(
+          StatementResult.query(
+              Statement.of("select * from random"),
+              new RandomResultSetGenerator(numRows).generate()));
+
+      int foundRows = 0;
+      try (Connection connection = DriverManager.getConnection(createUrl())) {
+        connection.createStatement().execute("begin transaction");
+        connection.createStatement().execute("declare c1 cursor for select * from random");
+        try (ResultSet resultSet =
+            connection.createStatement().executeQuery(String.format("fetch %s from c1", command))) {
+          while (resultSet.next()) {
+            foundRows++;
+          }
+        }
+        connection.createStatement().execute("close c1");
+      }
+      assertEquals(numRows, foundRows);
+    }
+  }
+
+  @Test
+  public void testNext() throws SQLException {
+    int numRows = new Random().nextInt(10) + 5;
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from random"),
+            new RandomResultSetGenerator(numRows).generate()));
+
+    int foundRows = 0;
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      connection.createStatement().execute("declare c1 cursor for select * from random");
+      while (true) {
+        try (ResultSet resultSet =
+            connection.createStatement().executeQuery("fetch next from c1")) {
+          if (foundRows == numRows) {
+            assertFalse(resultSet.next());
+            break;
+          } else {
+            assertTrue(resultSet.next());
+            assertFalse(resultSet.next());
+            foundRows++;
+          }
+        }
+      }
+      connection.createStatement().execute("close c1");
+    }
+    assertEquals(numRows, foundRows);
+  }
+
+  @Test
+  public void testForward() throws SQLException {
+    Random random = new Random();
+    for (String command : new String[] {"", "forward"}) {
+      int numRows = random.nextInt(20) + 5;
+      mockSpanner.putStatementResult(
+          StatementResult.query(
+              Statement.of("select * from random"),
+              new RandomResultSetGenerator(numRows).generate()));
+
+      int foundRows = 0;
+      try (Connection connection = DriverManager.getConnection(createUrl())) {
+        connection.createStatement().execute("begin transaction");
+        connection.createStatement().execute("declare c1 cursor for select * from random");
+        while (true) {
+          int count = random.nextInt(numRows) + 1;
+          try (ResultSet resultSet =
+              connection
+                  .createStatement()
+                  .executeQuery(String.format("fetch %s %d from c1", command, count))) {
+            if (foundRows == numRows) {
+              assertFalse(resultSet.next());
+              break;
+            } else {
+              while (resultSet.next()) {
+                foundRows++;
+              }
+            }
+          }
+        }
+        connection.createStatement().execute("close c1");
+      }
+      assertEquals(numRows, foundRows);
+    }
+  }
+
+  @Test
+  public void testDeclareCursorForNonQuery() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class,
+              () ->
+                  connection
+                      .createStatement()
+                      .execute("declare c1 cursor for update foo set bar=1 where true"));
+      assertEquals(SQLState.FeatureNotSupported.toString(), exception.getSQLState());
+    }
+  }
+
+  @Test
+  public void testScrollableCursor() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class,
+              () -> connection.createStatement().execute("declare c1 scroll cursor for select 1"));
+      assertEquals(SQLState.FeatureNotSupported.toString(), exception.getSQLState());
+    }
+  }
+
+  @Test
+  public void testHoldableCursor() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.createStatement().execute("begin transaction");
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class,
+              () ->
+                  connection.createStatement().execute("declare c1 cursor with hold for select 1"));
+      assertEquals(SQLState.FeatureNotSupported.toString(), exception.getSQLState());
+    }
+  }
+
+  @Test
+  public void testOnlyInTransaction() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class,
+              () ->
+                  connection
+                      .createStatement()
+                      .execute("declare c1 cursor for select * from random"));
+      assertEquals(SQLState.NoActiveSqlTransaction.toString(), exception.getSQLState());
+    }
+  }
+
+  @Test
+  public void testNotInAbortedTransaction() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+      assertThrows(PSQLException.class, () -> connection.createStatement().execute("select foo"));
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class,
+              () ->
+                  connection
+                      .createStatement()
+                      .execute("declare c1 cursor for select * from random"));
+      assertEquals(SQLState.InFailedSqlTransaction.toString(), exception.getSQLState());
+    }
+  }
+
+  @Test
+  public void testCommitAndRollbackClosesCursor() throws SQLException {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from random"), new RandomResultSetGenerator(10).generate()));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+
+      for (boolean commit : new boolean[] {true, false}) {
+        connection.createStatement().execute("declare c1 cursor for select * from random");
+        try (ResultSet resultSet = connection.createStatement().executeQuery("fetch 1 c1")) {
+          assertTrue(resultSet.next());
+          assertFalse(resultSet.next());
+        }
+        if (commit) {
+          connection.commit();
+        } else {
+          connection.rollback();
+        }
+        assertThrows(
+            PSQLException.class, () -> connection.createStatement().executeQuery("fetch 1 c1"));
+      }
+    }
+  }
+
+  @Test
+  public void testClose() throws SQLException {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of("select * from random"), new RandomResultSetGenerator(10).generate()));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      connection.setAutoCommit(false);
+
+      connection.createStatement().execute("declare c1 cursor for select * from random");
+      connection.createStatement().execute("declare c2 cursor for select * from random");
+      connection.createStatement().execute("declare c3 cursor for select * from random");
+
+      try (ResultSet resultSet = connection.createStatement().executeQuery("fetch 1 c1")) {
+        assertTrue(resultSet.next());
+        assertFalse(resultSet.next());
+      }
+      connection.createStatement().execute("close c1");
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class, () -> connection.createStatement().executeQuery("fetch 1 c1"));
+      assertEquals(SQLState.InvalidCursorName.toString(), exception.getSQLState());
+
+      connection.createStatement().execute("close all");
+      exception =
+          assertThrows(
+              PSQLException.class, () -> connection.createStatement().executeQuery("fetch 1 c2"));
+      assertEquals(SQLState.InvalidCursorName.toString(), exception.getSQLState());
+      exception =
+          assertThrows(
+              PSQLException.class, () -> connection.createStatement().executeQuery("fetch 1 c3"));
+      assertEquals(SQLState.InvalidCursorName.toString(), exception.getSQLState());
+    }
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
index c6502014..ff418dde 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
@@ -18,6 +18,8 @@ import static com.google.cloud.spanner.pgadapter.PgAdapterTestEnv.DEFAULT_DATA_M
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeFalse;
 import static org.junit.Assume.assumeTrue;
@@ -200,7 +202,7 @@ public class ITPsqlTest implements IntegrationTest {
           "-d",
           POSTGRES_DATABASE,
           "-c",
-          "drop table if exists all_types; drop table if exists numbers;"
+          "drop table if exists all_types; drop table if exists numbers; drop foreign table if exists f_all_types; drop foreign table if exists f_numbers;"
         };
     ProcessBuilder builder = new ProcessBuilder().command(dropTablesCommand);
     setPgPassword(builder);
@@ -441,6 +443,7 @@ public class ITPsqlTest implements IntegrationTest {
   public void testCopyBetweenPostgreSQLAndCloudSpanner() throws Exception {
     int numRows = 100;
 
+    truncatePostgreSQLAllTypesTable();
     logger.info("Copying initial data to PG");
     // Generate 99 random rows.
     copyRandomRowsToPostgreSQL(numRows - 1);
@@ -915,6 +918,13 @@ public class ITPsqlTest implements IntegrationTest {
     }
   }
 
+  private void truncatePostgreSQLAllTypesTable() throws SQLException {
+    try (Connection connection =
+        DriverManager.getConnection(createJdbcUrlForLocalPg(), POSTGRES_USER, POSTGRES_PASSWORD)) {
+      connection.createStatement().executeUpdate("truncate table all_types");
+    }
+  }
+
   private void copyRandomRowsToPostgreSQL(int numRows) throws Exception {
     // Generate random rows and copy these into the all_types table in a local PostgreSQL database.
     ProcessBuilder builder = new ProcessBuilder();
@@ -975,4 +985,212 @@ public class ITPsqlTest implements IntegrationTest {
     int res = process.waitFor();
     assertEquals(errors.toString(), 0, res);
   }
+
+  @Test
+  public void testForeignDataWrapper() throws Exception {
+    ProcessBuilder builder = new ProcessBuilder();
+    String[] psqlCommand =
+        new String[] {
+          "psql",
+          "-h",
+          POSTGRES_HOST,
+          "-p",
+          POSTGRES_PORT,
+          "-U",
+          POSTGRES_USER,
+          "-d",
+          POSTGRES_DATABASE
+        };
+    builder.command(psqlCommand);
+    setPgPassword(builder);
+    Process process = builder.start();
+    String errors;
+    String output;
+
+    String createForeignTablesCommand =
+        DEFAULT_DATA_MODEL.stream()
+            .map(s -> s.replace("create table ", "create foreign table f_"))
+            .map(s -> s.replace("primary key", ""))
+            .map(s -> s.replace("col_int int", "col_int bigint"))
+            .map(s -> s.replace("col_array_int int[]", "col_array_int bigint[]"))
+            .map(
+                s ->
+                    s
+                        + String.format(
+                            " server pgadapter options (table_name '%s');",
+                            s.substring(
+                                "create foreign table f_".length(),
+                                s.indexOf(' ', "create foreign table f_".length()))))
+            .collect(Collectors.joining("\n"));
+
+    try (OutputStreamWriter writer = new OutputStreamWriter(process.getOutputStream());
+        BufferedReader reader =
+            new BufferedReader(new InputStreamReader(process.getInputStream()));
+        BufferedReader errorReader =
+            new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+      writer.write("DROP EXTENSION IF EXISTS postgres_fdw CASCADE;\n");
+      writer.write("CREATE EXTENSION IF NOT EXISTS postgres_fdw;\n");
+      writer.write(
+          String.format(
+              "CREATE SERVER IF NOT EXISTS pgadapter FOREIGN DATA WRAPPER postgres_fdw OPTIONS (host '%s', port '%d', dbname '%s', options '-c spanner.read_only=true');\n",
+              (POSTGRES_HOST.startsWith("/") ? "/tmp" : testEnv.getPGAdapterHost()),
+              testEnv.getPGAdapterPort(),
+              database.getId().getDatabase()));
+      writer.write(
+          String.format(
+              "CREATE USER MAPPING IF NOT EXISTS FOR %s SERVER pgadapter OPTIONS (user '%s', password_required 'false');\n",
+              POSTGRES_USER, POSTGRES_USER));
+      writer.write(
+          String.format("GRANT USAGE ON FOREIGN SERVER pgadapter TO %s;\n", POSTGRES_USER));
+      writer.write(createForeignTablesCommand);
+      writer.write("truncate table all_types;\n");
+      writer.write("\\q\n");
+      writer.flush();
+      errors = errorReader.lines().collect(Collectors.joining("\n"));
+      output = reader.lines().collect(Collectors.joining("\n"));
+    }
+
+    assertTrue(errors, "".equals(errors) || !errors.contains("ERROR:"));
+    assertEquals(
+        "DROP EXTENSION\n"
+            + "CREATE EXTENSION\n"
+            + "CREATE SERVER\n"
+            + "CREATE USER MAPPING\n"
+            + "GRANT\n"
+            + "CREATE FOREIGN TABLE\n"
+            + "CREATE FOREIGN TABLE\n"
+            + "TRUNCATE TABLE",
+        output);
+    int res = process.waitFor();
+    assertEquals(0, res);
+
+    truncatePostgreSQLAllTypesTable();
+    // Generate 250 random rows.
+    int numRows = 250;
+    copyRandomRowsToPostgreSQL(numRows - 1);
+    // Also add one row with all nulls to ensure that nulls are also copied correctly.
+    long nullRowId = random.nextLong();
+    try (Connection connection =
+        DriverManager.getConnection(createJdbcUrlForLocalPg(), POSTGRES_USER, POSTGRES_PASSWORD)) {
+      try (PreparedStatement preparedStatement =
+          connection.prepareStatement("insert into all_types (col_bigint) values (?)")) {
+        preparedStatement.setLong(1, nullRowId);
+        assertEquals(1, preparedStatement.executeUpdate());
+      }
+    }
+
+    // Make sure the all_types table on Cloud Spanner is empty.
+    String databaseId = database.getId().getDatabase();
+    testEnv.write(databaseId, Collections.singleton(Mutation.delete("all_types", KeySet.all())));
+    // COPY the rows to Cloud Spanner.
+    builder = new ProcessBuilder();
+    builder.command(
+        "bash",
+        "-c",
+        "psql"
+            + " -h "
+            + POSTGRES_HOST
+            + " -p "
+            + POSTGRES_PORT
+            + " -U "
+            + POSTGRES_USER
+            + " -d "
+            + POSTGRES_DATABASE
+            + " -c \"copy all_types to stdout (format binary) \" "
+            + "  | psql "
+            + " -h "
+            + (POSTGRES_HOST.startsWith("/") ? "/tmp" : testEnv.getPGAdapterHost())
+            + " -p "
+            + testEnv.getPGAdapterPort()
+            + " -d "
+            + database.getId().getDatabase()
+            + " -c \"copy all_types from stdin (format binary)"
+            + ";\"\n");
+    setPgPassword(builder);
+    process = builder.start();
+    res = process.waitFor();
+    assertEquals(0, res);
+
+    // Make sure we can access the data in Cloud Spanner through the foreign data wrapper.
+    int foundRows = 0;
+    int foundNullRows = 0;
+    try (Connection connection =
+        DriverManager.getConnection(createJdbcUrlForLocalPg(), POSTGRES_USER, POSTGRES_PASSWORD)) {
+      connection
+          .createStatement()
+          .execute("set session characteristics as transaction isolation level serializable");
+      try (ResultSet resultSet =
+          connection.createStatement().executeQuery("select * from f_all_types")) {
+        while (resultSet.next()) {
+          foundRows++;
+          if (resultSet.getLong(1) == nullRowId) {
+            foundNullRows++;
+          }
+          for (int col = 2; col <= resultSet.getMetaData().getColumnCount(); col++) {
+            if (resultSet.getLong(1) == nullRowId) {
+              assertNull(resultSet.getObject(col));
+            } else {
+              assertNotNull(resultSet.getObject(col));
+            }
+          }
+        }
+      }
+      assertEquals(numRows, foundRows);
+      assertEquals(1, foundNullRows);
+
+      // Also verify that we can use parameterized queries with the foreign table.
+      // These parameters are however converted to literals by the PostgreSQL Foreign Data Wrapper.
+      // That means that the queries that are executed on Cloud Spanner are always unparameterized.
+      try (PreparedStatement preparedStatement =
+          connection.prepareStatement("select * from f_all_types where col_bigint=?")) {
+        preparedStatement.setLong(1, nullRowId);
+        try (ResultSet resultSet = preparedStatement.executeQuery()) {
+          assertTrue(resultSet.next());
+          for (int col = 2; col <= resultSet.getMetaData().getColumnCount(); col++) {
+            assertNull(resultSet.getObject(col));
+          }
+          assertFalse(resultSet.next());
+        }
+      }
+    }
+
+    // Create a user-defined function and use it on the data in Cloud Spanner.
+    builder = new ProcessBuilder();
+    builder.command(psqlCommand);
+    setPgPassword(builder);
+    process = builder.start();
+    try (OutputStreamWriter writer = new OutputStreamWriter(process.getOutputStream());
+        BufferedReader reader =
+            new BufferedReader(new InputStreamReader(process.getInputStream()));
+        BufferedReader errorReader =
+            new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+      writer.write("set session characteristics as transaction isolation level serializable;");
+      writer.write("DROP FUNCTION IF EXISTS date_timestamp_year_diff;\n");
+      writer.write(
+          "CREATE FUNCTION date_timestamp_year_diff(ts timestamptz, dt date) RETURNS integer AS $$\n"
+              + "    SELECT extract(year from ts) - extract(year from dt);\n"
+              + "$$ LANGUAGE SQL;\n");
+      writer.write(
+          "select date_timestamp_year_diff(col_timestamptz, col_date) as year_diff\n"
+              + "from f_all_types\n"
+              + "order by year_diff asc nulls first;\n");
+      writer.write("\\q\n");
+      writer.flush();
+      errors = errorReader.lines().collect(Collectors.joining("\n"));
+      output = reader.lines().collect(Collectors.joining("\n"));
+    }
+    assertTrue(errors, "".equals(errors) || !errors.contains("ERROR:"));
+    assertTrue(
+        output,
+        output.startsWith(
+            "SET\n"
+                + "DROP FUNCTION\n"
+                + "CREATE FUNCTION\n"
+                + " year_diff \n"
+                + "-----------\n"
+                + "          \n"));
+    assertTrue(output, output.endsWith("(250 rows)\n"));
+    res = process.waitFor();
+    assertEquals(0, res);
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/PsqlMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/PsqlMockServerTest.java
index 0c4563e1..facefcfc 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/PsqlMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/PsqlMockServerTest.java
@@ -247,6 +247,57 @@ public class PsqlMockServerTest extends AbstractMockServerTest {
     assertArrayEquals(byteArrayOutputStream.toByteArray(), copyOutput);
   }
 
+  @Test
+  public void testDeclareCursor() throws Exception {
+    assumeTrue("This test requires psql to be installed", isPsqlAvailable());
+    mockSpanner.putStatementResult(
+        StatementResult.query(Statement.of("select * from all_types"), ALL_TYPES_RESULTSET));
+
+    ProcessBuilder builder = new ProcessBuilder();
+    String[] psqlCommand =
+        new String[] {
+          "psql",
+          "-h",
+          "localhost",
+          "-p",
+          String.valueOf(pgServer.getLocalPort()),
+          "-c",
+          "begin",
+          "-c",
+          "declare foo cursor for select * from all_types",
+          "-c",
+          "fetch foo",
+          "-c",
+          "close foo",
+        };
+    builder.command(psqlCommand);
+    Process process = builder.start();
+    String errors;
+    String output;
+
+    try (BufferedReader reader =
+            new BufferedReader(new InputStreamReader(process.getInputStream()));
+        BufferedReader errorReader =
+            new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+      errors = errorReader.lines().collect(Collectors.joining("\n"));
+      output = reader.lines().collect(Collectors.joining("\n"));
+    }
+
+    assertEquals("", errors);
+    assertEquals(
+        "BEGIN\n"
+            + "DECLARE CURSOR\n"
+            + " col_bigint | col_bool | col_bytea  | col_float8 | col_int | col_numeric |        col_timestamptz        |  col_date  | col_varchar |    col_jsonb     | col_array_bigint | col_array_bool |              col_array_bytea               |  col_array_float8  |  col_array_int   | col_array_numeric  |                      col_array_timestamptz                      |          col_array_date          |     col_array_varchar      |                    col_array_jsonb                     \n"
+            + "------------+----------+------------+------------+---------+-------------+-------------------------------+------------+-------------+------------------+------------------+----------------+--------------------------------------------+--------------------+------------------+--------------------+-----------------------------------------------------------------+----------------------------------+----------------------------+--------------------------------------------------------\n"
+            + "          1 | t        | \\x74657374 |       3.14 |     100 |       6.626 | 2022-02-16 14:18:02.123456+01 | 2022-03-29 | test        | {\"key\": \"value\"} | {1,NULL,2}       | {t,NULL,f}     | {\"\\\\x627974657331\",NULL,\"\\\\x627974657332\"} | {3.14,NULL,-99.99} | {-100,NULL,-200} | {6.626,NULL,-3.14} | {\"2022-02-16 17:18:02.123456+01\",NULL,\"2000-01-01 01:00:00+01\"} | {\"2023-02-20\",NULL,\"2000-01-01\"} | {\"string1\",NULL,\"string2\"} | {\"{\\\"key\\\": \\\"value1\\\"}\",NULL,\"{\\\"key\\\": \\\"value2\\\"}\"}\n"
+            + "(1 row)\n"
+            + "\n"
+            + "CLOSE CURSOR",
+        output);
+    int res = process.waitFor();
+    assertEquals(0, res);
+  }
+
   @Test
   public void testArrayParameter() throws Exception {
     assumeTrue("This test requires psql to be installed", isPsqlAvailable());
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
index 23757835..c864580c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
@@ -77,6 +77,7 @@ public class BackendConnectionTest {
       AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
   private static final NotExecuted NOT_EXECUTED = new NotExecuted();
   private static final NoResult NO_RESULT = new NoResult();
+  private static final Runnable DO_NOTHING = () -> {};
 
   @Test
   public void testExtractDdlUpdateCounts() {
@@ -139,6 +140,7 @@ public class BackendConnectionTest {
     when(spannerConnection.runBatch()).thenThrow(expectedException);
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -174,6 +176,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -212,6 +215,7 @@ public class BackendConnectionTest {
 
     BackendConnection onlyDmlStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -224,6 +228,7 @@ public class BackendConnectionTest {
 
     BackendConnection onlyCopyStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -236,6 +241,7 @@ public class BackendConnectionTest {
 
     BackendConnection dmlAndCopyStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -249,6 +255,7 @@ public class BackendConnectionTest {
 
     BackendConnection onlySelectStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -261,6 +268,7 @@ public class BackendConnectionTest {
 
     BackendConnection onlyClientSideStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -275,6 +283,7 @@ public class BackendConnectionTest {
 
     BackendConnection onlyUnknownStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -287,6 +296,7 @@ public class BackendConnectionTest {
 
     BackendConnection dmlAndSelectStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -299,6 +309,7 @@ public class BackendConnectionTest {
 
     BackendConnection copyAndSelectStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -312,6 +323,7 @@ public class BackendConnectionTest {
 
     BackendConnection copyAndUnknownStatements =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             spannerConnection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -341,6 +353,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -377,6 +390,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -412,6 +426,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -437,6 +452,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -475,6 +491,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -503,6 +520,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -529,6 +547,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -556,6 +575,7 @@ public class BackendConnectionTest {
 
       BackendConnection backendConnection =
           new BackendConnection(
+              DO_NOTHING,
               DatabaseId.of("p", "i", "d"),
               connection,
               () -> WellKnownClient.UNSPECIFIED,
@@ -579,6 +599,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -608,6 +629,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -635,6 +657,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -661,6 +684,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -697,6 +721,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -732,6 +757,7 @@ public class BackendConnectionTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/CloseStatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/CloseStatementTest.java
new file mode 100644
index 00000000..d9084d79
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/CloseStatementTest.java
@@ -0,0 +1,134 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import static com.google.cloud.spanner.pgadapter.statements.CloseStatement.parse;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.error.PGExceptionFactory;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import java.util.concurrent.Future;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class CloseStatementTest {
+
+  @Test
+  public void testParse() {
+    assertEquals("foo", parse("close foo").name);
+    assertEquals("foo", parse("close FOO").name);
+    assertEquals("foo", parse("close\tfoo").name);
+    assertEquals("foo", parse("close\nfoo").name);
+    assertEquals("foo", parse("close/*comment*/foo").name);
+    assertEquals("foo", parse("close \"foo\"").name);
+    assertEquals("Foo", parse("close \"Foo\"").name);
+    assertEquals("prepare", parse("close \"prepare\"").name);
+    assertNull(parse("close all").name);
+    assertEquals("all", parse("close \"all\"").name);
+
+    assertThrows(PGException.class, () -> parse("prepare foo"));
+    assertThrows(PGException.class, () -> parse("close"));
+    assertThrows(PGException.class, () -> parse("close foo bar"));
+    assertThrows(PGException.class, () -> parse("close foo.bar"));
+  }
+
+  @Test
+  public void testCloseNamed() throws Exception {
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    OptionsMetadata optionsMetadata = mock(OptionsMetadata.class);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(mock(ConnectionMetadata.class));
+    Statement statement = Statement.of("close foo");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+    CloseStatement closeStatement =
+        new CloseStatement(connectionHandler, optionsMetadata, "", parsedStatement, statement);
+    assertFalse(closeStatement.executed);
+    closeStatement.executeAsync(mock(BackendConnection.class));
+    assertTrue(closeStatement.executed);
+    verify(connectionHandler).closePortal("foo");
+  }
+
+  @Test
+  public void testCloseAll() {
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    OptionsMetadata optionsMetadata = mock(OptionsMetadata.class);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(mock(ConnectionMetadata.class));
+    Statement statement = Statement.of("close all");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+    CloseStatement closeStatement =
+        new CloseStatement(connectionHandler, optionsMetadata, "", parsedStatement, statement);
+    assertFalse(closeStatement.executed);
+    closeStatement.executeAsync(mock(BackendConnection.class));
+    assertTrue(closeStatement.executed);
+    verify(connectionHandler).closeAllPortals();
+  }
+
+  @Test
+  public void testCloseFails() throws Exception {
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    OptionsMetadata optionsMetadata = mock(OptionsMetadata.class);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(mock(ConnectionMetadata.class));
+    Statement statement = Statement.of("close foo");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+    CloseStatement closeStatement =
+        new CloseStatement(connectionHandler, optionsMetadata, "", parsedStatement, statement);
+
+    doThrow(
+            PGExceptionFactory.newPGException(
+                "unrecognized portal name: foo", SQLState.InvalidCursorName))
+        .when(connectionHandler)
+        .closePortal("foo");
+    assertFalse(closeStatement.executed);
+    closeStatement.executeAsync(mock(BackendConnection.class));
+    assertTrue(closeStatement.hasException());
+    assertEquals(SQLState.InvalidCursorName, closeStatement.getException().getSQLState());
+    assertTrue(closeStatement.executed);
+  }
+
+  @Test
+  public void testDescribeIsNoOp() throws Exception {
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    OptionsMetadata optionsMetadata = mock(OptionsMetadata.class);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(mock(ConnectionMetadata.class));
+    Statement statement = Statement.of("close foo");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+    CloseStatement closeStatement =
+        new CloseStatement(connectionHandler, optionsMetadata, "", parsedStatement, statement);
+    Future<StatementResult> result = closeStatement.describeAsync(mock(BackendConnection.class));
+    assertNull(result.get());
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/DeclareStatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/DeclareStatementTest.java
new file mode 100644
index 00000000..46029cf8
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/DeclareStatementTest.java
@@ -0,0 +1,122 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import static com.google.cloud.spanner.pgadapter.statements.DeclareStatement.parse;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.DeclareStatement.Holdability;
+import com.google.cloud.spanner.pgadapter.statements.DeclareStatement.ParsedDeclareStatement;
+import com.google.cloud.spanner.pgadapter.statements.DeclareStatement.Scroll;
+import com.google.cloud.spanner.pgadapter.statements.DeclareStatement.Sensitivity;
+import java.util.concurrent.Future;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class DeclareStatementTest {
+  @Test
+  public void testParse() {
+    ParsedDeclareStatement statement;
+
+    statement = parse("declare foo cursor for select 1");
+    assertEquals("foo", statement.name);
+    assertFalse(statement.binary);
+    assertNull(statement.scroll);
+    assertNull(statement.sensitivity);
+    assertNull(statement.holdability);
+    assertEquals("select 1", statement.originalPreparedStatement.getSql());
+
+    statement = parse("declare \"my-cursor\" binary cursor for select * from foo where bar=1");
+    assertEquals("my-cursor", statement.name);
+    assertTrue(statement.binary);
+    assertEquals("select * from foo where bar=1", statement.originalPreparedStatement.getSql());
+
+    assertFalse(parse("declare foo cursor for select 1").binary);
+    assertTrue(parse("declare foo binary cursor for select 1").binary);
+
+    assertNull(parse("declare foo cursor for select 1").sensitivity);
+    assertEquals(
+        Sensitivity.ASENSITIVE, parse("declare foo asensitive cursor for select 1").sensitivity);
+    assertEquals(
+        Sensitivity.INSENSITIVE, parse("declare foo insensitive cursor for select 1").sensitivity);
+
+    assertNull(parse("declare foo cursor for select 1").scroll);
+    assertEquals(Scroll.NO_SCROLL, parse("declare foo no scroll cursor for select 1").scroll);
+    assertEquals(Scroll.SCROLL, parse("declare foo scroll cursor for select 1").scroll);
+
+    assertNull(parse("declare foo cursor for select 1").holdability);
+    assertEquals(Holdability.HOLD, parse("declare foo cursor with hold for select 1").holdability);
+    assertEquals(
+        Holdability.NO_HOLD, parse("declare foo cursor without hold for select 1").holdability);
+
+    statement = parse("declare foo scroll binary cursor for select 1");
+    assertEquals(Scroll.SCROLL, statement.scroll);
+    assertTrue(statement.binary);
+    statement = parse("declare foo no scroll insensitive cursor for select 1");
+    assertEquals(Scroll.NO_SCROLL, statement.scroll);
+    assertEquals(Sensitivity.INSENSITIVE, statement.sensitivity);
+    statement = parse("declare foo no scroll no scroll insensitive cursor for select 1");
+    assertEquals(Scroll.NO_SCROLL, statement.scroll);
+    assertEquals(Sensitivity.INSENSITIVE, statement.sensitivity);
+
+    assertThrows(PGException.class, () -> parse("foo"));
+    assertThrows(PGException.class, () -> parse("declare foo cursor for "));
+    assertThrows(PGException.class, () -> parse("declare foo cursor select 1"));
+    assertThrows(PGException.class, () -> parse("declare foo for select 1"));
+    assertThrows(PGException.class, () -> parse("declare foo for select 1"));
+    assertThrows(PGException.class, () -> parse("declare cursor for select 1"));
+    assertThrows(
+        PGException.class, () -> parse("declare foo insensitive asensitive cursor for select 1"));
+    assertThrows(
+        PGException.class, () -> parse("declare foo asensitive insensitive cursor for select 1"));
+    assertThrows(
+        PGException.class, () -> parse("declare foo scroll no scroll cursor for select 1"));
+    assertThrows(
+        PGException.class, () -> parse("declare foo no scroll scroll cursor for select 1"));
+    assertThrows(PGException.class, () -> parse("declare foo.bar cursor for select 1"));
+    assertThrows(PGException.class, () -> parse("declare $1 cursor for select 1"));
+  }
+
+  @Test
+  public void testDescribeIsNoOp() throws Exception {
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    OptionsMetadata optionsMetadata = mock(OptionsMetadata.class);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(mock(ConnectionMetadata.class));
+    Statement statement = Statement.of("declare foo cursor for select 1");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+    DeclareStatement declareStatement =
+        new DeclareStatement(connectionHandler, optionsMetadata, "", parsedStatement, statement);
+    Future<StatementResult> result = declareStatement.describeAsync(mock(BackendConnection.class));
+    assertNull(result.get());
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/FetchAndMoveStatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/FetchAndMoveStatementTest.java
new file mode 100644
index 00000000..177133a9
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/FetchAndMoveStatementTest.java
@@ -0,0 +1,148 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.StatementType;
+import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.cloud.spanner.pgadapter.error.PGException;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.cloud.spanner.pgadapter.metadata.ConnectionMetadata;
+import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.AbstractFetchOrMoveStatement.Direction;
+import com.google.cloud.spanner.pgadapter.statements.AbstractFetchOrMoveStatement.ParsedFetchOrMoveStatement;
+import com.google.cloud.spanner.pgadapter.statements.FetchStatement.ParsedFetchStatement;
+import com.google.cloud.spanner.pgadapter.statements.MoveStatement.ParsedMoveStatement;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class FetchAndMoveStatementTest {
+  @Test
+  public void testParse() {
+    for (String type : new String[] {"fetch", "move"}) {
+      ParsedFetchOrMoveStatement statement = parse(type, "%s foo");
+      assertEquals("foo", statement.name);
+      assertNull(statement.direction);
+      assertNull(statement.count);
+
+      assertEquals("foo", parse(type, "%s foo").name);
+      assertEquals("foo", parse(type, "%s \"foo\"").name);
+      assertEquals("my-cursor", parse(type, "%s \"my-cursor\"").name);
+      assertEquals("foo", parse(type, "%s from foo").name);
+      assertEquals("foo", parse(type, "%s in foo").name);
+      assertThrows(PGException.class, () -> parse(type, "%s from in foo"));
+
+      assertEquals(Direction.ABSOLUTE, parse(type, "%s absolute 10 foo").direction);
+      assertEquals(10, parse(type, "%s absolute 10 foo").count.intValue());
+      assertThrows(PGException.class, () -> parse(type, "%s absolute 99.4 foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s absolute foo"));
+      assertEquals(Direction.RELATIVE, parse(type, "%s relative 10 foo").direction);
+      assertThrows(PGException.class, () -> parse(type, "%s relative 99.4 foo"));
+      assertEquals(10, parse(type, "%s relative 10 foo").count.intValue());
+      assertThrows(PGException.class, () -> parse(type, "%s relative foo"));
+
+      assertEquals(1, parse(type, "%s 1 foo").count.intValue());
+      assertEquals(1, parse(type, "%s +1 foo").count.intValue());
+      assertEquals(1, parse(type, "%s 1 \"my-cursor\"").count.intValue());
+      assertEquals(99, parse(type, "%s   --comment\n99/*comment*/foo").count.intValue());
+      assertEquals(
+          Integer.MAX_VALUE, parse(type, "%s " + Integer.MAX_VALUE + " foo").count.intValue());
+      assertEquals(-1, parse(type, "%s -1 foo").count.intValue());
+      assertEquals(-1, parse(type, "%s relative -1 foo").count.intValue());
+      assertEquals(-1, parse(type, "%s absolute -1 foo").count.intValue());
+      assertThrows(PGException.class, () -> parse(type, "%s 99.9 foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s +inf foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s -inf foo"));
+
+      assertEquals(Direction.ALL, parse(type, "%s all foo").direction);
+      assertEquals(Direction.BACKWARD, parse(type, "%s backward foo").direction);
+      assertEquals(Direction.FORWARD, parse(type, "%s forward foo").direction);
+      assertEquals(Direction.FIRST, parse(type, "%s first foo").direction);
+      assertEquals(Direction.NEXT, parse(type, "%s next foo").direction);
+      assertEquals(Direction.BACKWARD_ALL, parse(type, "%s backward all foo").direction);
+      assertEquals(Direction.FORWARD_ALL, parse(type, "%s forward all foo").direction);
+      assertEquals(Direction.LAST, parse(type, "%s last foo").direction);
+      assertEquals(Direction.PRIOR, parse(type, "%s prior foo").direction);
+
+      assertThrows(PGException.class, () -> parse(type, "%s all 1 foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s first 1 foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s next 1 foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s last 1 foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s prior 1 foo"));
+      assertThrows(PGException.class, () -> parse(type, "%s $1"));
+      assertThrows(PGException.class, () -> parse(type, "%s foo.bar"));
+      assertThrows(
+          PGException.class, () -> parse(type, "%s " + ((long) Integer.MAX_VALUE + 1L) + " foo"));
+      assertThrows(
+          PGException.class, () -> parse(type, "%s " + ((long) Integer.MIN_VALUE - 1L) + " foo"));
+    }
+    assertEquals(ParsedFetchStatement.class, parse("fetch", "%s foo").getClass());
+    assertEquals(ParsedMoveStatement.class, parse("move", "%s foo").getClass());
+
+    assertThrows(PGException.class, () -> parse("fetch", "move foo"));
+    assertThrows(PGException.class, () -> parse("move", "fetch foo"));
+    assertThrows(PGException.class, () -> parse("fetch", "select foo"));
+    assertThrows(PGException.class, () -> parse("fetch", "fetch invalid_direction foo"));
+  }
+
+  @Test
+  public void testFetchStatement() throws Exception {
+    Statement statement = Statement.of("fetch c1");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+    ConnectionHandler connectionHandler = mock(ConnectionHandler.class);
+    ConnectionMetadata metadata = mock(ConnectionMetadata.class);
+    when(connectionHandler.getConnectionMetadata()).thenReturn(metadata);
+    FetchStatement fetchStatement =
+        new FetchStatement(
+            connectionHandler, mock(OptionsMetadata.class), "", parsedStatement, statement);
+    assertEquals(StatementType.CLIENT_SIDE, fetchStatement.getStatementType());
+    assertNull(fetchStatement.describeAsync(mock(BackendConnection.class)).get());
+  }
+
+  private static class InvalidFetchOrMoveStatement extends ParsedFetchOrMoveStatement {
+    InvalidFetchOrMoveStatement() {
+      super(null, null, null);
+    }
+  }
+
+  @Test
+  public void testInvalidParseClass() {
+    PGException exception =
+        assertThrows(
+            PGException.class,
+            () ->
+                AbstractFetchOrMoveStatement.parse(
+                    "fetch foo", "fetch", InvalidFetchOrMoveStatement.class));
+    assertEquals(SQLState.InternalError, exception.getSQLState());
+  }
+
+  ParsedFetchOrMoveStatement parse(String type, String expression) {
+    Class<? extends ParsedFetchOrMoveStatement> clazz =
+        "fetch".equalsIgnoreCase(type) ? ParsedFetchStatement.class : ParsedMoveStatement.class;
+    return AbstractFetchOrMoveStatement.parse(String.format(expression, type), type, clazz);
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java
index 8a76f47a..94b5669c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/SimpleParserTest.java
@@ -712,4 +712,20 @@ public class SimpleParserTest {
         Statement.of("select * from foo where id=$1 offset $2 limit 4611686018427387903"),
         addLimitIfParameterizedOffset(statement));
   }
+
+  @Test
+  public void testReadIntegerLiteral() {
+    assertEquals(1L, new SimpleParser("1").readIntegerLiteral().longValue());
+    assertEquals(-1L, new SimpleParser("-1").readIntegerLiteral().longValue());
+    assertEquals(100L, new SimpleParser("100").readIntegerLiteral().longValue());
+    assertEquals(1L, new SimpleParser(" 1").readIntegerLiteral().longValue());
+    assertEquals(1L, new SimpleParser("1 ").readIntegerLiteral().longValue());
+    assertEquals(1L, new SimpleParser("\n1\t").readIntegerLiteral().longValue());
+    assertEquals(1L, new SimpleParser("/* comment */1--foo").readIntegerLiteral().longValue());
+
+    assertNull(new SimpleParser("foo").readIntegerLiteral());
+    assertNull(new SimpleParser("").readIntegerLiteral());
+    assertNull(new SimpleParser("+-1").readIntegerLiteral());
+    assertNull(new SimpleParser("-+1").readIntegerLiteral());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
index 0afade94..e0ed7c3f 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
@@ -85,6 +85,7 @@ import org.postgresql.core.Oid;
 public class StatementTest {
   private static final AbstractStatementParser PARSER =
       AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
+  private static final Runnable DO_NOTHING = () -> {};
 
   private static ParsedStatement parse(String sql) {
     return PARSER.parse(Statement.of(sql));
@@ -180,6 +181,7 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             connectionHandler.getDatabaseId(),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -270,6 +272,7 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             connectionHandler.getDatabaseId(),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -309,6 +312,7 @@ public class StatementTest {
             .build();
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             connectionHandler.getDatabaseId(),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -416,6 +420,7 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             connectionHandler.getDatabaseId(),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -482,6 +487,7 @@ public class StatementTest {
 
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -554,6 +560,7 @@ public class StatementTest {
             ImmutableList.of());
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             connectionHandler.getDatabaseId(),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -579,6 +586,7 @@ public class StatementTest {
     setupQueryInformationSchemaResults();
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
@@ -632,6 +640,7 @@ public class StatementTest {
     setupQueryInformationSchemaResults();
     BackendConnection backendConnection =
         new BackendConnection(
+            DO_NOTHING,
             DatabaseId.of("p", "i", "d"),
             connection,
             () -> WellKnownClient.UNSPECIFIED,
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java
index b4092455..7a47aa9a 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/utils/ClientAutoDetectorTest.java
@@ -74,7 +74,8 @@ public class ClientAutoDetectorTest {
         ImmutableList.of(), WellKnownClient.UNSPECIFIED.createStartupNoticeResponses(connection));
     assertEquals(
         ImmutableList.of(), WellKnownClient.UNSPECIFIED.getErrorHints(mock(PGException.class)));
-    assertEquals(ImmutableMap.of(), WellKnownClient.UNSPECIFIED.getDefaultParameters());
+    assertEquals(
+        ImmutableMap.of(), WellKnownClient.UNSPECIFIED.getDefaultParameters(ImmutableMap.of()));
   }
 
   @Test
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
index 0d29bbe3..a4c1a48c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/wireprotocol/ProtocolTest.java
@@ -1104,6 +1104,7 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(backendConnection.getConnectionState()).thenReturn(ConnectionState.IDLE);
@@ -1138,6 +1139,7 @@ public class ProtocolTest {
     when(connectionHandler.getConnectionMetadata()).thenReturn(connectionMetadata);
     when(connectionMetadata.getInputStream()).thenReturn(inputStream);
     when(connectionMetadata.getOutputStream()).thenReturn(outputStream);
+
     when(connectionHandler.getExtendedQueryProtocolHandler())
         .thenReturn(extendedQueryProtocolHandler);
     when(backendConnection.getConnectionState()).thenReturn(ConnectionState.TRANSACTION);