Avoid adding tombstones of the same file to RangeDelAggregator multiple times

Summary:
RangeDelAggregator will remember the files whose range tombstones have been added,
so the caller can check whether the file has been added before call AddTombstones.

Closes https://github.com/facebook/rocksdb/pull/3635

Differential Revision: D7354604

Pulled By: ajkr

fbshipit-source-id: 9b9f7ec130556028df417e650711554b46d8d107

db/range_del_aggregator.cc

@@ -536,4 +536,11 @@ bool RangeDelAggregator::IsEmpty() {
   return true;
 }
 
+bool RangeDelAggregator::AddFile(uint64_t file_number) {
+  if (added_files_ == nullptr) {
+    added_files_.reset(new std::set<uint64_t>());
+  }
+  return added_files_->emplace(file_number).second;
+}
+
 }  // namespace rocksdb

db/range_del_aggregator.h

@@ -6,6 +6,7 @@
 #pragma once
 
 #include <map>
+#include <set>
 #include <string>
 #include <vector>
 
@@ -140,6 +141,7 @@ class RangeDelAggregator {
                     CompactionIterationStats* range_del_out_stats = nullptr,
                     bool bottommost_level = false);
   bool IsEmpty();
+  bool AddFile(uint64_t file_number);
 
  private:
   // Maps tombstone user start key -> tombstone object
@@ -180,6 +182,10 @@ class RangeDelAggregator {
   const InternalKeyComparator& icmp_;
   // collapse range deletions so they're binary searchable
   const bool collapse_deletions_;
+
+  // Record files whose tombstones have been added, to avoid duplicate adding.
+  // Same as rep_, we initializes it lazily.
+  std::unique_ptr<std::set<uint64_t>> added_files_;
 };
 
 }  // namespace rocksdb

db/table_cache.cc

@@ -247,6 +247,7 @@ InternalIterator* TableCache::NewIterator(
     }
   }
   if (s.ok() && range_del_agg != nullptr && !options.ignore_range_deletions) {
+    if (range_del_agg->AddFile(fd.GetNumber())) {
       std::unique_ptr<InternalIterator> range_del_iter(
           table_reader->NewRangeTombstoneIterator(options));
       if (range_del_iter != nullptr) {
@@ -256,6 +257,7 @@ InternalIterator* TableCache::NewIterator(
         s = range_del_agg->AddTombstones(std::move(range_del_iter));
       }
     }
+  }
 
   if (handle != nullptr) {
     ReleaseHandle(handle);