This is an automated email from the ASF dual-hosted git repository.
dengzhhu653 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 21dc6c134b2 HIVE-30053: Support multi-threaded CachedStore prewarm
(#6785)
21dc6c134b2 is described below
commit 21dc6c134b2e2e64ab62e5e4996587af146aac44
Author: neatHyperTxt-meesho <[email protected]>
AuthorDate: Mon Sep 21 05:47:08 2026 +0530
HIVE-30053: Support multi-threaded CachedStore prewarm (#6785)
---
.../hadoop/hive/metastore/conf/MetastoreConf.java | 6 +
.../hadoop/hive/metastore/cache/CachedStore.java | 220 ++--------
.../metastore/cache/DefaultMetaCachePreWarm.java | 455 +++++++++++++++++++++
.../hive/metastore/cache/MetaCachePreWarm.java | 56 +++
.../hadoop/hive/metastore/cache/SharedCache.java | 33 +-
.../hive/metastore/cache/TestCachedStore.java | 38 ++
6 files changed, 605 insertions(+), 203 deletions(-)
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 96f1db1db2b..83542c02c4e 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 1e091b7d671..dae29e47f08 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<String> catalogsToCache;
- try {
- catalogsToCache = catalogsToCache(rawStore);
- LOG.info("Going to cache catalogs: " +
org.apache.commons.lang3.StringUtils.join(catalogsToCache, ", "));
- List<Catalog> 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<Database> databases = new ArrayList<>();
- for (String catName : catalogsToCache) {
- try {
- List<String> 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<String> 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<String> colNames =
MetaStoreUtils.getColumnNamesForTable(table);
- try {
- ColumnStatistics tableColStats = null;
- List<Partition> partitions = null;
- List<ColumnStatistics> 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<String> 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<FieldSchema> partKeys = table.getPartitionKeys();
- String defaultPartitionValue =
- MetastoreConf.getVar(rawStore.getConf(),
ConfVars.DEFAULTPARTITIONNAME);
- List<String> partCols = new ArrayList<>();
- List<String> 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<String> tableNames = new Stack<>();
-
- private synchronized void addTableNamesForPrewarming(List<String>
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<String> catalogsToCache(RawStore rs) throws
MetaException {
+ static Collection<String> catalogsToCache(RawStore rs) {
Collection<String> 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 00000000000..f41b5251c9f
--- /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<RawStore> 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<String> 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<Database> 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<String> preWarmCatalogs() throws MetaException,
NoSuchObjectException {
+ Collection<String> catalogsToCache = CachedStore.catalogsToCache(rawStore);
+ LOG.info("Going to cache catalogs: {}",
org.apache.commons.lang3.StringUtils.join(catalogsToCache, ", "));
+ List<Catalog> 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<Database> listDatabases(Collection<String> catalogsToCache) {
+ List<Database> databases = new ArrayList<>();
+ for (String catName : catalogsToCache) {
+ try {
+ List<String> 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<Database> databases) {
+ int numberOfDatabasesCachedSoFar = 0;
+ for (Database db : databases) {
+ String catName = StringUtils.normalizeIdentifier(db.getCatalogName());
+ String dbName = StringUtils.normalizeIdentifier(db.getName());
+ List<String> 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<Future<?>> 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<Future<?>> 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<String> colNames = MetaStoreUtils.getColumnNamesForTable(table);
+ try {
+ ColumnStatistics tableColStats = null;
+ List<Partition> partitions = null;
+ List<ColumnStatistics> 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<String> 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<FieldSchema> partKeys = table.getPartitionKeys();
+ String defaultPartitionValue =
+ MetastoreConf.getVar(rawStore.getConf(),
ConfVars.DEFAULTPARTITIONNAME);
+ List<String> partCols = new ArrayList<>();
+ List<String> 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<String> tableNames = new Stack<>();
+
+ synchronized void addTableNamesForPrewarming(List<String> 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 00000000000..a6753ca7781
--- /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 97e4a4f375d..2aa2428317c 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<Class<?>, 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<Class<?>, ObjectEstimator> sizeEstimators = null;
+ private final Object sizeEstimatorsLock = new Object();
private Set<String> tableToUpdateSize = new ConcurrentHashSet<>();
private ScheduledExecutorService executor = null;
private Map<String, Integer> 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<Class<?>, 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<Class<?>, ObjectEstimator> updated = new HashMap<>(estimators);
+ IncrementalObjectSizeEstimator.createEstimators(clazz,
(HashMap<Class<?>, 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<Class<?>, 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<Class<?>, 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 bc051f5b7d9..56d14e76101 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<String> 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<Partition> db1Ptbl1Partitions =
+ cachedStore.getPartitions(DEFAULT_CATALOG_NAME, db1.getName(),
db1Ptbl1.getTableName(),
+ GetPartitionsArgs.getAllPartitions());
+ Assert.assertEquals(25, db1Ptbl1Partitions.size());
+ List<Partition> 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);
