Allow ignoring schema location cleanup

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveConfig.java

@@ -80,6 +80,7 @@ public class HiveConfig
     private boolean forceLocalScheduling;
     private boolean recursiveDirWalkerEnabled;
     private boolean ignoreAbsentPartitions;
+    private boolean ignoreSchemaLocationCleanupFailure;
 
     private int maxConcurrentFileRenames = 20;
     private int maxConcurrentMetastoreDrops = 20;
@@ -329,6 +330,19 @@ public HiveConfig setIgnoreAbsentPartitions(boolean ignoreAbsentPartitions)
         return this;
     }
 
+    public boolean isIgnoreSchemaLocationCleanupFailure()
+    {
+        return ignoreSchemaLocationCleanupFailure;
+    }
+
+    @Config("hive.ignore-schema-location-cleanup-failure")
+    @ConfigDescription("Allows to ignore failures related to file system cleanup during DROP SCHEMA for situations when schema location is misconfigured or no longer reachable")
+    public HiveConfig setIgnoreSchemaLocationCleanupFailure(boolean ignoreSchemaLocationCleanupFailure)
+    {
+        this.ignoreSchemaLocationCleanupFailure = ignoreSchemaLocationCleanupFailure;
+        return this;
+    }
+
     @NotNull
     public DataSize getMaxSplitSize()
     {

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveMetadata.java

@@ -809,7 +809,7 @@ public void createSchema(ConnectorSession session, String schemaName, Map<String
     @Override
     public void dropSchema(ConnectorSession session, String schemaName)
     {
-        metastore.dropDatabase(new HiveIdentity(session), schemaName);
+        metastore.dropDatabase(session, schemaName);
     }
 
     @Override

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveSessionProperties.java

@@ -103,6 +103,7 @@ public final class HiveSessionProperties
     private static final String TEMPORARY_STAGING_DIRECTORY_PATH = "temporary_staging_directory_path";
     private static final String DELEGATE_TRANSACTIONAL_MANAGED_TABLE_LOCATION_TO_METASTORE = "delegate_transactional_managed_table_location_to_metastore";
     private static final String IGNORE_ABSENT_PARTITIONS = "ignore_absent_partitions";
+    private static final String IGNORE_SCHEMA_LOCATION_CLEANUP_FAILURE = "ignore_schema_location_cleanup_failure";
     private static final String QUERY_PARTITION_FILTER_REQUIRED = "query_partition_filter_required";
     private static final String QUERY_PARTITION_FILTER_REQUIRED_SCHEMAS = "query_partition_filter_required_schemas";
     private static final String PROJECTION_PUSHDOWN_ENABLED = "projection_pushdown_enabled";
@@ -410,6 +411,11 @@ public HiveSessionProperties(
                         "Ignore partitions when the file system location does not exist rather than failing the query.",
                         hiveConfig.isIgnoreAbsentPartitions(),
                         false),
+                booleanProperty(
+                        IGNORE_SCHEMA_LOCATION_CLEANUP_FAILURE,
+                        "Allows to ignore failures related to file system cleanup during DROP SCHEMA for situations when schema location is misconfigured or no longer reachable",
+                        hiveConfig.isIgnoreSchemaLocationCleanupFailure(),
+                        false),
                 booleanProperty(
                         QUERY_PARTITION_FILTER_REQUIRED,
                         "Require filter on partition column",
@@ -747,6 +753,11 @@ public static boolean isIgnoreAbsentPartitions(ConnectorSession session)
         return session.getProperty(IGNORE_ABSENT_PARTITIONS, Boolean.class);
     }
 
+    public static boolean isIgnoreSchemaLocationCleanupFailure(ConnectorSession session)
+    {
+        return session.getProperty(IGNORE_SCHEMA_LOCATION_CLEANUP_FAILURE, Boolean.class);
+    }
+
     public static boolean isQueryPartitionFilterRequired(ConnectorSession session)
     {
         return session.getProperty(QUERY_PARTITION_FILTER_REQUIRED, Boolean.class);

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/SemiTransactionalHiveMetastore.java

@@ -95,6 +95,7 @@
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_PATH_ALREADY_EXISTS;
 import static io.trino.plugin.hive.HiveErrorCode.HIVE_TABLE_DROPPED_DURING_QUERY;
 import static io.trino.plugin.hive.HiveMetadata.PRESTO_QUERY_ID_NAME;
+import static io.trino.plugin.hive.HiveSessionProperties.isIgnoreSchemaLocationCleanupFailure;
 import static io.trino.plugin.hive.LocationHandle.WriteMode.DIRECT_TO_TARGET_NEW_DIRECTORY;
 import static io.trino.plugin.hive.ViewReaderUtil.isPrestoView;
 import static io.trino.plugin.hive.acid.AcidTransaction.NO_ACID_TRANSACTION;
@@ -366,8 +367,9 @@ public synchronized void createDatabase(HiveIdentity identity, Database database
         setExclusive((delegate, hdfsEnvironment) -> delegate.createDatabase(identity, database));
     }
 
-    public synchronized void dropDatabase(HiveIdentity identity, String schemaName)
+    public synchronized void dropDatabase(ConnectorSession session, String schemaName)
     {
+        HiveIdentity identity = new HiveIdentity(session);
         HdfsContext context = new HdfsContext(
                 identity.getUsername()
                         .map(ConnectorIdentity::ofUser)
@@ -393,10 +395,15 @@ public synchronized void dropDatabase(HiveIdentity identity, String schemaName)
                         log.info("Skipped deleting schema location with external files (%s)", path);
                     }
                 }
-                catch (IOException e) {
-                    throw new TrinoException(
-                            HIVE_FILESYSTEM_ERROR,
-                            format("Error checking or deleting schema directory '%s'", path), e);
+                catch (IOException | RuntimeException e) {
+                    if (isIgnoreSchemaLocationCleanupFailure(session)) {
+                        log.warn(e, "Failure when checking or deleting schema directory '%s'", path);
+                    }
+                    else {
+                        throw new TrinoException(
+                                HIVE_FILESYSTEM_ERROR,
+                                format("Error checking or deleting schema directory '%s'", path), e);
+                    }
                 }
             });
         });
