Implement single deletion in RocksDB.

If the target deletion range contains a single key, convert it into a
single deletion, to avoid extra cost in RocksDB due to range deletion.

fdbserver/KeyValueStoreRocksDB.actor.cpp

@@ -860,7 +860,11 @@ struct RocksDBKeyValueStore : IKeyValueStore {
 			writeBatch.reset(new rocksdb::WriteBatch());
 		}
 
-		writeBatch->DeleteRange(toSlice(keyRange.begin), toSlice(keyRange.end));
+		if (keyRange.singleKeyRange()) {
+			writeBatch->Delete(toSlice(keyRange.begin));
+		} else {
+			writeBatch->DeleteRange(toSlice(keyRange.begin), toSlice(keyRange.end));
+		}
 	}
 
 	Future<Void> commit(bool) override {
@@ -1015,7 +1019,63 @@ IKeyValueStore* keyValueStoreRocksDB(std::string const& path,
 
 namespace {
 
-TEST_CASE("noSim/fdbserver/KeyValueStoreRocksDB/Reopen") {
+TEST_CASE("noSim/fdbserver/KeyValueStoreRocksDB/RocksDBBasic") {
+	state const std::string rocksDBTestDir = "rocksdb-kvstore-basic-test-db";
+	platform::eraseDirectoryRecursive(rocksDBTestDir);
+
+	state IKeyValueStore* kvStore = new RocksDBKeyValueStore(rocksDBTestDir, deterministicRandom()->randomUniqueID());
+	wait(kvStore->init());
+
+	state StringRef foo = "foo"_sr;
+	state StringRef bar = "ibar"_sr;
+	kvStore->set({ foo, foo });
+	kvStore->set({ keyAfter(foo), keyAfter(foo) });
+	kvStore->set({ bar, bar });
+	kvStore->set({ keyAfter(bar), keyAfter(bar) });
+	wait(kvStore->commit(false));
+
+	{
+		Optional<Value> val = wait(kvStore->readValue(foo));
+		ASSERT(foo == val.get());
+	}
+
+	// Test single key deletion.
+	kvStore->clear(singleKeyRange(foo));
+	wait(kvStore->commit(false));
+
+	{
+		Optional<Value> val = wait(kvStore->readValue(foo));
+		ASSERT(!val.present());
+	}
+
+	{
+		Optional<Value> val = wait(kvStore->readValue(keyAfter(foo)));
+		ASSERT(keyAfter(foo) == val.get());
+	}
+
+	// Test range deletion.
+	kvStore->clear(KeyRangeRef(keyAfter(foo), keyAfter(bar)));
+	wait(kvStore->commit(false));
+
+	{
+		Optional<Value> val = wait(kvStore->readValue(bar));
+		ASSERT(!val.present());
+	}
+
+	{
+		Optional<Value> val = wait(kvStore->readValue(keyAfter(bar)));
+		ASSERT(keyAfter(bar) == val.get());
+	}
+
+	Future<Void> closed = kvStore->onClosed();
+	kvStore->close();
+	wait(closed);
+
+	platform::eraseDirectoryRecursive(rocksDBTestDir);
+	return Void();
+}
+
+TEST_CASE("noSim/fdbserver/KeyValueStoreRocksDB/RocksDBReopen") {
 	state const std::string rocksDBTestDir = "rocksdb-kvstore-reopen-test-db";
 	platform::eraseDirectoryRecursive(rocksDBTestDir);
 

tests/CMakeLists.txt

@@ -176,6 +176,7 @@ if(WITH_PYTHON)
   add_fdb_test(TEST_FILES fast/WriteDuringRead.toml)
   add_fdb_test(TEST_FILES fast/WriteDuringReadClean.toml)
   add_fdb_test(TEST_FILES noSim/RandomUnitTests.toml UNIT)
+  add_fdb_test(TEST_FILES noSim/KeyValueStoreRocksDBTest.toml)
   add_fdb_test(TEST_FILES rare/CheckRelocation.toml)
   add_fdb_test(TEST_FILES rare/ClogUnclog.toml)
   add_fdb_test(TEST_FILES rare/CloggedCycleWithKills.toml)

tests/noSim/KeyValueStoreRocksDBTest.toml

@@ -0,0 +1,9 @@
+[[test]]
+ testTitle = 'UnitTests'
+ useDB = false
+ startDelay = 0
+
+     [[test.workload]]
+     testName = 'UnitTests'
+     maxTestCases = 10
+     testsMatching = 'noSim/fdbserver/KeyValueStoreRocksDB/'