Copilot commented on code in PR #6441: URL: https://github.com/apache/hive/pull/6441#discussion_r3789621050
########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCachingCatalog.java: ########## @@ -9,96 +9,703 @@ * * 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. + * 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; -import com.github.benmanes.caffeine.cache.Ticker; +import static org.apache.iceberg.rest.HMSPrivilegeHelper.AccessLevel; + +import java.io.Closeable; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.lang.ref.SoftReference; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; -import org.apache.iceberg.CachingCatalog; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; + +import javax.management.JMException; +import javax.management.MBeanServer; +import javax.management.ObjectName; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Ticker; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.RangerPrivilegeHelper; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.iceberg.BaseMetadataTable; +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; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.catalog.ViewCatalog; +import org.apache.iceberg.exceptions.ForbiddenException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.hive.HiveCatalog; +import org.apache.iceberg.hive.MetadataLocator; import org.apache.iceberg.view.View; import org.apache.iceberg.view.ViewBuilder; - +import org.jetbrains.annotations.TestOnly; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** - * Class that wraps an Iceberg Catalog to cache tables. + * Caching wrapper around a {@link HiveCatalog} that adds two-level table caching and + * per-request authorization enforcement. + * + * <h3>Table caching (L2 + L1)</h3> + * <p><b>L2 — Caffeine cache.</b> The primary table store. Each {@link Table} object is keyed by + * its {@link TableIdentifier} and expires after the configured inactivity period + * ({@code ICEBERG_CATALOG_CACHE_EXPIRY}, in milliseconds). On a cache miss, the table is loaded + * from the underlying {@link HiveCatalog} and its current metadata location is recorded. + * Subsequent hits skip the HMS round-trip entirely.</p> + * + * <p><b>L1 — LinkedHashMap recency guard.</b> A small bounded map (default 32 entries, 3 s TTL; + * configurable via {@code hms.caching.catalog.l1.cache.size} and + * {@code hms.caching.catalog.l1.cache.ttl}) that tracks when each L2-cached table was last + * confirmed fresh. While the L1 entry is live, {@code loadTable} skips the metadata-location + * staleness check against HMS. Once the L1 entry expires, the next call re-validates the stored + * metadata location; if it has changed, the L2 entry is evicted ({@code onCacheInvalidate}) and + * a fresh load is performed. The L1 layer trades a small risk of serving a stale snapshot for a + * large reduction in HMS round-trips under repeated access to the same table.</p> + * + * <p>Both cache levels are invalidated together by {@link #invalidateTable(TableIdentifier)}, + * which also evicts all derived {@link org.apache.iceberg.MetadataTableType metadata-table} + * entries that share the base identifier.</p> + * + * <h3>Authorization</h3> + * <p>Every table and view operation enforces an access-level check against the authenticated user + * (resolved via {@link org.apache.hadoop.security.UserGroupInformation#getCurrentUser()}). + * Authorization is performed by the configured {@link HMSPrivilegeHelper} + * (typically {@link org.apache.hadoop.hive.metastore.RangerPrivilegeHelper}). If no Ranger + * authorizer is configured the helper returns {@link HMSPrivilegeHelper.AccessLevel#NONE} for + * all requests, so access is <em>denied</em> rather than open by default.</p> + * + * <p>Access levels are cached in a single Caffeine cache (configurable via + * {@code hms.caching.catalog.access.cache.size}, default 256) that expires entries after the same + * TTL as the table cache. The cache is keyed by {@link TableIdentifier}: table and view operations + * use the identifier directly; namespace operations use a synthetic + * {@code TableIdentifier(namespace, "*")} key — {@code "*"} is not a valid Hive identifier + * character, so there is no collision with real table entries.</p> + * <ul> + * <li>{@link HMSPrivilegeHelper.AccessLevel#READ_ONLY READ_ONLY} is required for + * {@code loadTable}/{@code loadView}/{@code listTables}/{@code listViews}.</li> + * <li>{@link HMSPrivilegeHelper.AccessLevel#READ_WRITE READ_WRITE} is required for + * {@code dropTable}/{@code dropView}/{@code renameTable}/{@code renameView}/ + * {@code registerTable}/{@code buildTable}/{@code buildView}.</li> + * </ul> + * <p>Authorization entries are invalidated alongside their object — table-level on + * {@link #invalidateTable(TableIdentifier)}, namespace-level on + * {@link #dropNamespace(org.apache.iceberg.catalog.Namespace)}.</p> + * + * <h3>Observability</h3> + * <p>This class implements {@link HMSCachingCatalogMXBean} and registers itself with the platform + * MBean server under the name {@code org.apache.hive:type=IcebergRESTCatalog,name=<catalogName>} + * so that cache hit/miss counts and invalidation counts can be monitored via JMX.</p> */ -public class HMSCachingCatalog extends CachingCatalog implements SupportsNamespaces, ViewCatalog { +public final class HMSCachingCatalog + implements Catalog, SupportsNamespaces, ViewCatalog, HMSCachingCatalogMXBean, Closeable { + private static final Logger LOG = LoggerFactory.getLogger(HMSCachingCatalog.class); + + @TestOnly + private static SoftReference<HMSCachingCatalog> cacheRef = new SoftReference<>(null); + + @TestOnly + @SuppressWarnings("unchecked") + public static <C extends Catalog> C getLatestCache(Function<HMSCachingCatalog, C> extractor) { + HMSCachingCatalog cache = cacheRef.get(); + if (cache == null) { + return null; + } + return extractor == null ? (C) cache : extractor.apply(cache); + } + + @TestOnly + public HiveCatalog getCatalog() { + return hiveCatalog; + } + + // The underlying HiveCatalog that this caching catalog wraps. private final HiveCatalog hiveCatalog; - - public HMSCachingCatalog(HiveCatalog catalog, long expiration) { - super(catalog, true, expiration, Ticker.systemTicker()); + // A helper that locates the metadata location for a given base table identifier. + private final MetadataLocator metadataLocator; + // An L2 table cache (Caffeine). + private final Cache<TableIdentifier, Table> tableCache; + // An L1 small latency cache. + // This is used to cache the last cached time for each table identifier, + // so that we can skip location check for repeated access to the same table within a short period of time, + // which can significantly reduce the latency for repeated access to the same table. + private final Map<TableIdentifier, Long> l1Cache; + // The TTL for L1 cache (3s). + private final int l1Ttl; + // The L1 cache size. + private final int l1CacheSize; + // Computes privileges for a given table identifier and user. + private final HMSPrivilegeHelper privilegeHelper; + // Unified authz cache: keyed by TableIdentifier for tables/views, or by namespaceIdent(ns) for namespaces. + private final Cache<TableIdentifier, ConcurrentMap<String, HMSPrivilegeHelper.AccessLevel>> accessLevelCache; + // Metrics counters. + private final AtomicLong cacheHitCount = new AtomicLong(0); + private final AtomicLong cacheMissCount = new AtomicLong(0); + private final AtomicLong cacheLoadCount = new AtomicLong(0); + private final AtomicLong cacheInvalidateCount = new AtomicLong(0); + private final AtomicLong cacheMetaLoadCount = new AtomicLong(0); + // L1 cache metrics: counted only when the L2 (Caffeine) cache already has the entry. + private final AtomicLong l1CacheHitCount = new AtomicLong(0); + private final AtomicLong l1CacheMissCount = new AtomicLong(0); + // JMX ObjectName under which this instance is registered (may be null if registration failed). + private ObjectName jmxObjectName; + + + /** + * Creates a new caching catalog that wraps the given HiveCatalog. + * @param catalog the underlying HiveCatalog + * @param expirationMs the expiration time for the L2 cache, in milliseconds + */ + public HMSCachingCatalog(HiveCatalog catalog, long expirationMs) { + this(catalog, expirationMs, RangerPrivilegeHelper.create(catalog.getConf())); + } + + /** + * Creates a new caching catalog that wraps the given HiveCatalog. + * @param catalog the underlying HiveCatalog + * @param expirationMs the expiration time for the L2 cache, in milliseconds + * @param privilegeHelper the helper to compute access levels for tables and namespaces + */ + HMSCachingCatalog(HiveCatalog catalog, long expirationMs, HMSPrivilegeHelper privilegeHelper) { this.hiveCatalog = catalog; + this.metadataLocator = new MetadataLocator(catalog); + this.tableCache = Caffeine.newBuilder() + .expireAfterAccess(expirationMs, TimeUnit.MILLISECONDS) + .ticker(Ticker.systemTicker()) + .build(); + Configuration conf = catalog.getConf(); + if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_IN_TEST)) { + // Only keep a reference to the latest cache for testing purpose, so that tests can manipulate the catalog. + cacheRef = new SoftReference<>(this); + } + int l1size = conf.getInt("hms.caching.catalog.l1.cache.size", 32); + int l1ttl = conf.getInt("hms.caching.catalog.l1.cache.ttl", 3_000); + if (l1size > 0 && l1ttl > 0) { + l1Cache = Collections.synchronizedMap(new LinkedHashMap<TableIdentifier, Long>() { + @Override + protected boolean removeEldestEntry(Map.Entry<TableIdentifier, Long> eldest) { + return size() > l1CacheSize; + } + }); + l1Ttl = l1ttl; + l1CacheSize = l1size; + } else { + l1Cache = Collections.emptyMap(); + l1Ttl = 0; + l1CacheSize = 0; + } + this.privilegeHelper = privilegeHelper; + // Covers both table/view and namespace entries; no need to be greater than the number of + // concurrent users × distinct objects, which is usually small (e.g., 256). + int accessLevelCacheSize = conf.getInt("hms.caching.catalog.access.cache.size", 256); + Caffeine<Object, Object> accessCacheBuilder = Caffeine.newBuilder() + .expireAfterWrite(Duration.ofMillis(expirationMs)) + .ticker(Ticker.systemTicker()); + if (accessLevelCacheSize > 0) { + accessCacheBuilder.maximumSize(accessLevelCacheSize); + } + this.accessLevelCache = accessCacheBuilder.build(); + // Register this instance as a JMX MBean for monitoring. + registerJmx(catalog.name()); + } + + private AccessLevel computeAccessLevel(TableIdentifier ident, String user) { + if (!privilegeHelper.isAvailable()) { + return AccessLevel.READ_WRITE; + } + try { + String dbName = ident.namespace().level(0); + String tableName = ident.name(); + return privilegeHelper.getAccessLevel(dbName, tableName, user); + } catch (Exception e) { + LOG.warn("Access level check failed for {}", ident, e); + return AccessLevel.NONE; + } + } + + private String currentUser() { + try { + return UserGroupInformation.getCurrentUser().getShortUserName(); + } catch (IOException e) { + LOG.warn("Failed to determine current user", e); + return null; + } + } + + private AccessLevel cachedAccessLevel(TableIdentifier ident) { + String user = currentUser(); + if (user == null) { + return AccessLevel.NONE; + } + ConcurrentMap<String, AccessLevel> perUser = accessLevelCache.get(ident, k -> new ConcurrentHashMap<>()); + return perUser.computeIfAbsent(user, u -> computeAccessLevel(ident, u)); + } + + private void checkReadAccess(TableIdentifier ident) { + if (cachedAccessLevel(ident) == AccessLevel.NONE) { + throw new ForbiddenException("Access denied on %s", ident); + } } + private void checkWriteAccess(TableIdentifier ident) { + if (cachedAccessLevel(ident) != AccessLevel.READ_WRITE) { + throw new ForbiddenException("Write access denied on %s", ident); + } + } + + private AccessLevel computeNamespaceAccessLevel(Namespace namespace, String user) { + if (namespace.isEmpty()) { + return AccessLevel.NONE; + } + if (!privilegeHelper.isAvailable()) { + return AccessLevel.READ_WRITE; + } + try { + return privilegeHelper.getNamespaceAccessLevel(namespace.level(0), user); + } catch (Exception e) { + LOG.warn("Namespace access level check failed for {}", namespace, e); + return AccessLevel.NONE; + } + } + + private TableIdentifier namespaceIdent(Namespace ns) { + return TableIdentifier.of(ns, "*"); + } + + private AccessLevel cachedNamespaceAccessLevel(Namespace namespace) { + String user = currentUser(); + if (user == null) { + return AccessLevel.NONE; + } + ConcurrentMap<String, AccessLevel> perUser = accessLevelCache.get(namespaceIdent(namespace), k -> new ConcurrentHashMap<>()); + return perUser.computeIfAbsent(user, u -> computeNamespaceAccessLevel(namespace, u)); + } + + private void checkNamespaceReadAccess(Namespace namespace) { + if (cachedNamespaceAccessLevel(namespace) == AccessLevel.NONE) { + throw new ForbiddenException("Access denied on namespace %s", namespace); + } + } + + private void checkNamespaceWriteAccess(Namespace namespace) { + if (cachedNamespaceAccessLevel(namespace) != AccessLevel.READ_WRITE) { + throw new ForbiddenException("Write access denied on namespace %s", namespace); + } + } + + /** + * Registers this instance as a JMX MBean. + * + * @param catalogName the catalog name, used to build the {@link ObjectName} + */ + private void registerJmx(String catalogName) { + try { + MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); + String sanitized = catalogName == null || catalogName.isEmpty() + ? "default" + : catalogName.replaceAll("[^a-zA-Z0-9.\\\\-]", "_"); + ObjectName name = new ObjectName("org.apache.iceberg.rest:type=HMSCachingCatalog,name=" + sanitized); + if (mbs.isRegistered(name)) { + mbs.unregisterMBean(name); + } + mbs.registerMBean(this, name); + this.jmxObjectName = name; + LOG.info("Registered JMX MBean: {}", name); + } catch (JMException e) { + LOG.warn("Failed to register JMX MBean for HMSCachingCatalog", e); + } + } + + /** + * Callback when cache invalidates the entry for a given table identifier. + * + * @param tid the table identifier to invalidate + */ + private void onCacheInvalidate(TableIdentifier tid) { + long count = cacheInvalidateCount.incrementAndGet(); + LOG.debug("Cache invalidate {}: {}", tid, count); + } + + /** + * Callback when cache loads a table for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheLoad(TableIdentifier tid) { + long count = cacheLoadCount.incrementAndGet(); + LOG.debug("Cache load {}: {}", tid, count); + } + + /** + * Callback when cache hit for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheHit(TableIdentifier tid) { + long count = cacheHitCount.incrementAndGet(); + LOG.debug("Cache hit {} : {}", tid, count); + } + + /** + * Callback when cache miss occurs for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheMiss(TableIdentifier tid) { + long count = cacheMissCount.incrementAndGet(); + LOG.debug("Cache miss {}: {}", tid, count); + } + + /** + * Callback when cache loads a metadata table for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheMetaLoad(TableIdentifier tid) { + long count = cacheMetaLoadCount.incrementAndGet(); + LOG.debug("Cache meta-load {}: {}", tid, count); + } + + /** + * Callback when an L1 cache hit occurs for a given table identifier. + * Only fired when the L2 cache also has the entry. + * + * @param tid the table identifier + */ + private void onL1CacheHit(TableIdentifier tid) { + long count = l1CacheHitCount.incrementAndGet(); + LOG.debug("L1 cache hit {}: {}", tid, count); + } + + /** + * Callback when an L1 cache miss occurs for a given table identifier. + * Only fired when the L2 cache has the entry but L1 is absent or expired. + * + * @param tid the table identifier + */ + private void onL1CacheMiss(TableIdentifier tid) { + long count = l1CacheMissCount.incrementAndGet(); + LOG.debug("L1 cache miss {}: {}", tid, count); + } + + // Getter methods for accessing metrics @Override - public Catalog.TableBuilder buildTable(TableIdentifier identifier, Schema schema) { - return hiveCatalog.buildTable(identifier, schema); + public long getCacheHitCount() { + return cacheHitCount.get(); + } + + @Override + public long getCacheMissCount() { + return cacheMissCount.get(); + } + + @Override + public long getCacheLoadCount() { + return cacheLoadCount.get(); + } + + @Override + public long getCacheInvalidateCount() { + return cacheInvalidateCount.get(); + } + + @Override + public long getCacheMetaLoadCount() { + return cacheMetaLoadCount.get(); + } + + @Override + public double getCacheHitRate() { + long hits = cacheHitCount.get(); + long total = hits + cacheMissCount.get(); + return total == 0 ? 0.0 : (double) hits / total; + } + + @Override + public long getL1CacheHitCount() { + return l1CacheHitCount.get(); + } + + @Override + public long getL1CacheMissCount() { + return l1CacheMissCount.get(); + } + + @Override + public double getL1CacheHitRate() { + long hits = l1CacheHitCount.get(); + long total = hits + l1CacheMissCount.get(); + return total == 0 ? 0.0 : (double) hits / total; + } + + @Override + public void resetCacheStats() { + cacheHitCount.set(0); + cacheMissCount.set(0); + cacheLoadCount.set(0); + cacheInvalidateCount.set(0); + cacheMetaLoadCount.set(0); + l1CacheHitCount.set(0); + l1CacheMissCount.set(0); + LOG.debug("Cache stats reset"); + } + + @Override + public void close() { + unregisterJmx(); + } + + /** + * Unregisters this instance from the platform MBeanServer. + */ + private void unregisterJmx() { + if (jmxObjectName != null) { + try { + MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); + if (mbs.isRegistered(jmxObjectName)) { + mbs.unregisterMBean(jmxObjectName); + LOG.info("Unregistered JMX MBean: {}", jmxObjectName); + } + } catch (JMException e) { + LOG.warn("Failed to unregister JMX MBean: {}", jmxObjectName, e); + } finally { + jmxObjectName = null; + } + } + } + + @Override + public String name() { + return hiveCatalog.name(); + } + + @Override + public List<TableIdentifier> listTables(Namespace namespace) { + checkNamespaceReadAccess(namespace); + return hiveCatalog.listTables(namespace); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + checkWriteAccess(identifier); + boolean dropped = hiveCatalog.dropTable(identifier, purge); + invalidateTable(identifier); + return dropped; + } + + @Override + public void renameTable(TableIdentifier from, TableIdentifier to) { + checkWriteAccess(from); + hiveCatalog.renameTable(from, to); + invalidateTable(from); + } + + @Override + public Table registerTable(TableIdentifier identifier, String metadataFileLocation) { + checkWriteAccess(identifier); + Table registered = hiveCatalog.registerTable(identifier, metadataFileLocation); + invalidateTable(identifier); + return registered; + } + + @Override + public void invalidateTable(TableIdentifier ident) { + hiveCatalog.invalidateTable(ident); + TableIdentifier canonicalized = ident; + tableCache.invalidate(canonicalized); + tableCache.invalidateAll(metadataTableIdentifiers(canonicalized)); + l1Cache.remove(canonicalized); + accessLevelCache.invalidate(canonicalized); + } + + /** + * Returns the identifiers of all metadata tables derived from the given base table identifier, + * in both upper-case and lower-case type-name forms so that eviction covers both variants. + */ + private List<TableIdentifier> metadataTableIdentifiers(TableIdentifier identifier) { + MetadataTableType[] types = MetadataTableType.values(); + List<TableIdentifier> result = new ArrayList<>(types.length * 2); + for (MetadataTableType type : types) { + result.add(TableIdentifier.parse(identifier + "." + type.name())); + result.add(TableIdentifier.parse(identifier + "." + type.name().toLowerCase(Locale.ROOT))); + } + return result; } @Override - public void createNamespace(Namespace nmspc, Map<String, String> map) { - hiveCatalog.createNamespace(nmspc, map); + public void createNamespace(Namespace namespace, Map<String, String> map) { + hiveCatalog.createNamespace(namespace, map); } @Override - public List<Namespace> listNamespaces(Namespace nmspc) throws NoSuchNamespaceException { - return hiveCatalog.listNamespaces(nmspc); + public List<Namespace> listNamespaces(Namespace namespace) throws NoSuchNamespaceException { + return hiveCatalog.listNamespaces(namespace); } @Override - public Map<String, String> loadNamespaceMetadata(Namespace nmspc) throws NoSuchNamespaceException { - return hiveCatalog.loadNamespaceMetadata(nmspc); + public void invalidateView(TableIdentifier identifier) { + hiveCatalog.invalidateView(identifier); } @Override - public boolean dropNamespace(Namespace nmspc) throws NamespaceNotEmptyException { - List<TableIdentifier> tables = listTables(nmspc); - for (TableIdentifier ident : tables) { + public Table loadTable(final TableIdentifier identifier) { + final TableIdentifier canonicalized = identifier; + checkReadAccess(canonicalized); Review Comment: Metadata-table identifiers have the form `db.base_table.metadata_type`, but this check authorizes `db.metadata_type` because `computeAccessLevel` uses namespace level 0 plus `identifier.name()`. Once an authorized user populates the global cache, a user permitted on an unrelated table named e.g. `db.snapshots` could receive `db.base_table.snapshots` without HMS authorization. Resolve metadata identifiers to their base table before checking access. Security disposition: VALID under THREAT_MODEL.md §13; this is an in-scope REST/HMS boundary authorization bypass (§3.3/§4) of the §5 authorization-scoping property. ########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCachingCatalog.java: ########## @@ -9,96 +9,703 @@ * * 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. + * 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; -import com.github.benmanes.caffeine.cache.Ticker; +import static org.apache.iceberg.rest.HMSPrivilegeHelper.AccessLevel; + +import java.io.Closeable; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.lang.ref.SoftReference; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; -import org.apache.iceberg.CachingCatalog; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; + +import javax.management.JMException; +import javax.management.MBeanServer; +import javax.management.ObjectName; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Ticker; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.RangerPrivilegeHelper; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.iceberg.BaseMetadataTable; +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; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.catalog.ViewCatalog; +import org.apache.iceberg.exceptions.ForbiddenException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.hive.HiveCatalog; +import org.apache.iceberg.hive.MetadataLocator; import org.apache.iceberg.view.View; import org.apache.iceberg.view.ViewBuilder; - +import org.jetbrains.annotations.TestOnly; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** - * Class that wraps an Iceberg Catalog to cache tables. + * Caching wrapper around a {@link HiveCatalog} that adds two-level table caching and + * per-request authorization enforcement. + * + * <h3>Table caching (L2 + L1)</h3> + * <p><b>L2 — Caffeine cache.</b> The primary table store. Each {@link Table} object is keyed by + * its {@link TableIdentifier} and expires after the configured inactivity period + * ({@code ICEBERG_CATALOG_CACHE_EXPIRY}, in milliseconds). On a cache miss, the table is loaded + * from the underlying {@link HiveCatalog} and its current metadata location is recorded. + * Subsequent hits skip the HMS round-trip entirely.</p> + * + * <p><b>L1 — LinkedHashMap recency guard.</b> A small bounded map (default 32 entries, 3 s TTL; + * configurable via {@code hms.caching.catalog.l1.cache.size} and + * {@code hms.caching.catalog.l1.cache.ttl}) that tracks when each L2-cached table was last + * confirmed fresh. While the L1 entry is live, {@code loadTable} skips the metadata-location + * staleness check against HMS. Once the L1 entry expires, the next call re-validates the stored + * metadata location; if it has changed, the L2 entry is evicted ({@code onCacheInvalidate}) and + * a fresh load is performed. The L1 layer trades a small risk of serving a stale snapshot for a + * large reduction in HMS round-trips under repeated access to the same table.</p> + * + * <p>Both cache levels are invalidated together by {@link #invalidateTable(TableIdentifier)}, + * which also evicts all derived {@link org.apache.iceberg.MetadataTableType metadata-table} + * entries that share the base identifier.</p> + * + * <h3>Authorization</h3> + * <p>Every table and view operation enforces an access-level check against the authenticated user + * (resolved via {@link org.apache.hadoop.security.UserGroupInformation#getCurrentUser()}). + * Authorization is performed by the configured {@link HMSPrivilegeHelper} + * (typically {@link org.apache.hadoop.hive.metastore.RangerPrivilegeHelper}). If no Ranger + * authorizer is configured the helper returns {@link HMSPrivilegeHelper.AccessLevel#NONE} for + * all requests, so access is <em>denied</em> rather than open by default.</p> + * + * <p>Access levels are cached in a single Caffeine cache (configurable via + * {@code hms.caching.catalog.access.cache.size}, default 256) that expires entries after the same + * TTL as the table cache. The cache is keyed by {@link TableIdentifier}: table and view operations + * use the identifier directly; namespace operations use a synthetic + * {@code TableIdentifier(namespace, "*")} key — {@code "*"} is not a valid Hive identifier + * character, so there is no collision with real table entries.</p> + * <ul> + * <li>{@link HMSPrivilegeHelper.AccessLevel#READ_ONLY READ_ONLY} is required for + * {@code loadTable}/{@code loadView}/{@code listTables}/{@code listViews}.</li> + * <li>{@link HMSPrivilegeHelper.AccessLevel#READ_WRITE READ_WRITE} is required for + * {@code dropTable}/{@code dropView}/{@code renameTable}/{@code renameView}/ + * {@code registerTable}/{@code buildTable}/{@code buildView}.</li> + * </ul> + * <p>Authorization entries are invalidated alongside their object — table-level on + * {@link #invalidateTable(TableIdentifier)}, namespace-level on + * {@link #dropNamespace(org.apache.iceberg.catalog.Namespace)}.</p> + * + * <h3>Observability</h3> + * <p>This class implements {@link HMSCachingCatalogMXBean} and registers itself with the platform + * MBean server under the name {@code org.apache.hive:type=IcebergRESTCatalog,name=<catalogName>} + * so that cache hit/miss counts and invalidation counts can be monitored via JMX.</p> */ -public class HMSCachingCatalog extends CachingCatalog implements SupportsNamespaces, ViewCatalog { +public final class HMSCachingCatalog + implements Catalog, SupportsNamespaces, ViewCatalog, HMSCachingCatalogMXBean, Closeable { + private static final Logger LOG = LoggerFactory.getLogger(HMSCachingCatalog.class); + + @TestOnly + private static SoftReference<HMSCachingCatalog> cacheRef = new SoftReference<>(null); + + @TestOnly + @SuppressWarnings("unchecked") + public static <C extends Catalog> C getLatestCache(Function<HMSCachingCatalog, C> extractor) { + HMSCachingCatalog cache = cacheRef.get(); + if (cache == null) { + return null; + } + return extractor == null ? (C) cache : extractor.apply(cache); + } + + @TestOnly + public HiveCatalog getCatalog() { + return hiveCatalog; + } + + // The underlying HiveCatalog that this caching catalog wraps. private final HiveCatalog hiveCatalog; - - public HMSCachingCatalog(HiveCatalog catalog, long expiration) { - super(catalog, true, expiration, Ticker.systemTicker()); + // A helper that locates the metadata location for a given base table identifier. + private final MetadataLocator metadataLocator; + // An L2 table cache (Caffeine). + private final Cache<TableIdentifier, Table> tableCache; + // An L1 small latency cache. + // This is used to cache the last cached time for each table identifier, + // so that we can skip location check for repeated access to the same table within a short period of time, + // which can significantly reduce the latency for repeated access to the same table. + private final Map<TableIdentifier, Long> l1Cache; + // The TTL for L1 cache (3s). + private final int l1Ttl; + // The L1 cache size. + private final int l1CacheSize; + // Computes privileges for a given table identifier and user. + private final HMSPrivilegeHelper privilegeHelper; + // Unified authz cache: keyed by TableIdentifier for tables/views, or by namespaceIdent(ns) for namespaces. + private final Cache<TableIdentifier, ConcurrentMap<String, HMSPrivilegeHelper.AccessLevel>> accessLevelCache; + // Metrics counters. + private final AtomicLong cacheHitCount = new AtomicLong(0); + private final AtomicLong cacheMissCount = new AtomicLong(0); + private final AtomicLong cacheLoadCount = new AtomicLong(0); + private final AtomicLong cacheInvalidateCount = new AtomicLong(0); + private final AtomicLong cacheMetaLoadCount = new AtomicLong(0); + // L1 cache metrics: counted only when the L2 (Caffeine) cache already has the entry. + private final AtomicLong l1CacheHitCount = new AtomicLong(0); + private final AtomicLong l1CacheMissCount = new AtomicLong(0); + // JMX ObjectName under which this instance is registered (may be null if registration failed). + private ObjectName jmxObjectName; + + + /** + * Creates a new caching catalog that wraps the given HiveCatalog. + * @param catalog the underlying HiveCatalog + * @param expirationMs the expiration time for the L2 cache, in milliseconds + */ + public HMSCachingCatalog(HiveCatalog catalog, long expirationMs) { + this(catalog, expirationMs, RangerPrivilegeHelper.create(catalog.getConf())); + } + + /** + * Creates a new caching catalog that wraps the given HiveCatalog. + * @param catalog the underlying HiveCatalog + * @param expirationMs the expiration time for the L2 cache, in milliseconds + * @param privilegeHelper the helper to compute access levels for tables and namespaces + */ + HMSCachingCatalog(HiveCatalog catalog, long expirationMs, HMSPrivilegeHelper privilegeHelper) { this.hiveCatalog = catalog; + this.metadataLocator = new MetadataLocator(catalog); + this.tableCache = Caffeine.newBuilder() + .expireAfterAccess(expirationMs, TimeUnit.MILLISECONDS) + .ticker(Ticker.systemTicker()) + .build(); + Configuration conf = catalog.getConf(); + if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_IN_TEST)) { + // Only keep a reference to the latest cache for testing purpose, so that tests can manipulate the catalog. + cacheRef = new SoftReference<>(this); + } + int l1size = conf.getInt("hms.caching.catalog.l1.cache.size", 32); + int l1ttl = conf.getInt("hms.caching.catalog.l1.cache.ttl", 3_000); + if (l1size > 0 && l1ttl > 0) { + l1Cache = Collections.synchronizedMap(new LinkedHashMap<TableIdentifier, Long>() { + @Override + protected boolean removeEldestEntry(Map.Entry<TableIdentifier, Long> eldest) { + return size() > l1CacheSize; + } + }); + l1Ttl = l1ttl; + l1CacheSize = l1size; + } else { + l1Cache = Collections.emptyMap(); + l1Ttl = 0; + l1CacheSize = 0; + } + this.privilegeHelper = privilegeHelper; + // Covers both table/view and namespace entries; no need to be greater than the number of + // concurrent users × distinct objects, which is usually small (e.g., 256). + int accessLevelCacheSize = conf.getInt("hms.caching.catalog.access.cache.size", 256); + Caffeine<Object, Object> accessCacheBuilder = Caffeine.newBuilder() + .expireAfterWrite(Duration.ofMillis(expirationMs)) + .ticker(Ticker.systemTicker()); + if (accessLevelCacheSize > 0) { + accessCacheBuilder.maximumSize(accessLevelCacheSize); + } + this.accessLevelCache = accessCacheBuilder.build(); + // Register this instance as a JMX MBean for monitoring. + registerJmx(catalog.name()); + } + + private AccessLevel computeAccessLevel(TableIdentifier ident, String user) { + if (!privilegeHelper.isAvailable()) { + return AccessLevel.READ_WRITE; + } + try { + String dbName = ident.namespace().level(0); + String tableName = ident.name(); + return privilegeHelper.getAccessLevel(dbName, tableName, user); + } catch (Exception e) { + LOG.warn("Access level check failed for {}", ident, e); + return AccessLevel.NONE; + } + } + + private String currentUser() { + try { + return UserGroupInformation.getCurrentUser().getShortUserName(); + } catch (IOException e) { + LOG.warn("Failed to determine current user", e); + return null; + } + } + + private AccessLevel cachedAccessLevel(TableIdentifier ident) { + String user = currentUser(); + if (user == null) { + return AccessLevel.NONE; + } + ConcurrentMap<String, AccessLevel> perUser = accessLevelCache.get(ident, k -> new ConcurrentHashMap<>()); + return perUser.computeIfAbsent(user, u -> computeAccessLevel(ident, u)); + } + + private void checkReadAccess(TableIdentifier ident) { + if (cachedAccessLevel(ident) == AccessLevel.NONE) { + throw new ForbiddenException("Access denied on %s", ident); + } } + private void checkWriteAccess(TableIdentifier ident) { + if (cachedAccessLevel(ident) != AccessLevel.READ_WRITE) { + throw new ForbiddenException("Write access denied on %s", ident); + } + } + + private AccessLevel computeNamespaceAccessLevel(Namespace namespace, String user) { + if (namespace.isEmpty()) { + return AccessLevel.NONE; + } + if (!privilegeHelper.isAvailable()) { + return AccessLevel.READ_WRITE; + } + try { + return privilegeHelper.getNamespaceAccessLevel(namespace.level(0), user); + } catch (Exception e) { + LOG.warn("Namespace access level check failed for {}", namespace, e); + return AccessLevel.NONE; + } + } + + private TableIdentifier namespaceIdent(Namespace ns) { + return TableIdentifier.of(ns, "*"); + } + + private AccessLevel cachedNamespaceAccessLevel(Namespace namespace) { + String user = currentUser(); + if (user == null) { + return AccessLevel.NONE; + } + ConcurrentMap<String, AccessLevel> perUser = accessLevelCache.get(namespaceIdent(namespace), k -> new ConcurrentHashMap<>()); + return perUser.computeIfAbsent(user, u -> computeNamespaceAccessLevel(namespace, u)); + } + + private void checkNamespaceReadAccess(Namespace namespace) { + if (cachedNamespaceAccessLevel(namespace) == AccessLevel.NONE) { + throw new ForbiddenException("Access denied on namespace %s", namespace); + } + } + + private void checkNamespaceWriteAccess(Namespace namespace) { + if (cachedNamespaceAccessLevel(namespace) != AccessLevel.READ_WRITE) { + throw new ForbiddenException("Write access denied on namespace %s", namespace); + } + } + + /** + * Registers this instance as a JMX MBean. + * + * @param catalogName the catalog name, used to build the {@link ObjectName} + */ + private void registerJmx(String catalogName) { + try { + MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); + String sanitized = catalogName == null || catalogName.isEmpty() + ? "default" + : catalogName.replaceAll("[^a-zA-Z0-9.\\\\-]", "_"); + ObjectName name = new ObjectName("org.apache.iceberg.rest:type=HMSCachingCatalog,name=" + sanitized); + if (mbs.isRegistered(name)) { + mbs.unregisterMBean(name); + } + mbs.registerMBean(this, name); + this.jmxObjectName = name; + LOG.info("Registered JMX MBean: {}", name); + } catch (JMException e) { + LOG.warn("Failed to register JMX MBean for HMSCachingCatalog", e); + } + } + + /** + * Callback when cache invalidates the entry for a given table identifier. + * + * @param tid the table identifier to invalidate + */ + private void onCacheInvalidate(TableIdentifier tid) { + long count = cacheInvalidateCount.incrementAndGet(); + LOG.debug("Cache invalidate {}: {}", tid, count); + } + + /** + * Callback when cache loads a table for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheLoad(TableIdentifier tid) { + long count = cacheLoadCount.incrementAndGet(); + LOG.debug("Cache load {}: {}", tid, count); + } + + /** + * Callback when cache hit for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheHit(TableIdentifier tid) { + long count = cacheHitCount.incrementAndGet(); + LOG.debug("Cache hit {} : {}", tid, count); + } + + /** + * Callback when cache miss occurs for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheMiss(TableIdentifier tid) { + long count = cacheMissCount.incrementAndGet(); + LOG.debug("Cache miss {}: {}", tid, count); + } + + /** + * Callback when cache loads a metadata table for a given table identifier. + * + * @param tid the table identifier + */ + private void onCacheMetaLoad(TableIdentifier tid) { + long count = cacheMetaLoadCount.incrementAndGet(); + LOG.debug("Cache meta-load {}: {}", tid, count); + } + + /** + * Callback when an L1 cache hit occurs for a given table identifier. + * Only fired when the L2 cache also has the entry. + * + * @param tid the table identifier + */ + private void onL1CacheHit(TableIdentifier tid) { + long count = l1CacheHitCount.incrementAndGet(); + LOG.debug("L1 cache hit {}: {}", tid, count); + } + + /** + * Callback when an L1 cache miss occurs for a given table identifier. + * Only fired when the L2 cache has the entry but L1 is absent or expired. + * + * @param tid the table identifier + */ + private void onL1CacheMiss(TableIdentifier tid) { + long count = l1CacheMissCount.incrementAndGet(); + LOG.debug("L1 cache miss {}: {}", tid, count); + } + + // Getter methods for accessing metrics @Override - public Catalog.TableBuilder buildTable(TableIdentifier identifier, Schema schema) { - return hiveCatalog.buildTable(identifier, schema); + public long getCacheHitCount() { + return cacheHitCount.get(); + } + + @Override + public long getCacheMissCount() { + return cacheMissCount.get(); + } + + @Override + public long getCacheLoadCount() { + return cacheLoadCount.get(); + } + + @Override + public long getCacheInvalidateCount() { + return cacheInvalidateCount.get(); + } + + @Override + public long getCacheMetaLoadCount() { + return cacheMetaLoadCount.get(); + } + + @Override + public double getCacheHitRate() { + long hits = cacheHitCount.get(); + long total = hits + cacheMissCount.get(); + return total == 0 ? 0.0 : (double) hits / total; + } + + @Override + public long getL1CacheHitCount() { + return l1CacheHitCount.get(); + } + + @Override + public long getL1CacheMissCount() { + return l1CacheMissCount.get(); + } + + @Override + public double getL1CacheHitRate() { + long hits = l1CacheHitCount.get(); + long total = hits + l1CacheMissCount.get(); + return total == 0 ? 0.0 : (double) hits / total; + } + + @Override + public void resetCacheStats() { + cacheHitCount.set(0); + cacheMissCount.set(0); + cacheLoadCount.set(0); + cacheInvalidateCount.set(0); + cacheMetaLoadCount.set(0); + l1CacheHitCount.set(0); + l1CacheMissCount.set(0); + LOG.debug("Cache stats reset"); + } + + @Override + public void close() { + unregisterJmx(); + } + + /** + * Unregisters this instance from the platform MBeanServer. + */ + private void unregisterJmx() { + if (jmxObjectName != null) { + try { + MBeanServer mbs = ManagementFactory.getPlatformMBeanServer(); + if (mbs.isRegistered(jmxObjectName)) { + mbs.unregisterMBean(jmxObjectName); + LOG.info("Unregistered JMX MBean: {}", jmxObjectName); + } + } catch (JMException e) { + LOG.warn("Failed to unregister JMX MBean: {}", jmxObjectName, e); + } finally { + jmxObjectName = null; + } + } + } + + @Override + public String name() { + return hiveCatalog.name(); + } + + @Override + public List<TableIdentifier> listTables(Namespace namespace) { + checkNamespaceReadAccess(namespace); + return hiveCatalog.listTables(namespace); + } + + @Override + public boolean dropTable(TableIdentifier identifier, boolean purge) { + checkWriteAccess(identifier); Review Comment: `checkWriteAccess` is based on `TABLE_WRITE_PRIVILEGES` (`UPDATE`, `WRITE`, `ALL`), while this is a DROP DDL operation and the helper explicitly says DROP is namespace-scoped. Consequently a legitimate namespace-DROP grant is rejected here, while an UPDATE grant is treated as sufficient until a downstream authorizer intervenes. Model operation-specific authorization (and apply the same correction to rename/register/view DDL) instead of one READ_WRITE level. ########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCachingCatalog.java: ########## @@ -9,96 +9,703 @@ * * 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. + * 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; -import com.github.benmanes.caffeine.cache.Ticker; +import static org.apache.iceberg.rest.HMSPrivilegeHelper.AccessLevel; + +import java.io.Closeable; +import java.io.IOException; +import java.lang.management.ManagementFactory; +import java.lang.ref.SoftReference; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; -import org.apache.iceberg.CachingCatalog; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; + +import javax.management.JMException; +import javax.management.MBeanServer; +import javax.management.ObjectName; + +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; +import com.github.benmanes.caffeine.cache.Ticker; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.conf.HiveConf; +import org.apache.hadoop.hive.metastore.RangerPrivilegeHelper; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.iceberg.BaseMetadataTable; +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; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.catalog.ViewCatalog; +import org.apache.iceberg.exceptions.ForbiddenException; import org.apache.iceberg.exceptions.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; import org.apache.iceberg.hive.HiveCatalog; +import org.apache.iceberg.hive.MetadataLocator; import org.apache.iceberg.view.View; import org.apache.iceberg.view.ViewBuilder; - +import org.jetbrains.annotations.TestOnly; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** - * Class that wraps an Iceberg Catalog to cache tables. + * Caching wrapper around a {@link HiveCatalog} that adds two-level table caching and + * per-request authorization enforcement. + * + * <h3>Table caching (L2 + L1)</h3> + * <p><b>L2 — Caffeine cache.</b> The primary table store. Each {@link Table} object is keyed by + * its {@link TableIdentifier} and expires after the configured inactivity period + * ({@code ICEBERG_CATALOG_CACHE_EXPIRY}, in milliseconds). On a cache miss, the table is loaded + * from the underlying {@link HiveCatalog} and its current metadata location is recorded. + * Subsequent hits skip the HMS round-trip entirely.</p> + * + * <p><b>L1 — LinkedHashMap recency guard.</b> A small bounded map (default 32 entries, 3 s TTL; + * configurable via {@code hms.caching.catalog.l1.cache.size} and + * {@code hms.caching.catalog.l1.cache.ttl}) that tracks when each L2-cached table was last + * confirmed fresh. While the L1 entry is live, {@code loadTable} skips the metadata-location + * staleness check against HMS. Once the L1 entry expires, the next call re-validates the stored + * metadata location; if it has changed, the L2 entry is evicted ({@code onCacheInvalidate}) and + * a fresh load is performed. The L1 layer trades a small risk of serving a stale snapshot for a + * large reduction in HMS round-trips under repeated access to the same table.</p> + * + * <p>Both cache levels are invalidated together by {@link #invalidateTable(TableIdentifier)}, + * which also evicts all derived {@link org.apache.iceberg.MetadataTableType metadata-table} + * entries that share the base identifier.</p> + * + * <h3>Authorization</h3> + * <p>Every table and view operation enforces an access-level check against the authenticated user + * (resolved via {@link org.apache.hadoop.security.UserGroupInformation#getCurrentUser()}). + * Authorization is performed by the configured {@link HMSPrivilegeHelper} + * (typically {@link org.apache.hadoop.hive.metastore.RangerPrivilegeHelper}). If no Ranger + * authorizer is configured the helper returns {@link HMSPrivilegeHelper.AccessLevel#NONE} for + * all requests, so access is <em>denied</em> rather than open by default.</p> + * + * <p>Access levels are cached in a single Caffeine cache (configurable via + * {@code hms.caching.catalog.access.cache.size}, default 256) that expires entries after the same + * TTL as the table cache. The cache is keyed by {@link TableIdentifier}: table and view operations + * use the identifier directly; namespace operations use a synthetic + * {@code TableIdentifier(namespace, "*")} key — {@code "*"} is not a valid Hive identifier + * character, so there is no collision with real table entries.</p> + * <ul> + * <li>{@link HMSPrivilegeHelper.AccessLevel#READ_ONLY READ_ONLY} is required for + * {@code loadTable}/{@code loadView}/{@code listTables}/{@code listViews}.</li> + * <li>{@link HMSPrivilegeHelper.AccessLevel#READ_WRITE READ_WRITE} is required for + * {@code dropTable}/{@code dropView}/{@code renameTable}/{@code renameView}/ + * {@code registerTable}/{@code buildTable}/{@code buildView}.</li> + * </ul> + * <p>Authorization entries are invalidated alongside their object — table-level on + * {@link #invalidateTable(TableIdentifier)}, namespace-level on + * {@link #dropNamespace(org.apache.iceberg.catalog.Namespace)}.</p> + * + * <h3>Observability</h3> + * <p>This class implements {@link HMSCachingCatalogMXBean} and registers itself with the platform + * MBean server under the name {@code org.apache.hive:type=IcebergRESTCatalog,name=<catalogName>} + * so that cache hit/miss counts and invalidation counts can be monitored via JMX.</p> */ -public class HMSCachingCatalog extends CachingCatalog implements SupportsNamespaces, ViewCatalog { +public final class HMSCachingCatalog + implements Catalog, SupportsNamespaces, ViewCatalog, HMSCachingCatalogMXBean, Closeable { + private static final Logger LOG = LoggerFactory.getLogger(HMSCachingCatalog.class); + + @TestOnly + private static SoftReference<HMSCachingCatalog> cacheRef = new SoftReference<>(null); + + @TestOnly + @SuppressWarnings("unchecked") + public static <C extends Catalog> C getLatestCache(Function<HMSCachingCatalog, C> extractor) { + HMSCachingCatalog cache = cacheRef.get(); + if (cache == null) { + return null; + } + return extractor == null ? (C) cache : extractor.apply(cache); + } + + @TestOnly + public HiveCatalog getCatalog() { + return hiveCatalog; + } + + // The underlying HiveCatalog that this caching catalog wraps. private final HiveCatalog hiveCatalog; - - public HMSCachingCatalog(HiveCatalog catalog, long expiration) { - super(catalog, true, expiration, Ticker.systemTicker()); + // A helper that locates the metadata location for a given base table identifier. + private final MetadataLocator metadataLocator; + // An L2 table cache (Caffeine). + private final Cache<TableIdentifier, Table> tableCache; + // An L1 small latency cache. + // This is used to cache the last cached time for each table identifier, + // so that we can skip location check for repeated access to the same table within a short period of time, + // which can significantly reduce the latency for repeated access to the same table. + private final Map<TableIdentifier, Long> l1Cache; + // The TTL for L1 cache (3s). + private final int l1Ttl; + // The L1 cache size. + private final int l1CacheSize; + // Computes privileges for a given table identifier and user. + private final HMSPrivilegeHelper privilegeHelper; + // Unified authz cache: keyed by TableIdentifier for tables/views, or by namespaceIdent(ns) for namespaces. + private final Cache<TableIdentifier, ConcurrentMap<String, HMSPrivilegeHelper.AccessLevel>> accessLevelCache; + // Metrics counters. + private final AtomicLong cacheHitCount = new AtomicLong(0); + private final AtomicLong cacheMissCount = new AtomicLong(0); + private final AtomicLong cacheLoadCount = new AtomicLong(0); + private final AtomicLong cacheInvalidateCount = new AtomicLong(0); + private final AtomicLong cacheMetaLoadCount = new AtomicLong(0); + // L1 cache metrics: counted only when the L2 (Caffeine) cache already has the entry. + private final AtomicLong l1CacheHitCount = new AtomicLong(0); + private final AtomicLong l1CacheMissCount = new AtomicLong(0); + // JMX ObjectName under which this instance is registered (may be null if registration failed). + private ObjectName jmxObjectName; + + + /** + * Creates a new caching catalog that wraps the given HiveCatalog. + * @param catalog the underlying HiveCatalog + * @param expirationMs the expiration time for the L2 cache, in milliseconds + */ + public HMSCachingCatalog(HiveCatalog catalog, long expirationMs) { + this(catalog, expirationMs, RangerPrivilegeHelper.create(catalog.getConf())); + } + + /** + * Creates a new caching catalog that wraps the given HiveCatalog. + * @param catalog the underlying HiveCatalog + * @param expirationMs the expiration time for the L2 cache, in milliseconds + * @param privilegeHelper the helper to compute access levels for tables and namespaces + */ + HMSCachingCatalog(HiveCatalog catalog, long expirationMs, HMSPrivilegeHelper privilegeHelper) { this.hiveCatalog = catalog; + this.metadataLocator = new MetadataLocator(catalog); + this.tableCache = Caffeine.newBuilder() + .expireAfterAccess(expirationMs, TimeUnit.MILLISECONDS) + .ticker(Ticker.systemTicker()) + .build(); + Configuration conf = catalog.getConf(); + if (HiveConf.getBoolVar(conf, HiveConf.ConfVars.HIVE_IN_TEST)) { + // Only keep a reference to the latest cache for testing purpose, so that tests can manipulate the catalog. + cacheRef = new SoftReference<>(this); + } + int l1size = conf.getInt("hms.caching.catalog.l1.cache.size", 32); + int l1ttl = conf.getInt("hms.caching.catalog.l1.cache.ttl", 3_000); + if (l1size > 0 && l1ttl > 0) { + l1Cache = Collections.synchronizedMap(new LinkedHashMap<TableIdentifier, Long>() { + @Override + protected boolean removeEldestEntry(Map.Entry<TableIdentifier, Long> eldest) { + return size() > l1CacheSize; + } + }); + l1Ttl = l1ttl; + l1CacheSize = l1size; + } else { + l1Cache = Collections.emptyMap(); + l1Ttl = 0; + l1CacheSize = 0; + } + this.privilegeHelper = privilegeHelper; + // Covers both table/view and namespace entries; no need to be greater than the number of + // concurrent users × distinct objects, which is usually small (e.g., 256). + int accessLevelCacheSize = conf.getInt("hms.caching.catalog.access.cache.size", 256); + Caffeine<Object, Object> accessCacheBuilder = Caffeine.newBuilder() + .expireAfterWrite(Duration.ofMillis(expirationMs)) + .ticker(Ticker.systemTicker()); + if (accessLevelCacheSize > 0) { + accessCacheBuilder.maximumSize(accessLevelCacheSize); + } + this.accessLevelCache = accessCacheBuilder.build(); + // Register this instance as a JMX MBean for monitoring. + registerJmx(catalog.name()); + } + + private AccessLevel computeAccessLevel(TableIdentifier ident, String user) { + if (!privilegeHelper.isAvailable()) { + return AccessLevel.READ_WRITE; + } + try { + String dbName = ident.namespace().level(0); + String tableName = ident.name(); + return privilegeHelper.getAccessLevel(dbName, tableName, user); + } catch (Exception e) { + LOG.warn("Access level check failed for {}", ident, e); + return AccessLevel.NONE; + } + } Review Comment: This turns any unavailable helper into `READ_WRITE`. In particular, `RangerPrivilegeHelper.create` returns an unavailable helper after authorizer initialization fails, whose own methods deliberately return `NONE`; this branch overrides that fail-closed result and lets users receive globally cached tables without an HMS round trip. Call the helper even when unavailable—the explicit pass-through helpers already return `READ_WRITE` when authorization is intentionally disabled. Security disposition: VALID under THREAT_MODEL.md §13; the REST/HMS client boundary is in scope (§3.3/§4) and authorization scoping is a claimed property (§5). This issue also appears on line 289 of the same file. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
