henrib commented on code in PR #6441: URL: https://github.com/apache/hive/pull/6441#discussion_r3791963838
########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/hive/MetadataLocator.java: ########## @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.iceberg.hive; + +import org.apache.hadoop.hive.metastore.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.iceberg.exceptions.NoSuchTableException; +import org.apache.thrift.TException; + +import java.util.Collections; +import java.util.List; + +/** + * 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 does not exist or is not a metadata table + * @throws NoSuchTableException if the table does not exist + */ + public String getLocation(TableIdentifier identifier) { + final ClientPool<IMetaStoreClient, TException> clients = catalog.clientPool(); + final String catName = catalog.name(); + final TableIdentifier baseTableIdentifier; + if (!catalog.isValidIdentifier(identifier)) { + if (!isValidMetadataIdentifier(identifier)) { + return null; + } else { + baseTableIdentifier = TableIdentifier.of(identifier.namespace().levels()); + } + } else { + baseTableIdentifier = identifier; + } + String database = baseTableIdentifier.namespace().level(0); + String tableName = baseTableIdentifier.name(); + try { + List<Table> tables = clients.run( + client -> client.getTables(catName, database, Collections.singletonList(tableName), PARAM_SPEC) Review Comment: Addressed in the current revision — getTables with PARAM_SPEC is used. ########## standalone-metastore/metastore-rest-catalog/src/main/java/org/apache/iceberg/hive/MetadataLocator.java: ########## @@ -0,0 +1,104 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.iceberg.hive; + +import org.apache.hadoop.hive.metastore.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.iceberg.exceptions.NoSuchTableException; +import org.apache.thrift.TException; + +import java.util.Collections; +import java.util.List; + +/** + * 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 does not exist or is not a metadata table + * @throws NoSuchTableException if the table does not exist + */ + public String getLocation(TableIdentifier identifier) { + final ClientPool<IMetaStoreClient, TException> clients = catalog.clientPool(); + final String catName = catalog.name(); + final TableIdentifier baseTableIdentifier; + if (!catalog.isValidIdentifier(identifier)) { + if (!isValidMetadataIdentifier(identifier)) { + return null; + } else { + baseTableIdentifier = TableIdentifier.of(identifier.namespace().levels()); + } + } else { + baseTableIdentifier = identifier; + } + String database = baseTableIdentifier.namespace().level(0); + String tableName = baseTableIdentifier.name(); + try { + List<Table> tables = clients.run( + client -> client.getTables(catName, database, Collections.singletonList(tableName), PARAM_SPEC) + ); + return tables == null || tables.isEmpty() + ? null + : tables.getFirst().getParameters().get(BaseMetastoreTableOperations.METADATA_LOCATION_PROP); + } catch (NoSuchTableException e) { + LOGGER.debug("Table {} not found: {}", baseTableIdentifier, e.getMessage()); + throw e; + } catch (NoSuchObjectException e) { + throw new NoSuchTableException("Table %s not found: %s", baseTableIdentifier, e.getMessage()); Review Comment: Addressed in the current revision. ########## 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: Obsolete — the authorization helpers were removed in this revision. HMSCachingCatalog is now a pure cache; authorization is enforced at the REST choke point in HIVE-29817. -- 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]
