From 02443dd93f32a67cb2b0c3545ebd2c5a1111b4fa Mon Sep 17 00:00:00 2001
From: Peter Dillinger <peterd@meta.com>
Date: Fri, 29 Sep 2023 11:21:59 -0700
Subject: [PATCH] Refactor, clean up, fixes, and more testing for
 SeqnoToTimeMapping (#11905)

Summary:
This change is before a planned DBImpl change to ensure all sufficiently recent sequence numbers since Open are covered by SeqnoToTimeMapping (bug fix with existing test work-arounds). **Intended follow-up**

However, I found enough issues with SeqnoToTimeMapping to warrant this PR first, including very small fixes in DB implementation related to API contract of SeqnoToTimeMapping.

Functional fixes / changes:
* This fixes some mishandling of boundary cases. For example, if the user decides to stop writing to DB, the last written sequence number would perpetually have its write time updated to "now" and would always be ineligible for migration to cold tier. Part of the problem is that the SeqnoToTimeMapping would return a seqno known to have been written before (immediately or otherwise) the requested time, but compaction_job.cc would include that seqno in the preserve/exclude set. That is fixed (in part) by adding one in compaction_job.cc
* That problem was worse because a whole range of seqnos could be updated perpetually with new times in SeqnoToTimeMapping::Append (if no writes to DB). That logic was apparently optimized for GetOldestApproximateTime (now GetProximalTimeBeforeSeqno), which is not used in production, to the detriment of GetOldestSequenceNum (now GetProximalSeqnoBeforeTime), which is used in production. (Perhaps plans changed during development?) This is fixed in Append to optimize for accuracy of GetProximalSeqnoBeforeTime. (Unit tests added and updated.)
* Related: SeqnoToTimeMapping did not have a clear contract about the relationships between seqnos and times, just the idea of a rough correspondence. Now the class description makes it clear that the write time of each recorded seqno comes before or at the associated time, to support getting best results for GetProximalSeqnoBeforeTime. And this makes it easier to make clear the contract of each API function.
  * Update `DBImpl::RecordSeqnoToTimeMapping()` to follow this ordering in gathering samples.

Some part of these changes has required an expanded test work-around for the problem (see intended follow-up above) that the DB does not immediately ensure recent seqnos are covered by its mapping. These work-arounds will be removed with that planned work.

An apparent compaction bug is revealed in
PrecludeLastLevelTest::RangeDelsCauseFileEndpointsToOverlap, so that test is disabled. Filed GitHub issue #11909

Cosmetic / code safety things (not exhaustive):
* Fix some confusing names.
  * `seqno_time_mapping` was used inconsistently in places. Now just `seqno_to_time_mapping` to correspond to class name.
  * Rename confusing `GetOldestSequenceNum` -> `GetProximalSeqnoBeforeTime` and `GetOldestApproximateTime` -> `GetProximalTimeBeforeSeqno`. Part of the motivation is that our times and seqnos here have the same underlying type, so we want to be clear about which is expected where to avoid mixing.
  * Rename `kUnknownSeqnoTime` to `kUnknownTimeBeforeAll` because the value is a bad choice for unknown if we ever add ProximalAfterBlah functions.
  * Arithmetic on SeqnoTimePair doesn't make sense except for delta encoding, so use better names / APIs with that in mind.
  * (OMG) Don't allow direct comparison between SeqnoTimePair and SequenceNumber. (There is no checking that it isn't compared against time by accident.)
  * A field name essentially matching the containing class name is a confusing pattern (`seqno_time_mapping_`).
  * Wrap calls to confusing (but useful) upper_bound and lower_bound functions to have clearer names and more code reuse.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/11905

Test Plan: GetOldestSequenceNum (now GetProximalSeqnoBeforeTime) and TruncateOldEntries were lacking unit tests, despite both being used in production (experimental feature). Added those and expanded others.

Reviewed By: jowlyzhang

Differential Revision: D49755592

Pulled By: pdillinger

fbshipit-source-id: f72a3baac74d24b963c77e538bba89a7fc8dce51
---
 db/builder.cc                           |   6 +-
 db/compaction/compaction_job.cc         |  26 ++-
 db/compaction/compaction_job.h          |   2 +-
 db/compaction/compaction_outputs.cc     |  14 +-
 db/compaction/compaction_outputs.h      |   2 +-
 db/compaction/tiered_compaction_test.cc |  44 ++++-
 db/db_impl/db_impl.cc                   |  10 +-
 db/db_impl/db_impl.h                    |   4 +-
 db/db_impl/db_impl_compaction_flush.cc  |   4 +-
 db/db_impl/db_impl_debug.cc             |   2 +-
 db/db_impl/db_impl_open.cc              |   4 +-
 db/flush_job.cc                         |  13 +-
 db/flush_job.h                          |   6 +-
 db/repair.cc                            |   7 +-
 db/seqno_time_test.cc                   | 235 ++++++++++++++++++++----
 db/seqno_to_time_mapping.cc             | 153 ++++++++-------
 db/seqno_to_time_mapping.h              | 118 +++++++-----
 17 files changed, 449 insertions(+), 201 deletions(-)

