This is an automated email from the ASF dual-hosted git repository.
PragmaTwice pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new cfeab096c feat(hash): add HSETEX and HGETEX support (#3554)
cfeab096c is described below
commit cfeab096c8795ebb0c3df229bd7b2707cf49c900
Author: Twice <[email protected]>
AuthorDate: Sat Jul 18 14:49:22 2026 +0800
feat(hash): add HSETEX and HGETEX support (#3554)
Adds Redis-compatible HSETEX and HGETEX commands on top of the hash
field expiration encoding. Implements batched field/value and metadata
transitions for persistent, expiring, expired, missing, and ghost
fields, including FXX/FNX, KEEPTTL, PERSIST, and legacy-encoding
rejection. Adds C++ unit and Go integration coverage for metadata
invariants, command parsing, replication, restart, and concurrency.
Proposal: https://github.com/apache/kvrocks/discussions/3432
Tracking issue: https://github.com/apache/kvrocks/issues/3436
Assisted by GPT 5.5 Sol.
---
src/commands/cmd_hash.cc | 278 ++++++
src/types/redis_hash.cc | 450 +++++++++
src/types/redis_hash.h | 36 +
tests/cppunit/types/hash_test.cc | 1021 ++++++++++++++++++++
.../hash_field_expiration_commands_test.go | 115 +++
.../unit/type/hash/hash_hsetex_hgetex_test.go | 796 +++++++++++++++
6 files changed, 2696 insertions(+)
diff --git a/src/commands/cmd_hash.cc b/src/commands/cmd_hash.cc
index c56aeae63..05d074e55 100644
--- a/src/commands/cmd_hash.cc
+++ b/src/commands/cmd_hash.cc
@@ -42,6 +42,27 @@ enum class HashFieldExpireTimeMode {
kAbsoluteMilliseconds,
};
+enum class HashFieldExpireCommandTTLAction {
+ kNone,
+ kKeep,
+ kPersist,
+ kSet,
+};
+
+enum class HashFieldSetCommandCondition {
+ kNone,
+ kFNX,
+ kFXX,
+};
+
+struct HashFieldExpireCommandArgs {
+ HashFieldExpireCommandTTLAction ttl_action =
HashFieldExpireCommandTTLAction::kNone;
+ HashFieldSetCommandCondition condition = HashFieldSetCommandCondition::kNone;
+ uint64_t expire_at_ms = 0;
+ std::vector<std::string> fields;
+ std::vector<FieldValue> field_values;
+};
+
std::vector<Slice> ToSlices(const std::vector<std::string> &values) {
std::vector<Slice> slices;
slices.reserve(values.size());
@@ -102,6 +123,164 @@ Status ConvertHashFieldExpireAtMs(int64_t expire_arg,
HashFieldExpireTimeMode ti
return Status::OK();
}
+template <typename Parser>
+std::optional<HashFieldExpireTimeMode> ParseHashFieldExpireTimeMode(Parser
&parser) {
+ if (parser.EatEqICase("EX")) return
HashFieldExpireTimeMode::kRelativeSeconds;
+ if (parser.EatEqICase("PX")) return
HashFieldExpireTimeMode::kRelativeMilliseconds;
+ if (parser.EatEqICase("EXAT")) return
HashFieldExpireTimeMode::kAbsoluteSeconds;
+ if (parser.EatEqICase("PXAT")) return
HashFieldExpireTimeMode::kAbsoluteMilliseconds;
+ return std::nullopt;
+}
+
+template <typename Parser>
+Status TakeHashFieldExpireArgument(Parser &parser, int64_t *expire_arg) {
+ auto parsed = parser.template TakeInt<int64_t>(10);
+ if (!parsed) {
+ return {Status::RedisParseErr, errValueNotInteger};
+ }
+ if (*parsed < 0) {
+ return {Status::RedisParseErr, "invalid expire time, must be >= 0"};
+ }
+ *expire_arg = *parsed;
+ return Status::OK();
+}
+
+Status ConvertHashFieldExpireAtMsForCommand(int64_t expire_arg,
HashFieldExpireTimeMode time_mode, uint64_t now_ms,
+ uint64_t *expire_at_ms) {
+ bool relative = time_mode == HashFieldExpireTimeMode::kRelativeSeconds ||
+ time_mode == HashFieldExpireTimeMode::kRelativeMilliseconds;
+ if ((!relative || now_ms <= kMaxHashFieldExpireAtMs) &&
+ ConvertHashFieldExpireAtMs(expire_arg, time_mode, now_ms,
expire_at_ms).IsOK()) {
+ return Status::OK();
+ }
+
+ return {Status::RedisParseErr, "invalid expire time"};
+}
+
+template <bool IsHSetEx, typename Parser>
+Status ParseHashFieldExpireFieldsBlock(Parser &parser,
HashFieldExpireCommandArgs *parsed) {
+ if (!parser.Good()) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ auto count = parser.template TakeInt<int64_t>(10);
+ if (!count || *count < 1 || *count > std::numeric_limits<int>::max()) {
+ return {Status::RedisParseErr, "invalid number of fields"};
+ }
+
+ constexpr size_t args_per_field = IsHSetEx ? 2 : 1;
+ auto field_count = static_cast<size_t>(*count);
+ if (field_count > parser.Remains() / args_per_field) {
+ return {Status::RedisParseErr, errWrongNumOfArguments};
+ }
+
+ if constexpr (IsHSetEx) {
+ parsed->field_values.reserve(field_count);
+ for (size_t i = 0; i < field_count; ++i) {
+ auto field = GET_OR_RET(parser.TakeStr());
+ auto value = GET_OR_RET(parser.TakeStr());
+ parsed->field_values.emplace_back(std::move(field), std::move(value));
+ }
+ } else {
+ parsed->fields.reserve(field_count);
+ for (size_t i = 0; i < field_count; ++i) {
+ parsed->fields.emplace_back(GET_OR_RET(parser.TakeStr()));
+ }
+ }
+
+ return Status::OK();
+}
+
+template <bool IsHSetEx>
+Status HashFieldExpireTTLConflict() {
+ auto option = IsHSetEx ? "KEEPTTL" : "PERSIST";
+ return {Status::RedisParseErr,
+ "Only one of EX, PX, EXAT, PXAT or " + std::string(option) + "
arguments can be specified"};
+}
+
+template <bool IsHSetEx>
+Status ParseHashFieldExpireCommandArgs(const std::vector<std::string> &args,
uint64_t now_ms,
+ HashFieldExpireCommandArgs *result) {
+ CommandParser parser(args, 2);
+ HashFieldExpireCommandArgs parsed;
+ bool fields_seen = false;
+
+ while (parser.Good()) {
+ if (parser.EatEqICase("FIELDS")) {
+ if (fields_seen) {
+ return {Status::RedisParseErr, "FIELDS keyword specified multiple
times"};
+ }
+ fields_seen = true;
+ GET_OR_RET(ParseHashFieldExpireFieldsBlock<IsHSetEx>(parser, &parsed));
+ continue;
+ }
+
+ auto time_mode = ParseHashFieldExpireTimeMode(parser);
+ if (time_mode) {
+ if (parsed.ttl_action != HashFieldExpireCommandTTLAction::kNone) {
+ return HashFieldExpireTTLConflict<IsHSetEx>();
+ }
+ if (!parser.Good()) {
+ return {Status::RedisParseErr, "missing expire time"};
+ }
+
+ int64_t expire_arg = 0;
+ GET_OR_RET(TakeHashFieldExpireArgument(parser, &expire_arg));
+ GET_OR_RET(ConvertHashFieldExpireAtMsForCommand(expire_arg, *time_mode,
now_ms, &parsed.expire_at_ms));
+ parsed.ttl_action = HashFieldExpireCommandTTLAction::kSet;
+ continue;
+ }
+
+ if constexpr (IsHSetEx) {
+ if (parser.EatEqICase("KEEPTTL")) {
+ if (parsed.ttl_action != HashFieldExpireCommandTTLAction::kNone) {
+ return HashFieldExpireTTLConflict<true>();
+ }
+ parsed.ttl_action = HashFieldExpireCommandTTLAction::kKeep;
+ continue;
+ }
+
+ HashFieldSetCommandCondition condition =
HashFieldSetCommandCondition::kNone;
+ if (parser.EatEqICase("FXX")) {
+ condition = HashFieldSetCommandCondition::kFXX;
+ } else if (parser.EatEqICase("FNX")) {
+ condition = HashFieldSetCommandCondition::kFNX;
+ }
+ if (condition != HashFieldSetCommandCondition::kNone) {
+ if (parsed.condition != HashFieldSetCommandCondition::kNone) {
+ return {Status::RedisParseErr, "Only one of FXX or FNX arguments can
be specified"};
+ }
+ parsed.condition = condition;
+ continue;
+ }
+ } else {
+ if (parser.EatEqICase("PERSIST")) {
+ if (parsed.ttl_action != HashFieldExpireCommandTTLAction::kNone) {
+ return HashFieldExpireTTLConflict<false>();
+ }
+ parsed.ttl_action = HashFieldExpireCommandTTLAction::kPersist;
+ continue;
+ }
+ }
+
+ return {Status::RedisParseErr, "unknown argument: " + parser.RawPeek()};
+ }
+
+ if (!fields_seen) {
+ return {Status::RedisParseErr, "missing FIELDS argument"};
+ }
+
+ *result = std::move(parsed);
+ return Status::OK();
+}
+
+Status HashCommandStatus(const rocksdb::Status &status) {
+ if (status.IsInvalidArgument() && status.ToString().find("WRONGTYPE") !=
std::string::npos) {
+ return {Status::RedisWrongType, "Operation against a key holding the wrong
kind of value"};
+ }
+ return {Status::RedisExecErr, status.ToString()};
+}
+
std::optional<HashFieldExpireCondition>
ParseHashExpireCondition(std::string_view token) {
if (util::EqualICase(token, "NX")) return HashFieldExpireCondition::kNX;
if (util::EqualICase(token, "XX")) return HashFieldExpireCondition::kXX;
@@ -704,6 +883,103 @@ class CommandHRandField : public Commander {
bool no_parameters_ = true;
};
+class CommandHSetEx : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ now_ms_ = util::GetTimeStampMS();
+ GET_OR_RET(ParseHashFieldExpireCommandArgs<true>(args, now_ms_, &parsed_));
+ return Commander::Parse(args);
+ }
+
+ Status Execute(engine::Context &ctx, Server *srv, Connection *conn,
std::string *output) override {
+ HashSetExOptions options;
+ switch (parsed_.condition) {
+ case HashFieldSetCommandCondition::kNone:
+ options.condition = HashFieldSetCondition::kNone;
+ break;
+ case HashFieldSetCommandCondition::kFNX:
+ options.condition = HashFieldSetCondition::kFNX;
+ break;
+ case HashFieldSetCommandCondition::kFXX:
+ options.condition = HashFieldSetCondition::kFXX;
+ break;
+ }
+
+ switch (parsed_.ttl_action) {
+ case HashFieldExpireCommandTTLAction::kNone:
+ options.ttl_action = HashSetExOptions::TTLAction::kDiscard;
+ break;
+ case HashFieldExpireCommandTTLAction::kKeep:
+ options.ttl_action = HashSetExOptions::TTLAction::kKeep;
+ break;
+ case HashFieldExpireCommandTTLAction::kSet:
+ options.ttl_action = HashSetExOptions::TTLAction::kSet;
+ options.expire_at_ms = parsed_.expire_at_ms;
+ break;
+ case HashFieldExpireCommandTTLAction::kPersist:
+ return {Status::RedisExecErr, errInvalidSyntax};
+ }
+
+ bool applied = false;
+ redis::Hash hash_db(srv->storage, conn->GetNamespace());
+ auto status = hash_db.SetFieldsWithExpire(ctx, args_[1],
parsed_.field_values, options, &applied, now_ms_);
+ if (!status.ok()) return HashCommandStatus(status);
+
+ *output = redis::Integer(applied ? 1 : 0);
+ return Status::OK();
+ }
+
+ private:
+ uint64_t now_ms_ = 0;
+ HashFieldExpireCommandArgs parsed_;
+};
+
+class CommandHGetEx : public Commander {
+ public:
+ Status Parse(const std::vector<std::string> &args) override {
+ now_ms_ = util::GetTimeStampMS();
+ GET_OR_RET(ParseHashFieldExpireCommandArgs<false>(args, now_ms_,
&parsed_));
+ return Commander::Parse(args);
+ }
+
+ Status Execute(engine::Context &ctx, Server *srv, Connection *conn,
std::string *output) override {
+ HashGetExOptions options;
+ switch (parsed_.ttl_action) {
+ case HashFieldExpireCommandTTLAction::kNone:
+ options.ttl_action = HashGetExOptions::TTLAction::kNone;
+ break;
+ case HashFieldExpireCommandTTLAction::kPersist:
+ options.ttl_action = HashGetExOptions::TTLAction::kPersist;
+ break;
+ case HashFieldExpireCommandTTLAction::kSet:
+ options.ttl_action = HashGetExOptions::TTLAction::kSet;
+ options.expire_at_ms = parsed_.expire_at_ms;
+ break;
+ case HashFieldExpireCommandTTLAction::kKeep:
+ return {Status::RedisExecErr, errInvalidSyntax};
+ }
+
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto fields = ToSlices(parsed_.fields);
+ redis::Hash hash_db(srv->storage, conn->GetNamespace());
+ auto status = hash_db.GetFieldsWithExpire(ctx, args_[1], fields, options,
&values, &statuses, now_ms_);
+ if (status.IsNotFound()) {
+ values.resize(fields.size());
+ statuses.resize(fields.size(), rocksdb::Status::NotFound());
+ } else if (!status.ok()) {
+ return HashCommandStatus(status);
+ }
+
+ *output = conn->MultiBulkString(values, statuses);
+ return Status::OK();
+ }
+
+ private:
+ uint64_t now_ms_ = 0;
+ HashFieldExpireCommandArgs parsed_;
+};
+
template <HashFieldExpireTimeMode kTimeMode>
class CommandHExpireGeneric : public Commander {
public:
@@ -809,6 +1085,8 @@ REDIS_REGISTER_COMMANDS(
MakeCmdAttr<CommandHScan>("hscan", -3, "read-only", 1, 1, 1),
MakeCmdAttr<CommandHRangeByLex>("hrangebylex", -4, "read-only", 1, 1, 1),
MakeCmdAttr<CommandHRandField>("hrandfield", -2, "read-only slow", 1, 1,
1),
+ MakeCmdAttr<CommandHSetEx>("hsetex", -6, "write", 1, 1, 1),
+ MakeCmdAttr<CommandHGetEx>("hgetex", -5, "write no-dbsize-check", 1, 1, 1),
MakeCmdAttr<CommandHExpireGeneric<HashFieldExpireTimeMode::kRelativeSeconds>>("hexpire",
-6, "write", 1, 1, 1),
MakeCmdAttr<CommandHExpireGeneric<HashFieldExpireTimeMode::kRelativeMilliseconds>>("hpexpire",
-6, "write", 1, 1,
1),
diff --git a/src/types/redis_hash.cc b/src/types/redis_hash.cc
index 1ec515b8d..5ba98b1f8 100644
--- a/src/types/redis_hash.cc
+++ b/src/types/redis_hash.cc
@@ -25,6 +25,7 @@
#include <algorithm>
#include <cctype>
#include <cmath>
+#include <limits>
#include <optional>
#include <random>
#include <unordered_map>
@@ -53,6 +54,9 @@ struct HashFieldState {
uint64_t expire = 0;
};
+constexpr const char *kHashFieldExpirationLegacyEncodingError =
+ "hash field expiration is not supported by legacy hash encoding";
+
bool IsFieldExpired(uint64_t expire, uint64_t now) { return expire != 0 &&
expire < now; }
bool IsImmediateExpire(uint64_t expire_at, uint64_t now) { return expire_at <=
now; }
@@ -85,6 +89,11 @@ void ApplyMissingToPersistent(HashMetadata *metadata) {
}
}
+void ApplyMissingToTTL(HashMetadata *metadata, uint64_t expire_at) {
+ ExpandExpireBounds(metadata, expire_at);
+ metadata->size += 1;
+}
+
void ApplyPersistentToTTL(HashMetadata *metadata, uint64_t expire_at) {
ExpandExpireBounds(metadata, expire_at);
metadata->persist = SaturatingSub(metadata->persist, 1);
@@ -171,6 +180,155 @@ bool HExpireConditionPasses(HashFieldExpireCondition
condition, const HashFieldS
return false;
}
+rocksdb::Status ValidateHashFieldExpirationMetadata(const HashMetadata
&metadata) {
+ if (metadata.persist > metadata.size) {
+ return rocksdb::Status::Corruption("invalid hash field expiration
metadata: persist exceeds size");
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status ValidateMissingFieldTransition(const HashMetadata &metadata) {
+ auto s = ValidateHashFieldExpirationMetadata(metadata);
+ if (!s.ok()) return s;
+ if (metadata.size == std::numeric_limits<uint64_t>::max()) {
+ return rocksdb::Status::Corruption("invalid hash field expiration
metadata: size overflow");
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status ValidatePersistentFieldTransition(const HashMetadata
&metadata) {
+ auto s = ValidateHashFieldExpirationMetadata(metadata);
+ if (!s.ok()) return s;
+ if (metadata.size == 0 || metadata.persist == 0) {
+ return rocksdb::Status::Corruption("invalid hash field expiration
metadata: no persistent field to update");
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status ValidateTTLFieldTransition(const HashMetadata &metadata) {
+ auto s = ValidateHashFieldExpirationMetadata(metadata);
+ if (!s.ok()) return s;
+ if (metadata.size == 0 || metadata.persist == metadata.size) {
+ return rocksdb::Status::Corruption("invalid hash field expiration
metadata: no TTL field to update");
+ }
+ return rocksdb::Status::OK();
+}
+
+bool HashMetadataEqual(const HashMetadata &lhs, const HashMetadata &rhs) {
+ return static_cast<const Metadata &>(lhs) == static_cast<const Metadata
&>(rhs) && lhs.mode == rhs.mode &&
+ lhs.persist == rhs.persist && lhs.lower == rhs.lower && lhs.upper ==
rhs.upper;
+}
+
+class HashBatchWriter {
+ public:
+ explicit HashBatchWriter(rocksdb::WriteBatchBase *batch) : batch_(batch) {}
+
+ rocksdb::Status Put(const Slice &key, const Slice &value) {
+ auto s = ensureLogData();
+ if (!s.ok()) return s;
+ s = batch_->Put(key, value);
+ if (s.ok()) storage_mutated_ = true;
+ return s;
+ }
+
+ rocksdb::Status Put(rocksdb::ColumnFamilyHandle *column_family, const Slice
&key, const Slice &value) {
+ auto s = ensureLogData();
+ if (!s.ok()) return s;
+ s = batch_->Put(column_family, key, value);
+ if (s.ok()) storage_mutated_ = true;
+ return s;
+ }
+
+ rocksdb::Status Delete(const Slice &key) {
+ auto s = ensureLogData();
+ if (!s.ok()) return s;
+ s = batch_->Delete(key);
+ if (s.ok()) storage_mutated_ = true;
+ return s;
+ }
+
+ rocksdb::Status Delete(rocksdb::ColumnFamilyHandle *column_family, const
Slice &key) {
+ auto s = ensureLogData();
+ if (!s.ok()) return s;
+ s = batch_->Delete(column_family, key);
+ if (s.ok()) storage_mutated_ = true;
+ return s;
+ }
+
+ bool HasStorageMutation() const { return storage_mutated_; }
+
+ private:
+ rocksdb::Status ensureLogData() {
+ if (has_log_data_) return rocksdb::Status::OK();
+ WriteBatchLogData log_data(kRedisHash);
+ auto s = batch_->PutLogData(log_data.Encode());
+ if (s.ok()) has_log_data_ = true;
+ return s;
+ }
+
+ rocksdb::WriteBatchBase *batch_;
+ bool has_log_data_ = false;
+ bool storage_mutated_ = false;
+};
+
+rocksdb::Status LoadFieldStates(engine::Storage *storage, engine::Context
&ctx, const HashMetadata &metadata,
+ const std::string &ns_key, const
std::vector<Slice> &fields, uint64_t now,
+ bool read_existing,
std::unordered_map<std::string, HashFieldState> *states) {
+ states->clear();
+ states->reserve(fields.size());
+
+ std::vector<std::string> unique_fields;
+ unique_fields.reserve(fields.size());
+ for (const auto &field : fields) {
+ bool inserted = states->emplace(field.ToString(), HashFieldState{}).second;
+ if (inserted) unique_fields.emplace_back(field.ToString());
+ }
+
+ if (!read_existing || unique_fields.empty()) return rocksdb::Status::OK();
+
+ std::vector<std::string> encoded_keys;
+ encoded_keys.reserve(unique_fields.size());
+ for (const auto &field : unique_fields) {
+ encoded_keys.emplace_back(InternalKey(ns_key, field, metadata.version,
storage->IsSlotIdEncoded()).Encode());
+ }
+
+ std::vector<rocksdb::Slice> keys;
+ keys.reserve(encoded_keys.size());
+ for (const auto &key : encoded_keys) keys.emplace_back(key);
+
+ std::vector<rocksdb::PinnableSlice> raw_values(keys.size());
+ std::vector<rocksdb::Status> statuses(keys.size());
+ storage->MultiGet(ctx, ctx.DefaultMultiGetOptions(),
storage->GetDB()->DefaultColumnFamily(), keys.size(),
+ keys.data(), raw_values.data(), statuses.data());
+
+ for (size_t i = 0; i < unique_fields.size(); ++i) {
+ if (statuses[i].IsNotFound()) continue;
+ if (!statuses[i].ok()) return statuses[i];
+ auto s = DecodeFieldState(metadata, Slice(raw_values[i]), now,
&states->at(unique_fields[i]));
+ if (!s.ok()) return s;
+ }
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status FinalizeHashMetadata(const HashMetadata &original_metadata,
bool metadata_existed,
+ HashMetadata *metadata,
rocksdb::ColumnFamilyHandle *metadata_cf_handle,
+ const std::string &ns_key,
HashBatchWriter *writer) {
+ auto s = ValidateHashFieldExpirationMetadata(*metadata);
+ if (!s.ok()) return s;
+
+ ClearBoundsIfNoTtlCandidates(metadata);
+ if (metadata_existed && metadata->size == 0) {
+ return writer->Delete(metadata_cf_handle, ns_key);
+ }
+ if (metadata->size == 0 || (metadata_existed &&
HashMetadataEqual(original_metadata, *metadata))) {
+ return rocksdb::Status::OK();
+ }
+
+ std::string encoded_metadata;
+ metadata->Encode(&encoded_metadata);
+ return writer->Put(metadata_cf_handle, ns_key, encoded_metadata);
+}
+
} // namespace
HashMetadata Hash::createMetadataForWrite(bool generate_version) const {
@@ -496,6 +654,298 @@ rocksdb::Status Hash::MGet(engine::Context &ctx, const
Slice &user_key, const st
return rocksdb::Status::OK();
}
+rocksdb::Status Hash::SetFieldsWithExpire(engine::Context &ctx, const Slice
&user_key,
+ const std::vector<FieldValue>
&field_values, const HashSetExOptions &options,
+ bool *applied,
std::optional<uint64_t> now_ms) {
+ *applied = false;
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+ HashMetadata metadata(false);
+ auto s = getMetadata(ctx, ns_key, &metadata);
+ bool metadata_existed = s.ok();
+ if (s.IsNotFound()) {
+ if (options.condition == HashFieldSetCondition::kFXX) return
rocksdb::Status::OK();
+ metadata = createMetadataForWrite();
+ if (metadata.IsLegacySubkeyEncoding()) {
+ return
rocksdb::Status::InvalidArgument(kHashFieldExpirationLegacyEncodingError);
+ }
+ } else if (!s.ok()) {
+ return s;
+ } else if (metadata.IsLegacySubkeyEncoding()) {
+ return
rocksdb::Status::InvalidArgument(kHashFieldExpirationLegacyEncodingError);
+ }
+
+ s = ValidateHashFieldExpirationMetadata(metadata);
+ if (!s.ok()) return s;
+ HashMetadata original_metadata = metadata;
+ uint64_t now = now_ms.value_or(util::GetTimeStampMS());
+
+ std::vector<Slice> fields;
+ fields.reserve(field_values.size());
+ for (const auto &field_value : field_values)
fields.emplace_back(field_value.field);
+
+ std::unordered_map<std::string, HashFieldState> state_cache;
+ s = LoadFieldStates(storage_, ctx, metadata, ns_key, fields, now,
metadata_existed, &state_cache);
+ if (!s.ok()) return s;
+
+ auto make_sub_key = [&](const std::string &field) {
+ return InternalKey(ns_key, field, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ };
+
+ std::vector<std::string> expired_cleanup_keys;
+ bool condition_passed = true;
+ if (options.condition != HashFieldSetCondition::kNone) {
+ for (const auto &field_value : field_values) {
+ HashFieldState &state = state_cache.at(field_value.field);
+ if (state.kind == HashFieldStateKind::kExpiredTTLPhysical) {
+ s = ValidateTTLFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyTTLToDeleted(&metadata);
+ expired_cleanup_keys.emplace_back(make_sub_key(field_value.field));
+ state = HashFieldState{};
+ }
+
+ bool exists = state.kind == HashFieldStateKind::kPersistent ||
state.kind == HashFieldStateKind::kLiveTTL;
+ if ((options.condition == HashFieldSetCondition::kFNX && exists) ||
+ (options.condition == HashFieldSetCondition::kFXX && !exists)) {
+ condition_passed = false;
+ break;
+ }
+ }
+ }
+
+ auto batch = storage_->GetWriteBatchBase();
+ HashBatchWriter writer(batch.Get());
+ for (const auto &sub_key : expired_cleanup_keys) {
+ s = writer.Delete(sub_key);
+ if (!s.ok()) return s;
+ }
+
+ if (!condition_passed) {
+ s = FinalizeHashMetadata(original_metadata, metadata_existed, &metadata,
metadata_cf_handle_, ns_key, &writer);
+ if (!s.ok()) return s;
+ if (writer.HasStorageMutation()) {
+ s = storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+ if (!s.ok()) return s;
+ }
+ return rocksdb::Status::OK();
+ }
+
+ std::unordered_set<std::string_view> seen_fields;
+ std::vector<const FieldValue *> unique_field_values;
+ seen_fields.reserve(field_values.size());
+ unique_field_values.reserve(field_values.size());
+ for (auto iter = field_values.rbegin(); iter != field_values.rend(); ++iter)
{
+ if (seen_fields.emplace(iter->field).second)
unique_field_values.emplace_back(&*iter);
+ }
+ std::reverse(unique_field_values.begin(), unique_field_values.end());
+
+ bool immediate =
+ options.ttl_action == HashSetExOptions::TTLAction::kSet &&
IsImmediateExpire(options.expire_at_ms, now);
+ for (const auto *field_value : unique_field_values) {
+ HashFieldState &state = state_cache.at(field_value->field);
+ std::string sub_key = make_sub_key(field_value->field);
+
+ if (immediate) {
+ if (state.kind == HashFieldStateKind::kMissing) continue;
+ if (state.kind == HashFieldStateKind::kPersistent) {
+ s = ValidatePersistentFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyPersistentToDeleted(&metadata);
+ } else {
+ s = ValidateTTLFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyTTLToDeleted(&metadata);
+ }
+ s = writer.Delete(sub_key);
+ if (!s.ok()) return s;
+ state = HashFieldState{};
+ continue;
+ }
+
+ uint64_t target_expire = 0;
+ switch (options.ttl_action) {
+ case HashSetExOptions::TTLAction::kDiscard:
+ break;
+ case HashSetExOptions::TTLAction::kKeep:
+ if (state.kind == HashFieldStateKind::kLiveTTL || state.kind ==
HashFieldStateKind::kExpiredTTLPhysical) {
+ target_expire = state.expire;
+ }
+ break;
+ case HashSetExOptions::TTLAction::kSet:
+ target_expire = options.expire_at_ms;
+ break;
+ }
+
+ switch (state.kind) {
+ case HashFieldStateKind::kMissing:
+ s = ValidateMissingFieldTransition(metadata);
+ if (!s.ok()) return s;
+ if (target_expire == 0) {
+ ApplyMissingToPersistent(&metadata);
+ } else {
+ ApplyMissingToTTL(&metadata, target_expire);
+ }
+ break;
+ case HashFieldStateKind::kPersistent:
+ if (target_expire != 0) {
+ s = ValidatePersistentFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyPersistentToTTL(&metadata, target_expire);
+ }
+ break;
+ case HashFieldStateKind::kLiveTTL:
+ case HashFieldStateKind::kExpiredTTLPhysical:
+ s = ValidateTTLFieldTransition(metadata);
+ if (!s.ok()) return s;
+ if (target_expire == 0) {
+ ApplyTTLToPersistent(&metadata);
+ } else {
+ ApplyTTLToTTL(&metadata, target_expire);
+ }
+ break;
+ }
+
+ s = writer.Put(sub_key, metadata.EncodeSubkeyValue(field_value->value,
target_expire));
+ if (!s.ok()) return s;
+ state.value = field_value->value;
+ state.expire = target_expire;
+ if (target_expire == 0) {
+ state.kind = HashFieldStateKind::kPersistent;
+ } else if (IsFieldExpired(target_expire, now)) {
+ state.kind = HashFieldStateKind::kExpiredTTLPhysical;
+ } else {
+ state.kind = HashFieldStateKind::kLiveTTL;
+ }
+ }
+
+ s = FinalizeHashMetadata(original_metadata, metadata_existed, &metadata,
metadata_cf_handle_, ns_key, &writer);
+ if (!s.ok()) return s;
+ if (writer.HasStorageMutation()) {
+ s = storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+ if (!s.ok()) return s;
+ }
+ *applied = true;
+ return rocksdb::Status::OK();
+}
+
+rocksdb::Status Hash::GetFieldsWithExpire(engine::Context &ctx, const Slice
&user_key, const std::vector<Slice> &fields,
+ const HashGetExOptions &options,
std::vector<std::string> *values,
+ std::vector<rocksdb::Status>
*statuses, std::optional<uint64_t> now_ms) {
+ values->clear();
+ statuses->clear();
+ values->reserve(fields.size());
+ statuses->reserve(fields.size());
+
+ std::string ns_key = AppendNamespacePrefix(user_key);
+ HashMetadata metadata(false);
+ auto s = getMetadata(ctx, ns_key, &metadata);
+ if (s.IsNotFound()) {
+ values->resize(fields.size());
+ statuses->resize(fields.size(), rocksdb::Status::NotFound());
+ return rocksdb::Status::OK();
+ }
+ if (!s.ok()) return s;
+ if (metadata.IsLegacySubkeyEncoding()) {
+ return
rocksdb::Status::InvalidArgument(kHashFieldExpirationLegacyEncodingError);
+ }
+
+ s = ValidateHashFieldExpirationMetadata(metadata);
+ if (!s.ok()) return s;
+ HashMetadata original_metadata = metadata;
+ uint64_t now = now_ms.value_or(util::GetTimeStampMS());
+
+ std::unordered_map<std::string, HashFieldState> state_cache;
+ s = LoadFieldStates(storage_, ctx, metadata, ns_key, fields, now, true,
&state_cache);
+ if (!s.ok()) return s;
+
+ auto batch = storage_->GetWriteBatchBase();
+ HashBatchWriter writer(batch.Get());
+ auto make_sub_key = [&](const std::string &field) {
+ return InternalKey(ns_key, field, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ };
+ bool immediate =
+ options.ttl_action == HashGetExOptions::TTLAction::kSet &&
IsImmediateExpire(options.expire_at_ms, now);
+
+ for (const auto &field_slice : fields) {
+ std::string field = field_slice.ToString();
+ HashFieldState &state = state_cache.at(field);
+ std::string sub_key = make_sub_key(field);
+
+ if (state.kind == HashFieldStateKind::kMissing) {
+ values->emplace_back();
+ statuses->emplace_back(rocksdb::Status::NotFound());
+ continue;
+ }
+ if (state.kind == HashFieldStateKind::kExpiredTTLPhysical) {
+ values->emplace_back();
+ statuses->emplace_back(rocksdb::Status::NotFound());
+ s = ValidateTTLFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyTTLToDeleted(&metadata);
+ s = writer.Delete(sub_key);
+ if (!s.ok()) return s;
+ state = HashFieldState{};
+ continue;
+ }
+
+ values->emplace_back(state.value);
+ statuses->emplace_back(rocksdb::Status::OK());
+
+ if (options.ttl_action == HashGetExOptions::TTLAction::kNone) continue;
+ if (options.ttl_action == HashGetExOptions::TTLAction::kPersist) {
+ if (state.kind == HashFieldStateKind::kPersistent) continue;
+ s = ValidateTTLFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyTTLToPersistent(&metadata);
+ s = writer.Put(sub_key, metadata.EncodeSubkeyValue(state.value));
+ if (!s.ok()) return s;
+ state.kind = HashFieldStateKind::kPersistent;
+ state.expire = 0;
+ continue;
+ }
+
+ if (immediate) {
+ if (state.kind == HashFieldStateKind::kPersistent) {
+ s = ValidatePersistentFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyPersistentToDeleted(&metadata);
+ } else {
+ s = ValidateTTLFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyTTLToDeleted(&metadata);
+ }
+ s = writer.Delete(sub_key);
+ if (!s.ok()) return s;
+ state = HashFieldState{};
+ continue;
+ }
+
+ if (state.kind == HashFieldStateKind::kLiveTTL && state.expire ==
options.expire_at_ms) continue;
+ if (state.kind == HashFieldStateKind::kPersistent) {
+ s = ValidatePersistentFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyPersistentToTTL(&metadata, options.expire_at_ms);
+ } else {
+ s = ValidateTTLFieldTransition(metadata);
+ if (!s.ok()) return s;
+ ApplyTTLToTTL(&metadata, options.expire_at_ms);
+ }
+ s = writer.Put(sub_key, metadata.EncodeSubkeyValue(state.value,
options.expire_at_ms));
+ if (!s.ok()) return s;
+ state.kind = HashFieldStateKind::kLiveTTL;
+ state.expire = options.expire_at_ms;
+ }
+
+ s = FinalizeHashMetadata(original_metadata, true, &metadata,
metadata_cf_handle_, ns_key, &writer);
+ if (!s.ok()) return s;
+ if (writer.HasStorageMutation()) {
+ s = storage_->Write(ctx, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
+ if (!s.ok()) return s;
+ }
+ return rocksdb::Status::OK();
+}
+
rocksdb::Status Hash::Set(engine::Context &ctx, const Slice &user_key, const
Slice &field, const Slice &value,
uint64_t *added_cnt) {
return MSet(ctx, user_key, {{field.ToString(), value.ToString()}}, false,
added_cnt);
diff --git a/src/types/redis_hash.h b/src/types/redis_hash.h
index 0728fa10f..659f43987 100644
--- a/src/types/redis_hash.h
+++ b/src/types/redis_hash.h
@@ -48,6 +48,35 @@ enum class HashFieldExpireCondition {
kLT,
};
+enum class HashFieldSetCondition {
+ kNone,
+ kFNX,
+ kFXX,
+};
+
+struct HashSetExOptions {
+ enum class TTLAction {
+ kDiscard,
+ kKeep,
+ kSet,
+ };
+
+ HashFieldSetCondition condition = HashFieldSetCondition::kNone;
+ TTLAction ttl_action = TTLAction::kDiscard;
+ uint64_t expire_at_ms = 0;
+};
+
+struct HashGetExOptions {
+ enum class TTLAction {
+ kNone,
+ kPersist,
+ kSet,
+ };
+
+ TTLAction ttl_action = TTLAction::kNone;
+ uint64_t expire_at_ms = 0;
+};
+
namespace redis {
class Hash : public SubKeyScanner {
@@ -71,6 +100,13 @@ class Hash : public SubKeyScanner {
std::vector<FieldValue> *field_values);
rocksdb::Status MGet(engine::Context &ctx, const Slice &user_key, const
std::vector<Slice> &fields,
std::vector<std::string> *values,
std::vector<rocksdb::Status> *statuses);
+ rocksdb::Status SetFieldsWithExpire(engine::Context &ctx, const Slice
&user_key,
+ const std::vector<FieldValue>
&field_values, const HashSetExOptions &options,
+ bool *applied, std::optional<uint64_t>
now_ms = std::nullopt);
+ rocksdb::Status GetFieldsWithExpire(engine::Context &ctx, const Slice
&user_key, const std::vector<Slice> &fields,
+ const HashGetExOptions &options,
std::vector<std::string> *values,
+ std::vector<rocksdb::Status> *statuses,
+ std::optional<uint64_t> now_ms =
std::nullopt);
rocksdb::Status GetAll(engine::Context &ctx, const Slice &user_key,
std::vector<FieldValue> *field_values,
HashFetchType type = HashFetchType::kAll);
rocksdb::Status Scan(engine::Context &ctx, const Slice &user_key, const
std::string &cursor, uint64_t limit,
diff --git a/tests/cppunit/types/hash_test.cc b/tests/cppunit/types/hash_test.cc
index e678aaf5c..53b8bd048 100644
--- a/tests/cppunit/types/hash_test.cc
+++ b/tests/cppunit/types/hash_test.cc
@@ -27,8 +27,11 @@
#include <filesystem>
#include <fstream>
#include <memory>
+#include <optional>
#include <random>
#include <string>
+#include <tuple>
+#include <utility>
#include <vector>
#include "parse_util.h"
@@ -146,6 +149,104 @@ class RedisHashFieldExpirationEncodingTest : public
::testing::Test {
return storage_->Write(*ctx_, storage_->DefaultWriteOptions(),
batch->GetWriteBatch());
}
+ rocksdb::Status getRawHashValue(const std::string &key, const std::string
&field, std::string *raw_value) {
+ HashMetadata metadata = hashMetadata(key);
+ return storage_->Get(*ctx_, ctx_->GetReadOptions(), hashSubKey(key, field,
metadata), raw_value);
+ }
+
+ std::pair<std::string, uint64_t> decodedHashValue(const std::string &key,
const std::string &field) {
+ HashMetadata metadata(false);
+ std::string raw_value = rawHashValue(key, field, &metadata);
+ Slice decoded_value(raw_value);
+ uint64_t expire = 0;
+ auto s = metadata.DecodeSubkeyValue(&decoded_value, &expire);
+ assert(s.ok());
+ return {decoded_value.ToString(), expire};
+ }
+
+ void expectHashMetadata(const std::string &key, uint64_t size, uint64_t
persist, uint64_t lower, uint64_t upper) {
+ HashMetadata metadata = hashMetadata(key);
+ EXPECT_EQ(metadata.mode, HashSubkeyEncodingMode::kFieldExpiration);
+ EXPECT_EQ(metadata.size, size);
+ EXPECT_EQ(metadata.persist, persist);
+ EXPECT_EQ(metadata.lower, lower);
+ EXPECT_EQ(metadata.upper, upper);
+ EXPECT_LE(metadata.persist, metadata.size);
+ if (metadata.size == metadata.persist) {
+ EXPECT_EQ(metadata.lower, 0);
+ EXPECT_EQ(metadata.upper, 0);
+ } else {
+ EXPECT_GT(metadata.lower, 0);
+ EXPECT_GE(metadata.upper, metadata.lower);
+ }
+ }
+
+ void createFourStateHash(const std::string &key, uint64_t expired_at,
uint64_t live_expire) {
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"P", "p"}, {"L", "l"}, {"X", "x"},
{"K", "k"}}, false, &ret);
+ assert(s.ok());
+ assert(ret == 4);
+ s = putRawHashValue(key, "L", live_expire, "l");
+ assert(s.ok());
+ s = putRawHashValue(key, "X", expired_at, "x");
+ assert(s.ok());
+
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 2;
+ metadata.lower = expired_at;
+ metadata.upper = live_expire;
+ s = putHashMetadata(key, metadata);
+ assert(s.ok());
+ }
+
+ void createKeeperAndGhost(const std::string &key, uint64_t ghost_expire) {
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"K", "k"}, {"G", "g"}}, false, &ret);
+ assert(s.ok());
+ assert(ret == 2);
+ s = putRawHashValue(key, "G", ghost_expire, "g");
+ assert(s.ok());
+
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 1;
+ metadata.lower = ghost_expire;
+ metadata.upper = ghost_expire;
+ s = putHashMetadata(key, metadata);
+ assert(s.ok());
+ s = deleteRawHashValue(key, "G");
+ assert(s.ok());
+ }
+
+ static void expectGetResults(const std::vector<std::string> &values, const
std::vector<rocksdb::Status> &statuses,
+ const std::vector<std::optional<std::string>>
&expected) {
+ ASSERT_EQ(values.size(), expected.size());
+ ASSERT_EQ(statuses.size(), expected.size());
+ for (size_t i = 0; i < expected.size(); ++i) {
+ if (expected[i]) {
+ EXPECT_TRUE(statuses[i].ok()) << statuses[i].ToString();
+ EXPECT_EQ(values[i], *expected[i]);
+ } else {
+ EXPECT_TRUE(statuses[i].IsNotFound()) << statuses[i].ToString();
+ }
+ }
+ }
+
+ static HashSetExOptions setExOptions(HashSetExOptions::TTLAction action,
uint64_t expire_at = 0,
+ HashFieldSetCondition condition =
HashFieldSetCondition::kNone) {
+ HashSetExOptions options;
+ options.ttl_action = action;
+ options.expire_at_ms = expire_at;
+ options.condition = condition;
+ return options;
+ }
+
+ static HashGetExOptions getExOptions(HashGetExOptions::TTLAction action,
uint64_t expire_at = 0) {
+ HashGetExOptions options;
+ options.ttl_action = action;
+ options.expire_at_ms = expire_at;
+ return options;
+ }
+
Config config_;
std::unique_ptr<engine::Storage> storage_;
std::unique_ptr<engine::Context> ctx_;
@@ -1034,6 +1135,926 @@ TEST_F(RedisHashFieldExpirationEncodingTest,
IncrementsKeepLiveTTLAndTreatExpire
EXPECT_EQ(metadata.upper, now + 120'000);
}
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetFieldsWithExpireCoversAllPhysicalStatesAndTTLChanges) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 1'000;
+ const uint64_t live_expire = now + 60'000;
+ const uint64_t new_expire = now + 120'000;
+ const std::vector<FieldValue> updates = {{"P", "new-p"}, {"L", "new-l"},
{"X", "new-x"}, {"M", "new-m"}};
+
+ {
+ const std::string key = "hsetex-state-future";
+ createFourStateHash(key, expired_at, live_expire);
+ HashMetadata before = hashMetadata(key);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kSet, new_expire);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, updates, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 5, 1, expired_at, new_expire);
+ HashMetadata after = hashMetadata(key);
+ EXPECT_EQ(after.flags, before.flags);
+ EXPECT_EQ(after.version, before.version);
+ EXPECT_EQ(after.expire, before.expire);
+ for (const auto &field : {"P", "L", "X", "M"}) {
+ EXPECT_EQ(decodedHashValue(key, field).second, new_expire);
+ }
+ EXPECT_EQ(decodedHashValue(key, "P").first, "new-p");
+ EXPECT_EQ(decodedHashValue(key, "L").first, "new-l");
+ EXPECT_EQ(decodedHashValue(key, "X").first, "new-x");
+ EXPECT_EQ(decodedHashValue(key, "M").first, "new-m");
+ EXPECT_EQ(decodedHashValue(key, "K"), (std::pair<std::string,
uint64_t>{"k", 0}));
+ }
+
+ {
+ const std::string key = "hsetex-state-discard";
+ createFourStateHash(key, expired_at, live_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, updates, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 5, 5, 0, 0);
+ for (const auto &field : {"P", "L", "X", "M", "K"}) {
+ EXPECT_EQ(decodedHashValue(key, field).second, 0);
+ }
+ }
+
+ {
+ const std::string key = "hsetex-state-keep";
+ createFourStateHash(key, expired_at, live_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kKeep);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, updates, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 5, 3, expired_at, live_expire);
+ EXPECT_EQ(decodedHashValue(key, "P"), (std::pair<std::string,
uint64_t>{"new-p", 0}));
+ EXPECT_EQ(decodedHashValue(key, "L"), (std::pair<std::string,
uint64_t>{"new-l", live_expire}));
+ EXPECT_EQ(decodedHashValue(key, "X"), (std::pair<std::string,
uint64_t>{"new-x", expired_at}));
+ EXPECT_EQ(decodedHashValue(key, "M"), (std::pair<std::string,
uint64_t>{"new-m", 0}));
+ std::string value;
+ EXPECT_TRUE(hash_->Get(*ctx_, key, "X", &value).IsNotFound());
+ }
+
+ {
+ const std::string key = "hsetex-state-immediate";
+ createFourStateHash(key, expired_at, live_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kSet, now);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, updates, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 1, 1, 0, 0);
+ std::string raw_value;
+ for (const auto &field : {"P", "L", "X", "M"}) {
+ EXPECT_TRUE(getRawHashValue(key, field, &raw_value).IsNotFound());
+ }
+ EXPECT_EQ(decodedHashValue(key, "K"), (std::pair<std::string,
uint64_t>{"k", 0}));
+ }
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
GetFieldsWithExpireCoversAllPhysicalStatesAndTTLChanges) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 1'000;
+ const uint64_t live_expire = now + 60'000;
+ const uint64_t new_expire = now + 120'000;
+ const std::vector<Slice> fields = {"P", "L", "X", "M"};
+ const std::vector<std::optional<std::string>> expected = {"p", "l",
std::nullopt, std::nullopt};
+
+ {
+ const std::string key = "hgetex-state-none";
+ createFourStateHash(key, expired_at, live_expire);
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kNone);
+ auto s = hash_->GetFieldsWithExpire(*ctx_, key, fields, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, expected);
+ expectHashMetadata(key, 3, 2, expired_at, live_expire);
+ EXPECT_EQ(decodedHashValue(key, "P"), (std::pair<std::string,
uint64_t>{"p", 0}));
+ EXPECT_EQ(decodedHashValue(key, "L"), (std::pair<std::string,
uint64_t>{"l", live_expire}));
+ }
+
+ {
+ const std::string key = "hgetex-state-future";
+ createFourStateHash(key, expired_at, live_expire);
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kSet, new_expire);
+ auto s = hash_->GetFieldsWithExpire(*ctx_, key, fields, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, expected);
+ expectHashMetadata(key, 3, 1, expired_at, new_expire);
+ EXPECT_EQ(decodedHashValue(key, "P").second, new_expire);
+ EXPECT_EQ(decodedHashValue(key, "L").second, new_expire);
+ }
+
+ {
+ const std::string key = "hgetex-state-persist";
+ createFourStateHash(key, expired_at, live_expire);
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kPersist);
+ auto s = hash_->GetFieldsWithExpire(*ctx_, key, fields, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, expected);
+ expectHashMetadata(key, 3, 3, 0, 0);
+ EXPECT_EQ(decodedHashValue(key, "P").second, 0);
+ EXPECT_EQ(decodedHashValue(key, "L").second, 0);
+ }
+
+ {
+ const std::string key = "hgetex-state-immediate";
+ createFourStateHash(key, expired_at, live_expire);
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kSet, now);
+ auto s = hash_->GetFieldsWithExpire(*ctx_, key, fields, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, expected);
+ expectHashMetadata(key, 1, 1, 0, 0);
+ std::string raw_value;
+ for (const auto &field : {"P", "L", "X"}) {
+ EXPECT_TRUE(getRawHashValue(key, field, &raw_value).IsNotFound());
+ }
+ }
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetAndGetFieldsWithExpireHandleMissingAndDeadlineBoundaries) {
+ const uint64_t now = util::GetTimeStampMS();
+ bool applied = false;
+ auto immediate_set = setExOptions(HashSetExOptions::TTLAction::kSet, now);
+ auto s =
+ hash_->SetFieldsWithExpire(*ctx_, "hsetex-missing-immediate", {{"M",
"value"}}, immediate_set, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ HashMetadata metadata(false);
+ EXPECT_TRUE(getHashMetadata("hsetex-missing-immediate",
&metadata).IsNotFound());
+
+ applied = false;
+ auto fnx_immediate = setExOptions(HashSetExOptions::TTLAction::kSet, now,
HashFieldSetCondition::kFNX);
+ s = hash_->SetFieldsWithExpire(*ctx_, "hsetex-missing-fnx-immediate", {{"M",
"value"}}, fnx_immediate, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ EXPECT_TRUE(getHashMetadata("hsetex-missing-fnx-immediate",
&metadata).IsNotFound());
+
+ const std::string key = "hgetex-equal-now";
+ uint64_t ret = 0;
+ s = hash_->MSet(*ctx_, key, {{"f", "value"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(ret, 1);
+ s = putRawHashValue(key, "f", now, "value");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ metadata = hashMetadata(key);
+ metadata.persist = 0;
+ metadata.lower = now;
+ metadata.upper = now;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto immediate_get = getExOptions(HashGetExOptions::TTLAction::kSet, now);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"f"}, immediate_get, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {"value"});
+ EXPECT_TRUE(getHashMetadata(key, &metadata).IsNotFound());
+
+ const std::string expired_key = "hgetex-only-expired";
+ s = hash_->MSet(*ctx_, expired_key, {{"X", "x"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = putRawHashValue(expired_key, "X", now - 1, "x");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ metadata = hashMetadata(expired_key);
+ metadata.persist = 0;
+ metadata.lower = now - 1;
+ metadata.upper = now - 1;
+ s = putHashMetadata(expired_key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ values.clear();
+ statuses.clear();
+ auto no_change = getExOptions(HashGetExOptions::TTLAction::kNone);
+ s = hash_->GetFieldsWithExpire(*ctx_, expired_key, {"X"}, no_change,
&values, &statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {std::nullopt});
+ EXPECT_TRUE(getHashMetadata(expired_key, &metadata).IsNotFound());
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetFieldsWithExpireConditionsAreAtomicAndCleanupInRequestOrder) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 1'000;
+ const uint64_t live_expire = now + 60'000;
+
+ {
+ const std::string key = "hsetex-fxx-success";
+ createFourStateHash(key, expired_at, live_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFXX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"P", "new-p"}, {"L",
"new-l"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 4, 3, expired_at, live_expire);
+ EXPECT_EQ(decodedHashValue(key, "P"), (std::pair<std::string,
uint64_t>{"new-p", 0}));
+ EXPECT_EQ(decodedHashValue(key, "L"), (std::pair<std::string,
uint64_t>{"new-l", 0}));
+ }
+
+ {
+ const std::string key = "hsetex-fxx-missing-failure";
+ createFourStateHash(key, expired_at, live_expire);
+ HashMetadata before = hashMetadata(key);
+ bool applied = true;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFXX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"P", "new-p"}, {"M",
"new-m"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_FALSE(applied);
+ EXPECT_EQ(hashMetadata(key), before);
+ EXPECT_EQ(decodedHashValue(key, "P").first, "p");
+ std::string raw_value;
+ EXPECT_TRUE(getRawHashValue(key, "M", &raw_value).IsNotFound());
+ }
+
+ {
+ const std::string key = "hsetex-fnx-expired-success";
+ createFourStateHash(key, expired_at, live_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFNX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"X", "new-x"}, {"M",
"new-m"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 5, 4, expired_at, live_expire);
+ EXPECT_EQ(decodedHashValue(key, "X"), (std::pair<std::string,
uint64_t>{"new-x", 0}));
+ EXPECT_EQ(decodedHashValue(key, "M"), (std::pair<std::string,
uint64_t>{"new-m", 0}));
+ }
+
+ {
+ const std::string key = "hsetex-fxx-expired-failure";
+ createFourStateHash(key, expired_at, live_expire);
+ bool applied = true;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFXX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"P", "new-p"}, {"X",
"new-x"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_FALSE(applied);
+ expectHashMetadata(key, 3, 2, expired_at, live_expire);
+ EXPECT_EQ(decodedHashValue(key, "P").first, "p");
+ std::string raw_value;
+ EXPECT_TRUE(getRawHashValue(key, "X", &raw_value).IsNotFound());
+ }
+
+ for (const auto &[name, condition, fields] :
+ std::vector<std::tuple<std::string, HashFieldSetCondition,
std::vector<FieldValue>>>{
+ {"fxx", HashFieldSetCondition::kFXX, {{"M", "new-m"}, {"X",
"new-x"}}},
+ {"fnx", HashFieldSetCondition::kFNX, {{"P", "new-p"}, {"X",
"new-x"}}},
+ }) {
+ const std::string key = "hsetex-ordered-unvisited-x-" + name;
+ createFourStateHash(key, expired_at, live_expire);
+ HashMetadata before = hashMetadata(key);
+ bool applied = true;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
condition);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, fields, options, &applied,
now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_FALSE(applied);
+ EXPECT_EQ(hashMetadata(key), before);
+ EXPECT_EQ(decodedHashValue(key, "X"), (std::pair<std::string,
uint64_t>{"x", expired_at}));
+ }
+
+ {
+ const std::string key = "hsetex-ordered-cleaned-x-before-failure";
+ createFourStateHash(key, expired_at, live_expire);
+ bool applied = true;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFNX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"X", "new-x"}, {"P",
"new-p"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_FALSE(applied);
+ expectHashMetadata(key, 3, 2, expired_at, live_expire);
+ EXPECT_EQ(decodedHashValue(key, "P").first, "p");
+ std::string raw_value;
+ EXPECT_TRUE(getRawHashValue(key, "X", &raw_value).IsNotFound());
+ }
+
+ {
+ const std::string key = "hsetex-missing-fxx";
+ bool applied = true;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFXX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"M", "value"}}, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_FALSE(applied);
+ HashMetadata metadata(false);
+ EXPECT_TRUE(getHashMetadata(key, &metadata).IsNotFound());
+ }
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetFieldsWithExpireConditionChangesExpiredKeepTTLTransition) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 1'000;
+ const uint64_t future = now + 60'000;
+
+ auto create_keeper_and_expired = [&](const std::string &key) {
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"K", "k"}, {"X", "x"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(ret, 2);
+ s = putRawHashValue(key, "X", expired_at, "x");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 1;
+ metadata.lower = expired_at;
+ metadata.upper = expired_at;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ };
+
+ {
+ const std::string key = "hsetex-x-keep-without-condition";
+ create_keeper_and_expired(key);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kKeep);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"X", "unconditional"}},
options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 2, 1, expired_at, expired_at);
+ EXPECT_EQ(decodedHashValue(key, "X"), (std::pair<std::string,
uint64_t>{"unconditional", expired_at}));
+ }
+
+ {
+ const std::string key = "hsetex-x-keep-with-fnx";
+ create_keeper_and_expired(key);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kKeep, 0,
HashFieldSetCondition::kFNX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"X", "conditional"}},
options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 2, 2, 0, 0);
+ EXPECT_EQ(decodedHashValue(key, "X"), (std::pair<std::string,
uint64_t>{"conditional", 0}));
+ }
+
+ {
+ const std::string key = "hsetex-x-future-with-fnx";
+ create_keeper_and_expired(key);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kSet, future,
HashFieldSetCondition::kFNX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"X", "conditional"}},
options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 2, 1, future, future);
+ EXPECT_EQ(decodedHashValue(key, "X"), (std::pair<std::string,
uint64_t>{"conditional", future}));
+ }
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetFieldsWithExpireDuplicatesUsePreWriteStateAndLastValue) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 1'000;
+ const uint64_t future = now + 60'000;
+
+ {
+ const std::string key = "hsetex-duplicate-missing-fnx";
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFNX);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"f", "first"}, {"f",
"last"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 1, 1, 0, 0);
+ EXPECT_EQ(decodedHashValue(key, "f"), (std::pair<std::string,
uint64_t>{"last", 0}));
+ }
+
+ {
+ const std::string key = "hsetex-duplicate-existing-fxx";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"f", "old"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kSet, future,
HashFieldSetCondition::kFXX);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"f", "first"}, {"f",
"last"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 1, 0, future, future);
+ EXPECT_EQ(decodedHashValue(key, "f"), (std::pair<std::string,
uint64_t>{"last", future}));
+ }
+
+ {
+ const std::string key = "hsetex-duplicate-expired-keep";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"K", "k"}, {"X", "x"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = putRawHashValue(key, "X", expired_at, "x");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 1;
+ metadata.lower = expired_at;
+ metadata.upper = expired_at;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kKeep);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"X", "first"}, {"X",
"last"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 2, 1, expired_at, expired_at);
+ EXPECT_EQ(decodedHashValue(key, "X"), (std::pair<std::string,
uint64_t>{"last", expired_at}));
+ }
+
+ {
+ const std::string key = "hsetex-duplicate-immediate";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"f", "old"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kSet, now);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"f", "first"}, {"f",
"last"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ HashMetadata metadata(false);
+ EXPECT_TRUE(getHashMetadata(key, &metadata).IsNotFound());
+ }
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
GetFieldsWithExpireDuplicatesUseCommandLocalState) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 1'000;
+ const uint64_t future = now + 60'000;
+
+ {
+ const std::string key = "hgetex-duplicate-immediate";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"f", "value"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kSet, now);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"f", "f"}, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {"value", std::nullopt});
+ HashMetadata metadata(false);
+ EXPECT_TRUE(getHashMetadata(key, &metadata).IsNotFound());
+ }
+
+ {
+ const std::string key = "hgetex-duplicate-persist";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"f", "value"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = putRawHashValue(key, "f", future, "value");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 0;
+ metadata.lower = future;
+ metadata.upper = future;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kPersist);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"f", "f"}, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {"value", "value"});
+ expectHashMetadata(key, 1, 1, 0, 0);
+ }
+
+ {
+ const std::string key = "hgetex-duplicate-future";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"f", "value"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kSet, future);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"f", "f"}, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {"value", "value"});
+ expectHashMetadata(key, 1, 0, future, future);
+ }
+
+ {
+ const std::string key = "hgetex-duplicate-expired";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"X", "x"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = putRawHashValue(key, "X", expired_at, "x");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 0;
+ metadata.lower = expired_at;
+ metadata.upper = expired_at;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kNone);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"X", "X"}, options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {std::nullopt, std::nullopt});
+ EXPECT_TRUE(getHashMetadata(key, &metadata).IsNotFound());
+ }
+
+ {
+ const std::string key = "hgetex-duplicate-ghost";
+ createKeeperAndGhost(key, expired_at);
+ HashMetadata before = hashMetadata(key);
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kNone);
+ auto s = hash_->GetFieldsWithExpire(*ctx_, key, {"G", "G"}, options,
&values, &statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {std::nullopt, std::nullopt});
+ EXPECT_EQ(hashMetadata(key), before);
+ }
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetAndGetFieldsWithExpireKeepConservativeBoundsUntilLastTTL) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t t10 = now + 10'000;
+ const uint64_t t20 = now + 20'000;
+ const uint64_t t25 = now + 25'000;
+ const uint64_t t30 = now + 30'000;
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kSet, t20);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, "hsetex-bounds", {{"a", "1"},
{"b", "2"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata("hsetex-bounds", 2, 0, t20, t20);
+
+ options.expire_at_ms = t30;
+ s = hash_->SetFieldsWithExpire(*ctx_, "hsetex-bounds", {{"a", "3"}},
options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectHashMetadata("hsetex-bounds", 2, 0, t20, t30);
+
+ options.expire_at_ms = t25;
+ s = hash_->SetFieldsWithExpire(*ctx_, "hsetex-bounds", {{"b", "4"}},
options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectHashMetadata("hsetex-bounds", 2, 0, t20, t30);
+
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto get_options = getExOptions(HashGetExOptions::TTLAction::kSet, t10);
+ s = hash_->GetFieldsWithExpire(*ctx_, "hsetex-bounds", {"a"}, get_options,
&values, &statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectHashMetadata("hsetex-bounds", 2, 0, t10, t30);
+
+ get_options.expire_at_ms = t20;
+ s = hash_->GetFieldsWithExpire(*ctx_, "hsetex-bounds", {"a"}, get_options,
&values, &statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectHashMetadata("hsetex-bounds", 2, 0, t10, t30);
+
+ get_options = getExOptions(HashGetExOptions::TTLAction::kPersist);
+ s = hash_->GetFieldsWithExpire(*ctx_, "hsetex-bounds", {"a"}, get_options,
&values, &statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectHashMetadata("hsetex-bounds", 2, 1, t10, t30);
+
+ options = setExOptions(HashSetExOptions::TTLAction::kDiscard);
+ s = hash_->SetFieldsWithExpire(*ctx_, "hsetex-bounds", {{"b",
"persistent"}}, options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectHashMetadata("hsetex-bounds", 2, 2, 0, 0);
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetFieldsWithExpireDoesNotConsumeCompactionGhosts) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t ghost_expire = now - 1'000;
+ const uint64_t future = now + 60'000;
+
+ {
+ const std::string key = "hsetex-ghost-persistent";
+ createKeeperAndGhost(key, ghost_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"G", "new-g"}}, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 3, 2, ghost_expire, ghost_expire);
+ EXPECT_EQ(decodedHashValue(key, "G"), (std::pair<std::string,
uint64_t>{"new-g", 0}));
+
+ uint64_t size = 0;
+ s = hash_->Size(*ctx_, key, &size, HashLengthMode::kAccurate);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_EQ(size, 2);
+ expectHashMetadata(key, 2, 2, 0, 0);
+ }
+
+ {
+ const std::string key = "hsetex-ghost-future";
+ createKeeperAndGhost(key, ghost_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kSet, future);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"G", "new-g"}}, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 3, 1, ghost_expire, future);
+ EXPECT_EQ(decodedHashValue(key, "G"), (std::pair<std::string,
uint64_t>{"new-g", future}));
+
+ uint64_t size = 0;
+ s = hash_->Size(*ctx_, key, &size, HashLengthMode::kAccurate);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_EQ(size, 2);
+ expectHashMetadata(key, 2, 1, future, future);
+ }
+
+ {
+ const std::string key = "hsetex-ghost-keep";
+ createKeeperAndGhost(key, ghost_expire);
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kKeep);
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"G", "new-g"}}, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expectHashMetadata(key, 3, 2, ghost_expire, ghost_expire);
+ EXPECT_EQ(decodedHashValue(key, "G"), (std::pair<std::string,
uint64_t>{"new-g", 0}));
+ }
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
GetFieldsWithExpireDistinguishesExpiredPhysicalFromGhost) {
+ const std::string key = "hgetex-expired-and-ghost";
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 2'000;
+ const uint64_t ghost_expire = now - 1'000;
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"X", "x"}, {"G", "g"}, {"K", "k"}},
false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(ret, 3);
+ s = putRawHashValue(key, "X", expired_at, "x");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = putRawHashValue(key, "G", ghost_expire, "g");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 1;
+ metadata.lower = expired_at;
+ metadata.upper = ghost_expire;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = deleteRawHashValue(key, "G");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto options = getExOptions(HashGetExOptions::TTLAction::kPersist);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"X", "G", "K"}, options,
&values, &statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expectGetResults(values, statuses, {std::nullopt, std::nullopt, "k"});
+ expectHashMetadata(key, 2, 1, expired_at, ghost_expire);
+ std::string raw_value;
+ EXPECT_TRUE(getRawHashValue(key, "X", &raw_value).IsNotFound());
+ EXPECT_TRUE(getRawHashValue(key, "G", &raw_value).IsNotFound());
+
+ uint64_t size = 0;
+ s = hash_->Size(*ctx_, key, &size, HashLengthMode::kAccurate);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_EQ(size, 1);
+ expectHashMetadata(key, 1, 1, 0, 0);
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetAndGetFieldsWithExpirePreserveExistingKeyMetadataIdentity) {
+ const std::string key = "hsetex-key-metadata-identity";
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t expired_at = now - 1'000;
+ const uint64_t live_expire = now + 60'000;
+ const uint64_t new_expire = now + 120'000;
+ const uint64_t key_expire = now + 600'000;
+ createFourStateHash(key, expired_at, live_expire);
+ HashMetadata metadata = hashMetadata(key);
+ metadata.expire = key_expire;
+ auto s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ const HashMetadata identity = hashMetadata(key);
+
+ auto expect_identity = [&]() {
+ HashMetadata current = hashMetadata(key);
+ EXPECT_EQ(current.flags, identity.flags);
+ EXPECT_EQ(current.mode, identity.mode);
+ EXPECT_EQ(current.version, identity.version);
+ EXPECT_EQ(current.expire, identity.expire);
+ };
+
+ bool applied = false;
+ auto set_options = setExOptions(HashSetExOptions::TTLAction::kSet,
new_expire);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"P", "new-p"}}, set_options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ expect_identity();
+
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ auto get_options = getExOptions(HashGetExOptions::TTLAction::kSet,
new_expire);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"L"}, get_options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expect_identity();
+
+ values.clear();
+ statuses.clear();
+ get_options = getExOptions(HashGetExOptions::TTLAction::kPersist);
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"L"}, get_options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expect_identity();
+
+ set_options = setExOptions(HashSetExOptions::TTLAction::kKeep);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"X", "kept-x"}}, set_options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ expect_identity();
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetFieldsWithExpireRetainsConservativeKeyTTLCornerState) {
+ const std::string key = "hsetex-key-ttl-conservative-upper";
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t short_expire = now - 1'000;
+ const uint64_t long_expire = now + 120'000;
+ const uint64_t key_expire = now + 60'000;
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"X", "x"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = putRawHashValue(key, "X", short_expire, "x");
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ metadata.persist = 0;
+ metadata.lower = short_expire;
+ metadata.upper = long_expire;
+ metadata.expire = key_expire;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ const HashMetadata original_metadata = hashMetadata(key);
+
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"new", "value"}}, options,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ metadata = hashMetadata(key);
+ EXPECT_EQ(metadata.size, 2);
+ EXPECT_EQ(metadata.persist, 1);
+ EXPECT_EQ(metadata.lower, short_expire);
+ EXPECT_EQ(metadata.upper, long_expire);
+ EXPECT_EQ(metadata.expire, original_metadata.expire);
+ EXPECT_EQ(metadata.version, original_metadata.version);
+ EXPECT_EQ(metadata.flags, original_metadata.flags);
+ EXPECT_EQ(decodedHashValue(key, "new"), (std::pair<std::string,
uint64_t>{"value", 0}));
+}
+
+TEST_F(RedisHashFieldExpirationEncodingTest,
SetFieldsWithExpireRecreatesExpiredKeyOrLeavesFXXMissing) {
+ const uint64_t now = util::GetTimeStampMS();
+ const uint64_t key_expired_at = now - 1'000;
+
+ {
+ const std::string key = "hsetex-recreate-expired-key";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"old", "value"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ const uint64_t old_version = metadata.version;
+ metadata.expire = key_expired_at;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+
+ bool applied = false;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"new", "new-value"}},
options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_TRUE(applied);
+ metadata = hashMetadata(key);
+ EXPECT_NE(metadata.version, old_version);
+ EXPECT_EQ(metadata.expire, 0);
+ EXPECT_EQ(metadata.size, 1);
+ EXPECT_EQ(metadata.persist, 1);
+ EXPECT_EQ(metadata.lower, 0);
+ EXPECT_EQ(metadata.upper, 0);
+ EXPECT_EQ(decodedHashValue(key, "new"), (std::pair<std::string,
uint64_t>{"new-value", 0}));
+ }
+
+ {
+ const std::string key = "hsetex-fxx-expired-key";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"old", "value"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ HashMetadata metadata = hashMetadata(key);
+ metadata.expire = key_expired_at;
+ s = putHashMetadata(key, metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ std::string raw_before;
+ s = db_->GetRawMetadata(*ctx_, db_->AppendNamespacePrefix(key),
&raw_before);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+
+ bool applied = true;
+ auto options = setExOptions(HashSetExOptions::TTLAction::kDiscard, 0,
HashFieldSetCondition::kFXX);
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"old", "new-value"}},
options, &applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_FALSE(applied);
+ std::string raw_after;
+ s = db_->GetRawMetadata(*ctx_, db_->AppendNamespacePrefix(key),
&raw_after);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_EQ(raw_after, raw_before);
+ }
+}
+
+TEST_F(RedisHashTest,
SetAndGetFieldsWithExpireRejectExistingLegacyHashWithoutMutation) {
+ redis::Database db(storage_.get(), "hash_ns");
+ const std::string key = "hsetex-legacy-existing";
+ uint64_t ret = 0;
+ auto s = hash_->MSet(*ctx_, key, {{"f", "value"}}, false, &ret);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(ret, 1);
+
+ HashMetadata metadata(false);
+ s = db.GetMetadata(*ctx_, {kRedisHash}, db.AppendNamespacePrefix(key),
&metadata);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(metadata.mode, HashSubkeyEncodingMode::kLegacy);
+ const std::string sub_key =
+ InternalKey(db.AppendNamespacePrefix(key), "f", metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ std::string raw_metadata_before;
+ std::string raw_value_before;
+ s = db.GetRawMetadata(*ctx_, db.AppendNamespacePrefix(key),
&raw_metadata_before);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = storage_->Get(*ctx_, ctx_->GetReadOptions(), sub_key, &raw_value_before);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(raw_value_before, "value");
+
+ std::vector<HashSetExOptions> set_options;
+ for (auto action : {HashSetExOptions::TTLAction::kDiscard,
HashSetExOptions::TTLAction::kKeep}) {
+ HashSetExOptions options;
+ options.ttl_action = action;
+ set_options.push_back(options);
+ }
+ for (uint64_t expire_at : {uint64_t{1}, util::GetTimeStampMS() + 60'000}) {
+ HashSetExOptions options;
+ options.ttl_action = HashSetExOptions::TTLAction::kSet;
+ options.expire_at_ms = expire_at;
+ set_options.push_back(options);
+ }
+ HashSetExOptions conditional;
+ conditional.ttl_action = HashSetExOptions::TTLAction::kDiscard;
+ conditional.condition = HashFieldSetCondition::kFXX;
+ set_options.push_back(conditional);
+
+ for (const auto &options : set_options) {
+ bool applied = false;
+ s = hash_->SetFieldsWithExpire(*ctx_, key, {{"f", "changed"}}, options,
&applied, util::GetTimeStampMS());
+ EXPECT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ }
+
+ std::vector<HashGetExOptions> get_options;
+ for (auto action : {HashGetExOptions::TTLAction::kNone,
HashGetExOptions::TTLAction::kPersist}) {
+ HashGetExOptions options;
+ options.ttl_action = action;
+ get_options.push_back(options);
+ }
+ for (uint64_t expire_at : {uint64_t{1}, util::GetTimeStampMS() + 60'000}) {
+ HashGetExOptions options;
+ options.ttl_action = HashGetExOptions::TTLAction::kSet;
+ options.expire_at_ms = expire_at;
+ get_options.push_back(options);
+ }
+ for (const auto &options : get_options) {
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"f"}, options, &values,
&statuses, util::GetTimeStampMS());
+ EXPECT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ }
+
+ std::string raw_metadata_after;
+ std::string raw_value_after;
+ s = db.GetRawMetadata(*ctx_, db.AppendNamespacePrefix(key),
&raw_metadata_after);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ s = storage_->Get(*ctx_, ctx_->GetReadOptions(), sub_key, &raw_value_after);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_EQ(raw_metadata_after, raw_metadata_before);
+ EXPECT_EQ(raw_value_after, raw_value_before);
+}
+
+TEST_F(RedisHashTest,
SetAndGetFieldsWithExpireHandleMissingKeyUnderLegacyConfig) {
+ redis::Database db(storage_.get(), "hash_ns");
+ const uint64_t now = util::GetTimeStampMS();
+ const std::string key = "hsetex-legacy-missing";
+
+ for (auto condition : {HashFieldSetCondition::kNone,
HashFieldSetCondition::kFNX}) {
+ for (auto action : {HashSetExOptions::TTLAction::kDiscard,
HashSetExOptions::TTLAction::kKeep,
+ HashSetExOptions::TTLAction::kSet}) {
+ HashSetExOptions options;
+ options.ttl_action = action;
+ options.expire_at_ms = action == HashSetExOptions::TTLAction::kSet ? now
: 0;
+ options.condition = condition;
+ bool applied = false;
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"f", "value"}},
options, &applied, now);
+ EXPECT_TRUE(s.IsInvalidArgument()) << s.ToString();
+ std::string raw_metadata;
+ EXPECT_TRUE(db.GetRawMetadata(*ctx_, db.AppendNamespacePrefix(key),
&raw_metadata).IsNotFound());
+ }
+ }
+
+ HashSetExOptions fxx;
+ fxx.ttl_action = HashSetExOptions::TTLAction::kDiscard;
+ fxx.condition = HashFieldSetCondition::kFXX;
+ bool applied = true;
+ auto s = hash_->SetFieldsWithExpire(*ctx_, key, {{"f", "value"}}, fxx,
&applied, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ EXPECT_FALSE(applied);
+
+ std::vector<std::string> values;
+ std::vector<rocksdb::Status> statuses;
+ HashGetExOptions get_options;
+ get_options.ttl_action = HashGetExOptions::TTLAction::kSet;
+ get_options.expire_at_ms = now + 60'000;
+ s = hash_->GetFieldsWithExpire(*ctx_, key, {"a", "b"}, get_options, &values,
&statuses, now);
+ ASSERT_TRUE(s.ok()) << s.ToString();
+ ASSERT_EQ(values.size(), 2);
+ ASSERT_EQ(statuses.size(), 2);
+ EXPECT_TRUE(statuses[0].IsNotFound());
+ EXPECT_TRUE(statuses[1].IsNotFound());
+}
+
TEST_F(RedisHashTest, HIncr) {
int64_t value = 0;
Slice field("hash-incrby-invalid-field");
diff --git
a/tests/gocase/integration/replication/hash_field_expiration_commands_test.go
b/tests/gocase/integration/replication/hash_field_expiration_commands_test.go
new file mode 100644
index 000000000..db3138268
--- /dev/null
+++
b/tests/gocase/integration/replication/hash_field_expiration_commands_test.go
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package replication
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/redis/go-redis/v9"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/kvrocks/tests/gocase/util"
+)
+
+func requireHFEArray(t *testing.T, got interface{}, want ...interface{}) {
+ t.Helper()
+
+ values, ok := got.([]interface{})
+ require.Truef(t, ok, "expected []interface{}, got %T", got)
+ require.Equal(t, want, values)
+}
+
+func waitForHashFieldToExpire(t *testing.T, rdb *redis.Client, ctx
context.Context, key, field string) {
+ t.Helper()
+
+ require.Eventually(t, func() bool {
+ return errors.Is(rdb.HGet(ctx, key, field).Err(), redis.Nil)
+ }, 5*time.Second, 50*time.Millisecond)
+}
+
+func TestHashFieldExpirationHSetExHGetExReplication(t *testing.T) {
+ configs := util.KvrocksServerConfigs{
+ "hash-encoding-mode": "field-expiration",
+ "rocksdb.disable_auto_compactions": "yes",
+ "resp3-enabled": "yes",
+ }
+ master := util.StartServer(t, configs)
+ defer master.Close()
+ masterClient := master.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+
+ replica := util.StartServer(t, configs)
+ defer replica.Close()
+ replicaClient := replica.NewClient()
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+
+ ctx := context.Background()
+ util.SlaveOf(t, replicaClient, master)
+ util.WaitForSync(t, replicaClient)
+
+ key := "hsetex-hgetex-replication"
+ result, err := masterClient.Do(ctx, "hsetex", key, "PX", 600000,
"FIELDS", 3,
+ "a", "1", "b", "2", "c", "3").Int64()
+ require.NoError(t, err)
+ require.Equal(t, int64(1), result)
+ util.WaitForOffsetSync(t, masterClient, replicaClient, 5*time.Second)
+
+ masterExpire := masterClient.Do(ctx, "hpexpiretime", key, "FIELDS", 3,
"a", "b", "c").Val()
+ replicaExpire := replicaClient.Do(ctx, "hpexpiretime", key, "FIELDS",
3, "a", "b", "c").Val()
+ require.Equal(t, masterExpire, replicaExpire)
+ expires := masterExpire.([]interface{})
+ require.Equal(t, expires[0], expires[1])
+ require.Equal(t, expires[1], expires[2])
+ require.Equal(t, util.GetKMetadata(t, masterClient, ctx, key),
util.GetKMetadata(t, replicaClient, ctx, key))
+
+ got, err := masterClient.Do(ctx, "hgetex", key, "PERSIST", "FIELDS", 2,
"a", "missing").Result()
+ require.NoError(t, err)
+ requireHFEArray(t, got, "1", nil)
+ result, err = masterClient.Do(ctx, "hsetex", key, "KEEPTTL", "FIELDS",
2, "b", "20", "d", "40").Int64()
+ require.NoError(t, err)
+ require.Equal(t, int64(1), result)
+ expireAt := time.Now().Add(20 * time.Minute).UnixMilli()
+ got, err = masterClient.Do(ctx, "hgetex", key, "PXAT", expireAt,
"FIELDS", 2, "a", "c").Result()
+ require.NoError(t, err)
+ requireHFEArray(t, got, "1", "3")
+ util.WaitForOffsetSync(t, masterClient, replicaClient, 5*time.Second)
+
+ require.Equal(t, masterClient.HGetAll(ctx, key).Val(),
replicaClient.HGetAll(ctx, key).Val())
+ require.Equal(t, masterClient.Do(ctx, "hpexpiretime", key, "FIELDS", 4,
"a", "b", "c", "d").Val(),
+ replicaClient.Do(ctx, "hpexpiretime", key, "FIELDS", 4, "a",
"b", "c", "d").Val())
+ require.Equal(t, util.GetKMetadata(t, masterClient, ctx, key),
util.GetKMetadata(t, replicaClient, ctx, key))
+
+ cleanupKey := "hsetex-condition-cleanup-replication"
+ require.Equal(t, int64(2), masterClient.HSet(ctx, cleanupKey,
"expired", "value", "keeper", "value").Val())
+ require.Equal(t, []interface{}{int64(1)}, masterClient.Do(ctx,
"hexpire", cleanupKey, 1, "FIELDS", 1, "expired").Val())
+ util.WaitForOffsetSync(t, masterClient, replicaClient, 5*time.Second)
+ waitForHashFieldToExpire(t, masterClient, ctx, cleanupKey, "expired")
+
+ result, err = masterClient.Do(ctx, "hsetex", cleanupKey, "FXX",
"FIELDS", 1, "expired", "new").Int64()
+ require.NoError(t, err)
+ require.Equal(t, int64(0), result)
+ util.WaitForOffsetSync(t, masterClient, replicaClient, 5*time.Second)
+ require.Equal(t, map[string]string{"keeper": "value"},
replicaClient.HGetAll(ctx, cleanupKey).Val())
+ require.Equal(t, util.GetKMetadata(t, masterClient, ctx, cleanupKey),
+ util.GetKMetadata(t, replicaClient, ctx, cleanupKey))
+}
diff --git a/tests/gocase/unit/type/hash/hash_hsetex_hgetex_test.go
b/tests/gocase/unit/type/hash/hash_hsetex_hgetex_test.go
new file mode 100644
index 000000000..e8af8d82c
--- /dev/null
+++ b/tests/gocase/unit/type/hash/hash_hsetex_hgetex_test.go
@@ -0,0 +1,796 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package hash
+
+import (
+ "context"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/redis/go-redis/v9"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/kvrocks/tests/gocase/util"
+)
+
+func requireIntegerReply(t *testing.T, rdb *redis.Client, ctx context.Context,
want int64, args ...interface{}) {
+ t.Helper()
+
+ got, err := rdb.Do(ctx, args...).Int64()
+ require.NoError(t, err)
+ require.Equal(t, want, got)
+}
+
+func requireOptionalStringArray(t *testing.T, got interface{}, want
...interface{}) {
+ t.Helper()
+
+ values, ok := got.([]interface{})
+ require.Truef(t, ok, "expected []interface{}, got %T", got)
+ require.Equal(t, want, values)
+}
+
+func requireMetadataHeaderUnchanged(t *testing.T, before, after
util.KMetadataResponse) {
+ t.Helper()
+
+ require.Equal(t, before.Flags, after.Flags)
+ require.Equal(t, before.Mode, after.Mode)
+ require.Equal(t, before.Version, after.Version)
+ require.Equal(t, before.Expire, after.Expire)
+}
+
+func TestHashFieldExpirationHSetExStateMatrix(t *testing.T) {
+ runWithFieldExpirationHashConfigs(t, util.KvrocksServerConfigs{
+ "rocksdb.disable_auto_compactions": "yes",
+ }, func(t *testing.T, rdb *redis.Client, ctx context.Context) {
+ t.Run("future deadline", func(t *testing.T) {
+ key := "hsetex-state-future"
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+ expireAt := time.Now().Add(10 * time.Minute).UnixMilli()
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"PXAT", expireAt, "FIELDS", 4,
+ hfePersistentField, "p2", hfeLiveField, "l2",
hfeExpiredField, "x2", hfeMissingField, "m2")
+
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, after, 5, 1)
+ require.Equal(t, before.Lower, after.Lower)
+ require.Equal(t, expireAt, after.Upper)
+ requireMetadataHeaderUnchanged(t, before, after)
+ requireHashValues(t, rdb, ctx, key, map[string]string{
+ hfePersistentField: "p2",
+ hfeLiveField: "l2",
+ hfeExpiredField: "x2",
+ hfeMissingField: "m2",
+ hfeKeeperField: "40",
+ })
+ requireIntArray(t, rdb.Do(ctx, "hpexpiretime", key,
"FIELDS", 4,
+ hfePersistentField, hfeLiveField,
hfeExpiredField, hfeMissingField).Val(),
+ []int64{expireAt, expireAt, expireAt, expireAt})
+ })
+
+ t.Run("discard ttl", func(t *testing.T) {
+ key := "hsetex-state-discard"
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FIELDS", 4,
+ hfePersistentField, "p2", hfeLiveField, "l2",
hfeExpiredField, "x2", hfeMissingField, "m2")
+
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, after, 5, 5)
+ requireMetadataHeaderUnchanged(t, before, after)
+ requireHashValues(t, rdb, ctx, key, map[string]string{
+ hfePersistentField: "p2",
+ hfeLiveField: "l2",
+ hfeExpiredField: "x2",
+ hfeMissingField: "m2",
+ hfeKeeperField: "40",
+ })
+ requireIntArray(t, rdb.Do(ctx, "hpexpiretime", key,
"FIELDS", 4,
+ hfePersistentField, hfeLiveField,
hfeExpiredField, hfeMissingField).Val(),
+ []int64{-1, -1, -1, -1})
+ })
+
+ t.Run("keep ttl", func(t *testing.T) {
+ key := "hsetex-state-keep"
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"KEEPTTL", "FIELDS", 4,
+ hfePersistentField, "p2", hfeLiveField, "l2",
hfeExpiredField, "x2", hfeMissingField, "m2")
+
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, after, 5, 3)
+ require.Equal(t, before.Lower, after.Lower)
+ require.Equal(t, before.Upper, after.Upper)
+ requireMetadataHeaderUnchanged(t, before, after)
+ require.Equal(t, "p2", rdb.HGet(ctx, key,
hfePersistentField).Val())
+ require.Equal(t, "l2", rdb.HGet(ctx, key,
hfeLiveField).Val())
+ require.ErrorIs(t, rdb.HGet(ctx, key,
hfeExpiredField).Err(), redis.Nil)
+ require.Equal(t, "m2", rdb.HGet(ctx, key,
hfeMissingField).Val())
+ })
+
+ t.Run("past deadline", func(t *testing.T) {
+ key := "hsetex-state-past"
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"PXAT", 1, "FIELDS", 4,
+ hfePersistentField, "p2", hfeLiveField, "l2",
hfeExpiredField, "x2", hfeMissingField, "m2")
+
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, after, 1, 1)
+ requireMetadataHeaderUnchanged(t, before, after)
+ require.Equal(t, map[string]string{hfeKeeperField:
"40"}, rdb.HGetAll(ctx, key).Val())
+ })
+ })
+}
+
+func TestHashFieldExpirationHGetExStateMatrix(t *testing.T) {
+ runWithFieldExpirationHashConfigs(t, util.KvrocksServerConfigs{
+ "rocksdb.disable_auto_compactions": "yes",
+ }, func(t *testing.T, rdb *redis.Client, ctx context.Context) {
+ tests := []struct {
+ name string
+ option []interface{}
+ wantSize int64
+ wantPersist int64
+ }{
+ {name: "no modifier", wantSize: 3, wantPersist: 2},
+ {name: "persist", option: []interface{}{"PERSIST"},
wantSize: 3, wantPersist: 3},
+ {name: "past deadline", option: []interface{}{"PXAT",
int64(1)}, wantSize: 1, wantPersist: 1},
+ }
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ key := "hgetex-state-" + test.name
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+ args := []interface{}{"hgetex", key}
+ args = append(args, test.option...)
+ args = append(args, "FIELDS", 4,
hfePersistentField, hfeLiveField, hfeExpiredField, hfeMissingField)
+
+ got, err := rdb.Do(ctx, args...).Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "10", "20",
nil, nil)
+
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, after, test.wantSize,
test.wantPersist)
+ requireMetadataHeaderUnchanged(t, before, after)
+ if test.name == "no modifier" {
+ require.Equal(t, before.Lower,
after.Lower)
+ require.Equal(t, before.Upper,
after.Upper)
+ }
+ })
+ }
+
+ t.Run("future deadline", func(t *testing.T) {
+ key := "hgetex-state-future"
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+ expireAt := time.Now().Add(10 * time.Minute).UnixMilli()
+
+ got, err := rdb.Do(ctx, "hgetex", key, "PXAT",
expireAt, "FIELDS", 4,
+ hfePersistentField, hfeLiveField,
hfeExpiredField, hfeMissingField).Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "10", "20", nil, nil)
+
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, after, 3, 1)
+ require.Equal(t, before.Lower, after.Lower)
+ require.Equal(t, expireAt, after.Upper)
+ requireMetadataHeaderUnchanged(t, before, after)
+ })
+
+ t.Run("empty string differs from missing", func(t *testing.T) {
+ key := "hgetex-empty"
+ require.Equal(t, int64(1), rdb.HSet(ctx, key, "empty",
"").Val())
+ got, err := rdb.Do(ctx, "hgetex", key, "FIELDS", 2,
"empty", "missing").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "", nil)
+ })
+ })
+}
+
+func TestHashFieldExpirationHSetExConditionsAndDuplicates(t *testing.T) {
+ runWithFieldExpirationHashConfigs(t, util.KvrocksServerConfigs{
+ "rocksdb.disable_auto_compactions": "yes",
+ }, func(t *testing.T, rdb *redis.Client, ctx context.Context) {
+ t.Run("FXX is atomic", func(t *testing.T) {
+ key := "hsetex-fxx"
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+
+ requireIntegerReply(t, rdb, ctx, 0, "hsetex", key,
"FXX", "FIELDS", 2,
+ hfePersistentField, "changed", hfeMissingField,
"created")
+ require.Equal(t, "10", rdb.HGet(ctx, key,
hfePersistentField).Val())
+ require.ErrorIs(t, rdb.HGet(ctx, key,
hfeMissingField).Err(), redis.Nil)
+ require.Equal(t, before, util.GetKMetadata(t, rdb, ctx,
key))
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FXX", "FIELDS", 2,
+ hfePersistentField, "p2", hfeLiveField, "l2")
+ require.Equal(t, "p2", rdb.HGet(ctx, key,
hfePersistentField).Val())
+ require.Equal(t, "l2", rdb.HGet(ctx, key,
hfeLiveField).Val())
+ })
+
+ t.Run("ordered expired cleanup", func(t *testing.T) {
+ key := "hsetex-ordered-cleanup"
+ createHashFieldStates(t, rdb, ctx, key)
+ before := util.GetKMetadata(t, rdb, ctx, key)
+
+ requireIntegerReply(t, rdb, ctx, 0, "hsetex", key,
"FNX", "FIELDS", 2,
+ hfeExpiredField, "x2", hfePersistentField, "p2")
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, after, 3, 2)
+ require.Equal(t, before.Lower, after.Lower)
+ require.Equal(t, before.Upper, after.Upper)
+ require.Equal(t, "10", rdb.HGet(ctx, key,
hfePersistentField).Val())
+ require.ErrorIs(t, rdb.HGet(ctx, key,
hfeExpiredField).Err(), redis.Nil)
+
+ key = "hsetex-no-late-cleanup"
+ createHashFieldStates(t, rdb, ctx, key)
+ before = util.GetKMetadata(t, rdb, ctx, key)
+ requireIntegerReply(t, rdb, ctx, 0, "hsetex", key,
"FNX", "FIELDS", 2,
+ hfePersistentField, "p2", hfeExpiredField, "x2")
+ require.Equal(t, before, util.GetKMetadata(t, rdb, ctx,
key))
+ })
+
+ t.Run("FNX rebuilds expired as missing", func(t *testing.T) {
+ key := "hsetex-fnx-expired"
+ createHashFieldStates(t, rdb, ctx, key)
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FNX", "KEEPTTL", "FIELDS", 2,
+ hfeExpiredField, "x2", hfeMissingField, "m2")
+ require.Equal(t, "x2", rdb.HGet(ctx, key,
hfeExpiredField).Val())
+ require.Equal(t, "m2", rdb.HGet(ctx, key,
hfeMissingField).Val())
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx,
key), 5, 4)
+ })
+
+ t.Run("duplicates use one transition and last value", func(t
*testing.T) {
+ key := "hsetex-duplicates"
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FNX", "FIELDS", 3,
+ "field", "first", "field", "second", "field",
"last")
+ require.Equal(t, "last", rdb.HGet(ctx, key,
"field").Val())
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx,
key), 1, 1)
+
+ expireAt := time.Now().Add(10 * time.Minute).UnixMilli()
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FXX", "PXAT", expireAt, "FIELDS", 2,
+ "field", "next", "field", "final")
+ require.Equal(t, "final", rdb.HGet(ctx, key,
"field").Val())
+ meta := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, meta, 1, 0)
+ require.Equal(t, expireAt, meta.Lower)
+ require.Equal(t, expireAt, meta.Upper)
+ })
+ })
+}
+
+func TestHashFieldExpirationHGetExDuplicates(t *testing.T) {
+ runWithFieldExpirationHashConfigs(t, util.KvrocksServerConfigs{
+ "rocksdb.disable_auto_compactions": "yes",
+ }, func(t *testing.T, rdb *redis.Client, ctx context.Context) {
+ t.Run("past deadline returns value then null", func(t
*testing.T) {
+ key := "hgetex-duplicate-delete"
+ require.Equal(t, int64(1), rdb.HSet(ctx, key, "field",
"value").Val())
+ got, err := rdb.Do(ctx, "hgetex", key, "PXAT", 1,
"FIELDS", 2, "field", "field").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "value", nil)
+ require.Equal(t, int64(0), rdb.Exists(ctx, key).Val())
+ })
+
+ t.Run("persist changes counter once", func(t *testing.T) {
+ key := "hgetex-duplicate-persist"
+ require.Equal(t, int64(1), rdb.HSet(ctx, key, "field",
"value").Val())
+ requireIntArray(t, rdb.Do(ctx, "hexpire", key, 300,
"FIELDS", 1, "field").Val(), []int64{1})
+ got, err := rdb.Do(ctx, "hgetex", key, "PERSIST",
"FIELDS", 2, "field", "field").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "value", "value")
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx,
key), 1, 1)
+ })
+
+ t.Run("expired physical cleans once", func(t *testing.T) {
+ key := "hgetex-duplicate-expired"
+ require.Equal(t, int64(2), rdb.HSet(ctx, key, "field",
"value", "keeper", "value").Val())
+ requireIntArray(t, rdb.Do(ctx, "hexpire", key, 1,
"FIELDS", 1, "field").Val(), []int64{1})
+ waitHashFieldExpired(t, rdb, ctx, key, "field")
+ got, err := rdb.Do(ctx, "hgetex", key, "FIELDS", 2,
"field", "field").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, nil, nil)
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx,
key), 1, 1)
+ })
+ })
+}
+
+func TestHashFieldExpirationHSetExHGetExParserCompatibility(t *testing.T) {
+ runWithFieldExpirationHash(t, func(t *testing.T, rdb *redis.Client, ctx
context.Context) {
+ key := "hsetex-hgetex-parser"
+ require.Equal(t, int64(2), rdb.HSet(ctx, key, "a", "1", "b",
"2").Val())
+
+ tests := []struct {
+ name string
+ args []interface{}
+ errContains string
+ exactError string
+ }{
+ {name: "hsetex repeated fields", args:
[]interface{}{"hsetex", key, "FIELDS", 1, "a", "x", "FIELDS", 1, "b", "y"},
errContains: "FIELDS keyword specified multiple times"},
+ {name: "hgetex repeated fields", args:
[]interface{}{"hgetex", key, "FIELDS", 1, "a", "FIELDS", 1, "b"}, errContains:
"FIELDS keyword specified multiple times"},
+ {name: "hsetex zero fields", args:
[]interface{}{"hsetex", key, "FIELDS", 0, "a", "x"}, errContains: "invalid
number of fields"},
+ {name: "hgetex negative fields", args:
[]interface{}{"hgetex", key, "FIELDS", -1, "a"}, errContains: "invalid number
of fields"},
+ {name: "hsetex invalid fields integer", args:
[]interface{}{"hsetex", key, "FIELDS", "invalid", "a", "x"}, errContains:
"invalid number of fields"},
+ {name: "hgetex fields integer overflow", args:
[]interface{}{"hgetex", key, "FIELDS", "9223372036854775808", "a"},
errContains: "invalid number of fields"},
+ {name: "hsetex short block", args:
[]interface{}{"hsetex", key, "FIELDS", 2, "a", "x"}, errContains: "wrong number
of arguments"},
+ {name: "hgetex short block", args:
[]interface{}{"hgetex", key, "FIELDS", 2, "a"}, errContains: "wrong number of
arguments"},
+ {name: "hsetex int max short block", args:
[]interface{}{"hsetex", key, "FIELDS", 2147483647, "a", "x"}, errContains:
"wrong number of arguments"},
+ {name: "hgetex int max short block", args:
[]interface{}{"hgetex", key, "FIELDS", 2147483647, "a"}, errContains: "wrong
number of arguments"},
+ {name: "hsetex fields above int max", args:
[]interface{}{"hsetex", key, "FIELDS", "2147483648", "a", "x"}, errContains:
"invalid number of fields"},
+ {name: "hgetex fields above int max", args:
[]interface{}{"hgetex", key, "FIELDS", "2147483648", "a"}, errContains:
"invalid number of fields"},
+ {name: "hsetex missing field count", args:
[]interface{}{"hsetex", key, "FNX", "EX", 10, "FIELDS"}, exactError: "ERR wrong
number of arguments"},
+ {name: "hgetex missing field count", args:
[]interface{}{"hgetex", key, "EX", 10, "FIELDS"}, exactError: "ERR wrong number
of arguments"},
+ {name: "hsetex duplicate fields before missing count",
args: []interface{}{"hsetex", key, "FIELDS", 1, "a", "x", "FIELDS"},
errContains: "FIELDS keyword specified multiple times"},
+ {name: "hgetex duplicate fields before missing count",
args: []interface{}{"hgetex", key, "FIELDS", 1, "a", "FIELDS"}, errContains:
"FIELDS keyword specified multiple times"},
+ {name: "hsetex unknown trailing token", args:
[]interface{}{"hsetex", key, "FIELDS", 1, "a", "x", "unknown"}, errContains:
"unknown argument: unknown"},
+ {name: "hgetex unknown trailing token", args:
[]interface{}{"hgetex", key, "FIELDS", 1, "a", "unknown"}, errContains:
"unknown argument: unknown"},
+ {name: "hsetex condition conflict", args:
[]interface{}{"hsetex", key, "FXX", "FNX", "FIELDS", 1, "a", "x"}, errContains:
"Only one of FXX or FNX"},
+ {name: "hsetex repeated fxx", args:
[]interface{}{"hsetex", key, "FXX", "FXX", "FIELDS", 1, "a", "x"}, errContains:
"Only one of FXX or FNX"},
+ {name: "hsetex repeated fnx", args:
[]interface{}{"hsetex", key, "FNX", "FNX", "FIELDS", 1, "missing", "x"},
errContains: "Only one of FXX or FNX"},
+ {name: "hsetex expiration conflict", args:
[]interface{}{"hsetex", key, "EX", 10, "KEEPTTL", "FIELDS", 1, "a", "x"},
errContains: "Only one of EX, PX, EXAT, PXAT or KEEPTTL"},
+ {name: "hsetex repeated expiration", args:
[]interface{}{"hsetex", key, "EX", 10, "EX", 20, "FIELDS", 1, "a", "x"},
errContains: "Only one of EX, PX, EXAT, PXAT or KEEPTTL"},
+ {name: "hsetex repeated keepttl", args:
[]interface{}{"hsetex", key, "KEEPTTL", "KEEPTTL", "FIELDS", 1, "a", "x"},
errContains: "Only one of EX, PX, EXAT, PXAT or KEEPTTL"},
+ {name: "hgetex expiration conflict", args:
[]interface{}{"hgetex", key, "PERSIST", "PX", 10, "FIELDS", 1, "a"},
errContains: "Only one of EX, PX, EXAT, PXAT or PERSIST"},
+ {name: "hgetex repeated expiration", args:
[]interface{}{"hgetex", key, "PX", 10, "PX", 20, "FIELDS", 1, "a"},
errContains: "Only one of EX, PX, EXAT, PXAT or PERSIST"},
+ {name: "hgetex repeated persist", args:
[]interface{}{"hgetex", key, "PERSIST", "PERSIST", "FIELDS", 1, "a"},
errContains: "Only one of EX, PX, EXAT, PXAT or PERSIST"},
+ {name: "hsetex missing fields", args:
[]interface{}{"hsetex", key, "EX", 10, "UNKNOWN", "x"}, errContains: "unknown
argument: UNKNOWN"},
+ {name: "hgetex missing fields", args:
[]interface{}{"hgetex", key, "EX", 10, "UNKNOWN"}, errContains: "unknown
argument: UNKNOWN"},
+ {name: "hsetex missing expire", args:
[]interface{}{"hsetex", key, "FIELDS", 1, "a", "x", "EX"}, errContains:
"missing expire time"},
+ {name: "hgetex missing expire", args:
[]interface{}{"hgetex", key, "FIELDS", 1, "a", "EX"}, errContains: "missing
expire time"},
+ {name: "hsetex conflict before missing expire", args:
[]interface{}{"hsetex", key, "EX", 10, "FIELDS", 1, "a", "x", "PX"},
errContains: "Only one of EX, PX, EXAT, PXAT or KEEPTTL"},
+ {name: "hgetex conflict before missing expire", args:
[]interface{}{"hgetex", key, "PERSIST", "FIELDS", 1, "a", "PX"}, errContains:
"Only one of EX, PX, EXAT, PXAT or PERSIST"},
+ {name: "hsetex fields consumed as time", args:
[]interface{}{"hsetex", key, "EX", "FIELDS", 1, "a", "x"}, errContains: "value
is not an integer or out of range"},
+ {name: "hgetex fields consumed as time", args:
[]interface{}{"hgetex", key, "EX", "FIELDS", 1, "a"}, errContains: "value is
not an integer or out of range"},
+ {name: "hsetex negative expire", args:
[]interface{}{"hsetex", key, "EX", -1, "FIELDS", 1, "a", "x"}, errContains:
"invalid expire time, must be >= 0"},
+ {name: "hgetex negative expire", args:
[]interface{}{"hgetex", key, "PX", -1, "FIELDS", 1, "a"}, errContains: "invalid
expire time, must be >= 0"},
+ {name: "hsetex expire integer overflow", args:
[]interface{}{"hsetex", key, "PXAT", "9223372036854775808", "FIELDS", 1, "a",
"x"}, errContains: "value is not an integer or out of range"},
+ {name: "hsetex absolute expire too large", args:
[]interface{}{"hsetex", key, "PXAT", hfeMaxAbsTimeMs + 1, "FIELDS", 1, "a",
"x"}, exactError: "ERR invalid expire time"},
+ {name: "hgetex absolute expire too large", args:
[]interface{}{"hgetex", key, "PXAT", hfeMaxAbsTimeMs + 1, "FIELDS", 1, "a"},
exactError: "ERR invalid expire time"},
+ {name: "hsetex option from hgetex", args:
[]interface{}{"hsetex", key, "PERSIST", "FIELDS", 1, "a", "x"}, errContains:
"unknown argument: PERSIST"},
+ {name: "hgetex option from hsetex", args:
[]interface{}{"hgetex", key, "KEEPTTL", "FIELDS", 1, "a"}, errContains:
"unknown argument: KEEPTTL"},
+ }
+
+ for _, test := range tests {
+ t.Run(test.name, func(t *testing.T) {
+ before := util.GetKMetadata(t, rdb, ctx, key)
+ valuesBefore := rdb.HGetAll(ctx, key).Val()
+ err := rdb.Do(ctx, test.args...).Err()
+ if test.exactError != "" {
+ require.EqualError(t, err,
test.exactError)
+ } else {
+ require.ErrorContains(t, err,
test.errContains)
+ }
+ require.Equal(t, valuesBefore, rdb.HGetAll(ctx,
key).Val())
+ require.Equal(t, before, util.GetKMetadata(t,
rdb, ctx, key))
+ })
+ }
+
+ t.Run("flexible ordering and option-like data", func(t
*testing.T) {
+ expireAt := time.Now().Add(10 * time.Minute).UnixMilli()
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"PXAT", expireAt, "FIELDS", 1, "a", "3")
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FIELDS", 1, "a", "5", "FXX", "KEEPTTL")
+ requireIntArray(t, rdb.Do(ctx, "hpexpiretime", key,
"FIELDS", 1, "a").Val(), []int64{expireAt})
+
+ requireIntegerReply(t, rdb, ctx, 0, "hsetex", key,
"FIELDS", 1, "missing-after-fields", "x", "FXX")
+ require.False(t, rdb.HExists(ctx, key,
"missing-after-fields").Val())
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FIELDS", 2, "EX", "60", "FIELDS", "value")
+ require.Equal(t, "60", rdb.HGet(ctx, key, "EX").Val())
+ require.Equal(t, "value", rdb.HGet(ctx, key,
"FIELDS").Val())
+
+ got, err := rdb.Do(ctx, "hgetex", key, "FIELDS", 2,
"EX", "FIELDS", "PERSIST").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "60", "value")
+
+ expireAt = time.Now().Add(20 * time.Minute).UnixMilli()
+ got, err = rdb.Do(ctx, "hgetex", key, "FIELDS", 1, "a",
"PXAT", expireAt).Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "5")
+ requireIntArray(t, rdb.Do(ctx, "hpexpiretime", key,
"FIELDS", 1, "a").Val(), []int64{expireAt})
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"FIELDS", 2,
+ "ordinary", "FIELDS", "another", "EX")
+ require.Equal(t, "FIELDS", rdb.HGet(ctx, key,
"ordinary").Val())
+ require.Equal(t, "EX", rdb.HGet(ctx, key,
"another").Val())
+ })
+
+ t.Run("keywords are case insensitive", func(t *testing.T) {
+ expireAt := time.Now().Add(30 * time.Minute).UnixMilli()
+ requireIntegerReply(t, rdb, ctx, 1, "HsEtEx", key,
"pXaT", expireAt, "fIeLdS", 1, "a", "case")
+ requireIntegerReply(t, rdb, ctx, 1, "HsEtEx", key,
"fXx", "fIeLdS", 1, "a", "4", "kEePtTl")
+ requireIntArray(t, rdb.Do(ctx, "hpexpiretime", key,
"FIELDS", 1, "a").Val(), []int64{expireAt})
+ got, err := rdb.Do(ctx, "HgEtEx", key, "fIeLdS", 1,
"a", "pErSiSt").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "4")
+ requireIntArray(t, rdb.Do(ctx, "hpexpiretime", key,
"FIELDS", 1, "a").Val(), []int64{-1})
+ })
+ })
+}
+
+func TestHashFieldExpirationHSetExHGetExParsingAndCommandFlags(t *testing.T) {
+ runWithFieldExpirationHash(t, func(t *testing.T, rdb *redis.Client, ctx
context.Context) {
+ require.EqualError(t, rdb.Do(ctx, "hsetex", "key").Err(),
+ "ERR wrong number of arguments")
+ require.EqualError(t, rdb.Do(ctx, "hgetex", "key",
"FIELDS").Err(),
+ "ERR wrong number of arguments")
+
+ wrongTypeKey := "hsetex-hgetex-wrongtype"
+ require.NoError(t, rdb.Set(ctx, wrongTypeKey, "value", 0).Err())
+
+ hsetErr := rdb.Do(ctx, "hsetex", wrongTypeKey, "FIELDS", 0,
"a", "x").Err()
+ require.ErrorContains(t, hsetErr, "invalid number of fields")
+ require.NotContains(t, hsetErr.Error(), "WRONGTYPE")
+
+ hgetErr := rdb.Do(ctx, "hgetex", wrongTypeKey, "FIELDS", 0,
"a").Err()
+ require.ErrorContains(t, hgetErr, "invalid number of fields")
+ require.NotContains(t, hgetErr.Error(), "WRONGTYPE")
+
+ require.ErrorContains(t, rdb.Do(ctx, "hsetex", wrongTypeKey,
"FIELDS", 1, "a", "x").Err(), "WRONGTYPE")
+ require.ErrorContains(t, rdb.Do(ctx, "hgetex", wrongTypeKey,
"FIELDS", 1, "a").Err(), "WRONGTYPE")
+
+ t.Run("parse error aborts multi", func(t *testing.T) {
+ key := "hgetex-parse-error-multi"
+ require.NoError(t, rdb.Del(ctx, key).Err())
+ require.NoError(t, rdb.Do(ctx, "MULTI").Err())
+ require.Equal(t, "QUEUED", rdb.Do(ctx, "SET", key,
"value").Val())
+ require.ErrorContains(t, rdb.Do(ctx, "HGETEX",
wrongTypeKey, "FIELDS", 0, "a").Err(),
+ "invalid number of fields")
+ require.EqualError(t, rdb.Do(ctx, "EXEC").Err(),
"EXECABORT Transaction discarded")
+ require.Zero(t, rdb.Exists(ctx, key).Val())
+ })
+
+ for _, test := range []struct {
+ command string
+ arity int64
+ flag string
+ }{
+ {command: "hsetex", arity: -6, flag: "write"},
+ {command: "hgetex", arity: -5, flag: "no-dbsize-check"},
+ } {
+ info, err := rdb.Do(ctx, "command", "info",
test.command).Slice()
+ require.NoError(t, err)
+ require.Len(t, info, 1)
+ commandInfo := info[0].([]interface{})
+ require.Equal(t, test.arity, commandInfo[1])
+ require.Contains(t, commandInfo[2].([]interface{}),
test.flag)
+ }
+
+ key := "hsetex-hgetex-eval-ro"
+ require.Equal(t, int64(1), rdb.HSet(ctx, key, "a", "1").Val())
+ require.ErrorContains(t,
+ rdb.Do(ctx, "eval_ro", `return redis.call('hsetex',
KEYS[1], 'FIELDS', 1, 'a', '2')`, 1, key).Err(),
+ "Write commands are not allowed from read-only scripts")
+ require.ErrorContains(t,
+ rdb.Do(ctx, "eval_ro", `return redis.call('hgetex',
KEYS[1], 'FIELDS', 1, 'a')`, 1, key).Err(),
+ "Write commands are not allowed from read-only scripts")
+ })
+}
+
+func TestHashFieldExpirationHSetExHGetExLegacyPolicy(t *testing.T) {
+ srv := util.StartServer(t, util.KvrocksServerConfigs{
+ "hash-encoding-mode": "legacy",
+ "resp3-enabled": "yes",
+ })
+ defer srv.Close()
+ ctx := context.Background()
+ rdb := srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+
+ key := "hsetex-hgetex-legacy"
+ require.Equal(t, int64(2), rdb.HSet(ctx, key, "a", "1", "b", "2").Val())
+ require.NoError(t, rdb.PExpire(ctx, key, time.Minute).Err())
+ before := util.GetKMetadata(t, rdb, ctx, key)
+ valuesBefore := rdb.HGetAll(ctx, key).Val()
+ for _, args := range [][]interface{}{
+ {"hsetex", key, "FIELDS", 1, "a", "x"},
+ {"hsetex", key, "KEEPTTL", "FIELDS", 1, "a", "x"},
+ {"hsetex", key, "PXAT", 1, "FIELDS", 1, "a", "x"},
+ {"hsetex", key, "FXX", "FIELDS", 1, "missing", "x"},
+ {"hsetex", key, "FNX", "FIELDS", 1, "a", "x"},
+ {"hgetex", key, "FIELDS", 1, "a"},
+ {"hgetex", key, "PERSIST", "FIELDS", 1, "a"},
+ {"hgetex", key, "PXAT", 1, "FIELDS", 1, "a"},
+ {"hgetex", key, "PX", 60000, "FIELDS", 1, "missing"},
+ } {
+ require.ErrorContains(t, rdb.Do(ctx, args...).Err(),
+ "hash field expiration is not supported by legacy hash
encoding")
+ require.Equal(t, valuesBefore, rdb.HGetAll(ctx, key).Val())
+ require.Equal(t, before, util.GetKMetadata(t, rdb, ctx, key))
+ }
+
+ missing := "hsetex-hgetex-legacy-missing"
+ got, err := rdb.Do(ctx, "hgetex", missing, "PX", 1000, "FIELDS", 2,
"a", "b").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, nil, nil)
+ requireIntegerReply(t, rdb, ctx, 0, "hsetex", missing, "FXX", "FIELDS",
1, "a", "1")
+ for _, args := range [][]interface{}{
+ {"hsetex", missing, "FIELDS", 1, "a", "1"},
+ {"hsetex", missing, "FNX", "PXAT", 1, "FIELDS", 1, "a", "1"},
+ {"hsetex", missing, "KEEPTTL", "FIELDS", 1, "a", "1"},
+ } {
+ require.ErrorContains(t, rdb.Do(ctx, args...).Err(), "hash
field expiration")
+ require.Equal(t, int64(0), rdb.Exists(ctx, missing).Val())
+ }
+}
+
+func TestHashFieldExpirationHSetExHGetExMetadataSequence(t *testing.T) {
+ runWithFieldExpirationHash(t, func(t *testing.T, rdb *redis.Client, ctx
context.Context) {
+ key := "hsetex-hgetex-metadata-sequence"
+ t1 := time.Now().Add(10 * time.Minute).UnixMilli()
+ t2 := t1 + int64((10*time.Minute)/time.Millisecond)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "PXAT", t1,
"FIELDS", 3,
+ "a", "1", "b", "2", "c", "3")
+ meta := util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, meta, 3, 0)
+ require.Equal(t, t1, meta.Lower)
+ require.Equal(t, t1, meta.Upper)
+ originalHeader := meta
+
+ got, err := rdb.Do(ctx, "hgetex", key, "PERSIST", "FIELDS", 2,
"a", "missing").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "1", nil)
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, meta, 3, 1)
+ require.Equal(t, t1, meta.Lower)
+ require.Equal(t, t1, meta.Upper)
+ requireMetadataHeaderUnchanged(t, originalHeader, meta)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "KEEPTTL",
"FIELDS", 3,
+ "a", "10", "b", "20", "d", "40")
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, meta, 4, 2)
+ require.Equal(t, t1, meta.Lower)
+ require.Equal(t, t1, meta.Upper)
+
+ got, err = rdb.Do(ctx, "hgetex", key, "PXAT", t2, "FIELDS", 3,
"a", "c", "missing").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "10", "3", nil)
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, meta, 4, 1)
+ require.Equal(t, t1, meta.Lower)
+ require.Equal(t, t2, meta.Upper)
+
+ require.Equal(t, int64(1), rdb.HDel(ctx, key, "b").Val())
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, meta, 3, 1)
+ require.Equal(t, t1, meta.Lower)
+ require.Equal(t, t2, meta.Upper)
+
+ require.Equal(t, int64(10), rdb.HIncrBy(ctx, key, "c", 7).Val())
+ require.Equal(t, meta, util.GetKMetadata(t, rdb, ctx, key))
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "FIELDS", 1,
"a", "100")
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ requireHashMetadata(t, meta, 3, 2)
+ require.Equal(t, t1, meta.Lower)
+ require.Equal(t, t2, meta.Upper)
+
+ got, err = rdb.Do(ctx, "hgetex", key, "PERSIST", "FIELDS", 1,
"c").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "10")
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx, key), 3,
3)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "PXAT", 1,
"FIELDS", 2,
+ "a", "0", "c", "0")
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx, key), 1,
1)
+ require.Equal(t, map[string]string{"d": "40"}, rdb.HGetAll(ctx,
key).Val())
+ require.Equal(t, int64(1), rdb.Do(ctx, "hlen", key,
"REPAIR").Val())
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx, key), 1,
1)
+ })
+}
+
+func TestHashFieldExpirationHSetExHGetExConservativeBounds(t *testing.T) {
+ runWithFieldExpirationHash(t, func(t *testing.T, rdb *redis.Client, ctx
context.Context) {
+ key := "hsetex-hgetex-bounds"
+ now := time.Now().UnixMilli()
+ t10 := now + 10*60*1000
+ t20 := now + 20*60*1000
+ t25 := now + 25*60*1000
+ t30 := now + 30*60*1000
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "PXAT", t20,
"FIELDS", 2,
+ "a", "1", "b", "2")
+ meta := util.GetKMetadata(t, rdb, ctx, key)
+ require.Equal(t, t20, meta.Lower)
+ require.Equal(t, t20, meta.Upper)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "PXAT", t30,
"FIELDS", 1, "a", "3")
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ require.Equal(t, t20, meta.Lower)
+ require.Equal(t, t30, meta.Upper)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "PXAT", t25,
"FIELDS", 1, "b", "4")
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ require.Equal(t, t20, meta.Lower)
+ require.Equal(t, t30, meta.Upper)
+
+ got, err := rdb.Do(ctx, "hgetex", key, "PXAT", t10, "FIELDS",
1, "a").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "3")
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ require.Equal(t, t10, meta.Lower)
+ require.Equal(t, t30, meta.Upper)
+
+ got, err = rdb.Do(ctx, "hgetex", key, "PXAT", t20, "FIELDS", 1,
"a").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "3")
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ require.Equal(t, t10, meta.Lower)
+ require.Equal(t, t30, meta.Upper)
+
+ got, err = rdb.Do(ctx, "hgetex", key, "PERSIST", "FIELDS", 1,
"a").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "3")
+ meta = util.GetKMetadata(t, rdb, ctx, key)
+ require.Equal(t, int64(1), meta.Persist)
+ require.Equal(t, t10, meta.Lower)
+ require.Equal(t, t30, meta.Upper)
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "FIELDS", 1,
"b", "5")
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx, key), 2,
2)
+ })
+}
+
+func TestHashFieldExpirationHSetExHGetExKeyLifecycle(t *testing.T) {
+ runWithFieldExpirationHash(t, func(t *testing.T, rdb *redis.Client, ctx
context.Context) {
+ t.Run("preserve key ttl and version", func(t *testing.T) {
+ key := "hsetex-hgetex-key-ttl"
+ require.Equal(t, int64(2), rdb.HSet(ctx, key, "a", "1",
"b", "2").Val())
+ require.NoError(t, rdb.PExpire(ctx, key,
time.Minute).Err())
+ before := util.GetKMetadata(t, rdb, ctx, key)
+ expireAt := time.Now().Add(10 * time.Minute).UnixMilli()
+
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"PXAT", expireAt, "FIELDS", 1, "a", "3")
+ after := util.GetKMetadata(t, rdb, ctx, key)
+ requireMetadataHeaderUnchanged(t, before, after)
+
+ got, err := rdb.Do(ctx, "hgetex", key, "PERSIST",
"FIELDS", 1, "a").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "3")
+ after = util.GetKMetadata(t, rdb, ctx, key)
+ requireMetadataHeaderUnchanged(t, before, after)
+ })
+
+ t.Run("past deadline on missing key leaves no tombstone",
func(t *testing.T) {
+ key := "hsetex-missing-past"
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key,
"PXAT", 1, "FIELDS", 2,
+ "a", "1", "b", "2")
+ require.Equal(t, int64(0), rdb.Exists(ctx, key).Val())
+ require.Error(t, rdb.Do(ctx, "kmetadata", key).Err())
+ })
+
+ t.Run("missing key replies", func(t *testing.T) {
+ key := "hsetex-hgetex-missing"
+ requireIntegerReply(t, rdb, ctx, 0, "hsetex", key,
"FXX", "FIELDS", 1, "a", "1")
+ got, err := rdb.Do(ctx, "hgetex", key, "PERSIST",
"FIELDS", 2, "a", "b").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, nil, nil)
+ require.Equal(t, int64(0), rdb.Exists(ctx, key).Val())
+ })
+
+ t.Run("last physical field deletes metadata", func(t
*testing.T) {
+ key := "hgetex-last-field"
+ require.Equal(t, int64(1), rdb.HSet(ctx, key, "a",
"1").Val())
+ got, err := rdb.Do(ctx, "hgetex", key, "PXAT", 1,
"FIELDS", 1, "a").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "1")
+ require.Equal(t, int64(0), rdb.Exists(ctx, key).Val())
+ require.Error(t, rdb.Do(ctx, "kmetadata", key).Err())
+ })
+ })
+}
+
+func TestHashFieldExpirationHSetExHGetExSingleTimeSnapshot(t *testing.T) {
+ runWithFieldExpirationHash(t, func(t *testing.T, rdb *redis.Client, ctx
context.Context) {
+ key := "hsetex-hgetex-one-clock"
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "PX",
600000, "FIELDS", 3,
+ "a", "1", "b", "2", "c", "3")
+ expires := rdb.Do(ctx, "hpexpiretime", key, "FIELDS", 3, "a",
"b", "c").Val().([]interface{})
+ require.Equal(t, expires[0], expires[1])
+ require.Equal(t, expires[1], expires[2])
+
+ got, err := rdb.Do(ctx, "hgetex", key, "PX", 700000, "FIELDS",
3, "a", "b", "c").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "1", "2", "3")
+ expires = rdb.Do(ctx, "hpexpiretime", key, "FIELDS", 3, "a",
"b", "c").Val().([]interface{})
+ require.Equal(t, expires[0], expires[1])
+ require.Equal(t, expires[1], expires[2])
+ })
+}
+
+func TestHashFieldExpirationHSetExFNXConcurrency(t *testing.T) {
+ runWithFieldExpirationHash(t, func(t *testing.T, rdb *redis.Client, ctx
context.Context) {
+ const clients = 16
+ key := "hsetex-fnx-concurrent"
+ start := make(chan struct{})
+ results := make(chan int64, clients)
+ errors := make(chan error, clients)
+ var wg sync.WaitGroup
+ for i := 0; i < clients; i++ {
+ wg.Add(1)
+ go func(value int) {
+ defer wg.Done()
+ <-start
+ result, err := rdb.Do(ctx, "hsetex", key,
"FNX", "FIELDS", 1, "field", value).Int64()
+ if err != nil {
+ errors <- err
+ return
+ }
+ results <- result
+ }(i)
+ }
+ close(start)
+ wg.Wait()
+ close(results)
+ close(errors)
+
+ for err := range errors {
+ require.NoError(t, err)
+ }
+ succeeded := 0
+ for result := range results {
+ if result == 1 {
+ succeeded++
+ } else {
+ require.Equal(t, int64(0), result)
+ }
+ }
+ require.Equal(t, 1, succeeded)
+ requireHashMetadata(t, util.GetKMetadata(t, rdb, ctx, key), 1,
1)
+ })
+}
+
+func TestHashFieldExpirationHSetExHGetExRestart(t *testing.T) {
+ srv := util.StartServer(t, util.KvrocksServerConfigs{
+ "hash-encoding-mode": "field-expiration",
+ "resp3-enabled": "yes",
+ })
+ defer srv.Close()
+ ctx := context.Background()
+ rdb := srv.NewClient()
+
+ key := "hsetex-hgetex-restart"
+ expireAt := time.Now().Add(time.Hour).UnixMilli()
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "PXAT", expireAt,
"FIELDS", 3,
+ "a", "1", "b", "2", "c", "3")
+ got, err := rdb.Do(ctx, "hgetex", key, "PERSIST", "FIELDS", 1,
"a").Result()
+ require.NoError(t, err)
+ requireOptionalStringArray(t, got, "1")
+ requireIntegerReply(t, rdb, ctx, 1, "hsetex", key, "KEEPTTL", "FIELDS",
2,
+ "b", "20", "d", "40")
+
+ valuesBefore := rdb.HGetAll(ctx, key).Val()
+ expiresBefore := rdb.Do(ctx, "hpexpiretime", key, "FIELDS", 4, "a",
"b", "c", "d").Val()
+ metadataBefore := util.GetKMetadata(t, rdb, ctx, key)
+ require.NoError(t, rdb.Close())
+
+ srv.Restart()
+ rdb = srv.NewClient()
+ defer func() { require.NoError(t, rdb.Close()) }()
+ require.Equal(t, valuesBefore, rdb.HGetAll(ctx, key).Val())
+ require.Equal(t, expiresBefore, rdb.Do(ctx, "hpexpiretime", key,
"FIELDS", 4, "a", "b", "c", "d").Val())
+ require.Equal(t, metadataBefore, util.GetKMetadata(t, rdb, ctx, key))
+}
