Treat subcommands as commands (#9504)

## Intro

The purpose is to allow having different flags/ACL categories for
subcommands (Example: CONFIG GET is ok-loading but CONFIG SET isn't)

We create a small command table for every command that has subcommands
and each subcommand has its own flags, etc. (same as a "regular" command)

This commit also unites the Redis and the Sentinel command tables

## Affected commands

CONFIG
Used to have "admin ok-loading ok-stale no-script"
Changes:
1. Dropped "ok-loading" in all except GET (this doesn't change behavior since
there were checks in the code doing that)

XINFO
Used to have "read-only random"
Changes:
1. Dropped "random" in all except CONSUMERS

XGROUP
Used to have "write use-memory"
Changes:
1. Dropped "use-memory" in all except CREATE and CREATECONSUMER

COMMAND
No changes.

MEMORY
Used to have "random read-only"
Changes:
1. Dropped "random" in PURGE and USAGE

ACL
Used to have "admin no-script ok-loading ok-stale"
Changes:
1. Dropped "admin" in WHOAMI, GENPASS, and CAT

LATENCY
No changes.

MODULE
No changes.

SLOWLOG
Used to have "admin random ok-loading ok-stale"
Changes:
1. Dropped "random" in RESET

OBJECT
Used to have "read-only random"
Changes:
1. Dropped "random" in ENCODING and REFCOUNT

SCRIPT
Used to have "may-replicate no-script"
Changes:
1. Dropped "may-replicate" in all except FLUSH and LOAD

CLIENT
Used to have "admin no-script random ok-loading ok-stale"
Changes:
1. Dropped "random" in all except INFO and LIST
2. Dropped "admin" in ID, TRACKING, CACHING, GETREDIR, INFO, SETNAME, GETNAME, and REPLY

STRALGO
No changes.

PUBSUB
No changes.

CLUSTER
Changes:
1. Dropped "admin in countkeysinslots, getkeysinslot, info, nodes, keyslot, myid, and slots

SENTINEL
No changes.

(note that DEBUG also fits, but we decided not to convert it since it's for
debugging and anyway undocumented)

## New sub-command
This commit adds another element to the per-command output of COMMAND,
describing the list of subcommands, if any (in the same structure as "regular" commands)
Also, it adds a new subcommand:
```
COMMAND LIST [FILTERBY (MODULE <module-name>|ACLCAT <cat>|PATTERN <pattern>)]
```
which returns a set of all commands (unless filters), but excluding subcommands.

## Module API
A new module API, RM_CreateSubcommand, was added, in order to allow
module writer to define subcommands

## ACL changes:
1. Now, that each subcommand is actually a command, each has its own ACL id.
2. The old mechanism of allowed_subcommands is redundant
(blocking/allowing a subcommand is the same as blocking/allowing a regular command),
but we had to keep it, to support the widespread usage of allowed_subcommands
to block commands with certain args, that aren't subcommands (e.g. "-select +select|0").
3. I have renamed allowed_subcommands to allowed_firstargs to emphasize the difference.
4. Because subcommands are commands in ACL too, you can now use "-" to block subcommands
(e.g. "+client -client|kill"), which wasn't possible in the past.
5. It is also possible to use the allowed_firstargs mechanism with subcommand.
For example: `+config -config|set +config|set|loglevel` will block all CONFIG SET except
for setting the log level.
6. All of the ACL changes above required some amount of refactoring.

## Misc
1. There are two approaches: Either each subcommand has its own function or all
   subcommands use the same function, determining what to do according to argv[0].
   For now, I took the former approaches only with CONFIG and COMMAND,
   while other commands use the latter approach (for smaller blamelog diff).
2. Deleted memoryGetKeys: It is no longer needed because MEMORY USAGE now uses the "range" key spec.
4. Bugfix: GETNAME was missing from CLIENT's help message.
5. Sentinel and Redis now use the same table, with the same function pointer.
   Some commands have a different implementation in Sentinel, so we redirect
   them (these are ROLE, PUBLISH, and INFO).
6. Command stats now show the stats per subcommand (e.g. instead of stats just
   for "config" you will have stats for "config|set", "config|get", etc.)
7. It is now possible to use COMMAND directly on subcommands:
   COMMAND INFO CONFIG|GET (The pipeline syntax was inspired from ACL, and
   can be used in functions lookupCommandBySds and lookupCommandByCString)
8. STRALGO is now a container command (has "help")

## Breaking changes:
1. Command stats now show the stats per subcommand (see (5) above)

14 files changed, 1270 insertions, 393 deletions

diff --git a/src/acl.c b/src/acl.c
index 6947fc204..199e050dd 100644
--- a/src/acl.c
+++ b/src/acl.c
@@ -101,9 +101,9 @@ struct ACLUserFlag {
     {NULL,0} /* Terminator. */
 };
 
-void ACLResetSubcommandsForCommand(user *u, unsigned long id);
-void ACLResetSubcommands(user *u);
-void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub);
+void ACLResetFirstArgsForCommand(user *u, unsigned long id);
+void ACLResetFirstArgs(user *u);
+void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub);
 void ACLFreeLogEntry(void *le);
 
 /* The length of the string representation of a hashed password. */
@@ -254,7 +254,7 @@ user *ACLCreateUser(const char *name, size_t namelen) {
     user *u = zmalloc(sizeof(*u));
     u->name = sdsnewlen(name,namelen);
     u->flags = USER_FLAG_DISABLED | server.acl_pubsub_default;
-    u->allowed_subcommands = NULL;
+    u->allowed_firstargs = NULL;
     u->passwords = listCreate();
     u->patterns = listCreate();
     u->channels = listCreate();
@@ -296,7 +296,7 @@ void ACLFreeUser(user *u) {
     listRelease(u->passwords);
     listRelease(u->patterns);
     listRelease(u->channels);
-    ACLResetSubcommands(u);
+    ACLResetFirstArgs(u);
     zfree(u);
 }
 
@@ -343,15 +343,15 @@ void ACLCopyUser(user *dst, user *src) {
     memcpy(dst->allowed_commands,src->allowed_commands,
            sizeof(dst->allowed_commands));
     dst->flags = src->flags;
-    ACLResetSubcommands(dst);
-    /* Copy the allowed subcommands array of array of SDS strings. */
-    if (src->allowed_subcommands) {
+    ACLResetFirstArgs(dst);
+    /* Copy the allowed first-args array of array of SDS strings. */
+    if (src->allowed_firstargs) {
         for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++) {
-            if (src->allowed_subcommands[j]) {
-                for (int i = 0; src->allowed_subcommands[j][i]; i++)
+            if (src->allowed_firstargs[j]) {
+                for (int i = 0; src->allowed_firstargs[j][i]; i++)
                 {
-                    ACLAddAllowedSubcommand(dst, j,
-                        src->allowed_subcommands[j][i]);
+                    ACLAddAllowedFirstArg(dst, j,
+                        src->allowed_firstargs[j][i]);
                 }
             }
         }
@@ -413,6 +413,38 @@ void ACLSetUserCommandBit(user *u, unsigned long id, int value) {
     }
 }
 
+/* This function is used to allow/block a specific command.
+ * Allowing/blocking a container command also applies for its subcommands */
+void ACLChangeCommandPerm(user *u, struct redisCommand *cmd, int allow) {
+    unsigned long id = cmd->id;
+    ACLSetUserCommandBit(u,id,allow);
+    ACLResetFirstArgsForCommand(u,id);
+    if (cmd->subcommands_dict) {
+        dictEntry *de;
+        dictIterator *di = dictGetSafeIterator(cmd->subcommands_dict);
+        while((de = dictNext(di)) != NULL) {
+            struct redisCommand *sub = (struct redisCommand *)dictGetVal(de);
+            ACLSetUserCommandBit(u,sub->id,allow);
+        }
+    }
+}
+
+void ACLSetUserCommandBitsForCategoryLogic(dict *commands, user *u, uint64_t cflag, int value) {
+    dictIterator *di = dictGetIterator(commands);
+    dictEntry *de;
+    while ((de = dictNext(di)) != NULL) {
+        struct redisCommand *cmd = dictGetVal(de);
+        if (cmd->flags & CMD_MODULE) continue; /* Ignore modules commands. */
+        if (cmd->flags & cflag) {
+            ACLChangeCommandPerm(u,cmd,value);
+        }
+        if (cmd->subcommands_dict) {
+            ACLSetUserCommandBitsForCategoryLogic(cmd->subcommands_dict, u, cflag, value);
+        }
+    }
+    dictReleaseIterator(di);
+}
+
 /* This is like ACLSetUserCommandBit(), but instead of setting the specified
  * ID, it will check all the commands in the category specified as argument,
  * and will set all the bits corresponding to such commands to the specified
@@ -422,18 +454,26 @@ void ACLSetUserCommandBit(user *u, unsigned long id, int value) {
 int ACLSetUserCommandBitsForCategory(user *u, const char *category, int value) {
     uint64_t cflag = ACLGetCommandCategoryFlagByName(category);
     if (!cflag) return C_ERR;
-    dictIterator *di = dictGetIterator(server.orig_commands);
+    ACLSetUserCommandBitsForCategoryLogic(server.orig_commands, u, cflag, value);
+    return C_OK;
+}
+
+void ACLCountCategoryBitsForCommands(dict *commands, user *u, unsigned long *on, unsigned long *off, uint64_t cflag) {
+    dictIterator *di = dictGetIterator(commands);
     dictEntry *de;
     while ((de = dictNext(di)) != NULL) {
         struct redisCommand *cmd = dictGetVal(de);
-        if (cmd->flags & CMD_MODULE) continue; /* Ignore modules commands. */
         if (cmd->flags & cflag) {
-            ACLSetUserCommandBit(u,cmd->id,value);
-            ACLResetSubcommandsForCommand(u,cmd->id);
+            if (ACLGetUserCommandBit(u,cmd->id))
+                (*on)++;
+            else
+                (*off)++;
+        }
+        if (cmd->subcommands_dict) {
+            ACLCountCategoryBitsForCommands(cmd->subcommands_dict, u, on, off, cflag);
         }
     }
     dictReleaseIterator(di);
-    return C_OK;
 }
 
 /* Return the number of commands allowed (on) and denied (off) for the user 'u'
@@ -447,19 +487,46 @@ int ACLCountCategoryBitsForUser(user *u, unsigned long *on, unsigned long *off,
     if (!cflag) return C_ERR;
 
     *on = *off = 0;
-    dictIterator *di = dictGetIterator(server.orig_commands);
+    ACLCountCategoryBitsForCommands(server.orig_commands, u, on, off, cflag);
+    return C_OK;
+}
+
+sds ACLDescribeUserCommandRulesSingleCommands(user *u, user *fakeuser, sds rules, dict *commands) {
+    dictIterator *di = dictGetIterator(commands);
     dictEntry *de;
     while ((de = dictNext(di)) != NULL) {
         struct redisCommand *cmd = dictGetVal(de);
-        if (cmd->flags & cflag) {
-            if (ACLGetUserCommandBit(u,cmd->id))
-                (*on)++;
-            else
-                (*off)++;
+        int userbit = ACLGetUserCommandBit(u,cmd->id);
+        int fakebit = ACLGetUserCommandBit(fakeuser,cmd->id);
+        if (userbit != fakebit) {
+            rules = sdscatlen(rules, userbit ? "+" : "-", 1);
+            sds fullname = getFullCommandName(cmd);
+            rules = sdscat(rules,fullname);
+            sdsfree(fullname);
+            rules = sdscatlen(rules," ",1);
+            ACLChangeCommandPerm(fakeuser,cmd,userbit);
+        }
+
+        if (cmd->subcommands_dict)
+            rules = ACLDescribeUserCommandRulesSingleCommands(u,fakeuser,rules,cmd->subcommands_dict);
+
+        /* Emit the first-args if there are any. */
+        if (userbit == 0 && u->allowed_firstargs &&
+            u->allowed_firstargs[cmd->id])
+        {
+            for (int j = 0; u->allowed_firstargs[cmd->id][j]; j++) {
+                rules = sdscatlen(rules,"+",1);
+                sds fullname = getFullCommandName(cmd);
+                rules = sdscat(rules,fullname);
+                sdsfree(fullname);
+                rules = sdscatlen(rules,"|",1);
+                rules = sdscatsds(rules,u->allowed_firstargs[cmd->id][j]);
+                rules = sdscatlen(rules," ",1);
+            }
         }
     }
     dictReleaseIterator(di);
-    return C_OK;
+    return rules;
 }
 
 /* This function returns an SDS string representing the specified user ACL
@@ -563,33 +630,7 @@ sds ACLDescribeUserCommandRules(user *u) {
     }
 
     /* Fix the final ACLs with single commands differences. */
