Pair<Long, NavigableMap<LogSegmentName, Long>> getValidDataSizeByLogSegment(TimeRange timeRange,
      long expiryReferenceTime, FileSpan fileSpanUnderCompaction) throws StoreException {
    if (!enabled.get()) {
      throw new StoreException(String.format("BlobStoreStats is not enabled or closing for store %s", storeId),
          StoreErrorCodes.StoreShuttingDown);
    }
    logger.info("Getting valid data size by log segment at time range {} for {}", timeRange, storeId);
    Pair<Long, NavigableMap<LogSegmentName, Long>> retValue = null;
    ScanResults currentScanResults = scanResults.get();
    long referenceTimeInMs = getLogSegmentDeleteRefTimeMs(currentScanResults, timeRange);
    if (enableBucketForLogSegmentReports) {
      if (referenceTimeInMs != REF_TIME_OUT_OF_BOUNDS) {
        retValue = currentScanResults.getValidSizePerLogSegment(referenceTimeInMs, expiryReferenceTime);
      } else {
        if (isScanning
            && getLogSegmentDeleteRefTimeMs(indexScanner.newScanResults, timeRange) != REF_TIME_OUT_OF_BOUNDS) {
          scanLock.lock();
          try {
            if (isScanning) {
              if (waitCondition.await(waitTimeoutInSecs, TimeUnit.SECONDS)) {
                currentScanResults = scanResults.get();
                referenceTimeInMs = getLogSegmentDeleteRefTimeMs(currentScanResults, timeRange);
                if (referenceTimeInMs != REF_TIME_OUT_OF_BOUNDS) {
                  retValue = currentScanResults.getValidSizePerLogSegment(referenceTimeInMs, expiryReferenceTime);
                }
              } else {
                metrics.blobStoreStatsIndexScannerErrorCount.inc();
                logger.error("Timed out while waiting for BlobStoreStats index scan to complete for store {}", storeId);
              }
            } else {
              currentScanResults = scanResults.get();
              referenceTimeInMs = getLogSegmentDeleteRefTimeMs(currentScanResults, timeRange);
              if (referenceTimeInMs != REF_TIME_OUT_OF_BOUNDS) {
                retValue = currentScanResults.getValidSizePerLogSegment(referenceTimeInMs, expiryReferenceTime);
              }
            }
          } catch (InterruptedException e) {
            metrics.blobStoreStatsIndexScannerErrorCount.inc();
            throw new IllegalStateException(
                String.format("Illegal state, wait for scan to complete is interrupted for store %s", storeId), e);
          } finally {
            scanLock.unlock();
          }
        }
      }
    }

    if (retValue == null) {
      // retValue could be null in three scenarios:
      // 1. timeRange is outside of current forecast coverage and there is no ongoing scan.
      // 2. timed out while waiting for an ongoing scan.
      // 3. rare edge case where currentScanResults updated twice since the start of the wait.
      referenceTimeInMs = timeRange.getEndTimeInMs();
      retValue = new Pair<>(referenceTimeInMs,
          collectValidDataSizeByLogSegment(referenceTimeInMs, expiryReferenceTime, fileSpanUnderCompaction));
    }
    // Before return the value, make sure all the log segments are in the final map
    for (LogSegment segment : index.getLogSegments()) {
      LogSegmentName logSegmentName = segment.getName();
      retValue.getSecond().putIfAbsent(logSegmentName, 0L);
    }
    return retValue;
  }
Same as {@link #getValidDataSizeByLogSegment(TimeRange, long)}, but provides a file span for under compaction log segments.
This file span would impact TTL_UPDATE's validity.
@param timeRange the delete reference {@link TimeRange} at which the data is requested. Defines both the reference time
                 and the acceptable resolution.
@param expiryReferenceTime the reference time for expired blobs. Blobs with expiration time less than it would be
                           considered as expired. Usually it's now.
@param fileSpanUnderCompaction the {@link FileSpan} of the under compaction log segments. This file span would impact
                               the validity of TTL_UPDATE. The rules can be found below. This file span could be null.
@return a {@link Pair} whose first element is the time at which stats was collected (in ms) and whose second
element is the valid data size for each segment in the form of a {@link NavigableMap} of segment names to
valid data sizes.
@throws StoreException
ambry-store/src/main/java/com/github/ambry/store/BlobStoreStats.java