Add overwrite param to registerTable procedure

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

@@ -347,6 +347,21 @@ default Table registerTable(TableIdentifier identifier, String metadataFileLocat
     throw new UnsupportedOperationException("Registering tables is not supported");
   }
 
+  /**
+   * Register a table with the catalog if it does not exist overwrite table metadata if the table
+   * already exists.
+   *
+   * @param identifier a table identifier
+   * @param metadataFileLocation the location of a metadata file
+   * @param overwrite whether to overwrite table metadata if the table already exists
+   * @return a Table instance
+   * @throws AlreadyExistsException if the table already exists in the catalog.
+   */
+  default Table registerTable(
+      TableIdentifier identifier, String metadataFileLocation, boolean overwrite) {
+    throw new UnsupportedOperationException("Registering 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

@@ -170,6 +170,25 @@ public Object wrappedIdentity() {
    */
   Table registerTable(SessionContext context, TableIdentifier ident, String metadataFileLocation);
 
+  /**
+   * Register a table with the catalog if it does not exist overwrite table metadata if the table
+   * already exists.
+   *
+   * @param context session context
+   * @param ident a table identifier
+   * @param metadataFileLocation the location of a metadata file
+   * @param overwrite whether to overwrite table metadata if the table already exists
+   * @return a Table instance
+   * @throws AlreadyExistsException if the table already exists in the catalog.
+   */
+  default Table registerTable(
+      SessionContext context,
+      TableIdentifier ident,
+      String metadataFileLocation,
+      boolean overwrite) {
+    return registerTable(context, ident, metadataFileLocation);
+  }
+
   /**
    * Check whether table exists.
    *

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

@@ -72,21 +72,30 @@ public Table loadTable(TableIdentifier identifier) {
 
   @Override
   public Table registerTable(TableIdentifier identifier, String metadataFileLocation) {
+    return registerTable(identifier, metadataFileLocation, false);
+  }
+
+  @Override
+  public Table registerTable(
+      TableIdentifier identifier, String metadataFileLocation, boolean overwrite) {
     Preconditions.checkArgument(
         identifier != null && isValidIdentifier(identifier), "Invalid identifier: %s", identifier);
     Preconditions.checkArgument(
         metadataFileLocation != null && !metadataFileLocation.isEmpty(),
         "Cannot register an empty metadata file location as a table");
 
-    // Throw an exception if this table already exists in the catalog.
-    if (tableExists(identifier)) {
+    // Throw an exception if this table already exists in the catalog and
+    // there is no intention to overwrite metadata.
+    if (!overwrite && tableExists(identifier)) {
       throw new AlreadyExistsException("Table already exists: %s", identifier);
     }
 
     TableOperations ops = newTableOps(identifier);
     InputFile metadataFile = ops.io().newInputFile(metadataFileLocation);
     TableMetadata metadata = TableMetadataParser.read(metadataFile);
-    ops.commit(null, metadata);
+
+    TableMetadata current = overwrite ? ops.current() : null;
+    ops.commit(current, metadata);
 
     return new BaseTable(ops, fullTableName(name(), identifier), metricsReporter());
   }

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

@@ -197,6 +197,14 @@ public Table registerTable(TableIdentifier identifier, String metadataFileLocati
     return table;
   }
 
+  @Override
+  public Table registerTable(
+      TableIdentifier identifier, String metadataFileLocation, boolean overwrite) {
+    Table table = catalog.registerTable(identifier, metadataFileLocation, overwrite);
+    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

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

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

@@ -295,7 +295,8 @@ public static LoadTableResponse registerTable(
     request.validate();
 
     TableIdentifier identifier = TableIdentifier.of(namespace, request.name());
-    Table table = catalog.registerTable(identifier, request.metadataLocation());
+    Table table =
+        catalog.registerTable(identifier, request.metadataLocation(), request.overwrite());
     if (table instanceof BaseTable) {
       return LoadTableResponse.builder()
           .withTableMetadata(((BaseTable) table).operations().current())

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

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

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

@@ -500,6 +500,15 @@ public void invalidateTable(SessionContext context, TableIdentifier ident) {}
   @Override
   public Table registerTable(
       SessionContext context, TableIdentifier ident, String metadataFileLocation) {
+    return registerTable(context, ident, metadataFileLocation, false);
+  }
+
+  @Override
+  public Table registerTable(
+      SessionContext context,
+      TableIdentifier ident,
+      String metadataFileLocation,
+      boolean overwrite) {
     Endpoint.check(endpoints, Endpoint.V1_REGISTER_TABLE);
     checkIdentifierIsValid(ident);
 
@@ -512,6 +521,7 @@ public Table registerTable(
         ImmutableRegisterTableRequest.builder()
             .name(ident.name())
             .metadataLocation(metadataFileLocation)
+            .overwrite(overwrite)
             .build();
 
     AuthSession contextualSession = authManager.contextualSession(context, catalogAuth);

core/src/main/java/org/apache/iceberg/rest/requests/RegisterTableRequest.java

@@ -28,6 +28,11 @@ public interface RegisterTableRequest extends RESTRequest {
 
   String metadataLocation();
 
+  @Value.Default
+  default boolean overwrite() {
+    return false;
+  }
+
   @Override
   default void validate() {
     // nothing to validate as it's not possible to create an invalid instance

core/src/main/java/org/apache/iceberg/rest/requests/RegisterTableRequestParser.java

@@ -28,6 +28,7 @@ public class RegisterTableRequestParser {
 
   private static final String NAME = "name";
   private static final String METADATA_LOCATION = "metadata-location";
+  private static final String OVERWRITE = "overwrite";
 
   private RegisterTableRequestParser() {}
 
@@ -46,6 +47,7 @@ public static void toJson(RegisterTableRequest request, JsonGenerator gen) throw
 
     gen.writeStringField(NAME, request.name());
     gen.writeStringField(METADATA_LOCATION, request.metadataLocation());
+    gen.writeBooleanField(OVERWRITE, request.overwrite());
 
     gen.writeEndObject();
   }
@@ -60,10 +62,12 @@ public static RegisterTableRequest fromJson(JsonNode json) {
 
     String name = JsonUtil.getString(NAME, json);
     String metadataLocation = JsonUtil.getString(METADATA_LOCATION, json);
+    Boolean overwrite = JsonUtil.getBool(OVERWRITE, json);
 
     return ImmutableRegisterTableRequest.builder()
         .name(name)
         .metadataLocation(metadataLocation)
+        .overwrite(Boolean.TRUE.equals(overwrite))
         .build();
   }
 }

core/src/test/java/org/apache/iceberg/rest/requests/TestRegisterTableRequestParser.java

@@ -67,7 +67,32 @@ public void roundTripSerde() {
     String expectedJson =
         "{\n"
             + "  \"name\" : \"table_1\",\n"
-            + "  \"metadata-location\" : \"file://tmp/NS/test_tbl/metadata/00000-d4f60d2f-2ad2-408b-8832-0ed7fbd851ee.metadata.json\"\n"
+            + "  \"metadata-location\" : \"file://tmp/NS/test_tbl/metadata/00000-d4f60d2f-2ad2-408b-8832-0ed7fbd851ee.metadata.json\",\n"
+            + "  \"overwrite\" : false\n"
+            + "}";
+
+    String json = RegisterTableRequestParser.toJson(request, true);
+    assertThat(json).isEqualTo(expectedJson);
+
+    assertThat(RegisterTableRequestParser.toJson(RegisterTableRequestParser.fromJson(json), true))
+        .isEqualTo(expectedJson);
+  }
+
+  @Test
+  public void roundTripSerdeWithOverwrite() {
+    RegisterTableRequest request =
+        ImmutableRegisterTableRequest.builder()
+            .name("table_1")
+            .metadataLocation(
+                "file://tmp/NS/test_tbl/metadata/00000-d4f60d2f-2ad2-408b-8832-0ed7fbd851ee.metadata.json")
+            .overwrite(true)
+            .build();
+
+    String expectedJson =
+        "{\n"
+            + "  \"name\" : \"table_1\",\n"
+            + "  \"metadata-location\" : \"file://tmp/NS/test_tbl/metadata/00000-d4f60d2f-2ad2-408b-8832-0ed7fbd851ee.metadata.json\",\n"
+            + "  \"overwrite\" : true\n"
             + "}";
 
     String json = RegisterTableRequestParser.toJson(request, true);

docs/docs/spark-procedures.md

@@ -725,9 +725,10 @@ Creates a catalog entry for a metadata.json file which already exists but does n
 #### Usage
 
 | Argument Name | Required? | Type | Description |
-|---------------|-----------|------|-------------|
-| `table`       | ✔️  | string | Table which is to be registered |
-| `metadata_file`| ✔️  | string | Metadata file which is to be registered as a new catalog identifier |
+|---------------|----------|------|-------------|
+| `table`       | ✔️ | string | Table which is to be registered |
+| `metadata_file`| ✔️ | string | Metadata file which is to be registered as a new catalog identifier |
+| `overwrite`| ️  | boolean | Whether to overwrite table metadata if the table already exists |
 
 !!! warning
     Having the same metadata.json registered in more than one catalog can lead to missing updates, loss of data, and table corruption.
@@ -750,6 +751,14 @@ CALL spark_catalog.system.register_table(
   metadata_file => 'path/to/metadata/file.json'
 );
 ```
+Re-register an existing table `db.tbl` in `spark_catalog` to point to new metadata.json file `path/to/metadata/file.json`.
+```sql
+CALL spark_catalog.system.register_table(
+  table => 'db.tbl',
+  metadata_file => 'path/to/metadata/file.json',
+  overwrite => true
+);
+```
 
 ## Metadata information