Spark 3.3: drop_namespace with CASCADE support

.palantir/revapi.yml

@@ -406,6 +406,14 @@ acceptedBreaks:
       justification: "Removing deprecations for 1.2.0"
   "1.2.0":
     org.apache.iceberg:iceberg-api:
+    - code: "java.method.addedToInterface"
+      new: "method boolean org.apache.iceberg.catalog.SessionCatalog::dropNamespace(org.apache.iceberg.catalog.SessionCatalog.SessionContext,\
+        \ org.apache.iceberg.catalog.Namespace, boolean)"
+      justification: "extending api to add new method, not removing any public api"
+    - code: "java.method.addedToInterface"
+      new: "method boolean org.apache.iceberg.catalog.SupportsNamespaces::dropNamespace(org.apache.iceberg.catalog.Namespace,\
+      \ boolean) throws org.apache.iceberg.exceptions.NamespaceNotEmptyException"
+      justification: "extending api to add new method, not removing any public api"
     - code: "java.field.constantValueChanged"
       old: "field org.apache.iceberg.actions.RewriteDataFiles.MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT"
       new: "field org.apache.iceberg.actions.RewriteDataFiles.MAX_CONCURRENT_FILE_GROUP_REWRITES_DEFAULT"
@@ -715,7 +723,7 @@ acceptedBreaks:
     - code: "java.method.addedToInterface"
       new: "method java.lang.String org.apache.iceberg.expressions.Reference<T>::name()"
       justification: "All subclasses implement name"
-    - code: "java.method.addedToInterface"
+    - code: "java.methodropNamespaced.addedToInterface"
       new: "method java.util.List<org.apache.iceberg.StatisticsFile> org.apache.iceberg.Table::statisticsFiles()"
       justification: "new API method"
     - code: "java.method.removed"

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

