kv/RocksDBStore: use bounded iterators in rm_range_keys

Fixes: https://tracker.ceph.com/issues/58463
Signed-off-by: Igor Fedotov <igor.fedotov@croit.io>
(cherry picked from commit b8e669d)

src/kv/RocksDBStore.cc

@@ -1721,7 +1721,7 @@ void RocksDBStore::RocksDBTransactionImpl::rmkeys_by_prefix(const string &prefix
       uint64_t cnt = db->get_delete_range_threshold();
       bat.SetSavePoint();
       auto it = db->new_shard_iterator(cf);
-      for (it->SeekToFirst(); it->Valid() && (--cnt) != 0; it->Next()) {
+      for (it->seek_to_first(); it->valid() && (--cnt) != 0; it->next()) {
 	bat.Delete(cf, it->key());
       }
       if (cnt == 0) {
@@ -1739,10 +1739,10 @@ void RocksDBStore::RocksDBTransactionImpl::rm_range_keys(const string &prefix,
                                                          const string &start,
                                                          const string &end)
 {
-  ldout(db->cct, 10) << __func__ 
+  ldout(db->cct, 10) << __func__
                      << " enter prefix=" << prefix
-                     << " start=" << start
-		     << " end=" << end << dendl;
+                     << " start=" << pretty_binary_string(start)
+		     << " end=" << pretty_binary_string(end) << dendl;
   auto p_iter = db->cf_handles.find(prefix);
   uint64_t cnt = db->get_delete_range_threshold();
   if (p_iter == db->cf_handles.end()) {
@@ -1754,8 +1754,8 @@ void RocksDBStore::RocksDBTransactionImpl::rm_range_keys(const string &prefix,
 	 it->next()) {
       bat.Delete(db->default_cf, combine_strings(prefix, it->key()));
     }
-    ldout(db->cct, 15) << __func__ << " count = "
-                       << cnt0 - cnt
+    ldout(db->cct, 15) << __func__
+                       << " count = " << cnt0 - cnt
                        << dendl;
     if (cnt == 0) {
       ldout(db->cct, 10) << __func__ << " p_iter == end(), resorting to DeleteRange"
@@ -1775,20 +1775,22 @@ void RocksDBStore::RocksDBTransactionImpl::rm_range_keys(const string &prefix,
 	bat.DeleteRange(cf, rocksdb::Slice(start), rocksdb::Slice(end));
     }
   } else {
+    auto bounds = KeyValueDB::IteratorBounds();
+    bounds.lower_bound = start;
+    bounds.upper_bound = end;
     ceph_assert(p_iter->second.handles.size() >= 1);
     for (auto cf : p_iter->second.handles) {
       cnt = db->get_delete_range_threshold();
       uint64_t cnt0 = cnt;
       bat.SetSavePoint();
-      rocksdb::Iterator* it = db->new_shard_iterator(cf);
-      ceph_assert(it != nullptr);
-      for (it->Seek(start);
-	   it->Valid() && db->comparator->Compare(it->key(), end) < 0 && (--cnt) != 0;
-	   it->Next()) {
+      auto it = db->new_shard_iterator(cf, prefix, bounds);
+      for (it->lower_bound(start);
+	   it->valid() && (--cnt) != 0;
+	   it->next()) {
 	bat.Delete(cf, it->key());
       }
-      ldout(db->cct, 10) << __func__ << " count = "
-                         << cnt0 - cnt
+      ldout(db->cct, 10) << __func__
+                         << " count = " << cnt0 - cnt
                          << dendl;
       if (cnt == 0) {
         ldout(db->cct, 10) << __func__ << " p_iter != end(), resorting to DeleteRange"
@@ -1798,7 +1800,6 @@ void RocksDBStore::RocksDBTransactionImpl::rm_range_keys(const string &prefix,
       } else {
 	bat.PopSavePoint();
       }
-      delete it;
     }
   }
   ldout(db->cct, 10) << __func__ << " end" << dendl;
@@ -3024,9 +3025,23 @@ KeyValueDB::Iterator RocksDBStore::get_iterator(const std::string& prefix, Itera
   }
 }
 
-rocksdb::Iterator* RocksDBStore::new_shard_iterator(rocksdb::ColumnFamilyHandle* cf)
+RocksDBStore::WholeSpaceIterator RocksDBStore::new_shard_iterator(rocksdb::ColumnFamilyHandle* cf)
+{
+  return std::make_shared<RocksDBWholeSpaceIteratorImpl>(
+    this,
+    cf,
+    0);
+}
+
+KeyValueDB::Iterator RocksDBStore::new_shard_iterator(rocksdb::ColumnFamilyHandle* cf,
+						      const std::string& prefix,
+						      IteratorBounds bounds)
 {
-  return db->NewIterator(rocksdb::ReadOptions(), cf);
+  return std::make_shared<CFIteratorImpl>(
+    this,
+    prefix,
+    cf,
+    std::move(bounds));
 }
 
 RocksDBStore::WholeSpaceIterator RocksDBStore::get_wholespace_iterator(IteratorOpts opts)

src/kv/RocksDBStore.h

@@ -389,7 +389,9 @@ class RocksDBStore : public KeyValueDB {
   Iterator get_iterator(const std::string& prefix, IteratorOpts opts = 0, IteratorBounds = IteratorBounds()) override;
 private:
   /// this iterator spans single cf
-  rocksdb::Iterator* new_shard_iterator(rocksdb::ColumnFamilyHandle* cf);
+  WholeSpaceIterator new_shard_iterator(rocksdb::ColumnFamilyHandle* cf);
+  Iterator new_shard_iterator(rocksdb::ColumnFamilyHandle* cf,
+			      const std::string& prefix, IteratorBounds bound);
 public:
   /// Utility
   static std::string combine_strings(const std::string &prefix, const std::string &value) {