private Map<Short, Map<Short, ContainerStorageStats>> collectContainerStorageStats(long referenceTimeInMs)
      throws StoreException {
    logger.trace("On demand index scanning to collect container valid data sizes for store {} wrt ref time {}", storeId,
        referenceTimeInMs);
    long startTimeMs = time.milliseconds();
    Map<StoreKey, IndexFinalState> keyFinalStates = new HashMap<>();
    Map<Short, Map<Short, Long>> validDataSizePerContainer = new HashMap<>();
    Map<Short, Map<Short, Long>> physicalDataSizePerContainer = new HashMap<>();
    Map<Short, Map<Short, Long>> storeKeysPerContainer = new HashMap<>();
    Map<Short, Map<Short, ContainerStorageStats>> result = new HashMap<>();
    int indexSegmentCount = 0;
    for (IndexSegment indexSegment : index.getIndexSegments().descendingMap().values()) {
      if (!enabled.get()) {
        throw new StoreException(String.format("BlobStoreStats is not enabled or closing for store %s", storeId),
            StoreErrorCodes.StoreShuttingDown);
      }
      long indexSegmentStartProcessTimeMs = time.milliseconds();
      diskIOScheduler.getSlice(BlobStoreStats.IO_SCHEDULER_JOB_TYPE, BlobStoreStats.IO_SCHEDULER_JOB_ID,
          indexSegment.size());
      forEachIndexEntry(indexSegment, referenceTimeInMs, time.milliseconds(), null, keyFinalStates, true,
          (entry, isValid) -> {
            IndexValue indexValue = entry.getValue();
            if (isValid && indexValue.isPut()) {
              // delete and TTL update records does not count towards valid data size for usage (containers)
              updateNestedMapHelper(validDataSizePerContainer, indexValue.getAccountId(), indexValue.getContainerId(),
                  indexValue.getSize());
            }
            updateNestedMapHelper(physicalDataSizePerContainer, indexValue.getAccountId(), indexValue.getContainerId(),
                indexValue.getSize());
            updateNestedMapHelper(storeKeysPerContainer, indexValue.getAccountId(), indexValue.getContainerId(),
                (long) (indexValue.isPut() ? 1 : 0));
          });
      metrics.statsOnDemandScanTimePerIndexSegmentMs.update(time.milliseconds() - indexSegmentStartProcessTimeMs,
          TimeUnit.MILLISECONDS);
      indexSegmentCount++;
      if (indexSegmentCount == 1 || indexSegmentCount % 10 == 0) {
        logger.info("Container Stats: Index segment {} processing complete (on-demand scanning) for store {}",
            indexSegment.getFile().getName(), storeId);
      }
    }
    for (short accountId : validDataSizePerContainer.keySet()) {
      for (short containerId : validDataSizePerContainer.get(accountId).keySet()) {
        result.computeIfAbsent(accountId, k -> new HashMap<>())
            .put(containerId,
                new ContainerStorageStats(containerId, validDataSizePerContainer.get(accountId).get(containerId),
                    physicalDataSizePerContainer.get(accountId).get(containerId),
                    storeKeysPerContainer.get(accountId).get(containerId)));
      }
    }
    // The remaining index entries in keyFinalStates are DELETE tombstones left by compaction (whose associated PUT is not found)
    updateDeleteTombstoneStats(keyFinalStates.values());
    metrics.statsOnDemandScanTotalTimeMs.update(time.milliseconds() - startTimeMs, TimeUnit.MILLISECONDS);
    return result;
  }
Walk through the entire index and collect storage stats per container.
@param referenceTimeInMs the reference time in ms until which deletes and expiration are relevant
@return a nested {@link Map} of serviceId to containerId to {@link ContainerStorageStats}.
ambry-store/src/main/java/com/github/ambry/store/BlobStoreStats.java