-    dictIterator *di = dictGetIterator(server.orig_commands);
-    dictEntry *de;
-    while ((de = dictNext(di)) != NULL) {
-        struct redisCommand *cmd = dictGetVal(de);
-        int userbit = ACLGetUserCommandBit(u,cmd->id);
-        int fakebit = ACLGetUserCommandBit(fakeuser,cmd->id);
-        if (userbit != fakebit) {
-            rules = sdscatlen(rules, userbit ? "+" : "-", 1);
-            rules = sdscat(rules,cmd->name);
-            rules = sdscatlen(rules," ",1);
-            ACLSetUserCommandBit(fakeuser,cmd->id,userbit);
-        }
-
-        /* Emit the subcommands if there are any. */
-        if (userbit == 0 && u->allowed_subcommands &&
-            u->allowed_subcommands[cmd->id])
-        {
-            for (int j = 0; u->allowed_subcommands[cmd->id][j]; j++) {
-                rules = sdscatlen(rules,"+",1);
-                rules = sdscat(rules,cmd->name);
-                rules = sdscatlen(rules,"|",1);
-                rules = sdscatsds(rules,u->allowed_subcommands[cmd->id][j]);
-                rules = sdscatlen(rules," ",1);
-            }
-        }
-    }
-    dictReleaseIterator(di);
+    rules = ACLDescribeUserCommandRulesSingleCommands(u,fakeuser,rules,server.orig_commands);
 
     /* Trim the final useless space. */
     sdsrange(rules,0,-2);
