Hfe serialization listpack (#13243)

Add RDB de/serialization for HFE

This PR adds two new RDB types: `RDB_TYPE_HASH_METADATA` and
`RDB_TYPE_HASH_LISTPACK_TTL` to save HFE data.
When the hash RAM encoding is dict, it will be saved in the former, and
when it is listpack it will be saved in the latter.
Both formats just add the TTL value for each field after the data that
was previously saved, i.e HASH_METADATA will save the number of entries
and, for each entry, key, value and TTL, whereas listpack is saved as a
blob.
On read, the usual dict <--> listpack conversion takes place if
required.
In addition, when reading a hash that was saved as a dict fields are
actively expired if expiry is due. Currently this slao holds for
listpack encoding, but it is supposed to be removed.

TODO:
Remove active expiry on load when loading from listpack format (unless
we'll decide to keep it)

src/cluster.c

@@ -237,7 +237,7 @@ void restoreCommand(client *c) {
 
  rioInitWithBuffer(&payload,c->argv[3]->ptr);
  if (((type = rdbLoadObjectType(&payload)) == -1) ||
- ((obj = rdbLoadObject(type,&payload,key->ptr,c->db->id,NULL)) == NULL))
+ ((obj = rdbLoadObject(type,&payload,key->ptr,c->db,0,NULL)) == NULL))
  {
  addReplyError(c,"Bad data format");
  return;

src/ebuckets.c

@@ -1403,7 +1403,8 @@ int ebRemove(ebuckets *eb, EbucketsType *type, eItem item) {
  * @param item - The eItem to be added to the ebucket.
  * @param expireTime - The expiration time of the item.
  *
- * @return 1 if the item was successfully added; Otherwise, return 0 on failure.
+ * @return 0 (C_OK) if the item was successfully added;
+ * Otherwise, return -1 (C_ERR) on failure.
  */
 int ebAdd(ebuckets *eb, EbucketsType *type, eItem item, uint64_t expireTime) {
  int res;

src/listpack.c

@@ -245,51 +245,61 @@ unsigned char* lpShrinkToFit(unsigned char *lp) {
 static inline void lpEncodeIntegerGetType(int64_t v, unsigned char *intenc, uint64_t *enclen) {
  if (v >= 0 && v <= 127) {
  /* Single byte 0-127 integer. */
- intenc[0] = v;
- *enclen = 1;
+ if (intenc != NULL) intenc[0] = v;
+ if (enclen != NULL) *enclen = 1;
  } else if (v >= -4096 && v <= 4095) {
  /* 13 bit integer. */
  if (v < 0) v = ((int64_t)1<<13)+v;
- intenc[0] = (v>>8)|LP_ENCODING_13BIT_INT;
- intenc[1] = v&0xff;
- *enclen = 2;
+ if (intenc != NULL) {
+ intenc[0] = (v>>8)|LP_ENCODING_13BIT_INT;
+ intenc[1] = v&0xff;
+ }
+ if (enclen != NULL) *enclen = 2;
  } else if (v >= -32768 && v <= 32767) {
  /* 16 bit integer. */
  if (v < 0) v = ((int64_t)1<<16)+v;
- intenc[0] = LP_ENCODING_16BIT_INT;
- intenc[1] = v&0xff;
- intenc[2] = v>>8;
- *enclen = 3;
+ if (intenc != NULL) {
+ intenc[0] = LP_ENCODING_16BIT_INT;
+ intenc[1] = v&0xff;
+ intenc[2] = v>>8;
+ }
+ if (enclen != NULL) *enclen = 3;
  } else if (v >= -8388608 && v <= 8388607) {
  /* 24 bit integer. */
  if (v < 0) v = ((int64_t)1<<24)+v;
- intenc[0] = LP_ENCODING_24BIT_INT;
- intenc[1] = v&0xff;
- intenc[2] = (v>>8)&0xff;
- intenc[3] = v>>16;
- *enclen = 4;
+ if (intenc != NULL) {
+ intenc[0] = LP_ENCODING_24BIT_INT;
+ intenc[1] = v&0xff;
+ intenc[2] = (v>>8)&0xff;
+ intenc[3] = v>>16;
+ }
+ if (enclen != NULL) *enclen = 4;
  } else if (v >= -2147483648 && v <= 2147483647) {
  /* 32 bit integer. */
  if (v < 0) v = ((int64_t)1<<32)+v;
- intenc[0] = LP_ENCODING_32BIT_INT;
- intenc[1] = v&0xff;
- intenc[2] = (v>>8)&0xff;
- intenc[3] = (v>>16)&0xff;
- intenc[4] = v>>24;
- *enclen = 5;
+ if (intenc != NULL) {
+ intenc[0] = LP_ENCODING_32BIT_INT;
+ intenc[1] = v&0xff;
+ intenc[2] = (v>>8)&0xff;
+ intenc[3] = (v>>16)&0xff;
+ intenc[4] = v>>24;
+ }
+ if (enclen != NULL) *enclen = 5;
  } else {
  /* 64 bit integer. */
  uint64_t uv = v;
- intenc[0] = LP_ENCODING_64BIT_INT;
- intenc[1] = uv&0xff;
- intenc[2] = (uv>>8)&0xff;
- intenc[3] = (uv>>16)&0xff;
- intenc[4] = (uv>>24)&0xff;
- intenc[5] = (uv>>32)&0xff;
- intenc[6] = (uv>>40)&0xff;
- intenc[7] = (uv>>48)&0xff;
- intenc[8] = uv>>56;
- *enclen = 9;
+ if (intenc != NULL) {
+ intenc[0] = LP_ENCODING_64BIT_INT;
+ intenc[1] = uv&0xff;
+ intenc[2] = (uv>>8)&0xff;
+ intenc[3] = (uv>>16)&0xff;
+ intenc[4] = (uv>>24)&0xff;
+ intenc[5] = (uv>>32)&0xff;
+ intenc[6] = (uv>>40)&0xff;
+ intenc[7] = (uv>>48)&0xff;
+ intenc[8] = uv>>56;
+ }
+ if (enclen != NULL) *enclen = 9;
  }
 }
 
@@ -1199,13 +1209,17 @@ size_t lpBytes(unsigned char *lp) {
  return lpGetTotalBytes(lp);
 }
 
-/* Returns the size of a listpack consisting of an integer repeated 'rep' times. */
-size_t lpEstimateBytesRepeatedInteger(long long lval, unsigned long rep) {
+/* Returns the size 'lval' will require when encoded, in bytes */
+size_t lpEntrySizeInteger(long long lval) {
  uint64_t enclen;
- unsigned char intenc[LP_MAX_INT_ENCODING_LEN];
- lpEncodeIntegerGetType(lval, intenc, &enclen);
+ lpEncodeIntegerGetType(lval, NULL, &enclen);
  unsigned long backlen = lpEncodeBacklen(NULL, enclen);
- return LP_HDR_SIZE + (enclen + backlen) * rep + 1;
+ return enclen + backlen;
+}
+
+/* Returns the size of a listpack consisting of an integer repeated 'rep' times. */
+size_t lpEstimateBytesRepeatedInteger(long long lval, unsigned long rep) {
+ return LP_HDR_SIZE + lpEntrySizeInteger(lval) * rep + 1;
 }
 
 /* Seek the specified element and returns the pointer to the seeked element.

src/listpack.h

@@ -61,6 +61,7 @@ unsigned char *lpLast(unsigned char *lp);
 unsigned char *lpNext(unsigned char *lp, unsigned char *p);
 unsigned char *lpPrev(unsigned char *lp, unsigned char *p);
 size_t lpBytes(unsigned char *lp);
+size_t lpEntrySizeInteger(long long lval);
 size_t lpEstimateBytesRepeatedInteger(long long lval, unsigned long rep);
 unsigned char *lpSeek(unsigned char *lp, long index);
 typedef int (*listpackValidateEntryCB)(unsigned char *p, unsigned int head_count, void *userdata);