Repository: GoogleCloudPlatform/pgadapter
Pull request #224 (2022-06-22 20:45:43)

feat: support IF [NOT] EXISTS for DDL statements
Adds support for `IF [NOT] EXISTS` in DDL statements. PGAdapter will first check whether the backend supports `IF [NOT] EXISTS`, and if it does, sends the statements unmodified to the backend. Otherwise, it will parse the statements and execute the check locally, and based on that send the statement without the `IF [NOT] EXISTS` clause to the backend.

codecov[bot]: # [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> Merging [#224](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (56731bf) into [postgresql-dialect](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/commit/0c3132325687a04b482bfbedaab84aaf9484d1f1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (0c31323) will **increase** coverage by `0.09%`.
> The diff coverage is `83.55%`.

```diff
@@                   Coverage Diff                    @@
##             postgresql-dialect     #224      +/-   ##
========================================================
+ Coverage                 80.56%   80.66%   +0.09%     
- Complexity                 1051     1093      +42     
========================================================
  Files                        91       92       +1     
  Lines                      3556     3703     +147     
  Branches                    426      446      +20     
========================================================
+ Hits                       2865     2987     +122     
- Misses                      525      540      +15     
- Partials                    166      176      +10     
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `80.66% <83.55%> (+0.09%)` | :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/224?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) | Coverage Δ | |
|---|---|---|
| [...loud/spanner/pgadapter/statements/DdlExecutor.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvRGRsRXhlY3V0b3IuamF2YQ==) | `81.57% <81.57%> (ø)` | |
| [...panner/pgadapter/statements/BackendConnection.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvQmFja2VuZENvbm5lY3Rpb24uamF2YQ==) | `91.66% <87.87%> (-0.76%)` | :arrow_down: |
| [...gle/cloud/spanner/pgadapter/ConnectionHandler.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL0Nvbm5lY3Rpb25IYW5kbGVyLmphdmE=) | `79.92% <100.00%> (+0.23%)` | :arrow_up: |
| [...apter/statements/ExtendedQueryProtocolHandler.java](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvRXh0ZW5kZWRRdWVyeVByb3RvY29sSGFuZGxlci5qYXZh) | `100.00% <100.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform). Last update [0c31323...56731bf](https://codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/224?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform).


Patch:
diff --git a/pom.xml b/pom.xml
index 8a580a7d..4238e018 100644
--- a/pom.xml
+++ b/pom.xml
@@ -411,6 +411,8 @@
         </executions>
         <configuration>
           <forkedProcessTimeoutInSeconds>3600</forkedProcessTimeoutInSeconds>
+          <forkCount>8</forkCount>
+          <reuseForks>true</reuseForks>
           <argLine>${failsafe.jacoco.args}</argLine>
           <skipITs>${skipITs}</skipITs>
         </configuration>
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
index 7eaddfa9..b752a0d5 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/BackendConnection.java
@@ -17,6 +17,7 @@ package com.google.cloud.spanner.pgadapter.statements;
 import static com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage.COPY;
 
 import com.google.api.core.InternalApi;
+import com.google.cloud.spanner.DatabaseId;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.SpannerBatchUpdateException;
@@ -29,6 +30,7 @@ import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.connection.StatementResult.ClientSideStatementType;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode;
+import com.google.cloud.spanner.pgadapter.statements.DdlExecutor.NotExecuted;
 import com.google.cloud.spanner.pgadapter.statements.local.LocalStatement;
 import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
 import com.google.cloud.spanner.pgadapter.utils.MutationWriter;
@@ -45,7 +47,10 @@ import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.google.common.util.concurrent.SettableFuture;
 import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
@@ -156,6 +161,8 @@ public class BackendConnection {
           result.set(NO_RESULT);
         } else if (statement.getSql().isEmpty()) {
           result.set(NO_RESULT);
+        } else if (parsedStatement.isDdl()) {
+          result.set(ddlExecutor.execute(parsedStatement, statement));
         } else {
           result.set(spannerConnection.execute(statement));
         }
@@ -226,8 +233,10 @@ public class BackendConnection {
   private final ImmutableMap<String, LocalStatement> localStatements;
   private ConnectionState connectionState = ConnectionState.IDLE;
   private TransactionMode transactionMode = TransactionMode.IMPLICIT;
+  private final String currentSchema = "public";
   private final LinkedList<BufferedStatement<?>> bufferedStatements = new LinkedList<>();
   private final Connection spannerConnection;
+  private final DdlExecutor ddlExecutor;
   private final DdlTransactionMode ddlTransactionMode;
 
   /**
@@ -235,10 +244,12 @@ public class BackendConnection {
    * DdlTransactionMode}.
    */
   BackendConnection(
+      DatabaseId databaseId,
       Connection spannerConnection,
       DdlTransactionMode ddlTransactionMode,
       ImmutableList<LocalStatement> localStatements) {
     this.spannerConnection = spannerConnection;
+    this.ddlExecutor = new DdlExecutor(databaseId, this);
     this.ddlTransactionMode = ddlTransactionMode;
     //noinspection UnstableApiUsage
     this.localStatements =
@@ -301,6 +312,14 @@ public class BackendConnection {
     }
   }
 
+  Connection getSpannerConnection() {
+    return this.spannerConnection;
+  }
+
+  String getCurrentSchema() {
+    return this.currentSchema;
+  }
+
   /**
    * Flushes all buffered statements to Cloud Spanner and commits/rolls back the transaction at the
    * end if isSync=true.
@@ -569,11 +588,22 @@ public class BackendConnection {
         throw SpannerExceptionFactory.newSpannerException(
             ErrorCode.INVALID_ARGUMENT, "Statement type is not supported for batching");
     }
+    List<StatementResult> statementResults = new ArrayList<>(getStatementCount());
     int index = fromIndex;
     while (index < getStatementCount()) {
       StatementType statementType = getStatementType(index);
       if (canBeBatchedTogether(batchType, statementType)) {
-        spannerConnection.execute(bufferedStatements.get(index).statement);
+        // Send DDL statements to the DdlExecutor instead of executing them directly on the
+        // connection, so we can support certain DDL constructs that are currently not supported by
+        // the backend, such as IF [NOT] EXISTS.
+        if (batchType == StatementType.DDL) {
+          statementResults.add(
+              ddlExecutor.execute(
+                  bufferedStatements.get(index).parsedStatement,
+                  bufferedStatements.get(index).statement));
+        } else {
+          spannerConnection.execute(bufferedStatements.get(index).statement);
+        }
         index++;
       } else {
         // End the batch here, as the statement type on this index can not be batched together with
@@ -583,9 +613,17 @@ public class BackendConnection {
     }
     try {
       long[] counts = spannerConnection.runBatch();
+      if (batchType == StatementType.DDL) {
+        counts = extractDdlUpdateCounts(statementResults, counts);
+      }
       updateBatchResultCount(fromIndex, counts);
     } catch (SpannerBatchUpdateException batchUpdateException) {
-      long[] counts = batchUpdateException.getUpdateCounts();
+      long[] counts;
+      if (batchType == StatementType.DDL) {
+        counts = extractDdlUpdateCounts(statementResults, batchUpdateException.getUpdateCounts());
+      } else {
+        counts = batchUpdateException.getUpdateCounts();
+      }
       updateBatchResultCount(fromIndex, counts);
       Execute failedExecute = (Execute) bufferedStatements.get(fromIndex + counts.length);
       failedExecute.result.setException(batchUpdateException);
@@ -594,6 +632,32 @@ public class BackendConnection {
     return index - fromIndex;
   }
 
+  /**
+   * Extracts the update count for a list of DDL statements. It could be that the DdlExecutor
+   * decided to skip some DDL statements. This is indicated by the executor returning a {@link
+   * NotExecuted} result for that statement. The result that we return to the client should still
+   * indicate that the statement was 'executed'.
+   */
+  static long[] extractDdlUpdateCounts(
+      List<StatementResult> statementResults, long[] returnedUpdateCounts) {
+    int index = 0;
+    int successfullyExecutedCount = 0;
+    while (index < returnedUpdateCounts.length
+        && successfullyExecutedCount < statementResults.size()) {
+      if (!(statementResults.get(successfullyExecutedCount) instanceof NotExecuted)) {
+        index++;
+      }
+      successfullyExecutedCount++;
+    }
+    while (successfullyExecutedCount < statementResults.size()
+        && statementResults.get(successfullyExecutedCount) instanceof NotExecuted) {
+      successfullyExecutedCount++;
+    }
+    long[] updateCounts = new long[successfullyExecutedCount];
+    Arrays.fill(updateCounts, 1L);
+    return updateCounts;
+  }
+
   /** Updates the results of the buffered statements after finishing executing a batch. */
   private void updateBatchResultCount(int fromIndex, long[] updateCounts) {
     for (int index = fromIndex; index < fromIndex + updateCounts.length; index++) {
@@ -613,7 +677,7 @@ public class BackendConnection {
   public static final class NoResult implements StatementResult {
     private final String commandTag;
 
-    private NoResult() {
+    NoResult() {
       this.commandTag = null;
     }
 
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java
new file mode 100644
index 00000000..58cb8220
--- /dev/null
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java
@@ -0,0 +1,340 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import com.google.cloud.spanner.AbstractLazyInitializer;
+import com.google.cloud.spanner.DatabaseAdminClient;
+import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.DatabaseId;
+import com.google.cloud.spanner.ErrorCode;
+import com.google.cloud.spanner.ResultSet;
+import com.google.cloud.spanner.Spanner;
+import com.google.cloud.spanner.SpannerException;
+import com.google.cloud.spanner.SpannerExceptionFactory;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
+import com.google.cloud.spanner.connection.Connection;
+import com.google.cloud.spanner.connection.ConnectionOptionsHelper;
+import com.google.cloud.spanner.connection.StatementResult;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableList;
+import java.util.concurrent.ExecutionException;
+import java.util.function.Function;
+
+/**
+ * {@link DdlExecutor} inspects DDL statements before executing these to support commonly used DDL
+ * features that are not (yet) supported by the backend.
+ */
+class DdlExecutor {
+  /**
+   * This class checks whether the backend supports 'if [not] exists' style statements. This is used
+   * to determine whether PGAdapter should do the existence check itself, or whether it should send
+   * the DDL statement unmodified to the backend. The check is only executed once for each
+   * connection, and only if at least one 'if [not] exists' statement is executed.
+   */
+  static class BackendSupportsIfExists extends AbstractLazyInitializer<Boolean> {
+    private final DatabaseId databaseId;
+    private final Connection connection;
+
+    BackendSupportsIfExists(DatabaseId databaseId, Connection connection) {
+      this.databaseId = databaseId;
+      this.connection = connection;
+    }
+
+    @Override
+    protected Boolean initialize() {
+      Spanner spanner = ConnectionOptionsHelper.getSpanner(this.connection);
+      DatabaseAdminClient adminClient = spanner.getDatabaseAdminClient();
+      try {
+        // NOTE: This update will *NEVER* succeed. Either, the first statement fails to parse
+        // because the backend does not support 'if [not] exists'-style statements, or the second
+        // statement fails because it misses a data type for the `id` column. The error that is
+        // returned indicates whether the backend supports `if [not] exists`, as Spanner will parse
+        // the statements in order, and return an error for the first statement that fails.
+        adminClient
+            .updateDatabaseDdl(
+                this.databaseId.getInstanceId().getInstance(),
+                this.databaseId.getDatabase(),
+                ImmutableList.of(
+                    "create table if not exists test (id bigint primary key)",
+                    "create table invalid (id primary key)"),
+                null)
+            .get();
+        // This statement should not be reachable, as at least one of the statements should fail.
+        return Boolean.TRUE;
+      } catch (ExecutionException exception) {
+        SpannerException spannerException =
+            SpannerExceptionFactory.asSpannerException(exception.getCause());
+        if (spannerException.getErrorCode() == ErrorCode.INVALID_ARGUMENT
+            && spannerException
+                .getMessage()
+                .contains("<IF NOT EXISTS> clause is not supported in <CREATE TABLE> statement")) {
+          return Boolean.FALSE;
+        }
+        return Boolean.TRUE;
+      } catch (InterruptedException interruptedException) {
+        throw SpannerExceptionFactory.propagateInterrupt(interruptedException);
+      }
+    }
+  }
+
+  /**
+   * {@link NotExecuted} is used to indicate that a statement was skipped because the 'if [not]
+   * exists' check indicated that the statement should be skipped.
+   */
+  static final class NotExecuted implements StatementResult {
+    @Override
+    public ResultType getResultType() {
+      return ResultType.NO_RESULT;
+    }
+
+    @Override
+    public ClientSideStatementType getClientSideStatementType() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public ResultSet getResultSet() {
+      throw new UnsupportedOperationException();
+    }
+
+    @Override
+    public Long getUpdateCount() {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  private static final NotExecuted NOT_EXECUTED = new NotExecuted();
+  private final AbstractLazyInitializer<Boolean> backendSupportsIfExists;
+  private final BackendConnection backendConnection;
+  private final Connection connection;
+
+  /**
+   * Constructor only intended for testing. This will create an executor that always assumes that
+   * the backend does not support 'if [not] exists'.
+   */
+  @VisibleForTesting
+  DdlExecutor(BackendConnection backendConnection) {
+    this(
+        new AbstractLazyInitializer<Boolean>() {
+          @Override
+          protected Boolean initialize() {
+            return Boolean.FALSE;
+          }
+        },
+        backendConnection);
+  }
+
+  /** Constructor for a {@link DdlExecutor} for the given database and connection. */
+  DdlExecutor(DatabaseId databaseId, BackendConnection backendConnection) {
+    this(
+        new BackendSupportsIfExists(databaseId, backendConnection.getSpannerConnection()),
+        backendConnection);
+  }
+
+  private DdlExecutor(
+      AbstractLazyInitializer<Boolean> backendSupportsIfExists,
+      BackendConnection backendConnection) {
+    this.backendConnection = backendConnection;
+    this.connection = backendConnection.getSpannerConnection();
+    this.backendSupportsIfExists = backendSupportsIfExists;
+  }
+
+  /**
+   * Removes any quotes around the given identifier, and folds the identifier to lower case if it
+   * was not quoted.
+   */
+  static String unquoteIdentifier(String identifier) {
+    if (identifier.length() < 2) {
+      return identifier;
+    }
+    if (identifier.startsWith("\"")) {
+      return identifier.substring(1, identifier.length() - 1);
+    }
+    return identifier.toLowerCase();
+  }
+
+  /**
+   * Executes the given DDL statement. This can include translating the statement into a check
+   * whether the to create/drop object exists, and skipping the actual execution of the DDL
+   * statement depending on the result of the `if [not] exists` check.
+   */
+  StatementResult execute(ParsedStatement parsedStatement, Statement statement) {
+    Statement translated = translate(parsedStatement, statement);
+    if (translated != null) {
+      return connection.execute(translated);
+    }
+    return NOT_EXECUTED;
+  }
+
+  /**
+   * Translates a DDL statement that can contain an `if [not] exists` statement into one that is
+   * supported by the backend.
+   */
+  Statement translate(ParsedStatement parsedStatement, Statement statement) {
+    SimpleParser parser = new SimpleParser(parsedStatement.getSqlWithoutComments());
+    if (parser.eat("create")) {
+      statement = translateCreate(parser, statement);
+    } else if (parser.eat("drop")) {
+      statement = translateDrop(parser, statement);
+    }
+
+    return statement;
+  }
+
+  Statement translateCreate(SimpleParser parser, Statement statement) {
+    if (parser.eat("table")) {
+      return translateCreateTable(parser, statement);
+    }
+    boolean unique = parser.eat("unique");
+    if (parser.eat("index")) {
+      return translateCreateIndex(parser, statement, unique);
+    }
+    return statement;
+  }
+
+  Statement translateDrop(SimpleParser parser, Statement statement) {
+    if (parser.eat("table")) {
+      return translateDropTable(parser, statement);
+    }
+    if (parser.eat("index")) {
+      return translateDropIndex(parser, statement);
+    }
+    return statement;
+  }
+
+  Statement translateCreateTable(SimpleParser parser, Statement statement) {
+    return translateCreateTableOrIndex(parser, statement, "table", this::tableExists);
+  }
+
+  Statement translateCreateIndex(SimpleParser parser, Statement statement, boolean unique) {
+    return translateCreateTableOrIndex(
+        parser, statement, unique ? "unique index" : "index", this::indexExists);
+  }
+
+  Statement translateDropTable(SimpleParser parser, Statement statement) {
+    return translateDropTableOrIndex(parser, statement, "table", this::tableExists);
+  }
+
+  Statement translateDropIndex(SimpleParser parser, Statement statement) {
+    return translateDropTableOrIndex(parser, statement, "index", this::indexExists);
+  }
+
+  private Statement translateCreateTableOrIndex(
+      SimpleParser parser,
+      Statement statement,
+      String type,
+      Function<TableOrIndexName, Boolean> existsFunction) {
+    if (!parser.eat("if", "not", "exists")) {
+      return statement;
+    }
+    return maybeExecuteStatement(
+        parser, statement, "create", type, name -> !existsFunction.apply(name));
+  }
+
+  private Statement translateDropTableOrIndex(
+      SimpleParser parser,
+      Statement statement,
+      String type,
+      Function<TableOrIndexName, Boolean> existsFunction) {
+    if (!parser.eat("if", "exists")) {
+      return statement;
+    }
+    return maybeExecuteStatement(parser, statement, "drop", type, existsFunction);
+  }
+
+  /** Executes the given DDL statement if the shouldExecuteFunction indicates so. */
+  Statement maybeExecuteStatement(
+      SimpleParser parser,
+      Statement statement,
+      String command,
+      String type,
+      Function<TableOrIndexName, Boolean> shouldExecuteFunction) {
+    if (backendSupportsIfExists()) {
+      return statement;
+    }
+    int identifierPosition = parser.getPos();
+    if (identifierPosition >= parser.getSql().length()) {
+      return statement;
+    }
+    TableOrIndexName name = parser.readTableOrIndexName();
+    if (name == null) {
+      return statement;
+    }
+    // Check if the table exists or not.
+    if (!shouldExecuteFunction.apply(name)) {
+      // Return null to indicate that the statement should be skipped.
+      return null;
+    }
+
+    // Return the DDL statement without the 'if [not] exists' clause.
+    return Statement.of(command + " " + type + parser.getSql().substring(identifierPosition));
+  }
+
+  boolean tableExists(TableOrIndexName tableName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                    .bind("p1")
+                    .to(
+                        unquoteIdentifier(
+                            tableName.schema == null
+                                ? backendConnection.getCurrentSchema()
+                                : tableName.schema))
+                    .bind("p2")
+                    .to(unquoteIdentifier(tableName.name))
+                    .build())) {
+      return resultSet.next();
+    }
+  }
+
+  boolean indexExists(TableOrIndexName indexName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                    .bind("p1")
+                    .to(
+                        unquoteIdentifier(
+                            indexName.schema == null
+                                ? backendConnection.getCurrentSchema()
+                                : indexName.schema))
+                    .bind("p2")
+                    .to(unquoteIdentifier(indexName.name))
+                    .build())) {
+      return resultSet.next();
+    }
+  }
+
+  /**
+   * Returns true if the backend supports `if [not] exists`-style statements. This will cause the
+   * {@link DdlExecutor} to send these statements unmodified to the backend.
+   */
+  boolean backendSupportsIfExists() {
+    try {
+      return backendSupportsIfExists.get();
+    } catch (Exception exception) {
+      throw SpannerExceptionFactory.asSpannerException(exception);
+    }
+  }
+}
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
index 64b69afa..861d02bf 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/ExtendedQueryProtocolHandler.java
@@ -41,6 +41,7 @@ public class ExtendedQueryProtocolHandler {
             : connectionHandler.getWellKnownClient().getLocalStatements(connectionHandler);
     this.backendConnection =
         new BackendConnection(
+            connectionHandler.getDatabaseId(),
             connectionHandler.getSpannerConnection(),
             connectionHandler.getServer().getOptions().getDdlTransactionMode(),
             localStatements);


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
index fa50f537..99ca458e 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/AbstractMockServerTest.java
@@ -106,6 +106,8 @@ public abstract class AbstractMockServerTest {
                           .build())
                   .build())
           .build();
+  protected static final com.google.spanner.v1.ResultSet EMPTY_RESULTSET =
+      com.google.spanner.v1.ResultSet.newBuilder().setMetadata(SELECT1_METADATA).build();
   protected static final com.google.spanner.v1.ResultSet SELECT1_RESULTSET =
       com.google.spanner.v1.ResultSet.newBuilder()
           .addRows(
@@ -456,4 +458,11 @@ public abstract class AbstractMockServerTest {
     mockDatabaseAdmin.addException(
         Status.INVALID_ARGUMENT.withDescription("Statement is invalid.").asRuntimeException());
   }
+
+  protected static void addIfNotExistsDdlException() {
+    mockDatabaseAdmin.addException(
+        Status.INVALID_ARGUMENT
+            .withDescription("<IF NOT EXISTS> clause is not supported in <CREATE TABLE> statement.")
+            .asRuntimeException());
+  }
 }
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
index f1e6c35c..6e6f1b92 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITJdbcTest.java
@@ -38,6 +38,7 @@ import java.sql.DriverManager;
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.sql.Statement;
 import java.sql.Types;
 import java.time.LocalDate;
 import java.util.ArrayList;
@@ -159,6 +160,31 @@ public class ITJdbcTest implements IntegrationTest {
     }
   }
 
+  @Test
+  public void testCreateTableIfNotExists() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        // This table already exists, so it should be a no-op.
+        assertFalse(
+            statement.execute("create table if not exists all_types (id bigint primary key)"));
+        assertFalse(statement.getMoreResults());
+      }
+    }
+  }
+
+  @Test
+  public void testCreateTableIfNotExists_withSchemaPrefix() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        // This table already exists, so it should be a no-op.
+        assertFalse(
+            statement.execute(
+                "create table if not exists public.all_types (id bigint primary key)"));
+        assertFalse(statement.getMoreResults());
+      }
+    }
+  }
+
   @Test
   public void testSelectWithParameters() throws SQLException {
     boolean isSimpleMode = "simple".equalsIgnoreCase(preferQueryMode);
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 5a5eb0a4..3e84c0c8 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/JdbcMockServerTest.java
@@ -677,6 +677,369 @@ public class JdbcMockServerTest extends AbstractMockServerTest {
     }
   }
 
