Use no_value dict type for stream_idmp_keys to explicitly mark it as a key-only set (#14987)

Fixes #14985

### Problem

dict stream_idmp_keys was using objectKeyPointerValueDictType, in this
dict type dicts are expected to have RObj as keys and Pointers as
values, but stream_idmp_keys was not using the value field at all.

### Solution
This PR fixes the above issue by implementing new dict type
(objectKeyNoValueDictType) for stream_idmp_keys

---------

Co-authored-by: debing.sun <debing.sun@redis.com>

src/cluster_asm.c

@@ -3033,7 +3033,7 @@ void asmTriggerBackgroundTrim(asmTrimCtx *trim_ctx, int migration_cleanup) {
                                      CLUSTER_SLOT_MASK_BITS,
                                      KVSTORE_ALLOCATE_DICTS_ON_DEMAND);
     estore *subexpires = estoreCreate(&subexpiresBucketsType, CLUSTER_SLOT_MASK_BITS);
-    dict *stream_idmp_keys = dictCreate(&objectKeyPointerValueDictType);
+    dict *stream_idmp_keys = dictCreate(&objectKeyNoValueDictType);
 
     size_t total_keys = 0;
 

src/db.c

@@ -1083,7 +1083,7 @@ redisDb *initTempDb(void) {
         tempDb[i].expires = kvstoreCreate(&kvstoreBaseType, &dbExpiresDictType,
                                           slot_count_bits, flags);
         tempDb[i].subexpires = estoreCreate(&subexpiresBucketsType, slot_count_bits);
-        tempDb[i].stream_idmp_keys = dictCreate(&objectKeyPointerValueDictType);
+        tempDb[i].stream_idmp_keys = dictCreate(&objectKeyNoValueDictType);
     }
 
     return tempDb;
@@ -1117,7 +1117,7 @@ void streamMoveIdmpKeys(dict *src, dict *dst, int slot) {
     while ((de = dictNext(di)) != NULL) {
         robj *key = dictGetKey(de);
         if (calculateKeySlot(key->ptr) == slot) {
-            if (dictAdd(dst, key, dictGetVal(de)) == DICT_OK) {
+            if (dictAddRaw(dst, key, NULL)) {
                 incrRefCount(key);
             }
             dictDelete(src, key);

src/lazyfree.c

@@ -332,7 +332,7 @@ void emptyDbAsync(redisDb *db) {
     db->keys = kvstoreCreate(&kvstoreExType, &dbDictType, slot_count_bits, flags);
     db->expires = kvstoreCreate(&kvstoreBaseType, &dbExpiresDictType, slot_count_bits, flags);
     db->subexpires = estoreCreate(&subexpiresBucketsType, slot_count_bits);
-    db->stream_idmp_keys = dictCreate(&objectKeyPointerValueDictType);
+    db->stream_idmp_keys = dictCreate(&objectKeyNoValueDictType);
     protectClientReplyObjects(); /* Protect client reply objects before async free. */
     emptyDbDataAsync(oldkeys, oldexpires, oldsubexpires, old_stream_idmp_keys, NULL);
 }

src/server.c

@@ -581,6 +581,19 @@ dictType objectKeyPointerValueDictType = {
     NULL                       /* allow to expand */
 };
 
+/* Dict type with robj pointer keys and no values. */
+dictType objectKeyNoValueDictType = {
+    dictEncObjHash,            /* hash function */
+    NULL,                      /* key dup */
+    NULL,                      /* val dup */
+    dictEncObjKeyCompare,      /* key compare */
+    dictObjectDestructor,      /* key destructor */
+    NULL,                      /* val destructor */
+    NULL,                      /* allow to expand */
+    .no_value = 1,             /* no values in this dict */
+    .keys_are_odd = 0,         /* robj pointers are not odd */
+};
+
 /* Like objectKeyPointerValueDictType(), but values can be destroyed, if
  * not NULL, calling zfree(). */
 dictType objectKeyHeapPointerValueDictType = {
@@ -2996,7 +3009,7 @@ void initServer(void) {
         server.db[j].blocking_keys = dictCreate(&keylistDictType);
         server.db[j].blocking_keys_unblock_on_nokey = dictCreate(&objectKeyPointerValueDictType);
         server.db[j].stream_claim_pending_keys = dictCreate(&objectKeyPointerValueDictType);
-        server.db[j].stream_idmp_keys = dictCreate(&objectKeyPointerValueDictType);
+        server.db[j].stream_idmp_keys = dictCreate(&objectKeyNoValueDictType);
         server.db[j].ready_keys = dictCreate(&objectKeyPointerValueDictType);
         server.db[j].watched_keys = dictCreate(&keylistDictType);
         server.db[j].id = j;

src/server.h

@@ -3004,6 +3004,7 @@ typedef struct {
 extern struct redisServer server;
 extern struct sharedObjectsStruct shared;
 extern dictType objectKeyPointerValueDictType;
+extern dictType objectKeyNoValueDictType;
 extern dictType objectKeyHeapPointerValueDictType;
 extern dictType setDictType;
 extern dictType BenchmarkDictType;

src/t_stream.c

@@ -5983,7 +5983,7 @@ void streamKeyLoaded(redisDb *db, robj *key, robj *val) {
     }
 }
 
-/* To be used when a steam key was removed from ram, un-redigster from stream_idmp_keys if needed */
+/* To be used when a stream key was removed from ram, un-register from stream_idmp_keys if needed */
 void streamKeyRemoved(redisDb *db, robj *key, robj *val) {
     UNUSED(val);
     dictDelete(db->stream_idmp_keys, key);