Repository: GoogleCloudPlatform/pgadapter
Pull request #841 (2023-05-18 05:01:17)

feat: support drop cascade
Adds support for the `CASCADE` option for dropping schemas and tables. The feature is conditional on executing `SET spanner.support_drop_cascade=true` (or similar) first.

* `DROP SCHEMA <schema> CASCADE` will drop all objects in the schema, but not the schema itself.
* `DROP TABLE <table> CASCADE` will first drop all dependent objects of the table and then the table itself. This happens in one DDL batch.
* `DROP TABLE <table>` will allow dropping a table, even if it has a secondary index.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/841?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) Report
> Merging [#841](https://app.codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/841?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (eb67651) into [postgresql-dialect](https://app.codecov.io/gh/GoogleCloudPlatform/pgadapter/commit/9be6b09d5e156d7f916d25467ca679e028aa19d1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform) (9be6b09) will **decrease** coverage by `1.06%`.
> The diff coverage is `2.08%`.

```diff
@@                   Coverage Diff                    @@
##             postgresql-dialect     #841      +/-   ##
========================================================
- Coverage                 89.75%   88.69%   -1.06%     
- Complexity                 2382     2383       +1     
========================================================
  Files                       131      131              
  Lines                      7896     7991      +95     
  Branches                   1133     1149      +16     
========================================================
+ Hits                       7087     7088       +1     
- Misses                      565      658      +93     
- Partials                    244      245       +1     
```

| Flag | Coverage Δ | |
|---|---|---|
| all_tests | `88.69% <2.08%> (-1.06%)` | :arrow_down: |

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

| [Impacted Files](https://app.codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/841?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://app.codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/841?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3N0YXRlbWVudHMvRGRsRXhlY3V0b3IuamF2YQ==) | `56.45% <1.05%> (-34.46%)` | :arrow_down: |
| [.../cloud/spanner/pgadapter/session/SessionState.java](https://app.codecov.io/gh/GoogleCloudPlatform/pgadapter/pull/841?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=GoogleCloudPlatform#diff-c3JjL21haW4vamF2YS9jb20vZ29vZ2xlL2Nsb3VkL3NwYW5uZXIvcGdhZGFwdGVyL3Nlc3Npb24vU2Vzc2lvblN0YXRlLmphdmE=) | `94.94% <100.00%> (+0.02%)` | :arrow_up: |

: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/benchmarks/latency-comparison/java/pom.xml b/benchmarks/latency-comparison/java/pom.xml
index 0ded0456..01df070b 100644
--- a/benchmarks/latency-comparison/java/pom.xml
+++ b/benchmarks/latency-comparison/java/pom.xml
@@ -21,7 +21,7 @@
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
-    <junixsocket.version>2.7.2</junixsocket.version>
+    <junixsocket.version>2.8.1</junixsocket.version>
   </properties>
 
   <modelVersion>4.0.0</modelVersion>
diff --git a/pom.xml b/pom.xml
index abd9b11b..09aa8e8e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -53,7 +53,7 @@
   <parent>
     <groupId>com.google.cloud</groupId>
     <artifactId>google-cloud-shared-config</artifactId>
-    <version>1.5.7</version>
+    <version>1.5.8</version>
   </parent>
   <scm>
     <connection>scm:git:git@github.com:GoogleCloudPlatform/pgadapter.git</connection>
@@ -76,14 +76,14 @@
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-spanner-bom</artifactId>
-        <version>6.47.0</version>
+        <version>6.51.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
       <dependency>
         <groupId>com.google.cloud</groupId>
         <artifactId>google-cloud-shared-dependencies</artifactId>
-        <version>3.15.0</version>
+        <version>3.17.0</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
diff --git a/samples/python/psycopg3/requirements.txt b/samples/python/psycopg3/requirements.txt
index 518f75d4..88e6aca5 100644
--- a/samples/python/psycopg3/requirements.txt
+++ b/samples/python/psycopg3/requirements.txt
@@ -1,3 +1,3 @@
-psycopg~=3.1.11
+psycopg~=3.1.12
 testcontainers~=3.7.1
 google~=3.0.0
diff --git a/samples/python/sqlalchemy-sample/requirements.txt b/samples/python/sqlalchemy-sample/requirements.txt
index 2bfae662..f6d1313b 100644
--- a/samples/python/sqlalchemy-sample/requirements.txt
+++ b/samples/python/sqlalchemy-sample/requirements.txt
@@ -1,3 +1,3 @@
 psycopg2~=2.9.3
 pytz~=2022.1
-sqlalchemy==1.4.45
+sqlalchemy==1.4.49
diff --git a/src/main/java/com/google/cloud/spanner/pgadapter/error/PGExceptionFactory.java b/src/main/java/com/google/cloud/spanner/pgadapter/error/PGExceptionFactory.java
index c0ec3625..4941aab2 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/error/PGExceptionFactory.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/error/PGExceptionFactory.java
@@ -27,6 +27,10 @@ import java.util.regex.Pattern;
 public class PGExceptionFactory {
   private static final Pattern RELATION_NOT_FOUND_PATTERN =
       Pattern.compile("relation .+ does not exist");
+  private static final Pattern CANNOT_DROP_TABLE_WITH_INDICES_PATTERN =
+      Pattern.compile("Cannot drop table test with indices");
+  private static final Pattern ONLY_RESTRICT_BEHAVIOR =
+      Pattern.compile("Only <RESTRICT> behavior is supported by <DROP> statement\\.");
 
   private PGExceptionFactory() {}
 
@@ -86,6 +90,20 @@ public class PGExceptionFactory {
       return PGException.newBuilder(extractMessage(spannerException))
           .setSQLState(SQLState.UndefinedTable)
           .build();
+    } else if (spannerException.getErrorCode() == ErrorCode.FAILED_PRECONDITION
+        && CANNOT_DROP_TABLE_WITH_INDICES_PATTERN.matcher(spannerException.getMessage()).find()) {
+      return PGException.newBuilder(extractMessage(spannerException))
+          .setSQLState(SQLState.FeatureNotSupported)
+          .setHints(
+              "Execute 'set spanner.support_drop_cascade=true' to enable dropping tables with indices")
+          .build();
+    } else if (spannerException.getErrorCode() == ErrorCode.INVALID_ARGUMENT
+        && ONLY_RESTRICT_BEHAVIOR.matcher(spannerException.getMessage()).find()) {
+      return PGException.newBuilder(extractMessage(spannerException))
+          .setSQLState(SQLState.FeatureNotSupported)
+          .setHints(
+              "Execute 'set spanner.support_drop_cascade=true' to enable 'drop {table|schema} cascade' statements.")
+          .build();
     }
     return newPGException(extractMessage(spannerException));
   }
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 a981ba84..c1988b43 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
@@ -399,6 +399,14 @@ public class SessionState {
     return getBoolSetting("spanner", "force_autocommit", false);
   }
 
+  /**
+   * Returns whether DROP ... CASCADE statements are supported. This also includes implicit
+   * cascading operations, such as dropping indexes of tables that are being dropped.
+   */
+  public boolean isSupportDropCascade() {
+    return getBoolSetting("spanner", "support_drop_cascade", false);
+  }
+
   /**
    * Returns whether statements with an OFFSET clause that uses a parameter should be automatically
    * appended with a LIMIT clause. The LIMIT clause will use the literal Long.MAX_VALUE for unbound
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 dca41885..5d898bd2 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
@@ -715,7 +715,7 @@ public class BackendConnection {
 
   private static final ImmutableMap<String, LocalStatement> EMPTY_LOCAL_STATEMENTS =
       ImmutableMap.of();
-  private static final StatementResult NO_RESULT = new NoResult();
+  static final StatementResult NO_RESULT = new NoResult();
   private static final StatementResult ROLLBACK_RESULT = new NoResult("ROLLBACK");
   private static final Statement ROLLBACK = Statement.of("ROLLBACK");
 
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
index e3e41b77..1b79f457 100644
--- a/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java
+++ b/src/main/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutor.java
@@ -14,8 +14,12 @@
 
 package com.google.cloud.spanner.pgadapter.statements;
 
+import static com.google.cloud.spanner.pgadapter.statements.BackendConnection.NO_RESULT;
+
 import com.google.cloud.Tuple;
+import com.google.cloud.spanner.DatabaseClient;
 import com.google.cloud.spanner.Dialect;
+import com.google.cloud.spanner.ResultSet;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser;
 import com.google.cloud.spanner.connection.AbstractStatementParser.ParsedStatement;
@@ -25,7 +29,12 @@ import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexNa
 import com.google.cloud.spanner.pgadapter.utils.QueryPartReplacer;
 import com.google.cloud.spanner.pgadapter.utils.QueryPartReplacer.ReplacementStatus;
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
 import java.util.function.Supplier;
 
 /**
@@ -33,6 +42,9 @@ import java.util.function.Supplier;
  * features that are not (yet) supported by the backend.
  */
 class DdlExecutor {
+  private static final Statement START_BATCH_DDL = Statement.of("START BATCH DDL");
+
+  private static final Statement RUN_BATCH = Statement.of("RUN BATCH");
 
   private final BackendConnection backendConnection;
   private final Connection connection;
@@ -66,6 +78,43 @@ class DdlExecutor {
    * make the statement compatible with Cloud Spanner.
    */
   StatementResult execute(ParsedStatement parsedStatement, Statement statement) {
+    Tuple<ParsedStatement, Statement> replaced = replace(parsedStatement, statement);
+    parsedStatement = replaced.x();
+    statement = replaced.y();
+
+    if (!backendConnection.getSessionState().isSupportDropCascade()) {
+      return connection.execute(translate(parsedStatement, statement));
+    } else {
+      ImmutableList<Statement> allStatements =
+          getDependentStatements(translate(parsedStatement, statement));
+      if (allStatements.size() == 1) {
+        return connection.execute(allStatements.get(0));
+      } else {
+        boolean startedBatch = false;
+        try {
+          if (!connection.isDdlBatchActive()) {
+            connection.execute(START_BATCH_DDL);
+            startedBatch = true;
+          }
+          for (Statement dropDependency : allStatements) {
+            connection.execute(dropDependency);
+          }
+          if (startedBatch) {
+            connection.execute(RUN_BATCH);
+          }
+          return NO_RESULT;
+        } catch (Throwable t) {
+          if (startedBatch && connection.isDdlBatchActive()) {
+            connection.abortBatch();
+          }
+          throw t;
+        }
+      }
+    }
+  }
+
+  private Tuple<ParsedStatement, Statement> replace(
+      ParsedStatement parsedStatement, Statement statement) {
     if (!ddlStatementReplacements.get().isEmpty()) {
       String replaced = applyReplacers(statement.getSql());
       if (!replaced.equals(statement.getSql())) {
@@ -73,7 +122,7 @@ class DdlExecutor {
         parsedStatement = AbstractStatementParser.getInstance(Dialect.POSTGRESQL).parse(statement);
       }
     }
-    return connection.execute(translate(parsedStatement, statement));
+    return Tuple.of(parsedStatement, statement);
   }
 
   @VisibleForTesting
@@ -146,4 +195,297 @@ class DdlExecutor {
     }
     return createTableStatement;
   }
+
+  ImmutableList<Statement> getDependentStatements(Statement statement) {
+    ImmutableList<Statement> defaultResult = ImmutableList.of(statement);
+    SimpleParser parser = new SimpleParser(statement.getSql());
+    if (!parser.eatKeyword("drop")) {
+      return defaultResult;
+    }
+    parser.eatKeyword("if", "exists");
+    if (parser.eatKeyword("table")) {
+      TableOrIndexName tableName = parser.readTableOrIndexName();
+      if (tableName == null) {
+        return defaultResult;
+      }
+      boolean cascade = false;
+      String sqlWithoutCascade = parser.getSql().substring(0, parser.getPos());
+      if (parser.eatKeyword("cascade")) {
+        cascade = true;
+      } else {
+        // This is the default, so no need to register it specifically.
+        parser.eatKeyword("restrict");
+      }
+      if (parser.hasMoreTokens()) {
+        return defaultResult;
+      }
+      ImmutableList.Builder<Statement> builder = ImmutableList.builder();
+      builder.addAll(getDropDependentIndexesStatements(tableName));
+      if (cascade) {
+        // We don't have any way to get the views that depend on this table.
+        builder.addAll(getDropDependentForeignKeyConstraintsStatements(tableName));
+      }
+      builder.add(cascade ? Statement.of(sqlWithoutCascade) : statement);
+      return builder.build();
+    } else if (parser.eatKeyword("schema")) {
+      TableOrIndexName schemaName = parser.readTableOrIndexName();
+      if (schemaName == null || schemaName.schema != null) {
+        return defaultResult;
+      }
+      if (!parser.eatKeyword("cascade")) {
+        return defaultResult;
+      }
+      ImmutableList.Builder<Statement> builder = ImmutableList.builder();
+      builder.addAll(getDropSchemaIndexesStatements(schemaName));
+      builder.addAll(getDropSchemaForeignKeysStatements(schemaName));
+      builder.addAll(getDropSchemaViewsStatements(schemaName));
+      builder.addAll(getDropSchemaTablesStatements(schemaName));
+      return builder.build();
+    }
+    return defaultResult;
+  }
+
+  ImmutableList<Statement> getDropDependentIndexesStatements(TableOrIndexName tableName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select table_schema, index_name "
+                            + "from information_schema.indexes "
+                            + "where table_schema=$1 and table_name=$2 "
+                            + "and not index_type = 'PRIMARY_KEY' "
+                            + "and spanner_is_managed = 'NO'")
+                    .bind("p1")
+                    .to(
+                        unquoteIdentifier(
+                            tableName.schema == null
+                                ? backendConnection.getCurrentSchema()
+                                : tableName.schema))
+                    .bind("p2")
+                    .to(unquoteIdentifier(tableName.name))
+                    .build())) {
+      ImmutableList.Builder<Statement> dropStatements = ImmutableList.builder();
+      while (resultSet.next()) {
+        dropStatements.add(
+            Statement.of(
+                String.format(
+                    "drop index \"%s\".\"%s\"",
+                    resultSet.getString("table_schema"), resultSet.getString("index_name"))));
+      }
+      return dropStatements.build();
+    }
+  }
+
+  ImmutableList<Statement> getDropDependentForeignKeyConstraintsStatements(
+      TableOrIndexName tableName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select child.table_schema, child.table_name, rc.constraint_name\n"
+                            + "from information_schema.referential_constraints rc\n"
+                            + "inner join information_schema.table_constraints parent\n"
+                            + "  on  rc.unique_constraint_catalog=parent.constraint_catalog\n"
+                            + "  and rc.unique_constraint_schema=parent.constraint_schema\n"
+                            + "  and rc.unique_constraint_name=parent.constraint_name\n"
+                            + "inner join information_schema.table_constraints child\n"
+                            + "  on  rc.constraint_catalog=child.constraint_catalog\n"
+                            + "  and rc.constraint_schema=child.constraint_schema\n"
+                            + "  and rc.constraint_name=child.constraint_name\n"
+                            + "where parent.table_schema=$1\n"
+                            + "and parent.table_name=$2")
+                    .bind("p1")
+                    .to(
+                        unquoteIdentifier(
+                            tableName.schema == null
+                                ? backendConnection.getCurrentSchema()
+                                : tableName.schema))
+                    .bind("p2")
+                    .to(unquoteIdentifier(tableName.name))
+                    .build())) {
+      ImmutableList.Builder<Statement> dropStatements = ImmutableList.builder();
+      while (resultSet.next()) {
+        dropStatements.add(
+            Statement.of(
+                String.format(
+                    "alter table \"%s\".\"%s\" drop constraint \"%s\"",
+                    resultSet.getString("table_schema"),
+                    resultSet.getString("table_name"),
+                    resultSet.getString("constraint_name"))));
+      }
+      return dropStatements.build();
+    }
+  }
+
+  ImmutableList<Statement> getDropSchemaForeignKeysStatements(TableOrIndexName schemaName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select child.table_schema, child.table_name, rc.constraint_name\n"
+                            + "from information_schema.referential_constraints rc\n"
+                            + "inner join information_schema.table_constraints parent\n"
+                            + "  on  rc.unique_constraint_catalog=parent.constraint_catalog\n"
+                            + "  and rc.unique_constraint_schema=parent.constraint_schema\n"
+                            + "  and rc.unique_constraint_name=parent.constraint_name\n"
+                            + "inner join information_schema.table_constraints child\n"
+                            + "  on  rc.constraint_catalog=child.constraint_catalog\n"
+                            + "  and rc.constraint_schema=child.constraint_schema\n"
+                            + "  and rc.constraint_name=child.constraint_name\n"
+                            + "where parent.table_schema=$1")
+                    .bind("p1")
+                    .to(unquoteIdentifier(schemaName.name))
+                    .build())) {
+      ImmutableList.Builder<Statement> dropStatements = ImmutableList.builder();
+      while (resultSet.next()) {
+        dropStatements.add(
+            Statement.of(
+                String.format(
+                    "alter table \"%s\".\"%s\" drop constraint \"%s\"",
+                    resultSet.getString("table_schema"),
+                    resultSet.getString("table_name"),
+                    resultSet.getString("constraint_name"))));
+      }
+      return dropStatements.build();
+    }
+  }
+
+  ImmutableList<Statement> getDropSchemaIndexesStatements(TableOrIndexName schemaName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select table_schema, index_name "
+                            + "from information_schema.indexes "
+                            + "where table_schema=$1 "
+                            + "and not index_type = 'PRIMARY_KEY' "
+                            + "and spanner_is_managed = 'NO'")
+                    .bind("p1")
+                    .to(unquoteIdentifier(schemaName.name))
+                    .build())) {
+      ImmutableList.Builder<Statement> dropStatements = ImmutableList.builder();
+      while (resultSet.next()) {
+        dropStatements.add(
+            Statement.of(
+                String.format(
+                    "drop index \"%s\".\"%s\"",
+                    resultSet.getString("table_schema"), resultSet.getString("index_name"))));
+      }
+      return dropStatements.build();
+    }
+  }
+
+  private static final class Table implements Comparable<Table> {
+    private final String schema;
+    private final String name;
+    private final String parent;
+
+    Table(String schema, String name, String parent) {
+      this.schema = Preconditions.checkNotNull(schema);
+      this.name = Preconditions.checkNotNull(name);
+      this.parent = parent;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(this.schema, this.name);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+      if (!(o instanceof Table)) {
+        return false;
+      }
+      Table other = (Table) o;
+      return Objects.equals(this.schema, other.schema) && Objects.equals(this.name, other.name);
+    }
+
+    @Override
+    public int compareTo(Table o) {
+      if (o.parent != null && Objects.equals(this.name, o.parent)) {
+        return 1;
+      }
+      if (this.parent != null && Objects.equals(this.parent, o.name)) {
+        return -1;
+      }
+      if (o.parent != null && this.parent == null) {
+        return 1;
+      }
+      if (this.parent != null && o.parent == null) {
+        return -1;
+      }
+      if (!Objects.equals(this.schema, o.schema)) {
+        return this.schema.compareTo(o.schema);
+      }
+      return this.name.compareTo(o.name);
+    }
+  }
+
+  ImmutableList<Statement> getDropSchemaViewsStatements(TableOrIndexName schemaName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select table_schema, table_name "
+                            + "from information_schema.tables "
+                            + "where table_schema=$1 "
+                            + "and table_type='VIEW'")
+                    .bind("p1")
+                    .to(unquoteIdentifier(schemaName.name))
+                    .build())) {
+      ImmutableList.Builder<Statement> dropStatements = ImmutableList.builder();
+      while (resultSet.next()) {
+        dropStatements.add(
+            Statement.of(
+                String.format(
+                    "drop view \"%s\".\"%s\"",
+                    resultSet.getString("table_schema"), resultSet.getString("table_name"))));
+      }
+      return dropStatements.build();
+    }
+  }
+
+  ImmutableList<Statement> getDropSchemaTablesStatements(TableOrIndexName schemaName) {
+    DatabaseClient client = connection.getDatabaseClient();
+    try (ResultSet resultSet =
+        client
+            .singleUse()
+            .executeQuery(
+                Statement.newBuilder(
+                        "select table_schema, table_name, parent_table_name "
+                            + "from information_schema.tables "
+                            + "where table_schema=$1 "
+                            + "and table_type='BASE TABLE'")
+                    .bind("p1")
+                    .to(unquoteIdentifier(schemaName.name))
+                    .build())) {
+      List<Table> tables = new ArrayList<>();
+      while (resultSet.next()) {
+        tables.add(
+            new Table(
+                resultSet.getString("table_schema"),
+                resultSet.getString("table_name"),
+                resultSet.isNull("parent_table_name")
+                    ? null
+                    : resultSet.getString("parent_table_name")));
+      }
+      Collections.sort(tables);
+      ImmutableList.Builder<Statement> dropStatements = ImmutableList.builder();
+      for (Table table : tables) {
+        dropStatements.add(
+            Statement.of(String.format("drop table \"%s\".\"%s\"", table.schema, table.name)));
+      }
+      return dropStatements.build();
+    }
+  }
 }


