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

@@ -39,16 +39,20 @@
 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;
@@ -297,6 +301,40 @@ 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);
+    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())) {
+      throw new CommitFailedException(
+          "Cannot unregister table %s: metadata location has changed or table was dropped",
+          identifier);
+    }
+
+    StaticTableOperations staticOps =
+        new StaticTableOperations(metadata, ops.io(), ops.locationProvider());
+    return new BaseTable(staticOps, identifier.name(), metricsReporter());
+  }
+
+  private boolean 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)
+        == 1;
+  }
+
   @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

@@ -95,11 +95,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 +490,20 @@ public static void dropTable(Catalog catalog, TableIdentifier ident) {
     }
   }
 
+  public static UnregisterTableResponse unregisterTable(Catalog catalog, TableIdentifier 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

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

@@ -43,6 +43,7 @@
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.StaticTableOperations;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableMetadata;
 import org.apache.iceberg.TableOperations;
@@ -100,6 +101,7 @@
 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.UnregisterTableResponse;
 import org.apache.iceberg.rest.responses.UpdateNamespacePropertiesResponse;
 import org.apache.iceberg.util.EnvironmentUtil;
 import org.apache.iceberg.util.PropertyUtil;
@@ -805,6 +807,40 @@ public Table registerTable(
         response.labels());
   }
 
+  /**
+   * Unregister a table from the catalog without removing its data or metadata files.
+   *
+   * <p>This is the opposite of {@link #registerTable(SessionContext, TableIdentifier, String)}. On
+   * success, the table no longer exists in the catalog and the returned table is fixed at the last
+   * metadata registered with the catalog.
+   *
+   * @param context session context
+   * @param identifier a table identifier
+   * @return a read-only table fixed at the metadata current when it was unregistered
+   */
+  @Override
+  public Table unregisterTable(SessionContext context, TableIdentifier identifier) {
+    Endpoint.check(endpoints, Endpoint.V1_UNREGISTER_TABLE);
+    checkIdentifierIsValid(identifier);
+
+    try {
+      AuthSession contextualSession = authManager.contextualSession(context, catalogAuth);
+      UnregisterTableResponse response =
+          client
+              .withAuthSession(contextualSession)
+              .post(
+                  paths.unregister(identifier),
+                  null,
+                  UnregisterTableResponse.class,
+                  mutationHeaders,
+                  ErrorHandlers.tableErrorHandler());
+      StaticTableOperations ops = new StaticTableOperations(response.metadata(), io);
+      return new BaseTable(ops, identifier.name());
+    } finally {
+      invalidateTable(context, identifier);
+    }
+  }
+
   @Override
   public void createNamespace(
       SessionContext context, Namespace namespace, Map<String, String> metadata) {

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

@@ -38,6 +38,8 @@ public class ResourcePaths {
   public static final String V1_TABLE_REMOTE_SIGN =
       "/v1/{prefix}/namespaces/{namespace}/tables/{table}/sign";
   public static final String V1_TABLE_REGISTER = "/v1/{prefix}/namespaces/{namespace}/register";
+  public static final String V1_TABLE_UNREGISTER =
+      "/v1/{prefix}/namespaces/{namespace}/tables/{table}/unregister";
   public static final String V1_TABLE_METRICS =
       "/v1/{prefix}/namespaces/{namespace}/tables/{table}/metrics";
   public static final String V1_TABLE_RENAME = "/v1/{prefix}/tables/rename";
@@ -108,6 +110,17 @@ public String register(Namespace ns) {
     return SLASH.join("v1", prefix, "namespaces", pathEncode(ns), "register");
   }
 
+  public String unregister(TableIdentifier ident) {
+    return SLASH.join(
+        "v1",
+        prefix,
+        "namespaces",
+        pathEncode(ident.namespace()),
+        "tables",
+        RESTUtil.encodePathSegment(ident.name()),
+        "unregister");
+  }
+
   public String rename() {
     return SLASH.join("v1", prefix, "tables", "rename");
   }