Repository: GoogleCloudPlatform/pgadapter
Pull request #331 (2022-08-21 13:28:54)

feat: replace pg_catalog tables with common table expressions
Replaces `pg_catalog` table references with common table expressions:
1. `pg_namespace` is added with two fixed entries.
2. `pg_type` is added with 14 fixed entries.

This setup can be used to add more `pg_catalog` tables that are needed by drivers / frameworks.

codecov[bot]: # [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> Merging [#331](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (b062684) into [postgresql-dialect](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/commit/a5e563419cd0cc83557719991469ff606429a976?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (a5e5634) will **increase** coverage by `0.14%`.
> The diff coverage is `89.61%`.

```diff
@@                   Coverage Diff                    @@
##             postgresql-dialect     #331      +/-   ##
========================================================
+ Coverage                 83.54%   83.69%   +0.14%     
- Complexity                 1505     1547      +42     
========================================================
  Files                       113      114       +1     
  Lines                      5203     5335     +132     
  Branches                    674      705      +31     
========================================================
+ Hits                       4347     4465     +118     
- Misses                      639      643       +4     
- Partials                    217      227      +10     
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `83.69% <89.61%> (+0.14%)` | :arrow_up: |

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/331?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | Coverage Δ | |
|---|---|---|
| [.../cloud/spanner/pgadapter/session/SessionState.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3Nlc3Npb24vU2Vzc2lvblN0YXRlLmphdmE=) | `99.03% <ø> (+1.36%)` | :arrow_up: |
| [...dapter/statements/JdbcMetadataStatementHelper.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvSmRiY01ldGFkYXRhU3RhdGVtZW50SGVscGVyLmphdmE=) | `79.89% <ø> (-0.11%)` | :arrow_down: |
| [...oud/spanner/pgadapter/statements/SimpleParser.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvU2ltcGxlUGFyc2VyLmphdmE=) | `90.52% <87.05%> (-4.55%)` | :arrow_down: |
| [...loud/spanner/pgadapter/statements/TableParser.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvVGFibGVQYXJzZXIuamF2YQ==) | `92.45% <92.45%> (ø)` | |
| [.../cloud/spanner/pgadapter/statements/PgCatalog.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvUGdDYXRhbG9nLmphdmE=) | `92.59% <92.59%> (ø)` | |
| [...oud/spanner/pgadapter/commands/DynamicCommand.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL2NvbW1hbmRzL0R5bmFtaWNDb21tYW5kLmphdmE=) | `92.30% <100.00%> (+0.64%)` | :arrow_up: |
| [...panner/pgadapter/statements/BackendConnection.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvQmFja2VuZENvbm5lY3Rpb24uamF2YQ==) | `90.63% <100.00%> (-0.03%)` | :arrow_down: |
| [...pter/statements/IntermediatePreparedStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvSW50ZXJtZWRpYXRlUHJlcGFyZWRTdGF0ZW1lbnQuamF2YQ==) | `84.32% <100.00%> (ø)` | |
| [...er/pgadapter/statements/IntermediateStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvSW50ZXJtZWRpYXRlU3RhdGVtZW50LmphdmE=) | `95.34% <100.00%> (ø)` | |
| [...ner/pgadapter/statements/SimpleQueryStatement.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvU2ltcGxlUXVlcnlTdGF0ZW1lbnQuamF2YQ==) | `93.54% <100.00%> (+0.10%)` | :arrow_up: |
| ... and [2 more](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/331/diff?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/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
index 73948eb1..19bf81cb 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadata.java
@@ -79,6 +79,8 @@ public class OptionsMetadata {
   private static final String OPTION_DISABLE_AUTO_DETECT_CLIENT = "disable_auto_detect_client";
   private static final String OPTION_DISABLE_DEFAULT_LOCAL_STATEMENTS =
       "disable_default_local_statements";
+  private static final String OPTION_DISABLE_PG_CATALOG_REPLACEMENTS =
+      "disable_pg_catalog_replacements";
   private static final String OPTION_PSQL_MODE = "q";
   private static final String OPTION_DDL_TRANSACTION_MODE = "ddl";
   private static final String OPTION_JDBC_MODE = "jdbc";
@@ -110,6 +112,7 @@ public class OptionsMetadata {
   private final boolean authenticate;
   private final boolean disableAutoDetectClient;
   private final boolean disableDefaultLocalStatements;
+  private final boolean disablePgCatalogReplacements;
   private final boolean requiresMatcher;
   private final DdlTransactionMode ddlTransactionMode;
   private final boolean replaceJdbcMetadataQueries;
@@ -162,6 +165,8 @@ public class OptionsMetadata {
     this.disableAutoDetectClient = commandLine.hasOption(OPTION_DISABLE_AUTO_DETECT_CLIENT);
     this.disableDefaultLocalStatements =
         commandLine.hasOption(OPTION_DISABLE_DEFAULT_LOCAL_STATEMENTS);
+    this.disablePgCatalogReplacements =
+        commandLine.hasOption(OPTION_DISABLE_PG_CATALOG_REPLACEMENTS);
     this.requiresMatcher =
         commandLine.hasOption(OPTION_PSQL_MODE)
             || commandLine.hasOption(OPTION_COMMAND_METADATA_FILE);
@@ -218,6 +223,7 @@ public class OptionsMetadata {
     this.authenticate = authenticate;
     this.disableAutoDetectClient = false;
     this.disableDefaultLocalStatements = false;
+    this.disablePgCatalogReplacements = false;
     this.requiresMatcher = requiresMatcher;
     this.ddlTransactionMode = DdlTransactionMode.AutocommitImplicitTransaction;
     this.replaceJdbcMetadataQueries = replaceJdbcMetadataQueries;
@@ -516,6 +522,14 @@ public class OptionsMetadata {
             + "currently not supported by Cloud Spanner (e.g. `select current_schema`). "
             + "Use this option if you do not want PGAdapter to automatically apply query "
             + "replacements for these statements.");
+    options.addOption(
+        null,
+        OPTION_DISABLE_PG_CATALOG_REPLACEMENTS,
+        false,
+        "This option turns off automatic replacement of pg_catalog tables with "
+            + "common table expressions for pg_catalog tables that are not yet supported by "
+            + "Cloud Spanner. Use this option if you do not want PGAdapter to automatically apply "
+            + "replacements for pg_catalog tables.");
     options.addOption(
         OPTION_PSQL_MODE,
         "psql-mode",
@@ -755,6 +769,10 @@ public class OptionsMetadata {
     return !this.disableDefaultLocalStatements;
   }
 
+  public boolean replacePgCatalogTables() {
+    return !this.disablePgCatalogReplacements;
+  }
+
   public boolean requiresMatcher() {
     return this.requiresMatcher;
   }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java b/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java
index 1d232403..35240efb 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/session/SessionState.java
@@ -18,9 +18,6 @@ import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.SpannerExceptionFactory;
-import com.google.cloud.spanner.Statement;
-import com.google.cloud.spanner.Value;
-import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
@@ -86,47 +83,6 @@ public class SessionState {
     this.settings.get("server_version_num").initSettingValue(options.getServerVersionNum());
   }
 
-  /**
-   * Potentially add any session state to the given statement if that is needed. This can for
-   * example include a CTE for pg_settings, if the statement references that table. This method can
-   * be extended in the future to include CTEs for more tables that require session state, or that
-   * are not yet supported in pg_catalog.
-   */
-  public Statement addSessionState(ParsedStatement parsedStatement, Statement statement) {
-    if (parsedStatement.isQuery()
-        && parsedStatement.getSqlWithoutComments().contains("pg_settings")) {
-      String pgSettingsCte = generatePGSettingsCte();
-      // Check whether we can safely use the original SQL statement.
-      String sql =
-          startsWithIgnoreCase(statement.getSql(), "select")
-                  || startsWithIgnoreCase(statement.getSql(), "with")
-              ? statement.getSql()
-              : parsedStatement.getSqlWithoutComments();
-      Statement.Builder builder = null;
-      if (startsWithIgnoreCase(sql, "select")) {
-        builder = Statement.newBuilder("with ").append(pgSettingsCte).append(" ").append(sql);
-      } else if (startsWithIgnoreCase(sql, "with")) {
-        builder =
-            Statement.newBuilder("with ")
-                .append(pgSettingsCte)
-                .append(", ")
-                .append(sql.substring("with".length()));
-      }
-      if (builder != null) {
-        Map<String, Value> parameters = statement.getParameters();
-        for (Entry<String, Value> param : parameters.entrySet()) {
-          builder.bind(param.getKey()).to(param.getValue());
-        }
-        statement = builder.build();
-      }
-    }
-    return statement;
-  }
-
-  static boolean startsWithIgnoreCase(String string, String prefix) {
-    return string.substring(0, prefix.length()).equalsIgnoreCase(prefix);
-  }
-
   /**
    * Generates a Common Table Expression that represents the pg_settings table. Note that the
    * generated query adds two additional CTEs that could in theory hide existing user tables. It is
@@ -136,7 +92,7 @@ public class SessionState {
    * with 'pg_' always risk being hidden by user tables, unless pg_catalog has been explicitly added
    * to the search_path after one or more user schemas.
    */
-  String generatePGSettingsCte() {
+  public String generatePGSettingsCte() {
     return "pg_settings_inmem_ as (\n"
         + getAll().stream()
             .filter(setting -> SUPPORTED_PG_SETTINGS_KEYS.contains(setting.getCasePreservingKey()))
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 ee1e2484..445dea5d 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
@@ -203,10 +203,12 @@ public class BackendConnection {
         } else if (parsedStatement.isDdl()) {
           result.set(ddlExecutor.execute(parsedStatement, statement));
         } else {
-          // Potentially add session state in the form of CTE(s) to the statement.
-          Statement statementWithSessionState =
-              sessionState.addSessionState(parsedStatement, statement);
-          result.set(spannerConnection.execute(statementWithSessionState));
+          // Potentially replace pg_catalog table references with common table expressions.
+          Statement updatedStatement =
+              optionsMetadata.replacePgCatalogTables()
+                  ? pgCatalog.replacePgCatalogTables(statement)
+                  : statement;
+          result.set(spannerConnection.execute(updatedStatement));
         }
       } catch (SpannerException spannerException) {
         // Executing queries against the information schema in a transaction is unsupported.
@@ -370,6 +372,7 @@ public class BackendConnection {
   private static final Statement ROLLBACK = Statement.of("ROLLBACK");
 
   private final SessionState sessionState;
+  private final PgCatalog pgCatalog;
   private final ImmutableMap<String, LocalStatement> localStatements;
   private ConnectionState connectionState = ConnectionState.IDLE;
   private TransactionMode transactionMode = TransactionMode.IMPLICIT;
@@ -390,6 +393,7 @@ public class BackendConnection {
       OptionsMetadata optionsMetadata,
       ImmutableList<LocalStatement> localStatements) {
     this.sessionState = new SessionState(optionsMetadata);
+    this.pgCatalog = new PgCatalog(this.sessionState);
     this.spannerConnection = spannerConnection;
     this.databaseId = databaseId;
     this.ddlExecutor = new DdlExecutor(databaseId, this);
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/JdbcMetadataStatementHelper.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/JdbcMetadataStatementHelper.java
index 031fd663..9dada6b5 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/JdbcMetadataStatementHelper.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/JdbcMetadataStatementHelper.java
@@ -155,7 +155,7 @@ class JdbcMetadataStatementHelper {
         .replace(
             " JOIN pg_catalog.pg_description",
             String.format(" JOIN (%s)", PgJdbcCatalog.PG_DESCRIPTION))
-        .replace(" pg_catalog.pg_type ", String.format(" (%s) ", PgJdbcCatalog.PG_TYPE))
+        // .replace(" pg_catalog.pg_type ", String.format(" (%s) ", PgJdbcCatalog.PG_TYPE))
         .replace(" pg_catalog.pg_am", String.format(" (%s)", PgJdbcCatalog.PG_AM))
         .replaceAll(
             "\\s+FROM\\s+pg_catalog\\.pg_settings",
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/PgCatalog.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/PgCatalog.java
new file mode 100644
index 00000000..2a96ef03
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/PgCatalog.java
@@ -0,0 +1,227 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import com.google.api.core.InternalApi;
+import com.google.cloud.Tuple;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.Value;
+import com.google.cloud.spanner.pgadapter.session.SessionState;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+
+@InternalApi
+public class PgCatalog {
+  private static final ImmutableMap<TableOrIndexName, TableOrIndexName> TABLE_REPLACEMENTS =
+      ImmutableMap.of(
+          new TableOrIndexName("pg_catalog", "pg_namespace"),
+              new TableOrIndexName(null, "pg_namespace"),
+          new TableOrIndexName(null, "pg_namespace"), new TableOrIndexName(null, "pg_namespace"),
+          new TableOrIndexName("pg_catalog", "pg_type"), new TableOrIndexName(null, "pg_type"),
+          new TableOrIndexName(null, "pg_type"), new TableOrIndexName(null, "pg_type"),
+          new TableOrIndexName("pg_catalog", "pg_settings"),
+              new TableOrIndexName(null, "pg_settings"),
+          new TableOrIndexName(null, "pg_settings"), new TableOrIndexName(null, "pg_settings"));
+
+  private final Map<TableOrIndexName, PgCatalogTable> pgCatalogTables =
+      ImmutableMap.of(
+          new TableOrIndexName(null, "pg_namespace"), new PgNamespace(),
+          new TableOrIndexName(null, "pg_type"), new PgType(),
+          new TableOrIndexName(null, "pg_settings"), new PgSettings());
+  private final SessionState sessionState;
+
+  public PgCatalog(SessionState sessionState) {
+    this.sessionState = sessionState;
+  }
+
+  /** Replace supported pg_catalog tables with Common Table Expressions. */
+  public Statement replacePgCatalogTables(Statement statement) {
+    Tuple<Set<TableOrIndexName>, Statement> replacedTablesStatement =
+        new TableParser(statement).detectAndReplaceTables(TABLE_REPLACEMENTS);
+    if (replacedTablesStatement.x().isEmpty()) {
+      return replacedTablesStatement.y();
+    }
+    // Add Common Table Expressions for the pg_catalog tables that were detected and replaced in the
+    // statement.
+    Set<TableOrIndexName> alreadyAdded = new HashSet<>();
+    ImmutableList.Builder<String> cteBuilder = ImmutableList.builder();
+    for (TableOrIndexName table : replacedTablesStatement.x()) {
+      addPgCatalogTable(table, getPgCatalogTable(table), cteBuilder, alreadyAdded);
+    }
+
+    return addCommonTableExpressions(replacedTablesStatement.y(), cteBuilder.build());
+  }
+
+  static Statement addCommonTableExpressions(
+      Statement statement, ImmutableList<String> tableExpressions) {
+    SimpleParser parser = new SimpleParser(statement.getSql());
+    boolean hadCommonTableExpressions = parser.eatKeyword("with");
+    String tableExpressionsSql = String.join(",\n", tableExpressions);
+    Statement.Builder builder =
+        Statement.newBuilder("with ")
+            .append(tableExpressionsSql)
+            .append(hadCommonTableExpressions ? ",\n" : "\n");
+    if (hadCommonTableExpressions) {
+      // Include the entire original statement except the 'with' keyword.
+      builder
+          .append(parser.getSql().substring(0, parser.getPos() - 4))
+          .append(parser.getSql().substring(parser.getPos()));
+    } else {
+      // Include the entire original statement (including any comments at the beginning).
+      builder.append(parser.getSql());
+    }
+    Map<String, Value> parameters = statement.getParameters();
+    for (Entry<String, Value> param : parameters.entrySet()) {
+      builder.bind(param.getKey()).to(param.getValue());
+    }
+    statement = builder.build();
+    return statement;
+  }
+
+  void addPgCatalogTable(
+      TableOrIndexName tableName,
+      PgCatalogTable pgCatalogTable,
+      ImmutableList.Builder<String> cteBuilder,
+      Set<TableOrIndexName> alreadyAdded) {
+    if (pgCatalogTable == null) {
+      return;
+    }
+    for (TableOrIndexName dependency : pgCatalogTable.getDependencies()) {
+      addPgCatalogTable(dependency, getPgCatalogTable(dependency), cteBuilder, alreadyAdded);
+    }
+    if (alreadyAdded.add(tableName)) {
+      cteBuilder.add(pgCatalogTable.getTableExpression());
+    }
+  }
+
+  PgCatalogTable getPgCatalogTable(TableOrIndexName tableOrIndexName) {
+    if (pgCatalogTables.containsKey(tableOrIndexName)) {
+      return pgCatalogTables.get(tableOrIndexName);
+    }
+    return null;
+  }
+
+  private interface PgCatalogTable {
+    String getTableExpression();
+
+    default ImmutableSet<TableOrIndexName> getDependencies() {
+      return ImmutableSet.of();
+    }
+  }
+
+  private static class PgNamespace implements PgCatalogTable {
+    private static final String PG_NAMESPACE_CTE =
+        "pg_namespace as (\n"
+            + "  select case schema_name when 'pg_catalog' then 11 when 'public' then 2200 else 0 end as oid,\n"
+            + "        schema_name as nspname, null as nspowner, null as nspacl\n"
+            + "  from information_schema.schemata\n"
+            + ")";
+
+    @Override
+    public String getTableExpression() {
+      return PG_NAMESPACE_CTE;
+    }
+  }
+
+  private static class PgType implements PgCatalogTable {
+    private static final ImmutableSet<TableOrIndexName> DEPENDENCIES =
+        ImmutableSet.of(new TableOrIndexName(null, "pg_namespace"));
+    private static final String GENERATION_SQL =
+        "select 'select '\n"
+            + "  || oid || ' as oid, '\n"
+            + "  || '''' || typname || ''' as typname, '\n"
+            + "  || '(select oid from pg_namespace where nspname=''pg_catalog'') as typnamespace, '\n"
+            + "  || 'null as typowner, '\n"
+            + "  || typlen || ' as typlen, '\n"
+            + "  || typbyval::text || ' as typbyval, '\n"
+            + "  || '''' || typtype || ''' as typtype, '\n"
+            + "  || '''' || typcategory || ''' as typcategory, '\n"
+            + "  || typispreferred::text || ' as typispreferred, '\n"
+            + "  || (case typname\n"
+            + "        when 'int2' then 'false'\n"
+            + "        when 'int4' then 'false'\n"
+            + "        when 'float4' then 'false'\n"
+            + "        when 'timestamp' then 'false'\n"
+            + "        else 'true'\n"
+            + "      end) || ' as typisdefined, '\n"
+            + "  || '''' || typdelim || ''' as typdelim, '\n"
+            + "  || typrelid || ' as typrelid, '\n"
+            + "  || typelem || ' as typelem, '\n"
+            + "  || typarray || ' as typarray, '\n"
+            + "  || '''' || typinput || ''' as typinput, '\n"
+            + "  || '''' || typoutput || ''' as typoutput, '\n"
+            + "  || '''' || typreceive || ''' as typreceive, '\n"
+            + "  || '''' || typsend || ''' as typsend, '\n"
+            + "  || '''' || typmodin || ''' as typmodin, '\n"
+            + "  || '''' || typmodout || ''' as typmodout, '\n"
+            + "  || '''' || typanalyze || ''' as typanalyze, '\n"
+            + "  || '''' || typalign || ''' as typalign, '\n"
+            + "  || '''' || typstorage || ''' as typstorage, '\n"
+            + "  || typnotnull::text || ' as typnotnull, '\n"
+            + "  || typbasetype || ' as typbasetype, '\n"
+            + "  || typtypmod || ' as typtypmod, '\n"
+            + "  || coalesce(typndims || ' as typndims, ', 'null as typndims, ')\n"
+            + "  || coalesce(typcollation || ' as typcollation, ', 'null as typcollation, ')\n"
+            + "  || coalesce('''' || typdefaultbin || ''' as typdefaultbin, ', 'null as typdefaultbin, ')\n"
+            + "  || coalesce('''' || typdefault || ''' as typdefault, ', 'null as typdefault, ')\n"
+            + "  || 'null as typacl '\n"
+            + "  || 'union all'\n"
+            + "from pg_type\n"
+            + "where typname in ('bool', 'bytea', 'int2', 'int4', 'int8', 'float4', 'float8',\n"
+            + "                  'numeric', 'varchar', 'text', 'jsonb', 'timestamp', 'timestamptz', 'date')\n"
+            + ";\n";
+    private static final String PG_TYPE_CTE =
+        "pg_type as (\n"
+            + "  select 16 as oid, 'bool' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 1 as typlen, true as typbyval, 'b' as typtype, 'B' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1000 as typarray, 'boolin' as typinput, 'boolout' as typoutput, 'boolrecv' as typreceive, 'boolsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'c' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 17 as oid, 'bytea' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1001 as typarray, 'byteain' as typinput, 'byteaout' as typoutput, 'bytearecv' as typreceive, 'byteasend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 20 as oid, 'int8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1016 as typarray, 'int8in' as typinput, 'int8out' as typoutput, 'int8recv' as typreceive, 'int8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 21 as oid, 'int2' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 2 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1005 as typarray, 'int2in' as typinput, 'int2out' as typoutput, 'int2recv' as typreceive, 'int2send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 's' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 23 as oid, 'int4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1007 as typarray, 'int4in' as typinput, 'int4out' as typoutput, 'int4recv' as typreceive, 'int4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 25 as oid, 'text' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1009 as typarray, 'textin' as typinput, 'textout' as typoutput, 'textrecv' as typreceive, 'textsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 700 as oid, 'float4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1021 as typarray, 'float4in' as typinput, 'float4out' as typoutput, 'float4recv' as typreceive, 'float4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 701 as oid, 'float8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1022 as typarray, 'float8in' as typinput, 'float8out' as typoutput, 'float8recv' as typreceive, 'float8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 1043 as oid, 'varchar' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1015 as typarray, 'varcharin' as typinput, 'varcharout' as typoutput, 'varcharrecv' as typreceive, 'varcharsend' as typsend, 'varchartypmodin' as typmodin, 'varchartypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 1082 as oid, 'date' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1182 as typarray, 'date_in' as typinput, 'date_out' as typoutput, 'date_recv' as typreceive, 'date_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 1114 as oid, 'timestamp' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1115 as typarray, 'timestamp_in' as typinput, 'timestamp_out' as typoutput, 'timestamp_recv' as typreceive, 'timestamp_send' as typsend, 'timestamptypmodin' as typmodin, 'timestamptypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 1184 as oid, 'timestamptz' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1185 as typarray, 'timestamptz_in' as typinput, 'timestamptz_out' as typoutput, 'timestamptz_recv' as typreceive, 'timestamptz_send' as typsend, 'timestamptztypmodin' as typmodin, 'timestamptztypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 1700 as oid, 'numeric' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1231 as typarray, 'numeric_in' as typinput, 'numeric_out' as typoutput, 'numeric_recv' as typreceive, 'numeric_send' as typsend, 'numerictypmodin' as typmodin, 'numerictypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'm' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+            + "  select 3802 as oid, 'jsonb' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 3807 as typarray, 'jsonb_in' as typinput, 'jsonb_out' as typoutput, 'jsonb_recv' as typreceive, 'jsonb_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl\n"
+            + ")";
+
+    @Override
+    public String getTableExpression() {
+      return PG_TYPE_CTE;
+    }
+
+    @Override
+    public ImmutableSet<TableOrIndexName> getDependencies() {
+      return DEPENDENCIES;
+    }
+  }
+
+  private class PgSettings implements PgCatalogTable {
+
+    @Override
+    public String getTableExpression() {
+      return sessionState.generatePGSettingsCte();
+    }
+  }
+}
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 0bab4e1d..9b4b2c50 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/SimpleParser.java
@@ -14,12 +14,16 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
-import com.google.api.client.util.Strings;
 import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.Value;
 import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Objects;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -66,16 +70,40 @@ public class SimpleParser {
         return false;
       }
       TableOrIndexName other = (TableOrIndexName) o;
-      return Objects.equals(this.schema, other.schema) && Objects.equals(this.name, other.name);
+      return Objects.equals(
+              unquoteOrFoldIdentifier(this.schema), unquoteOrFoldIdentifier(other.schema))
+          && Objects.equals(
+              unquoteOrFoldIdentifier(this.name), unquoteOrFoldIdentifier(other.name));
     }
 
     @Override
     public int hashCode() {
-      return Objects.hash(this.schema, this.name);
+      return Objects.hash(unquoteOrFoldIdentifier(this.schema), unquoteOrFoldIdentifier(this.name));
     }
   }
 
-  private final String sql;
+  static String unquoteOrFoldIdentifier(String identifier) {
+    if (Strings.isNullOrEmpty(identifier)) {
+      return null;
+    }
+    if (identifier.charAt(0) == '"'
+        && identifier.charAt(identifier.length() - 1) == '"'
+        && identifier.length() > 1) {
+      return identifier.substring(1, identifier.length() - 1);
+    }
+    return identifier.toLowerCase();
+  }
+
+  static Statement copyStatement(Statement original, String sql) {
+    Statement.Builder builder = Statement.newBuilder(sql);
+    Map<String, Value> parameters = original.getParameters();
+    for (Entry<String, Value> param : parameters.entrySet()) {
+      builder.bind(param.getKey()).to(param.getValue());
+    }
+    return builder.build();
+  }
+
+  private String sql;
   private int pos;
 
   SimpleParser(String sql) {
@@ -86,6 +114,10 @@ public class SimpleParser {
     return sql;
   }
 
+  void setSql(String sql) {
+    this.sql = sql;
+  }
+
   int getPos() {
     return pos;
   }
@@ -194,7 +226,8 @@ public class SimpleParser {
       String expression =
           parseExpressionUntilKeyword(
               keyword == null ? ImmutableList.of() : ImmutableList.of(keyword),
-              sameParensLevelAsStart);
+              sameParensLevelAsStart,
+              true);
       if (expression == null) {
         return null;
       }
@@ -214,11 +247,13 @@ public class SimpleParser {
   }
 
   String parseExpressionUntilKeyword(ImmutableList<String> keywords) {
-    return parseExpressionUntilKeyword(keywords, false);
+    return parseExpressionUntilKeyword(keywords, false, true);
   }
 
   String parseExpressionUntilKeyword(
-      ImmutableList<String> keywords, boolean sameParensLevelAsStart) {
+      ImmutableList<String> keywords,
+      boolean sameParensLevelAsStart,
+      boolean stopAtEndOfExpression) {
     skipWhitespaces();
     int start = pos;
     boolean valid;
@@ -228,10 +263,10 @@ public class SimpleParser {
         parens++;
       } else if (sql.charAt(pos) == ')') {
         parens--;
-        if (parens < 0) {
+        if (stopAtEndOfExpression && parens < 0) {
           break;
         }
-      } else if (parens == 0 && sql.charAt(pos) == ',') {
+      } else if (stopAtEndOfExpression && parens == 0 && sql.charAt(pos) == ',') {
         break;
       }
       if ((!sameParensLevelAsStart || parens == 0)
@@ -342,6 +377,28 @@ public class SimpleParser {
     return eat(true, false, token);
   }
 
+  /** Eats everything until an end parentheses at the same level as the current level. */
+  String eatSubExpression() {
+    int start = pos;
+    boolean valid;
+    int parens = 0;
+    while ((valid = skipCommentsAndLiterals()) && pos < sql.length()) {
+      if (sql.charAt(pos) == '(') {
+        parens++;
+      } else if (sql.charAt(pos) == ')') {
+        parens--;
+        if (parens < 0) {
+          break;
+        }
+      }
+      pos++;
+    }
+    if (pos == start || !valid || parens >= 0) {
+      return null;
+    }
+    return sql.substring(start, pos);
+  }
+
   boolean eatDotOperator() {
     if (eat(false, false, ".")) {
       if (pos == sql.length() || Character.isWhitespace(sql.charAt(pos))) {
@@ -352,6 +409,44 @@ public class SimpleParser {
     return false;
   }
 
+  boolean peekJoinKeyword() {
+    return peekKeyword("join")
+        || peekKeyword("left")
+        || peekKeyword("right")
+        || peekKeyword("full")
+        || peekKeyword("inner")
+        || peekKeyword("outer")
+        || peekKeyword("cross");
+  }
+
+  boolean eatJoinType() {
+    boolean foundValidJoin;
+    if (eatKeyword("left") || eatKeyword("right") || eatKeyword("full")) {
+      eatKeyword("outer");
+      foundValidJoin = eatKeyword("join");
+    } else if (eatKeyword("inner")) {
+      foundValidJoin = eatKeyword("join");
+    } else if (eatKeyword("cross")) {
+      foundValidJoin = eatKeyword("join");
+    } else {
+      foundValidJoin = eatKeyword("join");
+    }
+    return foundValidJoin;
+  }
+
+  boolean eatJoinCondition() {
+    if (eatKeyword("on")) {
+      parseExpressionUntilKeyword(
+          ImmutableList.of("left", "right", "full", "inner", "cross", "join", "where"), true, true);
+    } else if (eatKeyword("using")) {
+      if (eatToken("(")) {
+        parseExpressionList();
+        return eatToken(")");
+      }
+    }
+    return false;
+  }
+
   boolean eat(boolean skipWhitespaceBefore, boolean requireWhitespaceAfter, String... keywords) {
     boolean result = true;
     for (String keyword : keywords) {
@@ -525,4 +620,16 @@ public class SimpleParser {
     }
     return false;
   }
+
+  @Override
+  public String toString() {
+    if (sql.contains("\n")) {
+      return sql.substring(0, pos) + "|" + sql.substring(Math.min(pos, sql.length() - 1));
+    }
+    return sql
+        + "\n"
+        + Strings.repeat(" ", pos)
+        + "^"
+        + Strings.repeat(" ", Math.max(sql.length() - pos - 1, 0));
+  }
 }
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/TableParser.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/TableParser.java
new file mode 100644
index 00000000..852e7c97
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/TableParser.java
@@ -0,0 +1,173 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import com.google.cloud.Tuple;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.Set;
+
+class TableParser {
+  private static final ImmutableSet<TableOrIndexName> EMPTY_TABLE_SET = ImmutableSet.of();
+  private static final ImmutableList<String> KEYWORDS_BEFORE_TABLE =
+      ImmutableList.of("from", "join", "insert", "update", "delete");
+  private final Statement originalStatement;
+  private final SimpleParser parser;
+
+  TableParser(Statement statement) {
+    this.originalStatement = statement;
+    this.parser = new SimpleParser(statement.getSql());
+  }
+
+  Tuple<Set<TableOrIndexName>, Statement> detectAndReplaceTables(
+      ImmutableMap<TableOrIndexName, TableOrIndexName> detectAndReplaceMap) {
+    return detectAndReplaceTables(detectAndReplaceMap, true);
+  }
+
+  Tuple<Set<TableOrIndexName>, Statement> detectAndReplaceTables(
+      ImmutableMap<TableOrIndexName, TableOrIndexName> detectAndReplaceMap,
+      boolean searchForKeywordBeforeFirstTable) {
+    boolean potentialMatch = false;
+    String lowerCaseSql = parser.getSql().toLowerCase();
+    for (Entry<TableOrIndexName, TableOrIndexName> entry : detectAndReplaceMap.entrySet()) {
+      if (lowerCaseSql.contains(entry.getKey().name.toLowerCase())
+          && (entry.getKey().schema == null
+              || lowerCaseSql.contains(entry.getKey().schema.toLowerCase()))) {
+        potentialMatch = true;
+        break;
+      }
+    }
+    if (!potentialMatch) {
+      return Tuple.of(EMPTY_TABLE_SET, originalStatement);
+    }
+
+    ImmutableSet.Builder<TableOrIndexName> detectedTablesBuilder = ImmutableSet.builder();
+    boolean detectedOrReplacedTable = false;
+    while (parser.getPos() < parser.getSql().length()) {
+      boolean multipleTables = false;
+      if (searchForKeywordBeforeFirstTable) {
+        parser.parseExpressionUntilKeyword(KEYWORDS_BEFORE_TABLE, false, false);
+        if (parser.getPos() >= parser.getSql().length()) {
+          break;
+        }
+
+        if (parser.eatKeyword("insert")) {
+          parser.eatKeyword("into");
+        } else if (parser.eatKeyword("delete")) {
+          parser.eatKeyword("from");
+        } else if (parser.eatKeyword("update")) {
+          parser.eatKeyword("only");
+        } else if (parser.eatKeyword("from") || parser.eatKeyword("join")) {
+          // There could be multiple tables.
+          multipleTables = true;
+        } else {
+          // This shouldn't happen.
+          return Tuple.of(EMPTY_TABLE_SET, originalStatement);
+        }
+      }
+      searchForKeywordBeforeFirstTable = true;
+      detectedOrReplacedTable =
+          parseTableList(detectAndReplaceMap, detectedTablesBuilder, multipleTables)
+              || detectedOrReplacedTable;
+    }
+    return detectedOrReplacedTable
+        ? Tuple.of(
+            detectedTablesBuilder.build(),
+            SimpleParser.copyStatement(originalStatement, parser.getSql()))
+        : Tuple.of(EMPTY_TABLE_SET, originalStatement);
+  }
+
+  private boolean parseTableList(
+      ImmutableMap<TableOrIndexName, TableOrIndexName> detectAndReplaceMap,
+      ImmutableSet.Builder<TableOrIndexName> detectedTablesBuilder,
+      boolean multipleTables) {
+    boolean detectedOrReplacedTable = false;
+    boolean wasJoin = false;
+    do {
+      if (parser.eatToken("(")) {
+        int startPosition = parser.getPos();
+        String subExpression = parser.eatSubExpression();
+        TableParser subParser = new TableParser(Statement.of(subExpression));
+        Tuple<Set<TableOrIndexName>, Statement> subResult =
+            subParser.detectAndReplaceTables(
+                detectAndReplaceMap,
+                subParser.parser.peekKeyword("select") || subParser.parser.peekKeyword("with"));
+        if (!subResult.x().isEmpty()) {
+          detectedOrReplacedTable = true;
+          detectedTablesBuilder.addAll(subResult.x());
+          parser.setSql(
+              parser.getSql().substring(0, startPosition)
+                  + subResult.y().getSql()
+                  + parser.getSql().substring(parser.getPos()));
+          // Set the position of the parser to after the replaced sub-expression.
+          parser.setPos(startPosition + subResult.y().getSql().length());
+          parser.eatToken(")");
+        }
+      } else {
+        // Skip all whitespaces to get the actual position before the next table name.
+        parser.skipWhitespaces();
+        int positionBeforeName = parser.getPos();
+        TableOrIndexName tableOrIndexName = parser.readTableOrIndexName();
+        if (tableOrIndexName == null) {
+          break;
+        }
+        if (detectAndReplaceMap.containsKey(tableOrIndexName)) {
+          detectedOrReplacedTable = true;
+          // Add the translated table name to the set of discovered tables so that a CTE can be
+          // added
+          // for it.
+          detectedTablesBuilder.add(
+              Objects.requireNonNull(detectAndReplaceMap.get(tableOrIndexName)));
+          // Check if the entry in the table map contains a different replacement value than the
+          // original. Some tables may be added to the map of replacements with the same replacement
+          // value as the original with the sole purpose of detecting the use of the table.
+          if (!Objects.equals(detectAndReplaceMap.get(tableOrIndexName), tableOrIndexName)) {
+            parser.setSql(
+                parser.getSql().substring(0, positionBeforeName)
+                    + detectAndReplaceMap.get(tableOrIndexName)
+                    + parser.getSql().substring(parser.getPos()));
+            // Set the position of the parser to after the replaced table name.
+            parser.setPos(
+                positionBeforeName
+                    + Objects.requireNonNull(detectAndReplaceMap.get(tableOrIndexName))
+                        .toString()
+                        .length());
+          }
+        }
+      }
+      // Skip any aliases.
+      if (multipleTables && !parser.peekJoinKeyword()) {
+        parser.eatKeyword("as");
+        parser.readIdentifierPart();
+      }
+      if (wasJoin) {
+        // Skip the join condition.
+        parser.eatJoinCondition();
+      }
+    } while (multipleTables && (parser.eatToken(",") || (wasJoin = parser.eatJoinType())));
+
+    return detectedOrReplacedTable;
+  }
+
+  @Override
+  public String toString() {
+    return parser.toString();
+  }
+}


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
index c984dc7a..be8237f8 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
@@ -56,6 +56,7 @@ import org.junit.runners.Parameterized.Parameter;
 import org.junit.runners.Parameterized.Parameters;
 import org.postgresql.PGConnection;
 import org.postgresql.copy.CopyManager;
+import org.postgresql.core.Oid;
 
 @Category(IntegrationTest.class)
 @RunWith(Parameterized.class)
@@ -805,6 +806,82 @@ public class ITJdbcTest implements IntegrationTest {
     }
   }
 
+  @Test
+  public void testSelectNamespaces() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      try (ResultSet namespaces =
+          connection
+              .createStatement()
+              .executeQuery("select nspname from pg_namespace order by oid desc, nspname")) {
+        assertTrue(namespaces.next());
+        assertEquals("public", namespaces.getString(1));
+        assertTrue(namespaces.next());
+        assertEquals("pg_catalog", namespaces.getString(1));
+        assertTrue(namespaces.next());
+        assertEquals("information_schema", namespaces.getString(1));
+        assertTrue(namespaces.next());
+        assertEquals("spanner_sys", namespaces.getString(1));
+
+        assertFalse(namespaces.next());
+      }
+    }
+  }
+
+  @Test
+  public void testSelectTypes() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      try (ResultSet types =
+          connection
+              .createStatement()
+              .executeQuery("select oid, typname from pg_type order by oid")) {
+        assertTrue(types.next());
+        assertEquals(Oid.BOOL, types.getInt(1));
+        assertEquals("bool", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.BYTEA, types.getInt(1));
+        assertEquals("bytea", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.INT8, types.getInt(1));
+        assertEquals("int8", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.INT2, types.getInt(1));
+        assertEquals("int2", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.INT4, types.getInt(1));
+        assertEquals("int4", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.TEXT, types.getInt(1));
+        assertEquals("text", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.FLOAT4, types.getInt(1));
+        assertEquals("float4", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.FLOAT8, types.getInt(1));
+        assertEquals("float8", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.VARCHAR, types.getInt(1));
+        assertEquals("varchar", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.DATE, types.getInt(1));
+        assertEquals("date", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.TIMESTAMP, types.getInt(1));
+        assertEquals("timestamp", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.TIMESTAMPTZ, types.getInt(1));
+        assertEquals("timestamptz", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.NUMERIC, types.getInt(1));
+        assertEquals("numeric", types.getString(2));
+        assertTrue(types.next());
+        assertEquals(Oid.JSONB, types.getInt(1));
+        assertEquals("jsonb", types.getString(2));
+
+        assertFalse(types.next());
+      }
+    }
+  }
+
   private void writeExtraTestRows() {
     testEnv.write(
         database.getId().getDatabase(),
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 d628f377..aa9bc8ab 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITPsqlTest.java
@@ -22,11 +22,13 @@ import static org.junit.Assert.assertTrue;
 import static org.junit.Assume.assumeFalse;
 import static org.junit.Assume.assumeTrue;
 
+import com.google.cloud.Tuple;
 import com.google.cloud.spanner.Database;
 import com.google.cloud.spanner.KeySet;
 import com.google.cloud.spanner.Mutation;
 import com.google.common.base.Strings;
 import java.io.BufferedReader;
+import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStreamReader;
 import java.io.OutputStreamWriter;
@@ -216,6 +218,44 @@ public class ITPsqlTest implements IntegrationTest {
         testEnv.getPGAdapterHostAndPort(), database.getId().getDatabase());
   }
 
+  /**
+   * Runs the given SQL statement using psql and returns a String tuple containing the stdout and
+   * stderr text.
+   */
+  private Tuple<String, String> runUsingPsql(String sql) throws IOException, InterruptedException {
+    ProcessBuilder builder = new ProcessBuilder();
+    String[] psqlCommand =
+        new String[] {
+          "psql",
+          "-h",
+          (POSTGRES_HOST.startsWith("/") ? "/tmp" : testEnv.getPGAdapterHost()),
+          "-p",
+          String.valueOf(testEnv.getPGAdapterPort()),
+          "-d",
+          database.getId().getDatabase()
+        };
+
+    builder.command(psqlCommand);
+    setPgPassword(builder);
+    Process process = builder.start();
+    String errors;
+    String output;
+
+    try (OutputStreamWriter writer = new OutputStreamWriter(process.getOutputStream());
+        BufferedReader reader =
+            new BufferedReader(new InputStreamReader(process.getInputStream()));
+        BufferedReader errorReader =
+            new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+      writer.write(sql);
+      writer.write("\\q\n");
+      writer.flush();
+      errors = errorReader.lines().collect(Collectors.joining("\n"));
+      output = reader.lines().collect(Collectors.joining("\n"));
+    }
+    process.waitFor();
+    return Tuple.of(output, errors);
+  }
+
   @Test
   public void testRealPostgreSQLHelloWorld() throws Exception {
     ProcessBuilder builder = new ProcessBuilder();
@@ -255,6 +295,38 @@ public class ITPsqlTest implements IntegrationTest {
     assertEquals(0, res);
   }
 
+  @Test
+  public void testSelectPgCatalogTables() throws IOException, InterruptedException {
+    String sql =
+        "select pg_type.typname\n"
+            + "from pg_catalog.pg_type\n"
+            + "inner join pg_catalog.pg_namespace on pg_type.typnamespace=pg_namespace.oid\n"
+            + "order by pg_type.oid;\n";
+    Tuple<String, String> result = runUsingPsql(sql);
+    String output = result.x(), errors = result.y();
+
+    assertEquals("", errors);
+    assertEquals(
+        "   typname   \n"
+            + "-------------\n"
+            + " bool\n"
+            + " bytea\n"
+            + " int8\n"
+            + " int2\n"
+            + " int4\n"
+            + " text\n"
+            + " float4\n"
+            + " float8\n"
+            + " varchar\n"
+            + " date\n"
+            + " timestamp\n"
+            + " timestamptz\n"
+            + " numeric\n"
+            + " jsonb\n"
+            + "(14 rows)\n",
+        output);
+  }
+
   @Test
   public void testCopyFromPostgreSQLToCloudSpanner() throws Exception {
     int numRows = 100;
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
index c4b58e1c..d1a4c30b 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
@@ -2090,6 +2090,92 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     }
   }
 
+  @Test
+  public void testSelectPgType() throws SQLException {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(
+                "with pg_namespace as (\n"
+                    + "  select case schema_name when 'pg_catalog' then 11 when 'public' then 2200 else 0 end as oid,\n"
+                    + "        schema_name as nspname, null as nspowner, null as nspacl\n"
+                    + "  from information_schema.schemata\n"
+                    + "),\n"
+                    + "pg_type as (\n"
+                    + "  select 16 as oid, 'bool' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 1 as typlen, true as typbyval, 'b' as typtype, 'B' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1000 as typarray, 'boolin' as typinput, 'boolout' as typoutput, 'boolrecv' as typreceive, 'boolsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'c' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 17 as oid, 'bytea' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1001 as typarray, 'byteain' as typinput, 'byteaout' as typoutput, 'bytearecv' as typreceive, 'byteasend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 20 as oid, 'int8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1016 as typarray, 'int8in' as typinput, 'int8out' as typoutput, 'int8recv' as typreceive, 'int8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 21 as oid, 'int2' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 2 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1005 as typarray, 'int2in' as typinput, 'int2out' as typoutput, 'int2recv' as typreceive, 'int2send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 's' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 23 as oid, 'int4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1007 as typarray, 'int4in' as typinput, 'int4out' as typoutput, 'int4recv' as typreceive, 'int4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 25 as oid, 'text' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1009 as typarray, 'textin' as typinput, 'textout' as typoutput, 'textrecv' as typreceive, 'textsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 700 as oid, 'float4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1021 as typarray, 'float4in' as typinput, 'float4out' as typoutput, 'float4recv' as typreceive, 'float4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 701 as oid, 'float8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1022 as typarray, 'float8in' as typinput, 'float8out' as typoutput, 'float8recv' as typreceive, 'float8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1043 as oid, 'varchar' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1015 as typarray, 'varcharin' as typinput, 'varcharout' as typoutput, 'varcharrecv' as typreceive, 'varcharsend' as typsend, 'varchartypmodin' as typmodin, 'varchartypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1082 as oid, 'date' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1182 as typarray, 'date_in' as typinput, 'date_out' as typoutput, 'date_recv' as typreceive, 'date_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1114 as oid, 'timestamp' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1115 as typarray, 'timestamp_in' as typinput, 'timestamp_out' as typoutput, 'timestamp_recv' as typreceive, 'timestamp_send' as typsend, 'timestamptypmodin' as typmodin, 'timestamptypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1184 as oid, 'timestamptz' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1185 as typarray, 'timestamptz_in' as typinput, 'timestamptz_out' as typoutput, 'timestamptz_recv' as typreceive, 'timestamptz_send' as typsend, 'timestamptztypmodin' as typmodin, 'timestamptztypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1700 as oid, 'numeric' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1231 as typarray, 'numeric_in' as typinput, 'numeric_out' as typoutput, 'numeric_recv' as typreceive, 'numeric_send' as typsend, 'numerictypmodin' as typmodin, 'numerictypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'm' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 3802 as oid, 'jsonb' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 3807 as typarray, 'jsonb_in' as typinput, 'jsonb_out' as typoutput, 'jsonb_recv' as typreceive, 'jsonb_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl\n"
+                    + ")\n"
+                    + "select * from pg_type"),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (ResultSet resultSet =
+          connection.createStatement().executeQuery("select * from pg_catalog.pg_type")) {
+        // The result is not consistent with selecting from pg_type, but that's not relevant here.
+        // We just want to ensure that it includes the correct common table expressions.
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+    }
+  }
+
+  @Test
+  public void testSelectPgTypeAndPgNamespace() throws SQLException {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.of(
+                "with pg_namespace as (\n"
+                    + "  select case schema_name when 'pg_catalog' then 11 when 'public' then 2200 else 0 end as oid,\n"
+                    + "        schema_name as nspname, null as nspowner, null as nspacl\n"
+                    + "  from information_schema.schemata\n"
+                    + "),\n"
+                    + "pg_type as (\n"
+                    + "  select 16 as oid, 'bool' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 1 as typlen, true as typbyval, 'b' as typtype, 'B' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1000 as typarray, 'boolin' as typinput, 'boolout' as typoutput, 'boolrecv' as typreceive, 'boolsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'c' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 17 as oid, 'bytea' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1001 as typarray, 'byteain' as typinput, 'byteaout' as typoutput, 'bytearecv' as typreceive, 'byteasend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 20 as oid, 'int8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1016 as typarray, 'int8in' as typinput, 'int8out' as typoutput, 'int8recv' as typreceive, 'int8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 21 as oid, 'int2' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 2 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1005 as typarray, 'int2in' as typinput, 'int2out' as typoutput, 'int2recv' as typreceive, 'int2send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 's' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 23 as oid, 'int4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1007 as typarray, 'int4in' as typinput, 'int4out' as typoutput, 'int4recv' as typreceive, 'int4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 25 as oid, 'text' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1009 as typarray, 'textin' as typinput, 'textout' as typoutput, 'textrecv' as typreceive, 'textsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 700 as oid, 'float4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1021 as typarray, 'float4in' as typinput, 'float4out' as typoutput, 'float4recv' as typreceive, 'float4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 701 as oid, 'float8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1022 as typarray, 'float8in' as typinput, 'float8out' as typoutput, 'float8recv' as typreceive, 'float8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1043 as oid, 'varchar' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1015 as typarray, 'varcharin' as typinput, 'varcharout' as typoutput, 'varcharrecv' as typreceive, 'varcharsend' as typsend, 'varchartypmodin' as typmodin, 'varchartypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1082 as oid, 'date' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1182 as typarray, 'date_in' as typinput, 'date_out' as typoutput, 'date_recv' as typreceive, 'date_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1114 as oid, 'timestamp' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1115 as typarray, 'timestamp_in' as typinput, 'timestamp_out' as typoutput, 'timestamp_recv' as typreceive, 'timestamp_send' as typsend, 'timestamptypmodin' as typmodin, 'timestamptypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1184 as oid, 'timestamptz' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1185 as typarray, 'timestamptz_in' as typinput, 'timestamptz_out' as typoutput, 'timestamptz_recv' as typreceive, 'timestamptz_send' as typsend, 'timestamptztypmodin' as typmodin, 'timestamptztypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1700 as oid, 'numeric' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1231 as typarray, 'numeric_in' as typinput, 'numeric_out' as typoutput, 'numeric_recv' as typreceive, 'numeric_send' as typsend, 'numerictypmodin' as typmodin, 'numerictypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'm' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 3802 as oid, 'jsonb' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 3807 as typarray, 'jsonb_in' as typinput, 'jsonb_out' as typoutput, 'jsonb_recv' as typreceive, 'jsonb_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl\n"
+                    + ")\n"
+                    + "select * from pg_type join pg_namespace on pg_type.typnamespace=pg_namespace.oid"),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (ResultSet resultSet =
+          connection
+              .createStatement()
+              .executeQuery(
+                  "select * from pg_catalog.pg_type "
+                      + "join pg_catalog.pg_namespace on pg_type.typnamespace=pg_namespace.oid")) {
+        // The result is not consistent with selecting from pg_type, but that's not relevant here.
+        // We just want to ensure that it includes the correct common table expressions.
+        assertTrue(resultSet.next());
+        assertEquals(1L, resultSet.getLong(1));
+        assertFalse(resultSet.next());
+      }
+    }
+  }
+
   private void verifySettingIsNull(Connection connection, String setting) throws SQLException {
     try (ResultSet resultSet =
         connection.createStatement().executeQuery(String.format("show %s", setting))) {
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/csharp/AbstractNpgsqlMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/csharp/AbstractNpgsqlMockServerTest.java
index 29c43973..293f41fe 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/csharp/AbstractNpgsqlMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/csharp/AbstractNpgsqlMockServerTest.java
@@ -61,7 +61,28 @@ public abstract class AbstractNpgsqlMockServerTest extends AbstractMockServerTes
           .build();
   private static final Statement SELECT_TYPES =
       Statement.of(
-          "SELECT ns.nspname, t.oid, t.typname, t.typtype, t.typnotnull, t.elemtypoid\n"
+          "with pg_namespace as (\n"
+              + "  select case schema_name when 'pg_catalog' then 11 when 'public' then 2200 else 0 end as oid,\n"
+              + "        schema_name as nspname, null as nspowner, null as nspacl\n"
+              + "  from information_schema.schemata\n"
+              + "),\n"
+              + "pg_type as (\n"
+              + "  select 16 as oid, 'bool' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 1 as typlen, true as typbyval, 'b' as typtype, 'B' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1000 as typarray, 'boolin' as typinput, 'boolout' as typoutput, 'boolrecv' as typreceive, 'boolsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'c' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 17 as oid, 'bytea' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1001 as typarray, 'byteain' as typinput, 'byteaout' as typoutput, 'bytearecv' as typreceive, 'byteasend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 20 as oid, 'int8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1016 as typarray, 'int8in' as typinput, 'int8out' as typoutput, 'int8recv' as typreceive, 'int8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 21 as oid, 'int2' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 2 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1005 as typarray, 'int2in' as typinput, 'int2out' as typoutput, 'int2recv' as typreceive, 'int2send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 's' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 23 as oid, 'int4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1007 as typarray, 'int4in' as typinput, 'int4out' as typoutput, 'int4recv' as typreceive, 'int4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 25 as oid, 'text' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1009 as typarray, 'textin' as typinput, 'textout' as typoutput, 'textrecv' as typreceive, 'textsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 700 as oid, 'float4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1021 as typarray, 'float4in' as typinput, 'float4out' as typoutput, 'float4recv' as typreceive, 'float4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 701 as oid, 'float8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1022 as typarray, 'float8in' as typinput, 'float8out' as typoutput, 'float8recv' as typreceive, 'float8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1043 as oid, 'varchar' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1015 as typarray, 'varcharin' as typinput, 'varcharout' as typoutput, 'varcharrecv' as typreceive, 'varcharsend' as typsend, 'varchartypmodin' as typmodin, 'varchartypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1082 as oid, 'date' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1182 as typarray, 'date_in' as typinput, 'date_out' as typoutput, 'date_recv' as typreceive, 'date_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1114 as oid, 'timestamp' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1115 as typarray, 'timestamp_in' as typinput, 'timestamp_out' as typoutput, 'timestamp_recv' as typreceive, 'timestamp_send' as typsend, 'timestamptypmodin' as typmodin, 'timestamptypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1184 as oid, 'timestamptz' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1185 as typarray, 'timestamptz_in' as typinput, 'timestamptz_out' as typoutput, 'timestamptz_recv' as typreceive, 'timestamptz_send' as typsend, 'timestamptztypmodin' as typmodin, 'timestamptztypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1700 as oid, 'numeric' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1231 as typarray, 'numeric_in' as typinput, 'numeric_out' as typoutput, 'numeric_recv' as typreceive, 'numeric_send' as typsend, 'numerictypmodin' as typmodin, 'numerictypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'm' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 3802 as oid, 'jsonb' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 3807 as typarray, 'jsonb_in' as typinput, 'jsonb_out' as typoutput, 'jsonb_recv' as typreceive, 'jsonb_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl\n"
+              + ")\n"
+              + "SELECT ns.nspname, t.oid, t.typname, t.typtype, t.typnotnull, t.elemtypoid\n"
               + "FROM (\n"
               + "    -- Arrays have typtype=b - this subquery identifies them by their typreceive and converts their typtype to a\n"
               + "    -- We first do this for the type (innerest-most subquery), and then for its element type\n"
@@ -274,7 +295,28 @@ public abstract class AbstractNpgsqlMockServerTest extends AbstractMockServerTes
           .build();
   private static final Statement SELECT_ATTRIBUTES =
       Statement.of(
-          "-- Load field definitions for (free-standing) composite types\n"
+          "with pg_namespace as (\n"
+              + "  select case schema_name when 'pg_catalog' then 11 when 'public' then 2200 else 0 end as oid,\n"
+              + "        schema_name as nspname, null as nspowner, null as nspacl\n"
+              + "  from information_schema.schemata\n"
+              + "),\n"
+              + "pg_type as (\n"
+              + "  select 16 as oid, 'bool' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 1 as typlen, true as typbyval, 'b' as typtype, 'B' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1000 as typarray, 'boolin' as typinput, 'boolout' as typoutput, 'boolrecv' as typreceive, 'boolsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'c' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 17 as oid, 'bytea' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1001 as typarray, 'byteain' as typinput, 'byteaout' as typoutput, 'bytearecv' as typreceive, 'byteasend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 20 as oid, 'int8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1016 as typarray, 'int8in' as typinput, 'int8out' as typoutput, 'int8recv' as typreceive, 'int8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 21 as oid, 'int2' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 2 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1005 as typarray, 'int2in' as typinput, 'int2out' as typoutput, 'int2recv' as typreceive, 'int2send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 's' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 23 as oid, 'int4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1007 as typarray, 'int4in' as typinput, 'int4out' as typoutput, 'int4recv' as typreceive, 'int4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 25 as oid, 'text' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1009 as typarray, 'textin' as typinput, 'textout' as typoutput, 'textrecv' as typreceive, 'textsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 700 as oid, 'float4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1021 as typarray, 'float4in' as typinput, 'float4out' as typoutput, 'float4recv' as typreceive, 'float4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 701 as oid, 'float8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1022 as typarray, 'float8in' as typinput, 'float8out' as typoutput, 'float8recv' as typreceive, 'float8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1043 as oid, 'varchar' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1015 as typarray, 'varcharin' as typinput, 'varcharout' as typoutput, 'varcharrecv' as typreceive, 'varcharsend' as typsend, 'varchartypmodin' as typmodin, 'varchartypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1082 as oid, 'date' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1182 as typarray, 'date_in' as typinput, 'date_out' as typoutput, 'date_recv' as typreceive, 'date_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1114 as oid, 'timestamp' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1115 as typarray, 'timestamp_in' as typinput, 'timestamp_out' as typoutput, 'timestamp_recv' as typreceive, 'timestamp_send' as typsend, 'timestamptypmodin' as typmodin, 'timestamptypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1184 as oid, 'timestamptz' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1185 as typarray, 'timestamptz_in' as typinput, 'timestamptz_out' as typoutput, 'timestamptz_recv' as typreceive, 'timestamptz_send' as typsend, 'timestamptztypmodin' as typmodin, 'timestamptztypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 1700 as oid, 'numeric' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1231 as typarray, 'numeric_in' as typinput, 'numeric_out' as typoutput, 'numeric_recv' as typreceive, 'numeric_send' as typsend, 'numerictypmodin' as typmodin, 'numerictypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'm' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+              + "  select 3802 as oid, 'jsonb' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 3807 as typarray, 'jsonb_in' as typinput, 'jsonb_out' as typoutput, 'jsonb_recv' as typreceive, 'jsonb_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl\n"
+              + ")\n"
+              + "-- Load field definitions for (free-standing) composite types\n"
               + "SELECT typ.oid, att.attname, att.atttypid\n"
               + "FROM pg_type AS typ\n"
               + "JOIN pg_namespace AS ns ON (ns.oid = typ.typnamespace)\n"
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java
index 71e57e71..359c308b 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/metadata/OptionsMetadataTest.java
@@ -231,4 +231,14 @@ public class OptionsMetadataTest {
       System.setOut(originalOut);
     }
   }
+
+  @Test
+  public void testDisablePgCatalogReplacements() {
+    OptionsMetadata options = new OptionsMetadata(new String[] {"-p p", "-i i"});
+    assertTrue(options.replacePgCatalogTables());
+
+    options =
+        new OptionsMetadata(new String[] {"-p p", "-i i", "-disable_pg_catalog_replacements"});
+    assertFalse(options.replacePgCatalogTables());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java
index f5647bea..b702be9f 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/session/SessionStateTest.java
@@ -20,12 +20,10 @@ import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertThrows;
 import static org.mockito.Mockito.mock;
 
-import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.SpannerException;
 import com.google.cloud.spanner.Statement;
-import com.google.cloud.spanner.connection.AbstractStatementParser;
-import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata;
+import com.google.cloud.spanner.pgadapter.statements.PgCatalog;
 import java.util.List;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -450,33 +448,31 @@ public class SessionStateTest {
   @Test
   public void testAddSessionState() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
+    PgCatalog pgCatalog = new PgCatalog(state);
     Statement statement = Statement.of("select * from pg_settings");
-    ParsedStatement parsedStatement =
-        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
 
-    Statement withSessionState = state.addSessionState(parsedStatement, statement);
+    Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
 
     assertEquals(
-        "with " + getDefaultSessionStateExpression() + " " + statement.getSql(),
+        "with " + getDefaultSessionStateExpression() + "\n" + statement.getSql(),
         withSessionState.getSql());
   }
 
   @Test
   public void testAddSessionStateWithParameters() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
+    PgCatalog pgCatalog = new PgCatalog(state);
     Statement statement =
         Statement.newBuilder("select * from pg_settings where name=$1")
             .bind("p1")
             .to("some-name")
             .build();
-    ParsedStatement parsedStatement =
-        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
 
-    Statement withSessionState = state.addSessionState(parsedStatement, statement);
+    Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
 
     assertEquals(
         Statement.newBuilder(
-                "with " + getDefaultSessionStateExpression() + " " + statement.getSql())
+                "with " + getDefaultSessionStateExpression() + "\n" + statement.getSql())
             .bind("p1")
             .to("some-name")
             .build(),
@@ -486,11 +482,10 @@ public class SessionStateTest {
   @Test
   public void testAddSessionStateWithoutPgSettings() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
+    PgCatalog pgCatalog = new PgCatalog(state);
     Statement statement = Statement.of("select * from some_table");
-    ParsedStatement parsedStatement =
-        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
 
-    Statement withSessionState = state.addSessionState(parsedStatement, statement);
+    Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
 
     assertSame(statement, withSessionState);
   }
@@ -498,36 +493,47 @@ public class SessionStateTest {
   @Test
   public void testAddSessionStateWithComments() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
+    PgCatalog pgCatalog = new PgCatalog(state);
+    Statement statement = Statement.of("/* This comment is preserved */ select * from pg_settings");
+
+    Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
+
+    assertEquals(
+        "with " + getDefaultSessionStateExpression() + "\n" + statement.getSql(),
+        withSessionState.getSql());
+  }
+
+  @Test
+  public void testAddSessionStateWithExistingCTE() {
+    SessionState state = new SessionState(mock(OptionsMetadata.class));
+    PgCatalog pgCatalog = new PgCatalog(state);
     Statement statement =
-        Statement.of("/* This comment is not preserved */ select * from pg_settings");
-    ParsedStatement parsedStatement =
-        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+        Statement.of(
+            "with my_cte as (select col1, col2 from foo) select * from pg_settings inner join my_cte on my_cte.col1=pg_settings.name");
 
-    Statement withSessionState = state.addSessionState(parsedStatement, statement);
+    Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
 
     assertEquals(
         "with "
             + getDefaultSessionStateExpression()
-            + " "
-            + parsedStatement.getSqlWithoutComments(),
+            + ",\n my_cte as (select col1, col2 from foo) select * from pg_settings inner join my_cte on my_cte.col1=pg_settings.name",
         withSessionState.getSql());
   }
 
   @Test
-  public void testAddSessionStateWithExistingCte() {
+  public void testAddSessionStateWithCommentsAndExistingCTE() {
     SessionState state = new SessionState(mock(OptionsMetadata.class));
+    PgCatalog pgCatalog = new PgCatalog(state);
     Statement statement =
         Statement.of(
-            "with my_cte as (select col1, col2 from foo) select * from pg_settings inner join my_cte on my_cte.col1=pg_settings.name");
-    ParsedStatement parsedStatement =
-        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+            "/* This comment is preserved */ with foo as (select * from bar)\nselect * from pg_settings");
 
-    Statement withSessionState = state.addSessionState(parsedStatement, statement);
+    Statement withSessionState = pgCatalog.replacePgCatalogTables(statement);
 
     assertEquals(
         "with "
             + getDefaultSessionStateExpression()
-            + ",  my_cte as (select col1, col2 from foo) select * from pg_settings inner join my_cte on my_cte.col1=pg_settings.name",
+            + ",\n/* This comment is preserved */  foo as (select * from bar)\nselect * from pg_settings",
         withSessionState.getSql());
   }
 }
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 f48b67d1..041f7c40 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
@@ -392,4 +392,66 @@ public class BackendConnectionTest {
         assertThrows(ExecutionException.class, resultFuture1::get);
     assertSame(executionException.getCause(), error);
   }
+
+  @Test
+  public void testReplacePgCatalogTables() {
+    Connection connection = mock(Connection.class);
+    OptionsMetadata options = mock(OptionsMetadata.class);
+    when(options.replacePgCatalogTables()).thenReturn(true);
+    Statement statement = Statement.of("select * from pg_catalog.pg_type");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+
+    BackendConnection backendConnection =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, options, EMPTY_LOCAL_STATEMENTS);
+
+    backendConnection.execute(parsedStatement, statement);
+    backendConnection.flush();
+
+    verify(connection)
+        .execute(
+            Statement.of(
+                "with pg_namespace as (\n"
+                    + "  select case schema_name when 'pg_catalog' then 11 when 'public' then 2200 else 0 end as oid,\n"
+                    + "        schema_name as nspname, null as nspowner, null as nspacl\n"
+                    + "  from information_schema.schemata\n"
+                    + "),\n"
+                    + "pg_type as (\n"
+                    + "  select 16 as oid, 'bool' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 1 as typlen, true as typbyval, 'b' as typtype, 'B' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1000 as typarray, 'boolin' as typinput, 'boolout' as typoutput, 'boolrecv' as typreceive, 'boolsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'c' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 17 as oid, 'bytea' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1001 as typarray, 'byteain' as typinput, 'byteaout' as typoutput, 'bytearecv' as typreceive, 'byteasend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 20 as oid, 'int8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1016 as typarray, 'int8in' as typinput, 'int8out' as typoutput, 'int8recv' as typreceive, 'int8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 21 as oid, 'int2' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 2 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1005 as typarray, 'int2in' as typinput, 'int2out' as typoutput, 'int2recv' as typreceive, 'int2send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 's' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 23 as oid, 'int4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1007 as typarray, 'int4in' as typinput, 'int4out' as typoutput, 'int4recv' as typreceive, 'int4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 25 as oid, 'text' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1009 as typarray, 'textin' as typinput, 'textout' as typoutput, 'textrecv' as typreceive, 'textsend' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 700 as oid, 'float4' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1021 as typarray, 'float4in' as typinput, 'float4out' as typoutput, 'float4recv' as typreceive, 'float4send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 701 as oid, 'float8' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'N' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1022 as typarray, 'float8in' as typinput, 'float8out' as typoutput, 'float8recv' as typreceive, 'float8send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1043 as oid, 'varchar' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'S' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1015 as typarray, 'varcharin' as typinput, 'varcharout' as typoutput, 'varcharrecv' as typreceive, 'varcharsend' as typsend, 'varchartypmodin' as typmodin, 'varchartypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 100 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1082 as oid, 'date' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 4 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1182 as typarray, 'date_in' as typinput, 'date_out' as typoutput, 'date_recv' as typreceive, 'date_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1114 as oid, 'timestamp' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, false as typispreferred, false as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1115 as typarray, 'timestamp_in' as typinput, 'timestamp_out' as typoutput, 'timestamp_recv' as typreceive, 'timestamp_send' as typsend, 'timestamptypmodin' as typmodin, 'timestamptypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1184 as oid, 'timestamptz' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, 8 as typlen, true as typbyval, 'b' as typtype, 'D' as typcategory, true as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1185 as typarray, 'timestamptz_in' as typinput, 'timestamptz_out' as typoutput, 'timestamptz_recv' as typreceive, 'timestamptz_send' as typsend, 'timestamptztypmodin' as typmodin, 'timestamptztypmodout' as typmodout, '-' as typanalyze, 'd' as typalign, 'p' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 1700 as oid, 'numeric' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'N' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 1231 as typarray, 'numeric_in' as typinput, 'numeric_out' as typoutput, 'numeric_recv' as typreceive, 'numeric_send' as typsend, 'numerictypmodin' as typmodin, 'numerictypmodout' as typmodout, '-' as typanalyze, 'i' as typalign, 'm' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl union all\n"
+                    + "  select 3802 as oid, 'jsonb' as typname, (select oid from pg_namespace where nspname='pg_catalog') as typnamespace, null as typowner, -1 as typlen, false as typbyval, 'b' as typtype, 'U' as typcategory, false as typispreferred, true as typisdefined, ',' as typdelim, 0 as typrelid, 0 as typelem, 3807 as typarray, 'jsonb_in' as typinput, 'jsonb_out' as typoutput, 'jsonb_recv' as typreceive, 'jsonb_send' as typsend, '-' as typmodin, '-' as typmodout, '-' as typanalyze, 'i' as typalign, 'x' as typstorage, false as typnotnull, 0 as typbasetype, -1 as typtypmod, 0 as typndims, 0 as typcollation, null as typdefaultbin, null as typdefault, null as typacl\n"
+                    + ")\n"
+                    + "select * from pg_type"));
+  }
+
+  @Test
+  public void testDisableReplacePgCatalogTables() {
+    Connection connection = mock(Connection.class);
+    OptionsMetadata options = mock(OptionsMetadata.class);
+    when(options.replacePgCatalogTables()).thenReturn(false);
+    Statement statement = Statement.of("select * from pg_catalog.pg_type");
+    ParsedStatement parsedStatement =
+        AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
+
+    BackendConnection backendConnection =
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, options, EMPTY_LOCAL_STATEMENTS);
+
+    backendConnection.execute(parsedStatement, statement);
+    backendConnection.flush();
+
+    verify(connection).execute(statement);
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/PgCatalogTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/PgCatalogTest.java
new file mode 100644
index 00000000..3f19ebbd
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/PgCatalogTest.java
@@ -0,0 +1,25 @@
+// 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 org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class PgCatalogTest {
+  @Test
+  public void testReplaceCatalogTables() {}
+}
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 9616db5a..e95560d9 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
@@ -17,6 +17,7 @@ package com.google.cloud.spanner.pgadapter.statements;
 import static com.google.cloud.spanner.pgadapter.statements.SimpleParser.parseCommand;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -64,6 +65,12 @@ public class SimpleParserTest {
     return parser.getPos();
   }
 
+  @Test
+  public void testTableOrIndexName() {
+    assertEquals(new TableOrIndexName(null, "foo"), new TableOrIndexName(null, "foo"));
+    assertNotEquals(new TableOrIndexName(null, "foo"), new TableOrIndexName(null, "bar"));
+  }
+
   @Test
   public void testEatKeyword() {
     assertTrue(new SimpleParser("insert").eatKeyword("insert"));
@@ -387,6 +394,9 @@ public class SimpleParserTest {
     assertEquals("WITH", parseCommand("with 1 as (select * from foo update bar set col1='one'"));
     assertEquals(
         "WITH", parseCommand("with my_cte as select * from foo update bar set col1='one'"));
+
+    assertEquals("WITH", parseCommand("with my_cte as (select * from foo)"));
+    assertEquals("", parseCommand("/*only a comment*/"));
     assertEquals("WITH", parseCommand("with my_cte () as (select * from foo)"));
     assertEquals("WITH", parseCommand("with my_cte (,) as (select * from foo)"));
     assertEquals("WITH", parseCommand("with my_cte (bar"));
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/TableParserTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/TableParserTest.java
new file mode 100644
index 00000000..97ec9fb6
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/TableParserTest.java
@@ -0,0 +1,866 @@
+// 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.assertSame;
+
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableMap.Builder;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class TableParserTest {
+
+  @Test
+  public void testReplaceTables() {
+    Statement original = Statement.of("select * from foo");
+    assertSame(original, new TableParser(original).detectAndReplaceTables(ImmutableMap.of()).y());
+    assertSame(
+        original,
+        new TableParser(original)
+            .detectAndReplaceTables(
+                ImmutableMap.of(
+                    new TableOrIndexName(null, "bar"), new TableOrIndexName(null, "foo")))
+            .y());
+
+    assertEquals(Statement.of("select * from bar"), replace(original, "foo", "bar"));
+    assertEquals(Statement.of("select * from bar"), replace(original, "FOO", "bar"));
+    assertEquals(
+        Statement.of("select * from other_table"), replace(original, "Foo", "other_table"));
+    assertEquals(
+        Statement.of("select * from my_schema.my_table"),
+        replace(original, null, "foo", "my_schema", "my_table"));
+
+    assertEquals(
+        Statement.of("select * from bar"),
+        replace(Statement.of("select * from \"foo\""), "foo", "bar"));
+    assertEquals(
+        Statement.of("select * from \"Foo\""),
+        replace(Statement.of("select * from \"Foo\""), "foo", "bar"));
+    assertEquals(
+        Statement.of("select * from my_schema.bar"),
+        replace(Statement.of("select * from public.foo"), "public", "foo", "my_schema", "bar"));
+    assertEquals(
+        Statement.of("select * from my_schema.bar"),
+        replace(Statement.of("select * from public.\"foo\""), "public", "foo", "my_schema", "bar"));
+    assertEquals(
+        Statement.of("select * from my_schema.bar"),
+        replace(Statement.of("select * from \"public\".foo"), "public", "foo", "my_schema", "bar"));
+    assertEquals(
+        Statement.of("select * from my_schema.bar"),
+        replace(
+            Statement.of("select * from \"public\".\"foo\""), "public", "foo", "my_schema", "bar"));
+    assertEquals(
+        Statement.of("select * from \"Public\".\"foo\""),
+        replace(
+            Statement.of("select * from \"Public\".\"foo\""), "public", "foo", "my_schema", "bar"));
+    assertEquals(
+        Statement.of("select * from \"public\".\"Foo\""),
+        replace(
+            Statement.of("select * from \"public\".\"Foo\""), "public", "foo", "my_schema", "bar"));
+
+    assertEquals(
+        Statement.of("select * from pg_type"),
+        replace(
+            Statement.of("select * from pg_catalog.pg_type"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("SELECT * FROM pg_type"),
+        replace(
+            Statement.of("SELECT * FROM PG_CATALOG.PG_TYPE"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("SELECT * FROM pg_type"),
+        replace(
+            Statement.of("SELECT * FROM \"pg_catalog\".\"pg_type\""),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("SELECT * FROM pg_catalog.\"Pg_Type\""),
+        replace(
+            Statement.of("SELECT * FROM pg_catalog.\"Pg_Type\""),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+
+    assertEquals(
+        Statement.of("select namespace, name, some_function(name) as foo from pg_type"),
+        replace(
+            Statement.of(
+                "select namespace, name, some_function(name) as foo from pg_catalog.pg_type"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of(
+            "select namespace, name, '(select max(oid) from pg_catalog.pg_type)' as foo from pg_type"),
+        replace(
+            Statement.of(
+                "select namespace, name, '(select max(oid) from pg_catalog.pg_type)' as foo from pg_catalog.pg_type"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("select namespace, name, \"from pg_catalog.pg_type\" as foo from pg_type"),
+        replace(
+            Statement.of(
+                "select namespace, name, \"from pg_catalog.pg_type\" as foo from pg_catalog.pg_type"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+
+    assertEquals(
+        Statement.of("insert into pg_type (name) values ('foo')"),
+        replace(
+            Statement.of("insert into pg_catalog.pg_type (name) values ('foo')"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("insert into pg_type(name)values('foo')"),
+        replace(
+            Statement.of("insert into pg_catalog.pg_type(name)values('foo')"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("insert   pg_type (name) values ('foo')"),
+        replace(
+            Statement.of("insert   pg_catalog.pg_type (name) values ('foo')"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+
+    assertEquals(
+        Statement.of("update pg_type set name='foo' where value='pg_type'"),
+        replace(
+            Statement.of("update pg_catalog.pg_type set name='foo' where value='pg_type'"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("update pg_type set name='foo' where value=(select max(value) from pg_type)"),
+        replace(
+            Statement.of(
+                "update pg_catalog.pg_type set name='foo' where value=(select max(value) from pg_catalog.pg_type)"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("update pg_type set name=(select max(name) from pg_type)"),
+        replace(
+            Statement.of(
+                "update pg_catalog.pg_type set name=(select max(name) from \"pg_catalog\".\"pg_type\")"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+
+    assertEquals(
+        Statement.of("delete from pg_type"),
+        replace(
+            Statement.of("delete from pg_catalog.pg_type"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of("delete pg_type"),
+        replace(
+            Statement.of("delete pg_catalog.pg_type"), "pg_catalog", "pg_type", null, "pg_type"));
+
+    assertEquals(
+        Statement.of("select * from pg_type t1 left outer join pg_type t2 using (name)"),
+        replace(
+            Statement.of(
+                "select * from pg_catalog.pg_type t1 left outer join pg_catalog.pg_type t2 using (name)"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of(
+            "select * from pg_type t1 left outer join (select name from foo) t2 using (name)"),
+        replace(
+            Statement.of(
+                "select * from pg_catalog.pg_type t1 left outer join (select name from foo) t2 using (name)"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of(
+            "select * from pg_type t1 left outer join (select min(name) from pg_type) t2 using (name)"),
+        replace(
+            Statement.of(
+                "select * from pg_catalog.pg_type t1 left outer join (select min(name) from pg_catalog.pg_type) t2 using (name)"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of(
+            "select * from pg_type t1 left outer join (select min(name) from foo as \"pg_catalog.pg_type\") t2 using (name)"),
+        replace(
+            Statement.of(
+                "select * from pg_catalog.pg_type t1 left outer join (select min(name) from foo as \"pg_catalog.pg_type\") t2 using (name)"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of(
+            "select * from pg_type t1 left outer join (select min(name) from foo as \"pg_catalog.pg_type\") t2 using (name) where t1.name in (select name from pg_type)"),
+        replace(
+            Statement.of(
+                "select * from pg_catalog.pg_type t1 left outer join (select min(name) from foo as \"pg_catalog.pg_type\") t2 using (name) where t1.name in (select name from pg_catalog.pg_type)"),
+            "pg_catalog",
+            "pg_type",
+            null,
+            "pg_type"));
+    assertEquals(
+        Statement.of(
+            "SELECT is(\n"
+                + "    ARRAY(\n"
+                + "        SELECT stuff_id\n"
+                + "          FROM replaced_domain_stuff\n"
+                + "         WHERE domain_id = $1\n"
+                + "           AND src_id = $2\n"
+                + "         ORDER BY stuff_id\n"
+                + "    ),\n"
+                + "    ARRAY[ 1, 2, 3 ],\n"
+                + "    'some-string-literal'\n"
+                + ");"),
+        replace(
+            Statement.of(
+                "SELECT is(\n"
+                    + "    ARRAY(\n"
+                    + "        SELECT stuff_id\n"
+                    + "          FROM replaced_domain_stuff\n"
+                    + "         WHERE domain_id = $1\n"
+                    + "           AND src_id = $2\n"
+                    + "         ORDER BY stuff_id\n"
+                    + "    ),\n"
+                    + "    ARRAY[ 1, 2, 3 ],\n"
+                    + "    'some-string-literal'\n"
+                    + ");"),
+            "domain_stuff",
+            "replaced_domain_stuff"));
+    assertEquals(
+        Statement.of(
+            "select abc, sub.key from "
+                + "(select key, 'abc' as abc from replaced) sub "
+                + "order by 1"),
+        replace(
+            Statement.of(
+                "select abc, sub.key from "
+                    + "(select key, 'abc' as abc from keyvalue) sub "
+                    + "order by 1"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "insert into \"AllSpangresTypes\"(string_value, bool_value, "
+                + "int64_value) "
+                + "values ('hello world', default, 1234567890123), (default, true, "
+                + "9876543210987)"),
+        replace(
+            Statement.of(
+                "insert into keyvalue(string_value, bool_value, "
+                    + "int64_value) "
+                    + "values ('hello world', default, 1234567890123), (default, true, "
+                    + "9876543210987)"),
+            "keyvalue",
+            "\"AllSpangresTypes\""));
+    assertEquals(
+        Statement.of(
+            "update replaced set value='v'where key=12345678912 "
+                + "returning key + 1 as newkey, value as newvalue"),
+        replace(
+            Statement.of(
+                "update keyvalue set value='v'where key=12345678912 "
+                    + "returning key + 1 as newkey, value as newvalue"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "select\n"
+                + "            subquery.value, subquery.key, subquery.f1\n"
+                + "          from (\n"
+                + "            select\n"
+                + "              key, value, 123456789012 as f1\n"
+                + "            from\n"
+                + "              replaced\n"
+                + "          ) subquery"),
+        replace(
+            Statement.of(
+                "select\n"
+                    + "            subquery.value, subquery.key, subquery.f1\n"
+                    + "          from (\n"
+                    + "            select\n"
+                    + "              key, value, 123456789012 as f1\n"
+                    + "            from\n"
+                    + "              keyvalue\n"
+                    + "          ) subquery"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of("select exists(select 1000000000000 as `?column?`) " + "as exists_col"),
+        replace(
+            Statement.of("select exists(select 1000000000000 as `?column?`) " + "as exists_col"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "select * from replaced where key > 5 and not(key > 10) and "
+                + "value='hello' and (key = 6 or value='world')"),
+        replace(
+            Statement.of(
+                "select * from keyvalue where key > 5 and not(key > 10) and "
+                    + "value='hello' and (key = 6 or value='world')"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "select CASE pg.map_double_to_int(double_value) WHEN "
+                + "pg.map_double_to_int(double_value) THEN "
+                + "double_value END as result from replaced"),
+        replace(
+            Statement.of(
+                "select CASE pg.map_double_to_int(double_value) WHEN "
+                    + "pg.map_double_to_int(double_value) THEN "
+                    + "double_value END as result from AllSpangresTypes"),
+            "allspangrestypes",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "select 1 as one from replaced join /*@ join_hint = 1 */ replaced kv2 on true"),
+        replace(
+            Statement.of(
+                "select 1 as one from keyvalue join /*@ join_hint = 1 "
+                    + "*/ keyvalue kv2 on true"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "with t1 as (select key kk, value vv from replaced), "
+                + "t2 as (select kk xx, vv from t1) "
+                + "select xx, vv, '--' as col, * from t2"),
+        replace(
+            Statement.of(
+                "with t1 as (select key kk, value vv from keyvalue), "
+                    + "t2 as (select kk xx, vv from t1) "
+                    + "select xx, vv, '--' as col, * from t2"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "with t as (select key from replaced), "
+                + "t2 as (select 123 as col from replaced) "
+                + "select value from replaced"),
+        replace(
+            Statement.of(
+                "with t as (select key from keyvalue), "
+                    + "t2 as (select 123 as col from keyvalue) "
+                    + "select value from keyvalue"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of("SELECT key FROM replaced FULL JOIN \"KeyValue2\" USING (key);"),
+        replace(
+            Statement.of("SELECT key FROM keyvalue FULL JOIN \"KeyValue2\" USING (key);"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "SELECT keyvalue.key FROM (replaced LEFT JOIN \"KeyValue2\" ON ((keyvalue.key = \"KeyValue2\".key)))"),
+        replace(
+            Statement.of(
+                "SELECT keyvalue.key FROM (keyvalue LEFT JOIN \"KeyValue2\" ON ((keyvalue.key = \"KeyValue2\".key)))"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "SELECT keyvalue.key FROM (replaced LEFT JOIN \"KeyValue2\" ON ((keyvalue.key = \"KeyValue2\".key))), replaced as kv2 join replaced kv3 using (id)"),
+        replace(
+            Statement.of(
+                "SELECT keyvalue.key FROM (keyvalue LEFT JOIN \"KeyValue2\" ON ((keyvalue.key = \"KeyValue2\".key))), keyvalue as kv2 join KeyValue kv3 using (id)"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "select key from replaced\n"
+                + "inner join\n"
+                + "(select key as k1, key as k2 from \"KeyValue2\") as subquery1\n"
+                + "on subquery1.k1 = key;\n"
+                + "select key from replaced\n"
+                + "inner join\n"
+                + "(select key as k1, key as k2 from KeyValue2) as subquery1\n"
+                + "on subquery1.k1 = key;"),
+        replace(
+            Statement.of(
+                "select key from keyvalue\n"
+                    + "inner join\n"
+                    + "(select key as k1, key as k2 from \"KeyValue2\") as subquery1\n"
+                    + "on subquery1.k1 = key;\n"
+                    + "select key from keyvalue\n"
+                    + "inner join\n"
+                    + "(select key as k1, key as k2 from KeyValue2) as subquery1\n"
+                    + "on subquery1.k1 = key;"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "select * from (select 1 as col1) s1 join replaced on true join withpseudo on true"),
+        replace(
+            Statement.of(
+                "select * from (select 1 as col1) s1 join keyvalue on true join withpseudo on true"),
+            "keyvalue",
+            "replaced"));
+  }
+
+  @Test
+  public void testReplacePgCatalogTables() {
+    assertEquals(
+        Statement.of(
+            "SELECT n.nspname as \"Schema\",\n"
+                + "          c.relname as \"Name\",\n"
+                + "          CASE c.relkind WHEN 'r' THEN 'table' WHEN 'v' THEN 'view' WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index' WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special' WHEN 'f' THEN 'foreign table' WHEN 'p' THEN 'partitioned table' WHEN 'I' THEN 'partitioned index' END as \"Type\",\n"
+                + "          pg_catalog.pg_get_userbyid(c.relowner) as \"Owner\"\n"
+                + "        FROM pg_class c\n"
+                + "             LEFT JOIN pg_namespace n ON n.oid = c.relnamespace\n"
+                + "        WHERE c.relkind IN ('r','p','v','m','S','f','')\n"
+                + "              AND n.nspname <> 'pg_catalog'\n"
+                + "              AND n.nspname <> 'information_schema'\n"
+                + "              AND n.nspname !~ '^pg_toast'\n"
+                + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                + "        ORDER BY 1,2;"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT n.nspname as \"Schema\",\n"
+                    + "          c.relname as \"Name\",\n"
+                    + "          CASE c.relkind WHEN 'r' THEN 'table' WHEN 'v' THEN 'view' WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index' WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special' WHEN 'f' THEN 'foreign table' WHEN 'p' THEN 'partitioned table' WHEN 'I' THEN 'partitioned index' END as \"Type\",\n"
+                    + "          pg_catalog.pg_get_userbyid(c.relowner) as \"Owner\"\n"
+                    + "        FROM pg_catalog.pg_class c\n"
+                    + "             LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace\n"
+                    + "        WHERE c.relkind IN ('r','p','v','m','S','f','')\n"
+                    + "              AND n.nspname <> 'pg_catalog'\n"
+                    + "              AND n.nspname <> 'information_schema'\n"
+                    + "              AND n.nspname !~ '^pg_toast'\n"
+                    + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                    + "        ORDER BY 1,2;"),
+            "pg_class",
+            "pg_namespace"));
+    assertEquals(
+        Statement.of(
+            "SELECT c.oid,\n"
+                + "          n.nspname,\n"
+                + "          c.relname\n"
+                + "        FROM pg_class c\n"
+                + "             LEFT JOIN pg_namespace n ON n.oid = c.relnamespace\n"
+                + "        WHERE c.relname OPERATOR(pg_catalog.~) '^(test)$' COLLATE pg_catalog.default\n"
+                + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                + "        ORDER BY 2, 3;\n"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT c.oid,\n"
+                    + "          n.nspname,\n"
+                    + "          c.relname\n"
+                    + "        FROM pg_catalog.pg_class c\n"
+                    + "             LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace\n"
+                    + "        WHERE c.relname OPERATOR(pg_catalog.~) '^(test)$' COLLATE pg_catalog.default\n"
+                    + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                    + "        ORDER BY 2, 3;\n"),
+            "pg_class",
+            "pg_namespace"));
+    assertEquals(
+        Statement.of(
+            "SELECT n.nspname as \"Schema\",\n"
+                + "          c.relname as \"Name\",\n"
+                + "          CASE c.relkind WHEN 'r' THEN 'table' WHEN 'v' THEN 'view' WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index' WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special' WHEN 'f' THEN 'foreign table' WHEN 'p' THEN 'partitioned table' WHEN 'I' THEN 'partitioned index' END as \"Type\",\n"
+                + "          pg_catalog.pg_get_userbyid(c.relowner) as \"Owner\"\n"
+                + "        FROM pg_class c\n"
+                + "             LEFT JOIN pg_namespace n ON n.oid = c.relnamespace\n"
+                + "        WHERE c.relkind IN ('r','p','s','')\n"
+                + "              AND n.nspname !~ '^pg_toast'\n"
+                + "          AND c.relname OPERATOR(pg_catalog.~) '^(test)$' COLLATE pg_catalog.default\n"
+                + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                + "        ORDER BY 1,2;"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT n.nspname as \"Schema\",\n"
+                    + "          c.relname as \"Name\",\n"
+                    + "          CASE c.relkind WHEN 'r' THEN 'table' WHEN 'v' THEN 'view' WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index' WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special' WHEN 'f' THEN 'foreign table' WHEN 'p' THEN 'partitioned table' WHEN 'I' THEN 'partitioned index' END as \"Type\",\n"
+                    + "          pg_catalog.pg_get_userbyid(c.relowner) as \"Owner\"\n"
+                    + "        FROM pg_catalog.pg_class c\n"
+                    + "             LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace\n"
+                    + "        WHERE c.relkind IN ('r','p','s','')\n"
+                    + "              AND n.nspname !~ '^pg_toast'\n"
+                    + "          AND c.relname OPERATOR(pg_catalog.~) '^(test)$' COLLATE pg_catalog.default\n"
+                    + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                    + "        ORDER BY 1,2;"),
+            "pg_class",
+            "pg_namespace"));
+    assertEquals(
+        Statement.of(
+            "SELECT n.nspname as \"Schema\",\n"
+                + "          c.relname as \"Name\",\n"
+                + "          CASE c.relkind WHEN 'r' THEN 'table' WHEN 'v' THEN 'view' WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index' WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special' WHEN 'f' THEN 'foreign table' WHEN 'p' THEN 'partitioned table' WHEN 'I' THEN 'partitioned index' END as \"Type\",\n"
+                + "          pg_catalog.pg_get_userbyid(c.relowner) as \"Owner\",\n"
+                + "          c2.relname as \"Table\"\n"
+                + "        FROM pg_class c\n"
+                + "             LEFT JOIN pg_namespace n ON n.oid = c.relnamespace\n"
+                + "             LEFT JOIN pg_index i ON i.indexrelid = c.oid\n"
+                + "             LEFT JOIN pg_class c2 ON i.indrelid = c2.oid\n"
+                + "        WHERE c.relkind IN ('i','I','s','')\n"
+                + "              AND n.nspname !~ '^pg_toast'\n"
+                + "          AND c.relname OPERATOR(pg_catalog.~) '^(test)$' COLLATE pg_catalog.default\n"
+                + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                + "        ORDER BY 1,2;"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT n.nspname as \"Schema\",\n"
+                    + "          c.relname as \"Name\",\n"
+                    + "          CASE c.relkind WHEN 'r' THEN 'table' WHEN 'v' THEN 'view' WHEN 'm' THEN 'materialized view' WHEN 'i' THEN 'index' WHEN 'S' THEN 'sequence' WHEN 's' THEN 'special' WHEN 'f' THEN 'foreign table' WHEN 'p' THEN 'partitioned table' WHEN 'I' THEN 'partitioned index' END as \"Type\",\n"
+                    + "          pg_catalog.pg_get_userbyid(c.relowner) as \"Owner\",\n"
+                    + "          c2.relname as \"Table\"\n"
+                    + "        FROM pg_catalog.pg_class c\n"
+                    + "             LEFT JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace\n"
+                    + "             LEFT JOIN pg_catalog.pg_index i ON i.indexrelid = c.oid\n"
+                    + "             LEFT JOIN pg_catalog.pg_class c2 ON i.indrelid = c2.oid\n"
+                    + "        WHERE c.relkind IN ('i','I','s','')\n"
+                    + "              AND n.nspname !~ '^pg_toast'\n"
+                    + "          AND c.relname OPERATOR(pg_catalog.~) '^(test)$' COLLATE pg_catalog.default\n"
+                    + "          AND pg_catalog.pg_table_is_visible(c.oid)\n"
+                    + "        ORDER BY 1,2;"),
+            "pg_class",
+            "pg_namespace",
+            "pg_index"));
+    assertEquals(
+        Statement.of(
+            "SELECT d.datname as \"Name\",\n"
+                + "               pg_catalog.pg_get_userbyid(d.datdba) as \"Owner\",\n"
+                + "               pg_catalog.pg_encoding_to_char(d.encoding) as \"Encoding\",\n"
+                + "               d.datcollate as \"Collate\",\n"
+                + "               d.datctype as \"Ctype\",\n"
+                + "               pg_catalog.array_to_string(d.datacl, E'\\n') AS \"Access privileges\"\n"
+                + "        FROM pg_database d\n"
+                + "        ORDER BY 1;"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT d.datname as \"Name\",\n"
+                    + "               pg_catalog.pg_get_userbyid(d.datdba) as \"Owner\",\n"
+                    + "               pg_catalog.pg_encoding_to_char(d.encoding) as \"Encoding\",\n"
+                    + "               d.datcollate as \"Collate\",\n"
+                    + "               d.datctype as \"Ctype\",\n"
+                    + "               pg_catalog.array_to_string(d.datacl, E'\\n') AS \"Access privileges\"\n"
+                    + "        FROM pg_catalog.pg_database d\n"
+                    + "        ORDER BY 1;"),
+            "pg_database"));
+    assertEquals(
+        Statement.of(
+            "SELECT pg_catalog.quote_ident(c.relname) FROM pg_class c WHERE c.relkind IN ('r', 'S', 'v', 'm', 'f', 'p') AND substring(pg_catalog.quote_ident(c.relname),1,0)='' AND pg_catalog.pg_table_is_visible(c.oid) AND c.relnamespace <> (SELECT oid FROM pg_namespace WHERE nspname = 'pg_catalog')\n"
+                + "        UNION\n"
+                + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' FROM pg_namespace n WHERE substring(pg_catalog.quote_ident(n.nspname) || '.',1,0)='' AND (SELECT pg_catalog.count(*) FROM pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) > 1\n"
+                + "        UNION\n"
+                + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname) FROM pg_class c, pg_namespace n WHERE c.relnamespace = n.oid AND c.relkind IN ('r', 'S', 'v', 'm', 'f', 'p') AND substring(pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname),1,0)='' AND substring(pg_catalog.quote_ident(n.nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(n.nspname))+1) AND (SELECT pg_catalog.count(*) FROM pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) = 1\n"
+                + "        LIMIT 1000\n"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c WHERE c.relkind IN ('r', 'S', 'v', 'm', 'f', 'p') AND substring(pg_catalog.quote_ident(c.relname),1,0)='' AND pg_catalog.pg_table_is_visible(c.oid) AND c.relnamespace <> (SELECT oid FROM pg_catalog.pg_namespace WHERE nspname = 'pg_catalog')\n"
+                    + "        UNION\n"
+                    + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' FROM pg_catalog.pg_namespace n WHERE substring(pg_catalog.quote_ident(n.nspname) || '.',1,0)='' AND (SELECT pg_catalog.count(*) FROM pg_catalog.pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) > 1\n"
+                    + "        UNION\n"
+                    + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n WHERE c.relnamespace = n.oid AND c.relkind IN ('r', 'S', 'v', 'm', 'f', 'p') AND substring(pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname),1,0)='' AND substring(pg_catalog.quote_ident(n.nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(n.nspname))+1) AND (SELECT pg_catalog.count(*) FROM pg_catalog.pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) = 1\n"
+                    + "        LIMIT 1000\n"),
+            "pg_class",
+            "pg_namespace"));
+    assertEquals(
+        Statement.of(
+            "SELECT a.attname,\n"
+                + "          pg_catalog.format_type(a.atttypid, a.atttypmod),\n"
+                + "          (SELECT pg_catalog.pg_get_expr(d.adbin, d.adrelid, true)\n"
+                + "           FROM pg_attrdef d\n"
+                + "           WHERE d.adrelid = a.attrelid AND d.adnum = a.attnum AND a.atthasdef),\n"
+                + "          a.attnotnull,\n"
+                + "          (SELECT c.collname FROM pg_collation c, pg_type t\n"
+                + "           WHERE c.oid = a.attcollation AND t.oid = a.atttypid AND a.attcollation <> t.typcollation) AS attcollation,\n"
+                + "          a.attidentity,\n"
+                + "          a.attgenerated\n"
+                + "        FROM pg_attribute a\n"
+                + "        WHERE a.attrelid = '16404' AND a.attnum > 0 AND NOT a.attisdropped\n"
+                + "        ORDER BY a.attnum;"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT a.attname,\n"
+                    + "          pg_catalog.format_type(a.atttypid, a.atttypmod),\n"
+                    + "          (SELECT pg_catalog.pg_get_expr(d.adbin, d.adrelid, true)\n"
+                    + "           FROM pg_catalog.pg_attrdef d\n"
+                    + "           WHERE d.adrelid = a.attrelid AND d.adnum = a.attnum AND a.atthasdef),\n"
+                    + "          a.attnotnull,\n"
+                    + "          (SELECT c.collname FROM pg_catalog.pg_collation c, pg_catalog.pg_type t\n"
+                    + "           WHERE c.oid = a.attcollation AND t.oid = a.atttypid AND a.attcollation <> t.typcollation) AS attcollation,\n"
+                    + "          a.attidentity,\n"
+                    + "          a.attgenerated\n"
+                    + "        FROM pg_catalog.pg_attribute a\n"
+                    + "        WHERE a.attrelid = '16404' AND a.attnum > 0 AND NOT a.attisdropped\n"
+                    + "        ORDER BY a.attnum;"),
+            "pg_attribute",
+            "pg_collation",
+            "pg_type",
+            "pg_attrdef"));
+    assertEquals(
+        Statement.of(
+            "SELECT c2.relname, i.indisprimary, i.indisunique, i.indisclustered, i.indisvalid, pg_catalog.pg_get_indexdef(i.indexrelid, 0, true),\n"
+                + "          pg_catalog.pg_get_constraintdef(con.oid, true), contype, condeferrable, condeferred, i.indisreplident, c2.reltablespace\n"
+                + "        FROM pg_class c, pg_class c2, pg_index i\n"
+                + "          LEFT JOIN pg_constraint con ON (conrelid = i.indrelid AND conindid = i.indexrelid AND contype IN ('p','u','x'))\n"
+                + "        WHERE c.oid = '16404' AND c.oid = i.indrelid AND i.indexrelid = c2.oid\n"
+                + "        ORDER BY i.indisprimary DESC, c2.relname;"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT c2.relname, i.indisprimary, i.indisunique, i.indisclustered, i.indisvalid, pg_catalog.pg_get_indexdef(i.indexrelid, 0, true),\n"
+                    + "          pg_catalog.pg_get_constraintdef(con.oid, true), contype, condeferrable, condeferred, i.indisreplident, c2.reltablespace\n"
+                    + "        FROM pg_catalog.pg_class c, pg_catalog.pg_class c2, pg_catalog.pg_index i\n"
+                    + "          LEFT JOIN pg_catalog.pg_constraint con ON (conrelid = i.indrelid AND conindid = i.indexrelid AND contype IN ('p','u','x'))\n"
+                    + "        WHERE c.oid = '16404' AND c.oid = i.indrelid AND i.indexrelid = c2.oid\n"
+                    + "        ORDER BY i.indisprimary DESC, c2.relname;"),
+            "pg_class",
+            "pg_index",
+            "pg_constraint"));
+    assertEquals(
+        Statement.of(
+            "SELECT pol.polname, pol.polpermissive,\n"
+                + "          CASE WHEN pol.polroles = '{0}' THEN NULL ELSE pg_catalog.array_to_string(array(select rolname from pg_roles where oid = any (pol.polroles) order by 1),',') END,\n"
+                + "          pg_catalog.pg_get_expr(pol.polqual, pol.polrelid),\n"
+                + "          pg_catalog.pg_get_expr(pol.polwithcheck, pol.polrelid),\n"
+                + "          CASE pol.polcmd\n"
+                + "            WHEN 'r' THEN 'SELECT'\n"
+                + "            WHEN 'a' THEN 'INSERT'\n"
+                + "            WHEN 'w' THEN 'UPDATE'\n"
+                + "            WHEN 'd' THEN 'DELETE'\n"
+                + "            END AS cmd\n"
+                + "        FROM pg_policy pol\n"
+                + "        WHERE pol.polrelid = '16404' ORDER BY 1;"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT pol.polname, pol.polpermissive,\n"
+                    + "          CASE WHEN pol.polroles = '{0}' THEN NULL ELSE pg_catalog.array_to_string(array(select rolname from pg_catalog.pg_roles where oid = any (pol.polroles) order by 1),',') END,\n"
+                    + "          pg_catalog.pg_get_expr(pol.polqual, pol.polrelid),\n"
+                    + "          pg_catalog.pg_get_expr(pol.polwithcheck, pol.polrelid),\n"
+                    + "          CASE pol.polcmd\n"
+                    + "            WHEN 'r' THEN 'SELECT'\n"
+                    + "            WHEN 'a' THEN 'INSERT'\n"
+                    + "            WHEN 'w' THEN 'UPDATE'\n"
+                    + "            WHEN 'd' THEN 'DELETE'\n"
+                    + "            END AS cmd\n"
+                    + "        FROM pg_catalog.pg_policy pol\n"
+                    + "        WHERE pol.polrelid = '16404' ORDER BY 1;"),
+            "pg_class",
+            "pg_index",
+            "pg_constraint",
+            "pg_policy",
+            "pg_roles"));
+    assertEquals(
+        Statement.of(
+            "SELECT oid, stxrelid::pg_catalog.regclass, stxnamespace::pg_catalog.regnamespace AS nsp, stxname,\n"
+                + "          (SELECT pg_catalog.string_agg(pg_catalog.quote_ident(attname),', ')\n"
+                + "           FROM pg_catalog.unnest(stxkeys) s(attnum)\n"
+                + "           JOIN pg_attribute a ON (stxrelid = a.attrelid AND\n"
+                + "                a.attnum = s.attnum AND NOT attisdropped)) AS columns,\n"
+                + "          'd' = any(stxkind) AS ndist_enabled,\n"
+                + "          'f' = any(stxkind) AS deps_enabled,\n"
+                + "          'm' = any(stxkind) AS mcv_enabled,\n"
+                + "          stxstattarget\n"
+                + "        FROM pg_statistic_ext stat\n"
+                + "        WHERE stxrelid = '16404'\n"
+                + "        ORDER BY 1"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT oid, stxrelid::pg_catalog.regclass, stxnamespace::pg_catalog.regnamespace AS nsp, stxname,\n"
+                    + "          (SELECT pg_catalog.string_agg(pg_catalog.quote_ident(attname),', ')\n"
+                    + "           FROM pg_catalog.unnest(stxkeys) s(attnum)\n"
+                    + "           JOIN pg_catalog.pg_attribute a ON (stxrelid = a.attrelid AND\n"
+                    + "                a.attnum = s.attnum AND NOT attisdropped)) AS columns,\n"
+                    + "          'd' = any(stxkind) AS ndist_enabled,\n"
+                    + "          'f' = any(stxkind) AS deps_enabled,\n"
+                    + "          'm' = any(stxkind) AS mcv_enabled,\n"
+                    + "          stxstattarget\n"
+                    + "        FROM pg_catalog.pg_statistic_ext stat\n"
+                    + "        WHERE stxrelid = '16404'\n"
+                    + "        ORDER BY 1"),
+            "pg_class",
+            "pg_index",
+            "pg_constraint",
+            "pg_policy",
+            "pg_roles",
+            "pg_statistic_ext",
+            "pg_attribute"));
+    assertEquals(
+        Statement.of(
+            "SELECT pg_catalog.quote_ident(c.relname) FROM pg_class c WHERE c.relkind IN ('r', 'p') AND substring(pg_catalog.quote_ident(c.relname),1,0)='' AND pg_catalog.pg_table_is_visible(c.oid) AND c.relnamespace <> (SELECT oid FROM pg_namespace WHERE nspname = 'pg_catalog')\n"
+                + "        UNION\n"
+                + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' FROM pg_namespace n WHERE substring(pg_catalog.quote_ident(n.nspname) || '.',1,0)='' AND (SELECT pg_catalog.count(*) FROM pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) > 1\n"
+                + "        UNION\n"
+                + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname) FROM pg_class c, pg_namespace n WHERE c.relnamespace = n.oid AND c.relkind IN ('r', 'p') AND substring(pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname),1,0)='' AND substring(pg_catalog.quote_ident(n.nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(n.nspname))+1) AND (SELECT pg_catalog.count(*) FROM pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) = 1\n"
+                + "        LIMIT 1000"),
+        removePgCatalog(
+            Statement.of(
+                "SELECT pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c WHERE c.relkind IN ('r', 'p') AND substring(pg_catalog.quote_ident(c.relname),1,0)='' AND pg_catalog.pg_table_is_visible(c.oid) AND c.relnamespace <> (SELECT oid FROM pg_catalog.pg_namespace WHERE nspname = 'pg_catalog')\n"
+                    + "        UNION\n"
+                    + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' FROM pg_catalog.pg_namespace n WHERE substring(pg_catalog.quote_ident(n.nspname) || '.',1,0)='' AND (SELECT pg_catalog.count(*) FROM pg_catalog.pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) > 1\n"
+                    + "        UNION\n"
+                    + "        SELECT pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname) FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n WHERE c.relnamespace = n.oid AND c.relkind IN ('r', 'p') AND substring(pg_catalog.quote_ident(n.nspname) || '.' || pg_catalog.quote_ident(c.relname),1,0)='' AND substring(pg_catalog.quote_ident(n.nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(n.nspname))+1) AND (SELECT pg_catalog.count(*) FROM pg_catalog.pg_namespace WHERE substring(pg_catalog.quote_ident(nspname) || '.',1,0) = substring('',1,pg_catalog.length(pg_catalog.quote_ident(nspname))+1)) = 1\n"
+                    + "        LIMIT 1000"),
+            "pg_class",
+            "pg_index",
+            "pg_constraint",
+            "pg_policy",
+            "pg_roles",
+            "pg_statistic_ext",
+            "pg_attribute",
+            "pg_namespace"));
+  }
+
+  @Test
+  public void testReplaceMultipleTables() {
+    assertEquals(
+        Statement.of("select * from foo"),
+        new TableParser(Statement.of("select * from foo"))
+            .detectAndReplaceTables(
+                ImmutableMap.of(
+                    new TableOrIndexName("pg_catalog", "pg_type"),
+                        new TableOrIndexName(null, "pg_type"),
+                    new TableOrIndexName("pg_catalog", "pg_settings"),
+                        new TableOrIndexName(null, "pg_settings"),
+                    new TableOrIndexName("pg_catalog", "pg_class"),
+                        new TableOrIndexName(null, "pg_class")))
+            .y());
+    assertEquals(
+        Statement.of("select * from pg_settings"),
+        new TableParser(Statement.of("select * from pg_catalog.pg_settings"))
+            .detectAndReplaceTables(
+                ImmutableMap.of(
+                    new TableOrIndexName("pg_catalog", "pg_type"),
+                        new TableOrIndexName(null, "pg_type"),
+                    new TableOrIndexName("pg_catalog", "pg_settings"),
+                        new TableOrIndexName(null, "pg_settings"),
+                    new TableOrIndexName("pg_catalog", "pg_class"),
+                        new TableOrIndexName(null, "pg_class")))
+            .y());
+    assertEquals(
+        Statement.of("select s.name, (select min(name) from pg_type) from pg_settings s"),
+        new TableParser(
+                Statement.of(
+                    "select s.name, (select min(name) from pg_catalog.pg_type) from pg_catalog.pg_settings s"))
+            .detectAndReplaceTables(
+                ImmutableMap.of(
+                    new TableOrIndexName("pg_catalog", "pg_type"),
+                        new TableOrIndexName(null, "pg_type"),
+                    new TableOrIndexName("pg_catalog", "pg_settings"),
+                        new TableOrIndexName(null, "pg_settings"),
+                    new TableOrIndexName("pg_catalog", "pg_class"),
+                        new TableOrIndexName(null, "pg_class")))
+            .y());
+    assertEquals(
+        Statement.of(
+            "select s.name, (select min(name) from pg_type) from pg_settings s inner join pg_class c on s.name=c.name where s.setting in (select name from pg_class)"),
+        new TableParser(
+                Statement.of(
+                    "select s.name, (select min(name) from pg_catalog.pg_type) from pg_catalog.pg_settings s inner join pg_catalog.pg_class c on s.name=c.name where s.setting in (select name from pg_catalog.pg_class)"))
+            .detectAndReplaceTables(
+                ImmutableMap.of(
+                    new TableOrIndexName("pg_catalog", "pg_type"),
+                        new TableOrIndexName(null, "pg_type"),
+                    new TableOrIndexName("pg_catalog", "pg_settings"),
+                        new TableOrIndexName(null, "pg_settings"),
+                    new TableOrIndexName("pg_catalog", "pg_class"),
+                        new TableOrIndexName(null, "pg_class")))
+            .y());
+  }
+
+  @Test
+  public void testNestedTableExpressions() {
+    assertEquals(
+        Statement.of(
+            "select kv.key, kv2.value2, kv3.value, kv4.key from\n"
+                + "  replaced kv left join \"KeyValue2\" kv2 on kv.value = kv2.value2,\n"
+                + "  (replaced kv3 join replaced kv4 on kv3.key = kv4.key\n"
+                + "    join replaced kv5 on kv4.value = kv5.value);\n"),
+        replace(
+            Statement.of(
+                "select kv.key, kv2.value2, kv3.value, kv4.key from\n"
+                    + "  keyvalue kv left join \"KeyValue2\" kv2 on kv.value = kv2.value2,\n"
+                    + "  (keyvalue kv3 join keyvalue kv4 on kv3.key = kv4.key\n"
+                    + "    join keyvalue kv5 on kv4.value = kv5.value);\n"),
+            "keyvalue",
+            "replaced"));
+    assertEquals(
+        Statement.of(
+            "SELECT kv.key, kv2.value2, kv3.value, kv4.key "
+                + "FROM (replaced kv LEFT JOIN \"KeyValue2\" kv2 ON ((kv.value = kv2.value2))), "
+                + "((replaced kv3 JOIN replaced kv4 ON ((kv3.key = kv4.key))) JOIN replaced kv5 ON ((kv4.value = kv5.value)))"),
+        replace(
+            Statement.of(
+                "SELECT kv.key, kv2.value2, kv3.value, kv4.key "
+                    + "FROM (keyvalue kv LEFT JOIN \"KeyValue2\" kv2 ON ((kv.value = kv2.value2))), "
+                    + "((keyvalue kv3 JOIN keyvalue kv4 ON ((kv3.key = kv4.key))) JOIN keyvalue kv5 ON ((kv4.value = kv5.value)))"),
+            "keyvalue",
+            "replaced"));
+  }
+
+  static Statement removePgCatalog(Statement original, String... tables) {
+    Builder<TableOrIndexName, TableOrIndexName> builder = ImmutableMap.builder();
+    for (String table : tables) {
+      builder.put(new TableOrIndexName("pg_catalog", table), new TableOrIndexName(null, table));
+    }
+    return new TableParser(original).detectAndReplaceTables(builder.build()).y();
+  }
+
+  static Statement replace(Statement original, String table, String withTable) {
+    return replace(original, null, table, null, withTable);
+  }
+
+  static Statement replace(
+      Statement original, String schema, String table, String withSchema, String withTable) {
+    return new TableParser(original)
+        .detectAndReplaceTables(
+            ImmutableMap.of(
+                new TableOrIndexName(schema, table), new TableOrIndexName(withSchema, withTable)))
+        .y();
+  }
+}