[ADH-8380] Iceberg table rename PoC: implement metadata file rename

.github/workflows/api-binary-compatibility.yml

@@ -47,7 +47,7 @@ jobs:
   revapi:
     runs-on: ubuntu-22.04
     env:
-      GITHUB_ACTOR: ${{ github.actor }}
+      GITHUB_USERNAME: ${{ github.actor }}
       GITHUB_TOKEN: ${{ github.token }}
     steps:
       - uses: actions/checkout@v4

build.gradle

@@ -119,21 +119,14 @@ allprojects {
   version = projectVersion
   repositories {
     maven {
+      name = 'arenadata'
       url = 'https://maven.pkg.github.com/arenadata/*'
       credentials {
         username = project.findProperty("gpr.user") ?: System.getenv("GITHUB_USERNAME")
         password = project.findProperty("gpr.key") ?: System.getenv("GITHUB_TOKEN")
       }
     }
     mavenCentral()
-    maven {
-      name = 'arenadata'
-      url = uri('https://maven.pkg.github.com/arenadata/*')
-      credentials {
-        username = System.getenv('GITHUB_ACTOR') ?: ''
-        password = System.getenv('GITHUB_TOKEN') ?: ''
-      }
-    }
     mavenLocal()
   }
 }

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

@@ -162,4 +162,19 @@ private CatalogProperties() {}
 
   public static final String ENCRYPTION_KMS_TYPE = "encryption.kms-type";
   public static final String ENCRYPTION_KMS_IMPL = "encryption.kms-impl";
+
+  /**
+   * Controls whether a table's metadata {@code location} is updated to the new default warehouse
+   * path when a managed table is renamed. When disabled (default), rename only repoints the
+   * metastore entry, matching the Iceberg specification. When enabled, a default-located table is
+   * relocated to the new name's default path so that re-creating a table with the old name does not
+   * share a directory with the renamed table.
+   *
+   * <p><b>Limit.</b> Only writes made <i>after</i> the rename land under the new directory. Data
+   * files written before the rename keep their absolute paths under the old directory, so an
+   * external directory-level purge of the old directory will still destroy those pre-rename files.
+   */
+  public static final String RENAME_UPDATE_METADATA_LOCATION = "rename.metadata.location.update";
+
+  public static final boolean RENAME_UPDATE_LOCATION_DEFAULT = false;
 }

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

@@ -24,6 +24,7 @@
 import static org.apache.iceberg.TableProperties.GC_ENABLED_DEFAULT;
 
 import java.io.IOException;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -37,6 +38,7 @@
 import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.hadoop.Configurable;
 import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.FileInfo;
 import org.apache.iceberg.io.StorageCredential;
 import org.apache.iceberg.io.SupportsBulkOperations;
 import org.apache.iceberg.io.SupportsPrefixOperations;
@@ -101,15 +103,26 @@ private CatalogUtil() {}
    * @param metadata the last valid TableMetadata instance for a dropped table.
    */
   public static void dropTableData(FileIO io, TableMetadata metadata) {
-    if (supportsDirectoryDelete(io, metadata)) {
-      deleteTableDirectory((SupportsPrefixOperations) io, metadata);
-    } else {
-      deleteTableFiles(io, metadata);
-    }
+    deleteTableFiles(io, metadata);
+    deleteTableDirectoryIfEmpty(io, metadata);
   }
 
