private NavigableMap<LogSegmentName, Long> collectValidDataSizeByLogSegment(long deleteReferenceTimeInMs,
      long expiryReferenceTimeInMs, FileSpan fileSpanUnderCompaction) throws StoreException {
    logger.trace("On demand index scanning to collect compaction data stats for store {} wrt ref time {}", storeId,
        deleteReferenceTimeInMs);
    long startTimeMs = time.milliseconds();
    Map<StoreKey, IndexFinalState> keyFinalStates = new HashMap<>();
    NavigableMap<LogSegmentName, Long> validSizePerLogSegment = new TreeMap<>();
    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();
      LogSegmentName logSegmentName = indexSegment.getLogSegmentName();
      diskIOScheduler.getSlice(BlobStoreStats.IO_SCHEDULER_JOB_TYPE, BlobStoreStats.IO_SCHEDULER_JOB_ID,
          indexSegment.size());
      forEachValidIndexEntry(indexSegment, deleteReferenceTimeInMs, expiryReferenceTimeInMs, fileSpanUnderCompaction,
          keyFinalStates, true,
          entry -> updateMapHelper(validSizePerLogSegment, logSegmentName, entry.getValue().getSize()));
      metrics.statsOnDemandScanTimePerIndexSegmentMs.update(time.milliseconds() - indexSegmentStartProcessTimeMs,
          TimeUnit.MILLISECONDS);
      indexSegmentCount++;
      if (indexSegmentCount == 1 || indexSegmentCount % 10 == 0) {
        logger.debug("Compaction Stats: Index segment {} processing complete (on-demand scanning) for store {}",
            indexSegment.getFile().getName(), storeId);
      }
    }
    // 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);
    if (validSizePerLogSegment.isEmpty()) {
      validSizePerLogSegment.put(index.getStartOffset().getName(), 0L);
    }
    if (enablePurgeDeleteTombstone) {
      removeDeleteTombStonesFromValidSize(keyFinalStates.values(), validSizePerLogSegment, expiryReferenceTimeInMs);
    }
    return validSizePerLogSegment;
  }
Walk through the entire index and collect valid size information per log segment (size of delete records included).
@param deleteReferenceTimeInMs the reference time in ms until which deletes are relevant
@param expiryReferenceTimeInMs the reference tie in ms until which expiration is relevant
@return a {@link NavigableMap} of log segment name to valid data size
ambry-store/src/main/java/com/github/ambry/store/BlobStoreStats.java