Add support for isMaterializedView to Iceberg

This commit adds, among other things, test cases where we `DROP`
materialized view with corrupted metadata, which would otherwise fail.

In case of Hive catalogs we can bypass the metadata cache entirely, in
case of Glue catalogs the benefits are not that pronounced. But we still
avoid some potentially expensive operations and/or ones potentially
causing failures.

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

@@ -2796,6 +2796,12 @@ public Optional<ConnectorMaterializedViewDefinition> getMaterializedView(Connect
         return catalog.getMaterializedView(session, viewName);
     }
 
+    @Override
+    public boolean isMaterializedView(ConnectorSession session, SchemaTableName viewName)
+    {
+        return catalog.isMaterializedView(session, viewName);
+    }
+
     @Override
     public void renameMaterializedView(ConnectorSession session, SchemaTableName source, SchemaTableName target)
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/TrinoCatalog.java

@@ -159,6 +159,8 @@ void createMaterializedView(
 
     Optional<ConnectorMaterializedViewDefinition> getMaterializedView(ConnectorSession session, SchemaTableName viewName);
 
+    boolean isMaterializedView(ConnectorSession session, SchemaTableName viewName);
+
     void renameMaterializedView(ConnectorSession session, SchemaTableName source, SchemaTableName target);
 
     void updateColumnComment(ConnectorSession session, SchemaTableName schemaTableName, ColumnIdentity columnIdentity, Optional<String> comment);

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/glue/TrinoGlueCatalog.java

@@ -1246,6 +1246,28 @@ protected Optional<ConnectorMaterializedViewDefinition> doGetMaterializedView(Co
         return createMaterializedViewDefinition(session, viewName, table);
     }
 
+    @Override
+    public boolean isMaterializedView(ConnectorSession session, SchemaTableName viewName)
+    {
+        ConnectorMaterializedViewDefinition materializedViewDefinition = materializedViewCache.get(viewName);
+        if (materializedViewDefinition != null) {
+            return true;
+        }
+
+        if (tableMetadataCache.containsKey(viewName) || viewCache.containsKey(viewName)) {
+            // Entries in these caches are not materialized views.
+            return false;
+        }
+
+        Optional<com.amazonaws.services.glue.model.Table> maybeTable = getTableAndCacheMetadata(session, viewName);
+        if (maybeTable.isEmpty()) {
+            return false;
+        }
+
+        com.amazonaws.services.glue.model.Table table = maybeTable.get();
+        return isTrinoMaterializedView(getTableType(table), getTableParameters(table));
+    }
+
     private Optional<ConnectorMaterializedViewDefinition> createMaterializedViewDefinition(
             ConnectorSession session,
             SchemaTableName viewName,

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

@@ -663,6 +663,18 @@ protected Optional<ConnectorMaterializedViewDefinition> doGetMaterializedView(Co
                 storageTableName));
     }
 
+    @Override
+    public boolean isMaterializedView(ConnectorSession session, SchemaTableName viewName)
+    {
+        Optional<io.trino.plugin.hive.metastore.Table> tableOptional = metastore.getTable(viewName.getSchemaName(), viewName.getTableName());
+        if (tableOptional.isEmpty()) {
+            return false;
+        }
+
+        io.trino.plugin.hive.metastore.Table table = tableOptional.get();
+        return isTrinoMaterializedView(table.getTableType(), table.getParameters());
+    }
+
     @Override
     public void renameMaterializedView(ConnectorSession session, SchemaTableName source, SchemaTableName target)
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/jdbc/TrinoJdbcCatalog.java

@@ -412,6 +412,12 @@ public Optional<ConnectorMaterializedViewDefinition> getMaterializedView(Connect
         return Optional.empty();
     }
 
+    @Override
+    public boolean isMaterializedView(ConnectorSession session, SchemaTableName viewName)
+    {
+        return false;
+    }
+
     @Override
     public void renameMaterializedView(ConnectorSession session, SchemaTableName source, SchemaTableName target)
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/nessie/TrinoNessieCatalog.java

@@ -375,6 +375,12 @@ public Optional<ConnectorMaterializedViewDefinition> getMaterializedView(Connect
         return Optional.empty();
     }
 