Test patch:
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/DropCascadeMockServerTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/DropCascadeMockServerTest.java
new file mode 100644
index 00000000..ce04f512
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/DropCascadeMockServerTest.java
@@ -0,0 +1,771 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.spanner.MockSpannerServiceImpl.StatementResult;
+import com.google.common.collect.ImmutableList;
+import com.google.protobuf.ListValue;
+import com.google.protobuf.NullValue;
+import com.google.protobuf.Value;
+import com.google.spanner.admin.database.v1.UpdateDatabaseDdlRequest;
+import com.google.spanner.v1.TypeCode;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.postgresql.util.PSQLException;
+
+@RunWith(JUnit4.class)
+public class DropCascadeMockServerTest extends AbstractMockServerTest {
+
+  @BeforeClass
+  public static void loadPgJdbcDriver() throws Exception {
+    // Make sure the PG JDBC driver is loaded.
+    Class.forName("org.postgresql.Driver");
+  }
+
+  protected String createUrl() {
+    return String.format(
+        "jdbc:postgresql://localhost:%d/?options=-c%%20spanner.support_drop_cascade=true",
+        pgServer.getLocalPort());
+  }
+
+  @Test
+  public void testDropTableStatementWithoutIndexesAndForeignKeys() throws SQLException {
+    String sql = "DROP TABLE foo";
+    addDdlResponseToSpannerAdmin();
+
+    addTableIndexesResult(new TableName("public", "foo"), ImmutableList.of());
+    addTableForeignKeysResult(new TableName("public", "foo"), ImmutableList.of());
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (ResultSet resultSet =
+          connection.createStatement().executeQuery("show spanner.support_drop_cascade")) {
+        assertTrue(resultSet.next());
+        assertTrue(resultSet.getBoolean(1));
+        assertFalse(resultSet.next());
+      }
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, 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(sql, updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testDropTableStatementWithIndex() throws SQLException {
+    String sql = "DROP TABLE foo";
+    addDdlResponseToSpannerAdmin();
+
+    addTableIndexesResult(
+        new TableName("public", "foo"), ImmutableList.of(new IndexName("public", "idx_foo")));
+    addTableForeignKeysResult(new TableName("public", "foo"), ImmutableList.of());
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, 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(2, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "drop index \"public\".\"idx_foo\"", updateDatabaseDdlRequests.get(0).getStatements(0));
+    assertEquals(sql, updateDatabaseDdlRequests.get(0).getStatements(1));
+  }
+
+  @Test
+  public void testDropTableCascadeStatementWithForeignKey() throws SQLException {
+    String sql = "DROP TABLE foo cascade";
+    addDdlResponseToSpannerAdmin();
+
+    addTableIndexesResult(new TableName("public", "foo"), ImmutableList.of());
+    addTableForeignKeysResult(
+        new TableName("public", "foo"),
+        ImmutableList.of(new ForeignKey(new TableName("public", "bar"), "fk_foo_bar")));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, 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(2, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "alter table \"public\".\"bar\" drop constraint \"fk_foo_bar\"",
+        updateDatabaseDdlRequests.get(0).getStatements(0));
+    assertEquals("DROP TABLE foo", updateDatabaseDdlRequests.get(0).getStatements(1));
+  }
+
+  @Test
+  public void testDropTableCascadeStatementWithIndexAndForeignKey() throws SQLException {
+    String sql = "DROP TABLE foo cascade";
+    addDdlResponseToSpannerAdmin();
+
+    addTableIndexesResult(
+        new TableName("public", "foo"), ImmutableList.of(new IndexName("public", "idx_foo")));
+    addTableForeignKeysResult(
+        new TableName("public", "foo"),
+        ImmutableList.of(new ForeignKey(new TableName("public", "bar"), "fk_foo_bar")));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, 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(3, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "drop index \"public\".\"idx_foo\"", updateDatabaseDdlRequests.get(0).getStatements(0));
+    assertEquals(
+        "alter table \"public\".\"bar\" drop constraint \"fk_foo_bar\"",
+        updateDatabaseDdlRequests.get(0).getStatements(1));
+    assertEquals("DROP TABLE foo", updateDatabaseDdlRequests.get(0).getStatements(2));
+  }
+
+  @Test
+  public void testDropTableStatementWithIndexAndForeignKey() throws SQLException {
+    // DROP TABLE <table> [RESTRICT] removes all indexes but not foreign key constraints.
+    for (String sql : new String[] {"DROP TABLE foo", "DROP TABLE foo RESTRICT"}) {
+      addDdlResponseToSpannerAdmin();
+
+      addTableIndexesResult(
+          new TableName("public", "foo"), ImmutableList.of(new IndexName("public", "idx_foo")));
+      addTableForeignKeysResult(
+          new TableName("public", "foo"),
+          ImmutableList.of(new ForeignKey(new TableName("public", "bar"), "fk_foo_bar")));
+
+      try (Connection connection = DriverManager.getConnection(createUrl())) {
+        try (Statement statement = connection.createStatement()) {
+          assertFalse(statement.execute(sql));
+          assertEquals(0, statement.getUpdateCount());
+          assertFalse(statement.getMoreResults());
+          assertEquals(-1, 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(2, updateDatabaseDdlRequests.get(0).getStatementsCount());
+      assertEquals(
+          "drop index \"public\".\"idx_foo\"", updateDatabaseDdlRequests.get(0).getStatements(0));
+      assertEquals(sql, updateDatabaseDdlRequests.get(0).getStatements(1));
+      mockDatabaseAdmin.getRequests().clear();
+    }
+  }
+
+  @Test
+  public void testDropTableCascadeStatementWithSchemaAndIndexAndForeignKey() throws SQLException {
+    String sql = "DROP TABLE my_schema.foo cascade";
+    addDdlResponseToSpannerAdmin();
+
+    addTableIndexesResult(
+        new TableName("my_schema", "foo"), ImmutableList.of(new IndexName("my_schema", "idx_foo")));
+    addTableForeignKeysResult(
+        new TableName("my_schema", "foo"),
+        ImmutableList.of(new ForeignKey(new TableName("my_schema", "bar"), "fk_foo_bar")));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, 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(3, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals(
+        "drop index \"my_schema\".\"idx_foo\"", updateDatabaseDdlRequests.get(0).getStatements(0));
+    assertEquals(
+        "alter table \"my_schema\".\"bar\" drop constraint \"fk_foo_bar\"",
+        updateDatabaseDdlRequests.get(0).getStatements(1));
+    assertEquals("DROP TABLE my_schema.foo", updateDatabaseDdlRequests.get(0).getStatements(2));
+  }
+
+  @Test
+  public void testDropTableCascadeStatementWithIndexAndForeignKeyInBatch() throws SQLException {
+    String sql = "DROP TABLE foo cascade";
+    addDdlResponseToSpannerAdmin();
+
+    addTableIndexesResult(new TableName("public", "baz"), ImmutableList.of());
+    addTableForeignKeysResult(new TableName("public", "baz"), ImmutableList.of());
+    addTableIndexesResult(
+        new TableName("public", "foo"), ImmutableList.of(new IndexName("public", "idx_foo")));
+    addTableForeignKeysResult(
+        new TableName("public", "foo"),
+        ImmutableList.of(new ForeignKey(new TableName("public", "bar"), "fk_foo_bar")));
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        statement.addBatch("drop table baz");
+        statement.addBatch(sql);
+        int[] result = statement.executeBatch();
+        assertArrayEquals(new int[] {0, 0}, result);
+      }
+    }
+
+    List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+        mockDatabaseAdmin.getRequests().stream()
+            .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+            .map(UpdateDatabaseDdlRequest.class::cast)
+            .collect(Collectors.toList());
+    assertEquals(1, updateDatabaseDdlRequests.size());
+    assertEquals(4, updateDatabaseDdlRequests.get(0).getStatementsCount());
+    assertEquals("drop table baz", updateDatabaseDdlRequests.get(0).getStatements(0));
+    assertEquals(
+        "drop index \"public\".\"idx_foo\"", updateDatabaseDdlRequests.get(0).getStatements(1));
+    assertEquals(
+        "alter table \"public\".\"bar\" drop constraint \"fk_foo_bar\"",
+        updateDatabaseDdlRequests.get(0).getStatements(2));
+    assertEquals("DROP TABLE foo", updateDatabaseDdlRequests.get(0).getStatements(3));
+  }
+
+  @Test
+  public void testDropSchemaCascade() throws SQLException {
+    String sql = "DROP SCHEMA public cascade ";
+    for (boolean fail : new boolean[] {true, false}) {
+      mockDatabaseAdmin.getRequests().clear();
+      if (fail) {
+        addDdlExceptionToSpannerAdmin();
+      } else {
+        addDdlResponseToSpannerAdmin();
+      }
+
+      addSchemaIndexesResult(
+          "public",
+          ImmutableList.of(
+              new IndexName("public", "idx_albums_title"),
+              new IndexName("public", "idx_concerts_start_time"),
+              new IndexName("public", "idx_singers_last_name")));
+      addSchemaForeignKeysResult(
+          "public",
+          ImmutableList.of(
+              new ForeignKey(new TableName("public", "albums"), "fk_albums_singers"),
+              new ForeignKey(new TableName("public", "concerts"), "fk_concerts_singers"),
+              new ForeignKey(new TableName("public", "concerts"), "fk_concerts_venues")));
+      addSchemaViewsResult(
+          "public",
+          ImmutableList.of(
+              new TableName("public", "v_albums"),
+              new TableName("public", "v_singers"),
+              new TableName("public", "v_tracks")));
+
+      Random random = new Random();
+      //noinspection ComparatorMethodParameterNotUsed
+      addSchemaTablesResult(
+          "public",
+          ImmutableList.copyOf(
+              ImmutableList.of(
+                      new Table(new TableName("public", "albums"), "singers"),
+                      new Table(new TableName("public", "concerts"), null),
+                      new Table(new TableName("public", "singers"), null),
+                      new Table(new TableName("public", "tracks"), "albums"),
+                      new Table(new TableName("public", "venues"), null))
+                  .stream()
+                  .sorted((t1, t2) -> (random.nextInt(3) - 1))
+                  .collect(Collectors.toList())));
+
+      try (Connection connection = DriverManager.getConnection(createUrl())) {
+        try (Statement statement = connection.createStatement()) {
+          if (fail) {
+            assertThrows(PSQLException.class, () -> statement.execute(sql));
+            // Verify that the connection is usable after this error.
+            try (ResultSet resultSet =
+                connection.createStatement().executeQuery(SELECT1.getSql())) {
+              assertTrue(resultSet.next());
+              assertEquals(1L, resultSet.getLong(1));
+              assertFalse(resultSet.next());
+            }
+          } else {
+            assertFalse(statement.execute(sql));
+            assertEquals(0, statement.getUpdateCount());
+            assertFalse(statement.getMoreResults());
+            assertEquals(-1, statement.getUpdateCount());
+          }
+        }
+      }
+
+      List<UpdateDatabaseDdlRequest> updateDatabaseDdlRequests =
+          mockDatabaseAdmin.getRequests().stream()
+              .filter(request -> request instanceof UpdateDatabaseDdlRequest)
+              .map(UpdateDatabaseDdlRequest.class::cast)
+              .collect(Collectors.toList());
+      if (fail) {
+        assertEquals(0, updateDatabaseDdlRequests.size());
+      } else {
+        assertEquals(1, updateDatabaseDdlRequests.size());
+        assertEquals(14, updateDatabaseDdlRequests.get(0).getStatementsCount());
+        int index = -1;
+        assertEquals(
+            "drop index \"public\".\"idx_albums_title\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop index \"public\".\"idx_concerts_start_time\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop index \"public\".\"idx_singers_last_name\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "alter table \"public\".\"albums\" drop constraint \"fk_albums_singers\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "alter table \"public\".\"concerts\" drop constraint \"fk_concerts_singers\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "alter table \"public\".\"concerts\" drop constraint \"fk_concerts_venues\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop view \"public\".\"v_albums\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop view \"public\".\"v_singers\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop view \"public\".\"v_tracks\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop table \"public\".\"tracks\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop table \"public\".\"albums\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop table \"public\".\"concerts\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop table \"public\".\"singers\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+        assertEquals(
+            "drop table \"public\".\"venues\"",
+            updateDatabaseDdlRequests.get(0).getStatements(++index));
+      }
+    }
+  }
+
+  @Test
+  public void testDropSchema() throws SQLException {
+    for (String sql : new String[] {"DROP SCHEMA public", "drop schema public restrict"}) {
+      addDdlResponseToSpannerAdmin();
+
+      try (Connection connection = DriverManager.getConnection(createUrl())) {
+        try (Statement statement = connection.createStatement()) {
+          assertFalse(statement.execute(sql));
+          assertEquals(0, statement.getUpdateCount());
+          assertFalse(statement.getMoreResults());
+          assertEquals(-1, 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());
+      // DROP SCHEMA without a cascade should be a pass-through.
+      assertEquals(sql, updateDatabaseDdlRequests.get(0).getStatements(0));
+      mockDatabaseAdmin.getRequests().clear();
+    }
+  }
+
+  @Test
+  public void testDropSchemaRestrict() throws SQLException {
+    String sql = "DROP SCHEMA public restrict";
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, 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());
+    // DROP SCHEMA ... RESTRICT should be a pass-through.
+    assertEquals(sql, updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  @Test
+  public void testCreateTable() throws SQLException {
+    String sql = "CREATE TABLE foo (id bigint primary key)";
+    addDdlResponseToSpannerAdmin();
+
+    try (Connection connection = DriverManager.getConnection(createUrl())) {
+      try (Statement statement = connection.createStatement()) {
+        assertFalse(statement.execute(sql));
+        assertEquals(0, statement.getUpdateCount());
+        assertFalse(statement.getMoreResults());
+        assertEquals(-1, 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());
+    // CREATE TABLE should be a pass-through.
+    assertEquals(sql, updateDatabaseDdlRequests.get(0).getStatements(0));
+  }
+
+  private static final class Table {
+    final TableName tableName;
+    final String parent_table_name;
+
+    Table(TableName tableName, String parent_table_name) {
+      this.tableName = tableName;
+      this.parent_table_name = parent_table_name;
+    }
+  }
+
+  private static class TableName {
+    final String table_schema;
+    final String table_name;
+
+    TableName(String table_schema, String table_name) {
+      this.table_schema = table_schema;
+      this.table_name = table_name;
+    }
+  }
+
+  private static class IndexName {
+    final String table_schema;
+    final String index_name;
+
+    IndexName(String table_schema, String index_name) {
+      this.table_schema = table_schema;
+      this.index_name = index_name;
+    }
+  }
+
+  private static class ForeignKey {
+    final TableName tableName;
+    final String constraint_name;
+
+    ForeignKey(TableName tableName, String constraint_name) {
+      this.tableName = tableName;
+      this.constraint_name = constraint_name;
+    }
+  }
+
+  private static void addSchemaViewsResult(String schemaName, ImmutableList<TableName> views) {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            com.google.cloud.spanner.Statement.newBuilder(
+                    "select table_schema, table_name "
+                        + "from information_schema.tables "
+                        + "where table_schema=$1 "
+                        + "and table_type='VIEW'")
+                .bind("p1")
+                .to(schemaName)
+                .build(),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(TypeCode.STRING, TypeCode.STRING),
+                        ImmutableList.of("table_schema", "table_name")))
+                .addAllRows(
+                    views.stream()
+                        .map(
+                            view ->
+                                ListValue.newBuilder()
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(view.table_schema)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder().setStringValue(view.table_name).build())
+                                    .build())
+                        .collect(Collectors.toList()))
+                .build()));
+  }
+
+  private static void addSchemaTablesResult(String schemaName, ImmutableList<Table> tables) {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            com.google.cloud.spanner.Statement.newBuilder(
+                    "select table_schema, table_name, parent_table_name "
+                        + "from information_schema.tables "
+                        + "where table_schema=$1 "
+                        + "and table_type='BASE TABLE'")
+                .bind("p1")
+                .to(schemaName)
+                .build(),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(TypeCode.STRING, TypeCode.STRING, TypeCode.STRING),
+                        ImmutableList.of("table_schema", "table_name", "parent_table_name")))
+                .addAllRows(
+                    tables.stream()
+                        .map(
+                            table ->
+                                ListValue.newBuilder()
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(table.tableName.table_schema)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(table.tableName.table_name)
+                                            .build())
+                                    .addValues(
+                                        table.parent_table_name == null
+                                            ? Value.newBuilder()
+                                                .setNullValue(NullValue.NULL_VALUE)
+                                                .build()
+                                            : Value.newBuilder()
+                                                .setStringValue(table.parent_table_name)
+                                                .build())
+                                    .build())
+                        .collect(Collectors.toList()))
+                .build()));
+  }
+
+  private static void addTableIndexesResult(TableName tableName, ImmutableList<IndexName> indexes) {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            com.google.cloud.spanner.Statement.newBuilder(
+                    "select table_schema, index_name from information_schema.indexes "
+                        + "where table_schema=$1 "
+                        + "and table_name=$2 "
+                        + "and not index_type = 'PRIMARY_KEY' "
+                        + "and spanner_is_managed = 'NO'")
+                .bind("p1")
+                .to(tableName.table_schema)
+                .bind("p2")
+                .to(tableName.table_name)
+                .build(),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(TypeCode.STRING, TypeCode.STRING),
+                        ImmutableList.of("table_schema", "index_name")))
+                .addAllRows(
+                    indexes.stream()
+                        .map(
+                            index ->
+                                ListValue.newBuilder()
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(index.table_schema)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder().setStringValue(index.index_name).build())
+                                    .build())
+                        .collect(Collectors.toList()))
+                .build()));
+  }
+
+  private static void addSchemaIndexesResult(String schemaName, ImmutableList<IndexName> indexes) {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            com.google.cloud.spanner.Statement.newBuilder(
+                    "select table_schema, index_name from information_schema.indexes "
+                        + "where table_schema=$1 "
+                        + "and not index_type = 'PRIMARY_KEY' "
+                        + "and spanner_is_managed = 'NO'")
+                .bind("p1")
+                .to(schemaName)
+                .build(),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(TypeCode.STRING, TypeCode.STRING),
+                        ImmutableList.of("table_schema", "index_name")))
+                .addAllRows(
+                    indexes.stream()
+                        .map(
+                            index ->
+                                ListValue.newBuilder()
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(index.table_schema)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder().setStringValue(index.index_name).build())
+                                    .build())
+                        .collect(Collectors.toList()))
+                .build()));
+  }
+
+  private static void addTableForeignKeysResult(
+      TableName tableName, ImmutableList<ForeignKey> foreignKeys) {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            com.google.cloud.spanner.Statement.newBuilder(
+                    "select child.table_schema, child.table_name, rc.constraint_name\n"
+                        + "from information_schema.referential_constraints rc\n"
+                        + "inner join information_schema.table_constraints parent\n"
+                        + "  on  rc.unique_constraint_catalog=parent.constraint_catalog\n"
+                        + "  and rc.unique_constraint_schema=parent.constraint_schema\n"
+                        + "  and rc.unique_constraint_name=parent.constraint_name\n"
+                        + "inner join information_schema.table_constraints child\n"
+                        + "  on  rc.constraint_catalog=child.constraint_catalog\n"
+                        + "  and rc.constraint_schema=child.constraint_schema\n"
+                        + "  and rc.constraint_name=child.constraint_name\n"
+                        + "where parent.table_schema=$1\n"
+                        + "and parent.table_name=$2")
+                .bind("p1")
+                .to(tableName.table_schema)
+                .bind("p2")
+                .to(tableName.table_name)
+                .build(),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(TypeCode.STRING, TypeCode.STRING, TypeCode.STRING),
+                        ImmutableList.of("table_schema", "table_name", "constraint_name")))
+                .addAllRows(
+                    foreignKeys.stream()
+                        .map(
+                            foreignKey ->
+                                ListValue.newBuilder()
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(foreignKey.tableName.table_schema)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(foreignKey.tableName.table_name)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(foreignKey.constraint_name)
+                                            .build())
+                                    .build())
+                        .collect(Collectors.toList()))
+                .build()));
+  }
+
+  private static void addSchemaForeignKeysResult(
+      String schemaName, ImmutableList<ForeignKey> foreignKeys) {
+    mockSpanner.putStatementResult(
+        StatementResult.query(
+            com.google.cloud.spanner.Statement.newBuilder(
+                    "select child.table_schema, child.table_name, rc.constraint_name\n"
+                        + "from information_schema.referential_constraints rc\n"
+                        + "inner join information_schema.table_constraints parent\n"
+                        + "  on  rc.unique_constraint_catalog=parent.constraint_catalog\n"
+                        + "  and rc.unique_constraint_schema=parent.constraint_schema\n"
+                        + "  and rc.unique_constraint_name=parent.constraint_name\n"
+                        + "inner join information_schema.table_constraints child\n"
+                        + "  on  rc.constraint_catalog=child.constraint_catalog\n"
+                        + "  and rc.constraint_schema=child.constraint_schema\n"
+                        + "  and rc.constraint_name=child.constraint_name\n"
+                        + "where parent.table_schema=$1")
+                .bind("p1")
+                .to(schemaName)
+                .build(),
+            com.google.spanner.v1.ResultSet.newBuilder()
+                .setMetadata(
+                    createMetadata(
+                        ImmutableList.of(TypeCode.STRING, TypeCode.STRING, TypeCode.STRING),
+                        ImmutableList.of("table_schema", "table_name", "constraint_name")))
+                .addAllRows(
+                    foreignKeys.stream()
+                        .map(
+                            foreignKey ->
+                                ListValue.newBuilder()
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(foreignKey.tableName.table_schema)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(foreignKey.tableName.table_name)
+                                            .build())
+                                    .addValues(
+                                        Value.newBuilder()
+                                            .setStringValue(foreignKey.constraint_name)
+                                            .build())
+                                    .build())
+                        .collect(Collectors.toList()))
+                .build()));
+  }
+}
diff --git a/src/test/java/com/google/cloud/spanner/pgadapter/ITDdlTest.java b/src/test/java/com/google/cloud/spanner/pgadapter/ITDdlTest.java
new file mode 100644
index 00000000..0096ddf2
--- /dev/null
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/ITDdlTest.java
@@ -0,0 +1,231 @@
+// Copyright 2023 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package com.google.cloud.spanner.pgadapter;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import com.google.cloud.spanner.Database;
+import com.google.cloud.spanner.pgadapter.error.SQLState;
+import com.google.common.collect.ImmutableList;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.postgresql.util.PSQLException;
+
+@Category(IntegrationTest.class)
+@RunWith(JUnit4.class)
+public class ITDdlTest implements IntegrationTest {
+  private static final PgAdapterTestEnv testEnv = new PgAdapterTestEnv();
+
+  @BeforeClass
+  public static void setup() {
+    testEnv.setUp();
+    Database database = testEnv.createDatabase(ImmutableList.of());
+    testEnv.startPGAdapterServerWithDefaultDatabase(database.getId(), ImmutableList.of());
+  }
+
+  @AfterClass
+  public static void teardown() {
+    testEnv.stopPGAdapterServer();
+    testEnv.cleanUp();
+  }
+
+  private String getConnectionUrl() {
+    return String.format("jdbc:postgresql://%s/", testEnv.getPGAdapterHostAndPort());
+  }
+
+  @Test
+  public void testDropTableWithIndex() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      // First create a table with a secondary index.
+      try (Statement statement = connection.createStatement()) {
+        statement.addBatch("create table test (id bigint primary key, value varchar)");
+        statement.addBatch("create index idx_text on test (value)");
+        assertArrayEquals(new int[] {0, 0}, statement.executeBatch());
+      }
+
+      // Try to drop the table with an index. This will fail.
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class, () -> connection.createStatement().execute("drop table test"));
+      assertEquals(SQLState.FeatureNotSupported.toString(), exception.getSQLState());
+      assertNotNull(exception.getServerErrorMessage());
+      assertEquals(
+          "Execute 'set spanner.support_drop_cascade=true' to enable dropping tables with indices",
+          exception.getServerErrorMessage().getHint());
+
+      // Now enable drop_cascade.
+      connection.createStatement().execute("set spanner.support_drop_cascade to on");
+      // Dropping the table should now work.
+      assertEquals(0, connection.createStatement().executeUpdate("drop table test"));
+    }
+  }
+
+  @Test
+  public void testDropTableWithForeignKey() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      // First create a table with a secondary index.
+      try (Statement statement = connection.createStatement()) {
+        statement.addBatch("create table parent (id bigint primary key, value varchar)");
+        statement.addBatch(
+            "create table child (id bigint primary key, parent_id bigint, constraint fk_child_parent foreign key (parent_id) references parent (id))");
+        assertArrayEquals(new int[] {0, 0}, statement.executeBatch());
+      }
+
+      // Try to drop the table that is referenced by a foreign key. This will fail.
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class, () -> connection.createStatement().execute("drop table parent"));
+      String expectedMessage =
+          "Cannot drop table `parent`. It is referenced by one or more foreign keys: `fk_child_parent`. You must drop the foreign keys before dropping the table.";
+      assertNotNull(exception.getServerErrorMessage());
+      assertNotNull(exception.getServerErrorMessage().getMessage());
+      assertTrue(
+          exception.getServerErrorMessage().getMessage(),
+          exception.getServerErrorMessage().getMessage().endsWith(expectedMessage));
+
+      exception =
+          assertThrows(
+              PSQLException.class,
+              () -> connection.createStatement().execute("drop table parent cascade"));
+      System.out.println(exception.getErrorCode());
+      System.out.println(exception.getMessage());
+      assertEquals(SQLState.FeatureNotSupported.toString(), exception.getSQLState());
+      assertNotNull(exception.getServerErrorMessage());
+      assertEquals(
+          "Execute 'set spanner.support_drop_cascade=true' to enable 'drop {table|schema} cascade' statements.",
+          exception.getServerErrorMessage().getHint());
+
+      // Now enable drop_cascade.
+      connection.createStatement().execute("set spanner.support_drop_cascade to on");
+      // Dropping the table should now work.
+      assertEquals(0, connection.createStatement().executeUpdate("drop table parent cascade"));
+      // Drop the child table as well to clean up.
+      assertEquals(0, connection.createStatement().executeUpdate("drop table child cascade"));
+    }
+  }
+
+  @Test
+  public void testDropSchema() throws SQLException {
+    try (Connection connection = DriverManager.getConnection(getConnectionUrl())) {
+      // Create a set of tables, views, indexes etc.
+      try (Statement statement = connection.createStatement()) {
+        statement.addBatch(
+            "create table if not exists singers (\n"
+                + "    id         varchar not null primary key,\n"
+                + "    first_name varchar,\n"
+                + "    last_name  varchar not null,\n"
+                + "    full_name  varchar generated always as (coalesce(concat(first_name, ' '::varchar, last_name), last_name)) stored,\n"
+                + "    active     boolean,\n"
+                + "    created_at timestamptz,\n"
+                + "    updated_at timestamptz\n"
+                + ");");
+        statement.addBatch(
+            "create table if not exists albums (\n"
+                + "    id               varchar not null primary key,\n"
+                + "    title            varchar not null,\n"
+                + "    marketing_budget numeric,\n"
+                + "    release_date     date,\n"
+                + "    cover_picture    bytea,\n"
+                + "    singer_id        varchar not null,\n"
+                + "    created_at       timestamptz,\n"
+                + "    updated_at       timestamptz,\n"
+                + "    constraint fk_albums_singers foreign key (singer_id) references singers (id)\n"
+                + ");");
+        statement.addBatch(
+            "create table if not exists tracks (\n"
+                + "    id           varchar not null,\n"
+                + "    track_number bigint not null,\n"
+                + "    title        varchar not null,\n"
+                + "    sample_rate  float8 not null,\n"
+                + "    created_at   timestamptz,\n"
+                + "    updated_at   timestamptz,\n"
+                + "    primary key (id, track_number)\n"
+                + ") interleave in parent albums on delete cascade;");
+        statement.addBatch(
+            "create table if not exists venues (\n"
+                + "    id          varchar not null primary key,\n"
+                + "    name        varchar not null,\n"
+                + "    description varchar not null,\n"
+                + "    created_at  timestamptz,\n"
+                + "    updated_at  timestamptz\n"
+                + ");");
+        statement.addBatch(
+            "create table if not exists concerts (\n"
+                + "    id          varchar not null primary key,\n"
+                + "    venue_id    varchar not null,\n"
+                + "    singer_id   varchar not null,\n"
+                + "    name        varchar not null,\n"
+                + "    start_time  timestamptz not null,\n"
+                + "    end_time    timestamptz not null,\n"
+                + "    created_at  timestamptz,\n"
+                + "    updated_at  timestamptz,\n"
+                + "    constraint fk_concerts_venues  foreign key (venue_id)  references venues  (id),\n"
+                + "    constraint fk_concerts_singers foreign key (singer_id) references singers (id),\n"
+                + "    constraint chk_end_time_after_start_time check (end_time > start_time)\n"
+                + ");");
+        statement.addBatch("create index idx_concerts_start_time on concerts (start_time)");
+        statement.addBatch("create unique index idx_tracks_title on tracks (id, title)");
+        statement.addBatch(
+            "create view v_singers sql security invoker as select first_name, last_name from singers order by last_name");
+        statement.addBatch(
+            "create view v_venues sql security invoker as select name, description from venues order by name");
+
+        assertArrayEquals(new int[] {0, 0, 0, 0, 0, 0, 0, 0, 0}, statement.executeBatch());
+      }
+
+      // Try to drop the schema.
+      PSQLException exception =
+          assertThrows(
+              PSQLException.class,
+              () -> connection.createStatement().execute("drop schema public cascade"));
+      System.out.println(exception.getMessage());
+      assertEquals(SQLState.FeatureNotSupported.toString(), exception.getSQLState());
+      assertNotNull(exception.getServerErrorMessage());
+      assertEquals(
+          "Execute 'set spanner.support_drop_cascade=true' to enable 'drop {table|schema} cascade' statements.",
+          exception.getServerErrorMessage().getHint());
+
+      // Now enable drop_cascade.
+      connection.createStatement().execute("set spanner.support_drop_cascade to on");
+      // Dropping the schema should now work. This will not actually drop the schema, but it will
+      // drop all objects in the schema.
+      assertEquals(0, connection.createStatement().executeUpdate("drop schema public cascade"));
+
+      try (ResultSet tablesCount =
+          connection
+              .createStatement()
+              .executeQuery(
+                  "select count(1) from information_schema.tables where table_schema='public'")) {
+        assertTrue(tablesCount.next());
+        assertEquals(0L, tablesCount.getLong(1));
+        assertFalse(tablesCount.next());
+      }
+    }
+  }
+}
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
index dde07792..e14b126c 100644
--- a/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java
+++ b/src/test/java/com/google/cloud/spanner/pgadapter/statements/DdlExecutorTest.java
@@ -18,10 +18,13 @@ import static com.google.cloud.spanner.pgadapter.statements.DdlExecutor.unquoteI
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertSame;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import com.google.cloud.spanner.Dialect;
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.connection.AbstractStatementParser;
+import com.google.cloud.spanner.pgadapter.session.SessionState;
+import com.google.cloud.spanner.pgadapter.statements.SimpleParser.TableOrIndexName;
 import com.google.cloud.spanner.pgadapter.utils.RegexQueryPartReplacer;
 import com.google.common.base.Suppliers;
 import com.google.common.collect.ImmutableList;
@@ -114,6 +117,101 @@ public class DdlExecutorTest {
                 "CREATE TABLE \"user\" (\"id\" integer NOT NULL, \"firstName\" character varying NOT NULL, \"lastName\" character varying NOT NULL, \"age\" integer NOT NULL, CONSTRAINT \"PK_user\" PRIMARY KEY (\"id\"))")));
   }
 
+  @Test
+  public void testGetDependentStatements() {
+    BackendConnection connection = mock(BackendConnection.class);
+    SessionState sessionState = mock(SessionState.class);
+    when(sessionState.isSupportDropCascade()).thenReturn(true);
+    when(connection.getSessionState()).thenReturn(sessionState);
+    DdlExecutor ddlExecutor =
+        new DdlExecutor(connection, Suppliers.ofInstance(ImmutableList.of())) {
+          @Override
+          ImmutableList<Statement> getDropDependentIndexesStatements(TableOrIndexName tableName) {
+            return ImmutableList.of(Statement.of("drop dependent indexes of " + tableName));
+          }
+
+          @Override
+          ImmutableList<Statement> getDropDependentForeignKeyConstraintsStatements(
+              TableOrIndexName tableName) {
+            return ImmutableList.of(Statement.of("drop dependent foreign keys of " + tableName));
+          }
+
+          @Override
+          ImmutableList<Statement> getDropSchemaTablesStatements(TableOrIndexName schemaName) {
+            return ImmutableList.of(Statement.of("drop all tables in schema " + schemaName));
+          }
+
+          @Override
+          ImmutableList<Statement> getDropSchemaForeignKeysStatements(TableOrIndexName schemaName) {
+            return ImmutableList.of(Statement.of("drop all foreign keys in schema " + schemaName));
+          }
+
+          @Override
+          ImmutableList<Statement> getDropSchemaIndexesStatements(TableOrIndexName schemaName) {
+            return ImmutableList.of(Statement.of("drop all indexes in schema " + schemaName));
+          }
+
+          @Override
+          ImmutableList<Statement> getDropSchemaViewsStatements(TableOrIndexName schemaName) {
+            return ImmutableList.of(Statement.of("drop all views in schema " + schemaName));
+          }
+        };
+
+    assertGetDependentStatementsReturnsSame(
+        ddlExecutor, Statement.of("create table foo (id bigint primary key)"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop view foo"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop table"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop table foo invalid"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema foo invalid"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema foo"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema foo restrict"));
+
+    assertEquals(
+        ImmutableList.of(
+            Statement.of("drop dependent indexes of foo"), Statement.of("drop table foo")),
+        ddlExecutor.getDependentStatements(Statement.of("drop table foo")));
+    assertEquals(
+        ImmutableList.of(
+            Statement.of("drop dependent indexes of foo"),
+            Statement.of("drop dependent foreign keys of foo"),
+            Statement.of("drop table foo")),
+        ddlExecutor.getDependentStatements(Statement.of("drop table foo cascade")));
+    assertEquals(
+        ImmutableList.of(
+            Statement.of("drop all indexes in schema foo"),
+            Statement.of("drop all foreign keys in schema foo"),
+            Statement.of("drop all views in schema foo"),
+            Statement.of("drop all tables in schema foo")),
+        ddlExecutor.getDependentStatements(Statement.of("drop schema foo cascade")));
+  }
+
+  private void assertGetDependentStatementsReturnsSame(
+      DdlExecutor ddlExecutor, Statement statement) {
+    ImmutableList<Statement> dependentStatements = ddlExecutor.getDependentStatements(statement);
+    assertEquals(1, dependentStatements.size());
+    assertSame(dependentStatements.get(0), statement);
+  }
+
+  @Test
+  public void testGetDependentStatements_DropTable() {
+    BackendConnection connection = mock(BackendConnection.class);
+    SessionState sessionState = mock(SessionState.class);
+    when(sessionState.isSupportDropCascade()).thenReturn(true);
+    when(connection.getSessionState()).thenReturn(sessionState);
+    DdlExecutor ddlExecutor = new DdlExecutor(connection, Suppliers.ofInstance(ImmutableList.of()));
+
+    assertGetDependentStatementsReturnsSame(
+        ddlExecutor, Statement.of("create table foo (id bigint primary key)"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop view foo"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop table"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop table foo invalid"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema foo invalid"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema foo"));
+    assertGetDependentStatementsReturnsSame(ddlExecutor, Statement.of("drop schema foo restrict"));
+  }
+
   @Test
   public void testDdlReplacements() {
     DdlExecutor ddlExecutorWithoutReplacements =