Core: Implement unregister table in catalogs and RCKs

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

@@ -382,6 +382,21 @@ default Table registerTable(
     throw new UnsupportedOperationException("Registering tables with overwrite is not supported");
   }
 
+  /**
+   * Unregister a table without deleting its data or metadata files.
+   *
+   * <p>The returned table is fixed at the last metadata file registered with the catalog and cannot
+   * be modified. Its metadata file location can be used to {@link #registerTable(TableIdentifier,
+   * String) register} the table again.
+   *
+   * @param identifier a table identifier
+   * @return a read-only table fixed at the metadata current when it was unregistered
+   * @throws NoSuchTableException if the table does not exist
+   */
+  default Table unregisterTable(TableIdentifier identifier) {
+    throw new UnsupportedOperationException("Unregistering tables is not supported");
+  }
+
   /**
    * Instantiate a builder to either create a table or start a create/replace transaction.
    *

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

@@ -196,6 +196,21 @@ default Table registerTable(
     throw new UnsupportedOperationException("Registering tables with overwrite is not supported");
   }
 
+  /**
+   * Unregister a table without deleting its data or metadata files.
+   *
+   * <p>The returned table is fixed at the last metadata file registered with the catalog and cannot
+   * be modified.
+   *
+   * @param context session context
+   * @param ident a table identifier
+   * @return a read-only table fixed at the metadata current when it was unregistered
+   * @throws NoSuchTableException if the table does not exist
+   */
+  default Table unregisterTable(SessionContext context, TableIdentifier ident) {
+    throw new UnsupportedOperationException("Unregistering tables is not supported");
+  }
+
   /**
    * Check whether table exists.
    *

core/src/main/java/org/apache/iceberg/CachingCatalog.java

@@ -204,6 +204,13 @@ public Table registerTable(
     return table;
   }
 
+  @Override
+  public Table unregisterTable(TableIdentifier identifier) {
+    Table table = catalog.unregisterTable(identifier);
+    invalidateTable(identifier);
+    return table;
+  }
+
   private Iterable<TableIdentifier> metadataTableIdentifiers(TableIdentifier ident) {
     ImmutableList.Builder<TableIdentifier> builder = ImmutableList.builder();
 

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

@@ -95,6 +95,11 @@ public Table registerTable(
       return BaseSessionCatalog.this.registerTable(context, ident, metadataFileLocation, overwrite);
     }
 
+    @Override
+    public Table unregisterTable(TableIdentifier ident) {
+      return BaseSessionCatalog.this.unregisterTable(context, ident);
+    }
+
     @Override
     public boolean tableExists(TableIdentifier ident) {
       return BaseSessionCatalog.this.tableExists(context, ident);

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

@@ -29,8 +29,11 @@
 import java.util.concurrent.ConcurrentMap;
 import java.util.stream.Collectors;
 import org.apache.iceberg.BaseMetastoreTableOperations;
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.catalog.Namespace;
@@ -125,6 +128,27 @@ private String defaultNamespaceLocation(Namespace namespace) {
     }
   }
 
+  @Override
+  public Table unregisterTable(TableIdentifier tableIdentifier) {
+    TableOperations ops = newTableOps(tableIdentifier);
+    TableMetadata metadata;
+
+    synchronized (this) {
+      metadata = ops.current();
+      if (metadata == null) {
+        throw new NoSuchTableException("Table does not exist: %s", tableIdentifier);
+      }
+
+      if (tables.remove(tableIdentifier) == null) {
+        throw new NoSuchTableException("Table does not exist: %s", tableIdentifier);
+      }
+    }
+
+    StaticTableOperations staticOps =
+        new StaticTableOperations(metadata, ops.io(), ops.locationProvider());
+    return new BaseTable(staticOps, tableIdentifier.name(), metricsReporter());
+  }
+
   @Override
   public boolean dropTable(TableIdentifier tableIdentifier, boolean purge) {
     TableOperations ops = newTableOps(tableIdentifier);

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

@@ -18,6 +18,15 @@
  */
 package org.apache.iceberg.jdbc;
 
