Add KVStore::CompactRange to free disk space in HA

Summary:
Manually call `CompactRange` on KVStore (rocksDB) to free disk space
upon Raft log compaction.

Raft log takes surprisingly large amount of disk space (1MB for 1000 entries) so
we need to free disk space as we compact the log into snapshot.

When `log_size_snapshot_threshold` set to `5000` the disk usage of the
`durability_dir` goes from `4.8 MB` to `240 KB`.

Reviewers: ipaljak

Reviewed By: ipaljak

Subscribers: pullbot

Differential Revision: https://phabricator.memgraph.io/D1939

src/raft/raft_server.cpp

@@ -1018,6 +1018,10 @@ void RaftServer::SnapshotThread() {
                ++i) {
             disk_storage_.Delete(LogEntryKey(i));
           }
+          // After we deleted entries from the persistent store, make sure we
+          // compact the files and actually reduce used disk space.
+          disk_storage_.CompactRange(LogEntryKey(log_compaction_start_index),
+                                     LogEntryKey(last_included_index));
         }
       }
     }

src/storage/common/kvstore/kvstore.cpp

@@ -167,4 +167,13 @@ size_t KVStore::Size(const std::string &prefix) {
   return size;
 }
 
+bool KVStore::CompactRange(const std::string &begin_prefix,
+                           const std::string &end_prefix) {
+  rocksdb::CompactRangeOptions options;
+  rocksdb::Slice begin(begin_prefix);
+  rocksdb::Slice end(end_prefix);
+  auto s = pimpl_->db->CompactRange(options, &begin, &end);
+  return s.ok();
+}
+
 }  // namespace storage

src/storage/common/kvstore/kvstore.hpp

@@ -130,6 +130,20 @@ class KVStore final {
    */
   size_t Size(const std::string &prefix = "");
 
+  /**
+   * Compact the underlying storage for the key range [begin_prefix,
+   * end_prefix].
+   * The actual compaction interval might be superset of
+   * [begin_prefix, end_prefix].
+   *
+   * @param begin_prefix - Compaction interval start.
+   * @param end_prefix   - Compaction interval end.
+   *
+   * @return - true if the compaction finished successfully, false otherwise.
+   */
+  bool CompactRange(const std::string &begin_prefix,
+                    const std::string &end_prefix);
+
   /**
    * Custom prefix-based iterator over kvstore.
    *

src/storage/common/kvstore/kvstore_dummy.cpp

@@ -97,4 +97,10 @@ bool KVStore::iterator::IsValid() { return false; }
 
 size_t KVStore::Size(const std::string &prefix) { return 0; }
 
+bool KVStore::CompactRange(const std::string &begin_prefix,
+                           const std::string &end_prefix) {
+  CHECK(false) << "Unsupported operation (KVStore::Compact) -- this is a "
+                  "dummy kvstore";
+}
+
 }  // namespace storage