1 files changed, 3793 insertions, 0 deletions

diff --git a/examples/redis-unstable/src/db.c b/examples/redis-unstable/src/db.c
new file mode 100644
index 0000000..2a64147
--- /dev/null
+++ b/examples/redis-unstable/src/db.c
@@ -0,0 +1,3793 @@
+/*
+ * Copyright (c) 2009-Present, Redis Ltd.
+ * All rights reserved.
+ *
+ * Copyright (c) 2024-present, Valkey contributors.
+ * All rights reserved.
+ *
+ * Licensed under your choice of (a) the Redis Source Available License 2.0
+ * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the
+ * GNU Affero General Public License v3 (AGPLv3).
+ *
+ * Portions of this file are available under BSD3 terms; see REDISCONTRIBUTIONS for more information.
+ */
+
+#include "server.h"
+#include "cluster.h"
+#include "atomicvar.h"
+#include "latency.h"
+#include "script.h"
+#include "functions.h"
+#include "cluster_asm.h"
+#include "redisassert.h"
+
+#include <signal.h>
+#include <ctype.h>
+#include "bio.h"
+#include "keymeta.h"
+
+/*-----------------------------------------------------------------------------
+ * C-level DB API
+ *----------------------------------------------------------------------------*/
+
+static_assert(MAX_KEYSIZES_TYPES == OBJ_TYPE_BASIC_MAX, "Must be equal");
+
+/* Flags for expireIfNeeded */
+#define EXPIRE_FORCE_DELETE_EXPIRED 1
+#define EXPIRE_AVOID_DELETE_EXPIRED 2
+#define EXPIRE_ALLOW_ACCESS_EXPIRED 4
+#define EXPIRE_ALLOW_ACCESS_TRIMMED 8
+
+/* Return values for expireIfNeeded */
+typedef enum {
+    KEY_VALID = 0, /* Could be volatile and not yet expired, non-volatile, or even non-existing key. */
+    KEY_EXPIRED, /* Logically expired but not yet deleted. */
+    KEY_DELETED, /* The key was deleted now. */
+    KEY_TRIMMED  /* Logically trimmed but not yet deleted. */
+} keyStatus;
+
+static keyStatus expireIfNeeded(redisDb *db, robj *key, kvobj *kv, int flags);
+
+/* Update LFU when an object is accessed.
+ * Firstly, decrement the counter if the decrement time is reached.
+ * Then logarithmically increment the counter, and update the access time. */
+void updateLFU(robj *val) {
+    unsigned long counter = LFUDecrAndReturn(val);
+    counter = LFULogIncr(counter);
+    val->lru = (LFUGetTimeInMinutes()<<8) | counter;
+}
+
+/* Update LRM when an object is modified. */
+void updateLRM(robj *o) {
+    if (o->refcount == OBJ_SHARED_REFCOUNT)
+        return;
+    if (server.maxmemory_policy & MAXMEMORY_FLAG_LRM) {
+        o->lru = LRU_CLOCK();
+    }
+}
+
+/* 
+ * Update histogram of keys-sizes
+ * 
+ * It is used to track the distribution of key sizes in the dataset. It is updated 
+ * every time key's length is modified. Available to user via INFO command. 
+ * 
+ * The histogram is a base-2 logarithmic histogram, with 64 bins. The i'th bin 
+ * represents the number of keys with a size in the range 2^i and 2^(i+1) 
+ * exclusive. oldLen/newLen must be smaller than 2^48, and if their value 
+ * equals -1, it means that the key is being created/deleted, respectively. Each
+ * data type has its own histogram and it is per database (In addition, there is 
+ * histogram per slot for future cluster use).
+ *
+ * Example mapping of key lengths to bins:
+ *               [1,2)->1 [2,4)->2 [4,8)->3 [8,16)->4 ...
+ *
+ * Since strings can be zero length, the histogram also tracks:
+ *               [0,1)->0
+ */
+void updateKeysizesHist(redisDb *db, int didx, uint32_t type, int64_t oldLen, int64_t newLen) {
+    if(unlikely(type >= OBJ_TYPE_BASIC_MAX))
+        return;
+
+    kvstoreDictMetadata *dictMeta = kvstoreGetDictMeta(db->keys, didx, 0);
+    kvstoreMetadata *kvstoreMeta = kvstoreGetMetadata(db->keys);
+
+    if (oldLen > 0) {
+        int old_bin = log2ceil(oldLen) + 1;
+        debugServerAssert(old_bin < MAX_KEYSIZES_BINS);
+        /* If following a key deletion it is last one in slot's dict, then
+         * slot's dict might get released as well. Verify if metadata is not NULL. */
+        if(dictMeta) {
+            dictMeta->keysizes_hist[type][old_bin]--;
+            debugServerAssert(dictMeta->keysizes_hist[type][old_bin] >= 0);
+        }
+        kvstoreMeta->keysizes_hist[type][old_bin]--;
+        debugServerAssert(kvstoreMeta->keysizes_hist[type][old_bin] >= 0);
+    } else {
+        /* here, oldLen can be either 0 or -1 */
+        if (oldLen == 0) {
+            /* Only strings can be empty. Yet, a command flow might temporarily
+             * dbAdd() empty collection, and only after add elements. */
+
+            if (dictMeta) {
+                dictMeta->keysizes_hist[type][0]--;
+                debugServerAssert(dictMeta->keysizes_hist[type][0] >= 0);
+            }
+            kvstoreMeta->keysizes_hist[type][0]--;
+            debugServerAssert(kvstoreMeta->keysizes_hist[type][0] >= 0);
+        }
+    }
+    
+    if (newLen > 0) {
+        int new_bin = log2ceil(newLen) + 1;
+        debugServerAssert(new_bin < MAX_KEYSIZES_BINS);
+        /* If following a key deletion it is last one in slot's dict, then
+         * slot's dict might get released as well. Verify if metadata is not NULL. */
+        if(dictMeta) dictMeta->keysizes_hist[type][new_bin]++;
+        kvstoreMeta->keysizes_hist[type][new_bin]++;
+    } else {
+        /* here, newLen can be either 0 or -1 */
+        if (newLen == 0) {
+            /* Only strings can be empty. Yet, a command flow might temporarily
+             * dbAdd() empty collection, and only after add elements. */
+
+            if (dictMeta) dictMeta->keysizes_hist[type][0]++;
+            kvstoreMeta->keysizes_hist[type][0]++;
+        }
+    }
+}
+
+void updateSlotAllocSize(redisDb *db, int didx, size_t oldsize, size_t newsize) {
+    debugServerAssert(server.memory_tracking_per_slot);
+    kvstoreDictMetadata *dictMeta = kvstoreGetDictMeta(db->keys, didx, 0);
+    if (!dictMeta) return;
+    debugServerAssert(oldsize <= dictMeta->alloc_size);
+    dictMeta->alloc_size -= oldsize;
+    dictMeta->alloc_size += newsize;
+}
+
+/* Assert keysizes histogram (For debugging only)
+ *
+ * Triggered by DEBUG KEYSIZES-HIST-ASSERT 1 and tested after each command.
+ */
+void dbgAssertKeysizesHist(redisDb *db) {
+    /* Scan DB and build expected histogram by scanning all keys */
+    int64_t scanHist[MAX_KEYSIZES_TYPES][MAX_KEYSIZES_BINS] = {{0}};
+    dictEntry *de;
+    kvstoreIterator kvs_it;
+    kvstoreIteratorInit(&kvs_it, db->keys);
+    while ((de = kvstoreIteratorNext(&kvs_it)) != NULL) {
+        kvobj *kv = dictGetKV(de);
+        if (kv->type < OBJ_TYPE_BASIC_MAX) {
+            int64_t len = getObjectLength(kv);
+            scanHist[kv->type][(len == 0) ? 0 : log2ceil(len) + 1]++;
+        }
+    }
+    kvstoreIteratorReset(&kvs_it);
+    for (int type = 0; type < OBJ_TYPE_BASIC_MAX; type++) {
+        kvstoreMetadata *meta = kvstoreGetMetadata(db->keys);
+        volatile int64_t *keysizesHist = meta->keysizes_hist[type];
+        for (int i = 0; i < MAX_KEYSIZES_BINS; i++) {
+            if (scanHist[type][i] == keysizesHist[i])
+                continue;
+
+            /* print scanStr vs. expected histograms for debugging */
+            char scanStr[500], keysizesStr[500];
+            int l1 = 0, l2 = 0;
+            for (int j = 0; (j < MAX_KEYSIZES_BINS) && (l1 < 500) && (l2 < 500); j++) {
+                if (scanHist[type][j])
+                    l1 += snprintf(scanStr + l1, sizeof(scanStr) - l1,
+                                        "[%d]=%"PRId64" ", j, scanHist[type][j]);
+                if (keysizesHist[j])
+                    l2 += snprintf(keysizesStr + l2, sizeof(keysizesStr) - l2,
+                                            "[%d]=%"PRId64" ", j, keysizesHist[j]);
+            }
+            serverPanic("dbgAssertKeysizesHist: type=%d\nscanStr=%s\nkeysizes=%s\n",
+                        type, scanStr, keysizesStr);
+        }
+    }
+}
+
+/* Assert per-slot alloc_size (For debugging only)
+ *
+ * Triggered by DEBUG ALLOCSIZE-SLOTS-ASSERT 1 and tested after each command.
+ */
+void dbgAssertAllocSizePerSlot(redisDb *db) {
+    if (!server.memory_tracking_per_slot) return;
+    size_t slot_sizes[CLUSTER_SLOTS] = {0};
+    dictEntry *de;
+    kvstoreIterator kvs_it;
+    kvstoreIteratorInit(&kvs_it, db->keys);
+    while ((de = kvstoreIteratorNext(&kvs_it)) != NULL) {
+        int slot = kvstoreIteratorGetCurrentDictIndex(&kvs_it);
+        kvobj *kv = dictGetKV(de);
+        slot_sizes[slot] += kvobjAllocSize(kv);
+    }
+    kvstoreIteratorReset(&kvs_it);
+
+    int num_slots = kvstoreNumDicts(db->keys);
+    for (int slot = 0; slot < num_slots; slot++) {
+        kvstoreDictMetadata *dictMeta = kvstoreGetDictMeta(db->keys, slot, 0);
+        size_t want = slot_sizes[slot];
+        size_t have = dictMeta ? dictMeta->alloc_size : 0;
+        if (have == want) continue;
+        serverPanic("dbgAssertAllocSizePerSlot: slot=%d expected=%zu actual=%zu",
+                    slot, want, have);
+    }
+}
+
+/* Lookup a kvobj for read or write operations, or return NULL if the it is not
+ * found in the specified DB. This function implements the functionality of
+ * lookupKeyRead(), lookupKeyWrite() and their ...WithFlags() variants.
+ *
+ * link - If key found, return the link of the key.
+ *        If key not found, return the bucket link, where the key should be added.
+ *        Or NULL if dict wasn't allocated yet.
+ *
+ * Side-effects of calling this function:
+ *
+ * 1. A key gets expired if it reached it's TTL.
+ * 2. The key's last access time is updated.
+ * 3. The global keys hits/misses stats are updated (reported in INFO).
+ * 4. If keyspace notifications are enabled, a "keymiss" notification is fired.
+ *
+ * Flags change the behavior of this command:
+ *
+ *  LOOKUP_NONE (or zero): No special flags are passed.
+ *  LOOKUP_NOTOUCH: Don't alter the last access time of the key.
+ *  LOOKUP_NONOTIFY: Don't trigger keyspace event on key miss.
+ *  LOOKUP_NOSTATS: Don't increment key hits/misses counters.
+ *  LOOKUP_WRITE: Prepare the key for writing (delete expired keys even on
+ *                replicas, use separate keyspace stats and events (TODO)).
+ *  LOOKUP_NOEXPIRE: Perform expiration check, but avoid deleting the key,
+ *                   so that we don't have to propagate the deletion.
+ *
+ * Note: this function also returns NULL if the key is logically expired but
+ * still existing, in case this is a replica and the LOOKUP_WRITE is not set.
+ * Even if the key expiry is master-driven, we can correctly report a key is
+ * expired on replicas even if the master is lagging expiring our key via DELs
+ * in the replication link. */
+kvobj *lookupKey(redisDb *db, robj *key, int flags, dictEntryLink *link) {
+
+    kvobj *val = dbFindByLink(db, key->ptr, link);
+
+    if (val) {
+        /* Forcing deletion of expired keys on a replica makes the replica
+         * inconsistent with the master. We forbid it on readonly replicas, but
+         * we have to allow it on writable replicas to make write commands
+         * behave consistently.
+         *
+         * It's possible that the WRITE flag is set even during a readonly
+         * command, since the command may trigger events that cause modules to
+         * perform additional writes. */
+        int is_ro_replica = server.masterhost && server.repl_slave_ro;
+        int expire_flags = 0;
+        if (flags & LOOKUP_WRITE && !is_ro_replica)
+            expire_flags |= EXPIRE_FORCE_DELETE_EXPIRED;
+        if (flags & LOOKUP_NOEXPIRE)
+            expire_flags |= EXPIRE_AVOID_DELETE_EXPIRED;
+        if (flags & LOOKUP_ACCESS_EXPIRED)
+            expire_flags |= EXPIRE_ALLOW_ACCESS_EXPIRED;
+        if (flags & LOOKUP_ACCESS_TRIMMED)
+            expire_flags |= EXPIRE_ALLOW_ACCESS_TRIMMED;
+        if (expireIfNeeded(db, key, val, expire_flags) != KEY_VALID) {
+            /* The key is no longer valid. */
+            val = NULL;
+            if (link) *link = NULL;
+        }
+    }
+
+    if (val) {
+        /* Update the access time for the ageing algorithm.
+         * Don't do it if we have a saving child, as this will trigger
+         * a copy on write madness. */
+        if (((flags & LOOKUP_NOTOUCH) == 0) &&
+            (server.current_client && server.current_client->flags & CLIENT_NO_TOUCH) &&
+            (server.executing_client && server.executing_client->cmd->proc != touchCommand))
+            flags |= LOOKUP_NOTOUCH;
+        if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){
+            if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
+                updateLFU(val);
+            } else if (!(server.maxmemory_policy & MAXMEMORY_FLAG_LRM)) {
+                /* LRM policy should NOT update timestamp on reads. */
+                val->lru = LRU_CLOCK();
+            }
+        }
+
+        if (!(flags & (LOOKUP_NOSTATS | LOOKUP_WRITE)))
+            server.stat_keyspace_hits++;
+        /* TODO: Use separate hits stats for WRITE */
+    } else {
+        if (!(flags & (LOOKUP_NONOTIFY | LOOKUP_WRITE)))
+            notifyKeyspaceEvent(NOTIFY_KEY_MISS, "keymiss", key, db->id);
+        if (!(flags & (LOOKUP_NOSTATS | LOOKUP_WRITE)))
+            server.stat_keyspace_misses++;
+        /* TODO: Use separate misses stats and notify event for WRITE */
+    }
+
+    return val;
+}
+
+/* Lookup a key for read operations, or return NULL if the key is not found
+ * in the specified DB.
+ *
+ * This API should not be used when we write to the key after obtaining
+ * the object linked to the key, but only for read only operations.
+ *
+ * This function is equivalent to lookupKey(). The point of using this function
+ * rather than lookupKey() directly is to indicate that the purpose is to read
+ * the key. */
+kvobj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
+    serverAssert(!(flags & LOOKUP_WRITE));
+    return lookupKey(db, key, flags, NULL);
+}
+
+/* Like lookupKeyReadWithFlags(), but does not use any flag, which is the
+ * common case. */
+kvobj *lookupKeyRead(redisDb *db, robj *key) {
+    return lookupKeyReadWithFlags(db,key,LOOKUP_NONE);
+}
+
+/* Lookup a key for write operations, and as a side effect, if needed, expires
+ * the key if its TTL is reached. It's equivalent to lookupKey() with the
+ * LOOKUP_WRITE flag added.
+ *
+ * Returns the linked value object if the key exists or NULL if the key
+ * does not exist in the specified DB. */
+kvobj *lookupKeyWriteWithFlags(redisDb *db, robj *key, int flags) {
+    return lookupKey(db, key, flags | LOOKUP_WRITE, NULL);
+}
+
+kvobj *lookupKeyWrite(redisDb *db, robj *key) {
+    return lookupKeyWriteWithFlags(db, key, LOOKUP_NONE);
+}
+
+/* Like lookupKeyWrite(), but accepts ref to optional `link`
+ *
+ * link - If key found, updated to link the key.
+ *        If key not found, updated to the bucket where the key should be added.
+ *        If key not found and dict is empty, it is set to NULL
+ */
+kvobj *lookupKeyWriteWithLink(redisDb *db, robj *key, dictEntryLink *link) {
+    return lookupKey(db, key, LOOKUP_NONE | LOOKUP_WRITE, link);
+}
+
+kvobj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) {
+    kvobj *kv = lookupKeyRead(c->db, key);
+    if (!kv) addReplyOrErrorObject(c, reply);
+    return kv;
+}
+
+kvobj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply) {
+    kvobj *kv = lookupKeyWrite(c->db, key);
+    if (!kv) addReplyOrErrorObject(c, reply);
+    return kv;
+}
+
+/* Add a key-value entry to the DB.
+ *
+ * A copy of 'key' is stored in the database. The caller must ensure the
+ * `key` is properly freed by calling decrRefcount(key).
+ *
+ * The value may (if its reference counter == 1) be reallocated and become
+ * invalid after a call to this function. The (possibly reallocated) value is
+ * stored in the database and the 'valref' pointer is updated to point to the
+ * new allocation.
+ *
+ * The reference counter of the value pointed to by valref is not incremented,
+ * so the caller should not free the value using decrRefcount after calling this
+ * function.
+ *
+ * link - Optional link to bucket where the key should be added.
+ *          On return, get updated, by need, to the inserted key.
+ *          
+ * keymeta - Defines metadata to be attached to the key. Including optional 
+ *           expiration and modules metadata to be copied (REQUIRED).
+ */
+kvobj *dbAddInternal(redisDb *db, robj *key, robj **valref, dictEntryLink *link, 
+                     const KeyMetaSpec *keymeta) 
+{
+    int slot = getKeySlot(key->ptr);
+    dictEntryLink tmp = NULL;
+    if (link == NULL) link = &tmp;
+    robj *val = *valref;
+    kvobj *kv = kvobjSet(key->ptr, val, keymeta->metabits);
+    initObjectLRUOrLFU(kv);
+    kvstoreDictSetAtLink(db->keys, slot, kv, link, 1);
+    
+    /* Handle metadata (expiration and modules metadata) */
+    if (keymeta->metabits) {
+        if (keymeta->metabits & KEY_META_MASK_EXPIRE) {
+            /* Expiry is always the first meta (from last) */
+            long long expire = keymeta->meta[KEY_META_ID_MAX - 1];
+            kvobj *newkv = setExpireByLink(NULL, db, key->ptr, expire, *link);
+            serverAssert(newkv == kv);
+        }
+        
+        /* memcpy modules metadata to beginning of kvobj */
+        if (keymeta->metabits & KEY_META_MASK_MODULES)
+            /* Also trivial overwrite expire */
+            memcpy(kvobjGetAllocPtr(kv), 
+                   keymeta->meta + KEY_META_ID_MAX - keymeta->numMeta, 
+                   keymeta->numMeta * sizeof(uint64_t));
+    }
+
+    signalKeyAsReady(db, key, kv->type);
+    notifyKeyspaceEvent(NOTIFY_NEW,"new",key,db->id);
+    updateKeysizesHist(db, slot, kv->type, -1, getObjectLength(kv)); /* add hist */
+    if (server.memory_tracking_per_slot)
+        updateSlotAllocSize(db, slot, 0, kvobjAllocSize(kv));
+    *valref = kv;
+    return kv;
+}
+
+/* Read dbAddInternal() comment */
+kvobj *dbAdd(redisDb *db, robj *key, robj **valref) {
+    KeyMetaSpec keyMetaEmpty; /* No metadata added */
+    keyMetaSpecInit(&keyMetaEmpty);
+    return dbAddInternal(db, key, valref, NULL, &keyMetaEmpty);
+}
+
+kvobj *dbAddByLink(redisDb *db, robj *key, robj **valref, dictEntryLink *link) {
+    KeyMetaSpec keyMetaEmpty; /* No metadata added */
+    keyMetaSpecInit(&keyMetaEmpty);
+    return dbAddInternal(db, key, valref, link, &keyMetaEmpty);
+}
+
+/* Returns key's hash slot when cluster mode is enabled, or 0 when disabled.
+ * The only difference between this function and getKeySlot, is that it's not using cached key slot from the current_client
+ * and always calculates CRC hash.
+ * This is useful when slot needs to be calculated for a key that user didn't request for, such as in case of eviction. */
+int calculateKeySlot(sds key) {
+    return server.cluster_enabled ? keyHashSlot(key, (int) sdslen(key)) : 0;
+}
+
+/* Return slot-specific dictionary for key based on key's hash slot when cluster mode is enabled, else 0.*/
+int getKeySlot(sds key) {
+    if (!server.cluster_enabled) return 0;
+    /* This is performance optimization that uses pre-set slot id from the current command,
+     * in order to avoid calculation of the key hash.
+     *
+     * This optimization is only used when current_client flag `CLIENT_EXECUTING_COMMAND` is set.
+     * It only gets set during the execution of command under `call` method. Other flows requesting
+     * the key slot would fallback to calculateKeySlot.
+     */
+    if (server.current_client && server.current_client->slot >= 0 && server.current_client->flags & CLIENT_EXECUTING_COMMAND) {
+        debugServerAssertWithInfo(server.current_client, NULL,
+                                  (int)keyHashSlot(key, (int)sdslen(key)) == server.current_client->slot);
+        return server.current_client->slot;
+    }
+    int slot = keyHashSlot(key, (int)sdslen(key));
+    return slot;
+}
+
+/* Return the slot of the key in the command.
+ * INVALID_CLUSTER_SLOT if no keys, CLUSTER_CROSSSLOT if cross slot, otherwise the slot number. */
+int getSlotFromCommand(struct redisCommand *cmd, robj **argv, int argc) {
+    if (!cmd || !server.cluster_enabled) return INVALID_CLUSTER_SLOT;
+
+    /* Get the keys from the command */
+    getKeysResult result = GETKEYS_RESULT_INIT;
+    getKeysFromCommand(cmd, argv, argc, &result);
+
+    /* Extract slot from the keys result. */
+    int slot = extractSlotFromKeysResult(argv, &result);
+    getKeysFreeResult(&result);
+    return slot;
+}
+
+/* This is a special version of dbAdd() that is used only when loading
+ * keys from the RDB file: the key is passed as an SDS string that is
+ * copied by the function and freed by the caller.
+ *
+ * Moreover this function will not abort if the key is already busy, to
+ * give more control to the caller, nor will signal the key as ready
+ * since it is not useful in this context.
+ *
+ * If added to db, returns pointer to the object, Otherwise NULL is returned.
+ */
+kvobj *dbAddRDBLoad(redisDb *db, sds key, robj **valref, const KeyMetaSpec *keyMetaSpec) {
+    /* Add new kvobj to the db. */
+    int slot = getKeySlot(key);
+
+    dictEntryLink link, bucket;
+    link = kvstoreDictFindLink(db->keys, slot, key, &bucket);
+
+    /* If already exists, return NULL */
+    if (link != NULL)
+        return NULL;
+
+    /* Create kvobj with metadata bits from KeyMetaSpec */
+    robj *val = *valref;
+    kvobj *kv = kvobjSet(key, val, keyMetaSpec->metabits);
+    initObjectLRUOrLFU(kv);
+    kvstoreDictSetAtLink(db->keys, slot, kv, &bucket, 1);
+
+    /* Handle metadata (expiration and modules metadata) */
+    if (keyMetaSpec->metabits) {
+        if (keyMetaSpec->metabits & KEY_META_MASK_EXPIRE) {
+            /* Expiry is always the first meta (from last) */
+            long long expire = keyMetaSpec->meta[KEY_META_ID_MAX - 1];
+            kvobj *newkv = setExpireByLink(NULL, db, key, expire, bucket);
+            serverAssert(newkv == kv);
+        }
+
+        /* memcpy modules metadata to beginning of kvobj */
+        if (keyMetaSpec->metabits & KEY_META_MASK_MODULES)
+            memcpy(kvobjGetAllocPtr(kv),
+                   keyMetaSpec->meta + KEY_META_ID_MAX - keyMetaSpec->numMeta,
+                   keyMetaSpec->numMeta * sizeof(uint64_t));
+    }
+
+    updateKeysizesHist(db, slot, kv->type, -1, (int64_t) getObjectLength(kv));
+    if (server.memory_tracking_per_slot)
+        updateSlotAllocSize(db, slot, 0, kvobjAllocSize(kv));
+    return *valref = kv;
+}
+
+/**
+ * Overwrite an existing key's value in db with a new value.
+ *
+ * - If the reference count of 'valref' is 1 the ownership of the value is
+ *   transferred to this function. The value may be reallocated, potentially
+ *   invalidating any external references to it. The (potentially reallocated)
+ *   value is stored in the database, and the 'valref' pointer is updated to
+ *   reflect the new allocation, if one occurs.
+ * - The reference counter of the value referenced by 'valref' is not incremented
+ *   so the caller must refrain from releasing it using decrRefCount after this
+ *   function is called.
+ * - This function does not modify the expire time of the existing key.
+ * - The 'overwrite' flag is an indication whether this is done as part of a
+ *   complete replacement of their key, which can be thought as a deletion and
+ *   replacement (in which case we need to emit deletion signals), or just an
+ *   update of a value of an existing key (when false).
+ * - The `link` is optional, can save lookup, if provided.
+ */
+static void dbSetValue(redisDb *db, robj *key, robj **valref, dictEntryLink link, 
+                       int overwrite, int updateKeySizes, int keepTTL) {
+    int freeModuleMeta = 0;
+    robj *val = *valref;
+    int slot = getKeySlot(key->ptr);
+    size_t oldsize = 0;
+    if (!link) {
+        link = kvstoreDictFindLink(db->keys, slot, key->ptr, NULL);
+        serverAssertWithInfo(NULL, key, link != NULL); /* expected to exist */
+    }
+    kvobj *old = dictGetKV(*link);
+    kvobj *kvNew;
+
+    int64_t oldlen = (int64_t) getObjectLength(old);
+    int oldtype = old->type;
+
+    /* if hash with HFEs, take care to remove from global HFE DS before attempting
+     * to manipulate and maybe free kvOld object */
+    if (old->type == OBJ_HASH)
+        estoreRemove(db->subexpires, slot, old);
+
+    long long oldExpire = getExpire(db, key->ptr, old);
+
+    /* All metadata will be kept if not `overwrite` for the new object  */
+    uint32_t newKeyMetaBits = old->metabits;
+    /* clear expire if not keepTTL or no old expire */
+    if ((!keepTTL) || (oldExpire == -1))
+        newKeyMetaBits &= ~KEY_META_MASK_EXPIRE; 
+
+    if (overwrite) {
+        /* On overwrite, discard module metadata excluding expire if set */
+        newKeyMetaBits &= KEY_META_MASK_EXPIRE;
+        /* RM_StringDMA may call dbUnshareStringValue which may free val, so we
+         * need to incr to retain old */
+        incrRefCount(old);
+
+        /* Free related metadata. Ignore builtin metadata (currently only expire) */
+        if (getModuleMetaBits(old->metabits)) {
+            keyMetaOnUnlink(db, key, old);
+            freeModuleMeta = 1;
+        }
+
+        /* Although the key is not really deleted from the database, we regard
+         * overwrite as two steps of unlink+add, so we still need to call the unlink
+         * callback of the module. */
+        moduleNotifyKeyUnlink(key,old,db->id,DB_FLAG_KEY_OVERWRITE);
+        /* We want to try to unblock any module clients or clients using a blocking XREADGROUP */
+        signalDeletedKeyAsReady(db,key,old->type);
+        decrRefCount(old);
+        /* Because of RM_StringDMA, old may be changed, so we need get old again */
+        old = dictGetKV(*link);
+    }
+    if (server.memory_tracking_per_slot)
+        oldsize = kvobjAllocSize(old);
+
+    if ((old->refcount == 1 && old->encoding != OBJ_ENCODING_EMBSTR) &&
+        (val->refcount == 1 && val->encoding != OBJ_ENCODING_EMBSTR) && (!freeModuleMeta))
+    {
+        /* Keep old object in the database. Just swap it's ptr, type and
+         * encoding with the content of val. */
+        robj tmp = *old;
+        old->type = val->type;
+        old->encoding = val->encoding;
+        old->ptr = val->ptr;
+        val->type = tmp.type;
+        val->encoding = tmp.encoding;
+        val->ptr = tmp.ptr;
+        /* Set new to old to keep the old object. Set old to val to be freed below. */
+        kvNew = old;
+        old = val;
+
+        /* Handle TTL in the optimization path */
+        if ((!keepTTL) && (oldExpire >= 0))
+            removeExpire(db, key);
+    } else {
+        /* Replace the old value at its location in the key space. */
+        val->lru = old->lru;
+        
+        kvNew = kvobjSet(key->ptr, val, newKeyMetaBits);
+        kvstoreDictSetAtLink(db->keys, slot, kvNew, &link, 0);
+
+        /* if expiry replace the old value at its location in the expire space. */
+        if (oldExpire != -1) {
+            if (keepTTL) {
+                kvobjSetExpire(kvNew, oldExpire); /* kvNew not reallocated here */
+                dictEntryLink exLink = kvstoreDictFindLink(db->expires, slot,
+                                                           key->ptr, NULL);
+                serverAssertWithInfo(NULL, key, exLink != NULL);
+                kvstoreDictSetAtLink(db->expires, slot, kvNew, &exLink, 0);
+            } else {
+                kvstoreDictDelete(db->expires, slot, key->ptr);
+            }
+        }
+
+        if (newKeyMetaBits & KEY_META_MASK_MODULES)
+            keyMetaTransition(old, kvNew);
+    }
+
+    /* Remove old key and add new key to KEYSIZES histogram */
+    int64_t newlen = (int64_t) getObjectLength(kvNew);
+    if (updateKeySizes) {
+        /* Save one call if old and new are the same type */
+        if (oldtype == kvNew->type) {
+            updateKeysizesHist(db, slot, oldtype, oldlen, newlen);
+        } else {
+            updateKeysizesHist(db, slot, oldtype, oldlen, -1);
+            updateKeysizesHist(db, slot, kvNew->type, -1, newlen);
+        }
+    }
+
+    if (server.memory_tracking_per_slot)
+        updateSlotAllocSize(db, slot, oldsize, kvobjAllocSize(kvNew));
+
+    if (server.io_threads_num > 1 && old->encoding == OBJ_ENCODING_RAW) {
+        /* In multi-threaded mode, the OBJ_ENCODING_RAW string object usually is
+         * allocated in the IO thread, so we defer the free to the IO thread.
+         * Besides, we never free a string object in BIO threads, so, even with
+         * lazyfree-lazy-server-del enabled, a fallback to main thread freeing
+         * due to defer free failure doesn't go against the config intention. */
+        tryDeferFreeClientObject(server.current_client, DEFERRED_OBJECT_TYPE_ROBJ, old);
+    } else if (server.lazyfree_lazy_server_del) {
+        freeObjAsync(key, old, db->id);
+    } else {
+        decrRefCount(old);
+    }
+    *valref = kvNew;
+}
+
+/* Replace an existing key with a new value, we just replace value and don't
+ * emit any events */
+void dbReplaceValue(redisDb *db, robj *key, robj **valref, int updateKeySizes) {
+    dbSetValue(db, key, valref, NULL, 0, updateKeySizes, 1);
+}
+
+/* Replace an existing key with a new value (don't emit any events)
+ *
+ * parameter 'link' is optional. If provided, saves lookup.
+ */
+void dbReplaceValueWithLink(redisDb *db, robj *key, robj **val, dictEntryLink link) {
+    dbSetValue(db, key, val, link, 0, 1, 1);
+}
+
+/* High level Set operation. This function can be used in order to set
+ * a key, whatever it was existing or not, to a new object.
+ *
+ * 1) The value may be reallocated when adding it to the database. The value
+ *    pointer 'valref' is updated to point to the reallocated object. The
+ *    reference count of the value object is *not* incremented.
+ * 2) clients WATCHing for the destination key notified.
+ * 3) The expire time of the key is reset (the key is made persistent),
+ *    unless 'SETKEY_KEEPTTL' is enabled in flags.
+ * 4) The key lookup can take place outside this interface outcome will be
+ *    delivered with 'SETKEY_ALREADY_EXIST' or 'SETKEY_DOESNT_EXIST'
+ *
+ * All the new keys in the database should be created via this interface.
+ * The client 'c' argument may be set to NULL if the operation is performed
+ * in a context where there is no clear client performing the operation. */
+void setKey(client *c, redisDb *db, robj *key, robj **valref, int flags) {
+    setKeyByLink(c, db, key, valref, flags, NULL);
+}
+
+/* Like setKey(), but accepts an optional link
+ *
+ * - If flags is set with SETKEY_ALREADY_EXIST, then `link` must be provided
+ * - If flags is set with SETKEY_DOESNT_EXIST, then `link` is optional. If
+ *   provided, it will point to the bucket where the key should be added.
+ * - If flag is not set (0) then add or update key, and `link` must be NULL
+ * On return, link get updated, by need, to the inserted kvobj.
+ */
+void setKeyByLink(client *c, redisDb *db, robj *key, robj **valref, int flags, dictEntryLink *plink) {
+    dictEntryLink dummy = NULL, *link = plink ? plink : &dummy;
+    int exists;
+    kvobj *oldval = NULL;
+
+    if (flags & SETKEY_ALREADY_EXIST) {
+        debugServerAssert((*link) != NULL);
+        oldval = dictGetKV(**link);
+        exists = 1;
+    } else if (flags & SETKEY_DOESNT_EXIST) {
+        /* link is optional */
+        exists = 0;
+    } else {
+        /* Add or update key */
+        oldval = lookupKeyWriteWithLink(db, key, link);
+        exists = oldval != NULL;
+    }
+
+    if (exists) {
+        int oldtype = oldval->type;
+        int newtype = (*valref)->type;
+
+        /* Update the value of an existing key */
+        dbSetValue(db, key, valref, *link, 1, 1, flags & SETKEY_KEEPTTL);
+
+        /* Notify keyspace events for override and type change */
+        notifyKeyspaceEvent(NOTIFY_OVERWRITTEN, "overwritten", key, db->id);
+        if (oldtype != newtype)
+            notifyKeyspaceEvent(NOTIFY_TYPE_CHANGED, "type_changed", key, db->id);
+    } else {
+        /* Add the new key to the database */
+        dbAddByLink(db, key, valref, link);
+    }
+
+    /* Signal key modification and update LRM timestamp. */
+    keyModified(c,db,key,*valref,!(flags & SETKEY_NO_SIGNAL));
+}
+
+/* During atomic slot migration, keys that are being imported are in an
+ * intermediate state. we cannot access them and therefore skip them.
+ *
+ * This callback function now is used by:
+ * - dbRandomKey
+ * - keysCommand
+ * - scanCommand
+ */
+static int accessKeysShouldSkipDictIndex(int didx) {
+    return !clusterCanAccessKeysInSlot(didx);
+}
+
+/* Return a random key, in form of a Redis object.
+ * If there are no keys, NULL is returned.
+ *
+ * The function makes sure to return keys not already expired. */
+robj *dbRandomKey(redisDb *db) {
+    dictEntry *de;
+    int maxtries = 100;
+    int allvolatile = kvstoreSize(db->keys) == kvstoreSize(db->expires);
+
+    while(1) {
+        robj *keyobj;
+        int randomSlot = kvstoreGetFairRandomDictIndex(db->keys, accessKeysShouldSkipDictIndex, 16, 1);
+        if (randomSlot == -1) return NULL;
+        de = kvstoreDictGetFairRandomKey(db->keys, randomSlot);
+        if (de == NULL) return NULL;
+
+        kvobj *kv = dictGetKV(de);
+        sds key = kvobjGetKey(kv);
+        keyobj = createStringObject(key,sdslen(key));
+        if (allvolatile && (server.masterhost || isPausedActions(PAUSE_ACTION_EXPIRE)) && --maxtries == 0) {
+            /* If the DB is composed only of keys with an expire set,
+             * it could happen that all the keys are already logically
+             * expired in the slave, so the function cannot stop because
+             * expireIfNeeded() is false, nor it can stop because
+             * dictGetFairRandomKey() returns NULL (there are keys to return).
+             * To prevent the infinite loop we do some tries, but if there
+             * are the conditions for an infinite loop, eventually we
+             * return a key name that may be already expired. */
+            return keyobj;
+        }
+        if (expireIfNeeded(db, keyobj, kv, 0) != KEY_VALID) {
+            decrRefCount(keyobj);
+            continue; /* search for another key. This expired. */
+        }
+
+        return keyobj;
+    }
+}
+
+/* Helper for sync and async delete. */
+int dbGenericDelete(redisDb *db, robj *key, int async, int flags) {
+    dictEntryLink link;
+    int table;
+    int slot = getKeySlot(key->ptr);
+    link = kvstoreDictTwoPhaseUnlinkFind(db->keys, slot, key->ptr, &table);
+
+    if (link) {
+        kvobj *kv = dictGetKV(*link);
+
+        int64_t oldlen = (int64_t) getObjectLength(kv);
+        int type = kv->type;
+
+        /* If hash object with expiry on fields, remove it from HFE DS of DB */
+        if (type == OBJ_HASH)
+            estoreRemove(db->subexpires, slot, kv);
+
+        /* RM_StringDMA may call dbUnshareStringValue which may free kv, so we
+         * need to incr to retain kv */
+        incrRefCount(kv); /* refcnt=1->2 */
+        /* Metadata hook: notify unlink for key metadata cleanup. */
+        if (getModuleMetaBits(kv->metabits)) keyMetaOnUnlink(db, key, kv);
+        /* Tells the module that the key has been unlinked from the database. */
+        moduleNotifyKeyUnlink(key, kv, db->id, flags);
+        /* We want to try to unblock any module clients or clients using a blocking XREADGROUP */
+        signalDeletedKeyAsReady(db,key,type);
+        /* We should call decr before freeObjAsync. If not, the refcount may be
+         * greater than 1, so freeObjAsync doesn't work */
+        decrRefCount(kv);
+
+        /* Because of dbUnshareStringValue, the val in db may change. */
+        kv = dictGetKV(*link);
+        
+        /* if expirable, delete an entry from the expires dict is not decrRefCount of kvobj */
+        if (kvobjGetExpire(kv) != -1)
+            kvstoreDictDelete(db->expires, slot, key->ptr);
+
+        if (async) {
+            if (server.memory_tracking_per_slot)
+                updateSlotAllocSize(db, slot, kvobjAllocSize(kv), 0);
+            freeObjAsync(key, kv, db->id);
+            /* Set the key to NULL in the main dictionary. */
+            kvstoreDictSetAtLink(db->keys, slot, NULL, &link, 0);
+        }
+        kvstoreDictTwoPhaseUnlinkFree(db->keys, slot, link, table);
+
+        /* remove key from histogram */
+        if(!(flags & DB_FLAG_NO_UPDATE_KEYSIZES))
+            updateKeysizesHist(db, slot, type, oldlen, -1);
+        return 1;
+    } else {
+        return 0;
+    }
+}
+
+/* Delete a key, value, and associated expiration entry if any, from the DB */
+int dbSyncDelete(redisDb *db, robj *key) {
+    return dbGenericDelete(db, key, 0, DB_FLAG_KEY_DELETED);
+}
+
+/* Delete a key, value, and associated expiration entry if any, from the DB. If
+ * the value consists of many allocations, it may be freed asynchronously. */
+int dbAsyncDelete(redisDb *db, robj *key) {
+    return dbGenericDelete(db, key, 1, DB_FLAG_KEY_DELETED);
+}
+
+/* This is a wrapper whose behavior depends on the Redis lazy free
+ * configuration. Deletes the key synchronously or asynchronously. */
+int dbDelete(redisDb *db, robj *key) {
+    return dbGenericDelete(db, key, server.lazyfree_lazy_server_del, DB_FLAG_KEY_DELETED);
+}
+
+/* Similar to dbDelete(), but does not update the keysizes histogram.
+ * This is used when we want to delete a key without affecting the histogram,
+ * typically in cases where a command flow deletes elements from a collection
+ * and then deletes the collection itself. In such cases, using dbDelete()
+ * would incorrectly decrement bin #0. A corresponding test should be added
+ * to `info-keysizes.tcl`. */
+int dbDeleteSkipKeysizesUpdate(redisDb *db, robj *key) {
+    return dbGenericDelete(db, key, server.lazyfree_lazy_server_del,
+                    DB_FLAG_KEY_DELETED | DB_FLAG_NO_UPDATE_KEYSIZES);
+}
+
+/* Prepare the string object stored at 'key' to be modified destructively
+ * to implement commands like SETBIT or APPEND.
+ *
+ * An object is usually ready to be modified unless one of the two conditions
+ * are true:
+ *
+ * 1) The object 'o' is shared (refcount > 1), we don't want to affect
+ *    other users.
+ * 2) The object encoding is not "RAW".
+ *
+ * If the object is found in one of the above conditions (or both) by the
+ * function, an unshared / not-encoded copy of the string object is stored
+ * at 'key' in the specified 'db'. Otherwise the object 'o' itself is
+ * returned.
+ *
+ * USAGE:
+ *
+ * The object 'o' is what the caller already obtained by looking up 'key'
+ * in 'db', the usage pattern looks like this:
+ *
+ * o = lookupKeyWrite(db,key);
+ * if (checkType(c,o,OBJ_STRING)) return;
+ * o = dbUnshareStringValue(db,key,o);
+ *
+ * At this point the caller is ready to modify the object, for example
+ * using an sdscat() call to append some data, or anything else.
+ */
+kvobj *dbUnshareStringValue(redisDb *db, robj *key, kvobj *kv) {
+    return dbUnshareStringValueByLink(db,key,kv,NULL);
+}
+
+/* Like dbUnshareStringValue(), but accepts a optional link,
+ * which can be used if we already have one, thus saving the dbFind call. */
+kvobj *dbUnshareStringValueByLink(redisDb *db, robj *key, kvobj *o, dictEntryLink link) {
+    serverAssert(o->type == OBJ_STRING);
+    if (o->refcount != 1 || o->encoding != OBJ_ENCODING_RAW) {
+        robj *decoded = getDecodedObject(o);
+        o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr));
+        decrRefCount(decoded);
+        dbReplaceValueWithLink(db, key, &o, link);
+    }
+    return o;
+}
+
+/* Remove all keys from the database(s) structure. The dbarray argument
+ * may not be the server main DBs (could be a temporary DB).
+ *
+ * The dbnum can be -1 if all the DBs should be emptied, or the specified
+ * DB index if we want to empty only a single database.
+ * The function returns the number of keys removed from the database(s). */
+long long emptyDbStructure(redisDb *dbarray, int dbnum, int async,
+                           void(callback)(dict*))
+{
+    long long removed = 0;
+    int startdb, enddb;
+
+    if (dbnum == -1) {
+        startdb = 0;
+        enddb = server.dbnum-1;
+    } else {
+        startdb = enddb = dbnum;
+    }
+
+    for (int j = startdb; j <= enddb; j++) {
+        removed += kvstoreSize(dbarray[j].keys);
+        if (async) {
+            emptyDbAsync(&dbarray[j]);
+        } else {
+            /* Destroy sub-expires before deleting the kv-objects since ebuckets
+             * data structure is embedded in the stored kv-objects. */
+            estoreEmpty(dbarray[j].subexpires);
+            kvstoreEmpty(dbarray[j].keys, callback);
+            kvstoreEmpty(dbarray[j].expires, callback);
+        }
+        /* Because all keys of database are removed, reset average ttl. */
+        dbarray[j].avg_ttl = 0;
+        dbarray[j].expires_cursor = 0;
+    }
+
+    return removed;
+}
+
+/* Remove all data (keys and functions) from all the databases in a
+ * Redis server. If callback is given the function is called from
+ * time to time to signal that work is in progress.
+ *
+ * The dbnum can be -1 if all the DBs should be flushed, or the specified
+ * DB number if we want to flush only a single Redis database number.
+ *
+ * Flags are be EMPTYDB_NO_FLAGS if no special flags are specified or
+ * EMPTYDB_ASYNC if we want the memory to be freed in a different thread
+ * and the function to return ASAP. EMPTYDB_NOFUNCTIONS can also be set
+ * to specify that we do not want to delete the functions.
+ *
+ * On success the function returns the number of keys removed from the
+ * database(s). Otherwise -1 is returned in the specific case the
+ * DB number is out of range, and errno is set to EINVAL. */
+long long emptyData(int dbnum, int flags, void(callback)(dict*)) {
+    int async = (flags & EMPTYDB_ASYNC);
+    int with_functions = !(flags & EMPTYDB_NOFUNCTIONS);
+    RedisModuleFlushInfoV1 fi = {REDISMODULE_FLUSHINFO_VERSION,!async,dbnum};
+    long long removed = 0;
+
+    if (dbnum < -1 || dbnum >= server.dbnum) {
+        errno = EINVAL;
+        return -1;
+    }
+
+    if (dbnum == -1 || dbnum == 0)
+        asmCancelTrimJobs();
+
+    /* Fire the flushdb modules event. */
+    moduleFireServerEvent(REDISMODULE_EVENT_FLUSHDB,
+                          REDISMODULE_SUBEVENT_FLUSHDB_START,
+                          &fi);
+
+    /* Make sure the WATCHed keys are affected by the FLUSH* commands.
+     * Note that we need to call the function while the keys are still
+     * there. */
+    signalFlushedDb(dbnum, async, NULL);
+
+    /* Empty redis database structure. */
+    removed = emptyDbStructure(server.db, dbnum, async, callback);
+
+    if (dbnum == -1) flushSlaveKeysWithExpireList();
+
+    if (with_functions) {
+        serverAssert(dbnum == -1);
+        functionsLibCtxClearCurrent(async);
+    }
+
+    /* Also fire the end event. Note that this event will fire almost
+     * immediately after the start event if the flush is asynchronous. */
+    moduleFireServerEvent(REDISMODULE_EVENT_FLUSHDB,
+                          REDISMODULE_SUBEVENT_FLUSHDB_END,
+                          &fi);
+
+    return removed;
+}
+
+/* Initialize temporary db on replica for use during diskless replication. */
+redisDb *initTempDb(void) {
+    int slot_count_bits = 0;
+    int flags = KVSTORE_ALLOCATE_DICTS_ON_DEMAND;
+    if (server.cluster_enabled) {
+        slot_count_bits = CLUSTER_SLOT_MASK_BITS;
+        flags |= KVSTORE_FREE_EMPTY_DICTS;
+    }
+    redisDb *tempDb = zcalloc(sizeof(redisDb)*server.dbnum);
+    for (int i=0; i<server.dbnum; i++) {
+        tempDb[i].id = i;
+        tempDb[i].keys = kvstoreCreate(&kvstoreExType, &dbDictType, slot_count_bits,
+                                       flags);
+        tempDb[i].expires = kvstoreCreate(&kvstoreBaseType, &dbExpiresDictType,
+                                          slot_count_bits, flags);
+        tempDb[i].subexpires = estoreCreate(&subexpiresBucketsType, slot_count_bits);
+    }
+
+    return tempDb;
+}
+
+/* Discard tempDb, this can be slow (similar to FLUSHALL), but it's always async. */
+void discardTempDb(redisDb *tempDb) {
+    int async = 1;
+
+    /* Release temp DBs. */
+    emptyDbStructure(tempDb, -1, async, NULL);
+    for (int i=0; i<server.dbnum; i++) {
+        /* Destroy sub-expires before deleting the kv-objects since ebuckets
+         * data structure is embedded in the stored kv-objects. */
+        estoreRelease(tempDb[i].subexpires);
+        kvstoreRelease(tempDb[i].keys);
+        kvstoreRelease(tempDb[i].expires);
+    }
+
+    zfree(tempDb);
+}
+
+int selectDb(client *c, int id) {
+    if (id < 0 || id >= server.dbnum)
+        return C_ERR;
+    c->db = &server.db[id];
+    return C_OK;
+}
+
+long long dbTotalServerKeyCount(void) {
+    long long total = 0;
+    int j;
+    for (j = 0; j < server.dbnum; j++) {
+        total += kvstoreSize(server.db[j].keys);
+    }
+    return total;
+}
+
+/*-----------------------------------------------------------------------------
+ * Hooks for key space changes.
+ *
+ * Every time a key in the database is modified the function
+ * keyModified() is called.
+ *
+ * Every time a DB is flushed the function signalFlushDb() is called.
+ *----------------------------------------------------------------------------*/
+
+/* Called when a key is modified to update LRM timestamp
+ * and optionally signal watchers/tracking clients.
+ *
+ * Arguments:
+ * - c: client (may be NULL if the key was modified out of a context of a client)
+ * - db: database containing the key
+ * - key: the key that was modified
+ * - val: the value object (if NULL, LRM won't be updated, e.g., for deleted keys)
+ * - signal: if true, trigger WATCH and client-side tracking invalidation
+ */
+void keyModified(client *c, redisDb *db, robj *key, robj *val, int signal) {
+    if (val) updateLRM(val);
+    if (signal) {
+        touchWatchedKey(db,key);
+        trackingInvalidateKey(c,key,1);
+    }
+}
+
+void signalFlushedDb(int dbid, int async, slotRangeArray *slots) {
+    int startdb, enddb;
+    if (dbid == -1) {
+        startdb = 0;
+        enddb = server.dbnum-1;
+    } else {
+        startdb = enddb = dbid;
+    }
+
+    for (int j = startdb; j <= enddb; j++) {
+        scanDatabaseForDeletedKeys(&server.db[j], NULL, slots);
+        touchAllWatchedKeysInDb(&server.db[j], NULL, slots);
+    }
+
+    trackingInvalidateKeysOnFlush(async);
+
+    /* Changes in this method may take place in swapMainDbWithTempDb as well,
+     * where we execute similar calls, but with subtle differences as it's
+     * not simply flushing db. */
+}
+
+/*-----------------------------------------------------------------------------
+ * Type agnostic commands operating on the key space
+ *----------------------------------------------------------------------------*/
+
+/* Return the set of flags to use for the emptyData() call for FLUSHALL
+ * and FLUSHDB commands.
+ *
+ * sync: flushes the database in an sync manner.
+ * async: flushes the database in an async manner.
+ * no option: determine sync or async according to the value of lazyfree-lazy-user-flush.
+ *
+ * On success C_OK is returned and the flags are stored in *flags, otherwise
+ * C_ERR is returned and the function sends an error to the client. */
+int getFlushCommandFlags(client *c, int *flags) {
+    /* Parse the optional ASYNC option. */
+    if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"sync")) {
+        *flags = EMPTYDB_NO_FLAGS;
+    } else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"async")) {
+        *flags = EMPTYDB_ASYNC;
+    } else if (c->argc == 1) {
+        *flags = server.lazyfree_lazy_user_flush ? EMPTYDB_ASYNC : EMPTYDB_NO_FLAGS;
+    } else {
+        addReplyErrorObject(c,shared.syntaxerr);
+        return C_ERR;
+    }
+    return C_OK;
+}
+
+/* Flushes the whole server data set. */
+void flushAllDataAndResetRDB(int flags) {
+    server.dirty += emptyData(-1,flags,NULL);
+    if (server.child_type == CHILD_TYPE_RDB) killRDBChild();
+    if (server.saveparamslen > 0) {
+        rdbSaveInfo rsi, *rsiptr;
+        rsiptr = rdbPopulateSaveInfo(&rsi);
+        rdbSave(SLAVE_REQ_NONE,server.rdb_filename,rsiptr,RDBFLAGS_NONE);
+    }
+
+#if defined(USE_JEMALLOC)
+    /* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
+     * for large databases, flushdb blocks for long anyway, so a bit more won't
+     * harm and this way the flush and purge will be synchronous. */
+    if (!(flags & EMPTYDB_ASYNC)) {
+        /* Only clear the current thread cache.
+         * Ignore the return call since this will fail if the tcache is disabled. */
+        je_mallctl("thread.tcache.flush", NULL, NULL, NULL, 0);
+
+        jemalloc_purge();
+    }
+#endif
+}
+
+/* CB function on blocking ASYNC FLUSH completion
+ *
+ * Utilized by commands SFLUSH, FLUSHALL and FLUSHDB.
+ */
+void flushallSyncBgDone(uint64_t client_id, void *userdata) {
+    slotRangeArray *slots = userdata;
+    client *c = lookupClientByID(client_id);
+
+    /* Verify that client still exists and being blocked. */
+    if (!(c && c->flags & CLIENT_BLOCKED)) {
+        slotRangeArrayFree(slots);
+        return;
+    }
+
+    /* Update current_client (Called functions might rely on it) */
+    client *old_client = server.current_client;
+    server.current_client = c;
+
+    /* Don't update blocked_us since command was processed in bg by lazy_free thread */
+    updateStatsOnUnblock(c, 0 /*blocked_us*/, elapsedUs(c->bstate.lazyfreeStartTime), 0);
+
+    /* Only SFLUSH command pass user data pointer. */
+    if (slots)
+        replySlotsFlushAndFree(c, slots);
+    else
+        addReply(c, shared.ok);
+
+    /* mark client as unblocked */
+    unblockClient(c, 1);
+
+    if (c->flags & CLIENT_PENDING_COMMAND) {
+        c->flags &= ~CLIENT_PENDING_COMMAND;
+        /* The FLUSH command won't be reprocessed, FLUSH command is finished, but
+         * we still need to complete its full processing flow, including updating
+         * the replication offset. */
+        commandProcessed(c);
+    }
+
+    /* On flush completion, update the client's memory */
+    updateClientMemUsageAndBucket(c);
+
+    /* restore current_client */
+    server.current_client = old_client;
+}
+
+/* Common flush command implementation for FLUSHALL, FLUSHDB and SFLUSH.
+ *
+ * Return 1 indicates that flush SYNC is actually running in bg as blocking ASYNC
+ * Return 0 otherwise
+ *
+ * slots - provided only by SFLUSH command, otherwise NULL. Will be used on
+ *         completion to reply with the slots flush result. Ownership is passed
+ *         to the completion job in case of `blocking_async`.
+ */
+int flushCommandCommon(client *c, int type, int flags, slotRangeArray *slots) {
+    int blocking_async = 0; /* Flush SYNC option to run as blocking ASYNC */
+
+    /* in case of SYNC, check if we can optimize and run it in bg as blocking ASYNC */
+    if ((!(flags & EMPTYDB_ASYNC)) && (!(c->flags & CLIENT_AVOID_BLOCKING_ASYNC_FLUSH))) {
+        /* Run as ASYNC */
+        flags |= EMPTYDB_ASYNC;
+        blocking_async = 1;
+    }
+
+    /* Cancel all ASM tasks that overlap with the given slot ranges. */
+    clusterAsmCancelBySlotRangeArray(slots, c->argv[0]->ptr);
+
+    if (type == FLUSH_TYPE_ALL)
+        flushAllDataAndResetRDB(flags | EMPTYDB_NOFUNCTIONS);
+    else
+        server.dirty += emptyData(c->db->id,flags | EMPTYDB_NOFUNCTIONS,NULL);
+
+    /* Without the forceCommandPropagation, when DB(s) was already empty,
+     * FLUSHALL\FLUSHDB will not be replicated nor put into the AOF. */
+    forceCommandPropagation(c, PROPAGATE_REPL | PROPAGATE_AOF);
+
+    /* if blocking ASYNC, block client and add completion job request to BIO lazyfree
+     * worker's queue. To be called and reply with OK only after all preceding pending
+     * lazyfree jobs in queue were processed */
+    if (blocking_async) {
+        /* measure bg job till completion as elapsed time of flush command */
+        elapsedStart(&c->bstate.lazyfreeStartTime);
+
+        c->bstate.timeout = 0;
+        /* We still need to perform cleanup operations for the command, including
+         * updating the replication offset, so mark this command as pending to
+         * avoid command from being reset during unblock. */
+        c->flags |= CLIENT_PENDING_COMMAND;
+        blockClient(c,BLOCKED_LAZYFREE);
+        bioCreateCompRq(BIO_WORKER_LAZY_FREE, flushallSyncBgDone, c->id, slots);
+    }
+
+#if defined(USE_JEMALLOC)
+    /* jemalloc 5 doesn't release pages back to the OS when there's no traffic.
+     * for large databases, flushdb blocks for long anyway, so a bit more won't
+     * harm and this way the flush and purge will be synchronous.
+     *
+     * Take care purge only FLUSHDB for sync flow. FLUSHALL sync flow already
+     * applied at flushAllDataAndResetRDB. Async flow will apply only later on */
+    if ((type != FLUSH_TYPE_ALL) && (!(flags & EMPTYDB_ASYNC))) {
+        /* Only clear the current thread cache.
+         * Ignore the return call since this will fail if the tcache is disabled. */
+        je_mallctl("thread.tcache.flush", NULL, NULL, NULL, 0);
+
+        jemalloc_purge();
+    }
+#endif
+    return blocking_async;
+}
+
+/* FLUSHALL [SYNC|ASYNC]
+ *
+ * Flushes the whole server data set. */
+void flushallCommand(client *c) {
+    int flags;
+    if (getFlushCommandFlags(c,&flags) == C_ERR) return;
+
+    /* If FLUSH SYNC isn't running as blocking async, then reply */
+    if (flushCommandCommon(c, FLUSH_TYPE_ALL, flags, NULL) == 0)
+        addReply(c, shared.ok);
+}
+
+/* FLUSHDB [SYNC|ASYNC]
+ *
+ * Flushes the currently SELECTed Redis DB. */
+void flushdbCommand(client *c) {
+    int flags;
+    if (getFlushCommandFlags(c,&flags) == C_ERR) return;
+
+    /* If FLUSH SYNC isn't running as blocking async, then reply */
+    if (flushCommandCommon(c, FLUSH_TYPE_DB,flags, NULL) == 0)
+        addReply(c, shared.ok);
+
+}
+
+/* This command implements DEL and UNLINK. */
+void delGenericCommand(client *c, int lazy) {
+    int numdel = 0, j;
+
+    for (j = 1; j < c->argc; j++) {
+        if (expireIfNeeded(c->db, c->argv[j], NULL, 0) == KEY_DELETED)
+            continue;
+        int deleted  = lazy ? dbAsyncDelete(c->db,c->argv[j]) :
+                              dbSyncDelete(c->db,c->argv[j]);
+        if (deleted) {
+            keyModified(c,c->db,c->argv[j],NULL,1);
+            notifyKeyspaceEvent(NOTIFY_GENERIC,
+                "del",c->argv[j],c->db->id);
+            server.dirty++;
+            numdel++;
+        }
+    }
+    addReplyLongLong(c,numdel);
+}
+
+void delCommand(client *c) {
+    delGenericCommand(c,server.lazyfree_lazy_user_del);
+}
+
+/* DELEX key [IFEQ match-value|IFNE match-value|IFDEQ match-digest|IFDNE match-digest]
+ *
+ * Conditionally removes the specified key. A key is ignored if it does not
+ * exist.
+ * If no condition is specified the behavior is the same as DEL command.
+ * If condition is specified the key must be of STRING type.
+ *
+ * IFEQ/IFNE conditions check the match-value against the value of the key
+ * IFDEQ/IFDNE conditions check the match-digest against the digest of the key's value.*/
+void delexCommand(client *c) {
+    kvobj *o;
+    int deleted = 0, should_delete = 0;
+
+    /* If there are no conditions specified we just delete the key */
+    if (c->argc == 2) {
+        delGenericCommand(c, server.lazyfree_lazy_server_del);
+        return;
+    }
+
+    /* If we have more than two arguments the next two are condition and
+     * match-value */
+    if (c->argc != 4) {
+        addReplyErrorArity(c);
+        return;
+    }
+
+    robj *key = c->argv[1];
+    o = lookupKeyRead(c->db, key);
+    if (o == NULL) {
+        addReplyLongLong(c, 0);
+        return;
+    }
+
+    /* If any conditions are specified the only supported key type for now is
+     * string */
+    if (o->type != OBJ_STRING) {
+        addReplyError(c, "Key should be of string type if conditions are specified");
+        return;
+    }
+
+    char *condition = c->argv[2]->ptr;
+    if (!strcasecmp("ifeq", condition)) {
+        robj *valueobj = getDecodedObject(o);
+        sds match_value = c->argv[3]->ptr;
+        if (sdscmp(valueobj->ptr, match_value) == 0)
+            should_delete = 1;
+
+        decrRefCount(valueobj);
+    } else if (!strcasecmp("ifne", condition)) {
+        robj *valueobj = getDecodedObject(o);
+        sds match_value = c->argv[3]->ptr;
+        if (sdscmp(valueobj->ptr, match_value) != 0)
+           should_delete = 1;
+
+        decrRefCount(valueobj);
+    } else if (!strcasecmp("ifdeq", condition)) {
+        if (validateHexDigest(c, c->argv[3]->ptr) != C_OK)
+            return;
+
+        sds current_digest = stringDigest(o);
+        if (strcasecmp(current_digest, c->argv[3]->ptr) == 0)
+            should_delete = 1;
+
+        sdsfree(current_digest);
+    } else if (!strcasecmp("ifdne", condition)) {
+        if (validateHexDigest(c, c->argv[3]->ptr) != C_OK)
+            return;
+
+        sds current_digest = stringDigest(o);
+        if (strcasecmp(current_digest, c->argv[3]->ptr) != 0)
+            should_delete = 1;
+
+        sdsfree(current_digest);
+    } else {
+        addReplyError(c, "Invalid condition. Use IFEQ, IFNE, IFDEQ, or IFDNE");
+        return;
+    }
+
+    if (should_delete) {
+        deleted = server.lazyfree_lazy_server_del ?
+                  dbAsyncDelete(c->db, key) :
+                  dbSyncDelete(c->db, key);
+    }
+
+    if (deleted) {
+        rewriteClientCommandVector(c, 2, shared.del, key);
+        keyModified(c, c->db, key, NULL, 1);
+        notifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, c->db->id);
+        server.dirty++;
+    }
+
+    addReplyLongLong(c, deleted);
+}
+
+void unlinkCommand(client *c) {
+    delGenericCommand(c,1);
+}
+
+/* EXISTS key1 key2 ... key_N.
+ * Return value is the number of keys existing. */
+void existsCommand(client *c) {
+    long long count = 0;
+    int j;
+
+    for (j = 1; j < c->argc; j++) {
+        if (lookupKeyReadWithFlags(c->db,c->argv[j],LOOKUP_NOTOUCH)) count++;
+    }
+    addReplyLongLong(c,count);
+}
+
+void selectCommand(client *c) {
+    int id;
+
+    if (getIntFromObjectOrReply(c, c->argv[1], &id, NULL) != C_OK)
+        return;
+
+    if (server.cluster_enabled && id != 0) {
+        addReplyError(c,"SELECT is not allowed in cluster mode");
+        return;
+    }
+
+    if (id != 0) {
+        server.stat_cluster_incompatible_ops++;
+    }
+
+    if (selectDb(c,id) == C_ERR) {
+        addReplyError(c,"DB index is out of range");
+    } else {
+        addReply(c,shared.ok);
+    }
+}
+
+void randomkeyCommand(client *c) {
+    robj *key;
+
+    if ((key = dbRandomKey(c->db)) == NULL) {
+        addReplyNull(c);
+        return;
+    }
+
+    addReplyBulk(c,key);
+    decrRefCount(key);
+}
+
+void keysCommand(client *c) {
+    dictEntry *de;
+    sds pattern = c->argv[1]->ptr;
+    int plen = sdslen(pattern), allkeys, pslot = -1;
+    unsigned long numkeys = 0;
+    void *replylen = addReplyDeferredLen(c);
+    allkeys = (pattern[0] == '*' && plen == 1);
+    if (server.cluster_enabled && !allkeys) {
+        pslot = patternHashSlot(pattern, plen);
+    }
+    int has_slot = pslot != -1;
+    union {
+        kvstoreDictIterator kvs_di;
+        kvstoreIterator kvs_it;
+    } it;
+    if (has_slot) {
+        if (!kvstoreDictSize(c->db->keys, pslot) || accessKeysShouldSkipDictIndex(pslot)) {
+            /* Requested slot is empty */
+            setDeferredArrayLen(c,replylen,0);
+            return;
+        }
+        kvstoreInitDictSafeIterator(&it.kvs_di, c->db->keys, pslot);
+    } else {
+        kvstoreIteratorInit(&it.kvs_it, c->db->keys);
+    }
+
+    while ((de = has_slot ? kvstoreDictIteratorNext(&it.kvs_di) : kvstoreIteratorNext(&it.kvs_it)) != NULL) {
+        if (!has_slot && accessKeysShouldSkipDictIndex(kvstoreIteratorGetCurrentDictIndex(&it.kvs_it))) {
+            continue;
+        }
+
+        kvobj *kv = dictGetKV(de);
+        sds key = kvobjGetKey(kv);
+
+        if (allkeys || stringmatchlen(pattern,plen,key,sdslen(key),0)) {
+            if (!keyIsExpired(c->db, NULL, kv)) {
+                addReplyBulkCBuffer(c, key, sdslen(key));
+                numkeys++;
+            }
+        }
+        if (c->flags & CLIENT_CLOSE_ASAP)
+            break;
+    }
+    if (has_slot)
+        kvstoreResetDictIterator(&it.kvs_di);
+    else
+        kvstoreIteratorReset(&it.kvs_it);
+    setDeferredArrayLen(c,replylen,numkeys);
+}
+
+/* Data used by the dict scan callback. */
+typedef struct {
+    list *keys;   /* elements that collect from dict */
+    robj *o;      /* o must be a hash/set/zset object, NULL means current db */
+    long long type; /* the particular type when scan the db */
+    sds pattern;  /* pattern string, NULL means no pattern */
+    long sampled; /* cumulative number of keys sampled */
+    int no_values; /* set to 1 means to return keys only */
+    sds typename; /* typename string, NULL means no type filter */
+    redisDb *db;  /* database reference for expiration checks */
+} scanData;
+
+/* Helper function to compare key type in scan commands */
+int objectTypeCompare(robj *o, long long target) {
+    if (o->type != OBJ_MODULE) {
+        if (o->type != target) 
+            return 0;
+        else 
+            return 1;
+    }
+    /* module type compare */
+    moduleType *type = ((moduleValue *)o->ptr)->type;
+    long long mt = (long long)REDISMODULE_TYPE_SIGN(type->entity.id);
+    if (target != -mt)
+        return 0;
+    else 
+        return 1;
+}
+/* This callback is used by scanGenericCommand in order to collect elements
+ * returned by the dictionary iterator into a list. */
+void scanCallback(void *privdata, const dictEntry *de, dictEntryLink plink) {
+    UNUSED(plink);
+    Entry *hashEntry = NULL;
+    scanData *data = (scanData *)privdata;
+    list *keys = data->keys;
+    robj *o = data->o;
+    sds val = NULL;
+    void *key = NULL;  /* if OBJ_HASH then key is of type `hfield`. Otherwise, `sds` */
+    void *keyStr;
+    data->sampled++;
+
+    /* o and typename can not have values at the same time. */
+    serverAssert(!((data->type != LLONG_MAX) && o));
+
+    kvobj *kv = NULL;
+    zskiplistNode *znode = NULL;
+    if (!o) { /* If scanning keyspace */
+        kv = dictGetKV(de);
+        keyStr = kvobjGetKey(kv);
+    } else if (o->type == OBJ_HASH) {
+        hashEntry = dictGetKey(de);
+        keyStr = entryGetField(hashEntry);
+    } else if (o->type == OBJ_ZSET) {
+        znode = dictGetKey(de);
+        keyStr = zslGetNodeElement(znode);
+    } else {
+        keyStr = dictGetKey(de);
+    }
+    
+    /* Filter element if it does not match the pattern. */
+    if (data->pattern) {
+        if (!stringmatchlen(data->pattern, sdslen(data->pattern), keyStr, sdslen(keyStr), 0)) {
+            return;
+        }
+    }
+    
+    if (!o) {
+        /* Expiration check first - only for database keyspace scanning.
+         * Use kv obj to avoid robj creation. */
+        if (expireIfNeeded(data->db, NULL, kv, 0) != KEY_VALID)
+            return;
+
+        /* Type filtering - only for database keyspace scanning */
+        if (data->typename) {
+            /* For unknown types (LLONG_MAX), skip all keys */
+            if (data->type == LLONG_MAX)
+                return;
+            /* For known types, skip keys that don't match */
+            if (!objectTypeCompare(kv, data->type))
+                return;
+        }
+    }
+
+    if (o == NULL) {
+        key = keyStr;
+    } else if (o->type == OBJ_SET) {
+        key = keyStr;
+    } else if (o->type == OBJ_HASH) {
+        key = keyStr;
+        val = entryGetValue(hashEntry);
+
+        /* If field is expired, then ignore */
+        if (entryIsExpired(hashEntry))
+            return;
+
+    } else if (o->type == OBJ_ZSET) {
+        char buf[MAX_LONG_DOUBLE_CHARS];
+        int len = ld2string(buf, sizeof(buf), znode->score, LD_STR_AUTO);
+        key = sdsdup(keyStr);
+        val = sdsnewlen(buf, len);
+    } else {
+        serverPanic("Type not handled in SCAN callback.");
+    }
+
+    listAddNodeTail(keys, key);
+    if (val && !data->no_values) listAddNodeTail(keys, val);
+}
+
+/* Try to parse a SCAN cursor stored at object 'o':
+ * if the cursor is valid, store it as unsigned integer into *cursor and
+ * returns C_OK. Otherwise return C_ERR and send an error to the
+ * client. */
+int parseScanCursorOrReply(client *c, robj *o, unsigned long long *cursor) {
+    if (!string2ull(o->ptr, cursor)) {
+        addReplyError(c, "invalid cursor");
+        return C_ERR;
+    }
+    return C_OK;
+}
+
+char *obj_type_name[OBJ_TYPE_MAX] = {
+    "string", 
+    "list", 
+    "set", 
+    "zset", 
+    "hash", 
+    NULL, /* module type is special */
+    "stream"
+};
+
+/* Helper function to get type from a string in scan commands */
+long long getObjectTypeByName(char *name) {
+
+    for (long long i = 0; i < OBJ_TYPE_MAX; i++) {
+        if (obj_type_name[i] && !strcasecmp(name, obj_type_name[i])) {
+            return i;
+        }
+    }
+
+    moduleType *mt = moduleTypeLookupModuleByNameIgnoreCase(name);
+    if (mt != NULL) return -(REDISMODULE_TYPE_SIGN(mt->entity.id));
+
+    return LLONG_MAX;
+}
+
+char *getObjectTypeName(robj *o) {
+    if (o == NULL) {
+        return "none";
+    }
+
+    serverAssert(o->type >= 0 && o->type < OBJ_TYPE_MAX);
+
+    if (o->type == OBJ_MODULE) {
+        moduleValue *mv = o->ptr;
+        return mv->type->entity.name;
+    } else {
+        return obj_type_name[o->type];
+    }
+}
+
+static int scanShouldSkipDict(dict *d, int didx) {
+    UNUSED(d);
+    return accessKeysShouldSkipDictIndex(didx);
+}
+
+/* This command implements SCAN, HSCAN and SSCAN commands.
+ * If object 'o' is passed, then it must be a Hash, Set or Zset object, otherwise
+ * if 'o' is NULL the command will operate on the dictionary associated with
+ * the current database.
+ *
+ * When 'o' is not NULL the function assumes that the first argument in
+ * the client arguments vector is a key so it skips it before iterating
+ * in order to parse options.
+ *
+ * In the case of a Hash object the function returns both the field and value
+ * of every element on the Hash. */
+void scanGenericCommand(client *c, robj *o, unsigned long long cursor) {
+    int i, j;
+    listNode *node;
+    long count = 10;
+    sds pat = NULL;
+    sds typename = NULL;
+    long long type = LLONG_MAX;
+    int patlen = 0, use_pattern = 0, no_values = 0;
+    dict *ht;
+
+    /* Object must be NULL (to iterate keys names), or the type of the object
+     * must be Set, Sorted Set, or Hash. */
+    serverAssert(o == NULL || o->type == OBJ_SET || o->type == OBJ_HASH ||
+                o->type == OBJ_ZSET);
+
+    /* Set i to the first option argument. The previous one is the cursor. */
+    i = (o == NULL) ? 2 : 3; /* Skip the key argument if needed. */
+
+    /* Step 1: Parse options. */
+    while (i < c->argc) {
+        j = c->argc - i;
+        if (!strcasecmp(c->argv[i]->ptr, "count") && j >= 2) {
+            if (getLongFromObjectOrReply(c, c->argv[i+1], &count, NULL)
+                != C_OK)
+            {
+                return;
+            }
+
+            if (count < 1) {
+                addReplyErrorObject(c,shared.syntaxerr);
+                return;
+            }
+
+            i += 2;
+        } else if (!strcasecmp(c->argv[i]->ptr, "match") && j >= 2) {
+            pat = c->argv[i+1]->ptr;
+            patlen = sdslen(pat);
+
+            /* The pattern always matches if it is exactly "*", so it is
+             * equivalent to disabling it. */
+            use_pattern = !(patlen == 1 && pat[0] == '*');
+
+            i += 2;
+        } else if (!strcasecmp(c->argv[i]->ptr, "type") && o == NULL && j >= 2) {
+            /* SCAN for a particular type only applies to the db dict */
+            typename = c->argv[i+1]->ptr;
+            type = getObjectTypeByName(typename);
+            if (type == LLONG_MAX) {
+                /* TODO: uncomment in redis 8.0
+                addReplyErrorFormat(c, "unknown type name '%s'", typename);
+                return; */
+            }
+            i+= 2;
+        } else if (!strcasecmp(c->argv[i]->ptr, "novalues")) {
+            if (!o || o->type != OBJ_HASH) {
+                addReplyError(c, "NOVALUES option can only be used in HSCAN");
+                return;
+            }
+            no_values = 1;
+            i++;
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return;
+        }
+    }
+
+    /* Step 2: Iterate the collection.
+     *
+     * Note that if the object is encoded with a listpack, intset, or any other
+     * representation that is not a hash table, we are sure that it is also
+     * composed of a small number of elements. So to avoid taking state we
+     * just return everything inside the object in a single call, setting the
+     * cursor to zero to signal the end of the iteration. */
+
+    /* Handle the case of a hash table. */
+    ht = NULL;
+    if (o == NULL) {
+        ht = NULL;
+    } else if (o->type == OBJ_SET && o->encoding == OBJ_ENCODING_HT) {
+        ht = o->ptr;
+    } else if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_HT) {
+        ht = o->ptr;
+    } else if (o->type == OBJ_ZSET && o->encoding == OBJ_ENCODING_SKIPLIST) {
+        zset *zs = o->ptr;
+        ht = zs->dict;
+    }
+
+    list *keys = listCreate();
+    /* Set a free callback for the contents of the collected keys list.
+     * For the main keyspace dict, and when we scan a key that's dict encoded
+     * (we have 'ht'), we don't need to define free method because the strings
+     * in the list are just a shallow copy from the pointer in the dictEntry.
+     * When scanning a key with other encodings (e.g. listpack), we need to
+     * free the temporary strings we add to that list.
+     * The exception to the above is ZSET, where we do allocate temporary
+     * strings even when scanning a dict. */
+    if (o && (!ht || o->type == OBJ_ZSET)) {
+        listSetFreeMethod(keys, sdsfreegeneric);
+    }
+
+    /* For main dictionary scan or data structure using hashtable. */
+    if (!o || ht) {
+        /* We set the max number of iterations to ten times the specified
+         * COUNT, so if the hash table is in a pathological state (very
+         * sparsely populated) we avoid to block too much time at the cost
+         * of returning no or very few elements. */
+        long maxiterations = count*10;
+
+        /* We pass scanData which have three pointers to the callback:
+         * 1. data.keys: the list to which it will add new elements;
+         * 2. data.o: the object containing the dictionary so that
+         * it is possible to fetch more data in a type-dependent way;
+         * 3. data.type: the specified type scan in the db, LLONG_MAX means
+         * type matching is no needed;
+         * 4. data.pattern: the pattern string;
+         * 5. data.sampled: the maxiteration limit is there in case we're
+         * working on an empty dict, one with a lot of empty buckets, and
+         * for the buckets are not empty, we need to limit the spampled number
+         * to prevent a long hang time caused by filtering too many keys;
+         * 6. data.no_values: to control whether values will be returned or
+         * only keys are returned. */
+        scanData data = {
+            .keys = keys,
+            .o = o,
+            .type = type,
+            .pattern = use_pattern ? pat : NULL,
+            .sampled = 0,
+            .no_values = no_values,
+            .typename = typename,
+            .db = c->db,
+        };
+
+        /* A pattern may restrict all matching keys to one cluster slot. */
+        int onlydidx = -1;
+        if (o == NULL && use_pattern && server.cluster_enabled) {
+            onlydidx = patternHashSlot(pat, patlen);
+        }
+        do {
+            /* In cluster mode there is a separate dictionary for each slot.
+             * If cursor is empty, we should try exploring next non-empty slot. */
+            if (o == NULL) {
+                cursor = kvstoreScan(c->db->keys, cursor, onlydidx, scanCallback, scanShouldSkipDict, &data);
+            } else {
+                cursor = dictScan(ht, cursor, scanCallback, &data);
+            }
+        } while (cursor && maxiterations-- && data.sampled < count);
+    } else if (o->type == OBJ_SET) {
+        unsigned long array_reply_len = 0;
+        void *replylen = NULL;
+        listRelease(keys);
+        char *str;
+        char buf[LONG_STR_SIZE];
+        size_t len;
+        int64_t llele;
+        /* Reply to the client. */
+        addReplyArrayLen(c, 2);
+        /* Cursor is always 0 given we iterate over all set */
+        addReplyBulkLongLong(c,0);
+        /* If there is no pattern the length is the entire set size, otherwise we defer the reply size */
+        if (use_pattern)
+            replylen = addReplyDeferredLen(c);
+        else {
+            array_reply_len = setTypeSize(o);
+            addReplyArrayLen(c, array_reply_len);
+        }
+
+        setTypeIterator si;
+        unsigned long cur_length = 0;
+        setTypeInitIterator(&si, o);
+        while (setTypeNext(&si, &str, &len, &llele) != -1) {
+            if (str == NULL) {
+                len = ll2string(buf, sizeof(buf), llele);
+            }
+            char *key = str ? str : buf;
+            if (use_pattern && !stringmatchlen(pat, patlen, key, len, 0)) {
+                continue;
+            }
+            addReplyBulkCBuffer(c, key, len);
+            cur_length++;
+        }
+        setTypeResetIterator(&si);
+        if (use_pattern)
+            setDeferredArrayLen(c,replylen,cur_length);
+        else
+            serverAssert(cur_length == array_reply_len); /* fail on corrupt data */
+        return;
+    } else if ((o->type == OBJ_HASH || o->type == OBJ_ZSET) &&
+               o->encoding == OBJ_ENCODING_LISTPACK)
+    {
+        unsigned char *p = lpFirst(o->ptr);
+        unsigned char *str;
+        int64_t len;
+        unsigned long array_reply_len = 0;
+        unsigned char intbuf[LP_INTBUF_SIZE];
+        void *replylen = NULL;
+        listRelease(keys);
+
+        /* Reply to the client. */
+        addReplyArrayLen(c, 2);
+        /* Cursor is always 0 given we iterate over all set */
+        addReplyBulkLongLong(c,0);
+        /* If there is no pattern the length is the entire set size, otherwise we defer the reply size */
+        if (use_pattern)
+            replylen = addReplyDeferredLen(c);
+        else {
+            array_reply_len = o->type == OBJ_HASH ? hashTypeLength(o, 0) : zsetLength(o);
+            if (!no_values) {
+                array_reply_len *= 2;
+            }
+            addReplyArrayLen(c, array_reply_len);
+        }
+        unsigned long cur_length = 0;
+        while(p) {
+            str = lpGet(p, &len, intbuf);
+            /* point to the value */
+            p = lpNext(o->ptr, p);
+            if (use_pattern && !stringmatchlen(pat, patlen, (char *)str, len, 0)) {
+                /* jump to the next key/val pair */
+                p = lpNext(o->ptr, p);
+                continue;
+            }
+            /* add key object */
+            addReplyBulkCBuffer(c, str, len);
+            cur_length++;
+            /* add value object */
+            if (!no_values) {
+                str = lpGet(p, &len, intbuf);
+                addReplyBulkCBuffer(c, str, len);
+                cur_length++;
+            }
+            p = lpNext(o->ptr, p);
+        }
+        if (use_pattern)
+            setDeferredArrayLen(c,replylen,cur_length);
+        else
+            serverAssert(cur_length == array_reply_len); /* fail on corrupt data */
+        return;
+    } else if (o->type == OBJ_HASH && o->encoding == OBJ_ENCODING_LISTPACK_EX) {
+        int64_t len;
+        long long expire_at;
+        unsigned char *lp = hashTypeListpackGetLp(o);
+        unsigned char *p = lpFirst(lp);
+        unsigned char *str, *val;
+        unsigned char intbuf[LP_INTBUF_SIZE];
+        void *replylen = NULL;
+
+        listRelease(keys);
+        /* Reply to the client. */
+        addReplyArrayLen(c, 2);
+        /* Cursor is always 0 given we iterate over all set */
+        addReplyBulkLongLong(c,0);
+        /* In the case of OBJ_ENCODING_LISTPACK_EX we always defer the reply size given some fields might be expired */
+        replylen = addReplyDeferredLen(c);
+        unsigned long cur_length = 0;
+
+        while (p) {
+            str = lpGet(p, &len, intbuf);
+            p = lpNext(lp, p);
+            val = p; /* Keep pointer to value */
+
+            p = lpNext(lp, p);
+            serverAssert(p && lpGetIntegerValue(p, &expire_at));
+
+            if (hashTypeIsExpired(o, expire_at) ||
+               (use_pattern && !stringmatchlen(pat, patlen, (char *)str, len, 0)))
+            {
+                /* jump to the next key/val pair */
+                p = lpNext(lp, p);
+                continue;
+            }
+
+            /* add key object */
+            addReplyBulkCBuffer(c, str, len);
+            cur_length++;
+            /* add value object */
+            if (!no_values) {
+                str = lpGet(val, &len, intbuf);
+                addReplyBulkCBuffer(c, str, len);
+                cur_length++;
+            }
+            p = lpNext(lp, p);
+        }
+        setDeferredArrayLen(c,replylen,cur_length);
+        return;
+    } else {
+        serverPanic("Not handled encoding in SCAN.");
+    }
+
+    /* Step 3: Reply to the client. */
+    addReplyArrayLen(c, 2);
+    addReplyBulkLongLong(c,cursor);
+
+    addReplyArrayLen(c, listLength(keys));
+    while ((node = listFirst(keys)) != NULL) {
+        void *key = listNodeValue(node);
+        addReplyBulkCBuffer(c, key, sdslen(key));
+        listDelNode(keys, node);
+    }
+
+    listRelease(keys);
+}
+
+/* The SCAN command completely relies on scanGenericCommand. */
+void scanCommand(client *c) {
+    unsigned long long cursor;
+    if (parseScanCursorOrReply(c,c->argv[1],&cursor) == C_ERR) return;
+    scanGenericCommand(c,NULL,cursor);
+}
+
+void dbsizeCommand(client *c) {
+    addReplyLongLong(c,dbSize(c->db));
+}
+
+void lastsaveCommand(client *c) {
+    addReplyLongLong(c,server.lastsave);
+}
+
+void typeCommand(client *c) {
+    kvobj *kv = lookupKeyReadWithFlags(c->db,c->argv[1],LOOKUP_NOTOUCH);
+    addReplyStatus(c, getObjectTypeName(kv));
+}
+
+void shutdownCommand(client *c) {
+    int flags = SHUTDOWN_NOFLAGS;
+    int abort = 0;
+    for (int i = 1; i < c->argc; i++) {
+        if (!strcasecmp(c->argv[i]->ptr,"nosave")) {
+            flags |= SHUTDOWN_NOSAVE;
+        } else if (!strcasecmp(c->argv[i]->ptr,"save")) {
+            flags |= SHUTDOWN_SAVE;
+        } else if (!strcasecmp(c->argv[i]->ptr, "now")) {
+            flags |= SHUTDOWN_NOW;
+        } else if (!strcasecmp(c->argv[i]->ptr, "force")) {
+            flags |= SHUTDOWN_FORCE;
+        } else if (!strcasecmp(c->argv[i]->ptr, "abort")) {
+            abort = 1;
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return;
+        }
+    }
+    if ((abort && flags != SHUTDOWN_NOFLAGS) ||
+        (flags & SHUTDOWN_NOSAVE && flags & SHUTDOWN_SAVE))
+    {
+        /* Illegal combo. */
+        addReplyErrorObject(c,shared.syntaxerr);
+        return;
+    }
+
+    if (abort) {
+        if (abortShutdown() == C_OK)
+            addReply(c, shared.ok);
+        else
+            addReplyError(c, "No shutdown in progress.");
+        return;
+    }
+
+    if (!(flags & SHUTDOWN_NOW) && c->flags & CLIENT_DENY_BLOCKING) {
+        addReplyError(c, "SHUTDOWN without NOW or ABORT isn't allowed for DENY BLOCKING client");
+        return;
+    }
+
+    if (!(flags & SHUTDOWN_NOSAVE) && isInsideYieldingLongCommand()) {
+        /* Script timed out. Shutdown allowed only with the NOSAVE flag. See
+         * also processCommand where these errors are returned. */
+        if (server.busy_module_yield_flags && server.busy_module_yield_reply) {
+            addReplyErrorFormat(c, "-BUSY %s", server.busy_module_yield_reply);
+        } else if (server.busy_module_yield_flags) {
+            addReplyErrorObject(c, shared.slowmoduleerr);
+        } else if (scriptIsEval()) {
+            addReplyErrorObject(c, shared.slowevalerr);
+        } else {
+            addReplyErrorObject(c, shared.slowscripterr);
+        }
+        return;
+    }
+
+    blockClientShutdown(c);
+    if (prepareForShutdown(flags) == C_OK) exit(0);
+    /* If we're here, then shutdown is ongoing (the client is still blocked) or
+     * failed (the client has received an error). */
+}
+
+void renameGenericCommand(client *c, int nx) {
+    kvobj *o;
+    int samekey = 0;
+    uint64_t minHashExpireTime = EB_EXPIRE_TIME_INVALID;
+
+    /* When source and dest key is the same, no operation is performed,
+     * if the key exists, however we still return an error on unexisting key. */
+    if (sdscmp(c->argv[1]->ptr,c->argv[2]->ptr) == 0) samekey = 1;
+
+    if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.nokeyerr)) == NULL)
+        return;
+
+    if (samekey) {
+        addReply(c,nx ? shared.czero : shared.ok);
+        return;
+    }
+
+    incrRefCount(o);
+    kvobj *destval = lookupKeyWrite(c->db,c->argv[2]);
+    int overwritten = 0;
+    int desttype = -1;
+    if (destval != NULL) {
+        if (nx) {
+            decrRefCount(o);
+            addReply(c,shared.czero);
+            return;
+        }
+
+        /* Overwrite: delete the old key before creating the new one
+         * with the same name. */
+        desttype = destval->type;
+        dbDelete(c->db,c->argv[2]);
+        overwritten = 1;
+    }
+
+    /* If hash with expiration on fields then remove it from global HFE DS and
+     * keep next expiration time. Otherwise, dbDelete() will remove it from the
+     * global HFE DS and we will lose the expiration time. */
+    int srctype = o->type;
+    if (srctype == OBJ_HASH)
+        minHashExpireTime = estoreRemove(c->db->subexpires, getKeySlot(c->argv[1]->ptr), o);
+
+    /* Prepare metadata for the renamed key */
+    KeyMetaSpec keymeta;
+    keyMetaSpecInit(&keymeta);
+    if (o->metabits) keyMetaOnRename(c->db, o, c->argv[1], c->argv[2], &keymeta);
+
+    dbDelete(c->db,c->argv[1]);
+    
+    dbAddInternal(c->db, c->argv[2], &o, NULL, &keymeta);
+
+    /* If hash with HFEs, register in DB subexpires */
+    if (minHashExpireTime != EB_EXPIRE_TIME_INVALID)
+        estoreAdd(c->db->subexpires, getKeySlot(c->argv[2]->ptr), o, minHashExpireTime);
+
+    keyModified(c,c->db,c->argv[1],NULL,1);
+    keyModified(c,c->db,c->argv[2],NULL,1); /* LRM already updated by dbAddInternal */
+    notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_from",
+        c->argv[1],c->db->id);
+    notifyKeyspaceEvent(NOTIFY_GENERIC,"rename_to",
+        c->argv[2],c->db->id);
+    if (overwritten) {
+        notifyKeyspaceEvent(NOTIFY_OVERWRITTEN, "overwritten", c->argv[2], c->db->id);
+        if (desttype != srctype)
+            notifyKeyspaceEvent(NOTIFY_TYPE_CHANGED, "type_changed", c->argv[2], c->db->id);
+    }
+    server.dirty++;
+    addReply(c,nx ? shared.cone : shared.ok);
+}
+
+void renameCommand(client *c) {
+    renameGenericCommand(c,0);
+}
+
+void renamenxCommand(client *c) {
+    renameGenericCommand(c,1);
+}
+
+void moveCommand(client *c) {
+    redisDb *src, *dst;
+    int srcid, dbid;
+    uint64_t hashExpireTime = EB_EXPIRE_TIME_INVALID;
+
+    if (server.cluster_enabled) {
+        addReplyError(c,"MOVE is not allowed in cluster mode");
+        return;
+    }
+
+    /* Obtain source and target DB pointers */
+    src = c->db;
+    srcid = c->db->id;
+
+    if (getIntFromObjectOrReply(c, c->argv[2], &dbid, NULL) != C_OK)
+        return;
+
+    if (selectDb(c,dbid) == C_ERR) {
+        addReplyError(c,"DB index is out of range");
+        return;
+    }
+    dst = c->db;
+    selectDb(c,srcid); /* Back to the source DB */
+
+    /* If the user is moving using as target the same
+     * DB as the source DB it is probably an error. */
+    if (src == dst) {
+        addReplyErrorObject(c,shared.sameobjecterr);
+        return;
+    }
+
+    /* Record incompatible operations in cluster mode */
+    server.stat_cluster_incompatible_ops++;
+
+    /* Check if the element exists and get a reference */
+    kvobj *kv = lookupKeyWrite(c->db,c->argv[1]);
+    if (!kv) {
+        addReply(c,shared.czero);
+        return;
+    }
+
+    /* Return zero if the key already exists in the target DB */
+    dictEntryLink dstBucket;
+    if (lookupKey(dst, c->argv[1], LOOKUP_WRITE, &dstBucket) != NULL)  {
+        addReply(c,shared.czero);
+        return;
+    }
+
+    int slot = getKeySlot(c->argv[1]->ptr);
+
+    /* If hash with expiration on fields, remove it from DB subexpires and keep
+     * aside registered expiration time. Must be before removal of the
+     * object since it embeds ExpireMeta that is used by subexpires */
+    if (kv->type == OBJ_HASH)
+        hashExpireTime = estoreRemove(src->subexpires, slot, kv);
+    
+    /* Move a side metadata before dbDelete() */
+    KeyMetaSpec keymeta;
+    keyMetaSpecInit(&keymeta);
+    keyMetaOnMove(kv, c->argv[1], srcid, dbid, &keymeta);
+
+    incrRefCount(kv);            /* ref counter = 1->2 */
+    dbDelete(src,c->argv[1]);    /* ref counter = 2->1 */
+
+    dbAddInternal(dst, c->argv[1], &kv, &dstBucket, &keymeta);
+
+    /* If object of type hash with expiration on fields. Taken care to add the
+     * hash to subexpires of `dst` only after dbDelete(). */
+    if (hashExpireTime != EB_EXPIRE_TIME_INVALID)
+        estoreAdd(dst->subexpires, slot, kv, hashExpireTime);
+
+    keyModified(c,src,c->argv[1],NULL,1);
+    keyModified(c,dst,c->argv[1],NULL,1); /* LRM already updated by dbAddInternal */
+    notifyKeyspaceEvent(NOTIFY_GENERIC,
+                "move_from",c->argv[1],src->id);
+    notifyKeyspaceEvent(NOTIFY_GENERIC,
+                "move_to",c->argv[1],dst->id);
+
+    server.dirty++;
+    addReply(c,shared.cone);
+}
+
+void copyCommand(client *c) {
+    kvobj *o;
+    redisDb *src, *dst;
+    int srcid, dbid;
+    int j, replace = 0, delete = 0;
+
+    /* Obtain source and target DB pointers 
+     * Default target DB is the same as the source DB 
+     * Parse the REPLACE option and targetDB option. */
+    src = c->db;
+    dst = c->db;
+    srcid = c->db->id;
+    dbid = c->db->id;
+    for (j = 3; j < c->argc; j++) {
+        int additional = c->argc - j - 1;
+        if (!strcasecmp(c->argv[j]->ptr,"replace")) {
+            replace = 1;
+        } else if (!strcasecmp(c->argv[j]->ptr, "db") && additional >= 1) {
+            if (getIntFromObjectOrReply(c, c->argv[j+1], &dbid, NULL) != C_OK)
+                return;
+
+            if (selectDb(c, dbid) == C_ERR) {
+                addReplyError(c,"DB index is out of range");
+                return;
+            }
+            dst = c->db;
+            selectDb(c,srcid); /* Back to the source DB */
+            j++; /* Consume additional arg. */
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return;
+        }
+    }
+
+    if ((server.cluster_enabled == 1) && (srcid != 0 || dbid != 0)) {
+        addReplyError(c,"Copying to another database is not allowed in cluster mode");
+        return;
+    }
+
+    /* If the user select the same DB as
+     * the source DB and using newkey as the same key
+     * it is probably an error. */
+    robj *key = c->argv[1];
+    robj *newkey = c->argv[2];
+    if (src == dst && (sdscmp(key->ptr, newkey->ptr) == 0)) {
+        addReplyErrorObject(c,shared.sameobjecterr);
+        return;
+    }
+
+    if (srcid != 0 || dbid != 0) {
+        server.stat_cluster_incompatible_ops++;
+    }
+
+    /* Check if the element exists and get a reference */
+    o = lookupKeyRead(c->db, key);
+    if (!o) {
+        addReply(c,shared.czero);
+        return;
+    }
+
+    /* Return zero if the key already exists in the target DB. 
+     * If REPLACE option is selected, delete newkey from targetDB. */
+    kvobj *destval = lookupKeyWrite(dst,newkey);
+    if (destval != NULL) {
+        if (replace) {
+            delete = 1;
+        } else {
+            addReply(c,shared.czero);
+            return;
+        }
+    }
+    int destoldtype = destval ? destval->type : -1;
+    int destnewtype = o->type;
+
+    /* Duplicate object according to object's type. */
+    robj *newobj;
+    uint64_t minHashExpire = EB_EXPIRE_TIME_INVALID; /* HFE feature */
+    switch(o->type) {
+        case OBJ_STRING: newobj = dupStringObject(o); break;
+        case OBJ_LIST: newobj = listTypeDup(o); break;
+        case OBJ_SET: newobj = setTypeDup(o); break;
+        case OBJ_ZSET: newobj = zsetDup(o); break;
+        case OBJ_HASH: newobj = hashTypeDup(o, &minHashExpire); break;
+        case OBJ_STREAM: newobj = streamDup(o); break;
+        case OBJ_MODULE:
+            newobj = moduleTypeDupOrReply(c, key, newkey, dst->id, o);
+            if (!newobj) return;
+            break;
+        default:
+            addReplyError(c, "unknown type object");
+            return;
+    }
+
+    if (delete) {
+        dbDelete(dst,newkey);
+    }
+
+    /* Prepare metadata for the new key */
+    KeyMetaSpec keymeta;
+    keyMetaSpecInit(&keymeta);
+    if (o->metabits) keyMetaOnCopy(o, key, newkey, c->db->id, dst->id, &keymeta);
+
+    kvobj *kvCopy = dbAddInternal(dst, newkey, &newobj, NULL, &keymeta);
+
+    /* If minExpiredField was set, then the object is hash with expiration
+     * on fields and need to register it in global HFE DS */
+    if (minHashExpire != EB_EXPIRE_TIME_INVALID)
+        estoreAdd(dst->subexpires, getKeySlot(newkey->ptr), kvCopy, minHashExpire);
+
+    /* OK! key copied. Signal modification (LRM already updated by dbAddInternal) */
+    keyModified(c,dst,c->argv[2],NULL,1);
+    notifyKeyspaceEvent(NOTIFY_GENERIC,"copy_to",c->argv[2],dst->id);
+
+    /* `delete` implies the destination key was overwritten */
+    if (delete) {
+        notifyKeyspaceEvent(NOTIFY_OVERWRITTEN, "overwritten", c->argv[2], dst->id);
+        if (destoldtype != destnewtype)
+            notifyKeyspaceEvent(NOTIFY_TYPE_CHANGED, "type_changed", c->argv[2], dst->id);
+    }
+
+    server.dirty++;
+    addReply(c,shared.cone);
+}
+
+/* Helper function for dbSwapDatabases(): scans the list of keys that have
+ * one or more blocked clients for B[LR]POP or other blocking commands
+ * and signal the keys as ready if they are of the right type. See the comment
+ * where the function is used for more info. */
+void scanDatabaseForReadyKeys(redisDb *db) {
+    dictEntry *de;
+    dictIterator di;
+    dictInitSafeIterator(&di, db->blocking_keys);
+    while((de = dictNext(&di)) != NULL) {
+        robj *key = dictGetKey(de);
+        kvobj *kv = dbFind(db, key->ptr);
+        if (kv)
+            signalKeyAsReady(db, key, kv->type);
+    }
+    dictResetIterator(&di);
+}
+
+/* Since we are unblocking XREADGROUP clients in the event the key was
+ * deleted/overwritten we must do the same in case the database was
+ * flushed/swapped. If 'slots' is not NULL, only keys in the specified slot
+ * range are considered. */
+void scanDatabaseForDeletedKeys(redisDb *emptied, redisDb *replaced_with, slotRangeArray *slots) {
+    dictEntry *de;
+    dictIterator di;
+
+    dictInitSafeIterator(&di, emptied->blocking_keys);
+    while((de = dictNext(&di)) != NULL) {
+        robj *key = dictGetKey(de);
+        /* Check if key belongs to the slot range. */
+        if (slots && !slotRangeArrayContains(slots, keyHashSlot(key->ptr, sdslen(key->ptr))))
+            continue;
+        int existed = 0, exists = 0;
+        int original_type = -1, curr_type = -1;
+
+        kvobj *kv = dbFind(emptied, key->ptr);
+        if (kv) {
+            original_type = kv->type;
+            existed = 1;
+        }
+
+        if (replaced_with) {
+            kv = dbFind(replaced_with, key->ptr);
+            if (kv) {
+                curr_type = kv->type;
+                exists = 1;
+            }
+        }
+        /* We want to try to unblock any client using a blocking XREADGROUP */
+        if ((existed && !exists) || original_type != curr_type)
+            signalDeletedKeyAsReady(emptied, key, original_type);
+    }
+    dictResetIterator(&di);
+}
+
+/* Swap two databases at runtime so that all clients will magically see
+ * the new database even if already connected. Note that the client
+ * structure c->db points to a given DB, so we need to be smarter and
+ * swap the underlying referenced structures, otherwise we would need
+ * to fix all the references to the Redis DB structure.
+ *
+ * Returns C_ERR if at least one of the DB ids are out of range, otherwise
+ * C_OK is returned. */
+int dbSwapDatabases(int id1, int id2) {
+    if (id1 < 0 || id1 >= server.dbnum ||
+        id2 < 0 || id2 >= server.dbnum) return C_ERR;
+    if (id1 == id2) return C_OK;
+    redisDb aux = server.db[id1];
+    redisDb *db1 = &server.db[id1], *db2 = &server.db[id2];
+
+    /* Swapdb should make transaction fail if there is any
+     * client watching keys */
+    touchAllWatchedKeysInDb(db1, db2, NULL);
+    touchAllWatchedKeysInDb(db2, db1, NULL);
+
+    /* Try to unblock any XREADGROUP clients if the key no longer exists. */
+    scanDatabaseForDeletedKeys(db1, db2, NULL);
+    scanDatabaseForDeletedKeys(db2, db1, NULL);
+
+    /* Swap hash tables. Note that we don't swap blocking_keys,
+     * ready_keys and watched_keys, since we want clients to
+     * remain in the same DB they were. */
+    db1->keys = db2->keys;
+    db1->expires = db2->expires;
+    db1->subexpires = db2->subexpires;
+    db1->avg_ttl = db2->avg_ttl;
+    db1->expires_cursor = db2->expires_cursor;
+
+    db2->keys = aux.keys;
+    db2->expires = aux.expires;
+    db2->subexpires = aux.subexpires;
+    db2->avg_ttl = aux.avg_ttl;
+    db2->expires_cursor = aux.expires_cursor;
+
+    /* Now we need to handle clients blocked on lists: as an effect
+     * of swapping the two DBs, a client that was waiting for list
+     * X in a given DB, may now actually be unblocked if X happens
+     * to exist in the new version of the DB, after the swap.
+     *
+     * However normally we only do this check for efficiency reasons
+     * in dbAdd() when a list is created. So here we need to rescan
+     * the list of clients blocked on lists and signal lists as ready
+     * if needed. */
+    scanDatabaseForReadyKeys(db1);
+    scanDatabaseForReadyKeys(db2);
+    return C_OK;
+}
+
+/* Logically, this discards (flushes) the old main database, and apply the newly loaded
+ * database (temp) as the main (active) database, the actual freeing of old database
+ * (which will now be placed in the temp one) is done later. */
+void swapMainDbWithTempDb(redisDb *tempDb) {
+    for (int i=0; i<server.dbnum; i++) {
+        redisDb aux = server.db[i];
+        redisDb *activedb = &server.db[i], *newdb = &tempDb[i];
+
+        /* Swapping databases should make transaction fail if there is any
+         * client watching keys. */
+        touchAllWatchedKeysInDb(activedb, newdb, NULL);
+
+        /* Try to unblock any XREADGROUP clients if the key no longer exists. */
+        scanDatabaseForDeletedKeys(activedb, newdb, NULL);
+
+        /* Swap hash tables. Note that we don't swap blocking_keys,
+         * ready_keys and watched_keys, since clients 
+         * remain in the same DB they were. */
+        activedb->keys = newdb->keys;
+        activedb->expires = newdb->expires;
+        activedb->subexpires = newdb->subexpires;
+        activedb->avg_ttl = newdb->avg_ttl;
+        activedb->expires_cursor = newdb->expires_cursor;
+
+        newdb->keys = aux.keys;
+        newdb->expires = aux.expires;
+        newdb->subexpires = aux.subexpires;
+        newdb->avg_ttl = aux.avg_ttl;
+        newdb->expires_cursor = aux.expires_cursor;
+
+        /* Now we need to handle clients blocked on lists: as an effect
+         * of swapping the two DBs, a client that was waiting for list
+         * X in a given DB, may now actually be unblocked if X happens
+         * to exist in the new version of the DB, after the swap.
+         *
+         * However normally we only do this check for efficiency reasons
+         * in dbAdd() when a list is created. So here we need to rescan
+         * the list of clients blocked on lists and signal lists as ready
+         * if needed. */
+        scanDatabaseForReadyKeys(activedb);
+    }
+
+    trackingInvalidateKeysOnFlush(1);
+    flushSlaveKeysWithExpireList();
+}
+
+/* SWAPDB db1 db2 */
+void swapdbCommand(client *c) {
+    int id1, id2;
+
+    /* Not allowed in cluster mode: we have just DB 0 there. */
+    if (server.cluster_enabled) {
+        addReplyError(c,"SWAPDB is not allowed in cluster mode");
+        return;
+    }
+
+    /* Get the two DBs indexes. */
+    if (getIntFromObjectOrReply(c, c->argv[1], &id1,
+        "invalid first DB index") != C_OK)
+        return;
+
+    if (getIntFromObjectOrReply(c, c->argv[2], &id2,
+        "invalid second DB index") != C_OK)
+        return;
+
+    /* Swap... */
+    if (dbSwapDatabases(id1,id2) == C_ERR) {
+        addReplyError(c,"DB index is out of range");
+        return;
+    } else {
+        RedisModuleSwapDbInfo si = {REDISMODULE_SWAPDBINFO_VERSION,id1,id2};
+        moduleFireServerEvent(REDISMODULE_EVENT_SWAPDB,0,&si);
+        server.dirty++;
+        server.stat_cluster_incompatible_ops++;
+        addReply(c,shared.ok);
+    }
+}
+
+/*-----------------------------------------------------------------------------
+ * Expires API
+ *----------------------------------------------------------------------------*/
+
+/* Remove expiry from key
+ *
+ *  Remove the object from db->expires and set to -1 attached TTL to KV
+ */
+int removeExpire(redisDb *db, robj *key) {
+    int table;
+    int slot = getKeySlot(key->ptr);
+    dictEntryLink link = kvstoreDictTwoPhaseUnlinkFind(db->expires, slot, key->ptr, &table);
+
+    if (link == NULL) return 0;
+    dictEntry *de = *link;
+    kvobj *kv = dictGetKV(de);
+    kvobj *newkv = kvobjSetExpire(kv, -1);
+    serverAssert(newkv == kv);
+    kvstoreDictTwoPhaseUnlinkFree(db->expires, slot, link, table);
+    return 1;
+}
+
+
+/* Set an expire to the specified key. If the expire is set in the context
+ * of an user calling a command 'c' is the client, otherwise 'c' is set
+ * to NULL. The 'when' parameter is the absolute unix time in milliseconds
+ * after which the key will no longer be considered valid.
+ * 
+ * Note: It may reallocate kvobj. The returned ref may point to a new object. */
+kvobj *setExpire(client *c, redisDb *db, robj *key, long long when) {
+    return setExpireByLink(c,db,key->ptr,when,NULL);
+}
+
+/* Like setExpire(), but accepts an optional `keyLink` to save lookup */
+kvobj *setExpireByLink(client *c, redisDb *db, sds key, long long when, dictEntryLink keyLink) {
+    /* Reuse the sds from the main dict in the expire dict */
+    int slot = getKeySlot(key);
+    size_t oldsize = 0;
+    if (!keyLink) {
+        keyLink = kvstoreDictFindLink(db->keys, slot, key, NULL);
+        serverAssert(keyLink != NULL);
+    }
+    kvobj *kv = dictGetKV(*keyLink);
+    long long old_when = kvobjGetExpire(kv);
+
+    if (old_when != -1) { /* old expire */
+        kvobj *kvnew = kvobjSetExpire(kv, when); /* release kv if reallocated */
+        /* Val already had an expire field, so it was not reallocated. */
+        serverAssert(kv == kvnew);
+    } else { /* No old expire */
+        if (server.memory_tracking_per_slot)
+            oldsize = kvobjAllocSize(kv);
+        uint64_t subexpiry = EB_EXPIRE_TIME_INVALID;
+        /* If hash with HFEs, take care to remove from global HFE DS before attempting
+         * to manipulate and maybe free kv object */
+        if (kv->type == OBJ_HASH)
+            subexpiry = estoreRemove(db->subexpires, slot, kv);
+
+        kvobj *kvnew = kvobjSetExpire(kv, when); /* release kv if reallocated */
+        /* if kvobj was reallocated, update dict */
+        if (kv != kvnew) {
+            kvstoreDictSetAtLink(db->keys, slot, kvnew, &keyLink, 0);
+            if (server.memory_tracking_per_slot)
+                updateSlotAllocSize(db, slot, oldsize, kvobjAllocSize(kvnew));
+            kv = kvnew;
+        }
+        /* Now add to expires */
+        dictEntry *de = kvstoreDictAddRaw(db->expires, slot, kv, NULL);
+        serverAssert(de != NULL);
+
+        if (subexpiry != EB_EXPIRE_TIME_INVALID)
+            estoreAdd(db->subexpires, slot, kv, subexpiry);
+    }
+
+    int writable_slave = server.masterhost && server.repl_slave_ro == 0;
+    if (c && writable_slave && !(c->flags & CLIENT_MASTER))
+        rememberSlaveKeyWithExpire(db,key);
+    return kv;
+}
+
+/* Retrieve the expiration time for the specified key.
+ * Returns -1 if the key has no expiration set or doesn't exists
+ *
+ * To avoid lookup, pass key-value object (`kv`) instead of `key`.
+ */
+long long getExpire(redisDb *db, sds key, kvobj *kv) {
+    if (kv == NULL) kv = dbFindExpires(db, key);
+    if (kv == NULL) return -1;
+    return kvobjGetExpire(kv);
+}
+
+/* Delete the specified expired or evicted key and propagate to replicas.
+ * Currently notify_type can only be NOTIFY_EXPIRED or NOTIFY_EVICTED,
+ * and it affects other aspects like the latency monitor event name and,
+ * which config to look for lazy free, stats var to increment, and so on.
+ *
+ * key_mem_freed is an out parameter which contains the estimated
+ * amount of memory freed due to the trimming (may be NULL) */
+static void deleteKeyAndPropagate(redisDb *db, robj *keyobj, int notify_type, long long *key_mem_freed) {
+    mstime_t latency;
+    int del_flag = notify_type == NOTIFY_EXPIRED ? DB_FLAG_KEY_EXPIRED : DB_FLAG_KEY_EVICTED;
+    int lazy_flag = notify_type == NOTIFY_EXPIRED ? server.lazyfree_lazy_expire : server.lazyfree_lazy_eviction;
+    char *latency_name = notify_type == NOTIFY_EXPIRED ? "expire-del" : "evict-del";
+    char *notify_name = notify_type == NOTIFY_EXPIRED ? "expired" : "evicted";
+
+    /* The key needs to be converted from static to heap before deleted */
+    int static_key = keyobj->refcount == OBJ_STATIC_REFCOUNT;
+    if (static_key) {
+        keyobj = createStringObject(keyobj->ptr, sdslen(keyobj->ptr));
+    }
+
+    serverLog(LL_DEBUG,"key %s %s: deleting it", (char*)keyobj->ptr, notify_type == NOTIFY_EXPIRED ? "expired" : "evicted");
+
+    /* We compute the amount of memory freed by db*Delete() alone.
+     * It is possible that actually the memory needed to propagate
+     * the DEL in AOF and replication link is greater than the one
+     * we are freeing removing the key, but we can't account for
+     * that otherwise we would never exit the loop.
+     *
+     * Same for CSC invalidation messages generated by keyModified.
+     *
+     * AOF and Output buffer memory will be freed eventually so
+     * we only care about memory used by the key space.
+     *
+     * The code here used to first propagate and then record delta
+     * using only zmalloc_used_memory but in CRDT we can't do that
+     * so we use freeMemoryGetNotCountedMemory to avoid counting
+     * AOF and slave buffers */
+    if (key_mem_freed) *key_mem_freed = (long long) zmalloc_used_memory() - freeMemoryGetNotCountedMemory();
+    latencyStartMonitor(latency);
+    dbGenericDelete(db, keyobj, lazy_flag, del_flag);
+    latencyEndMonitor(latency);
+    latencyAddSampleIfNeeded(latency_name, latency);
+    if (key_mem_freed) *key_mem_freed -= (long long) zmalloc_used_memory() - freeMemoryGetNotCountedMemory();
+
+    notifyKeyspaceEvent(notify_type, notify_name,keyobj, db->id);
+    keyModified(NULL, db, keyobj, NULL, 1);
+    propagateDeletion(db, keyobj, lazy_flag);
+
+    if (notify_type == NOTIFY_EXPIRED)
+        server.stat_expiredkeys++;
+    else
+        server.stat_evictedkeys++;
+
+    if (static_key)
+        decrRefCount(keyobj);
+}
+
+/* Delete the specified expired key and propagate. */
+void deleteExpiredKeyAndPropagate(redisDb *db, robj *keyobj) {
+    deleteKeyAndPropagate(db, keyobj, NOTIFY_EXPIRED, NULL);
+}
+
+/* Delete the specified evicted key and propagate. */
+void deleteEvictedKeyAndPropagate(redisDb *db, robj *keyobj, long long *key_mem_freed) {
+    deleteKeyAndPropagate(db, keyobj, NOTIFY_EVICTED, key_mem_freed);
+}
+
+/* Propagate an implicit key deletion into replicas and the AOF file.
+ * When a key was deleted in the master by eviction, expiration or a similar
+ * mechanism a DEL/UNLINK operation for this key is sent
+ * to all the replicas and the AOF file if enabled.
+ *
+ * This way the key deletion is centralized in one place, and since both
+ * AOF and the replication link guarantee operation ordering, everything
+ * will be consistent even if we allow write operations against deleted
+ * keys.
+ *
+ * This function may be called from:
+ * 1. Within call(): Example: Lazy-expire on key access.
+ *    In this case the caller doesn't have to do anything
+ *    because call() handles server.also_propagate(); or
+ * 2. Outside of call(): Example: Active-expire, eviction, slot ownership changed.
+ *    In this the caller must remember to call
+ *    postExecutionUnitOperations, preferably just after a
+ *    single deletion batch, so that DEL/UNLINK will NOT be wrapped
+ *    in MULTI/EXEC */
+void propagateDeletion(redisDb *db, robj *key, int lazy) {
+    robj *argv[2];
+
+    argv[0] = lazy ? shared.unlink : shared.del;
+    argv[1] = key;
+    incrRefCount(argv[0]);
+    incrRefCount(argv[1]);
+
+    /* If the master decided to delete a key we must propagate it to replicas no matter what.
+     * Even if module executed a command without asking for propagation. */
+    int prev_replication_allowed = server.replication_allowed;
+    server.replication_allowed = 1;
+    alsoPropagate(db->id,argv,2,PROPAGATE_AOF|PROPAGATE_REPL);
+    server.replication_allowed = prev_replication_allowed;
+
+    decrRefCount(argv[0]);
+    decrRefCount(argv[1]);
+}
+
+/* Check if the key is expired
+ *
+ * Provide either the key name for a lookup or KV object (to save lookup)
+ */
+int keyIsExpired(redisDb *db, sds key, kvobj *kv) {
+    /* Don't expire anything while loading. It will be done later. */
+    if (server.loading || server.allow_access_expired) return 0;
+    mstime_t when = getExpire(db, key, kv);
+    if (when < 0) return 0; /* No expire for this key */
+    const mstime_t now = commandTimeSnapshot();
+    /* The key expired if the current (virtual or real) time is greater
+     * than the expire time of the key. */
+    return now > when;
+}
+
+/* Check if user configuration allows key to be deleted due to expiary */
+int confAllowsExpireDel(void) {
+    if (server.lazyexpire_nested_arbitrary_keys)
+        return 1;
+
+    /* This configuration specifically targets nested commands, to align with RE's feature of replication between dbs.
+     * transactions (from scripts or multi-exec) containing commands like SCAN and RANDOMKEY will execute locally, but their
+     * lazy-expiration DELs may induce CROSS-SLOT on remote proxy in mode replica-of (RED-161574) */
+    return !(server.execution_nesting > 1 && server.executing_client->cmd->flags & CMD_TOUCHES_ARBITRARY_KEYS);
+}
+
+/* This function is called when we are going to perform some operation
+ * in a given key, but such key may be already logically expired even if
+ * it still exists in the database. The main way this function is called
+ * is via lookupKey*() family of functions.
+ *
+ * The behavior of the function depends on the replication role of the
+ * instance, because by default replicas do not delete expired keys. They
+ * wait for DELs from the master for consistency matters. However even
+ * replicas will try to have a coherent return value for the function,
+ * so that read commands executed in the replica side will be able to
+ * behave like if the key is expired even if still present (because the
+ * master has yet to propagate the DEL).
+ *
+ * In masters as a side effect of finding a key which is expired, such
+ * key will be evicted from the database. Also this may trigger the
+ * propagation of a DEL/UNLINK command in AOF / replication stream.
+ *
+ * On replicas, this function does not delete expired keys by default, but
+ * it still returns KEY_EXPIRED if the key is logically expired. To force deletion
+ * of logically expired keys even on replicas, use the EXPIRE_FORCE_DELETE_EXPIRED
+ * flag. Note though that if the current client is executing
+ * replicated commands from the master, keys are never considered expired.
+ *
+ * On the other hand, if you just want expiration check, but need to avoid
+ * the actual key deletion and propagation of the deletion, use the
+ * EXPIRE_AVOID_DELETE_EXPIRED flag. If also needed to read expired key (that
+ * hasn't being deleted yet) then use EXPIRE_ALLOW_ACCESS_EXPIRED.
+ *
+ * The return value of the function is KEY_VALID if the key is still valid.
+ * The function returns KEY_EXPIRED if the key is expired BUT not deleted,
+ * or returns KEY_DELETED if the key is expired and deleted. If the key is in a
+ * trim job due to slot migration, the function returns KEY_TRIMMED, unless
+ * EXPIRE_ALLOW_ACCESS_TRIMMED is set, in which case it returns KEY_VALID.
+ *
+ * You can optionally pass `kv` to save a lookup.
+ */
+keyStatus expireIfNeeded(redisDb *db, robj *key, kvobj *kv, int flags) {
+    debugAssert(key != NULL || kv != NULL);
+
+    /* NOTE: Keys in slots scheduled for trimming can still exist for a while.
+     * We don't delete it here, return KEY_VALID if allowing access to trimmed
+     * keys, and return KEY_TRIMMED otherwise. */
+    sds key_name = key ? key->ptr : kvobjGetKey(kv);
+    if (asmIsKeyInTrimJob(key_name)) {
+        if (server.allow_access_trimmed || (flags & EXPIRE_ALLOW_ACCESS_TRIMMED))
+            return KEY_VALID;
+
+        return KEY_TRIMMED;
+    }
+
+    if ((flags & EXPIRE_ALLOW_ACCESS_EXPIRED) ||
+        (!keyIsExpired(db,  key ? key->ptr : NULL, kv)))
+        return KEY_VALID;
+
+    /* If we are running in the context of a replica, instead of
+     * evicting the expired key from the database, we return ASAP:
+     * the replica key expiration is controlled by the master that will
+     * send us synthesized DEL operations for expired keys. The
+     * exception is when write operations are performed on writable
+     * replicas.
+     *
+     * In cluster mode, we also return ASAP if we are importing data
+     * from the source, to avoid deleting keys that are still in use.
+     * We create a fake master client for data import, which can be
+     * identified using the CLIENT_MASTER flag.
+     *
+     * Still we try to return the right information to the caller,
+     * that is, KEY_VALID if we think the key should still be valid,
+     * KEY_EXPIRED if we think the key is expired but don't want to delete it at this time.
+     *
+     * When replicating commands from the master, keys are never considered
+     * expired. */
+    if (server.masterhost != NULL || server.cluster_enabled) {
+        if (server.current_client && (server.current_client->flags & CLIENT_MASTER)) return KEY_VALID;
+        if (server.masterhost != NULL && !(flags & EXPIRE_FORCE_DELETE_EXPIRED)) return KEY_EXPIRED;
+    }
+
+    /* Check if user configuration disables lazy-expire deletions in current state.
+     * This will only apply if the server doesn't mandate key deletion to operate correctly (write commands). */
+    if (!(flags & EXPIRE_FORCE_DELETE_EXPIRED) && !confAllowsExpireDel())
+        return KEY_EXPIRED;
+
+    /* In some cases we're explicitly instructed to return an indication of a
+     * missing key without actually deleting it, even on masters. */
+    if (flags & EXPIRE_AVOID_DELETE_EXPIRED)
+        return KEY_EXPIRED;
+
+    /* If 'expire' action is paused, for whatever reason, then don't expire any key.
+     * Typically, at the end of the pause we will properly expire the key OR we
+     * will have failed over and the new primary will send us the expire. */
+    if (isPausedActionsWithUpdate(PAUSE_ACTION_EXPIRE)) return KEY_EXPIRED;
+
+    /* Perform deletion */
+    if (key) {
+        deleteExpiredKeyAndPropagate(db, key);
+    } else {
+        sds keyname = kvobjGetKey(kv);
+        robj *tmpkey = createStringObject(keyname, sdslen(keyname));
+        deleteExpiredKeyAndPropagate(db, tmpkey);
+        decrRefCount(tmpkey);
+    }
+    return KEY_DELETED;
+}
+
+/* CB passed to kvstoreExpand.
+ * The purpose is to skip expansion of unused dicts in cluster mode (all
+ * dicts not mapped to *my* slots) */
+static int dbExpandSkipSlot(int slot) {
+    return !clusterNodeCoversSlot(getMyClusterNode(), slot);
+}
+
+/*
+ * This functions increases size of the main/expires db to match desired number.
+ * In cluster mode resizes all individual dictionaries for slots that this node owns.
+ *
+ * Based on the parameter `try_expand`, appropriate dict expand API is invoked.
+ * if try_expand is set to 1, `dictTryExpand` is used else `dictExpand`.
+ * The return code is either `DICT_OK`/`DICT_ERR` for both the API(s).
+ * `DICT_OK` response is for successful expansion. However ,`DICT_ERR` response signifies failure in allocation in
+ * `dictTryExpand` call and in case of `dictExpand` call it signifies no expansion was performed.
+ */
+static int dbExpandGeneric(kvstore *kvs, uint64_t db_size, int try_expand) {
+    int ret;
+    if (server.cluster_enabled) {
+        /* We don't know exact number of keys that would fall into each slot, but we can
+         * approximate it, assuming even distribution, divide it by the number of slots. */
+        int slots = getMyShardSlotCount();
+        if (slots == 0) return C_OK;
+        db_size = db_size / slots;
+        ret = kvstoreExpand(kvs, db_size, try_expand, dbExpandSkipSlot);
+    } else {
+        ret = kvstoreExpand(kvs, db_size, try_expand, NULL);
+    }
+
+    return ret? C_OK : C_ERR;
+}
+
+int dbExpand(redisDb *db, uint64_t db_size, int try_expand) {
+    return dbExpandGeneric(db->keys, db_size, try_expand);
+}
+
+int dbExpandExpires(redisDb *db, uint64_t db_size, int try_expand) {
+    return dbExpandGeneric(db->expires, db_size, try_expand);
+}
+
+static kvobj *dbFindGeneric(kvstore *kvs, sds key) {
+    dictEntry *res = kvstoreDictFind(kvs, getKeySlot(key), key);
+    return (res) ? dictGetKey(res) : NULL;
+}
+
+kvobj *dbFind(redisDb *db, sds key) {
+    return dbFindGeneric(db->keys, key);
+}
+
+/* Find a KV in the main db. Return also link to it.
+ *
+ * plink - If found, set to the link of the key in the dict.
+ *         If not found, set to the bucket where the key should be added.
+ *         If set to NULL, then HT of dict not allocated yet.
+ */
+kvobj *dbFindByLink(redisDb *db, sds key, dictEntryLink *plink) {
+    int slot = getKeySlot(key);
+    dictEntryLink link, bucket;
+
+    link = kvstoreDictFindLink(db->keys, slot, key, &bucket);
+    if (link == NULL) {
+        if (plink) *plink = bucket;
+        return NULL;
+    } else {
+        if (plink) *plink = link;
+        return dictGetKV(*link);
+    }
+}
+
+kvobj *dbFindExpires(redisDb *db, sds key) {
+    return dbFindGeneric(db->expires, key);
+}
+
+unsigned long long dbSize(redisDb *db) {
+    unsigned long long total = kvstoreSize(db->keys);
+
+    if (server.cluster_enabled) {
+        /* If we are the master and there is no import or trim in progress,
+         * then we can return the total count. If not, we need to subtract
+         * the number of keys in slots that are not accessible, as below. */
+        if (clusterNodeIsMaster(getMyClusterNode()) &&
+            !asmImportInProgress() &&
+            !asmIsTrimInProgress())
+        {
+            return total;
+        }
+
+        /* Besides, we don't know the slot migration states on replicas, so we
+         * need to check each slot to see if it's accessible. */
+        for (int i = 0; i < CLUSTER_SLOTS; i++) {
+            dict *d = kvstoreGetDict(db->keys, i);
+            if (d && !clusterCanAccessKeysInSlot(i)) {
+                total -= kvstoreDictSize(db->keys, i);
+            }
+        }
+    }
+
+    return total;
+}
+
+unsigned long long dbScan(redisDb *db, unsigned long long cursor, dictScanFunction *scan_cb, void *privdata) {
+    return kvstoreScan(db->keys, cursor, -1, scan_cb, scanShouldSkipDict, privdata);
+}
+
+/* -----------------------------------------------------------------------------
+ * API to get key arguments from commands
+ * ---------------------------------------------------------------------------*/
+
+/* Prepare the getKeysResult struct to hold numkeys, either by using the
+ * pre-allocated keysbuf or by allocating a new array on the heap.
+ *
+ * This function must be called at least once before starting to populate
+ * the result, and can be called repeatedly to enlarge the result array.
+ */
+keyReference *getKeysPrepareResult(getKeysResult *result, int numkeys) {
+    /* GETKEYS_RESULT_INIT initializes keys to NULL, point it to the pre-allocated stack
+     * buffer here. */
+    if (!result->keys) {
+        serverAssert(!result->numkeys);
+        result->keys = result->keysbuf;
+    }
+
+    /* Resize if necessary */
+    if (numkeys > result->size) {
+        if (result->keys != result->keysbuf) {
+            /* We're not using a static buffer, just (re)alloc */
+            result->keys = zrealloc(result->keys, numkeys * sizeof(keyReference));
+        } else {
+            /* We are using a static buffer, copy its contents */
+            result->keys = zmalloc(numkeys * sizeof(keyReference));
+            if (result->numkeys)
+                memcpy(result->keys, result->keysbuf, result->numkeys * sizeof(keyReference));
+        }
+        result->size = numkeys;
+    }
+
+    return result->keys;
+}
+
+/* Returns a bitmask with all the flags found in any of the key specs of the command.
+ * The 'inv' argument means we'll return a mask with all flags that are missing in at least one spec. */
+int64_t getAllKeySpecsFlags(struct redisCommand *cmd, int inv) {
+    int64_t flags = 0;
+    for (int j = 0; j < cmd->key_specs_num; j++) {
+        keySpec *spec = cmd->key_specs + j;
+        flags |= inv? ~spec->flags : spec->flags;
+    }
+    return flags;
+}
+
+/* Fetch the keys based of the provided key specs. Returns the number of keys found, or -1 on error.
+ * There are several flags that can be used to modify how this function finds keys in a command.
+ * 
+ * GET_KEYSPEC_INCLUDE_NOT_KEYS: Return 'fake' keys as if they were keys.
+ * GET_KEYSPEC_RETURN_PARTIAL:   Skips invalid and incomplete keyspecs but returns the keys
+ *                               found in other valid keyspecs. 
+ */
+int getKeysUsingKeySpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result) {
+    long j, i, last, first, step;
+    keyReference *keys;
+    serverAssert(result->numkeys == 0); /* caller should initialize or reset it */
+
+    for (j = 0; j < cmd->key_specs_num; j++) {
+        keySpec *spec = cmd->key_specs + j;
+        serverAssert(spec->begin_search_type != KSPEC_BS_INVALID);
+        /* Skip specs that represent 'fake' keys */
+        if ((spec->flags & CMD_KEY_NOT_KEY) && !(search_flags & GET_KEYSPEC_INCLUDE_NOT_KEYS)) {
+            continue;
+        }
+
+        first = 0;
+        if (spec->begin_search_type == KSPEC_BS_INDEX) {
+            first = spec->bs.index.pos;
+        } else if (spec->begin_search_type == KSPEC_BS_KEYWORD) {
+            int start_index = spec->bs.keyword.startfrom > 0 ? spec->bs.keyword.startfrom : argc+spec->bs.keyword.startfrom;
+            int end_index = spec->bs.keyword.startfrom > 0 ? argc-1: 1;
+            for (i = start_index; i != end_index; i = start_index <= end_index ? i + 1 : i - 1) {
+                if (i >= argc || i < 1)
+                    break;
+                if (!strcasecmp((char*)argv[i]->ptr,spec->bs.keyword.keyword)) {
+                    first = i+1;
+                    break;
+                }
+            }
+            /* keyword not found */
+            if (!first) {
+                continue;
+            }
+        } else {
+            /* unknown spec */
+            goto invalid_spec;
+        }
+
+        if (spec->find_keys_type == KSPEC_FK_RANGE) {
+            step = spec->fk.range.keystep;
+            if (spec->fk.range.lastkey >= 0) {
+                last = first + spec->fk.range.lastkey;
+            } else {
+                if (!spec->fk.range.limit) {
+                    last = argc + spec->fk.range.lastkey;
+                } else {
+                    serverAssert(spec->fk.range.lastkey == -1);
+                    last = first + ((argc-first)/spec->fk.range.limit + spec->fk.range.lastkey);
+                }
+            }
+        } else if (spec->find_keys_type == KSPEC_FK_KEYNUM) {
+            step = spec->fk.keynum.keystep;
+            long long numkeys;
+            if (spec->fk.keynum.keynumidx >= argc)
+                goto invalid_spec;
+
+            sds keynum_str = argv[first + spec->fk.keynum.keynumidx]->ptr;
+            if (!string2ll(keynum_str,sdslen(keynum_str),&numkeys) || numkeys < 0) {
+                /* Unable to parse the numkeys argument or it was invalid */
+                goto invalid_spec;
+            }
+
+            first += spec->fk.keynum.firstkey;
+            last = first + ((long)numkeys - 1) * step;
+        } else {
+            /* unknown spec */
+            goto invalid_spec;
+        }
+
+        /* First or last is out of bounds, which indicates a syntax error */
+        if (last >= argc || last < first || first >= argc) {
+            goto invalid_spec;
+        }
+
+        int count = ((last - first)+1);
+        keys = getKeysPrepareResult(result, result->numkeys + count);
+
+        for (i = first; i <= last; i += step) {
+            if (i >= argc || i < first) {
+                /* Modules commands, and standard commands with a not fixed number
+                 * of arguments (negative arity parameter) do not have dispatch
+                 * time arity checks, so we need to handle the case where the user
+                 * passed an invalid number of arguments here. In this case we
+                 * return no keys and expect the command implementation to report
+                 * an arity or syntax error. */
+                if (cmd->flags & CMD_MODULE || cmd->arity < 0) {
+                    continue;
+                } else {
+                    serverPanic("Redis built-in command declared keys positions not matching the arity requirements.");
+                }
+            }
+            keys[result->numkeys].pos = i;
+            keys[result->numkeys].flags = spec->flags;
+            result->numkeys++;
+        }
+
+        /* Handle incomplete specs (only after we added the current spec
+         * to `keys`, just in case GET_KEYSPEC_RETURN_PARTIAL was given) */
+        if (spec->flags & CMD_KEY_INCOMPLETE) {
+            goto invalid_spec;
+        }
+
+        /* Done with this spec */
+        continue;
+
+invalid_spec:
+        if (search_flags & GET_KEYSPEC_RETURN_PARTIAL) {
+            continue;
+        } else {
+            result->numkeys = 0;
+            return -1;
+        }
+    }
+
+    return result->numkeys;
+}
+
+/* Return all the arguments that are keys in the command passed via argc / argv. 
+ * This function will eventually replace getKeysFromCommand.
+ *
+ * The command returns the positions of all the key arguments inside the array,
+ * so the actual return value is a heap allocated array of integers. The
+ * length of the array is returned by reference into *numkeys.
+ * 
+ * Along with the position, this command also returns the flags that are
+ * associated with how Redis will access the key.
+ *
+ * 'cmd' must be point to the corresponding entry into the redisCommand
+ * table, according to the command name in argv[0]. */
+int getKeysFromCommandWithSpecs(struct redisCommand *cmd, robj **argv, int argc, int search_flags, getKeysResult *result) {
+    /* The command has at least one key-spec not marked as NOT_KEY */
+    int has_keyspec = (getAllKeySpecsFlags(cmd, 1) & CMD_KEY_NOT_KEY);
+    /* The command has at least one key-spec marked as VARIABLE_FLAGS */
+    int has_varflags = (getAllKeySpecsFlags(cmd, 0) & CMD_KEY_VARIABLE_FLAGS);
+
+    /* We prefer key-specs if there are any, and their flags are reliable. */
+    if (has_keyspec && !has_varflags) {
+        int ret = getKeysUsingKeySpecs(cmd,argv,argc,search_flags,result);
+        if (ret >= 0)
+            return ret;
+        /* If the specs returned with an error (probably an INVALID or INCOMPLETE spec),
+         * fallback to the callback method. */
+    }
+
+    /* Resort to getkeys callback methods. */
+    if (cmd->flags & CMD_MODULE_GETKEYS)
+        return moduleGetCommandKeysViaAPI(cmd,argv,argc,result);
+
+    /* We use native getkeys as a last resort, since not all these native getkeys provide
+     * flags properly (only the ones that correspond to INVALID, INCOMPLETE or VARIABLE_FLAGS do.*/
+    if (cmd->getkeys_proc)
+        return cmd->getkeys_proc(cmd,argv,argc,result);
+    return 0;
+}
+
+/* This function returns a sanity check if the command may have keys. */
+int doesCommandHaveKeys(struct redisCommand *cmd) {
+    return cmd->getkeys_proc ||                                 /* has getkeys_proc (non modules) */
+        (cmd->flags & CMD_MODULE_GETKEYS) ||                    /* module with GETKEYS */
+        (getAllKeySpecsFlags(cmd, 1) & CMD_KEY_NOT_KEY);        /* has at least one key-spec not marked as NOT_KEY */
+}
+
+/* A simplified channel spec table that contains all of the redis commands
+ * and which channels they have and how they are accessed. */
+typedef struct ChannelSpecs {
+    redisCommandProc *proc; /* Command procedure to match against */
+    uint64_t flags;         /* CMD_CHANNEL_* flags for this command */
+    int start;              /* The initial position of the first channel */
+    int count;              /* The number of channels, or -1 if all remaining
+                             * arguments are channels. */
+} ChannelSpecs;
+
+ChannelSpecs commands_with_channels[] = {
+    {subscribeCommand, CMD_CHANNEL_SUBSCRIBE, 1, -1},
+    {ssubscribeCommand, CMD_CHANNEL_SUBSCRIBE, 1, -1},
+    {unsubscribeCommand, CMD_CHANNEL_UNSUBSCRIBE, 1, -1},
+    {sunsubscribeCommand, CMD_CHANNEL_UNSUBSCRIBE, 1, -1},
+    {psubscribeCommand, CMD_CHANNEL_PATTERN | CMD_CHANNEL_SUBSCRIBE, 1, -1},
+    {punsubscribeCommand, CMD_CHANNEL_PATTERN | CMD_CHANNEL_UNSUBSCRIBE, 1, -1},
+    {publishCommand, CMD_CHANNEL_PUBLISH, 1, 1},
+    {spublishCommand, CMD_CHANNEL_PUBLISH, 1, 1},
+    {NULL,0} /* Terminator. */
+};
+
+/* Returns 1 if the command may access any channels matched by the flags
+ * argument. */
+int doesCommandHaveChannelsWithFlags(struct redisCommand *cmd, int flags) {
+    /* If a module declares get channels, we are just going to assume
+     * has channels. This API is allowed to return false positives. */
+    if (cmd->flags & CMD_MODULE_GETCHANNELS) {
+        return 1;
+    }
+    for (ChannelSpecs *spec = commands_with_channels; spec->proc != NULL; spec += 1) {
+        if (cmd->proc == spec->proc) {
+            return !!(spec->flags & flags);
+        }
+    }
+    return 0;
+}
+
+/* Return all the arguments that are channels in the command passed via argc / argv. 
+ * This function behaves similar to getKeysFromCommandWithSpecs, but with channels 
+ * instead of keys.
+ * 
+ * The command returns the positions of all the channel arguments inside the array,
+ * so the actual return value is a heap allocated array of integers. The
+ * length of the array is returned by reference into *numkeys.
+ * 
+ * Along with the position, this command also returns the flags that are
+ * associated with how Redis will access the channel.
+ *
+ * 'cmd' must be point to the corresponding entry into the redisCommand
+ * table, according to the command name in argv[0]. */
+int getChannelsFromCommand(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    keyReference *keys;
+    /* If a module declares get channels, use that. */
+    if (cmd->flags & CMD_MODULE_GETCHANNELS) {
+        return moduleGetCommandChannelsViaAPI(cmd, argv, argc, result);
+    }
+    /* Otherwise check the channel spec table */
+    for (ChannelSpecs *spec = commands_with_channels; spec != NULL; spec += 1) {
+        if (cmd->proc == spec->proc) {
+            int start = spec->start;
+            int stop = (spec->count == -1) ? argc : start + spec->count;
+            if (stop > argc) stop = argc;
+            int count = 0;
+            keys = getKeysPrepareResult(result, stop - start);
+            for (int i = start; i < stop; i++ ) {
+                keys[count].pos = i;
+                keys[count++].flags = spec->flags;
+            }
+            result->numkeys = count;
+            return count;
+        }
+    }
+    return 0;
+}
+
+/* Extract keys/channels from a command and calculate the cluster slot.
+ * Returns the number of keys/channels extracted.
+ * The slot number is returned by reference into *slot.
+ * If is_incomplete is not NULL, it will be set for key extraction.
+ *
+ * This function handles both regular commands (keys) and sharded pubsub
+ * commands (channels), but excludes regular pubsub commands which don't
+ * have slots.
+ */
+int extractKeysAndSlot(struct redisCommand *cmd, robj **argv, int argc,
+                       getKeysResult *result, int *slot) {
+    int num_keys = -1;
+
+    if (!doesCommandHaveChannelsWithFlags(cmd, CMD_CHANNEL_PUBLISH | CMD_CHANNEL_SUBSCRIBE)) {
+        num_keys = getKeysFromCommandWithSpecs(cmd, argv, argc, GET_KEYSPEC_DEFAULT, result);
+    } else {
+        /* Only extract channels for commands that have key_specs (sharded pubsub).
+         * Regular pubsub commands (PUBLISH, SUBSCRIBE) don't have slots. */
+        if (cmd->key_specs_num > 0) {
+            num_keys = getChannelsFromCommand(cmd, argv, argc, result);
+        } else {
+            num_keys = 0;
+        }
+    }
+
+    *slot = extractSlotFromKeysResult(argv, result);
+    return num_keys;
+}
+
+/* The base case is to use the keys position as given in the command table
+ * (firstkey, lastkey, step).
+ * This function works only on command with the legacy_range_key_spec,
+ * all other commands should be handled by getkeys_proc. 
+ * 
+ * If the commands keyspec is incomplete, no keys will be returned, and the provided
+ * keys function should be called instead.
+ * 
+ * NOTE: This function does not guarantee populating the flags for 
+ * the keys, in order to get flags you should use getKeysUsingKeySpecs. */
+int getKeysUsingLegacyRangeSpec(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    int j, i = 0, last, first, step;
+    keyReference *keys;
+    UNUSED(argv);
+
+    if (cmd->legacy_range_key_spec.begin_search_type == KSPEC_BS_INVALID) {
+        result->numkeys = 0;
+        return 0;
+    }
+
+    first = cmd->legacy_range_key_spec.bs.index.pos;
+    last = cmd->legacy_range_key_spec.fk.range.lastkey;
+    if (last >= 0)
+        last += first;
+    step = cmd->legacy_range_key_spec.fk.range.keystep;
+
+    if (last < 0) last = argc+last;
+
+    int count = ((last - first)+1);
+    keys = getKeysPrepareResult(result, count);
+
+    for (j = first; j <= last; j += step) {
+        if (j >= argc || j < first) {
+            /* Modules commands, and standard commands with a not fixed number
+             * of arguments (negative arity parameter) do not have dispatch
+             * time arity checks, so we need to handle the case where the user
+             * passed an invalid number of arguments here. In this case we
+             * return no keys and expect the command implementation to report
+             * an arity or syntax error. */
+            if (cmd->flags & CMD_MODULE || cmd->arity < 0) {
+                result->numkeys = 0;
+                return 0;
+            } else {
+                serverPanic("Redis built-in command declared keys positions not matching the arity requirements.");
+            }
+        }
+        keys[i].pos = j;
+        /* Flags are omitted from legacy key specs */
+        keys[i++].flags = 0;
+    }
+    result->numkeys = i;
+    return i;
+}
+
+/* Return all the arguments that are keys in the command passed via argc / argv.
+ *
+ * The command returns the positions of all the key arguments inside the array,
+ * so the actual return value is a heap allocated array of integers. The
+ * length of the array is returned by reference into *numkeys.
+ *
+ * 'cmd' must be point to the corresponding entry into the redisCommand
+ * table, according to the command name in argv[0].
+ *
+ * This function uses the command table if a command-specific helper function
+ * is not required, otherwise it calls the command-specific function. */
+int getKeysFromCommand(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    if (cmd->flags & CMD_MODULE_GETKEYS) {
+        return moduleGetCommandKeysViaAPI(cmd,argv,argc,result);
+    } else if (cmd->getkeys_proc) {
+        return cmd->getkeys_proc(cmd,argv,argc,result);
+    } else {
+        return getKeysUsingLegacyRangeSpec(cmd,argv,argc,result);
+    }
+}
+
+/* Free the result of getKeysFromCommand. */
+void getKeysFreeResult(getKeysResult *result) {
+    if (result && result->keys != result->keysbuf)
+        zfree(result->keys);
+}
+
+/* Helper function to extract keys from following commands:
+ * COMMAND [destkey] <num-keys> <key> [...] <key> [...] ... <options>
+ *
+ * eg:
+ * ZUNION <num-keys> <key> <key> ... <key> <options>
+ * ZUNIONSTORE <destkey> <num-keys> <key> <key> ... <key> <options>
+ *
+ * 'storeKeyOfs': destkey index, 0 means destkey not exists.
+ * 'keyCountOfs': num-keys index.
+ * 'firstKeyOfs': firstkey index.
+ * 'keyStep': the interval of each key, usually this value is 1.
+ * 
+ * The commands using this function have a fully defined keyspec, so returning flags isn't needed. */
+int genericGetKeys(int storeKeyOfs, int keyCountOfs, int firstKeyOfs, int keyStep,
+                    robj **argv, int argc, getKeysResult *result) {
+    int i, num;
+    keyReference *keys;
+
+    num = atoi(argv[keyCountOfs]->ptr);
+    /* Sanity check. Don't return any key if the command is going to
+     * reply with syntax error. (no input keys). */
+    if (num < 1 || num > (argc - firstKeyOfs)/keyStep) {
+        result->numkeys = 0;
+        return 0;
+    }
+
+    int numkeys = storeKeyOfs ? num + 1 : num;
+    keys = getKeysPrepareResult(result, numkeys);
+    result->numkeys = numkeys;
+
+    /* Add all key positions for argv[firstKeyOfs...n] to keys[] */
+    for (i = 0; i < num; i++) {
+        keys[i].pos = firstKeyOfs+(i*keyStep);
+        keys[i].flags = 0;
+    } 
+
+    if (storeKeyOfs) {
+        keys[num].pos = storeKeyOfs;
+        keys[num].flags = 0;
+    } 
+    return result->numkeys;
+}
+
+int sintercardGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 1, 2, 1, argv, argc, result);
+}
+
+int zunionInterDiffStoreGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(1, 2, 3, 1, argv, argc, result);
+}
+
+int zunionInterDiffGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 1, 2, 1, argv, argc, result);
+}
+
+int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 2, 3, 1, argv, argc, result);
+}
+
+int functionGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 2, 3, 1, argv, argc, result);
+}
+
+int lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 1, 2, 1, argv, argc, result);
+}
+
+int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 2, 3, 1, argv, argc, result);
+}
+
+int zmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 1, 2, 1, argv, argc, result);
+}
+
+int bzmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    UNUSED(cmd);
+    return genericGetKeys(0, 2, 3, 1, argv, argc, result);
+}
+
+/* Helper function to extract keys from the SORT RO command.
+ *
+ * SORT <sort-key>
+ *
+ * The second argument of SORT is always a key, however an arbitrary number of
+ * keys may be accessed while doing the sort (the BY and GET args), so the
+ * key-spec declares incomplete keys which is why we have to provide a concrete
+ * implementation to fetch the keys.
+ *
+ * This command declares incomplete keys, so the flags are correctly set for this function */
+int sortROGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    keyReference *keys;
+    UNUSED(cmd);
+    UNUSED(argv);
+    UNUSED(argc);
+
+    keys = getKeysPrepareResult(result, 1);
+    keys[0].pos = 1; /* <sort-key> is always present. */
+    keys[0].flags = CMD_KEY_RO | CMD_KEY_ACCESS;
+    result->numkeys = 1;
+    return result->numkeys;
+}
+
+/* Helper function to extract keys from the SORT command.
+ *
+ * SORT <sort-key> ... STORE <store-key> ...
+ *
+ * The first argument of SORT is always a key, however a list of options
+ * follow in SQL-alike style. Here we parse just the minimum in order to
+ * correctly identify keys in the "STORE" option. 
+ * 
+ * This command declares incomplete keys, so the flags are correctly set for this function */
+int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    int i, j, num, found_store = 0;
+    keyReference *keys;
+    UNUSED(cmd);
+
+    num = 0;
+    keys = getKeysPrepareResult(result, 2); /* Alloc 2 places for the worst case. */
+    keys[num].pos = 1; /* <sort-key> is always present. */
+    keys[num++].flags = CMD_KEY_RO | CMD_KEY_ACCESS;
+
+    /* Search for STORE option. By default we consider options to don't
+     * have arguments, so if we find an unknown option name we scan the
+     * next. However there are options with 1 or 2 arguments, so we
+     * provide a list here in order to skip the right number of args. */
+    struct {
+        char *name;
+        int skip;
+    } skiplist[] = {
+        {"limit", 2},
+        {"get", 1},
+        {"by", 1},
+        {NULL, 0} /* End of elements. */
+    };
+
+    for (i = 2; i < argc; i++) {
+        for (j = 0; skiplist[j].name != NULL; j++) {
+            if (!strcasecmp(argv[i]->ptr,skiplist[j].name)) {
+                i += skiplist[j].skip;
+                break;
+            } else if (!strcasecmp(argv[i]->ptr,"store") && i+1 < argc) {
+                /* Note: we don't increment "num" here and continue the loop
+                 * to be sure to process the *last* "STORE" option if multiple
+                 * ones are provided. This is same behavior as SORT. */
+                found_store = 1;
+                keys[num].pos = i+1; /* <store-key> */
+                keys[num].flags = CMD_KEY_OW | CMD_KEY_UPDATE;
+                break;
+            }
+        }
+    }
+    result->numkeys = num + found_store;
+    return result->numkeys;
+}
+
+/* This command declares incomplete keys, so the flags are correctly set for this function */
+int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    int i, j, num, first;
+    keyReference *keys;
+    UNUSED(cmd);
+
+    /* Assume the obvious form. */
+    first = 3;
+    num = 1;
+
+    /* But check for the extended one with the KEYS option. */
+    struct {
+        char* name;
+        int skip;
+    } skip_keywords[] = {       
+        {"copy", 0},
+        {"replace", 0},
+        {"auth", 1},
+        {"auth2", 2},
+        {NULL, 0}
+    };
+    if (argc > 6) {
+        for (i = 6; i < argc; i++) {
+            if (!strcasecmp(argv[i]->ptr, "keys")) {
+                if (sdslen(argv[3]->ptr) > 0) {
+                    /* This is a syntax error. So ignore the keys and leave
+                     * the syntax error to be handled by migrateCommand. */
+                    num = 0; 
+                } else {
+                    first = i + 1;
+                    num = argc - first;
+                }
+                break;
+            }
+            for (j = 0; skip_keywords[j].name != NULL; j++) {
+                if (!strcasecmp(argv[i]->ptr, skip_keywords[j].name)) {
+                    i += skip_keywords[j].skip;
+                    break;
+                }
+            }
+        }
+    }
+
+    keys = getKeysPrepareResult(result, num);
+    for (i = 0; i < num; i++) {
+        keys[i].pos = first+i;
+        keys[i].flags = CMD_KEY_RW | CMD_KEY_ACCESS | CMD_KEY_DELETE;
+    } 
+    result->numkeys = num;
+    return num;
+}
+
+/* Helper function to extract keys from following commands:
+ * GEORADIUS key x y radius unit [WITHDIST] [WITHHASH] [WITHCOORD] [ASC|DESC]
+ *                             [COUNT count] [STORE key|STOREDIST key]
+ * GEORADIUSBYMEMBER key member radius unit ... options ...
+ * 
+ * This command has a fully defined keyspec, so returning flags isn't needed. */
+int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    int i, num;
+    keyReference *keys;
+    UNUSED(cmd);
+
+    /* Check for the presence of the stored key in the command */
+    int stored_key = -1;
+    for (i = 5; i < argc; i++) {
+        char *arg = argv[i]->ptr;
+        /* For the case when user specifies both "store" and "storedist" options, the
+         * second key specified would override the first key. This behavior is kept
+         * the same as in georadiusCommand method.
+         */
+        if ((!strcasecmp(arg, "store") || !strcasecmp(arg, "storedist")) && ((i+1) < argc)) {
+            stored_key = i+1;
+            i++;
+        }
+    }
+    num = 1 + (stored_key == -1 ? 0 : 1);
+
+    /* Keys in the command come from two places:
+     * argv[1] = key,
+     * argv[5...n] = stored key if present
+     */
+    keys = getKeysPrepareResult(result, num);
+
+    /* Add all key positions to keys[] */
+    keys[0].pos = 1;
+    keys[0].flags = 0;
+    if(num > 1) {
+         keys[1].pos = stored_key;
+         keys[1].flags = 0;
+    }
+    result->numkeys = num;
+    return num;
+}
+
+/* XREAD [BLOCK <milliseconds>] [COUNT <count>] [GROUP <groupname> <ttl>]
+ *       STREAMS key_1 key_2 ... key_N ID_1 ID_2 ... ID_N
+ *
+ * This command has a fully defined keyspec, so returning flags isn't needed. */
+int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    int i, num = 0;
+    keyReference *keys;
+    UNUSED(cmd);
+
+    /* We need to parse the options of the command in order to seek the first
+     * "STREAMS" string which is actually the option. This is needed because
+     * "STREAMS" could also be the name of the consumer group and even the
+     * name of the stream key. */
+    int streams_pos = -1;
+    for (i = 1; i < argc; i++) {
+        char *arg = argv[i]->ptr;
+        if (!strcasecmp(arg, "block")) {
+            i++; /* Skip option argument. */
+        } else if (!strcasecmp(arg, "count")) {
+            i++; /* Skip option argument. */
+        } else if (!strcasecmp(arg, "group")) {
+            i += 2; /* Skip option argument. */
+        } else if (!strcasecmp(arg, "noack")) {
+            /* Nothing to do. */
+        } else if (!strcasecmp(arg, "streams")) {
+            streams_pos = i;
+            break;
+        } else {
+            break; /* Syntax error. */
+        }
+    }
+    if (streams_pos != -1) num = argc - streams_pos - 1;
+
+    /* Syntax error. */
+    if (streams_pos == -1 || num == 0 || num % 2 != 0) {
+        result->numkeys = 0;
+        return 0;
+    }
+    num /= 2; /* We have half the keys as there are arguments because
+                 there are also the IDs, one per key. */
+
+    keys = getKeysPrepareResult(result, num);
+    for (i = streams_pos+1; i < argc-num; i++) {
+        keys[i-streams_pos-1].pos = i;
+        keys[i-streams_pos-1].flags = 0; 
+    } 
+    result->numkeys = num;
+    return num;
+}
+
+/* Helper function to extract keys from the SET command, which may have
+ * an RW flag if the GET, IF* arguments are present, OW otherwise. */
+int setGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    keyReference *keys;
+    UNUSED(cmd);
+
+    keys = getKeysPrepareResult(result, 1);
+    keys[0].pos = 1; /* We always know the position */
+    result->numkeys = 1;
+    int actual = CMD_KEY_OW;
+    int logical = CMD_KEY_UPDATE;
+
+    for (int i = 3; i < argc; i++) {
+        char *arg = argv[i]->ptr;
+        if ((arg[0] == 'g' || arg[0] == 'G') &&
+            (arg[1] == 'e' || arg[1] == 'E') &&
+            (arg[2] == 't' || arg[2] == 'T') && arg[3] == '\0')
+        {
+            actual = CMD_KEY_RW;
+            logical |= CMD_KEY_ACCESS;
+        } else if (!strcasecmp(arg, "ifeq") || !strcasecmp(arg, "ifne") ||
+                   !strcasecmp(arg, "ifdeq") || !strcasecmp(arg, "ifdne"))
+        {
+            actual = CMD_KEY_RW;
+        }
+    }
+
+    keys[0].flags = actual | logical;
+
+    return 1;
+}
+
+/* Helper function to extract keys from the DELEX command, which may have
+ * an RW flag if the IF* arguments are present, RM otherwise. */
+int delexGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    keyReference *keys;
+    UNUSED(cmd);
+
+    keys = getKeysPrepareResult(result, 1);
+    keys[0].pos = 1; /* We always know the position */
+    result->numkeys = 1;
+    int actual = CMD_KEY_RM;
+    int logical = CMD_KEY_DELETE;
+
+    for (int i = 2; i < argc; i++) {
+        char *arg = argv[i]->ptr;
+        if (!strcasecmp(arg, "ifeq") || !strcasecmp(arg, "ifne") ||
+            !strcasecmp(arg, "ifdeq") || !strcasecmp(arg, "ifdne"))
+        {
+            actual = CMD_KEY_RW;
+        }
+    }
+
+    keys[0].flags = actual | logical;
+
+    return 1;
+}
+
+/* Helper function to extract keys from the BITFIELD command, which may be
+ * read-only if the BITFIELD GET subcommand is used. */
+int bitfieldGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
+    keyReference *keys;
+    int readonly = 1;
+    UNUSED(cmd);
+
+    keys = getKeysPrepareResult(result, 1);
+    keys[0].pos = 1; /* We always know the position */
+    result->numkeys = 1;
+
+    for (int i = 2; i < argc; i++) {
+        int remargs = argc - i - 1; /* Remaining args other than current. */
+        char *arg = argv[i]->ptr;
+        if (!strcasecmp(arg, "get") && remargs >= 2) {
+            i += 2;
+        } else if ((!strcasecmp(arg, "set") || !strcasecmp(arg, "incrby")) && remargs >= 3) {
+            readonly = 0;
+            i += 3;
+            break;
+        } else if (!strcasecmp(arg, "overflow") && remargs >= 1) {
+            i += 1;
+        } else {
+            readonly = 0; /* Syntax error. safer to assume non-RO. */
+            break;
+        }
+    }
+
+    if (readonly) {
+        keys[0].flags = CMD_KEY_RO | CMD_KEY_ACCESS;
+    } else {
+        keys[0].flags = CMD_KEY_RW | CMD_KEY_ACCESS | CMD_KEY_UPDATE;
+    }
+    return 1;
+}