private List<IndexEntry> getValidIndexEntries(IndexSegment indexSegment) throws StoreException {
      // Validity of a IndexValue is determined by itself and it's latest state. For example, if the current IndexValue
      // is a Put, and the latest IndexValue for this blob is a Delete, and the operation is done out of retention duration,
      // then this Put IndexValue is considered as invalid.
      //
      // There is one exception, TtlUpdate IndexValue. A TtlUpdate IndexValue's validity is not only depends on the latest
      // state of the blob, it's also affected by the Put IndexValue. If the latest IndexValue for this blob is a Delete,
      // and it's operation time is out of the retention duration, this TtlUpdate record would be considered as invalid.
      // However if the Put IndexValue exists for this TtlUpdate and is not under compaction, then we will keep this TtlUpdate
      // anyway.
      //
      // This is a table that shows how the validity of the current IndexValue is determined.
      // ----------------------------------------------------------------------------------------
      // Current IndexValue  | Latest IndexValue | Is Valid                                     |
      // --------------------+-------------------+----------------------------------------------|
      // Put(version c)      | Put(version f)    | isExpired(Pc)?false:true                     |
      //                     | Delete(f)         | reachRetention(Df)||isExpired(Df)?false:true |
      //                     | Undelete(f)       | isExpired(Uf)?false:true                     |
      // --------------------+-------------------+----------------------------------------------|
      // TtlUpdate(c)        | Put(f)            | Exception                                    |
      //                     | Delete(f)         | reachRetention(Df)?false:true                |
      //                     | Undelete(f)       | true                                         |
      // --------------------+-------------------+----------------------------------------------|
      // Delete(c)           | Put(f)            | Exception                                    |
      //                     | Delete(f)         | c==f && !isRemovable(Df) ? true : false      |
      //                     | Undelete(f)       | false                                        |
      // --------------------+-------------------+----------------------------------------------|
      // Undelete(c)         | Put(f)            | Exception                                    |
      //                     | Delete(f)         | false                                        |
      //                     | Undelete(f)       | c==f&&!isExpired(Uf)?true:false              |
      // ----------------------------------------------------------------------------------------
      List<IndexEntry> validEntries = new ArrayList<>();
      StoreKey previousKey = null;
      IndexValue previousLatestState = null;
      for (IndexEntry entry : indexSegment) {
        StoreKey currentKey = entry.getKey();
        IndexValue currentValue = entry.getValue();
        IndexValue currentLatestState;

        if (previousKey == null) {
          // save a token for restart (the key gets ignored but is required to be non null for construction)
          StoreFindToken safeToken =
              new StoreFindToken(currentKey, indexSegment.getStartOffset(), sessionId, incarnationId, null, null,
                  UNINITIALIZED_RESET_KEY_VERSION);
          compactionLog.setSafeToken(safeToken);
          logger.debug("Set safe token for compaction in {} to {}", storeId, safeToken);
        }
        // If an IndexSegment contains more than one IndexValue for the same StoreKey, then they must follow each other
        // since IndexSegment stores IndexValues based on StoreKey. If the current key equals to the previous key, then
        // we don't have to query the latest state again.
        if (currentKey.equals(previousKey)) {
          currentLatestState = previousLatestState;
        } else {
          previousKey = currentKey;
          previousLatestState = currentLatestState = srcIndex.findKey(currentKey);
        }

        if (currentValue.isUndelete()) {
          if (currentLatestState.isPut()) {
            throw new IllegalStateException(
                "Undelete's latest state can't be put for key" + currentKey + " in store " + dataDir);
          }
          if (currentLatestState.isUndelete() && currentLatestState.getLifeVersion() == currentValue.getLifeVersion()
              && !srcIndex.isExpired(currentLatestState)) {
            validEntries.add(entry);
          }
        } else if (currentValue.isDelete()) {
          if (currentLatestState.isPut()) {
            throw new IllegalStateException(
                "Delete's latest state can't be put for key" + currentKey + " in store " + dataDir);
          }
          if (currentLatestState.isDelete() && currentLatestState.getLifeVersion() == currentValue.getLifeVersion()) {
            // check if this is a removable delete tombstone.
            if (config.storeCompactionPurgeDeleteTombstone && isDeleteTombstoneRemovable(entry, indexSegment)
                && getPutValueFromSrc(currentKey, currentValue, indexSegment) == null) {
              logger.debug(
                  "Delete tombstone of {} (with expiration time {} ms) is removable and won't be copied to target log segment",
                  currentKey, currentValue.getExpiresAtMs());
            } else {
              validEntries.add(entry);
            }
          }
        } else if (currentValue.isTtlUpdate()) {
          if (currentLatestState == null) {
            // Under only one circumstance would this happen, an TTL_UPDATE index value has no PUT before and no DELETE after.
            // This is because the PUT is already compacted due to DELETE is out of retention and DELETE is compacted due to
            // DELETE tombstone clean up. But TTL_UPDATE is in different log segment from PUT and DELETE so it never gets compacted.
            continue;
          }
          if (currentLatestState.isPut()) {
            // If isPut returns true, when the latest state doesn't carry ttl_update flag, this is wrong
            throw new IllegalStateException(
                "TtlUpdate's latest state can't be put for key" + currentKey + " in store " + dataDir);
          }
          // This is a TTL_UPDATE record, then the blob can't be expired. Only check if it's deleted or not.
          if (currentLatestState.isDelete()) {
            if (currentLatestState.getOperationTimeInMs() >= compactionLog.getCompactionDetails()
                .getReferenceTimeMs()) {
              validEntries.add(entry);
            } else if (isTtlUpdateEntryValidWhenLatestStateIsDeleteAndRetention(currentKey,
                indexSegment.getStartOffset())) {
              validEntries.add(entry);
            }
          } else {
            validEntries.add(entry);
          }
        } else {
          if (srcIndex.isExpired(currentLatestState)) {
            logger.trace("{} in index segment with start offset {} in {} is not valid because it is an expired PUT",
                entry, indexSegment.getStartOffset(), storeId);
            continue;
          }
          if (currentLatestState.isPut()) {
            if (currentLatestState.getLifeVersion() != currentValue.getLifeVersion()) {
              throw new IllegalStateException(
                  "Two different lifeVersions  for puts key" + currentKey + " in store " + dataDir);
            }
            validEntries.add(entry);
          } else if (currentLatestState.isDelete()) {
            if (currentLatestState.getOperationTimeInMs() >= compactionLog.getCompactionDetails()
                .getReferenceTimeMs()) {
              validEntries.add(entry);
            } else {
              logger.trace("{} in index segment with start offset {} in {} is not valid because it is a deleted PUT",
                  entry, indexSegment.getStartOffset(), storeId);
            }
          } else {
            validEntries.add(entry);
          }
        }
      }
      return validEntries;
    }
Gets all the valid index entries in the given list of index entries.
@param indexSegment the {@link IndexSegment} that index entries are from.
@return the list of valid entries generated from {@code allIndexEntries}. May contain entries not in
{@code allIndexEntries}.
@throws StoreException if {@link BlobReadOptions} could not be obtained from the store for deleted blobs.
ambry-store/src/main/java/com/github/ambry/store/BlobStoreCompactor.java