+  @Test
+  public void testCreateTableIfNotExists_tableDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "create table foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testCreateTableIfNotExists_tableExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(0, updateDatabaseDdlRequests.size());
+  }
+
+  @Test
+  public void testCreateIndexIfNotExists_indexDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE INDEX IF NOT EXISTS foo on bar (value)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "create index foo on bar (value)", updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testCreateIndexIfNotExists_indexExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "CREATE INDEX IF NOT EXISTS foo on bar (value)";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(0, updateDatabaseDdlRequests.size());
+  }
+
+  @Test
+  public void testDropTableIfExists_tableDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP TABLE IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(0, updateDatabaseDdlRequests.size());
+  }
+
+  @Test
+  public void testDropTableIfExists_tableExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP TABLE IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals("drop table foo", updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testDropIndexIfExists_indexDoesNotExist() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP INDEX IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            EMPTY_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(0, updateDatabaseDdlRequests.size());
+  }
+
+  @Test
+  public void testDropIndexIfExists_indexExists() throws SQLException {
+    addIfNotExistsDdlException();
+    String sql = "DROP INDEX IF EXISTS foo";
+    addDdlResponseToSpannerAdmin();
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals("drop index foo", updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testDdlBatchWithIfNotExists() throws SQLException {
+    addIfNotExistsDdlException();
+    ImmutableList<String> statements =
+        ImmutableList.of(
+            "CREATE TABLE IF NOT EXISTS \"Foo\" (id bigint primary key)",
+            "CREATE TABLE IF NOT EXISTS bar (id bigint primary key, value text)",
+            "CREATE INDEX IF NOT EXISTS idx_foo ON bar (text)");
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("Foo")
+                .build(),
+            SELECT1_RESULTSET));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.tables where table_schema=$1 and table_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("bar")
+                .build(),
+            SELECT1_RESULTSET));
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            Statement.newBuilder(
+                    "select 1 from information_schema.indexes where table_schema=$1 and index_name=$2")
+                .bind("p1")
+                .to("public")
+                .bind("p2")
+                .to("idx_foo")
+                .build(),
+            SELECT1_RESULTSET));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        for (String sql : statements) {
+          statement.addBatch(sql);
+        }
+        int[] updateCounts = statement.executeBatch();
+        assertArrayEquals(new int[] {0, 0, 0}, updateCounts);
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(0, updateDatabaseDdlRequests.size());
+  }
+
+  @Test
+  public void testCreateTableIfNotExists_withBackendSupport() throws SQLException {
+    // Add a generic error that is returned for DDL statements. This will cause PGAdapter to think
+    // that the backend supports `IF [NOT] EXISTS`, as it does not receive a specific error
+    // regarding an `IF NOT EXISTS` statement.
+    addDdlExceptionToSpannerAdmin();
+    String sql = "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)";
+    // Add a response for the DDL statement that is sent to Spanner.
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (java.sql.Statement statement = connection.createStatement()) {
+        // Statement#execute(String) returns false if the result was either an update count or there
+        // was no result. Statement#getUpdateCount() returns 0 if there was no result.
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(1, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "CREATE TABLE IF NOT EXISTS foo (id bigint primary key)",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
   @Test
   public void testPreparedStatement() throws SQLException {
     mockSpanner.putStatementResult(
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
index 9fe7b413..f1636edd 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/BackendConnectionTest.java
@@ -14,6 +14,8 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
+import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.extractDdlUpdateCounts;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertSame;
@@ -26,6 +28,7 @@ import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import com.google.cloud.spanner.DatabaseId;
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.Statement;
@@ -35,7 +38,9 @@ import com.google.cloud.spanner.connection.Connection;
 import com.google.cloud.spanner.connection.StatementResult;
 import com.google.cloud.spanner.connection.StatementResult.ResultType;
 import com.google.cloud.spanner.pgadapter.metadata.OptionsMetadata.DdlTransactionMode;
+import com.google.cloud.spanner.pgadapter.statements.BackendConnection.NoResult;
 import com.google.cloud.spanner.pgadapter.statements.BackendConnection.QueryResult;
+import com.google.cloud.spanner.pgadapter.statements.DdlExecutor.NotExecuted;
 import com.google.cloud.spanner.pgadapter.statements.local.ListDatabasesStatement;
 import com.google.cloud.spanner.pgadapter.statements.local.LocalStatement;
 import com.google.cloud.spanner.pgadapter.utils.CopyDataReceiver;
@@ -53,6 +58,59 @@ import org.junit.runners.JUnit4;
 public class BackendConnectionTest {
   private final AbstractStatementParser PARSER =
       AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
+  private static final NotExecuted NOT_EXECUTED = new NotExecuted();
+  private static final NoResult NO_RESULT = new NoResult();
+
+  @Test
+  public void testExtractDdlUpdateCounts() {
+    assertArrayEquals(new long[] {}, extractDdlUpdateCounts(ImmutableList.of(), new long[] {}));
+    assertArrayEquals(
+        new long[] {1L}, extractDdlUpdateCounts(ImmutableList.of(NO_RESULT), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(ImmutableList.of(NO_RESULT, NOT_EXECUTED), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(ImmutableList.of(NOT_EXECUTED, NO_RESULT), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NO_RESULT, NO_RESULT), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NO_RESULT, NOT_EXECUTED), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NOT_EXECUTED, NO_RESULT, NOT_EXECUTED),
+            new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NOT_EXECUTED, NO_RESULT, NOT_EXECUTED, NO_RESULT),
+            new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NO_RESULT, NO_RESULT, NOT_EXECUTED), new long[] {1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L, 1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(
+                NOT_EXECUTED,
+                NO_RESULT,
+                NO_RESULT,
+                NOT_EXECUTED,
+                NOT_EXECUTED,
+                NO_RESULT,
+                NO_RESULT),
+            new long[] {1L, 1L, 1L}));
+    assertArrayEquals(
+        new long[] {1L, 1L, 1L},
+        extractDdlUpdateCounts(
+            ImmutableList.of(NOT_EXECUTED, NOT_EXECUTED, NOT_EXECUTED), new long[] {}));
+  }
 
   @Test
   public void testCopyPropagatesNonSpannerException() {
@@ -67,7 +125,11 @@ public class BackendConnectionTest {
     doThrow(rejectedExecutionException).when(executor).execute(any(Runnable.class));
 
     BackendConnection backendConnection =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     Future<StatementResult> result =
         backendConnection.executeCopy(parsedStatement, statement, receiver, writer, executor);
     backendConnection.flush();
@@ -100,21 +162,33 @@ public class BackendConnectionTest {
     ParsedStatement parsedUnknownStatement = PARSER.parse(unknownStatement);
 
     BackendConnection onlyDmlStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     onlyDmlStatements.execute(parsedUpdateStatement, updateStatement);
     onlyDmlStatements.execute(parsedUpdateStatement, updateStatement);
     assertTrue(onlyDmlStatements.hasDmlOrCopyStatementsAfter(0));
     assertTrue(onlyDmlStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection onlyCopyStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     onlyCopyStatements.executeCopy(parsedCopyStatement, copyStatement, receiver, writer, executor);
     onlyCopyStatements.executeCopy(parsedCopyStatement, copyStatement, receiver, writer, executor);
     assertTrue(onlyCopyStatements.hasDmlOrCopyStatementsAfter(0));
     assertTrue(onlyCopyStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection dmlAndCopyStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     dmlAndCopyStatements.execute(parsedUpdateStatement, updateStatement);
     dmlAndCopyStatements.executeCopy(
         parsedCopyStatement, copyStatement, receiver, writer, executor);
@@ -122,35 +196,55 @@ public class BackendConnectionTest {
     assertTrue(dmlAndCopyStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection onlySelectStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     onlySelectStatements.execute(parsedSelectStatement, selectStatement);
     onlySelectStatements.execute(parsedSelectStatement, selectStatement);
     assertFalse(onlySelectStatements.hasDmlOrCopyStatementsAfter(0));
     assertFalse(onlySelectStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection onlyClientSideStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     onlyClientSideStatements.execute(parsedClientSideStatement, clientSideStatement);
     onlyClientSideStatements.execute(parsedClientSideStatement, clientSideStatement);
     assertFalse(onlyClientSideStatements.hasDmlOrCopyStatementsAfter(0));
     assertFalse(onlyClientSideStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection onlyUnknownStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     onlyUnknownStatements.execute(parsedUnknownStatement, unknownStatement);
     onlyUnknownStatements.execute(parsedUnknownStatement, unknownStatement);
     assertFalse(onlyUnknownStatements.hasDmlOrCopyStatementsAfter(0));
     assertFalse(onlyUnknownStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection dmlAndSelectStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     dmlAndSelectStatements.execute(parsedUpdateStatement, updateStatement);
     dmlAndSelectStatements.execute(parsedSelectStatement, selectStatement);
     assertTrue(dmlAndSelectStatements.hasDmlOrCopyStatementsAfter(0));
     assertFalse(dmlAndSelectStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection copyAndSelectStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     copyAndSelectStatements.executeCopy(
         parsedCopyStatement, copyStatement, receiver, writer, executor);
     copyAndSelectStatements.execute(parsedSelectStatement, selectStatement);
@@ -158,7 +252,11 @@ public class BackendConnectionTest {
     assertFalse(copyAndSelectStatements.hasDmlOrCopyStatementsAfter(1));
 
     BackendConnection copyAndUnknownStatements =
-        new BackendConnection(spannerConnection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"),
+            spannerConnection,
+            DdlTransactionMode.Batch,
+            ImmutableList.of());
     copyAndUnknownStatements.executeCopy(
         parsedCopyStatement, copyStatement, receiver, writer, executor);
     copyAndUnknownStatements.execute(parsedUnknownStatement, unknownStatement);
@@ -180,7 +278,8 @@ public class BackendConnectionTest {
         .thenReturn(ListDatabasesStatement.LIST_DATABASES_SQL);
 
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, localStatements);
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, localStatements);
     Future<StatementResult> resultFuture =
         backendConnection.execute(
             parsedListDatabasesStatement, Statement.of(ListDatabasesStatement.LIST_DATABASES_SQL));
@@ -208,7 +307,8 @@ public class BackendConnectionTest {
     when(connection.execute(statement)).thenReturn(statementResult);
 
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, localStatements);
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, localStatements);
     Future<StatementResult> resultFuture = backendConnection.execute(parsedStatement, statement);
     backendConnection.flush();
 
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java
new file mode 100644
index 00000000..72840d1a
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java
@@ -0,0 +1,392 @@
+// Copyright 2022 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter.statements;
+
+import static com.google.cloud.spanner.pgadapter.statements.DdlExecutor.unquoteIdentifier;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+
+import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.Statement;
+import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class DdlExecutorTest {
+  private static final AbstractStatementParser PARSER =
+      AbstractStatementParser.getInstance(Dialect.POSTGRESQL);
+
+  @Test
+  public void testUnquoteIdentifier() {
+    assertEquals("foo", unquoteIdentifier("foo"));
+    assertEquals("foo", unquoteIdentifier("FOO"));
+    assertEquals("foo", unquoteIdentifier("Foo"));
+
+    assertEquals("foo", unquoteIdentifier("\"foo\""));
+    assertEquals("FOO", unquoteIdentifier("\"FOO\""));
+    assertEquals("Foo", unquoteIdentifier("\"Foo\""));
+
+    assertEquals("", unquoteIdentifier(""));
+    assertEquals("a", unquoteIdentifier("a"));
+    assertEquals("", unquoteIdentifier("\"\""));
+  }
+
+  private Statement translate(String sql, DdlExecutor executor) {
+    Statement statement = Statement.of(sql);
+    return executor.translate(PARSER.parse(statement), statement);
+  }
+
+  @Test
+  public void testTranslateCreateTableNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create table foo (id int)"),
+        translate("create table foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"Foo\" (id int)"),
+        translate("create table \"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.foo (id int)"),
+        translate("create table s.foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.\"Foo\" (id int)"),
+        translate("create table s.\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"s\".\"Foo\" (id int)"),
+        translate("create table \"s\".\"Foo\" (id int)", executor));
+    assertEquals(Statement.of("create table"), translate("create table", executor));
+    assertEquals(Statement.of("create table "), translate("create table ", executor));
+
+    assertEquals(
+        Statement.of("create table foo (id int)"),
+        translate("create table if not exists foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"Foo\" (id int)"),
+        translate("create table if not exists \"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.foo (id int)"),
+        translate("create table if not exists s.foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.\"Foo\" (id int)"),
+        translate("create table if not exists s.\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"s\".\"Foo\" (id int)"),
+        translate("create table if not exists \"s\".\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table if not exists"),
+        translate("create table if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateTableExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create table foo (id int)"),
+        translate("create table foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"Foo\" (id int)"),
+        translate("create table \"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table s.foo (id int)"),
+        translate("create table s.foo (id int)", executor));
+    assertEquals(
+        Statement.of("create table \"S\".\"Foo\" (id int)"),
+        translate("create table \"S\".\"Foo\" (id int)", executor));
+    assertEquals(Statement.of("create table"), translate("create table", executor));
+
+    assertNull(translate("create table if not exists foo (id int)", executor));
+    assertNull(translate("create table if not exists \"Foo\" (id int)", executor));
+    assertNull(translate("create table if not exists s.foo (id int)", executor));
+    assertNull(translate("create table if not exists \"S\".\"Foo\" (id int)", executor));
+    assertEquals(
+        Statement.of("create table if not exists"),
+        translate("create table if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateIndexNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create index foo on bar(id)"),
+        translate("create index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"Foo\" on bar(id)"),
+        translate("create index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index s.foo on bar(id)"),
+        translate("create index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"s\".\"Foo\" on bar(id)"),
+        translate("create index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create index"), translate("create index", executor));
+    assertEquals(Statement.of("create index "), translate("create index ", executor));
+
+    assertEquals(
+        Statement.of("create index foo on bar(id)"),
+        translate("create index if not exists foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"Foo\" on bar(id)"),
+        translate("create index if not exists \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index s.foo on bar(id)"),
+        translate("create index if not exists s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"s\".\"Foo\" on bar(id)"),
+        translate("create index if not exists \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index if not exists"),
+        translate("create index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateIndexExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create index foo on bar(id)"),
+        translate("create index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"Foo\" on bar(id)"),
+        translate("create index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index s.foo on bar(id)"),
+        translate("create index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create index \"s\".\"Foo\" on bar(id)"),
+        translate("create index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create index"), translate("create index", executor));
+
+    assertNull(translate("create index if not exists foo on bar(id int)", executor));
+    assertNull(translate("create index if not exists \"Foo\" on bar(id int)", executor));
+    assertNull(translate("create index if not exists s.foo on bar(id int)", executor));
+    assertNull(translate("create index if not exists \"s\".\"Foo\" on bar(id int)", executor));
+    assertEquals(
+        Statement.of("create index if not exists"),
+        translate("create index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateUniqueIndexNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create unique index foo on bar(id)"),
+        translate("create unique index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"Foo\" on bar(id)"),
+        translate("create unique index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index s.foo on bar(id)"),
+        translate("create unique index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"s\".\"Foo\" on bar(id)"),
+        translate("create unique index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create unique index"), translate("create unique index", executor));
+    assertEquals(Statement.of("create unique index "), translate("create unique index ", executor));
+
+    assertEquals(
+        Statement.of("create unique index foo on bar(id)"),
+        translate("create unique index if not exists foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"Foo\" on bar(id)"),
+        translate("create unique index if not exists \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index s.foo on bar(id)"),
+        translate("create unique index if not exists s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"s\".\"Foo\" on bar(id)"),
+        translate("create unique index if not exists \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index if not exists"),
+        translate("create unique index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateCreateUniqueIndexExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(
+        Statement.of("create unique index foo on bar(id)"),
+        translate("create unique index foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"Foo\" on bar(id)"),
+        translate("create unique index \"Foo\" on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index s.foo on bar(id)"),
+        translate("create unique index s.foo on bar(id)", executor));
+    assertEquals(
+        Statement.of("create unique index \"s\".\"Foo\" on bar(id)"),
+        translate("create unique index \"s\".\"Foo\" on bar(id)", executor));
+    assertEquals(Statement.of("create unique index"), translate("create unique index", executor));
+
+    assertNull(translate("create unique index if not exists foo on bar(id int)", executor));
+    assertNull(translate("create unique index if not exists \"Foo\" on bar(id int)", executor));
+    assertNull(translate("create unique index if not exists s.foo on bar(id int)", executor));
+    assertNull(
+        translate("create unique index if not exists \"s\".\"Foo\" on bar(id int)", executor));
+    assertEquals(
+        Statement.of("create unique index if not exists"),
+        translate("create unique index if not exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropTableNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(Statement.of("drop table foo"), translate("drop table foo", executor));
+    assertEquals(Statement.of("drop table \"Foo\""), translate("drop table \"Foo\"", executor));
+    assertEquals(Statement.of("drop table s.foo"), translate("drop table s.foo", executor));
+    assertEquals(
+        Statement.of("drop table \"s\".\"Foo\""), translate("drop table \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table"), translate("drop table", executor));
+    assertEquals(Statement.of("drop table "), translate("drop table ", executor));
+
+    assertNull(translate("drop table if exists foo", executor));
+    assertNull(translate("drop table if exists \"Foo\"", executor));
+    assertNull(translate("drop table if exists sfoo", executor));
+    assertNull(translate("drop table if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table if exists"), translate("drop table if exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropTableExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean tableExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(Statement.of("drop table foo"), translate("drop table foo", executor));
+    assertEquals(Statement.of("drop table \"Foo\""), translate("drop table \"Foo\"", executor));
+    assertEquals(Statement.of("drop table s.foo"), translate("drop table s.foo", executor));
+    assertEquals(
+        Statement.of("drop table \"s\".\"Foo\""), translate("drop table \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table"), translate("drop table", executor));
+
+    assertEquals(Statement.of("drop table foo"), translate("drop table if exists foo", executor));
+    assertEquals(
+        Statement.of("drop table \"Foo\""), translate("drop table if exists \"Foo\"", executor));
+    assertEquals(
+        Statement.of("drop table s.foo"), translate("drop table if exists s.foo", executor));
+    assertEquals(
+        Statement.of("drop table \"s\".\"Foo\""),
+        translate("drop table if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop table if exists"), translate("drop table if exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropIndexNotExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return false;
+          }
+        };
+
+    assertEquals(Statement.of("drop index foo"), translate("drop index foo", executor));
+    assertEquals(Statement.of("drop index \"Foo\""), translate("drop index \"Foo\"", executor));
+    assertEquals(Statement.of("drop index s.foo"), translate("drop index s.foo", executor));
+    assertEquals(
+        Statement.of("drop index \"s\".\"Foo\""), translate("drop index \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index"), translate("drop index", executor));
+    assertEquals(Statement.of("drop index "), translate("drop index ", executor));
+
+    assertNull(translate("drop index if exists foo", executor));
+    assertNull(translate("drop index if exists \"Foo\"", executor));
+    assertNull(translate("drop index if exists s.foo", executor));
+    assertNull(translate("drop index if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index if exists"), translate("drop index if exists", executor));
+  }
+
+  @Test
+  public void testTranslateDropIndexExists() {
+    DdlExecutor executor =
+        new DdlExecutor(mock(BackendConnection.class)) {
+          @Override
+          boolean indexExists(TableOrIndexName name) {
+            return true;
+          }
+        };
+
+    assertEquals(Statement.of("drop index foo"), translate("drop index foo", executor));
+    assertEquals(Statement.of("drop index \"Foo\""), translate("drop index \"Foo\"", executor));
+    assertEquals(Statement.of("drop index s.foo"), translate("drop index s.foo", executor));
+    assertEquals(
+        Statement.of("drop index \"s\".\"Foo\""), translate("drop index \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index"), translate("drop index", executor));
+
+    assertEquals(Statement.of("drop index foo"), translate("drop index if exists foo", executor));
+    assertEquals(
+        Statement.of("drop index \"Foo\""), translate("drop index if exists \"Foo\"", executor));
+    assertEquals(
+        Statement.of("drop index s.foo"), translate("drop index if exists s.foo", executor));
+    assertEquals(
+        Statement.of("drop index \"s\".\"Foo\""),
+        translate("drop index if exists \"s\".\"Foo\"", executor));
+    assertEquals(Statement.of("drop index if exists"), translate("drop index if exists", executor));
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
index f44c88bb..85755ed1 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/StatementTest.java
@@ -27,6 +27,7 @@ import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 import com.google.cloud.spanner.DatabaseClient;
+import com.google.cloud.spanner.DatabaseId;
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.ErrorCode;
 import com.google.cloud.spanner.ReadContext;
@@ -167,7 +168,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sql), Statement.of(sql));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     assertFalse(intermediateStatement.isExecuted());
     assertEquals("UPDATE", intermediateStatement.getCommand());
@@ -239,7 +244,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sql), Statement.of(sql));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.executeAsync(backendConnection);
     backendConnection.flush();
@@ -264,7 +273,11 @@ public class StatementTest {
             .to(30)
             .build();
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     IntermediatePreparedStatement intermediateStatement =
         new IntermediatePreparedStatement(
@@ -331,7 +344,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sqlStatement), Statement.of(sqlStatement));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.describeAsync(backendConnection);
     backendConnection.flush();
@@ -375,7 +392,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sqlStatement), Statement.of(sqlStatement));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     when(connection.execute(Statement.of(sqlStatement)))
         .thenThrow(
@@ -426,7 +447,8 @@ public class StatementTest {
             connectionHandler, mock(OptionsMetadata.class), "", parse(sql), Statement.of(sql));
 
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, ImmutableList.of());
     statement.executeAsync(backendConnection);
     backendConnection.flush();
 
@@ -479,7 +501,11 @@ public class StatementTest {
         new IntermediatePortalStatement(
             connectionHandler, options, "", parse(sql), Statement.of(sql));
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, EMPTY_LOCAL_STATEMENTS);
+        new BackendConnection(
+            connectionHandler.getDatabaseId(),
+            connection,
+            DdlTransactionMode.Batch,
+            EMPTY_LOCAL_STATEMENTS);
 
     intermediateStatement.executeAsync(backendConnection);
 
@@ -490,7 +516,8 @@ public class StatementTest {
   public void testCopyBatchSizeLimit() throws Exception {
     setupQueryInformationSchemaResults();
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, ImmutableList.of());
 
     byte[] payload = Files.readAllBytes(Paths.get("./src/test/resources/batch-size-test.txt"));
 
@@ -526,7 +553,8 @@ public class StatementTest {
   public void testCopyDataRowLengthMismatchLimit() throws Exception {
     setupQueryInformationSchemaResults();
     BackendConnection backendConnection =
-        new BackendConnection(connection, DdlTransactionMode.Batch, ImmutableList.of());
+        new BackendConnection(
+            DatabaseId.of("p", "i", "d"), connection, DdlTransactionMode.Batch, ImmutableList.of());
 
     byte[] payload = "1\t'one'\n2".getBytes();