[Alternative] Support drop corrupted iceberg table

core/trino-main/src/main/java/io/trino/execution/DropTableTask.java

@@ -20,16 +20,23 @@
 import io.trino.metadata.QualifiedObjectName;
 import io.trino.metadata.RedirectionAwareTableHandle;
 import io.trino.security.AccessControl;
+import io.trino.spi.TrinoException;
+import io.trino.spi.connector.CatalogSchemaTableName;
+import io.trino.spi.connector.DropRedirected;
+import io.trino.spi.connector.DropResult;
 import io.trino.sql.tree.DropTable;
 import io.trino.sql.tree.Expression;
 
 import javax.inject.Inject;
 
 import java.util.List;
+import java.util.Optional;
 
 import static com.google.common.util.concurrent.Futures.immediateVoidFuture;
 import static io.trino.metadata.MetadataUtil.createQualifiedObjectName;
+import static io.trino.metadata.QualifiedObjectName.convertFromSchemaTableName;
 import static io.trino.spi.StandardErrorCode.GENERIC_USER_ERROR;
+import static io.trino.spi.StandardErrorCode.METADATA_NOT_FOUND;
 import static io.trino.spi.StandardErrorCode.TABLE_NOT_FOUND;
 import static io.trino.sql.analyzer.SemanticExceptions.semanticException;
 import static java.util.Objects.requireNonNull;
@@ -76,18 +83,45 @@ public ListenableFuture<Void> execute(
                     "Table '%s' does not exist, but a view with that name exists. Did you mean DROP VIEW %s?", originalTableName, originalTableName);
         }
 
-        RedirectionAwareTableHandle redirectionAwareTableHandle = metadata.getRedirectionAwareTableHandle(session, originalTableName);
-        if (redirectionAwareTableHandle.getTableHandle().isEmpty()) {
-            if (!statement.isExists()) {
-                throw semanticException(TABLE_NOT_FOUND, statement, "Table '%s' does not exist", originalTableName);
+        Optional<RedirectionAwareTableHandle> redirectionAwareTableHandle = Optional.empty();
+        try {
+            redirectionAwareTableHandle = Optional.of(metadata.getRedirectionAwareTableHandle(session, originalTableName));
+            if (redirectionAwareTableHandle.orElseThrow().getTableHandle().isEmpty()) {
+                if (!statement.isExists()) {
+                    throw semanticException(TABLE_NOT_FOUND, statement, "Table '%s' does not exist", originalTableName);
+                }
+                return immediateVoidFuture();
             }
-            return immediateVoidFuture();
         }
-        QualifiedObjectName tableName = redirectionAwareTableHandle.getRedirectedTableName().orElse(originalTableName);
-        accessControl.checkCanDropTable(session.toSecurityContext(), tableName);
+        catch (TrinoException e) {
+            if (!e.getErrorCode().equals(METADATA_NOT_FOUND.toErrorCode())) {
+                throw e;
+            }
+            // Ignore Exception when metadata not found and force table drop.
+        }
 
-        metadata.dropTable(session, redirectionAwareTableHandle.getTableHandle().get());
+        QualifiedObjectName tableName = originalTableName;
+        if (redirectionAwareTableHandle.isPresent()) {
+            tableName = redirectionAwareTableHandle.get().getRedirectedTableName().orElse(originalTableName);
+        }
+
+        DropResult dropResult = dropTable(session, tableName);
+
+        if (dropResult.isRedirected()) {
+            // When getRedirectionAwareTableHandle fails to load the table due to metadata missing, and table is redirected.
+            // we will get the redirected target name from dropResult.
+            CatalogSchemaTableName catalogSchemaTableName = ((DropRedirected) dropResult).getRedirectedTarget();
+            tableName = convertFromSchemaTableName(catalogSchemaTableName.getCatalogName())
+                    .apply(catalogSchemaTableName.getSchemaTableName());
+            dropTable(session, tableName);
+        }
 
         return immediateVoidFuture();
     }
+
+    private DropResult dropTable(Session session, QualifiedObjectName tableName)
+    {
+        accessControl.checkCanDropTable(session.toSecurityContext(), tableName);
+        return metadata.dropTable(session, tableName);
+    }
 }

core/trino-main/src/main/java/io/trino/metadata/Metadata.java

@@ -30,6 +30,7 @@
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.Constraint;
 import io.trino.spi.connector.ConstraintApplicationResult;
+import io.trino.spi.connector.DropResult;
 import io.trino.spi.connector.JoinApplicationResult;
 import io.trino.spi.connector.JoinStatistics;
 import io.trino.spi.connector.JoinType;
