Iceberg does not support drop namespace cascade

[pan3793]

The semantic of drop namespace cascade is different between Iceberg and Spark, then run Spark SQL DROP DATABASE xxx CASCADE cause failure if the database is not empty.

Iceberg interface SupportsNamespaces

  /**
   * Drop a namespace. If the namespace exists and was dropped, this will return true.
   *
   * @param namespace a namespace. {@link Namespace}
   * @return true if the namespace was dropped, false otherwise.
   * @throws NamespaceNotEmptyException If the namespace does not empty
   */
  boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyException;

Spark interface SupportsNamespaces extends CatalogPlugin

  /**
   * Drop a namespace from the catalog, recursively dropping all objects within the namespace.
   * <p>
   * If the catalog implementation does not support this operation, it may throw
   * {@link UnsupportedOperationException}.
   *
   * @param namespace a multi-part namespace
   * @return true if the namespace was dropped
   * @throws NoSuchNamespaceException If the namespace does not exist (optional)
   * @throws UnsupportedOperationException If drop is not a supported operation
   */
  boolean dropNamespace(String[] namespace) throws NoSuchNamespaceException;

The CASCADE logic is handled by DropNamespaceExec

    if (nsCatalog.namespaceExists(ns)) {
      // The default behavior of `SupportsNamespace.dropNamespace()` is cascading,
      // so make sure the namespace to drop is empty.
      if (!cascade) {
        if (catalog.asTableCatalog.listTables(ns).nonEmpty
          || nsCatalog.listNamespaces(ns).nonEmpty) {
          throw QueryExecutionErrors.cannotDropNonemptyNamespaceError(namespace)
        }
      }

      if (!nsCatalog.dropNamespace(ns)) {
        throw QueryExecutionErrors.cannotDropNonemptyNamespaceError(namespace)
      }

[rdblue]

From the code, it looks like returning false is allowed and will result in an error that the namespace can't be dropped because it is not empty.

[kbendick]

Hi @pan3793 - Thanks for reporting this!

I spent some time looking into this, and it doesn't seem as though Spark exposes CASCADE to the interfaces that external catalogs are supposed to implement.

In fact, their javadocs do say that the default behavior is to CASCADE, but SupportsNamespace doesn't capture any of the information related to cascading from DropNamespaceExec.

Even in their tests for external catalogs, they only have a test that throws if the namespace isn't empty: https://github.com/apache/spark/blob/5c0762b5d2004464c36df353a4d483f8f57eebeb/sql/[…]/spark/sql/execution/datasources/v2/V2SessionCatalogSuite.scala

That's sort of what they're implying in the comment:

// The default behavior of `SupportsNamespace.dropNamespace()` is cascading,
// so make sure the namespace to drop is empty.

Even their implementation DropNamespaceExec (which you posted here) doesn't seem to handle CASCADE behavior at all which is why they actually check and throw vs list and drop.

And poking around the query planner, it doesn't seem to be the case that any instantiated DropNamespace node actually causes cascading either, at least for V2Catalog interfaces.

I'm going to poke around a bit more to see if there is an easy way to get at the CASCADE information - but I think that their documentation is a little off, specifically for external catalogs.

I will get back and confirm that. Since it's not exposed in Spark's SupportsNamespace interface presently, it will probably a little more difficult to get this behavior until it can be fixed upstream.

But hopefully I'm wrong and it's a simple change!

[kbendick]

Ok. On further inspection, it seems that the V2SessionCatalog falls back to the V1 definition DropDatabaseExec:

https://github.com/apache/spark/blob/8ab9d6327d7db20a4257f9fe6d0b17919576be5e/sql/core/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveSessionCatalog.scala#L254-L255

So we'd likely have to inject our own analyzer rule, or get that information exposed upstream in Spark (having Spark expose it being the better choice but longer task): https://github.com/apache/spark/blob/4df40ead99f07b87e3207c9d18c1b4f1f92a3b86/sql/core/src/main/scala/org/apache/spark/sql/internal/BaseSessionStateBuilder.scala#L182-L190

[pan3793]

Thanks for your reply, the cascade logic is handled by DropNamespaceExec, I updated issue description.

[kbendick]

Thanks for your reply, the cascade logic is handled by DropNamespaceExec, I updated issue description.

Appreciate the update.

Unfortunately, there's not really presently a way to access whether or not the user has set CASCADE in the query based on the interfaces that Spark provides.

Spark gives us essentially the same SupportsNamespaces interfaces, and dropNamespace only provides access to the Namespace that the user passed.

Our code (at least in the HiveCatalog) simply implements the interface Spark provides, which doesn't provide access to whether or not the CASCADE flag is set.

[pan3793]

Spark already told us the dropNamespace should be act as CASCADE

Spark interface SupportsNamespaces extends CatalogPlugin

  /**
   * Drop a namespace from the catalog, **recursively dropping all objects within the namespace.**
   * .....
   */
  boolean dropNamespace(String[] namespace) throws NoSuchNamespaceException;

And in DropNamespaceExec, Spark know the CASCADE flag, and handled by

      if (!cascade) {
        if (catalog.asTableCatalog.listTables(ns).nonEmpty
          || nsCatalog.listNamespaces(ns).nonEmpty) {
          throw QueryExecutionErrors.cannotDropNonemptyNamespaceError(namespace)
        }
      }

      if (!nsCatalog.dropNamespace(ns)) {
        throw QueryExecutionErrors.cannotDropNonemptyNamespaceError(namespace)
      }

[pan3793]

Seems the second throw QueryExecutionErrors.cannotDropNonemptyNamespaceError(namespace) is misleading

[pan3793]

How about the following change in Iceberg interface SupportsNamespaces

default boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyException {
    dropNamespace(namespace, false);
}

boolean dropNamespace(Namespace namespace, boolean cascade) throws NamespaceNotEmptyException;

And expose dropNamespace(namespace, cascade=true) to SparkCatalog

[kbendick]

Yeah. That's what I mean.

I know the Spark docs mention it, but they're actually the ones that even throw the exception. I did a pretty deep dive and I can see how Hive tables are able to handle it. They follow an older pathway DropDatabaseCommand, which relies on HMS to handle the Cascade.

The problem with the approach that you mentioned is that we still wouldn't have the ability to get the CASCADE information from the query. Just setting it to true by default is most likely not going to work for most cases. It's just not exposed in the interface that Spark provides to external catalog developers.

We'd have to implement our own antlr grammar to catch drop table ... cascade. That's something that could be done, but it's not a trivial amount of work necessarily.

[kbendick]

I can sync up with some people in the Spark community that I know and see if there is a way to get access to whether or not that flag is set. I somewhat doubt it, but I'll do my best to find out.

If I don't get back to you in a few days, feel free to message me on the Iceberg slack 🙂

[kbendick]

May I ask too, how big of an issue is this for you in practice? What's the workflow that you use that requires Cascade support?

I 100% agree it would be great to support, but it would be helpful to know your specific pain point and how you're using CASCADE in your workflows if you don't mind sharing. Is it in Dev environments to easily stand up and tear things down? Or do you use a namespace perhaps as a staging environment during your jobs?

[pan3793]

The problem with the approach that you mentioned is that we still wouldn't have the ability to get the CASCADE information from the query.

I can't agree. Spark already handled the nonempty detection, an exception will throw if drop a nonempty database without CASCADE. Why Iceberg need to know the CASCADE flag?

      if (!cascade) {
        if (catalog.asTableCatalog.listTables(ns).nonEmpty
          || nsCatalog.listNamespaces(ns).nonEmpty) {
          throw QueryExecutionErrors.cannotDropNonemptyNamespaceError(namespace)
        }
      }

[pan3793]

May I ask too, how big of an issue is this for you in practice? What's the workflow that you use that requires Cascade support?

I encounter this issue when change the TPCDS test suite from Delta table format to Iceberg. The drop database cascade sql works with delta but failed in iceberg.

https://github.com/apache/incubator-kyuubi/blob/2baabf46cacf745e5ceaf78192f9f2359dc9e477/kyuubi-server/src/test/scala/org/apache/kyuubi/operation/tpcds/OutputSchemaTPCDSSuite.scala#L70

[kbendick]

Spark has access to that information, but presently it just throws an exception after doing the lists.

So if it was changed on the spark side to list the tables and drop them if cascade is used, Iceberg wouldn't need to know at all.

But unfortunately that's not what Spark does. I'll try syncing up with some folks that are more deeply integrated with the core Spark community to see if something can be done, but I'm not super confident much will come from it. At least not very quickly.