diff --git a/db/builder.cc b/db/builder.cc
index a3a6bc47e6..d3040ee9e2 100644
--- a/db/builder.cc
+++ b/db/builder.cc
@@ -294,12 +294,12 @@ Status BuildTable(
     if (!s.ok() || empty) {
       builder->Abandon();
     } else {
-      std::string seqno_time_mapping_str;
+      std::string seqno_to_time_mapping_str;
       seqno_to_time_mapping.Encode(
-          seqno_time_mapping_str, meta->fd.smallest_seqno,
+          seqno_to_time_mapping_str, meta->fd.smallest_seqno,
           meta->fd.largest_seqno, meta->file_creation_time);
       builder->SetSeqnoTimeTableProperties(
-          seqno_time_mapping_str,
+          seqno_to_time_mapping_str,
           ioptions.compaction_style == CompactionStyle::kCompactionStyleFIFO
               ? meta->file_creation_time
               : meta->oldest_ancester_time);
diff --git a/db/compaction/compaction_job.cc b/db/compaction/compaction_job.cc
index a5de19a39b..bf8ce25a67 100644
--- a/db/compaction/compaction_job.cc
+++ b/db/compaction/compaction_job.cc
@@ -288,23 +288,23 @@ void CompactionJob::Prepare() {
 
   if (preserve_time_duration > 0) {
     const ReadOptions read_options(Env::IOActivity::kCompaction);
-    // setup seqno_time_mapping_
-    seqno_time_mapping_.SetMaxTimeDuration(preserve_time_duration);
+    // setup seqno_to_time_mapping_
+    seqno_to_time_mapping_.SetMaxTimeDuration(preserve_time_duration);
     for (const auto& each_level : *c->inputs()) {
       for (const auto& fmd : each_level.files) {
         std::shared_ptr<const TableProperties> tp;
         Status s =
             cfd->current()->GetTableProperties(read_options, &tp, fmd, nullptr);
         if (s.ok()) {
-          seqno_time_mapping_.Add(tp->seqno_to_time_mapping)
+          seqno_to_time_mapping_.Add(tp->seqno_to_time_mapping)
               .PermitUncheckedError();
-          seqno_time_mapping_.Add(fmd->fd.smallest_seqno,
-                                  fmd->oldest_ancester_time);
+          seqno_to_time_mapping_.Add(fmd->fd.smallest_seqno,
+                                     fmd->oldest_ancester_time);
         }
       }
     }
 
-    auto status = seqno_time_mapping_.Sort();
+    auto status = seqno_to_time_mapping_.Sort();
     if (!status.ok()) {
       ROCKS_LOG_WARN(db_options_.info_log,
                      "Invalid sequence number to time mapping: Status: %s",
@@ -320,13 +320,17 @@ void CompactionJob::Prepare() {
       preserve_time_min_seqno_ = 0;
       preclude_last_level_min_seqno_ = 0;
     } else {
-      seqno_time_mapping_.TruncateOldEntries(_current_time);
+      seqno_to_time_mapping_.TruncateOldEntries(_current_time);
       uint64_t preserve_time =
           static_cast<uint64_t>(_current_time) > preserve_time_duration
               ? _current_time - preserve_time_duration
               : 0;
+      // GetProximalSeqnoBeforeTime tells us the last seqno known to have been
+      // written at or before the given time. + 1 to get the minimum we should
+      // preserve without excluding anything that might have been written on or
+      // after the given time.
       preserve_time_min_seqno_ =
-          seqno_time_mapping_.GetOldestSequenceNum(preserve_time);
+          seqno_to_time_mapping_.GetProximalSeqnoBeforeTime(preserve_time) + 1;
       if (c->immutable_options()->preclude_last_level_data_seconds > 0) {
         uint64_t preclude_last_level_time =
             static_cast<uint64_t>(_current_time) >
@@ -335,7 +339,9 @@ void CompactionJob::Prepare() {
                       c->immutable_options()->preclude_last_level_data_seconds
                 : 0;
         preclude_last_level_min_seqno_ =
-            seqno_time_mapping_.GetOldestSequenceNum(preclude_last_level_time);
+            seqno_to_time_mapping_.GetProximalSeqnoBeforeTime(
+                preclude_last_level_time) +
+            1;
       }
     }
   }
@@ -1570,7 +1576,7 @@ Status CompactionJob::FinishCompactionOutputFile(
 
   const uint64_t current_entries = outputs.NumEntries();
 
-  s = outputs.Finish(s, seqno_time_mapping_);
+  s = outputs.Finish(s, seqno_to_time_mapping_);
 
   if (s.ok()) {
     // With accurate smallest and largest key, we can get a slightly more
diff --git a/db/compaction/compaction_job.h b/db/compaction/compaction_job.h
index 72d256735a..e812cfc72a 100644
--- a/db/compaction/compaction_job.h
+++ b/db/compaction/compaction_job.h
@@ -350,7 +350,7 @@ class CompactionJob {
 
   // Stores the sequence number to time mapping gathered from all input files
   // it also collects the smallest_seqno -> oldest_ancester_time from the SST.
-  SeqnoToTimeMapping seqno_time_mapping_;
+  SeqnoToTimeMapping seqno_to_time_mapping_;
 
   // Minimal sequence number for preserving the time information. The time info
   // older than this sequence number won't be preserved after the compaction and
diff --git a/db/compaction/compaction_outputs.cc b/db/compaction/compaction_outputs.cc
index 3e21484c46..eb76cd849a 100644
--- a/db/compaction/compaction_outputs.cc
+++ b/db/compaction/compaction_outputs.cc
@@ -18,16 +18,18 @@ void CompactionOutputs::NewBuilder(const TableBuilderOptions& tboptions) {
   builder_.reset(NewTableBuilder(tboptions, file_writer_.get()));
 }
 
-Status CompactionOutputs::Finish(const Status& intput_status,
-                                 const SeqnoToTimeMapping& seqno_time_mapping) {
+Status CompactionOutputs::Finish(
+    const Status& intput_status,
+    const SeqnoToTimeMapping& seqno_to_time_mapping) {
   FileMetaData* meta = GetMetaData();
   assert(meta != nullptr);
   Status s = intput_status;
   if (s.ok()) {
-    std::string seqno_time_mapping_str;
-    seqno_time_mapping.Encode(seqno_time_mapping_str, meta->fd.smallest_seqno,
-                              meta->fd.largest_seqno, meta->file_creation_time);
-    builder_->SetSeqnoTimeTableProperties(seqno_time_mapping_str,
+    std::string seqno_to_time_mapping_str;
+    seqno_to_time_mapping.Encode(
+        seqno_to_time_mapping_str, meta->fd.smallest_seqno,
+        meta->fd.largest_seqno, meta->file_creation_time);
+    builder_->SetSeqnoTimeTableProperties(seqno_to_time_mapping_str,
                                           meta->oldest_ancester_time);
     s = builder_->Finish();
 
diff --git a/db/compaction/compaction_outputs.h b/db/compaction/compaction_outputs.h
index 6c3e3b6b33..18246cf2fa 100644
--- a/db/compaction/compaction_outputs.h
+++ b/db/compaction/compaction_outputs.h
@@ -107,7 +107,7 @@ class CompactionOutputs {
 
   // Finish the current output file
   Status Finish(const Status& intput_status,
-                const SeqnoToTimeMapping& seqno_time_mapping);
+                const SeqnoToTimeMapping& seqno_to_time_mapping);
 
   // Update output table properties from table builder
   void UpdateTableProperties() {
diff --git a/db/compaction/tiered_compaction_test.cc b/db/compaction/tiered_compaction_test.cc
index d8aa229dfb..51f931603a 100644
--- a/db/compaction/tiered_compaction_test.cc
+++ b/db/compaction/tiered_compaction_test.cc
@@ -1249,8 +1249,11 @@ TEST_F(PrecludeLastLevelTest, MigrationFromPreserveTimeManualCompaction) {
   options.num_levels = kNumLevels;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kKeyPerSec)); });
 
@@ -1311,8 +1314,11 @@ TEST_F(PrecludeLastLevelTest, MigrationFromPreserveTimeAutoCompaction) {
   options.num_levels = kNumLevels;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kKeyPerSec)); });
 
@@ -1387,8 +1393,11 @@ TEST_F(PrecludeLastLevelTest, MigrationFromPreserveTimePartial) {
   options.num_levels = kNumLevels;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kKeyPerSec)); });
 
@@ -1514,8 +1523,11 @@ TEST_F(PrecludeLastLevelTest, LastLevelOnlyCompactionPartial) {
   options.num_levels = kNumLevels;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kKeyPerSec)); });
 
@@ -1592,8 +1604,11 @@ TEST_P(PrecludeLastLevelTestWithParms, LastLevelOnlyCompactionNoPreclude) {
   options.num_levels = kNumLevels;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kKeyPerSec)); });
 
@@ -1906,8 +1921,11 @@ TEST_F(PrecludeLastLevelTest, PartialPenultimateLevelCompaction) {
   options.num_levels = kNumLevels;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(10)); });
 
@@ -1996,7 +2014,13 @@ TEST_F(PrecludeLastLevelTest, PartialPenultimateLevelCompaction) {
   Close();
 }
 
