Do not delete iterators for immutable memtables.

Summary:
The immutable memtable iterators are allocated from an arena and there
is no benefit from deleting these. Also the immutable memtables
themselves will continue to be in memory until the version set
containing it is alive. We will not remove immutable memtable iterators
over the upper bound. We now add immutable iterators to the test.

Test Plan: db_tailing_iter_test.TailingIteratorTrimSeekToNext

Reviewers: tnovak, sdong

Reviewed By: sdong

Subscribers: dhruba, leveldb

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

db/db_tailing_iter_test.cc

@@ -121,7 +121,12 @@ TEST_F(DBTestTailingIterator, TailingIteratorSeekToNext) {
 }
 
 TEST_F(DBTestTailingIterator, TailingIteratorTrimSeekToNext) {
-  CreateAndReopenWithCF({"pikachu"}, CurrentOptions());
+  const uint64_t k20KB = 20 * 1024;
+  Options options;
+  options.write_buffer_size = k20KB;
+  options.max_write_buffer_number = 6;
+  options.min_write_buffer_number_to_merge = 5;
+  CreateAndReopenWithCF({"pikachu"}, options);
   ReadOptions read_options;
   read_options.tailing = true;
 

db/forward_iterator.cc

@@ -238,20 +238,11 @@ void ForwardIterator::SeekInternal(const Slice& internal_key,
     }
     for (size_t i = 0; i < imm_iters_.size(); i++) {
       auto* m = imm_iters_[i];
-      if (!m) {
-        continue;
-      }
       seek_to_first ? m->SeekToFirst() : m->Seek(internal_key);
       if (!m->status().ok()) {
         immutable_status_ = m->status();
       } else if (m->Valid()) {
-        if (!IsOverUpperBound(m->key())) {
+        immutable_min_heap_.push(m);
-          immutable_min_heap_.push(m);
-        } else {
-          has_iter_trimmed_for_upper_bound_ = true;
-          delete m;
-          imm_iters_[i] = nullptr;
-        }
       }
     }
 
@@ -624,19 +615,6 @@ bool ForwardIterator::NeedToRebuildTrimmed(const Slice& target) {
 }
 
 void ForwardIterator::DeleteCurrentIter() {
-  for (size_t i = 0; i < imm_iters_.size(); i++) {
-    auto& m = imm_iters_[i];
-    if (!m) {
-      continue;
-    }
-    if (m == current_) {
-      has_iter_trimmed_for_upper_bound_ = true;
-      delete m;
-      m = nullptr;
-      return;
-    }
-  }
-
   const VersionStorageInfo* vstorage = sv_->current->storage_info();
   const std::vector<FileMetaData*>& l0 = vstorage->LevelFiles(0);
   for (uint32_t i = 0; i < l0.size(); ++i) {
@@ -667,12 +645,6 @@ bool ForwardIterator::TEST_CheckDeletedIters() {
   if (!has_iter_trimmed_for_upper_bound_) {
     return false;
   }
-  for (size_t i = 0; i < imm_iters_.size(); i++) {
-    auto& m = imm_iters_[i];
-    if (!m) {
-      return true;
-    }
-  }
 
   const VersionStorageInfo* vstorage = sv_->current->storage_info();
   const std::vector<FileMetaData*>& l0 = vstorage->LevelFiles(0);