diff --git a/plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/TestClickHouseConnectorTest.java b/plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/TestClickHouseConnectorTest.java
index 6955303b2c28..c03a858f675d 100644
--- a/plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/TestClickHouseConnectorTest.java
+++ b/plugin/trino-clickhouse/src/test/java/io/trino/plugin/clickhouse/TestClickHouseConnectorTest.java
@@ -33,6 +33,7 @@
 import static io.trino.testing.assertions.Assert.assertEquals;
 import static io.trino.testing.sql.TestTable.randomTableSuffix;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
@@ -80,6 +81,15 @@ protected boolean hasBehavior(TestingConnectorBehavior connectorBehavior)
         }
     }
 
+    @Override
+    public void testDropNonEmptySchema()
+    {
+        assertThatThrownBy(super::testDropNonEmptySchema)
+                .isInstanceOf(AssertionError.class)
+                .hasMessageStartingWith("Expected query to fail: DROP SCHEMA ");
+        throw new SkipException("DROP SCHEMA erases all tables in ClickHouse connector"); // TODO (https://github.com/trinodb/trino/issues/8634)
+    }
+
     @Override
     @Test(dataProvider = "testColumnNameDataProvider")
     public void testColumnName(String columnName)
diff --git a/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduDistributedQueries.java b/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduDistributedQueries.java
index b86d2694ae3a..49c6e310e1d2 100644
--- a/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduDistributedQueries.java
+++ b/plugin/trino-kudu/src/test/java/io/trino/plugin/kudu/TestKuduDistributedQueries.java
@@ -118,6 +118,13 @@ public void testInsertUnicode()
         throw new SkipException("TODO");
     }
 
+    @Override
+    public void testDropNonEmptySchema()
+    {
+        // TODO Support these test once kudu connector can create tables with default partitions
+        throw new SkipException("TODO");
+    }
+
     @Override
     public void testDelete()
     {
diff --git a/plugin/trino-memsql/src/test/java/io/trino/plugin/memsql/TestMemSqlConnectorTest.java b/plugin/trino-memsql/src/test/java/io/trino/plugin/memsql/TestMemSqlConnectorTest.java
index e8a23122b304..1637ba3ff2a4 100644
--- a/plugin/trino-memsql/src/test/java/io/trino/plugin/memsql/TestMemSqlConnectorTest.java
+++ b/plugin/trino-memsql/src/test/java/io/trino/plugin/memsql/TestMemSqlConnectorTest.java
@@ -40,6 +40,7 @@
 import static java.util.stream.Collectors.joining;
 import static java.util.stream.IntStream.range;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
@@ -160,6 +161,15 @@ public void testInsertUnicode()
         throw new SkipException("MemSQL doesn't support utf8mb4");
     }
 
+    @Override
+    public void testDropNonEmptySchema()
+    {
+        assertThatThrownBy(super::testDropNonEmptySchema)
+                .isInstanceOf(AssertionError.class)
+                .hasMessageStartingWith("Expected query to fail: DROP SCHEMA ");
+        throw new SkipException("DROP SCHEMA erases all tables in MemSQL connector"); // TODO (https://github.com/trinodb/trino/issues/8634)
+    }
+
     @Test
     public void testDropTable()
     {
diff --git a/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java b/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
index 90573d273ccf..768176d7df88 100644
--- a/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
+++ b/plugin/trino-mysql/src/test/java/io/trino/plugin/mysql/BaseMySqlConnectorTest.java
@@ -21,6 +21,7 @@
 import io.trino.testing.TestingConnectorBehavior;
 import io.trino.testing.sql.TestTable;
 import org.intellij.lang.annotations.Language;
+import org.testng.SkipException;
 import org.testng.annotations.Test;
 
 import java.util.Optional;
@@ -33,6 +34,7 @@
 import static io.trino.testing.assertions.Assert.assertEquals;
 import static java.lang.String.format;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertTrue;
 
@@ -178,6 +180,15 @@ public void testShowCreateTable()
                         ")");
     }
 
+    @Override
+    public void testDropNonEmptySchema()
+    {
+        assertThatThrownBy(super::testDropNonEmptySchema)
+                .isInstanceOf(AssertionError.class)
+                .hasMessageStartingWith("Expected query to fail: DROP SCHEMA ");
+        throw new SkipException("DROP SCHEMA erases all tables in MySQL connector"); // TODO (https://github.com/trinodb/trino/issues/8634)
+    }
+
     @Test
     public void testDropTable()
     {
diff --git a/testing/trino-testing/src/main/java/io/trino/testing/AbstractTestDistributedQueries.java b/testing/trino-testing/src/main/java/io/trino/testing/AbstractTestDistributedQueries.java
index 62a032b1a55b..2aa4562ecdd0 100644
--- a/testing/trino-testing/src/main/java/io/trino/testing/AbstractTestDistributedQueries.java
+++ b/testing/trino-testing/src/main/java/io/trino/testing/AbstractTestDistributedQueries.java
@@ -76,6 +76,12 @@ protected boolean supportsCreateSchema()
         return true;
     }
 
+    private boolean supportsDropSchema()
+    {
+        // A connector either supports CREATE SCHEMA and DROP SCHEMA or none of them.
+        return supportsCreateSchema();
+    }
+
     protected boolean supportsCreateTable()
     {
         return true;
@@ -1126,6 +1132,26 @@ public void testCreateSchema()
         assertQueryFails("DROP SCHEMA " + schemaName, format("line 1:1: Schema '.*\\.%s' does not exist", schemaName));
     }
 
+    @Test
+    public void testDropNonEmptySchema()
+    {
+        if (!supportsCreateSchema()) {
+            assertQueryFails("DROP SCHEMA " + getSession().getSchema().orElseThrow(), "This connector does not support dropping schemas");
+            return;
+        }
+
+        String schemaName = "test_drop_non_empty_schema_" + randomTableSuffix();
+        assertUpdate("CREATE SCHEMA " + schemaName);
+        assertUpdate("CREATE TABLE " + schemaName + ".t(x int)");
+        assertQueryFails("DROP SCHEMA " + schemaName, "(?si).*" +
+                "(Schema not empty|" +
+                "Cannot drop schema|" +
+                "schema has (\\w+ )?tables|" +
+                "Cannot drop .{0,3}\\Q" + schemaName + "\\E).*");
+        assertUpdate("DROP TABLE " + schemaName + ".t");
+        assertUpdate("DROP SCHEMA " + schemaName);
+    }
+
     @Test
     public void testInsertForDefaultColumn()
     {