henrib commented on code in PR #6702: URL: https://github.com/apache/hive/pull/6702#discussion_r3791902935
########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCachingCatalog.java: ########## @@ -9,89 +9,566 @@ * * 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 java.io.Closeable; +import java.lang.management.ManagementFactory; +import java.lang.ref.SoftReference; +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.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.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.NamespaceNotEmptyException; import org.apache.iceberg.exceptions.NoSuchNamespaceException; +import org.apache.iceberg.exceptions.NoSuchTableException; 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. + * + * <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>Observability</h3> + * <p>This class implements {@link HMSCachingCatalogMXBean} and registers itself with the platform + * MBean server under the name {@code org.apache.iceberg.rest:type=HMSCachingCatalog,name=<catalogName>} + * so that cache hit/miss counts and invalidation counts can be monitored via JMX. The + * {@code catalogName} is {@link org.apache.iceberg.catalog.Catalog#name()} of the wrapped catalog + * (the metastore's configured default catalog name), sanitized for use in an {@link ObjectName}.</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; + // 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.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; + } + }); Review Comment: Good catch. Made the L1 map access-ordered (LRU) — `new LinkedHashMap<>(l1size, 0.75f, true)` — so `l1MarkFresh` on an existing key moves it to the tail and `removeEldestEntry` evicts the least-recently-used entry. Fixed in ad771f3240 (base PR #6441). ########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/IcebergAuthorizer.java: ########## @@ -151,14 +155,177 @@ void validateStageCreateTable(String catalogName, Namespace namespace, Map<Strin tableOwnerName, PrincipalType.USER) ); + check(authorizer, HiveOperationType.CREATETABLE, inputs, outputs, "create table " + request.name()); + } + + /** + * Authorizes loading a table. Table cache hits never reach Hive Metastore, so this check makes + * read authorization explicit at the REST layer rather than dependent on cache state. Mirrors + * the {@code QUERY} check {@code ReadTableEvent} performs on a metastore {@code get_table}. + * + * @param catalogName the Hive catalog name + * @param identifier the table identifier (a metadata-table identifier is checked against its + * base table) + * @throws ForbiddenException if the user does not have the required privileges + * @throws IllegalStateException if the authorization plugin fails + */ + void authorizeLoadTable(String catalogName, TableIdentifier identifier) { + var base = baseTableIdentifier(identifier); + check(HiveOperationType.QUERY, List.of(tableOrView(catalogName, base)), List.of(), "select"); + } + + /** + * Authorizes loading a view. Mirrors the {@code QUERY} check performed on a metastore + * {@code get_table} for the view. + * + * @param catalogName the Hive catalog name + * @param identifier the view identifier + * @throws ForbiddenException if the user does not have the required privileges + * @throws IllegalStateException if the authorization plugin fails + */ + void authorizeLoadView(String catalogName, TableIdentifier identifier) { + check(HiveOperationType.QUERY, List.of(tableOrView(catalogName, identifier)), List.of(), "select"); + } + + /** + * Filters a table listing down to the entries the user may see, mirroring how Hive filters a + * {@code SHOW TABLES} result. Hive Metastore performs no pre-event authorization for + * {@code get_tables}, so without this filter a user would see tables they cannot read. A user + * with no privileges receives an empty list rather than an error. + * + * @param catalogName the Hive catalog name + * @param identifiers the full listing to filter + * @return the subset the user is allowed to see (input order preserved) + * @throws IllegalStateException if the authorization plugin fails + */ + List<TableIdentifier> filterTables(String catalogName, List<TableIdentifier> identifiers) { + return filterTableOrViews(catalogName, identifiers, "show tables"); + } + + /** + * Filters a view listing down to the entries the user may see. See {@link #filterTables}. + * + * @param catalogName the Hive catalog name + * @param identifiers the full listing to filter + * @return the subset the user is allowed to see (input order preserved) + * @throws IllegalStateException if the authorization plugin fails + */ + List<TableIdentifier> filterViews(String catalogName, List<TableIdentifier> identifiers) { + return filterTableOrViews(catalogName, identifiers, "show views"); + } + + private List<TableIdentifier> filterTableOrViews(String catalogName, List<TableIdentifier> identifiers, + String commandString) { + var authorizer = authorizerSupplier.get(); + if (authorizer == null) { + return identifiers; + } + List<HivePrivilegeObject> objects = identifiers.stream() + .map(identifier -> tableOrView(catalogName, identifier)) + .collect(Collectors.toList()); + Set<String> allowed = filterListCmd(authorizer, objects, commandString).stream() + .map(object -> tableKey(object.getDbname(), object.getObjectName())) + .collect(Collectors.toSet()); + return identifiers.stream() + .filter(identifier -> allowed.contains(tableKey(identifier.namespace().level(0), identifier.name()))) + .collect(Collectors.toList()); + } + + /** + * Filters a namespace listing down to the databases the user may see, mirroring how Hive filters + * a {@code SHOW DATABASES} result. Hive Metastore performs no pre-event authorization for + * {@code get_databases}, so without this filter a user would see databases they cannot access. + * Multi-level namespaces cannot map to a Hive database and are dropped rather than failing the + * whole listing. + * + * @param catalogName the Hive catalog name + * @param namespaces the full listing to filter + * @return the subset the user is allowed to see (input order preserved) + * @throws IllegalStateException if the authorization plugin fails + */ + List<Namespace> filterNamespaces(String catalogName, List<Namespace> namespaces) { + var authorizer = authorizerSupplier.get(); + if (authorizer == null) { + return namespaces; + } + List<Namespace> singleLevel = namespaces.stream() + .filter(namespace -> namespace.levels().length == 1) + .collect(Collectors.toList()); + List<HivePrivilegeObject> objects = singleLevel.stream() + .map(namespace -> database(catalogName, namespace)) + .collect(Collectors.toList()); + Set<String> allowed = filterListCmd(authorizer, objects, "show databases").stream() + .map(HivePrivilegeObject::getDbname) + .collect(Collectors.toSet()); + return singleLevel.stream() + .filter(namespace -> allowed.contains(namespace.level(0))) + .collect(Collectors.toList()); + } + + private List<HivePrivilegeObject> filterListCmd(HiveAuthorizer authorizer, List<HivePrivilegeObject> objects, + String commandString) { + var builder = new HiveAuthzContext.Builder(); + builder.setCommandString(commandString); + try { + List<HivePrivilegeObject> allowed = authorizer.filterListCmdObjects(objects, builder.build()); + return allowed == null ? List.of() : allowed; + } catch (HiveAccessControlException e) { + throw new ForbiddenException(e, e.getMessage()); + } catch (HiveAuthzPluginException e) { + throw new IllegalStateException("Failed to filter " + commandString + " results", e); + } + } + + private static String tableKey(String database, String name) { + return database + '�' + name; + } Review Comment: Resolved: `tableKey` was removed entirely when the list filters were simplified to map the authorizer's allowed subset straight back to identifiers, so there is no string key/delimiter anymore. ########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/hive/MetadataLocator.java: ########## @@ -0,0 +1,110 @@ +/* + * 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 java.util.Collections; +import java.util.List; + +import org.apache.hadoop.hive.metastore.IMetaStoreClient; +import org.apache.hadoop.hive.metastore.api.GetProjectionsSpec; +import org.apache.hadoop.hive.metastore.api.NoSuchObjectException; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.hive.metastore.client.builder.GetTableProjectionsSpecBuilder; +import org.apache.iceberg.BaseMetastoreTableOperations; +import org.apache.iceberg.ClientPool; +import org.apache.iceberg.MetadataTableType; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.thrift.TException; + +/** + * Fetches the location of a given metadata table. + * <p>Since the location mutates with each transaction, this allows determining if a cached version of the + * table is the latest known in the HMS database.</p> + */ +public class MetadataLocator { + private static final org.slf4j.Logger LOGGER = org.slf4j.LoggerFactory.getLogger(MetadataLocator.class); + private static final GetProjectionsSpec PARAM_SPEC = + new GetTableProjectionsSpecBuilder() + .includeParameters() // only fetches table.parameters + .build(); + private final HiveCatalog catalog; + + public MetadataLocator(HiveCatalog catalog) { + this.catalog = catalog; + } + + public HiveCatalog getCatalog() { + return catalog; + } + + /** + * Returns the location of the metadata table identified by the given identifier, or null if the table is + * not a metadata table. + * <p>This uses the Thrift API to fetch the table parameters, which is more efficient than fetching the entire table object.</p> + * @param identifier the identifier of the metadata table to fetch the location for + * @return the location of the metadata table, or null if the table (or its database/catalog) does + * not exist, or the identifier is not a valid (metadata) table identifier + * @throws RuntimeException if the HMS lookup fails for any reason other than the object not existing + */ Review Comment: Fixed the javadoc in ad771f3240 (base PR #6441): it now states the method accepts a base- or metadata-table identifier and returns the current metadata-file location, and returns null only when the object does not exist or the identifier is neither a valid table nor metadata-table identifier. ########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/rest/HMSCatalogAdapter.java: ########## @@ -25,6 +24,7 @@ import java.io.IOException; import java.time.Clock; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.Map; Review Comment: Removed the unused `java.util.Collections` import in bc8da42090. -- 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]