+import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_MAX_RETRY_WAIT_MS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_MIN_RETRY_WAIT_MS_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES;
+import static org.apache.iceberg.TableProperties.COMMIT_NUM_RETRIES_DEFAULT;
+import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS;
+import static org.apache.iceberg.TableProperties.COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT;
+
 import java.io.IOException;
 import java.io.UncheckedIOException;
 import java.sql.Connection;
@@ -34,21 +43,26 @@
 import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
+import org.apache.iceberg.BaseTable;
 import org.apache.iceberg.CatalogProperties;
 import org.apache.iceberg.CatalogUtil;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.StaticTableOperations;
+import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.Transaction;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NamespaceNotEmptyException;
 import org.apache.iceberg.exceptions.NoSuchNamespaceException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
@@ -66,6 +80,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.util.LocationUtil;
 import org.apache.iceberg.util.PropertyUtil;
+import org.apache.iceberg.util.Tasks;
 import org.apache.iceberg.view.BaseMetastoreViewCatalog;
 import org.apache.iceberg.view.ViewMetadata;
 import org.apache.iceberg.view.ViewOperations;
@@ -297,6 +312,60 @@ protected String defaultWarehouseLocation(TableIdentifier table) {
     return SLASH.join(defaultNamespaceLocation(table.namespace()), tableLocation);
   }
 