-TEST_F(PrecludeLastLevelTest, RangeDelsCauseFileEndpointsToOverlap) {
+// FIXME broken test:
+// dbfull()->TEST_WaitForCompact()
+// Corruption: force_consistency_checks(DEBUG): VersionBuilder: L5 has
+// overlapping ranges:
+// file #14 largest key: '6B6579303030303134' seq:32, type:1 vs.
+// file #19 smallest key: '6B6579303030303130' seq:10, type:1
+TEST_F(PrecludeLastLevelTest, DISABLED_RangeDelsCauseFileEndpointsToOverlap) {
   const int kNumLevels = 7;
   const int kSecondsPerKey = 10;
   const int kNumFiles = 3;
@@ -2017,8 +2041,11 @@ TEST_F(PrecludeLastLevelTest, RangeDelsCauseFileEndpointsToOverlap) {
   options.target_file_size_base = kFileBytes;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun([&] {
     mock_clock_->MockSleepForSeconds(static_cast<int>(kSecondsPerKey));
   });
@@ -2139,7 +2166,6 @@ TEST_F(PrecludeLastLevelTest, RangeDelsCauseFileEndpointsToOverlap) {
   Close();
 }
 
-
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {
diff --git a/db/db_impl/db_impl.cc b/db/db_impl/db_impl.cc
index 1af0a1feca..204d56e7d1 100644
--- a/db/db_impl/db_impl.cc
+++ b/db/db_impl/db_impl.cc
@@ -830,9 +830,9 @@ Status DBImpl::RegisterRecordSeqnoTimeWorker() {
       }
     }
     if (min_time_duration == std::numeric_limits<uint64_t>::max()) {
-      seqno_time_mapping_.Resize(0, 0);
+      seqno_to_time_mapping_.Resize(0, 0);
     } else {
-      seqno_time_mapping_.Resize(min_time_duration, max_time_duration);
+      seqno_to_time_mapping_.Resize(min_time_duration, max_time_duration);
     }
   }
 
@@ -6371,16 +6371,18 @@ Status DBImpl::GetCreationTimeOfOldestFile(uint64_t* creation_time) {
 }
 
 void DBImpl::RecordSeqnoToTimeMapping() {
+  // TECHNICALITY: Sample last sequence number *before* time, as prescribed
+  // for SeqnoToTimeMapping
+  SequenceNumber seqno = GetLatestSequenceNumber();
   // Get time first then sequence number, so the actual time of seqno is <=
   // unix_time recorded
   int64_t unix_time = 0;
   immutable_db_options_.clock->GetCurrentTime(&unix_time)
       .PermitUncheckedError();  // Ignore error
-  SequenceNumber seqno = GetLatestSequenceNumber();
   bool appended = false;
   {
     InstrumentedMutexLock l(&mutex_);
-    appended = seqno_time_mapping_.Append(seqno, unix_time);
+    appended = seqno_to_time_mapping_.Append(seqno, unix_time);
   }
   if (!appended) {
     ROCKS_LOG_WARN(immutable_db_options_.info_log,
diff --git a/db/db_impl/db_impl.h b/db/db_impl/db_impl.h
index 98565bc70a..3f114fa369 100644
--- a/db/db_impl/db_impl.h
+++ b/db/db_impl/db_impl.h
@@ -2748,9 +2748,9 @@ class DBImpl : public DB {
   // Pointer to WriteBufferManager stalling interface.
   std::unique_ptr<StallInterface> wbm_stall_;
 
-  // seqno_time_mapping_ stores the sequence number to time mapping, it's not
+  // seqno_to_time_mapping_ stores the sequence number to time mapping, it's not
   // thread safe, both read and write need db mutex hold.
-  SeqnoToTimeMapping seqno_time_mapping_;
+  SeqnoToTimeMapping seqno_to_time_mapping_;
 
   // Stop write token that is acquired when first LockWAL() is called.
   // Destroyed when last UnlockWAL() is called. Controlled by DB mutex.
diff --git a/db/db_impl/db_impl_compaction_flush.cc b/db/db_impl/db_impl_compaction_flush.cc
index b610b8d33d..49da5acd33 100644
--- a/db/db_impl/db_impl_compaction_flush.cc
+++ b/db/db_impl/db_impl_compaction_flush.cc
@@ -250,7 +250,7 @@ Status DBImpl::FlushMemTableToOutputFile(
       GetCompressionFlush(*cfd->ioptions(), mutable_cf_options), stats_,
       &event_logger_, mutable_cf_options.report_bg_io_stats,
       true /* sync_output_directory */, true /* write_manifest */, thread_pri,
-      io_tracer_, seqno_time_mapping_, db_id_, db_session_id_,
+      io_tracer_, seqno_to_time_mapping_, db_id_, db_session_id_,
       cfd->GetFullHistoryTsLow(), &blob_callback_);
   FileMetaData file_meta;
 
@@ -522,7 +522,7 @@ Status DBImpl::AtomicFlushMemTablesToOutputFiles(
         GetCompressionFlush(*cfd->ioptions(), mutable_cf_options), stats_,
         &event_logger_, mutable_cf_options.report_bg_io_stats,
         false /* sync_output_directory */, false /* write_manifest */,
-        thread_pri, io_tracer_, seqno_time_mapping_, db_id_, db_session_id_,
+        thread_pri, io_tracer_, seqno_to_time_mapping_, db_id_, db_session_id_,
         cfd->GetFullHistoryTsLow(), &blob_callback_));
   }
 
diff --git a/db/db_impl/db_impl_debug.cc b/db/db_impl/db_impl_debug.cc
index be63637a26..670bc78872 100644
--- a/db/db_impl/db_impl_debug.cc
+++ b/db/db_impl/db_impl_debug.cc
@@ -306,7 +306,7 @@ const PeriodicTaskScheduler& DBImpl::TEST_GetPeriodicTaskScheduler() const {
 
 SeqnoToTimeMapping DBImpl::TEST_GetSeqnoToTimeMapping() const {
   InstrumentedMutexLock l(&mutex_);
-  return seqno_time_mapping_;
+  return seqno_to_time_mapping_;
 }
 
 
diff --git a/db/db_impl/db_impl_open.cc b/db/db_impl/db_impl_open.cc
index de164cc208..172a55879e 100644
--- a/db/db_impl/db_impl_open.cc
+++ b/db/db_impl/db_impl_open.cc
@@ -1649,7 +1649,7 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
           TableFileCreationReason::kRecovery, 0 /* oldest_key_time */,
           0 /* file_creation_time */, db_id_, db_session_id_,
           0 /* target_file_size */, meta.fd.GetNumber());
-      SeqnoToTimeMapping empty_seqno_time_mapping;
+      SeqnoToTimeMapping empty_seqno_to_time_mapping;
       Version* version = cfd->current();
       version->Ref();
       const ReadOptions read_option(Env::IOActivity::kDBOpen);
@@ -1661,7 +1661,7 @@ Status DBImpl::WriteLevel0TableForRecovery(int job_id, ColumnFamilyData* cfd,
           snapshot_seqs, earliest_write_conflict_snapshot, kMaxSequenceNumber,
           snapshot_checker, paranoid_file_checks, cfd->internal_stats(), &io_s,
           io_tracer_, BlobFileCreationReason::kRecovery,
-          empty_seqno_time_mapping, &event_logger_, job_id, Env::IO_HIGH,
+          empty_seqno_to_time_mapping, &event_logger_, job_id, Env::IO_HIGH,
           nullptr /* table_properties */, write_hint,
           nullptr /*full_history_ts_low*/, &blob_callback_, version,
           &num_input_entries);
diff --git a/db/flush_job.cc b/db/flush_job.cc
index 69303081fb..a3e168823a 100644
--- a/db/flush_job.cc
+++ b/db/flush_job.cc
@@ -100,7 +100,7 @@ FlushJob::FlushJob(
     Statistics* stats, EventLogger* event_logger, bool measure_io_stats,
     const bool sync_output_directory, const bool write_manifest,
     Env::Priority thread_pri, const std::shared_ptr<IOTracer>& io_tracer,
-    const SeqnoToTimeMapping& seqno_time_mapping, const std::string& db_id,
+    const SeqnoToTimeMapping& seqno_to_time_mapping, const std::string& db_id,
     const std::string& db_session_id, std::string full_history_ts_low,
     BlobFileCompletionCallback* blob_callback)
     : dbname_(dbname),
@@ -136,7 +136,7 @@ FlushJob::FlushJob(
       clock_(db_options_.clock),
       full_history_ts_low_(std::move(full_history_ts_low)),
       blob_callback_(blob_callback),
-      db_impl_seqno_time_mapping_(seqno_time_mapping) {
+      db_impl_seqno_to_time_mapping_(seqno_to_time_mapping) {
   // Update the thread status to indicate flush.
   ReportStartedFlush();
   TEST_SYNC_POINT("FlushJob::FlushJob()");
@@ -851,10 +851,11 @@ Status FlushJob::WriteLevel0Table() {
   Status s;
 
   SequenceNumber smallest_seqno = mems_.front()->GetEarliestSequenceNumber();
-  if (!db_impl_seqno_time_mapping_.Empty()) {
-    // make a local copy, as the seqno_time_mapping from db_impl is not thread
-    // safe, which will be used while not holding the db_mutex.
-    seqno_to_time_mapping_ = db_impl_seqno_time_mapping_.Copy(smallest_seqno);
+  if (!db_impl_seqno_to_time_mapping_.Empty()) {
+    // make a local copy, as the seqno_to_time_mapping from db_impl is not
+    // thread safe, which will be used while not holding the db_mutex.
+    seqno_to_time_mapping_ =
+        db_impl_seqno_to_time_mapping_.Copy(smallest_seqno);
   }
 
   std::vector<BlobFileAddition> blob_file_additions;
diff --git a/db/flush_job.h b/db/flush_job.h
index db5dbd719f..aef33ef423 100644
--- a/db/flush_job.h
+++ b/db/flush_job.h
@@ -210,9 +210,9 @@ class FlushJob {
   const std::string full_history_ts_low_;
   BlobFileCompletionCallback* blob_callback_;
 
-  // reference to the seqno_time_mapping_ in db_impl.h, not safe to read without
-  // db mutex
-  const SeqnoToTimeMapping& db_impl_seqno_time_mapping_;
+  // reference to the seqno_to_time_mapping_ in db_impl.h, not safe to read
+  // without db mutex
+  const SeqnoToTimeMapping& db_impl_seqno_to_time_mapping_;
   SeqnoToTimeMapping seqno_to_time_mapping_;
 
   // Keeps track of the newest user-defined timestamp for this flush job if
diff --git a/db/repair.cc b/db/repair.cc
index 1af738fca7..e303eae643 100644
--- a/db/repair.cc
+++ b/db/repair.cc
@@ -471,7 +471,7 @@ class Repairer {
           0 /* file_creation_time */, "DB Repairer" /* db_id */, db_session_id_,
           0 /*target_file_size*/, meta.fd.GetNumber());
 
-      SeqnoToTimeMapping empty_seqno_time_mapping;
+      SeqnoToTimeMapping empty_seqno_to_time_mapping;
       status = BuildTable(
           dbname_, /* versions */ nullptr, immutable_db_options_, tboptions,
           file_options_, read_options, table_cache_.get(), iter.get(),
@@ -479,8 +479,9 @@ class Repairer {
           {}, kMaxSequenceNumber, kMaxSequenceNumber, snapshot_checker,
           false /* paranoid_file_checks*/, nullptr /* internal_stats */, &io_s,
           nullptr /*IOTracer*/, BlobFileCreationReason::kRecovery,
-          empty_seqno_time_mapping, nullptr /* event_logger */, 0 /* job_id */,
-          Env::IO_HIGH, nullptr /* table_properties */, write_hint);
+          empty_seqno_to_time_mapping, nullptr /* event_logger */,
+          0 /* job_id */, Env::IO_HIGH, nullptr /* table_properties */,
+          write_hint);
       ROCKS_LOG_INFO(db_options_.info_log,
                      "Log #%" PRIu64 ": %d ops saved to Table #%" PRIu64 " %s",
                      log, counter, meta.fd.GetNumber(),
diff --git a/db/seqno_time_test.cc b/db/seqno_time_test.cc
index b18b255128..e4c5751676 100644
--- a/db/seqno_time_test.cc
+++ b/db/seqno_time_test.cc
@@ -12,7 +12,6 @@
 #include "rocksdb/utilities/debug.h"
 #include "test_util/mock_time_env.h"
 
-
 namespace ROCKSDB_NAMESPACE {
 
 class SeqnoTimeTest : public DBTestBase {
@@ -77,8 +76,11 @@ TEST_F(SeqnoTimeTest, TemperatureBasicUniversal) {
   options.num_levels = kNumLevels;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(kKeyPerSec)); });
 
@@ -189,8 +191,11 @@ TEST_F(SeqnoTimeTest, TemperatureBasicLevel) {
   options.disable_auto_compactions = true;
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
   // pass some time first, otherwise the first a few keys write time are going
-  // to be zero, and internally zero has special meaning: kUnknownSeqnoTime
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
   dbfull()->TEST_WaitForPeriodicTaskRun(
       [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(10)); });
 
@@ -320,7 +325,7 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
   DestroyAndReopen(options);
 
   std::set<uint64_t> checked_file_nums;
-  SequenceNumber start_seq = dbfull()->GetLatestSequenceNumber();
+  SequenceNumber start_seq = dbfull()->GetLatestSequenceNumber() + 1;
   // Write a key every 10 seconds
   for (int i = 0; i < 200; i++) {
     ASSERT_OK(Put(Key(i), "value"));
@@ -341,15 +346,15 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
   // passes 2k time.
   ASSERT_GE(seqs.size(), 19);
   ASSERT_LE(seqs.size(), 21);
-  SequenceNumber seq_end = dbfull()->GetLatestSequenceNumber();
+  SequenceNumber seq_end = dbfull()->GetLatestSequenceNumber() + 1;
   for (auto i = start_seq; i < start_seq + 10; i++) {
-    ASSERT_LE(tp_mapping.GetOldestApproximateTime(i), (i + 1) * 10);
+    ASSERT_LE(tp_mapping.GetProximalTimeBeforeSeqno(i), (i + 1) * 10);
   }
   start_seq += 10;
   for (auto i = start_seq; i < seq_end; i++) {
     // The result is within the range
-    ASSERT_GE(tp_mapping.GetOldestApproximateTime(i), (i - 10) * 10);
-    ASSERT_LE(tp_mapping.GetOldestApproximateTime(i), (i + 10) * 10);
+    ASSERT_GE(tp_mapping.GetProximalTimeBeforeSeqno(i), (i - 10) * 10);
+    ASSERT_LE(tp_mapping.GetProximalTimeBeforeSeqno(i), (i + 10) * 10);
   }
   checked_file_nums.insert(it->second->orig_file_number);
   start_seq = seq_end;
@@ -360,7 +365,7 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
     dbfull()->TEST_WaitForPeriodicTaskRun(
         [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(1)); });
   }
-  seq_end = dbfull()->GetLatestSequenceNumber();
+  seq_end = dbfull()->GetLatestSequenceNumber() + 1;
   ASSERT_OK(Flush());
   tables_props.clear();
   ASSERT_OK(dbfull()->GetPropertiesOfAllTables(&tables_props));
@@ -384,8 +389,8 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
   for (auto i = start_seq; i < seq_end; i++) {
     // The result is not very accurate, as there is more data write within small
     // range of time
-    ASSERT_GE(tp_mapping.GetOldestApproximateTime(i), (i - start_seq) + 1000);
-    ASSERT_LE(tp_mapping.GetOldestApproximateTime(i), (i - start_seq) + 3000);
+    ASSERT_GE(tp_mapping.GetProximalTimeBeforeSeqno(i), (i - start_seq) + 1000);
+    ASSERT_LE(tp_mapping.GetProximalTimeBeforeSeqno(i), (i - start_seq) + 3000);
   }
   checked_file_nums.insert(it->second->orig_file_number);
   start_seq = seq_end;
@@ -396,7 +401,7 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
     dbfull()->TEST_WaitForPeriodicTaskRun(
         [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(200)); });
   }
-  seq_end = dbfull()->GetLatestSequenceNumber();
+  seq_end = dbfull()->GetLatestSequenceNumber() + 1;
   ASSERT_OK(Flush());
   tables_props.clear();
   ASSERT_OK(dbfull()->GetPropertiesOfAllTables(&tables_props));
@@ -419,14 +424,14 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
   ASSERT_LE(seqs.size(), 101);
   for (auto i = start_seq; i < seq_end - 99; i++) {
     // likely the first 100 entries reports 0
-    ASSERT_LE(tp_mapping.GetOldestApproximateTime(i), (i - start_seq) + 3000);
+    ASSERT_LE(tp_mapping.GetProximalTimeBeforeSeqno(i), (i - start_seq) + 3000);
   }
   start_seq += 101;
 
   for (auto i = start_seq; i < seq_end; i++) {
-    ASSERT_GE(tp_mapping.GetOldestApproximateTime(i),
+    ASSERT_GE(tp_mapping.GetProximalTimeBeforeSeqno(i),
               (i - start_seq) * 200 + 22200);
-    ASSERT_LE(tp_mapping.GetOldestApproximateTime(i),
+    ASSERT_LE(tp_mapping.GetProximalTimeBeforeSeqno(i),
               (i - start_seq) * 200 + 22600);
   }
   checked_file_nums.insert(it->second->orig_file_number);
@@ -438,7 +443,7 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
     dbfull()->TEST_WaitForPeriodicTaskRun(
         [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(100)); });
   }
-  seq_end = dbfull()->GetLatestSequenceNumber();
+  seq_end = dbfull()->GetLatestSequenceNumber() + 1;
   ASSERT_OK(Flush());
   tables_props.clear();
   ASSERT_OK(dbfull()->GetPropertiesOfAllTables(&tables_props));
@@ -486,15 +491,15 @@ TEST_P(SeqnoTimeTablePropTest, BasicSeqnoToTimeMapping) {
   ASSERT_LE(seqs.size(), 101);
   for (auto i = start_seq; i < seq_end - 99; i++) {
     // likely the first 100 entries reports 0
-    ASSERT_LE(tp_mapping.GetOldestApproximateTime(i),
+    ASSERT_LE(tp_mapping.GetProximalTimeBeforeSeqno(i),
               (i - start_seq) * 100 + 50000);
   }
   start_seq += 101;
 
   for (auto i = start_seq; i < seq_end; i++) {
-    ASSERT_GE(tp_mapping.GetOldestApproximateTime(i),
+    ASSERT_GE(tp_mapping.GetProximalTimeBeforeSeqno(i),
               (i - start_seq) * 100 + 52200);
-    ASSERT_LE(tp_mapping.GetOldestApproximateTime(i),
+    ASSERT_LE(tp_mapping.GetProximalTimeBeforeSeqno(i),
               (i - start_seq) * 100 + 52400);
   }
   ASSERT_OK(db_->Close());
@@ -707,6 +712,14 @@ TEST_P(SeqnoTimeTablePropTest, SeqnoToTimeMappingUniversal) {
 
   DestroyAndReopen(options);
 
+  // bootstrap DB sequence numbers (FIXME: make these steps unnecessary)
+  ASSERT_OK(Put("foo", "bar"));
+  ASSERT_OK(SingleDelete("foo"));
+  // pass some time first, otherwise the first a few keys write time are going
+  // to be zero, and internally zero has special meaning: kUnknownTimeBeforeAll
+  dbfull()->TEST_WaitForPeriodicTaskRun(
+      [&] { mock_clock_->MockSleepForSeconds(static_cast<int>(10)); });
+
   std::atomic_uint64_t num_seqno_zeroing{0};
 
   SyncPoint::GetInstance()->DisableProcessing();
@@ -843,8 +856,9 @@ TEST_F(SeqnoTimeTest, MappingAppend) {
   ASSERT_FALSE(test.Append(8, 12));
   ASSERT_EQ(size, test.Size());
 
-  // Append with the same seqno, newer time will be accepted
-  ASSERT_TRUE(test.Append(10, 12));
+  // Append with the same seqno, newer time is rejected because that makes
+  // GetProximalSeqnoBeforeTime queries worse (see later test)
+  ASSERT_FALSE(test.Append(10, 12));
   ASSERT_EQ(size, test.Size());
   // older time will be ignored
   ASSERT_FALSE(test.Append(10, 9));
@@ -853,25 +867,176 @@ TEST_F(SeqnoTimeTest, MappingAppend) {
   // new seqno with old time will be ignored
   ASSERT_FALSE(test.Append(12, 8));
   ASSERT_EQ(size, test.Size());
+
+  // new seqno with same time is accepted by replacing last entry
+  // (improves GetProximalSeqnoBeforeTime queries without blowing up size)
+  ASSERT_TRUE(test.Append(12, 11));
+  ASSERT_EQ(size, test.Size());
 }
 
-TEST_F(SeqnoTimeTest, GetOldestApproximateTime) {
+TEST_F(SeqnoTimeTest, ProximalFunctions) {
   SeqnoToTimeMapping test(/*max_time_duration=*/100, /*max_capacity=*/10);
 
-  ASSERT_EQ(test.GetOldestApproximateTime(10), kUnknownSeqnoTime);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(1), kUnknownTimeBeforeAll);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(1000000000000U),
+            kUnknownTimeBeforeAll);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(1), kUnknownSeqnoBeforeAll);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(1000000000000U),
+            kUnknownSeqnoBeforeAll);
 
-  test.Append(3, 10);
+  // (Taken from example in SeqnoToTimeMapping class comment)
+  // Time 500 is after seqno 10 and before seqno 11
+  EXPECT_TRUE(test.Append(10, 500));
 
-  ASSERT_EQ(test.GetOldestApproximateTime(2), kUnknownSeqnoTime);
-  ASSERT_EQ(test.GetOldestApproximateTime(3), 10);
-  ASSERT_EQ(test.GetOldestApproximateTime(10), 10);
+  // Seqno too early
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(9), kUnknownTimeBeforeAll);
+  // We only know that 500 is after 10
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(10), kUnknownTimeBeforeAll);
+  // Found
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(11), 500U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(1000000000000U), 500U);
 
-  test.Append(10, 100);
+  // Time too early
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(499), kUnknownSeqnoBeforeAll);
+  // Found
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(500), 10U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(501), 10U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(1000000000000U), 10U);
 
-  test.Append(100, 1000);
-  ASSERT_EQ(test.GetOldestApproximateTime(10), 100);
-  ASSERT_EQ(test.GetOldestApproximateTime(40), 100);
-  ASSERT_EQ(test.GetOldestApproximateTime(111), 1000);
+  // More samples
+  EXPECT_TRUE(test.Append(20, 600));
+  EXPECT_TRUE(test.Append(30, 700));
+
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(10), kUnknownTimeBeforeAll);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(11), 500U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(20), 500U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(21), 600U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(30), 600U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(31), 700U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(1000000000000U), 700U);
+
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(499), kUnknownSeqnoBeforeAll);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(500), 10U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(501), 10U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(599), 10U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(600), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(601), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(699), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(700), 30U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(701), 30U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(1000000000000U), 30U);
+
+  // Redundant sample ignored
+  EXPECT_EQ(test.Size(), 3U);
+  EXPECT_FALSE(test.Append(30, 700));
+  EXPECT_EQ(test.Size(), 3U);
+
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(30), 600U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(31), 700U);
+
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(699), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(700), 30U);
+
+  // Later sample with same seqno is ignored, to provide best results
+  // for GetProximalSeqnoBeforeTime function while saving entries
+  // in SeqnoToTimeMapping.
+  EXPECT_FALSE(test.Append(30, 800));
+
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(30), 600U);
+  // Could return 800, but saving space in SeqnoToTimeMapping instead.
+  // Can reconsider if/when GetProximalTimeBeforeSeqno is used in
+  // production.
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(31), 700U);
+
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(699), 20U);
+  // If the existing {30, 700} entry were replaced with {30, 800}, this
+  // would return seqno 20 instead of 30, which would preclude more than
+  // necessary for "preclude_last_level_data_seconds" feature.
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(700), 30U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(800), 30U);
+
+  // Still OK
+  EXPECT_TRUE(test.Append(40, 900));
+
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(30), 600U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(41), 900U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(899), 30U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(900), 40U);
+
+  // Burst of writes during a short time creates an opportunity
+  // for better results from GetProximalSeqnoBeforeTime(), at the
+  // expense of GetProximalTimeBeforeSeqno().
+  EXPECT_TRUE(test.Append(50, 900));
+
+  // These are subject to later revision depending on priorities
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(49), 700U);
+  EXPECT_EQ(test.GetProximalTimeBeforeSeqno(51), 900U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(899), 30U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(900), 50U);
+}
+
+TEST_F(SeqnoTimeTest, TruncateOldEntries) {
+  constexpr uint64_t kMaxTimeDuration = 42;
+  SeqnoToTimeMapping test(kMaxTimeDuration, /*max_capacity=*/10);
+
+  EXPECT_EQ(test.Size(), 0U);
+
+  // Safe on empty mapping
+  test.TruncateOldEntries(500);
+
+  EXPECT_EQ(test.Size(), 0U);
+
+  // (Taken from example in SeqnoToTimeMapping class comment)
+  // Time 500 is after seqno 10 and before seqno 11
+  EXPECT_TRUE(test.Append(10, 500));
+  EXPECT_TRUE(test.Append(20, 600));
+  EXPECT_TRUE(test.Append(30, 700));
+  EXPECT_TRUE(test.Append(40, 800));
+  EXPECT_TRUE(test.Append(50, 900));
+
+  EXPECT_EQ(test.Size(), 5U);
+
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(500), 10U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(599), 10U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(600), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(699), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(700), 30U);
+  // etc.
+
+  // Must keep first entry
+  test.TruncateOldEntries(500 + kMaxTimeDuration);
+  EXPECT_EQ(test.Size(), 5U);
+  test.TruncateOldEntries(599 + kMaxTimeDuration);
+  EXPECT_EQ(test.Size(), 5U);
+
+  // Purges first entry
+  test.TruncateOldEntries(600 + kMaxTimeDuration);
+  EXPECT_EQ(test.Size(), 4U);
+
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(500), kUnknownSeqnoBeforeAll);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(599), kUnknownSeqnoBeforeAll);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(600), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(699), 20U);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(700), 30U);
+
+  // No effect
+  test.TruncateOldEntries(600 + kMaxTimeDuration);
+  EXPECT_EQ(test.Size(), 4U);
+  test.TruncateOldEntries(699 + kMaxTimeDuration);
+  EXPECT_EQ(test.Size(), 4U);
+
+  // Purges next two
+  test.TruncateOldEntries(899 + kMaxTimeDuration);
+  EXPECT_EQ(test.Size(), 2U);
+
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(799), kUnknownSeqnoBeforeAll);
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(899), 40U);
+
+  // Always keep last entry, to have a non-trivial seqno bound
+  test.TruncateOldEntries(10000000);
+  EXPECT_EQ(test.Size(), 1U);
+
+  EXPECT_EQ(test.GetProximalSeqnoBeforeTime(10000000), 50U);
 }
 
 TEST_F(SeqnoTimeTest, Sort) {
@@ -930,10 +1095,10 @@ TEST_F(SeqnoTimeTest, EncodeDecodeBasic) {
   for (SequenceNumber seq = 0; seq <= 1000; seq++) {
     // test has the more accurate time mapping, encode only pick
     // kMaxSeqnoTimePairsPerSST number of entries, which is less accurate
-    uint64_t target_time = test.GetOldestApproximateTime(seq);
-    ASSERT_GE(decoded.GetOldestApproximateTime(seq),
+    uint64_t target_time = test.GetProximalTimeBeforeSeqno(seq);
+    ASSERT_GE(decoded.GetProximalTimeBeforeSeqno(seq),
               target_time < 200 ? 0 : target_time - 200);
-    ASSERT_LE(decoded.GetOldestApproximateTime(seq), target_time);
+    ASSERT_LE(decoded.GetProximalTimeBeforeSeqno(seq), target_time);
   }
 }
 
diff --git a/db/seqno_to_time_mapping.cc b/db/seqno_to_time_mapping.cc
index c692099294..ec7b2d9cbd 100644
--- a/db/seqno_to_time_mapping.cc
+++ b/db/seqno_to_time_mapping.cc
@@ -11,14 +11,34 @@
 
 namespace ROCKSDB_NAMESPACE {
 
-uint64_t SeqnoToTimeMapping::GetOldestApproximateTime(
-    const SequenceNumber seqno) const {
+SeqnoToTimeMapping::pair_const_iterator SeqnoToTimeMapping::FindGreaterTime(
+    uint64_t time) const {
+  return std::upper_bound(pairs_.cbegin(), pairs_.cend(),
+                          SeqnoTimePair{0, time}, SeqnoTimePair::TimeLess);
+}
+
+SeqnoToTimeMapping::pair_const_iterator SeqnoToTimeMapping::FindGreaterEqSeqno(
+    SequenceNumber seqno) const {
+  return std::lower_bound(pairs_.cbegin(), pairs_.cend(),
+                          SeqnoTimePair{seqno, 0}, SeqnoTimePair::SeqnoLess);
+}
+
+SeqnoToTimeMapping::pair_const_iterator SeqnoToTimeMapping::FindGreaterSeqno(
+    SequenceNumber seqno) const {
+  return std::upper_bound(pairs_.cbegin(), pairs_.cend(),
+                          SeqnoTimePair{seqno, 0}, SeqnoTimePair::SeqnoLess);
+}
+
+uint64_t SeqnoToTimeMapping::GetProximalTimeBeforeSeqno(
+    SequenceNumber seqno) const {
   assert(is_sorted_);
-  auto it = std::upper_bound(seqno_time_mapping_.begin(),
-                             seqno_time_mapping_.end(), seqno);
-  if (it == seqno_time_mapping_.begin()) {
-    return 0;
+  // Find the last entry with a seqno strictly less than the given seqno.
+  // First, find the first entry >= the given seqno (or end)
+  auto it = FindGreaterEqSeqno(seqno);
+  if (it == pairs_.cbegin()) {
+    return kUnknownTimeBeforeAll;
   }
+  // Then return data from previous.
   it--;
   return it->time;
 }
@@ -28,44 +48,47 @@ void SeqnoToTimeMapping::Add(SequenceNumber seqno, uint64_t time) {
     return;
   }
   is_sorted_ = false;
-  seqno_time_mapping_.emplace_back(seqno, time);
+  pairs_.emplace_back(seqno, time);
 }
 
 void SeqnoToTimeMapping::TruncateOldEntries(const uint64_t now) {
   assert(is_sorted_);
 
   if (max_time_duration_ == 0) {
+    // No cutoff time
     return;
   }
 
-  const uint64_t cut_off_time =
-      now > max_time_duration_ ? now - max_time_duration_ : 0;
-  assert(cut_off_time <= now);  // no overflow
-
-  auto it = std::upper_bound(
-      seqno_time_mapping_.begin(), seqno_time_mapping_.end(), cut_off_time,
-      [](uint64_t target, const SeqnoTimePair& other) -> bool {
-        return target < other.time;
-      });
-  if (it == seqno_time_mapping_.begin()) {
+  if (now < max_time_duration_) {
+    // Would under-flow
     return;
   }
-  it--;
-  seqno_time_mapping_.erase(seqno_time_mapping_.begin(), it);
+
+  const uint64_t cut_off_time = now - max_time_duration_;
+  assert(cut_off_time <= now);  // no under/overflow
+
+  auto it = FindGreaterTime(cut_off_time);
+  if (it == pairs_.cbegin()) {
+    return;
+  }
+  // Move back one, to the entry that would be used to return a good seqno from
+  // GetProximalSeqnoBeforeTime(cut_off_time)
+  --it;
+  // Remove everything strictly before that entry
+  pairs_.erase(pairs_.cbegin(), std::move(it));
 }
 
-SequenceNumber SeqnoToTimeMapping::GetOldestSequenceNum(uint64_t time) {
+SequenceNumber SeqnoToTimeMapping::GetProximalSeqnoBeforeTime(uint64_t time) {
   assert(is_sorted_);
 
-  auto it = std::upper_bound(
-      seqno_time_mapping_.begin(), seqno_time_mapping_.end(), time,
-      [](uint64_t target, const SeqnoTimePair& other) -> bool {
-        return target < other.time;
-      });
-  if (it == seqno_time_mapping_.begin()) {
-    return 0;
+  // Find the last entry with a time <= the given time.
+  // First, find the first entry > the given time (or end).
+  auto it = FindGreaterTime(time);
+  if (it == pairs_.cbegin()) {
+    return kUnknownSeqnoBeforeAll;
   }
-  it--;
+  // Then return data from previous.
+  --it;
   return it->seqno;
 }
 
@@ -84,15 +107,13 @@ void SeqnoToTimeMapping::Encode(std::string& dest, const SequenceNumber start,
     return;
   }
 
-  auto start_it = std::upper_bound(seqno_time_mapping_.begin(),
-                                   seqno_time_mapping_.end(), start);
-  if (start_it != seqno_time_mapping_.begin()) {
+  auto start_it = FindGreaterSeqno(start);
+  if (start_it != pairs_.begin()) {
     start_it--;
   }
 
-  auto end_it = std::upper_bound(seqno_time_mapping_.begin(),
-                                 seqno_time_mapping_.end(), end);
-  if (end_it == seqno_time_mapping_.begin()) {
+  auto end_it = FindGreaterSeqno(end);
+  if (end_it == pairs_.begin()) {
     return;
   }
   if (start_it >= end_it) {
@@ -108,7 +129,7 @@ void SeqnoToTimeMapping::Encode(std::string& dest, const SequenceNumber start,
     }
   }
   // to include the first element
-  if (start_it != seqno_time_mapping_.begin()) {
+  if (start_it != pairs_.begin()) {
     start_it--;
   }
 
@@ -166,14 +187,14 @@ void SeqnoToTimeMapping::Encode(std::string& dest, const SequenceNumber start,
   SeqnoTimePair base;
   for (auto it = start_it; it < end_it; it++) {
     assert(base < *it);
-    SeqnoTimePair val = *it - base;
+    SeqnoTimePair val = it->ComputeDelta(base);
     base = *it;
     val.Encode(dest);
   }
 }
 
-Status SeqnoToTimeMapping::Add(const std::string& seqno_time_mapping_str) {
-  Slice input(seqno_time_mapping_str);
+Status SeqnoToTimeMapping::Add(const std::string& pairs_str) {
+  Slice input(pairs_str);
   if (input.empty()) {
     return Status::OK();
   }
@@ -189,8 +210,8 @@ Status SeqnoToTimeMapping::Add(const std::string& seqno_time_mapping_str) {
     if (!s.ok()) {
       return s;
     }
-    val.Add(base);
-    seqno_time_mapping_.emplace_back(val);
+    val.ApplyDelta(base);
+    pairs_.emplace_back(val);
     base = val;
   }
   return Status::OK();
@@ -222,19 +243,22 @@ bool SeqnoToTimeMapping::Append(SequenceNumber seqno, uint64_t time) {
       return false;
     }
     if (seqno == Last().seqno) {
-      Last().time = time;
-      return true;
+      // Updating Last() would hurt GetProximalSeqnoBeforeTime() queries, so
+      // NOT doing it (for now)
+      return false;
     }
     if (time == Last().time) {
-      // new sequence has the same time as old one, no need to add new mapping
-      return false;
+      // Updating Last() here helps GetProximalSeqnoBeforeTime() queries, so
+      // doing it (for now)
+      Last().seqno = seqno;
+      return true;
     }
   }
 
-  seqno_time_mapping_.emplace_back(seqno, time);
+  pairs_.emplace_back(seqno, time);
 
-  if (seqno_time_mapping_.size() > max_capacity_) {
-    seqno_time_mapping_.pop_front();
+  if (pairs_.size() > max_capacity_) {
+    pairs_.pop_front();
   }
   return true;
 }
@@ -245,10 +269,9 @@ bool SeqnoToTimeMapping::Resize(uint64_t min_time_duration,
       CalculateMaxCapacity(min_time_duration, max_time_duration);
   if (new_max_capacity == max_capacity_) {
     return false;
-  } else if (new_max_capacity < seqno_time_mapping_.size()) {
-    uint64_t delta = seqno_time_mapping_.size() - new_max_capacity;
-    seqno_time_mapping_.erase(seqno_time_mapping_.begin(),
-                              seqno_time_mapping_.begin() + delta);
+  } else if (new_max_capacity < pairs_.size()) {
+    uint64_t delta = pairs_.size() - new_max_capacity;
+    pairs_.erase(pairs_.begin(), pairs_.begin() + delta);
   }
   max_capacity_ = new_max_capacity;
   return true;
@@ -258,16 +281,16 @@ Status SeqnoToTimeMapping::Sort() {
   if (is_sorted_) {
     return Status::OK();
   }
-  if (seqno_time_mapping_.empty()) {
+  if (pairs_.empty()) {
     is_sorted_ = true;
     return Status::OK();
   }
 
-  std::deque<SeqnoTimePair> copy = std::move(seqno_time_mapping_);
+  std::deque<SeqnoTimePair> copy = std::move(pairs_);
 
   std::sort(copy.begin(), copy.end());
 
-  seqno_time_mapping_.clear();
+  pairs_.clear();
 
   // remove seqno = 0, which may have special meaning, like zeroed out data
   while (copy.front().seqno == 0) {
@@ -285,12 +308,12 @@ Status SeqnoToTimeMapping::Sort() {
       assert(it.seqno > prev.seqno);
       // If a larger sequence number has an older time which is not useful, skip
       if (it.time > prev.time) {
-        seqno_time_mapping_.push_back(prev);
+        pairs_.push_back(prev);
         prev = it;
       }
     }
   }
-  seqno_time_mapping_.emplace_back(prev);
+  pairs_.emplace_back(prev);
 
   is_sorted_ = true;
   return Status::OK();
@@ -298,7 +321,7 @@ Status SeqnoToTimeMapping::Sort() {
 
 std::string SeqnoToTimeMapping::ToHumanString() const {
   std::string ret;
-  for (const auto& seq_time : seqno_time_mapping_) {
+  for (const auto& seq_time : pairs_) {
     AppendNumberTo(&ret, seq_time.seqno);
     ret.append("->");
     AppendNumberTo(&ret, seq_time.time);
@@ -310,13 +333,11 @@ std::string SeqnoToTimeMapping::ToHumanString() const {
 SeqnoToTimeMapping SeqnoToTimeMapping::Copy(
     SequenceNumber smallest_seqno) const {
   SeqnoToTimeMapping ret;
-  auto it = std::upper_bound(seqno_time_mapping_.begin(),
-                             seqno_time_mapping_.end(), smallest_seqno);
-  if (it != seqno_time_mapping_.begin()) {
+  auto it = FindGreaterSeqno(smallest_seqno);
+  if (it != pairs_.begin()) {
     it--;
   }
-  std::copy(it, seqno_time_mapping_.end(),
-            std::back_inserter(ret.seqno_time_mapping_));
+  std::copy(it, pairs_.end(), std::back_inserter(ret.pairs_));
   return ret;
 }
 
@@ -330,12 +351,4 @@ uint64_t SeqnoToTimeMapping::CalculateMaxCapacity(uint64_t min_time_duration,
       max_time_duration * kMaxSeqnoTimePairsPerCF / min_time_duration);
 }
 
-SeqnoToTimeMapping::SeqnoTimePair SeqnoToTimeMapping::SeqnoTimePair::operator-(
-    const SeqnoTimePair& other) const {
-  SeqnoTimePair res;
-  res.seqno = seqno - other.seqno;
-  res.time = time - other.time;
-  return res;
-}
-
 }  // namespace ROCKSDB_NAMESPACE
diff --git a/db/seqno_to_time_mapping.h b/db/seqno_to_time_mapping.h
index 4ffc9c1992..132d74957e 100644
--- a/db/seqno_to_time_mapping.h
+++ b/db/seqno_to_time_mapping.h
@@ -18,20 +18,32 @@
 
 namespace ROCKSDB_NAMESPACE {
 
-constexpr uint64_t kUnknownSeqnoTime = 0;
+constexpr uint64_t kUnknownTimeBeforeAll = 0;
+constexpr SequenceNumber kUnknownSeqnoBeforeAll = 0;
 
-// SeqnoToTimeMapping stores the sequence number to time mapping, so given a
-// sequence number it can estimate the oldest possible time for that sequence
-// number. For example:
-//   10 -> 100
-//   50 -> 300
-// then if a key has seqno 19, the OldestApproximateTime would be 100, for 51 it
-// would be 300.
-// As it's a sorted list, the new entry is inserted from the back. The old data
-// will be popped from the front if they're no longer used.
+// SeqnoToTimeMapping stores a sampled mapping from sequence numbers to
+// unix times (seconds since epoch). This information provides rough bounds
+// between sequence numbers and their write times, but is primarily designed
+// for getting a best lower bound on the sequence number of data written no
+// later than a specified time.
 //
-// Note: the data struct is not thread safe, both read and write need to be
-//  synchronized by caller.
+// For ease of sampling, it is assumed that the recorded time in each pair
+// comes at or after the sequence number and before the next sequence number,
+// so this example:
+//
+// Seqno: 10,       11, ... 20,       21, ... 30,       31, ...
+// Time:  ...  500      ...      600      ...      700      ...
+//
+// would be represented as
+//   10 -> 500
+//   20 -> 600
+//   30 -> 700
+//
+// In typical operation, the list is sorted, both among seqnos and among times,
+// with a bounded number of entries, but some public working states violate
+// these constraints.
+//
+// NOT thread safe - requires external synchronization.
 class SeqnoToTimeMapping {
  public:
   // Maximum number of entries can be encoded into SST. The data is delta encode
@@ -63,28 +75,33 @@ class SeqnoToTimeMapping {
     // Decode the value from input Slice and remove it from the input
     Status Decode(Slice& input);
 
-    // subtraction of 2 SeqnoTimePair
-    SeqnoTimePair operator-(const SeqnoTimePair& other) const;
-
-    // Add 2 values together
-    void Add(const SeqnoTimePair& obj) {
-      seqno += obj.seqno;
-      time += obj.time;
+    // For delta encoding
+    SeqnoTimePair ComputeDelta(const SeqnoTimePair& base) const {
+      return {seqno - base.seqno, time - base.time};
     }
 
-    // Compare SeqnoTimePair with a sequence number, used for binary search a
-    // sequence number in a list of SeqnoTimePair
-    bool operator<(const SequenceNumber& other) const { return seqno < other; }
+    // For delta decoding
+    void ApplyDelta(const SeqnoTimePair& delta_or_base) {
+      seqno += delta_or_base.seqno;
+      time += delta_or_base.time;
+    }
 
-    // Compare 2 SeqnoTimePair
+    // Ordering used for Sort()
     bool operator<(const SeqnoTimePair& other) const {
       return std::tie(seqno, time) < std::tie(other.seqno, other.time);
     }
 
-    // Check if 2 SeqnoTimePair is the same
     bool operator==(const SeqnoTimePair& other) const {
       return std::tie(seqno, time) == std::tie(other.seqno, other.time);
     }
+
+    static bool SeqnoLess(const SeqnoTimePair& a, const SeqnoTimePair& b) {
+      return a.seqno < b.seqno;
+    }
+
+    static bool TimeLess(const SeqnoTimePair& a, const SeqnoTimePair& b) {
+      return a.time < b.time;
+    }
   };
 
   // constractor of SeqnoToTimeMapping
@@ -103,16 +120,31 @@ class SeqnoToTimeMapping {
   // existing ones. It maintains the internal sorted status.
   bool Append(SequenceNumber seqno, uint64_t time);
 
-  // Given a sequence number, estimate it's oldest time
-  uint64_t GetOldestApproximateTime(SequenceNumber seqno) const;
+  // Given a sequence number, return the best (largest / newest) known time
+  // that is no later than the write time of that given sequence number.
+  // If no such specific time is known, returns kUnknownTimeBeforeAll.
+  // Using the example in the class comment above,
+  //  GetProximalTimeBeforeSeqno(10) -> kUnknownTimeBeforeAll
+  //  GetProximalTimeBeforeSeqno(11) -> 500
+  //  GetProximalTimeBeforeSeqno(20) -> 500
+  //  GetProximalTimeBeforeSeqno(21) -> 600
+  uint64_t GetProximalTimeBeforeSeqno(SequenceNumber seqno) const;
 
-  // Truncate the old entries based on the current time and max_time_duration_
+  // Remove any entries not needed for GetProximalSeqnoBeforeTime queries of
+  // times older than `now - max_time_duration_`
   void TruncateOldEntries(uint64_t now);
 
-  // Given a time, return it's oldest possible sequence number
-  SequenceNumber GetOldestSequenceNum(uint64_t time);
+  // Given a time, return the best (largest) sequence number whose write time
+  // is no later than that given time. If no such specific sequence number is
+  // known, returns kUnknownSeqnoBeforeAll. Using the example in the class
+  // comment above,
+  //  GetProximalSeqnoBeforeTime(499) -> kUnknownSeqnoBeforeAll
+  //  GetProximalSeqnoBeforeTime(500) -> 10
+  //  GetProximalSeqnoBeforeTime(599) -> 10
+  //  GetProximalSeqnoBeforeTime(600) -> 20
+  SequenceNumber GetProximalSeqnoBeforeTime(uint64_t time);
 
-  // Encode to a binary string
+  // Encode to a binary string. start and end seqno are both inclusive.
   void Encode(std::string& des, SequenceNumber start, SequenceNumber end,
               uint64_t now,
               uint64_t output_size = kMaxSeqnoTimePairsPerSST) const;
@@ -122,10 +154,10 @@ class SeqnoToTimeMapping {
   void Add(SequenceNumber seqno, uint64_t time);
 
   // Decode and add the entries to the current obj. The list will be unsorted
-  Status Add(const std::string& seqno_time_mapping_str);
+  Status Add(const std::string& pairs_str);
 
   // Return the number of entries
-  size_t Size() const { return seqno_time_mapping_.size(); }
+  size_t Size() const { return pairs_.size(); }
 
   // Reduce the size of internal list
   bool Resize(uint64_t min_time_duration, uint64_t max_time_duration);
@@ -145,10 +177,10 @@ class SeqnoToTimeMapping {
   SeqnoToTimeMapping Copy(SequenceNumber smallest_seqno) const;
 
   // If the internal list is empty
-  bool Empty() const { return seqno_time_mapping_.empty(); }
+  bool Empty() const { return pairs_.empty(); }
 
   // clear all entries
-  void Clear() { seqno_time_mapping_.clear(); }
+  void Clear() { pairs_.clear(); }
 
   // return the string for user message
   // Note: Not efficient, okay for print
@@ -156,7 +188,7 @@ class SeqnoToTimeMapping {
 
 #ifndef NDEBUG
   const std::deque<SeqnoTimePair>& TEST_GetInternalMapping() const {
-    return seqno_time_mapping_;
+    return pairs_;
   }
 #endif
 
@@ -167,7 +199,7 @@ class SeqnoToTimeMapping {
   uint64_t max_time_duration_;
   uint64_t max_capacity_;
 
-  std::deque<SeqnoTimePair> seqno_time_mapping_;
+  std::deque<SeqnoTimePair> pairs_;
 
   bool is_sorted_ = true;
 
@@ -176,14 +208,14 @@ class SeqnoToTimeMapping {
 
   SeqnoTimePair& Last() {
     assert(!Empty());
-    return seqno_time_mapping_.back();
+    return pairs_.back();
   }
+
+  using pair_const_iterator =
+      std::deque<SeqnoToTimeMapping::SeqnoTimePair>::const_iterator;
+  pair_const_iterator FindGreaterTime(uint64_t time) const;
+  pair_const_iterator FindGreaterSeqno(SequenceNumber seqno) const;
+  pair_const_iterator FindGreaterEqSeqno(SequenceNumber seqno) const;
 };
 
-// for searching the sequence number from SeqnoToTimeMapping
-inline bool operator<(const SequenceNumber& seqno,
-                      const SeqnoToTimeMapping::SeqnoTimePair& other) {
-  return seqno < other.seqno;
-}
-
 }  // namespace ROCKSDB_NAMESPACE