Fix ASAN: avoid arena mode with range deletions

db/external_sst_file_ingestion_job.cc

@@ -381,13 +381,16 @@ Status ExternalSstFileIngestionJob::IngestedFilesOverlapWithMemtables(
   sv->imm->AddIterators(ro, &merge_iter_builder);
   ScopedArenaIterator memtable_iter(merge_iter_builder.Finish());
 
-  MergeIteratorBuilder merge_range_del_iter_builder(
-      &cfd_->internal_comparator(), &arena);
-  merge_range_del_iter_builder.AddIterator(
-      sv->mem->NewRangeTombstoneIterator(ro));
-  sv->imm->AddRangeTombstoneIterators(ro, &merge_range_del_iter_builder);
-  ScopedArenaIterator memtable_range_del_iter(
-      merge_range_del_iter_builder.Finish());
+  std::vector<InternalIterator*> memtable_range_del_iters;
+  auto* active_range_del_iter = sv->mem->NewRangeTombstoneIterator(ro);
+  if (active_range_del_iter != nullptr) {
+    memtable_range_del_iters.push_back(active_range_del_iter);
+  }
+  sv->imm->AddRangeTombstoneIterators(ro, &memtable_range_del_iters);
+  std::unique_ptr<InternalIterator> memtable_range_del_iter(NewMergingIterator(
+      &cfd_->internal_comparator(),
+      memtable_range_del_iters.empty() ? nullptr : &memtable_range_del_iters[0],
+      static_cast<int>(memtable_range_del_iters.size())));
 
   Status status;
   *overlap = false;
@@ -632,15 +635,18 @@ Status ExternalSstFileIngestionJob::IngestedFileOverlapWithLevel(
   ro.total_order_seek = true;
   MergeIteratorBuilder merge_iter_builder(&cfd_->internal_comparator(),
                                           &arena);
-  MergeIteratorBuilder merge_range_del_iter_builder(
-      &cfd_->internal_comparator(), &arena);
   sv->current->AddIteratorsForLevel(ro, env_options_, &merge_iter_builder, lvl,
                                     nullptr /* range_del_agg */);
-  sv->current->AddRangeDelIteratorsForLevel(ro, env_options_,
-                                            &merge_range_del_iter_builder, lvl);
   ScopedArenaIterator level_iter(merge_iter_builder.Finish());
-  ScopedArenaIterator level_range_del_iter(
-      merge_range_del_iter_builder.Finish());
+
+  std::vector<InternalIterator*> level_range_del_iters;
+  sv->current->AddRangeDelIteratorsForLevel(ro, env_options_, lvl,
+                                            &level_range_del_iters);
+  std::unique_ptr<InternalIterator> level_range_del_iter(NewMergingIterator(
+      &cfd_->internal_comparator(),
+      level_range_del_iters.empty() ? nullptr : &level_range_del_iters[0],
+      static_cast<int>(level_range_del_iters.size())));
+
   Status status = IngestedFileOverlapWithIteratorRange(
       file_to_ingest, level_iter.get(), overlap_with_level);
   if (status.ok() && *overlap_with_level == false) {

db/memtable_list.cc

@@ -169,9 +169,13 @@ Status MemTableListVersion::AddRangeTombstoneIterators(
 }
 
 Status MemTableListVersion::AddRangeTombstoneIterators(
-    const ReadOptions& read_opts, MergeIteratorBuilder* merge_iter_builder) {
+    const ReadOptions& read_opts,
+    std::vector<InternalIterator*>* range_del_iters) {
   for (auto& m : memlist_) {
-    merge_iter_builder->AddIterator(m->NewRangeTombstoneIterator(read_opts));
+    auto* range_del_iter = m->NewRangeTombstoneIterator(read_opts);
+    if (range_del_iter != nullptr) {
+      range_del_iters->push_back(range_del_iter);
+    }
   }
   return Status::OK();
 }

db/memtable_list.h

@@ -84,8 +84,9 @@ class MemTableListVersion {
 
   Status AddRangeTombstoneIterators(const ReadOptions& read_opts, Arena* arena,
                                     RangeDelAggregator* range_del_agg);
-  Status AddRangeTombstoneIterators(const ReadOptions& read_opts,
-                                    MergeIteratorBuilder* merge_iter_builder);
+  Status AddRangeTombstoneIterators(
+      const ReadOptions& read_opts,
+      std::vector<InternalIterator*>* range_del_iters);
 
   void AddIterators(const ReadOptions& options,
                     std::vector<InternalIterator*>* iterator_list,

db/version_set.cc

@@ -866,15 +866,18 @@ void Version::AddIteratorsForLevel(const ReadOptions& read_options,
 }
 
 void Version::AddRangeDelIteratorsForLevel(
-    const ReadOptions& read_options, const EnvOptions& soptions,
-    MergeIteratorBuilder* merge_iter_builder, int level) {
+    const ReadOptions& read_options, const EnvOptions& soptions, int level,
+    std::vector<InternalIterator*>* range_del_iters) {
+  range_del_iters->clear();
   for (size_t i = 0; i < storage_info_.LevelFilesBrief(level).num_files; i++) {
     const auto& file = storage_info_.LevelFilesBrief(level).files[i];
-    merge_iter_builder->AddIterator(
-        cfd_->table_cache()->NewRangeTombstoneIterator(
-            read_options, soptions, cfd_->internal_comparator(), file.fd,
-            cfd_->internal_stats()->GetFileReadHist(level),
-            false /* skip_filters */, level));
+    auto* range_del_iter = cfd_->table_cache()->NewRangeTombstoneIterator(
+        read_options, soptions, cfd_->internal_comparator(), file.fd,
+        cfd_->internal_stats()->GetFileReadHist(level),
+        false /* skip_filters */, level);
+    if (range_del_iter != nullptr) {
+      range_del_iters->push_back(range_del_iter);
+    }
   }
 }
 

db/version_set.h

@@ -463,10 +463,9 @@ class Version {
                             MergeIteratorBuilder* merger_iter_builder,
                             int level, RangeDelAggregator* range_del_agg);
 
-  void AddRangeDelIteratorsForLevel(const ReadOptions& read_options,
-                                    const EnvOptions& soptions,
-                                    MergeIteratorBuilder* merge_iter_builder,
-                                    int level);
+  void AddRangeDelIteratorsForLevel(
+      const ReadOptions& read_options, const EnvOptions& soptions, int level,
+      std::vector<InternalIterator*>* range_del_iters);
 
   // Lookup the value for key.  If found, store it in *val and
   // return OK.  Else return a non-OK status.