@@ -683,67 +724,66 @@ sds ACLDescribeUser(user *u) {
 struct redisCommand *ACLLookupCommand(const char *name) {
     struct redisCommand *cmd;
     sds sdsname = sdsnew(name);
-    cmd = dictFetchValue(server.orig_commands, sdsname);
+    cmd = lookupCommandBySdsLogic(server.orig_commands,sdsname);
     sdsfree(sdsname);
     return cmd;
 }
 
-/* Flush the array of allowed subcommands for the specified user
+/* Flush the array of allowed first-args for the specified user
  * and command ID. */
-void ACLResetSubcommandsForCommand(user *u, unsigned long id) {
-    if (u->allowed_subcommands && u->allowed_subcommands[id]) {
-        for (int i = 0; u->allowed_subcommands[id][i]; i++)
-            sdsfree(u->allowed_subcommands[id][i]);
-        zfree(u->allowed_subcommands[id]);
-        u->allowed_subcommands[id] = NULL;
+void ACLResetFirstArgsForCommand(user *u, unsigned long id) {
+    if (u->allowed_firstargs && u->allowed_firstargs[id]) {
+        for (int i = 0; u->allowed_firstargs[id][i]; i++)
+            sdsfree(u->allowed_firstargs[id][i]);
+        zfree(u->allowed_firstargs[id]);
+        u->allowed_firstargs[id] = NULL;
     }
 }
 
-/* Flush the entire table of subcommands. This is useful on +@all, -@all
+/* Flush the entire table of first-args. This is useful on +@all, -@all
  * or similar to return back to the minimal memory usage (and checks to do)
  * for the user. */
-void ACLResetSubcommands(user *u) {
-    if (u->allowed_subcommands == NULL) return;
+void ACLResetFirstArgs(user *u) {
+    if (u->allowed_firstargs == NULL) return;
     for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++) {
-        if (u->allowed_subcommands[j]) {
-            for (int i = 0; u->allowed_subcommands[j][i]; i++)
-                sdsfree(u->allowed_subcommands[j][i]);
-            zfree(u->allowed_subcommands[j]);
+        if (u->allowed_firstargs[j]) {
+            for (int i = 0; u->allowed_firstargs[j][i]; i++)
+                sdsfree(u->allowed_firstargs[j][i]);
+            zfree(u->allowed_firstargs[j]);
         }
     }
-    zfree(u->allowed_subcommands);
-    u->allowed_subcommands = NULL;
+    zfree(u->allowed_firstargs);
+    u->allowed_firstargs = NULL;
 }
 
-/* Add a subcommand to the list of subcommands for the user 'u' and
+/* Add a first-arh to the list of subcommands for the user 'u' and
  * the command id specified. */
-void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
-    /* If this is the first subcommand to be configured for
-     * this user, we have to allocate the subcommands array. */
-    if (u->allowed_subcommands == NULL) {
-        u->allowed_subcommands = zcalloc(USER_COMMAND_BITS_COUNT *
-                                 sizeof(sds*));
+void ACLAddAllowedFirstArg(user *u, unsigned long id, const char *sub) {
+    /* If this is the first first-arg to be configured for
+     * this user, we have to allocate the first-args array. */
+    if (u->allowed_firstargs == NULL) {
+        u->allowed_firstargs = zcalloc(USER_COMMAND_BITS_COUNT * sizeof(sds*));
     }
 
     /* We also need to enlarge the allocation pointing to the
      * null terminated SDS array, to make space for this one.
      * To start check the current size, and while we are here
-     * make sure the subcommand is not already specified inside. */
+     * make sure the first-arg is not already specified inside. */
     long items = 0;
-    if (u->allowed_subcommands[id]) {
-        while(u->allowed_subcommands[id][items]) {
+    if (u->allowed_firstargs[id]) {
+        while(u->allowed_firstargs[id][items]) {
             /* If it's already here do not add it again. */
-            if (!strcasecmp(u->allowed_subcommands[id][items],sub)) return;
+            if (!strcasecmp(u->allowed_firstargs[id][items],sub))
+                return;
             items++;
         }
     }
 
     /* Now we can make space for the new item (and the null term). */
     items += 2;
-    u->allowed_subcommands[id] = zrealloc(u->allowed_subcommands[id],
-                                 sizeof(sds)*items);
-    u->allowed_subcommands[id][items-2] = sdsnew(sub);
-    u->allowed_subcommands[id][items-1] = NULL;
+    u->allowed_firstargs[id] = zrealloc(u->allowed_firstargs[id], sizeof(sds)*items);
+    u->allowed_firstargs[id][items-2] = sdsnew(sub);
+    u->allowed_firstargs[id][items-1] = NULL;
 }
 
 /* Set user properties according to the string "op". The following
@@ -753,8 +793,10 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  * off          Disable the user: it's no longer possible to authenticate
  *              with this user, however the already authenticated connections
  *              will still work.
- * +<command>   Allow the execution of that command
- * -<command>   Disallow the execution of that command
+ * +<command>   Allow the execution of that command.
+ *              May be used with `|` for allowing subcommands (e.g "+config|get")
+ * -<command>   Disallow the execution of that command.
+ *              May be used with `|` for blocking subcommands (e.g "-config|set")
  * +@<category> Allow the execution of all the commands in such category
  *              with valid categories are like @admin, @set, @sortedset, ...
  *              and so forth, see the full list in the server.c file where
@@ -762,10 +804,10 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  *              The special category @all means all the commands, but currently
  *              present in the server, and that will be loaded in the future
  *              via modules.
- * +<command>|subcommand    Allow a specific subcommand of an otherwise
- *                          disabled command. Note that this form is not
- *                          allowed as negative like -DEBUG|SEGFAULT, but
- *                          only additive starting with "+".
+ * +<command>|first-arg    Allow a specific first argument of an otherwise
+ *                         disabled command. Note that this form is not
+ *                         allowed as negative like -SELECT|1, but
+ *                         only additive starting with "+".
  * allcommands  Alias for +@all. Note that it implies the ability to execute
  *              all the future commands loaded via the modules system.
  * nocommands   Alias for -@all.
@@ -866,13 +908,13 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     {
         memset(u->allowed_commands,255,sizeof(u->allowed_commands));
         u->flags |= USER_FLAG_ALLCOMMANDS;
-        ACLResetSubcommands(u);
+        ACLResetFirstArgs(u);
     } else if (!strcasecmp(op,"nocommands") ||
                !strcasecmp(op,"-@all"))
     {
         memset(u->allowed_commands,0,sizeof(u->allowed_commands));
         u->flags &= ~USER_FLAG_ALLCOMMANDS;
-        ACLResetSubcommands(u);
+        ACLResetFirstArgs(u);
     } else if (!strcasecmp(op,"nopass")) {
         u->flags |= USER_FLAG_NOPASS;
         listEmpty(u->passwords);
@@ -952,24 +994,25 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
             sdsfree(newpat);
         u->flags &= ~USER_FLAG_ALLCHANNELS;
     } else if (op[0] == '+' && op[1] != '@') {
-        if (strchr(op,'|') == NULL) {
-            if (ACLLookupCommand(op+1) == NULL) {
+        if (strrchr(op,'|') == NULL) {
+            struct redisCommand *cmd = ACLLookupCommand(op+1);
+            if (cmd == NULL) {
                 errno = ENOENT;
                 return C_ERR;
             }
-            unsigned long id = ACLGetCommandID(op+1);
-            ACLSetUserCommandBit(u,id,1);
-            ACLResetSubcommandsForCommand(u,id);
+            ACLChangeCommandPerm(u,cmd,1);
         } else {
             /* Split the command and subcommand parts. */
             char *copy = zstrdup(op+1);
-            char *sub = strchr(copy,'|');
+            char *sub = strrchr(copy,'|');
             sub[0] = '\0';
             sub++;
 
+            struct redisCommand *cmd = ACLLookupCommand(copy);
+
             /* Check if the command exists. We can't check the
              * subcommand to see if it is valid. */
-            if (ACLLookupCommand(copy) == NULL) {
+            if (cmd == NULL) {
                 zfree(copy);
                 errno = ENOENT;
                 return C_ERR;
@@ -983,22 +1026,38 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
                 return C_ERR;
             }
 
-            unsigned long id = ACLGetCommandID(copy);
-            /* Add the subcommand to the list of valid ones, if the command is not set. */
-            if (ACLGetUserCommandBit(u,id) == 0) {
-                ACLAddAllowedSubcommand(u,id,sub);
+            if (cmd->subcommands_dict) {
+                /* If user is trying to allow a valid subcommand we can just add its unique ID */
+                struct redisCommand *cmd = ACLLookupCommand(op+1);
+                if (cmd == NULL) {
+                    zfree(copy);
+                    errno = ENOENT;
+                    return C_ERR;
+                }
+                ACLChangeCommandPerm(u,cmd,1);
+            } else {
+                /* If user is trying to use the ACL mech to block SELECT except SELECT 0 or
+                 * block DEBUG except DEBUG OBJECT (DEBUG subcommands are not considered
+                 * subcommands for now) we use the allowed_firstargs mechanism. */
+                struct redisCommand *cmd = ACLLookupCommand(copy);
+                if (cmd == NULL) {
+                    zfree(copy);
+                    errno = ENOENT;
+                    return C_ERR;
+                }
+                /* Add the first-arg to the list of valid ones. */
+                ACLAddAllowedFirstArg(u,cmd->id,sub);
             }
 
             zfree(copy);
         }
     } else if (op[0] == '-' && op[1] != '@') {
-        if (ACLLookupCommand(op+1) == NULL) {
+        struct redisCommand *cmd = ACLLookupCommand(op+1);
+        if (cmd == NULL) {
             errno = ENOENT;
             return C_ERR;
         }
-        unsigned long id = ACLGetCommandID(op+1);
-        ACLSetUserCommandBit(u,id,0);
-        ACLResetSubcommandsForCommand(u,id);
+        ACLChangeCommandPerm(u,cmd,0);
     } else if ((op[0] == '+' || op[0] == '-') && op[1] == '@') {
         int bitval = op[0] == '+' ? 1 : 0;
         if (ACLSetUserCommandBitsForCategory(u,op+2,bitval) == C_ERR) {
@@ -1138,7 +1197,6 @@ int ACLAuthenticateUser(client *c, robj *username, robj *password) {
  * command name, so that a command retains the same ID in case of modules that
  * are unloaded and later reloaded. */
 unsigned long ACLGetCommandID(const char *cmdname) {
-
     sds lowername = sdsnew(cmdname);
     sdstolower(lowername);
     if (commandId == NULL) commandId = raxNew();
@@ -1225,23 +1283,23 @@ int ACLCheckCommandPerm(const user *u, struct redisCommand *cmd, robj **argv, in
     if (!(u->flags & USER_FLAG_ALLCOMMANDS) && !(cmd->flags & CMD_NO_AUTH))
     {
         /* If the bit is not set we have to check further, in case the
-         * command is allowed just with that specific subcommand. */
+         * command is allowed just with that specific first argument. */
         if (ACLGetUserCommandBit(u,id) == 0) {
-            /* Check if the subcommand matches. */
+            /* Check if the first argument matches. */
             if (argc < 2 ||
-                u->allowed_subcommands == NULL ||
-                u->allowed_subcommands[id] == NULL)
+                u->allowed_firstargs == NULL ||
+                u->allowed_firstargs[id] == NULL)
             {
                 return ACL_DENIED_CMD;
             }
 
             long subid = 0;
             while (1) {
-                if (u->allowed_subcommands[id][subid] == NULL)
+                if (u->allowed_firstargs[id][subid] == NULL)
                     return ACL_DENIED_CMD;
-                if (!strcasecmp(argv[1]->ptr,
-                                u->allowed_subcommands[id][subid]))
-                    break; /* Subcommand match found. Stop here. */
+                int idx = cmd->parent ? 2 : 1;
+                if (!strcasecmp(argv[idx]->ptr,u->allowed_firstargs[id][subid]))
+                    break; /* First argument match found. Stop here. */
                 subid++;
             }
         }
diff --git a/src/aof.c b/src/aof.c
index f9df2b1c8..0090c9472 100644
--- a/src/aof.c
+++ b/src/aof.c
@@ -793,7 +793,7 @@ int loadAppendOnlyFile(char *filename) {
         }
 
         /* Command lookup */
-        cmd = lookupCommand(argv[0]->ptr);
+        cmd = lookupCommand(argv,argc);
         if (!cmd) {
             serverLog(LL_WARNING,
                 "Unknown command '%s' reading the append only file",
diff --git a/src/config.c b/src/config.c
index 29c2a2195..c044ca516 100644
--- a/src/config.c
+++ b/src/config.c
@@ -573,7 +573,7 @@ void loadServerConfigFromString(char *config) {
         } else if (!strcasecmp(argv[0],"list-max-ziplist-value") && argc == 2) {
             /* DEAD OPTION */
         } else if (!strcasecmp(argv[0],"rename-command") && argc == 3) {
-            struct redisCommand *cmd = lookupCommand(argv[1]);
+            struct redisCommand *cmd = lookupCommandBySds(argv[1]);
             int retval;
 
             if (!cmd) {
@@ -2735,18 +2735,11 @@ standardConfig configs[] = {
 };
 
 /*-----------------------------------------------------------------------------
- * CONFIG command entry point
+ * CONFIG HELP
  *----------------------------------------------------------------------------*/
 
-void configCommand(client *c) {
-    /* Only allow CONFIG GET while loading. */
-    if (server.loading && strcasecmp(c->argv[1]->ptr,"get")) {
-        addReplyError(c,"Only CONFIG GET is allowed during loading");
-        return;
-    }
-
-    if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
-        const char *help[] = {
+void configHelpCommand(client *c) {
+    const char *help[] = {
 "GET <pattern>",
 "    Return parameters matching the glob-like <pattern> and their values.",
 "SET <directive> <value>",
@@ -2756,32 +2749,36 @@ void configCommand(client *c) {
 "REWRITE",
 "    Rewrite the configuration file.",
 NULL
-        };
-
-        addReplyHelp(c, help);
-    } else if (!strcasecmp(c->argv[1]->ptr,"set") && c->argc == 4) {
-        configSetCommand(c);
-    } else if (!strcasecmp(c->argv[1]->ptr,"get") && c->argc == 3) {
-        configGetCommand(c);
-    } else if (!strcasecmp(c->argv[1]->ptr,"resetstat") && c->argc == 2) {
-        resetServerStats();
-        resetCommandTableStats();
-        resetErrorTableStats();
-        addReply(c,shared.ok);
-    } else if (!strcasecmp(c->argv[1]->ptr,"rewrite") && c->argc == 2) {
-        if (server.configfile == NULL) {
-            addReplyError(c,"The server is running without a config file");
-            return;
-        }
-        if (rewriteConfig(server.configfile, 0) == -1) {
-            serverLog(LL_WARNING,"CONFIG REWRITE failed: %s", strerror(errno));
-            addReplyErrorFormat(c,"Rewriting config file: %s", strerror(errno));
-        } else {
-            serverLog(LL_WARNING,"CONFIG REWRITE executed with success.");
-            addReply(c,shared.ok);
-        }
-    } else {
-        addReplySubcommandSyntaxError(c);
+    };
+
+    addReplyHelp(c, help);
+}
+
+/*-----------------------------------------------------------------------------
+ * CONFIG RESETSTAT
+ *----------------------------------------------------------------------------*/
+
+void configResetStatCommand(client *c) {
+    resetServerStats();
+    resetCommandTableStats(server.commands);
+    resetErrorTableStats();
+    addReply(c,shared.ok);
+}
+
+/*-----------------------------------------------------------------------------
+ * CONFIG REWRITE
+ *----------------------------------------------------------------------------*/
+
+void configRewriteCommand(client *c) {
+    if (server.configfile == NULL) {
+        addReplyError(c,"The server is running without a config file");
         return;
     }
+    if (rewriteConfig(server.configfile, 0) == -1) {
+        serverLog(LL_WARNING,"CONFIG REWRITE failed: %s", strerror(errno));
+        addReplyErrorFormat(c,"Rewriting config file: %s", strerror(errno));
+    } else {
+        serverLog(LL_WARNING,"CONFIG REWRITE executed with success.");
+        addReply(c,shared.ok);
+    }
 }
diff --git a/src/db.c b/src/db.c
index 23cf220db..b0d379a16 100644
--- a/src/db.c
+++ b/src/db.c
@@ -1873,21 +1873,6 @@ int lcsGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *r
     return result->numkeys;
 }
 
-/* Helper function to extract keys from memory command.
- * MEMORY USAGE <key> */
-int memoryGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
-    UNUSED(cmd);
-
-    getKeysPrepareResult(result, 1);
-    if (argc >= 3 && !strcasecmp(argv[1]->ptr,"usage")) {
-        result->keys[0] = 2;
-        result->numkeys = 1;
-        return result->numkeys;
-    }
-    result->numkeys = 0;
-    return 0;
-}
-
 /* XREAD [BLOCK <milliseconds>] [COUNT <count>] [GROUP <groupname> <ttl>]
  *       STREAMS key_1 key_2 ... key_N ID_1 ID_2 ... ID_N */
 int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
diff --git a/src/module.c b/src/module.c
index 9cf30938d..029b5ddca 100644
--- a/src/module.c
+++ b/src/module.c
@@ -830,6 +830,8 @@ int64_t commandKeySpecsFlagsFromString(const char *s) {
     return flags;
 }
 
+RedisModuleCommandProxy *moduleCreateCommandProxy(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc cmdfunc, int64_t flags, int firstkey, int lastkey, int keystep);
+
 /* Register a new command in the Redis server, that will be handled by
  * calling the function pointer 'cmdfunc' using the RedisModule calling
  * convention. The function returns REDISMODULE_ERR if the specified command
@@ -888,7 +890,7 @@ int64_t commandKeySpecsFlagsFromString(const char *s) {
  *                     Normally this is used by a command that is used
  *                     to authenticate a client.
  * * **"may-replicate"**: This command may generate replication traffic, even
- *                        though it's not a write command.  
+ *                        though it's not a write command.
  *
  * The last three parameters specify which arguments of the new command are
  * Redis keys. See https://redis.io/commands/command for more information.
@@ -917,16 +919,24 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c
     if ((flags & CMD_MODULE_NO_CLUSTER) && server.cluster_enabled)
         return REDISMODULE_ERR;
 
+    /* Check if the command name is busy. */
+    if (lookupCommandByCString(name) != NULL)
+        return REDISMODULE_ERR;
+
+    RedisModuleCommandProxy *cp = moduleCreateCommandProxy(ctx, name, cmdfunc, flags, firstkey, lastkey, keystep);
+    cp->rediscmd->arity = cmdfunc ? -1 : -2;
+
+    dictAdd(server.commands,sdsnew(name),cp->rediscmd);
+    dictAdd(server.orig_commands,sdsnew(name),cp->rediscmd);
+    cp->rediscmd->id = ACLGetCommandID(name); /* ID used for ACL. */
+    return REDISMODULE_OK;
+}
+
+RedisModuleCommandProxy *moduleCreateCommandProxy(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc cmdfunc, int64_t flags, int firstkey, int lastkey, int keystep) {
     struct redisCommand *rediscmd;
     RedisModuleCommandProxy *cp;
     sds cmdname = sdsnew(name);
 
-    /* Check if the command name is busy. */
-    if (lookupCommand(cmdname) != NULL) {
-        sdsfree(cmdname);
-        return REDISMODULE_ERR;
-    }
-
     /* Create a command "proxy", which is a structure that is referenced
      * in the command table, so that the generic command that works as
      * binding between modules and Redis, can know what function to call
@@ -940,7 +950,6 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c
     cp->rediscmd = zmalloc(sizeof(*rediscmd));
     cp->rediscmd->name = cmdname;
     cp->rediscmd->proc = RedisModuleCommandDispatcher;
-    cp->rediscmd->arity = -1;
     cp->rediscmd->flags = flags | CMD_MODULE;
     cp->rediscmd->getkeys_proc = (redisGetKeysProc*)(unsigned long)cp;
     cp->rediscmd->key_specs_max = STATIC_KEY_SPECS_NUM;
@@ -967,12 +976,70 @@ int RM_CreateCommand(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc c
     cp->rediscmd->calls = 0;
     cp->rediscmd->rejected_calls = 0;
     cp->rediscmd->failed_calls = 0;
-    dictAdd(server.commands,sdsdup(cmdname),cp->rediscmd);
-    dictAdd(server.orig_commands,sdsdup(cmdname),cp->rediscmd);
-    cp->rediscmd->id = ACLGetCommandID(cmdname); /* ID used for ACL. */
+    return cp;
+}
+
+/* Very similar to RedisModule_CreateCommand except that it is used to create
+ * a subcommand, associated with another, container, command.
+ *
+ * Example: If a module has a configuration command, MODULE.CONFIG, then
+ * GET and SET should be individual subcommands, while MODULE.CONFIG is
+ * a command, but should not be registered with a valid `funcptr`:
+ *
+ * if (RedisModule_CreateCommand(ctx,"module.config",NULL,"",0,0,0) == REDISMODULE_ERR)
+ *     return REDISMODULE_ERR;
+ *
+ *  if (RedisModule_CreateSubcommand(ctx,"container.config","set",cmd_config_set,"",0,0,0) == REDISMODULE_ERR)
+ *     return REDISMODULE_ERR;
+ *
+ *  if (RedisModule_CreateSubcommand(ctx,"container.config","get",cmd_config_get,"",0,0,0) == REDISMODULE_ERR)
+ *     return REDISMODULE_ERR;
+ *
+ */
+int RM_CreateSubcommand(RedisModuleCtx *ctx, const char *parent_name, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep) {
+    int64_t flags = strflags ? commandFlagsFromString((char*)strflags) : 0;
+    if (flags == -1) return REDISMODULE_ERR;
+    if ((flags & CMD_MODULE_NO_CLUSTER) && server.cluster_enabled)
+        return REDISMODULE_ERR;
+
+    struct redisCommand *parent_cmd = lookupCommandByCString(parent_name);
+
+    if (!parent_cmd || !(parent_cmd->flags & CMD_MODULE))
+        return REDISMODULE_ERR;
+
+    if (parent_cmd->parent)
+        return REDISMODULE_ERR; /* We don't allow more than one level of subcommands */
+
+    RedisModuleCommandProxy *parent_cp = (void*)(unsigned long)parent_cmd->getkeys_proc;
+    if (parent_cp->module != ctx->module)
+        return REDISMODULE_ERR;
+
+    /* Check if the command name is busy within the parent command. */
+    if (parent_cmd->subcommands_dict && lookupCommandByCStringLogic(parent_cmd->subcommands_dict, name) != NULL)
+        return REDISMODULE_ERR;
+
+    RedisModuleCommandProxy *cp = moduleCreateCommandProxy(ctx, name, cmdfunc, flags, firstkey, lastkey, keystep);
+    cp->rediscmd->arity = -2;
+
+    commandAddSubcommand(parent_cmd, cp->rediscmd);
     return REDISMODULE_OK;
 }
 
+/* Return `struct RedisModule *` as `void *` to avoid exposing it outside of module.c. */
+void *moduleGetHandleByName(char *modulename) {
+    return dictFetchValue(modules,modulename);
+}
+
+/* Returns 1 if `cmd` is a command of the module `modulename`. 0 otherwise. */
+int moduleIsModuleCommand(void *module_handle, struct redisCommand *cmd) {
+    if (cmd->proc != RedisModuleCommandDispatcher)
+        return 0;
+    if (module_handle == NULL)
+        return 0;
+    RedisModuleCommandProxy *cp = (void*)(unsigned long)cmd->getkeys_proc;
+    return (cp->module == module_handle);
+}
+
 void extendKeySpecsIfNeeded(struct redisCommand *cmd) {
     /* We extend even if key_specs_num == key_specs_max because
      * this function is called prior to adding a new spec */
@@ -1095,7 +1162,7 @@ int moduleSetCommandKeySpecFindKeys(RedisModuleCtx *ctx, const char *name, int i
  *
  * Example:
  *
- * if (RedisModule_CreateCommand(ctx,"kspec.smove",kspec_legacy,"",0,0,0) == REDISMODULE_ERR)
+ *  if (RedisModule_CreateCommand(ctx,"kspec.smove",kspec_legacy,"",0,0,0) == REDISMODULE_ERR)
  *      return REDISMODULE_ERR;
  *
  *  if (RedisModule_AddCommandKeySpec(ctx,"kspec.smove","read write",&spec_id) == REDISMODULE_ERR)
@@ -1112,6 +1179,11 @@ int moduleSetCommandKeySpecFindKeys(RedisModuleCtx *ctx, const char *name, int i
  *  if (RedisModule_SetCommandKeySpecFindKeysRange(ctx,"kspec.smove",spec_id,0,1,0) == REDISMODULE_ERR)
  *      return REDISMODULE_ERR;
  *
+ * It is also possible to use this API on subcommands (See RedisModule_CreateSubcommand).
+ * The name of the subcommand should be the name of the parent command + "|" + name of subcommand.
+ * Example:
+ *   RedisModule_AddCommandKeySpec(ctx,"module.config|get","read",&spec_id)
+ *
  * Returns REDISMODULE_OK on success
  */
 int RM_AddCommandKeySpec(RedisModuleCtx *ctx, const char *name, const char *specflags, int *spec_id) {
@@ -4858,7 +4930,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
     /* Lookup command now, after filters had a chance to make modifications
      * if necessary.
      */
-    cmd = lookupCommand(c->argv[0]->ptr);
+    cmd = lookupCommand(c->argv,c->argc);
     if (!cmd) {
         errno = ENOENT;
         goto cleanup;
@@ -7373,7 +7445,7 @@ int RM_ACLCheckCommandPermissions(RedisModuleUser *user, RedisModuleString **arg
     struct redisCommand *cmd;
 
     /* Find command */
-    if ((cmd = lookupCommand(argv[0]->ptr)) == NULL) {
+    if ((cmd = lookupCommand(argv, argc)) == NULL) {
         errno = ENOENT;
         return REDISMODULE_ERR;
     }
@@ -9745,8 +9817,7 @@ void moduleCommand(client *c) {
 NULL
         };
         addReplyHelp(c, help);
-    } else
-    if (!strcasecmp(subcmd,"load") && c->argc >= 3) {
+    } else if (!strcasecmp(subcmd,"load") && c->argc >= 3) {
         robj **argv = NULL;
         int argc = 0;
 
@@ -9964,7 +10035,7 @@ int *RM_GetCommandKeys(RedisModuleCtx *ctx, RedisModuleString **argv, int argc,
     int *res = NULL;
 
     /* Find command */
-    if ((cmd = lookupCommand(argv[0]->ptr)) == NULL) {
+    if ((cmd = lookupCommand(argv,argc)) == NULL) {
         errno = ENOENT;
         return NULL;
     }
@@ -10243,6 +10314,7 @@ void moduleRegisterCoreAPI(void) {
     REGISTER_API(Free);
     REGISTER_API(Strdup);
     REGISTER_API(CreateCommand);
+    REGISTER_API(CreateSubcommand);
     REGISTER_API(SetModuleAttribs);
     REGISTER_API(IsModuleNameBusy);
     REGISTER_API(WrongArity);
diff --git a/src/networking.c b/src/networking.c
index 9c1b46052..d514c11a5 100644
--- a/src/networking.c
+++ b/src/networking.c
@@ -2399,7 +2399,8 @@ sds catClientInfoString(sds s, client *client) {
     /* Compute the total memory consumed by this client. */
     size_t obufmem, total_mem = getClientMemoryUsage(client, &obufmem);
 
-    return sdscatfmt(s,
+    sds cmdname = client->lastcmd ? getFullCommandName(client->lastcmd) : NULL;
+    sds ret = sdscatfmt(s,
         "id=%U addr=%s laddr=%s %s name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U argv-mem=%U multi-mem=%U obl=%U oll=%U omem=%U tot-mem=%U events=%s cmd=%s user=%s redir=%I resp=%i",
         (unsigned long long) client->id,
         getClientPeerId(client),
@@ -2422,10 +2423,13 @@ sds catClientInfoString(sds s, client *client) {
         (unsigned long long) obufmem, /* should not include client->buf since we want to see 0 for static clients. */
         (unsigned long long) total_mem,
         events,
-        client->lastcmd ? client->lastcmd->name : "NULL",
+        cmdname ? cmdname : "NULL",
         client->user ? client->user->name : "(superuser)",
         (client->flags & CLIENT_TRACKING) ? (long long) client->client_tracking_redirection : -1,
         client->resp);
+    if (cmdname)
+        sdsfree(cmdname);
+    return ret;
 }
 
 sds getAllClientsInfoString(int type) {
@@ -2568,6 +2572,8 @@ void clientCommand(client *c) {
 "    Control the replies sent to the current connection.",
 "SETNAME <name>",
 "    Assign the name <name> to the current connection.",
+"GETNAME",
+"    Get the name of the current connection.",
 "UNBLOCK <clientid> [TIMEOUT|ERROR]",
 "    Unblock the specified blocked client.",
 "TRACKING (ON|OFF) [REDIRECT <id>] [BCAST] [PREFIX <prefix> [...]]",
@@ -2575,6 +2581,8 @@ void clientCommand(client *c) {
 "    Control server assisted client side caching.",
 "TRACKINGINFO",
 "    Report tracking status for the current connection.",
+"NO-EVICT (ON|OFF)",
+"    Protect current client connection from eviction.",
 NULL
         };
         addReplyHelp(c, help);
@@ -2638,7 +2646,7 @@ NULL
             return;
         }
     } else if (!strcasecmp(c->argv[1]->ptr,"no-evict") && c->argc == 3) {
-        /* CLIENT PROTECT ON|OFF */
+        /* CLIENT NO-EVICT ON|OFF */
         if (!strcasecmp(c->argv[2]->ptr,"on")) {
             c->flags |= CLIENT_NO_EVICT;
             addReply(c,shared.ok);
@@ -3195,7 +3203,7 @@ void replaceClientCommandVector(client *c, int argc, robj **argv) {
     for (j = 0; j < c->argc; j++)
         if (c->argv[j])
             c->argv_len_sum += getStringObjectLen(c->argv[j]);
-    c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr);
+    c->cmd = lookupCommandOrOriginal(c->argv,c->argc);
     serverAssertWithInfo(c,NULL,c->cmd != NULL);
 }
 
@@ -3227,7 +3235,7 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval) {
 
     /* If this is the command name make sure to fix c->cmd. */
     if (i == 0) {
-        c->cmd = lookupCommandOrOriginal(c->argv[0]->ptr);
+        c->cmd = lookupCommandOrOriginal(c->argv,c->argc);
         serverAssertWithInfo(c,NULL,c->cmd != NULL);
     }
 }
diff --git a/src/pubsub.c b/src/pubsub.c
index e0bbc6d94..6da5b18cf 100644
--- a/src/pubsub.c
+++ b/src/pubsub.c
@@ -406,6 +406,11 @@ void punsubscribeCommand(client *c) {
 
 /* PUBLISH <channel> <message> */
 void publishCommand(client *c) {
+    if (server.sentinel_mode) {
+        sentinelPublishCommand(c);
+        return;
+    }
+
     int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
     if (server.cluster_enabled)
         clusterPropagatePublish(c->argv[1],c->argv[2]);
diff --git a/src/redismodule.h b/src/redismodule.h
index 85c963d58..68272e3f1 100644
--- a/src/redismodule.h
+++ b/src/redismodule.h
@@ -604,6 +604,7 @@ REDISMODULE_API void * (*RedisModule_Calloc)(size_t nmemb, size_t size) REDISMOD
 REDISMODULE_API char * (*RedisModule_Strdup)(const char *str) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_GetApi)(const char *, void *) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_CreateCommand)(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep) REDISMODULE_ATTR;
+REDISMODULE_API int (*RedisModule_CreateSubcommand)(RedisModuleCtx *ctx, const char *parent_name, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep) REDISMODULE_ATTR;
 REDISMODULE_API void (*RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, const char *name, int ver, int apiver) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_IsModuleNameBusy)(const char *name) REDISMODULE_ATTR;
 REDISMODULE_API int (*RedisModule_WrongArity)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
@@ -924,6 +925,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(Realloc);
     REDISMODULE_GET_API(Strdup);
     REDISMODULE_GET_API(CreateCommand);
+    REDISMODULE_GET_API(CreateSubcommand);
     REDISMODULE_GET_API(SetModuleAttribs);
     REDISMODULE_GET_API(IsModuleNameBusy);
     REDISMODULE_GET_API(WrongArity);
diff --git a/src/replication.c b/src/replication.c
index dcb407491..eb574d0ea 100644
--- a/src/replication.c
+++ b/src/replication.c
@@ -2812,6 +2812,11 @@ void replicaofCommand(client *c) {
  * (master or slave) and additional information related to replication
  * in an easy to process format. */
 void roleCommand(client *c) {
+    if (server.sentinel_mode) {
+        sentinelRoleCommand(c);
+        return;
+    }
+
     if (server.masterhost == NULL) {
         listIter li;
         listNode *ln;
diff --git a/src/scripting.c b/src/scripting.c
index 5b7b529ff..374ebaf91 100644
--- a/src/scripting.c
+++ b/src/scripting.c
@@ -817,7 +817,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     }
 
     /* Command lookup */
-    cmd = lookupCommand(argv[0]->ptr);
+    cmd = lookupCommand(argv,argc);
     if (!cmd || ((cmd->arity > 0 && cmd->arity != argc) ||
                    (argc < -cmd->arity)))
     {
diff --git a/src/sentinel.c b/src/sentinel.c
index 2a2068eef..dfbb8018f 100644
--- a/src/sentinel.c
+++ b/src/sentinel.c
@@ -456,32 +456,10 @@ dictType renamedCommandsDictType = {
 
 /* =========================== Initialization =============================== */
 
-void sentinelCommand(client *c);
-void sentinelInfoCommand(client *c);
 void sentinelSetCommand(client *c);
-void sentinelPublishCommand(client *c);
-void sentinelRoleCommand(client *c);
 void sentinelConfigGetCommand(client *c);
 void sentinelConfigSetCommand(client *c);
 
-struct redisCommand sentinelcmds[] = {
-    {"ping",pingCommand,1,"fast @connection"},
-    {"sentinel",sentinelCommand,-2,"admin"},
-    {"subscribe",subscribeCommand,-2,"pub-sub"},
-    {"unsubscribe",unsubscribeCommand,-1,"pub-sub"},
-    {"psubscribe",psubscribeCommand,-2,"pub-sub"},
-    {"punsubscribe",punsubscribeCommand,-1,"pub-sub"},
-    {"publish",sentinelPublishCommand,3,"pub-sub fast"},
-    {"info",sentinelInfoCommand,-1,"random @dangerous"},
-    {"role",sentinelRoleCommand,1,"fast read-only @dangerous"},
-    {"client",clientCommand,-2,"admin random @connection"},
-    {"shutdown",shutdownCommand,-1,"admin"},
-    {"auth",authCommand,-2,"no-auth fast @connection"},
-    {"hello",helloCommand,-1,"no-auth fast @connection"},
-    {"acl",aclCommand,-2,"admin"},
-    {"command",commandCommand,-1, "random @connection"}
-};
-
 /* this array is used for sentinel config lookup, which need to be loaded
  * before monitoring masters config to avoid dependency issues */
 const char *preMonitorCfgName[] = { 
@@ -507,28 +485,6 @@ void freeSentinelLoadQueueEntry(void *item);
 
 /* Perform the Sentinel mode initialization. */
 void initSentinel(void) {
-    unsigned int j;
-
-    /* Remove usual Redis commands from the command table, then just add
-     * the SENTINEL command. */
-    dictEmpty(server.commands,NULL);
-    dictEmpty(server.orig_commands,NULL);
-    ACLClearCommandID();
-    for (j = 0; j < sizeof(sentinelcmds)/sizeof(sentinelcmds[0]); j++) {
-        int retval;
-        struct redisCommand *cmd = sentinelcmds+j;
-        cmd->id = ACLGetCommandID(cmd->name); /* Assign the ID used for ACL. */
-        retval = dictAdd(server.commands, sdsnew(cmd->name), cmd);
-        serverAssert(retval == DICT_OK);
-        retval = dictAdd(server.orig_commands, sdsnew(cmd->name), cmd);
-        serverAssert(retval == DICT_OK);
-
-        /* Translate the command string flags description into an actual
-         * set of flags. */
-        if (populateSingleCommand(cmd,cmd->sflags) == C_ERR)
-            serverPanic("Unsupported command flag");
-    }
-
     /* Initialize various data structures. */
     sentinel.current_epoch = 0;
     sentinel.masters = dictCreate(&instancesDictType);
diff --git a/src/server.c b/src/server.c
index 78ef2925f..17672fbd6 100644
--- a/src/server.c
+++ b/src/server.c
@@ -171,6 +171,10 @@ struct redisServer server; /* Server global state */
  *                or may just execute read commands. A command can not be marked 
  *                both "write" and "may-replicate"
  *
+ * sentinel: This command is present in sentinel mode too.
+ *
+ * sentinel-only: This command is present only when in sentinel mode.
+ *
  * The following additional flags are only used in order to put commands
  * in a specific ACL category. Commands can have multiple ACL categories.
  * See redis.conf for the exact meaning of each.
@@ -193,11 +197,499 @@ struct redisServer server; /* Server global state */
  *    TYPE, EXPIRE*, PEXPIRE*, TTL, PTTL, ...
  */
 
+struct redisCommand configSubcommands[] = {
+    {"set",configSetCommand,4,
+     "admin ok-stale no-script"},
 
-struct redisCommand redisCommandTable[] = {
-    {"module",moduleCommand,-2,
+    {"get",configGetCommand,3,
+     "admin ok-loading ok-stale no-script"},
+
+    {"resetstat",configResetStatCommand,2,
+     "admin ok-stale no-script"},
+
+    {"rewrite",configRewriteCommand,2,
+     "admin ok-stale no-script"},
+
+    {"help",configHelpCommand,2,
+     "ok-stale ok-loading"},
+
+    {NULL},
+};
+
+struct redisCommand xinfoSubcommands[] = {
+    {"consumers",xinfoCommand,4,
+     "read-only random @stream",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"groups",xinfoCommand,3,
+     "read-only @stream",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"stream",xinfoCommand,-3,
+     "read-only @stream",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"help",xinfoCommand,2,
+     "ok-stale ok-loading @stream"},
+
+    {NULL},
+};
+
+struct redisCommand xgroupSubcommands[] = {
+    {"create",xgroupCommand,-5,
+     "write use-memory @stream",
+     {{"write",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"setid",xgroupCommand,5,
+     "write @stream",
+     {{"write",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"destroy",xgroupCommand,4,
+     "write @stream",
+     {{"write",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"createconsumer",xgroupCommand,5,
+     "write use-memory @stream",
+     {{"write",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"delconsumer",xgroupCommand,5,
+     "write @stream",
+     {{"write",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"help",xgroupCommand,2,
+     "ok-stale ok-loading @stream"},
+
+    {NULL},
+};
+
+struct redisCommand commandSubcommands[] = {
+    {"count",commandCountCommand,2,
+     "ok-loading ok-stale @connection"},
+
+    {"list",commandListCommand,-2,
+     "ok-loading ok-stale @connection"},
+
+    {"info",commandInfoCommand,-3,
+     "ok-loading ok-stale @connection"},
+
+    {"getkeys",commandGetKeysCommand,-4,
+     "ok-loading ok-stale @connection"},
+
+    {"help",commandHelpCommand,2,
+     "ok-loading ok-stale @connection"},
+
+    {NULL},
+};
+
+struct redisCommand memorySubcommands[] = {
+    {"doctor",memoryCommand,2,
+     "random"},
+
+    {"stats",memoryCommand,2,
+     "random"},
+
+    {"malloc-stats",memoryCommand,2,
+     "random"},
+
+    {"purge",memoryCommand,2,
+     ""},
+
+    {"usage",memoryCommand,-3,
+     "read-only",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"help",memoryCommand,2,
+     "ok-stale ok-loading"},
+
+    {NULL},
+};
+
+struct redisCommand aclSubcommands[] = {
+    {"cat",aclCommand,-2,
+     "no-script ok-loading ok-stale sentinel"},
+
+    {"deluser",aclCommand,-3,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"genpass",aclCommand,-2,
+     "no-script ok-loading ok-stale sentinel"},
+
+    {"getuser",aclCommand,3,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"list",aclCommand,2,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"load",aclCommand,2,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"log",aclCommand,-2,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"save",aclCommand,2,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"setuser",aclCommand,-3,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"users",aclCommand,2,
+     "admin no-script ok-loading ok-stale sentinel"},
+
+    {"whoami",aclCommand,2,
+     "no-script ok-loading ok-stale sentinel"},
+
+    {"help",aclCommand,2,
+     "ok-stale ok-loading sentinel"},
+
+    {NULL},
+};
+
+struct redisCommand latencySubcommands[] = {
+    {"doctor",latencyCommand,2,
+     "admin no-script ok-loading ok-stale"},
+
+    {"graph",latencyCommand,3,
+     "admin no-script ok-loading ok-stale"},
+
+    {"history",latencyCommand,3,
+     "admin no-script ok-loading ok-stale"},
+
+    {"latest",latencyCommand,2,
+     "admin no-script ok-loading ok-stale"},
+
+    {"reset",latencyCommand,-2,
+     "admin no-script ok-loading ok-stale"},
+
+    {"help",latencyCommand,2,
+     "ok-stale ok-loading"},
+
+    {NULL},
+};
+
+struct redisCommand moduleSubcommands[] = {
+    {"list",moduleCommand,2,
      "admin no-script"},
 
+    {"load",moduleCommand,-3,
+     "admin no-script"},
+
+    {"unload",moduleCommand,3,
+     "admin no-script"},
+
+    {"help",moduleCommand,2,
+     "ok-stale ok-loading"},
+
+    {NULL},
+};
+
+struct redisCommand slowlogSubcommands[] = {
+    {"get",slowlogCommand,-2,
+     "admin random ok-loading ok-stale"},
+
+    {"len",slowlogCommand,2,
+     "admin random ok-loading ok-stale"},
+
+    {"reset",slowlogCommand,2,
+     "admin ok-loading ok-stale"},
+
+    {"help",slowlogCommand,2,
+     "ok-stale ok-loading"},
+
+    {NULL},
+};
+
+struct redisCommand objectSubcommands[] = {
+    {"encoding",objectCommand,3,
+     "read-only @keyspace",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"freq",objectCommand,3,
+     "read-only random @keyspace",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"idletime",objectCommand,3,
+     "read-only random @keyspace",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"refcount",objectCommand,3,
+     "read-only @keyspace",
+     {{"read",
+       KSPEC_BS_INDEX,.bs.index={2},
+       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+
+    {"help",objectCommand,2,
+     "ok-stale ok-loading @keyspace"},
+
+    {NULL},
+};
+
+struct redisCommand scriptSubcommands[] = {
+    {"debug",scriptCommand,3,
+     "no-script @scripting"},
+
+    {"exists",scriptCommand,-3,
+     "no-script @scripting"},
+
+    {"flush",scriptCommand,-2,
+     "may-replicate no-script @scripting"},
+
+    {"kill",scriptCommand,2,
+     "no-script @scripting"},
+
+    {"load",scriptCommand,3,
+     "may-replicate no-script @scripting"},
+
+    {"help",scriptCommand,2,
+     "ok-loading ok-stale @scripting"},
+
+    {NULL},
+};
+
+struct redisCommand clientSubcommands[] = {
+    {"caching",clientCommand,3,
+     "no-script ok-loading ok-stale @connection"},
+
+    {"getredir",clientCommand,2,
+     "no-script ok-loading ok-stale @connection"},
+
+    {"id",clientCommand,2,
+     "no-script ok-loading ok-stale @connection"},
+
+    {"info",clientCommand,2,
+     "no-script random ok-loading ok-stale @connection"},
+
+    {"kill",clientCommand,-3,
+     "admin no-script ok-loading ok-stale @connection"},
+
+    {"list",clientCommand,-2,
+     "admin no-script random ok-loading ok-stale @connection"},
+
+    {"unpause",clientCommand,2,
+     "admin no-script ok-loading ok-stale @connection"},
+
+    {"pause",clientCommand,-3,
+     "admin no-script ok-loading ok-stale @connection"},
+
+    {"reply",clientCommand,3,
+     "no-script ok-loading ok-stale @connection"},
+
+    {"setname",clientCommand,3,
+     "no-script ok-loading ok-stale @connection"},
+
+    {"getname",clientCommand,2,
+     "no-script ok-loading ok-stale @connection"},
+
+    {"unblock",clientCommand,-3,
+     "admin no-script ok-loading ok-stale @connection"},
+
+    {"tracking",clientCommand,-3,
+     "no-script ok-loading ok-stale @connection"},
+
+    {"trackinginfo",clientCommand,2,
+     "admin no-script ok-loading ok-stale @connection"},
+
+    {"no-evict",clientCommand,3,
+     "admin no-script ok-loading ok-stale @connection"},
+
+    {"help",clientCommand,2,
+     "ok-loading ok-stale @connection"},
+
+    {NULL},
+};
+
+struct redisCommand stralgoSubcommands[] = {
+    {"lcs",stralgoCommand,-5,
+     "read-only @string",
+      {{"read incomplete", /* We can't use "keyword" here because we may give false information. */
+        KSPEC_BS_UNKNOWN,{{0}},
+        KSPEC_FK_UNKNOWN,{{0}}}},
+     lcsGetKeys},
+
+    {"help",stralgoCommand,2,
+     "ok-loading ok-stale @string"},
+
+    {NULL},
+};
+
+struct redisCommand pubsubSubcommands[] = {
+    {"channels",pubsubCommand,-2,
+     "pub-sub ok-loading ok-stale"},
+
+    {"numpat",pubsubCommand,2,
+     "pub-sub ok-loading ok-stale"},
+
+    {"numsub",pubsubCommand,-2,
+     "pub-sub ok-loading ok-stale"},
+
+    {"help",pubsubCommand,2,
+     "ok-loading ok-stale"},
+
+    {NULL},
+};
+
+struct redisCommand clusterSubcommands[] = {
+    {"addslots",clusterCommand,-3,
+     "admin ok-stale random"},
+
+    {"bumpepoch",clusterCommand,2,
+     "admin ok-stale random"},
+
+    {"count-failure-reports",clusterCommand,3,
+     "admin ok-stale random"},
+
+    {"countkeysinslots",clusterCommand,3,
+     "ok-stale random"},
+
+    {"delslots",clusterCommand,-3,
+     "admin ok-stale random"},
+
+    {"failover",clusterCommand,-2,
+     "admin ok-stale random"},
+
+    {"forget",clusterCommand,3,
+     "admin ok-stale random"},
+
+    {"getkeysinslot",clusterCommand,4,
+     "ok-stale random"},
+
+    {"flushslots",clusterCommand,2,
+     "admin ok-stale random"},
+
+    {"info",clusterCommand,2,
+     "ok-stale random"},
+
+    {"keyslot",clusterCommand,3,
+     "ok-stale random"},
+
+    {"meet",clusterCommand,-4,
+     "admin ok-stale random"},
+
+    {"myid",clusterCommand,2,
+     "ok-stale random"},
+
+    {"nodes",clusterCommand,2,
+     "ok-stale random"},
+
+    {"replicate",clusterCommand,3,
+     "admin ok-stale random"},
+
+    {"reset",clusterCommand,3,
+     "admin ok-stale random"},
+
+    {"set-config-epoch",clusterCommand,3,
+     "admin ok-stale random"},
+
+    {"setslot",clusterCommand,-4,
+     "admin ok-stale random"},
+
+    {"replicas",clusterCommand,3,
+     "admin ok-stale random"},
+
+    {"saveconfig",clusterCommand,2,
+     "admin ok-stale random"},
+
+    {"slots",clusterCommand,2,
+     "ok-stale random"},
+
+    {"help",clusterCommand,2,
+     "ok-loading ok-stale"},
+
+    {NULL},
+};
+
+struct redisCommand sentinelSubcommands[] = {
+    {"ckquorum",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"config",sentinelCommand,-3,
+     "admin only-sentinel"},
+
+    {"debug",sentinelCommand,2,
+     "admin only-sentinel"},
+
+    {"get-master-addr-by-name",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"failover",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"flushconfig",sentinelCommand,2,
+     "admin only-sentinel"},
+
+    {"info-cache",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"is-master-down-by-addr",sentinelCommand,6,
+     "admin only-sentinel"},
+
+    {"master",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"masters",sentinelCommand,2,
+     "admin only-sentinel"},
+
+    {"monitor",sentinelCommand,6,
+     "admin only-sentinel"},
+
+    {"myid",sentinelCommand,2,
+     "admin only-sentinel"},
+
+    {"pending-scripts",sentinelCommand,2,
+     "admin only-sentinel"},
+
+    {"remove",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"replicas",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"reset",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"sentinels",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"set",sentinelCommand,5,
+     "admin only-sentinel"},
+
+    {"simulate-failure",sentinelCommand,3,
+     "admin only-sentinel"},
+
+    {"help",sentinelCommand,2,
+     "ok-loading ok-stale only-sentinel"},
+
+    {NULL},
+};
+
+struct redisCommand redisCommandTable[] = {
+    {"module",NULL,-2,
+     "",
+     .subcommands=moduleSubcommands},
+
     {"get",getCommand,2,
      "read-only fast @string",
      {{"read",
@@ -1028,13 +1520,17 @@ struct redisCommand redisCommandTable[] = {
      "read-only fast @keyspace"},
 
     {"auth",authCommand,-2,
-     "no-auth no-script ok-loading ok-stale fast @connection"},
+     "no-auth no-script ok-loading ok-stale fast sentinel @connection"},
 
     /* We don't allow PING during loading since in Redis PING is used as
      * failure detection, and a loading server is considered to be
      * not available. */
     {"ping",pingCommand,-1,
-     "ok-stale fast @connection"},
+     "ok-stale fast sentinel @connection"},
+
+    {"sentinel",NULL,-2,
+     "admin only-sentinel",
+     .subcommands=sentinelSubcommands},
 
     {"echo",echoCommand,2,
      "fast @connection"},
@@ -1049,7 +1545,7 @@ struct redisCommand redisCommandTable[] = {
      "admin no-script"},
 
     {"shutdown",shutdownCommand,-1,
-     "admin no-script ok-loading ok-stale"},
+     "admin no-script ok-loading ok-stale sentinel"},
 
     {"lastsave",lastsaveCommand,1,
      "random fast ok-loading ok-stale @admin @dangerous"},
@@ -1101,7 +1597,7 @@ struct redisCommand redisCommandTable[] = {
        KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
 
     {"info",infoCommand,-1,
-     "ok-loading ok-stale random @dangerous"},
+     "ok-loading ok-stale random sentinel @dangerous"},
 
     {"monitor",monitorCommand,1,
      "admin no-script ok-loading ok-stale"},
@@ -1149,31 +1645,33 @@ struct redisCommand redisCommandTable[] = {
      "admin no-script ok-stale"},
 
     {"role",roleCommand,1,
-     "ok-loading ok-stale no-script fast @admin @dangerous"},
+     "ok-loading ok-stale no-script fast sentinel @admin @dangerous"},
 
     {"debug",debugCommand,-2,
      "admin no-script ok-loading ok-stale"},
 
-    {"config",configCommand,-2,
-     "admin ok-loading ok-stale no-script"},
+    {"config",NULL,-2,
+     "",
+     .subcommands=configSubcommands},
 
     {"subscribe",subscribeCommand,-2,
-     "pub-sub no-script ok-loading ok-stale"},
+     "pub-sub no-script ok-loading ok-stale sentinel"},
 
     {"unsubscribe",unsubscribeCommand,-1,
-     "pub-sub no-script ok-loading ok-stale"},
+     "pub-sub no-script ok-loading ok-stale sentinel"},
 
     {"psubscribe",psubscribeCommand,-2,
-     "pub-sub no-script ok-loading ok-stale"},
+     "pub-sub no-script ok-loading ok-stale sentinel"},
 
     {"punsubscribe",punsubscribeCommand,-1,
-     "pub-sub no-script ok-loading ok-stale"},
+     "pub-sub no-script ok-loading ok-stale sentinel"},
 
     {"publish",publishCommand,3,
-     "pub-sub ok-loading ok-stale fast may-replicate"},
+     "pub-sub ok-loading ok-stale fast may-replicate sentinel"},
 
-    {"pubsub",pubsubCommand,-2,
-     "pub-sub ok-loading ok-stale random"},
+    {"pubsub",NULL,-2,
+     "",
+     .subcommands=pubsubSubcommands},
 
     {"watch",watchCommand,-2,
      "no-script fast ok-loading ok-stale @transaction",
@@ -1184,8 +1682,9 @@ struct redisCommand redisCommandTable[] = {
     {"unwatch",unwatchCommand,1,
      "no-script fast ok-loading ok-stale @transaction"},
 
-    {"cluster",clusterCommand,-2,
-     "admin ok-stale random"},
+    {"cluster",NULL,-2,
+     "",
+     .subcommands=clusterSubcommands},
 
     {"restore",restoreCommand,-4,
      "write use-memory @keyspace @dangerous",
@@ -1224,24 +1723,20 @@ struct redisCommand redisCommandTable[] = {
        KSPEC_BS_INDEX,.bs.index={1},
        KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
 
-    {"object",objectCommand,-2,
-     "read-only random @keyspace",
-     {{"read",
-       KSPEC_BS_INDEX,.bs.index={2},
-       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+    {"object",NULL,-2,
+     "",
+    .subcommands=objectSubcommands},
 
-    {"memory",memoryCommand,-2,
-     "random read-only",
-     {{"read",
-       KSPEC_BS_KEYWORD,.bs.keyword={"USAGE",1},
-       KSPEC_FK_RANGE,.fk.range={0,1,0}}},
-     memoryGetKeys},
+    {"memory",NULL,-2,
+     "",
+     .subcommands=memorySubcommands},
 
-    {"client",clientCommand,-2,
-     "admin no-script random ok-loading ok-stale @connection"},
+    {"client",NULL,-2,
+     "sentinel",
+     .subcommands=clientSubcommands},
 
     {"hello",helloCommand,-1,
-     "no-auth no-script fast ok-loading ok-stale @connection"},
+     "no-auth no-script fast ok-loading ok-stale sentinel @connection"},
 
     /* EVAL can modify the dataset, however it is not flagged as a write
      * command since we do the check while running commands from Lua.
@@ -1277,11 +1772,13 @@ struct redisCommand redisCommandTable[] = {
        KSPEC_FK_KEYNUM,.fk.keynum={0,1,1}}},
      evalGetKeys},
 
-    {"slowlog",slowlogCommand,-2,
-     "admin random ok-loading ok-stale"},
+    {"slowlog",NULL,-2,
+     "",
+     .subcommands=slowlogSubcommands},
 
-    {"script",scriptCommand,-2,
-     "no-script may-replicate @scripting"},
+    {"script",NULL,-2,
+     "",
+     .subcommands=scriptSubcommands},
 
     {"time",timeCommand,1,
      "random fast ok-loading ok-stale"},
@@ -1311,7 +1808,8 @@ struct redisCommand redisCommandTable[] = {
      "no-script @connection"},
 
     {"command",commandCommand,-1,
-     "ok-loading ok-stale random @connection"},
+     "ok-loading ok-stale random sentinel @connection",
+     .subcommands=commandSubcommands},
 
     {"geoadd",geoaddCommand,-5,
      "write use-memory @geo",
@@ -1465,11 +1963,9 @@ struct redisCommand redisCommandTable[] = {
        KSPEC_FK_RANGE,.fk.range={-1,1,2}}},
      xreadGetKeys},
 
-    {"xgroup",xgroupCommand,-2,
-     "write use-memory @stream",
-     {{"write",
-       KSPEC_BS_INDEX,.bs.index={2},
-       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+    {"xgroup",NULL,-2,
+     "",
+     .subcommands=xgroupSubcommands},
 
     {"xsetid",xsetidCommand,3,
      "write use-memory fast @stream",
@@ -1501,11 +1997,9 @@ struct redisCommand redisCommandTable[] = {
        KSPEC_BS_INDEX,.bs.index={1},
        KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
 
-    {"xinfo",xinfoCommand,-2,
-     "read-only random @stream",
-     {{"write",
-       KSPEC_BS_INDEX,.bs.index={2},
-       KSPEC_FK_RANGE,.fk.range={0,1,0}}}},
+    {"xinfo",NULL,-2,
+     "",
+     .subcommands=xinfoSubcommands},
 
     {"xdel",xdelCommand,-3,
      "write fast @stream",
@@ -1525,21 +2019,20 @@ struct redisCommand redisCommandTable[] = {
     {"host:",securityWarningCommand,-1,
      "ok-loading ok-stale read-only"},
 
-    {"latency",latencyCommand,-2,
-     "admin no-script ok-loading ok-stale"},
+    {"latency",NULL,-2,
+     "",
+     .subcommands=latencySubcommands},
 
     {"lolwut",lolwutCommand,-1,
      "read-only fast"},
 
-    {"acl",aclCommand,-2,
-     "admin no-script ok-loading ok-stale"},
+    {"acl",NULL,-2,
+     "",
+     .subcommands=aclSubcommands},
 
-    {"stralgo",stralgoCommand,-2,
-     "read-only @string",
-      {{"read incomplete", /* We can't use "keyword" here because we may give false information. */
-        KSPEC_BS_UNKNOWN,{{0}},
-        KSPEC_FK_UNKNOWN,{{0}}}},
-     lcsGetKeys},
+    {"stralgo",NULL,-2,
+     "",
+     .subcommands=stralgoSubcommands},
 
     {"reset",resetCommand,1,
      "no-script ok-stale ok-loading fast @connection"},
@@ -3942,6 +4435,18 @@ void populateCommandLegacyRangeSpec(struct redisCommand *c) {
     c->legacy_range_key_spec.fk.range.limit = 0;
 }
 
+void commandAddSubcommand(struct redisCommand *parent, struct redisCommand *subcommand) {
+    if (!parent->subcommands_dict)
+        parent->subcommands_dict = dictCreate(&commandTableDictType);
+
+    subcommand->parent = parent; /* Assign the parent command */
+    sds fullname = getFullCommandName(subcommand);
+    subcommand->id = ACLGetCommandID(fullname); /* Assign the ID used for ACL. */
+    sdsfree(fullname);
+
+    serverAssert(dictAdd(parent->subcommands_dict, sdsnew(subcommand->name), subcommand) == DICT_OK);
+}
+
 /* Parse the flags string description 'strflags' and set them to the
  * command 'c'. If the flags are all valid C_OK is returned, otherwise
  * C_ERR is returned (yet the recognized flags are set in the command). */
@@ -3987,6 +4492,10 @@ int populateSingleCommand(struct redisCommand *c, char *strflags) {
             c->flags |= CMD_NO_AUTH;
         } else if (!strcasecmp(flag,"may-replicate")) {
             c->flags |= CMD_MAY_REPLICATE;
+        } else if (!strcasecmp(flag,"sentinel")) {
+            c->flags |= CMD_SENTINEL;
+        } else if (!strcasecmp(flag,"only-sentinel")) {
+            c->flags |= CMD_ONLY_SENTINEL;
         } else {
             /* Parse ACL categories here if the flag name starts with @. */
             uint64_t catflag;
@@ -4041,6 +4550,20 @@ int populateSingleCommand(struct redisCommand *c, char *strflags) {
     /* Handle the "movablekeys" flag (must be done after populating all key specs). */
     populateCommandMovableKeys(c);
 
+    /* Handle subcommands */
+    if (c->subcommands) {
+        for (int j = 0; c->subcommands[j].name; j++) {
+            struct redisCommand *sub = c->subcommands+j;
+
+            /* Translate the command string flags description into an actual
+             * set of flags. */
+            if (populateSingleCommand(sub,sub->sflags) == C_ERR)
+                serverPanic("Unsupported command flag or key spec flag");
+
+            commandAddSubcommand(c,sub);
+        }
+    }
+
     return C_OK;
 }
 
@@ -4052,14 +4575,23 @@ void populateCommandTable(void) {
 
     for (j = 0; j < numcommands; j++) {
         struct redisCommand *c = redisCommandTable+j;
+
+        if (!(c->flags & CMD_SENTINEL) && server.sentinel_mode)
+            continue;
+
+        if (c->flags & CMD_ONLY_SENTINEL && !server.sentinel_mode)
+            continue;
+
         int retval1, retval2;
 
+        /* Assign the ID used for ACL. */
+        c->id = ACLGetCommandID(c->name);
+
         /* Translate the command string flags description into an actual
          * set of flags. */
         if (populateSingleCommand(c,c->sflags) == C_ERR)
             serverPanic("Unsupported command flag or key spec flag");
 
-        c->id = ACLGetCommandID(c->name); /* Assign the ID used for ACL. */
         retval1 = dictAdd(server.commands, sdsnew(c->name), c);
         /* Populate an additional dictionary that will be unaffected
          * by rename-command statements in redis.conf. */
@@ -4068,18 +4600,20 @@ void populateCommandTable(void) {
     }
 }
 
-void resetCommandTableStats(void) {
+void resetCommandTableStats(dict* commands) {
     struct redisCommand *c;
     dictEntry *de;
     dictIterator *di;
 
-    di = dictGetSafeIterator(server.commands);
+    di = dictGetSafeIterator(commands);
     while((de = dictNext(di)) != NULL) {
         c = (struct redisCommand *) dictGetVal(de);
         c->microseconds = 0;
         c->calls = 0;
         c->rejected_calls = 0;
         c->failed_calls = 0;
+        if (c->subcommands_dict)
+            resetCommandTableStats(c->subcommands_dict);
     }
     dictReleaseIterator(di);
 
@@ -4127,19 +4661,62 @@ void redisOpArrayFree(redisOpArray *oa) {
 
 /* ====================== Commands lookup and execution ===================== */
 
-struct redisCommand *lookupCommand(sds name) {
-    return dictFetchValue(server.commands, name);
+struct redisCommand *lookupCommandLogic(dict *commands, robj **argv, int argc) {
+    struct redisCommand *base_cmd = dictFetchValue(commands, argv[0]->ptr);
+    int has_subcommands = base_cmd && base_cmd->subcommands_dict;
+    if (argc == 1 || !has_subcommands) {
+        /* Note: It is possible that base_cmd->proc==NULL (e.g. CONFIG) */
+        return base_cmd;
+    } else {
+        /* Note: Currently we support just one level of subcommands */
+        return dictFetchValue(base_cmd->subcommands_dict, argv[1]->ptr);
+    }
 }
 
-struct redisCommand *lookupCommandByCString(const char *s) {
+struct redisCommand *lookupCommand(robj **argv, int argc) {
+    return lookupCommandLogic(server.commands,argv,argc);
+}
+
+struct redisCommand *lookupCommandBySdsLogic(dict *commands, sds s) {
+    int argc, j;
+    sds *strings = sdssplitlen(s,sdslen(s),"|",1,&argc);
+    if (strings == NULL)
+        return NULL;
+    if (argc > 2) {
+        /* Currently we support just one level of subcommands */
+        sdsfreesplitres(strings,argc);
+        return NULL;
+    }
+
+    robj objects[argc];
+    robj *argv[argc];
+    for (j = 0; j < argc; j++) {
+        initStaticStringObject(objects[j],strings[j]);
+        argv[j] = &objects[j];
+    }
+
+    struct redisCommand *cmd = lookupCommandLogic(commands,argv,argc);
+    sdsfreesplitres(strings,argc);
+    return cmd;
+}
+
+struct redisCommand *lookupCommandBySds(sds s) {
+    return lookupCommandBySdsLogic(server.commands,s);
+}
+
+struct redisCommand *lookupCommandByCStringLogic(dict *commands, const char *s) {
     struct redisCommand *cmd;
     sds name = sdsnew(s);
 
-    cmd = dictFetchValue(server.commands, name);
+    cmd = lookupCommandBySdsLogic(commands,name);
     sdsfree(name);
     return cmd;
 }
 
+struct redisCommand *lookupCommandByCString(const char *s) {
+    return lookupCommandByCStringLogic(server.commands,s);
+}
+
 /* Lookup the command in the current table, if not found also check in
  * the original table containing the original command names unaffected by
  * redis.conf rename-command statement.
@@ -4147,10 +4724,10 @@ struct redisCommand *lookupCommandByCString(const char *s) {
  * This is used by functions rewriting the argument vector such as
  * rewriteClientCommandVector() in order to set client->cmd pointer
  * correctly even if the command was renamed. */
-struct redisCommand *lookupCommandOrOriginal(sds name) {
-    struct redisCommand *cmd = dictFetchValue(server.commands, name);
+struct redisCommand *lookupCommandOrOriginal(robj **argv ,int argc) {
+    struct redisCommand *cmd = lookupCommandLogic(server.commands, argv, argc);
 
-    if (!cmd) cmd = dictFetchValue(server.orig_commands,name);
+    if (!cmd) cmd = lookupCommandLogic(server.orig_commands, argv, argc);
     return cmd;
 }
 
@@ -4608,19 +5185,26 @@ int processCommand(client *c) {
 
     /* Now lookup the command and check ASAP about trivial error conditions
      * such as wrong arity, bad command name and so forth. */
-    c->cmd = c->lastcmd = lookupCommand(c->argv[0]->ptr);
+    c->cmd = c->lastcmd = lookupCommand(c->argv,c->argc);
     if (!c->cmd) {
+        if (lookupCommandBySds(c->argv[0]->ptr)) {
+            /* If we can't find the command but argv[0] by itself is a command
+             * it means we're dealing with an invalid subcommand. Print Help. */
+            addReplySubcommandSyntaxError(c);
+            return C_OK;
+        }
         sds args = sdsempty();
         int i;
         for (i=1; i < c->argc && sdslen(args) < 128; i++)
-            args = sdscatprintf(args, "`%.*s`, ", 128-(int)sdslen(args), (char*)c->argv[i]->ptr);
-        rejectCommandFormat(c,"unknown command `%s`, with args beginning with: %s",
+            args = sdscatprintf(args, "'%.*s' ", 128-(int)sdslen(args), (char*)c->argv[i]->ptr);
+        rejectCommandFormat(c,"unknown command '%s', with args beginning with: %s",
             (char*)c->argv[0]->ptr, args);
         sdsfree(args);
         return C_OK;
     } else if ((c->cmd->arity > 0 && c->cmd->arity != c->argc) ||
-               (c->argc < -c->cmd->arity)) {
-        rejectCommandFormat(c,"wrong number of arguments for '%s' command",
+               (c->argc < -c->cmd->arity))
+    {
+        rejectCommandFormat(c,"wrong number of arguments for '%s' command or subcommand",
             c->cmd->name);
         return C_OK;
     }
@@ -5134,6 +5718,7 @@ void addReplyFlagsForCommand(client *c, struct redisCommand *cmd) {
     flagcount += addReplyCommandFlag(c,cmd->flags,CMD_FAST, "fast");
     flagcount += addReplyCommandFlag(c,cmd->flags,CMD_NO_AUTH, "no_auth");
     flagcount += addReplyCommandFlag(c,cmd->flags,CMD_MAY_REPLICATE, "may_replicate");
+    /* "sentinel" and "only-sentinel" are hidden on purpose. */
     if (cmd->movablekeys) {
         addReplyStatus(c, "movablekeys");
         flagcount += 1;
@@ -5238,6 +5823,24 @@ void addReplyCommandKeyArgs(client *c, struct redisCommand *cmd) {
     }
 }
 
+void addReplyCommand(client *c, struct redisCommand *cmd);
+
+void addReplyCommandSubCommands(client *c, struct redisCommand *cmd) {
+    if (!cmd->subcommands_dict) {
+        addReplySetLen(c, 0);
+        return;
+    }
+
+    addReplyArrayLen(c, dictSize(cmd->subcommands_dict));
+    dictEntry *de;
+    dictIterator *di = dictGetSafeIterator(cmd->subcommands_dict);
+    while((de = dictNext(di)) != NULL) {
+        struct redisCommand *sub = (struct redisCommand *)dictGetVal(de);
+        addReplyCommand(c,sub);
+    }
+    dictReleaseIterator(di);
+}
+
 /* Output the representation of a Redis command. Used by the COMMAND command. */
 void addReplyCommand(client *c, struct redisCommand *cmd) {
     if (!cmd) {
@@ -5251,8 +5854,8 @@ void addReplyCommand(client *c, struct redisCommand *cmd) {
                 lastkey += firstkey;
             keystep = cmd->legacy_range_key_spec.fk.range.keystep;
         }
-        /* We are adding: command name, arg count, flags, first, last, offset, categories, key args */
-        addReplyArrayLen(c, 8);
+        /* We are adding: command name, arg count, flags, first, last, offset, categories, key args, subcommands */
+        addReplyArrayLen(c, 9);
         addReplyBulkCString(c, cmd->name);
         addReplyLongLong(c, cmd->arity);
         addReplyFlagsForCommand(c, cmd);
@@ -5261,70 +5864,188 @@ void addReplyCommand(client *c, struct redisCommand *cmd) {
         addReplyLongLong(c, keystep);
         addReplyCommandCategories(c,cmd);
         addReplyCommandKeyArgs(c,cmd);
+        addReplyCommandSubCommands(c,cmd);
     }
 }
 
-/* COMMAND <subcommand> <args> */
+/* Helper for COMMAND(S) command
+ *
+ * COMMAND(S) GETKEYS arg0 arg1 arg2 ... */
+void getKeysSubcommand(client *c) {
+    struct redisCommand *cmd = lookupCommand(c->argv+2,c->argc-2);
+    getKeysResult result = GETKEYS_RESULT_INIT;
+    int j;
+
+    if (!cmd) {
+        addReplyError(c,"Invalid command specified");
+        return;
+    } else if (cmd->getkeys_proc == NULL && cmd->key_specs_num == 0) {
+        addReplyError(c,"The command has no key arguments");
+        return;
+    } else if ((cmd->arity > 0 && cmd->arity != c->argc-2) ||
+               ((c->argc-2) < -cmd->arity))
+    {
+        addReplyError(c,"Invalid number of arguments specified for command");
+        return;
+    }
+
+    if (!getKeysFromCommand(cmd,c->argv+2,c->argc-2,&result)) {
+        addReplyError(c,"Invalid arguments specified for command");
+    } else {
+        addReplyArrayLen(c,result.numkeys);
+        for (j = 0; j < result.numkeys; j++) addReplyBulk(c,c->argv[result.keys[j]+2]);
+    }
+    getKeysFreeResult(&result);
+}
+
+/* COMMAND (no args) */
 void commandCommand(client *c) {
     dictIterator *di;
     dictEntry *de;
 
-    if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
-        const char *help[] = {
+    addReplyArrayLen(c, dictSize(server.commands));
+    di = dictGetIterator(server.commands);
+    while ((de = dictNext(di)) != NULL) {
+        addReplyCommand(c, dictGetVal(de));
+    }
+    dictReleaseIterator(di);
+}
+
+/* COMMAND COUNT */
+void commandCountCommand(client *c) {
+    addReplyLongLong(c, dictSize(server.commands));
+}
+
+typedef enum {
+    COMMAND_LIST_FILTER_MODULE,
+    COMMAND_LIST_FILTER_ACLCAT,
+    COMMAND_LIST_FILTER_PATTERN,
+} commandListFilterType;
+
+typedef struct {
+    commandListFilterType type;
+    sds arg;
+    struct {
+        int valid;
+        union {
+            uint64_t aclcat;
+            void *module_handle;
+        } u;
+    } cache;
+} commandListFilter;
+
+int shouldFilterFromCommandList(struct redisCommand *cmd, commandListFilter *filter) {
+    switch (filter->type) {
+        case (COMMAND_LIST_FILTER_MODULE):
+            if (!filter->cache.valid) {
+                filter->cache.u.module_handle = moduleGetHandleByName(filter->arg);
+                filter->cache.valid = 1;
+            }
+            return !moduleIsModuleCommand(filter->cache.u.module_handle, cmd);
+        case (COMMAND_LIST_FILTER_ACLCAT): {
+            if (!filter->cache.valid) {
+                filter->cache.u.aclcat = ACLGetCommandCategoryFlagByName(filter->arg);
+                filter->cache.valid = 1;
+            }
+            uint64_t cat = filter->cache.u.aclcat;
+            if (cat == 0)
+                return 1; /* Invalid ACL category */
+            return (!(cmd->flags & cat));
+            break;
+        }
+        case (COMMAND_LIST_FILTER_PATTERN):
+            return !stringmatchlen(filter->arg, sdslen(filter->arg), cmd->name, strlen(cmd->name), 1);
+        default:
+            serverPanic("Invalid filter type %d", filter->type);
+    }
+}
+
+/* COMMAND LIST [FILTERBY (MODULE <module-name>|ACLCAT <cat>|PATTERN <pattern>)] */
+void commandListCommand(client *c) {
+
+    /* Parse options. */
+    int i = 2, got_filter = 0;
+    commandListFilter filter = {0};
+    for (; i < c->argc; i++) {
+        int moreargs = (c->argc-1) - i; /* Number of additional arguments. */
+        char *opt = c->argv[i]->ptr;
+        if (!strcasecmp(opt,"filterby") && moreargs == 2) {
+            char *filtertype = c->argv[i+1]->ptr;
+            if (!strcasecmp(filtertype,"module")) {
+                filter.type = COMMAND_LIST_FILTER_MODULE;
+            } else if (!strcasecmp(filtertype,"aclcat")) {
+                filter.type = COMMAND_LIST_FILTER_ACLCAT;
+            } else if (!strcasecmp(filtertype,"pattern")) {
+                filter.type = COMMAND_LIST_FILTER_PATTERN;
+            } else {
+                addReplyErrorObject(c,shared.syntaxerr);
+                return;
+            }
+            got_filter = 1;
+            filter.arg = c->argv[i+2]->ptr;
+            i += 2;
+        } else {
+            addReplyErrorObject(c,shared.syntaxerr);
+            return;
+        }
+    }
+
+    dictIterator *di;
+    dictEntry *de;
+
+    di = dictGetIterator(server.commands);
+    if (!got_filter) {
+        addReplySetLen(c, dictSize(server.commands));
+        while ((de = dictNext(di)) != NULL) {
+            struct redisCommand *cmd = dictGetVal(de);
+            addReplyBulkCString(c,cmd->name);
+        }
+    } else {
+        int numcmds = 0;
+        void *replylen = addReplyDeferredLen(c);
+        while ((de = dictNext(di)) != NULL) {
+            struct redisCommand *cmd = dictGetVal(de);
+            if (!shouldFilterFromCommandList(cmd,&filter)) {
+                addReplyBulkCString(c,cmd->name);
+                numcmds++;
+            }
+        }
+        setDeferredArrayLen(c,replylen,numcmds);
+    }
+    dictReleaseIterator(di);
+}
+
+/* COMMAND INFO <command-name> [<command-name> ...] */
+void commandInfoCommand(client *c) {
+    int i;
+    addReplyArrayLen(c, c->argc-2);
+    for (i = 2; i < c->argc; i++) {
+        addReplyCommand(c, lookupCommandBySds(c->argv[i]->ptr));
+    }
+}
+
+/* COMMAND GETKEYS arg0 arg1 arg2 ... */
+void commandGetKeysCommand(client *c) {
+    getKeysSubcommand(c);
+}
+
+/* COMMAND HELP */
+void commandHelpCommand(client *c) {
+    const char *help[] = {
 "(no subcommand)",
 "    Return details about all Redis commands.",
 "COUNT",
 "    Return the total number of commands in this Redis server.",
+"LIST",
+"    Return a list of all commands in this Redis server.",
+"INFO <command-name> [<command-name> ...]",
+"    Return details about multiple Redis commands.",
 "GETKEYS <full-command>",
 "    Return the keys from a full Redis command.",
-"INFO [<command-name> ...]",
-"    Return details about multiple Redis commands.",
 NULL
-        };
-        addReplyHelp(c, help);
-    } else if (c->argc == 1) {
-        addReplyArrayLen(c, dictSize(server.commands));
-        di = dictGetIterator(server.commands);
-        while ((de = dictNext(di)) != NULL) {
-            addReplyCommand(c, dictGetVal(de));
-        }
-        dictReleaseIterator(di);
-    } else if (!strcasecmp(c->argv[1]->ptr, "info")) {
-        int i;
-        addReplyArrayLen(c, c->argc-2);
-        for (i = 2; i < c->argc; i++) {
-            addReplyCommand(c, dictFetchValue(server.commands, c->argv[i]->ptr));
-        }
-    } else if (!strcasecmp(c->argv[1]->ptr, "count") && c->argc == 2) {
-        addReplyLongLong(c, dictSize(server.commands));
-    } else if (!strcasecmp(c->argv[1]->ptr,"getkeys") && c->argc >= 3) {
-        struct redisCommand *cmd = lookupCommand(c->argv[2]->ptr);
-        getKeysResult result = GETKEYS_RESULT_INIT;
-        int j;
-
-        if (!cmd) {
-            addReplyError(c,"Invalid command specified");
-            return;
-        } else if (cmd->getkeys_proc == NULL && cmd->key_specs_num == 0) {
-            addReplyError(c,"The command has no key arguments");
-            return;
-        } else if ((cmd->arity > 0 && cmd->arity != c->argc-2) ||
-                   ((c->argc-2) < -cmd->arity))
-        {
-            addReplyError(c,"Invalid number of arguments specified for command");
-            return;
-        }
+    };
 
-        if (!getKeysFromCommand(cmd,c->argv+2,c->argc-2,&result)) {
-            addReplyError(c,"Invalid arguments specified for command");
-        } else {
-            addReplyArrayLen(c,result.numkeys);
-            for (j = 0; j < result.numkeys; j++) addReplyBulk(c,c->argv[result.keys[j]+2]);
-        }
-        getKeysFreeResult(&result);
-    } else {
-        addReplySubcommandSyntaxError(c);
-    }
+    addReplyHelp(c, help);
 }
 
 /* Convert an amount of bytes into a human readable string in the form
@@ -5356,6 +6077,14 @@ void bytesToHuman(char *s, unsigned long long n) {
     }
 }
 
+sds getFullCommandName(struct redisCommand *cmd) {
+    if (!cmd->parent) {
+        return sdsnew(cmd->name);
+    } else {
+        return sdscatfmt(sdsempty(),"%s|%s",cmd->parent->name,cmd->name);
+    }
+}
+
 /* Characters we sanitize on INFO output to maintain expected format. */
 static char unsafe_info_chars[] = "#:\n\r";
 static char unsafe_info_chars_substs[] = "____";   /* Must be same length as above */
@@ -5375,6 +6104,35 @@ const char *getSafeInfoString(const char *s, size_t len, char **tmp) {
                        sizeof(unsafe_info_chars)-1);
 }
 
+sds genRedisInfoStringCommandStats(sds info, dict *commands) {
+    struct redisCommand *c;
+    dictEntry *de;
+    dictIterator *di;
+    di = dictGetSafeIterator(commands);
+    while((de = dictNext(di)) != NULL) {
+        char *tmpsafe;
+        c = (struct redisCommand *) dictGetVal(de);
+        if (c->calls || c->failed_calls || c->rejected_calls) {
+            sds cmdnamesds = getFullCommandName(c);
+
+            info = sdscatprintf(info,
+                "cmdstat_%s:calls=%lld,usec=%lld,usec_per_call=%.2f"
+                ",rejected_calls=%lld,failed_calls=%lld\r\n",
+                getSafeInfoString(cmdnamesds, sdslen(cmdnamesds), &tmpsafe), c->calls, c->microseconds,
+                (c->calls == 0) ? 0 : ((float)c->microseconds/c->calls),
+                c->rejected_calls, c->failed_calls);
+            if (tmpsafe != NULL) zfree(tmpsafe);
+            sdsfree(cmdnamesds);
+        }
+        if (c->subcommands_dict) {
+            info = genRedisInfoStringCommandStats(info, c->subcommands_dict);
+        }
+    }
+    dictReleaseIterator(di);
+
+    return info;
+}
+
 /* Create the string returned by the INFO command. This is decoupled
  * by the INFO command itself as we need to report the same information
  * on memory corruption problems. */
@@ -6041,25 +6799,7 @@ sds genRedisInfoString(const char *section) {
     if (allsections || !strcasecmp(section,"commandstats")) {
         if (sections++) info = sdscat(info,"\r\n");
         info = sdscatprintf(info, "# Commandstats\r\n");
-
-        struct redisCommand *c;
-        dictEntry *de;
-        dictIterator *di;
-        di = dictGetSafeIterator(server.commands);
-        while((de = dictNext(di)) != NULL) {
-            char *tmpsafe;
-            c = (struct redisCommand *) dictGetVal(de);
-            if (!c->calls && !c->failed_calls && !c->rejected_calls)
-                continue;
-            info = sdscatprintf(info,
-                "cmdstat_%s:calls=%lld,usec=%lld,usec_per_call=%.2f"
-                ",rejected_calls=%lld,failed_calls=%lld\r\n",
-                getSafeInfoString(c->name, strlen(c->name), &tmpsafe), c->calls, c->microseconds,
-                (c->calls == 0) ? 0 : ((float)c->microseconds/c->calls),
-                c->rejected_calls, c->failed_calls);
-            if (tmpsafe != NULL) zfree(tmpsafe);
-        }
-        dictReleaseIterator(di);
+        info = genRedisInfoStringCommandStats(info, server.commands);
     }
     /* Error statistics */
     if (allsections || defsections || !strcasecmp(section,"errorstats")) {
@@ -6120,6 +6860,11 @@ sds genRedisInfoString(const char *section) {
 }
 
 void infoCommand(client *c) {
+    if (server.sentinel_mode) {
+        sentinelInfoCommand(c);
+        return;
+    }
+
     char *section = c->argc == 2 ? c->argv[1]->ptr : "default";
 
     if (c->argc > 2) {
diff --git a/src/server.h b/src/server.h
index 4b6e776c5..b911697a1 100644
--- a/src/server.h
+++ b/src/server.h
@@ -233,6 +233,9 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
 #define CMD_CATEGORY_TRANSACTION (1ULL<<38)
 #define CMD_CATEGORY_SCRIPTING (1ULL<<39)
 
+#define CMD_SENTINEL (1ULL<<40)        /* "sentinel" flag */
+#define CMD_ONLY_SENTINEL (1ULL<<41)   /* "only-sentinel" flag */
+
 /* AOF states */
 #define AOF_OFF 0             /* AOF is off */
 #define AOF_ON 1              /* AOF is on */
@@ -886,20 +889,29 @@ typedef struct {
     uint64_t flags; /* See USER_FLAG_* */
 
     /* The bit in allowed_commands is set if this user has the right to
-     * execute this command. In commands having subcommands, if this bit is
-     * set, then all the subcommands are also available.
+     * execute this command.
      *
      * If the bit for a given command is NOT set and the command has
-     * subcommands, Redis will also check allowed_subcommands in order to
+     * allowed first-args, Redis will also check allowed_firstargs in order to
      * understand if the command can be executed. */
     uint64_t allowed_commands[USER_COMMAND_BITS_COUNT/64];
 
-    /* This array points, for each command ID (corresponding to the command
+    /* NOTE: allowed_firstargs is a transformation of the old mechanism for allowing
+     * subcommands (now, subcommands are actually commands, with their own
+     * ACL ID)
+     * We had to keep allowed_firstargs (previously called allowed_subcommands)
+     * in order to support the widespread abuse of ACL rules to block a command
+     * with a specific argv[1] (which is not a subcommand at all).
+     * For example, a user can use the rule "-select +select|0" to block all
+     * SELECT commands, except "SELECT 0".
+     * It can also be applied for subcommands: "+config -config|set +config|set|loglevel"
+     *
+     * This array points, for each command ID (corresponding to the command
      * bit set in allowed_commands), to an array of SDS strings, terminated by
-     * a NULL pointer, with all the sub commands that can be executed for
-     * this command. When no subcommands matching is used, the field is just
+     * a NULL pointer, with all the first-args that are allowed for
+     * this command. When no first-arg matching is used, the field is just
      * set to NULL to avoid allocating USER_COMMAND_BITS_COUNT pointers. */
-    sds **allowed_subcommands;
+    sds **allowed_firstargs;
     list *passwords; /* A list of SDS valid passwords for this user. */
     list *patterns;  /* A list of allowed key patterns. If this field is NULL
                         the user cannot mention any key in a command, unless
@@ -1824,8 +1836,10 @@ struct redisCommand {
     char *sflags;   /* Flags as string representation, one char per flag. */
     keySpec key_specs_static[STATIC_KEY_SPECS_NUM];
     /* Use a function to determine keys arguments in a command line.
-     * Used for Redis Cluster redirect. */
+     * Used for Redis Cluster redirect (may be NULL) */
     redisGetKeysProc *getkeys_proc;
+    /* Array of subcommands (may be NULL) */
+    struct redisCommand *subcommands;
 
     /* Runtime data */
     uint64_t flags; /* The actual flags, obtained from the 'sflags' field. */
@@ -1844,6 +1858,8 @@ struct redisCommand {
     int key_specs_num;
     int key_specs_max;
     int movablekeys; /* See populateCommandMovableKeys */
+    dict *subcommands_dict;
+    struct redisCommand *parent;
 };
 
 struct redisError {
@@ -1980,6 +1996,8 @@ robj *moduleTypeDupOrReply(client *c, robj *fromkey, robj *tokey, int todb, robj
 int moduleDefragValue(robj *key, robj *obj, long *defragged, int dbid);
 int moduleLateDefrag(robj *key, robj *value, unsigned long *cursor, long long endtime, long long *defragged, int dbid);
 long moduleDefragGlobals(void);
+void *moduleGetHandleByName(char *modulename);
+int moduleIsModuleCommand(void *module_handle, struct redisCommand *cmd);
 
 /* Utils */
 long long ustime(void);
@@ -2420,9 +2438,12 @@ void removeSignalHandlers(void);
 int createSocketAcceptHandler(socketFds *sfd, aeFileProc *accept_handler);
 int changeListenPort(int port, socketFds *sfd, aeFileProc *accept_handler);
 int changeBindAddr(sds *addrlist, int addrlist_len);
-struct redisCommand *lookupCommand(sds name);
+struct redisCommand *lookupCommand(robj **argv ,int argc);
+struct redisCommand *lookupCommandBySdsLogic(dict *commands, sds s);
+struct redisCommand *lookupCommandBySds(sds s);
+struct redisCommand *lookupCommandByCStringLogic(dict *commands, const char *s);
 struct redisCommand *lookupCommandByCString(const char *s);
-struct redisCommand *lookupCommandOrOriginal(sds name);
+struct redisCommand *lookupCommandOrOriginal(robj **argv ,int argc);
 void call(client *c, int flags);
 void propagate(int dbid, robj **argv, int argc, int flags);
 void alsoPropagate(int dbid, robj **argv, int argc, int target);
@@ -2448,7 +2469,7 @@ void usage(void);
 void updateDictResizePolicy(void);
 int htNeedsResize(dict *dict);
 void populateCommandTable(void);
-void resetCommandTableStats(void);
+void resetCommandTableStats(dict* commands);
 void resetErrorTableStats(void);
 void adjustOpenFilesLimit(void);
 void incrementErrorCount(const char *fullerr, size_t namelen);
@@ -2606,7 +2627,6 @@ int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *
 int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
-int memoryGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int lcsGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
 int blmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
@@ -2624,6 +2644,10 @@ void queueSentinelConfig(sds *argv, int argc, int linenum, sds line);
 void loadSentinelConfigFromQueue(void);
 void sentinelIsRunning(void);
 void sentinelCheckConfigFile(void);
+void sentinelCommand(client *c);
+void sentinelInfoCommand(client *c);
+void sentinelPublishCommand(client *c);
+void sentinelRoleCommand(client *c);
 
 /* redis-check-rdb & aof */
 int redis_check_rdb(char *rdbfilename, FILE *fp);
@@ -2694,6 +2718,11 @@ void authCommand(client *c);
 void pingCommand(client *c);
 void echoCommand(client *c);
 void commandCommand(client *c);
+void commandCountCommand(client *c);
+void commandListCommand(client *c);
+void commandInfoCommand(client *c);
+void commandGetKeysCommand(client *c);
+void commandHelpCommand(client *c);
 void setCommand(client *c);
 void setnxCommand(client *c);
 void setexCommand(client *c);
@@ -2846,7 +2875,11 @@ void hgetallCommand(client *c);
 void hexistsCommand(client *c);
 void hscanCommand(client *c);
 void hrandfieldCommand(client *c);
-void configCommand(client *c);
+void configSetCommand(client *c);
+void configGetCommand(client *c);
+void configResetStatCommand(client *c);
+void configRewriteCommand(client *c);
+void configHelpCommand(client *c);
 void hincrbyCommand(client *c);
 void hincrbyfloatCommand(client *c);
 void subscribeCommand(client *c);
@@ -2937,6 +2970,7 @@ void _serverPanic(const char *file, int line, const char *msg, ...);
 #endif
 void serverLogObjectDebugInfo(const robj *o);
 void sigsegvHandler(int sig, siginfo_t *info, void *secret);
+sds getFullCommandName(struct redisCommand *cmd);
 const char *getSafeInfoString(const char *s, size_t len, char **tmp);
 sds genRedisInfoString(const char *section);
 sds genModulesInfoString(sds info);
@@ -2948,6 +2982,7 @@ int memtest_preserving_test(unsigned long *m, size_t bytes, int passes);
 void mixDigest(unsigned char *digest, void *ptr, size_t len);
 void xorDigest(unsigned char *digest, void *ptr, size_t len);
 int populateSingleCommand(struct redisCommand *c, char *strflags);
+void commandAddSubcommand(struct redisCommand *parent, struct redisCommand *subcommand);
 void populateCommandMovableKeys(struct redisCommand *cmd);
 void debugDelay(int usec);
 void killIOThreads(void);
diff --git a/src/t_string.c b/src/t_string.c
index fd938a9ec..f1c1fdca2 100644
--- a/src/t_string.c
+++ b/src/t_string.c
@@ -724,11 +724,20 @@ void strlenCommand(client *c) {
  * STRALGO <algorithm> ... arguments ... */
 void stralgoLCS(client *c);     /* This implements the LCS algorithm. */
 void stralgoCommand(client *c) {
-    /* Select the algorithm. */
-    if (!strcasecmp(c->argv[1]->ptr,"lcs")) {
+    char *subcmd = c->argv[1]->ptr;
+
+    if (c->argc == 2 && !strcasecmp(subcmd,"help")) {
+        const char *help[] = {
+"LCS",
+"    Run the longest common subsequence algorithm.",
+NULL
+        };
+        addReplyHelp(c, help);
+    } else if (!strcasecmp(subcmd,"lcs")) {
         stralgoLCS(c);
     } else {
-        addReplyErrorObject(c,shared.syntaxerr);
+        addReplySubcommandSyntaxError(c);
+        return;
     }
 }