Add multi-database support to cluster mode

src/cluster.c

@@ -814,8 +814,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) {
@@ -897,7 +905,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;
@@ -909,11 +917,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));
@@ -1028,6 +1036,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++) {
@@ -1045,6 +1055,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) {
+            int id;
+            if (getLongLongFromObject(margv[i], &id) != C_OK || (id < 0 || id >= server.dbnum)) {
+                if (error_code) *error_code = CLUSTER_REDIR_UNSTABLE;
+                return NULL;
+            }
+            currentDb = server.db+id;
+        }
+
+
         for (j = 0; j < numkeys; j++) {
             robj *thiskey = margv[keyindex[j].pos];
             int thisslot = keyHashSlot((char *)thiskey->ptr, sdslen(thiskey->ptr));
@@ -1078,6 +1098,20 @@ getNodeByQuery(client *c, struct serverCommand *cmd, robj **argv, int argc, int
                         importing_slot = 1;
                     }
                 }
+
+                /* Block COPY and MOVE during slot migration to prevent multi-DB inconsistencies. */
+                if ((migrating_slot || importing_slot) && (mcmd->proc == copyCommand || mcmd->proc == moveCommand)) {
+                    if (error_code) *error_code = CLUSTER_REDIR_UNSTABLE;
+                    getKeysFreeResult(&result);
+                    return NULL;
+                }
+
+                if (n == NULL) {
+                    getKeysFreeResult(&result);
+                    if (error_code) *error_code = CLUSTER_REDIR_UNSTABLE;
+                    return NULL;
+                }
+
             } else {
                 /* If it is not the first key/channel, make sure it is exactly
                  * the same key/channel as the first we saw. */
@@ -1101,8 +1135,9 @@ getNodeByQuery(client *c, struct serverCommand *cmd, robj **argv, int argc, int
              * 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 ((!c->flag.multi || c->flag.multi && cmd->proc == execCommand) && // Multi/Exec validation happens on exec
+                (migrating_slot || importing_slot) && !pubsubshard_included) {
+                if (lookupKeyReadWithFlags(currentDb, thiskey, flags) == NULL)
                     missing_keys++;
                 else
                     existing_keys++;

src/cluster.h

@@ -115,6 +115,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

@@ -5794,11 +5794,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++) {
@@ -6430,29 +6425,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;
@@ -6912,7 +6909,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;
         }
@@ -7053,7 +7050,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;
@@ -7187,7 +7184,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

@@ -607,13 +607,6 @@ void loadServerConfigFromString(char *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.config_hz < CONFIG_MIN_HZ) server.config_hz = CONFIG_MIN_HZ;
     if (server.config_hz > CONFIG_MAX_HZ) server.config_hz = CONFIG_MAX_HZ;

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 0;
+        }
+    }
+    return 1;
+}

src/server.h

@@ -3488,6 +3488,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

@@ -707,7 +707,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++;