neatHyperTxt-meesho commented on code in PR #6785:
URL: https://github.com/apache/hive/pull/6785#discussion_r4039427410
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/cache/CachedStore.java:
##########
@@ -492,129 +517,242 @@ static void prewarm(RawStore rawStore) {
}
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;
+ int prewarmThreads = Math.max(1,
+ MetastoreConf.getIntVar(rawStore.getConf(),
ConfVars.CACHED_RAW_STORE_PREWARM_THREADS));
+ ExecutorService prewarmPool = null;
+ List<RawStore> workerStores = new ArrayList<>();
+ if (prewarmThreads > 1) {
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;
+ // 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(createRawStoreForPrewarm(rawStore.getConf()));
+ }
+ 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);
+ shutdownPrewarmWorkers(null, workerStores);
+ workerStores = new ArrayList<>();
}
- tblsPendingPrewarm.addTableNamesForPrewarming(tblNames);
- int totalTablesToCache = tblNames.size();
- int numberOfTablesCachedSoFar = 0;
- while (tblsPendingPrewarm.hasMoreTablesToPrewarm()) {
+ }
+ try {
+ int numberOfDatabasesCachedSoFar = 0;
+ for (Database db : databases) {
+ String catName =
StringUtils.normalizeIdentifier(db.getCatalogName());
+ String dbName = StringUtils.normalizeIdentifier(db.getName());
+ List<String> tblNames;
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;
+ 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 cacheMemoryFull = 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, cacheMemoryFull, tablesCachedSoFar,
+ totalTablesToCache)));
}
- 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);
+ 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);
completePrewarm(startTime, false);
Review Comment:
Fixed. completePrewarm is no longer called inside the try block — the
interrupt and memory-full paths now just set the shared stop flag and break, so
completion is published only after the finally block has terminated the
workers. shutdownPrewarmWorkers now awaits termination (bounded) before closing
the worker RawStores, and the drain loop also checks Thread.isInterrupted().
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
