Support drop iceberg table when metadata or snapshot is missing and refactor code for delta-lake

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

@@ -18,8 +18,8 @@
 import io.trino.execution.warnings.WarningCollector;
 import io.trino.metadata.Metadata;
 import io.trino.metadata.QualifiedObjectName;
-import io.trino.metadata.RedirectionAwareTableHandle;
 import io.trino.security.AccessControl;
+import io.trino.spi.connector.TableNotFoundException;
 import io.trino.sql.tree.DropTable;
 import io.trino.sql.tree.Expression;
 
@@ -76,17 +76,17 @@ 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()) {
+        QualifiedObjectName targetTable = metadata.getRedirectedTableName(session, originalTableName);
+
+        try {
+            accessControl.checkCanDropTable(session.toSecurityContext(), targetTable);
+            metadata.dropTable(session, targetTable);
+        }
+        catch (TableNotFoundException e) {
             if (!statement.isExists()) {
                 throw semanticException(TABLE_NOT_FOUND, statement, "Table '%s' does not exist", originalTableName);
             }
-            return immediateVoidFuture();
         }
-        QualifiedObjectName tableName = redirectionAwareTableHandle.getRedirectedTableName().orElse(originalTableName);
-        accessControl.checkCanDropTable(session.toSecurityContext(), tableName);
-
-        metadata.dropTable(session, redirectionAwareTableHandle.getTableHandle().get(), tableName.asCatalogSchemaTableName());
 
         return immediateVoidFuture();
     }

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

@@ -264,9 +264,18 @@ Optional<TableExecuteHandle> getTableHandleForExecute(
      * Drops the specified table
      *
      * @throws RuntimeException if the table cannot be dropped or table handle is no longer valid
+     *
+     * @deprecated use {@link #dropTable(Session, QualifiedObjectName)}
      */
     void dropTable(Session session, TableHandle tableHandle, CatalogSchemaTableName tableName);
 
+    /**
+     * Drops the specified table
+     *
+     * @throws RuntimeException if the table cannot be dropped
+     */
+    void dropTable(Session session, QualifiedObjectName tableName);
+
     /**
      * Truncates the specified table
      */
@@ -708,6 +717,11 @@ default boolean isMaterializedView(Session session, QualifiedObjectName viewName
      */
     RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session session, QualifiedObjectName tableName, Optional<TableVersion> startVersion, Optional<TableVersion> endVersion);
 
+    /**
+     * Get the table name after performing redirection.
+     */
+    QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName tableName);
+
     /**
      * Returns true if the connector reports number of written bytes for an existing table. Otherwise, it returns false.
      */

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

@@ -799,6 +799,20 @@ public void dropTable(Session session, TableHandle tableHandle, CatalogSchemaTab
         }
     }
 
