From e004b9c78b026619831eded166e974b0a7c38e14 Mon Sep 17 00:00:00 2001
From: Adam Kupczyk <akupczyk@redhat.com>
Date: Thu, 9 Apr 2020 11:46:07 +0200
Subject: [PATCH] kv/RocksDBStore: Extended compact_range on column families

Modified compact_range to column families.
Now specified range is applied to all appropried shards.

Signed-off-by: Adam Kupczyk <akupczyk@redhat.com>
---
 src/kv/RocksDBStore.cc | 39 ++++++++++++++++++++++++++++++++++++++-
 1 file changed, 38 insertions(+), 1 deletion(-)

diff --git a/src/kv/RocksDBStore.cc b/src/kv/RocksDBStore.cc
index a47b51fa93fa3..6442273c7621d 100644
--- a/src/kv/RocksDBStore.cc
+++ b/src/kv/RocksDBStore.cc
@@ -1732,12 +1732,49 @@ bool RocksDBStore::check_omap_dir(string &omap_dir)
   db = nullptr;
   return status.ok();
 }
+
 void RocksDBStore::compact_range(const string& start, const string& end)
 {
   rocksdb::CompactRangeOptions options;
   rocksdb::Slice cstart(start);
   rocksdb::Slice cend(end);
-  db->CompactRange(options, &cstart, &cend);
+  string prefix_start, key_start;
+  string prefix_end, key_end;
+  string key_highest = "\xff\xff\xff\xff"; //cheating
+  string key_lowest = "";
+
+  auto compact_range = [&] (const decltype(cf_handles)::iterator column_it,
+			    const std::string& start,
+			    const std::string& end) {
+    rocksdb::Slice cstart(start);
+    rocksdb::Slice cend(end);
+    for (const auto& shard_it : column_it->second.handles) {
+      db->CompactRange(options, shard_it, &cstart, &cend);
+    }
+  };
+  db->CompactRange(options, default_cf, &cstart, &cend);
+  split_key(cstart, &prefix_start, &key_start);
+  split_key(cend, &prefix_end, &key_end);
+  if (prefix_start == prefix_end) {
+    const auto& column = cf_handles.find(prefix_start);
+    if (column != cf_handles.end()) {
+      compact_range(column, key_start, key_end);
+    }
+  } else {
+    auto column = cf_handles.find(prefix_start);
+    if (column != cf_handles.end()) {
+      compact_range(column, key_start, key_highest);
+      ++column;
+    }
+    const auto& column_end = cf_handles.find(prefix_end);
+    while (column != column_end) {
+      compact_range(column, key_lowest, key_highest);
+      column++;
+    }
+    if (column != cf_handles.end()) {
+      compact_range(column, key_lowest, key_end);
+    }
+  }
 }
 
 RocksDBStore::RocksDBWholeSpaceIteratorImpl::~RocksDBWholeSpaceIteratorImpl()
-- 
2.39.5