Abstract out SetMaxPossibleForUserKey() and SetMinPossibleForUserKey

Summary:
Based on feedback from D37083.

Are all of these correct? In some spaces it seems like we're doing SetMaxPossibleForUserKey() although we want the smallest possible internal key for user key.

Test Plan: make check

Reviewers: sdong

Reviewed By: sdong

Subscribers: dhruba, leveldb

Differential Revision: https://reviews.facebook.net/D37341

db/db_impl.cc

@@ -1686,7 +1686,7 @@ Status DBImpl::RunManualCompaction(ColumnFamilyData* cfd, int input_level,
       cfd->ioptions()->compaction_style == kCompactionStyleFIFO) {
     manual.begin = nullptr;
   } else {
-    begin_storage = InternalKey(*begin, kMaxSequenceNumber, kValueTypeForSeek);
+    begin_storage.SetMaxPossibleForUserKey(*begin);
     manual.begin = &begin_storage;
   }
   if (end == nullptr ||
@@ -1694,7 +1694,7 @@ Status DBImpl::RunManualCompaction(ColumnFamilyData* cfd, int input_level,
       cfd->ioptions()->compaction_style == kCompactionStyleFIFO) {
     manual.end = nullptr;
   } else {
-    end_storage = InternalKey(*end, 0, static_cast<ValueType>(0));
+    end_storage.SetMinPossibleForUserKey(*end);
     manual.end = &end_storage;
   }
 
@@ -3555,8 +3555,9 @@ void DBImpl::GetApproximateSizes(ColumnFamilyHandle* column_family,
 
   for (int i = 0; i < n; i++) {
     // Convert user_key into a corresponding internal key.
-    InternalKey k1(range[i].start, kMaxSequenceNumber, kValueTypeForSeek);
+    InternalKey k1, k2;
-    InternalKey k2(range[i].limit, kMaxSequenceNumber, kValueTypeForSeek);
+    k1.SetMaxPossibleForUserKey(range[i].start);
+    k2.SetMaxPossibleForUserKey(range[i].limit);
     uint64_t start = versions_->ApproximateOffsetOf(v, k1);
     uint64_t limit = versions_->ApproximateOffsetOf(v, k2);
     sizes[i] = (limit >= start ? limit - start : 0);

db/db_impl_experimental.cc

@@ -30,10 +30,10 @@ Status DBImpl::SuggestCompactRange(ColumnFamilyHandle* column_family,
   auto cfd = cfh->cfd();
   InternalKey start_key, end_key;
   if (begin != nullptr) {
-    start_key = InternalKey(*begin, kMaxSequenceNumber, kValueTypeForSeek);
+    start_key.SetMaxPossibleForUserKey(*begin);
   }
   if (end != nullptr) {
-    end_key = InternalKey(*end, 0, static_cast<ValueType>(0));
+    end_key.SetMinPossibleForUserKey(*end);
   }
   {
     InstrumentedMutexLock l(&mutex_);

db/dbformat.h

@@ -135,6 +135,20 @@ class InternalKey {
     AppendInternalKey(&rep_, ParsedInternalKey(_user_key, s, t));
   }
 
+  // sets the internal key to be bigger or equal to all internal keys with this
+  // user key
+  void SetMaxPossibleForUserKey(const Slice& _user_key) {
+    AppendInternalKey(&rep_, ParsedInternalKey(_user_key, kMaxSequenceNumber,
+                                               kValueTypeForSeek));
+  }
+
+  // sets the internal key to be smaller or equal to all internal keys with this
+  // user key
+  void SetMinPossibleForUserKey(const Slice& _user_key) {
+    AppendInternalKey(
+        &rep_, ParsedInternalKey(_user_key, 0, static_cast<ValueType>(0)));
+  }
+
   bool Valid() const {
     ParsedInternalKey parsed;
     return ParseInternalKey(Slice(rep_), &parsed);

db/version_set.cc

@@ -397,7 +397,8 @@ bool SomeFileOverlapsRange(
   uint32_t index = 0;
   if (smallest_user_key != nullptr) {
     // Find the earliest possible internal key for smallest_user_key
-    InternalKey small(*smallest_user_key, kMaxSequenceNumber,kValueTypeForSeek);
+    InternalKey small;
+    small.SetMaxPossibleForUserKey(*smallest_user_key);
     index = FindFile(icmp, file_level, small.Encode());
   }
 
@@ -1230,7 +1231,8 @@ int VersionStorageInfo::PickLevelForMemTableOutput(
   if (!OverlapInLevel(0, &smallest_user_key, &largest_user_key)) {
     // Push to next level if there is no overlap in next level,
     // and the #bytes overlapping in the level after that are limited.
-    InternalKey start(smallest_user_key, kMaxSequenceNumber, kValueTypeForSeek);
+    InternalKey start;
+    start.SetMaxPossibleForUserKey(smallest_user_key);
     InternalKey limit(largest_user_key, 0, static_cast<ValueType>(0));
     std::vector<FileMetaData*> overlaps;
     while (mutable_cf_options.max_mem_compaction_level > 0 &&

util/ldb_cmd.cc

@@ -796,7 +796,8 @@ void InternalDumpCommand::DoCommand() {
   }
 
   if (has_from_) {
-    InternalKey ikey(from_, kMaxSequenceNumber, kValueTypeForSeek);
+    InternalKey ikey;
+    ikey.SetMaxPossibleForUserKey(from_);
     iter->Seek(ikey.Encode());
   } else {
     iter->SeekToFirst();

util/sst_dump_tool.cc

@@ -187,7 +187,8 @@ Status SstFileReader::ReadSequential(bool print_kv,
                                                          false));
   uint64_t i = 0;
   if (has_from) {
-    InternalKey ikey(from_key, kMaxSequenceNumber, kValueTypeForSeek);
+    InternalKey ikey;
+    ikey.SetMaxPossibleForUserKey(from_key);
     iter->Seek(ikey.Encode());
   } else {
     iter->SeekToFirst();