+  @Override
+  public Table unregisterTable(TableIdentifier identifier) {
+    Preconditions.checkArgument(
+        identifier != null && isValidIdentifier(identifier), "Invalid identifier: %s", identifier);
+
+    TableMetadata initialMetadata = newTableOps(identifier).current();
+    if (initialMetadata == null) {
+      throw new NoSuchTableException("Table does not exist: %s", identifier);
+    }
+
+    AtomicReference<Table> unregistered = new AtomicReference<>();
+    Tasks.foreach(identifier)
+        .retry(initialMetadata.propertyAsInt(COMMIT_NUM_RETRIES, COMMIT_NUM_RETRIES_DEFAULT))
+        .exponentialBackoff(
+            initialMetadata.propertyAsInt(
+                COMMIT_MIN_RETRY_WAIT_MS, COMMIT_MIN_RETRY_WAIT_MS_DEFAULT),
+            initialMetadata.propertyAsInt(
+                COMMIT_MAX_RETRY_WAIT_MS, COMMIT_MAX_RETRY_WAIT_MS_DEFAULT),
+            initialMetadata.propertyAsInt(
+                COMMIT_TOTAL_RETRY_TIME_MS, COMMIT_TOTAL_RETRY_TIME_MS_DEFAULT),
+            2.0 /* exponential */)
+        .onlyRetryOn(CommitFailedException.class)
+        .run(tableIdentifier -> unregistered.set(unregisterTableOnce(tableIdentifier)));
+    return unregistered.get();
+  }
+
+  private Table unregisterTableOnce(TableIdentifier identifier) {
+    TableOperations ops = newTableOps(identifier);
+    TableMetadata metadata = ops.current();
+    if (metadata == null) {
+      throw new NoSuchTableException("Table does not exist: %s", identifier);
+    }
+
+    if (dropTableIfMetadataMatches(identifier, metadata.metadataFileLocation()) == 0) {
+      throw new CommitFailedException(
+          "Cannot unregister table %s: metadata location changed concurrently", identifier);
+    }
+
+    StaticTableOperations staticOps =
+        new StaticTableOperations(metadata, ops.io(), ops.locationProvider());
+    return new BaseTable(staticOps, identifier.name(), metricsReporter());
+  }
+
+  int dropTableIfMetadataMatches(TableIdentifier identifier, String metadataLocation) {
+    return execute(
+        (schemaVersion == JdbcUtil.SchemaVersion.V1)
+            ? JdbcUtil.V1_UNREGISTER_TABLE_SQL
+            : JdbcUtil.V0_UNREGISTER_TABLE_SQL,
+        catalogName,
+        JdbcUtil.namespaceToString(identifier.namespace()),
+        identifier.name(),
+        metadataLocation);
+  }
+
   @Override
   public boolean dropTable(TableIdentifier identifier, boolean purge) {
     TableOperations ops = newTableOps(identifier);

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

@@ -321,6 +321,8 @@ enum SchemaVersion {
           + " OR "
           + RECORD_TYPE
           + " IS NULL)";
+  static final String V1_UNREGISTER_TABLE_SQL =
+      V1_DROP_TABLE_SQL + " AND " + BaseMetastoreTableOperations.METADATA_LOCATION_PROP + " = ?";
   static final String V0_DROP_TABLE_SQL =
       "DELETE FROM "
           + CATALOG_TABLE_VIEW_NAME
@@ -331,6 +333,8 @@ enum SchemaVersion {
           + "  = ? AND "
           + TABLE_NAME
           + " = ?";
+  static final String V0_UNREGISTER_TABLE_SQL =
+      V0_DROP_TABLE_SQL + " AND " + BaseMetastoreTableOperations.METADATA_LOCATION_PROP + " = ?";
   private static final String GET_NAMESPACE_SQL =
       "SELECT "
           + TABLE_NAMESPACE

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

@@ -51,6 +51,7 @@
 import org.apache.iceberg.BaseTransaction;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.IncrementalAppendScan;
+import org.apache.iceberg.MetadataTableType;
 import org.apache.iceberg.MetadataUpdate.UpgradeFormatVersion;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.RetryableValidationException;
@@ -95,11 +96,13 @@
 import org.apache.iceberg.rest.responses.FetchScanTasksResponse;
 import org.apache.iceberg.rest.responses.GetNamespaceResponse;
 import org.apache.iceberg.rest.responses.ImmutableLoadViewResponse;
+import org.apache.iceberg.rest.responses.ImmutableUnregisterTableResponse;
 import org.apache.iceberg.rest.responses.ListNamespacesResponse;
 import org.apache.iceberg.rest.responses.ListTablesResponse;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
 import org.apache.iceberg.rest.responses.LoadViewResponse;
 import org.apache.iceberg.rest.responses.PlanTableScanResponse;
+import org.apache.iceberg.rest.responses.UnregisterTableResponse;
 import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.Tasks;
@@ -488,6 +491,24 @@ public static void dropTable(Catalog catalog, TableIdentifier ident) {
     }
   }
 
+  public static UnregisterTableResponse unregisterTable(Catalog catalog, TableIdentifier ident) {
+    if (MetadataTableType.from(ident.name()) != null) {
+      throw new NoSuchTableException("Table does not exist: %s", ident);
+    }
+
+    Table table = catalog.unregisterTable(ident);
+    if (!(table instanceof BaseTable)) {
+      throw new IllegalStateException("Cannot wrap catalog that does not produce BaseTable");
+    }
+
+    TableMetadata metadata = ((BaseTable) table).operations().current();
+
+    return ImmutableUnregisterTableResponse.builder()
+        .metadataLocation(metadata.metadataFileLocation())
+        .metadata(metadata)
+        .build();
+  }
+
   public static void purgeTable(Catalog catalog, TableIdentifier ident) {
     boolean dropped = catalog.dropTable(ident, true);
     if (!dropped) {

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

@@ -62,6 +62,8 @@ public class Endpoint {
       Endpoint.create("POST", ResourcePaths.V1_TABLE_RENAME);
   public static final Endpoint V1_REGISTER_TABLE =
       Endpoint.create("POST", ResourcePaths.V1_TABLE_REGISTER);
+  public static final Endpoint V1_UNREGISTER_TABLE =
+      Endpoint.create("POST", ResourcePaths.V1_TABLE_UNREGISTER);
   public static final Endpoint V1_REPORT_METRICS =
       Endpoint.create("POST", ResourcePaths.V1_TABLE_METRICS);
   public static final Endpoint V1_TABLE_CREDENTIALS =

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

@@ -252,6 +252,11 @@ public Table registerTable(
     return delegate.registerTable(ident, metadataFileLocation, overwrite);
   }
 
+  @Override
+  public Table unregisterTable(TableIdentifier ident) {
+    return delegate.unregisterTable(ident);
+  }
+
   @Override
   public void createNamespace(Namespace ns, Map<String, String> props) {
     nsDelegate.createNamespace(ns, props);

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

@@ -80,6 +80,7 @@
 import org.apache.iceberg.rest.responses.ImmutableLoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.ImmutableLoadViewResponse;
 import org.apache.iceberg.rest.responses.ImmutableRemoteSignResponse;
+import org.apache.iceberg.rest.responses.ImmutableUnregisterTableResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponse;
 import org.apache.iceberg.rest.responses.LoadCredentialsResponseParser;
 import org.apache.iceberg.rest.responses.LoadTableResponse;
@@ -91,6 +92,8 @@
 import org.apache.iceberg.rest.responses.PlanTableScanResponseParser;
 import org.apache.iceberg.rest.responses.RemoteSignResponse;
 import org.apache.iceberg.rest.responses.RemoteSignResponseParser;
+import org.apache.iceberg.rest.responses.UnregisterTableResponse;
+import org.apache.iceberg.rest.responses.UnregisterTableResponseParser;
 import org.apache.iceberg.util.JsonUtil;
 
 public class RESTSerializers {
@@ -153,6 +156,12 @@ public static void registerAll(ObjectMapper mapper) {
         .addDeserializer(ConfigResponse.class, new ConfigResponseDeserializer<>())
         .addSerializer(LoadTableResponse.class, new LoadTableResponseSerializer<>())
         .addDeserializer(LoadTableResponse.class, new LoadTableResponseDeserializer<>())
+        .addSerializer(UnregisterTableResponse.class, new UnregisterTableResponseSerializer<>())
+        .addSerializer(
+            ImmutableUnregisterTableResponse.class, new UnregisterTableResponseSerializer<>())
+        .addDeserializer(UnregisterTableResponse.class, new UnregisterTableResponseDeserializer<>())
+        .addDeserializer(
+            ImmutableUnregisterTableResponse.class, new UnregisterTableResponseDeserializer<>())
         .addSerializer(PlanTableScanRequest.class, new PlanTableScanRequestSerializer<>())
         .addDeserializer(PlanTableScanRequest.class, new PlanTableScanRequestDeserializer<>())
         .addSerializer(FetchScanTasksRequest.class, new FetchScanTasksRequestSerializer<>())
@@ -532,6 +541,24 @@ public void serialize(T request, JsonGenerator gen, SerializerProvider serialize
     }
   }
 
+  static class UnregisterTableResponseSerializer<T extends UnregisterTableResponse>
+      extends JsonSerializer<T> {
+    @Override
+    public void serialize(T response, JsonGenerator gen, SerializerProvider serializers)
+        throws IOException {
+      UnregisterTableResponseParser.toJson(response, gen);
+    }
+  }
+
+  static class UnregisterTableResponseDeserializer<T extends UnregisterTableResponse>
+      extends JsonDeserializer<T> {
+    @Override
+    public T deserialize(JsonParser p, DeserializationContext context) throws IOException {
+      JsonNode jsonNode = p.getCodec().readTree(p);
+      return (T) UnregisterTableResponseParser.fromJson(jsonNode);
+    }
+  }
+
   static class LoadTableResponseDeserializer<T extends LoadTableResponse>
       extends JsonDeserializer<T> {
     @Override