Implement LHLOCDEL command

CHANGELOG.md

 # Changelog
-All notable changes to this project will be documented in this file.
 
 ## Unreleased
-- ``deque-scan-back`` was returning wrong cursor to signal end of iteration: ``next:0``
-while it should have been ``0``.
+
+### Bug fixes
+- ``deque-scan-back`` was returning the wrong cursor to signal end of
+iteration: ``next:0`` while it should have been ``0``.
 - A race condition was sometimes causing elections to fail spuriously, making
 the election of a stable leader to require slightly more rounds than it should have.
+
+### New features
 - Implementation of health indicators through ``QUARKDB-HEALTH`` command.
 - Added support for RESPv3 push types, activated on a per-client basis through
 ``ACTIVATE-PUSH-TYPES`` command.
+- Implementation of ``LHLOCDEL`` command for conditionally deleting a locality hash
+field, only if the provided hint matches.
 - Add convenience command ``DEQUE-CLEAR``.
 - Add support for ``MATCHLOC`` in ``LHSCAN``, used to filter out results based
 on locality hint.
+- Add ``RECOVERY-SCAN`` command for scanning through complete keyspace, including
+internal rocksdb keys.
+- Add tool ``quarkdb-sst-inspect`` to allow low-level inspection of SST files.
+
+### Improvements
 - Protection for a strange case of corruption which brought down a development
 test cluster. (last-applied jumped ahead of commit-index by 1024, causing all
 writes to stall). From now on, similar kind of corruption should only take out
@@ -21,11 +31,10 @@ raft journal easier.
 - ``KEYS`` is now implemented in terms of ``SCAN``, making prefix matching of the
 keyspace just as efficient as with ``SCAN``. (Note: The use of ``KEYS`` is still
 generally discouraged due to potentially huge response size)
-- Add ``RECOVERY-SCAN`` command for scanning through complete keyspace, including
-internal rocksdb keys.
-- Add tool ``quarkdb-sst-inspect`` to allow low-level inspection of SST files.
 - Removed unused tool ``quarkdb-scrub``.
 
+
+
 ## 0.3.8 (2019-05-27)
 - Prevent elections from hanging on the TCP timeout when one of the member hosts
 is dropping packets, which could bring down an otherwise healthy cluster.

src/Commands.cc

@@ -97,6 +97,7 @@ struct cmdMapInit {
     redis_cmd_map["config_set"] = {RedisCommand::CONFIG_SET, CommandType::WRITE};
     redis_cmd_map["lhset"] = {RedisCommand::LHSET, CommandType::WRITE};
     redis_cmd_map["lhdel"] = {RedisCommand::LHDEL, CommandType::WRITE};
+    redis_cmd_map["lhlocdel"] = {RedisCommand::LHLOCDEL, CommandType::WRITE};
     redis_cmd_map["lhmset"] = {RedisCommand::LHMSET, CommandType::WRITE};
     redis_cmd_map["lhdel_with_fallback"] = {RedisCommand::LHDEL_WITH_FALLBACK, CommandType::WRITE};
     redis_cmd_map["lhset_and_del_fallback"] = {RedisCommand::LHSET_AND_DEL_FALLBACK, CommandType::WRITE};

src/Commands.hh

@@ -72,6 +72,7 @@ enum class RedisCommand {
   LHGET,
   LHLEN,
   LHDEL,
+  LHLOCDEL,
   LHSCAN,
 
   LHGET_WITH_FALLBACK,

src/Dispatcher.cc

@@ -327,6 +327,13 @@ RedisEncodedResponse RedisDispatcher::dispatchWrite(StagingArea &stagingArea, Re
       if(request.size() <= 2) return errArgs(request);
       return dispatchLHDEL(stagingArea, request[1], request.begin()+2, request.end());
     }
+    case RedisCommand::LHLOCDEL: {
+      if(request.size() != 4) return errArgs(request);
+      int64_t itemsRemoved;
+      rocksdb::Status st = store.lhlocdel(stagingArea, request[1], request[2], request[3], itemsRemoved);
+      if(!st.ok()) return Formatter::fromStatus(st);
+      return Formatter::integer(itemsRemoved);
+    }
     case RedisCommand::LHMSET: {
       if(request.size() <= 4 || (request.size()-2) % 3 != 0) return Formatter::errArgs(request[0]);
       rocksdb::Status st = store.lhmset(stagingArea, request[1], request.begin()+2, request.end());

src/StateMachine.cc

@@ -416,6 +416,27 @@ rocksdb::Status StateMachine::lhset(StagingArea &stagingArea, std::string_view k
   return operation.finalize(operation.keySize() + fieldcreated);
 }
 
+rocksdb::Status StateMachine::lhlocdel(StagingArea &stagingArea, std::string_view key, std::string_view field, std::string_view hint, int64_t &removed) {
+  if(!assertKeyType(stagingArea, key, KeyType::kLocalityHash)) return wrong_type();
+
+  LocalityFieldLocator locator(key, hint, field);
+
+  rocksdb::Status st = stagingArea.exists(locator.toView());
+  ASSERT_OK_OR_NOTFOUND(st);
+  if(st.ok()) {
+    // Yes, we got a match, field + locality hint match, proceed to deletion.
+    RedisRequest req;
+    req.push_back(field);
+
+    rocksdb::Status st = lhdel(stagingArea, key, req.begin(), req.end(), removed);
+    qdb_assert(removed == 1);
+    return st;
+  }
+
+  removed = 0;
+  return rocksdb::Status::OK();
+}
+
 rocksdb::Status StateMachine::lhdel(StagingArea &stagingArea, std::string_view key, const ReqIterator &start, const ReqIterator &end, int64_t &removed) {
   removed = 0;
 

src/StateMachine.hh

@@ -80,6 +80,7 @@ public:
   // locality hashes
   rocksdb::Status lhset(StagingArea &stagingArea, std::string_view key, std::string_view field, std::string_view hint, std::string_view value, bool &fieldcreated);
   rocksdb::Status lhdel(StagingArea &stagingArea, std::string_view key, const ReqIterator &start, const ReqIterator &end, int64_t &removed);
+  rocksdb::Status lhlocdel(StagingArea &stagingArea, std::string_view key, std::string_view field, std::string_view hint, int64_t &removed);
   rocksdb::Status lhmset(StagingArea &stagingArea, std::string_view key, const ReqIterator &start, const ReqIterator &end);
 
   // sets

test/e2e.cc

@@ -1809,9 +1809,25 @@ TEST_F(Raft_e2e, LHSCAN) {
     "   2) \"f4\"\n"
     "   3) \"v4\"\n"
   );
-
 }
 
+TEST_F(Raft_e2e, lhlocdel) {
+  spinup(0); spinup(1); spinup(2);
+  RETRY_ASSERT_TRUE(checkStateConsensus(0, 1, 2));
+
+  int leaderID = getLeaderID();
+
+  ASSERT_REPLY(tunnel(leaderID)->exec("lhset", "myhash", "f1", "h1", "v1"), 1);
+  ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "myhash", "f1"), "v1");
+
+  // wrong locality hint, no deletion
+  ASSERT_REPLY(tunnel(leaderID)->exec("lhlocdel", "myhash", "f1", "h2"), 0);
+  ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "myhash", "f1"), "v1");
+
+  // correct locality hint, bye bye
+  ASSERT_REPLY(tunnel(leaderID)->exec("lhlocdel", "myhash", "f1", "h1"), 1);
+  ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "myhash", "f1"), "");
+}
 
 TEST_F(Raft_e2e, RawGetAllVersions) {
   spinup(0); spinup(1); spinup(2);