Skip to content

HIVE-30053: Support multi-threaded CachedStore prewarm #6785

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
dengzhhu653 merged 7 commits into from
Sep 21, 2026
Merged

HIVE-30053: Support multi-threaded CachedStore prewarm #6785

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c2aca89
HIVE-30053: Support multi-threaded CachedStore prewarm
neatHyperTxt-meesho Sep 15, 2026
e7df672
HIVE-30053: Address Sonar findings (log format specifiers, line lengt…
neatHyperTxt-meesho Sep 15, 2026
d3f88cd
HIVE-30053: Publish prewarm completion only after workers terminate; …
neatHyperTxt-meesho Sep 17, 2026
23ee560
HIVE-30053: Extract prewarm into MetaCachePreWarm interface per review
neatHyperTxt-meesho Sep 18, 2026
2a73ed6
HIVE-30053: Preemptive cleanup in DefaultMetaCachePreWarm (log format…
neatHyperTxt-meesho Sep 18, 2026
11fb74e
HIVE-30053: Address Sonar findings (cognitive complexity, collection …
neatHyperTxt-meesho Sep 18, 2026
f7de929
HIVE-30053: Move TablesPendingPrewarm into the prewarmer; add priorit…
neatHyperTxt-meesho Sep 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions ...e/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/conf/MetastoreConf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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."),
Expand Down
220 changes: 24 additions & 196 deletions ...re/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;

@dengzhhu653 dengzhhu653 Sep 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you separate the prewarm from this class please, and create an interface to warm up the cache?
An interface for example:

interface MetaCachePreWarm extends AutoClosable, Configurable {
   void initialize() throws MetaException;
   void preWarm() throws MetaException;
....
}

@neatHyperTxt-meesho neatHyperTxt-meesho Sep 18, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done, prewarm now lives behind a MetaCachePreWarm interface @dengzhhu653

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);
}

/**
Expand All @@ -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);
}
Expand All @@ -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();
Expand Down Expand Up @@ -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);
Expand Down
Loading
Loading