@Override
  public HostPartitionClassStorageStatsWrapper queryHostPartitionClassStorageStatsByHost(String hostname, int port,
      Map<String, Set<Integer>> partitionNameAndIds) throws SQLException {
    long startTimeMs = System.currentTimeMillis();
    hostname = hostnameHelper.simplifyHostname(hostname, port);
    Map<Integer, Map<Short, Map<Short, ContainerStorageStats>>> partitionAccountContainerUsage = new HashMap<>();
    AtomicLong timestamp = new AtomicLong(0);
    accountReportsDao.queryStorageUsageForHost(clusterName, hostname,
        (partitionId, accountId, containerStats, updatedAtMs) -> {
          partitionAccountContainerUsage.computeIfAbsent(partitionId, pid -> new HashMap<>())
              .computeIfAbsent(accountId, aid -> new HashMap<>())
              .put(containerStats.getContainerId(), containerStats);
          timestamp.set(Math.max(timestamp.get(), updatedAtMs));
        });
    // Here partitionAccountContainerUsage has partition id, account id and container id as keys of map at each level,
    // the value is the storage usage.

    // As indicated by the comments above, we have to know the partition class name for each partition id. Luckily, we
    // have all the partition ids and we have a map partitionNameAndIds whose key is the partition class name and the
    //value is the list of all partition ids belong to the partition class name.
    Set<Integer> partitionIds = partitionAccountContainerUsage.keySet();
    Map<String, Set<Integer>> partitionNameAndIdsForHost = new HashMap<>();
    for (int partitionId : partitionIds) {
      boolean found = false;
      for (Map.Entry<String, Set<Integer>> namesAndIdsEntry : partitionNameAndIds.entrySet()) {
        if (namesAndIdsEntry.getValue().contains(partitionId)) {
          partitionNameAndIdsForHost.computeIfAbsent(namesAndIdsEntry.getKey(), k -> new HashSet<>()).add(partitionId);
          found = true;
          break;
        }
      }
      if (!found) {
        storeMetrics.missingPartitionClassNameErrorCount.inc();
        logger.error("Can't find partition class name for partition id {}", partitionId);
      }
    }
    HostPartitionClassStorageStats hostPartitionClassStorageStats = new HostPartitionClassStorageStats();
    for (Map.Entry<String, Set<Integer>> nameAndIdsEntry : partitionNameAndIdsForHost.entrySet()) {
      String partitionClassName = nameAndIdsEntry.getKey();
      for (int partitionId : nameAndIdsEntry.getValue()) {
        Map<Short, Map<Short, ContainerStorageStats>> accountContainerUsage =
            partitionAccountContainerUsage.get(partitionId);
        for (short accountId : accountContainerUsage.keySet()) {
          Map<Short, ContainerStorageStats> containerUsage = accountContainerUsage.get(accountId);
          containerUsage.values()
              .stream()
              .forEach(containerStats -> hostPartitionClassStorageStats.addContainerStorageStats(partitionClassName,
                  partitionId, accountId, containerStats));
        }
      }
    }

    storeMetrics.queryPartitionClassStatsTimeMs.update(System.currentTimeMillis() - startTimeMs);
    return new HostPartitionClassStorageStatsWrapper(
        new StatsHeader(StatsHeader.StatsDescription.STORED_DATA_SIZE, timestamp.get(),
            partitionAccountContainerUsage.size(), partitionAccountContainerUsage.size(), null),
        hostPartitionClassStorageStats);
  }
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