diff --git a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
index 637493a32a2a..2bbce3a03c49 100644
--- a/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
+++ b/plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java
@@ -389,7 +389,7 @@ public synchronized void dropDatabase(ConnectorSession session, String schemaNam
             boolean deleteData = location.map(path -> {
                 HdfsContext context = new HdfsContext(session);
                 try (FileSystem fs = hdfsEnvironment.getFileSystem(context, path)) {
-                    return !fs.listFiles(path, false).hasNext();
+                    return !fs.listLocatedStatus(path).hasNext();
                 }
                 catch (IOException | RuntimeException e) {
                     log.warn(e, "Could not check schema directory '%s'", path);
diff --git a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/TrinoHiveCatalog.java b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/TrinoHiveCatalog.java
index 93b1ebc53832..5f4389c4f4e6 100644
--- a/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/TrinoHiveCatalog.java
+++ b/plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/TrinoHiveCatalog.java
@@ -231,7 +231,7 @@ public boolean dropNamespace(ConnectorSession session, String namespace)
         boolean deleteData = location.map(path -> {
             HdfsContext context = new HdfsContext(session);
             try (FileSystem fs = hdfsEnvironment.getFileSystem(context, path)) {
-                return !fs.listFiles(path, false).hasNext();
+                return !fs.listLocatedStatus(path).hasNext();
             }
             catch (IOException e) {
                 log.warn(e, "Could not check schema directory '%s'", path);
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestCreateDropSchema.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestCreateDropSchema.java
index b5454be1c527..837f65b902be 100644
--- a/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestCreateDropSchema.java
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/hive/TestCreateDropSchema.java
@@ -54,51 +54,68 @@ public void testCreateDropSchema()
     }
 
     @Test
-    public void testDropSchemaWithLocationWithoutExternalFiles()
+    public void testDropSchemaFiles()
     {
-        String schemaName = "schema_with_empty_location_" + randomTableSuffix();
-        String schemaDir = warehouseDirectory + "/schema-with-empty-location/";
+        String schemaName = "schema_without_location_" + randomTableSuffix();
+        String schemaDir = format("%s/%s.db/", warehouseDirectory, schemaName);
 
-        onTrino().executeQuery(format("CREATE SCHEMA %s WITH (location = '%s')", schemaName, schemaDir));
+        onTrino().executeQuery(format("CREATE SCHEMA %s", schemaName));
         assertFileExistence(schemaDir, true, "schema directory exists after creating schema");
         onTrino().executeQuery("DROP SCHEMA " + schemaName);
         assertFileExistence(schemaDir, false, "schema directory exists after dropping schema");
     }
 
     @Test
-    public void testDropSchemaFilesWithoutLocation()
+    public void testDropSchemaFilesWithLocation()
     {
-        String schemaName = "schema_without_location_" + randomTableSuffix();
-        String schemaDir = format("%s/%s.db/", warehouseDirectory, schemaName);
+        String schemaName = "schema_with_empty_location_" + randomTableSuffix();
+        String schemaDir = warehouseDirectory + "/schema-with-empty-location/";
 
-        onTrino().executeQuery(format("CREATE SCHEMA %s", schemaName));
+        onTrino().executeQuery(format("CREATE SCHEMA %s WITH (location = '%s')", schemaName, schemaDir));
         assertFileExistence(schemaDir, true, "schema directory exists after creating schema");
         onTrino().executeQuery("DROP SCHEMA " + schemaName);
         assertFileExistence(schemaDir, false, "schema directory exists after dropping schema");
     }
 
-    @Test
-    public void testDropSchemaFilesWithLocationWithExternalFile()
+    @Test // specified location, external file in subdir
+    public void testDropWithExternalFilesInSubdirectory()
     {
         String schemaName = "schema_with_nonempty_location_" + randomTableSuffix();
         String schemaDir = warehouseDirectory + "/schema-with-nonempty-location/";
+        // Use subdirectory to make sure file check is recursive
+        String subDir = schemaDir + "subdir/";
+        String externalFile = subDir + "external-file";
 
-        // Create file in schema directory before creating schema
-        String externalFile = schemaDir + "external-file";
-        hdfsClient.createDirectory(schemaDir);
+        // Create file below schema directory before creating schema
+        hdfsClient.createDirectory(subDir);
         hdfsClient.saveFile(externalFile, "");
 
         onTrino().executeQuery(format("CREATE SCHEMA %s WITH (location = '%s')", schemaName, schemaDir));
-        assertFileExistence(schemaDir, true, "schema directory exists after creating schema");
+        assertFileExistence(externalFile, true, "external file exists after creating schema");
         onTrino().executeQuery("DROP SCHEMA " + schemaName);
-        assertFileExistence(schemaDir, true, "schema directory exists after dropping schema");
         assertFileExistence(externalFile, true, "external file exists after dropping schema");
 
-        hdfsClient.delete(externalFile);
+        hdfsClient.delete(schemaDir);
     }
 
-    // Tests create/drop schema transactions with default schema location
-    @Test
+    @Test // default location, empty external subdir
+    public void testDropSchemaFilesWithEmptyExternalSubdir()
+    {
+        String schemaName = "schema_with_empty_subdirectory_" + randomTableSuffix();
+        String schemaDir = format("%s/%s.db/", warehouseDirectory, schemaName);
+        String externalSubdir = schemaDir + "external-subdir/";
+
+        hdfsClient.createDirectory(externalSubdir);
+
+        onTrino().executeQuery("CREATE SCHEMA " + schemaName);
+        assertFileExistence(externalSubdir, true, "external subdirectory exists after creating schema");
+        onTrino().executeQuery("DROP SCHEMA " + schemaName);
+        assertFileExistence(externalSubdir, true, "external subdirectory exists after dropping schema");
+
+        hdfsClient.delete(schemaDir);
+    }
+
+    @Test // default location, transactions without external files
     public void testDropSchemaFilesTransactions()
     {
         String schemaName = "schema_directory_transactions_" + randomTableSuffix();
@@ -122,8 +139,8 @@ public void testDropSchemaFilesTransactions()
         assertFileExistence(schemaDir, false, "schema directory exists after dropping schema");
     }
 
-    @Test
-    public void testDropSchemaFilesTransactionsWithExternalFile()
+    @Test // specified location, transaction with top-level external file
+    public void testDropTransactionsWithExternalFiles()
     {
         String schemaName = "schema_transactions_with_external_files_" + randomTableSuffix();
         String schemaDir = warehouseDirectory + "/schema-transactions-with-external-files/";
diff --git a/testing/trino-product-tests/src/main/java/io/trino/tests/product/iceberg/TestCreateDropSchema.java b/testing/trino-product-tests/src/main/java/io/trino/tests/product/iceberg/TestCreateDropSchema.java
index b7f9fdefd194..ec9228c0da7c 100644
--- a/testing/trino-product-tests/src/main/java/io/trino/tests/product/iceberg/TestCreateDropSchema.java
+++ b/testing/trino-product-tests/src/main/java/io/trino/tests/product/iceberg/TestCreateDropSchema.java
@@ -46,52 +46,69 @@ public void useIceberg()
     }
 
     @Test(groups = ICEBERG)
-    public void testDropSchemaWithLocationWithoutExternalFiles()
+    public void testDropSchemaFiles()
     {
-        String schemaName = "schema_with_empty_location_" + randomTableSuffix();
-        String schemaDir = warehouseDirectory + "/schema-with-empty-location/";
+        String schemaName = "schema_without_location_" + randomTableSuffix();
+        String schemaDir = format("%s/%s.db/", warehouseDirectory, schemaName);
 
-        onTrino().executeQuery(format("CREATE SCHEMA %s WITH (location = '%s')", schemaName, schemaDir));
+        onTrino().executeQuery(format("CREATE SCHEMA %s", schemaName));
         assertFileExistence(schemaDir, true, "schema directory exists after creating schema");
         onTrino().executeQuery("DROP SCHEMA " + schemaName);
         assertFileExistence(schemaDir, false, "schema directory exists after dropping schema");
     }
 
     @Test(groups = ICEBERG)
-    public void testDropSchemaFilesWithoutLocation()
+    public void testDropSchemaFilesWithLocation()
     {
-        String schemaName = "schema_without_location_" + randomTableSuffix();
-        String schemaDir = format("%s/%s.db/", warehouseDirectory, schemaName);
+        String schemaName = "schema_with_empty_location_" + randomTableSuffix();
+        String schemaDir = warehouseDirectory + "/schema-with-empty-location/";
 
-        onTrino().executeQuery(format("CREATE SCHEMA %s", schemaName));
+        onTrino().executeQuery(format("CREATE SCHEMA %s WITH (location = '%s')", schemaName, schemaDir));
         assertFileExistence(schemaDir, true, "schema directory exists after creating schema");
         onTrino().executeQuery("DROP SCHEMA " + schemaName);
         assertFileExistence(schemaDir, false, "schema directory exists after dropping schema");
     }
 
-    @Test(groups = ICEBERG)
-    public void testDropSchemaFilesWithLocationWithExternalFile()
+    @Test(groups = ICEBERG) // specified location, external file in subdir
+    public void testDropWithExternalFilesInSubdirectory()
     {
         String schemaName = "schema_with_nonempty_location_" + randomTableSuffix();
         String schemaDir = warehouseDirectory + "/schema-with-nonempty-location/";
+        // Use subdirectory to make sure file check is recursive
+        String subDir = schemaDir + "subdir/";
+        String externalFile = subDir + "external-file";
 
-        // Create file in schema directory before creating schema
-        String externalFile = schemaDir + "external-file";
-        hdfsClient.createDirectory(schemaDir);
+        // Create file below schema directory before creating schema
+        hdfsClient.createDirectory(subDir);
         hdfsClient.saveFile(externalFile, "");
 
         onTrino().executeQuery(format("CREATE SCHEMA %s WITH (location = '%s')", schemaName, schemaDir));
-        assertFileExistence(schemaDir, true, "schema directory exists after creating schema");
+        assertFileExistence(externalFile, true, "external file exists after creating schema");
         onTrino().executeQuery("DROP SCHEMA " + schemaName);
-        assertFileExistence(schemaDir, true, "schema directory exists after dropping schema");
         assertFileExistence(externalFile, true, "external file exists after dropping schema");
 
-        hdfsClient.delete(externalFile);
         hdfsClient.delete(schemaDir);
     }
 
-    @Test(groups = ICEBERG)
-    public void testDropSchemaWithExternalFileWithoutLocation()
+    @Test(groups = ICEBERG) // make sure empty directories are noticed as well
+    public void testDropSchemaFilesWithEmptyExternalSubdir()
+    {
+        String schemaName = "schema_with_empty_subdirectory_" + randomTableSuffix();
+        String schemaDir = format("%s/%s.db/", warehouseDirectory, schemaName);
+        String externalSubdir = schemaDir + "external-subdir/";
+
+        hdfsClient.createDirectory(externalSubdir);
+
+        onTrino().executeQuery("CREATE SCHEMA " + schemaName);
+        assertFileExistence(externalSubdir, true, "external subdirectory exists after creating schema");
+        onTrino().executeQuery("DROP SCHEMA " + schemaName);
+        assertFileExistence(externalSubdir, true, "external subdirectory exists after dropping schema");
+
+        hdfsClient.delete(schemaDir);
+    }
+
+    @Test(groups = ICEBERG) // default location, external file at top level
+    public void testDropWithExternalFiles()
     {
         String schemaName = "schema_with_files_in_default_location_" + randomTableSuffix();
         String schemaDir = format("%s/%s.db/", warehouseDirectory, schemaName);