@@ -309,6 +309,23 @@ default List<Namespace> listNamespaces(SessionContext context) {
    */
   boolean dropNamespace(SessionContext context, Namespace namespace);
 
+  /**
+   * Drop a namespace. If the namespace exists and was dropped, this will return true.
+   *
+   * @param context session context
+   * @param namespace a {@link Namespace namespace}
+   * @param cascade – When true, deletes all objects under the namespace
+   * @return true if the namespace was dropped, false otherwise.
+   * @throws NamespaceNotEmptyException If the namespace is not empty
+   */
+  default boolean dropNamespace(SessionContext context, Namespace namespace, boolean cascade) {
+    if (cascade) {
+      throw new UnsupportedOperationException(
+          "dropNamespace with cascade not supported with this catalog");
+    }
+    return dropNamespace(context, namespace);
+  }
+
   /**
    * Set a collection of properties on a namespace in the catalog.
    *

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

@@ -103,6 +103,23 @@ default List<Namespace> listNamespaces() {
    */
   boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyException;
 
+  /**
+   * Drop a namespace. If the namespace exists and was dropped, this will return true.
+   *
+   * @param namespace a namespace. {@link Namespace}
+   * @param cascade – When true, deletes all objects under the namespace
+   * @return true if the namespace was dropped, false otherwise.
+   * @throws NamespaceNotEmptyException If the namespace is not empty
+   */
+  default boolean dropNamespace(Namespace namespace, boolean cascade)
+      throws NamespaceNotEmptyException {
+    if (cascade) {
+      throw new UnsupportedOperationException(
+          "dropNamespace with cascade not supported with this catalog");
+    }
+    return dropNamespace(namespace);
+  }
+
   /**
    * Set a collection of properties on a namespace in the catalog.
    *

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

@@ -143,6 +143,16 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept
       return BaseSessionCatalog.this.dropNamespace(context, namespace);
     }
 
+    @Override
+    public boolean dropNamespace(Namespace namespace, boolean cascade)
+        throws NamespaceNotEmptyException {
+      if (cascade) {
+        return BaseSessionCatalog.this.dropNamespace(context, namespace, true);
+      } else {
+        return dropNamespace(namespace);
+      }
+    }
+
     @Override
     public boolean setProperties(Namespace namespace, Map<String, String> updates) {
       return BaseSessionCatalog.this.updateNamespaceMetadata(

core/src/main/java/org/apache/iceberg/hadoop/HadoopCatalog.java

@@ -349,6 +349,17 @@ public boolean dropNamespace(Namespace namespace) {
     }
   }
 
+  @Override
+  public boolean dropNamespace(Namespace namespace, boolean cascade)
+      throws NamespaceNotEmptyException {
+    if (cascade) {
+      // recursively delete all nested namespaces
+      listNamespaces(namespace).forEach(n -> dropNamespace(n, true));
+      listTables(namespace).forEach(this::dropTable);
+    }
+    return dropNamespace(namespace);
+  }
+
   @Override
   public boolean setProperties(Namespace namespace, Map<String, String> properties) {
     throw new UnsupportedOperationException(

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

@@ -203,6 +203,15 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept
     return namespaces.remove(namespace) != null;
   }
 
+  @Override
+  public boolean dropNamespace(Namespace namespace, boolean cascade)
+      throws NamespaceNotEmptyException {
+    if (cascade) {
+      listTables(namespace).forEach(this::dropTable);
+    }
+    return dropNamespace(namespace);
+  }
+
   @Override
   public boolean setProperties(Namespace namespace, Map<String, String> properties)
       throws NoSuchNamespaceException {

core/src/main/java/org/apache/iceberg/jdbc/JdbcCatalog.java

@@ -419,6 +419,17 @@ public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyExcept
     return deletedRows > 0;
   }
 
+  @Override
+  public boolean dropNamespace(Namespace namespace, boolean cascade)
+      throws NamespaceNotEmptyException {
+    if (cascade) {
+      // recursively delete all nested namespaces
+      listNamespaces(namespace).forEach(n -> dropNamespace(n, true));
+      listTables(namespace).forEach(this::dropTable);
+    }
+    return dropNamespace(namespace);
+  }
+
   @Override
   public boolean setProperties(Namespace namespace, Map<String, String> properties)
       throws NoSuchNamespaceException {

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

@@ -129,6 +129,14 @@ public static void dropNamespace(SupportsNamespaces catalog, Namespace namespace
     }
   }
 
+  public static void dropNamespace(
+      SupportsNamespaces catalog, Namespace namespace, boolean cascade) {
+    boolean dropped = catalog.dropNamespace(namespace, cascade);
+    if (!dropped) {
+      throw new NoSuchNamespaceException("Namespace does not exist: %s", namespace);
+    }
+  }
+
   public static UpdateNamespacePropertiesResponse updateNamespaceProperties(
       SupportsNamespaces catalog, Namespace namespace, UpdateNamespacePropertiesRequest request) {
     request.validate();

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

@@ -228,6 +228,12 @@ public boolean dropNamespace(Namespace ns) throws NamespaceNotEmptyException {
     return nsDelegate.dropNamespace(ns);
   }
 
+  @Override
+  public boolean dropNamespace(Namespace namespace, boolean cascade)
+      throws NamespaceNotEmptyException {
+    return nsDelegate.dropNamespace(namespace, cascade);
+  }
+
   @Override
   public boolean setProperties(Namespace ns, Map<String, String> props)
       throws NoSuchNamespaceException {

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

@@ -444,17 +444,30 @@ public Map<String, String> loadNamespaceMetadata(SessionContext context, Namespa
 
   @Override
   public boolean dropNamespace(SessionContext context, Namespace ns) {
+    return dropNamespaceInternal(context, ns, false);
+  }
+
+  private boolean dropNamespaceInternal(SessionContext context, Namespace ns, boolean cascade) {
     checkNamespaceIsValid(ns);
 
     try {
       client.delete(
-          paths.namespace(ns), null, headers(context), ErrorHandlers.namespaceErrorHandler());
+          paths.namespace(ns),
+          ImmutableMap.of("cascade", Boolean.toString(cascade)),
+          null,
+          headers(context),
+          ErrorHandlers.namespaceErrorHandler());
       return true;
     } catch (NoSuchNamespaceException e) {
       return false;
     }
   }
 
+  @Override
+  public boolean dropNamespace(SessionContext context, Namespace namespace, boolean cascade) {
+    return dropNamespaceInternal(context, namespace, cascade);
+  }
+
   @Override
   public boolean updateNamespaceMetadata(
       SessionContext context, Namespace ns, Map<String, String> updates, Set<String> removals) {

core/src/test/java/org/apache/iceberg/hadoop/TestHadoopCatalog.java

@@ -468,6 +468,53 @@ public void testDropNamespace() throws IOException {
     Assert.assertFalse(fs.isDirectory(new Path(metaLocation)));
   }
 
+  @Test
+  public void testDropNamespaceCascade() throws IOException {
+    String warehouseLocation = temp.newFolder().getAbsolutePath();
+    HadoopCatalog catalog = new HadoopCatalog();
+    catalog.setConf(new Configuration());
+    catalog.initialize(
+        "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation));
+    Namespace namespace1 = Namespace.of("db");
+    Namespace namespace2 = Namespace.of("db", "ns1");
+
+    TableIdentifier tbl1 = TableIdentifier.of(namespace1, "tbl1");
+    TableIdentifier tbl2 = TableIdentifier.of(namespace2, "tbl1");
+
+    Lists.newArrayList(tbl1, tbl2)
+        .forEach(t -> catalog.createTable(t, SCHEMA, PartitionSpec.unpartitioned()));
+
+    catalog.dropNamespace(namespace1, true);
+    String metaLocation = warehouseLocation + "/" + "db";
+    FileSystem fs = Util.getFs(new Path(metaLocation), catalog.getConf());
+    Assert.assertFalse(fs.isDirectory(new Path(metaLocation)));
+  }
+
+  @Test
+  public void testDropNamespaceCascadeFalse() throws IOException {
+    String warehouseLocation = temp.newFolder().getAbsolutePath();
+    HadoopCatalog catalog = new HadoopCatalog();
+    catalog.setConf(new Configuration());
+    catalog.initialize(
+        "hadoop", ImmutableMap.of(CatalogProperties.WAREHOUSE_LOCATION, warehouseLocation));
+    Namespace namespace1 = Namespace.of("db");
+    Namespace namespace2 = Namespace.of("db", "ns1");
+
+    TableIdentifier tbl1 = TableIdentifier.of(namespace1, "tbl1");
+    TableIdentifier tbl2 = TableIdentifier.of(namespace2, "tbl1");
+
+    Lists.newArrayList(tbl1, tbl2)
+        .forEach(t -> catalog.createTable(t, SCHEMA, PartitionSpec.unpartitioned()));
+
+    AssertHelpers.assertThrows(
+        "Should fail to drop namespace is not empty " + namespace1,
+        NamespaceNotEmptyException.class,
+        "Namespace " + namespace1 + " is not empty.",
+        () -> {
+          catalog.dropNamespace(Namespace.of("db"), false);
+        });
+  }
+
   @Test
   public void testVersionHintFileErrorWithFile() throws Exception {
     addVersionsToTable(table);

core/src/test/java/org/apache/iceberg/jdbc/TestJdbcCatalog.java

@@ -625,6 +625,43 @@ public void testDropNamespace() {
         () -> catalog.dropNamespace(tbl4.namespace()));
   }
 
+  @Test
+  public void testDropNamespaceCascade() {
+    TableIdentifier tbl0 = TableIdentifier.of("db", "ns1", "ns2", "tbl2");
+    TableIdentifier tbl1 = TableIdentifier.of("db", "ns1", "ns2", "tbl1");
+    TableIdentifier tbl2 = TableIdentifier.of("db", "ns1", "tbl2");
+    TableIdentifier tbl3 = TableIdentifier.of("db", "ns3", "tbl4");
+    TableIdentifier tbl4 = TableIdentifier.of("db", "tbl");
+
+    Lists.newArrayList(tbl0, tbl1, tbl2, tbl3, tbl4)
+        .forEach(t -> catalog.createTable(t, SCHEMA, PartitionSpec.unpartitioned()));
+
+    catalog.dropNamespace(tbl4.namespace(), true);
+    Assert.assertFalse(catalog.namespaceExists(tbl1.namespace()));
+  }
+
+  @Test
+  public void testDropNamespaceCascadeFalse() {
+    Assert.assertFalse(
+        "Should return false if drop does not modify state",
+        catalog.dropNamespace(Namespace.of("db", "ns1_not_exitss")));
+
+    TableIdentifier tbl0 = TableIdentifier.of("db", "ns1", "ns2", "tbl2");
+    TableIdentifier tbl1 = TableIdentifier.of("db", "ns1", "ns2", "tbl1");
+    TableIdentifier tbl2 = TableIdentifier.of("db", "ns1", "tbl2");
+    TableIdentifier tbl3 = TableIdentifier.of("db", "ns3", "tbl4");
+    TableIdentifier tbl4 = TableIdentifier.of("db", "tbl");
+
+    Lists.newArrayList(tbl0, tbl1, tbl2, tbl3, tbl4)
+        .forEach(t -> catalog.createTable(t, SCHEMA, PartitionSpec.unpartitioned()));
+
+    AssertHelpers.assertThrows(
+        "Should fail to drop namespace has tables",
+        NamespaceNotEmptyException.class,
+        "is not empty. 1 tables exist.",
+        () -> catalog.dropNamespace(tbl4.namespace(), false));
+  }
+
   @Test
   public void testCreateNamespace() {
     Namespace testNamespace = Namespace.of("testDb", "ns1", "ns2");

core/src/test/java/org/apache/iceberg/rest/RESTCatalogAdapter.java

@@ -284,7 +284,7 @@ public <T extends RESTResponse> T handleRequest(
 
       case DROP_NAMESPACE:
         if (asNamespaceCatalog != null) {
-          CatalogHandlers.dropNamespace(asNamespaceCatalog, namespaceFromPathVars(vars));
+          dropNamespace(vars);
           return null;
         }
         break;
@@ -363,6 +363,12 @@ public <T extends RESTResponse> T handleRequest(
     return null;
   }
 
+  private void dropNamespace(Map<String, String> vars) {
+    Namespace namespace = namespaceFromPathVars(vars);
+    boolean cascade = PropertyUtil.propertyAsBoolean(vars, "cascade", false);
+    CatalogHandlers.dropNamespace(asNamespaceCatalog, namespace, cascade);
+  }
+
   public <T extends RESTResponse> T execute(
       HTTPMethod method,
       String path,

docs/spark-ddl.md

@@ -140,6 +140,47 @@ AS SELECT ...
 The schema and partition spec will be replaced if changed. To avoid modifying the table's schema and partitioning, use `INSERT OVERWRITE` instead of `REPLACE TABLE`.
 The new table properties in the `REPLACE TABLE` command will be merged with any existing table properties. The existing table properties will be updated if changed else they are preserved.
 
+## `DROP NAMESPACE`
+
+### `DROP EMPTY NAMESPACE`
+
+To drop an _empty_ namespace, run:
+
+```sql
+DROP database prod.db.sample
+```
+If the namespace is not empty, this will fail with _NamespaceNotEmptyException_.
+
+
+### `DROP NON_EMPTY NAMESPACE`
+
+To drop a namespace and all its contents including tables, run:
+
+```sql
+DROP TABLE prod.db.sample CASCADE
+```
+**WARNING**: 
+drop table purge behaviour with cascade depends on the type of catalog managing the namespace.
+see below mapping of purge behaviour for different catalogs.
+- If the database is managed by **HiveCatalog**, this will _purge_ all the tables in the database.
+
+#### `drop namespace table purge behaviour by catalog for CASCADE`
+
+When namespace is dropped with _CASCADE_, all tables are dropped and contents are purged based on the type of
+catalog. 
+
+| Catalog   | Table Purge  | Nested Namespaces      |
+|-----------|--------------|------------------------|
+| Hive      | - [ x ]      | - [ ]                  |
+| Hadoop    | - [  ]       | - [ x ]                |
+| JDBC      | - [  ]       | - [ x ]                |
+| ECS       | - [  ]       | - [ x ]                |
+| Nessie    | NotSupported       | NotSupported                 |
+| DynamoDb  | NotSupported       | NotSupported                 |
+| Glue      | NotSupported       | NotSupported                 |
+| Snowflake | NotSupported | NotSupported           |
+
+
 ## `DROP TABLE`
 
 The drop table behavior changed in 0.14.