Commit 42e7cfda authored by Georgios Bitzes's avatar Georgios Bitzes
Browse files

Implement field deletion in locality hashes

parent 8ae99c7c
Pipeline #384346 passed with stages
in 38 minutes and 46 seconds
......@@ -76,6 +76,7 @@ struct cmdMapInit {
redis_cmd_map["rpop"] = {RedisCommand::RPOP, CommandType::WRITE};
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["exec"] = {RedisCommand::EXEC, CommandType::CONTROL};
redis_cmd_map["discard"] = {RedisCommand::DISCARD, CommandType::CONTROL};
......
......@@ -63,6 +63,7 @@ enum class RedisCommand {
LHSET,
LHGET,
LHLEN,
LHDEL,
SADD,
SISMEMBER,
......
......@@ -218,6 +218,13 @@ RedisEncodedResponse RedisDispatcher::dispatchWrite(StagingArea &stagingArea, Re
return Formatter::integer(fieldcreated);
}
case RedisCommand::LHDEL: {
if(request.size() <= 2) return errArgs(request);
int64_t count = 0;
rocksdb::Status st = store.lhdel(stagingArea, request[1], request.begin()+2, request.end(), count);
if(!st.ok()) return Formatter::fromStatus(st);
return Formatter::integer(count);
}
default: {
qdb_throw("internal dispatching error in RedisDispatcher for " << request);
}
......
......@@ -333,6 +333,25 @@ rocksdb::Status StateMachine::lhset(StagingArea &stagingArea, const std::string
return operation.finalize(operation.keySize() + 1);
}
rocksdb::Status StateMachine::lhdel(StagingArea &stagingArea, const std::string &key, const VecIterator &start, const VecIterator &end, int64_t &removed) {
removed = 0;
WriteOperation operation(stagingArea, key, KeyType::kLocalityHash);
if(!operation.valid()) return wrong_type();
for(VecIterator it = start; it != end; it++) {
std::string hint;
bool exists = operation.getAndDeleteLocalityIndex(*it, hint);
if(exists) {
removed++;
qdb_assert(operation.deleteLocalityField(hint, *it));
}
}
int64_t newsize = operation.keySize() - removed;
return operation.finalize(newsize);
}
rocksdb::Status StateMachine::lhget(StagingArea &stagingArea, const std::string &key, const std::string &field, const std::string &hint, std::string &value) {
if(!assertKeyType(stagingArea, key, KeyType::kLocalityHash)) return wrong_type();
......@@ -759,6 +778,21 @@ bool StateMachine::WriteOperation::getLocalityIndex(const std::string &field, st
return st.ok();
}
bool StateMachine::WriteOperation::getAndDeleteLocalityIndex(const std::string &field, std::string &out) {
assertWritable();
qdb_assert(keyinfo.getKeyType() == KeyType::kLocalityHash);
LocalityIndexLocator locator(redisKey, field);
rocksdb::Status st = stagingArea.get(locator.toSlice(), out);
ASSERT_OK_OR_NOTFOUND(st);
if(st.ok()) {
stagingArea.del(locator.toSlice());
}
return st.ok();
}
int64_t StateMachine::WriteOperation::keySize() {
return keyinfo.getSize();
}
......
......@@ -65,6 +65,7 @@ public:
rocksdb::Status hdel(StagingArea &stagingArea, const std::string &key, const VecIterator &start, const VecIterator &end, int64_t &removed);
rocksdb::Status lhset(StagingArea &stagingArea, const std::string &key, const std::string &field, const std::string &hint, const std::string &value, bool &fieldcreated);
rocksdb::Status lhdel(StagingArea &stagingArea, const std::string &key, const VecIterator &start, const VecIterator &end, int64_t &removed);
rocksdb::Status sadd(StagingArea &stagingArea, const std::string &key, const VecIterator &start, const VecIterator &end, int64_t &added);
rocksdb::Status srem(StagingArea &stagingArea, const std::string &key, const VecIterator &start, const VecIterator &end, int64_t &removed);
......@@ -216,6 +217,7 @@ private:
bool deleteField(const std::string &field);
bool deleteLocalityField(const std::string &hint, const std::string &field);
bool getAndDeleteLocalityIndex(const std::string &field, std::string &hint);
rocksdb::Status finalize(int64_t newsize);
......
......@@ -1002,8 +1002,43 @@ TEST_F(Raft_e2e, LocalityHash) {
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f1", "hint1"), "v2");
// Delete key.
ASSERT_REPLY(tunnel(leaderID)->exec("exists", "mykey"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("exists", "mykey", "mykey"), 2);
ASSERT_REPLY(tunnel(leaderID)->exec("del", "mykey"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("exists", "mykey"), 0);
ASSERT_REPLY(tunnel(leaderID)->exec("lhlen", "mykey"), 0);
ASSERT_REPLY(tunnel(leaderID)->exec("del", "mykey"), 0);
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f3", "aaaaa"), "");
// Recreate with five fields.
ASSERT_REPLY(tunnel(leaderID)->exec("lhset", "mykey", "f1", "hint1", "v1"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhset", "mykey", "f2", "hint2", "v2"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhset", "mykey", "f3", "hint3", "v3"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhset", "mykey", "f4", "hint4", "v4"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhset", "mykey", "f5", "hint5", "v5"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("exists", "mykey"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhlen", "mykey"), 5);
ASSERT_REPLY(tunnel(leaderID)->exec("lhdel", "mykey", "f2", "hint1"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhlen", "mykey"), 4);
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f2"), "");
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f2", "hint2"), "");
ASSERT_REPLY(tunnel(leaderID)->exec("lhdel", "mykey", "f2", "hint1"), 0);
ASSERT_REPLY(tunnel(leaderID)->exec("lhdel", "mykey", "f1", "f3"), 2);
ASSERT_REPLY(tunnel(leaderID)->exec("lhlen", "mykey"), 2);
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f4"), "v4");
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f5"), "v5");
ASSERT_REPLY(tunnel(leaderID)->exec("lhdel", "mykey", "f4", "f4", "f4", "f4"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f4"), "");
ASSERT_REPLY(tunnel(leaderID)->exec("lhlen", "mykey"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("get", "mykey"), "ERR Invalid argument: WRONGTYPE Operation against a key holding the wrong kind of value");
ASSERT_REPLY(tunnel(leaderID)->exec("lhdel", "mykey", "f4"), 0);
ASSERT_REPLY(tunnel(leaderID)->exec("lhlen", "mykey"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f5", "hint5"), "v5");
ASSERT_REPLY(tunnel(leaderID)->exec("lhdel", "mykey", "f5"), 1);
ASSERT_REPLY(tunnel(leaderID)->exec("lhget", "mykey", "f5", "hint5"), "");
ASSERT_REPLY(tunnel(leaderID)->exec("lhlen", "mykey"), 0);
}
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment