Fix issue that cause false alarm corruption report (facebook#12626)

Summary:
The state of `saved_seq_for_penul_check_` is not correctly maintained with the current flow. It's supposed to store the original sequence number for a `kTypeValuePreferredSeqno` entry for use in the `DecideOutputLevel` function. However, it's not always properly cleared.

Pull Request resolved: facebook#12626

Test Plan:
Added unit test that would fail before the fix
./tiered_compaction_test --gtest_filter="*InterleavedTimedPutAndPut*"

Reviewed By: pdillinger

Differential Revision: D57123469

Pulled By: jowlyzhang

fbshipit-source-id: 8d73214b3b6dc152daf19b6bd6ee9063581dc277

db/compaction/compaction_iterator.cc

@@ -1280,23 +1280,28 @@ void CompactionIterator::DecideOutputLevel() {
   }
 #endif  // NDEBUG
 
+  // saved_seq_for_penul_check_ is populated in `NextFromInput` when the
+  // entry's sequence number is non zero and validity context for output this
+  // entry is kSwapPreferredSeqno for use in `DecideOutputLevel`. It should be
+  // cleared out here unconditionally. Otherwise, it may end up getting consumed
+  // incorrectly by a different entry.
+  SequenceNumber seq_for_range_check =
+      (saved_seq_for_penul_check_.has_value() &&
+       saved_seq_for_penul_check_.value() != kMaxSequenceNumber)
+          ? saved_seq_for_penul_check_.value()
+          : ikey_.sequence;
+  saved_seq_for_penul_check_ = std::nullopt;
+  ParsedInternalKey ikey_for_range_check = ikey_;
+  if (seq_for_range_check != ikey_.sequence) {
+    ikey_for_range_check.sequence = seq_for_range_check;
+  }
   if (output_to_penultimate_level_) {
     // If it's decided to output to the penultimate level, but unsafe to do so,
     // still output to the last level. For example, moving the data from a lower
     // level to a higher level outside of the higher-level input key range is
     // considered unsafe, because the key may conflict with higher-level SSTs
     // not from this compaction.
     // TODO: add statistic for declined output_to_penultimate_level
-    SequenceNumber seq_for_range_check =
-        (saved_seq_for_penul_check_.has_value() &&
-         saved_seq_for_penul_check_.value() != kMaxSequenceNumber)
-            ? saved_seq_for_penul_check_.value()
-            : ikey_.sequence;
-    ParsedInternalKey ikey_for_range_check = ikey_;
-    if (seq_for_range_check != ikey_.sequence) {
-      ikey_for_range_check.sequence = seq_for_range_check;
-      saved_seq_for_penul_check_ = std::nullopt;
-    }
     bool safe_to_penultimate_level =
         compaction_->WithinPenultimateLevelOutputRange(ikey_for_range_check);
     if (!safe_to_penultimate_level) {
@@ -1310,7 +1315,7 @@ void CompactionIterator::DecideOutputLevel() {
       // snapshot is released before enabling `last_level_temperature` feature
       // We will migrate the feature to `last_level_temperature` and maybe make
       // it not dynamically changeable.
-      if (ikey_.sequence > earliest_snapshot_) {
+      if (seq_for_range_check > earliest_snapshot_) {
         status_ = Status::Corruption(
             "Unsafe to store Seq later than snapshot in the last level if "
             "per_key_placement is enabled");

db/compaction/tiered_compaction_test.cc

@@ -1651,6 +1651,31 @@ TEST_P(TimedPutPrecludeLastLevelTest, FastTrackTimedPutToLastLevel) {
   Close();
 }
 
+TEST_P(TimedPutPrecludeLastLevelTest, InterleavedTimedPutAndPut) {
+  Options options = CurrentOptions();
+  options.compaction_style = kCompactionStyleUniversal;
+  options.disable_auto_compactions = true;
+  options.preclude_last_level_data_seconds = 1 * 24 * 60 * 60;
+  options.env = mock_env_.get();
+  options.num_levels = 7;
+  options.last_level_temperature = Temperature::kCold;
+  options.default_write_temperature = Temperature::kHot;
+  DestroyAndReopen(options);
+  WriteOptions wo;
+  wo.protection_bytes_per_key = GetParam();
+
+  // Start time: kMockStartTime = 10000000;
+  ASSERT_OK(TimedPut(0, Key(0), "v0", kMockStartTime - 1 * 24 * 60 * 60, wo));
+  ASSERT_OK(Put(Key(1), "v1", wo));
+  ASSERT_OK(Flush());
+
+  ASSERT_OK(db_->CompactRange(CompactRangeOptions(), nullptr, nullptr));
+  ASSERT_EQ("0,0,0,0,0,1,1", FilesPerLevel());
+  ASSERT_GT(GetSstSizeHelper(Temperature::kHot), 0);
+  ASSERT_GT(GetSstSizeHelper(Temperature::kCold), 0);
+  Close();
+}
+
 TEST_P(TimedPutPrecludeLastLevelTest, PreserveTimedPutOnPenultimateLevel) {
   Options options = CurrentOptions();
   options.compaction_style = kCompactionStyleUniversal;