@VisibleForTesting
  /**
   * This initializes the caches in SharedCache by getting the objects from Metastore DB via
   * ObjectStore and populating the respective caches
   */
  static void prewarm(RawStore rawStore) {
    if (isCachePrewarmed.get()) {
      return;
    }
    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, -1);
                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());
      }
      sharedCache.clearDirtyFlags();
      completePrewarm(startTime, true);
    }
  }
@Test public void testPrewarm() 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");
    MetaStoreTestUtils.setConfForStandloneMode(conf);
    CachedStore cachedStore = new CachedStore();
    CachedStore.clearSharedCache();
    cachedStore.setConfForTest(conf);
    ObjectStore objectStore = (ObjectStore) cachedStore.getRawStore();
    // Prewarm CachedStore
    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()));
    List<String> db1Tables = cachedStore.getAllTables(DEFAULT_CATALOG_NAME, db1.getName());
    Assert.assertEquals(2, db1Tables.size());
    Assert.assertTrue(db1Tables.contains(db1Utbl1.getTableName()));
    Assert.assertTrue(db1Tables.contains(db1Ptbl1.getTableName()));
    List<String> db2Tables = cachedStore.getAllTables(DEFAULT_CATALOG_NAME, db2.getName());
    Assert.assertEquals(2, db2Tables.size());
    Assert.assertTrue(db2Tables.contains(db2Utbl1.getTableName()));
    Assert.assertTrue(db2Tables.contains(db2Ptbl1.getTableName()));
    // cs_db1_ptntbl1
    List<Partition> db1Ptbl1Partitions =
        cachedStore.getPartitions(DEFAULT_CATALOG_NAME, db1.getName(), db1Ptbl1.getTableName(), -1);
    Assert.assertEquals(25, db1Ptbl1Partitions.size());
    Deadline.startTimer("");
    List<Partition> db1Ptbl1PartitionsOS =
        objectStore.getPartitions(DEFAULT_CATALOG_NAME, db2.getName(), db1Ptbl1.getTableName(), -1);
    Assert.assertTrue(db1Ptbl1Partitions.containsAll(db1Ptbl1PartitionsOS));
    // cs_db2_ptntbl1
    List<Partition> db2Ptbl1Partitions =
        cachedStore.getPartitions(DEFAULT_CATALOG_NAME, db2.getName(), db2Ptbl1.getTableName(), -1);
    Assert.assertEquals(25, db2Ptbl1Partitions.size());
    Deadline.startTimer("");
    List<Partition> db2Ptbl1PartitionsOS =
        objectStore.getPartitions(DEFAULT_CATALOG_NAME, db2.getName(), db2Ptbl1.getTableName(), -1);
    Assert.assertTrue(db2Ptbl1Partitions.containsAll(db2Ptbl1PartitionsOS));
    cachedStore.shutdown();
  }