@@ -2246,7 +2253,7 @@ private void rollbackShared()
                                     .map(Column::getName)
                                     .collect(toImmutableList());
                             List<String> partitionNames = delegate.getPartitionNamesByFilter(
-                                    identity, schemaTableName.getSchemaName(), schemaTableName.getTableName(), partitionColumnNames, TupleDomain.all())
+                                            identity, schemaTableName.getSchemaName(), schemaTableName.getTableName(), partitionColumnNames, TupleDomain.all())
                                     .orElse(ImmutableList.of());
                             for (List<String> partitionNameBatch : Iterables.partition(partitionNames, 10)) {
                                 Collection<Optional<Partition>> partitions = delegate.getPartitionsByNames(identity, schemaTableName.getSchemaName(), schemaTableName.getTableName(), partitionNameBatch).values();

plugin/trino-hive/src/test/java/io/trino/plugin/hive/TestHiveConfig.java

@@ -57,6 +57,7 @@ public void testDefaults()
                 .setMaxConcurrentMetastoreUpdates(20)
                 .setRecursiveDirWalkerEnabled(false)
                 .setIgnoreAbsentPartitions(false)
+                .setIgnoreSchemaLocationCleanupFailure(false)
                 .setHiveStorageFormat(HiveStorageFormat.ORC)
                 .setHiveCompressionCodec(HiveCompressionCodec.GZIP)
                 .setRespectTableFormat(true)
@@ -130,6 +131,7 @@ public void testExplicitPropertyMappings()
                 .put("hive.writer-sort-buffer-size", "13MB")
                 .put("hive.recursive-directories", "true")
                 .put("hive.ignore-absent-partitions", "true")
+                .put("hive.ignore-schema-location-cleanup-failure", "true")
                 .put("hive.storage-format", "SEQUENCEFILE")
                 .put("hive.compression-codec", "NONE")
                 .put("hive.respect-table-format", "false")
@@ -208,6 +210,7 @@ public void testExplicitPropertyMappings()
                 .setMaxConcurrentMetastoreUpdates(100)
                 .setRecursiveDirWalkerEnabled(true)
                 .setIgnoreAbsentPartitions(true)
+                .setIgnoreSchemaLocationCleanupFailure(true)
                 .setHiveStorageFormat(HiveStorageFormat.SEQUENCEFILE)
                 .setHiveCompressionCodec(HiveCompressionCodec.NONE)
                 .setRespectTableFormat(false)