-  public static void deleteTableDirectory(SupportsPrefixOperations io, TableMetadata metadata) {
-    io.deletePrefix(metadata.location());
+  /** Remove the table base directory if it's empty. */
+  public static void deleteTableDirectoryIfEmpty(FileIO io, TableMetadata metadata) {
+    if (!supportsDirectoryDelete(io, metadata)) {
+      return;
+    }
+
+    SupportsPrefixOperations prefixIo = (SupportsPrefixOperations) io;
+    Iterator<FileInfo> remaining = prefixIo.listPrefix(metadata.location()).iterator();
+    if (!remaining.hasNext()) {
+      prefixIo.deletePrefix(metadata.location());
+    } else {
+      LOG.info(
+          "Skipping base-directory purge for {}: file {} is not referenced by this table",
+          metadata.location(),
+          remaining.next().location());
+    }
   }
 
   private static void deleteTableFiles(FileIO io, TableMetadata metadata) {

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

@@ -18,8 +18,10 @@
  */
 package org.apache.iceberg;
 
+import java.util.List;
 import java.util.Set;
 import org.apache.iceberg.deletes.DeleteGranularity;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 
 public class TableProperties {
@@ -278,6 +280,15 @@ private TableProperties() {}
   public static final String WRITE_PARTITION_SUMMARY_LIMIT = "write.summary.partition-limit";
   public static final int WRITE_PARTITION_SUMMARY_LIMIT_DEFAULT = 0;
 
+  public static final List<String> CUSTOM_WRITE_PATH_PROPERTIES =
+      ImmutableList.of(
+          TableProperties.WRITE_DATA_LOCATION,
+          TableProperties.WRITE_METADATA_LOCATION,
+          TableProperties.WRITE_LOCATION_PROVIDER_IMPL,
+          // legacy aliases still honored by LocationProviders
+          TableProperties.OBJECT_STORE_PATH,
+          TableProperties.WRITE_FOLDER_STORAGE_LOCATION);
+
   /**
    * @deprecated will be removed in 2.0.0, writing manifest lists is always enabled
    */

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java

@@ -18,6 +18,8 @@
  */
 package org.apache.iceberg.hive;
 
+import static org.apache.iceberg.TableProperties.CUSTOM_WRITE_PATH_PROPERTIES;
+
 import java.io.IOException;
 import java.util.List;
 import java.util.Map;
@@ -65,6 +67,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 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.view.BaseMetastoreViewCatalog;
 import org.apache.iceberg.view.View;
 import org.apache.iceberg.view.ViewBuilder;
@@ -366,6 +369,11 @@ private void renameTableOrView(
     String fromName = from.name();
 
     try {
+      if (contentType == HiveOperationsBase.ContentType.TABLE && tryRelocateRename(from, to)) {
+        LOG.info("Updated metadata and renamed Table from {}, to {}", from, to);
+        return;
+      }
+
       Table table = clients.run(client -> client.getTable(fromDatabase, fromName));
       validateTableIsIcebergTableOrView(contentType, table, CatalogUtil.fullTableName(name, from));
 
@@ -406,6 +414,74 @@ private void renameTableOrView(
     }
   }
 
+  /**
+   * Renames a table by committing a metadata location update through {@link HiveTableOperations} so
+   * that the rename of {@code dbName} / {@code tableName} and the update of {@code
+   * metadata_location} happen in a single HMS {@code alter_table} RPC.
+   *
+   * <p>Controlled by {@link CatalogProperties#RENAME_UPDATE_METADATA_LOCATION}. Only tables sitting
+   * at the default warehouse location of the source identifier are relocated; tables created with
+   * an explicit {@code LOCATION} return {@code false} so the caller falls back to the plain rename
+   * path.
+   *
+   * @return {@code true} if the rename was performed by the relocate path; {@code false} if the
+   *     caller should fall back to the standard {@code alter_table}-only rename.
+   */
+  private boolean tryRelocateRename(TableIdentifier from, TableIdentifier to) {
+    if (!PropertyUtil.propertyAsBoolean(
+        catalogProperties,
+        CatalogProperties.RENAME_UPDATE_METADATA_LOCATION,
+        CatalogProperties.RENAME_UPDATE_LOCATION_DEFAULT)) {
+      return false;
+    }
+
+    String oldTableDefaultLocation =
+        LocationUtil.stripTrailingSlash(defaultWarehouseLocation(from));
+    String newTableDefaultLocation = LocationUtil.stripTrailingSlash(defaultWarehouseLocation(to));
+    if (oldTableDefaultLocation.equals(newTableDefaultLocation)) {
+      return false;
+    }
+
+    HiveTableOperations ops = newTableOps(from, hmsTable -> updateHmsTableName(hmsTable, to));
+
+    TableMetadata current;
+    try {
+      current = ops.current();
+    } catch (NoSuchTableException e) {
+      // let base path handle this exception
+      return false;
+    }
+
+    if (current == null) {
+      return false;
+    }
+
+    if (!LocationUtil.stripTrailingSlash(current.location()).equals(oldTableDefaultLocation)) {
+      LOG.info(
+          "Not updating location of renamed table {} -> {}: table has an explicit location {}",
+          from,
+          to,
+          current.location());
+      return false;
+    }
+
+    List<String> customWriteProperties = customWritePathProperties(current.properties());
+    if (!customWriteProperties.isEmpty()) {
+      LOG.info(
+          "Not updating location of renamed table {} -> {}: table sets custom write-path "
+              + "properties {} that override metadata.location for new writes",
+          from,
+          to,
+          customWriteProperties);
+      return false;
+    }
+
+    // update metadata location + table name via HmsTablePreCommitHandler
+    ops.commit(current, current.updateLocation(newTableDefaultLocation));
+    LOG.info("Updated location of renamed table {} -> {} to {}", from, to, newTableDefaultLocation);
+    return true;
+  }
+
   private void validateTableIsIcebergTableOrView(
       HiveOperationsBase.ContentType contentType, Table table, String fullName) {
     switch (contentType) {
@@ -417,6 +493,17 @@ private void validateTableIsIcebergTableOrView(
     }
   }
 
+  private void updateHmsTableName(Table hmsTable, TableIdentifier to) {
+    hmsTable.setDbName(to.namespace().level(0));
+    hmsTable.setTableName(to.name());
+  }
+
+  private static List<String> customWritePathProperties(Map<String, String> properties) {
+    return CUSTOM_WRITE_PATH_PROPERTIES.stream()
+        .filter(properties::containsKey)
+        .collect(Collectors.toList());
+  }
+
   /**
    * Check whether table or metadata table exists.
    *
@@ -692,10 +779,22 @@ private boolean isValidateNamespace(Namespace namespace) {
 
   @Override
   public TableOperations newTableOps(TableIdentifier tableIdentifier) {
+    return newTableOps(tableIdentifier, HmsTablePreCommitHandler.NO_OP_HANDLER);
+  }
+
+  protected HiveTableOperations newTableOps(
+      TableIdentifier tableIdentifier, HmsTablePreCommitHandler hmsTablePreCommitHandler) {
     String dbName = tableIdentifier.namespace().level(0);
     String tableName = tableIdentifier.name();
     return new HiveTableOperations(
-        conf, clients, fileIO, keyManagementClient, name, dbName, tableName);
+        conf,
+        clients,
+        fileIO,
+        keyManagementClient,
+        hmsTablePreCommitHandler,
+        name,
+        dbName,
+        tableName);
   }
 
   @Override

hive-metastore/src/main/java/org/apache/iceberg/hive/HiveTableOperations.java

@@ -81,6 +81,7 @@ public class HiveTableOperations extends BaseMetastoreTableOperations
   private final FileIO fileIO;
   private final KeyManagementClient keyManagementClient;
   private final ClientPool<IMetaStoreClient, TException> metaClients;
+  private final HmsTablePreCommitHandler hmsTablePreCommitHandler;
 
   private EncryptionManager encryptionManager;
   private EncryptingFileIO encryptingFileIO;
@@ -96,10 +97,31 @@ protected HiveTableOperations(
       String catalogName,
       String database,
       String table) {
+    this(
+        conf,
+        metaClients,
+        fileIO,
+        keyManagementClient,
+        HmsTablePreCommitHandler.NO_OP_HANDLER,
+        catalogName,
+        database,
+        table);
+  }
+
+  protected HiveTableOperations(
+      Configuration conf,
+      ClientPool<IMetaStoreClient, TException> metaClients,
+      FileIO fileIO,
+      KeyManagementClient keyManagementClient,
+      HmsTablePreCommitHandler hmsTablePreCommitHandler,
+      String catalogName,
+      String database,
+      String table) {
     this.conf = conf;
     this.metaClients = metaClients;
     this.fileIO = fileIO;
     this.keyManagementClient = keyManagementClient;
+    this.hmsTablePreCommitHandler = hmsTablePreCommitHandler;
     this.fullName = catalogName + "." + database + "." + table;
     this.catalogName = catalogName;
     this.database = database;
@@ -301,6 +323,8 @@ protected void doCommit(TableMetadata base, TableMetadata metadata) {
           maxHiveTablePropertySize,
           currentMetadataLocation());
 
+      hmsTablePreCommitHandler.handle(tbl);
+
       if (!keepHiveStats) {
         tbl.getParameters().remove(StatsSetupConst.COLUMN_STATS_ACCURATE);
         tbl.getParameters().put(StatsSetupConst.DO_NOT_UPDATE_STATS, StatsSetupConst.TRUE);

hive-metastore/src/main/java/org/apache/iceberg/hive/HmsTablePreCommitHandler.java

@@ -0,0 +1,27 @@
+/*
+ * 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.hive;
+
+import org.apache.hadoop.hive.metastore.api.Table;
+
+public interface HmsTablePreCommitHandler {
+  HmsTablePreCommitHandler NO_OP_HANDLER = hmsTable -> {};
+
+  void handle(Table hmsTable);
+}