Delete Iceberg table data on DROP

plugin/trino-hive/src/main/java/io/trino/plugin/hive/metastore/file/FileHiveMetastore.java

@@ -524,12 +524,10 @@ public synchronized void dropTable(String databaseName, String tableName, boolea
 
         Path tableMetadataDirectory = getTableMetadataDirectory(databaseName, tableName);
 
-        // It is safe to delete the whole meta directory for external tables and views
-        if (!table.getTableType().equals(MANAGED_TABLE.name()) || deleteData) {
+        if (deleteData) {
             deleteDirectoryAndSchema(TABLE, tableMetadataDirectory);
         }
         else {
-            // in this case we only want to delete the metadata of a managed table
             deleteSchemaFile(TABLE, tableMetadataDirectory);
             deleteTablePrivileges(table);
         }

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/TrinoHiveCatalog.java

@@ -82,6 +82,7 @@
 import static io.trino.plugin.hive.metastore.StorageFormat.VIEW_STORAGE_FORMAT;
 import static io.trino.plugin.hive.util.HiveUtil.isHiveSystemSchema;
 import static io.trino.plugin.hive.util.HiveWriteUtils.getTableDefaultLocation;
+import static io.trino.plugin.iceberg.IcebergErrorCode.ICEBERG_FILESYSTEM_ERROR;
 import static io.trino.plugin.iceberg.IcebergMaterializedViewDefinition.decodeMaterializedViewData;
 import static io.trino.plugin.iceberg.IcebergMaterializedViewDefinition.encodeMaterializedViewData;
 import static io.trino.plugin.iceberg.IcebergMaterializedViewDefinition.fromConnectorMaterializedViewDefinition;
@@ -96,11 +97,13 @@
 import static io.trino.spi.StandardErrorCode.NOT_SUPPORTED;
 import static io.trino.spi.StandardErrorCode.SCHEMA_NOT_EMPTY;
 import static io.trino.spi.StandardErrorCode.TABLE_NOT_FOUND;
+import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static java.util.UUID.randomUUID;
 import static org.apache.hadoop.hive.metastore.TableType.VIRTUAL_VIEW;
 import static org.apache.iceberg.BaseMetastoreTableOperations.ICEBERG_TABLE_TYPE_VALUE;
 import static org.apache.iceberg.BaseMetastoreTableOperations.TABLE_TYPE_PROP;
+import static org.apache.iceberg.CatalogUtil.dropTableData;
 import static org.apache.iceberg.TableMetadata.newTableMetadata;
 import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
 import static org.apache.iceberg.TableProperties.OBJECT_STORE_PATH;
@@ -302,14 +305,40 @@ public List<SchemaTableName> listTables(ConnectorSession session, Optional<Strin
     public void dropTable(ConnectorSession session, SchemaTableName schemaTableName)
     {
         // TODO: support path override in Iceberg table creation: https://github.com/trinodb/trino/issues/8861
-        Table table = loadTable(session, schemaTableName);
+        BaseTable table = (BaseTable) loadTable(session, schemaTableName);
+        TableMetadata metadata = table.operations().current();
         if (table.properties().containsKey(OBJECT_STORE_PATH) ||
                 table.properties().containsKey(WRITE_NEW_DATA_LOCATION) ||
                 table.properties().containsKey(WRITE_METADATA_LOCATION) ||
                 table.properties().containsKey(WRITE_DATA_LOCATION)) {
             throw new TrinoException(NOT_SUPPORTED, "Table " + schemaTableName + " contains Iceberg path override properties and cannot be dropped from Trino");
         }
-        metastore.dropTable(schemaTableName.getSchemaName(), schemaTableName.getTableName(), true);
+
+        io.trino.plugin.hive.metastore.Table metastoreTable = metastore.getTable(schemaTableName.getSchemaName(), schemaTableName.getTableName())
+                .orElseThrow(() -> new TableNotFoundException(schemaTableName));
+        metastore.dropTable(
+                schemaTableName.getSchemaName(),
+                schemaTableName.getTableName(),
+                false /* do not delete data */);
+        // Use the Iceberg routine for dropping the table data because the data files
+        // of the Iceberg table may be located in different locations
+        dropTableData(table.io(), metadata);
+        deleteTableDirectory(session, metastoreTable);
+    }
+
+    private void deleteTableDirectory(ConnectorSession session, io.trino.plugin.hive.metastore.Table metastoreTable)
+    {
+        Path tablePath = new Path(metastoreTable.getStorage().getLocation());
+        try {
+            FileSystem fileSystem = hdfsEnvironment.getFileSystem(new HdfsContext(session), tablePath);
+            fileSystem.delete(tablePath, true);
+        }
+        catch (IOException e) {
+            throw new TrinoException(
+                    ICEBERG_FILESYSTEM_ERROR,
+                    format("Failed to delete directory %s of the table %s.%s", tablePath, metastoreTable.getDatabaseName(), metastoreTable.getTableName()),
+                    e);
+        }
     }
 
     @Override

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/BaseIcebergConnectorTest.java

@@ -1097,6 +1097,9 @@ private void testCreateTableLikeForFormat(IcebergFileFormat otherFormat)
         File tempDir = getDistributedQueryRunner().getCoordinator().getBaseDataDir().toFile();
         String tempDirPath = tempDir.toURI().toASCIIString() + randomTableSuffix();
 
+        // LIKE source INCLUDING PROPERTIES copies all the properties of the source table, including the `location`.
+        // For this reason the source and the copied table will share the same directory.
+        // This test does not drop intentionally the created tables to avoid affecting the source table or the information_schema.
         assertUpdate(format("CREATE TABLE test_create_table_like_original (col1 INTEGER, aDate DATE) WITH(format = '%s', location = '%s', partitioning = ARRAY['aDate'])", format, tempDirPath));
         assertEquals(getTablePropertiesString("test_create_table_like_original"), "WITH (\n" +
                 format("   format = '%s',\n", format) +
@@ -1107,12 +1110,10 @@ private void testCreateTableLikeForFormat(IcebergFileFormat otherFormat)
         assertUpdate("CREATE TABLE test_create_table_like_copy0 (LIKE test_create_table_like_original, col2 INTEGER)");
         assertUpdate("INSERT INTO test_create_table_like_copy0 (col1, aDate, col2) VALUES (1, CAST('1950-06-28' AS DATE), 3)", 1);
         assertQuery("SELECT * from test_create_table_like_copy0", "VALUES(1, CAST('1950-06-28' AS DATE), 3)");
-        dropTable("test_create_table_like_copy0");
 
         assertUpdate("CREATE TABLE test_create_table_like_copy1 (LIKE test_create_table_like_original)");
         assertEquals(getTablePropertiesString("test_create_table_like_copy1"), "WITH (\n" +
                 format("   format = '%s',\n   location = '%s'\n)", format, tempDir + "/iceberg_data/tpch/test_create_table_like_copy1"));
-        dropTable("test_create_table_like_copy1");
 
         assertUpdate("CREATE TABLE test_create_table_like_copy2 (LIKE test_create_table_like_original EXCLUDING PROPERTIES)");
         assertEquals(getTablePropertiesString("test_create_table_like_copy2"), "WITH (\n" +
@@ -1125,17 +1126,13 @@ private void testCreateTableLikeForFormat(IcebergFileFormat otherFormat)
                 format("   location = '%s',\n", tempDirPath) +
                 "   partitioning = ARRAY['adate']\n" +
                 ")");
-        dropTable("test_create_table_like_copy3");
 
         assertUpdate(format("CREATE TABLE test_create_table_like_copy4 (LIKE test_create_table_like_original INCLUDING PROPERTIES) WITH (format = '%s')", otherFormat));
         assertEquals(getTablePropertiesString("test_create_table_like_copy4"), "WITH (\n" +
                 format("   format = '%s',\n", otherFormat) +
                 format("   location = '%s',\n", tempDirPath) +
                 "   partitioning = ARRAY['adate']\n" +
                 ")");
-        dropTable("test_create_table_like_copy4");
-
-        dropTable("test_create_table_like_original");
     }
 
     private String getTablePropertiesString(String tableName)
@@ -3367,4 +3364,16 @@ public void testTargetMaxFileSize()
                 // so just to be safe we check if it is not much bigger
                 .forEach(row -> assertThat((Long) row.getField(0)).isBetween(1L, maxSize.toBytes() * 3));
     }
+
+    @Test
+    public void testDroppingIcebergAndCreatingANewTableWithTheSameNameShouldBePossible()
+    {
+        assertUpdate("CREATE TABLE test_iceberg_recreate (a_int) AS VALUES (1)", 1);
+        assertThat(query("SELECT min(a_int) FROM test_iceberg_recreate")).matches("VALUES 1");
+        dropTable("test_iceberg_recreate");
+
+        assertUpdate("CREATE TABLE test_iceberg_recreate (a_varchar) AS VALUES ('Trino')", 1);
+        assertThat(query("SELECT min(a_varchar) FROM test_iceberg_recreate")).matches("VALUES CAST('Trino' AS varchar)");
+        dropTable("test_iceberg_recreate");
+    }
 }

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/DataFileRecord.java

@@ -0,0 +1,134 @@
+/*
+ * 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.plugin.iceberg;
+
+import com.google.common.collect.ImmutableMap;
+import io.trino.testing.MaterializedRow;
+
+import java.util.Map;
+
+import static org.testng.Assert.assertEquals;
+
+public class DataFileRecord
+{
+    private final int content;
+    private final String filePath;
+    private final String fileFormat;
+    private final long recordCount;
+    private final long fileSizeInBytes;
+    private final Map<Integer, Long> columnSizes;
+    private final Map<Integer, Long> valueCounts;
+    private final Map<Integer, Long> nullValueCounts;
+    private final Map<Integer, Long> nanValueCounts;
+    private final Map<Integer, String> lowerBounds;
+    private final Map<Integer, String> upperBounds;
+
+    public static DataFileRecord toDataFileRecord(MaterializedRow row)
+    {
+        assertEquals(row.getFieldCount(), 14);
+        return new DataFileRecord(
+                (int) row.getField(0),
+                (String) row.getField(1),
+                (String) row.getField(2),
+                (long) row.getField(3),
+                (long) row.getField(4),
+                row.getField(5) != null ? ImmutableMap.copyOf((Map<Integer, Long>) row.getField(5)) : null,
+                row.getField(6) != null ? ImmutableMap.copyOf((Map<Integer, Long>) row.getField(6)) : null,
+                row.getField(7) != null ? ImmutableMap.copyOf((Map<Integer, Long>) row.getField(7)) : null,
+                row.getField(8) != null ? ImmutableMap.copyOf((Map<Integer, Long>) row.getField(8)) : null,
+                row.getField(9) != null ? ImmutableMap.copyOf((Map<Integer, String>) row.getField(9)) : null,
+                row.getField(10) != null ? ImmutableMap.copyOf((Map<Integer, String>) row.getField(10)) : null);
+    }
+
+    private DataFileRecord(
+            int content,
+            String filePath,
+            String fileFormat,
+            long recordCount,
+            long fileSizeInBytes,
+            Map<Integer, Long> columnSizes,
+            Map<Integer, Long> valueCounts,
+            Map<Integer, Long> nullValueCounts,
+            Map<Integer, Long> nanValueCounts,
+            Map<Integer, String> lowerBounds,
+            Map<Integer, String> upperBounds)
+    {
+        this.content = content;
+        this.filePath = filePath;
+        this.fileFormat = fileFormat;
+        this.recordCount = recordCount;
+        this.fileSizeInBytes = fileSizeInBytes;
+        this.columnSizes = columnSizes;
+        this.valueCounts = valueCounts;
+        this.nullValueCounts = nullValueCounts;
+        this.nanValueCounts = nanValueCounts;
+        this.lowerBounds = lowerBounds;
+        this.upperBounds = upperBounds;
+    }
+
+    public int getContent()
+    {
+        return content;
+    }
+
+    public String getFilePath()
+    {
+        return filePath;
+    }
+
+    public String getFileFormat()
+    {
+        return fileFormat;
+    }
+
+    public long getRecordCount()
+    {
+        return recordCount;
+    }
+
+    public long getFileSizeInBytes()
+    {
+        return fileSizeInBytes;
+    }
+
+    public Map<Integer, Long> getColumnSizes()
+    {
+        return columnSizes;
+    }
+
+    public Map<Integer, Long> getValueCounts()
+    {
+        return valueCounts;
+    }
+
+    public Map<Integer, Long> getNullValueCounts()
+    {
+        return nullValueCounts;
+    }
+
+    public Map<Integer, Long> getNanValueCounts()
+    {
+        return nanValueCounts;
+    }
+
+    public Map<Integer, String> getLowerBounds()
+    {
+        return lowerBounds;
+    }
+
+    public Map<Integer, String> getUpperBounds()
+    {
+        return upperBounds;
+    }
+}