Unverified Commit dcf02298 authored by Moti Cohen's avatar Moti Cohen Committed by GitHub
Browse files

HFE - RDB serialize also hash min expiration (For ROF flow) (#13391)

* Following this feature, Redis (ROF) may implement flow that allows hashes to be
  dumped directly from RDB to FLUSH without parsing. In this scenario, it is still
  essential to determine when to update hashes due to expired fields. By writing
  and reading the next minimum hash-field expiration before serializing objects
  to and from RDB, we can effectively track and expire hash fields without the need
  to parse the hash during loading.

    Before:
    #define RDB_TYPE_HASH_METADATA 22
    #define RDB_TYPE_HASH_LISTPACK_EX 23
    
    After:
    /* Hash with HFEs. Doesn't attach min TTL at start */
    #define RDB_TYPE_HASH_METADATA_PRE_GA 22      
    /* Hash LP with HFEs. Doesn't attach min TTL at start */
    #define RDB_TYPE_HASH_LISTPACK_EX_PRE_GA 23   
    /* Hash with HFEs. Attach min TTL at start */
    #define RDB_TYPE_HASH_METADATA 24             
    /* Hash LP with HFEs. Attach min TTL at start */
    #define RDB_TYPE_HASH_LISTPACK_EX 25          


* Manually test loading RDB file before the change and verify hash and its HFEs 
  is as expected.
* Added `subexpires` counter to `redis-check-rdb`
parent 58edea73
......@@ -947,12 +947,22 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
if ((o->encoding == OBJ_ENCODING_LISTPACK) ||
(o->encoding == OBJ_ENCODING_LISTPACK_EX))
{
/* Save min/next HFE expiration time if needed */
if (o->encoding == OBJ_ENCODING_LISTPACK_EX) {
uint64_t minExpire = hashTypeGetMinExpire(o, 0);
/* if invalid time then save 0 */
if (minExpire == EB_EXPIRE_TIME_INVALID)
minExpire = 0;
if (rdbSaveMillisecondTime(rdb, minExpire) == -1)
return -1;
}
unsigned char *lp_ptr = hashTypeListpackGetLp(o);
size_t l = lpBytes(lp_ptr);
if ((n = rdbSaveRawString(rdb,lp_ptr,l)) == -1) return -1;
nwritten += n;
} else if (o->encoding == OBJ_ENCODING_HT) {
int hashWithMeta = 0; /* RDB_TYPE_HASH_METADATA */
dictIterator *di = dictGetIterator(o->ptr);
dictEntry *de;
/* Determine the hash layout to use based on the presence of at least
......@@ -960,7 +970,17 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
* RDB_TYPE_HASH_METADATA layout, including tuples of [ttl][field][value].
* Otherwise, use the standard RDB_TYPE_HASH layout containing only
* the tuples [field][value]. */
int with_ttl = (hashTypeGetMinExpire(o, 0) != EB_EXPIRE_TIME_INVALID);
uint64_t minExpire = hashTypeGetMinExpire(o, 0);
/* if RDB_TYPE_HASH_METADATA (Can have TTLs on fields) */
if (minExpire != EB_EXPIRE_TIME_INVALID) {
hashWithMeta = 1;
/* Save next field expire time of hash */
if (rdbSaveMillisecondTime(rdb, minExpire) == -1) {
dictReleaseIterator(di);
return -1;
}
}
/* save number of fields in hash */
if ((n = rdbSaveLen(rdb,dictSize((dict*)o->ptr))) == -1) {
......@@ -975,7 +995,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key, int dbid) {
sds value = dictGetVal(de);
/* save the TTL */
if (with_ttl) {
if (hashWithMeta) {
uint64_t ttl = hfieldGetExpireTime(field);
/* 0 is used to indicate no TTL is set for this field */
if (ttl == EB_EXPIRE_TIME_INVALID) ttl = 0;
......@@ -2238,12 +2258,24 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
/* All pairs should be read by now */
serverAssert(len == 0);
} else if (rdbtype == RDB_TYPE_HASH_METADATA) {
} else if (rdbtype == RDB_TYPE_HASH_METADATA || rdbtype == RDB_TYPE_HASH_METADATA_PRE_GA) {
sds value;
hfield field;
uint64_t expireAt;
dict *dupSearchDict = NULL;
/* If hash with TTLs, load next/min expiration time */
if (rdbtype == RDB_TYPE_HASH_METADATA) {
uint64_t minExpire = rdbLoadMillisecondTime(rdb, RDB_VERSION);
/* This value was serialized for future use-case of streaming the object
* directly to FLASH (while keeping in mem its next expiration time) */
UNUSED(minExpire);
if (rioGetReadError(rdb)) {
rdbReportCorruptRDB("Hash failed loading minExpire");
return NULL;
}
}
len = rdbLoadLen(rdb, NULL);
if (len == RDB_LENERR) return NULL;
if (len == 0) goto emptykey;
......@@ -2456,9 +2488,23 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
rdbtype == RDB_TYPE_ZSET_LISTPACK ||
rdbtype == RDB_TYPE_HASH_ZIPLIST ||
rdbtype == RDB_TYPE_HASH_LISTPACK ||
rdbtype == RDB_TYPE_HASH_LISTPACK_EX_PRE_GA ||
rdbtype == RDB_TYPE_HASH_LISTPACK_EX)
{
size_t encoded_len;
/* If Hash TTLs, Load next/min expiration time before the `encoded` */
if (rdbtype == RDB_TYPE_HASH_LISTPACK_EX) {
uint64_t minExpire = rdbLoadMillisecondTime(rdb, RDB_VERSION);
/* This value was serialized for future use-case of streaming the object
* directly to FLASH (while keeping in mem its next expiration time) */
UNUSED(minExpire);
if (rioGetReadError(rdb)) {
rdbReportCorruptRDB( "Hash listpackex integrity check failed.");
return NULL;
}
}
unsigned char *encoded =
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,&encoded_len);
if (encoded == NULL) return NULL;
......@@ -2665,11 +2711,13 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error)
break;
}
case RDB_TYPE_HASH_LISTPACK:
case RDB_TYPE_HASH_LISTPACK_EX_PRE_GA:
case RDB_TYPE_HASH_LISTPACK_EX:
/* listpack-encoded hash with TTL requires its own struct
* pointed to by o->ptr */
o->type = OBJ_HASH;
if (rdbtype == RDB_TYPE_HASH_LISTPACK_EX) {
if ( (rdbtype == RDB_TYPE_HASH_LISTPACK_EX) ||
(rdbtype == RDB_TYPE_HASH_LISTPACK_EX_PRE_GA) ) {
listpackEx *lpt = listpackExCreate();
lpt->lp = encoded;
lpt->key = key;
......
......@@ -73,12 +73,14 @@
#define RDB_TYPE_STREAM_LISTPACKS_2 19
#define RDB_TYPE_SET_LISTPACK 20
#define RDB_TYPE_STREAM_LISTPACKS_3 21
#define RDB_TYPE_HASH_METADATA 22
#define RDB_TYPE_HASH_LISTPACK_EX 23
#define RDB_TYPE_HASH_METADATA_PRE_GA 22 /* Hash with HFEs. Doesn't attach min TTL at start (7.4 RC) */
#define RDB_TYPE_HASH_LISTPACK_EX_PRE_GA 23 /* Hash LP with HFEs. Doesn't attach min TTL at start (7.4 RC) */
#define RDB_TYPE_HASH_METADATA 24 /* Hash with HFEs. Attach min TTL at start */
#define RDB_TYPE_HASH_LISTPACK_EX 25 /* Hash LP with HFEs. Attach min TTL at start */
/* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType(), and rdb_type_string[] */
/* Test if a type is an object type. */
#define rdbIsObjectType(t) (((t) >= 0 && (t) <= 7) || ((t) >= 9 && (t) <= 23))
#define rdbIsObjectType(t) (((t) >= 0 && (t) <= 7) || ((t) >= 9 && (t) <= 25))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_SLOT_INFO 244 /* Individual slot info, such as slot id and size (cluster mode only). */
......
......@@ -26,6 +26,7 @@ struct {
unsigned long keys; /* Number of keys processed. */
unsigned long expires; /* Number of keys with an expire. */
unsigned long already_expired; /* Number of keys already expired. */
unsigned long subexpires; /* Number of keys with subexpires */
int doing; /* The state while reading the RDB. */
int error_set; /* True if error is populated. */
char error[1024];
......@@ -80,6 +81,8 @@ char *rdb_type_string[] = {
"stream-v2",
"set-listpack",
"stream-v3",
"hash-hashtable-md-pre-release",
"hash-listpack-md-pre-release",
"hash-hashtable-md",
"hash-listpack-md",
};
......@@ -89,6 +92,7 @@ void rdbShowGenericInfo(void) {
printf("[info] %lu keys read\n", rdbstate.keys);
printf("[info] %lu expires\n", rdbstate.expires);
printf("[info] %lu already expired\n", rdbstate.already_expired);
printf("[info] %lu subexpires\n", rdbstate.subexpires);
}
/* Called on RDB errors. Provides details about the RDB and the offset
......@@ -339,6 +343,10 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
if (expiretime != -1 && expiretime < now)
rdbstate.already_expired++;
if (expiretime != -1) rdbstate.expires++;
/* If hash with HFEs then with expiration on fields then need to count it */
if ((val->type == OBJ_HASH) && (hashTypeGetMinExpire(val, 1) != EB_EXPIRE_TIME_INVALID))
rdbstate.subexpires++;
rdbstate.key = NULL;
decrRefCount(key);
decrRefCount(val);
......
......@@ -1917,8 +1917,9 @@ static int hashTypeExpireIfNeeded(redisDb *db, robj *o) {
/* Return the next/minimum expiry time of the hash-field.
* accurate=1 - Return the exact time by looking into the object DS.
* accurate=0 - Return the minimum expiration time maintained in expireMeta which
* might not be accurate due to optimization reasons.
* accurate=0 - Return the minimum expiration time maintained in expireMeta
* (Verify it is not trash before using it) which might not be
* accurate due to optimization reasons.
*
* If not found, return EB_EXPIRE_TIME_INVALID
*/
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment