diff --git a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduClientSession.java b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduClientSession.java
index fa88ab9c3ef7..452afc08e631 100644
--- a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduClientSession.java
+++ b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduClientSession.java
@@ -261,9 +261,9 @@ public void createSchema(String schemaName)
         schemaEmulation.createSchema(client, schemaName);
     }
 
-    public void dropSchema(String schemaName)
+    public void dropSchema(String schemaName, boolean cascade)
     {
-        schemaEmulation.dropSchema(client, schemaName);
+        schemaEmulation.dropSchema(client, schemaName, cascade);
     }
 
     public void dropTable(SchemaTableName schemaTableName)
diff --git a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java
index 7348bdd6b54c..e8ac4922ebe0 100755
--- a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java
+++ b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/KuduMetadata.java
@@ -242,9 +242,9 @@ public void createSchema(ConnectorSession session, String schemaName, Map<String
     }
 
     @Override
-    public void dropSchema(ConnectorSession session, String schemaName)
+    public void dropSchema(ConnectorSession session, String schemaName, boolean cascade)
     {
-        clientSession.dropSchema(schemaName);
+        clientSession.dropSchema(schemaName, cascade);
     }
 
     @Override
diff --git a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/NoSchemaEmulation.java b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/NoSchemaEmulation.java
index b3e4b44f8b0d..ad3c828445d3 100644
--- a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/NoSchemaEmulation.java
+++ b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/NoSchemaEmulation.java
@@ -37,7 +37,7 @@ public void createSchema(KuduClientWrapper client, String schemaName)
     }
 
     @Override