+    @Override
+    public void dropTable(Session session, QualifiedObjectName tableName)
+    {
+        CatalogMetadata catalogMetadata = getCatalogMetadataForWrite(session, tableName.getCatalogName());
+        ConnectorSession connectorSession = session.toConnectorSession(catalogMetadata.getCatalogHandle());
+        ConnectorMetadata metadata = catalogMetadata.getMetadata(session);
+
+        metadata.dropTable(connectorSession, tableName.asSchemaTableName());
+
+        if (catalogMetadata.getSecurityManagement() == SYSTEM) {
+            systemSecurityMetadata.tableDropped(session, tableName.asCatalogSchemaTableName());
+        }
+    }
+
     @Override
     public void truncateTable(Session session, TableHandle tableHandle)
     {
@@ -1486,7 +1500,8 @@ public Optional<TableScanRedirectApplicationResult> applyTableScanRedirect(Sessi
         return metadata.applyTableScanRedirect(connectorSession, tableHandle.getConnectorHandle());
     }
 
-    private QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName originalTableName)
+    @Override
+    public QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName originalTableName)
     {
         requireNonNull(session, "session is null");
         requireNonNull(originalTableName, "originalTableName is null");

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

@@ -186,7 +186,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

@@ -577,6 +577,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

@@ -26,6 +26,7 @@
 import io.trino.metadata.MaterializedViewPropertyManager;
 import io.trino.metadata.MetadataManager;
 import io.trino.metadata.QualifiedObjectName;
+import io.trino.metadata.QualifiedTablePrefix;
 import io.trino.metadata.ResolvedFunction;
 import io.trino.metadata.TableHandle;
 import io.trino.metadata.TableMetadata;
@@ -43,6 +44,7 @@
 import io.trino.spi.connector.ConnectorTableMetadata;
 import io.trino.spi.connector.MaterializedViewNotFoundException;
 import io.trino.spi.connector.SchemaTableName;
+import io.trino.spi.connector.TableNotFoundException;
 import io.trino.spi.connector.TestingColumnHandle;
 import io.trino.spi.function.OperatorType;
 import io.trino.spi.resourcegroups.ResourceGroupId;
@@ -318,6 +320,15 @@ public void dropTable(Session session, TableHandle tableHandle, CatalogSchemaTab
             tables.remove(tableName.getSchemaTableName());
         }
 
+        @Override
+        public void dropTable(Session session, QualifiedObjectName tableName)
+        {
+            SchemaTableName schemaTableName = tableName.asSchemaTableName();
+            if (tables.remove(schemaTableName) == null) {
+                throw new TableNotFoundException(schemaTableName);
+            }
+        }
+
         @Override
         public void renameTable(Session session, TableHandle tableHandle, CatalogSchemaTableName currentTableName, QualifiedObjectName newTableName)
         {
@@ -380,6 +391,18 @@ private SchemaTableName getTableName(TableHandle tableHandle)
             return ((TestingTableHandle) tableHandle.getConnectorHandle()).getTableName();
         }
 
+        @Override
+        public List<QualifiedObjectName> listTables(Session session, QualifiedTablePrefix prefix)
+        {
+            return tables.keySet().stream().map(table -> QualifiedObjectName.convertFromSchemaTableName(TEST_CATALOG_NAME).apply(table)).toList();
+        }
+
+        @Override
+        public QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName tableName)
+        {
+            return tableName;
+        }
+
         @Override
         public Map<String, ColumnHandle> getColumnHandles(Session session, TableHandle tableHandle)
         {

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

@@ -336,6 +336,12 @@ public void dropTable(Session session, TableHandle tableHandle, CatalogSchemaTab
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public void dropTable(Session session, QualifiedObjectName tableName)
+    {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public void truncateTable(Session session, TableHandle tableHandle)
     {
@@ -878,6 +884,12 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio
         throw new UnsupportedOperationException();
     }
 
+    @Override
+    public QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName tableName)
+    {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     public boolean supportsReportingWrittenBytes(Session session, TableHandle tableHandle)
     {

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

@@ -340,6 +340,12 @@ public void dropTable(Session session, TableHandle tableHandle, CatalogSchemaTab
         delegate.dropTable(session, tableHandle, tableName);
     }
 
+    @Override
+    public void dropTable(Session session, QualifiedObjectName tableName)
+    {
+        delegate.dropTable(session, tableName);
+    }
+
     @Override
     public void truncateTable(Session session, TableHandle tableHandle)
     {
@@ -857,6 +863,12 @@ public RedirectionAwareTableHandle getRedirectionAwareTableHandle(Session sessio
         return delegate.getRedirectionAwareTableHandle(session, tableName, startVersion, endVersion);
     }
 
+    @Override
+    public QualifiedObjectName getRedirectedTableName(Session session, QualifiedObjectName tableName)
+    {
+        return delegate.getRedirectedTableName(session, tableName);
+    }
+
     @Override
     public boolean supportsReportingWrittenBytes(Session session, TableHandle tableHandle)
     {

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

@@ -357,11 +357,24 @@ 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
      *
      * @throws RuntimeException if the table cannot be dropped or table handle is no longer valid
+     *
+     * @deprecated use {@link #dropTable(ConnectorSession, SchemaTableName)}
      */
+    @Deprecated
     default void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
         throw new TrinoException(NOT_SUPPORTED, "This connector does not support dropping tables");

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

@@ -393,6 +393,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)
     {

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeMetadata.java

@@ -172,6 +172,7 @@
 import static io.trino.plugin.deltalake.DeltaLakeColumnType.SYNTHESIZED;
 import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_BAD_WRITE;
 import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_INVALID_SCHEMA;
+import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_INVALID_TABLE;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.getHiveCatalogName;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isCollectExtendedStatisticsColumnStatisticsOnWrite;
 import static io.trino.plugin.deltalake.DeltaLakeSessionProperties.isExtendedStatisticsEnabled;
@@ -415,8 +416,10 @@ public DeltaLakeTableHandle getTableHandle(ConnectorSession session, SchemaTable
         }
 
         TableSnapshot tableSnapshot = metastore.getSnapshot(dataTableName, session);
-        Optional<MetadataEntry> metadata = metastore.getMetadata(tableSnapshot, session);
-        metadata.ifPresent(metadataEntry -> verifySupportedColumnMapping(getColumnMappingMode(metadataEntry)));
+        MetadataEntry metadata = metastore.getMetadata(tableSnapshot, session)
+                .orElseThrow(() -> new TrinoException(DELTA_LAKE_INVALID_TABLE, "Metadata not found in transaction log for table " + tableName));
+
+        verifySupportedColumnMapping(getColumnMappingMode(metadata));
         return new DeltaLakeTableHandle(
                 dataTableName.getSchemaName(),
                 dataTableName.getTableName(),
@@ -1766,12 +1769,16 @@ public Optional<Object> getInfo(ConnectorTableHandle table)
     @Override
     public void dropTable(ConnectorSession session, ConnectorTableHandle tableHandle)
     {
-        DeltaLakeTableHandle handle = (DeltaLakeTableHandle) tableHandle;
+        dropTable(session, ((DeltaLakeTableHandle) tableHandle).getSchemaTableName());
+    }
 
-        Table table = metastore.getTable(handle.getSchemaName(), handle.getTableName())
-                .orElseThrow(() -> new TableNotFoundException(handle.getSchemaTableName()));
+    @Override
+    public void dropTable(ConnectorSession session, SchemaTableName schemaTableName)
+    {
+        Table table = metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName())
+                .orElseThrow(() -> new TableNotFoundException(schemaTableName));
 
-        metastore.dropTable(session, handle.getSchemaName(), handle.getTableName(), table.getTableType().equals(MANAGED_TABLE.toString()));
+        metastore.dropTable(session, schemaTableName.getSchemaName(), schemaTableName.getTableName(), table.getTableType().equals(MANAGED_TABLE.toString()));
     }
 
     @Override
@@ -2030,7 +2037,7 @@ public Optional<ConstraintApplicationResult<ConnectorTableHandle>> applyFilter(C
                 tableName.getSchemaName(),
                 tableName.getTableName(),
                 tableHandle.getLocation(),
-                Optional.of(tableHandle.getMetadataEntry()),
+                tableHandle.getMetadataEntry(),
                 // Do not simplify the enforced constraint, the connector is guaranteeing the constraint will be applied as is.
                 // The unenforced constraint will still be checked by the engine.
                 tableHandle.getEnforcedPartitionConstraint()
@@ -2157,7 +2164,7 @@ public ConnectorAnalyzeMetadata getStatisticsCollectionMetadata(ConnectorSession
                 handle.getSchemaTableName().getSchemaName(),
                 handle.getSchemaTableName().getTableName(),
                 handle.getLocation(),
-                Optional.of(metadata),
+                metadata,
                 TupleDomain.all(),
                 TupleDomain.all(),
                 Optional.empty(),

plugin/trino-delta-lake/src/main/java/io/trino/plugin/deltalake/DeltaLakeTableHandle.java

@@ -18,7 +18,6 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.airlift.units.DataSize;
 import io.trino.plugin.deltalake.transactionlog.MetadataEntry;
-import io.trino.spi.TrinoException;
 import io.trino.spi.connector.ColumnHandle;
 import io.trino.spi.connector.ConnectorTableHandle;
 import io.trino.spi.connector.SchemaTableName;
@@ -30,7 +29,6 @@
 import java.util.Set;
 
 import static com.google.common.base.Preconditions.checkArgument;
-import static io.trino.plugin.deltalake.DeltaLakeErrorCode.DELTA_LAKE_INVALID_SCHEMA;
 import static io.trino.plugin.deltalake.DeltaLakeTableHandle.WriteType.UPDATE;
 import static java.util.Objects.requireNonNull;
 
@@ -47,7 +45,7 @@ public enum WriteType
     private final String schemaName;
     private final String tableName;
     private final String location;
-    private final Optional<MetadataEntry> metadataEntry;
+    private final MetadataEntry metadataEntry;
     private final TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint;
     private final TupleDomain<DeltaLakeColumnHandle> nonPartitionConstraint;
     private final Optional<WriteType> writeType;
@@ -72,7 +70,7 @@ public DeltaLakeTableHandle(
             @JsonProperty("schemaName") String schemaName,
             @JsonProperty("tableName") String tableName,
             @JsonProperty("location") String location,
-            @JsonProperty("metadataEntry") Optional<MetadataEntry> metadataEntry,
+            @JsonProperty("metadataEntry") MetadataEntry metadataEntry,
             @JsonProperty("enforcedPartitionConstraint") TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint,
             @JsonProperty("nonPartitionConstraint") TupleDomain<DeltaLakeColumnHandle> nonPartitionConstraint,
             @JsonProperty("writeType") Optional<WriteType> writeType,
@@ -105,7 +103,7 @@ public DeltaLakeTableHandle(
             String schemaName,
             String tableName,
             String location,
-            Optional<MetadataEntry> metadataEntry,
+            MetadataEntry metadataEntry,
             TupleDomain<DeltaLakeColumnHandle> enforcedPartitionConstraint,
             TupleDomain<DeltaLakeColumnHandle> nonPartitionConstraint,
             Optional<WriteType> writeType,
@@ -143,7 +141,7 @@ public DeltaLakeTableHandle withProjectedColumns(Set<ColumnHandle> projectedColu
                 getSchemaName(),
                 getTableName(),
                 getLocation(),
-                Optional.of(getMetadataEntry()),
+                getMetadataEntry(),
                 getEnforcedPartitionConstraint(),
                 getNonPartitionConstraint(),
                 getWriteType(),
@@ -196,7 +194,7 @@ public String getLocation()
     @JsonProperty
     public MetadataEntry getMetadataEntry()
     {
-        return metadataEntry.orElseThrow(() -> new TrinoException(DELTA_LAKE_INVALID_SCHEMA, "Metadata not found in transaction log for " + tableName));
+        return metadataEntry;
     }
 
     @JsonProperty