+    @Override
+    public boolean isMaterializedView(ConnectorSession session, SchemaTableName viewName)
+    {
+        return false;
+    }
+
     @Override
     protected Optional<ConnectorMaterializedViewDefinition> doGetMaterializedView(ConnectorSession session, SchemaTableName schemaViewName)
     {

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/catalog/rest/TrinoRestCatalog.java

@@ -415,6 +415,12 @@ public Optional<ConnectorMaterializedViewDefinition> getMaterializedView(Connect
         return Optional.empty();
     }
 
+    @Override
+    public boolean isMaterializedView(ConnectorSession session, SchemaTableName viewName)
+    {
+        return false;
+    }
+
     @Override
     public void renameMaterializedView(ConnectorSession session, SchemaTableName source, SchemaTableName target)
     {

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

@@ -519,6 +519,24 @@ public void testDropTableWithMissingMetadataFile()
         assertFalse(fileSystem.listFiles(tableLocation).hasNext(), "Table location should not exist");
     }
 
+    @Test
+    public void testDropMaterializedViewWithMissingMetadataFile()
+            throws Exception
+    {
+        String viewName = "test_drop_mv_with_missing_metadata_file_" + randomNameSuffix();
+        assertUpdate("CREATE MATERIALIZED VIEW " + viewName + " AS SELECT 1 x, 'INDIA' y");
+
+        Location metadataLocation = Location.of(getMetadataLocation(viewName));
+
+        // Delete current metadata file
+        fileSystem.deleteFile(metadataLocation);
+        assertFalse(fileSystem.newInputFile(metadataLocation).exists(), "Current metadata file should not exist");
+
+        // try to drop mv
+        assertUpdate("DROP MATERIALIZED VIEW " + viewName);
+        assertFalse(getQueryRunner().tableExists(getSession(), viewName));
+    }
+
     @Test
     public void testDropTableWithMissingSnapshotFile()
             throws Exception

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

@@ -27,6 +27,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
@@ -232,13 +233,17 @@ protected void dropTableFromMetastore(String tableName)
     @Override
     protected String getMetadataLocation(String tableName)
     {
-        HiveMetastore metastore = new BridgingHiveMetastore(
+        Map<String, String> parameters = new BridgingHiveMetastore(
                 testingThriftHiveMetastoreBuilder()
                         .metastoreClient(hiveMinioDataLake.getHiveHadoop().getHiveMetastoreEndpoint())
-                        .build());
-        return metastore
+                        .build())
                 .getTable(schemaName, tableName).orElseThrow()
-                .getParameters().get("metadata_location");
+                .getParameters();
+        return Optional.ofNullable(parameters.get("storage_table"))
+                // this is a materialized view:
+                .map(this::getMetadataLocation)
+                // this is a plain table:
+                .orElseGet(() -> parameters.get("metadata_location"));
     }
 
     @Override

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

@@ -24,6 +24,8 @@
 import java.io.UncheckedIOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
+import java.util.Map;
+import java.util.Optional;
 
 import static com.google.common.io.MoreFiles.deleteRecursively;
 import static com.google.common.io.RecursiveDeleteOption.ALLOW_INSECURE;
@@ -80,9 +82,14 @@ protected void dropTableFromMetastore(String tableName)
     @Override
     protected String getMetadataLocation(String tableName)
     {
-        return metastore
+        Map<String, String> parameters = metastore
                 .getTable(getSession().getSchema().orElseThrow(), tableName).orElseThrow()
-                .getParameters().get("metadata_location");
+                .getParameters();
+        return Optional.ofNullable(parameters.get("storage_table"))
+                // this is a materialized view:
+                .map(this::getMetadataLocation)
+                // this is a plain table:
+                .orElseGet(() -> parameters.get("metadata_location"));
     }
 
     @Override

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

@@ -219,7 +219,7 @@ public void testDropMaterializedView()
 
         assertMetastoreInvocations("DROP MATERIALIZED VIEW test_drop_mview_view",
                 ImmutableMultiset.builder()
-                        .addCopies(GET_TABLE, 2)
+                        .add(GET_TABLE)
                         .addCopies(DROP_TABLE, 2)
                         .build());
     }

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/jdbc/TestIcebergJdbcCatalogConnectorSmokeTest.java

@@ -182,4 +182,11 @@ protected boolean isFileSorted(Location path, String sortColumnName)
         }
         return checkOrcFileSorting(fileSystem, path, sortColumnName);
     }
+
+    @Override
+    public void testDropMaterializedViewWithMissingMetadataFile()
+    {
+        assertThatThrownBy(super::testDropMaterializedViewWithMissingMetadataFile)
+                .hasMessageMatching("createMaterializedView is not supported for Iceberg JDBC catalogs");
+    }
 }

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/nessie/TestIcebergNessieCatalogConnectorSmokeTest.java

@@ -224,6 +224,13 @@ public void testDropTableWithMissingMetadataFile()
                 .hasMessageMatching("metadata location for register_table is not supported");
     }
 
+    @Override
+    public void testDropMaterializedViewWithMissingMetadataFile()
+    {
+        assertThatThrownBy(super::testDropMaterializedViewWithMissingMetadataFile)
+                .hasMessageMatching("createMaterializedView is not supported for Iceberg Nessie catalogs");
+    }
+
     @Override
     public void testDropTableWithMissingSnapshotFile()
     {

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/rest/TestIcebergTrinoRestCatalogConnectorSmokeTest.java

@@ -234,6 +234,13 @@ public void testDropTableWithMissingMetadataFile()
                 .hasMessageMatching("Failed to load table: (.*)");
     }
 
+    @Override
+    public void testDropMaterializedViewWithMissingMetadataFile()
+    {
+        assertThatThrownBy(super::testDropMaterializedViewWithMissingMetadataFile)
+                .hasMessageMatching("createMaterializedView is not supported for Iceberg REST catalog");
+    }
+
     @Override
     public void testDropTableWithMissingSnapshotFile()
     {