Skip to content

Commit b464d40

Browse files
kirito632kirito632
authored
Merge branch 'unstable' into feat/stream-xdelex
2 parents 95f9406 + 5264d59 commit b464d40

5 files changed

Lines changed: 502 additions & 12 deletions

File tree

‎src/commands/cmd_hash.cc‎

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919
*/
2020

2121
#include <limits>
22+
#include <optional>
2223

2324
#include "commander.h"
2425
#include "commands/command_parser.h"
@@ -53,6 +54,22 @@ Status ParseHashFieldListTail(Parser &parser, std::vector<std::string> *fields)
5354
return Status::OK();
5455
}
5556

57+
uint64_t GenerateHLenFlags(uint64_t flags, const std::vector<std::string> &args, const Config &config) {
58+
bool needs_repair = false;
59+
if (args.size() == 2) {
60+
needs_repair = config.hash_encoding_mode == HashSubkeyEncodingMode::kFieldExpiration &&
61+
config.hash_length_mode == HashLengthMode::kAccurate;
62+
} else if (args.size() == 3) {
63+
needs_repair = util::EqualICase(args[2], "REPAIR");
64+
}
65+
66+
if (!needs_repair) {
67+
return flags;
68+
}
69+
70+
return (flags | kCmdWrite | kCmdNoDBSizeCheck) & ~kCmdReadOnly;
71+
}
72+
5673
} // namespace
5774