@@ -252,9 +253,9 @@ Optional<TableExecuteHandle> getTableHandleForExecute(
     /**
      * Drops the specified table
      *
-     * @throws RuntimeException if the table cannot be dropped or table handle is no longer valid
+     * @throws RuntimeException if the table cannot be dropped
      */
-    void dropTable(Session session, TableHandle tableHandle);
+    DropResult dropTable(Session session, QualifiedObjectName tableName);
 
     /**
      * Truncates the specified table

core/trino-main/src/main/java/io/trino/metadata/MetadataManager.java

@@ -61,6 +61,9 @@
 import io.trino.spi.connector.ConnectorViewDefinition;
 import io.trino.spi.connector.Constraint;
 import io.trino.spi.connector.ConstraintApplicationResult;
+import io.trino.spi.connector.DropRedirected;
+import io.trino.spi.connector.DropResult;
+import io.trino.spi.connector.DropSuccess;
 import io.trino.spi.connector.JoinApplicationResult;
 import io.trino.spi.connector.JoinStatistics;
 import io.trino.spi.connector.JoinType;
@@ -758,14 +761,21 @@ public void setTableAuthorization(Session session, CatalogSchemaTableName table,
     }
 
     @Override
-    public void dropTable(Session session, TableHandle tableHandle)
+    public DropResult dropTable(Session session, QualifiedObjectName tableName)
     {
-        CatalogHandle catalogHandle = tableHandle.getCatalogHandle();
-        CatalogMetadata catalogMetadata = getCatalogMetadataForWrite(session, catalogHandle);
+        QualifiedObjectName targetTableName = getRedirectedTableName(session, tableName);
+        if (!targetTableName.equals(tableName)) {
+            return new DropRedirected(targetTableName.asCatalogSchemaTableName());
+        }
+        CatalogMetadata catalogMetadata = getCatalogMetadataForWrite(session, tableName.getCatalogName());
         ConnectorMetadata metadata = catalogMetadata.getMetadata(session);
-        Optional<CatalogSchemaTableName> tableName = getTableNameIfSystemSecurity(session, catalogMetadata, tableHandle);
-        metadata.dropTable(session.toConnectorSession(catalogHandle), tableHandle.getConnectorHandle());
-        tableName.ifPresent(name -> systemSecurityMetadata.tableDropped(session, name));
+        ConnectorSession connectorSession = session.toConnectorSession(catalogMetadata.getCatalogHandle());
+
+        metadata.dropTable(connectorSession, tableName.asSchemaTableName());
+        if (catalogMetadata.getSecurityManagement() == SYSTEM) {
+            systemSecurityMetadata.tableDropped(session, tableName.asCatalogSchemaTableName());
+        }
+        return DropSuccess.getInstance();
     }
 
     @Override

core/trino-main/src/main/java/io/trino/testing/TestingMetadata.java

@@ -180,7 +180,13 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
     @Override
     public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
-        tables.remove(getTableName(tableHandle));
+        dropTable(session, getTableName(tableHandle));
+    }
+
+    @Override
+    public void dropTable(ConnectorSession session, SchemaTableName tableName)
+    {
+        tables.remove(tableName);
     }
 
     @Override

core/trino-main/src/test/java/io/trino/connector/MockConnector.java

@@ -522,6 +522,9 @@ public void createTable(ConnectorSession session, ConnectorTableMetadata tableMe
         @Override
         public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle) {}
 
+        @Override
+        public void dropTable(ConnectorSession session, SchemaTableName tableName) {}
+
         @Override
         public void renameTable(ConnectorSession session, ConnectorTableHandle tableHandle, SchemaTableName newTableName) {}
 

core/trino-main/src/test/java/io/trino/execution/BaseDataDefinitionTaskTest.java

@@ -39,6 +39,9 @@
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ColumnMetadata;
 import io.trino.spi.connector.ConnectorTableMetadata;
+import io.trino.spi.connector.DropRedirected;
+import io.trino.spi.connector.DropResult;
+import io.trino.spi.connector.DropSuccess;
 import io.trino.spi.connector.MaterializedViewNotFoundException;
 import io.trino.spi.connector.SchemaTableName;
 import io.trino.spi.connector.TestingColumnHandle;
@@ -302,9 +305,13 @@ public void createTable(Session session, String catalogName, ConnectorTableMetad
         }
 
         @Override
-        public void dropTable(Session session, TableHandle tableHandle)
+        public DropResult dropTable(Session session, QualifiedObjectName tableName)
         {
-            tables.remove(getTableName(tableHandle));
+            ConnectorTableMetadata connectorTableMetadata = tables.remove(tableName.asSchemaTableName());
+            if (connectorTableMetadata == null) {
+                return new DropRedirected(tableName.asCatalogSchemaTableName());
+            }
+            return DropSuccess.getInstance();
         }
 
         @Override

core/trino-main/src/test/java/io/trino/metadata/AbstractMockMetadata.java

@@ -36,6 +36,7 @@
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.Constraint;
 import io.trino.spi.connector.ConstraintApplicationResult;
+import io.trino.spi.connector.DropResult;
 import io.trino.spi.connector.JoinApplicationResult;
 import io.trino.spi.connector.JoinStatistics;
 import io.trino.spi.connector.JoinType;
@@ -318,7 +319,7 @@ public void setTableAuthorization(Session session, CatalogSchemaTableName table,
     }
 
     @Override
-    public void dropTable(Session session, TableHandle tableHandle)
+    public DropResult dropTable(Session session, QualifiedObjectName tableName)
     {
         throw new UnsupportedOperationException();
     }

core/trino-main/src/test/java/io/trino/metadata/CountingAccessMetadata.java

@@ -32,6 +32,7 @@
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.Constraint;
 import io.trino.spi.connector.ConstraintApplicationResult;
+import io.trino.spi.connector.DropResult;
 import io.trino.spi.connector.JoinApplicationResult;
 import io.trino.spi.connector.JoinStatistics;
 import io.trino.spi.connector.JoinType;
@@ -322,9 +323,9 @@ public void dropColumn(Session session, TableHandle tableHandle, ColumnHandle co
     }
 
     @Override
-    public void dropTable(Session session, TableHandle tableHandle)
+    public DropResult dropTable(Session session, QualifiedObjectName tableName)
     {
-        delegate.dropTable(session, tableHandle);
+        return delegate.dropTable(session, tableName);
     }
 
     @Override

core/trino-spi/src/main/java/io/trino/spi/StandardErrorCode.java

@@ -189,6 +189,7 @@ public enum StandardErrorCode
     EXCEEDED_TASK_DESCRIPTOR_STORAGE_CAPACITY(131082, INSUFFICIENT_RESOURCES),
 
     UNSUPPORTED_TABLE_TYPE(133001, EXTERNAL),
+    METADATA_NOT_FOUND(133002, EXTERNAL),
     /**/;
 
     // Connectors can use error codes starting at the range 0x0100_0000

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorMetadata.java

@@ -353,6 +353,16 @@ default void createTable(ConnectorSession session, ConnectorTableMetadata tableM
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support creating tables");
     }
 
+    /**
+     * Drops the specified table
+     *
+     * @throws RuntimeException if the table cannot be dropped
+     */
+    default void dropTable(ConnectorSession session, SchemaTableName schemaTableName)
+    {
+        dropTable(session, getTableHandle(session, schemaTableName));
+    }
+
     /**
      * Drops the specified table
      *

core/trino-spi/src/main/java/io/trino/spi/connector/DropRedirected.java

@@ -0,0 +1,36 @@
+/*
+ * 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 io.trino.spi.connector;
+
+public final class DropRedirected
+        extends DropResult
+{
+    private final CatalogSchemaTableName redirectedTarget;
+
+    public DropRedirected(CatalogSchemaTableName target)
+    {
+        this.redirectedTarget = target;
+    }
+
+    public CatalogSchemaTableName getRedirectedTarget()
+    {
+        return redirectedTarget;
+    }
+
+    @Override
+    public boolean isRedirected()
+    {
+        return true;
+    }
+}

core/trino-spi/src/main/java/io/trino/spi/connector/DropResult.java

@@ -0,0 +1,20 @@
+/*
+ * 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 io.trino.spi.connector;
+
+public abstract sealed class DropResult
+        permits DropSuccess, DropRedirected
+{
+    public abstract boolean isRedirected();
+}

core/trino-spi/src/main/java/io/trino/spi/connector/DropSuccess.java

@@ -0,0 +1,36 @@
+/*
+ * 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 io.trino.spi.connector;
+
+public final class DropSuccess
+        extends DropResult
+{
+    private static DropSuccess instance;
+
+    private DropSuccess() {}
+
+    public static DropSuccess getInstance()
+    {
+        if (instance == null) {
+            instance = new DropSuccess();
+        }
+        return instance;
+    }
+
+    @Override
+    public boolean isRedirected()
+    {
+        return false;
+    }
+}

lib/trino-plugin-toolkit/src/main/java/io/trino/plugin/base/classloader/ClassLoaderSafeConnectorMetadata.java

@@ -383,6 +383,14 @@ public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle
         }
     }
 
+    @Override
+    public void dropTable(ConnectorSession session, SchemaTableName tableName)
+    {
+        try (ThreadContextClassLoader ignored = new ThreadContextClassLoader(classLoader)) {
+            delegate.dropTable(session, tableName);
+        }
+    }
+
     @Override
     public void truncateTable(ConnectorSession session, ConnectorTableHandle tableHandle)
     {