Add multi-database support to cluster mode (#1671)

## cluster: add multi-database support in cluster mode

Add multi-database support in cluster mode to align with standalone mode
and facilitate migration. Previously, cluster mode was restricted to a
single database (DB0). This change allows multiple databases while
preserving the existing slot-based key distribution.


### Key Features:
- Database-Agnostic Hashing. The hashing algorithm is unchanged.
  Identical keys always map to the same slot across all databases,
  ensuring consistent key distribution and compatibility with
  existing single-database setups.
- Multi-DB commands support. SELECT, MOVE, and COPY are now supported in
  cluster mode.
- Fully backward compatible with no API changes.
- SWAPDB is not supported in cluster mode. It is unsafe due to
inconsistency risks.

### Command-Level Changes:
- SELECT / MOVE / COPY are now supported in cluster mode.
- MOVE / COPY (with db) are rejected (TRYAGAIN error) during slot
migration to prevent multi-DB inconsistencies.
- SWAPDB will return an error if used when cluster mode is enabled.
- GETKEYSINSLOT, COUNTKEYSINSLOT and MIGRATE will operate in the context
of the selected database.
This means, for example, that migrating keys in a slot will require
iterating and repeating across all databases.

### Slot Migration Process:
- Multi-DB support in cluster mode affects slot migration. Operators
should now iterate over all the configured databases.
 
### Transaction Handling (MULTI/EXEC):
- getNodeByQuery key lookup behavior changed:
  - No key lookups when queuing commands in MULTI, only cross-slot
    validation.
  - Key lookups happen at EXEC time.
  - SELECT inside MULTI/EXEC is now checked, ensuring key validation
    uses the selected DB at lookup.

### Valkey-cli:
- valkey-cli has been updated to support resharding across all
databases.

### Configuration:
- Introduce new configuration `cluster-databases`.
The new configuration controls the maximal number of databases in
cluster mode.

Implements  #1319

---------

Signed-off-by: xbasel <103044017+xbasel@users.noreply.github.com>
Signed-off-by: zhaozhao.zz <zhaozhao.zz@alibaba-inc.com>
Co-authored-by: zhaozhao.zz <zhaozhao.zz@alibaba-inc.com>
Co-authored-by: Viktor Söderqvist <viktor.soderqvist@est.tech>
Co-authored-by: Madelyn Olson <madelyneolson@gmail.com>
Co-authored-by: Ran Shidlansik <ranshid@amazon.com>

Author: 5 people

src/cluster.c

@@ -817,8 +817,16 @@ static int shouldReturnTlsInfo(void) {
     }
 }
 
