diff --git a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java index 96f1db1db2b6..83542c02c4e2 100644 --- a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java +++ b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java @@ -405,6 +405,12 @@ public enum ConfVars { "The maximum memory in bytes that the cached objects can use. " + "Memory used is calculated based on estimated size of tables and partitions in the cache. " + "Setting it to a negative value disables memory estimation."), + CACHED_RAW_STORE_PREWARM_THREADS("metastore.cached.rawstore.prewarm.threads", + "hive.metastore.cached.rawstore.prewarm.threads", 1, + "Number of threads CachedStore uses to prewarm the cache from the backing database at startup. " + + "Each thread opens its own connection to the backing database, so this value should be kept " + + "below the connection pool size, and the effective concurrent load on the database during " + + "prewarm scales with it. The default of 1 preserves the original single-threaded prewarm."), CAPABILITY_CHECK("metastore.client.capability.check", "hive.metastore.client.capability.check", true, "Whether to check client capabilities for potentially breaking API usage."), diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java index 1e091b7d671c..dae29e47f089 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java @@ -21,14 +21,12 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.EmptyStackException; import java.util.HashMap; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Optional; -import java.util.Stack; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ThreadFactory; @@ -44,7 +42,6 @@ import org.apache.hadoop.conf.Configurable; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hive.common.DatabaseName; import org.apache.hadoop.hive.common.StatsSetupConst; import org.apache.hadoop.hive.common.TableName; import org.apache.hadoop.hive.metastore.Deadline; @@ -120,7 +117,9 @@ public class CachedStore implements RawStore, Configurable { // This is set to true only if we were able to cache all the metadata. // We may not be able to cache all metadata if we hit CACHED_RAW_STORE_MAX_CACHE_MEMORY limit. private static AtomicBoolean isCachedAllMetadata = new AtomicBoolean(false); - private static TablesPendingPrewarm tblsPendingPrewarm = new TablesPendingPrewarm(); + // The prewarmer while a prewarm is in flight, so that getTable can promote requested tables + // to the front of the prewarm queue; null once prewarm has completed + private static volatile MetaCachePreWarm activePreWarmer = null; private RawStore rawStore = null; private Configuration conf; private static boolean areTxnStatsSupported; @@ -469,171 +468,23 @@ static void prewarm(RawStore rawStore) { } long startTime = System.nanoTime(); LOG.info("Prewarming CachedStore"); - long sleepTime = 100; - while (!isCachePrewarmed.get()) { - // Prevents throwing exceptions in our raw store calls since we're not using RawStoreProxy - Deadline.registerIfNot(1000000); - Collection catalogsToCache; - try { - catalogsToCache = catalogsToCache(rawStore); - LOG.info("Going to cache catalogs: " + org.apache.commons.lang3.StringUtils.join(catalogsToCache, ", ")); - List catalogs = new ArrayList<>(catalogsToCache.size()); - for (String catName : catalogsToCache) { - catalogs.add(rawStore.getCatalog(catName)); - } - sharedCache.populateCatalogsInCache(catalogs); - } catch (MetaException | NoSuchObjectException e) { - LOG.warn("Failed to populate catalogs in cache, going to try again", e); - try { - Thread.sleep(sleepTime); - sleepTime = sleepTime * 2; - } catch (InterruptedException timerEx) { - LOG.info("sleep interrupted", timerEx.getMessage()); - } - // try again - continue; - } - LOG.info("Finished prewarming catalogs, starting on databases"); - List databases = new ArrayList<>(); - for (String catName : catalogsToCache) { - try { - List dbNames = rawStore.getAllDatabases(catName); - LOG.info("Number of databases to prewarm in catalog {}: {}", catName, dbNames.size()); - for (String dbName : dbNames) { - try { - databases.add(rawStore.getDatabase(catName, dbName)); - } catch (NoSuchObjectException e) { - // Continue with next database - LOG.warn("Failed to cache database " + DatabaseName.getQualified(catName, dbName) + ", moving on", e); - } - } - } catch (MetaException e) { - LOG.warn("Failed to cache databases in catalog " + catName + ", moving on", e); - } - } - sharedCache.populateDatabasesInCache(databases); - LOG.info("Databases cache is now prewarmed. Now adding tables, partitions and statistics to the cache"); - int numberOfDatabasesCachedSoFar = 0; - for (Database db : databases) { - String catName = StringUtils.normalizeIdentifier(db.getCatalogName()); - String dbName = StringUtils.normalizeIdentifier(db.getName()); - List tblNames; - try { - tblNames = rawStore.getAllTables(catName, dbName); - } catch (MetaException e) { - LOG.warn("Failed to cache tables for database " + DatabaseName.getQualified(catName, dbName) + ", moving on"); - // Continue with next database - continue; - } - tblsPendingPrewarm.addTableNamesForPrewarming(tblNames); - int totalTablesToCache = tblNames.size(); - int numberOfTablesCachedSoFar = 0; - while (tblsPendingPrewarm.hasMoreTablesToPrewarm()) { - try { - String tblName = StringUtils.normalizeIdentifier(tblsPendingPrewarm.getNextTableNameToPrewarm()); - if (!shouldCacheTable(catName, dbName, tblName)) { - continue; - } - Table table; - try { - table = rawStore.getTable(catName, dbName, tblName); - } catch (MetaException e) { - LOG.debug(ExceptionUtils.getStackTrace(e)); - // It is possible the table is deleted during fetching tables of the database, - // in that case, continue with the next table - continue; - } - List colNames = MetaStoreUtils.getColumnNamesForTable(table); - try { - ColumnStatistics tableColStats = null; - List partitions = null; - List partitionColStats = null; - AggrStats aggrStatsAllPartitions = null; - AggrStats aggrStatsAllButDefaultPartition = null; - TableCacheObjects cacheObjects = new TableCacheObjects(); - if (!table.getPartitionKeys().isEmpty()) { - Deadline.startTimer("getPartitions"); - partitions = rawStore.getPartitions(catName, dbName, tblName, GetPartitionsArgs.getAllPartitions()); - Deadline.stopTimer(); - cacheObjects.setPartitions(partitions); - List partNames = new ArrayList<>(partitions.size()); - for (Partition p : partitions) { - partNames.add(Warehouse.makePartName(table.getPartitionKeys(), p.getValues())); - } - if (!partNames.isEmpty()) { - // Get partition column stats for this table - Deadline.startTimer("getPartitionColumnStatistics"); - partitionColStats = - rawStore.getPartitionColumnStatistics(catName, dbName, tblName, partNames, colNames, CacheUtils.HIVE_ENGINE); - Deadline.stopTimer(); - cacheObjects.setPartitionColStats(partitionColStats); - // Get aggregate stats for all partitions of a table and for all but default - // partition - Deadline.startTimer("getAggrPartitionColumnStatistics"); - aggrStatsAllPartitions = rawStore.get_aggr_stats_for(catName, dbName, tblName, partNames, colNames, CacheUtils.HIVE_ENGINE); - Deadline.stopTimer(); - cacheObjects.setAggrStatsAllPartitions(aggrStatsAllPartitions); - // Remove default partition from partition names and get aggregate - // stats again - List partKeys = table.getPartitionKeys(); - String defaultPartitionValue = - MetastoreConf.getVar(rawStore.getConf(), ConfVars.DEFAULTPARTITIONNAME); - List partCols = new ArrayList<>(); - List partVals = new ArrayList<>(); - for (FieldSchema fs : partKeys) { - partCols.add(fs.getName()); - partVals.add(defaultPartitionValue); - } - String defaultPartitionName = FileUtils.makePartName(partCols, partVals); - partNames.remove(defaultPartitionName); - Deadline.startTimer("getAggrPartitionColumnStatistics"); - aggrStatsAllButDefaultPartition = - rawStore.get_aggr_stats_for(catName, dbName, tblName, partNames, colNames, CacheUtils.HIVE_ENGINE); - Deadline.stopTimer(); - cacheObjects.setAggrStatsAllButDefaultPartition(aggrStatsAllButDefaultPartition); - } - } else { - Deadline.startTimer("getTableColumnStatistics"); - tableColStats = rawStore.getTableColumnStatistics(catName, dbName, tblName, colNames, CacheUtils.HIVE_ENGINE); - Deadline.stopTimer(); - cacheObjects.setTableColStats(tableColStats); - } - - Deadline.startTimer("getAllTableConstraints"); - SQLAllTableConstraints tableConstraints = rawStore.getAllTableConstraints( - new AllTableConstraintsRequest(catName, dbName, tblName)); - Deadline.stopTimer(); - cacheObjects.setTableConstraints(tableConstraints); - - // If the table could not cached due to memory limit, stop prewarm - boolean isSuccess = sharedCache - .populateTableInCache(table, cacheObjects); - if (isSuccess) { - LOG.trace("Cached Database: {}'s Table: {}.", dbName, tblName); - } else { - LOG.info("Unable to cache Database: {}'s Table: {}, since the cache memory is full. " - + "Will stop attempting to cache any more tables.", dbName, tblName); - completePrewarm(startTime, false); - return; - } - } catch (MetaException | NoSuchObjectException e) { - LOG.debug(ExceptionUtils.getStackTrace(e)); - // Continue with next table - continue; - } - LOG.debug("Processed database: {}'s table: {}. Cached {} / {} tables so far.", dbName, tblName, - ++numberOfTablesCachedSoFar, totalTablesToCache); - } catch (EmptyStackException e) { - // We've prewarmed this database, continue with the next one - continue; - } - } - LOG.debug("Processed database: {}. Cached {} / {} databases so far.", dbName, ++numberOfDatabasesCachedSoFar, - databases.size()); - } + boolean cachedAllMetadata; + // The prewarmer is closed (its workers terminated, their stores shut down) before completion + // is published, so nothing can still be mutating the cache once isCachePrewarmed is set + try (MetaCachePreWarm preWarmer = new DefaultMetaCachePreWarm(rawStore, sharedCache)) { + preWarmer.setConf(rawStore.getConf()); + preWarmer.initialize(); + activePreWarmer = preWarmer; + cachedAllMetadata = preWarmer.preWarm(); + } catch (MetaException e) { + throw new RuntimeException("CachedStore prewarm failed", e); + } finally { + activePreWarmer = null; + } + if (cachedAllMetadata) { sharedCache.clearDirtyFlags(); - completePrewarm(startTime, true); } + completePrewarm(startTime, cachedAllMetadata); } /** @@ -658,32 +509,6 @@ static void completePrewarm(long startTime, boolean cachedAllMetadata) { sharedCache.completeTableCachePrewarm(); } - static class TablesPendingPrewarm { - private Stack tableNames = new Stack<>(); - - private synchronized void addTableNamesForPrewarming(List tblNames) { - tableNames.clear(); - if (tblNames != null) { - tableNames.addAll(tblNames); - } - } - - private synchronized boolean hasMoreTablesToPrewarm() { - return !tableNames.empty(); - } - - private synchronized String getNextTableNameToPrewarm() { - return tableNames.pop(); - } - - private synchronized void prioritizeTableForPrewarm(String tblName) { - // If the table is in the pending prewarm list, move it to the top - if (tableNames.remove(tblName)) { - tableNames.push(tblName); - } - } - } - @VisibleForTesting static void setCachePrewarmedState(boolean state) { isCachePrewarmed.set(state); } @@ -695,7 +520,7 @@ private static void initBlackListWhiteList(Configuration conf) { MetastoreConf.getAsString(conf, MetastoreConf.ConfVars.CACHED_RAW_STORE_CACHED_OBJECTS_BLACKLIST)); } - private static Collection catalogsToCache(RawStore rs) throws MetaException { + static Collection catalogsToCache(RawStore rs) { Collection confValue = MetastoreConf.getStringCollection(rs.getConf(), ConfVars.CATALOGS_TO_CACHE); if (confValue == null || confValue.isEmpty() || (confValue.size() == 1 && confValue.contains(""))) { return rs.getCatalogs(); @@ -1364,9 +1189,12 @@ public Table getTable(String catName, String dbName, String tblName, String vali if (tbl == null) { // This table is not yet loaded in cache // If the prewarm thread is working on this table's database, - // let's move this table to the top of tblNamesBeingPrewarmed stack, + // let's move this table to the top of the prewarm queue, // so that it gets loaded to the cache faster and is available for subsequent requests - tblsPendingPrewarm.prioritizeTableForPrewarm(tblName); + MetaCachePreWarm preWarmer = activePreWarmer; + if (preWarmer != null) { + preWarmer.prioritizeTableForPrewarm(new TableName(catName, dbName, tblName)); + } Table t = rawStore.getTable(catName, dbName, tblName, validWriteIds); if (t != null) { sharedCache.addTableToCache(catName, dbName, tblName, t); diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/DefaultMetaCachePreWarm.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/DefaultMetaCachePreWarm.java new file mode 100644 index 000000000000..f41b5251c9ff --- /dev/null +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/DefaultMetaCachePreWarm.java @@ -0,0 +1,455 @@ +/* + * 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.hadoop.hive.metastore.cache; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.EmptyStackException; +import java.util.List; +import java.util.Stack; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.hive.common.DatabaseName; +import org.apache.hadoop.hive.common.TableName; +import org.apache.hadoop.hive.metastore.Deadline; +import org.apache.hadoop.hive.metastore.ObjectStore; +import org.apache.hadoop.hive.metastore.RawStore; +import org.apache.hadoop.hive.metastore.Warehouse; +import org.apache.hadoop.hive.metastore.api.AggrStats; +import org.apache.hadoop.hive.metastore.api.AllTableConstraintsRequest; +import org.apache.hadoop.hive.metastore.api.Catalog; +import org.apache.hadoop.hive.metastore.api.ColumnStatistics; +import org.apache.hadoop.hive.metastore.api.Database; +import org.apache.hadoop.hive.metastore.api.FieldSchema; +import org.apache.hadoop.hive.metastore.api.MetaException; +import org.apache.hadoop.hive.metastore.api.NoSuchObjectException; +import org.apache.hadoop.hive.metastore.api.Partition; +import org.apache.hadoop.hive.metastore.api.SQLAllTableConstraints; +import org.apache.hadoop.hive.metastore.api.Table; +import org.apache.hadoop.hive.metastore.client.builder.GetPartitionsArgs; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf; +import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars; +import org.apache.hadoop.hive.metastore.utils.FileUtils; +import org.apache.hadoop.hive.metastore.utils.JavaUtils; +import org.apache.hadoop.hive.metastore.utils.MetaStoreUtils; +import org.apache.hadoop.hive.metastore.utils.StringUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Default {@link MetaCachePreWarm} implementation: walks the catalogs, databases and tables of + * the backing database through RawStore and populates the given {@link SharedCache}. Warming the + * tables can be spread over metastore.cached.rawstore.prewarm.threads worker threads, each with + * its own RawStore instance and hence its own connection to the backing database. + */ +class DefaultMetaCachePreWarm implements MetaCachePreWarm { + + private static final Logger LOG = LoggerFactory.getLogger(DefaultMetaCachePreWarm.class); + // How long to wait for the prewarm workers to terminate before giving up on a clean shutdown + private static final long WORKER_SHUTDOWN_TIMEOUT_MS = 10000; + + private final RawStore rawStore; + private final SharedCache sharedCache; + private final TablesPendingPrewarm tblsPendingPrewarm = new TablesPendingPrewarm(); + private Configuration conf; + private ExecutorService prewarmPool; + private final List workerStores = new ArrayList<>(); + + DefaultMetaCachePreWarm(RawStore rawStore, SharedCache sharedCache) { + this.rawStore = rawStore; + this.sharedCache = sharedCache; + } + + @Override public void prioritizeTableForPrewarm(TableName... tableNames) { + for (TableName tableName : tableNames) { + tblsPendingPrewarm.prioritizeTableForPrewarm(tableName.getTable()); + } + } + + @Override public void setConf(Configuration conf) { + this.conf = conf; + } + + @Override public Configuration getConf() { + return conf; + } + + @Override public void initialize() throws MetaException { + int prewarmThreads = Math.max(1, MetastoreConf.getIntVar(conf, ConfVars.CACHED_RAW_STORE_PREWARM_THREADS)); + if (prewarmThreads > 1) { + try { + // RawStore implementations (ObjectStore) are not thread safe, so each worker gets its + // own instance and hence its own connection to the backing database + for (int i = 0; i < prewarmThreads; i++) { + workerStores.add(createRawStore(conf)); + } + LOG.info("Prewarming table cache with {} threads", prewarmThreads); + prewarmPool = Executors.newFixedThreadPool(prewarmThreads, new ThreadFactory() { + private final AtomicInteger threadCount = new AtomicInteger(); + @Override public Thread newThread(Runnable r) { + Thread t = Executors.defaultThreadFactory().newThread(r); + t.setName("CachedStore-PrewarmWorker-" + threadCount.getAndIncrement()); + t.setDaemon(true); + return t; + } + }); + } catch (RuntimeException e) { + LOG.warn("Failed to create RawStores for prewarm workers, falling back to single threaded prewarm", e); + close(); + } + } + } + + @Override public boolean preWarm() throws MetaException { + long sleepTime = 100; + while (true) { + // Prevents throwing exceptions in our raw store calls since we're not using RawStoreProxy + Deadline.registerIfNot(1000000); + Collection catalogsToCache; + try { + catalogsToCache = preWarmCatalogs(); + } catch (MetaException | NoSuchObjectException e) { + LOG.warn("Failed to populate catalogs in cache, going to try again", e); + try { + Thread.sleep(sleepTime); + sleepTime = sleepTime * 2; + } catch (InterruptedException timerEx) { + Thread.currentThread().interrupt(); + LOG.warn("Interrupted while waiting to retry the catalog prewarm, stopping prewarm"); + return false; + } + // try again + continue; + } + LOG.info("Finished prewarming catalogs, starting on databases"); + List databases = listDatabases(catalogsToCache); + sharedCache.populateDatabasesInCache(databases); + LOG.info("Databases cache is now prewarmed. Now adding tables, partitions and statistics to the cache"); + return preWarmTables(databases); + } + } + + /** Caches all catalogs and returns their names. */ + private Collection preWarmCatalogs() throws MetaException, NoSuchObjectException { + Collection catalogsToCache = CachedStore.catalogsToCache(rawStore); + LOG.info("Going to cache catalogs: {}", org.apache.commons.lang3.StringUtils.join(catalogsToCache, ", ")); + List catalogs = new ArrayList<>(catalogsToCache.size()); + for (String catName : catalogsToCache) { + catalogs.add(rawStore.getCatalog(catName)); + } + sharedCache.populateCatalogsInCache(catalogs); + return catalogsToCache; + } + + /** Lists the databases of the given catalogs, skipping the ones that cannot be read. */ + private List listDatabases(Collection catalogsToCache) { + List databases = new ArrayList<>(); + for (String catName : catalogsToCache) { + try { + List dbNames = rawStore.getAllDatabases(catName); + LOG.info("Number of databases to prewarm in catalog {}: {}", catName, dbNames.size()); + for (String dbName : dbNames) { + try { + databases.add(rawStore.getDatabase(catName, dbName)); + } catch (NoSuchObjectException e) { + // Continue with next database + LOG.warn("Failed to cache database {}, moving on", DatabaseName.getQualified(catName, dbName), e); + } + } + } catch (MetaException e) { + LOG.warn("Failed to cache databases in catalog {}, moving on", catName, e); + } + } + return databases; + } + + /** + * Warms all cacheable tables of the given databases, using the worker pool when one was + * created. Returns true if all metadata was cached, false if warming stopped early because the + * cache memory limit was reached or the thread was interrupted. The caller must {@link #close()} + * this prewarmer, which awaits worker termination, before publishing completion. + */ + private boolean preWarmTables(List databases) { + int numberOfDatabasesCachedSoFar = 0; + for (Database db : databases) { + String catName = StringUtils.normalizeIdentifier(db.getCatalogName()); + String dbName = StringUtils.normalizeIdentifier(db.getName()); + List tblNames; + try { + tblNames = rawStore.getAllTables(catName, dbName); + } catch (MetaException e) { + LOG.warn("Failed to cache tables for database {}, moving on", DatabaseName.getQualified(catName, dbName)); + // Continue with next database + continue; + } + tblsPendingPrewarm.addTableNamesForPrewarming(tblNames); + int totalTablesToCache = tblNames.size(); + AtomicBoolean stopPrewarm = new AtomicBoolean(false); + AtomicInteger tablesCachedSoFar = new AtomicInteger(); + if (prewarmPool != null) { + List> workers = new ArrayList<>(workerStores.size()); + for (RawStore workerStore : workerStores) { + workers.add(prewarmPool.submit( + () -> drainTablesPendingPrewarm(workerStore, catName, dbName, stopPrewarm, tablesCachedSoFar, + totalTablesToCache))); + } + if (!awaitWorkers(workers, dbName, stopPrewarm)) { + return false; + } + } else { + drainTablesPendingPrewarm(rawStore, catName, dbName, stopPrewarm, tablesCachedSoFar, totalTablesToCache); + } + if (stopPrewarm.get()) { + // The cache is full: stop here and serve with whatever has been cached so far + return false; + } + LOG.debug("Processed database: {}. Cached {} / {} databases so far.", dbName, ++numberOfDatabasesCachedSoFar, + databases.size()); + } + return true; + } + + /** + * Waits for all the given workers to finish. On interruption, tells the remaining workers to + * stop (close() then awaits their termination) and returns false; a failed worker is only + * logged, the tables it could not cache are served from the raw store until the cache update + * service refreshes them. + */ + private boolean awaitWorkers(List> workers, String dbName, AtomicBoolean stopPrewarm) { + for (Future worker : workers) { + try { + worker.get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOG.warn("Interrupted while waiting for prewarm workers on database {}; " + + "completing prewarm with the metadata cached so far", dbName); + stopPrewarm.set(true); + return false; + } catch (ExecutionException e) { + LOG.warn("Prewarm worker failed for database {}, moving on", dbName, e); + } + } + return true; + } + + /** + * Drains tables from tblsPendingPrewarm for the given database, caching each one, until the + * pending list is empty or the shared cache reports its memory limit is reached. Safe to run + * from multiple threads concurrently: the pending-table stack hands out each table exactly once, + * and hot tables promoted by prioritizeTableForPrewarm are picked up by whichever worker pops next. + */ + private void drainTablesPendingPrewarm(RawStore rawStore, String catName, String dbName, + AtomicBoolean stopPrewarm, AtomicInteger tablesCachedSoFar, int totalTablesToCache) { + // Deadline is thread local; register it for prewarm worker threads + Deadline.registerIfNot(1000000); + while (!stopPrewarm.get() && !Thread.currentThread().isInterrupted() + && tblsPendingPrewarm.hasMoreTablesToPrewarm()) { + String tblName; + try { + tblName = StringUtils.normalizeIdentifier(tblsPendingPrewarm.getNextTableNameToPrewarm()); + } catch (EmptyStackException e) { + // Another worker drained the remaining tables between our check and pop + break; + } + if (CachedStore.shouldCacheTable(catName, dbName, tblName)) { + if (!preWarmTable(rawStore, catName, dbName, tblName)) { + LOG.info("Unable to cache Database: {}'s Table: {}, since the cache memory is full. " + + "Will stop attempting to cache any more tables.", dbName, tblName); + stopPrewarm.set(true); + return; + } + LOG.debug("Processed database: {}'s table: {}. Cached {} / {} tables so far.", dbName, tblName, + tablesCachedSoFar.incrementAndGet(), totalTablesToCache); + } + } + } + + /** + * Fetches one table with its partitions, statistics and constraints from the backing database + * and populates it in the shared cache. Returns false only when the shared cache reports that + * its memory limit is reached; a table that vanished or failed to load is skipped by returning + * true so that prewarm continues with the next table. + */ + private boolean preWarmTable(RawStore rawStore, String catName, String dbName, String tblName) { + Table table; + try { + table = rawStore.getTable(catName, dbName, tblName); + } catch (MetaException e) { + LOG.debug(ExceptionUtils.getStackTrace(e)); + // It is possible the table is deleted during fetching tables of the database, + // in that case, continue with the next table + return true; + } + List colNames = MetaStoreUtils.getColumnNamesForTable(table); + try { + ColumnStatistics tableColStats = null; + List partitions = null; + List partitionColStats = null; + AggrStats aggrStatsAllPartitions = null; + AggrStats aggrStatsAllButDefaultPartition = null; + TableCacheObjects cacheObjects = new TableCacheObjects(); + if (!table.getPartitionKeys().isEmpty()) { + Deadline.startTimer("getPartitions"); + partitions = rawStore.getPartitions(catName, dbName, tblName, GetPartitionsArgs.getAllPartitions()); + Deadline.stopTimer(); + cacheObjects.setPartitions(partitions); + List partNames = new ArrayList<>(partitions.size()); + for (Partition p : partitions) { + partNames.add(Warehouse.makePartName(table.getPartitionKeys(), p.getValues())); + } + if (!partNames.isEmpty()) { + // Get partition column stats for this table + Deadline.startTimer("getPartitionColumnStatistics"); + partitionColStats = rawStore.getPartitionColumnStatistics(catName, dbName, tblName, partNames, colNames, + CacheUtils.HIVE_ENGINE); + Deadline.stopTimer(); + cacheObjects.setPartitionColStats(partitionColStats); + // Get aggregate stats for all partitions of a table and for all but default + // partition + Deadline.startTimer("getAggrPartitionColumnStatistics"); + aggrStatsAllPartitions = rawStore.get_aggr_stats_for(catName, dbName, tblName, partNames, colNames, + CacheUtils.HIVE_ENGINE); + Deadline.stopTimer(); + cacheObjects.setAggrStatsAllPartitions(aggrStatsAllPartitions); + // Remove default partition from partition names and get aggregate + // stats again + List partKeys = table.getPartitionKeys(); + String defaultPartitionValue = + MetastoreConf.getVar(rawStore.getConf(), ConfVars.DEFAULTPARTITIONNAME); + List partCols = new ArrayList<>(); + List partVals = new ArrayList<>(); + for (FieldSchema fs : partKeys) { + partCols.add(fs.getName()); + partVals.add(defaultPartitionValue); + } + String defaultPartitionName = FileUtils.makePartName(partCols, partVals); + partNames.remove(defaultPartitionName); + Deadline.startTimer("getAggrPartitionColumnStatistics"); + aggrStatsAllButDefaultPartition = + rawStore.get_aggr_stats_for(catName, dbName, tblName, partNames, colNames, CacheUtils.HIVE_ENGINE); + Deadline.stopTimer(); + cacheObjects.setAggrStatsAllButDefaultPartition(aggrStatsAllButDefaultPartition); + } + } else { + Deadline.startTimer("getTableColumnStatistics"); + tableColStats = rawStore.getTableColumnStatistics(catName, dbName, tblName, colNames, CacheUtils.HIVE_ENGINE); + Deadline.stopTimer(); + cacheObjects.setTableColStats(tableColStats); + } + + Deadline.startTimer("getAllTableConstraints"); + SQLAllTableConstraints tableConstraints = rawStore.getAllTableConstraints( + new AllTableConstraintsRequest(catName, dbName, tblName)); + Deadline.stopTimer(); + cacheObjects.setTableConstraints(tableConstraints); + + // If the table could not be cached due to memory limit, stop prewarm + boolean isSuccess = sharedCache + .populateTableInCache(table, cacheObjects); + if (isSuccess) { + LOG.trace("Cached Database: {}'s Table: {}.", dbName, tblName); + } else { + return false; + } + } catch (MetaException | NoSuchObjectException e) { + LOG.debug(ExceptionUtils.getStackTrace(e)); + // Continue with next table + } + return true; + } + + /** + * Creates a fresh RawStore instance for a prewarm worker thread, mirroring the way + * CacheUpdateMasterWork creates its own store. + */ + private static RawStore createRawStore(Configuration conf) { + String rawStoreClassName = MetastoreConf.getVar(conf, ConfVars.CACHED_RAW_STORE_IMPL, ObjectStore.class.getName()); + try { + RawStore rs = JavaUtils.getClass(rawStoreClassName, RawStore.class).newInstance(); + rs.setConf(conf); + return rs; + } catch (InstantiationException | IllegalAccessException | MetaException e) { + throw new RuntimeException("Cannot instantiate " + rawStoreClassName, e); + } + } + + /** + * Stops the prewarm workers and waits for them to terminate before their RawStores are closed, + * so that no worker can still be reading from a closed store or writing to the shared cache + * once prewarm reports completion. Idempotent. + */ + @Override public void close() { + if (prewarmPool != null) { + prewarmPool.shutdownNow(); + try { + if (!prewarmPool.awaitTermination(WORKER_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { + LOG.warn("Prewarm workers did not terminate within {} ms", WORKER_SHUTDOWN_TIMEOUT_MS); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOG.warn("Interrupted while waiting for the prewarm workers to terminate"); + } + prewarmPool = null; + } + for (RawStore workerStore : workerStores) { + try { + workerStore.shutdown(); + } catch (RuntimeException e) { + LOG.warn("Failed to shut down a prewarm worker RawStore", e); + } + } + workerStores.clear(); + } + + /** The tables of the database currently being prewarmed that have not been cached yet. */ + private static class TablesPendingPrewarm { + private final Stack tableNames = new Stack<>(); + + synchronized void addTableNamesForPrewarming(List tblNames) { + tableNames.clear(); + if (tblNames != null) { + tableNames.addAll(tblNames); + } + } + + synchronized boolean hasMoreTablesToPrewarm() { + return !tableNames.empty(); + } + + synchronized String getNextTableNameToPrewarm() { + return tableNames.pop(); + } + + synchronized void prioritizeTableForPrewarm(String tblName) { + // If the table is in the pending prewarm list, move it to the top + if (tableNames.remove(tblName)) { + tableNames.push(tblName); + } + } + } +} diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/MetaCachePreWarm.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/MetaCachePreWarm.java new file mode 100644 index 000000000000..a6753ca77812 --- /dev/null +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/MetaCachePreWarm.java @@ -0,0 +1,56 @@ +/* + * 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.hadoop.hive.metastore.cache; + +import org.apache.hadoop.conf.Configurable; +import org.apache.hadoop.hive.common.TableName; +import org.apache.hadoop.hive.metastore.api.MetaException; + +/** + * Warms up a metadata cache from the backing database before the cache starts serving. + * Implementations own the resources they need (worker RawStores, thread pools) and release them + * in {@link #close()}, which is expected to be called before the cache reports itself warm. + */ +public interface MetaCachePreWarm extends AutoCloseable, Configurable { + + /** + * Creates the resources needed to prewarm the cache, e.g. worker stores and thread pools. + */ + void initialize() throws MetaException; + + /** + * Populates the cache from the backing database. + * @return true if all metadata was cached; false if prewarm stopped early (the cache memory + * limit was reached or the thread was interrupted) with only part of the metadata cached + */ + boolean preWarm() throws MetaException; + + /** + * Moves the given tables, when they are still pending prewarm, to the front of the prewarm + * queue, so that a table a client is asking for right now becomes available in the cache as + * soon as possible. + */ + void prioritizeTableForPrewarm(TableName... tableNames); + + /** + * Releases the prewarm resources, waiting for any still running workers to terminate so that + * nothing keeps mutating the cache after prewarm reports completion. + */ + @Override void close(); +} diff --git a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java index 97e4a4f375d0..2aa2428317cf 100644 --- a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java +++ b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/SharedCache.java @@ -116,7 +116,10 @@ public class SharedCache { private static final Logger LOG = LoggerFactory.getLogger(SharedCache.class.getName()); private AtomicLong cacheUpdateCount = new AtomicLong(0); private long maxCacheSizeInBytes = -1; - private HashMap, ObjectEstimator> sizeEstimators = null; + // volatile + copy-on-write in getMemorySizeEstimator: tables are cached concurrently during a + // multi threaded prewarm, so this map must never be mutated in place while others read it + private volatile Map, ObjectEstimator> sizeEstimators = null; + private final Object sizeEstimatorsLock = new Object(); private Set tableToUpdateSize = new ConcurrentHashSet<>(); private ScheduledExecutorService executor = null; private Map tableSizeMap = null; @@ -247,14 +250,29 @@ public Thread newThread(Runnable r) { } + /** + * Returns the estimator for the given class, creating it on first use. Estimators are added + * copy-on-write under {@link #sizeEstimatorsLock}: callers can run concurrently (prewarm caches + * tables in parallel), and readers always see a map that is no longer being mutated. + */ private ObjectEstimator getMemorySizeEstimator(Class clazz) { - if (sizeEstimators == null) { + Map, ObjectEstimator> estimators = sizeEstimators; + if (estimators == null) { return null; } - ObjectEstimator estimator = sizeEstimators.get(clazz); + ObjectEstimator estimator = estimators.get(clazz); if (estimator == null) { - IncrementalObjectSizeEstimator.createEstimators(clazz, sizeEstimators); - estimator = sizeEstimators.get(clazz); + synchronized (sizeEstimatorsLock) { + estimators = sizeEstimators; + estimator = estimators.get(clazz); + if (estimator == null) { + // IncrementalObjectSizeEstimator's API is HashMap typed, hence the casts at its boundary + Map, ObjectEstimator> updated = new HashMap<>(estimators); + IncrementalObjectSizeEstimator.createEstimators(clazz, (HashMap, ObjectEstimator>) updated); + estimator = updated.get(clazz); + sizeEstimators = updated; + } + } } return estimator; } @@ -266,7 +284,8 @@ public int getObjectSize(Class clazz, Object obj) { try { ObjectEstimator oe = getMemorySizeEstimator(clazz); - return oe.estimate(obj, sizeEstimators); + // Read the field again: getMemorySizeEstimator may have published a map with more entries + return oe.estimate(obj, (HashMap, ObjectEstimator>) sizeEstimators); } catch (Exception e) { LOG.error("Error while getting object size.", e); } @@ -364,7 +383,7 @@ private int getTableWrapperSizeWithoutMaps() { Object val = field.get(this); ObjectEstimator oe = getMemorySizeEstimator(field.getType()); if (oe != null) { - size += oe.estimate(val, sizeEstimators); + size += oe.estimate(val, (HashMap, ObjectEstimator>) sizeEstimators); } } catch (Exception ex) { LOG.error("Not able to estimate size.", ex); diff --git a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/cache/TestCachedStore.java b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/cache/TestCachedStore.java index bc051f5b7d9d..56d14e761018 100644 --- a/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/cache/TestCachedStore.java +++ b/standalone-metastore/metastore-server/src/test/java/org/apache/hadoop/hive/metastore/cache/TestCachedStore.java @@ -188,6 +188,44 @@ cachedStore.shutdown(); } + /** + * HIVE-30053: same coverage as testPrewarm, but with the multi-threaded prewarm enabled. + * All databases, tables and partitions must be cached regardless of how the tables are + * distributed among the worker threads. + */ + @Test public void testPrewarmMultiThreaded() throws Exception { + Configuration conf = MetastoreConf.newMetastoreConf(); + MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.HIVE_IN_TEST, true); + MetastoreConf.setVar(conf, MetastoreConf.ConfVars.CACHED_RAW_STORE_MAX_CACHE_MEMORY, "-1Kb"); + MetastoreConf.setLongVar(conf, MetastoreConf.ConfVars.CACHED_RAW_STORE_PREWARM_THREADS, 4); + MetaStoreTestUtils.setConfForStandloneMode(conf); + CachedStore cachedStore = new CachedStore(); + CachedStore.clearSharedCache(); + cachedStore.setConfForTest(conf); + ObjectStore objectStore = (ObjectStore) cachedStore.getRawStore(); + CachedStore.setCachePrewarmedState(false); + CachedStore.prewarm(objectStore); + List allDatabases = cachedStore.getAllDatabases(DEFAULT_CATALOG_NAME); + Assert.assertEquals(2, allDatabases.size()); + Assert.assertTrue(allDatabases.contains(db1.getName())); + Assert.assertTrue(allDatabases.contains(db2.getName())); + // All four tables must be in the shared cache after a parallel prewarm + SharedCache sharedCache = CachedStore.getSharedCache(); + Assert.assertNotNull(sharedCache.getTableFromCache(DEFAULT_CATALOG_NAME, db1.getName(), db1Utbl1.getTableName())); + Assert.assertNotNull(sharedCache.getTableFromCache(DEFAULT_CATALOG_NAME, db1.getName(), db1Ptbl1.getTableName())); + Assert.assertNotNull(sharedCache.getTableFromCache(DEFAULT_CATALOG_NAME, db2.getName(), db2Utbl1.getTableName())); + Assert.assertNotNull(sharedCache.getTableFromCache(DEFAULT_CATALOG_NAME, db2.getName(), db2Ptbl1.getTableName())); + List db1Ptbl1Partitions = + cachedStore.getPartitions(DEFAULT_CATALOG_NAME, db1.getName(), db1Ptbl1.getTableName(), + GetPartitionsArgs.getAllPartitions()); + Assert.assertEquals(25, db1Ptbl1Partitions.size()); + List db2Ptbl1Partitions = + cachedStore.getPartitions(DEFAULT_CATALOG_NAME, db2.getName(), db2Ptbl1.getTableName(), + GetPartitionsArgs.getAllPartitions()); + Assert.assertEquals(25, db2Ptbl1Partitions.size()); + cachedStore.shutdown(); + } + @Test public void testPrewarmBlackList() { Configuration conf = MetastoreConf.newMetastoreConf(); MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.HIVE_IN_TEST, true);