Core: Purge support on drop view operation

.palantir/revapi.yml

@@ -873,6 +873,11 @@ acceptedBreaks:
       new: "method void org.apache.iceberg.encryption.Ciphers::<init>()"
       justification: "Static utility class - should not have public constructor"
   "1.4.0":
+    org.apache.iceberg:iceberg-api:
+    - code: "java.method.addedToInterface"
+      new: "method boolean org.apache.iceberg.catalog.ViewCatalog::dropView(org.apache.iceberg.catalog.TableIdentifier,\
+        \ boolean)"
+      justification: "Added purge support for drop view"
     org.apache.iceberg:iceberg-core:
     - code: "java.class.defaultSerializationChanged"
       old: "class org.apache.iceberg.mapping.NameMapping"

api/src/main/java/org/apache/iceberg/catalog/ViewCatalog.java

@@ -83,7 +83,20 @@ default boolean viewExists(TableIdentifier identifier) {
    * @param identifier a view identifier
    * @return true if the view was dropped, false if the view did not exist
    */
-  boolean dropView(TableIdentifier identifier);
+  default boolean dropView(TableIdentifier identifier) {
+    return dropView(identifier, true);
+  }
+
+  /**
+   * Drop a view; optionally delete metadata files.
+   *
+   * <p>If purge is set to true the implementation should delete metadata files.
+   *
+   * @param identifier a table identifier
+   * @param purge if true, delete metadata files in the view
+   * @return true if the view was dropped, false if the view did not exist
+   */
+  boolean dropView(TableIdentifier identifier, boolean purge);
 
   /**
    * Rename a view.

api/src/main/java/org/apache/iceberg/catalog/ViewSessionCatalog.java

@@ -85,6 +85,16 @@ default boolean viewExists(SessionCatalog.SessionContext context, TableIdentifie
    */
   boolean dropView(SessionCatalog.SessionContext context, TableIdentifier identifier);
 
+  /**
+   * Drop a view and request that metadata files are immediately deleted.
+   *
+   * @param context session context
+   * @param ident a table identifier
+   * @return true if the view was dropped and purged, false if the table did not exist
+   * @throws UnsupportedOperationException if immediate delete is not supported
+   */
+  boolean purgeView(SessionCatalog.SessionContext context, TableIdentifier ident);
+
   /**
    * Rename a view.
    *

core/src/main/java/org/apache/iceberg/catalog/BaseViewSessionCatalog.java

@@ -73,6 +73,15 @@ public boolean dropView(TableIdentifier identifier) {
       return BaseViewSessionCatalog.this.dropView(context, identifier);
     }
 
+    @Override
+    public boolean dropView(TableIdentifier identifier, boolean purge) {
+      if (purge) {
+        return BaseViewSessionCatalog.this.purgeView(context, identifier);
+      } else {
+        return BaseViewSessionCatalog.this.dropView(context, identifier);
+      }
+    }
+
     @Override
     public void renameView(TableIdentifier from, TableIdentifier to) {
       BaseViewSessionCatalog.this.renameView(context, from, to);

core/src/main/java/org/apache/iceberg/inmemory/InMemoryCatalog.java

@@ -332,6 +332,28 @@ public boolean dropView(TableIdentifier identifier) {
     }
   }
 
+  @Override
+  public boolean dropView(TableIdentifier identifier, boolean purge) {
+    TableOperations ops = newTableOps(identifier);
+    TableMetadata lastMetadata;
+    if (purge && ops.current() != null) {
+      lastMetadata = ops.current();
+    } else {
+      lastMetadata = null;
+    }
+
+    synchronized (this) {
+      if (null == views.remove(identifier)) {
+        return false;
+      }
+    }
+
+    if (purge && lastMetadata != null) {
+      CatalogUtil.dropTableData(ops.io(), lastMetadata);
+    }
+    return true;
+  }
+
   @Override
   public void renameView(TableIdentifier from, TableIdentifier to) {
     if (from.equals(to)) {

core/src/main/java/org/apache/iceberg/rest/RESTCatalog.java

@@ -288,6 +288,11 @@ public boolean dropView(TableIdentifier identifier) {
     return viewSessionCatalog.dropView(identifier);
   }
 
+  @Override
+  public boolean dropView(TableIdentifier identifier, boolean purge) {
+    return viewSessionCatalog.dropView(identifier, purge);
+  }
+
   @Override
   public void renameView(TableIdentifier from, TableIdentifier to) {
     viewSessionCatalog.renameView(from, to);

core/src/main/java/org/apache/iceberg/rest/RESTSessionCatalog.java

@@ -1081,6 +1081,23 @@ public boolean dropView(SessionContext context, TableIdentifier identifier) {
     }
   }
 
+  @Override
+  public boolean purgeView(SessionContext context, TableIdentifier identifier) {
+    checkViewIdentifierIsValid(identifier);
+
+    try {
+      client.delete(
+          paths.view(identifier),
+          ImmutableMap.of("purgeRequested", "true"),
+          null,
+          headers(context),
+          ErrorHandlers.viewErrorHandler());
+      return true;
+    } catch (NoSuchViewException e) {
+      return false;
+    }
+  }
+
   @Override
   public void renameView(SessionContext context, TableIdentifier from, TableIdentifier to) {
     checkViewIdentifierIsValid(from);

nessie/src/main/java/org/apache/iceberg/nessie/NessieCatalog.java

@@ -360,10 +360,15 @@ public List<TableIdentifier> listViews(Namespace namespace) {
 
   @Override
   public boolean dropView(TableIdentifier identifier) {
+    return dropView(identifier, false);
+  }
+
+  @Override
+  public boolean dropView(TableIdentifier identifier, boolean purge) {
     TableReference tableReference = parseTableReference(identifier);
     return client
         .withReference(tableReference.getReference(), tableReference.getHash())
-        .dropView(identifierWithoutTableReference(identifier, tableReference), false);
+        .dropView(identifierWithoutTableReference(identifier, tableReference), purge);
   }
 
   @Override