5875
class CommandHGet : public Commander {
@@ -155,18 +172,39 @@ class CommandHExists : public Commander {
155172

156173
class CommandHLen : public Commander {
157174
public:
175+
Status Parse(const std::vector<std::string> &args) override {
176+
if (args.size() == 2) {
177+
length_mode_ = std::nullopt;
178+
return Commander::Parse(args);
179+
}
180+
if (args.size() != 3) {
181+
return {Status::RedisParseErr, errWrongNumOfArguments};
182+
}
183+
if (util::EqualICase(args[2], "APPROX")) {
184+
length_mode_ = HashLengthMode::kApproximate;
185+
} else if (util::EqualICase(args[2], "REPAIR")) {
186+
length_mode_ = HashLengthMode::kAccurate;
187+
} else {
188+
return {Status::RedisParseErr, errInvalidSyntax};
189+
}
190+
return Commander::Parse(args);
191+
}
192+
158193
Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
159194
uint64_t count = 0;
160195
redis::Hash hash_db(srv->storage, conn->GetNamespace());
161196

162-
auto s = hash_db.Size(ctx, args_[1], &count);
197+
auto s = length_mode_ ? hash_db.Size(ctx, args_[1], &count, *length_mode_) : hash_db.Size(ctx, args_[1], &count);
163198
if (!s.ok() && !s.IsNotFound()) {
164199
return {Status::RedisExecErr, s.ToString()};
165200
}
166201

167202
*output = s.IsNotFound() ? redis::Integer(0) : redis::Integer(count);
168203
return Status::OK();
169204
}
205+
206+
private:
207+
std::optional<HashLengthMode> length_mode_;
170208
};
171209

172210
class CommandHIncrBy : public Commander {
@@ -639,7 +677,7 @@ REDIS_REGISTER_COMMANDS(Hash, MakeCmdAttr<CommandHGet>("hget", 3, "read-only", 1
639677
MakeCmdAttr<CommandHDel>("hdel", -3, "write no-dbsize-check", 1, 1, 1),
640678
MakeCmdAttr<CommandHStrlen>("hstrlen", 3, "read-only", 1, 1, 1),
641679
MakeCmdAttr<CommandHExists>("hexists", 3, "read-only", 1, 1, 1),
642-
MakeCmdAttr<CommandHLen>("hlen", 2, "read-only", 1, 1, 1),
680+
MakeCmdAttr<CommandHLen>("hlen", -2, "read-only", 1, 1, 1, GenerateHLenFlags),
643681
MakeCmdAttr<CommandHMGet>("hmget", -3, "read-only", 1, 1, 1),
644682
MakeCmdAttr<CommandHMSet>("hmset", -4, "write", 1, 1, 1),
645683
MakeCmdAttr<CommandHKeys>("hkeys", 2, "read-only slow", 1, 1, 1),

‎src/types/redis_hash.cc‎

Lines changed: 105 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -189,13 +189,117 @@ rocksdb::Status Hash::decodeValue(const HashMetadata &metadata, Slice *value, ui
189189
}
190190

191191
rocksdb::Status Hash::Size(engine::Context &ctx, const Slice &user_key, uint64_t *size) {
192+
return Size(ctx, user_key, size, storage_->GetConfig()->hash_length_mode);
193+
}
194+
195+
rocksdb::Status Hash::Size(engine::Context &ctx, const Slice &user_key, uint64_t *size, HashLengthMode length_mode) {
192196
*size = 0;
193197

194198
std::string ns_key = AppendNamespacePrefix(user_key);
195199
HashMetadata metadata(false);
196200
rocksdb::Status s = getMetadata(ctx, ns_key, &metadata);
197201
if (!s.ok()) return s;
198-
*size = metadata.size;
202+
203+
if (metadata.IsLegacySubkeyEncoding() || length_mode == HashLengthMode::kApproximate) {
204+
*size = metadata.size;
205+
return rocksdb::Status::OK();
206+
}
207+
208+
if (metadata.persist > metadata.size) {
209+
return scanAndRepair(ctx, ns_key, &metadata, util::GetTimeStampMS(), size);
210+
}
211+
212+
uint64_t ttl_candidates = metadata.size - metadata.persist;
213+
if (ttl_candidates == 0) {
214+
*size = metadata.size;
215+
return rocksdb::Status::OK();
216+
}
217+
218+
uint64_t now = util::GetTimeStampMS();
219+
if (metadata.lower != 0 && now < metadata.lower) {
220+
*size = metadata.size;
221+
return rocksdb::Status::OK();
222+
}
223+
if (metadata.upper != 0 && now > metadata.upper && metadata.persist == 0) {
224+
auto batch = storage_->GetWriteBatchBase();
225+
WriteBatchLogData log_data(kRedisHash);
226+
s = batch->PutLogData(log_data.Encode());
227+
if (!s.ok()) return s;
228+
s = batch->Delete(metadata_cf_handle_, ns_key);
229+
if (!s.ok()) return s;
230+
s = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
231+
if (!s.ok()) return s;
232+
*size = 0;
233+
return rocksdb::Status::OK();
234+
}
235+
236+
return scanAndRepair(ctx, ns_key, &metadata, now, size);
237+
}
238+
239+
rocksdb::Status Hash::scanAndRepair(engine::Context &ctx, const Slice &ns_key, HashMetadata *metadata, uint64_t now,
240+
uint64_t *size) {
241+
*size = metadata->size;
242+
if (!metadata->IsFieldExpirationEncoding()) {
243+
return rocksdb::Status::OK();
244+
}
245+
246+
HashMetadata repaired_metadata = *metadata;
247+
repaired_metadata.size = 0;
248+
repaired_metadata.persist = 0;
249+
repaired_metadata.lower = 0;
250+
repaired_metadata.upper = 0;
251+
252+
auto batch = storage_->GetWriteBatchBase();
253+
WriteBatchLogData log_data(kRedisHash);
254+
auto s = batch->PutLogData(log_data.Encode());
255+
if (!s.ok()) return s;
256+
257+
std::string prefix_key = InternalKey(ns_key, "", metadata->version, storage_->IsSlotIdEncoded()).Encode();
258+
std::string next_version_prefix_key =
259+
InternalKey(ns_key, "", metadata->version + 1, storage_->IsSlotIdEncoded()).Encode();
260+
261+
rocksdb::ReadOptions read_options = ctx.DefaultScanOptions();
262+
rocksdb::Slice lower_bound(prefix_key);
263+
rocksdb::Slice upper_bound(next_version_prefix_key);
264+
read_options.iterate_lower_bound = &lower_bound;
265+
read_options.iterate_upper_bound = &upper_bound;
266+
267+
auto iter = util::UniqueIterator(ctx, read_options);
268+
for (iter->Seek(prefix_key); iter->Valid() && iter->key().starts_with(prefix_key); iter->Next()) {
269+
HashFieldState state;
270+
s = DecodeFieldState(*metadata, Slice(iter->value()), now, &state);
271+
if (!s.ok()) return s;
272+
if (state.kind == HashFieldStateKind::kExpiredTTLPhysical) {
273+
s = batch->Delete(iter->key());
274+
if (!s.ok()) return s;
275+
continue;
276+
}
277+
repaired_metadata.size += 1;
278+
if (state.kind == HashFieldStateKind::kPersistent) {
279+
repaired_metadata.persist += 1;
280+
} else if (state.kind == HashFieldStateKind::kLiveTTL) {
281+
if (repaired_metadata.lower == 0 || state.expire < repaired_metadata.lower) {
282+
repaired_metadata.lower = state.expire;
283+
}
284+
repaired_metadata.upper = std::max(repaired_metadata.upper, state.expire);
285+
}
286+
}
287+
s = iter->status();
288+
if (!s.ok()) return s;
289+
290+
*size = repaired_metadata.size;
291+
if (repaired_metadata.size == 0) {
292+
s = batch->Delete(metadata_cf_handle_, ns_key);
293+
} else {
294+
std::string bytes;
295+
repaired_metadata.Encode(&bytes);
296+
s = batch->Put(metadata_cf_handle_, ns_key, bytes);
297+
}
298+
if (!s.ok()) return s;
299+
300+
s = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch());
301+
if (!s.ok()) return s;
302+
*metadata = repaired_metadata;
199303
return rocksdb::Status::OK();
200304
}
201305

‎src/types/redis_hash.h‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,7 @@ class Hash : public SubKeyScanner {
5454
Hash(engine::Storage *storage, const std::string &ns) : SubKeyScanner(storage, ns) {}
5555

5656
rocksdb::Status Size(engine::Context &ctx, const Slice &user_key, uint64_t *size);
57+
rocksdb::Status Size(engine::Context &ctx, const Slice &user_key, uint64_t *size, HashLengthMode length_mode);
5758
rocksdb::Status Get(engine::Context &ctx, const Slice &user_key, const Slice &field, std::string *value);
5859
rocksdb::Status Set(engine::Context &ctx, const Slice &user_key, const Slice &field, const Slice &value,
5960
uint64_t *added_cnt);
@@ -87,6 +88,8 @@ class Hash : public SubKeyScanner {
8788
rocksdb::Status getMetadata(engine::Context &ctx, const Slice &ns_key, HashMetadata *metadata);
8889
rocksdb::Status getRawValue(engine::Context &ctx, const std::string &sub_key, std::string *value);
8990
static rocksdb::Status decodeValue(const HashMetadata &metadata, Slice *value, uint64_t *expire = nullptr);
91+
rocksdb::Status scanAndRepair(engine::Context &ctx, const Slice &ns_key, HashMetadata *metadata, uint64_t now,
92+
uint64_t *size);
9093

9194
friend struct FieldValueRetriever;
9295
};

‎tests/cppunit/types/hash_test.cc‎

Lines changed: 84 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -611,6 +611,90 @@ TEST_F(RedisHashFieldExpirationEncodingTest, CompactionGhostDoesNotDecrementMeta
611611
EXPECT_EQ(metadata.upper, before.upper);
612612
}
613613

614+
TEST_F(RedisHashFieldExpirationEncodingTest, SizeRepairsExpiredPhysicalAndGhostMetadata) {
615+
const Slice key = "hfe-size-repair";
616+
uint64_t ret = 0;
617+
auto s = hash_->MSet(*ctx_, key, {{"persistent", "1"}, {"live", "2"}, {"expired", "3"}, {"ghost", "4"}}, false, &ret);
618+
ASSERT_TRUE(s.ok()) << s.ToString();
619+
ASSERT_EQ(ret, 4);
620+
621+
std::vector<int64_t> results;
622+
uint64_t now = util::GetTimeStampMS();
623+
uint64_t live_expire = now + 60'000;
624+
uint64_t expired_at = now - 1;
625+
s = hash_->ExpireFields(*ctx_, key, {"live"}, live_expire, HashFieldExpireCondition::kNone, &results);
626+
ASSERT_TRUE(s.ok()) << s.ToString();
627+
s = hash_->ExpireFields(*ctx_, key, {"ghost"}, live_expire + 60'000, HashFieldExpireCondition::kNone, &results);
628+
ASSERT_TRUE(s.ok()) << s.ToString();
629+
630+
HashMetadata metadata = hashMetadata(key.ToString());
631+
s = putRawHashValue(key.ToString(), "expired", expired_at, "3");
632+
ASSERT_TRUE(s.ok()) << s.ToString();
633+
s = deleteRawHashValue(key.ToString(), "ghost");
634+
ASSERT_TRUE(s.ok()) << s.ToString();
635+
636+
metadata.size = 4;
637+
metadata.persist = 1;
638+
metadata.lower = expired_at;
639+
metadata.upper = live_expire + 60'000;
640+
s = putHashMetadata(key.ToString(), metadata);
641+
ASSERT_TRUE(s.ok()) << s.ToString();
642+
643+
s = hash_->Size(*ctx_, key, &ret);
644+
ASSERT_TRUE(s.ok()) << s.ToString();
645+
EXPECT_EQ(ret, 2);
646+
647+
metadata = hashMetadata(key.ToString());
648+
EXPECT_EQ(metadata.size, 2);
649+
EXPECT_EQ(metadata.persist, 1);
650+
EXPECT_EQ(metadata.lower, live_expire);
651+
EXPECT_EQ(metadata.upper, live_expire);
652+
653+
std::string value;
654+
s = hash_->Get(*ctx_, key, "expired", &value);
655+
EXPECT_TRUE(s.IsNotFound());
656+
s = hash_->Get(*ctx_, key, "ghost", &value);
657+
EXPECT_TRUE(s.IsNotFound());
658+
s = hash_->Get(*ctx_, key, "persistent", &value);
659+
ASSERT_TRUE(s.ok()) << s.ToString();
660+
EXPECT_EQ(value, "1");
661+
s = hash_->Get(*ctx_, key, "live", &value);
662+
ASSERT_TRUE(s.ok()) << s.ToString();
663+
EXPECT_EQ(value, "2");
664+
}
665+
666+
TEST_F(RedisHashFieldExpirationEncodingTest, SizeDeletesHashWhenAllTtlCandidatesExpired) {
667+
const Slice key = "hfe-size-delete-all-expired";
668+
uint64_t ret = 0;
669+
auto s = hash_->MSet(*ctx_, key, {{"a", "1"}, {"b", "2"}}, false, &ret);
670+
ASSERT_TRUE(s.ok()) << s.ToString();
671+
ASSERT_EQ(ret, 2);
672+
673+
uint64_t now = util::GetTimeStampMS();
674+
uint64_t lower = now - 2'000;
675+
uint64_t upper = now - 1'000;
676+
s = putRawHashValue(key.ToString(), "a", lower, "1");
677+
ASSERT_TRUE(s.ok()) << s.ToString();
678+
s = putRawHashValue(key.ToString(), "b", upper, "2");
679+
ASSERT_TRUE(s.ok()) << s.ToString();
680+
681+
HashMetadata metadata = hashMetadata(key.ToString());
682+
metadata.size = 2;
683+
metadata.persist = 0;
684+
metadata.lower = lower;
685+
metadata.upper = upper;
686+
s = putHashMetadata(key.ToString(), metadata);
687+
ASSERT_TRUE(s.ok()) << s.ToString();
688+
689+
s = hash_->Size(*ctx_, key, &ret);
690+
ASSERT_TRUE(s.ok()) << s.ToString();
691+
EXPECT_EQ(ret, 0);
692+
693+
metadata = HashMetadata(false);
694+
s = getHashMetadata(key.ToString(), &metadata);
695+
EXPECT_TRUE(s.IsNotFound());
696+
}
697+
614698
TEST_F(RedisHashFieldExpirationEncodingTest, DeleteHandlesPersistentLiveExpiredMissingAndDuplicateFields) {
615699
const Slice key = "hfe-delete-state-matrix";
616700
uint64_t ret = 0;

0 commit comments

Comments
 (0)