@Override
  public void storeHostPartitionClassStorageStats(HostPartitionClassStorageStatsWrapper statsWrapper)
      throws SQLException {
    long startTimeMs = System.currentTimeMillis();
    Map<String, Map<Long, Map<Short, Map<Short, ContainerStorageStats>>>> partitionClassStorageStats =
        statsWrapper.getStats().getStorageStats();
    // 1. Get all the partition class names in this statswrapper and the partition class names in DB
    Set<String> partitionClassNames = new HashSet<>(partitionClassStorageStats.keySet());
    Map<String, Short> partitionClassNamesInDB = partitionClassReportsDao.queryPartitionClassNames(clusterName);

    // 2. Add partition class names not in DB
    partitionClassNames.removeAll(partitionClassNamesInDB.keySet());
    if (!partitionClassNames.isEmpty()) {
      for (String partitionClassName : partitionClassNames) {
        partitionClassReportsDao.insertPartitionClassName(clusterName, partitionClassName);
      }
      // Refresh the partition class names
      partitionClassNamesInDB = partitionClassReportsDao.queryPartitionClassNames(clusterName);
    }

    // 3. Get partition ids under each partition class name
    // The result looks like
    //  {
    //    "default": [1, 2, 3],
    //    "new": [4, 5, 6]
    //  }
    // The "default" and "new" are the partition class names and the numbers are partition ids. Same partition
    // can't belong to different partition class names.
    Map<String, List<Long>> partitionIdsUnderClassNames = partitionClassStorageStats.entrySet()
        .stream()
        .collect(Collectors.toMap(Map.Entry::getKey, ent -> new ArrayList<>(ent.getValue().keySet())));

    // 4. Add partition ids not in DB
    Set<Integer> partitionIdsInDB = partitionClassReportsDao.queryPartitionIds(clusterName);
    for (String partitionClassName : partitionIdsUnderClassNames.keySet()) {
      short partitionClassId = partitionClassNamesInDB.get(partitionClassName);
      for (long pid : partitionIdsUnderClassNames.get(partitionClassName)) {
        if (!partitionIdsInDB.contains(pid)) {
          partitionClassReportsDao.insertPartitionId(clusterName, (int) pid, partitionClassId);
        }
      }
    }
    storeMetrics.storePartitionClassStatsTimeMs.update(System.currentTimeMillis() - startTimeMs);
  }
Copy the row of table {@link AggregatedAccountReportsDao#AGGREGATED_ACCOUNT_REPORTS_TABLE} to {@link AggregatedAccountReportsDao#MONTHLY_AGGREGATED_ACCOUNT_REPORTS_TABLE}
and update the {@code monthValue} in table {@link AggregatedAccountReportsDao#AGGREGATED_ACCOUNT_REPORTS_MONTH_TABLE}.
@param monthValue The month value.
@throws Exception
ambry-mysql/src/main/java/com/github/ambry/accountstats/AccountStatsMySqlStore.java