-    public void dropSchema(KuduClientWrapper client, String schemaName)
+    public void dropSchema(KuduClientWrapper client, String schemaName, boolean cascade)
     {
         if (DEFAULT_SCHEMA.equals(schemaName)) {
             throw new TrinoException(GENERIC_USER_ERROR, "Deleting default schema not allowed.");
diff --git a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulation.java b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulation.java
index ff983718fa5f..56ebfc743a2a 100644
--- a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulation.java
+++ b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulation.java
@@ -22,7 +22,7 @@ public interface SchemaEmulation
 {
     void createSchema(KuduClientWrapper client, String schemaName);
 
-    void dropSchema(KuduClientWrapper client, String schemaName);
+    void dropSchema(KuduClientWrapper client, String schemaName, boolean cascade);
 
     boolean existsSchema(KuduClientWrapper client, String schemaName);
 
diff --git a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulationByTableNameConvention.java b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulationByTableNameConvention.java
index 496e2a86205d..076001d11ee1 100644
--- a/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulationByTableNameConvention.java
+++ b/plugin/trino-kudu/src/main/java/io/trino/plugin/kudu/schema/SchemaEmulationByTableNameConvention.java
@@ -39,6 +39,7 @@
 import static io.trino.plugin.kudu.KuduClientSession.DEFAULT_SCHEMA;
 import static io.trino.spi.StandardErrorCode.GENERIC_INTERNAL_ERROR;
 import static io.trino.spi.StandardErrorCode.GENERIC_USER_ERROR;
+import static io.trino.spi.StandardErrorCode.SCHEMA_NOT_EMPTY;
 
 public class SchemaEmulationByTableNameConvention
         implements SchemaEmulation
@@ -81,14 +82,18 @@ public boolean existsSchema(KuduClientWrapper client, String schemaName)
     }
 
     @Override
-    public void dropSchema(KuduClientWrapper client, String schemaName)
+    public void dropSchema(KuduClientWrapper client, String schemaName, boolean cascade)
     {
         if (DEFAULT_SCHEMA.equals(schemaName)) {
             throw new TrinoException(GENERIC_USER_ERROR, "Deleting default schema not allowed.");
         }
         try (KuduOperationApplier operationApplier = KuduOperationApplier.fromKuduClientWrapper(client)) {
             String prefix = getPrefixForTablesOfSchema(schemaName);
-            for (String name : client.getTablesList(prefix).getTablesList()) {
+            List<String> tables = client.getTablesList(prefix).getTablesList();
+            if (!cascade && !tables.isEmpty()) {
+                throw new TrinoException(SCHEMA_NOT_EMPTY, "Cannot drop non-empty schema '%s'".formatted(schemaName));
+            }
+            for (String name : tables) {
                 client.deleteTable(name);
             }
 
diff --git a/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/BaseKuduConnectorSmokeTest.java b/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/BaseKuduConnectorSmokeTest.java
index ccc606c04ada..fd0b65062696 100644
--- a/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/BaseKuduConnectorSmokeTest.java
+++ b/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/BaseKuduConnectorSmokeTest.java
@@ -16,6 +16,7 @@
 import io.trino.testing.BaseConnectorSmokeTest;
 import io.trino.testing.QueryRunner;
 import io.trino.testing.TestingConnectorBehavior;
+import org.testng.SkipException;
 import org.testng.annotations.Test;
 
 import java.util.Optional;
@@ -26,6 +27,7 @@
 import static io.trino.plugin.kudu.TestingKuduServer.EARLIEST_TAG;
 import static io.trino.testing.TestingNames.randomNameSuffix;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 public abstract class BaseKuduConnectorSmokeTest
         extends BaseConnectorSmokeTest
@@ -162,4 +164,29 @@ public void testCreateTableWithTableComment()
 
         assertUpdate("DROP TABLE " + tableName);
     }
+
+    @Test
+    public void testDropSchemaCascade()
+    {
+        String schemaName = "test_drop_schema_cascade_" + randomNameSuffix();
+        String tableName = "test_table" + randomNameSuffix();
+        try {
+            if (getKuduSchemaEmulationPrefix().isEmpty()) {
+                assertThatThrownBy(() -> assertUpdate("CREATE SCHEMA " + schemaName))
+                        .hasMessageContaining("Creating schema in Kudu connector not allowed if schema emulation is disabled.");
+                throw new SkipException("Cannot test when schema emulation is disabled");
+            }
+            assertUpdate("CREATE SCHEMA " + schemaName);
+            assertUpdate("CREATE TABLE " + schemaName + "." + tableName + " AS SELECT 1 a", 1);
+
+            assertThat(computeActual("SHOW SCHEMAS").getOnlyColumnAsSet()).contains(schemaName);
+
+            assertUpdate("DROP SCHEMA " + schemaName + " CASCADE");
+            assertThat(computeActual("SHOW SCHEMAS").getOnlyColumnAsSet()).doesNotContain(schemaName);
+        }
+        finally {
+            assertUpdate("DROP TABLE IF EXISTS " + schemaName + "." + tableName);
+            assertUpdate("DROP SCHEMA IF EXISTS " + schemaName);
+        }
+    }
 }
diff --git a/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduConnectorTest.java b/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduConnectorTest.java
index 6fb5b5557a85..658689c327d2 100644
--- a/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduConnectorTest.java
+++ b/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduConnectorTest.java
@@ -72,7 +72,6 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                 return false;
 
             case SUPPORTS_RENAME_SCHEMA:
-            case SUPPORTS_DROP_SCHEMA_CASCADE:
                 return false;
 
             case SUPPORTS_CREATE_TABLE_WITH_COLUMN_COMMENT:
diff --git a/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java b/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java
index d8c5d8b031ae..37ca08535c28 100644
--- a/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java
+++ b/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoMetadata.java
@@ -147,9 +147,9 @@ public void createSchema(ConnectorSession session, String schemaName, Map<String
     }
 
     @Override
-    public void dropSchema(ConnectorSession session, String schemaName)
+    public void dropSchema(ConnectorSession session, String schemaName, boolean cascade)
     {
-        mongoSession.dropSchema(schemaName);
+        mongoSession.dropSchema(schemaName, cascade);
     }
 
     @Override
diff --git a/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoSession.java b/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoSession.java
index 366b4e478942..f507a9582d6b 100644
--- a/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoSession.java
+++ b/plugin/trino-mongodb/src/main/java/io/trino/plugin/mongodb/MongoSession.java
@@ -93,6 +93,7 @@
 import static io.trino.plugin.mongodb.ObjectIdType.OBJECT_ID;
 import static io.trino.plugin.mongodb.ptf.Query.parseFilter;
 import static io.trino.spi.HostAddress.fromParts;
+import static io.trino.spi.StandardErrorCode.SCHEMA_NOT_EMPTY;
 import static io.trino.spi.type.BigintType.BIGINT;
 import static io.trino.spi.type.BooleanType.BOOLEAN;
 import static io.trino.spi.type.Chars.padSpaces;
@@ -212,9 +213,20 @@ public void createSchema(String schemaName)
         client.getDatabase(schemaName).createCollection(schemaCollection);
     }
 
-    public void dropSchema(String schemaName)
+    public void dropSchema(String schemaName, boolean cascade)
     {
-        client.getDatabase(toRemoteSchemaName(schemaName)).drop();
+        MongoDatabase database = client.getDatabase(toRemoteSchemaName(schemaName));
+        if (!cascade) {
+            try (MongoCursor<String> collections = database.listCollectionNames().cursor()) {
+                while (collections.hasNext()) {
+                    if (collections.next().equals(schemaCollection)) {
+                        continue;
+                    }
+                    throw new TrinoException(SCHEMA_NOT_EMPTY, "Cannot drop non-empty schema '%s'".formatted(schemaName));
+                }
+            }
+        }
+        database.drop();
     }
 
     public Set<String> getAllTables(String schema)
diff --git a/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java b/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java
index ab98e1551a02..3eaa926f8246 100644
--- a/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java
+++ b/plugin/trino-mongodb/src/test/java/io/trino/plugin/mongodb/TestMongoConnectorTest.java
@@ -107,7 +107,6 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
                 return false;
 
             case SUPPORTS_RENAME_SCHEMA:
-            case SUPPORTS_DROP_SCHEMA_CASCADE:
                 return false;
 
             case SUPPORTS_ADD_FIELD: