Core: Add in RenameTableRequest for the RESTCatalog

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

@@ -194,7 +194,9 @@ public List<TableIdentifier> listTables(Namespace namespace) {
   public void renameTable(TableIdentifier from, TableIdentifier to) {
     int updatedRecords = execute(
         err -> {
-          if (err instanceof SQLIntegrityConstraintViolationException) {
+          // SQLite doesn't set SQLState or throw SQLIntegrityConstraintViolationException
+          if (err instanceof SQLIntegrityConstraintViolationException ||
+              err.getMessage() != null && err.getMessage().contains("constraint failed")) {
             throw new AlreadyExistsException("Table already exists: %s", to);
           }
         },

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

@@ -47,6 +47,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.CreateNamespaceResponse;
@@ -245,6 +246,10 @@ public static LoadTableResponse updateTable(Catalog catalog, TableIdentifier ide
         .build();
   }
 
+  public static void renameTable(Catalog catalog, RenameTableRequest request) {
+    catalog.renameTable(request.source(), request.destination());
+  }
+
   private static boolean isCreate(UpdateTableRequest request) {
     boolean isCreate = request.requirements().stream()
         .anyMatch(UpdateTableRequest.UpdateRequirement.AssertTableDoesNotExist.class::isInstance);

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

@@ -53,6 +53,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.responses.ConfigResponse;
 import org.apache.iceberg.rest.responses.CreateNamespaceResponse;
@@ -129,7 +130,13 @@ public boolean dropTable(TableIdentifier identifier, boolean purge) {
 
   @Override
   public void renameTable(TableIdentifier from, TableIdentifier to) {
+    RenameTableRequest request = RenameTableRequest.builder()
+        .withSource(from)
+        .withDestination(to)
+        .build();
 
+    // for now, ignore the response because there is no way to return it.
+    client.post("v1/tables/rename", request, null, ErrorHandlers.tableErrorHandler());
   }
 
   private LoadTableResponse loadInternal(TableIdentifier identifier) {

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

@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.rest.requests;
+
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.rest.RESTRequest;
+
+/**
+ * A REST request to rename a table.
+ */
+public class RenameTableRequest implements RESTRequest {
+
+  private TableIdentifier source;
+  private TableIdentifier destination;
+
+  @SuppressWarnings("unused")
+  public RenameTableRequest() {
+    // Needed for Jackson Deserialization.
+  }
+
+  private RenameTableRequest(TableIdentifier source, TableIdentifier destination) {
+    this.source = source;
+    this.destination = destination;
+    validate();
+  }
+
+  public void validate() {
+    Preconditions.checkArgument(source != null, "Invalid source table: null");
+    Preconditions.checkArgument(destination != null, "Invalid destination table: null");
+  }
+
+  public TableIdentifier source() {
+    return source;
+  }
+
+  public TableIdentifier destination() {
+    return destination;
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("source", source)
+        .add("destination", destination)
+        .toString();
+  }
+
+  public static Builder builder() {
+    return new Builder();
+  }
+
+  public static class Builder {
+    private TableIdentifier source;
+    private TableIdentifier destination;
+
+    private Builder() {
+    }
+
+    public Builder withSource(TableIdentifier sourceTable) {
+      Preconditions.checkNotNull(sourceTable, "Invalid source table identifier: null");
+      this.source = sourceTable;
+      return this;
+    }
+
+    public Builder withDestination(TableIdentifier destinationTable) {
+      Preconditions.checkNotNull(destinationTable, "Invalid destination table identifier: null");
+      this.destination = destinationTable;
+      return this;
+    }
+
+    public RenameTableRequest build() {
+      return new RenameTableRequest(source, destination);
+    }
+  }
+}

core/src/test/java/org/apache/iceberg/catalog/CatalogTests.java

@@ -31,6 +31,7 @@
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.FileScanTask;
+import org.apache.iceberg.HasTableOperations;
 import org.apache.iceberg.PartitionSpec;
 import org.apache.iceberg.ReplaceSortOrder;
 import org.apache.iceberg.Schema;
@@ -65,6 +66,7 @@
 public abstract class CatalogTests<C extends Catalog & SupportsNamespaces> {
   private static final Namespace NS = Namespace.of("newdb");
   private static final TableIdentifier TABLE = TableIdentifier.of(NS, "table");
+  private static final TableIdentifier RENAMED_TABLE = TableIdentifier.of(NS, "table_renamed");
 
   // Schema passed to create tables
   private static final Schema SCHEMA = new Schema(
@@ -562,6 +564,78 @@ public void testLoadMissingTable() {
         () -> catalog.loadTable(ident));
   }
 
+  @Test
+  public void testRenameTable() {
+    C catalog = catalog();
+
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(NS);
+    }
+
+    Assert.assertFalse("Source table should not exist before create", catalog.tableExists(TABLE));
+
+    catalog.buildTable(TABLE, SCHEMA).create();
+    Assert.assertTrue("Table should exist after create", catalog.tableExists(TABLE));
+
+    Assert.assertFalse("Destination table should not exist before rename", catalog.tableExists(RENAMED_TABLE));
+
+    catalog.renameTable(TABLE, RENAMED_TABLE);
+    Assert.assertTrue("Table should exist with new name", catalog.tableExists(RENAMED_TABLE));
+    Assert.assertFalse("Original table should no longer exist", catalog.tableExists(TABLE));
+
+    catalog.dropTable(RENAMED_TABLE);
+    assertEmpty("Should not contain table after drop", catalog, NS);
+  }
+
+  @Test
+  public void testRenameTableMissingSourceTable() {
+    C catalog = catalog();
+
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(NS);
+    }
+
+    Assert.assertFalse("Source table should not exist before rename", catalog.tableExists(TABLE));
+    Assert.assertFalse("Destination table should not exist before rename", catalog.tableExists(RENAMED_TABLE));
+
+    AssertHelpers.assertThrows("Should reject renaming a table that does not exist",
+        NoSuchTableException.class,
+        "Table does not exist",
+        () -> catalog.renameTable(TABLE, RENAMED_TABLE));
+
+    Assert.assertFalse("Destination table should not exist after failed rename", catalog.tableExists(RENAMED_TABLE));
+  }
+
+  @Test
+  public void testRenameTableDestinationTableAlreadyExists() {
+    C catalog = catalog();
+
+    if (requiresNamespaceCreate()) {
+      catalog.createNamespace(NS);
+    }
+
+    Assert.assertFalse("Source table should not exist before create", catalog.tableExists(TABLE));
+    catalog.buildTable(TABLE, SCHEMA).create();
+    Assert.assertTrue("Source table should exist after create", catalog.tableExists(TABLE));
+
+    Assert.assertFalse("Destination table should not exist before create", catalog.tableExists(RENAMED_TABLE));
+    catalog.buildTable(RENAMED_TABLE, SCHEMA).create();
+    Assert.assertTrue("Destination table should exist after create", catalog.tableExists(RENAMED_TABLE));
+
+    AssertHelpers.assertThrows("Should reject renaming a table if the new name already exists",
+        AlreadyExistsException.class,
+        "Table already exists",
+        () -> catalog.renameTable(TABLE, RENAMED_TABLE));
+
+    Assert.assertTrue("Source table should still exist after failed rename", catalog.tableExists(TABLE));
+    Assert.assertTrue("Destination table should still exist after failed rename", catalog.tableExists(RENAMED_TABLE));
+
+    String sourceTableUUID = ((HasTableOperations) catalog.loadTable(TABLE)).operations().current().uuid();
+    String destinationTableUUID = ((HasTableOperations) catalog.loadTable(RENAMED_TABLE)).operations().current().uuid();
+    Assert.assertNotEquals("Source and destination table should remain distinct after failed rename",
+        sourceTableUUID, destinationTableUUID);
+  }
+
   @Test
   public void testDropTable() {
     C catalog = catalog();

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

@@ -385,8 +385,8 @@ public void testRenameTable() {
     // rename table to existing table name!
     TableIdentifier from2 = TableIdentifier.of("db", "tbl2");
     catalog.createTable(from2, SCHEMA, PartitionSpec.unpartitioned());
-    AssertHelpers.assertThrows("should throw exception", UncheckedSQLException.class,
-        "Failed to execute", () -> catalog.renameTable(from2, to)
+    AssertHelpers.assertThrows("should throw exception", AlreadyExistsException.class,
+        "Table already exists", () -> catalog.renameTable(from2, to)
     );
   }
 

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

@@ -43,6 +43,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.rest.requests.CreateNamespaceRequest;
 import org.apache.iceberg.rest.requests.CreateTableRequest;
+import org.apache.iceberg.rest.requests.RenameTableRequest;
 import org.apache.iceberg.rest.requests.UpdateNamespacePropertiesRequest;
 import org.apache.iceberg.rest.requests.UpdateTableRequest;
 import org.apache.iceberg.rest.responses.ConfigResponse;
@@ -98,7 +99,8 @@ private enum Route {
     CREATE_TABLE(HTTPMethod.POST, "v1/namespaces/{namespace}/tables"),
     LOAD_TABLE(HTTPMethod.GET, "v1/namespaces/{namespace}/tables/{table}"),
     UPDATE_TABLE(HTTPMethod.POST, "v1/namespaces/{namespace}/tables/{table}"),
-    DROP_TABLE(HTTPMethod.DELETE, "v1/namespaces/{namespace}/tables/{table}");
+    DROP_TABLE(HTTPMethod.DELETE, "v1/namespaces/{namespace}/tables/{table}"),
+    RENAME_TABLE(HTTPMethod.POST, "v1/tables/rename");
 
     private final HTTPMethod method;
     private final int requriedLength;
@@ -225,6 +227,12 @@ public <T extends RESTResponse> T handleRequest(Route route, Map<String, String>
         return castResponse(responseType, CatalogHandlers.updateTable(catalog, ident, request));
       }
 
+      case RENAME_TABLE: {
+        RenameTableRequest request = castRequest(RenameTableRequest.class, body);
+        CatalogHandlers.renameTable(catalog, request);
+        return null;
+      }
+
       default:
     }