Reuse location when replacing view for Iceberg Rest

If replace and useUniqueTableLocation, load the existing view and reuse its location.

Author: codluca

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

@@ -609,12 +609,19 @@ public void createView(ConnectorSession session, SchemaTableName schemaViewName,
         definition.getComment().ifPresent(comment -> properties.put(COMMENT, comment));
         Schema schema = IcebergUtil.schemaFromViewColumns(typeManager, definition.getColumns());
         ViewBuilder viewBuilder = restSessionCatalog.buildView(convert(session), toRemoteView(session, schemaViewName, true));
+        String tableLocation = defaultTableLocation(session, schemaViewName);
+        if (replace && useUniqueTableLocation) {
+            Optional<View> view = getIcebergView(session, schemaViewName, true);
+            if (view.isPresent()) {
+                tableLocation = view.get().location();
+            }
+        }
         viewBuilder = viewBuilder.withSchema(schema)
                 .withQuery("trino", definition.getOriginalSql())
                 .withDefaultNamespace(toRemoteNamespace(session, toNamespace(schemaViewName.getSchemaName())))
                 .withDefaultCatalog(definition.getCatalog().orElse(null))
                 .withProperties(properties.buildOrThrow())
-                .withLocation(defaultTableLocation(session, schemaViewName));
+                .withLocation(tableLocation);
         try {
             if (replace) {
                 viewBuilder.createOrReplace();

plugin/trino-iceberg/src/test/java/io/trino/plugin/iceberg/catalog/BaseTrinoCatalogTest.java

@@ -423,6 +423,70 @@ public void testView()
         }
     }
 
+    @Test
+    public void testViewUnique()
+            throws IOException
+    {
+        TrinoCatalog catalog = createTrinoCatalog(true);
+        Path tmpDirectory = Files.createTempDirectory("iceberg_catalog_test_create_view_");
+        tmpDirectory.toFile().deleteOnExit();
+
+        String namespace = "test_create_view_" + randomNameSuffix();
+        String viewName = "viewName";
+        String renamedViewName = "renamedViewName";
+        SchemaTableName schemaTableName = new SchemaTableName(namespace, viewName);
+        SchemaTableName renamedSchemaTableName = new SchemaTableName(namespace, renamedViewName);
+        ConnectorViewDefinition viewDefinition = new ConnectorViewDefinition(
+                "SELECT name FROM local.tiny.nation",
+                Optional.empty(),
+                Optional.empty(),
+                ImmutableList.of(
+                        new ConnectorViewDefinition.ViewColumn("name", VarcharType.createUnboundedVarcharType().getTypeId(), Optional.empty())),
+                Optional.empty(),
+                Optional.of(SESSION.getUser()),
+                false,
+                ImmutableList.of());
+
+        try {
+            catalog.createNamespace(SESSION, namespace, defaultNamespaceProperties(namespace), new TrinoPrincipal(PrincipalType.USER, SESSION.getUser()));
+            catalog.createView(SESSION, schemaTableName, viewDefinition, false);
+
+            assertThat(catalog.listTables(SESSION, Optional.of(namespace)).stream()).contains(new TableInfo(schemaTableName, getViewType()));
+
+            Map<SchemaTableName, ConnectorViewDefinition> views = catalog.getViews(SESSION, Optional.of(schemaTableName.getSchemaName()));
+            assertThat(views).hasSize(1);
+            assertViewDefinition(views.get(schemaTableName), viewDefinition);
+            assertViewDefinition(catalog.getView(SESSION, schemaTableName).orElseThrow(), viewDefinition);
+
+            catalog.createView(SESSION, schemaTableName, viewDefinition, true);
+            assertThat(catalog.listTables(SESSION, Optional.of(namespace)).stream()).contains(new TableInfo(schemaTableName, getViewType()));
+            views = catalog.getViews(SESSION, Optional.of(schemaTableName.getSchemaName()));
+            assertThat(views).hasSize(1);
+            assertViewDefinition(views.get(schemaTableName), viewDefinition);
+            assertViewDefinition(catalog.getView(SESSION, schemaTableName).orElseThrow(), viewDefinition);
+
+            catalog.renameView(SESSION, schemaTableName, renamedSchemaTableName);
+            assertThat(catalog.listTables(SESSION, Optional.of(namespace)).stream().map(TableInfo::tableName).toList()).doesNotContain(schemaTableName);
+            views = catalog.getViews(SESSION, Optional.of(schemaTableName.getSchemaName()));
+            assertThat(views).hasSize(1);
+            assertViewDefinition(views.get(renamedSchemaTableName), viewDefinition);
+            assertViewDefinition(catalog.getView(SESSION, renamedSchemaTableName).orElseThrow(), viewDefinition);
+            assertThat(catalog.getView(SESSION, schemaTableName)).isEmpty();
+
+            catalog.dropView(SESSION, renamedSchemaTableName);
+            assertThat(catalog.listTables(SESSION, Optional.empty()).stream().map(TableInfo::tableName).toList())
+                    .doesNotContain(renamedSchemaTableName);
+        }
+        finally {
+            try {
+                catalog.dropNamespace(SESSION, namespace);
+            }
+            catch (Exception e) {
+                LOG.warn("Failed to clean up namespace: %s", namespace);
+            }
+        }
+    }
+
     protected ExtendedRelationType getViewType()
     {
         return TRINO_VIEW;

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

@@ -160,6 +160,14 @@ public void testView()
                 .hasMessageContaining("createView is not supported for Iceberg Nessie catalogs");
     }
 
+    @Test
+    @Override
+    public void testViewUnique()
+    {
+        assertThatThrownBy(super::testView)
+                .hasMessageContaining("createView is not supported for Iceberg Nessie catalogs");
+    }
+
     @Test
     @Override
     public void testNonLowercaseNamespace()

plugin/trino-iceberg/src/test/java/org/apache/iceberg/snowflake/TestTrinoSnowflakeCatalog.java

@@ -349,6 +349,26 @@ public void testView()
                 .hasMessageContaining("Views are not supported for the Snowflake Iceberg catalog");
     }
 
+    // Overridden to assert correct method calls as the tested feature is not supported
+    @Test
+    @Override
+    public void testViewUnique()
+    {
+        ConnectorViewDefinition viewDefinition = new ConnectorViewDefinition(
+                "SELECT name FROM local.tiny.%s".formatted(TpchTable.NATION.getTableName()),
+                Optional.empty(),
+                Optional.empty(),
+                ImmutableList.of(
+                        new ConnectorViewDefinition.ViewColumn("name", VarcharType.createVarcharType(25).getTypeId(), Optional.empty())),
+                Optional.empty(),
+                Optional.of(SESSION.getUser()),
+                false,
+                ImmutableList.of());
+        TrinoCatalog catalog = createTrinoCatalog(true);
+        assertThatThrownBy(() -> catalog.createView(SESSION, SchemaTableName.schemaTableName(SNOWFLAKE_TEST_SCHEMA, TpchTable.NATION.getTableName()), viewDefinition, true))
+                .hasMessageContaining("Views are not supported for the Snowflake Iceberg catalog");
+    }
+
     @Test
     @Override
     public void testListTables()