HIVE-29035: Fixing cache handling for REST catalog

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

@@ -21,6 +21,7 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
 import org.apache.hadoop.conf.Configurable;
@@ -31,6 +32,7 @@
 import org.apache.hadoop.hive.metastore.TableType;
 import org.apache.hadoop.hive.metastore.api.AlreadyExistsException;
 import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.GetTableRequest;
 import org.apache.hadoop.hive.metastore.api.InvalidOperationException;
 import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
 import org.apache.hadoop.hive.metastore.api.PrincipalType;
@@ -408,23 +410,46 @@ private void validateTableIsIcebergTableOrView(
    */
   @Override
   public boolean tableExists(TableIdentifier identifier) {
+    return Objects.nonNull(fetchTable(identifier));
+  }
+
+  /**
+   * Check whether table exists and return its current metadata location.
+   *
+   * <p>Note: If a hive table with the same identifier exists in catalog, this method will return
+   * {@code null}.
+   *
+   * @param identifier a table identifier
+   * @return the location of the table if it exists, null otherwise
+   */
+  public String getTableMetadataLocation(TableIdentifier identifier) {
+    Table table =  fetchTable(identifier);
+    if (table == null) {
+      return null;
+    }
+    return table.getParameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP);
+  }
+
+  private Table fetchTable(TableIdentifier identifier) {
     TableIdentifier baseTableIdentifier = identifier;
     if (!isValidIdentifier(identifier)) {
       if (!isValidMetadataIdentifier(identifier)) {
-        return false;
+        return null;
       } else {
         baseTableIdentifier = TableIdentifier.of(identifier.namespace().levels());
       }
     }
-
     String database = baseTableIdentifier.namespace().level(0);
     String tableName = baseTableIdentifier.name();
     try {
-      Table table = clients.run(client -> client.getTable(database, tableName));
+      GetTableRequest request = new GetTableRequest();
+      request.setDbName(database);
+      request.setTblName(tableName);
+      Table table = clients.run(client -> client.getTable(request));
       HiveOperationsBase.validateTableIsIceberg(table, fullTableName(name, baseTableIdentifier));
-      return true;
+      return table;
     } catch (NoSuchTableException | NoSuchObjectException e) {
-      return false;
+      return null;
     } catch (TException e) {
       throw new RuntimeException("Failed to check table existence of " + baseTableIdentifier, e);
     } catch (InterruptedException e) {
@@ -434,6 +459,7 @@ public boolean tableExists(TableIdentifier identifier) {
     }
   }
 
+
   @Override
   public boolean viewExists(TableIdentifier viewIdentifier) {
     if (!isValidIdentifier(viewIdentifier)) {

standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java

@@ -1881,7 +1881,7 @@ public enum ConfVars {
         "HMS Iceberg Catalog servlet path component of URL endpoint."
     ),
     ICEBERG_CATALOG_CACHE_EXPIRY("metastore.iceberg.catalog.cache.expiry",
-        "hive.metastore.iceberg.catalog.cache.expiry", -1,
+        "hive.metastore.iceberg.catalog.cache.expiry", 600_000L,
         "HMS Iceberg Catalog cache expiry."
     ),
     HTTPSERVER_THREADPOOL_MIN("hive.metastore.httpserver.threadpool.min",

standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCachingCatalog.java

@@ -19,12 +19,19 @@
 
 package org.apache.iceberg.rest;
 
+import com.github.benmanes.caffeine.cache.Cache;
 import com.github.benmanes.caffeine.cache.Ticker;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.apache.iceberg.BaseMetadataTable;
 import org.apache.iceberg.CachingCatalog;
+import org.apache.iceberg.HasTableOperations;
+import org.apache.iceberg.MetadataTableType;
+import org.apache.iceberg.MetadataTableUtils;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableOperations;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.SupportsNamespaces;
@@ -35,63 +42,261 @@
 import org.apache.iceberg.hive.HiveCatalog;
 import org.apache.iceberg.view.View;
 import org.apache.iceberg.view.ViewBuilder;
-
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * Class that wraps an Iceberg Catalog to cache tables.
  */
 public class HMSCachingCatalog extends CachingCatalog implements SupportsNamespaces, ViewCatalog {
-  private final HiveCatalog hiveCatalog;
+  protected static final Logger LOG = LoggerFactory.getLogger(HMSCachingCatalog.class);
+  protected final HiveCatalog hiveCatalog;
 
   public HMSCachingCatalog(HiveCatalog catalog, long expiration) {
-    super(catalog, true, expiration, Ticker.systemTicker());
+    super(catalog, false, expiration, Ticker.systemTicker());
     this.hiveCatalog = catalog;
   }
 
   @Override
-  public Catalog.TableBuilder buildTable(TableIdentifier identifier, Schema schema) {
-    return hiveCatalog.buildTable(identifier, schema);
+  public void createNamespace(Namespace namespace, Map<String, String> map) {
+    hiveCatalog.createNamespace(namespace, map);
   }
 
   @Override
-  public void createNamespace(Namespace nmspc, Map<String, String> map) {
-    hiveCatalog.createNamespace(nmspc, map);
+  public List<Namespace> listNamespaces(Namespace namespace) throws NoSuchNamespaceException {
+    return hiveCatalog.listNamespaces(namespace);
+  }
+
+  /**
+   * Callback when cache invalidates the entry for a given table identifier.
+   *
+   * @param tid the table identifier to invalidate
+   */
+  protected void onCacheInvalidate(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+
+  /**
+   * Callback when cache loads a table for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheLoad(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+
+  /**
+   * Callback when cache hit for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheHit(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+
+  /**
+   * Callback when cache miss occurs for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheMiss(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+  /**
+   * Callback when cache loads a metadata table for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheMetaLoad(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
   }
 
   @Override
-  public List<Namespace> listNamespaces(Namespace nmspc) throws NoSuchNamespaceException {
-    return hiveCatalog.listNamespaces(nmspc);
+  public Table loadTable(final TableIdentifier identifier) {
+    final TableIdentifier canonicalized = identifier.toLowerCase();
+    final Table cachedTable = tableCache.getIfPresent(canonicalized);
+    if (cachedTable != null) {
+      final String location = hiveCatalog.getTableMetadataLocation(canonicalized);
+      if (location == null) {
+        LOG.debug("Table {} has no location, returning cached table without location", canonicalized);
+      } else {
+        String cachedLocation = cachedTable instanceof HasTableOperations tableOps
+                ? tableOps.operations().current().metadataFileLocation()
+                : null;
+        if (!location.equals(cachedLocation)) {
+          LOG.debug("Invalidate table {}, cached {} != actual {}", canonicalized, cachedLocation, location);
+          // Invalidate the cached table if the location is different
+          invalidateTable(canonicalized);
+          onCacheInvalidate(canonicalized);
+        } else {
+          LOG.debug("Returning cached table: {}", canonicalized);
+          onCacheHit(canonicalized);
+          return cachedTable;
+        }
+      }
+    } else {
+      LOG.debug("Cache miss for table: {}", canonicalized);
+      onCacheMiss(canonicalized);
+    }
+    final Table table = tableCache.get(canonicalized, hiveCatalog::loadTable);
+    if (table instanceof BaseMetadataTable) {
+      // Cache underlying table
+      TableIdentifier originTableIdentifier =
+              TableIdentifier.of(canonicalized.namespace().levels());
+      Table originTable = tableCache.get(originTableIdentifier, hiveCatalog::loadTable);
+      // Share TableOperations instance of origin table for all metadata tables, so that metadata
+      // table instances are refreshed as well when origin table instance is refreshed.
+      if (originTable instanceof HasTableOperations tableOps) {
+        TableOperations ops = tableOps.operations();
+        MetadataTableType type = MetadataTableType.from(canonicalized.name());
+        Table metadataTable =
+                MetadataTableUtils.createMetadataTableInstance(
+                        ops, hiveCatalog.name(), originTableIdentifier, canonicalized, type);
+        tableCache.put(canonicalized, metadataTable);
+        onCacheMetaLoad(canonicalized);
+        LOG.debug("Loaded metadata table: {} for origin table: {}", canonicalized, originTableIdentifier);
+        // Return the metadata table instead of the original table
+        return metadataTable;
+      }
+    }
+    onCacheLoad(canonicalized);
+    LOG.debug("Loaded table: {} ", canonicalized);
+    return table;
+  }
+
+  /**
+   * Callback when cache invalidates the entry for a given table identifier.
+   *
+   * @param tid the table identifier to invalidate
+   */
+  protected void onCacheInvalidate(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+
+  /**
+   * Callback when cache loads a table for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheLoad(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+
+  /**
+   * Callback when cache hit for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheHit(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+
+  /**
+   * Callback when cache miss occurs for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheMiss(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+  /**
+   * Callback when cache loads a metadata table for a given table identifier.
+   *
+   * @param tid the table identifier
+   */
+  protected void onCacheMetaLoad(TableIdentifier tid) {
+    // This method is intentionally left empty. It can be overridden in subclasses if needed.
+  }
+
+  @Override
+  public Table loadTable(final TableIdentifier identifier) {
+    final Cache<TableIdentifier, Table> cache = this.tableCache;
+    final HiveCatalog catalog = this.hiveCatalog;
+    final TableIdentifier canonicalized = identifier.toLowerCase();
+    Table cachedTable = cache.getIfPresent(canonicalized);
+    if (cachedTable != null) {
+      final String location = catalog.getTableMetadataLocation(canonicalized);
+      if (location == null) {
+        LOG.debug("Table {} has no location, returning cached table without location", canonicalized);
+      } else {
+        String cachedLocation = cachedTable instanceof HasTableOperations tableOps
+                ? tableOps.operations().current().metadataFileLocation()
+                : null;
+        if (!location.equals(cachedLocation)) {
+          LOG.debug("Invalidate table {}, cached {} != actual {}", canonicalized, cachedLocation, location);
+          // Invalidate the cached table if the location is different
+          invalidateTable(canonicalized);
+          onCacheInvalidate(canonicalized);
+        } else {
+          LOG.debug("Returning cached table: {}", canonicalized);
+          onCacheHit(canonicalized);
+          return cachedTable;
+        }
+      }
+    } else {
+      LOG.debug("Cache miss for table: {}", canonicalized);
+      onCacheMiss(canonicalized);
+    }
+    Table table = cache.get(canonicalized, catalog::loadTable);
+    if (table instanceof BaseMetadataTable) {
+      // Cache underlying table
+      TableIdentifier originTableIdentifier =
+              TableIdentifier.of(canonicalized.namespace().levels());
+      Table originTable = cache.get(originTableIdentifier, catalog::loadTable);
+      // Share TableOperations instance of origin table for all metadata tables, so that metadata
+      // table instances are refreshed as well when origin table instance is refreshed.
+      if (originTable instanceof HasTableOperations tableOps) {
+        TableOperations ops = tableOps.operations();
+        MetadataTableType type = MetadataTableType.from(canonicalized.name());
+        Table metadataTable =
+                MetadataTableUtils.createMetadataTableInstance(
+                        ops, catalog.name(), originTableIdentifier, canonicalized, type);
+        cache.put(canonicalized, metadataTable);
+        onCacheMetaLoad(canonicalized);
+        LOG.debug("Loaded metadata table: {} for origin table: {}", canonicalized, originTableIdentifier);
+        // Return the metadata table instead of the original table
+        return metadataTable;
+      }
+    }
+    onCacheLoad(canonicalized);
+    LOG.debug("Loaded table: {} ", canonicalized);
+    return table;
   }
 
   @Override
-  public Map<String, String> loadNamespaceMetadata(Namespace nmspc) throws NoSuchNamespaceException {
-    return hiveCatalog.loadNamespaceMetadata(nmspc);
+  public Map<String, String> loadNamespaceMetadata(Namespace namespace) throws NoSuchNamespaceException {
+    return hiveCatalog.loadNamespaceMetadata(namespace);
   }
 
   @Override
-  public boolean dropNamespace(Namespace nmspc) throws NamespaceNotEmptyException {
-    List<TableIdentifier> tables = listTables(nmspc);
+  public boolean dropNamespace(Namespace namespace) throws NamespaceNotEmptyException {
+    List<TableIdentifier> tables = listTables(namespace);
     for (TableIdentifier ident : tables) {
       invalidateTable(ident);
     }
-    return hiveCatalog.dropNamespace(nmspc);
+    return hiveCatalog.dropNamespace(namespace);
   }
 
   @Override
-  public boolean setProperties(Namespace nmspc, Map<String, String> map) throws NoSuchNamespaceException {
-    return hiveCatalog.setProperties(nmspc, map);
+  public boolean setProperties(Namespace namespace, Map<String, String> map) throws NoSuchNamespaceException {
+    return hiveCatalog.setProperties(namespace, map);
   }
 
   @Override
-  public boolean removeProperties(Namespace nmspc, Set<String> set) throws NoSuchNamespaceException {
-    return hiveCatalog.removeProperties(nmspc, set);
+  public boolean removeProperties(Namespace namespace, Set<String> set) throws NoSuchNamespaceException {
+    return hiveCatalog.removeProperties(namespace, set);
   }
 
   @Override
   public boolean namespaceExists(Namespace namespace) {
     return hiveCatalog.namespaceExists(namespace);
   }
 
+  @Override
+  public Catalog.TableBuilder buildTable(TableIdentifier identifier, Schema schema) {
+    return hiveCatalog.buildTable(identifier, schema);
+  }
+
   @Override
   public List<TableIdentifier> listViews(Namespace namespace) {
     return hiveCatalog.listViews(namespace);