+unsigned int countKeysInSlotForDb(unsigned int hashslot, serverDb *db) {
+    return kvstoreHashtableSize(db->keys, hashslot);
+}
+
 unsigned int countKeysInSlot(unsigned int slot) {
-    return kvstoreHashtableSize(server.db->keys, slot);
+    unsigned int result = 0;
+    for (int i = 0; i < server.dbnum; i++) {
+        result += countKeysInSlotForDb(slot, server.db + i);
+    }
+    return result;
 }
 
 void clusterCommandHelp(client *c) {
@@ -900,7 +908,7 @@ void clusterCommand(client *c) {
             addReplyError(c, "Invalid slot");
             return;
         }
-        addReplyLongLong(c, countKeysInSlot(slot));
+        addReplyLongLong(c, countKeysInSlotForDb(slot, c->db));
     } else if (!strcasecmp(c->argv[1]->ptr, "getkeysinslot") && c->argc == 4) {
         /* CLUSTER GETKEYSINSLOT <slot> <count> */
         long long maxkeys, slot;
@@ -912,11 +920,11 @@ void clusterCommand(client *c) {
             return;
         }
 
-        unsigned int keys_in_slot = countKeysInSlot(slot);
+        unsigned int keys_in_slot = countKeysInSlotForDb(slot, c->db);
         unsigned int numkeys = maxkeys > keys_in_slot ? keys_in_slot : maxkeys;
         addReplyArrayLen(c, numkeys);
         kvstoreHashtableIterator *kvs_di = NULL;
-        kvs_di = kvstoreGetHashtableIterator(server.db->keys, slot, 0);
+        kvs_di = kvstoreGetHashtableIterator(c->db->keys, slot, 0);
         for (unsigned int i = 0; i < numkeys; i++) {
             void *next;
             serverAssert(kvstoreHashtableIteratorNext(kvs_di, &next));
@@ -1031,6 +1039,8 @@ getNodeByQuery(client *c, struct serverCommand *cmd, robj **argv, int argc, int
     int pubsubshard_included =
         (cmd_flags & CMD_PUBSUB) || (c->cmd->proc == execCommand && (c->mstate->cmd_flags & CMD_PUBSUB));
 
+    serverDb *currentDb = c->db;
+
     /* Check that all the keys are in the same hash slot, and obtain this
      * slot and the node associated. */
     for (i = 0; i < ms->count; i++) {
@@ -1048,6 +1058,16 @@ getNodeByQuery(client *c, struct serverCommand *cmd, robj **argv, int argc, int
         numkeys = getKeysFromCommand(mcmd, margv, margc, &result);
         keyindex = result.keys;
 
+        if (mcmd->proc == selectCommand) {
+            /* Failed SELECT is ignored since it doesn't modify the database. */
+            serverDb *origDb = currentDb;
+            long long id;
+            if (getLongLongFromObject(margv[1], &id) == C_OK && selectDb(c, id) == C_OK) {
+                currentDb = c->db;
+                selectDb(c, origDb->id);
+            }
+        }
+
         for (j = 0; j < numkeys; j++) {
             robj *thiskey = margv[keyindex[j].pos];
             int thisslot = keyHashSlot((char *)thiskey->ptr, sdslen(thiskey->ptr));
@@ -1081,6 +1101,7 @@ getNodeByQuery(client *c, struct serverCommand *cmd, robj **argv, int argc, int
                         importing_slot = 1;
                     }
                 }
+
             } else {
                 /* If it is not the first key/channel, make sure it is exactly
                  * the same key/channel as the first we saw. */
@@ -1097,15 +1118,39 @@ getNodeByQuery(client *c, struct serverCommand *cmd, robj **argv, int argc, int
                 }
             }
 
-            /* Migrating / Importing slot? Count keys we don't have.
+            /* Block MOVE command as the destination key is not expected to exist, and we don't know if it was migrated */
+            if ((migrating_slot || importing_slot) && mcmd->proc == moveCommand) {
+                if (error_code) *error_code = CLUSTER_REDIR_UNSTABLE;
+                getKeysFreeResult(&result);
+                return NULL;
+            }
+
+            /* Block the COPY command if it's cross-DB to keep the code simple.
+             * Allowing cross-DB COPY is possible, but it would require looking up the second key in the target DB.
+             * The command should only be allowed if the key exists. We may revisit this decision in the future. */
+            if ((migrating_slot || importing_slot) &&
+                mcmd->proc == copyCommand &&
+                margc >= 4 && !strcasecmp(margv[3]->ptr, "db")) {
+                long long value;
+                if (getLongLongFromObject(margv[4], &value) != C_OK || value != currentDb->id) {
+                    if (error_code) *error_code = CLUSTER_REDIR_UNSTABLE;
+                    getKeysFreeResult(&result);
+                    return NULL;
+                }
+            }
+
+            /* Migrating / Importing slot? During exec we count keys we don't have.
              * If it is pubsubshard command, it isn't required to check
              * the channel being present or not in the node during the
              * slot migration, the channel will be served from the source
              * node until the migration completes with CLUSTER SETSLOT <slot>
              * NODE <node-id>. */
             int flags = LOOKUP_NOTOUCH | LOOKUP_NOSTATS | LOOKUP_NONOTIFY | LOOKUP_NOEXPIRE;
-            if ((migrating_slot || importing_slot) && !pubsubshard_included) {
-                if (lookupKeyReadWithFlags(&server.db[0], thiskey, flags) == NULL)
+            if ((migrating_slot || importing_slot) &&
+                !pubsubshard_included &&
+                (!c->flag.multi || (c->flag.multi && cmd->proc == execCommand)) // Multi/Exec validation happens on exec
+            ) {
+                if (lookupKeyReadWithFlags(currentDb, thiskey, flags) == NULL)
                     missing_keys++;
                 else
                     existing_keys++;

src/cluster.h

@@ -113,6 +113,7 @@ int detectAndUpdateCachedNodeHealth(void);
 client *createCachedResponseClient(int resp);
 void deleteCachedResponseClient(client *recording_client);
 void clearCachedClusterSlotsResponse(void);
+unsigned int countKeysInSlotForDb(unsigned int hashslot, serverDb *db);
 unsigned int countKeysInSlot(unsigned int hashslot);
 int getSlotOrReply(client *c, robj *o);
 

src/cluster_legacy.c

@@ -5878,11 +5878,6 @@ int verifyClusterConfigWithData(void) {
      * completely depend on the replication stream. */
     if (nodeIsReplica(myself)) return C_OK;
 
-    /* Make sure we only have keys in DB0. */
-    for (j = 1; j < server.dbnum; j++) {
-        if (kvstoreSize(server.db[j].keys)) return C_ERR;
-    }
-
     /* Check that all the slots we see populated memory have a corresponding
      * entry in the cluster table. Otherwise fix the table. */
     for (j = 0; j < CLUSTER_SLOTS; j++) {
@@ -6542,29 +6537,31 @@ unsigned int delKeysInSlot(unsigned int hashslot) {
     server.server_del_keys_in_slot = 1;
     unsigned int j = 0;
 
-    kvstoreHashtableIterator *kvs_di = NULL;
-    void *next;
-    kvs_di = kvstoreGetHashtableIterator(server.db->keys, hashslot, HASHTABLE_ITER_SAFE);
-    while (kvstoreHashtableIteratorNext(kvs_di, &next)) {
-        robj *valkey = next;
-        enterExecutionUnit(1, 0);
-        sds sdskey = objectGetKey(valkey);
-        robj *key = createStringObject(sdskey, sdslen(sdskey));
-        dbDelete(&server.db[0], key);
-        propagateDeletion(&server.db[0], key, server.lazyfree_lazy_server_del);
-        signalModifiedKey(NULL, &server.db[0], key);
-        /* The keys are not actually logically deleted from the database, just moved to another node.
-         * The modules needs to know that these keys are no longer available locally, so just send the
-         * keyspace notification to the modules, but not to clients. */
-        moduleNotifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, server.db[0].id);
-        exitExecutionUnit();
-        postExecutionUnitOperations();
-        decrRefCount(key);
-        j++;
-        server.dirty++;
+    for (int i = 0; i < server.dbnum; i++) {
+        kvstoreHashtableIterator *kvs_di = NULL;
+        void *next;
+        serverDb db = server.db[i];
+        kvs_di = kvstoreGetHashtableIterator(db.keys, hashslot, HASHTABLE_ITER_SAFE);
+        while (kvstoreHashtableIteratorNext(kvs_di, &next)) {
+            robj *valkey = next;
+            enterExecutionUnit(1, 0);
+            sds sdskey = objectGetKey(valkey);
+            robj *key = createStringObject(sdskey, sdslen(sdskey));
+            dbDelete(&db, key);
+            propagateDeletion(&db, key, server.lazyfree_lazy_server_del);
+            signalModifiedKey(NULL, &db, key);
+            /* The keys are not actually logically deleted from the database, just moved to another node.
+             * The modules needs to know that these keys are no longer available locally, so just send the
+             * keyspace notification to the modules, but not to clients. */
+            moduleNotifyKeyspaceEvent(NOTIFY_GENERIC, "del", key, db.id);
+            exitExecutionUnit();
+            postExecutionUnitOperations();
+            decrRefCount(key);
+            j++;
+            server.dirty++;
+        }
+        kvstoreReleaseHashtableIterator(kvs_di);
     }
-    kvstoreReleaseHashtableIterator(kvs_di);
-
     server.server_del_keys_in_slot = 0;
     serverAssert(server.execution_nesting == 0);
     return j;
@@ -7033,7 +7030,7 @@ int clusterCommandSpecial(client *c) {
         }
     } else if (!strcasecmp(c->argv[1]->ptr, "flushslots") && c->argc == 2) {
         /* CLUSTER FLUSHSLOTS */
-        if (kvstoreSize(server.db[0].keys) != 0) {
+        if (!dbHasNoKeys()) {
             addReplyError(c, "DB must be empty to perform CLUSTER FLUSHSLOTS.");
             return 1;
         }
@@ -7203,7 +7200,7 @@ int clusterCommandSpecial(client *c) {
         /* If the instance is currently a primary, it should have no assigned
          * slots nor keys to accept to replicate some other node.
          * Replicas can switch to another primary without issues. */
-        if (clusterNodeIsPrimary(myself) && (myself->numslots != 0 || kvstoreSize(server.db[0].keys) != 0)) {
+        if (clusterNodeIsPrimary(myself) && (myself->numslots != 0 || !dbHasNoKeys())) {
             addReplyError(c, "To set a master the node must be empty and "
                              "without assigned slots.");
             return 1;
@@ -7355,7 +7352,7 @@ int clusterCommandSpecial(client *c) {
 
         /* Replicas can be reset while containing data, but not primary nodes
          * that must be empty. */
-        if (clusterNodeIsPrimary(myself) && kvstoreSize(c->db->keys) != 0) {
+        if (clusterNodeIsPrimary(myself) && !dbHasNoKeys()) {
             addReplyError(c, "CLUSTER RESET can't be called with "
                              "master nodes containing keys");
             return 1;

src/config.c

@@ -608,13 +608,6 @@ void loadServerConfigFromString(sds config) {
         goto loaderr;
     }
 
-    /* in case cluster mode is enabled dbnum must be 1 */
-    if (server.cluster_enabled && server.dbnum > 1) {
-        serverLog(LL_WARNING, "WARNING: Changing databases number from %d to 1 since we are in cluster mode",
-                  server.dbnum);
-        server.dbnum = 1;
-    }
-
     /* To ensure backward compatibility and work while hz is out of range */
     if (server.hz < CONFIG_MIN_HZ) server.hz = CONFIG_MIN_HZ;
     if (server.hz > CONFIG_MAX_HZ) server.hz = CONFIG_MAX_HZ;
@@ -3263,7 +3256,8 @@ standardConfig static_configs[] = {
     createEnumConfig("rdb-version-check", NULL, MODIFIABLE_CONFIG, rdb_version_check_enum, server.rdb_version_check, RDB_VERSION_CHECK_STRICT, NULL, NULL),
 
     /* Integer configs */
-    createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.dbnum, 16, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.config_databases, 16, INTEGER_CONFIG, NULL, NULL),
+    createIntConfig("cluster-databases", NULL, IMMUTABLE_CONFIG, 1, INT_MAX, server.config_databases_cluster, 1, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("port", NULL, MODIFIABLE_CONFIG, 0, 65535, server.port, 6379, INTEGER_CONFIG, NULL, updatePort),                                   /* TCP port. */
     createIntConfig("io-threads", NULL, DEBUG_CONFIG | IMMUTABLE_CONFIG, 1, IO_THREADS_MAX_NUM, server.io_threads_num, 1, INTEGER_CONFIG, NULL, NULL), /* Single threaded by default */
     createIntConfig("events-per-io-thread", NULL, MODIFIABLE_CONFIG | HIDDEN_CONFIG, 0, INT_MAX, server.events_per_io_thread, 2, INTEGER_CONFIG, NULL, NULL),

src/db.c

@@ -860,10 +860,6 @@ void selectCommand(client *c) {
 
     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 (selectDb(c, id) == C_ERR) {
         addReplyError(c, "DB index is out of range");
     } else {
@@ -1429,11 +1425,6 @@ void moveCommand(client *c) {
     int srcid, dbid;
     long long expire;
 
-    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;
@@ -1518,11 +1509,6 @@ void copyCommand(client *c) {
         }
     }
 
-    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. */
@@ -2851,3 +2837,12 @@ int bitfieldGetKeys(struct serverCommand *cmd, robj **argv, int argc, getKeysRes
     }
     return 1;
 }
+
+bool dbHasNoKeys(void) {
+    for (int i = 0; i < server.dbnum; i++) {
+        if (kvstoreSize(server.db[i].keys) != 0) {
+            return false;
+        }
+    }
+    return true;
+}

src/server.c

@@ -2813,6 +2813,8 @@ void initServer(void) {
         serverLog(LL_WARNING, "Failed creating the event loop. Error message: '%s'", strerror(errno));
         exit(1);
     }
+
+    server.dbnum = server.cluster_enabled ? server.config_databases_cluster : server.config_databases;
     server.db = zmalloc(sizeof(serverDb) * server.dbnum);
 
     /* Create the databases, and initialize other internal state. */

src/server.h

@@ -46,6 +46,7 @@
 #include <limits.h>
 #include <unistd.h>
 #include <errno.h>
+#include <stdbool.h>
 #include <inttypes.h>
 #include <pthread.h>
 #include <syslog.h>
@@ -1793,7 +1794,9 @@ struct valkeyServer {
     unsigned long active_defrag_max_scan_fields; /* maximum number of fields of set/hash/zset/list to process from
                                                     within the main dict scan */
     size_t client_max_querybuf_len;              /* Limit for client query buffer length */
-    int dbnum;                                   /* Total number of configured DBs */
+    int config_databases;                        /* Total number of configured DBs in standalone */
+    int config_databases_cluster;                /* Total number of configured DBs in cluster mode */
+    int dbnum;                                   /* Total number of initialized DBs */
     int supervised;                              /* 1 if supervised, 0 otherwise. */
     int supervised_mode;                         /* See SUPERVISED_* */
     int daemonize;                               /* True if running as a daemon */
@@ -3500,6 +3503,7 @@ int zmpopGetKeys(struct serverCommand *cmd, robj **argv, int argc, getKeysResult
 int bzmpopGetKeys(struct serverCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int setGetKeys(struct serverCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int bitfieldGetKeys(struct serverCommand *cmd, robj **argv, int argc, getKeysResult *result);
+bool dbHasNoKeys(void);
 
 unsigned short crc16(const char *buf, int len);
 

src/valkey-benchmark.c

@@ -717,7 +717,7 @@ static client createClient(char *cmd, size_t len, client from, int thread_id) {
      * buffer with the SELECT command, that will be discarded the first
      * time the replies are received, so if the client is reused the
      * SELECT command will not be used again. */
-    if (config.conn_info.input_dbnum != 0 && !is_cluster_client) {
+    if (config.conn_info.input_dbnum) {
         c->obuf = sdscatprintf(c->obuf, "*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n", (int)sdslen(config.input_dbnumstr),
                                config.input_dbnumstr);
         c->prefix_pending++;