slipenkomk/valkey
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support RDB compatability with Redis 7.2.4 RDB format (#665)

Browse Source 
This PR makes our current RDB format compatible with the Redis 7.2.4 RDB
format. there are 2 changes introduced in this PR:
1. Move back the RDB version to 11
2. Make slot info section persist as AUX data instead of dedicated
section.

We have introduced slot-info as part of the work to replace cluster
metadata with slot specific dictionaries. This caused us to bump the RDB
version and thus we prevent downgrade (which is conceptualy O.K but
better be prevented). We do not require the slot-info section to exist,
so making it an AUX section will help suppport version downgrade from
Valkey 8.

fixes: [#645](https://github.com/valkey-io/valkey/issues/645)

NOTE: tested manually by:
1. connecting Redis 7.2.4 replica to a Valkey 8(RC) 
2. upgrade/downgrade Redis 7.2.4 cluster and Valkey 8(RC) cluster

---------

Signed-off-by: ranshid <ranshid@amazon.com>
Co-authored-by: Viktor Söderqvist <viktor.soderqvist@est.tech>
This commit is contained in:
ranshid 2024-06-18 22:04:06 +03:00 committed by GitHub
parent a2cc2fe26d
commit be2c321682
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 26 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
src/rdb.c
View File

@ -1349,15 +1349,14 @@ ssize_t rdbSaveDb(rio *rdb, int dbid, int rdbflags, long *key_counter) {
int curr_slot = kvstoreIteratorGetCurrentDictIndex(kvs_it);
/* Save slot info. */
if (server.cluster_enabled && curr_slot != last_slot) {
if ((res = rdbSaveType(rdb, RDB_OPCODE_SLOT_INFO)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, curr_slot)) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, kvstoreDictSize(db->keys, curr_slot))) < 0) goto werr;
written += res;
if ((res = rdbSaveLen(rdb, kvstoreDictSize(db->expires, curr_slot))) < 0) goto werr;
written += res;
sds slot_info = sdscatprintf(sdsempty(), "%i,%lu,%lu", curr_slot, kvstoreDictSize(db->keys, curr_slot),
kvstoreDictSize(db->expires, curr_slot));
if ((res = rdbSaveAuxFieldStrStr(rdb, "slot-info", slot_info)) < 0) {
sdsfree(slot_info);
goto werr;
}
last_slot = curr_slot;
sdsfree(slot_info);
}
sds keystr = dictGetKey(de);
robj key, *o = dictGetVal(de);
@ -3078,20 +3077,6 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
if ((expires_size = rdbLoadLen(rdb, NULL)) == RDB_LENERR) goto eoferr;
should_expand_db = 1;
continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_SLOT_INFO) {
uint64_t slot_id, slot_size, expires_slot_size;
if ((slot_id = rdbLoadLen(rdb, NULL)) == RDB_LENERR) goto eoferr;
if ((slot_size = rdbLoadLen(rdb, NULL)) == RDB_LENERR) goto eoferr;
if ((expires_slot_size = rdbLoadLen(rdb, NULL)) == RDB_LENERR) goto eoferr;
if (!server.cluster_enabled) {
continue; /* Ignore gracefully. */
}
/* In cluster mode we resize individual slot specific dictionaries based on the number of keys that slot
* holds. */
kvstoreDictExpand(db->keys, slot_id, slot_size);
kvstoreDictExpand(db->expires, slot_id, expires_slot_size);
should_expand_db = 0;
continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_AUX) {
/* AUX: generic string-string fields. Use to add state to RDB
* which is backward compatible. Implementations of RDB loading
@ -3141,6 +3126,24 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
if (isbase) serverLog(LL_NOTICE, "RDB is base AOF");
} else if (!strcasecmp(auxkey->ptr, "redis-bits")) {
/* Just ignored. */
} else if (!strcasecmp(auxkey->ptr, "slot-info")) {
int slot_id;
unsigned long slot_size, expires_slot_size;
/* Try to parse the slot information. In case the number of parsed arguments is smaller than expected
* we'll fail the RDB load. */
if (sscanf(auxval->ptr, "%i,%lu,%lu", &slot_id, &slot_size, &expires_slot_size) < 3) {
decrRefCount(auxkey);
decrRefCount(auxval);
goto eoferr;
}
if (server.cluster_enabled) {
/* In cluster mode we resize individual slot specific dictionaries based on the number of keys that
* slot holds. */
kvstoreDictExpand(db->keys, slot_id, slot_size);
kvstoreDictExpand(db->expires, slot_id, expires_slot_size);
should_expand_db = 0;
}
} else {
/* Check if this is a dynamic aux field */
int handled = 0;

3
src/rdb.h
View File

@ -38,7 +38,7 @@
/* The current RDB version. When the format changes in a way that is no longer
* backward compatible this number gets incremented. */
#define RDB_VERSION 12
#define RDB_VERSION 11
/* Defines related to the dump file format. To store 32 bits lengths for short
* keys requires a lot of space, so we check the most significant 2 bits of
@ -101,7 +101,6 @@
#define rdbIsObjectType(t) (((t) >= 0 && (t) <= 7) || ((t) >= 9 && (t) <= 21))
/* 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). */
#define RDB_OPCODE_FUNCTION2 245 /* function library data */
#define RDB_OPCODE_FUNCTION_PRE_GA 246 /* old function library data for 7.0 rc1 and rc2 */
#define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */

6
src/valkey-check-rdb.c
View File

@ -256,12 +256,6 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
if ((db_size = rdbLoadLen(&rdb, NULL)) == RDB_LENERR) goto eoferr;
if ((expires_size = rdbLoadLen(&rdb, NULL)) == RDB_LENERR) goto eoferr;
continue; /* Read type again. */
} else if (type == RDB_OPCODE_SLOT_INFO) {
uint64_t slot_id, slot_size, expires_slot_size;
if ((slot_id = rdbLoadLen(&rdb, NULL)) == RDB_LENERR) goto eoferr;
if ((slot_size = rdbLoadLen(&rdb, NULL)) == RDB_LENERR) goto eoferr;
if ((expires_slot_size = rdbLoadLen(&rdb, NULL)) == RDB_LENERR) goto eoferr;
continue; /* Read type again. */
} else if (type == RDB_OPCODE_AUX) {
/* AUX: generic string-string fields. Use to add state to RDB
* which is backward compatible. Implementations of RDB loading