Fix possible double-free on TruncatedRangeDelIterator

Summary: Not sure where or how it happens, but using a recent CircleCI
failure I got a reliable db_stress reproducer.

Using std::unique_ptr appropriately for managing them has apparently
(and unsurprisingly) fixed the problem without needing to know exactly
where the problem was.

Test Plan: Reproducer passes, plus ASAN test and crash test runs:

```
./db_stress --WAL_size_limit_MB=1 --WAL_ttl_seconds=60 --acquire_snapshot_one_in=10000 --adaptive_readahead=1 --adm_policy=2 --advise_random_on_open=1 --allow_data_in_errors=True --allow_fallocate=1 --async_io=0 --auto_readahead_size=1 --avoid_flush_during_recovery=0 --avoid_flush_during_shutdown=1 --avoid_unnecessary_blocking_io=1 --backup_max_size=104857600 --backup_one_in=100000 --batch_protection_bytes_per_key=0 --bgerror_resume_retry_interval=1000000 --block_align=1 --block_protection_bytes_per_key=4 --block_size=16384 --bloom_before_level=2147483646 --bloom_bits=15 --bottommost_compression_type=none --bottommost_file_compaction_delay=3600 --bytes_per_sync=262144 --cache_index_and_filter_blocks=0 --cache_index_and_filter_blocks_with_high_priority=0 --cache_size=33554432 --cache_type=tiered_lru_cache --charge_compression_dictionary_building_buffer=0 --charge_file_metadata=1 --charge_filter_construction=0 --charge_table_reader=0 --check_multiget_consistency=1 --check_multiget_entity_consistency=1 --checkpoint_one_in=10000 --checksum_type=kxxHash --clear_column_family_one_in=0 --compact_files_one_in=1000000 --compact_range_one_in=1000 --compaction_pri=0 --compaction_readahead_size=0 --compaction_ttl=0 --compress_format_version=2 --compressed_secondary_cache_ratio=0.2 --compressed_secondary_cache_size=0 --compression_checksum=0 --compression_max_dict_buffer_bytes=0 --compression_max_dict_bytes=0 --compression_parallel_threads=1 --compression_type=none --compression_use_zstd_dict_trainer=0 --compression_zstd_max_train_bytes=0 --continuous_verification_interval=0 --daily_offpeak_time_utc= --data_block_index_type=0 --db=/dev/shm/rocksdb.gpxs/rocksdb_crashtest_blackbox --db_write_buffer_size=0 --default_temperature=kWarm --default_write_temperature=kCold --delete_obsolete_files_period_micros=21600000000 --delpercent=4 --delrangepercent=1 --destroy_db_initially=0 --detect_filter_construct_corruption=0 --disable_file_deletions_one_in=10000 --disable_manual_compaction_one_in=1000000 --disable_wal=0 --dump_malloc_stats=1 --enable_checksum_handoff=1 --enable_compaction_filter=0 --enable_custom_split_merge=0 --enable_do_not_compress_roles=0 --enable_index_compression=0 --enable_memtable_insert_with_hint_prefix_extractor=0 --enable_pipelined_write=1 --enable_sst_partitioner_factory=0 --enable_thread_tracking=1 --enable_write_thread_adaptive_yield=0 --error_recovery_with_no_fault_injection=0 --expected_values_dir=/dev/shm/rocksdb.gpxs/rocksdb_crashtest_expected --fail_if_options_file_error=0 --fifo_allow_compaction=0 --file_checksum_impl=none --fill_cache=1 --flush_one_in=1000000 --format_version=3 --get_all_column_family_metadata_one_in=1000000 --get_current_wal_file_one_in=0 --get_live_files_apis_one_in=10000 --get_properties_of_all_tables_one_in=100000 --get_property_one_in=100000 --get_sorted_wal_files_one_in=0 --hard_pending_compaction_bytes_limit=274877906944 --high_pri_pool_ratio=0 --index_block_restart_interval=4 --index_shortening=0 --index_type=0 --ingest_external_file_one_in=0 --initial_auto_readahead_size=16384 --inplace_update_support=0 --iterpercent=10 --key_len_percent_dist=1,30,69 --key_may_exist_one_in=100 --last_level_temperature=kHot --level_compaction_dynamic_level_bytes=0 --lock_wal_one_in=1000000 --log_file_time_to_roll=0 --log_readahead_size=0 --long_running_snapshots=1 --low_pri_pool_ratio=0 --lowest_used_cache_tier=2 --manifest_preallocation_size=5120 --manual_wal_flush_one_in=1000 --mark_for_compaction_one_file_in=10 --max_auto_readahead_size=16384 --max_background_compactions=20 --max_bytes_for_level_base=10485760 --max_key=2500000 --max_key_len=3 --max_log_file_size=0 --max_manifest_file_size=1073741824 --max_sequential_skip_in_iterations=1 --max_total_wal_size=0 --max_write_batch_group_size_bytes=16 --max_write_buffer_number=3 --max_write_buffer_size_to_maintain=0 --memtable_insert_hint_per_batch=1 --memtable_max_range_deletions=100 --memtable_prefix_bloom_size_ratio=0 --memtable_protection_bytes_per_key=4 --memtable_whole_key_filtering=0 --memtablerep=skip_list --metadata_charge_policy=0 --metadata_read_fault_one_in=32 --metadata_write_fault_one_in=0 --min_write_buffer_number_to_merge=2 --mmap_read=1 --mock_direct_io=False --nooverwritepercent=1 --num_file_reads_for_auto_readahead=0 --open_files=100 --open_metadata_read_fault_one_in=0 --open_metadata_write_fault_one_in=8 --open_read_fault_one_in=0 --open_write_fault_one_in=16 --ops_per_thread=100000000 --optimize_filters_for_hits=1 --optimize_filters_for_memory=0 --optimize_multiget_for_io=1 --paranoid_file_checks=1 --partition_filters=0 --partition_pinning=1 --pause_background_one_in=1000000 --periodic_compaction_seconds=0 --prefix_size=-1 --prefixpercent=0 --prepopulate_block_cache=1 --preserve_internal_time_seconds=60 --progress_reports=0 --promote_l0_one_in=0 --read_amp_bytes_per_bit=0 --read_fault_one_in=32 --readahead_size=524288 --readpercent=50 --recycle_log_file_num=1 --reopen=0 --report_bg_io_stats=1 --reset_stats_one_in=10000 --sample_for_compression=5 --secondary_cache_fault_one_in=32 --secondary_cache_uri= --set_options_one_in=10000 --skip_stats_update_on_db_open=0 --snapshot_hold_ops=100000 --soft_pending_compaction_bytes_limit=68719476736 --sqfc_name=bar --sqfc_version=1 --sst_file_manager_bytes_per_sec=104857600 --sst_file_manager_bytes_per_truncate=0 --stats_dump_period_sec=0 --stats_history_buffer_size=1048576 --strict_bytes_per_sync=1 --subcompactions=3 --sync=0 --sync_fault_injection=1 --table_cache_numshardbits=0 --target_file_size_base=524288 --target_file_size_multiplier=2 --test_batches_snapshots=0 --test_cf_consistency=1 --top_level_index_pinning=1 --uncache_aggressiveness=5 --universal_max_read_amp=-1 --unpartitioned_pinning=2 --use_adaptive_mutex=0 --use_adaptive_mutex_lru=0 --use_attribute_group=1 --use_delta_encoding=1 --use_direct_io_for_flush_and_compaction=0 --use_direct_reads=0 --use_full_merge_v1=0 --use_get_entity=0 --use_merge=0 --use_multi_cf_iterator=0 --use_multi_get_entity=0 --use_multiget=1 --use_put_entity_one_in=1 --use_sqfc_for_range_queries=1 --use_timed_put_one_in=0 --use_write_buffer_manager=0 --user_timestamp_size=0 --value_size_mult=32 --verification_only=0 --verify_checksum=1 --verify_checksum_one_in=1000000 --verify_compression=1 --verify_db_one_in=100000 --verify_file_checksums_one_in=0 --verify_iterator_with_expected_state_one_in=0 --verify_sst_unique_id_in_manifest=1 --wal_bytes_per_sync=0 --wal_compression=none --write_buffer_size=1048576 --write_dbid_to_manifest=1 --write_fault_one_in=0 --writepercent=35
```

db/arena_wrapped_db_iter.cc

@@ -138,10 +138,10 @@ Status ArenaWrappedDBIter::Refresh(const Snapshot* snapshot) {
             reinit_internal_iter();
             break;
           } else {
-            delete *memtable_range_tombstone_iter_;
-            *memtable_range_tombstone_iter_ = new TruncatedRangeDelIterator(
-                std::unique_ptr<FragmentedRangeTombstoneIterator>(t),
-                &cfd->internal_comparator(), nullptr, nullptr);
+            *memtable_range_tombstone_iter_ =
+                std::make_unique<TruncatedRangeDelIterator>(
+                    std::unique_ptr<FragmentedRangeTombstoneIterator>(t),
+                    &cfd->internal_comparator(), nullptr, nullptr);
           }
         }
         db_impl->ReturnAndCleanupSuperVersion(cfd, sv);

db/arena_wrapped_db_iter.h

@@ -55,7 +55,8 @@ class ArenaWrappedDBIter : public Iterator {
     db_iter_->SetIter(iter);
   }
 
-  void SetMemtableRangetombstoneIter(TruncatedRangeDelIterator** iter) {
+  void SetMemtableRangetombstoneIter(
+      std::unique_ptr<TruncatedRangeDelIterator>* iter) {
     memtable_range_tombstone_iter_ = iter;
   }
 
@@ -110,7 +111,8 @@ class ArenaWrappedDBIter : public Iterator {
   bool allow_refresh_ = true;
   // If this is nullptr, it means the mutable memtable does not contain range
   // tombstone when added under this DBIter.
-  TruncatedRangeDelIterator** memtable_range_tombstone_iter_ = nullptr;
+  std::unique_ptr<TruncatedRangeDelIterator>* memtable_range_tombstone_iter_ =
+      nullptr;
 };
 
 // Generate the arena wrapped iterator class.

db/db_impl/db_impl.cc

@@ -2064,19 +2064,19 @@ InternalIterator* DBImpl::NewInternalIterator(
       read_options, super_version->GetSeqnoToTimeMapping(), arena);
   Status s;
   if (!read_options.ignore_range_deletions) {
-    TruncatedRangeDelIterator* mem_tombstone_iter = nullptr;
+    std::unique_ptr<TruncatedRangeDelIterator> mem_tombstone_iter;
     auto range_del_iter = super_version->mem->NewRangeTombstoneIterator(
         read_options, sequence, false /* immutable_memtable */);
     if (range_del_iter == nullptr || range_del_iter->empty()) {
       delete range_del_iter;
     } else {
-      mem_tombstone_iter = new TruncatedRangeDelIterator(
+      mem_tombstone_iter = std::make_unique<TruncatedRangeDelIterator>(
           std::unique_ptr<FragmentedRangeTombstoneIterator>(range_del_iter),
           &cfd->ioptions()->internal_comparator, nullptr /* smallest */,
           nullptr /* largest */);
     }
-    merge_iter_builder.AddPointAndTombstoneIterator(mem_iter,
-                                                    mem_tombstone_iter);
+    merge_iter_builder.AddPointAndTombstoneIterator(
+        mem_iter, std::move(mem_tombstone_iter));
   } else {
     merge_iter_builder.AddIterator(mem_iter);
   }

db/memtable_list.cc

@@ -235,19 +235,19 @@ void MemTableListVersion::AddIterators(
       SequenceNumber read_seq = options.snapshot != nullptr
                                     ? options.snapshot->GetSequenceNumber()
                                     : kMaxSequenceNumber;
-      TruncatedRangeDelIterator* mem_tombstone_iter = nullptr;
+      std::unique_ptr<TruncatedRangeDelIterator> mem_tombstone_iter;
       auto range_del_iter = m->NewRangeTombstoneIterator(
           options, read_seq, true /* immutale_memtable */);
       if (range_del_iter == nullptr || range_del_iter->empty()) {
         delete range_del_iter;
       } else {
-        mem_tombstone_iter = new TruncatedRangeDelIterator(
+        mem_tombstone_iter = std::make_unique<TruncatedRangeDelIterator>(
             std::unique_ptr<FragmentedRangeTombstoneIterator>(range_del_iter),
             &m->GetInternalKeyComparator(), nullptr /* smallest */,
             nullptr /* largest */);
       }
-      merge_iter_builder->AddPointAndTombstoneIterator(mem_iter,
-                                                       mem_tombstone_iter);
+      merge_iter_builder->AddPointAndTombstoneIterator(
+          mem_iter, std::move(mem_tombstone_iter));
     }
   }
 }

db/table_cache.cc

@@ -230,7 +230,7 @@ InternalIterator* TableCache::NewIterator(
     const InternalKey* smallest_compaction_key,
     const InternalKey* largest_compaction_key, bool allow_unprepared_value,
     uint8_t block_protection_bytes_per_key, const SequenceNumber* read_seqno,
-    TruncatedRangeDelIterator** range_del_iter) {
+    std::unique_ptr<TruncatedRangeDelIterator>* range_del_iter) {
   PERF_TIMER_GUARD(new_table_iterator_nanos);
 
   Status s;
@@ -285,7 +285,7 @@ InternalIterator* TableCache::NewIterator(
         delete new_range_del_iter;
         *range_del_iter = nullptr;
       } else {
-        *range_del_iter = new TruncatedRangeDelIterator(
+        *range_del_iter = std::make_unique<TruncatedRangeDelIterator>(
             std::unique_ptr<FragmentedRangeTombstoneIterator>(
                 new_range_del_iter),
             &icomparator, &file_meta.smallest, &file_meta.largest);

db/table_cache.h

@@ -100,7 +100,7 @@ class TableCache {
       const InternalKey* largest_compaction_key, bool allow_unprepared_value,
       uint8_t protection_bytes_per_key,
       const SequenceNumber* range_del_read_seqno = nullptr,
-      TruncatedRangeDelIterator** range_del_iter = nullptr);
+      std::unique_ptr<TruncatedRangeDelIterator>* range_del_iter = nullptr);
 
   // If a seek to internal key "k" in specified file finds an entry,
   // call get_context->SaveValue() repeatedly until

db/version_set.cc

@@ -976,7 +976,8 @@ class LevelIterator final : public InternalIterator {
       const std::vector<AtomicCompactionUnitBoundary>* compaction_boundaries =
           nullptr,
       bool allow_unprepared_value = false,
-      TruncatedRangeDelIterator**** range_tombstone_iter_ptr_ = nullptr)
+      std::unique_ptr<TruncatedRangeDelIterator>*** range_tombstone_iter_ptr_ =
+          nullptr)
       : table_cache_(table_cache),
         read_options_(read_options),
         file_options_(file_options),
@@ -1116,9 +1117,8 @@ class LevelIterator final : public InternalIterator {
   }
 
   void ClearRangeTombstoneIter() {
-    if (range_tombstone_iter_ && *range_tombstone_iter_) {
-      delete *range_tombstone_iter_;
-      *range_tombstone_iter_ = nullptr;
+    if (range_tombstone_iter_) {
+      range_tombstone_iter_->reset();
     }
   }
 
@@ -1201,7 +1201,7 @@ class LevelIterator final : public InternalIterator {
   // iterator end).
   //
   // *range_tombstone_iter_ points to range tombstones of the current SST file
-  TruncatedRangeDelIterator** range_tombstone_iter_;
+  std::unique_ptr<TruncatedRangeDelIterator>* range_tombstone_iter_;
 
   // The sentinel key to be returned
   Slice sentinel_;
@@ -1929,7 +1929,7 @@ InternalIterator* Version::TEST_GetLevelIterator(
     int level, bool allow_unprepared_value) {
   auto* arena = merge_iter_builder->GetArena();
   auto* mem = arena->AllocateAligned(sizeof(LevelIterator));
-  TruncatedRangeDelIterator*** tombstone_iter_ptr = nullptr;
+  std::unique_ptr<TruncatedRangeDelIterator>** tombstone_iter_ptr = nullptr;
   auto level_iter = new (mem) LevelIterator(
       cfd_->table_cache(), read_options, file_options_,
       cfd_->internal_comparator(), &storage_info_.LevelFilesBrief(level),
@@ -2029,7 +2029,7 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
   auto* arena = merge_iter_builder->GetArena();
   if (level == 0) {
     // Merge all level zero files together since they may overlap
-    TruncatedRangeDelIterator* tombstone_iter = nullptr;
+    std::unique_ptr<TruncatedRangeDelIterator> tombstone_iter = nullptr;
     for (size_t i = 0; i < storage_info_.LevelFilesBrief(0).num_files; i++) {
       const auto& file = storage_info_.LevelFilesBrief(0).files[i];
       auto table_iter = cfd_->table_cache()->NewIterator(
@@ -2046,8 +2046,8 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
       if (read_options.ignore_range_deletions) {
         merge_iter_builder->AddIterator(table_iter);
       } else {
-        merge_iter_builder->AddPointAndTombstoneIterator(table_iter,
-                                                         tombstone_iter);
+        merge_iter_builder->AddPointAndTombstoneIterator(
+            table_iter, std::move(tombstone_iter));
       }
     }
     if (should_sample) {
@@ -2064,7 +2064,7 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
     // walks through the non-overlapping files in the level, opening them
     // lazily.
     auto* mem = arena->AllocateAligned(sizeof(LevelIterator));
-    TruncatedRangeDelIterator*** tombstone_iter_ptr = nullptr;
+    std::unique_ptr<TruncatedRangeDelIterator>** tombstone_iter_ptr = nullptr;
     auto level_iter = new (mem) LevelIterator(
         cfd_->table_cache(), read_options, soptions,
         cfd_->internal_comparator(), &storage_info_.LevelFilesBrief(level),
@@ -7078,8 +7078,8 @@ InternalIterator* VersionSet::MakeInputIterator(
   // that will be initialized to where CompactionMergingIterator stores
   // pointer to its range tombstones. This is used by LevelIterator
   // to update pointer to range tombstones as it traverse different SST files.
-  std::vector<
-      std::pair<TruncatedRangeDelIterator*, TruncatedRangeDelIterator***>>
+  std::vector<std::pair<std::unique_ptr<TruncatedRangeDelIterator>,
+                        std::unique_ptr<TruncatedRangeDelIterator>**>>
       range_tombstones;
   size_t num = 0;
   for (size_t which = 0; which < c->num_input_levels(); which++) {
@@ -7101,7 +7101,8 @@ InternalIterator* VersionSet::MakeInputIterator(
                   *end, fmd.smallest.user_key()) < 0) {
             continue;
           }
-          TruncatedRangeDelIterator* range_tombstone_iter = nullptr;
+          std::unique_ptr<TruncatedRangeDelIterator> range_tombstone_iter =
+              nullptr;
           list[num++] = cfd->table_cache()->NewIterator(
               read_options, file_options_compactions,
               cfd->internal_comparator(), fmd, range_del_agg,
@@ -7118,11 +7119,13 @@ InternalIterator* VersionSet::MakeInputIterator(
               c->mutable_cf_options()->block_protection_bytes_per_key,
               /*range_del_read_seqno=*/nullptr,
               /*range_del_iter=*/&range_tombstone_iter);
-          range_tombstones.emplace_back(range_tombstone_iter, nullptr);
+          range_tombstones.emplace_back(std::move(range_tombstone_iter),
+                                        nullptr);
         }
       } else {
         // Create concatenating iterator for the files from this level
-        TruncatedRangeDelIterator*** tombstone_iter_ptr = nullptr;
+        std::unique_ptr<TruncatedRangeDelIterator>** tombstone_iter_ptr =
+            nullptr;
         list[num++] = new LevelIterator(
             cfd->table_cache(), read_options, file_options_compactions,
             cfd->internal_comparator(), c->input_levels(which),

table/compaction_merging_iterator.cc

@@ -11,8 +11,8 @@ class CompactionMergingIterator : public InternalIterator {
   CompactionMergingIterator(
       const InternalKeyComparator* comparator, InternalIterator** children,
       int n, bool is_arena_mode,
-      std::vector<
-          std::pair<TruncatedRangeDelIterator*, TruncatedRangeDelIterator***>>
+      std::vector<std::pair<std::unique_ptr<TruncatedRangeDelIterator>,
+                            std::unique_ptr<TruncatedRangeDelIterator>**>>&
           range_tombstones)
       : is_arena_mode_(is_arena_mode),
         comparator_(comparator),
@@ -27,7 +27,7 @@ class CompactionMergingIterator : public InternalIterator {
     }
     assert(range_tombstones.size() == static_cast<size_t>(n));
     for (auto& p : range_tombstones) {
-      range_tombstone_iters_.push_back(p.first);
+      range_tombstone_iters_.push_back(std::move(p.first));
     }
     pinned_heap_item_.resize(n);
     for (int i = 0; i < n; ++i) {
@@ -47,10 +47,7 @@ class CompactionMergingIterator : public InternalIterator {
   }
 
   ~CompactionMergingIterator() override {
-    // TODO: use unique_ptr for range_tombstone_iters_
-    for (auto child : range_tombstone_iters_) {
-      delete child;
-    }
+    range_tombstone_iters_.clear();
 
     for (auto& child : children_) {
       child.iter.DeleteIter(is_arena_mode_);
@@ -197,7 +194,8 @@ class CompactionMergingIterator : public InternalIterator {
   // nullptr means the sorted run of children_[i] does not have range
   // tombstones (or the current SSTable does not have range tombstones in the
   // case of LevelIterator).
-  std::vector<TruncatedRangeDelIterator*> range_tombstone_iters_;
+  std::vector<std::unique_ptr<TruncatedRangeDelIterator>>
+      range_tombstone_iters_;
   // Used as value for range tombstone keys
   std::string dummy_tombstone_val{};
 
@@ -349,8 +347,9 @@ void CompactionMergingIterator::AddToMinHeapOrCheckStatus(HeapItem* child) {
 
 InternalIterator* NewCompactionMergingIterator(
     const InternalKeyComparator* comparator, InternalIterator** children, int n,
-    std::vector<std::pair<TruncatedRangeDelIterator*,
-                          TruncatedRangeDelIterator***>>& range_tombstone_iters,
+    std::vector<std::pair<std::unique_ptr<TruncatedRangeDelIterator>,
+                          std::unique_ptr<TruncatedRangeDelIterator>**>>&
+        range_tombstone_iters,
     Arena* arena) {
   assert(n >= 0);
   if (n == 0) {

table/compaction_merging_iterator.h

@@ -38,7 +38,8 @@ class CompactionMergingIterator;
 
 InternalIterator* NewCompactionMergingIterator(
     const InternalKeyComparator* comparator, InternalIterator** children, int n,
-    std::vector<std::pair<TruncatedRangeDelIterator*,
-                          TruncatedRangeDelIterator***>>& range_tombstone_iters,
+    std::vector<std::pair<std::unique_ptr<TruncatedRangeDelIterator>,
+                          std::unique_ptr<TruncatedRangeDelIterator>**>>&
+        range_tombstone_iters,
     Arena* arena = nullptr);
 }  // namespace ROCKSDB_NAMESPACE

table/merging_iterator.cc

@@ -96,8 +96,9 @@ class MergingIterator : public InternalIterator {
   // could be updated. In that case, this merging iterator is only responsible
   // for freeing the new range tombstone iterator that it has pointers to in
   // range_tombstone_iters_.
-  void AddRangeTombstoneIterator(TruncatedRangeDelIterator* iter) {
-    range_tombstone_iters_.emplace_back(iter);
+  void AddRangeTombstoneIterator(
+      std::unique_ptr<TruncatedRangeDelIterator>&& iter) {
+    range_tombstone_iters_.emplace_back(std::move(iter));
   }
 
   // Called by MergingIteratorBuilder when all point iterators and range
@@ -125,9 +126,7 @@ class MergingIterator : public InternalIterator {
   }
 
   ~MergingIterator() override {
-    for (auto child : range_tombstone_iters_) {
-      delete child;
-    }
+    range_tombstone_iters_.clear();
 
     for (auto& child : children_) {
       child.iter.DeleteIter(is_arena_mode_);
@@ -624,7 +623,8 @@ class MergingIterator : public InternalIterator {
   // Invariant(rti): pinned_heap_item_[i] is in minHeap_ iff
   // range_tombstone_iters_[i]->Valid() and at most one pinned_heap_item_[i] is
   // in minHeap_.
-  std::vector<TruncatedRangeDelIterator*> range_tombstone_iters_;
+  std::vector<std::unique_ptr<TruncatedRangeDelIterator>>
+      range_tombstone_iters_;
 
   // Levels (indices into range_tombstone_iters_/children_ ) that currently have
   // "active" range tombstones. See comments above MergingIterator for meaning
@@ -841,7 +841,8 @@ void MergingIterator::SeekImpl(const Slice& target, size_t starting_level,
         prefetched_target.emplace_back(
             level, current_search_key.GetInternalKey().ToString());
       }
-      auto range_tombstone_iter = range_tombstone_iters_[level];
+      UnownedPtr<TruncatedRangeDelIterator> range_tombstone_iter =
+          range_tombstone_iters_[level].get();
       if (range_tombstone_iter) {
         range_tombstone_iter->SeekInternalKey(
             current_search_key.GetInternalKey());
@@ -1125,7 +1126,8 @@ void MergingIterator::SeekForPrevImpl(const Slice& target,
         prefetched_target.emplace_back(
             level, current_search_key.GetInternalKey().ToString());
       }
-      auto range_tombstone_iter = range_tombstone_iters_[level];
+      UnownedPtr<TruncatedRangeDelIterator> range_tombstone_iter =
+          range_tombstone_iters_[level].get();
       if (range_tombstone_iter) {
         range_tombstone_iter->SeekForPrev(current_search_key.GetUserKey());
         if (range_tombstone_iter->Valid()) {
@@ -1349,7 +1351,8 @@ void MergingIterator::SwitchToForward() {
     ParseInternalKey(target, &pik, false /* log_err_key */)
         .PermitUncheckedError();
     for (size_t i = 0; i < range_tombstone_iters_.size(); ++i) {
-      auto iter = range_tombstone_iters_[i];
+      UnownedPtr<TruncatedRangeDelIterator> iter =
+          range_tombstone_iters_[i].get();
       if (iter) {
         iter->Seek(pik.user_key);
         // The while loop is needed as the Seek() call above is only for user
@@ -1395,7 +1398,8 @@ void MergingIterator::SwitchToBackward() {
   ParseInternalKey(target, &pik, false /* log_err_key */)
       .PermitUncheckedError();
   for (size_t i = 0; i < range_tombstone_iters_.size(); ++i) {
-    auto iter = range_tombstone_iters_[i];
+    UnownedPtr<TruncatedRangeDelIterator> iter =
+        range_tombstone_iters_[i].get();
     if (iter) {
       iter->SeekForPrev(pik.user_key);
       // Since the SeekForPrev() call above is only for user key,
@@ -1690,8 +1694,9 @@ void MergeIteratorBuilder::AddIterator(InternalIterator* iter) {
 }
 
 void MergeIteratorBuilder::AddPointAndTombstoneIterator(
-    InternalIterator* point_iter, TruncatedRangeDelIterator* tombstone_iter,
-    TruncatedRangeDelIterator*** tombstone_iter_ptr) {
+    InternalIterator* point_iter,
+    std::unique_ptr<TruncatedRangeDelIterator>&& tombstone_iter,
+    std::unique_ptr<TruncatedRangeDelIterator>** tombstone_iter_ptr) {
   // tombstone_iter_ptr != nullptr means point_iter is a LevelIterator.
   bool add_range_tombstone = tombstone_iter ||
                              !merge_iter->range_tombstone_iters_.empty() ||
@@ -1711,7 +1716,7 @@ void MergeIteratorBuilder::AddPointAndTombstoneIterator(
              merge_iter->children_.size() - 1) {
         merge_iter->AddRangeTombstoneIterator(nullptr);
       }
-      merge_iter->AddRangeTombstoneIterator(tombstone_iter);
+      merge_iter->AddRangeTombstoneIterator(std::move(tombstone_iter));
     }
 
     if (tombstone_iter_ptr) {

table/merging_iterator.h

@@ -70,8 +70,10 @@ class MergeIteratorBuilder {
   // point iterators are not LevelIterator, then range tombstone iterator is
   // only added to the merging iter if there is a non-null `tombstone_iter`.
   void AddPointAndTombstoneIterator(
-      InternalIterator* point_iter, TruncatedRangeDelIterator* tombstone_iter,
-      TruncatedRangeDelIterator*** tombstone_iter_ptr = nullptr);
+      InternalIterator* point_iter,
+      std::unique_ptr<TruncatedRangeDelIterator>&& tombstone_iter,
+      std::unique_ptr<TruncatedRangeDelIterator>** tombstone_iter_ptr =
+          nullptr);
 
   // Get arena used to build the merging iterator. It is called one a child
   // iterator needs to be allocated.
@@ -91,7 +93,7 @@ class MergeIteratorBuilder {
   Arena* arena;
   // Used to set LevelIterator.range_tombstone_iter_.
   // See AddRangeTombstoneIterator() implementation for more detail.
-  std::vector<std::pair<size_t, TruncatedRangeDelIterator***>>
+  std::vector<std::pair<size_t, std::unique_ptr<TruncatedRangeDelIterator>**>>
       range_del_iter_ptrs_;
 };